[
https://issues.apache.org/jira/browse/SPARK-59080?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth updated SPARK-59080:
-------------------------------
Affects Version/s: 4.1.0
4.2.0
4.3.0
4.0.0
Description:
Under `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled`,
`KeyedPartitioning.createShuffleSpec` projects each member of a
`PartitioningCollection` onto *its own* join-key subset and drops the duplicate
keys that projection creates. The members of the resulting
`ShuffleSpecCollection` can therefore end up with different `numPartitions`.
`EnsureRequirements` then asks that collection for a shuffle template, and
`ShuffleSpecCollection.createPartitioning` requires all members to agree:
{noformat}
expected all specs in the collection to have the same number of partitions
{noformat}
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.
h3. Why the collection cannot answer the question
`ShuffleSpecCollection.isCompatibleWith` succeeds when *any* member matches, so
the collection alone never said which member the two sides agreed on.
`createPartitioning` falls back to `specs.head`, which is whichever member the
alias cross-product enumerated first, and `numPartitions` reads the head's
count too.
Narrowing the collection to its finest members would satisfy the `require`, but
it would still be a local guess. The right member is the one the *other* side
matched, and that is not knowable inside the collection.
h3. Fix shape
Resolve the member in `EnsureRequirements`, where both sides are visible. Pick
the member that every matched child is compatible with, preferring the finest
when several qualify, and use it to build the re-shuffled child's partitioning.
When no member serves every matched child there is no shared layout, so every
child takes the ordinary shuffle. The `require` stays as a guard on a method
that then has no production caller.
The `joinKeyPositions` pushed into a compatible child should come from that
child's own matching member, since they index into that child's partition
expressions. The same change fixes that. It is latent: no query reaches the
wrong case today, because a join is handled by `checkKeyGroupCompatible`, which
already pushes each side's own positions, and a cogroup's grouping key is
synthesized so neither side stays keyed.
was:
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: Pick one ShuffleSpecCollection member for the SPJ
pushdown and the re-shuffle (was: ShuffleSpecCollection members can disagree
on numPartitions, so planning throws)
> Pick one ShuffleSpecCollection member for the SPJ pushdown and the re-shuffle
> -----------------------------------------------------------------------------
>
> Key: SPARK-59080
> URL: https://issues.apache.org/jira/browse/SPARK-59080
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.1.0, 4.0.0, 4.2.0, 4.3.0, 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 and drops the
> duplicate keys that projection creates. The members of the resulting
> `ShuffleSpecCollection` can therefore end up with different `numPartitions`.
> `EnsureRequirements` then asks that collection for a shuffle template, and
> `ShuffleSpecCollection.createPartitioning` requires all members to agree:
> {noformat}
> expected all specs in the collection to have the same number of partitions
> {noformat}
> 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.
> h3. Why the collection cannot answer the question
> `ShuffleSpecCollection.isCompatibleWith` succeeds when *any* member matches,
> so the collection alone never said which member the two sides agreed on.
> `createPartitioning` falls back to `specs.head`, which is whichever member
> the alias cross-product enumerated first, and `numPartitions` reads the
> head's count too.
> Narrowing the collection to its finest members would satisfy the `require`,
> but it would still be a local guess. The right member is the one the *other*
> side matched, and that is not knowable inside the collection.
> h3. Fix shape
> Resolve the member in `EnsureRequirements`, where both sides are visible.
> Pick the member that every matched child is compatible with, preferring the
> finest when several qualify, and use it to build the re-shuffled child's
> partitioning. When no member serves every matched child there is no shared
> layout, so every child takes the ordinary shuffle. The `require` stays as a
> guard on a method that then has no production caller.
> The `joinKeyPositions` pushed into a compatible child should come from that
> child's own matching member, since they index into that child's partition
> expressions. The same change fixes that. It is latent: no query reaches the
> wrong case today, because a join is handled by `checkKeyGroupCompatible`,
> which already pushes each side's own positions, and a cogroup's grouping key
> is synthesized so neither side stays keyed.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
