You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/16 17:09:58 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-4974-c3316fbe39e597f311dd999efc012616a71d966e created (now 86d2d78257)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-4974-c3316fbe39e597f311dd999efc012616a71d966e
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at 86d2d78257 feat: native RANGE window frames with explicit offset on DATE ORDER BY (#4974)
This branch includes the following new commits:
new 86d2d78257 feat: native RANGE window frames with explicit offset on DATE ORDER BY (#4974)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: feat: native RANGE window frames with explicit offset on DATE ORDER BY (#4974)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-4974-c3316fbe39e597f311dd999efc012616a71d966e
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 86d2d78257df81d3e2443dab194405bef8a5c4ca
Author: ChenChen Lai <72...@users.noreply.github.com>
AuthorDate: Fri Oct 16 14:52:41 2026 +0000
feat: native RANGE window frames with explicit offset on DATE ORDER BY (#4974)
* feat: native RANGE window frames with explicit offset on DATE ORDER BY
* fix fmt
* update documentation
* Modified SQL test
* fix: fall back for DATE RANGE offsets that can wrap an Int
---
docs/source/contributor-guide/roadmap.md | 2 +-
.../user-guide/latest/compatibility/operators.md | 11 +-
native/core/src/execution/planner.rs | 58 +++++-
.../apache/spark/sql/comet/CometWindowExec.scala | 57 ++++--
.../sql-tests/windows/window_functions.sql | 11 +-
.../apache/comet/exec/CometWindowExecSuite.scala | 208 +++++++++++++++++++++
6 files changed, 320 insertions(+), 27 deletions(-)
diff --git a/docs/source/contributor-guide/roadmap.md b/docs/source/contributor-guide/roadmap.md
index 7913230f10..06bff3435d 100644
--- a/docs/source/contributor-guide/roadmap.md
+++ b/docs/source/contributor-guide/roadmap.md
@@ -28,7 +28,7 @@ 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]), `RANGE` frames with explicit date or
+`corr`, `covar`) and `collect_list` / `collect_set` as window functions ([#4766]), `RANGE` frames with explicit
decimal offsets ([#4834]), `first_value` / `last_value` on `RANGE` frames with a literal offset ([#4835]), and
non-literal `lag` / `lead` default values ([#4268]). See the
[window function compatibility guide](../user-guide/latest/compatibility/operators.md) for the complete list of
diff --git a/docs/source/user-guide/latest/compatibility/operators.md b/docs/source/user-guide/latest/compatibility/operators.md
index 75c2210ef8..1b9a61bf7f 100644
--- a/docs/source/user-guide/latest/compatibility/operators.md
+++ b/docs/source/user-guide/latest/compatibility/operators.md
@@ -142,8 +142,12 @@ incorrect result. When any single window expression in a `WindowExec` falls back
([#7007](https://github.com/apache/datafusion-comet/issues/7007)). From precision 28 the native average also
differs from Spark once the running sum overflows precision 38
([#5418](https://github.com/apache/datafusion-comet/issues/5418)).
-- `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 `DECIMAL`
([#4834](https://github.com/apache/datafusion-comet/issues/4834)).
+- `RANGE` frame with an offset above 2144550751 days when the `ORDER BY` column is `DATE`. Spark adds the offset
+ to the date as an `Int` that wraps on overflow, while the native frame treats an overflowing bound as the edge of
+ the partition. Smaller offsets cannot wrap for any date between 0001-01-01 and 9999-12-31
+ ([#5022](https://github.com/apache/datafusion-comet/issues/5022)).
- `RANGE` frame bounded by `CURRENT ROW` when an `ORDER BY` key is an array of arrays or structs, or a struct
holding an array, such as `array(named_struct('x', x))`. DataFusion cannot compare those values to find the
frame's bounds ([apache/datafusion#24937](https://github.com/apache/datafusion/issues/24937)). Ranking functions
@@ -159,6 +163,11 @@ incorrect result. When any single window expression in a `WindowExec` falls back
- A `ROWS` offset that is not an integer or long, or a `RANGE` offset that is not numeric.
- Any `PARTITION BY` or `ORDER BY` expression that Comet cannot serialize.
+A `RANGE` frame with an explicit offset on a `DATE` `ORDER BY` runs natively, but over a date outside Spark's
+supported range of 0001-01-01 to 9999-12-31, such as one produced by `date_add` with a very large day count, the
+native frame can treat the bound as the edge of the partition where Spark compares against it
+([#5022](https://github.com/apache/datafusion-comet/issues/5022)).
+
`WindowGroupLimitExec` (window-based limit pushdown for `ROW_NUMBER`, `RANK`, and `DENSE_RANK`)
runs natively; it is controlled by `spark.comet.exec.windowGroupLimit.enabled` (default: true).
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index d871c941dd..0c9bf36e1e 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -53,7 +53,8 @@ use crate::execution::{
use crate::jvm_bridge::{jni_call, JVMClasses, ShufflePartitionPusher};
use arrow::compute::CastOptions;
use arrow::datatypes::{
- DataType, Field, FieldRef, Fields, IntervalUnit, Schema, TimeUnit, DECIMAL128_MAX_PRECISION,
+ DataType, Field, FieldRef, Fields, IntervalDayTime, IntervalUnit, Schema, TimeUnit,
+ DECIMAL128_MAX_PRECISION,
};
use arrow::ffi_stream::{ArrowArrayStreamReader, FFI_ArrowArrayStream};
use datafusion::functions_aggregate::bit_and_or_xor::{bit_and_udaf, bit_or_udaf, bit_xor_udaf};
@@ -3780,6 +3781,20 @@ impl PhysicalPlanner {
WindowFrameType::Range => WindowFrameUnits::Range,
};
+ // For RANGE frames the offset scalar must be arithmetic-compatible with
+ // the ORDER BY column, not just numeric. arrow-arith rejects e.g.
+ // `Date32 + Int32` — it requires an Interval RHS. Capture the ORDER BY
+ // datatype once so both bound branches can coerce accordingly.
+ let range_order_by_type = if matches!(units, WindowFrameUnits::Range) {
+ sort_exprs
+ .first()
+ .map(|s| s.expr.data_type(&input_schema))
+ .transpose()
+ .map_err(|e| GeneralError(format!("Failed to resolve ORDER BY type: {e}")))?
+ } else {
+ None
+ };
+
let lower_bound: WindowFrameBound = match spark_window_frame
.lower_bound
.as_ref()
@@ -3815,10 +3830,14 @@ impl PhysicalPlanner {
}
}
WindowFrameUnits::Range => {
- let scalar = match offset.range_offset.as_ref() {
+ let raw = match offset.range_offset.as_ref() {
Some(lit) => numeric_literal_to_scalar(lit)?,
None => ScalarValue::Int64(Some(offset.offset.abs())),
};
+ let scalar = coerce_range_offset_for_order_by(
+ raw,
+ range_order_by_type.as_ref(),
+ )?;
WindowFrameBound::Preceding(scalar)
}
WindowFrameUnits::Groups => {
@@ -3874,10 +3893,12 @@ impl PhysicalPlanner {
}
}
WindowFrameUnits::Range => {
- let scalar = match offset.range_offset.as_ref() {
+ let raw = match offset.range_offset.as_ref() {
Some(lit) => numeric_literal_to_scalar(lit)?,
None => ScalarValue::Int64(Some(offset.offset)),
};
+ let scalar =
+ coerce_range_offset_for_order_by(raw, range_order_by_type.as_ref())?;
WindowFrameBound::Following(scalar)
}
WindowFrameUnits::Groups => {
@@ -4418,6 +4439,37 @@ fn numeric_literal_to_scalar(
Ok(scalar)
}
+/// arrow-arith only provides `Date32 + Interval` kernels, not `Date32 + Int32`.
+/// Spark ships `n PRECEDING/FOLLOWING` as an IntegerType literal, so a DATE
+/// ORDER BY frame would blow up at execution with
+/// "Invalid date arithmetic operation: Date32 + Int32". Wrap integer offsets
+/// as IntervalDayTime (days component) so the frame comparator can compute
+/// `current_date +/- offset` natively.
+fn coerce_range_offset_for_order_by(
+ scalar: ScalarValue,
+ order_by_type: Option<&DataType>,
+) -> Result<ScalarValue, ExecutionError> {
+ let Some(dt) = order_by_type else {
+ return Ok(scalar);
+ };
+ match (dt, &scalar) {
+ (DataType::Date32, ScalarValue::Int32(Some(v))) => Ok(ScalarValue::IntervalDayTime(Some(
+ IntervalDayTime::new(*v, 0),
+ ))),
+ (DataType::Date32, ScalarValue::Int64(Some(v))) => {
+ let days = i32::try_from(*v).map_err(|_| {
+ GeneralError(format!(
+ "DATE RANGE frame offset {v} does not fit in i32 days"
+ ))
+ })?;
+ Ok(ScalarValue::IntervalDayTime(Some(IntervalDayTime::new(
+ days, 0,
+ ))))
+ }
+ _ => Ok(scalar),
+ }
+}
+
/// A physical join filter rewritter which rewrites the column indices in the expression
/// to use the new column indices. See `rewrite_physical_expr`.
struct JoinFilterRewriter<'a> {
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 464b7c1d2e..33c2da2117 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
@@ -394,30 +394,35 @@ 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.
+ // ScalarValue::Int64, which the DECIMAL comparator can't handle: 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 for DECIMAL until the native side normalizes
+ // precision. DATE is now handled natively by wrapping the offset as
+ // IntervalDayTime in the planner. UNBOUNDED / CURRENT ROW bounds don't
+ // evaluate the offending arithmetic and stay native regardless.
+ //
+ // For DATE, Spark computes the bound with DateAdd, an Int addition that
+ // wraps silently on overflow, while DataFusion treats an overflowing bound
+ // as the partition edge. Over dates in Spark's supported range the two only
+ // disagree when the offset is large enough to wrap, so fall back for those
+ // offsets. https://github.com/apache/datafusion-comet/issues/5022
f match {
case SpecifiedWindowFrame(RangeFrame, lb, ub)
if !lb.isInstanceOf[SpecialFrameBoundary] ||
!ub.isInstanceOf[SpecialFrameBoundary] =>
windowExpr.windowSpec.orderSpec.headOption.map(_.dataType) match {
- case Some(DateType) =>
+ case Some(_: DecimalType) =>
withFallbackReason(
windowExpr,
- "RANGE frame with explicit offset on DATE ORDER BY is not supported")
+ "RANGE frame with explicit offset on DECIMAL ORDER BY is not supported")
return None
- case Some(_: DecimalType) =>
+ case Some(DateType) if !Seq(lb, ub).forall(isDateRangeOffsetSafe) =>
withFallbackReason(
windowExpr,
- "RANGE frame with explicit offset on DECIMAL ORDER BY is not supported")
+ "RANGE frame on DATE ORDER BY with an offset above " +
+ s"$MaxSafeDateRangeOffset days is not supported")
return None
case _ =>
}
@@ -671,6 +676,30 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] {
case _ => true
}
+ // The largest absolute day number of a date in Spark's supported range,
+ // 0001-01-01 to 9999-12-31. 9999-12-31 is further from the epoch.
+ private val MaxSupportedDateDays: Long = java.time.LocalDate.of(9999, 12, 31).toEpochDay
+
+ // The largest DATE RANGE offset for which `current +/- offset` cannot wrap
+ // an Int for any date in Spark's supported range.
+ private val MaxSafeDateRangeOffset: Long = Int.MaxValue.toLong - MaxSupportedDateDays
+
+ private def isDateRangeOffsetSafe(bound: Expression): Boolean = bound match {
+ case _: SpecialFrameBoundary => true
+ case e =>
+ try {
+ e.eval() match {
+ case i: java.lang.Integer => Math.abs(i.longValue()) <= MaxSafeDateRangeOffset
+ case l: java.lang.Long =>
+ l.longValue() != Long.MinValue && Math.abs(l.longValue()) <= MaxSafeDateRangeOffset
+ // Other offsets are rejected when the bound is serialized.
+ case _ => true
+ }
+ } catch {
+ case _: Exception => true
+ }
+ }
+
// Folds a RANGE frame bound expression to a constant and serializes its
// magnitude as a typed Literal proto. Spark encodes PRECEDING/FOLLOWING via
// the sign of the literal (negative => PRECEDING, positive => FOLLOWING),
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 998b6dfa5b..02fcbf6c82 100644
--- a/spark/src/test/resources/sql-tests/windows/window_functions.sql
+++ b/spark/src/test/resources/sql-tests/windows/window_functions.sql
@@ -232,16 +232,11 @@ FROM emp
-- N days, so MAX(val_date) OVER (... RANGE BETWEEN CURRENT ROW AND 2 FOLLOWING)
-- returns the largest val_date within the next 2 days of each row.
-- Mirrors the "max(val_date) ... RANGE BETWEEN CURRENT ROW AND 2 FOLLOWING"
--- case from Spark's window.sql. Comet falls back because its native window
--- planner ships RANGE frame offsets as Int64 while arrow-arith requires an
--- Interval RHS for Date32 arithmetic. Once the native planner coerces Int ->
--- Interval for DATE sort keys (as DataFusion's type-coercion analyzer does),
--- 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
--- native execution.
+-- case from Spark's window.sql. The native planner coerces the Int offset to
+-- IntervalDayTime for Date32 sort keys so this runs natively.
-- ============================================================
-query expect_fallback(RANGE frame with explicit offset on DATE ORDER BY is not supported)
+query
SELECT val_date, cate,
MAX(val_date) OVER (PARTITION BY cate ORDER BY val_date
RANGE BETWEEN CURRENT ROW AND 2 FOLLOWING) AS mx_d
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 4a0174a11e..cf881d42c4 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala
@@ -1644,4 +1644,212 @@ class CometWindowExecSuite extends CometTestBase {
checkSparkAnswerAndOperator(df)
}
}
+
+ // ---------------------------------------------------------------------------
+ // Tests for issue #4834: RANGE frame with explicit PRECEDING/FOLLOWING offset
+ // on DATE ORDER BY. The planner wraps the offset as IntervalDayTime so these
+ // run natively; `checkSparkAnswerAndOperator` pins that.
+ // ---------------------------------------------------------------------------
+
+ private def writeDateWindowFixture(dir: java.io.File): Unit = {
+ (0 until 30)
+ .map { i =>
+ val day = (i % 28) + 1
+ (i % 3, java.sql.Date.valueOf(f"2024-01-$day%02d"), i * 2)
+ }
+ .toDF("a", "d", "c")
+ .repartition(3)
+ .write
+ .mode("overwrite")
+ .parquet(dir.toString)
+ }
+
+ test("window: RANGE BETWEEN N PRECEDING AND N FOLLOWING with DATE ORDER BY") {
+ withTempDir { dir =>
+ writeDateWindowFixture(dir)
+ spark.read.parquet(dir.toString).createOrReplaceTempView("window_test")
+ 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
+ """)
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ test("window: RANGE BETWEEN N PRECEDING AND CURRENT ROW with DATE ORDER BY") {
+ withTempDir { dir =>
+ writeDateWindowFixture(dir)
+ spark.read.parquet(dir.toString).createOrReplaceTempView("window_test")
+ 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
+ """)
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ test("window: RANGE BETWEEN CURRENT ROW AND N FOLLOWING with DATE ORDER BY") {
+ withTempDir { dir =>
+ writeDateWindowFixture(dir)
+ spark.read.parquet(dir.toString).createOrReplaceTempView("window_test")
+ 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
+ """)
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ test("window: RANGE BETWEEN UNBOUNDED PRECEDING AND N FOLLOWING with DATE ORDER BY") {
+ withTempDir { dir =>
+ writeDateWindowFixture(dir)
+ spark.read.parquet(dir.toString).createOrReplaceTempView("window_test")
+ 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
+ """)
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ test("window: RANGE BETWEEN with DATE ORDER BY and duplicate dates") {
+ // RANGE (not ROWS) must aggregate all peer rows that share the same date.
+ // With duplicates we can tell the two apart: ROWS would only include the
+ // literal N rows, RANGE includes every row whose date is within +/- N days.
+ withTempDir { dir =>
+ Seq(
+ (1, java.sql.Date.valueOf("2024-01-01"), 10),
+ (1, java.sql.Date.valueOf("2024-01-01"), 20),
+ (1, java.sql.Date.valueOf("2024-01-02"), 30),
+ (1, java.sql.Date.valueOf("2024-01-02"), 40),
+ (1, java.sql.Date.valueOf("2024-01-05"), 50),
+ (1, java.sql.Date.valueOf("2024-01-05"), 60))
+ .toDF("a", "d", "c")
+ .write
+ .mode("overwrite")
+ .parquet(dir.toString)
+
+ spark.read.parquet(dir.toString).createOrReplaceTempView("window_test")
+ 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
+ """)
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ test("window: RANGE BETWEEN with DATE ORDER BY and NULL dates") {
+ // Null dates land at the head under ASC NULLS FIRST (Spark's default).
+ // Verifies the native frame comparator handles Date32 nulls the same way
+ // Spark does when computing the boundary.
+ withTempDir { dir =>
+ Seq(
+ (1, Option.empty[java.sql.Date], 10),
+ (1, Option(java.sql.Date.valueOf("2024-01-01")), 20),
+ (1, Option(java.sql.Date.valueOf("2024-01-02")), 30),
+ (1, Option.empty[java.sql.Date], 40),
+ (1, Option(java.sql.Date.valueOf("2024-01-05")), 50))
+ .toDF("a", "d", "c")
+ .write
+ .mode("overwrite")
+ .parquet(dir.toString)
+
+ spark.read.parquet(dir.toString).createOrReplaceTempView("window_test")
+ 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
+ """)
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ test("window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW with DATE ORDER BY") {
+ // Regression baseline: no explicit offset, so this ran natively even
+ // before DATE offsets did.
+ withTempDir { dir =>
+ writeDateWindowFixture(dir)
+ spark.read.parquet(dir.toString).createOrReplaceTempView("window_test")
+ val df = sql("""
+ SELECT a, d, c,
+ SUM(c) OVER (PARTITION BY a ORDER BY d
+ RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) as sum_c
+ FROM window_test
+ """)
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ test("window: RANGE offset on DATE ORDER BY that wraps an Int falls back") {
+ // Spark adds the offset to the date as a plain Int, so 2147483647 FOLLOWING
+ // wraps below the current row and every frame is empty, while DataFusion
+ // would treat the overflowing bound as the end of the partition.
+ // https://github.com/apache/datafusion-comet/issues/5022
+ withTempDir { dir =>
+ Seq("2024-01-01", "2024-01-02", "2024-01-05")
+ .map(java.sql.Date.valueOf)
+ .toDF("d")
+ .repartition(1)
+ .write
+ .mode("overwrite")
+ .parquet(dir.toString)
+ spark.read.parquet(dir.toString).createOrReplaceTempView("date_wrap")
+ for (order <- Seq("ASC", "DESC")) {
+ val df = sql(s"""
+ SELECT d, COUNT(*) OVER (ORDER BY d $order
+ RANGE BETWEEN CURRENT ROW AND 2147483647 FOLLOWING) AS cnt
+ FROM date_wrap
+ """)
+ val (_, cometPlan) = checkSparkAnswerAndFallbackReason(
+ df,
+ "RANGE frame on DATE ORDER BY with an offset above 2144550751 days is not supported")
+ assert(collect(cometPlan) { case w: CometWindowExec => w }.isEmpty, cometPlan)
+ // Descending, the bound is `d - offset`, which does not wrap for these dates.
+ if (order == "ASC") {
+ checkAnswer(df.select("cnt"), Seq(Row(0L), Row(0L), Row(0L)))
+ }
+ }
+ }
+ }
+
+ test("window: largest RANGE offset on DATE ORDER BY that cannot wrap runs natively") {
+ // 2144550751 + the day number of 9999-12-31 is Int.MaxValue, so no date in
+ // Spark's supported range wraps. Past chrono's date range the native bound
+ // becomes the partition edge, which agrees with Spark's unwrapped bound
+ // lying beyond every row.
+ withTempDir { dir =>
+ Seq("0001-01-01", "1970-01-01", "2024-01-05", "9999-12-31")
+ .map(java.sql.Date.valueOf)
+ .toDF("d")
+ .repartition(1)
+ .write
+ .mode("overwrite")
+ .parquet(dir.toString)
+ spark.read.parquet(dir.toString).createOrReplaceTempView("date_edge")
+ for {
+ order <- Seq("ASC", "DESC")
+ frame <- Seq(
+ "RANGE BETWEEN 2144550751 PRECEDING AND CURRENT ROW",
+ "RANGE BETWEEN CURRENT ROW AND 2144550751 FOLLOWING",
+ "RANGE BETWEEN 2144550751 PRECEDING AND 2144550751 FOLLOWING")
+ } {
+ val df = sql(s"""
+ SELECT d, COUNT(*) OVER (ORDER BY d $order $frame) AS cnt
+ FROM date_edge
+ """)
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org