dongjoon-hyun opened a new pull request, #60197:
URL: https://github.com/apache/spark/pull/60197
### What changes were proposed in this pull request?
This PR fixes a correctness issue: a self-join of a checkpointed Dataset can
return wrong results.
`LogicalRDD.newInstance` re-issues the expression IDs of `output` and
rewrites `outputPartitioning`, `outputOrdering`, `originStats` and
`originConstraints` through a map from the old output attributes to the new
ones. The map was built with `Utils.toMap`, so a lookup uses
`AttributeReference.equals`, which also compares the qualifier and the
nullability. However, `LogicalRDD.fromDataset` takes `output` from the analyzed
plan, the partitioning and the ordering from the executed plan, and the
constraints from the optimized plan. Their attributes can differ from `output`
in qualifier or nullability, so they miss the map and keep their old expression
IDs, which are the expression IDs of the other copy in a self-join.
This PR builds the map with `AttributeMap`, which looks up attributes by
expression ID, like `InMemoryRelation.withOutput` already does.
### Why are the changes needed?
When a checkpointed Dataset is joined with itself, `DeduplicateRelations`
calls `LogicalRDD.newInstance` on one copy. The renewed copy then reports:
1. A partitioning over the column of the other copy. When a broadcast hash
join streams the renewed copy, it passes this partitioning up, and an aggregate
grouped by the column of the other copy skips its shuffle. The aggregate
returns duplicated groups with split counts. This needs AQE to be disabled when
the Dataset is checkpointed, because `AdaptiveSparkPlanExec` reports
`UnknownPartitioning`.
2. Constraints over the column of the other copy. For a filter over a `RIGHT
OUTER` join whose right side is the renewed copy, `EliminateOuterJoin` reads
such a constraint as a null-filtering predicate on the left side and turns the
join into an `INNER` join, so the null-extended rows are lost. This happens
with the default configurations.
```scala
// Issue 1: It should return 10, but returns 50.
spark.conf.set("spark.sql.adaptive.enabled", "false")
val cp1 = spark.range(10).repartition($"id").as("s").localCheckpoint()
broadcast(cp1.as("x")).join(cp1.as("y"), $"x.id" % 2 === $"y.id" %
2).groupBy($"x.id").count().count()
// Issue 2: It should return 4, but returns 2.
spark.conf.unset("spark.sql.adaptive.enabled")
val cp2 = spark.range(10).as("t").where($"t.id" >
5).as("s").localCheckpoint()
cp2.as("x").join(cp2.as("y"), $"x.id" === $"y.id" + 2,
"right_outer").where($"y.id" > 0).count()
```
In both cases, the attributes of the partitioning or the constraints have a
different qualifier (none or `t`) from the output attribute (`s`). A
nullability difference triggers the same issue, e.g., when `EliminateOuterJoin`
turns a `FULL OUTER` join into a `LEFT OUTER` join, the partitioning is over
the non-nullable join key while the analyzed output is nullable.
### Does this PR introduce _any_ user-facing change?
Yes, this fixes wrong results. The partitioning issue exists since Apache
Spark 2.1.0 ([v2.1.0](https://github.com/apache/spark/releases/tag/v2.1.0),
2016-12-15, SPARK-17972) and the constraints issue exists since Apache Spark
3.4.0 ([v3.4.0](https://github.com/apache/spark/releases/tag/v3.4.0),
2023-04-07, SPARK-39834).
### How was this patch tested?
Pass the CIs with the newly added test case, which fails without this fix.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.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: [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]
