peter-toth opened a new pull request, #58858:
URL: https://github.com/apache/spark/pull/58858
### What changes were proposed in this pull request?
`EnsureRequirements` stops applying partition filtering to a join pair that
holds a marked
`KeyedPartitioning`. `checkKeyGroupCompatible` decides filtering once, in a
new `filtersKeys`, which
both `rank` and the merge call read, and `mergeAndDedupPartitions` takes
that answer as a
`filterPartitions` parameter instead of reading the config itself.
### Why are the changes needed?
A marked layout pins the rows whose key it does not declare to `hash(key) %
numPartitions` over the
keys it does declare. `GroupPartitionsExec` keeps that claim only for an
identity regrouping, and
since SPARK-59272 a forfeited claim makes the pairing gate decline, so the
join shuffles both sides.
Filtering is the only thing that shrinks the merged key list below a side's
own declared keys. The
merging arms take the union, and `areKeysCompatible` pairs a marked layout
only with one whose keys
are a subset of its declared keys, so the union is the marked side's own key
set. An intersection
drops groups the marked side holds, the count the hash is taken modulo
changes with them, and the
claim goes. The pair then trades a whole storage-partitioned join for
pruning one key group.
It surfaces as a failing test as soon as
`spark.sql.sources.v2.bucketing.partition.filter.enabled` is on, which
#58681 proposes to make the
default:
```
- SPARK-59050: SPJ: unknown-keyed partitioning still joins a subset-keyed
partner *** FAILED ***
2 did not equal 1 expected 1 shuffles, got 2
```
`r` is marked and declares `[1, 2]`, `u` declares `[1]`, the inner join
intersects to `[1]`, `r`
regroups 2 groups into 1 and the gate declines.
This does not walk back the SPARK-59272 gate. There the merged list is the
union of two key orders
and `mergeAndDedupPartitions` sorts it, so the join could not be kept and
the plan left valid at the
same time, and the extra shuffle bought back every `AQEShuffleReadRule` and
`OptimizeSkewedJoin` on
that stage. Here nothing forced the list to be narrowed.
Sorting is the other half and is not fixed here: a marked layout whose
declared order is not the
sorted one is relabelled even where the set is unchanged. Handing it its own
list verbatim is not
the same fix, because `KeyedPartitioning.createShuffleSpec` sorts an
unmarked side through
`toGrouped` under `v2BucketingAllowKeysSubsetOfPartitionKeys` while handing
a marked one back
unprojected, so the two children hold one partitioning and report two specs
`describesSameKeys`
calls different. I tried it, and the SPARK-59272 sweep fails on cell
`left=id/12 right=id/312/marked Inner`. That half is the same regression as
this one rather than an
improvement on top, and I am filing it with the measurement.
### Does this PR introduce _any_ user-facing change?
**Yes**, one plan change, reachable only with
`spark.sql.sources.v2.bucketing.shuffle.enabled` on,
the only producer of a marked layout, together with
`spark.sql.sources.v2.bucketing.partition.filter.enabled`. Both default to
`false`, so nothing
changes out of the box. With both on, such a join keeps its
storage-partitioned join and reads the
groups the filter would have pruned, instead of shuffling both sides.
### How was this patch tested?
Three tests, each failing with the change reverted. No existing expectation
is changed: the two
`KeyGroupedPartitioningSuite` cases keep the assertions they had and now run
them under one more
setting.
- `EnsureRequirementsSuite`, new `SPARK-59272: partition filtering leaves a
marked pair's merged
keys alone`: a marked `[1, 2, 3]` against a plain `[1, 2]`, over both
settings, as `Inner` with
the marker on the left and as `LeftOuter` with it on the right. The two
join types take different
arms of the filter and of `rank`. It pins no shuffle, three pushed keys a
side, and
`ValidateRequirements`.
- `KeyGroupedPartitioningSuite`, `SPARK-59050: SPJ: unknown-keyed
partitioning still joins a
subset-keyed partner`, now over both settings. Reverted, the `true` arm
gives
`2 did not equal 1 expected 1 shuffles`.
- `KeyGroupedPartitioningSuite`, `SPARK-59050: SPJ: a reducer-free identity
regrouping keeps the
unknown-keyed claim`, now over both settings, plus `ValidateRequirements`
and the marked side's
three keys. Reverted, the `true` arm gives `identityGpe.isDefined was
false`. This is the case
#58681 currently pins to `false`; with this in, the flip does not reach it.
The SPARK-59272 sweep, 24 shapes a side over four join types and 16
configuration cells asserting
`ValidateRequirements` on every plan the rule produces, passes. `SPARK-59199:
mergeAndDedupPartitions filters partitions per join type` now passes
`filterPartitions` instead of
toggling the config, which is where the read moved to.
Green: `KeyGroupedPartitioningSuite`, `EnsureRequirementsSuite`,
`GroupPartitionsExecSuite`,
`DistributionSuite`, `ShuffleSpecSuite`, `ValidateRequirementsSuite`,
`PlannerSuite`,
`ProjectedOrderingAndPartitioningSuite`, `ExchangeSuite`, `JoinSuite` and
`AdaptiveQueryExecSuite`,
668 tests. `dev/lint-scala` is clean.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
--
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]
