diff --git a/docs/source/contributor-guide/roadmap.md b/docs/source/contributor-guide/roadmap.md index b8ae8a310f..9121456df8 100644 --- a/docs/source/contributor-guide/roadmap.md +++ b/docs/source/contributor-guide/roadmap.md @@ -28,8 +28,8 @@ Native window execution runs by default (`spark.comet.exec.window.enabled`). The `dense_rank`, `row_number`, `percent_rank`, `cume_dist`, `ntile`), value functions (`lag`, `lead`, `nth_value`, `first_value`, `last_value`), and the `count`, `min`, `max`, `sum`, and `avg` aggregates are accelerated. Remaining work is to close the gaps that still fall back to Spark: statistical aggregates (`stddev`, `variance`, -`corr`, `covar`) and `collect_list` / `collect_set` as window functions ([#4766]), `GROUPS` frames ([#4836]), `RANGE` frames with explicit date or -decimal offsets ([#4834]), `first_value` / `last_value` on `RANGE` frames with a literal offset ([#4835]), +`corr`, `covar`) and `collect_list` / `collect_set` as window functions ([#4766]), `GROUPS` frames ([#4836]), `RANGE` frames with explicit date +offsets ([#4834]), `first_value` / `last_value` on `RANGE` frames with a literal offset ([#4835]), non-literal `lag` / `lead` default values ([#4268]), and `WindowGroupLimitExec` ([#4837]). See the [window function compatibility guide](../user-guide/latest/compatibility/operators.md) for the complete list of supported functions, frames, and fallback cases. diff --git a/docs/source/user-guide/latest/compatibility/operators.md b/docs/source/user-guide/latest/compatibility/operators.md index eec21d283b..22c4e3b93a 100644 --- a/docs/source/user-guide/latest/compatibility/operators.md +++ b/docs/source/user-guide/latest/compatibility/operators.md @@ -45,7 +45,7 @@ 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`. -- `RANGE` frame with an explicit offset when the `ORDER BY` column is `DATE` or `DECIMAL` +- `RANGE` frame with an explicit offset when the `ORDER BY` column is `DATE` ([#4834](https://github.com/apache/datafusion-comet/issues/4834)). - `first_value` / `last_value` on a `RANGE` frame with a literal offset ([#4835](https://github.com/apache/datafusion-comet/issues/4835)). @@ -55,6 +55,13 @@ incorrect result. When any single window expression in a `WindowExec` falls back window are not supported by Spark either. - Any `PARTITION BY` or `ORDER BY` expression that Comet cannot serialize. +**Known differences (runs natively but may diverge from Spark on edge cases):** + +- `RANGE` frame boundary arithmetic on `DECIMAL` near max precision: when `current +/- offset` overflows the + underlying integer, DataFusion collapses the bound to the partition edge, whereas Spark evaluates the boundary + through normal decimal arithmetic (returning `NULL` under non-ANSI, throwing under ANSI). The two behaviors + agree for values away from `Decimal128` limits. + `WindowGroupLimitExec` (window-based limit pushdown) is not yet supported and falls back to Spark ([#4837](https://github.com/apache/datafusion-comet/issues/4837)). 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 500f9996b9..4809a43245 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 @@ -363,17 +363,12 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] { } } - // Comet's native window planner ships RANGE frame offsets as - // ScalarValue::Int64, but a couple of ORDER BY types don't tolerate that: - // - DATE: arrow-arith requires an Interval RHS for Date32 arithmetic, - // so execution fails with - // Invalid date arithmetic operation: Date32 + Int32 - // - DECIMAL: Spark decimal arithmetic widens precision on +/-, so the - // computed boundary (e.g. Decimal(11,0)) doesn't match the current - // value's precision (e.g. Decimal(10,0)) and the comparator fails - // with "Uncomparable values". - // Fall back to Spark for those shapes. UNBOUNDED / CURRENT ROW bounds - // don't evaluate the offending arithmetic and stay native. + // The RANGE frame offset type follows Spark's WindowFrameTypeCoercion: + // DECIMAL ORDER BY casts the offset to the same Decimal type (handled + // natively), but DATE keeps the offset as IntegerType. arrow-arith + // rejects Date32 + Int32 (it requires an Interval RHS), so DATE ORDER BY + // falls back with "Invalid date arithmetic operation: Date32 + Int32". + // UNBOUNDED / CURRENT ROW bounds skip the arithmetic and stay native. f match { case SpecifiedWindowFrame(RangeFrame, lb, ub) if !lb.isInstanceOf[SpecialFrameBoundary] || @@ -384,11 +379,6 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] { windowExpr, "RANGE frame with explicit offset on DATE ORDER BY is not supported") return None - case Some(_: DecimalType) => - withFallbackReason( - windowExpr, - "RANGE frame with explicit offset on DECIMAL ORDER BY is not supported") - return None case _ => } diff --git a/spark/src/test/resources/sql-tests/windows/window_functions.sql b/spark/src/test/resources/sql-tests/windows/window_functions.sql index d8fb8a4df2..89aeac284f 100644 --- a/spark/src/test/resources/sql-tests/windows/window_functions.sql +++ b/spark/src/test/resources/sql-tests/windows/window_functions.sql @@ -253,16 +253,14 @@ ORDER BY cate, val_date -- Ported from Spark's typeCoercion/native/windowFrameCoercion.sql: -- SELECT COUNT(*) OVER (PARTITION BY 1 ORDER BY cast(1 as decimal(10, 0)) DESC -- RANGE BETWEEN CURRENT ROW AND 1 FOLLOWING) FROM t --- Comet falls back because Spark decimal arithmetic widens precision on --- +/-, so the native frame-boundary arithmetic produces e.g. Decimal(11,0) --- while the current row stays Decimal(10,0), and DataFusion's comparator --- fails with "Uncomparable values". Once the native planner preserves the --- ORDER BY column's precision when computing RANGE boundaries, the guard --- in CometWindowExec can be removed and this test will start failing --- because Comet stops falling back — that's the signal to re-enable it. +-- Runs natively: Spark's WindowFrameTypeCoercion casts the frame offset to +-- the same Decimal type as the ORDER BY column, so the boundary arithmetic +-- keeps the current row's precision and DataFusion's comparator has matching +-- operands. Contrast with the DATE case in 1.11, which still falls back +-- because the offset stays IntegerType and arrow-arith rejects Date32 + Int32. -- ============================================================ -query expect_fallback(RANGE frame with explicit offset on DECIMAL ORDER BY is not supported) +query SELECT COUNT(*) OVER (PARTITION BY 1 ORDER BY CAST(1 AS DECIMAL(10, 0)) DESC RANGE BETWEEN CURRENT ROW AND 1 FOLLOWING) AS c 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 0e6fcf8946..2e68d9a102 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala @@ -1274,4 +1274,259 @@ class CometWindowExecSuite extends CometTestBase { checkSparkAnswerAndOperator(df) } } + + // --------------------------------------------------------------------------- + // Tests for issue #4834: RANGE frame with explicit PRECEDING/FOLLOWING offset + // on DECIMAL ORDER BY. Comet previously fell back to Spark because DataFusion + // rejected boundary arithmetic that widened precision (e.g. Decimal(10,0) + + // offset -> Decimal(11,0) vs current Decimal(10,0), "Uncomparable values"). + // The upstream fix (apache/datafusion#22174, closes #22113) makes + // ScalarValue::partial_cmp ignore precision for Decimal128 when scales match, + // and is included in DataFusion 54.0.0 which Comet already pins. These tests + // pin the behavior with `checkSparkAnswerAndOperator` so results must match + // Spark AND run natively. + // --------------------------------------------------------------------------- + + private def writeDecimalWindowFixture(dir: java.io.File, decimalType: String): Unit = { + (0 until 30) + .map { i => + // Fractional part rotates 0.00/0.25/0.50/0.75 so RANGE peers vary; c + // stays as the aggregate payload. + val whole = i % 10 + val frac = (i % 4) * 25 + (i % 3, f"$whole%d.$frac%02d", i * 2) + } + .toDF("a", "d_str", "c") + .selectExpr("a", s"CAST(d_str AS $decimalType) AS d", "c") + .repartition(3) + .write + .mode("overwrite") + .parquet(dir.toString) + } + + test("window: RANGE BETWEEN N PRECEDING AND N FOLLOWING with DECIMAL ORDER BY") { + withTempDir { dir => + writeDecimalWindowFixture(dir, "DECIMAL(10, 2)") + spark.read.parquet(dir.toString).createOrReplaceTempView("window_test_dec") + val df = sql(""" + SELECT a, d, c, + SUM(c) OVER (PARTITION BY a ORDER BY d + RANGE BETWEEN 2 PRECEDING AND 2 FOLLOWING) as sum_c + FROM window_test_dec + """) + checkSparkAnswerAndOperator(df) + } + } + + test("window: RANGE BETWEEN N PRECEDING AND CURRENT ROW with DECIMAL ORDER BY") { + withTempDir { dir => + writeDecimalWindowFixture(dir, "DECIMAL(10, 2)") + spark.read.parquet(dir.toString).createOrReplaceTempView("window_test_dec") + val df = sql(""" + SELECT a, d, c, + SUM(c) OVER (PARTITION BY a ORDER BY d + RANGE BETWEEN 3 PRECEDING AND CURRENT ROW) as sum_c + FROM window_test_dec + """) + checkSparkAnswerAndOperator(df) + } + } + + test("window: RANGE BETWEEN CURRENT ROW AND N FOLLOWING with DECIMAL ORDER BY") { + withTempDir { dir => + writeDecimalWindowFixture(dir, "DECIMAL(10, 2)") + spark.read.parquet(dir.toString).createOrReplaceTempView("window_test_dec") + val df = sql(""" + SELECT a, d, c, + SUM(c) OVER (PARTITION BY a ORDER BY d + RANGE BETWEEN CURRENT ROW AND 3 FOLLOWING) as sum_c + FROM window_test_dec + """) + checkSparkAnswerAndOperator(df) + } + } + + test("window: RANGE BETWEEN UNBOUNDED PRECEDING AND N FOLLOWING with DECIMAL ORDER BY") { + withTempDir { dir => + writeDecimalWindowFixture(dir, "DECIMAL(10, 2)") + spark.read.parquet(dir.toString).createOrReplaceTempView("window_test_dec") + val df = sql(""" + SELECT a, d, c, + SUM(c) OVER (PARTITION BY a ORDER BY d + RANGE BETWEEN UNBOUNDED PRECEDING AND 2 FOLLOWING) as sum_c + FROM window_test_dec + """) + checkSparkAnswerAndOperator(df) + } + } + + test("window: RANGE BETWEEN with DECIMAL ORDER BY and duplicate values") { + // RANGE (not ROWS) must aggregate all peer rows that share the same value. + // Duplicates let us tell ROWS from RANGE and expose peer-boundary bugs. + withTempDir { dir => + Seq( + (1, "1.00", 10), + (1, "1.00", 20), + (1, "2.00", 30), + (1, "2.00", 40), + (1, "5.00", 50), + (1, "5.00", 60)) + .toDF("a", "d_str", "c") + .selectExpr("a", "CAST(d_str AS DECIMAL(10, 2)) AS d", "c") + .write + .mode("overwrite") + .parquet(dir.toString) + + spark.read.parquet(dir.toString).createOrReplaceTempView("window_test_dec") + val df = sql(""" + SELECT a, d, c, + SUM(c) OVER (PARTITION BY a ORDER BY d + RANGE BETWEEN 1 PRECEDING AND 1 FOLLOWING) as sum_c + FROM window_test_dec + """) + checkSparkAnswerAndOperator(df) + } + } + + test("window: RANGE BETWEEN with DECIMAL ORDER BY and NULL values") { + // Null decimals land at the head under ASC NULLS FIRST (Spark's default). + // Ensures the native comparator handles Decimal nulls the same way. + withTempDir { dir => + Seq( + (1, Option.empty[String], 10), + (1, Option("1.00"), 20), + (1, Option("2.00"), 30), + (1, Option.empty[String], 40), + (1, Option("5.00"), 50)) + .toDF("a", "d_str", "c") + .selectExpr("a", "CAST(d_str AS DECIMAL(10, 2)) AS d", "c") + .write + .mode("overwrite") + .parquet(dir.toString) + + spark.read.parquet(dir.toString).createOrReplaceTempView("window_test_dec") + val df = sql(""" + SELECT a, d, c, + SUM(c) OVER (PARTITION BY a ORDER BY d + RANGE BETWEEN 1 PRECEDING AND 1 FOLLOWING) as sum_c + FROM window_test_dec + """) + checkSparkAnswerAndOperator(df) + } + } + + test("window: RANGE BETWEEN with DECIMAL ORDER BY DESC") { + // Descending order swaps the direction of `current +/- offset` in + // calculate_index_of_row; a bug specific to DESC would only show here. + withTempDir { dir => + writeDecimalWindowFixture(dir, "DECIMAL(10, 2)") + spark.read.parquet(dir.toString).createOrReplaceTempView("window_test_dec") + val df = sql(""" + SELECT a, d, c, + SUM(c) OVER (PARTITION BY a ORDER BY d DESC + RANGE BETWEEN 2 PRECEDING AND 2 FOLLOWING) as sum_c + FROM window_test_dec + """) + checkSparkAnswerAndOperator(df) + } + } + + test("window: RANGE BETWEEN with DECIMAL(10, 0) ORDER BY (zero scale)") { + withTempDir { dir => + writeDecimalWindowFixture(dir, "DECIMAL(10, 0)") + spark.read.parquet(dir.toString).createOrReplaceTempView("window_test_dec") + val df = sql(""" + SELECT a, d, c, + SUM(c) OVER (PARTITION BY a ORDER BY d + RANGE BETWEEN 1 PRECEDING AND 1 FOLLOWING) as sum_c + FROM window_test_dec + """) + checkSparkAnswerAndOperator(df) + } + } + + test("window: RANGE BETWEEN with DECIMAL(18, 6) ORDER BY (high precision & scale)") { + withTempDir { dir => + writeDecimalWindowFixture(dir, "DECIMAL(18, 6)") + spark.read.parquet(dir.toString).createOrReplaceTempView("window_test_dec") + val df = sql(""" + SELECT a, d, c, + SUM(c) OVER (PARTITION BY a ORDER BY d + RANGE BETWEEN 2 PRECEDING AND 2 FOLLOWING) as sum_c + FROM window_test_dec + """) + checkSparkAnswerAndOperator(df) + } + } + + test("window: RANGE BETWEEN with negative DECIMAL ORDER BY values") { + // Negative current + positive offset crosses zero, which is a common + // arithmetic edge for decimal signed comparisons. + withTempDir { dir => + Seq( + (1, "-5.50", 10), + (1, "-3.00", 20), + (1, "-1.25", 30), + (1, "0.00", 40), + (1, "1.25", 50), + (1, "3.00", 60), + (1, "5.50", 70)) + .toDF("a", "d_str", "c") + .selectExpr("a", "CAST(d_str AS DECIMAL(10, 2)) AS d", "c") + .write + .mode("overwrite") + .parquet(dir.toString) + + spark.read.parquet(dir.toString).createOrReplaceTempView("window_test_dec") + val df = sql(""" + SELECT a, d, c, + SUM(c) OVER (PARTITION BY a ORDER BY d + RANGE BETWEEN 2 PRECEDING AND 2 FOLLOWING) as sum_c + FROM window_test_dec + """) + checkSparkAnswerAndOperator(df) + } + } + + test("window: RANGE BETWEEN with fractional offset on DECIMAL ORDER BY") { + // Fractional offsets (0.5 PRECEDING) exercise scale-aware boundary + // arithmetic. Spark accepts decimal literals as offsets when the ORDER BY + // column is decimal. + withTempDir { dir => + writeDecimalWindowFixture(dir, "DECIMAL(10, 2)") + spark.read.parquet(dir.toString).createOrReplaceTempView("window_test_dec") + val df = sql(""" + SELECT a, d, c, + SUM(c) OVER (PARTITION BY a ORDER BY d + RANGE BETWEEN 0.5 PRECEDING AND 0.5 FOLLOWING) as sum_c + FROM window_test_dec + """) + checkSparkAnswerAndOperator(df) + } + } + + test("window: RANGE BETWEEN with DECIMAL(38, 0) ORDER BY (max precision)") { + // Max-precision Decimal storage. Values are kept well below Decimal128 + // max so `current +/- offset` never overflows; that keeps the test on + // topic (native RANGE frame arithmetic on the widest decimal precision) + // and off the overflow-semantics divergence between Spark ANSI and + // DataFusion's `add_checked`. + withTempDir { dir => + Seq((1, "10", 10), (1, "13", 20), (1, "15", 30), (1, "18", 40), (1, "22", 50)) + .toDF("a", "d_str", "c") + .selectExpr("a", "CAST(d_str AS DECIMAL(38, 0)) AS d", "c") + .write + .mode("overwrite") + .parquet(dir.toString) + + spark.read.parquet(dir.toString).createOrReplaceTempView("window_test_dec") + val df = sql(""" + SELECT a, d, c, + SUM(c) OVER (PARTITION BY a ORDER BY d + RANGE BETWEEN 3 PRECEDING AND 3 FOLLOWING) as sum_c + FROM window_test_dec + """) + checkSparkAnswerAndOperator(df) + } + } }