You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "rich7420 (via GitHub)" <gi...@apache.org> on 2026/09/21 13:03:44 UTC
[PR] fix: match Spark variance updates for large nearby values [datafusion-comet]
rich7420 opened a new pull request, #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076
## 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](https://github.com/rich7420/datafusion-comet/pull/31/checks) passes across Spark 3.4–4.2, including the Spark 4.1 SQL suites. Clippy and formatting checks pass.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: match Spark statistical aggregate updates and merges [datafusion-comet]
Posted by "kazuyukitanimura (via GitHub)" <gi...@apache.org>.
kazuyukitanimura merged PR #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: match Spark statistical aggregate updates and merges [datafusion-comet]
Posted by "kazuyukitanimura (via GitHub)" <gi...@apache.org>.
kazuyukitanimura commented on PR #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076#issuecomment-6150318714
Thank you @rich7420 @sunchao @andygrove
Merge queued
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: match Spark statistical aggregate updates and merges [datafusion-comet]
Posted by "rich7420 (via GitHub)" <gi...@apache.org>.
rich7420 commented on PR #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076#issuecomment-6035262852
@andygrove the R2 underflow/ANSI fix and exact aggregate regressions are included in current head `37506bfd4`. The benchmark caller now also supplies `EvalMode::Legacy`, and workspace/all-targets Clippy passes. [Current-head CI](https://github.com/apache/datafusion-comet/actions/runs/37474391984) is green; sunchao has re-approved this head. #6481 remains open for the inherited ANSI-on corr behavior. Could you re-review the update and, if the earlier requests are addressed, update your review?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: match Spark statistical aggregate updates and merges [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076#issuecomment-5955529206
I would like to do a deeper review of this PR before it is merged.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: match Spark variance updates for large nearby values [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076#issuecomment-5778787594
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.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: match Spark variance updates for large nearby values [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076#issuecomment-5847801204
Re-reviewed head **`72f12f2` with five independent agents**. **One P2 Spark-compatibility issue remains; I would hold approval.**
**[P2] Apply the corrected merge to empty partials too** — [welford.rs:83](https://github.com/apache/datafusion-comet/blob/72f12f2cb71c7130a3e842b302a369b256a0ba15/native/spark-expr/src/agg_funcs/welford.rs#L83).
With one partial containing **100 copies of `1e155`**, followed by an all-null partial for the same group:
| Implementation | Population variance/covariance |
|---|---:|
| Spark 4.1.3 | `NaN` |
| Previous PR revision | `NaN` |
| Current native accumulators | `0` |
The new arithmetic preserves zero variance during the first merge. Spark then processes the empty partial, where an intermediate multiplication overflows before multiplying by zero, producing `NaN`. Comet’s existing scalar/grouped variance and covariance callers skip that partial.
This is specifically a compatibility regression at extreme finite magnitudes; zero is mathematically reasonable. Apply the canonical merge to zero-count partials too, and add coverage for both partition orders. I independently reproduced the native results using the actual modules; another agent confirmed Spark’s behavior through repeated SQL executions.
Elsewhere:
- **Correctness:** The original nearby-value bug and the previously reported nonempty-merge issue are fixed. Central-moment versus Pearson routing matches Spark.
- **Performance:** No new Spark fallback, buffers, or per-group allocations. Focused kernel measurements found no material regression; no whole-query speedup is established.
- **Design and complexity:** The small update-mode enum represents a necessary numerical distinction. No additional abstraction concerns.
**Validation:** 35 native module tests passed. Separately, 46,212 kernel comparisons against actual Spark Catalyst expressions found no semantic mismatch; the finding above concerns callers skipping the kernel. [CI](https://github.com/apache/datafusion-comet/actions/runs/35847885378) passed 1,603 Rust tests and 989 Spark 4.1 execution tests, including both new regressions.
CI tested a merge commit whose changed native files and new regression block match this head. Dedicated SQL/macOS suites were skipped. Local validation did not include a full Comet/Spark integration run because of disk constraints.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: match Spark statistical aggregate updates and merges [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076#discussion_r4150537612
##########
spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala:
##########
@@ -2343,6 +2343,131 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
}
}
+ test("statistical aggregates with large nearby values") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "1",
+ "spark.sql.files.minPartitionNum" -> "1",
+ CometConf.COMET_SHUFFLE_ENABLED.key -> "true",
+ CometConf.COMET_SHUFFLE_MODE.key -> "native") {
+ for (values <- Seq(Seq(1e16, 1e16 + 2), Seq(1e16 + 2, 1e16), Seq(-1e16, -1e16 - 2))) {
+ // One ordered file keeps both values in the same partial accumulator. Splitting
+ // them across files would only exercise merging two single-row states.
+ withTempPath { path =>
+ (Seq(Some(values.head), None, Some(values.last)))
+ .map(v => (0, v))
+ .toDF("g", "v")
+ .coalesce(1)
+ .write
+ .parquet(path.getCanonicalPath)
+ withParquetTable(path.getCanonicalPath, "large_moments") {
+ for (groupBy <- Seq("", " GROUP BY g")) {
+ val query = "SELECT var_pop(v), var_samp(v), stddev_pop(v), stddev_samp(v) " +
+ "FROM large_moments" + groupBy
+ val (_, cometPlan) = checkSparkAnswerAndOperator(query)
+ val aggregates = cometPlan.collect { case a: CometHashAggregateExec => a }
+ assert(aggregates.exists(_.modes.contains(Partial)))
+ assert(aggregates.exists(_.modes.contains(Final)))
+ checkAnswer(sql(query), Seq(Row(1.0, 2.0, 1.0, math.sqrt(2.0))))
+
+ // CORR and REGR_R2 use PearsonCorrelation's update, while REGR_SXX/SYY
+ // and the variance used by slope/intercept follow CentralMomentAgg.
+ checkSparkAnswerWithTolAndNumOfAggregates(
+ "SELECT corr(v, v), regr_r2(v, v), regr_sxx(v, v), regr_syy(v, v), " +
+ "regr_slope(v, v), regr_intercept(v, v) FROM large_moments" + groupBy,
+ 2)
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("statistical aggregates merge large nearby values across partitions") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "1",
+ SQLConf.FILES_MAX_PARTITION_BYTES.key -> "1048576",
+ SQLConf.FILES_OPEN_COST_IN_BYTES.key -> "1048576",
+ CometConf.COMET_SHUFFLE_ENABLED.key -> "true",
+ CometConf.COMET_SHUFFLE_MODE.key -> "native") {
+ withTempPath { path =>
+ // Two constant-valued files produce separate partials with zero M2. The
+ // old merge returns 576 instead of 1024, regardless of which partial arrives first.
+ for (value <- Seq(1e17 - 96, 1e17 - 32)) {
+ (Seq.fill(3)((0, Option(value))) ++ Seq((0, None), (1, None)))
+ .toDF("g", "v")
+ .coalesce(1)
+ .write
+ .mode("append")
+ .parquet(path.getCanonicalPath)
+ }
+ withParquetTable(path.getCanonicalPath, "merged_moments") {
+ assert(spark.table("merged_moments").rdd.getNumPartitions == 2)
+ for (groupBy <- Seq("", " GROUP BY g")) {
+ val query = "SELECT var_pop(v), var_samp(v), stddev_pop(v), stddev_samp(v) " +
+ "FROM merged_moments" + groupBy
+ val (_, cometPlan) = checkSparkAnswerAndOperator(query)
+ val aggregates = cometPlan.collect { case a: CometHashAggregateExec => a }
+ assert(aggregates.exists(_.modes.contains(Partial)))
+ assert(aggregates.exists(_.modes.contains(Final)))
+ val expected = Seq(Row(1024.0, 1228.8, 32.0, math.sqrt(1228.8))) ++
+ (if (groupBy.isEmpty) Seq.empty else Seq(Row(null, null, null, null)))
+ checkAnswer(sql(query), expected)
+ checkSparkAnswerWithTolAndNumOfAggregates(
+ "SELECT covar_pop(v, -v), covar_samp(v, -v), corr(v, -v), regr_r2(v, -v), " +
+ "regr_sxx(v, -v), regr_syy(v, -v), regr_sxy(v, -v), " +
+ "regr_slope(v, -v), regr_intercept(v, -v) FROM merged_moments" + groupBy,
+ 2)
+ }
+ }
+ }
+ }
+ }
+
+ test("statistical aggregates merge fractional constants across partitions") {
Review Comment:
The merge fix also resolves #6481's seven aggregate results with ANSI off: exactly `NULL, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0`. Validation passed for grouped and ungrouped queries and both native merge orders. Could you extend this test to cover those aggregates exactly and reference #6481?
ANSI on still returns NULL for `corr` where Spark raises `DIVIDE_BY_ZERO`, so that remaining behavior needs a linked follow-up before closing the issue. The fork CI note can also be updated: its run at d89c385 finished green.
##########
spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala:
##########
@@ -2343,6 +2343,131 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
}
}
+ test("statistical aggregates with large nearby values") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "1",
+ "spark.sql.files.minPartitionNum" -> "1",
+ CometConf.COMET_SHUFFLE_ENABLED.key -> "true",
+ CometConf.COMET_SHUFFLE_MODE.key -> "native") {
+ for (values <- Seq(Seq(1e16, 1e16 + 2), Seq(1e16 + 2, 1e16), Seq(-1e16, -1e16 - 2))) {
+ // One ordered file keeps both values in the same partial accumulator. Splitting
+ // them across files would only exercise merging two single-row states.
+ withTempPath { path =>
+ (Seq(Some(values.head), None, Some(values.last)))
+ .map(v => (0, v))
+ .toDF("g", "v")
+ .coalesce(1)
+ .write
+ .parquet(path.getCanonicalPath)
+ withParquetTable(path.getCanonicalPath, "large_moments") {
+ for (groupBy <- Seq("", " GROUP BY g")) {
+ val query = "SELECT var_pop(v), var_samp(v), stddev_pop(v), stddev_samp(v) " +
+ "FROM large_moments" + groupBy
+ val (_, cometPlan) = checkSparkAnswerAndOperator(query)
+ val aggregates = cometPlan.collect { case a: CometHashAggregateExec => a }
+ assert(aggregates.exists(_.modes.contains(Partial)))
+ assert(aggregates.exists(_.modes.contains(Final)))
+ checkAnswer(sql(query), Seq(Row(1.0, 2.0, 1.0, math.sqrt(2.0))))
+
+ // CORR and REGR_R2 use PearsonCorrelation's update, while REGR_SXX/SYY
+ // and the variance used by slope/intercept follow CentralMomentAgg.
+ checkSparkAnswerWithTolAndNumOfAggregates(
Review Comment:
The scalar `corr` evaluation difference is confirmed. This nearby-value test returns `0.9999999999999998` versus Spark's `1.0`, hidden by the `1e-6` tolerance. On `[1e100, 2e100]`, scalar Comet returns `1.0000000000000002` versus Spark/grouped Comet's `0.0`.
`CorrelationAccumulator::evaluate` divides normalized covariance by the two standard deviations, while Spark and the grouped path use `ck / sqrt(m2_1 * m2_2)`. This predates the PR. Could you either align scalar evaluation and make `corr`'s assertion exact, or link a follow-up explicitly separating this from the update/merge fix?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: match Spark statistical aggregate updates and merges [datafusion-comet]
Posted by "rich7420 (via GitHub)" <gi...@apache.org>.
rich7420 commented on PR #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076#issuecomment-5958142803
@andygrove and @sunchao thanks for the review! I'll update as soon as possilbe
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: match Spark statistical aggregate updates and merges [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076#discussion_r4193325561
##########
native/spark-expr/src/agg_funcs/regr.rs:
##########
@@ -88,6 +91,7 @@ impl Regr {
name: impl Into<String>,
filter_var_by_pair_nulls: bool,
r2_constant_dependent_is_perfect_fit: bool,
+ eval_mode: EvalMode,
Review Comment:
[P2] Adding the required `eval_mode` argument leaves `native/spark-expr/benches/aggregate_misc.rs:176` calling `Regr::new` with four arguments. Consequently, benchmark compilation and the required all-targets Clippy check fail with `E0061`, blocking CI before its Rust tests run. Could you update that caller to pass `EvalMode::Legacy` (the mode is unused for `RegrType::Slope`) and rerun the all-targets check?
Evidence: At the reviewed head, running `cargo check --offline -p datafusion-comet-spark-expr --bench aggregate_misc` from `native/` exits 101: `this function takes 5 arguments but 4 arguments were supplied`, pointing to `spark-expr/benches/aggregate_misc.rs:176`. The benchmark is unchanged relative to the supplied base. Exact-head upstream job https://github.com/apache/datafusion-comet/actions/runs/37423867266/job/112140226269 and fork CI report the same failure.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: match Spark statistical aggregate updates and merges [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076#issuecomment-5950401917
#6451 is now on `main`. It marks the five `regr_*` serdes `Incompatible` until the merge matches Spark, which is exactly what this PR does. With it merged in, three of your tests fail because the `regr_*` queries fall back to Spark. `statistical aggregates with large nearby values` and `statistical aggregates merge large nearby values across partitions` find 0 Comet aggregates instead of 2, and `statistical aggregates merge fractional constants across partitions` reports the #6423 fallback reason. Could you lift that marking in this PR? That means removing the `getSupportLevel` and `getIncompatibleReasons` overrides from `CometRegrBase` in `aggregates.scala`, restoring the six `regr_*` rows in `expressions.md`, dropping the five `allowIncompatible` lines from `regr.sql`, and deleting `regr_fallback.sql`. Your fractional-constants test already covers that data with two partitions asserted. The description can then say this PR is the real fix for #6423 and replaces #6451's fallback.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: match Spark statistical aggregate updates and merges [datafusion-comet]
Posted by "rich7420 (via GitHub)" <gi...@apache.org>.
rich7420 commented on code in PR #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076#discussion_r4152816912
##########
spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala:
##########
@@ -2343,6 +2343,131 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
}
}
+ test("statistical aggregates with large nearby values") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "1",
+ "spark.sql.files.minPartitionNum" -> "1",
+ CometConf.COMET_SHUFFLE_ENABLED.key -> "true",
+ CometConf.COMET_SHUFFLE_MODE.key -> "native") {
+ for (values <- Seq(Seq(1e16, 1e16 + 2), Seq(1e16 + 2, 1e16), Seq(-1e16, -1e16 - 2))) {
+ // One ordered file keeps both values in the same partial accumulator. Splitting
+ // them across files would only exercise merging two single-row states.
+ withTempPath { path =>
+ (Seq(Some(values.head), None, Some(values.last)))
+ .map(v => (0, v))
+ .toDF("g", "v")
+ .coalesce(1)
+ .write
+ .parquet(path.getCanonicalPath)
+ withParquetTable(path.getCanonicalPath, "large_moments") {
+ for (groupBy <- Seq("", " GROUP BY g")) {
+ val query = "SELECT var_pop(v), var_samp(v), stddev_pop(v), stddev_samp(v) " +
+ "FROM large_moments" + groupBy
+ val (_, cometPlan) = checkSparkAnswerAndOperator(query)
+ val aggregates = cometPlan.collect { case a: CometHashAggregateExec => a }
+ assert(aggregates.exists(_.modes.contains(Partial)))
+ assert(aggregates.exists(_.modes.contains(Final)))
+ checkAnswer(sql(query), Seq(Row(1.0, 2.0, 1.0, math.sqrt(2.0))))
+
+ // CORR and REGR_R2 use PearsonCorrelation's update, while REGR_SXX/SYY
+ // and the variance used by slope/intercept follow CentralMomentAgg.
+ checkSparkAnswerWithTolAndNumOfAggregates(
+ "SELECT corr(v, v), regr_r2(v, v), regr_sxx(v, v), regr_syy(v, v), " +
+ "regr_slope(v, v), regr_intercept(v, v) FROM large_moments" + groupBy,
+ 2)
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("statistical aggregates merge large nearby values across partitions") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "1",
+ SQLConf.FILES_MAX_PARTITION_BYTES.key -> "1048576",
+ SQLConf.FILES_OPEN_COST_IN_BYTES.key -> "1048576",
+ CometConf.COMET_SHUFFLE_ENABLED.key -> "true",
+ CometConf.COMET_SHUFFLE_MODE.key -> "native") {
+ withTempPath { path =>
+ // Two constant-valued files produce separate partials with zero M2. The
+ // old merge returns 576 instead of 1024, regardless of which partial arrives first.
+ for (value <- Seq(1e17 - 96, 1e17 - 32)) {
+ (Seq.fill(3)((0, Option(value))) ++ Seq((0, None), (1, None)))
+ .toDF("g", "v")
+ .coalesce(1)
+ .write
+ .mode("append")
+ .parquet(path.getCanonicalPath)
+ }
+ withParquetTable(path.getCanonicalPath, "merged_moments") {
+ assert(spark.table("merged_moments").rdd.getNumPartitions == 2)
+ for (groupBy <- Seq("", " GROUP BY g")) {
+ val query = "SELECT var_pop(v), var_samp(v), stddev_pop(v), stddev_samp(v) " +
+ "FROM merged_moments" + groupBy
+ val (_, cometPlan) = checkSparkAnswerAndOperator(query)
+ val aggregates = cometPlan.collect { case a: CometHashAggregateExec => a }
+ assert(aggregates.exists(_.modes.contains(Partial)))
+ assert(aggregates.exists(_.modes.contains(Final)))
+ val expected = Seq(Row(1024.0, 1228.8, 32.0, math.sqrt(1228.8))) ++
+ (if (groupBy.isEmpty) Seq.empty else Seq(Row(null, null, null, null)))
+ checkAnswer(sql(query), expected)
+ checkSparkAnswerWithTolAndNumOfAggregates(
+ "SELECT covar_pop(v, -v), covar_samp(v, -v), corr(v, -v), regr_r2(v, -v), " +
+ "regr_sxx(v, -v), regr_syy(v, -v), regr_sxy(v, -v), " +
+ "regr_slope(v, -v), regr_intercept(v, -v) FROM merged_moments" + groupBy,
+ 2)
+ }
+ }
+ }
+ }
+ }
+
+ test("statistical aggregates merge fractional constants across partitions") {
Review Comment:
Added in c4baa2611: #6481's seven aggregates now assert exactly `NULL, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0` with ANSI off, for grouped/ungrouped queries and both file-writing orders, with native Partial/Final aggregates checked. All four targeted Spark 4.1.3 tests pass. #6481 stays open for the ANSI-on `corr` difference (NULL versus Spark's `DIVIDE_BY_ZERO`). I also corrected the fork CI note: d89c385 finished green, but the pushed head needs its own verdict; the additional profiles and Spark 4.1 SQL checks are running.
##########
spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala:
##########
@@ -2343,6 +2343,131 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
}
}
+ test("statistical aggregates with large nearby values") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "1",
+ "spark.sql.files.minPartitionNum" -> "1",
+ CometConf.COMET_SHUFFLE_ENABLED.key -> "true",
+ CometConf.COMET_SHUFFLE_MODE.key -> "native") {
+ for (values <- Seq(Seq(1e16, 1e16 + 2), Seq(1e16 + 2, 1e16), Seq(-1e16, -1e16 - 2))) {
+ // One ordered file keeps both values in the same partial accumulator. Splitting
+ // them across files would only exercise merging two single-row states.
+ withTempPath { path =>
+ (Seq(Some(values.head), None, Some(values.last)))
+ .map(v => (0, v))
+ .toDF("g", "v")
+ .coalesce(1)
+ .write
+ .parquet(path.getCanonicalPath)
+ withParquetTable(path.getCanonicalPath, "large_moments") {
+ for (groupBy <- Seq("", " GROUP BY g")) {
+ val query = "SELECT var_pop(v), var_samp(v), stddev_pop(v), stddev_samp(v) " +
+ "FROM large_moments" + groupBy
+ val (_, cometPlan) = checkSparkAnswerAndOperator(query)
+ val aggregates = cometPlan.collect { case a: CometHashAggregateExec => a }
+ assert(aggregates.exists(_.modes.contains(Partial)))
+ assert(aggregates.exists(_.modes.contains(Final)))
+ checkAnswer(sql(query), Seq(Row(1.0, 2.0, 1.0, math.sqrt(2.0))))
+
+ // CORR and REGR_R2 use PearsonCorrelation's update, while REGR_SXX/SYY
+ // and the variance used by slope/intercept follow CentralMomentAgg.
+ checkSparkAnswerWithTolAndNumOfAggregates(
Review Comment:
Fixed in c4baa2611: scalar `corr` now uses `ck / sqrt(m2_1 * m2_2)`, matching Spark and grouped Comet. Exact Spark assertions cover 1.0 for the nearby positive pair, NULL for its reversed order, and 0.0 for `[1e100, 2e100]`; the nearby-value and overflow tests fail with the old library and pass with this fix. The boundary checks also fix raw-product underflow to NULL and preserve NaN for zero times infinity in both paths, with direct/merged native coverage, null-pair filtering, both signs and legacy-statistics settings. All 138 native aggregate tests and four targeted Spark tests pass. ANSI-on division errors remain tracked in #6481.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: match Spark statistical aggregate updates and merges [datafusion-comet]
Posted by "rich7420 (via GitHub)" <gi...@apache.org>.
rich7420 commented on code in PR #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076#discussion_r4205366742
##########
native/spark-expr/src/agg_funcs/regr.rs:
##########
@@ -88,6 +91,7 @@ impl Regr {
name: impl Into<String>,
filter_var_by_pair_nulls: bool,
r2_constant_dependent_is_perfect_fit: bool,
+ eval_mode: EvalMode,
Review Comment:
Fixed in `37506bfd4`. The slope benchmark now passes `EvalMode::Legacy`; slope does not consume the evaluation mode. I checked both `Regr::new` call sites, and the planner already passes the captured mode. Full workspace/all-targets Clippy with warnings denied and Rustfmt pass locally. [Current-head CI](https://github.com/apache/datafusion-comet/actions/runs/37474391984) now passes, including Required Checks.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: match Spark variance updates for large nearby values [datafusion-comet]
Posted by "rich7420 (via GitHub)" <gi...@apache.org>.
rich7420 commented on PR #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076#issuecomment-5808171634
could you take another look? thanks!
cc @andygrove , @sunchao
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: match Spark statistical aggregate updates and merges [datafusion-comet]
Posted by "rich7420 (via GitHub)" <gi...@apache.org>.
rich7420 commented on PR #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076#issuecomment-6009058226
@andygrove could you take a look please?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: match Spark statistical aggregate updates and merges [datafusion-comet]
Posted by "rich7420 (via GitHub)" <gi...@apache.org>.
rich7420 commented on PR #6076:
URL: https://github.com/apache/datafusion-comet/pull/6076#issuecomment-6154915631
thanks for the review @andygrove , @sunchao , @kazuyukitanimura !!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org