peter-toth commented on code in PR #58527:
URL: https://github.com/apache/spark/pull/58527#discussion_r3940598901
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -247,30 +246,50 @@ case class EnsureRequirements(
}
}
+ // A `ShuffleSpecCollection` answers `isCompatibleWith` if *any* of its
members does, so the
+ // collection alone does not say which member the sides agreed on. The
projection pushed into
+ // a compatible child and the partitioning built for a re-shuffled child
both have to come
+ // from one member, otherwise the sides end up grouped on different
keys, or on a key set the
+ // child does not even have. Pick that member once, preferring the
finest when several
+ // qualify. Only the branch that shuffles a child reads these, hence
`lazy`.
+ lazy val matchedIndexes = bestSpecOpt.toSeq.flatMap { best =>
+ childrenIndexes.filter(i => best.isCompatibleWith(specs(i)))
+ }
+ lazy val bestMemberOpt = bestSpecOpt.flatMap { best =>
Review Comment:
Agreed, and this is exactly what I had queued as the follow-up:
apache/spark#58531 (SPARK-59256). It takes the max over the flattened members
at the call site, for the reason you give — changing `numPartitions` itself
would affect other readers.
Worth naming the other reader, since it turned out to want a different
aggregation: `SinglePartitionShuffleSpec.isCompatibleWith` reads
`other.numPartitions == 1`, reached from `ValidateRequirements`, and there the
right answer is `exists`, not `max`. Two consumers wanting two different
aggregations is what convinced me the collection should not answer
`numPartitions` at all rather than answer it better, so that PR removes it
along with `createPartitioning`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -247,30 +246,50 @@ case class EnsureRequirements(
}
}
+ // A `ShuffleSpecCollection` answers `isCompatibleWith` if *any* of its
members does, so the
+ // collection alone does not say which member the sides agreed on. The
projection pushed into
+ // a compatible child and the partitioning built for a re-shuffled child
both have to come
+ // from one member, otherwise the sides end up grouped on different
keys, or on a key set the
+ // child does not even have. Pick that member once, preferring the
finest when several
+ // qualify. Only the branch that shuffles a child reads these, hence
`lazy`.
+ lazy val matchedIndexes = bestSpecOpt.toSeq.flatMap { best =>
+ childrenIndexes.filter(i => best.isCompatibleWith(specs(i)))
+ }
+ lazy val bestMemberOpt = bestSpecOpt.flatMap { best =>
+ val matchedMembers = matchedIndexes.map(i => flattenSpec(specs(i)))
+ // No member serving every matched child means there is no layout to
align them on, so they
+ // all take the ordinary shuffle. That needs three or more clustered
children, since with
+ // two the member that reported the match serves both, and no operator
has three today.
+ flattenSpec(best)
+ .filter(m => matchedMembers.forall(_.exists(m.isCompatibleWith)))
+ .maxByOption(_.numPartitions)
Review Comment:
Good catch, and I took it into apache/spark#58531 rather than a separate
ticket. `createKeyedShuffleSpecs` now returns every member's spec and
`checkKeyGroupCompatible` picks the pair, with your constraint respected: not
an independent per-side finest, but the pair that agrees on the keys and offers
the most parallelism.
Two things I measured that are worth passing back.
Your reachability framing is slightly off, in a way that does not change
your conclusion. `requireAllClusterKeysForCoPartition = true` does not always
keep a collection down to one candidate: I instrumented
`createKeyedShuffleSpec` over `KeyGroupedPartitioningSuite` and
`EnsureRequirementsSuite` and got 31 collection reaches, of which 2 had two
qualifying members under the default. But in every one of those the qualifying
members produced *identical* specs, so the pick was not observable. The
observable case does need the relaxed conf, as you said — just not for the
reason that the members are filtered out.
The shape is narrower than it looks, too. Two members can only differ in
coverage if the clustering is wider than the partitioning arity, since the
collection requires its members to have matching arity. With that, the repro
is: each side offering a member for a different clustering-key subset, in the
opposite order, so taking each side's first member pairs two that do not agree.
On the fix that pairs correctly and no shuffle is needed at all; without it
`checkKeyGroupCompatible` declines and each side is merely grouped on its own
keys.
I also checked the sibling `collectFirst` at the top of the
partially-clustered branch, which recovers the original `KeyedPartitioning`
while the positions come from the picked member. That one is harmless and I am
not touching it: `projectKeys` reads only `partitionKeys` and the types derived
from them, and the collection invariant makes `partitionKeys` a shared
reference, so any member gives the same answer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
