peter-toth commented on code in PR #58527:
URL: https://github.com/apache/spark/pull/58527#discussion_r3947301890
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/ShuffleSpecSuite.scala:
##########
@@ -690,4 +690,34 @@ class ShuffleSpecSuite extends SparkFunSuite with
SQLHelper {
expected = false
)
}
+
+ test("SPARK-59080: a collection whose members cover different key subsets
disagrees") {
Review Comment:
Exactly right, and your list of the three is exactly the set: the
coarse-member-first plan test, the Pandas cogroup one with the swapped key
order, and the e2e. All three fail on the pre-PR code; this fourth one passes
there by design, which the PR description says too.
It became the guard's test rather than the fix's when the design moved: an
earlier shape narrowed the collection so the `require` was satisfiable, and
this test asserted that narrowing. The fix stopped touching catalyst, so what
was left worth pinning is the contract — the members disagree on purpose, every
one of them stays available for `isCompatibleWith`, and asking the collection
for a single partitioning throws.
##########
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)))
Review Comment:
Agreed, and thank you for checking the `requiredChildDistribution`
implementations independently — I measured the same thing and got the same
answer: joins and cogroups are the only operators with more than one clustered
child, and both have exactly two.
One thing worth adding, since it changes how alarming the untested branch
is. That fallback used to *guess* a member: it took the finest of all of them
when no single one served every matched child. A blind review round pointed out
that a child then gets declared co-partitioned on a layout it does not have,
with no shuffle inserted — silent wrong results. So it was changed to bail out
instead, which is what you see. The untested arm is now the conservative one,
and if an operator with three clustered children ever lands, the failure mode
is a lost SPJ rather than a wrong answer. A test then, agreed.
--
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]
