ulysses-you commented on code in PR #59317:
URL: https://github.com/apache/spark/pull/59317#discussion_r4269575803
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -667,6 +667,14 @@ private[sql] object GroupPartitionsExec {
val splits = keyMap.getOrElse(key, Seq.empty)
numMatchedPartitions += splits.size
if (distributePartitions) {
+ // `padTo` only pads, never truncates, so a key holding more splits
than its expected
+ // count would be emitted more times than the count says, while
`isGrouped` can still read
+ // as grouped. A join split would then fail the `committed` gate in
`EnsureRequirements`
+ // and silently fall back to a shuffle, and an ordering split would
report duplicate keys
+ // as grouped. Fail fast here, where the alignment is owned, instead
of either.
+ assert(splits.size <= numSplits,
Review Comment:
The precondition here needs a condition attached. The counts compared
against come from `EnsureRequirements.scala:898-903`, which groups the spread
side's keys projected but not reduced, while `mergedPartitionKeys` (`:799-802`)
and this node's grouping (`:706-715`) are in the reduced key space. A reducer
that merges key values is reachable with partial clustering on: a same-function
pair is admitted at `partitioning.scala:2309` without consulting
`canReduceKeys`, and `reducersBothWays` runs unconditionally at
`EnsureRequirements.scala:774`. Then a reduced key's count is read off an
unrelated raw key.
Two tables of four rows partitioned by a self-reducing transform reproduce
it. On master the query returns the right rows after a silent shuffle (the
planner logs `Not storage-partitioning this join after all: the two sides no
longer declare the same partition keys once regrouped onto the merged partition
values`, and both sides get `Exchange hashpartitioning(id, 5),
ENSURE_REQUIREMENTS` with no `GroupPartitionsExec` left); with this patch it
fails at planning with `more splits (2) than its expected count (1)`.
Reducing the spread side's keys with the same reducers before that `groupBy`
puts the counts in this node's space, keeps the pairing, and makes the
precondition hold by construction. Otherwise the comment should say what the
guard trades: for that input, a correct-but-shuffled plan becomes a failed
query.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -667,6 +667,14 @@ private[sql] object GroupPartitionsExec {
val splits = keyMap.getOrElse(key, Seq.empty)
numMatchedPartitions += splits.size
if (distributePartitions) {
+ // `padTo` only pads, never truncates, so a key holding more splits
than its expected
+ // count would be emitted more times than the count says, while
`isGrouped` can still read
+ // as grouped. A join split would then fail the `committed` gate in
`EnsureRequirements`
+ // and silently fall back to a shuffle, and an ordering split would
report duplicate keys
Review Comment:
The `ordering split` half of this sentence describes a state its producer
cannot reach. `EnsureRequirements.scala:153-155` derives the expected counts by
grouping the same `partitionKeys` this node groups at `:706-715`, so there is
no independent count a repeated key could survive, and a repeated key already
sets `isGrouped` false at `:666`. The join half is accurate; suggest dropping
the ordering half or stating the two producers' counts directly.
--
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]
