fix: preserve ExprFunctionExt options when chaining

datafusion-examples/examples/udf/advanced_udwf.rs

@@ -197,6 +197,7 @@ impl WindowUDFImpl for SimplifySmoothItUdf {
                     partition_by: window_function.params.partition_by,
                     order_by: window_function.params.order_by,
                     window_frame: window_function.params.window_frame,
+                    window_frame_explicit: window_function.params.window_frame_explicit,
                     null_treatment: window_function.params.null_treatment,
                     distinct: window_function.params.distinct,
                     filter: window_function.params.filter,

datafusion/core/src/physical_planner.rs

@@ -2927,6 +2927,7 @@ pub fn create_window_expr_with_name(
                         null_treatment,
                         distinct,
                         filter,
+                        ..
                     },
             } = window_fun.as_ref();
             let physical_args = create_physical_exprs(

datafusion/core/tests/expr_api/mod.rs

@@ -24,12 +24,12 @@ use arrow::datatypes::{DataType, Field};
 use arrow::util::pretty::{pretty_format_batches, pretty_format_columns};
 use datafusion::prelude::*;
 use datafusion_common::{DFSchema, ScalarValue};
-use datafusion_expr::ExprFunctionExt;
-use datafusion_expr::expr::NullTreatment;
+use datafusion_expr::expr::{NullTreatment, WindowFunction};
 use datafusion_expr::simplify::SimplifyContext;
+use datafusion_expr::{ExprFunctionExt, WindowFrame};
 use datafusion_functions::core::expr_ext::FieldAccessor;
 use datafusion_functions_aggregate::first_last::first_value_udaf;
-use datafusion_functions_aggregate::sum::sum_udaf;
+use datafusion_functions_aggregate::sum::{sum, sum_udaf};
 use datafusion_functions_nested::expr_ext::{IndexAccessor, SliceAccessor};
 use datafusion_optimizer::simplify_expressions::ExprSimplifier;
 /// Tests of using and evaluating `Expr`s outside the context of a LogicalPlan
@@ -39,6 +39,152 @@ mod parse_sql_expr;
 #[expect(clippy::needless_pass_by_value)]
 mod simplification;
 
+#[test]
+fn test_chaining_aggregate_options_preserves_existing_options() {
+    let order_by = vec![col("i").sort(true, false)];
+    let filter = col("i").gt(lit(0));
+    let aggregate = sum(col("i"))
+        .order_by(order_by.clone())
+        .filter(filter.clone())
+        .build()
+        .unwrap();
+
+    let Expr::AggregateFunction(aggregate) = aggregate.distinct().build().unwrap() else {
+        panic!("expected aggregate function");
+    };
+    assert_eq!(aggregate.params.order_by, order_by);
+    assert_eq!(aggregate.params.filter.as_deref(), Some(&filter));
+    assert!(aggregate.params.distinct);
+}
+
+#[test]
+fn test_chaining_window_options_preserves_explicit_frame() {
+    let order_by = vec![col("i").sort(true, false)];
+    let frame = WindowFrame::new(None);
+    let window = Expr::from(WindowFunction::new(first_value_udaf(), vec![col("i")]))
+        .order_by(order_by.clone())
+        .window_frame(frame.clone())
+        .build()
+        .unwrap();
+
+    let Expr::WindowFunction(window) =
+        window.partition_by(vec![col("id")]).build().unwrap()
+    else {
+        panic!("expected window function");
+    };
+    assert_eq!(window.params.order_by, order_by);
+    assert_eq!(window.params.partition_by, vec![col("id")]);
+    assert_eq!(window.params.window_frame, frame);
+    assert!(window.params.window_frame_explicit);
+}
+
+#[test]
+fn test_chaining_window_options_rederives_default_frame() {
+    let window = Expr::from(WindowFunction::new(first_value_udaf(), vec![col("i")]))
+        .partition_by(vec![col("id")])
+        .build()
+        .unwrap();
+    let Expr::WindowFunction(ref original) = window else {
+        panic!("expected window function");
+    };
+    assert!(!original.params.window_frame_explicit);
+    assert_eq!(original.params.window_frame, WindowFrame::new(None));
+
+    let order_by = vec![col("i").sort(true, false)];
+    let Expr::WindowFunction(chained) =
+        window.order_by(order_by.clone()).build().unwrap()
+    else {
+        panic!("expected window function");
+    };
+    assert_eq!(chained.params.order_by, order_by);
+    assert_eq!(chained.params.partition_by, vec![col("id")]);
+    assert_eq!(chained.params.window_frame, WindowFrame::new(Some(false)));
+    assert!(!chained.params.window_frame_explicit);
+}
+
+#[test]
+fn test_window_frame_provenance_does_not_change_expression_equality() {
+    use std::hash::{DefaultHasher, Hash, Hasher};
+
+    let derived = Expr::from(WindowFunction::new(first_value_udaf(), vec![col("i")]));
+    let explicit = Expr::from(WindowFunction::new(first_value_udaf(), vec![col("i")]))
+        .window_frame(WindowFrame::new(None))
+        .build()
+        .unwrap();
+    assert_eq!(derived, explicit);
+
+    let mut derived_hash = DefaultHasher::new();
+    derived.hash(&mut derived_hash);
+    let mut explicit_hash = DefaultHasher::new();
+    explicit.hash(&mut explicit_hash);
+    assert_eq!(derived_hash.finish(), explicit_hash.finish());
+}
+
+#[tokio::test]
+async fn test_chaining_implicit_window_frame_includes_peers()
+-> datafusion::error::Result<()> {
+    let ctx = SessionContext::new();
+    let batch = RecordBatch::try_from_iter(vec![
+        ("k", Arc::new(Int64Array::from(vec![1, 1, 2])) as ArrayRef),
+        (
+            "v",
+            Arc::new(Int64Array::from(vec![10, 20, 30])) as ArrayRef,
+        ),
+    ])?;
+    let window = Expr::from(WindowFunction::new(sum_udaf(), vec![col("v")]))
+        .order_by(vec![col("k").sort(true, false)])
+        .build()?
+        .partition_by(vec![])
+        .build()?;
+    let result = ctx
+        .read_batch(batch)?
+        .select(vec![col("k"), window.alias("total")])?
+        .sort(vec![col("k").sort(true, false)])?
+        .collect()
+        .await?;
+    datafusion_common::assert_batches_eq!(
+        [
+            "+---+-------+",
+            "| k | total |",
+            "+---+-------+",
+            "| 1 | 30    |",
+            "| 1 | 30    |",
+            "| 2 | 60    |",
+            "+---+-------+",
+        ],
+        &result
+    );
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_sql_window_frame_provenance() -> datafusion::error::Result<()> {
+    fn frame_is_explicit(plan: &datafusion_expr::LogicalPlan) -> Option<bool> {
+        match plan {
+            datafusion_expr::LogicalPlan::Window(window) => {
+                window.window_expr.iter().find_map(|expr| match expr {
+                    Expr::WindowFunction(fun) => Some(fun.params.window_frame_explicit),
+                    _ => None,
+                })
+            }
+            _ => plan.inputs().into_iter().find_map(frame_is_explicit),
+        }
+    }
+
+    let ctx = SessionContext::new();
+    for (frame_clause, expected) in [
+        ("", false),
+        ("ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", true),
+    ] {
+        let sql = format!(
+            "SELECT row_number() OVER (ORDER BY x {frame_clause}) FROM (VALUES (1), (2)) AS t(x)"
+        );
+        let plan = ctx.sql(&sql).await?.into_unoptimized_plan();
+        assert_eq!(frame_is_explicit(&plan), Some(expected), "{sql}");
+    }
+    Ok(())
+}
+
 #[test]
 fn test_octet_length() {
     #[rustfmt::skip]

datafusion/expr/src/expr.rs

@@ -1250,7 +1250,7 @@ pub struct WindowFunction {
     pub params: WindowFunctionParams,
 }
 
-#[derive(Clone, PartialEq, Eq, PartialOrd, Hash, Debug)]
+#[derive(Clone, Debug)]
 pub struct WindowFunctionParams {
     /// List of expressions to feed to the functions as arguments
     pub args: Vec<Expr>,
@@ -1260,6 +1260,8 @@ pub struct WindowFunctionParams {
     pub order_by: Vec<Sort>,
     /// Window frame
     pub window_frame: WindowFrame,
+    /// Whether the frame was specified explicitly rather than derived from ORDER BY.
+    pub window_frame_explicit: bool,
     /// Optional filter expression (FILTER (WHERE ...))
     pub filter: Option<Box<Expr>>,
     /// Specifies how NULL value is treated: ignore or respect
@@ -1268,6 +1270,52 @@ pub struct WindowFunctionParams {
     pub distinct: bool,
 }
 
+type WindowFunctionSemanticFields<'a> = (
+    &'a Vec<Expr>,
+    &'a Vec<Expr>,
+    &'a Vec<Sort>,
+    &'a WindowFrame,
+    &'a Option<Box<Expr>>,
+    &'a Option<NullTreatment>,
+    &'a bool,
+);
+
+impl WindowFunctionParams {
+    // Provenance affects future builder calls, but not the meaning of the
+    // already-resolved window expression.
+    fn semantic_fields(&self) -> WindowFunctionSemanticFields<'_> {
+        (
+            &self.args,
+            &self.partition_by,
+            &self.order_by,
+            &self.window_frame,
+            &self.filter,
+            &self.null_treatment,
+            &self.distinct,
+        )
+    }
+}
+
+impl PartialEq for WindowFunctionParams {
+    fn eq(&self, other: &Self) -> bool {
+        self.semantic_fields() == other.semantic_fields()
+    }
+}
+
+impl Eq for WindowFunctionParams {}
+
+impl PartialOrd for WindowFunctionParams {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        self.semantic_fields().partial_cmp(&other.semantic_fields())
+    }
+}
+
+impl Hash for WindowFunctionParams {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.semantic_fields().hash(state);
+    }
+}
+
 impl WindowFunction {
     /// Create a new Window expression with the specified argument an
     /// empty `OVER` clause
@@ -1279,6 +1327,7 @@ impl WindowFunction {
                 partition_by: Vec::default(),
                 order_by: Vec::default(),
                 window_frame: WindowFrame::new(None),
+                window_frame_explicit: false,
                 filter: None,
                 null_treatment: None,
                 distinct: false,
@@ -2664,6 +2713,7 @@ impl NormalizeEq for Expr {
                             filter: self_filter,
                             null_treatment: self_null_treatment,
                             distinct: self_distinct,
+                            ..
                         },
                 } = left.as_ref();
                 let WindowFunction {
@@ -2677,6 +2727,7 @@ impl NormalizeEq for Expr {
                             filter: other_filter,
                             null_treatment: other_null_treatment,
                             distinct: other_distinct,
+                            ..
                         },
                 } = other.as_ref();
 
@@ -2937,6 +2988,7 @@ impl HashNode for Expr {
                             filter,
                             null_treatment,
                             distinct,
+                            ..
                         },
                 } = window_fun.as_ref();
                 fun.hash(state);
@@ -3295,6 +3347,7 @@ impl Display for SchemaDisplay<'_> {
                             filter,
                             null_treatment,
                             distinct,
+                            ..
                         } = params;
 
                         // Write function name and open parenthesis
@@ -3727,6 +3780,7 @@ impl Display for Expr {
                             filter,
                             null_treatment,
                             distinct,
+                            ..
                         } = params;
 
                         fmt_function(f, &fun.to_string(), *distinct, args, true)?;

datafusion/expr/src/expr_fn.rs

@@ -822,15 +822,36 @@ pub struct ExprFuncBuilder {
 impl ExprFuncBuilder {
     /// Create a new `ExprFuncBuilder`, see [`ExprFunctionExt`]
     fn new(fun: Option<ExprFuncKind>) -> Self {
-        Self {
+        let mut builder = Self {
             fun,
             order_by: None,
             filter: None,
             distinct: false,
             null_treatment: None,
             partition_by: None,
             window_frame: None,
+        };
+        match &builder.fun {
+            Some(ExprFuncKind::Aggregate(udaf)) => {
+                builder.order_by = Some(udaf.params.order_by.clone());
+                builder.filter = udaf.params.filter.as_deref().cloned();
+                builder.distinct = udaf.params.distinct;
+                builder.null_treatment = udaf.params.null_treatment;
+            }
+            Some(ExprFuncKind::Window(udwf)) => {
+                builder.order_by = Some(udwf.params.order_by.clone());
+                builder.filter = udwf.params.filter.as_deref().cloned();
+                builder.distinct = udwf.params.distinct;
+                builder.null_treatment = udwf.params.null_treatment;
+                builder.partition_by = Some(udwf.params.partition_by.clone());
+                builder.window_frame = udwf
+                    .params
+                    .window_frame_explicit
+                    .then(|| udwf.params.window_frame.clone());
+            }
+            None => {}
         }
+        builder
     }
 
     /// Updates and returns the in progress [`Expr::AggregateFunction`] or [`Expr::WindowFunction`]
@@ -865,11 +886,19 @@ impl ExprFuncBuilder {
                 Expr::AggregateFunction(udaf)
             }
             ExprFuncKind::Window(mut udwf) => {
-                let has_order_by = order_by.as_ref().map(|o| !o.is_empty());
+                let order_by = order_by.unwrap_or_default();
+                let window_frame_explicit = window_frame.is_some();
+                let window_frame = window_frame.unwrap_or_else(|| {
+                    if udwf.params.order_by == order_by {
+                        udwf.params.window_frame.clone()
+                    } else {
+                        WindowFrame::new((!order_by.is_empty()).then_some(false))
+                    }
+                });
                 udwf.params.partition_by = partition_by.unwrap_or_default();
-                udwf.params.order_by = order_by.unwrap_or_default();
-                udwf.params.window_frame =
-                    window_frame.unwrap_or_else(|| WindowFrame::new(has_order_by));
+                udwf.params.order_by = order_by;
+                udwf.params.window_frame = window_frame;
+                udwf.params.window_frame_explicit = window_frame_explicit;
                 udwf.params.filter = filter.map(Box::new);
                 udwf.params.null_treatment = null_treatment;
                 udwf.params.distinct = distinct;

datafusion/expr/src/planner.rs

@@ -357,6 +357,7 @@ pub struct RawWindowExpr {
     pub partition_by: Vec<Expr>,
     pub order_by: Vec<SortExpr>,
     pub window_frame: WindowFrame,
+    pub window_frame_explicit: bool,
     pub filter: Option<Box<Expr>>,
     pub null_treatment: Option<NullTreatment>,
     pub distinct: bool,

datafusion/expr/src/tree_node.rs

@@ -269,6 +269,7 @@ impl TreeNode for Expr {
                             partition_by,
                             order_by,
                             window_frame,
+                            window_frame_explicit,
                             filter,
                             null_treatment,
                             distinct,
@@ -286,6 +287,7 @@ impl TreeNode for Expr {
                                     partition_by: new_partition_by,
                                     order_by: new_order_by,
                                     window_frame,
+                                    window_frame_explicit,
                                     filter: new_filter,
                                     null_treatment,
                                     distinct,