fix: match Spark variance updates for large nearby values

[rich7420]

Which issue does this PR close?

Closes #6044.

Rationale for this change

For nearby DOUBLE values around 1e16, native var_pop returns 2 instead of Spark's 1. Subtracting the rounded mean loses precision and also affects standard deviation.

What changes are included in this PR?

Use Spark's CentralMomentAgg update for variance and standard deviation, while preserving the Pearson update used by corr and regr_r2. Add scalar and grouped regression coverage, including the regression aggregates that share these accumulators.

How are these changes tested?

The new Rust regression fails before the fix. All 128 aggregate Rust tests and 8 targeted Scala/SQL tests pass locally; the regression asserts native Partial and Final execution. Fork CI passes across Spark 3.4–4.2, including the Spark 4.1 SQL suites. Clippy and formatting checks pass.

[sunchao]

Correctness

The prior update computes delta * (value - new_mean). Rounding new_mean for the pair [1e16, 1e16 + 2] makes population variance 2 instead of Spark's 1. This patch uses delta * (delta - delta / new_count) for variance and standard deviation, matching the operation order in Spark's CentralMomentAgg on the checked 3.5 and 4.0 sources.

The distinction between the two formulas is necessary. Scalar and grouped corr, and the two variance children of regr_r2, explicitly select the existing Pearson update. regr_sxx, regr_syy, and the variance used by slope/intercept retain the central-moment default. Null filtering, sample/population denominators, the legacy single-row null/NaN choice, and the three-double intermediate state layout are unchanged. I found no new correctness issue in the authored changes.

The regression exercises positive, reversed, and negative nearby pairs with an intervening null, both scalar and grouped accumulators, and batch boundaries. The Scala test writes one file so the values reach the same partial accumulator, then asserts native Partial and Final execution. The existing merge arithmetic is unchanged and retains its existing floating-point limitations across partition layouts.

Validation

Eight isolated recurrence checks passed, covering the reported pairs, IEEE special values and signed zero, count boundaries, Pearson preservation, batch splitting, and two nonempty partial states. These checks use exact copied helpers with modeled drivers. My direct Spark source comparison covered 3.5/4.0; maintained 3.4/4.1 sources were unavailable.

I verified the CI checkout badd9430 has parents 5ca14992 and 1299da8b, and that all six authored files match the reviewed head. The Rust job passed 1,594 tests, including large_offset_variance. The Spark 4.1 exec job passed 975 tests, including the new Scala regression. Its native-library artifact matches the same merge's build job. Dedicated Spark SQL workflows and macOS were skipped. The successful expression job had one canceled Spark-4.0-only regex test.

Performance

The recurrence keeps one division per non-null value and introduces a selection between two subtraction formulas. The selection is fixed for the lifetime of an accumulator. There are no new row allocations, state columns, input passes, or per-group vectors. The fix removes the accuracy loss in the reported inputs, but the review has no measured query-time or throughput improvement to report. The added mode selection is the only new hot-loop decision, and I found no evidence of a material performance regression.

Design

Choosing the update formula at accumulator construction keeps the numerical contract with the aggregate that owns it. The default follows central moments, while Pearson users opt in explicitly. This preserves the shared state representation and merge path without changing SQL registration, serialization, or version flags.

The test belongs in the Scala aggregate suite because it controls file layout and checks both native aggregation modes. That setup directly targets the failure mechanism. The change is limited to six files, and the supplied base's separate documentation/CI changes integrate without altering the authored contribution.

Abstraction & complexity

VarianceUpdate is a small, local enum that represents a real difference in Spark's algorithms. The sibling-only with_pearson_update methods expose that choice without adding a public configuration or duplicating accumulator implementations. The helper comment explains why the formulas must remain distinct. The abstraction is proportionate to the two update contracts, and I have no additional change requests.

[andygrove]

This closes the update path cleanly. I read CentralMomentAgg and Corr in 3.4.3 and 4.1.3 as well, which fills the gap @sunchao flagged, and both endpoints match the two formulas you picked. The routing lines up with Spark's class hierarchy in every case too: RegrR2 extends PearsonCorrelation, RegrSXX and RegrSYY reduce to RegrReplacement which is a CentralMomentAgg, and RegrSlope and RegrIntercept hold a literal VariancePop. All six construction sites in Comet agree with that.

I think the merge path keeps #6044 open, though. variance_merge computes ma*ca/nc + mb*cb/nc for the mean and d*d*ca*cb/nc for m2, where Spark's mergeExpressions computes avg.left + deltaN*n2 and delta*deltaN*n1*n2. The different operation order costs a ULP on the running mean at the first merge into the empty Final buffer, and the next merge squares it. Your Scala regression writes a single file so every value reaches one partial accumulator, which is why it does not show.

On your head, real Spark 4.1 against real Comet, five doubles clustered at 1e17 split across two partitions with CometHashAggregate in both Partial and Final:

Spark var_pop    = 61.44000000000001
Comet var_pop    = 245.76
Spark stddev_pop = 7.8383671769061705
Comet stddev_pop = 15.67673435381234
exact var_pop    = 61.44

A differential search over the two merge formulas diverges on about 24% of randomly clustered inputs. The single-partition control is bit-identical to Spark after your fix, so this is the half the change does not reach rather than anything wrong with it.

Would you rather fold the merge into this PR, or land this and open a follow-up so #6044 stays open for the merge half? Either is fine by me. If it becomes a follow-up, a two-file version of the Scala regression would keep it honest.