fix: preserve native sliding integer sum overflow semantics

[rich7420]

Which issue does this PR close?

Closes #6043.

Rationale for this change

Native sliding integer sums currently wrap on overflow. Spark's ANSI sum throws and try_sum returns NULL, with recovery once the overflowing values leave the frame. This change preserves native window execution while matching those semantics.

What changes are included in this PR?

Add a retractable integer SUM accumulator for ANSI/TRY sliding windows. Two monotonic queues track the smallest and largest exact prefix sums, detecting intermediate overflow without rescanning each frame. Evaluate overflow after update and retraction so temporary overlap between frames cannot cause a false error. Account for queue capacity in accumulator memory usage.

Disable reversal for ANSI/TRY sums because checked addition depends on input order. Legacy sliding sums retain the existing DataFusion path. Grouped and ever-expanding sums retain their existing accumulators.

Replace the proposed Spark fallback with native execution assertions. Extend SQL and Scala coverage for overflow, cancellation, recovery, NULL/empty frames, batch boundaries and RANGE peers. Add a benchmark that consumes the window results and verifies both execution paths.

How are these changes tested?

Candidate 3e8625af8 passed fork CI, including all Comet suites, Rust tests, strict Spark 3.5/JDK17 compilation, benchmark checks, TPC-H/TPC-DS and all nine Spark 4.1 SQL shards.

Local verification after merging main 481e6bce2:

• Spark 4.1.3/JDK21: full Window suite and expanded SQL fixture, 70 tests passed.
• Spark 3.5.9/JDK21 with strict warnings: focused window and SQL tests, 4 passed.
• Release Rust SUM tests: 10 passed, including all 7,776 five-value sequences drawn from NULL, 0, +/-1 and i64 bounds, at four frame widths and three evaluation modes.
• Release build, all-target workspace Clippy with -D warnings, formatting and license checks passed.

Local benchmark: M3 Pro, Spark 4.1.3/JDK21, local[1], release native library, 65,536 rows. SUM and TRY_SUM, frame widths 16/1024 and NULL fractions 0/50/100%. All 12 cases compare consumed results and verify the intended window operator. Native mean query times were 27–33 ms versus 36–990 ms for Spark-window fallback, a 1.29–31.94x speedup in this matrix. These are microbenchmarks, not production-workload guarantees.

Upstream head ce6e3a90d merges main c7aa38bd0. Its tree matches the conflict-free merge preview and retains a six-file change. The SUM accumulator, SQL/Scala regressions and benchmark are unchanged from the verified fork candidate 3e8625af8; the local/fork runtime results above apply to that candidate.

This PR is ready for review of the native implementation. Current-head upstream validation is incomplete: the execution job was cancelled during CometJoinSuite, after the sliding integral ROWS tests passed in both ANSI modes. No ScalaTest failure was recorded before cancellation, but this is not a complete execution-suite pass. Required Checks failed because of the cancellation. A maintainer has been asked to rerun CI and apply run-spark-4.1-tests for current-head Spark SQL coverage. This account cannot rerun the upstream workflow or add labels. The incomplete checks still need to pass before merging.

[sunchao]

Correctness

Reviewed 46ef424d0261f73ea943af7160ca410c77c9854b against 6065705c16340c0be293212a71decfd9df4daae4. No verified P1/P2 findings.

The prior sliding path uses DataFusion's integer sum accumulator, which wraps on overflow without honoring the aggregate's ANSI/TRY mode. For a one-row-preceding frame over [Long.MaxValue, 1, -1], Spark's try_sum returns [Long.MaxValue, NULL, 0]; the wrapping path produces Long.MinValue at the second row. Spark recomputes changing frames, so the result can recover after the overflowing values leave.

The new admission guard correctly combines a lower bound other than UNBOUNDED PRECEDING, a LongType sum result, and a non-legacy aggregate evaluation mode. Spark promotes byte, short, int and long sums to long. Reading the mode through the existing shim also catches try_sum when session ANSI is disabled. Returning None follows the existing whole-WindowExec fallback contract, including shrinking, singleton and empty frames. Legacy integral sums, floating-point sums, and expanding integral sums retain their existing native paths; expanding sums use Comet's mode-aware accumulator.

Validation

• Compared Spark's sum state and moving/growing/shrinking frame evaluation on the maintained 3.5 and 4.0 branches. Checked all four Comet version shims and the checksum-verified DataFusion 55.1.0 sliding accumulator.
• The current expressions CI job actually ran windows/sliding_integer_sum.sql. Its nine fallback queries check the expected reason and Spark-equivalent answers; five ordinary queries assert native coverage; three error queries compare overflow failures. Coverage includes both overflow directions, recovery, null/empty frames, ROWS/RANGE, narrow integral inputs and native controls.
• The current execution CI job passed the affected window tests. Both jobs checked out merge 908f9931454e, whose tree equals the reviewed head, and consumed the native artifact built by the same run with matching upload/download SHA-256 digests.
• Local validation comprised source assertions and an independent signed-64 arithmetic model. The required fallback reason is absent from the base implementation, establishing the fixture's pre-fix failure at the source-contract level; I did not run the pre-fix Spark/JNI implementation. Current runtime evidence is Spark 4.1 CI. Canonical maintained 3.4 and 4.1 source branches were unavailable, so source-semantic coverage is limited to 3.5/4.0. The expressions job's one canceled test is an unrelated Spark-4.0-only regexp reproducer.

Performance

The extra work is a small planning-time type/mode check, with no per-row allocation or arithmetic added. Affected windows now execute in Spark, and one rejected expression causes the whole window operator to fall back. That can increase query time and JVM buffering/recomputation costs, including the quadratic shrinking-frame case, but it is the explicit cost of preventing incorrect overflow results. The guard is independent of input values and therefore also applies when a particular dataset would not overflow. Native legacy and expanding controls keep the unaffected cases covered. No performance measurement or speedup claim is attached to this change.

Design

The fix fits the existing sliding-decimal fallback and uses the same frame classification as native accumulator selection. Keeping the decision in window admission avoids changing grouped sums or pretending that a sticky overflow flag can support retraction. A native alternative would need to reproduce Spark's overflow and recovery semantics as frame contents change, which is a larger change than this compatibility guard. The fallback reason and compatibility documentation explain the resulting boundary.

Abstraction & complexity

The implementation reuses the existing evaluation-mode shim and fallback machinery. It introduces no new runtime abstraction or ownership path. The small test helper adapts existing plain-sum assertions to the session's ANSI mode, while the SQL fixture separately tests explicit TRY mode with ANSI both disabled and enabled. The focused production change, regression fixture and documentation form a coherent scope; I found no actionable simplification.

[andygrove]

Thanks for this. I traced the premise through the native side and it checks out. process_agg_func in native/core/src/execution/planner.rs only builds the mode-aware SumInteger UDAF when is_ever_expanding is true, and sliding frames fall through to DataFusion's built-in sum, which wraps and has no notion of eval mode. So the divergence was real, and the admission guard is at the right layer, matching the decimal guard from #4732.

A few things I checked that look fine, noted here so they do not get re-litigated. s.dataType == LongType is the right predicate, because Spark promotes byte, short, int and long sums to LongType, and interval sums are already rejected earlier by AggSerde.sumDataTypeSupported. Shim coverage is not a concern either, since sumEvalMode already exists across the 3.4, 3.5, 4.0 and 4.1+ shims. I also suspected a matching gap in integral avg, but Spark's Average accumulates integral input in DoubleType, so there is no ANSI overflow there and nothing to fix.

The fixture is thorough on the behaviour it covers, including both overflow directions, recovery, null and empty frames, and native controls on the paths that stay native. Both of my comments are about the test harness rather than the fix itself.

[andygrove]

Thanks for the comment explaining why this is here, it made the intent easy to follow. One concern though. CometSqlFileTestSuite excludes ConstantFolding on purpose, and appends that exclusion after each file's own configs so that a header cannot override it. Clearing it with a SET statement turns folding back on for every query in the rest of the file, including the plain query entries that assert native coverage. This is the only fixture in the repo that does this.

window_functions.sql lines 198 to 203 hit the same ROWS ... N PRECEDING problem and took the other route, keeping SQL fixtures to bounds that parse directly and covering N PRECEDING in CometWindowExecSuite through the DataFrame API. Could we follow that convention here? The RANGE ... 1 PRECEDING cases already work unfolded, as window_functions.sql line 226 shows, so only the ROWS ones would need to move and the SET could go away entirely.

If you think the fixture really does need folding, could we add a supported file-level directive for it instead, so the harness stays in control of what it guarantees?

[andygrove]

I follow why this is needed. The default profile is Spark 4.1.3 where ANSI is on, so without this the sliding sums in this matrix would now fall back and checkSparkAnswerAndOperator would fail its native coverage assertion.

The side effect is that this matrix stops exercising ANSI at all, and it covers a good deal more than sums. Would routing the sum assertions through checkSlidingIntegralSum work here, the way the three tests below do, so the matrix keeps running in both modes? If pinning really is the cleaner option, could the comment mention that the matrix no longer covers ANSI? As written it reads as being only about legacy sums, and the next person may not realise the broader coverage went with it.

[andygrove]

Is spark.comet.operator.WindowExec.allowIncompatible=false doing anything here? CometWindowExec does not declare a support level, and the only other fixture using that key, lag_lead.sql, sets it to true. If this is just documenting the default, it may be worth dropping so that it does not get copied into future fixtures as boilerplate.

[sunchao]

Rechecked the unchanged 46ef424d after the new harness feedback. I agree with the existing comments: the fixture-wide SET clears the harness’s folding exclusion, the window matrix now forces legacy mode for non-sum cases too, and WindowExec.allowIncompatible has no effect on this conversion guard. Those threads cover the needed test changes. I found no additional P1/P2 to raise.

One coverage qualification: the matrix previously inherited the session’s ANSI mode. It was not an explicit two-mode matrix. The new fixture still tests ANSI/TRY fallback and errors, and its table-column sums are not folded away. Its five ordinary queries retain native-plan assertions.

Fresh log checks confirm that the SQL fixture and window suite passed on CI merge 908f9931, whose parents are the reviewed base/head and whose tree equals this head. This follow-up used source checks, with no new local Spark/JNI run. Maintained Spark 3.5/4.0 semantics were rechecked; the 3.4/4.1 source gaps remain. My existing approval is preserved.

[rich7420]

cc @andygrove , @sunchao could you take another look? thanks!

[sunchao]

LGTM

[sunchao]

@rich7420 actually instead of fallback to Spark which would sacrifice performance, should we consider to fix this behavior in Comet native path instead?

[rich7420]

Updated this PR with the native sliding accumulator from 3e8625af8. Current head ce6e3a90d merges main without changes to the SUM implementation or its regressions and benchmark. It checks ordered prefix extrema after retraction, so it detects intermediate overflow and recovers when the offending values leave the frame. ANSI/TRY sums cannot be reversed because that changes overflow behavior.

Fork CI passed, including all nine Spark 4.1 SQL shards. Local Spark 4.1 Window/SQL tests (70), focused Spark 3.5 tests (4) and Rust SUM tests (10) also passed. The 12-case local benchmark was faster than Spark-window fallback in every case, at 1.29–31.94x. The PR description records the workload and limits.

This PR is now ready for a fresh review of the native implementation. Upstream run 36725911344 passed the Rust tests, builds, strict compilation, scans/shuffle/expressions and TPC-H/TPC-DS checks. Its execution job was cancelled during CometJoinSuite, after the sliding integral ROWS tests passed with ANSI both enabled and disabled. The log records no ScalaTest failure before cancellation, but the execution-suite verdict is incomplete and Required Checks is red.

Could a maintainer rerun the cancelled execution job with its dependent checks, and apply run-spark-4.1-tests for the updated upstream merge? This account cannot rerun the upstream workflow or add labels.