This is an automated email from the ASF dual-hosted git repository.
peter-toth 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 0b67dc215b67 [SPARK-59009][SQL] Re-map outputOrdering in
InMemoryRelation.newInstance()
0b67dc215b67 is described below
commit 0b67dc215b67e2ee76d21305b7582a1a62d04b3f
Author: James Willis <[email protected]>
AuthorDate: Fri Aug 28 12:37:27 2026 +0200
[SPARK-59009][SQL] Re-map outputOrdering in InMemoryRelation.newInstance()
### What changes were proposed in this pull request?
`InMemoryRelation.newInstance()` now goes through `withOutput`, so that
`outputOrdering` is re-mapped onto the freshly-instantiated attributes instead
of being carried over unchanged.
`withOutput` also re-keys `statsOfPlanToCache` onto the new attributes via
`LogicalRDD.rewriteStatistics`. It previously passed the stats through
unchanged, so `Statistics.attributeStats` stayed keyed by the old attributes
and every column-stat lookup missed on the new relation, silently dropping CBO
estimates back to the un-filtered defaults. This matches what
`LogicalRDD.newInstance()` already does. Thanks to peter-toth for catching it.
### Why are the changes needed?
`InMemoryRelation` has an implicit invariant that `outputOrdering` may only
reference attributes present in `output`. `newInstance()` violates it: it gives
`output` fresh exprIds but passes `outputOrdering` through unchanged, so the
returned relation's ordering still points at the old attributes.
That was harmless until
[SPARK-53738](https://issues.apache.org/jira/browse/SPARK-53738), which routed
`doCanonicalize` through `withOutput` and made `withOutput` re-map the ordering
with a strict `AttributeMap` lookup. Since then, any `InMemoryRelation` that
has been through `newInstance()` fails as soon as anything canonicalizes it:
```
java.util.NoSuchElementException: key not found: k#1L
at scala.collection.MapOps.default(Map.scala:289)
at
org.apache.spark.sql.catalyst.expressions.AttributeMap.apply(AttributeMap.scala:41)
at
org.apache.spark.sql.execution.columnar.InMemoryRelation.$anonfun$withOutput$1(InMemoryRelation.scala:711)
at
org.apache.spark.sql.execution.columnar.InMemoryRelation.withOutput(InMemoryRelation.scala:711)
at
org.apache.spark.sql.execution.columnar.InMemoryRelation.doCanonicalize(InMemoryRelation.scala:672)
...
at
org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.createNonResultQueryStages(AdaptiveSparkPlanExec.scala:589)
```
This is reachable from ordinary SQL. Cache substitution
(`CacheManager.useCachedData`) runs on the analyzed plan, before the optimizer.
`InlineCTE` then inlines a CTE that is referenced more than once and, to get a
fresh-exprId copy, runs `DeduplicateRelations` over a synthetic self-join. By
that point the plan already contains the `InMemoryRelation`, which is a
`MultiInstanceRelation`, so `newInstance()` is called on it.
A user-facing repro — a persisted DataFrame with a global `ORDER BY`,
window-ranked and then self-joined:
```python
spark.range(0, 20).selectExpr("id", "id % 3 AS
k").createOrReplaceTempView("t")
b = spark.sql("SELECT id, k FROM t ORDER BY k, id")
b.persist()
b.count()
b.createOrReplaceTempView("b")
spark.sql("""
WITH r AS (SELECT *, row_number() OVER (PARTITION BY k ORDER BY id DESC)
AS rn FROM b)
SELECT x.id AS p, y.id AS q
FROM r x JOIN r y ON x.k = y.k AND x.rn = 1 AND y.rn = 2
""").show()
```
This fails on 4.0.2 and later. It succeeds on 4.0.1, which predates
SPARK-53738. The user sees only an internal `NoSuchElementException` at the
first action, with nothing actionable in it — the query itself is well formed.
I verified the failure on 4.0.4, 4.1.3 and 4.2.0.
See [SPARK-59009](https://issues.apache.org/jira/browse/SPARK-59009) for
the full analysis.
### Does this PR introduce _any_ user-facing change?
No, other than the bug fix itself: queries that reference a cached relation
with a non-empty `outputOrdering` more than once now succeed instead of failing
with an internal error.
`newInstance()` also preserves the ordering now rather than returning a
relation with a stale one, so the ordering remains usable as an optimization
hint for the new instance.
### How was this patch tested?
Two new tests, both of which fail on unmodified `master` and pass with the
change:
- `InMemoryRelationSuite`, a unit test asserting that after `newInstance()`
the ordering references the new attributes and that the result canonicalizes
without throwing.
- `CachedTableSuite`, an end-to-end test running the CTE self-join over a
cached, ordered relation and checking the answer. Without the change it fails
with `NoSuchElementException: key not found: k#...`.
`InMemoryRelationSuite`, `CachedTableSuite` and `DatasetCacheSuite` are
green with the change.
```
build/sbt "sql/testOnly
org.apache.spark.sql.execution.columnar.InMemoryRelationSuite"
build/sbt "sql/testOnly org.apache.spark.sql.CachedTableSuite"
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (model claude-opus-5)
Closes #58293 from james-willis/SPARK-59009.
Authored-by: James Willis <[email protected]>
Signed-off-by: Peter Toth <[email protected]>
(cherry picked from commit 38fc867ca29a024384c63abe166a13acb9c327df)
Signed-off-by: Peter Toth <[email protected]>
---
.../sql/execution/columnar/InMemoryRelation.scala | 22 +++++++++++-------
.../org/apache/spark/sql/CachedTableSuite.scala | 27 ++++++++++++++++++++++
.../execution/columnar/InMemoryRelationSuite.scala | 17 ++++++++++++++
3 files changed, 58 insertions(+), 8 deletions(-)
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 e6bf65ec89e6..584eaa9ecbc8 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
@@ -710,16 +710,22 @@ case class InMemoryRelation(
val newOutputOrdering = outputOrdering
.map(_.transform { case a: Attribute => map(a) })
.asInstanceOf[Seq[SortOrder]]
- InMemoryRelation(newOutput, cacheBuilder, newOutputOrdering,
statsOfPlanToCache)
+ // `attributeStats` is keyed by attribute, so it has to be re-keyed onto
`newOutput` as well,
+ // otherwise every column stat lookup misses for the new relation and the
estimates silently
+ // fall back to the un-filtered defaults. `statsOfPlanToCache` is a `var`
that starts as null.
+ val newStatsOfPlanToCache = if (statsOfPlanToCache == null) {
+ null
+ } else {
+ LogicalRDD.rewriteStatistics(statsOfPlanToCache, map)
+ }
+ InMemoryRelation(newOutput, cacheBuilder, newOutputOrdering,
newStatsOfPlanToCache)
}
- override def newInstance(): this.type = {
- InMemoryRelation(
- output.map(_.newInstance()),
- cacheBuilder,
- outputOrdering,
- statsOfPlanToCache).asInstanceOf[this.type]
- }
+ // Goes through `withOutput` so that `outputOrdering` is re-mapped onto the
fresh exprIds.
+ // Returning a relation whose `outputOrdering` still references the old
attributes would break
+ // canonicalization, which re-maps the ordering through the relation's own
`output`.
+ override def newInstance(): this.type =
+ withOutput(output.map(_.newInstance())).asInstanceOf[this.type]
// override `clone` since the default implementation won't carry over
mutable states.
override def clone(): LogicalPlan = {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
index 90f991479801..66732a9e52b0 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
@@ -2727,6 +2727,33 @@ class CachedTableSuite extends SharedSparkSession
}
}
+ test("SPARK-59009: cached relation with outputOrdering can be referenced
more than once") {
+ withTempView("t", "ordered_t") {
+ spark.range(0, 20).selectExpr("id", "id % 3 AS
k").createOrReplaceTempView("t")
+ val ordered = sql("SELECT id, k FROM t ORDER BY k, id")
+ ordered.persist()
+ try {
+ ordered.createOrReplaceTempView("ordered_t")
+ // The CTE below is referenced twice, so InlineCTE deduplicates the
second copy and calls
+ // newInstance() on the cached relation. The join then canonicalizes
it, which used to
+ // throw because newInstance() left `outputOrdering` on the old
attributes.
+ checkAnswer(
+ sql(
+ """
+ |WITH r AS (
+ | SELECT *, row_number() OVER (PARTITION BY k ORDER BY id DESC)
AS rn
+ | FROM ordered_t
+ |)
+ |SELECT x.id AS x_id, y.id AS y_id
+ |FROM r x JOIN r y ON x.k = y.k AND x.rn = 1 AND y.rn = 2
+ """.stripMargin),
+ Row(18L, 15L) :: Row(19L, 16L) :: Row(17L, 14L) :: Nil)
+ } finally {
+ ordered.unpersist()
+ }
+ }
+ }
+
private def cacheManager = spark.sharedState.cacheManager
private def pinTable(
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 2c73622739a5..6a9c8964e4db 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
@@ -18,6 +18,7 @@
package org.apache.spark.sql.execution.columnar
import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.expressions.AttributeSet
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.functions.expr
@@ -34,6 +35,22 @@ class InMemoryRelationSuite extends SparkFunSuite
assert(r1.sameResult(r2))
}
+ test("SPARK-59009: newInstance() re-maps outputOrdering onto the new
attributes") {
+ val d = spark.range(10).selectExpr("id", "id % 3 AS k").orderBy("k", "id")
+ val r1 = InMemoryRelation(StorageLevel.MEMORY_ONLY, d.queryExecution, None)
+ assert(r1.outputOrdering.nonEmpty)
+
+ val r2 = r1.newInstance()
+ assert(r2.output.map(_.exprId) != r1.output.map(_.exprId))
+ // The ordering must be re-mapped onto the new attributes, not left
referencing the old ones.
+ assert(r2.outputOrdering.nonEmpty)
+
assert(AttributeSet(r2.outputOrdering.flatMap(_.references)).subsetOf(AttributeSet(r2.output)))
+ // `sameResult` is what CacheManager lookups and exchange reuse rely on.
It goes through
+ // `doCanonicalize`, which re-maps `outputOrdering` through `output`, so a
stale ordering
+ // throws here rather than merely producing an unequal plan.
+ assert(r1.sameResult(r2))
+ }
+
test("SPARK-47177: Cached SQL plan do not display final AQE plan in explain
string") {
def findIMRInnerChild(p: SparkPlan): SparkPlan = {
val tableCache = find(p) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
