LuciferYang commented on code in PR #58527:
URL: https://github.com/apache/spark/pull/58527#discussion_r3945716755
##########
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:
Coverage note, no change requested: this fallback (no member serving every
matched child, so all children take the ordinary shuffle) has no test and leans
on the comment that no operator has three clustered children today - which
holds at this commit, having checked the `requiredChildDistribution`
implementations under sql/core. A future operator with three clustered children
would exercise this branch silently, so if one ever lands it deserves a test
then.
##########
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:
Just a note on this test so the next reader doesn't mistake it for the
regression test: it pins the current contract (members with divergent
`numPartitions` keep `createPartitioning` throwing) and passes both before and
after this PR, since catalyst is untouched here. Regression protection rests on
the other three tests - the two `EnsureRequirementsSuite` ones
(coarse-member-first, swapped-key-order cogroup) and the e2e - all of which
fail on the pre-PR code. No change needed.
--
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]
