diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index 9398d14eb94e8..2e866508c8675 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala @@ -2647,6 +2647,18 @@ object SQLConf { .enumConf(classOf[Level]) .createWithDefault(Level.TRACE) + val DATAFRAME_CACHE_PLAN_ID_NAME_ENABLED = + buildConf("spark.sql.dataframeCache.planIdName.enabled") + .internal() + .doc("When true and the cached table has no name, use the physical plan id, e.g. " + + "'CachedRDD (plan_id=42)', as the cached name instead of the abbreviated plan tree " + + "string. Rendering the plan tree string can be expensive for large plans. The name " + + "is resolved when the cache is first materialized.") + .version("4.4.0") + .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) + .booleanConf + .createWithDefault(false) + val DROP_TABLE_VIEW_ENABLED = buildConf("spark.sql.dropTableOnView.enabled") .doc("When true, DROP TABLE command will work on VIEW as well.") diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala index 584eaa9ecbc82..3de3e41e110bc 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala @@ -289,8 +289,15 @@ case class CachedRDDBuilder( // late updates from making a rebuilt cache appear complete. private var partitionStats = newPartitionStats() - val cachedName = tableName.map(n => s"In-memory table $n") - .getOrElse(Utils.abbreviate(cachedPlan.toString, 1024)) + // Resolved on first access (cache materialization for anonymous caches). For adaptive plans, + // the name reflects the final plan. + lazy val cachedName: String = tableName.map(n => s"In-memory table $n").getOrElse { + if (cachedPlan.conf.getConf(SQLConf.DATAFRAME_CACHE_PLAN_ID_NAME_ENABLED)) { + s"CachedRDD (plan_id=${cachedPlan.id})" + } else { + Utils.abbreviate(cachedPlan.toString, 1024) + } + } val supportsColumnarInput: Boolean = { cachedPlan.supportsColumnar && diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryRelationSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryRelationSuite.scala index 6a9c8964e4dbf..706e81d8588cd 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryRelationSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryRelationSuite.scala @@ -17,13 +17,19 @@ package org.apache.spark.sql.execution.columnar +import java.util.concurrent.atomic.AtomicInteger + import org.apache.spark.SparkFunSuite -import org.apache.spark.sql.catalyst.expressions.AttributeSet -import org.apache.spark.sql.execution.SparkPlan +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet} +import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan} import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.functions.expr +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSparkSessionBase import org.apache.spark.storage.StorageLevel +import org.apache.spark.util.Utils class InMemoryRelationSuite extends SparkFunSuite with SharedSparkSessionBase with AdaptiveSparkPlanHelper { @@ -51,6 +57,38 @@ class InMemoryRelationSuite extends SparkFunSuite assert(r1.sameResult(r2)) } + test("SPARK-59024: plan id cached name for anonymous cached tables") { + val d = spark.range(1) + withSQLConf(SQLConf.DATAFRAME_CACHE_PLAN_ID_NAME_ENABLED.key -> "true") { + val r1 = InMemoryRelation(StorageLevel.MEMORY_ONLY, d.queryExecution, None) + // Caches of the same physical plan instance share the plan id. + val r1Again = InMemoryRelation(StorageLevel.MEMORY_ONLY, d.queryExecution, None) + val r2 = InMemoryRelation(StorageLevel.MEMORY_ONLY, spark.range(2).queryExecution, None) + assert(r1.cacheBuilder.cachedName.matches("CachedRDD \\(plan_id=\\d+\\)")) + assert(r1Again.cacheBuilder.cachedName == r1.cacheBuilder.cachedName) + assert(r1.cacheBuilder.cachedName != r2.cacheBuilder.cachedName) + // Named tables keep the usual name. + val r3 = InMemoryRelation(StorageLevel.MEMORY_ONLY, d.queryExecution, Some("t1")) + assert(r3.cacheBuilder.cachedName == "In-memory table t1") + } + // When disabled, the cached name keeps the abbreviated plan tree string. + withSQLConf(SQLConf.DATAFRAME_CACHE_PLAN_ID_NAME_ENABLED.key -> "false") { + val r4 = InMemoryRelation(StorageLevel.MEMORY_ONLY, d.queryExecution, None) + assert(r4.cacheBuilder.cachedName == + Utils.abbreviate(r4.cacheBuilder.cachedPlan.toString, 1024)) + } + } + + test("SPARK-59024: anonymous cached name is not rendered before materialization") { + val plan = ToStringCountingPlan() + val relation = InMemoryRelation(new DefaultCachedBatchSerializer, StorageLevel.MEMORY_ONLY, + plan, None, spark.range(1).queryExecution.optimizedPlan) + assert(plan.toStringCount == 0) + // Forcing the name renders the tree string exactly once. + relation.cacheBuilder.cachedName + assert(plan.toStringCount == 1) + } + test("SPARK-47177: Cached SQL plan do not display final AQE plan in explain string") { def findIMRInnerChild(p: SparkPlan): SparkPlan = { val tableCache = find(p) { @@ -74,3 +112,19 @@ class InMemoryRelationSuite extends SparkFunSuite .contains("AdaptiveSparkPlan isFinalPlan=true")) } } + +case class ToStringCountingPlan() extends LeafExecNode { + private val _toStringCount = new AtomicInteger(0) + + def toStringCount: Int = _toStringCount.get() + + override def output: Seq[Attribute] = Seq.empty + + override protected def doExecute(): RDD[InternalRow] = + throw new UnsupportedOperationException + + override def toString: String = { + _toStringCount.incrementAndGet() + "ToStringCountingPlan" + } +}