[
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 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, and it is reachable rather than
latent. A join is handled by `checkKeyGroupCompatible`, which already pushes
each side's own positions, and the Scala `CoGroupExec`'s key comes from an
`AppendColumns` that no `KeyedPartitioning` satisfies - but a Pandas or Arrow
cogroup groups on real columns, so both of its children stay keyed and reach
this branch. With their partition expressions laid out differently, the second
side is handed the first side's positions and ends up grouped on its other
partition column.
The only part that stays latent is the case of three or more clustered
children, which no operator has.
was:
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.
> 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
> Labels: pull-request-available
>
> 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, and it is reachable rather than
> latent. A join is handled by `checkKeyGroupCompatible`, which already pushes
> each side's own positions, and the Scala `CoGroupExec`'s key comes from an
> `AppendColumns` that no `KeyedPartitioning` satisfies - but a Pandas or Arrow
> cogroup groups on real columns, so both of its children stay keyed and reach
> this branch. With their partition expressions laid out differently, the
> second side is handed the first side's positions and ends up grouped on its
> other partition column.
> The only part that stays latent is the case of three or more clustered
> children, which no operator has.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
