[
https://issues.apache.org/jira/browse/SPARK-59256?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth updated SPARK-59256:
-------------------------------
Description:
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.
was:
`ShuffleSpecCollection` answers two different kinds of question with one type.
* *Matching*: `isCompatibleWith` succeeds when **any** member matches. That is
a collection-level question and it is the reason the type exists.
* *Shuffle template*: `createPartitioning` and `numPartitions` need **one**
member, and the collection picks `specs.head` - whichever member the alias
cross-product enumerated first.
The second answer is a guess. Which member is right depends on what the *other*
side matched, and that is not knowable inside the collection.
SPARK-59080 fixed the consequence: `EnsureRequirements` now resolves the agreed
member itself and calls `createPartitioning` on that member, so the collection
is never asked. This ticket removes the ability to ask.
h3. What changes
Split the trait so that the shuffle-template contract is a separate type, and
`ShuffleSpecCollection` carries only `isCompatibleWith`.
* `createPartitioning` leaves the collection. After SPARK-59080 it has no
production caller, and the only thing standing between a future caller and a
wrong partitioning is a runtime `require`. The compiler should enforce that
instead.
* `numPartitions` leaves the collection. Its consumers get an answer that does
not read an arbitrary member:
** `EnsureRequirements`, `finalCandidateSpecs.values.maxBy(_.numPartitions)`,
which ranks the children to pick the reference layout. A head that understates
the count makes a child lose that ranking, so the join lands on a coarser
layout: an extra shuffle or less parallelism, never wrong results.
** `SinglePartitionShuffleSpec.isCompatibleWith`, which reads
`other.numPartitions == 1`. Reached from `ValidateRequirements`, where
`specs.tail.forall(_.isCompatibleWith(specs.head))` can put a collection on the
right-hand side. A collection whose head projects to one partition while
another member does not then answers on the head's authority.
Neither consumer is measured, and I could not build a query that reaches the
second one.
h3. Scope
Master only. No user-facing change and no known bug, so there is nothing to
backport. The behavioural part is limited to the ranking answer; everything
else is a type move.
Summary: Choose a ShuffleSpecCollection member by pairing, not by
enumeration order (was: Remove createPartitioning and numPartitions from
ShuffleSpecCollection)
> 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
> 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]
