peter-toth commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4062966334
##########
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:
**Finding 2.** Three comments justify a choice by naming
`ValidateRequirements` as the caller that constrains them, and after this PR
none of them is reached from it:
- `partitioning.scala:1389-1395`, in
`PartitioningCollection.createShuffleSpec`: "The set matters because
`ValidateRequirements` builds a spec from a finished plan through here." It
does not any more, since `specsForPairing` flattens the collection and builds
the member specs itself.
- `partitioning.scala:1442-1450`, in `maySatisfyAfterProjection`: "the
caller feeds `ValidateRequirements` as well as the planner, so it does not
widen what a finished plan is checked against." The finished-plan check now
uses a wider admission, so that sentence no longer describes the code (finding
1).
- `partitioning.scala:1690-1696`, in
`SinglePartitionShuffleSpec.isCompatibleWith`: "The one production caller that
can put a collection on the `other` side is `ValidateRequirements`'
`specs.tail.forall(_.isCompatibleWith(specs.head))`, and there the stricter
answer is the safer one." That line is the one this PR deletes, and
`specsForPairing` returns leaf specs only, so the `case
ShuffleSpecCollection(specs) => specs.forall(isCompatibleWith)` arm is now
reachable only from `ShuffleSpecSuite`. The `forall`-versus-`exists` divergence
it documents has no production caller left to be safe for.
The description says the first two are left as they are. Each states an
invariant rather than a pointer, so all three are worth correcting here.
##########
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:
**Finding 4.** Not a request to change this PR: the narrowing in finding 1
uses a configuration as a proxy for the property this comment names, and the
property itself could be carried instead.
What the pairing cannot tell apart is which role an ungrouped side plays,
and the node that made it ungrouped does know: `GroupPartitionsExec` with
`distributePartitions = true` spreads a key's splits, while the other side
repeats its whole group. If `KeyLayout` carried that role, the way it already
carries `mayContainUnknownPartitionKeys` for another per-row property, then
`isCompatibleWith` could require at most one spread side with a repeating
partner, and the validator would accept the sound pair and refuse the unsound
one on its own. No configuration would be read, and it would cover
SPARK-59436's skew-split side for free rather than needing another disjunct.
The counter-arguments are real, which is why this is not a request.
`KeyLayout` is master and `branch-4.x` only, so a fix that should reach
`branch-4.2` cannot be built on it. The marker has to be produced by
`GroupPartitionsExec` and kept across `copy`, and every consumer that compares
layouts has to agree what it means. And it is a larger change than the
validator question that prompted it. Worth its own ticket if you think the
direction is right; happy for it to sit on either of our lists.
##########
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:
**Finding 1.** `keysMaySatisfy` is `if (isGrouped) keysCanSatisfy else
mayGroupToSatisfy`, so for an ungrouped member it answers "yes, once something
groups it", and this then builds the spec from the ungrouped layout. The
planner may ask that question, because it is about to insert the node:
`createKeyedShuffleSpecs.tryCreate` builds the spec from
`partitioning.toGrouped` (`EnsureRequirements.scala:1135`). A finished plan has
nobody left to insert one.
That is the clause `maySatisfyAfterProjection` keeps, and it was a decision
rather than an omission. SPARK-59289 made `satisfies` strict about a projecting
node and had to widen this filter back for it; it widened the projection and
not the grouping, and wrote down why (`partitioning.scala:1442-1450`): "A
partitioning that is not grouped is not admitted, even though a node would also
group it. ... the caller feeds `ValidateRequirements` as well as the planner,
so it does not widen what a finished plan is checked against."
Your first unit test states the cost: two sides reporting `[1, 1, 2]`,
neither grouped, no partially-clustered conf set, and `validate` says yes. That
layout pair has the two readings your own comment in `validate` names. One side
spread and the other replicating the whole group, which is sound and is what
the rule builds. Or both sides splitting the key between two partitions, which
is not: rows of key `1` in left partition 0 never meet rows of key `1` in right
partition 1. The second is not reachable from `EnsureRequirements` today, which
is why this is Non-blocking, and it is also why the guarantee now rests
entirely on the producer with nothing enforcing it.
The projection half has the same shape: under
`allowKeysSubsetOfPartitionKeys` an ungrouped member contributes
`project(...).toGrouped`, so two sides whose key multiplicities differ can
project onto the same grouped key list and pair, while the plan holds neither
the projection nor the grouping.
Narrowing the widening to the shape that needs it keeps the rest as strict
as it was:
```scala
case k: KeyedPartitioning =>
// An ungrouped side is a plan only partially clustered distribution
builds.
val mayBeUngrouped =
SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled
Option.when((k.isGrouped || mayBeUngrouped) &&
distribution.requiredNumPartitions.forall(_ == k.numPartitions) &&
k.keysMaySatisfy(distribution))(k.createShuffleSpec(distribution))
```
The per-side exemption in `validateInternal` wants the same condition, and a
test for the refusal in the default configuration would pin it, which is what
your first test asserts the other way round today. One more reason to spell the
condition out rather than admit everything: SPARK-59436 (#58771) will want the
same relaxation for a skew-split side, and its own config then belongs in that
disjunction.
##########
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:
**Finding 5.** The doc, and the description, say these specs are "built the
way the planner builds them" and that the admission is the planner's "less the
coverage of every operation key it also requires there". There is a second
difference, and it is the one this change turns on:
`createKeyedShuffleSpecs.tryCreate` builds the spec from the *grouped* form of
an ungrouped member, `val grouped = if (partitioning.isGrouped) partitioning
else partitioning.toGrouped` at `EnsureRequirements.scala:1135`, because it is
about to insert the node that groups it. This helper deliberately does not,
which is the right call for a finished plan and worth saying outright instead
of implying an equivalence that does not hold.
Same for "the question `EnsureRequirements` commits a pair on (`committed`,
over the pair `agreeingPairs` picked)": on the push path `committed` is
`sidesDeclareSameKeys`, a `describesSameKeys` comparison of the two children's
declared layouts, and only the `compatibleAsIs` path asks `isCompatibleWith`.
--
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]
