[
https://issues.apache.org/jira/browse/SPARK-59080?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth updated SPARK-59080:
-------------------------------
Description:
Under `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled`,
`KeyedPartitioning.createShuffleSpec` projects each member of a
`PartitioningCollection` onto *its own* join-key subset, so the members of the
resulting `ShuffleSpecCollection` can end up with different `numPartitions`.
`ShuffleSpecCollection.numPartitions` then reports the head member's arbitrary
count, and that count also feeds the `maxBy(_.numPartitions)` that picks the
best spec, while `ShuffleSpecCollection.createPartitioning` requires all
members to agree and throws:
{noformat}
expected all specs in the collection to have the same number of partitions
{noformat}
thrown from `EnsureRequirements`, so planning fails outright.
Reproduced with `items` partitioned by `[identity(id), identity(arrive_time)]`
and one row per split, rows `(1,'aa',40,01-01), (1,'ab',30,01-02),
(3,'bb',10,01-01), (4,'cc',15.5,02-01)`, `purchases` unpartitioned, and
`v2BucketingShuffleEnabled=true`, `partiallyClusteredDistribution=false`,
`allowKeysSubsetOfPartitionKeys=true`:
{code:sql}
SELECT /*+ MERGE(i, p) */ id, t1, t2, i.price AS purchase_price, p.price AS
sale_price
FROM (SELECT id, arrive_time AS t1, arrive_time AS t2, price FROM
testcat.ns.items) i
JOIN testcat.ns.purchases p ON i.id = p.item_id AND i.t1 = p.time
{code}
The two aliases of `arrive_time` make the alias cross-product produce
collection members that cover different numbers of join keys, hence the
differing counts. Independent of SPARK-59025: it fails identically before and
after that change.
Fix shape: in `PartitioningCollection.createShuffleSpec`, keep only the members
whose spec has the maximum `numPartitions` - the finest granularity, i.e. the
members covering the most join keys. That also makes the head member principled
rather than dependent on the order in which the alias cross-product was
enumerated.
Summary: ShuffleSpecCollection members can disagree on numPartitions,
so planning throws (was: Push each join child's own key positions in the
multi-child SPJ block)
> ShuffleSpecCollection members can disagree on numPartitions, so planning
> throws
> -------------------------------------------------------------------------------
>
> Key: SPARK-59080
> URL: https://issues.apache.org/jira/browse/SPARK-59080
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 5.0.0
> Reporter: Peter Toth
> Priority: Major
>
> Under
> `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled`,
> `KeyedPartitioning.createShuffleSpec` projects each member of a
> `PartitioningCollection` onto *its own* join-key subset, so the members of
> the resulting `ShuffleSpecCollection` can end up with different
> `numPartitions`.
> `ShuffleSpecCollection.numPartitions` then reports the head member's
> arbitrary count, and that count also feeds the `maxBy(_.numPartitions)` that
> picks the best spec, while `ShuffleSpecCollection.createPartitioning`
> requires all members to agree and throws:
> {noformat}
> expected all specs in the collection to have the same number of partitions
> {noformat}
> thrown from `EnsureRequirements`, so planning fails outright.
> Reproduced with `items` partitioned by `[identity(id),
> identity(arrive_time)]` and one row per split, rows `(1,'aa',40,01-01),
> (1,'ab',30,01-02), (3,'bb',10,01-01), (4,'cc',15.5,02-01)`, `purchases`
> unpartitioned, and `v2BucketingShuffleEnabled=true`,
> `partiallyClusteredDistribution=false`, `allowKeysSubsetOfPartitionKeys=true`:
> {code:sql}
> SELECT /*+ MERGE(i, p) */ id, t1, t2, i.price AS purchase_price, p.price AS
> sale_price
> FROM (SELECT id, arrive_time AS t1, arrive_time AS t2, price FROM
> testcat.ns.items) i
> JOIN testcat.ns.purchases p ON i.id = p.item_id AND i.t1 = p.time
> {code}
> The two aliases of `arrive_time` make the alias cross-product produce
> collection members that cover different numbers of join keys, hence the
> differing counts. Independent of SPARK-59025: it fails identically before and
> after that change.
> Fix shape: in `PartitioningCollection.createShuffleSpec`, keep only the
> members whose spec has the maximum `numPartitions` - the finest granularity,
> i.e. the members covering the most join keys. That also makes the head member
> principled rather than dependent on the order in which the alias
> cross-product was enumerated.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
