This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 7659eb8b0883 [SPARK-59024][SQL] Use the physical plan id as the cached
name for anonymous cached tables
7659eb8b0883 is described below
commit 7659eb8b0883482a2bbcf4e53891f8d9e601cd69
Author: Cheng Pan <[email protected]>
AuthorDate: Mon Aug 31 09:33:06 2026 +0800
[SPARK-59024][SQL] Use the physical plan id as the cached name for
anonymous cached tables
### What changes were proposed in this pull request?
Add an internal SQL config `spark.sql.dataframeCache.planIdName.enabled`
(default `false`). When it is true and the cached table has no name,
`CachedRDDBuilder` uses the physical plan id, e.g. `CachedRDD (plan_id=42)`, as
the cached name instead of the abbreviated plan tree string. `cachedName` is
also made a `lazy val`, so the name is only computed when it is actually needed:
```scala
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)
}
}
```
### Why are the changes needed?
For anonymous cached tables, the cached name was built from the plan's tree
string (`cachedPlan.toString`, abbreviated to 1024 chars) at `CachedRDDBuilder`
construction time, even for caches that are never materialized. Rendering the
plan tree string can be expensive for large plans, and the name is only used
for display.
This is another spot, besides the SQL event plan description addressed in
SPARK-59023, that hurts the same customer job: it constructs a huge plan whose
`treeString` exceeds 280,000 lines, and rendering the plan tree string takes
minutes per iteration and contributes to driver OOM.
### Does this PR introduce _any_ user-facing change?
The new config is internal and defaults to `false`. One nuance: because
`cachedName` is now evaluated lazily, the name of anonymous caches is rendered
at materialization time, so for adaptive plans the Storage tab shows a name
derived from the final AQE plan instead of the pre-execution plan. No other
behavior changes.
### How was this patch tested?
New unit tests in `InMemoryRelationSuite`:
- `SPARK-59024: plan id cached name for anonymous cached tables` --
verifies the `CachedRDD (plan_id=<id>)` format, that caches of the same plan
share the name, that distinct plans get distinct names, that named tables keep
the `In-memory table <name>` name, and that the abbreviated plan tree string is
kept when the config is disabled
- `SPARK-59024: anonymous cached name is not rendered before
materialization` -- verifies the plan tree string is not rendered at cache
construction
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Qwen3.8 Max
Closes #58314 from pan3793/SPARK-59024.
Authored-by: Cheng Pan <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
(cherry picked from commit af0d5c3abd19ba4bc244da174d3df257d582f970)
Signed-off-by: Cheng Pan <[email protected]>
---
.../org/apache/spark/sql/internal/SQLConf.scala | 12 +++++
.../sql/execution/columnar/InMemoryRelation.scala | 11 +++-
.../execution/columnar/InMemoryRelationSuite.scala | 58 +++++++++++++++++++++-
3 files changed, 77 insertions(+), 4 deletions(-)
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 9f92028ea8a2..3bdeae6ee7a5 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 584eaa9ecbc8..3de3e41e110b 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 6a9c8964e4db..706e81d8588c 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"
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
