From 14919cc6fd602350d0d2dab0c01e11e29fdc2cf6 Mon Sep 17 00:00:00 2001 From: rich7420 Date: Sat, 19 Sep 2026 23:38:07 +0800 Subject: [PATCH 1/4] fix: fall back for ANSI and TRY integer sums over sliding windows --- .../expression-audits/window_funcs.md | 3 +- .../latest/compatibility/operators.md | 2 + .../spark/sql/comet/CometWindowExec.scala | 11 ++ .../sql-tests/windows/sliding_integer_sum.sql | 129 ++++++++++++++++++ .../comet/exec/CometWindowExecSuite.scala | 18 ++- 5 files changed, 159 insertions(+), 4 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/windows/sliding_integer_sum.sql diff --git a/docs/source/contributor-guide/expression-audits/window_funcs.md b/docs/source/contributor-guide/expression-audits/window_funcs.md index c692a42a555..e28fca5bd1e 100644 --- a/docs/source/contributor-guide/expression-audits/window_funcs.md +++ b/docs/source/contributor-guide/expression-audits/window_funcs.md @@ -35,6 +35,7 @@ ## sum (window) -- 4.1.1, audited 2026-06-25: `SUM()` over a sliding window frame (lower bound not `UNBOUNDED PRECEDING`) falls back to Spark. The native sliding path would use DataFusion's built-in `sum`, which wraps on overflow rather than returning Spark's NULL, and overflow cannot be detected at plan time, so the whole sliding decimal `SUM` case falls back. Ever-expanding frames use Comet's overflow-aware `SumDecimal` UDAF and run natively, matching Spark including overflow-to-NULL. Bigint sliding `SUM` overflow matches Spark (both wrap) and stays native. +- 4.1.1, audited 2026-06-25: `SUM()` over a sliding window frame (lower bound not `UNBOUNDED PRECEDING`) falls back to Spark. The native sliding path would use DataFusion's built-in `sum`, which wraps on overflow rather than returning Spark's NULL, and overflow cannot be detected at plan time, so the whole sliding decimal `SUM` case falls back. Ever-expanding frames use Comet's overflow-aware `SumDecimal` UDAF and run natively, matching Spark including overflow-to-NULL. Bigint sliding `SUM` overflow matches Spark in legacy mode (both wrap) and stays native. +- 3.5.9 and 4.1.3, audited 2026-09-19: Integral sliding `SUM` in ANSI or TRY mode also requires fallback: the DataFusion accumulator wraps instead of throwing or returning NULL. The guard uses the aggregate's evaluation mode, so `try_sum` falls back even with ANSI disabled. Legacy sliding sums and ever-expanding integral sums retain native execution. [Spark Expression Support]: ../../user-guide/latest/expressions.md diff --git a/docs/source/user-guide/latest/compatibility/operators.md b/docs/source/user-guide/latest/compatibility/operators.md index c637c197790..c4974834d06 100644 --- a/docs/source/user-guide/latest/compatibility/operators.md +++ b/docs/source/user-guide/latest/compatibility/operators.md @@ -86,6 +86,8 @@ incorrect result. When any single window expression in a `WindowExec` falls back support as the batch aggregates, so these fall back in both contexts. - `sum` or `avg` on `DECIMAL` with a sliding (non ever-expanding) frame, because the sliding path would wrap on overflow instead of returning Spark's `NULL`. +- Integral `sum` in ANSI mode and integral `try_sum` with a sliding frame, because the sliding path would wrap + on overflow instead of throwing or returning `NULL`. Legacy integral sums and ever-expanding frames stay native. - `RANGE` frame with an explicit offset when the `ORDER BY` column is `DATE` or `DECIMAL` ([#4834](https://github.com/apache/datafusion-comet/issues/4834)). - `first_value` / `last_value` on a `RANGE` frame with a literal offset diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala index 4d4abf3d2e0..018207e1f68 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala @@ -34,9 +34,11 @@ import com.google.common.base.Objects import org.apache.comet.{CometConf, ConfigEntry} import org.apache.comet.CometSparkSessionExtensions.withFallbackReason +import org.apache.comet.expressions.CometEvalMode import org.apache.comet.serde.{AggSerde, CometOperatorSerde, LiteralOuterClass, OperatorOuterClass} import org.apache.comet.serde.OperatorOuterClass.Operator import org.apache.comet.serde.QueryPlanSerde.{aggExprToProto, exprToProto, scalarFunctionExprToProto, serializeDataType} +import org.apache.comet.shims.CometEvalModeUtil object CometWindowExec extends CometOperatorSerde[WindowExec] { @@ -342,6 +344,15 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] { windowExpr, "SUM on DECIMAL with a sliding window frame is not supported") return None + // The sliding accumulator also wraps integer sums, ignoring ANSI/TRY overflow. + case s: Sum + if s.dataType == LongType && + CometEvalModeUtil.fromSparkEvalMode(CometEvalModeUtil.sumEvalMode(s)) != + CometEvalMode.LEGACY => + withFallbackReason( + windowExpr, + "ANSI/TRY SUM on integral types with a sliding window frame is not supported") + return None case a: Average if a.dataType.isInstanceOf[DecimalType] => withFallbackReason( windowExpr, diff --git a/spark/src/test/resources/sql-tests/windows/sliding_integer_sum.sql b/spark/src/test/resources/sql-tests/windows/sliding_integer_sum.sql new file mode 100644 index 00000000000..d53a9f1acb7 --- /dev/null +++ b/spark/src/test/resources/sql-tests/windows/sliding_integer_sum.sql @@ -0,0 +1,129 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to you under the Apache License, Version 2.0 (the +-- "License"); you may not use this file except in compliance +-- with the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, +-- software distributed under the License is distributed on an +-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +-- KIND, either express or implied. See the License for the +-- specific language governing permissions and limitations +-- under the License. + +-- Config: spark.sql.adaptive.enabled=false +-- Config: spark.sql.ansi.enabled=false +-- Config: spark.comet.operator.WindowExec.allowIncompatible=false + +-- Spark needs constant folding for PRECEDING bounds. Aggregate inputs remain columns. +statement +SET spark.sql.optimizer.excludedRules= + +statement +CREATE TABLE sliding_integer_sum(g INT, id INT, v BIGINT) USING parquet + +statement +INSERT INTO sliding_integer_sum VALUES + (1, 1, 9223372036854775807), (1, 2, 1), (1, 3, -1), (1, 4, NULL), (1, 5, NULL), + (2, 1, -9223372036854775808), (2, 2, -1), (2, 3, 1), (2, 4, NULL), (2, 5, NULL), + (3, 1, NULL), (3, 2, NULL), (3, 3, NULL) + +-- TRY mode must fall back even when ANSI is disabled. Cover positive/negative +-- overflow, recovery after it leaves the frame, all-NULL and empty frames. +query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) +SELECT g, id, try_sum(v) OVER ( + PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum + +query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) +SELECT g, id, try_sum(v) OVER ( + PARTITION BY g ORDER BY id RANGE BETWEEN 1 PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum + +query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) +SELECT g, id, try_sum(v) OVER ( + PARTITION BY g ORDER BY id ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING) +FROM sliding_integer_sum + +query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) +SELECT g, id, try_sum(v) OVER ( + PARTITION BY g ORDER BY id ROWS BETWEEN 1 FOLLOWING AND 1 FOLLOWING) +FROM sliding_integer_sum + +-- Legacy sliding sums still run natively, including wrapping overflow. +query +SELECT g, id, + sum(v) OVER (PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW), + sum(v) OVER (PARTITION BY g ORDER BY id RANGE BETWEEN 1 PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum + +-- Ever-expanding TRY sums retain native overflow-to-NULL behavior. +query +SELECT g, id, try_sum(v) OVER ( + PARTITION BY g ORDER BY id ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum + +statement +SET spark.sql.ansi.enabled=true + +query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) +SELECT g, id, try_sum(v) OVER ( + PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum + +-- Admission is independent of the data; even a non-overflowing ANSI sum falls back. +query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) +SELECT g, id, sum(v) OVER ( + PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum WHERE id > 1 + +query expect_error(ARITHMETIC_OVERFLOW) +SELECT sum(v) OVER (ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum WHERE g = 1 + +query expect_error(ARITHMETIC_OVERFLOW) +SELECT sum(v) OVER (ORDER BY id RANGE BETWEEN 1 PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum WHERE g = 2 + +-- Narrow integral inputs also produce BIGINT sums and require the same guard. +query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) +SELECT sum(CAST(id AS TINYINT)) OVER ( + PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum + +query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) +SELECT sum(CAST(id AS SMALLINT)) OVER ( + PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum + +query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) +SELECT sum(id) OVER ( + PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum + +-- Floating-point sums do not use checked integer arithmetic and remain native. +query +SELECT g, id, + sum(CAST(v AS DOUBLE)) OVER (PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW), + try_sum(CAST(v AS DOUBLE)) OVER (PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum WHERE id > 1 + +-- Native expanding sums remain enabled in ANSI and TRY mode. +query +SELECT g, id, + sum(v) OVER (PARTITION BY g ORDER BY id ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW), + try_sum(v) OVER (PARTITION BY g ORDER BY id ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum WHERE id > 1 + +query +SELECT g, id, try_sum(v) OVER ( + PARTITION BY g ORDER BY id ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum + +query expect_error(ARITHMETIC_OVERFLOW) +SELECT sum(v) OVER (ORDER BY id ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum WHERE g = 1 diff --git a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala index 8f3d785c726..88872b119cd 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala @@ -64,6 +64,16 @@ class CometWindowExecSuite extends CometTestBase { assert(cometWindowExecs.nonEmpty) } + private def checkSlidingIntegralSum(df: DataFrame): Unit = { + if (SQLConf.get.ansiEnabled) { + checkSparkAnswerAndFallbackReason( + df, + "ANSI/TRY SUM on integral types with a sliding window frame is not supported") + } else { + checkSparkAnswerAndOperator(df) + } + } + private def sparkWindowExpressions(plan: SparkPlan): Seq[Expression] = { collect(plan) { case w: SparkWindowExec => w.windowExpression @@ -528,6 +538,8 @@ class CometWindowExecSuite extends CometTestBase { test("Windows support") { Seq("true", "false").foreach(aqeEnabled => withSQLConf( + // This native coverage matrix includes legacy sliding integral sums. + SQLConf.ANSI_ENABLED.key -> "false", CometConf.COMET_SHUFFLE_ENABLED.key -> "true", SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqeEnabled) { withParquetTable((0 until 10).map(i => (i, 10 - i)), "t1") { // TODO: test nulls @@ -835,7 +847,7 @@ class CometWindowExecSuite extends CometTestBase { SUM(c) OVER (PARTITION BY a ORDER BY b, c ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING) as sum_c FROM window_test """) - checkSparkAnswerAndOperator(df) + checkSlidingIntegralSum(df) } } @@ -875,7 +887,7 @@ class CometWindowExecSuite extends CometTestBase { SUM(c) OVER (PARTITION BY a ORDER BY b, c ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) as sum_c FROM window_test """) - checkSparkAnswerAndOperator(df) + checkSlidingIntegralSum(df) } } @@ -1346,7 +1358,7 @@ class CometWindowExecSuite extends CometTestBase { SUM(c) OVER (PARTITION BY a ORDER BY b RANGE BETWEEN 2 PRECEDING AND 2 FOLLOWING) as sum_c FROM window_test """) - checkSparkAnswerAndOperator(df) + checkSlidingIntegralSum(df) } } From 748bf996caf4bf1785cc275e9982719108f29799 Mon Sep 17 00:00:00 2001 From: rich7420 Date: Tue, 22 Sep 2026 15:54:34 +0800 Subject: [PATCH 2/4] test: preserve window harness and ANSI coverage --- .../sql-tests/windows/sliding_integer_sum.sql | 50 +---------- .../comet/exec/CometWindowExecSuite.scala | 84 +++++++++++++++++-- 2 files changed, 78 insertions(+), 56 deletions(-) diff --git a/spark/src/test/resources/sql-tests/windows/sliding_integer_sum.sql b/spark/src/test/resources/sql-tests/windows/sliding_integer_sum.sql index d53a9f1acb7..1f25f233f92 100644 --- a/spark/src/test/resources/sql-tests/windows/sliding_integer_sum.sql +++ b/spark/src/test/resources/sql-tests/windows/sliding_integer_sum.sql @@ -17,11 +17,9 @@ -- Config: spark.sql.adaptive.enabled=false -- Config: spark.sql.ansi.enabled=false --- Config: spark.comet.operator.WindowExec.allowIncompatible=false --- Spark needs constant folding for PRECEDING bounds. Aggregate inputs remain columns. -statement -SET spark.sql.optimizer.excludedRules= +-- ROWS ... N PRECEDING is covered through the DataFrame API in CometWindowExecSuite +-- so this fixture can keep the harness's ConstantFolding exclusion. statement CREATE TABLE sliding_integer_sum(g INT, id INT, v BIGINT) USING parquet @@ -34,11 +32,6 @@ INSERT INTO sliding_integer_sum VALUES -- TRY mode must fall back even when ANSI is disabled. Cover positive/negative -- overflow, recovery after it leaves the frame, all-NULL and empty frames. -query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) -SELECT g, id, try_sum(v) OVER ( - PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) -FROM sliding_integer_sum - query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) SELECT g, id, try_sum(v) OVER ( PARTITION BY g ORDER BY id RANGE BETWEEN 1 PRECEDING AND CURRENT ROW) @@ -57,7 +50,6 @@ FROM sliding_integer_sum -- Legacy sliding sums still run natively, including wrapping overflow. query SELECT g, id, - sum(v) OVER (PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW), sum(v) OVER (PARTITION BY g ORDER BY id RANGE BETWEEN 1 PRECEDING AND CURRENT ROW) FROM sliding_integer_sum @@ -70,48 +62,10 @@ FROM sliding_integer_sum statement SET spark.sql.ansi.enabled=true -query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) -SELECT g, id, try_sum(v) OVER ( - PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) -FROM sliding_integer_sum - --- Admission is independent of the data; even a non-overflowing ANSI sum falls back. -query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) -SELECT g, id, sum(v) OVER ( - PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) -FROM sliding_integer_sum WHERE id > 1 - -query expect_error(ARITHMETIC_OVERFLOW) -SELECT sum(v) OVER (ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) -FROM sliding_integer_sum WHERE g = 1 - query expect_error(ARITHMETIC_OVERFLOW) SELECT sum(v) OVER (ORDER BY id RANGE BETWEEN 1 PRECEDING AND CURRENT ROW) FROM sliding_integer_sum WHERE g = 2 --- Narrow integral inputs also produce BIGINT sums and require the same guard. -query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) -SELECT sum(CAST(id AS TINYINT)) OVER ( - PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) -FROM sliding_integer_sum - -query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) -SELECT sum(CAST(id AS SMALLINT)) OVER ( - PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) -FROM sliding_integer_sum - -query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) -SELECT sum(id) OVER ( - PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) -FROM sliding_integer_sum - --- Floating-point sums do not use checked integer arithmetic and remain native. -query -SELECT g, id, - sum(CAST(v AS DOUBLE)) OVER (PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW), - try_sum(CAST(v AS DOUBLE)) OVER (PARTITION BY g ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) -FROM sliding_integer_sum WHERE id > 1 - -- Native expanding sums remain enabled in ANSI and TRY mode. query SELECT g, id, diff --git a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala index 88872b119cd..7842668b4ab 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala @@ -31,7 +31,7 @@ import org.apache.spark.sql.comet.{CometSortExec, CometWindowExec, CometWindowGr import org.apache.spark.sql.execution.SparkPlan import org.apache.spark.sql.execution.window.{WindowExec => SparkWindowExec} import org.apache.spark.sql.expressions.Window -import org.apache.spark.sql.functions.{count, lead, sum} +import org.apache.spark.sql.functions.{count, expr, lead, sum} import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.DecimalType @@ -536,10 +536,12 @@ class CometWindowExecSuite extends CometTestBase { } test("Windows support") { - Seq("true", "false").foreach(aqeEnabled => + for { + aqeEnabled <- Seq("true", "false") + ansiEnabled <- Seq("true", "false") + } { withSQLConf( - // This native coverage matrix includes legacy sliding integral sums. - SQLConf.ANSI_ENABLED.key -> "false", + SQLConf.ANSI_ENABLED.key -> ansiEnabled, CometConf.COMET_SHUFFLE_ENABLED.key -> "true", SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqeEnabled) { withParquetTable((0 until 10).map(i => (i, 10 - i)), "t1") { // TODO: test nulls @@ -557,17 +559,83 @@ class CometWindowExecSuite extends CometTestBase { s"SELECT $function OVER() FROM t1", s"SELECT $function OVER(order by _2) FROM t1", s"SELECT $function OVER(order by _2 desc) FROM t1", - s"SELECT $function OVER(partition by _2 order by _2) FROM t1", + s"SELECT $function OVER(partition by _2 order by _2) FROM t1") + queries.foreach { query => + checkSparkAnswerAndOperator(query) + } + + val slidingQueries = Seq( s"SELECT $function OVER(rows between 1 preceding and 1 following) FROM t1", s"SELECT $function OVER(order by _2 rows between 1 preceding and current row) FROM t1", s"SELECT $function OVER(order by _2 rows between current row and 1 following) FROM t1") - queries.foreach { query => - checkSparkAnswerAndOperator(query) + slidingQueries.foreach { query => + if (function == "SUM(_1)") { + checkSlidingIntegralSum(sql(query)) + } else { + checkSparkAnswerAndOperator(query) + } + } + } + } + } + } + } + + for (ansiEnabled <- Seq("true", "false")) { + test(s"sliding integral ROWS sums (ANSI=$ansiEnabled)") { + withSQLConf( + SQLConf.ANSI_ENABLED.key -> ansiEnabled, + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> + "org.apache.spark.sql.catalyst.optimizer.ConstantFolding") { + val values = Seq( + (1, 1, Some(Long.MaxValue)), + (1, 2, Some(1L)), + (1, 3, Some(-1L)), + (1, 4, None), + (1, 5, None), + (2, 1, Some(Long.MinValue)), + (2, 2, Some(-1L)), + (2, 3, Some(1L)), + (2, 4, None), + (2, 5, None), + (3, 1, None), + (3, 2, None), + (3, 3, None)) + withParquetTable(values, "sliding_rows_sum") { + val df = spark.table("sliding_rows_sum").toDF("g", "id", "v") + // DataFrame bounds are literals, so PRECEDING needs no constant folding. + val frame = Window.partitionBy("g").orderBy("id").rowsBetween(-1, Window.currentRow) + checkSparkAnswerAndFallbackReason( + df.select($"g", $"id", expr("try_sum(v)").over(frame)), + "ANSI/TRY SUM on integral types with a sliding window frame is not supported") + + // ANSI admission does not depend on whether the data actually overflows. + val sumInput = if (SQLConf.get.ansiEnabled) df.where($"id" > 1) else df + checkSlidingIntegralSum(sumInput.select($"g", $"id", sum("v").over(frame))) + for (dataType <- Seq("tinyint", "smallint", "int")) { + checkSlidingIntegralSum(df.select(sum($"id".cast(dataType)).over(frame))) + } + + if (SQLConf.get.ansiEnabled) { + for (group <- Seq(1, 2)) { + val (sparkError, cometError) = + checkSparkAnswerMaybeThrows(df.where($"g" === group).select(sum("v").over(frame))) + assert(sparkError.exists(_.getMessage.contains("ARITHMETIC_OVERFLOW"))) + assert(cometError.exists(_.getMessage.contains("ARITHMETIC_OVERFLOW"))) } } + + // Floating-point sums remain native in both modes. + checkSparkAnswerAndOperator( + df.where($"id" > 1) + .select( + sum($"v".cast("double")).over(frame), + expr("try_sum(CAST(v AS DOUBLE))").over(frame))) } - }) + } + } } test("window: simple COUNT(*) without frame") { From 0b6c0f7db497c882fcaa75b6d7ebee3ed4666e4a Mon Sep 17 00:00:00 2001 From: rich7420 Date: Wed, 30 Sep 2026 15:14:17 +0800 Subject: [PATCH 3/4] fix: preserve native sliding integer sum overflow semantics --- .../expression-audits/window_funcs.md | 2 +- .../latest/compatibility/operators.md | 2 - native/core/src/execution/planner.rs | 22 +- native/spark-expr/src/agg_funcs/sum_int.rs | 250 +++++++++++++++++- .../spark/sql/comet/CometWindowExec.scala | 11 - .../sql-tests/windows/sliding_integer_sum.sql | 31 ++- .../comet/exec/CometWindowExecSuite.scala | 38 +-- .../benchmark/CometSlidingSumBenchmark.scala | 92 +++++++ 8 files changed, 394 insertions(+), 54 deletions(-) create mode 100644 spark/src/test/scala/org/apache/spark/sql/benchmark/CometSlidingSumBenchmark.scala diff --git a/docs/source/contributor-guide/expression-audits/window_funcs.md b/docs/source/contributor-guide/expression-audits/window_funcs.md index e28fca5bd1e..7b6111f1fa9 100644 --- a/docs/source/contributor-guide/expression-audits/window_funcs.md +++ b/docs/source/contributor-guide/expression-audits/window_funcs.md @@ -36,6 +36,6 @@ ## sum (window) - 4.1.1, audited 2026-06-25: `SUM()` over a sliding window frame (lower bound not `UNBOUNDED PRECEDING`) falls back to Spark. The native sliding path would use DataFusion's built-in `sum`, which wraps on overflow rather than returning Spark's NULL, and overflow cannot be detected at plan time, so the whole sliding decimal `SUM` case falls back. Ever-expanding frames use Comet's overflow-aware `SumDecimal` UDAF and run natively, matching Spark including overflow-to-NULL. Bigint sliding `SUM` overflow matches Spark in legacy mode (both wrap) and stays native. -- 3.5.9 and 4.1.3, audited 2026-09-19: Integral sliding `SUM` in ANSI or TRY mode also requires fallback: the DataFusion accumulator wraps instead of throwing or returning NULL. The guard uses the aggregate's evaluation mode, so `try_sum` falls back even with ANSI disabled. Legacy sliding sums and ever-expanding integral sums retain native execution. +- 3.5.9 and 4.1.3, audited 2026-09-30: Integral sliding `SUM` uses a native retractable accumulator in ANSI and TRY mode. It checks ordered prefix sums, so intermediate overflow throws or returns NULL even when later values cancel it, and results recover once the offending values leave the frame. Legacy sliding sums retain the DataFusion path. [Spark Expression Support]: ../../user-guide/latest/expressions.md diff --git a/docs/source/user-guide/latest/compatibility/operators.md b/docs/source/user-guide/latest/compatibility/operators.md index c4974834d06..c637c197790 100644 --- a/docs/source/user-guide/latest/compatibility/operators.md +++ b/docs/source/user-guide/latest/compatibility/operators.md @@ -86,8 +86,6 @@ incorrect result. When any single window expression in a `WindowExec` falls back support as the batch aggregates, so these fall back in both contexts. - `sum` or `avg` on `DECIMAL` with a sliding (non ever-expanding) frame, because the sliding path would wrap on overflow instead of returning Spark's `NULL`. -- Integral `sum` in ANSI mode and integral `try_sum` with a sliding frame, because the sliding path would wrap - on overflow instead of throwing or returning `NULL`. Legacy integral sums and ever-expanding frames stay native. - `RANGE` frame with an explicit offset when the `ORDER BY` column is `DATE` or `DECIMAL` ([#4834](https://github.com/apache/datafusion-comet/issues/4834)). - `first_value` / `last_value` on a `RANGE` frame with a literal offset diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 80f8d890a34..68a7851aab8 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -2437,10 +2437,9 @@ impl PhysicalPlanner { // `PERCENT_RANK` / `CUME_DIST` / `NTILE` // (`!uses_bounded_memory()` — "Can not execute X in a streaming // fashion") and keeps the Spark-compatible Comet UDAFs - // (`SumDecimal` / `SumInteger` / `AvgDecimal` / `Avg`) on the - // non-streaming path since they don't implement `retract_batch`. - // Because `process_agg_func` already picks DataFusion's - // retract-capable built-ins for sliding aggregate frames, + // (`SumDecimal` / `AvgDecimal` / `Avg`) on the non-streaming path + // where needed. `process_agg_func` picks retract-capable UDAFs + // for sliding aggregate frames; // ever-expanding aggregate frames (all that route to // `BoundedWindowAggExec` as `PlainAggregateWindowExpr`) never // trigger a retract call. @@ -3267,9 +3266,9 @@ impl PhysicalPlanner { // DataFusion uses `PlainAggregateWindowExpr` which does not call // `retract_batch`, so we can safely use Comet's Spark-compatible // UDAFs (SumDecimal/SumInteger/AvgDecimal/Avg). Otherwise it uses - // `SlidingAggregateWindowExpr` which requires retract — Comet's UDAFs - // don't implement it, so the caller must fall back to DataFusion's - // built-ins (which do). + // `SlidingAggregateWindowExpr` which requires retract. process_agg_func + // selects Comet's checked integer accumulator or a DataFusion built-in + // that supports retraction. let is_ever_expanding = spark_expr .spec .as_ref() @@ -3485,9 +3484,9 @@ impl PhysicalPlanner { Some(AggExprStruct::Sum(expr)) => { // For ever-expanding frames, use Comet's Spark-compatible Sum UDAFs // (SumDecimal / SumInteger) which enforce Spark overflow semantics. - // For sliding frames, those UDAFs can't be used (no retract_batch), - // so delegate to DataFusion's built-in `sum`, which supports retract - // but doesn't enforce Spark's decimal precision overflow-to-NULL. + // Checked integral sliding sums also use Comet's retractable + // accumulator. Legacy sliding sums keep DataFusion's fast path; + // decimal sliding frames are rejected by the JVM planner. let child = self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&schema))?; let arrow_type = to_arrow_datatype(expr.datatype.as_ref().unwrap()); match arrow_type { @@ -3502,7 +3501,8 @@ impl PhysicalPlanner { Ok((udaf(func), vec![child])) } DataType::Int8 | DataType::Int16 | DataType::Int32 | DataType::Int64 - if is_ever_expanding => + if is_ever_expanding + || from_protobuf_eval_mode(expr.eval_mode)? != EvalMode::Legacy => { let eval_mode = from_protobuf_eval_mode(expr.eval_mode)?; let func = SumInteger::try_new(arrow_type, eval_mode)?; diff --git a/native/spark-expr/src/agg_funcs/sum_int.rs b/native/spark-expr/src/agg_funcs/sum_int.rs index 56a8c122c41..c74480e5c9f 100644 --- a/native/spark-expr/src/agg_funcs/sum_int.rs +++ b/native/spark-expr/src/agg_funcs/sum_int.rs @@ -29,6 +29,7 @@ use datafusion::logical_expr::Volatility::Immutable; use datafusion::logical_expr::{ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, ReversedUDAF, Signature, }; +use std::collections::VecDeque; use std::sync::Arc; #[derive(Debug, PartialEq, Eq, Hash)] @@ -72,6 +73,10 @@ impl AggregateUDFImpl for SumInteger { } } + fn create_sliding_accumulator(&self, _args: AccumulatorArgs) -> DFResult> { + Ok(Box::new(SlidingSumIntegerAccumulator::new(self.eval_mode))) + } + fn state_fields(&self, _args: StateFieldsArgs) -> DFResult> { if self.eval_mode == EvalMode::Try { Ok(vec![ @@ -99,7 +104,149 @@ impl AggregateUDFImpl for SumInteger { } fn reverse_expr(&self) -> ReversedUDAF { - ReversedUDAF::Identical + // Checked addition depends on input order: [MAX, 1, -1] overflows, + // while [-1, 1, MAX] does not. Reversing a frame must not change that. + if self.eval_mode == EvalMode::Legacy { + ReversedUDAF::Identical + } else { + ReversedUDAF::NotSupported + } + } +} + +/// Spark recomputes each sliding frame in input order. Checking only its final +/// sum misses an intermediate overflow followed by cancellation. Track the min +/// and max prefix sums instead: subtracting the prefix before the frame gives +/// every partial sum Spark would visit. Monotonic queues retain just the extrema +/// candidates; each entry is pushed/popped once, for amortized O(1) work per row. +/// +/// i128 holds the exact sum of at most usize::MAX i64 values on a 64-bit host. +/// Nulls need no entries because retraction receives the original input values. +#[derive(Debug)] +struct SlidingSumIntegerAccumulator { + eval_mode: EvalMode, + end: i128, + start: i128, + added: usize, + removed: usize, + minima: VecDeque<(usize, i128)>, + maxima: VecDeque<(usize, i128)>, +} + +impl SlidingSumIntegerAccumulator { + fn new(eval_mode: EvalMode) -> Self { + Self { + eval_mode, + end: 0, + start: 0, + added: 0, + removed: 0, + minima: VecDeque::new(), + maxima: VecDeque::new(), + } + } + + fn add(&mut self, value: i64) { + self.end += i128::from(value); + self.added += 1; + while self.minima.back().is_some_and(|&(_, v)| v >= self.end) { + self.minima.pop_back(); + } + while self.maxima.back().is_some_and(|&(_, v)| v <= self.end) { + self.maxima.pop_back(); + } + self.minima.push_back((self.added, self.end)); + self.maxima.push_back((self.added, self.end)); + } + + fn remove(&mut self, value: i64) { + self.start += i128::from(value); + self.removed += 1; + } + + fn visit(&mut self, values: &ArrayRef, retract: bool) -> DFResult<()> { + fn visit( + acc: &mut SlidingSumIntegerAccumulator, + values: &PrimitiveArray, + retract: bool, + ) -> DFResult<()> { + for value in values.iter().flatten() { + let value = value.to_i64().ok_or_else(|| { + DataFusionError::Internal("Expected an integral SUM input".into()) + })?; + if retract { + acc.remove(value); + } else { + acc.add(value); + } + } + Ok(()) + } + match values.data_type() { + DataType::Int8 => visit(self, as_primitive_array::(values), retract), + DataType::Int16 => visit(self, as_primitive_array::(values), retract), + DataType::Int32 => visit(self, as_primitive_array::(values), retract), + DataType::Int64 => visit(self, as_primitive_array::(values), retract), + other => not_impl_err!("Sliding integer SUM does not support {other}"), + } + } +} + +impl Accumulator for SlidingSumIntegerAccumulator { + fn update_batch(&mut self, values: &[ArrayRef]) -> DFResult<()> { + // DataFusion updates before retracting. Defer overflow checks until + // evaluate(), when the accumulator represents the actual output frame. + self.visit(&values[0], false) + } + + fn retract_batch(&mut self, values: &[ArrayRef]) -> DFResult<()> { + self.visit(&values[0], true)?; + while self.minima.front().is_some_and(|&(i, _)| i <= self.removed) { + self.minima.pop_front(); + } + while self.maxima.front().is_some_and(|&(i, _)| i <= self.removed) { + self.maxima.pop_front(); + } + Ok(()) + } + + fn supports_retract_batch(&self) -> bool { + true + } + + fn evaluate(&mut self) -> DFResult { + if self.added == self.removed { + return Ok(ScalarValue::Int64(None)); + } + let overflow = self + .minima + .front() + .is_some_and(|&(_, v)| v - self.start < i128::from(i64::MIN)) + || self + .maxima + .front() + .is_some_and(|&(_, v)| v - self.start > i128::from(i64::MAX)); + if overflow && self.eval_mode != EvalMode::Legacy { + return match self.eval_mode { + EvalMode::Ansi => Err(arithmetic_overflow_error("integer").into()), + _ => Ok(ScalarValue::Int64(None)), + }; + } + Ok(ScalarValue::Int64(Some((self.end - self.start) as i64))) + } + + fn size(&self) -> usize { + std::mem::size_of_val(self) + + (self.minima.capacity() + self.maxima.capacity()) + * std::mem::size_of::<(usize, i128)>() + } + + fn state(&mut self) -> DFResult> { + not_impl_err!("Sliding integer SUM is a window accumulator") + } + + fn merge_batch(&mut self, _states: &[ArrayRef]) -> DFResult<()> { + not_impl_err!("Sliding integer SUM does not merge partial aggregates") } } @@ -903,6 +1050,107 @@ mod tests { use arrow::array::Int64Array; use datafusion::logical_expr::{EmitTo, GroupsAccumulator}; + #[test] + fn sliding_sum_matches_ordered_checked_addition() { + let choices = [ + None, + Some(0), + Some(1), + Some(-1), + Some(i64::MIN), + Some(i64::MAX), + ]; + // Exhaust every 5-row sequence of nulls, boundaries, and cancellation. + for mut seed in 0..choices.len().pow(5) { + let values: Vec<_> = (0..5) + .map(|_| { + let value = choices[seed % choices.len()]; + seed /= choices.len(); + value + }) + .collect(); + let array: ArrayRef = Arc::new(Int64Array::from(values.clone())); + for width in [1, 2, 3, 5] { + for mode in [EvalMode::Legacy, EvalMode::Try, EvalMode::Ansi] { + let mut acc = SlidingSumIntegerAccumulator::new(mode); + for end in 1..=values.len() { + // Deliberately update first, as DataFusion does. The + // transient union may overflow while both frames fit. + acc.update_batch(&[array.slice(end - 1, 1)]).unwrap(); + let start = end.saturating_sub(width); + if end > width { + acc.retract_batch(&[array.slice(start - 1, 1)]).unwrap(); + } + let mut expected = None; + let mut overflow = false; + for &v in values[start..end].iter().flatten() { + let sum = expected.unwrap_or(0_i64); + overflow |= sum.checked_add(v).is_none(); + expected = Some(sum.wrapping_add(v)); + } + let actual = acc.evaluate(); + if overflow && mode == EvalMode::Ansi { + assert!(actual.is_err(), "{values:?}, {start}..{end}"); + } else { + if overflow && mode == EvalMode::Try { + expected = None; + } + assert_eq!( + actual.unwrap(), + ScalarValue::Int64(expected), + "{values:?}, {start}..{end}, {mode:?}" + ); + } + } + acc.retract_batch(&[array.slice(values.len().saturating_sub(width), width)]) + .unwrap(); + assert_eq!(acc.evaluate().unwrap(), ScalarValue::Int64(None)); + assert!(acc.minima.is_empty() && acc.maxima.is_empty()); + // Empty -> non-empty, with absolute prefix counters retained. + acc.update_batch(&[Arc::new(Int64Array::from(vec![7]))]) + .unwrap(); + assert_eq!(acc.evaluate().unwrap(), ScalarValue::Int64(Some(7))); + } + } + } + } + + #[test] + fn sliding_sum_checks_intermediate_overflow_and_recovers() { + for (values, expected) in [(vec![i64::MAX, 1, -1], 0), (vec![i64::MIN, -1, 1], 0)] { + let array: ArrayRef = Arc::new(Int64Array::from(values)); + let mut acc = SlidingSumIntegerAccumulator::new(EvalMode::Try); + acc.update_batch(&[Arc::clone(&array)]).unwrap(); + assert_eq!(acc.evaluate().unwrap(), ScalarValue::Int64(None)); + acc.retract_batch(&[array.slice(0, 1)]).unwrap(); + assert_eq!(acc.evaluate().unwrap(), ScalarValue::Int64(Some(expected))); + } + } + + #[test] + fn sliding_sum_accepts_all_integral_types() { + for datatype in [ + DataType::Int8, + DataType::Int16, + DataType::Int32, + DataType::Int64, + ] { + let array = arrow::compute::cast( + &Int64Array::from(vec![Some(100), None, Some(-100), Some(7)]), + &datatype, + ) + .unwrap(); + let mut acc = SlidingSumIntegerAccumulator::new(EvalMode::Ansi); + acc.update_batch(&[Arc::clone(&array)]).unwrap(); + acc.retract_batch(&[array.slice(0, 2)]).unwrap(); + assert_eq!(acc.evaluate().unwrap(), ScalarValue::Int64(Some(-93))); + } + for mode in [EvalMode::Ansi, EvalMode::Try] { + let udf = SumInteger::try_new(DataType::Int64, mode).unwrap(); + assert!(matches!(udf.reverse_expr(), ReversedUDAF::NotSupported)); + } + } + fn run_update_batch_with_filter( acc: &mut dyn GroupsAccumulator, values: Vec, diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala index 018207e1f68..4d4abf3d2e0 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala @@ -34,11 +34,9 @@ import com.google.common.base.Objects import org.apache.comet.{CometConf, ConfigEntry} import org.apache.comet.CometSparkSessionExtensions.withFallbackReason -import org.apache.comet.expressions.CometEvalMode import org.apache.comet.serde.{AggSerde, CometOperatorSerde, LiteralOuterClass, OperatorOuterClass} import org.apache.comet.serde.OperatorOuterClass.Operator import org.apache.comet.serde.QueryPlanSerde.{aggExprToProto, exprToProto, scalarFunctionExprToProto, serializeDataType} -import org.apache.comet.shims.CometEvalModeUtil object CometWindowExec extends CometOperatorSerde[WindowExec] { @@ -344,15 +342,6 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] { windowExpr, "SUM on DECIMAL with a sliding window frame is not supported") return None - // The sliding accumulator also wraps integer sums, ignoring ANSI/TRY overflow. - case s: Sum - if s.dataType == LongType && - CometEvalModeUtil.fromSparkEvalMode(CometEvalModeUtil.sumEvalMode(s)) != - CometEvalMode.LEGACY => - withFallbackReason( - windowExpr, - "ANSI/TRY SUM on integral types with a sliding window frame is not supported") - return None case a: Average if a.dataType.isInstanceOf[DecimalType] => withFallbackReason( windowExpr, diff --git a/spark/src/test/resources/sql-tests/windows/sliding_integer_sum.sql b/spark/src/test/resources/sql-tests/windows/sliding_integer_sum.sql index 1f25f233f92..2070fdae71f 100644 --- a/spark/src/test/resources/sql-tests/windows/sliding_integer_sum.sql +++ b/spark/src/test/resources/sql-tests/windows/sliding_integer_sum.sql @@ -30,19 +30,19 @@ INSERT INTO sliding_integer_sum VALUES (2, 1, -9223372036854775808), (2, 2, -1), (2, 3, 1), (2, 4, NULL), (2, 5, NULL), (3, 1, NULL), (3, 2, NULL), (3, 3, NULL) --- TRY mode must fall back even when ANSI is disabled. Cover positive/negative +-- TRY mode runs natively even when ANSI is disabled. Cover positive/negative -- overflow, recovery after it leaves the frame, all-NULL and empty frames. -query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) +query SELECT g, id, try_sum(v) OVER ( PARTITION BY g ORDER BY id RANGE BETWEEN 1 PRECEDING AND CURRENT ROW) FROM sliding_integer_sum -query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) +query SELECT g, id, try_sum(v) OVER ( PARTITION BY g ORDER BY id ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING) FROM sliding_integer_sum -query expect_fallback(ANSI/TRY SUM on integral types with a sliding window frame is not supported) +query SELECT g, id, try_sum(v) OVER ( PARTITION BY g ORDER BY id ROWS BETWEEN 1 FOLLOWING AND 1 FOLLOWING) FROM sliding_integer_sum @@ -81,3 +81,26 @@ FROM sliding_integer_sum query expect_error(ARITHMETIC_OVERFLOW) SELECT sum(v) OVER (ORDER BY id ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) FROM sliding_integer_sum WHERE g = 1 + +-- Updating before retracting must not raise on the transient union of two frames. +query +SELECT g, id, sum(v) OVER ( + PARTITION BY g ORDER BY id ROWS BETWEEN CURRENT ROW AND CURRENT ROW) +FROM sliding_integer_sum + +statement +CREATE TABLE sliding_integer_sum_peers(id INT, v BIGINT) USING parquet + +statement +INSERT INTO sliding_integer_sum_peers VALUES + (1, 9223372036854775807), (1, 1), (3, -1), (6, NULL), (9, 7) + +-- RANGE peers enter together; a gap removes both overflow-causing values. +query +SELECT id, try_sum(v) OVER (ORDER BY id RANGE BETWEEN 1 PRECEDING AND CURRENT ROW) +FROM sliding_integer_sum_peers + +-- Existing unsupported RANGE upper bounds retain their fallback boundary. +query expect_fallback(Unsupported RANGE frame upper offset) +SELECT id, try_sum(v) OVER (ORDER BY id RANGE BETWEEN 2 PRECEDING AND 1 PRECEDING) +FROM sliding_integer_sum_peers diff --git a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala index 7842668b4ab..924c9b1f580 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala @@ -64,16 +64,6 @@ class CometWindowExecSuite extends CometTestBase { assert(cometWindowExecs.nonEmpty) } - private def checkSlidingIntegralSum(df: DataFrame): Unit = { - if (SQLConf.get.ansiEnabled) { - checkSparkAnswerAndFallbackReason( - df, - "ANSI/TRY SUM on integral types with a sliding window frame is not supported") - } else { - checkSparkAnswerAndOperator(df) - } - } - private def sparkWindowExpressions(plan: SparkPlan): Seq[Expression] = { collect(plan) { case w: SparkWindowExec => w.windowExpression @@ -570,11 +560,7 @@ class CometWindowExecSuite extends CometTestBase { s"SELECT $function OVER(order by _2 rows between current row and 1 following) FROM t1") slidingQueries.foreach { query => - if (function == "SUM(_1)") { - checkSlidingIntegralSum(sql(query)) - } else { - checkSparkAnswerAndOperator(query) - } + checkSparkAnswerAndOperator(query) } } } @@ -607,15 +593,19 @@ class CometWindowExecSuite extends CometTestBase { val df = spark.table("sliding_rows_sum").toDF("g", "id", "v") // DataFrame bounds are literals, so PRECEDING needs no constant folding. val frame = Window.partitionBy("g").orderBy("id").rowsBetween(-1, Window.currentRow) - checkSparkAnswerAndFallbackReason( - df.select($"g", $"id", expr("try_sum(v)").over(frame)), - "ANSI/TRY SUM on integral types with a sliding window frame is not supported") + checkSparkAnswerAndOperator(df.select($"g", $"id", expr("try_sum(v)").over(frame))) + val widerFrame = + Window.partitionBy("g").orderBy("id").rowsBetween(-2, Window.currentRow) + withSQLConf(CometConf.COMET_BATCH_SIZE.key -> "2") { + checkSparkAnswerAndOperator( + df.select($"g", $"id", expr("try_sum(v)").over(widerFrame))) + } - // ANSI admission does not depend on whether the data actually overflows. + // Exercise successful ANSI execution separately from the overflow cases. val sumInput = if (SQLConf.get.ansiEnabled) df.where($"id" > 1) else df - checkSlidingIntegralSum(sumInput.select($"g", $"id", sum("v").over(frame))) + checkSparkAnswerAndOperator(sumInput.select($"g", $"id", sum("v").over(frame))) for (dataType <- Seq("tinyint", "smallint", "int")) { - checkSlidingIntegralSum(df.select(sum($"id".cast(dataType)).over(frame))) + checkSparkAnswerAndOperator(df.select(sum($"id".cast(dataType)).over(frame))) } if (SQLConf.get.ansiEnabled) { @@ -915,7 +905,7 @@ class CometWindowExecSuite extends CometTestBase { SUM(c) OVER (PARTITION BY a ORDER BY b, c ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING) as sum_c FROM window_test """) - checkSlidingIntegralSum(df) + checkSparkAnswerAndOperator(df) } } @@ -955,7 +945,7 @@ class CometWindowExecSuite extends CometTestBase { SUM(c) OVER (PARTITION BY a ORDER BY b, c ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) as sum_c FROM window_test """) - checkSlidingIntegralSum(df) + checkSparkAnswerAndOperator(df) } } @@ -1426,7 +1416,7 @@ class CometWindowExecSuite extends CometTestBase { SUM(c) OVER (PARTITION BY a ORDER BY b RANGE BETWEEN 2 PRECEDING AND 2 FOLLOWING) as sum_c FROM window_test """) - checkSlidingIntegralSum(df) + checkSparkAnswerAndOperator(df) } } diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometSlidingSumBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometSlidingSumBenchmark.scala new file mode 100644 index 00000000000..8a44b8c54ef --- /dev/null +++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometSlidingSumBenchmark.scala @@ -0,0 +1,92 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.spark.sql.benchmark + +import org.apache.spark.benchmark.Benchmark +import org.apache.spark.sql.Row +import org.apache.spark.sql.comet.CometWindowExec +import org.apache.spark.sql.internal.SQLConf + +import org.apache.comet.CometConf + +/** Compare native sliding SUM with a Spark window over the same Comet input. */ +object CometSlidingSumBenchmark extends CometBenchmarkBase { + override def runCometBenchmark(mainArgs: Array[String]): Unit = { + val rows = mainArgs.headOption.map(_.toLong).getOrElse(65536L) + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + SQLConf.SHUFFLE_PARTITIONS.key -> "1", + SQLConf.ANSI_ENABLED.key -> "true", + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_SHUFFLE_ENABLED.key -> "true", + "spark.comet.operator.WindowExec.allowIncompatible" -> "true") { + for (nullPercent <- Seq(0, 50, 100)) { + withTempPath { dir => + spark + .range(rows) + .selectExpr( + "id", + s"CASE WHEN pmod(id, 100) < $nullPercent THEN NULL ELSE id % 17 - 8 END AS v") + .write + .parquet(dir.getCanonicalPath) + withTempTable("sliding_sum_benchmark") { + spark.read + .parquet(dir.getCanonicalPath) + .createOrReplaceTempView("sliding_sum_benchmark") + for (width <- Seq(16, 1024); function <- Seq("sum", "try_sum")) { + val query = s"SELECT sum(s), count(s) FROM (SELECT $function(v) OVER " + + s"(ORDER BY id ROWS BETWEEN ${width - 1} PRECEDING AND CURRENT ROW) AS s " + + "FROM sliding_sum_benchmark)" + var expected: Option[Seq[Row]] = None + def run(native: Boolean, verify: Boolean): Unit = { + withSQLConf(CometConf.COMET_EXEC_WINDOW_ENABLED.key -> native.toString) { + val df = spark.sql(query) + if (verify) { + assert( + df.queryExecution.executedPlan.exists( + _.isInstanceOf[CometWindowExec]) == native, + df.queryExecution.executedPlan.toString) + } + val result = df.collect().toSeq + if (verify) { + expected.foreach(answer => assert(result == answer)) + expected = Some(result) + } + } + } + run(native = false, verify = true) + run(native = true, verify = true) + val benchmark = new Benchmark( + s"$function: frame=$width, NULL=$nullPercent%", + rows, + output = output) + benchmark.addCase("Spark window fallback") { _ => + run(native = false, verify = false) + } + benchmark.addCase("Comet native window") { _ => run(native = true, verify = false) } + benchmark.run() + } + } + } + } + } + } +} From fbd8f7221831a118f55506e9655f1dab6a4876a2 Mon Sep 17 00:00:00 2001 From: rich7420 Date: Sun, 4 Oct 2026 01:18:49 +0800 Subject: [PATCH 4/4] perf: avoid unnecessary sliding sum prefix queues --- native/spark-expr/src/agg_funcs/sum_int.rs | 120 +++++++++++++++++- .../benchmark/CometSlidingSumBenchmark.scala | 95 ++++++++++---- 2 files changed, 191 insertions(+), 24 deletions(-) diff --git a/native/spark-expr/src/agg_funcs/sum_int.rs b/native/spark-expr/src/agg_funcs/sum_int.rs index c74480e5c9f..b7c8dcf411c 100644 --- a/native/spark-expr/src/agg_funcs/sum_int.rs +++ b/native/spark-expr/src/agg_funcs/sum_int.rs @@ -120,6 +120,19 @@ impl AggregateUDFImpl for SumInteger { /// every partial sum Spark would visit. Monotonic queues retain just the extrema /// candidates; each entry is pushed/popped once, for amortized O(1) work per row. /// +/// A partial sum is bounded above by the positive sum and below by the negative +/// sum of retained values, including update-before-retract overlap. +/// Only enqueue a prefix when that bound exceeds the corresponding i64 limit. +/// A skipped prefix is safe for every later frame too: moving the left boundary +/// forward only removes values from those bounds. +/// Ordinary values therefore need no queue allocation. +/// +/// Worst-case queue space is still linear in the largest non-null frame, +/// including update-before-retract overlap. A suffix frame can span the entire +/// partition. Each entry occupies 32 bytes on a 64-bit host, and VecDeque retains +/// its grown capacity after retraction. size() reports this capacity, but +/// DataFusion 55.1's window operators do not account for accumulator sizes. +/// /// i128 holds the exact sum of at most usize::MAX i64 values on a 64-bit host. /// Nulls need no entries because retraction receives the original input values. #[derive(Debug)] @@ -127,6 +140,7 @@ struct SlidingSumIntegerAccumulator { eval_mode: EvalMode, end: i128, start: i128, + positive_sum: i128, added: usize, removed: usize, minima: VecDeque<(usize, i128)>, @@ -139,6 +153,7 @@ impl SlidingSumIntegerAccumulator { eval_mode, end: 0, start: 0, + positive_sum: 0, added: 0, removed: 0, minima: VecDeque::new(), @@ -148,6 +163,7 @@ impl SlidingSumIntegerAccumulator { fn add(&mut self, value: i64) { self.end += i128::from(value); + self.positive_sum += i128::from(value.max(0)); self.added += 1; while self.minima.back().is_some_and(|&(_, v)| v >= self.end) { self.minima.pop_back(); @@ -155,12 +171,20 @@ impl SlidingSumIntegerAccumulator { while self.maxima.back().is_some_and(|&(_, v)| v <= self.end) { self.maxima.pop_back(); } - self.minima.push_back((self.added, self.end)); - self.maxima.push_back((self.added, self.end)); + // Keep the dominance pops above even for a skipped prefix. It also + // bounds any older entries it dominates in every future frame. + let negative_sum = self.end - self.start - self.positive_sum; + if negative_sum < i128::from(i64::MIN) { + self.minima.push_back((self.added, self.end)); + } + if self.positive_sum > i128::from(i64::MAX) { + self.maxima.push_back((self.added, self.end)); + } } fn remove(&mut self, value: i64) { self.start += i128::from(value); + self.positive_sum -= i128::from(value.max(0)); self.removed += 1; } @@ -1115,6 +1139,98 @@ mod tests { } } + #[test] + fn sliding_sum_matches_advancing_frames() { + // Cross the bounds by accumulating ordinary values as well as MIN/MAX. + let half = 1_i64 << 62; + let choices = [ + None, + Some(half), + Some(-half), + Some(i64::MAX), + Some(i64::MIN), + ]; + for mut seed in 0..choices.len().pow(5) { + let values: Vec<_> = (0..5) + .map(|_| { + let value = choices[seed % choices.len()]; + seed /= choices.len(); + value + }) + .collect(); + let array: ArrayRef = Arc::new(Int64Array::from(values.clone())); + // Growing prefixes, shrinking suffixes, batch updates/retractions, + // and disjoint frames. + for frames in [ + vec![0..1, 0..2, 0..3, 0..4, 0..5], + vec![0..5, 1..5, 2..5, 3..5, 4..5, 5..5], + vec![0..2, 1..4, 2..5, 4..5, 5..5], + vec![0..1, 3..4, 4..5, 5..5], + ] { + for mode in [EvalMode::Ansi, EvalMode::Try] { + let mut acc = SlidingSumIntegerAccumulator::new(mode); + let mut previous = 0..0; + for frame in &frames { + acc.update_batch(&[array.slice(previous.end, frame.end - previous.end)]) + .unwrap(); + acc.retract_batch(&[ + array.slice(previous.start, frame.start - previous.start) + ]) + .unwrap(); + let expected = values[frame.clone()] + .iter() + .flatten() + .try_fold(None, |sum, &value| { + sum.unwrap_or(0_i64).checked_add(value).map(Some) + }); + match (expected, mode) { + (None, EvalMode::Ansi) => assert!(acc.evaluate().is_err()), + _ => assert_eq!( + acc.evaluate().unwrap(), + ScalarValue::Int64(expected.flatten()), + "{values:?}, {frame:?}, {mode:?}" + ), + } + previous = frame.clone(); + } + } + } + } + } + + #[test] + fn sliding_sum_ordinary_suffixes_need_no_queue_allocation() { + for value in [Some(8), Some(-8), Some(0), None] { + let array: ArrayRef = Arc::new(Int64Array::from(vec![value; 65536])); + let mut acc = SlidingSumIntegerAccumulator::new(EvalMode::Try); + acc.update_batch(&[Arc::clone(&array)]).unwrap(); + for start in (0..65536).step_by(1024) { + assert_eq!( + acc.evaluate().unwrap(), + ScalarValue::Int64(value.map(|v| v * (65536 - start) as i64)) + ); + assert_eq!(acc.size(), std::mem::size_of_val(&acc)); + acc.retract_batch(&[array.slice(start, 1024)]).unwrap(); + } + assert_eq!(acc.evaluate().unwrap(), ScalarValue::Int64(None)); + assert_eq!(acc.size(), std::mem::size_of_val(&acc)); + } + } + + #[test] + fn sliding_sum_exact_bounds_need_no_queue_allocation() { + for value in [i64::MIN, i64::MAX] { + let array: ArrayRef = Arc::new(Int64Array::from(vec![Some(value), None, Some(0)])); + let mut acc = SlidingSumIntegerAccumulator::new(EvalMode::Ansi); + acc.update_batch(&[Arc::clone(&array)]).unwrap(); + assert_eq!(acc.evaluate().unwrap(), ScalarValue::Int64(Some(value))); + assert_eq!(acc.size(), std::mem::size_of_val(&acc)); + acc.retract_batch(&[array.slice(0, 1)]).unwrap(); + assert_eq!(acc.evaluate().unwrap(), ScalarValue::Int64(Some(0))); + assert_eq!(acc.size(), std::mem::size_of_val(&acc)); + } + } + #[test] fn sliding_sum_checks_intermediate_overflow_and_recovers() { for (values, expected) in [(vec![i64::MAX, 1, -1], 0), (vec![i64::MIN, -1, 1], 0)] { diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometSlidingSumBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometSlidingSumBenchmark.scala index 8a44b8c54ef..f533de2c7de 100644 --- a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometSlidingSumBenchmark.scala +++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometSlidingSumBenchmark.scala @@ -26,38 +26,73 @@ import org.apache.spark.sql.internal.SQLConf import org.apache.comet.CometConf -/** Compare native sliding SUM with a Spark window over the same Comet input. */ +/** + * Compare DataFusion's legacy SUM, checked native SUM/TRY_SUM, and Spark windows. Arguments: row + * count (default 65536), frames (all, bounded, or suffix). For example, `4000000 suffix` measures + * a whole-partition initial frame. + */ object CometSlidingSumBenchmark extends CometBenchmarkBase { override def runCometBenchmark(mainArgs: Array[String]): Unit = { val rows = mainArgs.headOption.map(_.toLong).getOrElse(65536L) + val frames = mainArgs.lift(1).getOrElse("all") + require(rows > 0 && Set("all", "bounded", "suffix").contains(frames)) + val boundedFrames = + Seq(16, 1024).map(width => (s"rows=$width", s"${width - 1} PRECEDING AND CURRENT ROW")) + val suffixFrame = Seq(("suffix", "CURRENT ROW AND UNBOUNDED FOLLOWING")) + val selectedFrames = frames match { + case "bounded" => boundedFrames + case "suffix" => suffixFrame + case _ => boundedFrames ++ suffixFrame + } + val nativeCases = Seq( + ("DataFusion SUM (legacy)", "sum", false, true), + ("Comet SUM (ANSI)", "sum", true, true), + ("Comet TRY_SUM", "try_sum", true, true)) + val sparkCases = + Seq(("Spark SUM (ANSI)", "sum", true, false), ("Spark TRY_SUM", "try_sum", true, false)) withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", SQLConf.SHUFFLE_PARTITIONS.key -> "1", - SQLConf.ANSI_ENABLED.key -> "true", CometConf.COMET_ENABLED.key -> "true", CometConf.COMET_EXEC_ENABLED.key -> "true", CometConf.COMET_SHUFFLE_ENABLED.key -> "true", "spark.comet.operator.WindowExec.allowIncompatible" -> "true") { - for (nullPercent <- Seq(0, 50, 100)) { + for ((shape, value) <- Seq( + "cancelling" -> "id % 17 - 8", + "positive" -> "id % 17", + // Partial sums fit in i64, but both sign totals cross the bounds. + "near-limit" -> + "CASE WHEN id % 2 = 0 THEN 4611686018427387904L ELSE -4611686018427387903L END"); + nullPercent <- (if (shape == "cancelling") Seq(0, 50, 100) else Seq(0, 50))) { withTempPath { dir => spark .range(rows) .selectExpr( "id", - s"CASE WHEN pmod(id, 100) < $nullPercent THEN NULL ELSE id % 17 - 8 END AS v") + s"CASE WHEN pmod(id, 100) < $nullPercent THEN NULL ELSE $value END AS v") .write .parquet(dir.getCanonicalPath) withTempTable("sliding_sum_benchmark") { spark.read .parquet(dir.getCanonicalPath) .createOrReplaceTempView("sliding_sum_benchmark") - for (width <- Seq(16, 1024); function <- Seq("sum", "try_sum")) { - val query = s"SELECT sum(s), count(s) FROM (SELECT $function(v) OVER " + - s"(ORDER BY id ROWS BETWEEN ${width - 1} PRECEDING AND CURRENT ROW) AS s " + - "FROM sliding_sum_benchmark)" - var expected: Option[Seq[Row]] = None - def run(native: Boolean, verify: Boolean): Unit = { - withSQLConf(CometConf.COMET_EXEC_WINDOW_ENABLED.key -> native.toString) { + for ((frameName, bounds) <- selectedFrames) { + def run( + function: String, + ansi: Boolean, + native: Boolean, + inputRows: Long, + verify: Boolean): Seq[Row] = { + // Spark 3.5's withSQLConf returns Unit. + var result = Seq.empty[Row] + withSQLConf( + SQLConf.ANSI_ENABLED.key -> ansi.toString, + CometConf.COMET_EXEC_WINDOW_ENABLED.key -> native.toString) { + // Consume the window values with a sink unaffected by ANSI mode. + val filter = if (inputRows < rows) s" WHERE id < $inputRows" else "" + val query = s"SELECT bit_xor(s), count(s) FROM (SELECT $function(v) OVER " + + s"(ORDER BY id ROWS BETWEEN $bounds) AS s " + + s"FROM sliding_sum_benchmark$filter)" val df = spark.sql(query) if (verify) { assert( @@ -65,23 +100,39 @@ object CometSlidingSumBenchmark extends CometBenchmarkBase { _.isInstanceOf[CometWindowExec]) == native, df.queryExecution.executedPlan.toString) } - val result = df.collect().toSeq - if (verify) { - expected.foreach(answer => assert(result == answer)) - expected = Some(result) - } + result = df.collect().toSeq } + result } - run(native = false, verify = true) - run(native = true, verify = true) + // Spark recomputes suffix frames in O(rows^2). Check a small partition + // against Spark, then compare all native cases on the full partition. + val verificationRows = if (frameName == "suffix") math.min(rows, 1024L) else rows + val expected = run( + "sum", + ansi = true, + native = false, + inputRows = verificationRows, + verify = true) + for ((_, function, ansi, native) <- nativeCases ++ sparkCases.tail) { + assert(run(function, ansi, native, verificationRows, verify = true) == expected) + } + if (verificationRows != rows) { + val legacy = + run("sum", ansi = false, native = true, inputRows = rows, verify = true) + for ((_, function, ansi, native) <- nativeCases.tail) { + assert(run(function, ansi, native, rows, verify = true) == legacy) + } + } + val cases = if (frameName == "suffix") nativeCases else nativeCases ++ sparkCases val benchmark = new Benchmark( - s"$function: frame=$width, NULL=$nullPercent%", + s"$shape: frame=$frameName, NULL=$nullPercent%", rows, output = output) - benchmark.addCase("Spark window fallback") { _ => - run(native = false, verify = false) + for ((name, function, ansi, native) <- cases) { + benchmark.addCase(name) { _ => + val _ = run(function, ansi, native, rows, verify = false) + } } - benchmark.addCase("Comet native window") { _ => run(native = true, verify = false) } benchmark.run() } }