HyukjinKwon commented on code in PR #59317:
URL: https://github.com/apache/spark/pull/59317#discussion_r4280081486
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -667,12 +667,22 @@ 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` still reads as
Review Comment:
**Nit (P3):** This says an over-counted key is emitted too many times "while
`isGrouped` still reads as grouped", but two lines up `if (numSplits > 1)
isGrouped = false` clears the flag as soon as any expected count exceeds 1. So
an overflow is reported as grouped only when every count is 1. The previous
revision (aed95ec) had the accurate "can still read as grouped". The closing
sentence also lost its verb, and "committed-gate shuffle fallback" is a new
coinage: it reads as if the fallback happens when the gate commits, whereas it
happens when the `committed` gate in `EnsureRequirements` refuses the pair.
Something like:
```scala
// `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. No producer does this today: the join split
reduces its keys into this
// node's space before counting them (`EnsureRequirements`), and the
ordering split counts
// the same keys it groups, so every count matches its splits. The
assert surfaces any
// future miscount here instead of as a shuffle fallback at the
`committed` gate in
// `EnsureRequirements` or as a bad grouped layout.
```
**Verification:**
- **Inspection:** The guard comment's `isGrouped` clause is conditional, its
closing sentence names the `committed` gate in `EnsureRequirements`, and the
comment lines stay ASCII and within 100 characters.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala:
##########
@@ -74,6 +74,37 @@ class GroupPartitionsExecSuite extends SharedSparkSession {
}
}
+ test("SPARK-59234: a distributed key with more splits than its expected
count fails fast") {
+ // Keys [(1,1), (1,2)] projected onto position 0 both become key 1, so the
child holds two
+ // splits for it. Declaring an expected count of 1 for key 1 while the
node is ungrouped for a
+ // join split is the inconsistency this guard catches: `padTo` would emit
both splits while the
+ // count says one, leaving duplicate keys under a grouped report.
+ val childKp = KeyedPartitioning(Seq(exprA, exprB), Seq(row(1, 1), row(1,
2)))
+ def keyOf(a: Int): InternalRowComparableWrapper =
+ InternalRowComparableWrapper(row(a), Seq(exprA))
+ val e = intercept[AssertionError] {
+ GroupPartitionsExec(DummySparkPlan(outputPartitioning = childKp),
+ joinKeyPositions = Some(Seq(0)),
+ expectedPartitionKeys = Some(Seq(keyOf(1) -> 1)),
+ ungroupingOrigin = Some(SPLIT_FOR_JOIN))
+ }
+ assert(e.getMessage.contains("more splits"))
+ }
+
+ test("SPARK-59234: the replicate branch emits numSplits entries and does not
overflow") {
+ // The same over-count input as the distribute case, but ungrouped for a
replicate rather than a
+ // split. REPLICATED_FOR_JOIN takes the else branch, which emits exactly
numSplits entries per
+ // key whatever the split count, so it cannot overflow and builds without
the guard firing.
+ val childKp = KeyedPartitioning(Seq(exprA, exprB), Seq(row(1, 1), row(1,
2)))
+ def keyOf(a: Int): InternalRowComparableWrapper =
+ InternalRowComparableWrapper(row(a), Seq(exprA))
+ val gpe = GroupPartitionsExec(DummySparkPlan(outputPartitioning = childKp),
+ joinKeyPositions = Some(Seq(0)),
+ expectedPartitionKeys = Some(Seq(keyOf(1) -> 1)),
+ ungroupingOrigin = Some(REPLICATED_FOR_JOIN))
+ assert(gpe.groupedPartitions.size === 1, "the one expected partition
replicates both splits")
Review Comment:
**Nit (P3):** This checks only the slot count, but the message says the one
slot replicates both splits, and that is the property that matters here: the
replicated side has to keep the whole key group, or partial clustering loses
join rows. A later "symmetric" guard that truncated the group (e.g.
`splits.take(numSplits)`) would keep `groupedPartitions.size === 1` and pass.
Asserting the indices pins it:
```suggestion
assert(gpe.groupedPartitions.map(_._2) === Seq(Seq(0, 1)),
"the one expected partition replicates both splits")
```
**Verification:**
- **Regression:** A replicate-branch change that drops a split for an
over-counted key (for example, truncating the group to `numSplits` entries)
makes the fixture fail.
- **Behavior:** With the current implementation, a REPLICATED_FOR_JOIN node
over two splits under one key with an expected count of 1 builds and emits one
slot holding both splits.
--
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]
