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