dwsmith1983 opened a new pull request, #6041:
URL: https://github.com/apache/datafusion-comet/pull/6041

   ## Which issue does this PR close?
   
   Closes #6002.
   
   ## Rationale for this change
   
   Comet's native decimal `SUM` validated the running sum against the result 
precision on every update and latched the failure, returning `NULL` in legacy 
mode and raising `ARITHMETIC_OVERFLOW` under ANSI. Spark adds with 
`DecimalAddNoOverflowCheck` and checks only when a value leaves its buffer. For 
an ungrouped aggregate under whole-stage codegen, for an expanding window 
frame, and for `ObjectHashAggregateExec`, that buffer is unbounded, so an 
intermediate that exceeds the precision and later cancels out still produces a 
result. `SUM` over `DECIMAL(38,38)` values `0.6, 0.6, -0.6` returns `0.6` in 
Spark and `NULL` or an error in Comet.
   
   ## What changes are included in this PR?
   
   - `SumDecimalAccumulator`, which serves ungrouped aggregates and expanding 
window frames, keeps the running sum as an unbounded `i256` and no longer 
checks per row. `state()` emits a null sum with `is_empty = false` when the 
partial's sum does not fit the result precision, which is what Spark's 
`UnsafeRow` write does to a partial's output row. `merge_batch` keeps the 
sticky overflow rule and otherwise adds unbounded. `evaluate()` applies the 
check: a latched null raises `ARITHMETIC_OVERFLOW` and a value that does not 
fit raises `NUMERIC_VALUE_OUT_OF_RANGE` under ANSI, matching 
`CheckOverflowInSum` (`toPrecision` fails for a present value); legacy and 
`try_sum` return null.
   - `SumDecimalGroupsAccumulator` is unchanged: Spark's grouped 
`HashAggregateExec` buffers in an `UnsafeRow` that nulls a value which does not 
fit, so latching per row is the matching behaviour. The comments on both 
accumulators say why they differ.
   - `CometObjectHashAggregateExec` declines a grouped decimal `SUM` whose 
result precision is 38, since Spark's object aggregation buffer is unbounded 
there and Comet's grouped accumulator latches. Only precision 38 has no 
headroom above the input. Decimal `AVG` has the same gap and is tracked in 
#5509.
   - A compatibility note: with `spark.sql.codegen.wholeStage=false` Spark's 
ungrouped aggregate buffers in an `UnsafeRow` and latches like the grouped 
path, so Spark returns `NULL` where Comet now returns the recovered value. 
Spark's own `checkAggResultsForDecimalOverflow` runs with codegen on and off 
and accepts either outcome.
   
   Not mirrored: `UnsafeRow.setDecimal` skips the precision check for result 
precision 18 or below, which needs more than 10^10 rows in a group to observe.
   
   ## How are these changes tested?
   
   Rust unit tests in `sum_decimal.rs` (11 in the module, 6 new): recovery 
after an intermediate overflow in legacy, ANSI and try mode; recovery when the 
running sum exceeds `i128`; `state()` nulling a running sum that does not fit; 
the error class at `evaluate()` for a present value versus a latched null, and 
try mode returning null; unbounded merge of fitting partials with stickiness 
for a partial that overflowed; and the window pattern of updating one row and 
evaluating after each without latching. The new tests fail on the previous 
implementation.
   
   Scala tests in `CometAggregateSuite` and `CometWindowExecSuite`, each for 
ANSI on and off with native shuffle: the issue's repro without `GROUP BY` for 
`SUM` and `try_sum`, also with nulls interleaved; the same rows under `GROUP 
BY` still null or raise `ARITHMETIC_OVERFLOW` like Spark; a partial that does 
not fit at emission across two partitions gives `NULL` or 
`ARITHMETIC_OVERFLOW`; two fitting partials whose merged final does not fit 
give `NULL` or `NUMERIC_VALUE_OUT_OF_RANGE` (`.WITH_SUGGESTION` on Spark 4.x), 
checked with `checkSparkError` for error class and SQLSTATE parity; `SUM` with 
`collect_list` in the same `GROUP BY` falls back with the new reason at 
precision 38 and stays native at `DECIMAL(10,2)`; and an expanding window frame 
returns `0.6, NULL, 0.6` for `SUM` in legacy mode and for `try_sum` in both 
modes, and raises `NUMERIC_VALUE_OUT_OF_RANGE` for `SUM` under ANSI.
   
   Run locally on Spark 3.5 with JDK 17: `CometAggregateSuite` and 
`CometWindowExecSuite` (168 succeeded), `CometExecRuleSuite` (31 succeeded), 
the `datafusion-comet-spark-expr` crate (921 tests), `cargo clippy 
--all-targets --workspace -- -D warnings`, spotless and the semantic scalafix 
check.
   


-- 
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]

Reply via email to