ulysses-you commented on PR #59764: URL: https://github.com/apache/spark/pull/59764#issuecomment-6407574469
Thanks @cloud-fan for the followup. I reviewed head `7339014aee5` against base `cecbd5d8` independently: traced both execution paths at the new threshold, ran targeted probes at head, and ran the suite. No correctness issue: the change is mechanically sound on both paths and I could not construct a wrong-results, data-loss or OOM case. One non-blocking test issue below. ## What I checked **Threshold split, both paths.** The periodic check keeps `minCompaction`; the spill check uses `max(minCompaction, minCompactionToSpill)` (`HashAggregateExec.scala:218-219`, `:1043-1044`, `:1126`; `TungstenAggregationIterator.scala:233-234`, `:246-247`, `:278`). Both paths evaluate the same row set at the boundary (the failed-insert row is in neither the numerator nor the denominator), so codegen and interpreted stay in lockstep. The `minRows` floor and the reset-on-spill counters are untouched by the diff. **Pass-through invariant under the larger trigger set.** This is where the change has reach, so I wrote an order-sensitive probe: `first`/`last` over a key that holds ids 8 and 10 in the frozen map while id 16 is the failed insert and repeats it, `minCompaction = 1.0` to keep the periodic check inert, forced cap 16, four arms (codegen on/off x fused / Exchange-split): ``` PROBE1 derivedKey=false wholeStage=true refSkipped=0 prSkipped=8 lowSkipped=0 ref=[-1,8,16]|... pr=[-1,8,16]|... PROBE1 derivedKey=false wholeStage=false refSkipped=0 prSkipped=8 lowSkipped=0 ... PROBE1 derivedKey=true wholeStage=true refSkipped=0 prSkipped=8 lowSkipped=0 ... PROBE1 derivedKey=true wholeStage=false refSkipped=0 prSkipped=8 lowSkipped=0 ... ``` All four arms return `first=8, last=16`, equal to a non-spilled reference. `lowSkipped = 0` is the control: at `minCompactionToSpill = 1.05` the same input spills (16 rows / 15 keys = 1.07) instead of flipping, so the 8 pass-through rows provably come from the spill check. The freeze/queue ordering holds at a spill-check flip in both plan shapes; in the Exchange-split arms the map drain is interrupted by `shouldStop()` once per row and resumed across re-entries, which is the harder path. **Config and plumbing.** The new config validates like `minCompaction` (finite, `>= 1.0`) on both keys, the `max()` direction matches both docs pages, version 4.4.0 matches the parent, and `NOT_APPLICABLE` matches its siblings. No other registry needs the entry. No non-ASCII and no added line over 100 characters. No stale `numBypassingRows` reference remains anywhere. **Metric rename.** Only `HashAggregateExec`, `TungstenAggregationIterator` and the suite use it. The metric is registered only when the feature applies, so feature-off plans are unaffected and the "no user-facing change" claim holds. **`aggTime`.** I re-checked the round-1 items. Item 11 is not conditional: `BufferedRowIterator.shouldStop()` is `!currentRows.isEmpty()` and the aggregate appends nothing before the first pass-through row, so the first `doAgg` call always covers `finishHashMap`, and the double count fires on every codegen task that passes through. Item 2 holds as stated: fused, nothing is appended, so the whole build, pass-through rows included, sits inside the single timed `doAgg`. For items 1 and 3, bracketing only `nextPassThroughOutput()` and the resume call, then converting the summed nanos once at the end, recovers the measurement without counting downstream work. **Cost of the 2.0 default.** At a 1.5x compaction ratio (200k rows, 133,333 keys, forced full map): `partialOutputRows = 193,333` with 0 fallbacks at 2.0, versus `133,333` with 1 fallback at 1.05. That is 1.45x more rows to the shuffle and no spill, the trade the description describes, and it is defensible for the uniform shape. The shapes where it could invert are the skewed / repeating-key ones (items 12/13), and there is no benchmark case in the 1.05-2.0 band (item 17). That case is the one measurement I would still want. **Suite.** `build/sbt 'sql/testOnly *AdaptivePartialAggregationSuite'` at `7339014aee5`: 57 succeeded, 0 failed. ## The one issue `AdaptivePartialAggregationSuite.scala:891` (`forceSpill` loop at `:897-902`). The `forceSpill = true` cells no longer pass a single row through. `MIN_ROWS` is 32 while the forced cap is 16, so every map epoch spills at 16 rows and resets `processedRows`. The spill check requires `processedRows >= max(minRows, 1) = 32` (`HashAggregateExec.scala:1036`, `:1043`; `TungstenAggregationIterator.scala:230`, `:246`), and the periodic check requires `processedRows == 32`, so neither can fire. Measured on the exact query and configs, reading the partial aggregate's `numPassthroughRows` and `numTasksFallBacked`: ``` splits=1 wholeStage=true twoLevelMap=true passthrough=0 fallbacks=1 splits=1 wholeStage=true twoLevelMap=false passthrough=0 fallbacks=1 splits=1 wholeStage=false twoLevelMap=true passthrough=0 fallbacks=1 splits=1 wholeStage=false twoLevelMap=false passthrough=0 fallbacks=1 splits=2 wholeStage=true twoLevelMap=true passthrough=0 fallbacks=2 splits=2 wholeStage=true twoLevelMap=false passthrough=0 fallbacks=2 splits=2 wholeStage=false twoLevelMap=true passthrough=0 fallbacks=2 splits=2 wholeStage=false twoLevelMap=false passthrough=0 fallbacks=2 ``` `checkAnswer(adaptive, reference)` then compares two non-adapting runs, so the cell is vacuous. This was the only order-sensitive fixture whose flip came from the spill check: before `SPARK-58511` (7d16a2ba4af) applied the `minRows` floor, the check fired at 16 rows / 16 keys (ratio 1.0). That makes it material here, because `minCompactionToSpill = 2.0` turns that flip into the common trigger. Fix: raise the forced cap above `MIN_ROWS` (`"4, 64"`) so the periodic check flips at 32 rows on a 32-key map, add a cell whose flip is the spill check (`MIN_ROWS = 16` with cap 16, 16 rows over 16 keys is below 2.0), and assert `numPassthroughRows > 0` so the fixture cannot go vacuous silently again. Pre-existing: introduced by 7d16a2ba4af, not by this PR. -- 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]
