[jira] [Resolved] (SPARK-59256) Choose a ShuffleSpecCollection member by pairing, not by enumeration order

["Peter Toth (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/SPARK-59256?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Peter Toth resolved SPARK-59256.
--------------------------------
    Fix Version/s: 4.4.0
       Resolution: Fixed

Issue resolved by pull request 58531
[https://github.com/apache/spark/pull/58531]

> Choose a ShuffleSpecCollection member by pairing, not by enumeration order
> --------------------------------------------------------------------------
>
>                 Key: SPARK-59256
>                 URL: https://issues.apache.org/jira/browse/SPARK-59256
>             Project: Spark
>          Issue Type: Improvement
>          Components: SQL
>    Affects Versions: 5.0.0
>            Reporter: Peter Toth
>            Assignee: Peter Toth
>            Priority: Major
>              Labels: pull-request-available
>             Fix For: 4.4.0
>
>
> A `PartitioningCollection` offers several layouts, and which one is right depends on what the other side matched. Three places in `EnsureRequirements` decided it by enumeration order instead, and none of them could see both sides:
> # the per-child branch that shuffles a child read `specs.head` through `ShuffleSpecCollection.createPartitioning`. **Fixed by SPARK-59080.**
> # the child ranking, `finalCandidateSpecs.values.maxBy(_.numPartitions)`, reads the collection's `numPartitions`, which is `specs.head`'s.
> # `createKeyedShuffleSpec` collapses a collection to one member with `collectFirst`, per side, before either side has seen the other.
> This ticket is for 2 and 3, and for making the type say why a collection cannot answer alone.
> h3. Why the collection should not answer at all
> `isCompatibleWith` succeeds when *any* member matches, which is why the type exists. The two single-member questions are not wrong in the same way:
> * `createPartitioning` has *no local answer*. After SPARK-59080 it has no production caller, and only a runtime `require` stands between a future caller and a wrong partitioning.
> * `numPartitions` has *two consumers wanting two different aggregations*, which is why the answer belongs to each caller rather than to the collection:
> ** the child ranking wants the **max** over the members;
> ** `SinglePartitionShuffleSpec.isCompatibleWith`, reached from `ValidateRequirements`, wants **exists**.
> h3. What changes
> * A `LeafShuffleSpec` sub-trait carries `numPartitions` and `createPartitioning`; the seven concrete specs extend it and `ShuffleSpecCollection` extends `ShuffleSpec` alone. `ShuffleSpec` becomes sealed, so the two kinds are the only kinds and `flatten` can return `Seq[LeafShuffleSpec]` without a fallback case.
> * The ranking takes the max over the flattened members.
> * `SinglePartitionShuffleSpec.isCompatibleWith` unwraps a collection the way every other spec already does.
> * `createKeyedShuffleSpecs` returns every member's spec, and `checkKeyGroupCompatible` picks the *pair* that agrees on the keys and offers the most parallelism. Without that, two sides can pick members that do not agree, the check declines, and the join loses the storage-partitioned pushdown even though a pairing existed. The pick cannot be an independent per-side finest: a side whose only members are coarse would then fail to pair.
> h3. Scope
> Master only. No user-facing change and no known bug, so nothing to backport.
> The three behaviour changes are gated differently, which is worth not conflating. The ranking and the `SinglePartitionShuffleSpec` change are no-ops unless `allowKeysSubsetOfPartitionKeys` is on, because `PartitioningCollection` requires its members to agree on `numPartitions` and every spec reports its own partitioning's count, so `max` equals `head`. The pairing reads no count, so that argument does not cover it: it needs `requireAllClusterKeysForCoPartition` off, which is what otherwise refuses a member that does not cover every clustering key. Neither config is the default.
> h3. Credit
> Items 2 and 3 were both raised by [~LuciferYang] in review of apache/spark#58527.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org