ulysses-you commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4071636729
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1457,6 +1457,30 @@ object PartitioningCollection {
case other => other.satisfies(required)
}
+ /**
+ * The specs `p` offers for `distribution`, one per member that may serve
it, built the way the
+ * planner builds them (`createShuffleSpec`): a keyed member that covers
part of the operation's
+ * keys contributes the projection onto the keys it covers, which is the
layout the alignment
+ * about to be planned will emit for it. A member that is not keyed has no
projection to make and
+ * is asked for its own. A keyed member is admitted on `keysMaySatisfy`, any
other member on
+ * `satisfies`, and a count the operation pinned is asked as it stands, the
way
+ * `maySatisfyAfterProjection` asks it. That is the planner's admission of a
member
+ * (`EnsureRequirements.createKeyedShuffleSpecs`) less the coverage of every
operation key it
+ * also requires there: that requirement is the skew heuristic of
+ * `spark.sql.requireAllClusterKeysForCoPartition`, while a member covering
a subset of the
+ * operation keys is a sound pairing.
+ */
+ private[sql] def specsForPairing(
+ p: Partitioning,
+ distribution: ClusteredDistribution): Seq[ShuffleSpec] =
+ flatten(p).flatMap {
+ case k: KeyedPartitioning =>
+ Option.when(distribution.requiredNumPartitions.forall(_ ==
k.numPartitions) &&
+ k.keysMaySatisfy(distribution))(k.createShuffleSpec(distribution))
Review Comment:
Thanks @peter-toth! Narrowed in ce59a03fc24. `specsForPairing` now offers
the layout each member reports (`KeyedShuffleSpec(k, distribution)`, nothing
projected and nothing grouped), admitted on `satisfies`, so the member the plan
holds is the member judged. An ungrouped one is the single exception, and it is
confined to the configuration that builds one through
`PartitioningCollection.mayUngroupedMember`; `validateInternal`'s per-side
exemption reads the same predicate, so the two cannot drift. `keysSatisfy` is
asked from outside the class now, so the family doc lists it beside
`keysMaySatisfy`, and that is the question the ungrouped exception asks.
The refusal in the default configuration is pinned in
`ValidateRequirementsSuite` (the first test, after the conf block) and in
`ShuffleSpecSuite` (the member is offered only where something builds one), and
the shapes you would expect to still be refused are: a differing key order and
a differing partition count, now with both permissions on, so the answer is the
layouts' own rather than the admission's.
Classification: regression of this change. The base asked `satisfies` per
side, which an ungrouped keyed child fails in every configuration, so nothing
here is a pre-existing hole being made visible.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1457,6 +1457,30 @@ object PartitioningCollection {
case other => other.satisfies(required)
}
+ /**
+ * The specs `p` offers for `distribution`, one per member that may serve
it, built the way the
+ * planner builds them (`createShuffleSpec`): a keyed member that covers
part of the operation's
+ * keys contributes the projection onto the keys it covers, which is the
layout the alignment
+ * about to be planned will emit for it. A member that is not keyed has no
projection to make and
+ * is asked for its own. A keyed member is admitted on `keysMaySatisfy`, any
other member on
+ * `satisfies`, and a count the operation pinned is asked as it stands, the
way
+ * `maySatisfyAfterProjection` asks it. That is the planner's admission of a
member
+ * (`EnsureRequirements.createKeyedShuffleSpecs`) less the coverage of every
operation key it
+ * also requires there: that requirement is the skew heuristic of
+ * `spark.sql.requireAllClusterKeysForCoPartition`, while a member covering
a subset of the
+ * operation keys is a sound pairing.
+ */
+ private[sql] def specsForPairing(
Review Comment:
Thanks @peter-toth! All three corrected in ce59a03fc24, each now stating the
invariant rather than a caller: `PartitioningCollection.createShuffleSpec`
names the planner's `shuffleToCoPartition` as the reader of that admission set;
`maySatisfyAfterProjection` says the validator does not read it and points at
`specsForPairing`; `SinglePartitionShuffleSpec` says no production caller puts
a collection on the `other` side, which I checked by enumerating the
`isCompatibleWith` call sites: `pickCoPartitionTarget` flattens to leaves
before it pairs, and `specsForPairing` offers leaves.
One more site said the same thing and is corrected with them, outside the
list you gave: the SPARK-59289 test comment in `ShuffleSpecSuite`, which read
"the admission set here is what a finished plan is checked against by
`ValidateRequirements`".
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +45,60 @@ 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 with two of them is judged by
their pairing below,
+ // and every other child, an operator with a single clustered child
included, answers for
+ // itself.
+ val coPartitioning = children.length > 1 &&
+ requiredChildDistributions.forall(_.isInstanceOf[ClusteredDistribution])
+
val satisfied =
children.zip(requiredChildDistributions.zip(requiredChildOrderings)).forall {
case (child, (distribution, ordering))
- if !child.outputPartitioning.satisfies(distribution)
+ if (!child.outputPartitioning.satisfies(distribution) &&
+ !(coPartitioning &&
+
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 co-partitioning 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
Review Comment:
Thanks @peter-toth! Direction agreed, and it wants its own ticket rather
than this PR: `KeyLayout` is master and branch-4.x only, while the repair here
should reach further back.
With the narrowing, no configuration stands in for the role any more: what
is read says which shape a plan may report, not which part a side plays in the
alignment.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1457,6 +1457,30 @@ object PartitioningCollection {
case other => other.satisfies(required)
}
+ /**
+ * The specs `p` offers for `distribution`, one per member that may serve
it, built the way the
+ * planner builds them (`createShuffleSpec`): a keyed member that covers
part of the operation's
+ * keys contributes the projection onto the keys it covers, which is the
layout the alignment
+ * about to be planned will emit for it. A member that is not keyed has no
projection to make and
+ * is asked for its own. A keyed member is admitted on `keysMaySatisfy`, any
other member on
+ * `satisfies`, and a count the operation pinned is asked as it stands, the
way
+ * `maySatisfyAfterProjection` asks it. That is the planner's admission of a
member
Review Comment:
Thanks @peter-toth! Both corrected in ce59a03fc24. The helper's doc states
the two things the planner does and it deliberately does not, since they build
a layout the plan does not hold: no projection onto the operation keys under
`allowKeysSubsetOfPartitionKeys` (a plan the permission applies to holds the
projection already, in the grouping node `EnsureRequirements` inserted), and no
grouping of an ungrouped member or re-sorting of a grouped one, `toGrouped`
being what `tryCreate` builds from.
`satisfiesForPairing` now says this is the question the `compatibleAsIs`
path asks, and that the push path commits on the two sides it builds
(`committed`, a `describesSameKeys` comparison of the layouts the reduce left),
which is sound to ask here because a pair that took the reduce answers the
strict question through `hasSameReducedKeys`.
--
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]
