pan3793 opened a new pull request, #57762:
URL: https://github.com/apache/spark/pull/57762
### What changes were proposed in this pull request?
This PR removes `spark.sql.requireAllClusterKeysForCoPartition` as a gate
for storage-partitioned joins (V2 `KeyedPartitioning`). Previously,
`createKeyedShuffleSpec` in `EnsureRequirements` checked this config before
deciding whether a V2 data-source-partitioned side could avoid shuffle,
requiring an exact attribute-level match between partition keys and join keys
when `true` (the default). This forced users to set
`requireAllClusterKeysForCoPartition=false` solely to enable SPJ, even for safe
cases where all partition keys appear in the join keys (just not in exact
order/count).
The check is replaced with a direct `partitioning.satisfies(distribution)`
call, which delegates to the existing V2 partitioning satisfaction logic in
`KeyedPartitioning.groupedSatisfies`. That logic already encodes the correct
skew-protection threshold via `v2BucketingAllowKeysSubsetOfPartitionKeys`:
- `false` (default): all partition keys must appear in the join keys
(order-independent) -- no skew risk
- `true`: subset of partition keys allowed -- user explicitly opts into skew
risk
The config still gates hash partitioning shuffle reuse (e.g., V1 bucketing)
via `HashShuffleSpec.canCreatePartitioning` /
`NullAwareHashShuffleSpec.canCreatePartitioning` in `partitioning.scala`, which
are unchanged. Its `SQLConf` doc is updated to clarify this scope.
Docs and tests are updated accordingly: the migration guide documents the
change, the SPJ config table in `sql-performance-tuning.md` no longer lists the
config, the `allowKeysSubsetOfPartitionKeys` description and the SPJ example
drop the `requireAllClusterKeysForCoPartition=false` prerequisite, and
`KeyGroupedPartitioningSuite` removes the now-redundant
`requireAllClusterKeysForCoPartition=false` settings from all V2 SPJ tests.
### Why are the changes needed?
The `requireAllClusterKeysForCoPartition` exact-match gate for V2 SPJ was
unnecessarily strict and added no skew protection:
1. **Exact-match vs all-keys-in-join (order-independent)**: For V2
value-based partitioning, reordering join keys or having extra join keys does
not change partition assignment or introduce skew. The exact-match check
blocked safe SPJ cases (e.g., join keys in different order than partition keys)
without preventing any skew that the all-keys check wouldn't already prevent.
2. **Config burden without protection**: Setting
`requireAllClusterKeysForCoPartition=false` merely relaxed the exact-match
check to the all-partition-keys-in-join-keys rule that `partitioning.satisfies`
already implements, so the config forced users to flip a switch to get behavior
that is safe unconditionally. The only real skew threshold, joining on a subset
of partition keys, is owned by `v2BucketingAllowKeysSubsetOfPartitionKeys`,
which additionally had to be enabled for that case.
3. **Inconsistency with `requireAllClusterKeysForDistribution`**: That
sibling config (for aggregate/window) already delegates to
`partitioning.satisfies` via `ClusteredDistribution.requireAllClusterKeys`,
which is the pattern this PR adopts for co-partitioning V2 joins.
### Does this PR introduce _any_ user-facing change?
Yes. After this PR, `spark.sql.requireAllClusterKeysForCoPartition` no
longer affects storage-partitioned joins (V2 data sources). The behavioral
change vs. released Spark 4.3:
**Before**: A join on two V2 tables partitioned by `(a, b)` with join
condition `ON t1.b = t2.b AND t1.a = t2.a` (keys in reverse order) would
require a shuffle, because the exact-match check expected join keys in the same
order as partition keys. Users had to set
`spark.sql.requireAllClusterKeysForCoPartition=false` to avoid the shuffle.
**After**: The same join avoids shuffle by default, because both partition
keys (`a`, `b`) appear in the join keys regardless of order. No config change
is needed.
This is a safe relaxation: value-based V2 partitioning co-locates rows by
partition values, not by key ordering. The only real skew threshold -- whether
to allow joining on a *subset* of partition keys -- remains controlled by
`spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled`
(default `false`).
Users who previously set `requireAllClusterKeysForCoPartition=false` solely
for V2 SPJ no longer need to do so. The config still applies to hash
partitioning (e.g., V1 bucketing) and is documented accordingly.
### How was this patch tested?
- Updated `EnsureRequirementsSuite`:
- The "Check with KeyedPartitioning" case that expected shuffles under the
default config is replaced by a dedicated test "KeyedPartitioning: SPJ is not
affected by requireAllClusterKeysForCoPartition", which exercises both config
values (`true` and `false`) and verifies no shuffle is introduced either way.
- The "KeyedPartitioning with ... = false" test is renamed to
"KeyedPartitioning with subset of join keys" and no longer requires any config
override -- V2 SPJ works with default settings when all partition keys are in
the join keys.
- All V2 `KeyedPartitioning` test call sites that previously used
`applyEnsureRequirementsWithSubsetKeys` (which set
`requireAllClusterKeysForCoPartition=false`) now use `EnsureRequirements.apply`
directly.
- `KeyGroupedPartitioningSuite`: removed the now-redundant
`requireAllClusterKeysForCoPartition=false` settings from all V2 SPJ tests, so
they run with the default value. `ShuffleSpecSuite` is unchanged; it covers
`HashShuffleSpec` / `NullAwareHashShuffleSpec`, which still honor the config.
- Ran all three suites: `ShuffleSpecSuite` (14 tests),
`EnsureRequirementsSuite` (31 tests), `KeyGroupedPartitioningSuite` (98 tests)
-- all pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Fable 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]
