feat: native RANGE window frames with explicit offset on " DECIMAL " ORDER BY

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.

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)).
 

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 _ =>
         }
 

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

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)
+    }
+  }
 }