cloud-fan commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4085063362
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +47,81 @@ object ValidateRequirements extends Logging {
assert(requiredChildDistributions.length == children.length)
assert(requiredChildOrderings.length == children.length)
+ // A `ClusteredDistribution` is the one distribution an operator can owe
its children together
+ // rather than one by one, so an operator whose children all owe one is
judged on their mutual
+ // layout below, and every other child, an operator with a single
clustered child included,
+ // answers for itself. That is every such operator, not only a join: one
that zips corresponding
+ // partitions, a cogroup for instance, reads a layout both children have
to hold together.
+ val clusteredMultiChild = children.length > 1 &&
+ requiredChildDistributions.forall(_.isInstanceOf[ClusteredDistribution])
+
+ // The one member a finished plan may report without satisfying the
distribution is the shape
+ // partially clustered distribution spreads ungrouped, and one producer
builds it:
+ // `EnsureRequirements.checkKeyGroupCompatible`. Both halves of that
admission are asked here,
+ // and the answer is passed down to the pairing below so the waiver cannot
be read one way here
+ // and the other way there. Neither half is a second copy: the operator
kinds come from the
+ // producer itself. What is left to the member, its count and the
permission for the collapse it
+ // went through, is asked there.
+ val mayBeUngrouped = clusteredMultiChild &&
+ SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled &&
+ ShuffledJoin.partiallyClusteredJoinType(plan).isDefined
Review Comment:
**Non-blocking (P2):** `partiallyClusteredJoinType(plan).isDefined` proves
only that this is an SMJ/SHJ, not that this join type can produce the
spread/repeat layout the waiver requires. For `FullOuter`, both duplication
capabilities are false, so `EnsureRequirements` skips partial clustering, yet
matching repeated children can pass here and execute split groups
partition-by-partition, missing cross-split matches or emitting matched rows as
unmatched. Please make the shared predicate exclude join types with no
duplicable side and add a focused FullOuter refusal test.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +47,81 @@ object ValidateRequirements extends Logging {
assert(requiredChildDistributions.length == children.length)
assert(requiredChildOrderings.length == children.length)
+ // A `ClusteredDistribution` is the one distribution an operator can owe
its children together
+ // rather than one by one, so an operator whose children all owe one is
judged on their mutual
+ // layout below, and every other child, an operator with a single
clustered child included,
+ // answers for itself. That is every such operator, not only a join: one
that zips corresponding
+ // partitions, a cogroup for instance, reads a layout both children have
to hold together.
+ val clusteredMultiChild = children.length > 1 &&
+ requiredChildDistributions.forall(_.isInstanceOf[ClusteredDistribution])
+
+ // The one member a finished plan may report without satisfying the
distribution is the shape
+ // partially clustered distribution spreads ungrouped, and one producer
builds it:
+ // `EnsureRequirements.checkKeyGroupCompatible`. Both halves of that
admission are asked here,
+ // and the answer is passed down to the pairing below so the waiver cannot
be read one way here
+ // and the other way there. Neither half is a second copy: the operator
kinds come from the
+ // producer itself. What is left to the member, its count and the
permission for the collapse it
+ // went through, is asked there.
+ val mayBeUngrouped = clusteredMultiChild &&
+ SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled &&
+ ShuffledJoin.partiallyClusteredJoinType(plan).isDefined
+
val satisfied =
children.zip(requiredChildDistributions.zip(requiredChildOrderings)).forall {
case (child, (distribution, ordering))
- if !child.outputPartitioning.satisfies(distribution)
+ if (!child.outputPartitioning.satisfies(distribution) &&
+ !(mayBeUngrouped &&
+
PartitioningCollection.representativeOf(child.outputPartitioning).isDefined))
|| !SortOrder.orderingSatisfies(child.outputOrdering, ordering) =>
logDebug(s"ValidateRequirements failed: $distribution,
$ordering\n$plan")
false
case _ => true
}
- if (satisfied && children.length > 1 &&
-
requiredChildDistributions.forall(_.isInstanceOf[ClusteredDistribution])) {
- // Check the co-partitioning requirement.
- val specs =
children.map(_.outputPartitioning).zip(requiredChildDistributions).map {
- case (p, d) =>
p.createShuffleSpec(d.asInstanceOf[ClusteredDistribution])
- }
- if (specs.tail.forall(_.isCompatibleWith(specs.head))) {
- true
- } else {
+ // What a multi-child clustered operator reads is the pairing: a pair
aligned without grouping,
+ // which partially clustered distribution builds on purpose, is one the
sides agree on while
+ // neither is grouped. The pairing cannot tell how the two sides hold a
key's rows, since a
+ // spread side and one that repeats the whole group report the same keys
as two sides that
+ // split the key, so that rests on the producer, which is why the
ungrouped shape alone is
+ // waived above.
+ if (!satisfied) {
+ false
+ } else if (clusteredMultiChild) {
+ val paired = satisfiesForPairing(children, requiredChildDistributions,
mayBeUngrouped)
+ if (!paired) {
logDebug(s"ValidateRequirements failed: children not co-partitioned
in\n$plan")
- false
}
+ paired
} else {
- satisfied
+ true
+ }
+ }
+
+ /**
+ * Whether the sides of a multi-child clustered operator line up: every side
offers the layouts it
+ * reports ([[PartitioningCollection.specsForPairing]]), and one member of
the first side pairs
+ * with every other side. A plan holds what its members report, so no key is
deduped and none is
+ * re-sorted to make a pair: a side is judged on the partitions it has,
under the key the
+ * operation clusters on.
+ *
+ * This is the question `EnsureRequirements` asks of a pair it takes as it
stands, the
Review Comment:
**Nit (P3):** This is not quite the same question as `compatibleAsIs`. Here,
`reportedSpecOf` may relabel a grouped `[a, b]` member for an `[a]`
distribution when dropping `b` merges no partition; the planner represents that
candidate as projected, and `compatibleAsIs` requires both specs to be
unprojected. Please describe the relationship without claiming predicate
equivalence across this projection case.
--
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]
