[
https://issues.apache.org/jira/browse/SPARK-58594?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth updated SPARK-58594:
-------------------------------
Description:
`UnionExec.outputPartitioning` merges its children's `KeyedPartitioning`s only
when every child reports a single one and all of them carry semantically equal
expressions at equal arity. Two consequences:
* A child that reports a `PartitioningCollection` of `KeyedPartitioning`s is
not merged at all. An inner `ShuffledJoin` reports
`PartitioningCollection(left, right)`, so a storage-partitioned join inside a
`UNION ALL` leg produces exactly that shape, and so does a projection that
aliases a partition key to more than one output name.
* Legs whose partition expressions differ in arity or order are not merged,
even when they agree on a subset of key positions. `KP([k1, k2])` and
`KP([k2])` could merge into `KP([k2], isNarrowed = true)`, and `KP([k1, k2])`
with `KP([k2, k1])` could merge by reordering.
In both cases the union falls back to `UnknownPartitioning` and the
storage-partitioned join is lost.
This revises the merge to work per key position, the same way SPARK-46367 made
`PartitioningPreservingUnaryExecNode` narrow a `KeyedPartitioning` instead of
dropping it. A position survives when every leg has a semantically equal
expression for it; the first leg's `KeyedPartitioning`s are filtered and
narrowed to the surviving positions and deduplicated; each leg's partition keys
are projected to its own positions for those expressions and concatenated.
`isNarrowed` is set when any leg loses a position, or when any leg's
partitioning was already narrowed, which keeps the existing skew gating in
`groupedSatisfies` intact.
Note that narrowing only pays off without configuration when the projected keys
stay distinct. When narrowing introduces duplicate keys, `groupedSatisfies`
requires
`spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled`,
exactly as for the narrowing projection added by SPARK-46367.
Summary: Revise KeyedPartitioning handling in UnionExec output
partitioning (was: Union output partitioning should merge KeyedPartitioning in
PartitioningCollection children)
> Revise KeyedPartitioning handling in UnionExec output partitioning
> ------------------------------------------------------------------
>
> Key: SPARK-58594
> URL: https://issues.apache.org/jira/browse/SPARK-58594
> Project: Spark
> Issue Type: Improvement
> Components: SQL
> Affects Versions: 5.0.0
> Reporter: Peter Toth
> Priority: Major
> Labels: pull-request-available
>
> `UnionExec.outputPartitioning` merges its children's `KeyedPartitioning`s
> only when every child reports a single one and all of them carry semantically
> equal expressions at equal arity. Two consequences:
> * A child that reports a `PartitioningCollection` of `KeyedPartitioning`s is
> not merged at all. An inner `ShuffledJoin` reports
> `PartitioningCollection(left, right)`, so a storage-partitioned join inside a
> `UNION ALL` leg produces exactly that shape, and so does a projection that
> aliases a partition key to more than one output name.
> * Legs whose partition expressions differ in arity or order are not merged,
> even when they agree on a subset of key positions. `KP([k1, k2])` and
> `KP([k2])` could merge into `KP([k2], isNarrowed = true)`, and `KP([k1, k2])`
> with `KP([k2, k1])` could merge by reordering.
> In both cases the union falls back to `UnknownPartitioning` and the
> storage-partitioned join is lost.
> This revises the merge to work per key position, the same way SPARK-46367
> made `PartitioningPreservingUnaryExecNode` narrow a `KeyedPartitioning`
> instead of dropping it. A position survives when every leg has a semantically
> equal expression for it; the first leg's `KeyedPartitioning`s are filtered
> and narrowed to the surviving positions and deduplicated; each leg's
> partition keys are projected to its own positions for those expressions and
> concatenated. `isNarrowed` is set when any leg loses a position, or when any
> leg's partitioning was already narrowed, which keeps the existing skew gating
> in `groupedSatisfies` intact.
> Note that narrowing only pays off without configuration when the projected
> keys stay distinct. When narrowing introduces duplicate keys,
> `groupedSatisfies` requires
> `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled`,
> exactly as for the narrowing projection added by SPARK-46367.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
