ulysses-you commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4088936708
##########
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:
Thanks @cloud-fan! Fixed in 32ddb2663e4. The waiver asks the second gate the
producer applies before it spreads a side now (`canDuplicateLeftSide ||
canDuplicateRightSide`), folded in at the waiver rather than into
`partiallyClusteredJoinType`: that helper is the producer's entry to key-group
checking altogether, so a kind turned away there would lose the
storage-partitioned join it can still plan without spreading a side. The
producer-boundary test pins the join type as well as the operator kind, the
same pair refused for a full outer join and accepted for the inner pair the
waiver is for.
One correction on attribution: the over-admission is not new to the latest
revision. The waiver was `ShuffledJoin`-wide before it, so a `FullOuter` pair
was admitted then too, and the operator-kind revision neither introduced nor
closed it. It came in with the waiver itself.
##########
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:
Thanks @cloud-fan! Reworded in 32ddb2663e4: the doc says the planner asks
this of a pair it takes as it stands, though not by the same predicate, since
its `compatibleAsIs` path reads two unprojected specs while a member here may
be relabelled onto the key the operation clusters on.
--
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]
