ulysses-you commented on code in PR #59317:
URL: https://github.com/apache/spark/pull/59317#discussion_r4269577706
##########
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,
+ s"a distributed partition key holds more splits (${splits.size})
than its expected " +
+ s"count ($numSplits)")
// Distribute splits across expected partitions, padding with empty
sequences
Review Comment:
The replicate arm just below (line 686) is worth a clause in the comment
above it: it emits exactly `numSplits` entries per key however many splits it
holds, which is why the new guard is branch-local. The new test only covers the
throwing arm, so a non-throwing fixture with the same child and expected keys
and `ungroupingOrigin = Some(REPLICATED_FOR_JOIN)` would pin that the omission
is deliberate rather than overlooked.
--
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]
