Skip to content

Commit af0d5c3

Browse files
committed
[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 <pan3793@gmail.com> Signed-off-by: Cheng Pan <chengpan@apache.org>
1 parent f54433a commit af0d5c3

3 files changed

Lines changed: 77 additions & 4 deletions

File tree

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

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2647,6 +2647,18 @@ object SQLConf {
26472647
.enumConf(classOf[Level])
26482648
.createWithDefault(Level.TRACE)
26492649

2650+
val DATAFRAME_CACHE_PLAN_ID_NAME_ENABLED =
2651+
buildConf("spark.sql.dataframeCache.planIdName.enabled")
2652+
.internal()
2653+
.doc("When true and the cached table has no name, use the physical plan id, e.g. " +
2654+
"'CachedRDD (plan_id=42)', as the cached name instead of the abbreviated plan tree " +
2655+
"string. Rendering the plan tree string can be expensive for large plans. The name " +
2656+
"is resolved when the cache is first materialized.")
2657+
.version("4.4.0")
2658+
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
2659+
.booleanConf
2660+
.createWithDefault(false)
2661+
26502662
val DROP_TABLE_VIEW_ENABLED =
26512663
buildConf("spark.sql.dropTableOnView.enabled")
26522664
.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

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -289,8 +289,15 @@ case class CachedRDDBuilder(
289289
// late updates from making a rebuilt cache appear complete.
290290
private var partitionStats = newPartitionStats()
291291

292-
val cachedName = tableName.map(n => s"In-memory table $n")
293-
.getOrElse(Utils.abbreviate(cachedPlan.toString, 1024))
292+
// Resolved on first access (cache materialization for anonymous caches). For adaptive plans,
293+
// the name reflects the final plan.
294+
lazy val cachedName: String = tableName.map(n => s"In-memory table $n").getOrElse {
295+
if (cachedPlan.conf.getConf(SQLConf.DATAFRAME_CACHE_PLAN_ID_NAME_ENABLED)) {
296+
s"CachedRDD (plan_id=${cachedPlan.id})"
297+
} else {
298+
Utils.abbreviate(cachedPlan.toString, 1024)
299+
}
300+
}
294301

295302
val supportsColumnarInput: Boolean = {
296303
cachedPlan.supportsColumnar &&

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

Lines changed: 56 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,13 +17,19 @@
1717

1818
package org.apache.spark.sql.execution.columnar
1919

20+
import java.util.concurrent.atomic.AtomicInteger
21+
2022
import org.apache.spark.SparkFunSuite
21-
import org.apache.spark.sql.catalyst.expressions.AttributeSet
22-
import org.apache.spark.sql.execution.SparkPlan
23+
import org.apache.spark.rdd.RDD
24+
import org.apache.spark.sql.catalyst.InternalRow
25+
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet}
26+
import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan}
2327
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
2428
import org.apache.spark.sql.functions.expr
29+
import org.apache.spark.sql.internal.SQLConf
2530
import org.apache.spark.sql.test.SharedSparkSessionBase
2631
import org.apache.spark.storage.StorageLevel
32+
import org.apache.spark.util.Utils
2733

2834
class InMemoryRelationSuite extends SparkFunSuite
2935
with SharedSparkSessionBase with AdaptiveSparkPlanHelper {
@@ -51,6 +57,38 @@ class InMemoryRelationSuite extends SparkFunSuite
5157
assert(r1.sameResult(r2))
5258
}
5359

60+
test("SPARK-59024: plan id cached name for anonymous cached tables") {
61+
val d = spark.range(1)
62+
withSQLConf(SQLConf.DATAFRAME_CACHE_PLAN_ID_NAME_ENABLED.key -> "true") {
63+
val r1 = InMemoryRelation(StorageLevel.MEMORY_ONLY, d.queryExecution, None)
64+
// Caches of the same physical plan instance share the plan id.
65+
val r1Again = InMemoryRelation(StorageLevel.MEMORY_ONLY, d.queryExecution, None)
66+
val r2 = InMemoryRelation(StorageLevel.MEMORY_ONLY, spark.range(2).queryExecution, None)
67+
assert(r1.cacheBuilder.cachedName.matches("CachedRDD \\(plan_id=\\d+\\)"))
68+
assert(r1Again.cacheBuilder.cachedName == r1.cacheBuilder.cachedName)
69+
assert(r1.cacheBuilder.cachedName != r2.cacheBuilder.cachedName)
70+
// Named tables keep the usual name.
71+
val r3 = InMemoryRelation(StorageLevel.MEMORY_ONLY, d.queryExecution, Some("t1"))
72+
assert(r3.cacheBuilder.cachedName == "In-memory table t1")
73+
}
74+
// When disabled, the cached name keeps the abbreviated plan tree string.
75+
withSQLConf(SQLConf.DATAFRAME_CACHE_PLAN_ID_NAME_ENABLED.key -> "false") {
76+
val r4 = InMemoryRelation(StorageLevel.MEMORY_ONLY, d.queryExecution, None)
77+
assert(r4.cacheBuilder.cachedName ==
78+
Utils.abbreviate(r4.cacheBuilder.cachedPlan.toString, 1024))
79+
}
80+
}
81+
82+
test("SPARK-59024: anonymous cached name is not rendered before materialization") {
83+
val plan = ToStringCountingPlan()
84+
val relation = InMemoryRelation(new DefaultCachedBatchSerializer, StorageLevel.MEMORY_ONLY,
85+
plan, None, spark.range(1).queryExecution.optimizedPlan)
86+
assert(plan.toStringCount == 0)
87+
// Forcing the name renders the tree string exactly once.
88+
relation.cacheBuilder.cachedName
89+
assert(plan.toStringCount == 1)
90+
}
91+
5492
test("SPARK-47177: Cached SQL plan do not display final AQE plan in explain string") {
5593
def findIMRInnerChild(p: SparkPlan): SparkPlan = {
5694
val tableCache = find(p) {
@@ -74,3 +112,19 @@ class InMemoryRelationSuite extends SparkFunSuite
74112
.contains("AdaptiveSparkPlan isFinalPlan=true"))
75113
}
76114
}
115+
116+
case class ToStringCountingPlan() extends LeafExecNode {
117+
private val _toStringCount = new AtomicInteger(0)
118+
119+
def toStringCount: Int = _toStringCount.get()
120+
121+
override def output: Seq[Attribute] = Seq.empty
122+
123+
override protected def doExecute(): RDD[InternalRow] =
124+
throw new UnsupportedOperationException
125+
126+
override def toString: String = {
127+
_toStringCount.incrementAndGet()
128+
"ToStringCountingPlan"
129+
}
130+
}

0 commit comments

Comments
 (0)