You are viewing a plain text version of this content. The canonical link for it is here.

Posted to reviews@spark.apache.org by "james-willis (via GitHub)" <gi...@apache.org> on 2026/08/25 21:38:47 UTC

[PR] [SPARK-59009][SQL] Re-map outputOrdering in InMemoryRelation.newInstance() [spark]

james-willis opened a new pull request, #58293:
URL: https://github.com/apache/spark/pull/58293

   ### 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` is also made defensive: if `outputOrdering` references attributes that are not present in `output`, the ordering is dropped rather than the `AttributeMap` lookup throwing.
   
   ### 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`:
   
   - `InMemoryRelationSuite`, a unit test asserting that after `newInstance()` the ordering references the new attributes and that the result canonicalizes without throwing. This one pins the `newInstance()` behaviour specifically: it still fails if only the `withOutput` guard is applied.
   - `CachedTableSuite`, an end-to-end test running the CTE self-join over a cached, ordered relation and checking the answer. Without either change it fails with `NoSuchElementException: key not found: k#...`. Note that this test covers the user-visible symptom, which either change fixes on its own.
   
   ```
   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)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-59009][SQL] Re-map outputOrdering in InMemoryRelation.newInstance() [spark]

Posted by "peter-toth (via GitHub)" <gi...@apache.org>.

peter-toth commented on code in PR #58293:
URL: https://github.com/apache/spark/pull/58293#discussion_r3860906211


##########
sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala:
##########
@@ -713,13 +713,11 @@ case class InMemoryRelation(
     InMemoryRelation(newOutput, cacheBuilder, newOutputOrdering, statsOfPlanToCache)
   }
 
-  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]

Review Comment:
   **Finding 1.** `withOutput` re-maps `outputOrdering` but hands `statsOfPlanToCache` straight through, so `newInstance()` still returns a relation whose `Statistics.attributeStats` is keyed by the *old* attributes. `FilterEstimation` and `JoinEstimation` look those up by the plan's own attributes, so every column stat is silently dropped for the deduplicated copy.
   
   Measured on this commit, CBO on, over a 1000-row analyzed table cached with an ordering:
   
   ```
   ### relation stats (orig): sizeInBytes=15.6 KiB, rowCount=1.00E+3
   ### relation stats (new) : sizeInBytes=15.6 KiB, rowCount=1.00E+3
   ### Filter(id > 900) over ORIGINAL : sizeInBytes=1600.0 B, rowCount=100
   ### Filter(id > 900) over NEW COPY : sizeInBytes=15.6 KiB, rowCount=1.00E+3
   ```
   
   A 10x overestimate on the branch that went through `newInstance()`, which is enough to flip a join strategy or a build side. No error, just a worse plan.
   
   The sibling `MultiInstanceRelation` gets this right: `LogicalRDD.newInstance()` (`sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala:119`) re-maps partitioning, ordering *and* stats, and its helper is `private[sql]` and already in scope here. In `withOutput`:
   
   ```scala
   val newStats = if (statsOfPlanToCache == null) {
     null
   } else {
     LogicalRDD.rewriteStatistics(statsOfPlanToCache, map)
   }
   InMemoryRelation(newOutput, cacheBuilder, newOutputOrdering, newStats)
   ```
   
   I applied that: the lookup returns `Some(ColumnStat(...))`, the filter estimate goes back to 100 rows, and `CachedTableSuite` + `InMemoryRelationSuite` + `DatasetCacheSuite` stay green. The null guard is needed because `statsOfPlanToCache` is a `var` initialized to `null`.
   
   Pre-existing and CBO-only, so a follow-up ticket is fine if you'd rather keep this PR to the crash.
   



##########
sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryRelationSuite.scala:
##########
@@ -34,6 +35,20 @@ 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)))
+    // Canonicalization re-maps `outputOrdering` through `output`, so a stale ordering would throw.
+    r2.canonicalized

Review Comment:
   **Finding 2.** A bare `r2.canonicalized` only pins "doesn't throw". `sameResult` is the property that actually matters for a `MultiInstanceRelation` (it is what `CacheManager` lookups and exchange reuse rely on), it still fails on base because it throws inside `doCanonicalize`, and it mirrors the SPARK-46779 test right above.
   
   ```suggestion
       assert(r1.sameResult(r2))
   ```
   
   I checked `r1.sameResult(r2)` is `true` with the fix in place.
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org