[SPARK-59024][SQL] Use the physical plan id as the cached name for anonymous cached tables

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -2645,6 +2645,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.")

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, which happens at cache materialization; for adaptive plans the
+  // name therefore 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 &&

sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryRelationSuite.scala

@@ -17,12 +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.execution.SparkPlan
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+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 {
@@ -34,6 +41,35 @@ 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 plan 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()
+    InMemoryRelation(new DefaultCachedBatchSerializer, StorageLevel.MEMORY_ONLY, plan, None,
+      spark.range(1).queryExecution.optimizedPlan)
+    assert(plan.toStringCount == 0)
+  }
+
   test("SPARK-47177: Cached SQL plan do not display final AQE plan in explain string") {
     def findIMRInnerChild(p: SparkPlan): SparkPlan = {
       val tableCache = find(p) {
@@ -57,3 +93,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"
+  }
+}