[
https://issues.apache.org/jira/browse/SPARK-59080?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-59080:
-----------------------------------
Labels: pull-request-available (was: )
> 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. 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]
