peter-toth commented on code in PR #58339:
URL: https://github.com/apache/spark/pull/58339#discussion_r3922953551
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -219,6 +232,18 @@ case class GroupPartitionsExec(
PartitionGrouping(partitions, isGrouped, isCollapsed)
}
+ /**
+ * Whether this node's grouping leaves every partition where it was, i.e.
output partition i
+ * holds exactly input partition i. That is the only grouping that keeps a
marked layout's
+ * undeclared rows at hash(key) % numPartitions. The `forall` stops at the
first moved
+ * partition, so a reorder or coalesce is rejected without a full scan.
+ */
+ @transient private lazy val identityGrouping: Boolean =
Review Comment:
**Finding 23.** This answers the order and per-partition-membership halves
of
[r3920336519](https://github.com/apache/spark/pull/58339#discussion_r3920336519),
not the count half.
`alignToExpectedKeys` iterates `expectedPartitionKeys`, so a key the child
declares but the merged set does not is never emitted at all. When the dropped
ones are trailing, every kept group still satisfies `single == outputIndex`, so
`identityGrouping` is true while `numPartitions` shrank. The child's undeclared
rows sit at `nonNegativeMod(hash, childCount)` and the retained marker now
claims `nonNegativeMod(hash, groupedCount)`.
The merged set is a strict subset of a marked side's own keys on three
routes in `mergeAndDedupPartitions`: `Inner` under
`spark.sql.sources.v2.bucketing.partition.filter.enabled` intersects, and the
`LeftOuter` and `RightOuter` arms take one side's keys. A marked `[1, 2, 3]`
against an unmarked partner `[1, 2]` under the filter gives `expected = [1,
2]`, groups `[(1, [0]), (2, [1])]`, `identityGrouping = true`, and two
partitions where the hash was taken over three.
I re-checked the enumeration I gave at round 5 and it still holds, so this
is latent rather than live: on each of those routes the shrunk side is not the
side whose partitioning the join reports. `Inner` with one marked side clears
the mixed collection, `Inner` with both marked requires equal key sequences so
the intersection is the identity, `LeftOuter` shrinks only the right, and
`RightOuter` mirrors it. It closes in this same `lazy val`:
```scala
@transient private lazy val identityGrouping: Boolean =
grouping.partitions.size == child.outputPartitioning.numPartitions &&
grouping.partitions.zipWithIndex.forall {
case ((_, Seq(single)), outputIndex) => single == outputIndex
case _ => false
}
```
The doc sentence above it can then say "leaves every partition where it was,
and keeps all of them", which is what the marked contract needs.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -578,12 +582,42 @@ case class CoalescedNullAwareHashPartitioning(
* partitioning this one was derived from onto the same
key, so one key here can
* stand for several of the original ones. Sticky. See "Key
Collapse" above for
* what it gates and how it travels.
+ * @param mayContainUnknownPartitionKeys Whether the data may contain rows
whose partition key is
+ * not among the declared `partitionKeys`.
`KeyGroupedPartitioner`
+ * routes such rows by a deterministic hash
when a side is
+ * re-shuffled onto this partitioning (see
+ * `KeyedShuffleSpec.createPartitioning`), so
co-location holds
+ * for whole keys only: two marked
partitionings declaring the
+ * same keys in the same order and using the
same partition
+ * function per position still pair (equal
undeclared keys hash
+ * to the same partition), but a row of an
undeclared key sits in
+ * the partition of some other declared key,
away from rows
+ * sharing a subset of its columns.
`satisfies` and
+ * `KeyedShuffleSpec.areKeysCompatible`
therefore accept a marked
+ * partitioning only for full-key clustering,
never for a subset
+ * of its partition columns and never for a
global ordering
+ * across several partitions. Two carry rules:
(1) a node that
+ * changes the declared key set must drop the
keyed partitioning,
+ * whether it coarsens it (key-dropping
projection, reducer,
+ * join-key projection) or expands it over a
marked leg (a union,
+ * where another leg may declare exactly the
key that leg holds
+ * out-of-set); (2) a marked member beside an
unmarked sibling is
Review Comment:
**Finding 24.** Rule (2) lost its verb in the reorder: "(2) a marked member
beside an unmarked sibling is agreement is enforced by
`PartitioningCollection`".
The clause after it also says the opposite of your reply on
[r3915661530](https://github.com/apache/spark/pull/58339#discussion_r3915661530).
There you wrote that the clearing keeps the OR from spreading a spurious
marker. Here it reads "clears the markers first -- precision, so the OR spreads
a spurious marker onto the accurate side instead of a genuine one", which
parses as clearing *causing* the spread.
```scala
* out-of-set); (2) marker agreement is
enforced by
* `PartitioningCollection`: the constructor
requires it and
* `fromPartitionings` normalizes by OR, so
members are
* uniformly marked or unmarked and
consumers may read one
* member. `ShuffledJoin`'s `InnerLike` arm,
the only site that
* meets a marked input with an unmarked
one, clears the markers
* first. That is precision, not the
guarantee: without it the
* OR would spread the spurious marker onto
the accurate side.
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -78,6 +78,18 @@ case class GroupPartitionsExec(
// data types match the reduced partition keys for the
identity-vs-transform and
// single-side-transform reducers; for the both-sides-reduce shape no
single transform
// describes the keys (see `KeyedShuffleSpec.reducersBothWays`).
+ //
+ // A marked claim pins undeclared rows to hash(key) % numPartitions
(see
+ // `KeyedPartitioning.mayContainUnknownPartitionKeys`). Only an
identity grouping keeps
+ // that relationship: any other grouping -- a reorder, a coalesce, a
resize, or the
+ // collapse a reduction applies -- moves those rows. Clearing only the
marker would
+ // misreport the undeclared rows that remain, so give up the keyed
partitioning at the
+ // physical output count (one per group, padding included) that a
parent's
+ // `PartitioningCollection` requires for uniformity.
`identityGrouping` is a lazy val, so
+ // repeated `outputPartitioning` calls scan it at most once.
+ if (PartitioningCollection.keyedMarkerOf(p).contains(true) &&
!identityGrouping) {
Review Comment:
**Finding 22.** The give-up runs after `EnsureRequirements` has already
committed to the pairing, so the join it was planned for ends up with a child
that no longer satisfies its required distribution.
`checkKeyGroupCompatible` decides `isCompatible` from `areKeysCompatible`,
inserts the two `GroupPartitionsExec`s, and returns `Some(Seq(newLeft,
newRight))`. That sets `areChildrenCompatible`, and
`EnsureRequirements.scala:251` then leaves both children alone. Nothing re-asks
them afterwards. On your own new test the second join's left child reports
`UnknownPartitioning(4)` while the `SortMergeJoinExec` above it requires
`ClusteredDistribution(id)`.
Measured in a review worktree at this head, by printing inside the AQE-off
arm of `SPARK-59050: SPJ: regrouping a marked layout must not keep the
unknown-keyed claim`:
```
gpe outputPartitionings = UnknownPartitioning, KeyedPartitioning
ValidateRequirements.validate = false
```
Narrowing the clause back to `reducers.isDefined &&
PartitioningCollection.keyedMarkerOf(p).contains(true) && !identityGrouping`
and re-running the same print gives `KeyedPartitioning, KeyedPartitioning` and
`validate = true`. So the flip is this clause's.
`ValidateRequirements` is not decoration. It is AQE's veto on every
`AQEShuffleReadRule` (`AdaptiveSparkPlanExec.scala:209` reverts the rule when
it returns false) and on `OptimizeSkewedJoin` (`OptimizeSkewedJoin.scala:269`).
A stage holding such a join can lose coalescing, local read and skew handling
with only a `logDebug` to show it.
I could not exhibit that loss, so I am not resting the severity on it. This
query's three exchanges are all keyed, so `CoalesceShufflePartitions` has
nothing to work on and both arms print `shuffleReads=0`. I also wrapped the
query in a hash-shuffled join to give the top stage coalescable inputs. There
AQE did coalesce (`shuffleReads=2 coalesced=2`) and `validate` came back true,
because the give-up did not survive into that plan shape. The experiment that
would settle it is a plan carrying both the give-up and a coalescable hash
exchange in one stage.
The minimum ask is to say it here: that the node deliberately under-reports,
and that a plan containing it does not pass `ValidateRequirements`. As written
the comment reads as if the give-up were free.
I am not asking for more than that in this PR. Every other coarsening site
here closes at the producer, and this is the only consumer-side gate left, but
moving it needs `EnsureRequirements` to know the permutation that
`alignToExpectedKeys` derives, which is a refactor rather than a bugfix. If you
would rather close it now, `checkKeyGroupCompatible` already holds both new
children and both distributions, so it can decline the pairing it just built:
```scala
newLeft = applyGroupPartitions(left, ...)
newRight = applyGroupPartitions(right, ...)
// A marked side whose regrouping is not the identity gives up its
keyed partitioning, so
// the pairing above no longer holds. Fall back to the shuffle rather
than leave the join
// with a child that does not satisfy its distribution.
if
(!newLeft.outputPartitioning.satisfies(requiredChildDistribution.head) ||
!newRight.outputPartitioning.satisfies(requiredChildDistribution(1))) {
return None
}
```
That is not free either: it sends the marked side back to the ordinary
shuffle path, which changes the exchange count your new test pins. Your call
which trade you want. I have not measured the new count.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
