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/08/31 06:35:11 UTC
(datafusion) branch main updated: fix(substrait): preserve window function semantics (#22664)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 6e0d433c08 fix(substrait): preserve window function semantics (#22664)
6e0d433c08 is described below
commit 6e0d433c0899637a6f1f8d2761bbdd23f5ee5ce7
Author: Bruno Volpato <br...@gmail.com>
AuthorDate: Mon Aug 31 06:25:12 2026 +0000
fix(substrait): preserve window function semantics (#22664)
## Which issue does this PR close?
- Closes #22661.
## Rationale for this change
Substrait logical-plan conversion can currently return a valid window
plan with
different semantics from its input. On `main` (`496f2c2065c0`), a
round trip of:
```sql
SELECT array_agg(val) IGNORE NULLS OVER (
ORDER BY ts ROWS BETWEEN 1 PRECEDING AND CURRENT ROW
)
```
changes expected output `[A], [A], [C], [C], [E]` to
`[A], [A, NULL], [NULL, C], [C, NULL], [NULL, E]`.
Likewise, `window.slt:909` uses finite interval `RANGE` bounds; round
trip
changes bounded counts such as `6, 2, 1` into partition-wide counts `8,
8, 8`.
Returning changed results is worse than reporting unsupported
conversion.
## What changes are included in this PR?
- Encode and decode window `DISTINCT` through Substrait
`AggregationInvocation`.
- Reject explicit window null treatment and `FILTER`, which the current
Substrait window expression path does not encode.
- Reject finite frame offsets which cannot be represented as Substrait
integer
offsets instead of widening them to `UNBOUNDED`.
- Add logical round-trip coverage for `DISTINCT` and fail-closed
coverage for
`IGNORE NULLS`, `RESPECT NULLS`, `FILTER`, and interval `RANGE` frames.
## Are these changes tested?
- `cargo fmt --all`
- `cargo clippy --all-targets --all-features -- -D warnings`
- `cargo test -p datafusion-substrait`
Pre-fix reproduction:
```bash
cargo test -p datafusion-sqllogictest --test sqllogictests --features substrait -- --substrait-round-trip array_agg_sliding_window.slt:150 window.slt:909
```
- `array_agg_sliding_window.slt:150` produced null-containing arrays
despite
`IGNORE NULLS`.
- `window.slt:909` produced partition-wide counts for finite interval
bounds.
Post-fix targeted checks:
```bash
cargo test -p datafusion-sqllogictest --test sqllogictests --features substrait -- --substrait-round-trip array_agg_sliding_window.slt:150
cargo test -p datafusion-sqllogictest --test sqllogictests --features substrait -- --substrait-round-trip window.slt:909
```
- The first now fails conversion with `Window functions with IGNORE
NULLS are
not supported in Substrait`.
- The second now fails conversion with `Unsupported Substrait window
frame
offset: 1 DAY`.
- Selecting `window.slt:909` also runs unrelated existing `EXPLAIN`
records in
that fixture, which report separate plan-output mismatches after the
target
conversion error.
## Are there any user-facing changes?
Substrait producers now return explicit unsupported-feature errors for
window
null treatment, window `FILTER`, and finite non-integer window-frame
offsets
instead of producing plans with altered semantics. Window `DISTINCT` now
round-trips correctly.
---
.../src/logical_plan/consumer/expr/mod.rs | 25 +++-
.../logical_plan/consumer/expr/window_function.rs | 13 +-
.../logical_plan/producer/expr/window_function.rs | 132 +++++++++++++++------
.../tests/cases/roundtrip_logical_plan.rs | 39 ++++++
4 files changed, 171 insertions(+), 38 deletions(-)
diff --git a/datafusion/substrait/src/logical_plan/consumer/expr/mod.rs b/datafusion/substrait/src/logical_plan/consumer/expr/mod.rs
index 2fcc11f4e4..ad52d9f0dc 100644
--- a/datafusion/substrait/src/logical_plan/consumer/expr/mod.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/expr/mod.rs
@@ -206,7 +206,7 @@ mod tests {
use crate::extensions::Extensions;
use crate::logical_plan::consumer::utils::tests::test_consumer;
use crate::logical_plan::consumer::*;
- use datafusion::common::DFSchema;
+ use datafusion::common::{DFSchema, assert_contains};
use datafusion::logical_expr::Expr;
use substrait::proto::Expression;
use substrait::proto::expression::RexType;
@@ -270,4 +270,27 @@ mod tests {
Ok(())
}
+
+ #[tokio::test]
+ async fn window_function_with_invalid_invocation() {
+ let substrait = Expression {
+ rex_type: Some(RexType::WindowFunction(
+ substrait::proto::expression::WindowFunction {
+ function_reference: 0,
+ invocation: 3,
+ ..Default::default()
+ },
+ )),
+ };
+
+ let mut consumer = test_consumer();
+ let mut extensions = Extensions::default();
+ extensions.register_function("count");
+ consumer.extensions = &extensions;
+
+ let err = from_substrait_rex(&consumer, &substrait, &DFSchema::empty())
+ .await
+ .unwrap_err();
+ assert_contains!(err.to_string(), "Invalid window aggregation invocation 3");
+ }
}
diff --git a/datafusion/substrait/src/logical_plan/consumer/expr/window_function.rs b/datafusion/substrait/src/logical_plan/consumer/expr/window_function.rs
index 1f6f602a2a..3f32ed7268 100644
--- a/datafusion/substrait/src/logical_plan/consumer/expr/window_function.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/expr/window_function.rs
@@ -27,6 +27,7 @@ use datafusion::logical_expr::expr::WindowFunctionParams;
use datafusion::logical_expr::{
Expr, WindowFrameBound, WindowFrameUnits, WindowFunctionDefinition, expr,
};
+use substrait::proto::aggregate_function::AggregationInvocation;
use substrait::proto::expression::WindowFunction;
use substrait::proto::expression::window_function::{Bound, BoundsType};
use substrait::proto::expression::{
@@ -98,6 +99,16 @@ pub async fn from_window_function(
} else {
from_substrait_func_args(consumer, &window.arguments, input_schema).await?
};
+ let distinct =
+ match AggregationInvocation::try_from(window.invocation).map_err(|e| {
+ plan_datafusion_err!(
+ "Invalid window aggregation invocation {}: {e}",
+ window.invocation
+ )
+ })? {
+ AggregationInvocation::Unspecified | AggregationInvocation::All => false,
+ AggregationInvocation::Distinct => true,
+ };
Ok(Expr::from(expr::WindowFunction {
fun,
@@ -113,7 +124,7 @@ pub async fn from_window_function(
window_frame,
filter: None,
null_treatment: None,
- distinct: false,
+ distinct,
},
}))
}
diff --git a/datafusion/substrait/src/logical_plan/producer/expr/window_function.rs b/datafusion/substrait/src/logical_plan/producer/expr/window_function.rs
index 5d5f31cf11..d35771bf09 100644
--- a/datafusion/substrait/src/logical_plan/producer/expr/window_function.rs
+++ b/datafusion/substrait/src/logical_plan/producer/expr/window_function.rs
@@ -20,6 +20,7 @@ use crate::logical_plan::producer::utils::substrait_sort_field;
use datafusion::common::{DFSchemaRef, ScalarValue, not_impl_err};
use datafusion::logical_expr::expr::{WindowFunction, WindowFunctionParams};
use datafusion::logical_expr::{WindowFrame, WindowFrameBound, WindowFrameUnits};
+use substrait::proto::aggregate_function::AggregationInvocation;
use substrait::proto::expression::RexType;
use substrait::proto::expression::WindowFunction as SubstraitWindowFunction;
use substrait::proto::expression::window_function::bound as SubstraitBound;
@@ -41,11 +42,21 @@ pub fn from_window_function(
partition_by,
order_by,
window_frame,
- null_treatment: _,
- distinct: _,
- filter: _,
+ null_treatment,
+ distinct,
+ filter,
},
} = window_fn;
+ if let Some(null_treatment) = null_treatment {
+ return not_impl_err!(
+ "Window functions with {null_treatment} are not supported in Substrait"
+ );
+ }
+ if filter.is_some() {
+ return not_impl_err!(
+ "Window functions with FILTER are not supported in Substrait"
+ );
+ }
// function reference
let function_anchor = producer.register_function(fun.to_string());
// arguments
@@ -75,6 +86,7 @@ pub fn from_window_function(
order_by,
bounds,
bound_type,
+ *distinct,
))
}
@@ -85,6 +97,7 @@ fn make_substrait_window_function(
sorts: Vec<SortField>,
bounds: (Bound, Bound),
bounds_type: BoundsType,
+ distinct: bool,
) -> Expression {
#[expect(deprecated)]
Expression {
@@ -95,8 +108,12 @@ fn make_substrait_window_function(
sorts,
options: vec![],
output_type: None,
- phase: 0, // default to AGGREGATION_PHASE_UNSPECIFIED
- invocation: 0, // TODO: fix
+ phase: 0, // default to AGGREGATION_PHASE_UNSPECIFIED
+ invocation: if distinct {
+ AggregationInvocation::Distinct as i32
+ } else {
+ AggregationInvocation::All as i32
+ },
lower_bound: Some(bounds.0),
upper_bound: Some(bounds.1),
args: vec![],
@@ -120,45 +137,88 @@ fn to_substrait_bounds(
window_frame: &WindowFrame,
) -> datafusion::common::Result<(Bound, Bound)> {
Ok((
- to_substrait_bound(&window_frame.start_bound),
- to_substrait_bound(&window_frame.end_bound),
+ to_substrait_bound(&window_frame.start_bound)?,
+ to_substrait_bound(&window_frame.end_bound)?,
))
}
-fn to_substrait_bound(bound: &WindowFrameBound) -> Bound {
+fn to_substrait_bound(bound: &WindowFrameBound) -> datafusion::common::Result<Bound> {
+ if bound.is_unbounded() {
+ return Ok(Bound {
+ kind: Some(BoundKind::Unbounded(SubstraitBound::Unbounded {})),
+ });
+ }
match bound {
- WindowFrameBound::CurrentRow => Bound {
+ WindowFrameBound::CurrentRow => Ok(Bound {
kind: Some(BoundKind::CurrentRow(SubstraitBound::CurrentRow {})),
- },
- WindowFrameBound::Preceding(s) => match to_substrait_bound_offset(s) {
- Some(offset) => Bound {
- kind: Some(BoundKind::Preceding(SubstraitBound::Preceding { offset })),
- },
- None => Bound {
- kind: Some(BoundKind::Unbounded(SubstraitBound::Unbounded {})),
- },
- },
- WindowFrameBound::Following(s) => match to_substrait_bound_offset(s) {
- Some(offset) => Bound {
- kind: Some(BoundKind::Following(SubstraitBound::Following { offset })),
- },
- None => Bound {
- kind: Some(BoundKind::Unbounded(SubstraitBound::Unbounded {})),
- },
- },
+ }),
+ WindowFrameBound::Preceding(s) => Ok(Bound {
+ kind: Some(BoundKind::Preceding(SubstraitBound::Preceding {
+ offset: to_substrait_bound_offset(s)?,
+ })),
+ }),
+ WindowFrameBound::Following(s) => Ok(Bound {
+ kind: Some(BoundKind::Following(SubstraitBound::Following {
+ offset: to_substrait_bound_offset(s)?,
+ })),
+ }),
}
}
-fn to_substrait_bound_offset(value: &ScalarValue) -> Option<i64> {
+fn to_substrait_bound_offset(value: &ScalarValue) -> datafusion::common::Result<i64> {
match value {
- ScalarValue::UInt8(Some(v)) => Some(*v as i64),
- ScalarValue::UInt16(Some(v)) => Some(*v as i64),
- ScalarValue::UInt32(Some(v)) => Some(*v as i64),
- ScalarValue::UInt64(Some(v)) => Some(*v as i64),
- ScalarValue::Int8(Some(v)) => Some(*v as i64),
- ScalarValue::Int16(Some(v)) => Some(*v as i64),
- ScalarValue::Int32(Some(v)) => Some(*v as i64),
- ScalarValue::Int64(Some(v)) => Some(*v),
- _ => None,
+ ScalarValue::UInt8(Some(v)) => Ok((*v).into()),
+ ScalarValue::UInt16(Some(v)) => Ok((*v).into()),
+ ScalarValue::UInt32(Some(v)) => Ok((*v).into()),
+ ScalarValue::UInt64(Some(v)) => match i64::try_from(*v) {
+ Ok(v) => Ok(v),
+ Err(_) => not_impl_err!("Unsupported Substrait window frame offset: {value}"),
+ },
+ ScalarValue::Int8(Some(v)) => Ok((*v).into()),
+ ScalarValue::Int16(Some(v)) => Ok((*v).into()),
+ ScalarValue::Int32(Some(v)) => Ok((*v).into()),
+ ScalarValue::Int64(Some(v)) => Ok(*v),
+ _ => not_impl_err!("Unsupported Substrait window frame offset: {value}"),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use datafusion::common::assert_contains;
+
+ #[test]
+ fn window_frame_offsets() {
+ for value in [
+ ScalarValue::UInt8(Some(1)),
+ ScalarValue::UInt16(Some(1)),
+ ScalarValue::UInt32(Some(1)),
+ ScalarValue::UInt64(Some(1)),
+ ScalarValue::Int8(Some(1)),
+ ScalarValue::Int16(Some(1)),
+ ScalarValue::Int32(Some(1)),
+ ScalarValue::Int64(Some(1)),
+ ] {
+ let frame = WindowFrame::new_bounds(
+ WindowFrameUnits::Rows,
+ WindowFrameBound::CurrentRow,
+ WindowFrameBound::Following(value),
+ );
+ let (_, bound) = to_substrait_bounds(&frame).unwrap();
+ assert_eq!(
+ bound.kind,
+ Some(BoundKind::Following(SubstraitBound::Following {
+ offset: 1
+ }))
+ );
+ }
+
+ let frame = WindowFrame::new_bounds(
+ WindowFrameUnits::Rows,
+ WindowFrameBound::CurrentRow,
+ WindowFrameBound::Following(ScalarValue::UInt64(Some(u64::MAX))),
+ );
+ let err = to_substrait_bounds(&frame).unwrap_err();
+ assert_contains!(err.to_string(), "Unsupported Substrait window frame offset");
}
}
diff --git a/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs b/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs
index 3ee40cd5c6..9dcd702456 100644
--- a/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs
+++ b/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs
@@ -1587,6 +1587,45 @@ async fn window_with_rows() -> Result<()> {
roundtrip("SELECT sum(b) OVER (PARTITION BY a ROWS BETWEEN 4 PRECEDING AND 2 PRECEDING) FROM data;").await
}
+#[tokio::test]
+async fn window_distinct() -> Result<()> {
+ roundtrip(
+ "SELECT count(DISTINCT a) OVER (ORDER BY a ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) FROM data;",
+ )
+ .await
+}
+
+#[tokio::test]
+async fn unsupported_window_semantics_are_rejected() -> Result<()> {
+ let ctx = create_context().await?;
+ for (sql, expected_error) in [
+ (
+ "SELECT first_value(a) IGNORE NULLS OVER (ORDER BY a) FROM data;",
+ "IGNORE NULLS",
+ ),
+ (
+ "SELECT first_value(a) RESPECT NULLS OVER (ORDER BY a) FROM data;",
+ "RESPECT NULLS",
+ ),
+ (
+ "SELECT sum(a) FILTER (WHERE a > 0) OVER (PARTITION BY d) FROM data;",
+ "FILTER",
+ ),
+ (
+ "SELECT count(*) OVER (ORDER BY c RANGE BETWEEN INTERVAL '1' DAY PRECEDING AND CURRENT ROW) FROM data;",
+ "Unsupported Substrait window frame offset",
+ ),
+ ] {
+ let plan = ctx.sql(sql).await?.into_optimized_plan()?;
+ let err = to_substrait_plan(&plan, &ctx.state()).unwrap_err();
+ assert!(
+ err.to_string().contains(expected_error),
+ "expected error containing '{expected_error}', got '{err}'"
+ );
+ }
+ Ok(())
+}
+
#[tokio::test]
async fn qualified_schema_table_reference() -> Result<()> {
roundtrip("SELECT * FROM public.data;").await
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org