peter-toth opened a new pull request, #58928:
URL: https://github.com/apache/spark/pull/58928
### What changes were proposed in this pull request?
Five `Partitioning.satisfies0` implementations spelled out the same
two-branch test on `ClusteredDistribution.requireAllClusterKeys`:
```scala
if (requireAllClusterKeys) {
c.areAllClusterKeysMatched(expressions)
} else {
expressions.forall(x => requiredClustering.exists(_.semanticEquals(x)))
}
```
So the flag was read in five places and the membership test written out five
times. `ClusteredDistribution.matchesClusterKeys` is that pair, beside the
`areAllClusterKeysMatched` it wraps and the `isClusterKey` it is built from,
and the five callers ask it: `HashPartitioningLike`,
`NullAwareHashPartitioning`, `CoalescedNullAwareHashPartitioning`,
`RangePartitioning` and `ShufflePartitionIdPassThrough`.
Three hand-written zip-and-compare folds become one `Seq.corresponds` each:
`ClusteredDistribution.areAllClusterKeysMatched`,
`OrderedDistribution.areAllClusterKeysMatched`, and the
`StatefulOpClusteredDistribution` arm of `HashPartitioningLike.satisfies0` that
spelled the first one out at a call site rather than calling it.
`AQEUtils.getRequiredDistribution`'s mirror fold, which asks whether a
project list covers the clustering rather than the other way round, is
`allClusterKeysAmong`.
Net 28 lines removed, no new API beyond `matchesClusterKeys` on a
`private[sql]`-reachable case class in `sql.catalyst`.
### Why are the changes needed?
`requireAllClusterKeys` is one rule about what a partitioning has to be
partitioned on, and it was five copies of that rule. A change to it had to be
made in five places, and a reader comparing two `satisfies0` implementations
had to diff the copies to see that they agree. `isClusterKey` and
`allClusterKeysAmong` already exist on `ClusteredDistribution` for exactly this
reason, added by SPARK-59289; this finishes the set and uses them.
### Does this PR introduce _any_ user-facing change?
No. This is behaviour neutral, and three things are worth stating because
each is a place it could have failed to be:
- `Seq.corresponds` is same-length-and-element-wise, which is what each of
the three folds computed.
- The argument order of every `semanticEquals` is preserved, so nothing
relies on it being symmetric.
- Replacing the `case c @ ClusteredDistribution(...)` extractor patterns
with `case c: ClusteredDistribution` matches the same set: the physical
`ClusteredDistribution` is a leaf case class. The `ClusteredDistributionImpl`
in `connector.distributions` extends the connector interface of that name, not
this one.
### How was this patch tested?
No new tests: there is no new behaviour to pin, and the existing suites
already cover each rewritten site.
Green: `DistributionSuite`, `ShuffleSpecSuite`, `PlannerSuite`,
`EnsureRequirementsSuite`, `ValidateRequirementsSuite`,
`KeyGroupedPartitioningSuite`, `GroupPartitionsExecSuite`,
`AdaptiveQueryExecSuite` and `ExchangeSuite`, 593 tests.
The touched arms that are easy to miss got their own runs.
`StatefulOpClusteredDistribution` and the two null-aware partitionings:
`StreamingAggregationDistributionSuite`,
`StreamingDeduplicationDistributionSuite`,
`StreamingSessionWindowDistributionSuite`,
`FlatMapGroupsWithStateDistributionSuite` and
`StreamingQueryHashPartitionVerifySuite`, 12 tests. And since `satisfies0` runs
on every query, all five plan-stability suites, 172 tests, so no golden TPCDS
or TPCH plan changed.
`dev/lint-scala` is clean.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 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]
