ulysses-you opened a new pull request, #58942:
URL: https://github.com/apache/spark/pull/58942
### What changes were proposed in this pull request?
`ValidateRequirements` asks every child of a clustered operator to satisfy
its distribution on its own. A `ClusteredDistribution` is the one distribution
an operator can owe two children together rather than one by one, so this PR
judges such an operator's children by their pairing, and a pair aligned without
grouping (which partially clustered distribution builds on purpose) is one the
sides agree on while neither is grouped.
- `ValidateRequirements` asks the per-side check only of children that
answer for themselves or report no keyed layout; the rest are judged together,
on the specs the planner builds for them
(`PartitioningCollection.specsForPairing`, a new `private[sql]` helper on the
existing object: a keyed member on the planner's own admission
`keysMaySatisfy`, any other member on `satisfies`, and a pinned count asked as
it stands). One member of the first side has to pair with every other side,
which is the question `EnsureRequirements` commits a pair on (`committed`, over
the pair `agreeingPairs` picked).
- What those specs hold is what `createShuffleSpec` makes of a member: its
own layout, or under
`spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys` the projection
onto the keys it covers, the layout the alignment about to be planned emits.
That is the question the planner asks when it picks the member a pair is
planned on, less the coverage of every operation key it additionally requires
there: `spark.sql.requireAllClusterKeysForCoPartition` is a skew heuristic, and
a member covering a subset of the operation keys is a sound pairing.
- What the pairing cannot say is how the two sides hold a key's rows: a
spread side and one that repeats the whole group report the same keys as two
sides that split the key between them, and no layout distinguishes those, so
that rests on the producer, which is the join path: `checkKeyGroupCompatible`
is where such a pair is planned, and every other co-partitioning operator's
children are grouped before the rule is done.
The catalyst change is additive: a new helper beside the existing ones, and
no change to `PartitioningCollection.createShuffleSpec` or
`maySatisfyAfterProjection`. Those two still name `ValidateRequirements` as a
caller in their comments, which this change leaves as they are.
### Why are the changes needed?
A storage-partitioned join planned by partially clustered distribution
aligns its sides without grouping either of them: the side that keeps its
splits spreads them, the other replicates its group across them, so both report
keys that repeat on purpose. Such a pair fails the per-side check at the join
node even though the two sides agree key by key, and
`AdaptiveSparkPlanExec.optimizeQueryStage` validates a stage's whole candidate
plan before accepting an `AQEShuffleReadRule` change, so every shuffle read in
that stage stays uncoalesced, unrelated ones included.
Measured on `d39cc1784c0` (base) versus this head, both with
`spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled=true`:
| Probe | base | head |
|---|---|---|
| A pair the rule planned (`EnsureRequirements.apply`) |
`ValidateRequirements.validate` false | true |
| The aggregate's shuffle read in the join's stage | not coalesced
(`AQEShuffleReadExec.hasCoalescedPartition` false) | coalesced |
| A three-table chain whose outer join reads a projection over both key
columns | `validate` false | true, shuffle-free, still coalesced |
The planner-side half of this hazard landed in SPARK-59272, which declines a
pairing whose sides no longer declare the same aligned key sequence: that
closes the pairs a `GroupPartitionsExec` gives up on, which should not be built
at all. A pair aligned without grouping is the other half, and it is built on
purpose, so the validator has to read the pairing instead of each child on its
own.
### Does this PR introduce _any_ user-facing change?
No. No plan changes unless the plan already contains such an alignment, and
no new configuration.
### How was this patch tested?
New tests, by what each one is there for (the four marked base-failing were
re-run against `d39cc1784c0` and fail there):
- `ValidateRequirementsSuite`
- the exemption and its refusals (base-failing): a keyed pair that repeats
its keys position by position passes, while key sets that disagree, a differing
key order, a hashed side, a lone clustered child and two sides that each
satisfy on their own but do not line up all fail;
- the rule's own pair (base-failing): a pair planned by
`EnsureRequirements` under partially clustered distribution passes;
- the collection shape (base-failing): a side reporting several keyed
alternatives is judged on whichever of them pairs, not on the first one;
- guards, which pass on base as well and pin this change's clauses: a
subset-keyed pair still passes, a lone clustered child still owes its own
grouping, a pinned partition count is still asked of a pair, a pair that lines
up still owes its operator an ordering, and a collapsed pair is not admitted on
its pairing alone.
- `KeyGroupedPartitioningSuite`
- with partially clustered distribution on, the aggregate's shuffle read
in the stage holding a two-table join coalesces (base-failing: it is not
coalesced); the test also pins that the join side shuffles nothing, that the
chain shuffles once, and that the join shares the final stage with that read,
which is what makes the coalesce a decision the join can block;
- a three-table chain whose outer join reads a projection that keeps both
key columns stays shuffle-free, passes validation and keeps its coalescing
(base-failing at the validation assertion).
- `ShuffleSpecSuite`: the specs a side offers are the planner's own: a keyed
member offers the layout it reports, a member covering part of the operation's
keys offers the projection onto the keys it covers under the subset permission,
a member whose keys do not cover the clustering offers nothing, and a member
that is not keyed offers its own spec.
Ran locally: `ValidateRequirementsSuite`, `KeyGroupedPartitioningSuite`,
`EnsureRequirementsSuite`, `GroupPartitionsExecSuite`,
`AdaptiveQueryExecSuite`, `PushDownLocalSortSuite`, `ShuffleSpecSuite`: all
green, with scalastyle clean for catalyst and sql, main and test sources.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (qwen3.8-flash)
🤖 Generated with [Claude Code](https://claude.com/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]
