You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "ulysses-you (via GitHub)" <gi...@apache.org> on 2026/09/21 10:32:53 UTC
[PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
ulysses-you opened a new pull request, #58942:
URL: https://github.com/apache/spark/pull/58942
### What changes were proposed in this pull request?
`ValidateRequirements` asks every child of a clustered operator to satisfy its distribution on its own. A `ClusteredDistribution` is the one distribution an operator can owe two children together rather than one by one, so this PR judges such an operator's children by their pairing, and a pair aligned without grouping (which partially clustered distribution builds on purpose) is one the sides agree on while neither is grouped.
- `ValidateRequirements` asks the per-side check only of children that answer for themselves or report no keyed layout; the rest are judged together, on the specs the planner builds for them (`PartitioningCollection.specsForPairing`, a new `private[sql]` helper on the existing object: a keyed member on the planner's own admission `keysMaySatisfy`, any other member on `satisfies`, and a pinned count asked as it stands). One member of the first side has to pair with every other side, which is the question `EnsureRequirements` commits a pair on (`committed`, over the pair `agreeingPairs` picked).
- What those specs hold is what `createShuffleSpec` makes of a member: its own layout, or under `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys` the projection onto the keys it covers, the layout the alignment about to be planned emits. That is the question the planner asks when it picks the member a pair is planned on, less the coverage of every operation key it additionally requires there: `spark.sql.requireAllClusterKeysForCoPartition` is a skew heuristic, and a member covering a subset of the operation keys is a sound pairing.
- What the pairing cannot say is how the two sides hold a key's rows: a spread side and one that repeats the whole group report the same keys as two sides that split the key between them, and no layout distinguishes those, so that rests on the producer, which is the join path: `checkKeyGroupCompatible` is where such a pair is planned, and every other co-partitioning operator's children are grouped before the rule is done.
The catalyst change is additive: a new helper beside the existing ones, and no change to `PartitioningCollection.createShuffleSpec` or `maySatisfyAfterProjection`. Those two still name `ValidateRequirements` as a caller in their comments, which this change leaves as they are.
### Why are the changes needed?
A storage-partitioned join planned by partially clustered distribution aligns its sides without grouping either of them: the side that keeps its splits spreads them, the other replicates its group across them, so both report keys that repeat on purpose. Such a pair fails the per-side check at the join node even though the two sides agree key by key, and `AdaptiveSparkPlanExec.optimizeQueryStage` validates a stage's whole candidate plan before accepting an `AQEShuffleReadRule` change, so every shuffle read in that stage stays uncoalesced, unrelated ones included.
Measured on `d39cc1784c0` (base) versus this head, both with `spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled=true`:
| Probe | base | head |
|---|---|---|
| A pair the rule planned (`EnsureRequirements.apply`) | `ValidateRequirements.validate` false | true |
| The aggregate's shuffle read in the join's stage | not coalesced (`AQEShuffleReadExec.hasCoalescedPartition` false) | coalesced |
| A three-table chain whose outer join reads a projection over both key columns | `validate` false | true, shuffle-free, still coalesced |
The planner-side half of this hazard landed in SPARK-59272, which declines a pairing whose sides no longer declare the same aligned key sequence: that closes the pairs a `GroupPartitionsExec` gives up on, which should not be built at all. A pair aligned without grouping is the other half, and it is built on purpose, so the validator has to read the pairing instead of each child on its own.
### Does this PR introduce _any_ user-facing change?
No. No plan changes unless the plan already contains such an alignment, and no new configuration.
### How was this patch tested?
New tests, by what each one is there for (the four marked base-failing were re-run against `d39cc1784c0` and fail there):
- `ValidateRequirementsSuite`
- the exemption and its refusals (base-failing): a keyed pair that repeats its keys position by position passes, while key sets that disagree, a differing key order, a hashed side, a lone clustered child and two sides that each satisfy on their own but do not line up all fail;
- the rule's own pair (base-failing): a pair planned by `EnsureRequirements` under partially clustered distribution passes;
- the collection shape (base-failing): a side reporting several keyed alternatives is judged on whichever of them pairs, not on the first one;
- guards, which pass on base as well and pin this change's clauses: a subset-keyed pair still passes, a lone clustered child still owes its own grouping, a pinned partition count is still asked of a pair, a pair that lines up still owes its operator an ordering, and a collapsed pair is not admitted on its pairing alone.
- `KeyGroupedPartitioningSuite`
- with partially clustered distribution on, the aggregate's shuffle read in the stage holding a two-table join coalesces (base-failing: it is not coalesced); the test also pins that the join side shuffles nothing, that the chain shuffles once, and that the join shares the final stage with that read, which is what makes the coalesce a decision the join can block;
- a three-table chain whose outer join reads a projection that keeps both key columns stays shuffle-free, passes validation and keeps its coalescing (base-failing at the validation assertion).
- `ShuffleSpecSuite`: the specs a side offers are the planner's own: a keyed member offers the layout it reports, a member covering part of the operation's keys offers the projection onto the keys it covers under the subset permission, a member whose keys do not cover the clustering offers nothing, and a member that is not keyed offers its own spec.
Ran locally: `ValidateRequirementsSuite`, `KeyGroupedPartitioningSuite`, `EnsureRequirementsSuite`, `GroupPartitionsExecSuite`, `AdaptiveQueryExecSuite`, `PushDownLocalSortSuite`, `ShuffleSpecSuite`: all green, with scalastyle clean for catalyst and sql, main and test sources.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (qwen3.8-flash)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4071638339
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/ValidateRequirementsSuite.scala:
##########
@@ -158,4 +163,213 @@ class ValidateRequirementsSuite extends SharedSparkSession {
testNestedJoin(Seq((2, 2), (1, 1)), Seq((2, 2)), Seq(5, 5, 5), false)
testNestedJoin(Seq((2, 2), (1, 1)), Seq((2, 5)), Seq(5, 5, 5), false)
}
+
+ test("SPARK-59671: a co-partitioning operator judges keyed children by their pairing") {
+ // The sides of a storage-partitioned join aligned for skew repeat their spread keys on
+ // purpose: neither satisfies a clustering on its own, yet the two key sequences agree
+ // index by index, and the pairing is what the operator reads.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val rows = Seq(InternalRow(1), InternalRow(1), InternalRow(2))
+ val left = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(a), rows))
+ val right = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(b), rows))
+ val join = ShuffledHashJoinExec(Seq(a), Seq(b), Inner, BuildLeft, None, left, right)
+ assert(ValidateRequirements.validate(join),
+ s"aligned but ungrouped keyed sides pair, and that pairing is the requirement:\n$join")
+
+ // The same key rows in another order do not pair: position by position is the contract.
+ val off = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(1), InternalRow(2), InternalRow(1))))
+ assert(!ValidateRequirements.validate(join.copy(right = off)),
+ "the same keys in a different order are not aligned")
+
+ // Nor does a keyed side with one that never pairs: a hashed side pairs with neither the keys
+ // nor the layout of a keyed one.
+ val hashed = DummySparkPlan(outputPartitioning = HashPartitioning(Seq(b), 3))
+ assert(!ValidateRequirements.validate(join.copy(right = hashed)),
+ "a keyed side does not pair with a hashed one")
+
+ // And a pair each side satisfies on its own is still refused when the sides do not line up,
+ // which is the operator's requirement: the pairing, not the per-side answer.
+ val leftKeys = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(a), Seq(InternalRow(1), InternalRow(2), InternalRow(3))))
+ val rightKeys = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(4), InternalRow(5), InternalRow(6))))
+ assert(leftKeys.outputPartitioning.satisfies(ClusteredDistribution(Seq(a))) &&
+ rightKeys.outputPartitioning.satisfies(ClusteredDistribution(Seq(b))),
+ "test setup: each side answers its own clustering")
+ assert(!ValidateRequirements.validate(ShuffledHashJoinExec(
+ Seq(a), Seq(b), Inner, BuildLeft, None, leftKeys, rightKeys)),
+ "the sides do not line up, and nothing else the operator reads says otherwise")
+ }
+
+ test("SPARK-59671: a subset-keyed pair is judged on its own layouts") {
+ // With the subset permission on, the planner builds pairs whose sides are grouped on a subset
Review Comment:
Thanks @cloud-fan! Corrected in ce59a03fc24, in the fixture and in the helper text: the permission applies where the operation's keys are a subset of the source's partitioning keys, so a `[a, b]` partitioning may be projected for an `[a]` operation, and not the other way round.
The fixture now uses that direction: the sides report the projection's own layout, which passes, and the same pair one step earlier, reporting the source's keys that no node projected, is refused. The `ShuffleSpecSuite` arm had the same wording and is corrected with it.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #58942:
URL: https://github.com/apache/spark/pull/58942#issuecomment-5888520428
Thank you @peter-toth for the round — finding 7 is a real wrong-results regression this PR introduced, and the ablation pinned it to the line. Pushed `856fa0d43c7` addressing all three findings. The design moves in one paragraph: the alignment stamp is no longer read by `satisfies` at all.
**Why.** The pushed head had `satisfies(ClusteredDistribution)` answer `true` for a stamped ungrouped layout. That leaked through the four arms where `ShuffledJoin.outputPartitioning` reports one side verbatim: a single-child consumer, say an aggregate over a left outer join, read the spread side as clustered and split a key's group across its partitions ([1,1],[1,1],[2,1] instead of [1,2],[2,1]).
**What changed.**
- `satisfies0`'s clustering arm is back to the base `isGrouped && keysSatisfy`: every consumer but the pairing owes an ungrouped layout the settling node. The stamp is read only where the pair is judged — a new `KeyedPartitioning.pairsUngrouped` (role, count, key space) feeds `specsForPairing`, and `ValidateRequirements` delegates the per-child distribution clause of a multi-child clustered operator to that pairing, which asks each side the same question plus the stamp and sinks on an empty side, so nothing is lost by the delegation. The ordering clause stays per child.
- An ordering pairs nothing, so an ungrouped layout still answers it without settling — but the keys now have to ascend in the sequence the ordering reads (`keysSortedFor`), on a construction shared with `resolveChild`'s keep-versus-spread decision so the two cannot disagree. This also tightens the grouped half: a distinct-but-descending layout no longer claims an ordering.
- `stringArgs` appends `ungroupingOrigin` where a producer made a claim; an empty `Option` appends nothing, so unclaimed layouts keep their previous plan string.
**Tests.** The aggregate-over-spread repro above is an end-to-end test, pinning both the verbatim spread side under the join and the grouping node above it. The serves matrix pins both halves of the ordering gate (unsorted ungrouped, unsorted grouped) and the clustering refusals; each new gate was mutation-checked against the fixture that pins it. Full runs green: catalyst 34/34, sql 376/376, `dev/lint-scala` clean.
Per-thread replies carry the details for each finding.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4081502311
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1437,6 +1445,54 @@ object PartitioningCollection {
case other => other.satisfies(required)
}
+ /**
+ * Whether a finished plan may report an ungrouped keyed member, which only partially clustered
+ * distribution builds. Read by `specsForPairing` and by `ValidateRequirements`' per-side
+ * admission, so the two cannot drift.
+ */
+ private[sql] def mayUngroupedMember: Boolean =
+ SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled
+
+ /**
+ * The specs `p` offers for `distribution`, one per member that may serve it, each of them the
+ * layout that member reports. A keyed member is admitted on `satisfies`, the as-it-stands
+ * question, count included; a member that is not keyed is asked on `satisfies` as well and has no
+ * projection to make.
+ *
+ * The one member whose layout a finished plan may report without satisfying the distribution is
+ * an ungrouped keyed one: that is the shape partially clustered distribution spreads, and
+ * `mayUngroupedMember` confines it to the configuration that builds one. `keysSatisfy` is the
+ * as-it-stands question for it, since `satisfies` adds `isGrouped` on top.
+ *
+ * This is the planner's admission of a member (`EnsureRequirements.createKeyedShuffleSpecs`) less
+ * the coverage of every operation key it requires there
+ * (`spark.sql.requireAllClusterKeysForCoPartition`), which is a skew heuristic: a member whose
Review Comment:
Thanks @cloud-fan! Corrected in f95e44a9f42, in the helper's text and in the PR description: disabling the coverage check admits a member whose partitioning keys cover only a subset of the operation's keys, and the partition keys being a superset is the separate projection case under `allowKeysSubsetOfPartitionKeys`.
##########
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:
One correction to my reply above, from this round's producer-boundary review: the spec is not the member's layout verbatim. It is that layout viewed on the key the operation clusters on, so a partition expression carrying no cluster key is left out, with the member's own count and its own key order, no key deduped and none re-sorted. `PartitioningCollection.reportedSpecOf` states that and why, and the point of the reply stands: `createShuffleSpec`'s grouped layout is not what the validator reads. Head is f95e44a9f42.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #58942:
URL: https://github.com/apache/spark/pull/58942#issuecomment-5759846748
cc @dongjoon-hyun @peter-toth @cloud-fan thank you
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4141258562
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -432,6 +432,20 @@ case class CoalescedNullAwareHashPartitioning(
copy(from = from.copy(expressions = newChildren))
}
+// Describes why a keyed layout reports repeated partition keys on purpose, rather than leaving them
+// for the grouping node that would settle them.
+sealed trait UngroupingOrigin
+
+// One side of a partially clustered alignment: it keeps its splits and spreads them across the
+// slots its key is expected in.
+case object SPREADS_SPLITS extends UngroupingOrigin
Review Comment:
Thanks @szehon-ho — the roles came out of the 09-26 design round: record on the layout which side a producer built, rather than inferring it from the conf and the operator kind, which is what `UngroupingOrigin` is. Renamed to your names in 9a7b697a2b9, and the producer branch now reads `if (replicateLeftSide) REPLICATED_FOR_JOIN else SPLIT_FOR_JOIN`. Renaming only: no behavior change, and no test asserts on the value, although the plan string prints it. `UngroupingOrigin` keeps its name: it says why the keys repeat, not which side of the alignment this is.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #58942:
URL: https://github.com/apache/spark/pull/58942#issuecomment-5777524577
The as-held-layout fixes are in place, but I found two remaining boundary issues: the ungrouped waiver currently includes SortMergeAsOfJoinExec even though EnsureRequirements never builds its spread/replicate pairing, and non-join cogroup operators lost the old mutual compatibility check for already-grouped children. Please separate the producer-specific waiver from the generic multi-child compatibility validation and add focused AS-OF plus grouped-cogroup coverage.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"conversation:5776592901","thread_id":"conversation:5776592901","verdict_sha256":"0d655853a520afc3453e956a11e8a41dba7a05ec2c7ad3cb830af90b05ced131"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4131434427
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -854,8 +881,25 @@ case class KeyedPartitioning(
reducers: Seq[Option[KeyReducer]]): (Seq[DataType], Seq[InternalRowComparableWrapper]) =
KeyedPartitioning.reduceKeys(partitionKeys, keyDataTypes, reducers)
+ /**
+ * Whether the reason this layout is ungrouped is one `required` reads. A clustering pairs
+ * partitions index by index, so only the two alignment roles answer yes, and `None` does not:
+ * a layout that is not grouped and does not say why is waiting for the node that would settle
+ * it, and a finished plan has none left to insert. An ordering reads the keys without pairing
+ * anything, so settling is not what it asks: `None` and its own spread answer yes, while the
+ * two pairing roles answer no, built as they are for a clustering and not for an ordering.
+ */
+ private def ungroupingServes(required: Distribution): Boolean = required match {
+ case _: ClusteredDistribution =>
+ ungroupingOrigin.exists(r => r == SPREADS_SPLITS || r == REPEATS_GROUP)
Review Comment:
**Finding 7.** A `ClusteredDistribution` is not "pairs partitions index by index". It is "rows sharing a cluster key share a partition", and single-child operators ask it too, where nothing pairs. A `SPREADS_SPLITS` side breaks it by construction: key 1's rows sit in two partitions. `REPEATS_GROUP` breaks it the other way, by holding the same group in several.
`ShuffledJoin.outputPartitioning` returns one side verbatim on four arms (`joins/ShuffledJoin.scala:81-84`), and on each of them the reported side is the *spread* one, because the producer's duplication gate forces it: `canDuplicateLeftSide(LeftOuter)` is false, so the right repeats and the left spreads, and `LeftOuter` reports the left. Only the `InnerLike` arm is safe, and not by design — `fromPartitionings` runs `concatUngrouping` over two disagreeing roles and clears the stamp.
Measured on this head, with `partiallyClusteredDistribution.enabled=true`, `pushPartValues=true`, broadcast off:
```
probeL: id = 1, 1, 2 (the bigger side, so the heuristic replicates the right)
probeR: id = 1, 2
SELECT id, count(*) FROM (
SELECT /*+ MERGE(l, r) */ l.id FROM probeL l LEFT OUTER JOIN probeR r ON l.id = r.id
) GROUP BY id
join isGrouped=false origin=Some(SPREADS_SPLITS)
groupPartitions=List(Some(SPREADS_SPLITS), Some(REPEATS_GROUP))
shuffles=0
ROWS = [1,1], [1,1], [2,1] <- expected [1,2], [2,1]
```
Key 1 comes out as two groups. Ablating this arm alone, `&& !required.isInstanceOf[ClusteredDistribution]`, makes `EnsureRequirements` add the grouping node back (`groupPartitions` gains a `None` entry) and the rows come out `[1,2], [2,1]`. So the cause is this line and nothing else.
The PR's own assertions sit on both sides of it. `KeyGroupedPartitioningSuite` asserts "a zipped pair reports no claim of its own" over the joins in the chain, which is exactly the invariant the four one-sided arms break, and that chain only uses inner joins. `ShuffleSpecSuite` then pins the acceptance with `assert(stamped(Some(SPREADS_SPLITS)).satisfies(clustered))`.
The fix I would take is to keep the stamp out of `satisfies` and read it only where the pairing is decided:
```scala
private def ungroupingServes(required: Distribution): Boolean = required match {
// A clustering asks that rows sharing a cluster key share a partition, which an alignment
// role does not give on its own. `KeyedShuffleSpec.isCompatibleWith` is where the pair is
// judged. An ordering reads the keys without pairing anything.
case _: OrderedDistribution =>
ungroupingOrigin.isEmpty || ungroupingOrigin.contains(SPREADS_FOR_ORDERING)
case _ => false
}
```
`ValidateRequirements` then needs its per-child distribution clause skipped for a multi-child clustered operator, and that is sound for the reason the earlier round established: `specsForPairing` admits a member only on its own answer, and an empty side fails the pairing's `exists`, so the clause never changes the result there. `specsForPairing` would admit a stamped member explicitly instead of through `satisfies`.
The narrower alternative is to clear the stamp on the four one-sided arms, the way `InnerLike` ends up doing. That plugs the leak I can reach, but it leaves "a stamped layout is only ever read by the join it was built for" as an invariant nothing enforces, which is the same class as the local-read leak this redesign just closed.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -854,8 +881,25 @@ case class KeyedPartitioning(
reducers: Seq[Option[KeyReducer]]): (Seq[DataType], Seq[InternalRowComparableWrapper]) =
KeyedPartitioning.reduceKeys(partitionKeys, keyDataTypes, reducers)
+ /**
+ * Whether the reason this layout is ungrouped is one `required` reads. A clustering pairs
+ * partitions index by index, so only the two alignment roles answer yes, and `None` does not:
+ * a layout that is not grouped and does not say why is waiting for the node that would settle
+ * it, and a finished plan has none left to insert. An ordering reads the keys without pairing
+ * anything, so settling is not what it asks: `None` and its own spread answer yes, while the
+ * two pairing roles answer no, built as they are for a clustering and not for an ordering.
+ */
+ private def ungroupingServes(required: Distribution): Boolean = required match {
+ case _: ClusteredDistribution =>
+ ungroupingOrigin.exists(r => r == SPREADS_SPLITS || r == REPEATS_GROUP)
+ case _: OrderedDistribution =>
+ ungroupingOrigin.isEmpty || ungroupingOrigin.contains(SPREADS_FOR_ORDERING)
Review Comment:
**Finding 8.** `ungroupingOrigin.isEmpty` admits every unstamped ungrouped layout, and those keys are not sorted by construction. A grouped layout's are: `toGrouped` sorts with `keyOrdering`, and a scan sorts with `groupedKeyRowOrdering`. That is what made it safe for `keysSatisfy`'s ordered arm to leave the order to the caller. An unstamped ungrouped layout has no producer that sorts it.
Measured on this head:
```scala
val one = KeyedPartitioning(Seq(a), Seq(InternalRow(1), InternalRow(2)))
val unioned = KeyedPartitioning.concat(Seq(one, one))
// isGrouped = false, ungroupingOrigin = None, keys = [1], [2], [1], [2]
// satisfies(OrderedDistribution(a asc)) == true, with v2 bucketing sorting on
```
Nothing acts on that wrongly today, which is why I am not calling it Blocking. `resolveChild` re-checks the key order itself before returning the child (`EnsureRequirements.scala:149`), and `RemoveRedundantSorts` needs the child to report an ordering as well, which `UnionExec` does not. What is left is a layout answering `true` to a claim it cannot back, and `ValidateRequirements` reading it.
Two ways to close it. Check the key order in the ordered arm of `keysSatisfy`, which would also let `resolveChild` stop re-deriving it. Or stamp the already-sorted case, so the admission is a positive answer rather than the absence of one — `resolveChild` has decided the keys are ascending at that point, it just has no node to stamp.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -486,13 +500,20 @@ case class CoalescedNullAwareHashPartitioning(
* one, clears the markers first. That is precision, not the
* guarantee: without it the merge would spread the spurious
* marker onto the accurate side.
+ * @param ungroupingOrigin Why the keys repeat on purpose, when a producer built them that way
+ * rather than leaving them for the grouping node that would settle them.
+ * `None` means unsettled: a clustering refuses such a layout, and only a
+ * node about to be inserted can make it serve one, while an ordering reads
+ * the keys without asking them to settle. See [[UngroupingOrigin]].
*/
+
case class KeyLayout(
@transient partitionKeys: Seq[InternalRowComparableWrapper],
dataTypes: Seq[DataType],
isGrouped: Boolean,
isCollapsed: Boolean,
- mayContainUnknownPartitionKeys: Boolean = false) {
+ mayContainUnknownPartitionKeys: Boolean = false,
+ ungroupingOrigin: Option[UngroupingOrigin] = None) {
Review Comment:
**Finding 9.** `KeyLayout.stringArgs` (`partitioning.scala:723`) is a curated list, and its doc sets the rule for a new field: "A field the layout grows is therefore a decision here, and the default is to print it: the plan string is the only place a reader sees why a marked partitioning still shuffles." The same argument applies to this one. The plan string is where a reader would see why a keyed partitioning reports repeated keys rather than one per key, and `ungroupingOrigin` is not in the list.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #58942:
URL: https://github.com/apache/spark/pull/58942#issuecomment-5763610046
Two things to add on top of @peter-toth's review rather than repeat it.
**Finding 1 reproduces, with numbers.** On `bb183663ba8`, the projection half of
his finding is not hypothetical. Taking your own `ValidateRequirementsSuite`
case "the same keys in a different order are not aligned" and flipping only the
config:
```
[1,1,2] vs [1,2,1], subsetConf=false -> validate=false
[1,1,2] vs [1,2,1], subsetConf=true -> validate=true
[1,1,2] (3 partitions) vs [1,2] (2 partitions), subsetConf=true -> validate=true
```
At the catalyst level `specsForPairing` reports the divergence directly:
```
left child numPartitions = 3 left spec numPartitions = 2
right child numPartitions = 3 right spec numPartitions = 2
compatible = true
```
So the config default is the only reason that test still passes, and the second
line means the validator stops noticing that two children have different
partition counts at all.
**`coPartitioning` is not only the join path.** It is true for any operator with
more than one child all requiring `ClusteredDistribution`, which includes
`CoGroupExec` (`objects.scala:637`) and `FlatMapCoGroupsInBatchExec`
(`FlatMapCoGroupsInBatchExec.scala:56-59`). For a cogroup, "the two sides agree
key by key while neither is grouped" is not a sufficient input: it pairs left
group with right group per partition, so a key split across two partitions on
each side yields two partial cogroups rather than merely a missing shuffle.
The comment in `validateInternal` argues every non-join co-partitioning
operator's children are grouped before the rule is done, and that holds today -
`checkKeyGroupCompatible` returns `None` for a non-join parent
(`EnsureRequirements.scala:598-600`), so `resolveEachChild` groups them. But
`ValidateRequirements` is also the gate for third-party `AQEShuffleReadRule`s
registered through `adaptiveRulesHolder`
(`AdaptiveSparkPlanExec.optimizeQueryStage`), which is exactly where an
invariant `EnsureRequirements` maintains stops being self-evident. Whatever
shape the narrowing in finding 1 takes, scoping the per-side exemption to the
operators the pairing argument actually covers would keep the guard for the
rest.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4132743695
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -740,16 +756,19 @@ private[sql] object GroupPartitionsExec {
case ((_, Seq(single)), outputIndex) => single == outputIndex
case _ => false
}
- // A `copy` of the child's layout rather than a fresh one, so the marker and anything the layout
- // grows later are carried without this having to name them. Built here, where the child is in
- // scope, so `computeOutputPartitioning` has nothing left to re-apply.
+ // A `copy` of the child's layout rather than a fresh one, so the marker and anything the
+ // layout grows later are carried without this having to name them. The origin is the one
+ // field not carried: the output claims what this node's own producer stamped, never what
+ // the child's claimed. Built here, where the child is in scope, so
+ // `computeOutputPartitioning` has nothing left to re-apply.
PartitionGrouping(
partitions,
childKp.layout.copy(
partitionKeys = partitions.map(_._1),
dataTypes = reducedDataTypes,
isGrouped = isGrouped,
- isCollapsed = isCollapsed),
+ isCollapsed = isCollapsed,
+ ungroupingOrigin = ungroupingOrigin),
Review Comment:
**Finding 10.** This writes the producer's stamp into the layout right next to the `isGrouped` it just computed, with no relation between the two. `alignToExpectedKeys` sets `isGrouped = false` only when some key's `numSplits > 1`, so an alignment where every key gets exactly one slot comes out **grouped and stamped**. `toGrouped` clears the origin for exactly this reason ("the keys are unique afterwards, so a reason to be ungrouped has been spent"); this producer does not.
That was harmless at the previous head, where the ordered arm read `(isGrouped || ungroupingServes(required)) && keysSatisfy(required)` and `isGrouped` short-circuited. This commit dropped the `isGrouped ||`, so the stamp now decides the answer.
Measured on this head. `EnsureRequirements` over two sides holding one split per key, partial clustering and `pushPartValues` on:
```
node origin=Some(SPREADS_SPLITS) reported isGrouped=true origin=Some(SPREADS_SPLITS)
node origin=Some(REPEATS_GROUP) reported isGrouped=true origin=Some(REPEATS_GROUP)
```
and with `v2BucketingAllowSorting` on, over the same keys `[1], [2]`:
```
grouped, unstamped satisfies(OrderedDistribution(a asc)) = true
grouped, SPREADS_SPLITS satisfies(OrderedDistribution(a asc)) = false
```
So a partially clustered join whose keys each hold one split loses its global-ordering claim and pays an extra sort, or the spread `resolveChild` builds in its place. No wrong results, and only with the sorting config on, which is why this is not Blocking.
The narrow fix is to put `isGrouped ||` back in the ordered arm, but the stamp is the thing that is wrong: every doc in this change reads it as "why the keys repeat on purpose", which a grouped layout's keys do not. Clearing it at the producer keeps that invariant true everywhere it is read:
```suggestion
ungroupingOrigin = if (isGrouped) None else ungroupingOrigin),
```
I applied that and ran it. The two nodes above then report `origin=None`; `ValidateRequirementsSuite` + `GroupPartitionsExecSuite` + `EnsureRequirementsSuite` (133 tests) and all five `SPARK-59671` cases in `KeyGroupedPartitioningSuite` stay green. `isCompatibleWith` is unaffected either way: a grouped pair moves from `(Some(SPREADS_SPLITS), Some(REPEATS_GROUP))` to `(None, None)`, and both answer `true`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you closed pull request #58942: [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing
URL: https://github.com/apache/spark/pull/58942
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #58942:
URL: https://github.com/apache/spark/pull/58942#issuecomment-5763541408
Thanks for working on this — the AQE coalescing problem is real and the pairing
approach makes sense. One issue before this lands.
`specsForPairing` builds the spec with `k.createShuffleSpec(distribution)`.
Under `spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys=true`
that method returns `project(joinKeyPositions).toGrouped`, i.e. a **deduped and
re-sorted** layout the child does not actually hold. Since the per-side
`satisfies` check is waived for exactly these children, nothing else catches the
difference.
Reproduced on this branch (`bb183663ba8`). In `ValidateRequirementsSuite`, this
is your own "the same keys in a different order are not aligned" case with the
config flipped:
```scala
withSQLConf(SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true") {
val left = DummySparkPlan(outputPartitioning =
KeyedPartitioning(Seq(a), Seq(InternalRow(1), InternalRow(1), InternalRow(2))))
val right = DummySparkPlan(outputPartitioning =
KeyedPartitioning(Seq(b), Seq(InternalRow(1), InternalRow(2), InternalRow(1))))
ValidateRequirements.validate(
ShuffledHashJoinExec(Seq(a), Seq(b), Inner, BuildLeft, None, left, right))
}
```
```
misaligned pair: subsetConf=false -> validate=false
misaligned pair: subsetConf=true -> validate=true
3-partition child vs 2-partition sibling (subset conf on) -> validate=true
```
At the catalyst level the divergence is visible directly:
```
left child numPartitions = 3 left spec numPartitions = 2
right child numPartitions = 3 right spec numPartitions = 2
compatible = true
```
So two consequences:
1. Two ungrouped sides whose key sequences do **not** line up position by
position (`[1,1,2]` vs `[1,2,1]`) both collapse to `[1,2]` and pair. Your
test rejects this shape today only because it runs at the config's `false`
default.
2. An ungrouped 3-partition side pairs with a grouped 2-partition sibling, so
the validator no longer notices that the two children have different
partition counts.
Note the difference from `EnsureRequirements.createKeyedShuffleSpecs`: the
planner applies `toGrouped` because it is about to insert the
`GroupPartitionsExec` that makes it true. A validator reading a finished plan
has no such node to point at, so it should judge the pair on the layouts the
children actually report. Building the pairing specs without `project`/
`toGrouped` would fix both cases and, as far as I can tell, still admits the
partially-clustered pair this PR is after — both sides report their raw keys and
`describesSameKeys` compares them index by index.
Two smaller points:
- `coPartitioning` is true for any operator with >1 children all requiring
`ClusteredDistribution`, which includes `CoGroupExec` (`objects.scala:637`)
and `FlatMapCoGroupsInBatchExec`, not just joins. For a cogroup, "aligned but
neither side grouped" is not a sufficient input — it would emit partial
cogroups rather than merely miss a shuffle. The comment argues
`EnsureRequirements` groups them (it does — `checkKeyGroupCompatible` returns
`None` for non-join parents), but `ValidateRequirements` is also the gate for
third-party `AQEShuffleReadRule`s registered via `adaptiveRulesHolder`.
Restricting the exemption to the join path named in the comment would keep
the fix and the guard.
- `PartitioningCollection.createShuffleSpec` and `maySatisfyAfterProjection`
still justify their admission set with "ValidateRequirements builds a spec
from a finished plan through here" / "the caller feeds `ValidateRequirements`
as well as the planner". Neither is on that path anymore, so those should be
updated to name the planner.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4068736081
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/ValidateRequirementsSuite.scala:
##########
@@ -158,4 +163,213 @@ class ValidateRequirementsSuite extends SharedSparkSession {
testNestedJoin(Seq((2, 2), (1, 1)), Seq((2, 2)), Seq(5, 5, 5), false)
testNestedJoin(Seq((2, 2), (1, 1)), Seq((2, 5)), Seq(5, 5, 5), false)
}
+
+ test("SPARK-59671: a co-partitioning operator judges keyed children by their pairing") {
+ // The sides of a storage-partitioned join aligned for skew repeat their spread keys on
+ // purpose: neither satisfies a clustering on its own, yet the two key sequences agree
+ // index by index, and the pairing is what the operator reads.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val rows = Seq(InternalRow(1), InternalRow(1), InternalRow(2))
+ val left = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(a), rows))
+ val right = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(b), rows))
+ val join = ShuffledHashJoinExec(Seq(a), Seq(b), Inner, BuildLeft, None, left, right)
+ assert(ValidateRequirements.validate(join),
+ s"aligned but ungrouped keyed sides pair, and that pairing is the requirement:\n$join")
+
+ // The same key rows in another order do not pair: position by position is the contract.
+ val off = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(1), InternalRow(2), InternalRow(1))))
+ assert(!ValidateRequirements.validate(join.copy(right = off)),
+ "the same keys in a different order are not aligned")
+
+ // Nor does a keyed side with one that never pairs: a hashed side pairs with neither the keys
+ // nor the layout of a keyed one.
+ val hashed = DummySparkPlan(outputPartitioning = HashPartitioning(Seq(b), 3))
+ assert(!ValidateRequirements.validate(join.copy(right = hashed)),
+ "a keyed side does not pair with a hashed one")
+
+ // And a pair each side satisfies on its own is still refused when the sides do not line up,
+ // which is the operator's requirement: the pairing, not the per-side answer.
+ val leftKeys = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(a), Seq(InternalRow(1), InternalRow(2), InternalRow(3))))
+ val rightKeys = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(4), InternalRow(5), InternalRow(6))))
+ assert(leftKeys.outputPartitioning.satisfies(ClusteredDistribution(Seq(a))) &&
+ rightKeys.outputPartitioning.satisfies(ClusteredDistribution(Seq(b))),
+ "test setup: each side answers its own clustering")
+ assert(!ValidateRequirements.validate(ShuffledHashJoinExec(
+ Seq(a), Seq(b), Inner, BuildLeft, None, leftKeys, rightKeys)),
+ "the sides do not line up, and nothing else the operator reads says otherwise")
+ }
+
+ test("SPARK-59671: a subset-keyed pair is judged on its own layouts") {
+ // With the subset permission on, the planner builds pairs whose sides are grouped on a subset
Review Comment:
**Nit (P3):** This describes the subset in the opposite direction from the configuration contract. `allowKeysSubsetOfPartitionKeys` permits the operation keys to be a subset of the source partition keys, so an `[a, b]` partitioning may be projected for an `[a]` operation. It does not make the planner build an `[a]` partitioning for an `[a, b]` operation; with `requireAllClusterKeysForCoPartition` at its default, that shape is rejected. Please update this fixture comment and the matching helper text to use the operation-subset-of-source direction.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4072240006
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +46,73 @@ 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 a join 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.
+ // Only a shuffling join gets that reading: it matches within a partition and unions the
+ // matches, so an alignment that spreads one side against a repeating other is sound, while an
+ // operator that emits a per-key result from one partition, a cogroup for instance, would emit a
+ // partial one for every spread key. `EnsureRequirements.checkKeyGroupCompatible` is the join
+ // path that plans such a pair, and the only producer of one.
+ val coPartitioning = children.length > 1 && plan.isInstanceOf[ShuffledJoin] &&
Review Comment:
**Non-blocking (P2):** This waiver is broader than its producer invariant: `EnsureRequirements.checkKeyGroupCompatible` builds spread/replicate layouts only for `SortMergeJoinExec` and `ShuffledHashJoinExec`, while `SortMergeAsOfJoinExec` also matches `ShuffledJoin`. With partial clustering enabled, an AQE rule candidate over aligned repeated AS-OF children can therefore pass without either side being known to repeat the whole key group, and the partition-local AS-OF scan can miss or duplicate matches. Please restrict the waiver to the producer-supported join kinds or carry equivalent explicit provenance.
See **Shared repair plan 1** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +46,73 @@ 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 a join 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.
+ // Only a shuffling join gets that reading: it matches within a partition and unions the
+ // matches, so an alignment that spreads one side against a repeating other is sound, while an
+ // operator that emits a per-key result from one partition, a cogroup for instance, would emit a
+ // partial one for every spread key. `EnsureRequirements.checkKeyGroupCompatible` is the join
+ // path that plans such a pair, and the only producer of one.
+ val coPartitioning = children.length > 1 && plan.isInstanceOf[ShuffledJoin] &&
+ 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).exists(k =>
+ k.isGrouped || PartitioningCollection.mayUngroupedMember)))
|| !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
+ // and one that repeats the whole group report the same keys as two sides that split the key,
+ // so that rests on the producer: only a join is admitted above, and
+ // `checkKeyGroupCompatible` is where it plans such a pair.
+ if (!satisfied) {
+ false
+ } else if (coPartitioning) {
+ val paired = satisfiesForPairing(children, requiredChildDistributions)
+ if (!paired) {
logDebug(s"ValidateRequirements failed: children not co-partitioned in\n$plan")
- false
}
+ paired
} else {
- satisfied
+ true
Review Comment:
**Blocking (P1):** For a non-`ShuffledJoin`, this branch now returns `true` after per-child satisfaction without the merge-target validator's mutual shuffle-spec compatibility check. `CoGroupExec` and `FlatMapCoGroupsInBatchExec` zip corresponding partitions, so individually satisfying children with different counts can fail at execution, while equal-count but misaligned layouts can emit partial or missing groups. Please restore generic compatibility validation for multi-child clustered consumers and keep only the ungrouped waiver join-specific.
**Recommended change:** Restore a generic mutual compatibility check for every multi-child operator whose requirements are clustered, while retaining the producer-specific ungrouped exception only on the eligible shuffled-join path; add grouped cogroup coverage for incompatible counts and layouts.
**Why this works:** After individual distribution and ordering validation, compare compatible shuffle specs for all multi-child clustered consumers. Use the specialized as-held pairing only where an eligible join needs the ungrouped waiver; non-join consumers must still prove a mutually compatible grouped layout.
**Scope:** Reestablish cross-child compatibility as a generic validator responsibility without broadening the ungrouped join waiver.
**Compatibility:** The join-specific partially clustered waiver remains separate from ordinary grouped compatibility validation.
**Risks:** A generic check must not apply the join-only ungrouped semantics to operators that emit one result per key group.
**Constraints:** Single-child operators continue to owe only their own distribution and ordering. Valid grouped multi-child plans remain accepted.
**Success:** Two individually grouped cogroup children with different partition counts are rejected before execution. Equal-count but mutually misaligned grouped cogroup layouts are rejected. Compatible grouped non-join children and valid shuffled joins continue to validate.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1437,6 +1445,54 @@ object PartitioningCollection {
case other => other.satisfies(required)
}
+ /**
+ * Whether a finished plan may report an ungrouped keyed member, which only partially clustered
+ * distribution builds. Read by `specsForPairing` and by `ValidateRequirements`' per-side
+ * admission, so the two cannot drift.
+ */
+ private[sql] def mayUngroupedMember: Boolean =
+ SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled
+
+ /**
+ * The specs `p` offers for `distribution`, one per member that may serve it, each of them the
+ * layout that member reports. A keyed member is admitted on `satisfies`, the as-it-stands
+ * question, count included; a member that is not keyed is asked on `satisfies` as well and has no
+ * projection to make.
+ *
+ * The one member whose layout a finished plan may report without satisfying the distribution is
+ * an ungrouped keyed one: that is the shape partially clustered distribution spreads, and
+ * `mayUngroupedMember` confines it to the configuration that builds one. `keysSatisfy` is the
+ * as-it-stands question for it, since `satisfies` adds `isGrouped` on top.
+ *
+ * This is the planner's admission of a member (`EnsureRequirements.createKeyedShuffleSpecs`) less
+ * the coverage of every operation key it requires there
+ * (`spark.sql.requireAllClusterKeysForCoPartition`), which is a skew heuristic: a member whose
+ * partitioning keys are a superset of the operation's keys is a sound pairing.
+ *
+ * Two things the planner does are deliberately not done here, because they build a layout the
+ * plan does not hold, which is the one thing the validator must not read:
+ *
+ * - no projection onto the operation keys (`KeyedPartitioning.createShuffleSpec` makes one under
+ * `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys`). A plan in which the
+ * permission applies holds the projection already, in the grouping node
+ * `EnsureRequirements` inserted, so the member reports it and no projection is wanted.
+ * - no grouping of an ungrouped member (`tryCreate` builds from `toGrouped`), and no re-sorting
+ * of a grouped one: `toGrouped` dedups and sorts, both of which move the member's key rows.
+ */
+ private[sql] def specsForPairing(
+ p: Partitioning,
+ distribution: ClusteredDistribution): Seq[ShuffleSpec] =
+ flatten(p).flatMap {
+ case k: KeyedPartitioning =>
+ val pairsAsIs = k.satisfies(distribution) ||
+ (mayUngroupedMember &&
Review Comment:
**Non-blocking (P2):** With partial clustering enabled, this fallback asks `keysSatisfy`, which does not apply the `isCollapsed` gate. `EnsureRequirements` asks `keysMaySatisfy`; for an ungrouped collapsed layout, `mayGroupToSatisfy` rejects it while `allowKeysSubsetOfPartitionKeys` is off. Two matching collapsed sides can therefore validate even though the planner refuses that shape. The added negative leaves partial clustering disabled and exits before this branch, so please preserve the planner's collapsed-layout permission here and cover the two flags together.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1437,6 +1445,54 @@ object PartitioningCollection {
case other => other.satisfies(required)
}
+ /**
+ * Whether a finished plan may report an ungrouped keyed member, which only partially clustered
+ * distribution builds. Read by `specsForPairing` and by `ValidateRequirements`' per-side
+ * admission, so the two cannot drift.
+ */
+ private[sql] def mayUngroupedMember: Boolean =
+ SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled
+
+ /**
+ * The specs `p` offers for `distribution`, one per member that may serve it, each of them the
+ * layout that member reports. A keyed member is admitted on `satisfies`, the as-it-stands
+ * question, count included; a member that is not keyed is asked on `satisfies` as well and has no
+ * projection to make.
+ *
+ * The one member whose layout a finished plan may report without satisfying the distribution is
+ * an ungrouped keyed one: that is the shape partially clustered distribution spreads, and
+ * `mayUngroupedMember` confines it to the configuration that builds one. `keysSatisfy` is the
+ * as-it-stands question for it, since `satisfies` adds `isGrouped` on top.
+ *
+ * This is the planner's admission of a member (`EnsureRequirements.createKeyedShuffleSpecs`) less
+ * the coverage of every operation key it requires there
+ * (`spark.sql.requireAllClusterKeysForCoPartition`), which is a skew heuristic: a member whose
Review Comment:
**Nit (P3):** Disabling `requireAllClusterKeysForCoPartition` skips `allClusterKeysCovered`, so it can admit partitioning on `[a]` for an operation on `[a, b]`: the partitioning keys cover only a subset of the operation keys. A partition-key superset is the separate projection case controlled by `allowKeysSubsetOfPartitionKeys`. Please reverse this example/rationale so it describes the configuration used here.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -9228,6 +9229,100 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59671: a partially clustered join leaves AQE's shuffle coalescing alone") {
+ // AQE validates a stage's whole candidate plan before accepting a shuffle-read change, so a
Review Comment:
**Nit (P3):** This states the regression as present behavior, but the assertion below requires the unrelated `AQEShuffleReadExec` to have a coalesced partition. Please make this historical (for example, the join *used to* keep unrelated shuffles uncoalesced) so the explanation agrees with the behavior this test pins.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4140064718
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -432,6 +432,20 @@ case class CoalescedNullAwareHashPartitioning(
copy(from = from.copy(expressions = newChildren))
}
+// Describes why a keyed layout reports repeated partition keys on purpose, rather than leaving them
+// for the grouping node that would settle them.
+sealed trait UngroupingOrigin
+
+// One side of a partially clustered alignment: it keeps its splits and spreads them across the
+// slots its key is expected in.
+case object SPREADS_SPLITS extends UngroupingOrigin
Review Comment:
sorry for late review.
This reads a bit awkward, was there an earlier discussion?
Can we do something like:
*SPLIT_FOR_JOIN
* REPLICATED_FOR_JOIN
* SPLIT_FOR_ORDERING?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -432,6 +432,20 @@ case class CoalescedNullAwareHashPartitioning(
copy(from = from.copy(expressions = newChildren))
}
+// Describes why a keyed layout reports repeated partition keys on purpose, rather than leaving them
+// for the grouping node that would settle them.
+sealed trait UngroupingOrigin
+
+// One side of a partially clustered alignment: it keeps its splits and spreads them across the
+// slots its key is expected in.
+case object SPREADS_SPLITS extends UngroupingOrigin
Review Comment:
sorry for late review.
This reads a bit awkward, was there an earlier discussion?
Can we do something like:
* SPLIT_FOR_JOIN
* REPLICATED_FOR_JOIN
* SPLIT_FOR_ORDERING?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #58942:
URL: https://github.com/apache/spark/pull/58942#issuecomment-5805859336
Thanks @peter-toth and @cloud-fan! Both findings of this pass are in 32ddb2663e4.
- The waiver asks the join type's duplication capability now, the second gate the producer applies before it spreads a side, so a `FullOuter` pair is refused instead of admitted on its kind. The gate is asked at the waiver and not through `partiallyClusteredJoinType`, since that helper is the producer's entry to key-group checking altogether: a kind turned away there would lose the storage-partitioned join it can still plan without spreading a side.
- `satisfiesForPairing`'s doc describes the relationship to the planner's `compatibleAsIs` path without claiming the same predicate.
- The producer-boundary test pins the join type alongside the operator kind: an as-of join and a full outer join refuse the same pair, and the inner pair the waiver is for passes.
On attribution, one correction to the review's "newly introduced": the full-outer over-admission came in with the waiver itself, which was `ShuffledJoin`-wide before the operator-kind revision, so that revision neither introduced nor closed it.
The change is a second commit on top of f95e44a rather than an amend, so its delta reads on its own.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #58942:
URL: https://github.com/apache/spark/pull/58942#issuecomment-5799112116
The earlier AS-OF and non-join issues are addressed, but the shared predicate still returns FullOuter for SMJ/SHJ even though both duplication predicates are false, so validation admits a shape EnsureRequirements cannot produce safely. Please exclude join types with no duplicable side and add a focused full-outer refusal test.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"conversation:5793320348","thread_id":"conversation:5793320348","verdict_sha256":"7002bd93d18b8fa15754a55bc683c8e244b7f4075f2e4c2830048ddb01d81e20"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4105382249
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1437,6 +1445,54 @@ object PartitioningCollection {
case other => other.satisfies(required)
}
+ /**
+ * Whether a finished plan may report an ungrouped keyed member, which only partially clustered
+ * distribution builds. Read by `specsForPairing` and by `ValidateRequirements`' per-side
+ * admission, so the two cannot drift.
+ */
+ private[sql] def mayUngroupedMember: Boolean =
+ SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled
+
+ /**
+ * The specs `p` offers for `distribution`, one per member that may serve it, each of them the
+ * layout that member reports. A keyed member is admitted on `satisfies`, the as-it-stands
+ * question, count included; a member that is not keyed is asked on `satisfies` as well and has no
+ * projection to make.
+ *
+ * The one member whose layout a finished plan may report without satisfying the distribution is
+ * an ungrouped keyed one: that is the shape partially clustered distribution spreads, and
+ * `mayUngroupedMember` confines it to the configuration that builds one. `keysSatisfy` is the
+ * as-it-stands question for it, since `satisfies` adds `isGrouped` on top.
+ *
+ * This is the planner's admission of a member (`EnsureRequirements.createKeyedShuffleSpecs`) less
+ * the coverage of every operation key it requires there
+ * (`spark.sql.requireAllClusterKeysForCoPartition`), which is a skew heuristic: a member whose
Review Comment:
Confirmed: the coverage heuristic and projection permission now describe their two subset directions separately and correctly.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4072240041","thread_id":"inline:4072240041","verdict_sha256":"8e26d305a65705965995542a009188889aa25752be172203798719da4958df03"} -->
##########
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:
Confirmed: the duplication-capability gate excludes FullOuter, and the focused refusal test covers the same repeated-key shape.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4085063362","thread_id":"inline:4085063362","verdict_sha256":"8e26d305a65705965995542a009188889aa25752be172203798719da4958df03"} -->
##########
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:
Confirmed: the revised text distinguishes the related planner question from predicate equivalence in the relabeling case.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4085063382","thread_id":"inline:4085063382","verdict_sha256":"8e26d305a65705965995542a009188889aa25752be172203798719da4958df03"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4081500867
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +46,73 @@ 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 a join 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.
+ // Only a shuffling join gets that reading: it matches within a partition and unions the
+ // matches, so an alignment that spreads one side against a repeating other is sound, while an
+ // operator that emits a per-key result from one partition, a cogroup for instance, would emit a
+ // partial one for every spread key. `EnsureRequirements.checkKeyGroupCompatible` is the join
+ // path that plans such a pair, and the only producer of one.
+ val coPartitioning = children.length > 1 && plan.isInstanceOf[ShuffledJoin] &&
Review Comment:
Thanks @cloud-fan! Restricted in f95e44a9f42, and not by a second copy of the list: the operator kinds now live in one predicate, `ShuffledJoin.partiallyClusteredJoinType`, which `EnsureRequirements.checkKeyGroupCompatible` dispatches on and `ValidateRequirements` reads through `.isDefined`, so the two cannot drift apart. A `SortMergeAsOfJoinExec` is excluded with them, and the pair a shuffled-hash join accepts is pinned as refused for it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +46,73 @@ 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 a join 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.
+ // Only a shuffling join gets that reading: it matches within a partition and unions the
+ // matches, so an alignment that spreads one side against a repeating other is sound, while an
+ // operator that emits a per-key result from one partition, a cogroup for instance, would emit a
+ // partial one for every spread key. `EnsureRequirements.checkKeyGroupCompatible` is the join
+ // path that plans such a pair, and the only producer of one.
+ val coPartitioning = children.length > 1 && plan.isInstanceOf[ShuffledJoin] &&
+ 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).exists(k =>
+ k.isGrouped || PartitioningCollection.mayUngroupedMember)))
|| !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
+ // and one that repeats the whole group report the same keys as two sides that split the key,
+ // so that rests on the producer: only a join is admitted above, and
+ // `checkKeyGroupCompatible` is where it plans such a pair.
+ if (!satisfied) {
+ false
+ } else if (coPartitioning) {
+ val paired = satisfiesForPairing(children, requiredChildDistributions)
+ if (!paired) {
logDebug(s"ValidateRequirements failed: children not co-partitioned in\n$plan")
- false
}
+ paired
} else {
- satisfied
+ true
Review Comment:
Thanks @cloud-fan! Restored in f95e44a9f42. The pairing is asked of every multi-child clustered operator again, which is what the base did, and only the ungrouped waiver stays producer-scoped: the waiver's condition is the multi-child cluster, the partially clustered configuration and the producer's kinds, computed once and passed into the pairing. The cogroup test now covers the mutual check from three sides, two sides holding the same grouped layout passing, a differing partition count refused, and a differing key order refused.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -9228,6 +9229,100 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59671: a partially clustered join leaves AQE's shuffle coalescing alone") {
+ // AQE validates a stage's whole candidate plan before accepting a shuffle-read change, so a
Review Comment:
Thanks @cloud-fan! Reworded in f95e44a9f42: the failure is stated against the base now, "on the base a storage-partitioned join whose sides are aligned but not grouped kept every shuffle in its stage uncoalesced", and the closing sentence says the assertions below pin the read the join's presence must leave alone.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1437,6 +1445,54 @@ object PartitioningCollection {
case other => other.satisfies(required)
}
+ /**
+ * Whether a finished plan may report an ungrouped keyed member, which only partially clustered
+ * distribution builds. Read by `specsForPairing` and by `ValidateRequirements`' per-side
+ * admission, so the two cannot drift.
+ */
+ private[sql] def mayUngroupedMember: Boolean =
+ SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled
+
+ /**
+ * The specs `p` offers for `distribution`, one per member that may serve it, each of them the
+ * layout that member reports. A keyed member is admitted on `satisfies`, the as-it-stands
+ * question, count included; a member that is not keyed is asked on `satisfies` as well and has no
+ * projection to make.
+ *
+ * The one member whose layout a finished plan may report without satisfying the distribution is
+ * an ungrouped keyed one: that is the shape partially clustered distribution spreads, and
+ * `mayUngroupedMember` confines it to the configuration that builds one. `keysSatisfy` is the
+ * as-it-stands question for it, since `satisfies` adds `isGrouped` on top.
+ *
+ * This is the planner's admission of a member (`EnsureRequirements.createKeyedShuffleSpecs`) less
+ * the coverage of every operation key it requires there
+ * (`spark.sql.requireAllClusterKeysForCoPartition`), which is a skew heuristic: a member whose
+ * partitioning keys are a superset of the operation's keys is a sound pairing.
+ *
+ * Two things the planner does are deliberately not done here, because they build a layout the
+ * plan does not hold, which is the one thing the validator must not read:
+ *
+ * - no projection onto the operation keys (`KeyedPartitioning.createShuffleSpec` makes one under
+ * `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys`). A plan in which the
+ * permission applies holds the projection already, in the grouping node
+ * `EnsureRequirements` inserted, so the member reports it and no projection is wanted.
+ * - no grouping of an ungrouped member (`tryCreate` builds from `toGrouped`), and no re-sorting
+ * of a grouped one: `toGrouped` dedups and sorts, both of which move the member's key rows.
+ */
+ private[sql] def specsForPairing(
+ p: Partitioning,
+ distribution: ClusteredDistribution): Seq[ShuffleSpec] =
+ flatten(p).flatMap {
+ case k: KeyedPartitioning =>
+ val pairsAsIs = k.satisfies(distribution) ||
+ (mayUngroupedMember &&
Review Comment:
Thanks @cloud-fan! The permission is back in f95e44a9f42, read once and shared: `collapsedLayoutMayBeGrouped` is what `mayGroupToSatisfy` and the ungrouped admission both ask, so a member a finished plan reports ungrouped is admitted only where the planner would agree to group it. The negative covers the two flags together, partially clustered distribution on with the subset permission off, and a unit test pins that the member is not offered in that configuration.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #58942:
URL: https://github.com/apache/spark/pull/58942#issuecomment-6057212583
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/91f015743a1c48d96ed2e2fb76ec86eefe84ef98
- merged into branch-4.x https://github.com/apache/spark/commit/a615688368b6ad87a31f4c6a6bb25a6c1e4752a1
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4133003856
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -740,16 +756,19 @@ private[sql] object GroupPartitionsExec {
case ((_, Seq(single)), outputIndex) => single == outputIndex
case _ => false
}
- // A `copy` of the child's layout rather than a fresh one, so the marker and anything the layout
- // grows later are carried without this having to name them. Built here, where the child is in
- // scope, so `computeOutputPartitioning` has nothing left to re-apply.
+ // A `copy` of the child's layout rather than a fresh one, so the marker and anything the
+ // layout grows later are carried without this having to name them. The origin is the one
+ // field not carried: the output claims what this node's own producer stamped, never what
+ // the child's claimed. Built here, where the child is in scope, so
+ // `computeOutputPartitioning` has nothing left to re-apply.
PartitionGrouping(
partitions,
childKp.layout.copy(
partitionKeys = partitions.map(_._1),
dataTypes = reducedDataTypes,
isGrouped = isGrouped,
- isCollapsed = isCollapsed),
+ isCollapsed = isCollapsed,
+ ungroupingOrigin = ungroupingOrigin),
Review Comment:
Thanks — fixed in 8fe2b06dffb with your suggestion, at the producer: `computeGrouping` writes the node's origin into the layout only when the computed shape actually stays ungrouped (`if (isGrouped) None else ungroupingOrigin`), so it spends the claim exactly where `toGrouped` does. That restores the invariant globally — no producer or carrier leaves a grouped layout holding a claim — and the ordered arm stays in its strict form: a grouped layout carries `None`, answers through `forall` as before, and the descending-keys pin from the last round is untouched. The node keeps its own field, so the routing derivation and the plan string still show the producer's decision.
Pinned by "an alignment that settles every key spends its stamp" (`GroupPartitionsExecSuite`): the node keeps the stamp, the reported layout is grouped and unclaimed, and `satisfies(OrderedDistribution)` is restored. Mutation-checked: dropping the guard fails exactly that fixture. Suites green: 133 (`GroupPartitionsExecSuite` + `ValidateRequirementsSuite` + `EnsureRequirementsSuite`), the five SPARK-59671 cases in `KeyGroupedPartitioningSuite`, `ShuffleSpecSuite` 34/34, lint clean.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #58942:
URL: https://github.com/apache/spark/pull/58942#issuecomment-5793320348
Thanks @peter-toth, @dongjoon-hyun and @cloud-fan! All of the review's findings are in f95e44a9f42.
- The pairing is owed by every multi-child clustered operator again, so a cogroup keeps the mutual check it had on the base, and only the ungrouped waiver is producer-scoped.
- The operator kinds are one predicate now, `ShuffledJoin.partiallyClusteredJoinType`, read by the producer's dispatch and by the validator, rather than a copy in each file.
- The ungrouped waiver asks the collapse permission the planner applies, so a member whose keys were collapsed is admitted ungrouped only where the planner would agree to group it.
- A side is judged on the key the operation clusters on, as it reports it: the partition expressions carrying no cluster key are left out, and the count and the key order are the member's own. That takes in the subset shape without reading a projection the plan does not hold.
- The producer states the invariant the waiver rests on, where it is decided: one side spreads its splits and the other repeats the whole group, since `replicateRightSide` is the negation of `replicateLeftSide` and the branch is reached only when the picked side may replicate. That is what makes pairing the two index by index sound, and the pairing cannot tell that from two sides that split the key between them.
The tests are the suites the change touches: `ShuffleSpecSuite` and `DistributionSuite` (52), and `ValidateRequirementsSuite`, `EnsureRequirementsSuite`, `GroupPartitionsExecSuite`, `KeyGroupedPartitioningSuite` and `ProjectedOrderingAndPartitioningSuite` (362), with `dev/lint-scala` clean.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #58942:
URL: https://github.com/apache/spark/pull/58942#issuecomment-5776592901
Thanks @peter-toth, @dongjoon-hyun and @cloud-fan! Addressed in ce59a03fc24, which answers both comments above and the inline ones.
The validator judges as-is layouts now. A keyed member contributes the layout it reports -- `KeyedShuffleSpec(k, distribution)`, with no projection onto the operation keys, no grouping and no re-sorting -- and is admitted on `satisfies`. The explicitly safe ungrouped case is the one exception: it is confined to the partially clustered producer configuration through a single predicate, `PartitioningCollection.mayUngroupedMember`, which `specsForPairing` and the per-side exemption both read, so the two cannot drift. The co-partitioning reading is `ShuffledJoin`-only, so `CoGroupExec` and `FlatMapCoGroupsInBatchExec` keep the per-side check.
That closes the measured shapes: a projected key count no longer stands in for the child's own, so three partitions against two is refused; a projection onto distinct sorted keys no longer normalizes a key order; and an ungrouped side is admitted only where something builds one. Coverage for the two configurations asked for is in `ValidateRequirementsSuite`: subset-enabled/partial-disabled, where the projection's own layout passes and the source's keys, which no node projected, are refused, and a non-join operator, a cogroup over the pair a join reads, refused.
Two consequences, stated rather than left implicit:
- A join's distributions never pin `requiredNumPartitions` (`ShuffledJoin.scala:57`), so with the co-partitioning reading limited to the join, `specsForPairing`'s count clause is no longer reachable from the validator. The pinned-count test moved to `ShuffleSpecSuite`, as a direct test of the helper.
- Where the operation's keys are a subset of the source's partitioning keys and the projection would merge no partition, `satisfies` admits the member but the pairing refuses it, the expression the operation does not cluster on mapping to no cluster key. The base accepted that shape through a projection it did not hold; here the plan has to hold the grouping node. This is the one place the validator is stricter than the base.
The comment and doc sites raised inline are answered in their threads.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4109472422
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +47,91 @@ 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, and so does the join type's capability, the second gate it applies before spreading
+ // a side (`canDuplicateLeftSide` / `canDuplicateRightSide`; a `FullOuter` join may duplicate
+ // neither side, so nothing spreads such a pair). That gate is asked here rather than through
+ // `partiallyClusteredJoinType`, which is the producer's entry to key-group checking at all: a
+ // kind turned away there would lose the storage-partitioned join it can still plan without
+ // spreading. What is left to the member, its count and the permission for the collapse it went
+ // through, is asked in the spec.
+ val mayBeUngrouped = clusteredMultiChild &&
+ SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled &&
+ ShuffledJoin.partiallyClusteredJoinType(plan).exists { joinType =>
+ ShuffledJoin.canDuplicateLeftSide(joinType) ||
Review Comment:
**[Correctness] The waiver doesn't check which side replicated, or that any side did.**
This asks only whether *some* side may be duplicated for the join type. For `LeftOuter`/`LeftSemi`/`LeftAnti` only the right side may replicate, but a pair whose *left* side repeats its group passes here as well (`canDuplicateRightSide(LeftOuter)` is true, and the two reported layouts are identical); executing it would emit the unmatched left rows once per replica. Two sides that both split a key, with nothing replicated, also pass. The producer doesn't build the former today (`EnsureRequirements.scala:856-865`), but this validator is the safety net for plans the producer did not build.
Note that the new test `a co-partitioning operator judges keyed children by their pairing` (`ValidateRequirementsSuite.scala:178`) asserts `validate == true` for two producer-less `DummySparkPlan` sides with ungrouped keys `[1, 1, 2]`, which pins this acceptance as the contract.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +47,91 @@ 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, and so does the join type's capability, the second gate it applies before spreading
+ // a side (`canDuplicateLeftSide` / `canDuplicateRightSide`; a `FullOuter` join may duplicate
+ // neither side, so nothing spreads such a pair). That gate is asked here rather than through
+ // `partiallyClusteredJoinType`, which is the producer's entry to key-group checking at all: a
+ // kind turned away there would lose the storage-partitioned join it can still plan without
+ // spreading. What is left to the member, its count and the permission for the collapse it went
+ // through, is asked in the spec.
+ val mayBeUngrouped = clusteredMultiChild &&
+ SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled &&
+ ShuffledJoin.partiallyClusteredJoinType(plan).exists { joinType =>
+ ShuffledJoin.canDuplicateLeftSide(joinType) ||
+ ShuffledJoin.canDuplicateRightSide(joinType)
+ }
+
val satisfied = children.zip(requiredChildDistributions.zip(requiredChildOrderings)).forall {
case (child, (distribution, ordering))
- if !child.outputPartitioning.satisfies(distribution)
+ if (!child.outputPartitioning.satisfies(distribution) &&
Review Comment:
**[Correctness] The waiver can accept a plan that loses join matches.**
`mayBeUngrouped` is decided by the conf, the operator kind and the join type only. Nothing checks that the two sides were actually built by `checkKeyGroupCompatible` as a spread/replicate pair, and an ungrouped keyed pair can also reach an SMJ/SHJ through AQE's local shuffle read:
1. `partiallyClusteredDistribution.enabled=true`, and a query like `(t t1 JOIN s1 ON t1.id = s1.id) JOIN (t t2 JOIN s2 ON t2.id = s2.id) ON t1.id = t2.id`, where `t` is partitioned by `identity(id)` with splits keyed `[1, 1, 2]`, and `s1`/`s2` are large by estimate but small at runtime.
2. With partial clustering on, `KeyedShuffleSpec.canCreatePartitioning` is false, so each lower SMJ hash-shuffles `t`, and `shuffleToCoPartition` peels the `GroupPartitionsExec` (`EnsureRequirements.scala:327`), leaving the raw ungrouped scan under the shuffle. The upper SMJ is hash co-partitioned with no exchange of its own.
3. AQE turns both lower joins into BHJs, and `OptimizeShuffleWithLocalRead` builds one `PartialMapperPartitionSpec` per mapper when `numMappers <= parallelism < 2 * numMappers` (e.g. 150 splits vs 200 shuffle partitions). `AQEShuffleReadExec.outputPartitioning` then reports `s.child.outputPartitioning`, i.e. the scan's ungrouped `KeyedPartitioning([1, 1, 2])`, and BHJ/Sort pass it through.
4. Base: the per-child `satisfies` fails (`isGrouped`), so `optimizeQueryStage` reverts the local reads. Head: the waiver skips the per-child check, `specsForPairing` admits both sides via `mayServeUngrouped`, `describesSameKeys` holds, and `validate` returns `true`.
5. The upper SMJ zips mapper `i` with mapper `i`, so `split0 x split1` and `split1 x split0` for `id = 1` are never joined (e.g. 3 rows instead of 5). For `LeftOuter`/`LeftAnti`, wrong unmatched rows are emitted instead.
The PR's own comment in `EnsureRequirements` says the validator "cannot tell it from two sides that split the key between them". The local read is a second source of exactly that shape, so the soundness cannot rest on the producer alone. Could you add a regression test for this case?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1437,6 +1464,74 @@ object PartitioningCollection {
case other => other.satisfies(required)
}
+ /**
+ * The specs `p` offers for `distribution`, one per member that may serve it, each of them the
+ * layout that member reports (`reportedSpecOf`): the partitions it has, in the order it reports
+ * them, with the partition expressions that carry no cluster key left out. Leaving those out is
+ * what lets two members pair when the operation clusters on part of what they are partitioned
+ * on, and it invents nothing: a kept key is that partition's own, the count is the member's own,
+ * and the order is the member's own.
+ *
+ * A keyed member is admitted on `satisfies`, the as-it-stands question, count included; a member
+ * that is not keyed is asked on `satisfies` as well and has no projection to make. The one member
+ * whose layout a finished plan may report without satisfying the distribution is an ungrouped
+ * keyed one: that is the shape partially clustered distribution spreads. It is admitted only
+ * where the caller says so, through `mayBeUngrouped`, which carries both halves of that
+ * admission: the configuration that builds such a shape, and the operator whose producer spreads
+ * it. `mayServeUngrouped` is the as-it-stands question for it, since `satisfies`
+ * adds `isGrouped` on top.
+ *
+ * This is the planner's admission of a member (`EnsureRequirements.createKeyedShuffleSpecs`) less
+ * the coverage of every operation key it requires there
+ * (`spark.sql.requireAllClusterKeysForCoPartition`), which is a skew heuristic: a member whose
+ * partitioning keys cover only a subset of the operation's keys is a sound pairing.
+ *
+ * @param mayBeUngrouped whether an ungrouped keyed member of `p` may serve `distribution` here.
+ * Only the caller knows both halves of that; see above.
+ */
+ private[sql] def specsForPairing(
+ p: Partitioning,
+ distribution: ClusteredDistribution,
+ mayBeUngrouped: Boolean): Seq[ShuffleSpec] =
+ flatten(p).flatMap {
+ case k: KeyedPartitioning =>
+ val pairsAsIs = k.satisfies(distribution) ||
+ (mayBeUngrouped &&
+ distribution.requiredNumPartitions.forall(_ == k.numPartitions) &&
+ k.mayServeUngrouped(distribution))
Review Comment:
**[Correctness] Admitting ungrouped members widens the reach of `KeyedShuffleSpec.keyPositions`' `assert(refs.size == 1)`.**
`BroadcastHashJoinExec.expandOutputPartitioning` rewrites a `KeyedPartitioning`'s expressions with the build-side keys without any `supportsExpressions` check, e.g. `KP([f.id])` -> `KP([d1.a + d1.b])` for `f.id = d1.a + d1.b`. In the local-read shape described on `ValidateRequirements.scala:79`, with an upper join clustered on `d1.a + d1.b`, `mayServeUngrouped` admits that member (`isClusterKey(e)`), and then `isCompatibleWith` -> `areKeysCompatible` -> `keyPositions` fails the assertion. The `AssertionError` escapes `optimizeQueryStage` (there is no catch there) and fails the query instead of reverting the rule. On the base, the ungrouped member stopped at `satisfies` and `validate` returned `false`. The single-reference issue itself pre-exists for grouped layouts; this PR extends it to ungrouped ones.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +47,91 @@ 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, and so does the join type's capability, the second gate it applies before spreading
+ // a side (`canDuplicateLeftSide` / `canDuplicateRightSide`; a `FullOuter` join may duplicate
+ // neither side, so nothing spreads such a pair). That gate is asked here rather than through
+ // `partiallyClusteredJoinType`, which is the producer's entry to key-group checking at all: a
+ // kind turned away there would lose the storage-partitioned join it can still plan without
+ // spreading. What is left to the member, its count and the permission for the collapse it went
+ // through, is asked in the spec.
+ val mayBeUngrouped = clusteredMultiChild &&
Review Comment:
**[Design] Could the waiver key off what the plan records instead of a guess?**
The pair is built as two `GroupPartitionsExec` nodes with `distributePartitions = applyPartialClustering && !replicateXSide` and the same expected keys, so the plan already knows which side spreads and which replicates (`GroupPartitionsExec.scala:173` already identifies the replicating side as `!grouping.isGrouped && !distributePartitions`). Checking that exactly one child's `GroupPartitionsExec` (through a local `SortExec`) distributes, and that the other side replicates and is one `canDuplicateLeftSide`/`canDuplicateRightSide` allows for the join type, would be about 15 lines and would close the issues on lines 71, 73 and 79 without the conf read or the operator list. Alternatively, the role could be carried on the layout / required distribution, following the `allowNullKeySpreading` precedent, so that `satisfies`/`isCompatibleWith` answer it directly.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +47,91 @@ 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, and so does the join type's capability, the second gate it applies before spreading
+ // a side (`canDuplicateLeftSide` / `canDuplicateRightSide`; a `FullOuter` join may duplicate
+ // neither side, so nothing spreads such a pair). That gate is asked here rather than through
+ // `partiallyClusteredJoinType`, which is the producer's entry to key-group checking at all: a
+ // kind turned away there would lose the storage-partitioned join it can still plan without
+ // spreading. What is left to the member, its count and the permission for the collapse it went
+ // through, is asked in the spec.
+ val mayBeUngrouped = clusteredMultiChild &&
+ SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled &&
+ ShuffledJoin.partiallyClusteredJoinType(plan).exists { joinType =>
+ ShuffledJoin.canDuplicateLeftSide(joinType) ||
+ ShuffledJoin.canDuplicateRightSide(joinType)
+ }
+
val satisfied = children.zip(requiredChildDistributions.zip(requiredChildOrderings)).forall {
case (child, (distribution, ordering))
- if !child.outputPartitioning.satisfies(distribution)
+ if (!child.outputPartitioning.satisfies(distribution) &&
+ !(mayBeUngrouped &&
Review Comment:
**[Design] The same stage-wide symptom remains for another deliberately ungrouped shape.**
With `spark.sql.sources.v2.bucketing.sorting.enabled=true`, `resolveChild` (`EnsureRequirements.scala:149-158`) leaves an ungrouped `KeyedPartitioning` under an `OrderedDistribution` (keys like `[1, 2, 2, 3]` kept as they are, or spread with `distributePartitions = true`), and `KeyedPartitioning.satisfies` requires `isGrouped`, so that plan still fails `validate` (single child, so no waiver). For example, `(SELECT id FROM t ORDER BY id) UNION ALL (SELECT count(*) FROM s GROUP BY k)` still keeps the aggregate's shuffle read uncoalesced. This is pre-existing, but it is the symptom this PR is motivated by, and with the current structure each deliberately ungrouped producer needs its own conf-keyed waiver. A more general fix would be to have the AQE gates validate only what the rule changed, or to let the layout say that the ungrouping is intentional.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1437,6 +1464,74 @@ object PartitioningCollection {
case other => other.satisfies(required)
}
+ /**
+ * The specs `p` offers for `distribution`, one per member that may serve it, each of them the
+ * layout that member reports (`reportedSpecOf`): the partitions it has, in the order it reports
+ * them, with the partition expressions that carry no cluster key left out. Leaving those out is
+ * what lets two members pair when the operation clusters on part of what they are partitioned
+ * on, and it invents nothing: a kept key is that partition's own, the count is the member's own,
+ * and the order is the member's own.
+ *
+ * A keyed member is admitted on `satisfies`, the as-it-stands question, count included; a member
+ * that is not keyed is asked on `satisfies` as well and has no projection to make. The one member
+ * whose layout a finished plan may report without satisfying the distribution is an ungrouped
+ * keyed one: that is the shape partially clustered distribution spreads. It is admitted only
+ * where the caller says so, through `mayBeUngrouped`, which carries both halves of that
+ * admission: the configuration that builds such a shape, and the operator whose producer spreads
+ * it. `mayServeUngrouped` is the as-it-stands question for it, since `satisfies`
+ * adds `isGrouped` on top.
+ *
+ * This is the planner's admission of a member (`EnsureRequirements.createKeyedShuffleSpecs`) less
+ * the coverage of every operation key it requires there
+ * (`spark.sql.requireAllClusterKeysForCoPartition`), which is a skew heuristic: a member whose
+ * partitioning keys cover only a subset of the operation's keys is a sound pairing.
+ *
+ * @param mayBeUngrouped whether an ungrouped keyed member of `p` may serve `distribution` here.
+ * Only the caller knows both halves of that; see above.
+ */
+ private[sql] def specsForPairing(
+ p: Partitioning,
+ distribution: ClusteredDistribution,
+ mayBeUngrouped: Boolean): Seq[ShuffleSpec] =
+ flatten(p).flatMap {
+ case k: KeyedPartitioning =>
+ val pairsAsIs = k.satisfies(distribution) ||
+ (mayBeUngrouped &&
+ distribution.requiredNumPartitions.forall(_ == k.numPartitions) &&
Review Comment:
The count clause lives at the call site, outside `mayServeUngrouped`, which is documented as the as-it-stands question (and at lines 1481-1482 as differing from `satisfies` only by `isGrouped`) but ignores `requiredNumPartitions`. It is also untested: `pairingFour` in `ShuffleSpecSuite` uses `mayBeUngrouped = false`, so removing this line passes every test in the PR. Could it move into `mayServeUngrouped`, with a test that uses `mayBeUngrouped = true` and a pinned count?
Separately, the doc at lines 1484-1487 calls this "the planner's admission of a member (`EnsureRequirements.createKeyedShuffleSpecs`) less the coverage" requirement, but the planner admits on `keysMaySatisfy` (grouping and merging projections allowed), while this admits on the strict `satisfies`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1437,6 +1464,74 @@ object PartitioningCollection {
case other => other.satisfies(required)
}
+ /**
+ * The specs `p` offers for `distribution`, one per member that may serve it, each of them the
+ * layout that member reports (`reportedSpecOf`): the partitions it has, in the order it reports
+ * them, with the partition expressions that carry no cluster key left out. Leaving those out is
+ * what lets two members pair when the operation clusters on part of what they are partitioned
+ * on, and it invents nothing: a kept key is that partition's own, the count is the member's own,
+ * and the order is the member's own.
+ *
+ * A keyed member is admitted on `satisfies`, the as-it-stands question, count included; a member
+ * that is not keyed is asked on `satisfies` as well and has no projection to make. The one member
+ * whose layout a finished plan may report without satisfying the distribution is an ungrouped
+ * keyed one: that is the shape partially clustered distribution spreads. It is admitted only
+ * where the caller says so, through `mayBeUngrouped`, which carries both halves of that
+ * admission: the configuration that builds such a shape, and the operator whose producer spreads
+ * it. `mayServeUngrouped` is the as-it-stands question for it, since `satisfies`
+ * adds `isGrouped` on top.
+ *
+ * This is the planner's admission of a member (`EnsureRequirements.createKeyedShuffleSpecs`) less
+ * the coverage of every operation key it requires there
+ * (`spark.sql.requireAllClusterKeysForCoPartition`), which is a skew heuristic: a member whose
+ * partitioning keys cover only a subset of the operation's keys is a sound pairing.
+ *
+ * @param mayBeUngrouped whether an ungrouped keyed member of `p` may serve `distribution` here.
+ * Only the caller knows both halves of that; see above.
+ */
+ private[sql] def specsForPairing(
+ p: Partitioning,
+ distribution: ClusteredDistribution,
+ mayBeUngrouped: Boolean): Seq[ShuffleSpec] =
+ flatten(p).flatMap {
+ case k: KeyedPartitioning =>
+ val pairsAsIs = k.satisfies(distribution) ||
+ (mayBeUngrouped &&
+ distribution.requiredNumPartitions.forall(_ == k.numPartitions) &&
+ k.mayServeUngrouped(distribution))
+ Option.when(pairsAsIs)(reportedSpecOf(k, distribution))
+ case other =>
+ Option.when(other.satisfies(distribution))(other.createShuffleSpec(distribution))
+ }
+
+ /**
+ * The spec for a member a finished plan reports: its own layout with the partition expressions
+ * that carry no cluster key left out. No key is deduped and no key is re-sorted, so a side offers
+ * a view of the partitions the plan has rather than the layout
+ * `KeyedPartitioning.createShuffleSpec` builds for a node a planner would insert: that one dedups
+ * and sorts, and its count is the grouped one, which the plan does not hold.
+ *
+ * A projection here is a relabelling of the member's partitions, and this leans on the caller for
+ * that: only `keysSatisfy`'s projecting half admits such a member, and it holds the projection to
+ * one that merges no partition, handing the positions it checked to this method.
+ *
+ * A marked layout is not projected: its claim routes the rows of an undeclared key by a hash over
+ * these keys in this order, and dropping one breaks the claim. The clause cannot fire as the
+ * admission stands, and it is kept because the causality is not local: an admitted marked member
+ * has its keys covering the clustering structurally, since the projecting half of `keysSatisfy`
+ * excludes a marked one, so it leaves here unprojected either way.
+ */
+ private def reportedSpecOf(
+ k: KeyedPartitioning,
+ distribution: ClusteredDistribution): KeyedShuffleSpec = {
+ val positions = k.positionsCoveringClusterKeys(distribution).toSeq
+ if (positions.size == k.expressions.length || k.mayContainUnknownPartitionKeys) {
+ KeyedShuffleSpec(k, distribution)
+ } else {
+ KeyedShuffleSpec(k.project(positions), distribution, Some(positions))
Review Comment:
`KeyedShuffleSpec`'s `@param joinKeyPositions` says it is set only when the projection changed the layout, and that `None` means the partitioning is the one the child reports, i.e. a consumer needs no `GroupPartitionsExec`. Here `Some(positions)` is set on a view of partitions the child already holds. It is harmless today because `isCompatibleWith` ignores the field, but the `EnsureRequirements` readers treat `Some` as "insert or compose a grouping node" (`shuffleToCoPartition` -> `withJoinKeyPositions`, `bothUnprojected`, `alreadyCoPartitioned`), so a future reuse of `specsForPairing` there would misread it. Could the `@param` doc be updated for this producer?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/ValidateRequirementsSuite.scala:
##########
@@ -158,4 +163,341 @@ class ValidateRequirementsSuite extends SharedSparkSession {
testNestedJoin(Seq((2, 2), (1, 1)), Seq((2, 2)), Seq(5, 5, 5), false)
testNestedJoin(Seq((2, 2), (1, 1)), Seq((2, 5)), Seq(5, 5, 5), false)
}
+
+ test("SPARK-59671: a co-partitioning operator judges keyed children by their pairing") {
+ // The sides of a storage-partitioned join aligned for skew repeat their spread keys on
+ // purpose: neither satisfies a clustering on its own, yet the two key sequences agree
+ // index by index, and the pairing is what the operator reads.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val rows = Seq(InternalRow(1), InternalRow(1), InternalRow(2))
+ val left = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(a), rows))
+ val right = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(b), rows))
+ val join = ShuffledHashJoinExec(Seq(a), Seq(b), Inner, BuildLeft, None, left, right)
+ withSQLConf(SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> "true") {
+ assert(ValidateRequirements.validate(join),
+ s"aligned but ungrouped keyed sides pair, and that pairing is the requirement:\n$join")
+
+ // The same key rows in another order do not pair: position by position is the contract.
+ val off = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(1), InternalRow(2), InternalRow(1))))
+ assert(!ValidateRequirements.validate(join.copy(right = off)),
+ "the same keys in a different order are not aligned")
+
+ // Nor does a keyed side pair with one that never pairs: a hashed side matches neither the
+ // keys nor the layout of a keyed one.
+ val hashed = DummySparkPlan(outputPartitioning = HashPartitioning(Seq(b), 3))
+ assert(!ValidateRequirements.validate(join.copy(right = hashed)),
+ "a keyed side does not pair with a hashed one")
+ }
+
+ // An ungrouped side is a plan only partially clustered distribution builds, so with that off
+ // the per-side check stands: nothing is left to group the side, and a pair the plan does not
+ // hold is not one the operator reads.
+ assert(!ValidateRequirements.validate(join),
+ s"an ungrouped side is admitted only where something builds one:\n$join")
+
+ // And a pair each side satisfies on its own is still refused when the sides do not line up,
+ // which is the operator's requirement: the pairing, not the per-side answer.
+ val leftKeys = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(a), Seq(InternalRow(1), InternalRow(2), InternalRow(3))))
+ val rightKeys = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(4), InternalRow(5), InternalRow(6))))
+ assert(leftKeys.outputPartitioning.satisfies(ClusteredDistribution(Seq(a))) &&
+ rightKeys.outputPartitioning.satisfies(ClusteredDistribution(Seq(b))),
+ "test setup: each side answers its own clustering")
+ assert(!ValidateRequirements.validate(ShuffledHashJoinExec(
+ Seq(a), Seq(b), Inner, BuildLeft, None, leftKeys, rightKeys)),
+ "the sides do not line up, and nothing else the operator reads says otherwise")
+ }
+
+ test("SPARK-59671: a keyed pair is judged on the layouts the plan holds") {
+ // Both permissions on, so the shape is the one partially clustered distribution spreads.
+ // Neither side's keys are deduped or re-sorted to make the pair line up: reading it through
+ // `createShuffleSpec` would do both, so two sides whose layouts disagree as they stand would
+ // answer as though they agreed.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val rows = Seq(InternalRow(1), InternalRow(1), InternalRow(2))
+ withSQLConf(
+ SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> "true") {
+ val left = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(a), rows))
+ // The same keys in another order, which a projection onto distinct sorted keys would
+ // normalize away.
+ val off = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(1), InternalRow(2), InternalRow(1))))
+ assert(!ValidateRequirements.validate(ShuffledHashJoinExec(
+ Seq(a), Seq(b), Inner, BuildLeft, None, left, off)),
+ "the sides report different layouts, and nothing normalizes them")
+
+ // And a count: the other side holds two partitions, and neither side has the node that would
+ // make the two agree.
+ val twoKeys = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(1), InternalRow(2))))
+ assert(!ValidateRequirements.validate(ShuffledHashJoinExec(
+ Seq(a), Seq(b), Inner, BuildLeft, None, left, twoKeys)),
+ "a child holding three partitions does not pair with one holding two")
+ }
+ }
+
+ test("SPARK-59671: a pair is judged on the key the operation clusters on") {
+ // The subset permission applies where the operation's keys are a subset of the source's
+ // partitioning keys: a side partitioned on `[a, b]` can serve an operator on `[a]`. Such a side
+ // is judged on its own partitions under `[a]`, which is what its own key already gives it when
+ // dropping `b` merges no partition, whether or not a node stands over it. No key is deduped and
+ // none is re-sorted, so the pairs that line up are the ones whose keys line up as reported.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val c = AttributeReference("c", IntegerType)()
+ val d = AttributeReference("d", IntegerType)()
+ withSQLConf(
+ SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> "false") {
+ def joinOnKeys(left: SparkPlan, right: SparkPlan): SparkPlan =
+ ShuffledHashJoinExec(Seq(a), Seq(c), Inner, BuildLeft, None, left, right)
+
+ // The projection's result, which is what such a plan reports from the side it grouped.
+ val projected = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(a), Seq(InternalRow(1), InternalRow(2))))
+ val projectedRight = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(c), Seq(InternalRow(1), InternalRow(2))))
+ assert(ValidateRequirements.validate(joinOnKeys(projected, projectedRight)),
+ "the layout the grouping node leaves is what the pair is judged on")
+
+ // The source's keys, one step earlier: the second expression is the operation's to drop, and
+ // what is left is the same pair, so it stands.
+ val source = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(a, b), Seq(InternalRow(1, 1), InternalRow(2, 2))))
+ val sourceRight = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(c, d), Seq(InternalRow(1, 1), InternalRow(2, 2))))
+ assert(ValidateRequirements.validate(joinOnKeys(source, sourceRight)),
+ s"the key the operation clusters on is what the side offers:\n$source")
+
+ // And it is offered in the order it is reported, so a side whose keys run the other way is
+ // not aligned: a spec that sorted them would call these two the same pair.
+ val reversed = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(a, b), Seq(InternalRow(2, 2), InternalRow(1, 1))))
+ assert(!ValidateRequirements.validate(joinOnKeys(reversed, sourceRight)),
+ s"the keys are compared as reported, not as sorted:\n$reversed")
+ }
+ }
+
+ test("SPARK-59671: a single clustered child still owes its own grouping") {
+ // The pairing stands in for the per-child check only where children pair with each other.
+ // An operator with a single clustered child, an aggregate over a join output say, is judged
+ // per side: the ungrouped keyed layout does not satisfy it until a grouping stands under.
+ val a = AttributeReference("a", IntegerType)()
+ val child = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(a), Seq(InternalRow(1), InternalRow(1))))
+ val parent = DummySparkPlan(
+ children = Seq(child),
+ requiredChildDistribution = Seq(ClusteredDistribution(Seq(a))),
+ requiredChildOrdering = Seq(Nil))
+ assert(!ValidateRequirements.validate(parent), "an ungrouped child fails a lone clustered slot")
Review Comment:
**[Test] This doesn't guard the property its comment names.**
The conf is off and the parent is a `DummySparkPlan`, so `partiallyClusteredJoinType` is `None` and `mayBeUngrouped` is false independently of `clusteredMultiChild`. The test would still pass if that conjunct were dropped from `mayBeUngrouped`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1437,6 +1464,74 @@ object PartitioningCollection {
case other => other.satisfies(required)
}
+ /**
+ * The specs `p` offers for `distribution`, one per member that may serve it, each of them the
+ * layout that member reports (`reportedSpecOf`): the partitions it has, in the order it reports
+ * them, with the partition expressions that carry no cluster key left out. Leaving those out is
+ * what lets two members pair when the operation clusters on part of what they are partitioned
+ * on, and it invents nothing: a kept key is that partition's own, the count is the member's own,
+ * and the order is the member's own.
+ *
+ * A keyed member is admitted on `satisfies`, the as-it-stands question, count included; a member
+ * that is not keyed is asked on `satisfies` as well and has no projection to make. The one member
+ * whose layout a finished plan may report without satisfying the distribution is an ungrouped
+ * keyed one: that is the shape partially clustered distribution spreads. It is admitted only
+ * where the caller says so, through `mayBeUngrouped`, which carries both halves of that
+ * admission: the configuration that builds such a shape, and the operator whose producer spreads
+ * it. `mayServeUngrouped` is the as-it-stands question for it, since `satisfies`
+ * adds `isGrouped` on top.
+ *
+ * This is the planner's admission of a member (`EnsureRequirements.createKeyedShuffleSpecs`) less
+ * the coverage of every operation key it requires there
+ * (`spark.sql.requireAllClusterKeysForCoPartition`), which is a skew heuristic: a member whose
+ * partitioning keys cover only a subset of the operation's keys is a sound pairing.
+ *
+ * @param mayBeUngrouped whether an ungrouped keyed member of `p` may serve `distribution` here.
+ * Only the caller knows both halves of that; see above.
+ */
+ private[sql] def specsForPairing(
+ p: Partitioning,
+ distribution: ClusteredDistribution,
+ mayBeUngrouped: Boolean): Seq[ShuffleSpec] =
+ flatten(p).flatMap {
+ case k: KeyedPartitioning =>
+ val pairsAsIs = k.satisfies(distribution) ||
+ (mayBeUngrouped &&
+ distribution.requiredNumPartitions.forall(_ == k.numPartitions) &&
+ k.mayServeUngrouped(distribution))
+ Option.when(pairsAsIs)(reportedSpecOf(k, distribution))
+ case other =>
+ Option.when(other.satisfies(distribution))(other.createShuffleSpec(distribution))
+ }
+
+ /**
+ * The spec for a member a finished plan reports: its own layout with the partition expressions
+ * that carry no cluster key left out. No key is deduped and no key is re-sorted, so a side offers
+ * a view of the partitions the plan has rather than the layout
+ * `KeyedPartitioning.createShuffleSpec` builds for a node a planner would insert: that one dedups
+ * and sorts, and its count is the grouped one, which the plan does not hold.
+ *
+ * A projection here is a relabelling of the member's partitions, and this leans on the caller for
+ * that: only `keysSatisfy`'s projecting half admits such a member, and it holds the projection to
+ * one that merges no partition, handing the positions it checked to this method.
+ *
+ * A marked layout is not projected: its claim routes the rows of an undeclared key by a hash over
+ * these keys in this order, and dropping one breaks the claim. The clause cannot fire as the
+ * admission stands, and it is kept because the causality is not local: an admitted marked member
+ * has its keys covering the clustering structurally, since the projecting half of `keysSatisfy`
+ * excludes a marked one, so it leaves here unprojected either way.
+ */
+ private def reportedSpecOf(
+ k: KeyedPartitioning,
+ distribution: ClusteredDistribution): KeyedShuffleSpec = {
+ val positions = k.positionsCoveringClusterKeys(distribution).toSeq
+ if (positions.size == k.expressions.length || k.mayContainUnknownPartitionKeys) {
Review Comment:
nit: `|| k.mayContainUnknownPartitionKeys` can't change the outcome: a marked member is only admitted through the non-projecting half of `keysSatisfy`, and with one reference per expression that already gives full coverage, as the doc below concedes. Consider dropping the clause and its paragraph. Also, the doc at lines 1514-1516 says `keysSatisfy` hands the positions it checked to this method, but they are recomputed on line 1527.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -9228,6 +9229,101 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59671: a partially clustered join leaves AQE's shuffle coalescing alone") {
+ // AQE validates a stage's whole candidate plan before accepting a shuffle-read change: on the
+ // base, a storage-partitioned join whose sides are aligned but not grouped kept every shuffle
+ // in its stage uncoalesced, unrelated ones included. Partially clustered distribution plans
+ // such a pair: the side that keeps its splits spreads them, and the other replicates its
+ // group across them, so both report repeated keys on purpose. The assertions below pin the
+ // read that the join's presence must leave alone.
+ val idCols = Array(Column.create("id", IntegerType), Column.create("data", StringType))
+ createTable("pc1", idCols, Array(identity("id")))
+ createTable("pc2", idCols, Array(identity("id")))
+ sql("INSERT INTO testcat.ns.pc1 VALUES (1, 'a1'), (2, 'a2'), (3, 'a3')")
+ // Key 1 twice: the side holding it keeps and spreads its two splits, which is what makes the
+ // pair ungrouped while its keys still line up.
+ sql("INSERT INTO testcat.ns.pc2 VALUES (1, 'b1'), (1, 'b1b'), (2, 'b2'), (4, 'b4')")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> "true",
+ "spark.sql.autoBroadcastJoinThreshold" -> "-1") {
+ val df = sql(
+ s"""
+ |SELECT /*+ MERGE(a) */ a.id, b.data
+ |FROM testcat.ns.pc1 a JOIN testcat.ns.pc2 b ON a.id = b.id
+ |UNION ALL
+ |SELECT count(*), cast(id % 2 AS STRING)
+ |FROM testcat.ns.pc1 GROUP BY id % 2
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(1, "b1"), Row(1, "b1b"), Row(2, "b2"), Row(2, "1"), Row(1, "0")))
+
+ val plan = stripAQEPlan(df.queryExecution.executedPlan)
+ val joins = collect(plan) { case j: ShuffledJoin => j }
+ assert(joins.size == 1, s"test setup: one storage-partitioned join:\n$plan")
+ assert(collectGroupPartitions(plan).exists { g =>
+ PartitioningCollection.representativeOf(g.outputPartitioning).exists(!_.isGrouped)
+ }, s"test setup: the pair is spread, so a side repeats its keys:\n$plan")
+ // The join's side shuffles nothing (this suite's `collectShuffles` counts the exchanges a
+ // join reads through), and the chain shuffles once, for the aggregate: the stage holding
+ // both is the one whose read the join's presence could have kept uncoalesced.
+ assert(collectShuffles(plan).isEmpty, s"the join shuffles nothing:\n$plan")
+ assert(collect(plan) { case s: ShuffleExchangeExec => s }.size === 1,
Review Comment:
nit: This is `collectAllShuffles(plan).size === 1`. Similarly, lines 9318-9319 could use `keyedPartitioningsOf(Seq(p)).size >= 2`, and line 9314 could use `checkAnswer(df, expected)`, as the sibling test does at line 9260.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +47,91 @@ 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, and so does the join type's capability, the second gate it applies before spreading
+ // a side (`canDuplicateLeftSide` / `canDuplicateRightSide`; a `FullOuter` join may duplicate
+ // neither side, so nothing spreads such a pair). That gate is asked here rather than through
+ // `partiallyClusteredJoinType`, which is the producer's entry to key-group checking at all: a
+ // kind turned away there would lose the storage-partitioned join it can still plan without
+ // spreading. What is left to the member, its count and the permission for the collapse it went
+ // through, is asked in the spec.
+ val mayBeUngrouped = clusteredMultiChild &&
+ SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled &&
Review Comment:
**The waiver omits the producer's push precondition.**
The comment above says both halves of the producer's admission are asked here, but the producer only spreads a side inside the push branch, which is gated by `pushCommonValues = ... && (v2BucketingPushPartValuesEnabled || v2BucketingAllowKeysSubsetOfPartitionKeys)` (`EnsureRequirements.scala:717-720`). The dependency of partial clustering on `pushPartValues` is documented but not enforced, so with `partiallyClusteredDistribution=true`, `pushPartValues=false` and `allowKeysSubsetOfPartitionKeys=false`, no spread pair can ever be built, yet every Inner/Left/Right SMJ/SHJ still waives the per-child check (widening the issue on line 79). It is also read from the session conf at validation time rather than from anything the plan records.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/ValidateRequirementsSuite.scala:
##########
@@ -158,4 +163,341 @@ class ValidateRequirementsSuite extends SharedSparkSession {
testNestedJoin(Seq((2, 2), (1, 1)), Seq((2, 2)), Seq(5, 5, 5), false)
testNestedJoin(Seq((2, 2), (1, 1)), Seq((2, 5)), Seq(5, 5, 5), false)
}
+
+ test("SPARK-59671: a co-partitioning operator judges keyed children by their pairing") {
+ // The sides of a storage-partitioned join aligned for skew repeat their spread keys on
+ // purpose: neither satisfies a clustering on its own, yet the two key sequences agree
+ // index by index, and the pairing is what the operator reads.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val rows = Seq(InternalRow(1), InternalRow(1), InternalRow(2))
+ val left = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(a), rows))
+ val right = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(b), rows))
+ val join = ShuffledHashJoinExec(Seq(a), Seq(b), Inner, BuildLeft, None, left, right)
+ withSQLConf(SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> "true") {
+ assert(ValidateRequirements.validate(join),
+ s"aligned but ungrouped keyed sides pair, and that pairing is the requirement:\n$join")
+
+ // The same key rows in another order do not pair: position by position is the contract.
+ val off = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(1), InternalRow(2), InternalRow(1))))
+ assert(!ValidateRequirements.validate(join.copy(right = off)),
+ "the same keys in a different order are not aligned")
+
+ // Nor does a keyed side pair with one that never pairs: a hashed side matches neither the
+ // keys nor the layout of a keyed one.
+ val hashed = DummySparkPlan(outputPartitioning = HashPartitioning(Seq(b), 3))
+ assert(!ValidateRequirements.validate(join.copy(right = hashed)),
+ "a keyed side does not pair with a hashed one")
+ }
+
+ // An ungrouped side is a plan only partially clustered distribution builds, so with that off
+ // the per-side check stands: nothing is left to group the side, and a pair the plan does not
+ // hold is not one the operator reads.
+ assert(!ValidateRequirements.validate(join),
+ s"an ungrouped side is admitted only where something builds one:\n$join")
+
+ // And a pair each side satisfies on its own is still refused when the sides do not line up,
+ // which is the operator's requirement: the pairing, not the per-side answer.
+ val leftKeys = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(a), Seq(InternalRow(1), InternalRow(2), InternalRow(3))))
+ val rightKeys = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(4), InternalRow(5), InternalRow(6))))
+ assert(leftKeys.outputPartitioning.satisfies(ClusteredDistribution(Seq(a))) &&
+ rightKeys.outputPartitioning.satisfies(ClusteredDistribution(Seq(b))),
+ "test setup: each side answers its own clustering")
+ assert(!ValidateRequirements.validate(ShuffledHashJoinExec(
+ Seq(a), Seq(b), Inner, BuildLeft, None, leftKeys, rightKeys)),
+ "the sides do not line up, and nothing else the operator reads says otherwise")
+ }
+
+ test("SPARK-59671: a keyed pair is judged on the layouts the plan holds") {
+ // Both permissions on, so the shape is the one partially clustered distribution spreads.
+ // Neither side's keys are deduped or re-sorted to make the pair line up: reading it through
+ // `createShuffleSpec` would do both, so two sides whose layouts disagree as they stand would
+ // answer as though they agreed.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val rows = Seq(InternalRow(1), InternalRow(1), InternalRow(2))
+ withSQLConf(
+ SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> "true") {
+ val left = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(a), rows))
+ // The same keys in another order, which a projection onto distinct sorted keys would
+ // normalize away.
+ val off = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(1), InternalRow(2), InternalRow(1))))
+ assert(!ValidateRequirements.validate(ShuffledHashJoinExec(
+ Seq(a), Seq(b), Inner, BuildLeft, None, left, off)),
+ "the sides report different layouts, and nothing normalizes them")
+
+ // And a count: the other side holds two partitions, and neither side has the node that would
+ // make the two agree.
+ val twoKeys = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(1), InternalRow(2))))
+ assert(!ValidateRequirements.validate(ShuffledHashJoinExec(
+ Seq(a), Seq(b), Inner, BuildLeft, None, left, twoKeys)),
+ "a child holding three partitions does not pair with one holding two")
+ }
+ }
+
+ test("SPARK-59671: a pair is judged on the key the operation clusters on") {
+ // The subset permission applies where the operation's keys are a subset of the source's
+ // partitioning keys: a side partitioned on `[a, b]` can serve an operator on `[a]`. Such a side
+ // is judged on its own partitions under `[a]`, which is what its own key already gives it when
+ // dropping `b` merges no partition, whether or not a node stands over it. No key is deduped and
+ // none is re-sorted, so the pairs that line up are the ones whose keys line up as reported.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val c = AttributeReference("c", IntegerType)()
+ val d = AttributeReference("d", IntegerType)()
+ withSQLConf(
+ SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> "false") {
+ def joinOnKeys(left: SparkPlan, right: SparkPlan): SparkPlan =
+ ShuffledHashJoinExec(Seq(a), Seq(c), Inner, BuildLeft, None, left, right)
+
+ // The projection's result, which is what such a plan reports from the side it grouped.
+ val projected = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(a), Seq(InternalRow(1), InternalRow(2))))
+ val projectedRight = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(c), Seq(InternalRow(1), InternalRow(2))))
+ assert(ValidateRequirements.validate(joinOnKeys(projected, projectedRight)),
+ "the layout the grouping node leaves is what the pair is judged on")
+
+ // The source's keys, one step earlier: the second expression is the operation's to drop, and
+ // what is left is the same pair, so it stands.
+ val source = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(a, b), Seq(InternalRow(1, 1), InternalRow(2, 2))))
+ val sourceRight = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(c, d), Seq(InternalRow(1, 1), InternalRow(2, 2))))
+ assert(ValidateRequirements.validate(joinOnKeys(source, sourceRight)),
+ s"the key the operation clusters on is what the side offers:\n$source")
+
+ // And it is offered in the order it is reported, so a side whose keys run the other way is
+ // not aligned: a spec that sorted them would call these two the same pair.
+ val reversed = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(a, b), Seq(InternalRow(2, 2), InternalRow(1, 1))))
+ assert(!ValidateRequirements.validate(joinOnKeys(reversed, sourceRight)),
+ s"the keys are compared as reported, not as sorted:\n$reversed")
+ }
+ }
+
+ test("SPARK-59671: a single clustered child still owes its own grouping") {
+ // The pairing stands in for the per-child check only where children pair with each other.
+ // An operator with a single clustered child, an aggregate over a join output say, is judged
+ // per side: the ungrouped keyed layout does not satisfy it until a grouping stands under.
+ val a = AttributeReference("a", IntegerType)()
+ val child = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(a), Seq(InternalRow(1), InternalRow(1))))
+ val parent = DummySparkPlan(
+ children = Seq(child),
+ requiredChildDistribution = Seq(ClusteredDistribution(Seq(a))),
+ requiredChildOrdering = Seq(Nil))
+ assert(!ValidateRequirements.validate(parent), "an ungrouped child fails a lone clustered slot")
+ }
+
+ test("SPARK-59671: a partially clustered pair planned by the rule passes validation") {
+ // Partially clustered distribution spreads a side, so its replicate side reports a
+ // non-grouped layout whose keys repeat on purpose, and the validator's per-side check
+ // refused such a pair. AQE validates a stage's whole candidate plan before accepting a
+ // shuffle-read change, so the plan's stage took no coalescing either. The pairing takes
+ // it: the spread is deliberate, and the two sides agree index by index.
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> "true") {
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val left = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(a), Seq(InternalRow(1), InternalRow(1), InternalRow(2))))
+ val right = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(1), InternalRow(2))))
+ val plan = new EnsureRequirements().apply(
+ SortMergeJoinExec(Seq(a), Seq(b), Inner, None, left, right))
+ val join = plan.collectFirst { case j: SortMergeJoinExec => j }
+ .getOrElse(fail(s"expected the join back:\n${plan.treeString}"))
+ def groupingAt(plan: SparkPlan): Option[GroupPartitionsExec] = plan match {
+ case g: GroupPartitionsExec => Some(g)
+ case s: SortExec if !s.global => groupingAt(s.child)
+ case _ => None
+ }
+ val sideGroupings = join.children.flatMap(groupingAt)
+ assert(sideGroupings.size == 2,
+ s"test setup: both sides are aligned:\n${plan.treeString}")
+ assert(sideGroupings.exists { g =>
+ PartitioningCollection.representativeOf(g.outputPartitioning).exists(!_.isGrouped)
+ }, s"test setup: a side keeps its splits, so its keys repeat:\n${plan.treeString}")
+ assert(ValidateRequirements.validate(plan),
+ s"a spread pair that agrees on its keys is accepted:\n${plan.treeString}")
+ }
+ }
+
+ test("SPARK-59671: a collection of keyed members is judged by its pairing") {
+ // A side that reports several keyed alternatives (a projection over a join keeps one per join
+ // key column) is judged on the members the admission keeps, not on the collection's own spec
+ // build having to leave one behind: a side whose members all repeat their keys is a pair the
+ // planner aligns without grouping, and it holds up when the member keyed on the join key is
+ // the one that lines up with the other side.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val rows = Seq(InternalRow(1), InternalRow(1), InternalRow(2))
+ def keyed(attr: AttributeReference): DummySparkPlan =
+ DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(attr), rows))
+ val bothAlternatives = DummySparkPlan(outputPartitioning =
+ PartitioningCollection.fromPartitionings(Seq(
+ KeyedPartitioning(Seq(a), rows), KeyedPartitioning(Seq(b), rows))))
+ // The alternatives repeat their keys, so they are the layout partially clustered distribution
+ // spreads, and that is where the admission keeps an ungrouped member.
+ withSQLConf(SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> "true") {
+ assert(ValidateRequirements.validate(ShuffledHashJoinExec(
+ Seq(a), Seq(a), Inner, BuildLeft, None, bothAlternatives, keyed(a))),
+ "the member keyed on the join key pairs with the other side")
+
+ // The pairing still has to be there: a side whose members are keyed on something else offers
+ // nothing to pair with.
+ val wrongKeys = DummySparkPlan(outputPartitioning =
+ PartitioningCollection.fromPartitionings(Seq(
+ KeyedPartitioning(Seq(b), rows), KeyedPartitioning(Seq(b), rows))))
+ assert(!ValidateRequirements.validate(ShuffledHashJoinExec(
+ Seq(a), Seq(a), Inner, BuildLeft, None, wrongKeys, keyed(a))),
+ "a side offering no member keyed on the join keys does not pair")
+ }
+ }
+
+ test("SPARK-59671: a multi-child clustered operator is judged on its children's pairing") {
+ // Every operator whose children all owe a `ClusteredDistribution` reads one layout they have to
+ // hold together, whichever operator it is: a cogroup zips corresponding partitions, so two
+ // sides that each satisfy the distribution on their own are not enough.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val rows = Seq(InternalRow(1), InternalRow(1), InternalRow(2))
+ def cogroupOf(left: SparkPlan, right: SparkPlan): SparkPlan = CoGroupExec(
+ (key: Any, l: Iterator[Any], r: Iterator[Any]) => Nil,
+ Literal(1), Literal(1), Literal(1), Seq(a), Seq(b), Seq(a), Seq(b), Nil, Nil,
+ AttributeReference("obj", ObjectType(classOf[AnyRef]))(), left, right)
+ def grouped(attr: AttributeReference, keys: Seq[Int]): DummySparkPlan = DummySparkPlan(
+ outputOrdering = Seq(SortOrder(attr, Ascending)),
+ outputPartitioning = KeyedPartitioning(Seq(attr), keys.map(InternalRow(_))))
+
+ withSQLConf(SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> "true") {
+ // The ungrouped shape is waived only for the producer that spreads a side, so a cogroup over
+ // the very pair a join reads is refused: it would run its function once per spread part.
+ val left = DummySparkPlan(
+ outputOrdering = Seq(SortOrder(a, Ascending)),
+ outputPartitioning = KeyedPartitioning(Seq(a), rows))
+ val right = DummySparkPlan(
+ outputOrdering = Seq(SortOrder(b, Ascending)),
+ outputPartitioning = KeyedPartitioning(Seq(b), rows))
+ assert(ValidateRequirements.validate(
+ ShuffledHashJoinExec(Seq(a), Seq(b), Inner, BuildLeft, None, left, right)),
+ "test setup: the join reads this pair")
+ assert(!ValidateRequirements.validate(cogroupOf(left, right)),
+ "a cogroup reads no spread side, and this pair passes nothing else")
+ }
+
+ // What a cogroup shares with a join is the mutual check on the layouts its sides report.
+ def cogroup(aKeys: Seq[Int], bKeys: Seq[Int]): SparkPlan =
+ cogroupOf(grouped(a, aKeys), grouped(b, bKeys))
+ assert(ValidateRequirements.validate(cogroup(Seq(1, 2), Seq(1, 2))),
+ "two sides holding the same grouped layout are read as they stand")
+ assert(!ValidateRequirements.validate(cogroup(Seq(1, 2), Seq(1, 2, 3))),
+ "a side holding three partitions does not pair with one holding two")
+ assert(!ValidateRequirements.validate(cogroup(Seq(1, 2), Seq(2, 1))),
+ "the same keys in another order are not aligned")
+ }
+
+ test("SPARK-59671: an ungrouped pair is waived only where the producer spreads a side") {
+ // `EnsureRequirements.checkKeyGroupCompatible` is the path that spreads one side against a
+ // repeater, and it is entered for a sort-merge or shuffled-hash join. A sort-merge as-of join
+ // is a `ShuffledJoin` too and builds none, so a pair it reads ungrouped is one no producer made
+ // and the per-side check stands.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val rows = Seq(InternalRow(1), InternalRow(1), InternalRow(2))
+ withSQLConf(SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> "true") {
+ val left = DummySparkPlan(
+ outputOrdering = Seq(SortOrder(a, Ascending), SortOrder(a, Ascending)),
+ outputPartitioning = KeyedPartitioning(Seq(a), rows))
+ val right = DummySparkPlan(
+ outputOrdering = Seq(SortOrder(b, Ascending), SortOrder(b, Ascending)),
+ outputPartitioning = KeyedPartitioning(Seq(b), rows))
+ assert(ValidateRequirements.validate(
+ ShuffledHashJoinExec(Seq(a), Seq(b), Inner, BuildLeft, None, left, right)),
+ "test setup: the join the waiver is for reads this pair")
+ assert(!ValidateRequirements.validate(SortMergeAsOfJoinExec(
+ Seq(a), Seq(b), Seq(a), Seq(b), GreaterThan(a, b), a, Inner, None, left, right)),
+ "an as-of join builds no spread side, so the same pair is refused")
+ // Nor for a join type that may duplicate neither side: the producer applies that gate before
+ // it spreads, so a full outer join over the same pair is one no producer builds either.
+ assert(!ValidateRequirements.validate(
+ SortMergeJoinExec(Seq(a), Seq(b), FullOuter, None, left, right)),
+ "neither side of a full outer join may be duplicated, so nothing spreads this pair")
+ }
+ }
+
+ test("SPARK-59671: a side is judged on the member that pairs, whichever one it is") {
+ // A side that reports several keyed alternatives offers all of them: the one keyed on the
+ // join key is the one that lines up, and it need not be the first the side reports. Reading a
+ // single member would refuse this pair, which the planner builds whenever a projection keeps
+ // two key columns.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val rows = Seq(InternalRow(1), InternalRow(1), InternalRow(2))
+ def keyed(attr: AttributeReference): KeyedPartitioning = KeyedPartitioning(Seq(attr), rows)
+ val left = DummySparkPlan(outputPartitioning =
+ PartitioningCollection.fromPartitionings(Seq(keyed(b), keyed(a))))
+ val right = DummySparkPlan(outputPartitioning = keyed(a))
+ withSQLConf(SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> "true") {
+ assert(ValidateRequirements.validate(ShuffledHashJoinExec(
+ Seq(a), Seq(a), Inner, BuildLeft, None, left, right)),
+ "the second member is the one keyed on the join key, and it pairs")
+ }
+ }
+
+ test("SPARK-59671: the ordering requirement is not part of the exemption") {
+ // A pair can line up and still owe its operator an ordering: the exemption covers the
+ // distribution clause alone, so a sort-merge join over the aligned sides, with nothing
+ // ordering them, is refused.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val rows = Seq(InternalRow(1), InternalRow(1), InternalRow(2))
+ val left = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(a), rows))
+ val right = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(b), rows))
+ assert(left.outputOrdering.isEmpty && right.outputOrdering.isEmpty,
+ "test setup: nothing orders the sides")
+ withSQLConf(SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> "true") {
+ assert(!ValidateRequirements.validate(
+ SortMergeJoinExec(Seq(a), Seq(b), Inner, None, left, right)),
+ "the sides pair, and still owe the join its ordering")
+ }
+ }
+
+ test("SPARK-59671: a collapsed pair is not admitted on its pairing alone") {
+ // A layout whose keys were collapsed (a partition standing for several of the source's) serves
+ // a clustering through a grouping node only where that grouping is permitted, and the
+ // permission is a config. Without it nothing admits such a member, and a side offering nothing
+ // has nothing to pair with.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ def collapsed(attr: AttributeReference): DummySparkPlan = DummySparkPlan(
+ outputPartitioning = KeyedPartitioning(Seq(attr),
+ Seq(InternalRow(1), InternalRow(1), InternalRow(2)))
+ .withLayout(_.copy(isCollapsed = true)))
+ def pair: SparkPlan = ShuffledHashJoinExec(Seq(a), Seq(b), Inner, BuildLeft, None,
+ collapsed(a), collapsed(b))
+
+ withSQLConf(SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "false") {
+ assert(!ValidateRequirements.validate(pair),
+ s"a collapsed pair is refused while the subset permission is off:\n$pair")
Review Comment:
**[Test] This assertion passes for a reason unrelated to the collapse permission.**
Partial clustering is off in this block, so `mayBeUngrouped` is false and the per-child `satisfies` refuses the ungrouped sides before `collapsedLayoutMayBeGrouped` is ever consulted. It would still pass with `.withLayout(_.copy(isCollapsed = true))` removed. Only the second block exercises the permission.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +47,91 @@ 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, and so does the join type's capability, the second gate it applies before spreading
+ // a side (`canDuplicateLeftSide` / `canDuplicateRightSide`; a `FullOuter` join may duplicate
+ // neither side, so nothing spreads such a pair). That gate is asked here rather than through
+ // `partiallyClusteredJoinType`, which is the producer's entry to key-group checking at all: a
+ // kind turned away there would lose the storage-partitioned join it can still plan without
+ // spreading. What is left to the member, its count and the permission for the collapse it went
+ // through, is asked in the spec.
+ val mayBeUngrouped = clusteredMultiChild &&
+ SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled &&
+ ShuffledJoin.partiallyClusteredJoinType(plan).exists { joinType =>
+ ShuffledJoin.canDuplicateLeftSide(joinType) ||
+ ShuffledJoin.canDuplicateRightSide(joinType)
+ }
+
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) {
Review Comment:
**[Simplification]** For a multi-child clustered operator, the per-child distribution clause and its waiver (lines 79-81) never change the result: `specsForPairing` admits a member only on `satisfies` or the precise waiver, and an empty side fails the pairing's `exists`. So the loop could check `(!clusteredMultiChild && !child.outputPartitioning.satisfies(distribution)) || !SortOrder.orderingSatisfies(...)`, and this chain could become `satisfied && (!clusteredMultiChild || satisfiesForPairing(...))`, with `mayBeUngrouped` computed inside `satisfiesForPairing`. Today the waiver is spelled twice (a loose `representativeOf(...).isDefined` form in the loop and the precise one in `specsForPairing`), and `outputPartitioning`, a `def` on `ShuffledJoin` that rebuilds the collection, is evaluated up to three times per child.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledJoin.scala:
##########
@@ -173,4 +174,22 @@ object ShuffledJoin {
case _: InnerLike | RightOuter => true
case _ => false
}
+
+ /**
+ * The join type of the operators whose partially clustered alignment spreads one side against a
+ * repeater, and `None` for every other operator. `EnsureRequirements.checkKeyGroupCompatible` is
+ * the only producer of such a pair, and `ValidateRequirements` waives an ungrouped side only for
+ * the operators this names, so the two read one list rather than a copy each.
+ *
+ * A `SortMergeAsOfJoinExec` is a `ShuffledJoin` and builds none, which is why the kinds are named
+ * rather than read off the trait. One of these kinds can still be unable to spread for its join
+ * type, which is a separate question: no side of a `FullOuter` join may be duplicated, so the
+ * alignment spreads nothing there. The side that repeats is decided next to this, by
+ * `canDuplicateLeftSide` and `canDuplicateRightSide`.
+ */
+ def partiallyClusteredJoinType(plan: SparkPlan): Option[JoinType] = plan match {
Review Comment:
nit: `EnsureRequirements.checkKeyGroupCompatible` uses this as the entry gate for every storage-partitioned join (conf off and `FullOuter` included), not only for partially clustered ones, which is why the doc here and the paragraph in `ValidateRequirements` have to explain the mismatch. A name like `storagePartitionedJoinType` would fit both callers, and a `canDuplicateEitherSide(joinType)` helper would replace the three copies of that check (`ValidateRequirements`, `EnsureRequirements.scala:810-813`, `OptimizeSkewedJoin.scala:107-108`).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -904,6 +899,15 @@ case class EnsureRequirements(
}
// Now we need to push-down the common partition information to the `GroupPartitionsExec`s.
+ //
+ // The two arguments below say which side does what, and exactly one of them is true:
Review Comment:
nit: This reads as an unconditional invariant, but it only holds when `applyPartialClustering` is true. With partial clustering off (the default), when neither side may replicate (line 813), or when the picked side is vetoed (lines 856-865 reset the flag), both `distributePartitions` arguments below are `false` and the two flags are not negations of each other. Could it be scoped, e.g. "When `applyPartialClustering` holds, exactly one side replicates ..."?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #58942:
URL: https://github.com/apache/spark/pull/58942#issuecomment-5833876945
Thanks, I confirmed the FullOuter duplication-capability gate, the revised compatibleAsIs wording, and the focused refusal coverage in this revision.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"conversation:5805859336","thread_id":"conversation:5805859336","verdict_sha256":"8e26d305a65705965995542a009188889aa25752be172203798719da4958df03"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #58942:
URL: https://github.com/apache/spark/pull/58942#issuecomment-6051851478
rebased to master, cc @dongjoon-hyun @cloud-fan @peter-toth @szehon-ho @uros-b if you have any more comments, thank you
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #58942:
URL: https://github.com/apache/spark/pull/58942#issuecomment-5771749814
Confirmed. specsForPairing can project and group an ungrouped KeyedPartitioning even though the finished child does not report that layout, and the broad coPartitioning exemption is not limited to the join producer path. I am retaining this as a correctness finding; the repair should validate as-is layouts except for the explicitly safe producer-specific ungrouped case, with coverage for subset-enabled/partial-disabled and non-join operators.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"conversation:5763610046","thread_id":"conversation:5763610046","verdict_sha256":"ebf00ce694aa0f0f32134bea65b9cd85a904ba39ccb4a63c35685b1ff2e293f4"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4132431783
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -854,8 +881,25 @@ case class KeyedPartitioning(
reducers: Seq[Option[KeyReducer]]): (Seq[DataType], Seq[InternalRowComparableWrapper]) =
KeyedPartitioning.reduceKeys(partitionKeys, keyDataTypes, reducers)
+ /**
+ * Whether the reason this layout is ungrouped is one `required` reads. A clustering pairs
+ * partitions index by index, so only the two alignment roles answer yes, and `None` does not:
+ * a layout that is not grouped and does not say why is waiting for the node that would settle
+ * it, and a finished plan has none left to insert. An ordering reads the keys without pairing
+ * anything, so settling is not what it asks: `None` and its own spread answer yes, while the
+ * two pairing roles answer no, built as they are for a clustering and not for an ordering.
+ */
+ private def ungroupingServes(required: Distribution): Boolean = required match {
+ case _: ClusteredDistribution =>
+ ungroupingOrigin.exists(r => r == SPREADS_SPLITS || r == REPEATS_GROUP)
Review Comment:
Thank you — this one is on me: I had enumerated the consumers of the stamped admission and concluded a stamped layout is only ever read by its paired join, and the enumeration missed `ShuffledJoin.outputPartitioning`'s verbatim arms. Your ablation pinned the cause to the line.
Fixed in 856fa0d43c7, taking your primary fix: the stamp is out of `satisfies` entirely, and only the pairing reads it.
`satisfies0`'s clustering arm is back to the base `isGrouped && keysSatisfy`, so every consumer but the pairing owes an ungrouped layout the settling node, whichever arm it arrived through. The pairing admits the two roles through a new `KeyedPartitioning.pairsUngrouped` (role, count, key space), consulted by `specsForPairing`; `ValidateRequirements` delegates the per-child distribution clause of a multi-child clustered operator to that pairing, on the subsumption argument you gave, with the ordering clause still asked per child.
Your repro is an end-to-end test now, "an aggregate over a left outer join's spread side groups it": on the pre-fix head it returns [1,1],[1,1],[2,1]; it now returns [1,2],[2,1], and pins both the verbatim spread side under the join and the grouping node above it.
I did not take the narrower alternative, for the reason you gave: it would leave "a stamped layout is only read by the join it was built for" unenforced.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -854,8 +881,25 @@ case class KeyedPartitioning(
reducers: Seq[Option[KeyReducer]]): (Seq[DataType], Seq[InternalRowComparableWrapper]) =
KeyedPartitioning.reduceKeys(partitionKeys, keyDataTypes, reducers)
+ /**
+ * Whether the reason this layout is ungrouped is one `required` reads. A clustering pairs
+ * partitions index by index, so only the two alignment roles answer yes, and `None` does not:
+ * a layout that is not grouped and does not say why is waiting for the node that would settle
+ * it, and a finished plan has none left to insert. An ordering reads the keys without pairing
+ * anything, so settling is not what it asks: `None` and its own spread answer yes, while the
+ * two pairing roles answer no, built as they are for a clustering and not for an ordering.
+ */
+ private def ungroupingServes(required: Distribution): Boolean = required match {
+ case _: ClusteredDistribution =>
+ ungroupingOrigin.exists(r => r == SPREADS_SPLITS || r == REPEATS_GROUP)
+ case _: OrderedDistribution =>
+ ungroupingOrigin.isEmpty || ungroupingOrigin.contains(SPREADS_FOR_ORDERING)
Review Comment:
Thanks — fixed in 856fa0d43c7 with the key-order check, at the `satisfies` layer rather than inside `keysSatisfy`: `keysSatisfy` also feeds the planner's `keysMaySatisfy` family, which has to admit unsorted keys so that `resolveChild` can build the spread that sorts them.
`satisfies0`'s ordered arm now requires `keysSortedFor(o)`: the keys ascend in the sequence the ordering reads them, on the `RowOrdering.create(o.ordering, attrs)` construction. `resolveChild` calls the same predicate for its keep-versus-spread decision, so the two answers cannot disagree. Your concat counterexample is pinned (`[1],[2],[1],[2]` answers false), and so is the grouped half of the tightening: a distinct-but-descending `[2,1]` answers false where the pushed head answered true. Both arms are mutation-checked; under a grouped exemption only the new fixtures flip.
I chose the check over stamping the as-is case: that shape has no node to carry a stamp, and the check makes the claim self-contained instead of resting on "grouped keys are sorted by construction" holding at every producer.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -486,13 +500,20 @@ case class CoalescedNullAwareHashPartitioning(
* one, clears the markers first. That is precision, not the
* guarantee: without it the merge would spread the spurious
* marker onto the accurate side.
+ * @param ungroupingOrigin Why the keys repeat on purpose, when a producer built them that way
+ * rather than leaving them for the grouping node that would settle them.
+ * `None` means unsettled: a clustering refuses such a layout, and only a
+ * node about to be inserted can make it serve one, while an ordering reads
+ * the keys without asking them to settle. See [[UngroupingOrigin]].
*/
+
case class KeyLayout(
@transient partitionKeys: Seq[InternalRowComparableWrapper],
dataTypes: Seq[DataType],
isGrouped: Boolean,
isCollapsed: Boolean,
- mayContainUnknownPartitionKeys: Boolean = false) {
+ mayContainUnknownPartitionKeys: Boolean = false,
+ ungroupingOrigin: Option[UngroupingOrigin] = None) {
Review Comment:
Thanks — added in 856fa0d43c7: `stringArgs` appends `ungroupingOrigin`. The `Option` appends nothing when empty, so a layout with no claim keeps its previous plan string.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #58942:
URL: https://github.com/apache/spark/pull/58942#issuecomment-5885837792
thank you @dongjoon-hyun , this is a great catch, and sorry for the late update.
**The head is now a single commit, 2ced4575f9e, and the design changed substantially since the last round.** The validator-side waiver is gone; the fact it tried to guess is recorded on the layout itself.
## Why the origin lives on the layout
An ungrouped keyed layout — repeated partition keys — is ambiguous to any reader. The same key list describes two different things: a side a producer built for an alignment, where one side's split of a key meets the other side's whole repeated group, so pairing the two index by index loses no match; and a raw source report nothing has settled, where the same pairing loses every match that crosses partitions. `isGrouped = false` does not distinguish them, and no reader downstream can reconstruct which one it is. The waiver guessed — from the session conf, the operator kind and the join type — and guessed wrong on exactly the layouts no producer had built (an AQE local read re-reporting a scan's splits). Requiring `isGrouped` is not the answer either: it makes the deliberately ungrouped shapes unservable, and a finished plan holding one then fails validation as a whole, which is what reverted AQE's shuffle read optimizations across the plan — the coalescing symptom this PR
started from, and the sort-branch one. Only the producer knows why the keys repeat, and a finished plan is judged on the layouts it holds, so the reason is declared where the shape is built and read where the layout is consumed. An unstamped layout is inert by default: it serves no clustering until the node that would settle it arrives.
## What the last round had
`ValidateRequirements` waived the per-child check for an operator list (`partiallyClusteredJoinType`), a conf read (`partiallyClusteredDistribution.enabled`) and a join-type capability check, then paired the sides through `specsForPairing`. Every correctness finding in this round's review traced back to that waiver guessing what the producer had built.
## What it is now
- `KeyLayout.ungroupingOrigin: Option[UngroupingOrigin]` says why a producer left the keys repeating on purpose. Three values: `SPREADS_SPLITS` / `REPEATS_GROUP` — the two complementary sides of a partially clustered alignment — and `SPREADS_FOR_ORDERING`, a spread an ordering requirement is read through.
- Only `EnsureRequirements` stamps, at the two sites that build those shapes; the alignment stamp stays gated by the existing `canDuplicateLeftSide`/`canDuplicateRightSide` checks.
- `GroupPartitionsExec` derives its routing from the stamp — the `distributePartitions` constructor field is gone — and `computeGrouping` writes the node's *own* origin into the layout it builds, so a child's claim is never inherited.
- `KeyedPartitioning.satisfies` reads the stamp per requirement: a clustering pairs partitions index by index, so an ungrouped layout serves one only with an alignment role; an ordering reads the keys without pairing anything, so it is answered by the keys alone. The latter also fixes the as-is sorted shape from the [ordering thread](https://github.com/apache/spark/pull/58942#discussion_r4109472427) — that union query is now an end-to-end test.
- `KeyedShuffleSpec.isCompatibleWith` accepts an ungrouped pair only as one spread side against one repeating side.
- Carriers keep one agreed claim: `toGrouped` spends it, `concat` and `fromPartitionings` keep it only where every member carries the same, and a regrouping forwards its node's own.
- `validateInternal` is back to its base loop-and-gate shape; the only change against master is the spec comparison line, now the pairing question over the layouts the plan reports (`specsForPairing`/`reportedSpecOf` — no key deduped, none re-sorted). `ShuffledJoin.scala` is no longer touched by this PR.
## Where the findings land
- The local-read data-loss path is closed by construction: a read re-reporting a scan's layout carries no stamp, fails the per-child `satisfies`, and AQE reverts it. The end-to-end regression test loses exactly the cross-mapper pairs (3 rows instead of 5) under a relaxed stamp gate.
- The which-side-replicated, push-precondition and `keyPositions`-assert findings are closed the same way — the validator reads no join type, no conf and no operator list, and an unstamped ungrouped layout never reaches a spec.
- Per-thread replies carry the details for each comment, including the two test rewrites.
also @cloud-fan @peter-toth for another review since the code is changed a lot.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59671][SQL] Validate a co-partitioned pair by its children's pairing [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4105379420
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -904,6 +899,15 @@ case class EnsureRequirements(
}
// Now we need to push-down the common partition information to the `GroupPartitionsExec`s.
+ //
+ // The two arguments below say which side does what, and exactly one of them is true:
+ // `replicateRightSide` is the negation of `replicateLeftSide`, and the branch above is taken
Review Comment:
**Nit (P3):** When applyPartialClustering is false, both distributePartitions arguments below are false. Please scope the exactly-one statement to the path where partial clustering is actually applied; as written, the comment documents a false invariant for ordinary or rejected paths.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +47,91 @@ 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, and so does the join type's capability, the second gate it applies before spreading
+ // a side (`canDuplicateLeftSide` / `canDuplicateRightSide`; a `FullOuter` join may duplicate
+ // neither side, so nothing spreads such a pair). That gate is asked here rather than through
+ // `partiallyClusteredJoinType`, which is the producer's entry to key-group checking at all: a
+ // kind turned away there would lose the storage-partitioned join it can still plan without
+ // spreading. What is left to the member, its count and the permission for the collapse it went
+ // through, is asked in the spec.
+ val mayBeUngrouped = clusteredMultiChild &&
+ SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled &&
+ ShuffledJoin.partiallyClusteredJoinType(plan).exists { joinType =>
+ ShuffledJoin.canDuplicateLeftSide(joinType) ||
+ ShuffledJoin.canDuplicateRightSide(joinType)
+ }
+
val satisfied = children.zip(requiredChildDistributions.zip(requiredChildOrderings)).forall {
case (child, (distribution, ordering))
- if !child.outputPartitioning.satisfies(distribution)
+ if (!child.outputPartitioning.satisfies(distribution) &&
+ !(mayBeUngrouped &&
Review Comment:
**Non-blocking (P2):** Please add a focused case for a join type with exactly one duplication capability, such as LeftOuter or RightOuter. The current Inner and FullOuter cases cover only both predicates true and both false, so changing this OR to AND would leave the suite green while rejecting a valid one-sided partially clustered layout during AQE validation.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/ValidateRequirementsSuite.scala:
##########
@@ -158,4 +163,213 @@ class ValidateRequirementsSuite extends SharedSparkSession {
testNestedJoin(Seq((2, 2), (1, 1)), Seq((2, 2)), Seq(5, 5, 5), false)
testNestedJoin(Seq((2, 2), (1, 1)), Seq((2, 5)), Seq(5, 5, 5), false)
}
+
+ test("SPARK-59671: a co-partitioning operator judges keyed children by their pairing") {
+ // The sides of a storage-partitioned join aligned for skew repeat their spread keys on
+ // purpose: neither satisfies a clustering on its own, yet the two key sequences agree
+ // index by index, and the pairing is what the operator reads.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val rows = Seq(InternalRow(1), InternalRow(1), InternalRow(2))
+ val left = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(a), rows))
+ val right = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(b), rows))
+ val join = ShuffledHashJoinExec(Seq(a), Seq(b), Inner, BuildLeft, None, left, right)
+ assert(ValidateRequirements.validate(join),
+ s"aligned but ungrouped keyed sides pair, and that pairing is the requirement:\n$join")
+
+ // The same key rows in another order do not pair: position by position is the contract.
+ val off = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(1), InternalRow(2), InternalRow(1))))
+ assert(!ValidateRequirements.validate(join.copy(right = off)),
+ "the same keys in a different order are not aligned")
+
+ // Nor does a keyed side with one that never pairs: a hashed side pairs with neither the keys
+ // nor the layout of a keyed one.
+ val hashed = DummySparkPlan(outputPartitioning = HashPartitioning(Seq(b), 3))
+ assert(!ValidateRequirements.validate(join.copy(right = hashed)),
+ "a keyed side does not pair with a hashed one")
+
+ // And a pair each side satisfies on its own is still refused when the sides do not line up,
+ // which is the operator's requirement: the pairing, not the per-side answer.
+ val leftKeys = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(a), Seq(InternalRow(1), InternalRow(2), InternalRow(3))))
+ val rightKeys = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(4), InternalRow(5), InternalRow(6))))
+ assert(leftKeys.outputPartitioning.satisfies(ClusteredDistribution(Seq(a))) &&
+ rightKeys.outputPartitioning.satisfies(ClusteredDistribution(Seq(b))),
+ "test setup: each side answers its own clustering")
+ assert(!ValidateRequirements.validate(ShuffledHashJoinExec(
+ Seq(a), Seq(b), Inner, BuildLeft, None, leftKeys, rightKeys)),
+ "the sides do not line up, and nothing else the operator reads says otherwise")
+ }
+
+ test("SPARK-59671: a subset-keyed pair is judged on its own layouts") {
+ // With the subset permission on, the planner builds pairs whose sides are grouped on a subset
Review Comment:
Confirmed: the helper and fixture now describe and test operation keys as the subset of source partition keys.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4068736081","thread_id":"inline:4068736081","verdict_sha256":"8e26d305a65705965995542a009188889aa25752be172203798719da4958df03"} -->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +46,73 @@ 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 a join 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.
+ // Only a shuffling join gets that reading: it matches within a partition and unions the
+ // matches, so an alignment that spreads one side against a repeating other is sound, while an
+ // operator that emits a per-key result from one partition, a cogroup for instance, would emit a
+ // partial one for every spread key. `EnsureRequirements.checkKeyGroupCompatible` is the join
+ // path that plans such a pair, and the only producer of one.
+ val coPartitioning = children.length > 1 && plan.isInstanceOf[ShuffledJoin] &&
Review Comment:
Confirmed: the shared SMJ/SHJ classifier excludes AS-OF joins, and the focused refusal test covers that boundary.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4072240006","thread_id":"inline:4072240006","verdict_sha256":"8e26d305a65705965995542a009188889aa25752be172203798719da4958df03"} -->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ValidateRequirements.scala:
##########
@@ -45,29 +46,73 @@ 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 a join 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.
+ // Only a shuffling join gets that reading: it matches within a partition and unions the
+ // matches, so an alignment that spreads one side against a repeating other is sound, while an
+ // operator that emits a per-key result from one partition, a cogroup for instance, would emit a
+ // partial one for every spread key. `EnsureRequirements.checkKeyGroupCompatible` is the join
+ // path that plans such a pair, and the only producer of one.
+ val coPartitioning = children.length > 1 && plan.isInstanceOf[ShuffledJoin] &&
+ 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).exists(k =>
+ k.isGrouped || PartitioningCollection.mayUngroupedMember)))
|| !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
+ // and one that repeats the whole group report the same keys as two sides that split the key,
+ // so that rests on the producer: only a join is admitted above, and
+ // `checkKeyGroupCompatible` is where it plans such a pair.
+ if (!satisfied) {
+ false
+ } else if (coPartitioning) {
+ val paired = satisfiesForPairing(children, requiredChildDistributions)
+ if (!paired) {
logDebug(s"ValidateRequirements failed: children not co-partitioned in\n$plan")
- false
}
+ paired
} else {
- satisfied
+ true
Review Comment:
Confirmed: generic multi-child compatibility is restored, with cogroup coverage for the passing, count-mismatch, and order-mismatch cases.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4072240021","thread_id":"inline:4072240021","verdict_sha256":"8e26d305a65705965995542a009188889aa25752be172203798719da4958df03"} -->
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -9228,6 +9229,100 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59671: a partially clustered join leaves AQE's shuffle coalescing alone") {
+ // AQE validates a stage's whole candidate plan before accepting a shuffle-read change, so a
Review Comment:
Confirmed: the comment now describes the base regression historically and matches the coalescing assertion.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4072240029","thread_id":"inline:4072240029","verdict_sha256":"8e26d305a65705965995542a009188889aa25752be172203798719da4958df03"} -->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1437,6 +1445,54 @@ object PartitioningCollection {
case other => other.satisfies(required)
}
+ /**
+ * Whether a finished plan may report an ungrouped keyed member, which only partially clustered
+ * distribution builds. Read by `specsForPairing` and by `ValidateRequirements`' per-side
+ * admission, so the two cannot drift.
+ */
+ private[sql] def mayUngroupedMember: Boolean =
+ SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled
+
+ /**
+ * The specs `p` offers for `distribution`, one per member that may serve it, each of them the
+ * layout that member reports. A keyed member is admitted on `satisfies`, the as-it-stands
+ * question, count included; a member that is not keyed is asked on `satisfies` as well and has no
+ * projection to make.
+ *
+ * The one member whose layout a finished plan may report without satisfying the distribution is
+ * an ungrouped keyed one: that is the shape partially clustered distribution spreads, and
+ * `mayUngroupedMember` confines it to the configuration that builds one. `keysSatisfy` is the
+ * as-it-stands question for it, since `satisfies` adds `isGrouped` on top.
+ *
+ * This is the planner's admission of a member (`EnsureRequirements.createKeyedShuffleSpecs`) less
+ * the coverage of every operation key it requires there
+ * (`spark.sql.requireAllClusterKeysForCoPartition`), which is a skew heuristic: a member whose
+ * partitioning keys are a superset of the operation's keys is a sound pairing.
+ *
+ * Two things the planner does are deliberately not done here, because they build a layout the
+ * plan does not hold, which is the one thing the validator must not read:
+ *
+ * - no projection onto the operation keys (`KeyedPartitioning.createShuffleSpec` makes one under
+ * `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys`). A plan in which the
+ * permission applies holds the projection already, in the grouping node
+ * `EnsureRequirements` inserted, so the member reports it and no projection is wanted.
+ * - no grouping of an ungrouped member (`tryCreate` builds from `toGrouped`), and no re-sorting
+ * of a grouped one: `toGrouped` dedups and sorts, both of which move the member's key rows.
+ */
+ private[sql] def specsForPairing(
+ p: Partitioning,
+ distribution: ClusteredDistribution): Seq[ShuffleSpec] =
+ flatten(p).flatMap {
+ case k: KeyedPartitioning =>
+ val pairsAsIs = k.satisfies(distribution) ||
+ (mayUngroupedMember &&
Review Comment:
Confirmed: the ungrouped path shares the planner's collapse permission, and the combined-configuration negative test covers it.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4072240037","thread_id":"inline:4072240037","verdict_sha256":"8e26d305a65705965995542a009188889aa25752be172203798719da4958df03"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org