dongjoon-hyun opened a new pull request, #58316:
URL: https://github.com/apache/spark/pull/58316
### What changes were proposed in this pull request?
This PR fixes two spots in `partitioning.scala` where the
`KeyedPartitioning.isNarrowed` flag
was silently dropped:
- `KeyedShuffleSpec.createPartitioning` built the shuffled side's
`KeyedPartitioning` with the
3-argument constructor, so `isNarrowed` fell back to the default `false`
even when the
template partitioning was narrowed.
- `KeyedPartitioning.toGrouped` likewise dropped the flag when building the
grouped variant.
Both now propagate `isNarrowed` from the source partitioning.
### Why are the changes needed?
`isNarrowed` marks a `KeyedPartitioning` that was derived from a
finer-grained one by dropping
key positions. `AliasAwareOutputExpression` documents that this flag must be
sticky: a
narrowed, non-grouped partitioning must not satisfy `ClusteredDistribution`
via grouping
unless
`spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled` is
on,
because `GroupPartitionsExec` would merge partitions that held distinct keys
in the original
partitioning, carrying the same skew risk.
`KeyedShuffleSpec.createPartitioning` broke that stickiness. When a
narrowed-but-grouped
partitioning is used as the shuffle template for the other join leg (with
`spark.sql.sources.v2.bucketing.shuffle.enabled`), the shuffled side mirrors
the narrowed
side's partition keys but declared `isNarrowed=false`. If an outer join then
exposes only the
shuffled side and `UnionExec` merges it with an overlapping-key sibling into
an ungrouped
`KeyedPartitioning`, the `isNarrowed && !isGrouped` guard in
`groupedSatisfies` never fires,
and a downstream operator can coalesce the narrowed keys with
`GroupPartitionsExec` without
the opt-in config — bypassing the skew protection. The new end-to-end test
reproduces this
chain and fails without the fix.
The flag was lost in the refactoring of SPARK-55535/SPARK-55092
(a1c62ddde85).
### Does this PR introduce _any_ user-facing change?
Yes, a narrow plan-level fix. In the multi-stage scenario above, Spark
previously coalesced
partitions derived from a narrowed partitioning without
`allowJoinKeysSubsetOfPartitionKeys.enabled`, risking skewed partitions; it
now inserts a
shuffle unless the config is enabled. Query results are unchanged.
### How was this patch tested?
- New end-to-end test in `KeyGroupedPartitioningSuite` reproducing the full
chain
(narrowing projection -> shuffle from the narrowed template -> right outer
join -> union
merge -> aggregate), verified to fail without the fix.
- New unit test in `DistributionSuite` pinning flag propagation through
`toGrouped` and
`KeyedShuffleSpec.createPartitioning`.
- Passed `KeyGroupedPartitioningSuite` (100 tests), `DistributionSuite`,
`ProjectedOrderingAndPartitioningSuite`, and `EnsureRequirementsSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: 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]
