diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs index 4738904916d5a..efd5a61e4ed9f 100644 --- a/datafusion/optimizer/src/push_down_filter.rs +++ b/datafusion/optimizer/src/push_down_filter.rs @@ -1135,8 +1135,8 @@ impl OptimizerRule for PushDownFilter { // multiple window functions, each with potentially different partition keys. // Therefore, we need to ensure that any potential partition key returned is used in // ALL window functions. Otherwise, filters cannot be pushed by through that column. - fn extract_partition_keys(func: &WindowFunction) -> HashSet { - expr_columns(&func.params.partition_by) + fn extract_partition_keys(func: &WindowFunction) -> HashSet<&Expr> { + func.params.partition_by.iter().collect() } let potential_partition_keys = window @@ -1172,8 +1172,11 @@ impl OptimizerRule for PushDownFilter { let mut keep_predicates = vec![]; let mut push_predicates = vec![]; for expr in predicates { - let cols = expr.column_refs(); - if cols.iter().all(|c| potential_partition_keys.contains(c)) { + // A volatile predicate has to stay above the window: pushing it + // changes which rows the window function sees. + if !expr.is_volatile() + && reads_only_partition_keys(&expr, &potential_partition_keys)? + { push_predicates.push(expr); } else { keep_predicates.push(expr); @@ -1181,12 +1184,11 @@ impl OptimizerRule for PushDownFilter { } // Unlike with aggregations, there are no cases where we have to replace, e.g., - // `a+b` with Column(a)+Column(b). This is because partition expressions are not - // available as standalone columns to the user. For example, while an aggregation on - // `a+b` becomes Column(a + b), in a window partition it becomes - // `func() PARTITION BY [a + b] ...`. Thus, filters on expressions always remain in - // place, so we can use `push_predicates` directly. This is consistent with other - // optimizers, such as the one used by Postgres. + // `a+b` with Column(a+b). This is because partition expressions are not available + // as standalone columns to the user: while an aggregation on `a+b` becomes + // Column(a + b), in a window partition it stays `func() PARTITION BY [a + b] ...`. + // That is why the predicate is matched against the key expressions themselves and + // can be pushed unchanged. // If we have a filter to push, we push it down to the input of the aggregate let result = if let Some(predicate) = conjunction(push_predicates) { @@ -1585,6 +1587,54 @@ fn with_filters(predicates: Vec, plan: LogicalPlan) -> LogicalPlan { } } +/// Can `expr` be evaluated below a window with these `PARTITION BY` keys? +/// +/// A predicate that reads only the partition keys is constant within each +/// partition, so filtering before the window drops whole partitions and leaves +/// the surviving rows' window values unchanged. "Reads only the keys" is checked +/// structurally: every column reference must sit inside a subtree that is equal +/// to one of the keys. Given `PARTITION BY a, b + c`: +/// +/// * `a < 5`, `b + c = 4` and `(b + c) + 1 > 10` can be pushed down +/// * `d < 5` and `b < 5` cannot (`b` on its own is not a key), and neither can +/// `c + b = 4` (the match is structural, `c + b` is not `b + c`) +/// +/// A predicate containing a subquery is never pushed: what the subquery reads is +/// not visible from the expression tree. +fn reads_only_partition_keys( + expr: &Expr, + partition_keys: &HashSet<&Expr>, +) -> Result { + let mut reads_something_else = false; + expr.apply(|node| { + Ok(if partition_keys.contains(&node) { + // the whole key was matched, so whatever it reads is accounted for + TreeNodeRecursion::Jump + } else if reads_beyond_this_node(node) { + reads_something_else = true; + TreeNodeRecursion::Stop + } else { + TreeNodeRecursion::Continue + }) + })?; + Ok(!reads_something_else) +} + +/// Does this node read data that walking its children cannot account for? A +/// column reads itself; a subquery reads whatever its plan reads, which +/// `Expr::apply` does not visit. +fn reads_beyond_this_node(node: &Expr) -> bool { + matches!( + node, + Expr::Column(_) + | Expr::OuterReferenceColumn(..) + | Expr::ScalarSubquery(_) + | Expr::Exists(_) + | Expr::InSubquery(_) + | Expr::SetComparison(_) + ) +} + fn expr_columns(exprs: &[Expr]) -> HashSet { exprs .iter() @@ -1956,10 +2006,12 @@ mod tests { ) } - /// verifies that filters on partition expressions are not pushed, as the single expression - /// column is not available to the user, unlike with aggregations + /// verifies that a filter on an expression partition key is pushed; the + /// remaining shapes (mixed keys, operand order, subqueries, volatile + /// predicates, several windows) are covered in + /// `push_down_filter_regression.slt` #[test] - fn filter_expression_keep_window() -> Result<()> { + fn filter_expression_move_window() -> Result<()> { let table_scan = test_table_scan()?; let window = Expr::from(WindowFunction::new( @@ -1975,49 +2027,14 @@ mod tests { let plan = LogicalPlanBuilder::from(table_scan) .window(vec![window])? - // unlike with aggregations, single partition column "test.a + test.b" is not available - // to the plan, so we use multiple columns when filtering - .filter(add(col("a"), col("b")).gt(lit(10i64)))? - .build()?; - - assert_optimized_plan_equal!( - plan, - @r" - Filter: test.a + test.b > Int64(10) - WindowAggr: windowExpr=[[rank() PARTITION BY [test.a + test.b] ORDER BY [test.c ASC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] - TableScan: test - " - ) - } - - /// verifies that filters are not pushed on order by columns (that are not used in partitioning) - #[test] - fn filter_order_keep_window() -> Result<()> { - let table_scan = test_table_scan()?; - - let window = Expr::from(WindowFunction::new( - WindowFunctionDefinition::WindowUDF( - datafusion_functions_window::rank::rank_udwf(), - ), - vec![], - )) - .partition_by(vec![col("a")]) - .order_by(vec![col("c").sort(true, true)]) - .build() - .unwrap(); - - let plan = LogicalPlanBuilder::from(table_scan) - .window(vec![window])? - .filter(col("c").gt(lit(10i64)))? + .filter(add(col("a"), col("b")).gt(lit(10i64)))? // a + b > 10 .build()?; - assert_plan_not_transformed!(plan.clone()); assert_optimized_plan_equal!( plan, @r" - Filter: test.c > Int64(10) - WindowAggr: windowExpr=[[rank() PARTITION BY [test.a] ORDER BY [test.c ASC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] - TableScan: test + WindowAggr: windowExpr=[[rank() PARTITION BY [test.a + test.b] ORDER BY [test.c ASC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] + TableScan: test, full_filters=[test.a + test.b > Int64(10)] " ) } diff --git a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt index 20962e15486de..f85d568efc789 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt @@ -723,3 +723,291 @@ query I SELECT sum(c) FROM (SELECT random() < 0.5 AS k, count(*) AS c FROM generate_series(1, 10000) GROUP BY random() < 0.5) WHERE k OR NOT k; ---- 10000 + +# Window filter pushdown over expression PARTITION BY keys. +# +# A predicate that reads only the partition keys is constant within a partition, +# so it is pushed below the window; anything else stays above it. + +statement ok +set datafusion.explain.logical_plan_only = true; + +statement ok +create table window_expr_key(k varchar, v int) as values + ('a', 1), ('a', 2), ('a', 3), + ('', 4), ('', 5), + ('b', 6); + +# Partitions under NULLIF(k, '') are 'a' => {1,2,3}, NULL => {4,5}, 'b' => {6}. +query TII +SELECT k, v, s FROM ( + SELECT k, v, SUM(v) OVER (PARTITION BY NULLIF(k, '')) AS s + FROM window_expr_key +) +ORDER BY v; +---- +a 1 6 +a 2 6 +a 3 6 +(empty) 4 9 +(empty) 5 9 +b 6 6 + +# Pushed: the predicate reads only the partition key. +query TT +EXPLAIN SELECT k, v, s FROM ( + SELECT k, v, SUM(v) OVER (PARTITION BY NULLIF(k, '')) AS s + FROM window_expr_key +) +WHERE NULLIF(k, '') = 'a'; +---- +logical_plan +01)Projection: window_expr_key.k, window_expr_key.v, sum(window_expr_key.v) PARTITION BY [nullif(window_expr_key.k,Utf8(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS s +02)--WindowAggr: windowExpr=[[sum(CAST(window_expr_key.v AS Int64)) PARTITION BY [nullif(window_expr_key.k, Utf8View(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS sum(window_expr_key.v) PARTITION BY [nullif(window_expr_key.k,Utf8(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]] +03)----Filter: nullif(window_expr_key.k, Utf8View("")) = Utf8View("a") +04)------TableScan: window_expr_key projection=[k, v] + +query TII +SELECT k, v, s FROM ( + SELECT k, v, SUM(v) OVER (PARTITION BY NULLIF(k, '')) AS s + FROM window_expr_key +) +WHERE NULLIF(k, '') = 'a' +ORDER BY v; +---- +a 1 6 +a 2 6 +a 3 6 + +# Pushed: a predicate built on top of the key still reads only the key. +query TT +EXPLAIN SELECT k, v, s FROM ( + SELECT k, v, SUM(v) OVER (PARTITION BY NULLIF(k, '')) AS s + FROM window_expr_key +) +WHERE NULLIF(k, '') IS NOT NULL; +---- +logical_plan +01)Projection: window_expr_key.k, window_expr_key.v, sum(window_expr_key.v) PARTITION BY [nullif(window_expr_key.k,Utf8(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS s +02)--WindowAggr: windowExpr=[[sum(CAST(window_expr_key.v AS Int64)) PARTITION BY [nullif(window_expr_key.k, Utf8View(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS sum(window_expr_key.v) PARTITION BY [nullif(window_expr_key.k,Utf8(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]] +03)----Filter: nullif(window_expr_key.k, Utf8View("")) IS NOT NULL +04)------TableScan: window_expr_key projection=[k, v] + +query TII +SELECT k, v, s FROM ( + SELECT k, v, SUM(v) OVER (PARTITION BY NULLIF(k, '')) AS s + FROM window_expr_key +) +WHERE NULLIF(k, '') IS NOT NULL +ORDER BY v; +---- +a 1 6 +a 2 6 +a 3 6 +b 6 6 + +# Kept: `v` is not a partition key. Pushing it would recompute the sums over the +# surviving rows and report 5 instead of 6 for partition 'a'. +query TT +EXPLAIN SELECT k, v, s FROM ( + SELECT k, v, SUM(v) OVER (PARTITION BY NULLIF(k, '')) AS s + FROM window_expr_key +) +WHERE v > 1; +---- +logical_plan +01)Projection: window_expr_key.k, window_expr_key.v, sum(window_expr_key.v) PARTITION BY [nullif(window_expr_key.k,Utf8(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS s +02)--Filter: window_expr_key.v > Int32(1) +03)----WindowAggr: windowExpr=[[sum(CAST(window_expr_key.v AS Int64)) PARTITION BY [nullif(window_expr_key.k, Utf8View(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS sum(window_expr_key.v) PARTITION BY [nullif(window_expr_key.k,Utf8(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]] +04)------TableScan: window_expr_key projection=[k, v] + +query TII +SELECT k, v, s FROM ( + SELECT k, v, SUM(v) OVER (PARTITION BY NULLIF(k, '')) AS s + FROM window_expr_key +) +WHERE v > 1 +ORDER BY v; +---- +a 2 6 +a 3 6 +(empty) 4 9 +(empty) 5 9 +b 6 6 + +# Kept: `k` on its own is not a key, only NULLIF(k, '') is. +query TT +EXPLAIN SELECT k, v, s FROM ( + SELECT k, v, SUM(v) OVER (PARTITION BY NULLIF(k, '')) AS s + FROM window_expr_key +) +WHERE k = 'a'; +---- +logical_plan +01)Projection: window_expr_key.k, window_expr_key.v, sum(window_expr_key.v) PARTITION BY [nullif(window_expr_key.k,Utf8(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS s +02)--Filter: window_expr_key.k = Utf8View("a") +03)----WindowAggr: windowExpr=[[sum(CAST(window_expr_key.v AS Int64)) PARTITION BY [nullif(window_expr_key.k, Utf8View(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS sum(window_expr_key.v) PARTITION BY [nullif(window_expr_key.k,Utf8(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]] +04)------TableScan: window_expr_key projection=[k, v] + +# Kept: a subquery, even one whose left-hand side is the key. +query TT +EXPLAIN SELECT k, v, s FROM ( + SELECT k, v, SUM(v) OVER (PARTITION BY NULLIF(k, '')) AS s + FROM window_expr_key +) +WHERE NULLIF(k, '') IN (SELECT k FROM window_expr_key WHERE v > 5); +---- +logical_plan +01)LeftSemi Join: nullif(window_expr_key.k, Utf8View("")) = __correlated_sq_1.k +02)--Projection: window_expr_key.k, window_expr_key.v, sum(window_expr_key.v) PARTITION BY [nullif(window_expr_key.k,Utf8(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS s +03)----WindowAggr: windowExpr=[[sum(CAST(window_expr_key.v AS Int64)) PARTITION BY [nullif(window_expr_key.k, Utf8View(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS sum(window_expr_key.v) PARTITION BY [nullif(window_expr_key.k,Utf8(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]] +04)------TableScan: window_expr_key projection=[k, v] +05)--SubqueryAlias: __correlated_sq_1 +06)----Projection: window_expr_key.k +07)------Filter: window_expr_key.v > Int32(5) +08)--------TableScan: window_expr_key projection=[k, v] + +query TII +SELECT k, v, s FROM ( + SELECT k, v, SUM(v) OVER (PARTITION BY NULLIF(k, '')) AS s + FROM window_expr_key +) +WHERE NULLIF(k, '') IN (SELECT k FROM window_expr_key WHERE v > 5) +ORDER BY v; +---- +b 6 6 + +# Kept: a scalar subquery. +query TT +EXPLAIN SELECT k, v, s FROM ( + SELECT k, v, SUM(v) OVER (PARTITION BY NULLIF(k, '')) AS s + FROM window_expr_key +) +WHERE NULLIF(k, '') = (SELECT max(k) FROM window_expr_key); +---- +logical_plan +01)Projection: window_expr_key.k, window_expr_key.v, sum(window_expr_key.v) PARTITION BY [nullif(window_expr_key.k,Utf8(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS s +02)--Filter: nullif(window_expr_key.k, Utf8View("")) = () +03)----Subquery: +04)------Aggregate: groupBy=[[]], aggr=[[max(window_expr_key.k)]] +05)--------TableScan: window_expr_key projection=[k] +06)----WindowAggr: windowExpr=[[sum(CAST(window_expr_key.v AS Int64)) PARTITION BY [nullif(window_expr_key.k, Utf8View(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS sum(window_expr_key.v) PARTITION BY [nullif(window_expr_key.k,Utf8(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]] +07)------TableScan: window_expr_key projection=[k, v] + +# Kept: a volatile predicate, even one that reads no column at all. +query TT +EXPLAIN SELECT k, v, s FROM ( + SELECT k, v, SUM(v) OVER (PARTITION BY NULLIF(k, '')) AS s + FROM window_expr_key +) +WHERE random() < 2; +---- +logical_plan +01)Projection: window_expr_key.k, window_expr_key.v, sum(window_expr_key.v) PARTITION BY [nullif(window_expr_key.k,Utf8(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS s +02)--Filter: random() < Float64(2) +03)----WindowAggr: windowExpr=[[sum(CAST(window_expr_key.v AS Int64)) PARTITION BY [nullif(window_expr_key.k, Utf8View(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS sum(window_expr_key.v) PARTITION BY [nullif(window_expr_key.k,Utf8(""))] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]] +04)------TableScan: window_expr_key projection=[k, v] + +statement ok +drop table window_expr_key; + +statement ok +create table window_ab(a int, b int, c int) as values + (1, 1, 10), (1, 2, 20), (2, 2, 30), (3, 3, 40); + +# Pushed: an arithmetic expression key. +query TT +EXPLAIN SELECT a, b, s FROM ( + SELECT a, b, SUM(c) OVER (PARTITION BY a + b) AS s FROM window_ab +) +WHERE a + b > 2; +---- +logical_plan +01)Projection: window_ab.a, window_ab.b, sum(window_ab.c) PARTITION BY [window_ab.a + window_ab.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS s +02)--WindowAggr: windowExpr=[[sum(CAST(window_ab.c AS Int64)) PARTITION BY [window_ab.a + window_ab.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]] +03)----Filter: window_ab.a + window_ab.b > Int32(2) +04)------TableScan: window_ab projection=[a, b, c] + +# Pushed: the key appears twice in one predicate. +query TT +EXPLAIN SELECT a, b, s FROM ( + SELECT a, b, SUM(c) OVER (PARTITION BY a + b) AS s FROM window_ab +) +WHERE a + b > 5 OR a + b < 3; +---- +logical_plan +01)Projection: window_ab.a, window_ab.b, sum(window_ab.c) PARTITION BY [window_ab.a + window_ab.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS s +02)--WindowAggr: windowExpr=[[sum(CAST(window_ab.c AS Int64)) PARTITION BY [window_ab.a + window_ab.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]] +03)----Projection: window_ab.a, window_ab.b, window_ab.c +04)------Filter: __common_expr_3 > Int32(5) OR __common_expr_3 < Int32(3) +05)--------Projection: window_ab.a + window_ab.b AS __common_expr_3, window_ab.a, window_ab.b, window_ab.c +06)----------TableScan: window_ab projection=[a, b, c] + +# Pushed: a predicate over a column key and an expression key of the same window. +query TT +EXPLAIN SELECT a, b, s FROM ( + SELECT a, b, SUM(c) OVER (PARTITION BY a, a + b) AS s FROM window_ab +) +WHERE a > 1 OR a + b > 4; +---- +logical_plan +01)Projection: window_ab.a, window_ab.b, sum(window_ab.c) PARTITION BY [window_ab.a, window_ab.a + window_ab.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS s +02)--WindowAggr: windowExpr=[[sum(CAST(window_ab.c AS Int64)) PARTITION BY [window_ab.a, window_ab.a + window_ab.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]] +03)----Filter: window_ab.a > Int32(1) OR window_ab.a + window_ab.b > Int32(4) +04)------TableScan: window_ab projection=[a, b, c] + +# Kept: the match is structural, `b + a` is not the key `a + b`. +query TT +EXPLAIN SELECT a, b, s FROM ( + SELECT a, b, SUM(c) OVER (PARTITION BY a + b) AS s FROM window_ab +) +WHERE b + a > 2; +---- +logical_plan +01)Projection: window_ab.a, window_ab.b, sum(window_ab.c) PARTITION BY [window_ab.a + window_ab.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS s +02)--Filter: window_ab.b + window_ab.a > Int32(2) +03)----Projection: window_ab.a, window_ab.b, sum(window_ab.c) PARTITION BY [window_ab.a + window_ab.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING +04)------WindowAggr: windowExpr=[[sum(CAST(window_ab.c AS Int64)) PARTITION BY [window_ab.a + window_ab.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]] +05)--------TableScan: window_ab projection=[a, b, c] + +# Pushed: every window function partitions by the key. +query TT +EXPLAIN SELECT a, b, s, r FROM ( + SELECT a, b, + SUM(c) OVER (PARTITION BY a + b) AS s, + RANK() OVER (PARTITION BY a + b ORDER BY c) AS r + FROM window_ab +) +WHERE a + b > 2; +---- +logical_plan +01)Projection: window_ab.a, window_ab.b, sum(window_ab.c) PARTITION BY [window_ab.a + window_ab.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS s, rank() PARTITION BY [window_ab.a + window_ab.b] ORDER BY [window_ab.c ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS r +02)--WindowAggr: windowExpr=[[sum(CAST(window_ab.c AS Int64)) PARTITION BY [__common_expr_1 AS window_ab.a + window_ab.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]] +03)----WindowAggr: windowExpr=[[rank() PARTITION BY [__common_expr_1 AS window_ab.a + window_ab.b] ORDER BY [window_ab.c ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] +04)------Projection: window_ab.a + window_ab.b AS __common_expr_1, window_ab.a, window_ab.b, window_ab.c +05)--------Filter: window_ab.a + window_ab.b > Int32(2) +06)----------TableScan: window_ab projection=[a, b, c] + +# Kept: one window function partitions by something else. +query TT +EXPLAIN SELECT a, b, s, r FROM ( + SELECT a, b, + SUM(c) OVER (PARTITION BY a + b) AS s, + RANK() OVER (PARTITION BY a ORDER BY c) AS r + FROM window_ab +) +WHERE a + b > 2; +---- +logical_plan +01)Projection: window_ab.a, window_ab.b, sum(window_ab.c) PARTITION BY [window_ab.a + window_ab.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS s, rank() PARTITION BY [window_ab.a] ORDER BY [window_ab.c ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS r +02)--Filter: window_ab.a + window_ab.b > Int32(2) +03)----Projection: window_ab.a, window_ab.b, sum(window_ab.c) PARTITION BY [window_ab.a + window_ab.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING, rank() PARTITION BY [window_ab.a] ORDER BY [window_ab.c ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW +04)------WindowAggr: windowExpr=[[rank() PARTITION BY [window_ab.a] ORDER BY [window_ab.c ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] +05)--------WindowAggr: windowExpr=[[sum(CAST(window_ab.c AS Int64)) PARTITION BY [window_ab.a + window_ab.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]] +06)----------TableScan: window_ab projection=[a, b, c] + +statement ok +drop table window_ab; + +statement ok +set datafusion.explain.logical_plan_only = false;