LuciferYang commented on code in PR #58527:
URL: https://github.com/apache/spark/pull/58527#discussion_r3939094520
##########
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:
`createKeyedShuffleSpec` still resolves a `PartitioningCollection` with
`collectFirst`, taking the first satisfying member, while this PR makes the
per-child branch resolve the finest member both sides agreed on. The default
`requireAllClusterKeysForCoPartition=true` keeps partial-coverage members out
of the candidates, so this only shows up with
`requireAllClusterKeysForCoPartition=false` plus
`allowKeysSubsetOfPartitionKeys=true`: when both sides report alias
cross-products with a coarse member first, SPJ pairs the coarse members and
joins on them, leaving the finer pairing unused; with coarse first on one side
and fine first on the other, the picks fail to pair and the per-child branch
takes over, grouping each side on its finest matching member, so what is
skipped is the SPJ partition-value pushdown, not a shuffle. Results stay
correct. If you align it, the choice cannot be an independent per-side finest
(a side with only coarse members would then fail to pair); it has to pair
like `bestMemberOpt` does, by finest member compatible with some member of
the other side. Either fixing it here or a follow-up works for me.
##########
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:
`bestSpecOpt` is still picked with `ShuffleSpecCollection.numPartitions`,
which comes from `specs.head`. A collection like [coarse head with 2
partitions, fine member with 200] competes as 2 in the `maxBy`, loses to a
sibling child's plain 100-partition spec, and the side holding the fine member
gets re-shuffled onto 100 partitions when it could have stayed unshuffled as
the best. The selection line predates this PR, and in this shape it behaves
exactly as before (the losing side took the same full re-shuffle), so this is
purely pre-existing. Not blocking; worth a follow-up comparing best candidates
at member level at the call site (changing `numPartitions` itself would affect
other readers).
--
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]
