ulysses-you commented on PR #59317: URL: https://github.com/apache/spark/pull/59317#issuecomment-6166195213
Thank you for the patch, and for laying the fail-fast versus derive-isGrouped choice out explicitly. I dug into whether the guard can be reached today, on master and on this branch, and it can: the assert is not dead code. **Blocking: the precondition is not an invariant once a reducer is in play** The expected counts are read from `projectedOriginalPartitionKeys.groupBy(identity)` (`EnsureRequirements.scala:898-903`), the spread side's keys projected but **not reduced**, while `mergedPartitionKeys` (`:799-802`) and the node's grouping (`GroupPartitionsExec.scala:706-715`) are in the reduced key space. A reducer that merges key values makes the count for a reduced key be read off an unrelated raw key, so the overflow gets produced rather than merely being possible. Reproduction, end to end: two tables of four rows partitioned by a transform whose function reduces *itself* (legal under `ReducibleFunction`'s general `r1(f1(x)) = r2(f2(x))` form, which is vacuous when `f1 == f2`), with `spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled=true` and a join on the key column. - master: all four rows returned correctly; 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`; no `GroupPartitionsExec` survives; both sides are `Exchange hashpartitioning(id, 5), ENSURE_REQUIREMENTS`. A silent shuffle. - this branch: planning fails with `AssertionError: a distributed partition key holds more splits (2) than its expected count (1)`, surfaced to users as `SparkException: [INTERNAL_ERROR] The Spark SQL phase planning failed with an internal error ... SQLSTATE: XX000`, asking them to report a Spark bug. Same with AQE on and off, so it is not an AQE artifact. The numbers in between pin the mechanism: `After merging, there are 2 partitions` (four keys reduced to two), then `After applying partially clustered distribution, there are 2 partitions.` (the counts are still 1 + 1). So the spread side groups two splits under a key whose count says one, `padTo` emits both, the two sides' key sequences diverge, and the `committed` gate (`EnsureRequirements.scala:958-976`) drops the pairing. Why `canReduceKeys` does not cover this: it is consulted only inside `isExpressionCompatible` (`partitioning.scala:2308`, `:2311`). A same-function pair is admitted through the `else left.isSameFunction(right)` arm at `:2309`, which never asks, and `reducersBothWays` (`:2388`) is called unconditionally at `EnsureRequirements.scala:774`, so both sides reduce (the both-sides branch at `partitioning.scala:2415-2421`). No in-tree function reaches it (`BucketFunction.reducer` returns null when `gcd == thisNumBuckets`, `YearsFunction.reducer` is always null, `DaysFunction.reducer` answers only for `YearsFunction`, `FlipLowBitFunction` is not reducible), which is why CI is green; a connector whose `reducer` answers for its own function gets here. The construction that makes the precondition hold is in the same file: reduce the spread side's keys with the same reducers before the `groupBy`. The counts then live in the same space as the node's key map, the aligner emits one entry per slot on both sides, the pairing commits (the join is kept instead of silently lost), and the overflow disappears by construction, which is what would make this assert provably unreachable. As written, "nothing hits this today" needs a condition attached: for that input the guard changes a correct-but-shuffled plan into a failed query. **Nit: the ordering clause describes a state its producer cannot reach** (`GroupPartitionsExec.scala:673-674`) The `SPLIT_FOR_ORDERING` producer takes its expected counts from the child's own key list (`EnsureRequirements.scala:153-155` groups the same `partitionKeys` the node groups at `GroupPartitionsExec.scala:706-715`), so it has no independent count to get wrong and cannot report duplicate keys as grouped. The clause reads as a live consequence but cannot be reproduced; the join clause is accurate. **Nit: the replicate branch's asymmetry is neither stated nor pinned** (`GroupPartitionsExec.scala:686`) `Seq.fill(numSplits)((key, splits))` emits exactly `numSplits` entries per key however many splits it holds, so no guard is needed there. That is right, but nothing says so and the new test only covers the throwing arm. One clause in the comment plus a non-throwing fixture (same child and expected keys, `ungroupingOrigin = Some(REPLICATED_FOR_JOIN)`) separates deliberately omitted from 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]
