peter-toth commented on code in PR #59599:
URL: https://github.com/apache/spark/pull/59599#discussion_r4339031126
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -698,42 +709,131 @@ case class EnsureRequirements(
joinType == LeftAnti || joinType == LeftOuter
}
- // Populate the common partition information down to the scan nodes
- private def populateCommonPartitionInfo(
- plan: SparkPlan,
- values: Seq[(InternalRow, Int)],
+ /**
+ * Compose join-key positions from an outer SPJ with positions already
pushed by an inner SPJ.
+ * Outer positions index the layout the outer join sees; inner positions
index the scan's raw
+ * partition expressions, so they are composed rather than replaced.
+ */
+ private def composeJoinKeyPositions(
+ existing: Option[Seq[Int]],
+ newPositions: Option[Seq[Int]]): Option[Seq[Int]] = {
+ (existing, newPositions) match {
+ case (Some(old), Some(newPos)) => Some(newPos.map(old))
+ case (_, None) => existing
+ case (None, Some(newPos)) => Some(newPos)
+ }
+ }
+
+ /** Controls which SPJ fields are updated on a scan during pushdown. */
+ private case class SpjPushFlags(
+ pushJoinKeyPositions: Boolean = true,
+ pushCommonValues: Boolean = true)
+
+ /**
+ * Compute side-aware SPJ push flags for nested shuffle joins. With no
join-key positions,
+ * push common values to both inner sides. With subset join keys, compose
join-key positions
+ * on the matching side only and retain inner common-partition metadata on
both sides.
+ */
+ private def nestedSpjPushFlags(
joinKeyPositions: Option[Seq[Int]],
- reducers: Option[Seq[Option[Reducer[_, _]]]],
- applyPartialClustering: Boolean,
- replicatePartitions: Boolean): SparkPlan = plan match {
- case scan: BatchScanExec =>
- scan.copy(
- spjParams = scan.spjParams.copy(
- commonPartitionValues = Some(values),
- joinKeyPositions = joinKeyPositions,
- reducers = reducers,
- applyPartialClustering = applyPartialClustering,
- replicatePartitions = replicatePartitions
- )
- )
- case node =>
- node.mapChildren(child => populateCommonPartitionInfo(
- child, values, joinKeyPositions, reducers, applyPartialClustering,
replicatePartitions))
+ isMatchingSide: Boolean): SpjPushFlags = {
+ val syncCommonValuesToBothSides = joinKeyPositions.isEmpty
+ if (joinKeyPositions.isDefined && !syncCommonValuesToBothSides) {
+ SpjPushFlags(pushJoinKeyPositions = isMatchingSide, pushCommonValues =
false)
Review Comment:
**Finding 1.** `KeyGroupedPartitioning.createShuffleSpec` sets
`joinKeyPositions` on every spec once `allowJoinKeysSubsetOfPartitionKeys` is
on, even when the join uses all partition keys. So this branch runs for every
nested join under that conf, and neither inner side gets the outer join's
merged values. Only the outer join's other side gets them.
Two queries that are correct on `branch-4.1` break here. All tables are
`(store INT, id INT, data STRING) PARTITIONED BY (id)`, with
`allowJoinKeysSubsetOfPartitionKeys.enabled=true`, AQE on and off:
- **Wrong rows.** `e1` and `e2` hold ids 1 and 2. `e5` holds ids 2 and 3.
Also set `partition.filter.enabled=true`.
```sql
SELECT /*+ MERGE(e1, e2, e5) */ e1.id, e2.id, e5.id
FROM e1 JOIN e2 ON e1.id = e2.id
RIGHT JOIN e5 ON e1.id = e5.id
```
It returns `[null,null,2]`, `[null,null,3]` instead of `[2,2,2]`,
`[null,null,3]`. The merged values are `e5`'s (2, 3), while the nested join
keeps (1, 2). Both sides have 2 partitions, so the guard at line 191 passes and
id 1 is zipped with id 2.
- **SPJ lost.** `e3` holds ids 1, 2 and 3, and the query is `e1 JOIN e2 ON
e1.id = e2.id JOIN e3 ON e1.id = e3.id`. `branch-4.1` plans it with no shuffle.
Here both sides of the outer join are shuffled, because the nested join keeps 2
partitions against `e3`'s 3.
The `case other` branch below already does the right thing. It pushes the
values and composes the positions on both inner scans. It runs whenever a
`Project` sits between the two joins, which is the case in three of the four
new tests (all but the reducers one). With a `Project` added, both queries
above are correct with no shuffle. The two inner sides report the same key
layout after the inner join's SPJ, so the outer join's positions and values fit
both of them.
With the two join cases here disabled, all new tests and both queries above
pass. I'd drop `SpjPushFlags`, `nestedSpjPushFlags` and the join cases, and
call `populateSpjParams` from both call sites.
That also opens a way to fix the ignored "multiple names per id" test
without `GroupPartitionsExec`. When a push narrows the positions without
bringing new values, `populateSpjParams` can project the scan's common values
onto them and dedupe, instead of the leading-prefix lookup in
`KeyGroupedPartitionedScan`. I tried it and all five new tests pass, the
ignored one included.
--
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]
