ulysses-you opened a new pull request, #57491:
URL: https://github.com/apache/spark/pull/57491
### What changes were proposed in this pull request?
`UnionExec.outputPartitioning` (SPARK-52921) passes a child partitioning
through the union when every child is compatibly partitioned, letting a
downstream operator (e.g. an aggregate) skip a shuffle. This PR extends that
pass-through to children whose `outputPartitioning` is a
`PartitioningCollection`.
- `outputPartitioning` now treats each child's partitioning as a set of
candidate partitionings (a `PartitioningCollection` flattens to its members; a
single partitioning is a one-element set) and passes through the
**intersection** across all children: empty -> `UnknownPartitioning`; one
member -> that partitioning; many -> a `PartitioningCollection`. Only
index-co-locatable partitionings (`HashPartitioningLike` / `SinglePartition`)
participate.
- The existing `KeyedPartitioning` concatenation path is kept as a separate
case that fires only when every child is a single `KeyedPartitioning`. It is a
distinct physical strategy (`sparkContext.union`, `numPartitions = sum`) and
cannot be folded into the co-located intersection
(`SQLPartitioningAwareUnionRDD`, `numPartitions = N`), because a
`PartitioningCollection` requires uniform `numPartitions` across its members.
- `comparePartitioning` is documented as a leaf-only equivalence predicate;
collections are flattened before it is called.
### Why are the changes needed?
`comparePartitioning` had no case for `PartitioningCollection`, so whenever
a child reported one, the whole union collapsed to `UnknownPartitioning` and an
extra shuffle was inserted. This is hit by a common query shape - a `UNION ALL`
where one branch is an inner shuffled-hash join (whose `outputPartitioning` is
`PartitioningCollection(Hash(k1), Hash(k2))`) and the other is a left join
(single `HashPartitioning`), feeding a `GROUP BY`:
```sql
select c1, c2, c3, count(*) from (
select t1.c1, t1.c2, t1.c3 from t1 join t2 on t1.c1 = t2.c1
union all
select t3.c1, t3.c2, t3.c3 from t3 left join t4 on t3.c1 = t4.c1
) group by c1, c2, c3
```
The union failed to pass through and a redundant `ENSURE_REQUIREMENTS`
shuffle appeared before the aggregate.
### Does this PR introduce _any_ user-facing change?
No change in query results. Under the default-on
`spark.sql.unionOutputPartitioning`, affected plans drop a redundant shuffle.
### How was this patch tested?
Added tests to `DataFrameSetOperationsSuite` covering: the mixed inner+left
join union (intersection to a single `HashPartitioning`, shuffle eliminated);
all children reporting collections (pass-through as a
`PartitioningCollection`); empty intersection (fall back to
`UnknownPartitioning`); and the collection pass-through under AQE. Existing
`DataFrameSetOperationsSuite` and `KeyGroupedPartitioningSuite` (97 tests)
regressions pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
--
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]
