peter-toth commented on code in PR #60197:
URL: https://github.com/apache/spark/pull/60197#discussion_r4501298061
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala:
##########
@@ -117,7 +116,9 @@ case class LogicalRDD(
session :: originStats :: originConstraints :: Boolean.box(fromCheckpoint)
:: Nil
override def newInstance(): LogicalRDD.this.type = {
- val rewrite = Utils.toMap(output, output.map(_.newInstance()))
+ // Keyed by exprId, because the attributes in the partitioning, the
ordering, the stats and
+ // the constraints can differ from `output` in qualifier or nullability.
+ val rewrite = AttributeMap(output.zip(output.map(_.newInstance())))
Review Comment:
**Finding 2.** Reproduced. The `right_outer` join with the source returns 9
rows instead of 10, at this head and on the merge base `a11fd253b5a`.
The recursive-CTE caller fails today too. With default configs, this query
fails with `RECURSION_LEVEL_LIMIT_EXCEEDED` on the base and at this head,
instead of returning `[1000, 4]`:
```sql
WITH RECURSIVE r(k, lvl) AS (
SELECT id AS k, 0 AS lvl FROM range(200)
UNION ALL
SELECT * FROM (SELECT k, lvl + 1 AS lvl FROM r LIMIT 100000) t WHERE t.lvl
< 5)
SELECT count(*), max(lvl) FROM r
```
That is the symptom SPARK-52353 fixed, in a case its fix didn't cover.
Making both `buildOutputAssocForRewrite` helpers return an `AttributeMap`, as
`newInstance` now does, fixes both:
```scala
val rewrite = AttributeMap(source.zip(destination).flatMap { case
(attr1, attr2) =>
if (attr1.name == attr2.name && attr1.dataType == attr2.dataType) {
Some(attr1 -> attr2)
} else {
None
}
})
```
I made the same change in `buildOutputAssocForRewriteWithNewOutput`. With
both changes, the join returns 10 rows and the CTE returns `[1000, 4]`.
`DatasetSuite` and `cte-recursion.sql` still pass. A recursive-CTE case in the
test would pin this path.
##########
sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala:
##########
@@ -1956,6 +1956,20 @@ class DatasetSuite extends SharedSparkSession
}
}
+ test("SPARK-61088: self-join of a checkpointed Dataset should not use the
partitioning " +
+ "and constraints of the other copy") {
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ // The partitioning and the constraints refer to `t.id`, while the
output is `s.id`.
+ val cp = spark.range(16).as("t").where($"t.id" >
5).repartition($"t.id").as("s")
Review Comment:
**Finding 4.** Confirmed with your single-partition shape, `spark.range(0,
16, 1, 1).as("s").localCheckpoint()`:
- On the merge base, all 128 rows get a window count other than 8, with AQE
on and off.
- At this head, all of them get 8.
The plan-level assertion you suggest also covers the nullability mismatch
that the new comment and the description name. No result check reaches that
case. With AQE and broadcast joins off when checkpointing:
```scala
val cp = spark.range(10).toDF("k")
.join(spark.range(5).toDF("k2"), $"k" === $"k2", "full_outer")
.where($"k".isNotNull).localCheckpoint()
// the output `k` is nullable, while the partitioning uses the non-nullable
`k`
```
The self-join aggregate over it returns correct rows even on the base,
because column pruning puts a `Project` over the scan, and that drops the stale
partitioning. Asserting that the partitioning, ordering and constraints of
`newInstance()` refer only to its own `outputSet` fails on the base for three
checkpoints, and passes at this head:
- this one;
- the test's checkpoint;
- the test's checkpoint with `.sortWithinPartitions($"t.id")`.
Separately, the constraints half of the test also fails on the base with AQE
on, so it can move out of the `withSQLConf`. Only the aggregate half needs AQE
off, because `AdaptiveSparkPlanExec` reports `UnknownPartitioning` when
checkpointing.
--
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]
