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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
