[
https://issues.apache.org/jira/browse/SPARK-59256?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth reassigned SPARK-59256:
----------------------------------
Assignee: Peter Toth
> 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
>
> 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: [email protected]
For additional commands, e-mail: [email protected]
