HyukjinKwon commented on code in PR #59317:
URL: https://github.com/apache/spark/pull/59317#discussion_r4237482226
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -667,6 +667,13 @@ 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 emit more partitions than the pairing side expects
while `isGrouped` can
+ // still read as grouped. Fail fast here, where the alignment is
owned, rather than let
+ // the inconsistency surface downstream in
`PartitioningCollection.fromPartitionings`.
Review Comment:
**Nit (P3):** This comment says the assert stops the inconsistency from
surfacing "downstream in `PartitioningCollection.fromPartitionings`", but on
the current planner that isn't what an overflow would do. The assert runs
inside `GroupPartitionsExec.apply`, while `EnsureRequirements` builds the
pushed pair, and only after that does the `committed` gate compare
`declaredLayout(newLeft).describesSameKeys(declaredLayout(newRight))`
(`EnsureRequirements.scala` ~960). A `SPLIT_FOR_JOIN` side that overflowed
would emit `[k, k]` against the replicated side's `[k]`, so the gate would come
back false and the join would fall back to `shuffleToCoPartition` before any
join `outputPartitioning` (and so `fromPartitionings`) is built. Also,
`SPLIT_FOR_ORDERING` "pairs with nothing", so "the pairing side" doesn't apply
to that producer.
So on the join path, the guard swaps a silent shuffle fallback (correct
results, no SPJ) for a planning-time `AssertionError`. That's a reasonable
choice, and it's the trade-off you're asking @peter-toth and @ulysses-you to
weigh, but the comment presents it as just moving an error that would happen
anyway. Something like:
```suggestion
// `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.
```
**Verification:**
- **Inspection:** The guard comment describes the committed-gate shuffle
fallback on the join path and the grouped-with-duplicate-keys report on the
ordering path, and contains no claim of a downstream
PartitioningCollection.fromPartitionings failure.
- **Behavior:** A distribute-mode node whose key holds more splits than its
expected count still fails at construction with the same 'more splits'
assertion, and valid-count distribute alignments are unchanged.
--
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]
