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

[pan3793]

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:

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

[dongjoon-hyun]

Thanks for working on this — the underlying problem is real, and defaulting the config to false keeps this safe. A few comments.

1. Part of this can be fixed without a config

cachedName has only two consumers:

• InMemoryTableScanExec#nodeName — but only in the case Some(_) => branch, i.e. named tables only.
• CachedRDDBuilder#buildBuffers → cached.setName(cachedName).

So for anonymous caches, cachedName feeds exactly one thing: the RDD display name in the Storage tab. Yet it is a val in the case class body, so Utils.abbreviate(cachedPlan.toString, 1024) is evaluated at every CachedRDDBuilder construction — including caches that are never materialized (e.g. df.cache() that is never triggered, or a plan AQE ends up not using).

Making it lazy val removes that cost unconditionally, with no config and no behavior change:

lazy val cachedName: String = tableName.map(n => s"In-memory table $n").getOrElse { ... }

One thing to confirm if you take this: cachedPlan is @transient, so a lazy val forced after deserialization would NPE. Both consumers above look driver-side (setName, and override val nodeName which is itself eager), so it seems safe — but worth stating explicitly.

This would also narrow what the new config has to justify, down to "large anonymous caches that are materialized".

2. Off-by-one between the doc and the behavior

private val _nextCachedRDDId = new AtomicLong(0)
def nextCachedRDDId(): Long = _nextCachedRDDId.getAndIncrement

AtomicLong(0) + getAndIncrement means the first name is CachedRDD 0, but both the config doc and the PR description say 'CachedRDD 1'. Either use incrementAndGet or fix the doc.

Minor: the closest precedent in this area is SparkPlan.newPlanId():

private val nextPlanId = new AtomicInteger(0)
private[execution] def newPlanId(): Int = nextPlanId.getAndIncrement()

private val nextId reads better here than the underscore-prefixed name.

3. The config should be .internal(), and the name is off-convention

Since the only observable effect for anonymous caches is an RDD display name, this is a debugging/tuning knob — .internal() seems right. As written it will show up in the public SQL config docs table.

The name also doesn't match its neighbors in SQLConf, which is where it is (correctly) placed:

• spark.sql.defaultCacheStorageLevel
• spark.sql.dataframeCache.logLevel ← directly above the new entry
• spark.sql.useSequentialCacheName ← new

Something like spark.sql.dataframeCache.sequentialName.enabled would be more consistent, and follows the usual .enabled suffix for boolean confs.

4. Prefer cachedPlan.conf over cachedPlan.session.conf

if (cachedPlan.session.conf.get(SQLConf.USE_SEQUENTIAL_CACHE_NAME)) {

SparkPlan.conf already resolves to session.sessionState.conf when a session is active and falls back to SQLConf.get otherwise, and it is what this very file uses a few lines down (cachedPlan.conf.clone() in buildBuffers):

if (cachedPlan.conf.getConf(SQLConf.USE_SEQUENTIAL_CACHE_NAME)) {

(cachedPlan.session is getActiveSession.orNull, so the current form NPEs on a null session. newPartitionStats() already assumes non-null, so this isn't a new risk — but no reason to add another one.)

5. Side-effecting val in a case class body

CachedRDDBuilder is a case class, and cachedName now increments a global counter as a side effect of construction. Any future copy(...) would silently change the name and burn an id. There are no copy call sites on the builder today (InMemoryRelation.copy() shares the builder reference), so this is latent — but a short comment would help.

For what it's worth, the equality side is fine: cachedName is a body val, not a constructor param, so equals/hashCode/canonicalization are unaffected and sameResult / plan reuse can't be perturbed by this.

6. Test

• The other two tests in InMemoryRelationSuite are prefixed (SPARK-46779:, SPARK-47177:); please add SPARK-59024: for consistency.
• The disabled-config assertion is weak:

  assert(!r4.cacheBuilder.cachedName.startsWith("CachedRDD "))

  This only checks it isn't the new format, not that it is the abbreviated plan tree string. Asserting equality against Utils.abbreviate(...) (or at least that it starts with the plan's first line) would actually protect the fallback path.

Nits confirmed OK

• .version("4.4.0") matches branch-4.x, which is right for a normally-backported change.
• Placement inside the cache-related config cluster in SQLConf is good.
• Default false means no golden-file or explain-output impact.

[pan3793]

Thanks for the review!

1. Adopted. cachedName is now a lazy val. Both consumers are driver-side, and cacheBuilder is @transient in InMemoryRelation, so the builder is never deserialized on executors.

2. Kept getAndIncrement() -- the first name is CachedRDD 0 -- and fixed the doc and PR description instead. Also renamed the generator to follow the SparkPlan.newPlanId() precedent (nextCachedRDDId / newCachedRDDId()).

3. Made the conf internal, renamed it to spark.sql.dataframeCache.sequentialName.enabled, and added .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) -- the name only affects display. This was also behind the CI failure in SparkConfigBindingPolicySuite.

4. Adopted: cachedPlan.conf.getConf(...).

5. Added a short comment noting the sequential id is consumed when the lazy val is first forced.

6. Added the SPARK-59024: prefix; the disabled case now asserts equality against Utils.abbreviate(cachedPlan.toString, 1024).

[ulysses-you]

lgtm

[dongjoon-hyun]

Thanks for the quick turnaround — all six items from the previous round are addressed, and CI is fully green on fcf6236. NOT_APPLICABLE is the right binding policy here too: the test is whether the config changes what a view/UDF body resolves to, and this one only affects a display name.

Two follow-ups, one of which is a correction of something I said last round.

1. Correction: the lazy val does change the default-path name

I said making cachedName lazy has "no behavior change". That is not quite right, and it is worth a line in the PR description.

cached.setName(cachedName) runs after the plan has been executed:

val cb = try {
  ... serializer.convertInternalRowToCachedBatch(buildInputRDD(cachedPlan.execute()), ...)
  ...
}
val cached = cb.mapPartitionsWithIndexInternal { ... }.persist(storageLevel)
cached.setName(cachedName)   // <- forced here, post-execution

For an AQE plan, AdaptiveSparkPlanExec.generateTreeString renders the initial plan while isFinalPlan=false, and switches to a Final Plan section once execution has completed. So with the config off (the default), the Storage-tab name of an anonymous cache is now derived from the final AQE plan rather than the pre-execution plan it used to show.

Harmless in itself — it is a display string, and arguably the more useful one — but it does mean "Does this PR introduce any user-facing change? No" is slightly overstated. Suggest noting that the anonymous cached name is now rendered at materialization time.

2. Same for the config read

Because the conf is read inside the lazy val, it is evaluated at first materialization, not at cache() time:

df.cache()                                   // config = false
spark.conf.set("spark.sql.dataframeCache.sequentialName.enabled", "true")
df.count()                                   // -> "CachedRDD 7"

That seems fine (and is the only sensible option given the laziness), but a short sentence in the conf doc or the comment would save the next reader the trip.

3. Optional: SparkPlan.id already gives you a sequential id

SparkPlan carries a globally unique, monotonically increasing id:

// SparkPlan.scala
private val nextPlanId = new AtomicInteger(0)
private[execution] def newPlanId(): Int = nextPlanId.getAndIncrement()
...
val id: Int = SparkPlan.newPlanId()

So the sequential branch could be just:

s"CachedRDD ${cachedPlan.id}"

which drops the CachedRDDBuilder companion object, the extra AtomicLong, and the "a copy of the builder would draw a new id" caveat entirely — the id then belongs to the plan, not to the forcing of a lazy val.

The trade-off is that the numbers are no longer dense (they interleave with every other physical operator's id), so CachedRDD 0, CachedRDD 41, CachedRDD 97 instead of 0, 1, 2. If the dense numbering is what you want for readability, the current form is fine — your call.

4. Optional: the test does not cover the actual win

The new test asserts the value of cachedName, but the reason for this PR is that the tree string is never rendered for a cache that is never materialized. That is cheap to pin down with a wrapper plan that records whether toString was called:

val plan = new PlanWithCountedToString(child)   // increments a counter in toString
val r = InMemoryRelation(StorageLevel.MEMORY_ONLY, ..., None)
assert(plan.toStringCount == 0)                // not rendered at construction

Otherwise a future change that re-eagerizes the name would pass the suite.

Nothing blocking. With (1) reflected in the description this LGTM.

[dongjoon-hyun]

+1, LGTM except the above (1).

[pan3793]

Thanks for the second pass!

1. Correct -- noted in the PR description: the anonymous cached name is now rendered at materialization time, so for adaptive plans it is derived from the final AQE plan; the user-facing section is adjusted accordingly.

2. Added to the config doc: "The name is resolved when the cache is first materialized."

3. Switched to cachedPlan.id, with a self-describing name format: CachedRDD (plan_id=42). This drops the companion object, the AtomicLong, and the id-consumption caveat; the config is renamed to spark.sql.dataframeCache.planIdName.enabled.

4. Added SPARK-59024: anonymous cached name is not rendered before materialization -- a ToStringCountingPlan leaf asserts toString is never called at cache construction.

[dongjoon-hyun]

Thanks, this addresses everything from the last round, and CI is green on 57578fe.

The diff is now just a lazy val plus a three-line branch -- no companion object, no extra AtomicLong, no id-consumption caveat. The CachedRDD (plan_id=42) format also lines up with Exchange.stringArgs, which renders the same SparkPlan.id as [plan_id=$id].

Four comments, all non-blocking -- feel free to fold them into a follow-up or address them here.

1. plan_id is an overloaded term

Spark uses plan_id for two unrelated things:

• SparkPlan.id, rendered as [plan_id=N] by Exchange and EmptyRelationExec -- what this PR uses.
• LogicalPlan.PLAN_ID_TAG = TreeNodeTag[Long]("plan_id"), the Spark Connect plan id used for column resolution in ColumnResolutionHelper -- a completely separate numbering space.

On top of that, the physical [plan_id=N] suffix is only printed for Exchange / EmptyRelationExec, so the root of a cached plan never shows its id in explain() output. The name therefore suggests a cross-reference that a user cannot actually make.

CachedRDD 42 would carry the same information without the ambiguity. Since I am the one who suggested cachedPlan.id, I brought the ambiguity in -- your call whether the explicit label is worth it.

2. Test comment, and a behavior nuance worth knowing

// Caches of the same plan share the plan id.
val r1Again = InMemoryRelation(StorageLevel.MEMORY_ONLY, d.queryExecution, None)

This is the same executedPlan instance (both come from d.queryExecution), not merely the same plan shape. Two separately built but structurally identical plans get different ids -- and those used to share a name under the tree-string scheme, since the string was identical.

Display-only, and arguably better (distinct caches, distinct names), but "the same physical plan instance" would describe what the test actually pins.

3. The new comment does not hold for named caches

// Resolved on first access, which happens at cache materialization; for adaptive plans the
// name therefore reflects the final plan.

InMemoryTableScanExec#nodeName is an override val and reads cachedName in the case Some(_) => branch, so for a named cache the lazy val is forced at scan-exec construction, well before materialization. Something like "first access (cache materialization for anonymous caches)" would be accurate.

4. Optional: pin the other direction too

ToStringCountingPlan currently asserts only that nothing is rendered:

assert(plan.toStringCount == 0)

Forcing the name and asserting toStringCount == 1 would also lock in that the fallback path really does render the tree string, so a future change cannot quietly break either half.

LGTM.

[dongjoon-hyun]

Thank you, @pan3793 . Also, please resolve the merge conflicts.

[pan3793]

Thanks for the third pass!

1. Keeping the explicit CachedRDD (plan_id=N) label -- the config is internal, and the id remains a useful debugging handle even if explain does not show a cross-reference for every plan.

2. Reworded: "Caches of the same physical plan instance share the plan id."

3. Reworded the comment: "Resolved on first access (cache materialization for anonymous caches)."

4. Added the reverse assertion: forcing the name renders the tree string exactly once.

Also rebased onto master and resolved the conflict with the SPARK-59009 test.