[SPARK-59659][SQL] Consolidate the repeated cluster-key matching in Partitioning.satisfies0

[peter-toth]

What changes were proposed in this pull request?

Five Partitioning.satisfies0 implementations spelled out the same two-branch test on ClusteredDistribution.requireAllClusterKeys:

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)

[peter-toth]

@ulysses-you this is the matchesClusterKeys consolidation promised in #58659 (comment), taken out of that PR because it accounted for most of its partitioning.scala diff.

[dongjoon-hyun]

+1, LGTM.

I checked the equivalence of each rewrite and they all hold:

• corresponds is same-length-and-elementwise, so it computes exactly what the three zip-folds did, and the semanticEquals argument order is preserved in each.
• Replacing case c @ ClusteredDistribution(_, _, _, _) with case c: ClusteredDistribution matches the same set, since the extractor bound all four parameters of a leaf case class. The ClusteredDistribution in connector.distributions is a different type and is not imported here.
• allowNullKeySpreading is the fourth constructor parameter, so the guard is unchanged in the two null-aware partitionings.
• allClusterKeysAmong is literally the fold it replaces in AQEUtils, argument order included.

I also think the scope is right: KeyGroupedPartitioning.keysSatisfy reads the same flag but its else branch is isFunctionOfClusterKeys || (mayProjectToClusterKeys && ...), not the membership test, and the two canCreatePartitioning sites are on the co-partition flag. Folding either into matchesClusterKeys would have changed behaviour; leaving them alone is correct.

Two non-blocking doc nits, feel free to take them in a follow-up:

1. The new scaladoc says "This is the whole of what requireAllClusterKeys governs, so a satisfies0 reads it here rather than branching on the flag itself." KeyGroupedPartitioning.satisfies0 still branches on the flag via keysSatisfy, and mayProjectToClusterKeys reads it too. A reader who takes the sentence at face value might "finish the job" there and drop the projection logic. Narrowing it to these five satisfies0 implementations, or naming the exception, would prevent that.

2. The existing comment on areAllClusterKeysMatched still says a Partitioning should call it when requireAllClusterKeys is set. After this PR the partitionings call matchesClusterKeys instead, so it could point there.