ulysses-you opened a new pull request, #58771: URL: https://github.com/apache/spark/pull/58771
### What changes were proposed in this pull request? Adds a per-key skew split to the storage-partitioned join alignment, the SPJ counterpart of the AQE skew join: - A key group whose combined input partition count over both sides is larger than `max(skewedPartitionsPerGroupThreshold, median group count * skewedGroupFactor)` spreads the side holding more of its splits over output partitions holding about `advisoryInputPartitionsPerOutputPartition` each, and the other side replicates its group to each of those. Which side spreads is gated per key by what the join type may replicate, and the spreading side's own count must also exceed the threshold. - The mechanism is per-key alignment modes: `GroupPartitionsExec.expectedPartitionKeys` entries become `ExpectedPartitionKey(key, numSplits, distribute, skewed)`, replacing the side-wide `distributePartitions` flag. Distribute mode with a single reserved partition concatenates the key's splits as before; fewer reserved partitions than splits spread them over contiguous chunks of `ceil(size / numSplits)`, padding the leftover slots so the emitted count stays exactly the reserved one that pairs the two sides. - A pair that is compatible as it stands and reserves nothing goes back untouched. The split stands down when a reducer regroups the keys the counts cover, and over a child whose layout carries `mayContainUnknownPartitionKeys`, since any non-identity grouping makes `GroupPartitionsExec` give up the keyed partitioning (SPARK-59050) that the plan above the join relies on. - Counts are the only static evidence a data source reports at planning time, so a key spread over many small input partitions is split like a large one, and a key held in one huge input partition is not split at all; a split trades total reads for per-task reads. ### Why are the changes needed? A storage-partitioned join replaces the shuffle with an alignment that coalesces each key's input partitions into one output partition per side, so a key holding disproportionately many input partitions leaves its whole join to a single task. There is no fallback for it later in the plan: the AQE skew join splits skewed shuffle partitions once the map-side sizes are known, and a plan that never shuffles has neither those sizes nor a shuffle read to split. The fix has to be made where the alignment is decided. That place is `EnsureRequirements`, and it is a planning-time decision rather than a runtime one, because the alignment is pushed into the data sources: the per-key output partition counts become the expected partition keys the scans are replanned against, so the counts have to be settled before the scan partitioning is built. The inputs available there are the per-key input partition counts of the two sides, read from their pre-alignment plans by unwrapping the grouping this rule inserted in an earlier pass, together with the merged key set the alignment already computes. Reading the pre-alignment plans, rather than the aligned reports, is what makes the decision reproducible when the rule runs again over a plan it already produced, as AQE re-planning does. The alignment then reserves the same partition count per key on both sides, which is what keeps their output partitions paired by position: the side holding more of the key's splits spreads them over the reserved partitions, and the other side repeats its group once per reserved partition. Both sides derive their entries from that one shared list, and only the distribute/replicate mode is per side. Splits of a key whose spread lands on a single partition are not applied at all, since that layout is the one the default alignment already produces. ### Does this PR introduce _any_ user-facing change? No. The four configs it adds all default to off, so no plan changes unless they are enabled: | Config | Default | Meaning | |---|---|---| | `spark.sql.sources.v2.bucketing.skewJoin.enabled` | `false` | Whether to split skewed key groups instead of coalescing each key's input partitions into one output partition per side. | | `spark.sql.sources.v2.bucketing.skewJoin.skewedPartitionsPerGroupThreshold` | `5` | A key group is skewed when its combined input partition count over both sides is larger than the maximum of this threshold and the median group count times `skewedGroupFactor`. | | `spark.sql.sources.v2.bucketing.skewJoin.skewedGroupFactor` | `5.0` | The factor the median group count is multiplied by in that comparison, the partition-count counterpart of `spark.sql.adaptive.skewJoin.skewedPartitionFactor`. | | `spark.sql.sources.v2.bucketing.skewJoin.advisoryInputPartitionsPerOutputPartition` | `1` | The number of input partitions each output partition holds when a skewed key's splits are spread, the partition-count counterpart of `spark.sql.adaptive.advisoryPartitionSizeInBytes`. | ### How was this patch tested? New tests, all prefixed `SPARK-59436`: - `EnsureRequirementsSuite`: the decision itself, covering which side spreads, the join-type replication gate, the per-side trigger, advisory chunking, the no-op / marked-child / reducer stand-downs, re-run stability and the interaction with the partially clustered distribution. - `GroupPartitionsExecSuite`: the alignment modes, covering a single reserved partition in distribute mode, contiguous chunking, and per-key mixed modes within one alignment. - `KeyGroupedPartitioningSuite`: end to end on the in-memory catalog with AQE both on and off, join to an aggregate and to a second join, semi/left outer/right outer/full outer join types, partially clustered co-existence and per-key flip, and a marked child reached by a real plan. Ran `KeyGroupedPartitioningSuite`, `AdaptiveQueryExecSuite`, `EnsureRequirementsSuite`, `GroupPartitionsExecSuite` and `SQLConfSuite` locally; 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 (deepseek-v4.1-flash) -- 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]
