From 6daa8754515fd53d5d3aca9ef38aafe9e09e98f4 Mon Sep 17 00:00:00 2001 From: goutamadwant Date: Mon, 12 Oct 2026 23:30:33 -0700 Subject: [PATCH 1/2] Preserve window function options when chaining --- .../examples/udf/advanced_udwf.rs | 1 + datafusion/core/src/physical_planner.rs | 1 + datafusion/core/tests/expr_api/mod.rs | 115 +++++++++++++++++- datafusion/expr/src/expr.rs | 56 ++++++++- datafusion/expr/src/expr_fn.rs | 39 +++++- datafusion/expr/src/planner.rs | 1 + datafusion/expr/src/tree_node.rs | 2 + datafusion/expr/src/udaf.rs | 2 + datafusion/functions-window/src/planner.rs | 6 + .../optimizer/src/analyzer/type_coercion.rs | 2 + .../proto-models/proto/datafusion.proto | 1 + .../proto-models/src/generated/pbjson.rs | 18 +++ .../proto-models/src/generated/prost.rs | 2 + .../proto/src/logical_plan/from_proto.rs | 6 +- datafusion/proto/src/logical_plan/to_proto.rs | 2 + .../tests/cases/roundtrip_logical_plan.rs | 9 ++ datafusion/sql/src/expr/function.rs | 4 + datafusion/sql/src/unparser/expr.rs | 20 ++- datafusion/sql/tests/cases/plan_to_sql.rs | 12 ++ .../consumer/expr/window_function.rs | 2 + .../producer/expr/window_function.rs | 16 ++- .../tests/cases/roundtrip_logical_plan.rs | 28 ++++- .../library-user-guide/upgrading/56.0.0.md | 10 ++ 23 files changed, 335 insertions(+), 20 deletions(-) diff --git a/datafusion-examples/examples/udf/advanced_udwf.rs b/datafusion-examples/examples/udf/advanced_udwf.rs index 2508e6cd60e59..81ea342574ea1 100644 --- a/datafusion-examples/examples/udf/advanced_udwf.rs +++ b/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, diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs index f1bc734d7292b..aee82a093af81 100644 --- a/datafusion/core/src/physical_planner.rs +++ b/datafusion/core/src/physical_planner.rs @@ -2706,6 +2706,7 @@ pub fn create_window_expr_with_name( null_treatment, distinct, filter, + .. }, } = window_fun.as_ref(); let physical_args = create_physical_exprs( diff --git a/datafusion/core/tests/expr_api/mod.rs b/datafusion/core/tests/expr_api/mod.rs index 4ca48a5f4711f..0961885a0cf87 100644 --- a/datafusion/core/tests/expr_api/mod.rs +++ b/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,115 @@ 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_sql_window_frame_provenance() -> datafusion::error::Result<()> { + fn frame_is_explicit(plan: &datafusion_expr::LogicalPlan) -> Option { + 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] diff --git a/datafusion/expr/src/expr.rs b/datafusion/expr/src/expr.rs index 30fcf47f5815b..c3beea926b792 100644 --- a/datafusion/expr/src/expr.rs +++ b/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, @@ -1260,6 +1260,8 @@ pub struct WindowFunctionParams { pub order_by: Vec, /// 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>, /// Specifies how NULL value is treated: ignore or respect @@ -1268,6 +1270,52 @@ pub struct WindowFunctionParams { pub distinct: bool, } +type WindowFunctionSemanticFields<'a> = ( + &'a Vec, + &'a Vec, + &'a Vec, + &'a WindowFrame, + &'a Option>, + &'a Option, + &'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 { + self.semantic_fields().partial_cmp(&other.semantic_fields()) + } +} + +impl Hash for WindowFunctionParams { + fn hash(&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)?; diff --git a/datafusion/expr/src/expr_fn.rs b/datafusion/expr/src/expr_fn.rs index b1a5a12d155ce..6aaf0524d2a89 100644 --- a/datafusion/expr/src/expr_fn.rs +++ b/datafusion/expr/src/expr_fn.rs @@ -822,7 +822,7 @@ pub struct ExprFuncBuilder { impl ExprFuncBuilder { /// Create a new `ExprFuncBuilder`, see [`ExprFunctionExt`] fn new(fun: Option) -> Self { - Self { + let mut builder = Self { fun, order_by: None, filter: None, @@ -830,7 +830,28 @@ impl ExprFuncBuilder { 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; diff --git a/datafusion/expr/src/planner.rs b/datafusion/expr/src/planner.rs index 31aad67d17bfc..367f93f2dec09 100644 --- a/datafusion/expr/src/planner.rs +++ b/datafusion/expr/src/planner.rs @@ -353,6 +353,7 @@ pub struct RawWindowExpr { pub partition_by: Vec, pub order_by: Vec, pub window_frame: WindowFrame, + pub window_frame_explicit: bool, pub filter: Option>, pub null_treatment: Option, pub distinct: bool, diff --git a/datafusion/expr/src/tree_node.rs b/datafusion/expr/src/tree_node.rs index 941fd22ea179f..8fa11a227a99c 100644 --- a/datafusion/expr/src/tree_node.rs +++ b/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, diff --git a/datafusion/expr/src/udaf.rs b/datafusion/expr/src/udaf.rs index 7358568f0afe1..ef6c67f048dd9 100644 --- a/datafusion/expr/src/udaf.rs +++ b/datafusion/expr/src/udaf.rs @@ -1257,6 +1257,7 @@ impl<'a> UdafWindowFunctionSchemaNameBuilder<'a> { filter, null_treatment, distinct, + .. } = params; let mut schema_name = String::new(); @@ -1448,6 +1449,7 @@ impl<'a> UdafWindowFunctionDisplayNameBuilder<'a> { filter, null_treatment, distinct, + .. } = params; let mut display_name = String::new(); diff --git a/datafusion/functions-window/src/planner.rs b/datafusion/functions-window/src/planner.rs index 6f4eb2051f047..7e10337ced2af 100644 --- a/datafusion/functions-window/src/planner.rs +++ b/datafusion/functions-window/src/planner.rs @@ -40,6 +40,7 @@ impl ExprPlanner for WindowFunctionPlanner { partition_by, order_by, window_frame, + window_frame_explicit, filter, null_treatment, distinct, @@ -52,6 +53,7 @@ impl ExprPlanner for WindowFunctionPlanner { partition_by, order_by, window_frame, + window_frame_explicit, filter, null_treatment, distinct, @@ -71,6 +73,7 @@ impl ExprPlanner for WindowFunctionPlanner { partition_by, order_by, window_frame, + window_frame_explicit, null_treatment, distinct, filter, @@ -82,6 +85,7 @@ impl ExprPlanner for WindowFunctionPlanner { partition_by, order_by, window_frame, + window_frame_explicit, filter, null_treatment, distinct, @@ -100,6 +104,7 @@ impl ExprPlanner for WindowFunctionPlanner { partition_by, order_by, window_frame, + window_frame_explicit, filter, null_treatment, distinct, @@ -112,6 +117,7 @@ impl ExprPlanner for WindowFunctionPlanner { partition_by, order_by, window_frame, + window_frame_explicit, filter, null_treatment, distinct, diff --git a/datafusion/optimizer/src/analyzer/type_coercion.rs b/datafusion/optimizer/src/analyzer/type_coercion.rs index 47aba9e715486..a43d3ca1ecffa 100644 --- a/datafusion/optimizer/src/analyzer/type_coercion.rs +++ b/datafusion/optimizer/src/analyzer/type_coercion.rs @@ -988,6 +988,7 @@ impl TreeNodeRewriter for TypeCoercionRewriter<'_> { partition_by, order_by, window_frame, + window_frame_explicit, filter, null_treatment, distinct, @@ -1017,6 +1018,7 @@ impl TreeNodeRewriter for TypeCoercionRewriter<'_> { partition_by, order_by, window_frame, + window_frame_explicit, filter, null_treatment, distinct, diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto index b82efd3b2b583..c90c2610d8237 100644 --- a/datafusion/proto-models/proto/datafusion.proto +++ b/datafusion/proto-models/proto/datafusion.proto @@ -731,6 +731,7 @@ message WindowExprNode { optional NullTreatment null_treatment = 11; bool distinct = 12; LogicalExprNode filter = 13; + bool window_frame_explicit = 14; } message BetweenNode { diff --git a/datafusion/proto-models/src/generated/pbjson.rs b/datafusion/proto-models/src/generated/pbjson.rs index 01b7f025636b1..7ff414c7e0b6a 100644 --- a/datafusion/proto-models/src/generated/pbjson.rs +++ b/datafusion/proto-models/src/generated/pbjson.rs @@ -30131,6 +30131,9 @@ impl serde::Serialize for WindowExprNode { if self.filter.is_some() { len += 1; } + if self.window_frame_explicit { + len += 1; + } if self.window_function.is_some() { len += 1; } @@ -30163,6 +30166,9 @@ impl serde::Serialize for WindowExprNode { if let Some(v) = self.filter.as_ref() { struct_ser.serialize_field("filter", v)?; } + if self.window_frame_explicit { + struct_ser.serialize_field("windowFrameExplicit", &self.window_frame_explicit)?; + } if let Some(v) = self.window_function.as_ref() { match v { window_expr_node::WindowFunction::Udaf(v) => { @@ -30196,6 +30202,8 @@ impl<'de> serde::Deserialize<'de> for WindowExprNode { "nullTreatment", "distinct", "filter", + "window_frame_explicit", + "windowFrameExplicit", "udaf", "udwf", ]; @@ -30210,6 +30218,7 @@ impl<'de> serde::Deserialize<'de> for WindowExprNode { NullTreatment, Distinct, Filter, + WindowFrameExplicit, Udaf, Udwf, } @@ -30241,6 +30250,7 @@ impl<'de> serde::Deserialize<'de> for WindowExprNode { "nullTreatment" | "null_treatment" => Ok(GeneratedField::NullTreatment), "distinct" => Ok(GeneratedField::Distinct), "filter" => Ok(GeneratedField::Filter), + "windowFrameExplicit" | "window_frame_explicit" => Ok(GeneratedField::WindowFrameExplicit), "udaf" => Ok(GeneratedField::Udaf), "udwf" => Ok(GeneratedField::Udwf), _ => Err(serde::de::Error::unknown_field(value, FIELDS)), @@ -30270,6 +30280,7 @@ impl<'de> serde::Deserialize<'de> for WindowExprNode { let mut null_treatment__ = None; let mut distinct__ = None; let mut filter__ = None; + let mut window_frame_explicit__ = None; let mut window_function__ = None; while let Some(k) = map_.next_key()? { match k { @@ -30323,6 +30334,12 @@ impl<'de> serde::Deserialize<'de> for WindowExprNode { } filter__ = map_.next_value()?; } + GeneratedField::WindowFrameExplicit => { + if window_frame_explicit__.is_some() { + return Err(serde::de::Error::duplicate_field("windowFrameExplicit")); + } + window_frame_explicit__ = Some(map_.next_value()?); + } GeneratedField::Udaf => { if window_function__.is_some() { return Err(serde::de::Error::duplicate_field("udaf")); @@ -30346,6 +30363,7 @@ impl<'de> serde::Deserialize<'de> for WindowExprNode { null_treatment: null_treatment__, distinct: distinct__.unwrap_or_default(), filter: filter__, + window_frame_explicit: window_frame_explicit__.unwrap_or_default(), window_function: window_function__, }) } diff --git a/datafusion/proto-models/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs index 61ba256704f93..db26f122399c7 100644 --- a/datafusion/proto-models/src/generated/prost.rs +++ b/datafusion/proto-models/src/generated/prost.rs @@ -1143,6 +1143,8 @@ pub struct WindowExprNode { pub distinct: bool, #[prost(message, optional, boxed, tag = "13")] pub filter: ::core::option::Option<::prost::alloc::boxed::Box>, + #[prost(bool, tag = "14")] + pub window_frame_explicit: bool, #[prost(oneof = "window_expr_node::WindowFunction", tags = "3, 9")] pub window_function: ::core::option::Option, } diff --git a/datafusion/proto/src/logical_plan/from_proto.rs b/datafusion/proto/src/logical_plan/from_proto.rs index 2910ef356e5d1..0375f33b4d9bf 100644 --- a/datafusion/proto/src/logical_plan/from_proto.rs +++ b/datafusion/proto/src/logical_plan/from_proto.rs @@ -275,7 +275,11 @@ pub fn parse_expr( builder = builder.filter(filter); } - builder.build().map_err(Error::DataFusionError) + let mut window_expr = builder.build().map_err(Error::DataFusionError)?; + if let Expr::WindowFunction(window_fun) = &mut window_expr { + window_fun.params.window_frame_explicit = expr.window_frame_explicit; + } + Ok(window_expr) } ExprType::Alias(alias) => Ok(Expr::Alias(Alias::new( parse_required_expr(alias.expr.as_deref(), ctx, "expr", codec)?, diff --git a/datafusion/proto/src/logical_plan/to_proto.rs b/datafusion/proto/src/logical_plan/to_proto.rs index f72b07871bb1a..bc417d9b74139 100644 --- a/datafusion/proto/src/logical_plan/to_proto.rs +++ b/datafusion/proto/src/logical_plan/to_proto.rs @@ -177,6 +177,7 @@ pub fn serialize_expr( partition_by, order_by, window_frame, + window_frame_explicit, null_treatment, distinct, filter, @@ -209,6 +210,7 @@ pub fn serialize_expr( partition_by, order_by, window_frame, + window_frame_explicit: *window_frame_explicit, distinct: *distinct, filter: match filter { Some(e) => Some(Box::new(serialize_expr(e.as_ref(), codec)?)), diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs index 4a3e6669321c4..28eeb01b63c13 100644 --- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs @@ -3313,6 +3313,15 @@ fn roundtrip_substr() { fn roundtrip_window() { let ctx = SessionContext::new(); + let derived_frame = Expr::from(expr::WindowFunction::new( + WindowFunctionDefinition::WindowUDF(rank_udwf()), + vec![], + )) + .order_by(vec![col("col2").sort(true, false)]) + .build() + .unwrap(); + roundtrip_expr_test(derived_frame, ctx.clone()); + // 1. without window_frame let test_expr1 = Expr::from(expr::WindowFunction::new( WindowFunctionDefinition::WindowUDF(rank_udwf()), diff --git a/datafusion/sql/src/expr/function.rs b/datafusion/sql/src/expr/function.rs index f3e4c79740a95..91dd0cbd351f8 100644 --- a/datafusion/sql/src/expr/function.rs +++ b/datafusion/sql/src/expr/function.rs @@ -618,6 +618,7 @@ impl SqlToRel<'_, S> { Some(false) }); + let window_frame_explicit = window.window_frame.is_some(); let window_frame = window .window_frame .as_ref() @@ -675,6 +676,7 @@ impl SqlToRel<'_, S> { partition_by, order_by, window_frame, + window_frame_explicit, filter, null_treatment, distinct: function_args.distinct, @@ -693,6 +695,7 @@ impl SqlToRel<'_, S> { partition_by, order_by, window_frame, + window_frame_explicit, filter, null_treatment, distinct, @@ -705,6 +708,7 @@ impl SqlToRel<'_, S> { partition_by, order_by, window_frame, + window_frame_explicit, filter, null_treatment, distinct, diff --git a/datafusion/sql/src/unparser/expr.rs b/datafusion/sql/src/unparser/expr.rs index 7c7fec38bf070..cafc0d06ef17d 100644 --- a/datafusion/sql/src/unparser/expr.rs +++ b/datafusion/sql/src/unparser/expr.rs @@ -281,6 +281,7 @@ impl Unparser<'_> { partition_by, order_by, window_frame, + window_frame_explicit, filter, distinct, null_treatment, @@ -310,11 +311,12 @@ impl Unparser<'_> { let start_bound = self.convert_bound(&window_frame.start_bound)?; let end_bound = self.convert_bound(&window_frame.end_bound)?; - let window_frame = if self.dialect.window_func_support_window_frame( - func_name, - &start_bound, - &end_bound, - ) { + let window_frame = if *window_frame_explicit + && self.dialect.window_func_support_window_frame( + func_name, + &start_bound, + &end_bound, + ) { Some(ast::WindowFrame { units, start_bound, @@ -2304,12 +2306,13 @@ mod tests { partition_by: vec![], order_by: vec![], window_frame: WindowFrame::new(None), + window_frame_explicit: false, null_treatment: None, distinct: false, filter: None, }, }), - r#"row_number(col) OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)"#, + r#"row_number(col) OVER ()"#, ), ( #[expect(deprecated)] @@ -2331,6 +2334,7 @@ mod tests { ScalarValue::UInt32(Some(2)), ), ), + window_frame_explicit: true, null_treatment: None, distinct: false, filter: Some(Box::new(col("a").gt(lit(100)))), @@ -3475,6 +3479,7 @@ mod tests { let func = WindowFunctionDefinition::WindowUDF(rank_udwf()); let mut window_func = WindowFunction::new(func, vec![]); window_func.params.order_by = vec![Sort::new(col("a"), true, true)]; + window_func.params.window_frame_explicit = true; let expr = Expr::from(window_func); let ast = unparser.expr_to_sql(&expr)?; @@ -4015,6 +4020,7 @@ mod tests { partition_by: vec![], order_by: vec![Sort::new(col("b"), true, true)], window_frame: WindowFrame::new(None), + window_frame_explicit: true, null_treatment: Some(NullTreatment::IgnoreNulls), distinct: false, filter: None, @@ -4029,6 +4035,7 @@ mod tests { partition_by: vec![], order_by: vec![Sort::new(col("b"), true, true)], window_frame: WindowFrame::new(None), + window_frame_explicit: true, null_treatment: Some(NullTreatment::RespectNulls), distinct: false, filter: None, @@ -4043,6 +4050,7 @@ mod tests { partition_by: vec![], order_by: vec![Sort::new(col("b"), true, true)], window_frame: WindowFrame::new(None), + window_frame_explicit: true, null_treatment: None, distinct: false, filter: None, diff --git a/datafusion/sql/tests/cases/plan_to_sql.rs b/datafusion/sql/tests/cases/plan_to_sql.rs index c30e8d3e070c3..2a2842dccc740 100644 --- a/datafusion/sql/tests/cases/plan_to_sql.rs +++ b/datafusion/sql/tests/cases/plan_to_sql.rs @@ -208,6 +208,7 @@ fn roundtrip_statement() -> Result<()> { "SELECT id, count(*) over (PARTITION BY first_name ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING), last_name, sum(id) over (PARTITION BY first_name ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING), first_name from person", + "SELECT id, sum(id) OVER (PARTITION BY first_name ORDER BY id) FROM person", r#"SELECT id, count(distinct id) over (ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING), sum(id) OVER (PARTITION BY first_name ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) from person"#, "SELECT id, sum(id) OVER (PARTITION BY first_name ROWS BETWEEN 5 PRECEDING AND 2 FOLLOWING) from person", @@ -3161,6 +3162,7 @@ fn test_unparse_window() -> Result<()> { partition_by: vec![col("k")], order_by: vec![col("v").sort(true, true)], window_frame: WindowFrame::new(None), + window_frame_explicit: true, null_treatment: None, distinct: false, filter: None, @@ -3239,6 +3241,7 @@ fn test_unparse_window_over_aggregate_without_projection() -> Result<()> { partition_by: vec![], order_by: vec![col("time").sort(true, true)], window_frame: WindowFrame::new(None), + window_frame_explicit: true, null_treatment: None, distinct: false, filter: None, @@ -3272,6 +3275,7 @@ fn test_unparse_filter_on_window_over_aggregate_without_projection() -> Result<( partition_by: vec![], order_by: vec![col("time").sort(true, true)], window_frame: WindowFrame::new(None), + window_frame_explicit: true, null_treatment: None, distinct: false, filter: None, @@ -3307,6 +3311,7 @@ fn test_unparse_filter_on_aggregate_output_above_window_without_projection() -> partition_by: vec![], order_by: vec![col("time").sort(true, true)], window_frame: WindowFrame::new(None), + window_frame_explicit: true, null_treatment: None, distinct: false, filter: None, @@ -3341,6 +3346,7 @@ fn test_unparse_window_over_table_scan_without_projection() -> Result<()> { partition_by: vec![col("k")], order_by: vec![col("v").sort(true, true)], window_frame: WindowFrame::new(None), + window_frame_explicit: true, null_treatment: None, distinct: false, filter: None, @@ -3373,6 +3379,7 @@ fn test_unparse_stacked_windows_without_projection() -> Result<()> { partition_by: vec![col("k")], order_by: vec![col("v").sort(true, true)], window_frame: WindowFrame::new(None), + window_frame_explicit: true, null_treatment: None, distinct: false, filter: None, @@ -3386,6 +3393,7 @@ fn test_unparse_stacked_windows_without_projection() -> Result<()> { partition_by: vec![], order_by: vec![col("v").sort(false, false)], window_frame: WindowFrame::new(None), + window_frame_explicit: true, null_treatment: None, distinct: false, filter: None, @@ -3419,6 +3427,7 @@ fn test_unparse_window_over_distinct_without_projection() -> Result<()> { partition_by: vec![], order_by: vec![col("v").sort(true, true)], window_frame: WindowFrame::new(None), + window_frame_explicit: true, null_treatment: None, distinct: false, filter: None, @@ -3452,6 +3461,7 @@ fn test_unparse_window_over_limit_without_projection() -> Result<()> { partition_by: vec![], order_by: vec![col("v").sort(true, true)], window_frame: WindowFrame::new(None), + window_frame_explicit: true, null_treatment: None, distinct: false, filter: None, @@ -3485,6 +3495,7 @@ fn test_unparse_window_over_projection_without_projection() -> Result<()> { partition_by: vec![], order_by: vec![col("v_alias").sort(true, true)], window_frame: WindowFrame::new(None), + window_frame_explicit: true, null_treatment: None, distinct: false, filter: None, @@ -3521,6 +3532,7 @@ fn test_unparse_window_over_derived_aggregate_without_projection() -> Result<()> .sort(true, true), ], window_frame: WindowFrame::new(None), + window_frame_explicit: true, null_treatment: None, distinct: false, filter: None, 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 bd1a63b7040f5..3f5b040509417 100644 --- a/datafusion/substrait/src/logical_plan/consumer/expr/window_function.rs +++ b/datafusion/substrait/src/logical_plan/consumer/expr/window_function.rs @@ -124,6 +124,8 @@ pub async fn from_window_function( .await?, order_by, window_frame, + window_frame_explicit: window.lower_bound.is_some() + || window.upper_bound.is_some(), filter: None, null_treatment: None, 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 a262842ae39ce..850d30d93282c 100644 --- a/datafusion/substrait/src/logical_plan/producer/expr/window_function.rs +++ b/datafusion/substrait/src/logical_plan/producer/expr/window_function.rs @@ -44,6 +44,7 @@ pub fn from_window_function( partition_by, order_by, window_frame, + window_frame_explicit, null_treatment, distinct, filter, @@ -79,7 +80,16 @@ pub fn from_window_function( .map(|e| substrait_sort_field(producer, e, schema)) .collect::>>()?; // window frame - let bounds = to_substrait_bounds(window_frame)?; + // Substrait defaults missing bounds to the whole partition. Omit them only + // when that is also the resolved DataFusion frame; a running frame must + // retain its upper bound to preserve the query result. + let omit_bounds = !window_frame_explicit && window_frame == &WindowFrame::new(None); + let bounds = if omit_bounds { + (None, None) + } else { + let (lower, upper) = to_substrait_bounds(window_frame)?; + (Some(lower), Some(upper)) + }; let bound_type = to_substrait_bound_type(window_frame)?; // Substrait documents `output_type` as "Must be set to the return type of // the function, exactly as derived using the declaration in the extension", @@ -105,8 +115,8 @@ pub fn from_window_function( } else { AggregationInvocation::All as i32 }, - lower_bound: Some(bounds.0), - upper_bound: Some(bounds.1), + lower_bound: bounds.0, + upper_bound: bounds.1, args: vec![], bounds_type: bound_type as i32, })), diff --git a/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs b/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs index bcc7becf0af04..79cff3c85bd07 100644 --- a/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs +++ b/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs @@ -2381,7 +2381,33 @@ async fn roundtrip_window_udf() -> Result<()> { let ctx = create_context().await?; ctx.register_udwf(dummy_agg); - roundtrip_with_ctx("select dummy_window(a) OVER () from data", ctx).await?; + fn frame_is_explicit(plan: &LogicalPlan) -> Option { + match plan { + LogicalPlan::Window(window) => window.window_expr.iter().find_map(|expr| { + if let Expr::WindowFunction(fun) = expr { + Some(fun.params.window_frame_explicit) + } else { + None + } + }), + _ => plan.inputs().into_iter().find_map(frame_is_explicit), + } + } + for (sql, expected_explicit) in [ + ("select dummy_window(a) OVER () from data", false), + ( + "select dummy_window(a) OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) from data", + true, + ), + ] { + let plan = ctx.sql(sql).await?.into_optimized_plan()?; + let roundtripped = substrait_roundtrip(&plan, &ctx).await?; + assert_eq!(frame_is_explicit(&roundtripped), Some(expected_explicit)); + roundtrip_with_ctx(sql, ctx.clone()).await?; + } + // An ordered default frame needs an upper bound in Substrait to retain + // execution semantics, even though that format cannot retain provenance. + roundtrip_with_ctx("select dummy_window(a) OVER (ORDER BY a) from data", ctx).await?; Ok(()) } diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index 62af520611296..819ec96af9fa3 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -25,6 +25,16 @@ in this section pertains to features and changes that have already been merged to the main branch and are awaiting release in this version. +### Window frame provenance in Rust expressions + +`WindowFunctionParams` and `RawWindowExpr` now include +`window_frame_explicit: bool`. Code that constructs either struct with a literal +must set the new field. Use `true` when the frame was specified by the caller +(including a frame equal to the default), and `false` when it was derived from +the window ordering. The resolved `window_frame` remains required in both cases. +This lets later `ExprFunctionExt` calls preserve explicit frames while deriving +a new default frame when `ORDER BY` changes. + ### Session time zone now applies to mixed timestamp operations When `datafusion.execution.time_zone` is set, DataFusion now uses it to From 98566af35675a2b1d9e029ab5d864ac502e84990 Mon Sep 17 00:00:00 2001 From: goutamadwant Date: Sun, 25 Oct 2026 00:21:37 -0700 Subject: [PATCH 2/2] test: cover implicit window frames and refresh expectations --- datafusion/core/tests/expr_api/mod.rs | 37 ++++++++++++++++++++ datafusion/expr/src/udaf.rs | 2 ++ datafusion/optimizer/src/push_down_filter.rs | 14 ++++---- datafusion/sql/tests/cases/plan_to_sql.rs | 2 +- 4 files changed, 47 insertions(+), 8 deletions(-) diff --git a/datafusion/core/tests/expr_api/mod.rs b/datafusion/core/tests/expr_api/mod.rs index 0961885a0cf87..67981436b818e 100644 --- a/datafusion/core/tests/expr_api/mod.rs +++ b/datafusion/core/tests/expr_api/mod.rs @@ -120,6 +120,43 @@ fn test_window_frame_provenance_does_not_change_expression_equality() { 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 { diff --git a/datafusion/expr/src/udaf.rs b/datafusion/expr/src/udaf.rs index 0bac5979870c8..b464114ad4b20 100644 --- a/datafusion/expr/src/udaf.rs +++ b/datafusion/expr/src/udaf.rs @@ -1220,6 +1220,7 @@ pub fn udaf_default_human_display( /// # partition_by: vec![col("y")], /// # order_by: vec![], /// # window_frame: WindowFrame::new(None), +/// # window_frame_explicit: false, /// # filter: None, /// # null_treatment: None, /// # distinct: false, @@ -1412,6 +1413,7 @@ pub fn udaf_default_display_name( /// # partition_by: vec![col("y")], /// # order_by: vec![], /// # window_frame: WindowFrame::new(None), +/// # window_frame_explicit: false, /// # filter: None, /// # null_treatment: None, /// # distinct: false, diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs index becc789d81029..9d09fe50e2b0d 100644 --- a/datafusion/optimizer/src/push_down_filter.rs +++ b/datafusion/optimizer/src/push_down_filter.rs @@ -1980,7 +1980,7 @@ mod tests { assert_optimized_plan_equal!( plan, @r" - WindowAggr: windowExpr=[[rank() PARTITION BY [test.a, test.b] ORDER BY [test.c ASC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] + WindowAggr: windowExpr=[[rank() PARTITION BY [test.a, test.b] ORDER BY [test.c ASC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] TableScan: test, full_filters=[test.b > Int64(10)] " ) @@ -2015,7 +2015,7 @@ mod tests { assert_optimized_plan_equal!( plan, @r" - WindowAggr: windowExpr=[[rank() PARTITION BY [test.$a, test.$b] ORDER BY [test.$c ASC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] + WindowAggr: windowExpr=[[rank() PARTITION BY [test.$a, test.$b] ORDER BY [test.$c ASC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] TableScan: test, full_filters=[test.$b > Int64(10)] " ) @@ -2046,7 +2046,7 @@ mod tests { assert_optimized_plan_equal!( plan, @r" - WindowAggr: windowExpr=[[rank() PARTITION BY [test.a, test.b] ORDER BY [test.c ASC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] + WindowAggr: windowExpr=[[rank() PARTITION BY [test.a, test.b] ORDER BY [test.c ASC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] TableScan: test, full_filters=[test.a > Int64(10), test.b = Int64(1)] " ) @@ -2077,7 +2077,7 @@ mod tests { plan, @r" Filter: test.b = Int64(1) - WindowAggr: windowExpr=[[rank() PARTITION BY [test.a] ORDER BY [test.c ASC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] + WindowAggr: windowExpr=[[rank() PARTITION BY [test.a] ORDER BY [test.c ASC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] TableScan: test, full_filters=[test.a > Int64(10)] " ) @@ -2110,7 +2110,7 @@ mod tests { assert_optimized_plan_equal!( plan, @r" - WindowAggr: windowExpr=[[rank() PARTITION BY [test.a + test.b] ORDER BY [test.c ASC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] + WindowAggr: windowExpr=[[rank() PARTITION BY [test.a + test.b] ORDER BY [test.c ASC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] TableScan: test, full_filters=[test.a + test.b > Int64(10)] " ) @@ -2152,7 +2152,7 @@ mod tests { assert_optimized_plan_equal!( plan, @r" - WindowAggr: windowExpr=[[rank() PARTITION BY [test.a] ORDER BY [test.c ASC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW, rank() PARTITION BY [test.b, test.a] ORDER BY [test.c ASC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] + WindowAggr: windowExpr=[[rank() PARTITION BY [test.a] ORDER BY [test.c ASC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW, rank() PARTITION BY [test.b, test.a] ORDER BY [test.c ASC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] TableScan: test, full_filters=[test.a > Int64(10)] " ) @@ -2195,7 +2195,7 @@ mod tests { plan, @r" Filter: test.b > Int64(10) - WindowAggr: windowExpr=[[rank() PARTITION BY [test.a] ORDER BY [test.c ASC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW, rank() PARTITION BY [test.b, test.a] ORDER BY [test.c ASC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] + WindowAggr: windowExpr=[[rank() PARTITION BY [test.a] ORDER BY [test.c ASC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW, rank() PARTITION BY [test.b, test.a] ORDER BY [test.c ASC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] TableScan: test " ) diff --git a/datafusion/sql/tests/cases/plan_to_sql.rs b/datafusion/sql/tests/cases/plan_to_sql.rs index 2a2842dccc740..2d135d9a8d4a4 100644 --- a/datafusion/sql/tests/cases/plan_to_sql.rs +++ b/datafusion/sql/tests/cases/plan_to_sql.rs @@ -2351,7 +2351,7 @@ fn test_aggregation_to_sql() { let statement = generate_round_trip_statement(GenericDialect {}, sql); assert_snapshot!( statement, - @"SELECT person.id, person.first_name, sum(person.id) AS total_sum, sum(person.id) OVER (PARTITION BY person.first_name ROWS BETWEEN 5 PRECEDING AND 2 FOLLOWING) AS moving_sum, sum(person.id) FILTER (WHERE ((person.id > 50) AND (person.first_name = 'John'))) OVER (PARTITION BY person.first_name ROWS BETWEEN 5 PRECEDING AND 2 FOLLOWING) AS filtered_sum, max(sum(person.id)) OVER (PARTITION BY person.first_name ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS max_total, rank() OVER (PARTITION BY (grouping(person.id) + grouping(person.age)), CASE WHEN (grouping(person.age) = 0) THEN person.id END ORDER BY sum(person.id) DESC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS rank_within_parent_1, rank() OVER (PARTITION BY (grouping(person.age) + grouping(person.id)), CASE WHEN (CAST(grouping(person.age) AS BIGINT) = 0) THEN person.id END ORDER BY sum(person.id) DESC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS rank_within_parent_2 FROM person GROUP BY person.id, person.first_name", + @"SELECT person.id, person.first_name, sum(person.id) AS total_sum, sum(person.id) OVER (PARTITION BY person.first_name ROWS BETWEEN 5 PRECEDING AND 2 FOLLOWING) AS moving_sum, sum(person.id) FILTER (WHERE ((person.id > 50) AND (person.first_name = 'John'))) OVER (PARTITION BY person.first_name ROWS BETWEEN 5 PRECEDING AND 2 FOLLOWING) AS filtered_sum, max(sum(person.id)) OVER (PARTITION BY person.first_name ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS max_total, rank() OVER (PARTITION BY (grouping(person.id) + grouping(person.age)), CASE WHEN (grouping(person.age) = 0) THEN person.id END ORDER BY sum(person.id) DESC NULLS FIRST) AS rank_within_parent_1, rank() OVER (PARTITION BY (grouping(person.age) + grouping(person.id)), CASE WHEN (CAST(grouping(person.age) AS BIGINT) = 0) THEN person.id END ORDER BY sum(person.id) DESC NULLS FIRST) AS rank_within_parent_2 FROM person GROUP BY person.id, person.first_name", ); }