From 00b626b58af61df80cec61feadcfb121a6facf43 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Mon, 28 Sep 2026 10:46:44 -0500 Subject: [PATCH 1/8] fix: pull a correlated filter only through plan nodes known to be safe `PullUpCorrelatedExpr::f_down` moved a correlated filter above every plan node that it did not block explicitly. Each node it did not model was a possible wrong result. An ASOF join is one: a correlated filter on its right side selects the rows the join can match, and above the join it removes rows instead (https://github.com/apache/datafusion/issues/25834). The match is now exhaustive. A node is a pass-through only when it is in the list of safe nodes. Window, DISTINCT ON, Unnest, RecursiveQuery and the nodes that are not queries keep the subquery correlated when their input holds an outer reference. An ASOF join keeps the subquery correlated when its right input holds an outer reference. Co-Authored-By: Claude Opus 5.5 --- datafusion/optimizer/src/decorrelate.rs | 85 +++++++++++++++++++++---- 1 file changed, 71 insertions(+), 14 deletions(-) diff --git a/datafusion/optimizer/src/decorrelate.rs b/datafusion/optimizer/src/decorrelate.rs index 4f89532e50525..b43c0471cb336 100644 --- a/datafusion/optimizer/src/decorrelate.rs +++ b/datafusion/optimizer/src/decorrelate.rs @@ -36,8 +36,8 @@ use datafusion_expr::utils::{ collect_subquery_cols, conjunction, find_join_exprs, split_conjunction, }; use datafusion_expr::{ - BinaryExpr, Cast, EmptyRelation, Expr, ExprSchemable, FetchType, LogicalPlan, - LogicalPlanBuilder, Operator, expr, lit, + BinaryExpr, Cast, Distinct, EmptyRelation, Expr, ExprSchemable, FetchType, + LogicalPlan, LogicalPlanBuilder, Operator, expr, lit, }; /// This struct rewrite the sub query plan by pull up the correlated @@ -136,6 +136,9 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { type Node = LogicalPlan; fn f_down(&mut self, plan: LogicalPlan) -> Result> { + // This match is exhaustive on purpose. A correlated filter moves above + // a node only when the node is known to keep the result the same. Add + // each new `LogicalPlan` variant to one of the groups below. match plan { LogicalPlan::Filter(_) => Ok(Transformed::no(plan)), // Subquery nodes are scope boundaries for correlation. A nested @@ -151,16 +154,19 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { // join does not output cannot give its columns to a pulled up // filter either. LogicalPlan::Join(ref join) if !correlated_inputs_are_preserved(join) => { - // the unsupported case - self.can_pull_up = false; - Ok(Transformed::new(plan, false, TreeNodeRecursion::Jump)) + Ok(self.stop_pull_up(plan)) + } + // An ASOF join keeps each left row once, so a filter on the left + // side can move above it. A filter on the right side selects the + // rows the join can match, so it must stay below the join. + LogicalPlan::AsOfJoin(ref join) if holds_outer_reference(&join.right) => { + Ok(self.stop_pull_up(plan)) } LogicalPlan::Union(_) | LogicalPlan::Sort(_) | LogicalPlan::Extension(_) => { let plan_hold_outer = !plan.all_out_ref_exprs().is_empty(); if plan_hold_outer { // the unsupported case - self.can_pull_up = false; - Ok(Transformed::new(plan, false, TreeNodeRecursion::Jump)) + Ok(self.stop_pull_up(plan)) } else { Ok(Transformed::no(plan)) } @@ -170,18 +176,62 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { match (self.exists_sub_query, plan_hold_outer) { (false, true) => { // the unsupported case - self.can_pull_up = false; - Ok(Transformed::new(plan, false, TreeNodeRecursion::Jump)) + Ok(self.stop_pull_up(plan)) } _ => Ok(Transformed::no(plan)), } } - _ if plan.contains_outer_reference() => { - // the unsupported cases, the plan expressions contain out reference columns(like window expressions) - self.can_pull_up = false; - Ok(Transformed::new(plan, false, TreeNodeRecursion::Jump)) + // A correlated filter below these nodes can move above them. `f_up` + // adds the columns the pulled up filter needs to a Projection or an + // Aggregate, and checks that the filter can move above an + // Aggregate. A `DISTINCT` gives the same rows when a filter runs + // before it or after it, and its output takes the columns the node + // below it adds. The node itself must not hold an outer reference, + // because only a Filter gives its outer references to the join. + LogicalPlan::Projection(_) + | LogicalPlan::Aggregate(_) + | LogicalPlan::Distinct(Distinct::All(_)) + | LogicalPlan::Join(_) + | LogicalPlan::AsOfJoin(_) + | LogicalPlan::Repartition(_) + | LogicalPlan::SubqueryAlias(_) + | LogicalPlan::TableScan(_) + | LogicalPlan::EmptyRelation(_) + | LogicalPlan::Values(_) => { + if plan.contains_outer_reference() { + // the unsupported cases, the plan expressions contain out + // reference columns + Ok(self.stop_pull_up(plan)) + } else { + Ok(Transformed::no(plan)) + } + } + // A correlated filter below these nodes cannot move above them: + // - A Window function reads the rows of its partition, so the + // filter changes the rows each function sees. + // - `DISTINCT ON` keeps the first row of each group, so the filter + // changes which row is first. + // - The pull up does not carry the columns of the pulled up filter + // through an Unnest, and fails with a schema error. + // - A RecursiveQuery repeats its recursive term. + // - The other nodes are not queries and are not in a subquery. + LogicalPlan::Window(_) + | LogicalPlan::Distinct(Distinct::On(_)) + | LogicalPlan::Unnest(_) + | LogicalPlan::RecursiveQuery(_) + | LogicalPlan::Statement(_) + | LogicalPlan::Explain(_) + | LogicalPlan::Analyze(_) + | LogicalPlan::Dml(_) + | LogicalPlan::Ddl(_) + | LogicalPlan::Copy(_) + | LogicalPlan::DescribeTable(_) => { + if holds_outer_reference(&plan) { + Ok(self.stop_pull_up(plan)) + } else { + Ok(Transformed::no(plan)) + } } - _ => Ok(Transformed::no(plan)), } } @@ -457,6 +507,13 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { } impl PullUpCorrelatedExpr { + /// Keeps the correlated expressions below `plan` where they are. The caller + /// then keeps the subquery correlated. + fn stop_pull_up(&mut self, plan: LogicalPlan) -> Transformed { + self.can_pull_up = false; + Transformed::new(plan, false, TreeNodeRecursion::Jump) + } + /// Whether the pull up can add its columns to `group_expr` without changing /// what the aggregate returns. /// From 4c30c7aa262da480b6f664bd3201b11607ba19db Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Sun, 27 Sep 2026 00:46:20 -0500 Subject: [PATCH 2/8] test: cover correlated filters below and above a window function Co-Authored-By: Claude Opus 5.5 --- .../sqllogictest/test_files/lateral_join.slt | 57 ++++++++++++++ .../sqllogictest/test_files/subquery.slt | 76 +++++++++++++++++++ 2 files changed, 133 insertions(+) diff --git a/datafusion/sqllogictest/test_files/lateral_join.slt b/datafusion/sqllogictest/test_files/lateral_join.slt index 02d4d9f0ea1fd..1129e22b9f17e 100644 --- a/datafusion/sqllogictest/test_files/lateral_join.slt +++ b/datafusion/sqllogictest/test_files/lateral_join.slt @@ -753,6 +753,63 @@ FROM t1, LATERAL ( ) AS sub ORDER BY t1.id, sub.x; +# --- Correlated filter below a window function --- +# https://github.com/apache/datafusion/issues/25792 +# Pulling `t2.t1_id = t1.id` above the window would number the t2 rows of all +# t1 rows together instead of the t2 rows of each t1 row. +# Expected result once supported: (1, 10, 1), (1, 20, 2), (2, 30, 1). +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression OuterReferenceColumn +SELECT t1.id, sub.id, sub.rn +FROM t1, LATERAL ( + SELECT t2.id, row_number() OVER (ORDER BY t2.id) AS rn + FROM (SELECT * FROM t2 WHERE t2.t1_id = t1.id) AS t2 +) AS sub +ORDER BY t1.id, sub.id; + +# A correlated filter above the window is still decorrelated. The window +# numbers all t2 rows, then the filter keeps the rows of each t1 row. +query TT +EXPLAIN SELECT t1.id, sub.id, sub.rn +FROM t1, LATERAL ( + SELECT w.id, w.rn + FROM (SELECT t2.id, t2.t1_id, row_number() OVER (ORDER BY t2.id) AS rn FROM t2) AS w + WHERE w.t1_id = t1.id +) AS sub +ORDER BY t1.id, sub.id; +---- +logical_plan +01)Sort: t1.id ASC NULLS LAST, sub.id ASC NULLS LAST +02)--Projection: t1.id, sub.id, sub.rn +03)----Inner Join: t1.id = sub.t1_id +04)------TableScan: t1 projection=[id] +05)------SubqueryAlias: sub +06)--------Projection: w.id, w.rn, w.t1_id +07)----------SubqueryAlias: w +08)------------Projection: t2.id, t2.t1_id, row_number() ORDER BY [t2.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS rn +09)--------------WindowAggr: windowExpr=[[row_number() ORDER BY [t2.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] +10)----------------TableScan: t2 projection=[id, t1_id] +physical_plan +01)SortExec: expr=[id@0 ASC NULLS LAST, id@1 ASC NULLS LAST], preserve_partitioning=[false] +02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(t1_id@2, id@0)], projection=[id@3, id@0, rn@1] +03)----ProjectionExec: expr=[id@0 as id, row_number() ORDER BY [t2.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as rn, t1_id@1 as t1_id] +04)------BoundedWindowAggExec: wdw=[row_number() ORDER BY [t2.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() ORDER BY [t2.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] +05)--------SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false] +06)----------DataSourceExec: partitions=1, partition_sizes=[1] +07)----DataSourceExec: partitions=1, partition_sizes=[1] + +query III +SELECT t1.id, sub.id, sub.rn +FROM t1, LATERAL ( + SELECT w.id, w.rn + FROM (SELECT t2.id, t2.t1_id, row_number() OVER (ORDER BY t2.id) AS rn FROM t2) AS w + WHERE w.t1_id = t1.id +) AS sub +ORDER BY t1.id, sub.id; +---- +1 10 1 +1 20 2 +2 30 3 + # --- Implicit LATERAL (comma without LATERAL keyword) --- # DuckDB auto-detects lateral correlation; DataFusion requires # the explicit LATERAL keyword. diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index d27c244e0d4df..714ebb6de2593 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -3022,3 +3022,79 @@ DROP TABLE oj_a; statement ok DROP TABLE oj_b; + +# Regression test for #25792: a correlated filter below a window function must +# not be pulled above the window. The filter selects the rows the window +# function sees for each outer row, so once it is above the window the window +# function runs over all rows and computes different values. +statement ok +CREATE TABLE win_outer(k INT) AS VALUES (1), (2), (5); + +statement ok +CREATE TABLE win_inner(id INT, y INT) AS VALUES (1, 1), (2, 2); + +statement ok +set datafusion.explain.logical_plan_only = true; + +# The correlated filter is below the window, so the subquery stays correlated. +query TT +EXPLAIN SELECT win_outer.k FROM win_outer WHERE EXISTS (SELECT 1 FROM (SELECT b.id, row_number() OVER (ORDER BY b.id) AS rn FROM (SELECT * FROM win_inner WHERE win_inner.y = win_outer.k) AS b) AS w WHERE w.rn = 1); +---- +logical_plan +01)Filter: EXISTS () +02)--Subquery: +03)----Projection: Int64(1) +04)------SubqueryAlias: w +05)--------Projection: b.id, row_number() ORDER BY [b.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS rn +06)----------Filter: row_number() ORDER BY [b.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW = UInt64(1) +07)------------WindowAggr: windowExpr=[[row_number() ORDER BY [b.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] +08)--------------SubqueryAlias: b +09)----------------Projection: win_inner.id, win_inner.y +10)------------------Filter: win_inner.y = outer_ref(win_outer.k) +11)--------------------TableScan: win_inner +12)--TableScan: win_outer projection=[k] + +# It errors rather than returning wrong results. The right answer is `1, 2`. +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression Exists +SELECT win_outer.k FROM win_outer WHERE EXISTS (SELECT 1 FROM (SELECT b.id, row_number() OVER (ORDER BY b.id) AS rn FROM (SELECT * FROM win_inner WHERE win_inner.y = win_outer.k) AS b) AS w WHERE w.rn = 1) ORDER BY win_outer.k; + +# The same for NOT EXISTS. The right answer is `5`. +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression Exists +SELECT win_outer.k FROM win_outer WHERE NOT EXISTS (SELECT 1 FROM (SELECT b.id, row_number() OVER (ORDER BY b.id) AS rn FROM (SELECT * FROM win_inner WHERE win_inner.y = win_outer.k) AS b) AS w WHERE w.rn = 1) ORDER BY win_outer.k; + +# The same for IN. The right answer is `1, 2`. +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression InSubquery +SELECT win_outer.k FROM win_outer WHERE 1 IN (SELECT row_number() OVER (ORDER BY win_inner.id) FROM win_inner WHERE win_inner.y = win_outer.k) ORDER BY win_outer.k; + +# And for a scalar subquery. The right answer is `(1, 1), (2, 1), (5, NULL)`. +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression ScalarSubquery +SELECT win_outer.k, (SELECT max(w.rn) FROM (SELECT row_number() OVER (ORDER BY win_inner.id) AS rn FROM win_inner WHERE win_inner.y = win_outer.k) AS w) AS m FROM win_outer ORDER BY win_outer.k; + +# A correlated filter above the window is still pulled up, because the window +# input does not depend on the outer row. +query TT +EXPLAIN SELECT win_outer.k FROM win_outer WHERE EXISTS (SELECT 1 FROM (SELECT win_inner.id, row_number() OVER (ORDER BY win_inner.id) AS rn FROM win_inner) AS w WHERE w.id = win_outer.k AND w.rn = 2); +---- +logical_plan +01)LeftSemi Join: win_outer.k = __correlated_sq_1.id +02)--TableScan: win_outer projection=[k] +03)--SubqueryAlias: __correlated_sq_1 +04)----SubqueryAlias: w +05)------Projection: win_inner.id +06)--------Filter: row_number() ORDER BY [win_inner.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW = UInt64(2) +07)----------WindowAggr: windowExpr=[[row_number() ORDER BY [win_inner.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] +08)------------TableScan: win_inner projection=[id] + +query I +SELECT win_outer.k FROM win_outer WHERE EXISTS (SELECT 1 FROM (SELECT win_inner.id, row_number() OVER (ORDER BY win_inner.id) AS rn FROM win_inner) AS w WHERE w.id = win_outer.k AND w.rn = 2) ORDER BY win_outer.k; +---- +2 + +statement ok +RESET datafusion.explain.logical_plan_only; + +statement ok +DROP TABLE win_outer; + +statement ok +DROP TABLE win_inner; From abd41e1b5e29f767f2fbca26927995e5a9ea67b1 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Mon, 28 Sep 2026 10:48:49 -0500 Subject: [PATCH 3/8] test: cover correlated filters around an ASOF join A correlated filter on the right side of an ASOF join keeps the subquery correlated, for LATERAL and for EXISTS. A correlated filter on the left side, or above the join, is still pulled up. Co-Authored-By: Claude Opus 5.5 --- .../sqllogictest/test_files/asof_join.slt | 103 ++++++++++++++++++ 1 file changed, 103 insertions(+) diff --git a/datafusion/sqllogictest/test_files/asof_join.slt b/datafusion/sqllogictest/test_files/asof_join.slt index e745a95b0ec6e..2391f758326d1 100644 --- a/datafusion/sqllogictest/test_files/asof_join.slt +++ b/datafusion/sqllogictest/test_files/asof_join.slt @@ -495,3 +495,106 @@ FROM asof_left l ASOF JOIN asof_right r MATCH_CONDITION (l.ts) ON l.grp = r.grp; + +# Correlated subqueries around an ASOF join. See +# https://github.com/apache/datafusion/issues/25834. +statement ok +CREATE TABLE asof_corr_o(k INT) AS VALUES (1), (2); + +statement ok +CREATE TABLE asof_corr_l(id INT, ts INT) AS VALUES (1, 10); + +statement ok +CREATE TABLE asof_corr_r(ts INT, val INT) AS VALUES (5, 1), (8, 2); + +statement ok +set datafusion.explain.logical_plan_only = true; + +# A correlated filter on the right side of an ASOF join selects the rows the +# join can match. Above the join, it removes rows instead, so the subquery stays +# correlated. The derived table uses the table name as its alias: with a +# different alias, the lateral join rewrite fails with a schema error first. +query TT +EXPLAIN SELECT o.k, s.rts FROM asof_corr_o o, LATERAL (SELECT asof_corr_r.ts AS rts FROM asof_corr_l l ASOF JOIN (SELECT * FROM asof_corr_r WHERE asof_corr_r.val = o.k) AS asof_corr_r MATCH_CONDITION (l.ts >= asof_corr_r.ts)) AS s ORDER BY o.k; +---- +logical_plan +01)Sort: o.k ASC NULLS LAST +02)--Cross Join: +03)----SubqueryAlias: o +04)------TableScan: asof_corr_o projection=[k] +05)----SubqueryAlias: s +06)------Subquery: +07)--------Projection: asof_corr_r.ts AS rts +08)----------AsOf Join: match=[l.ts >= asof_corr_r.ts], constraint=On +09)------------SubqueryAlias: l +10)--------------TableScan: asof_corr_l projection=[ts] +11)------------SubqueryAlias: asof_corr_r +12)--------------Projection: asof_corr_r.ts +13)----------------Filter: asof_corr_r.val = outer_ref(o.k) +14)------------------TableScan: asof_corr_r projection=[ts, val] + +# It errors rather than returning wrong results. The right answer is +# `(1, 5), (2, 8)`. +query error DataFusion error: This feature is not implemented: Physical plan does not support logical expression OuterReferenceColumn +SELECT o.k, s.rts FROM asof_corr_o o, LATERAL (SELECT asof_corr_r.ts AS rts FROM asof_corr_l l ASOF JOIN (SELECT * FROM asof_corr_r WHERE asof_corr_r.val = o.k) AS asof_corr_r MATCH_CONDITION (l.ts >= asof_corr_r.ts)) AS s ORDER BY o.k; + +# The same for EXISTS. The right answer is `1`. +query error DataFusion error: This feature is not implemented: Physical plan does not support logical expression Exists +SELECT o.k FROM asof_corr_o o WHERE EXISTS (SELECT 1 FROM asof_corr_l l ASOF JOIN (SELECT * FROM asof_corr_r WHERE asof_corr_r.val = o.k) AS r MATCH_CONDITION (l.ts >= r.ts) WHERE r.ts = 5); + +# An ASOF join keeps each left row once, so a correlated filter on the left +# side is still pulled up. +query TT +EXPLAIN SELECT o.k FROM asof_corr_o o WHERE EXISTS (SELECT 1 FROM (SELECT * FROM asof_corr_l WHERE asof_corr_l.id = o.k) AS l ASOF JOIN asof_corr_r r MATCH_CONDITION (l.ts >= r.ts)); +---- +logical_plan +01)LeftSemi Join: o.k = __correlated_sq_1.id +02)--SubqueryAlias: o +03)----TableScan: asof_corr_o projection=[k] +04)--SubqueryAlias: __correlated_sq_1 +05)----Projection: l.id +06)------AsOf Join: match=[l.ts >= r.ts], constraint=On +07)--------SubqueryAlias: l +08)----------TableScan: asof_corr_l projection=[id, ts] +09)--------SubqueryAlias: r +10)----------TableScan: asof_corr_r projection=[ts] + +query I +SELECT o.k FROM asof_corr_o o WHERE EXISTS (SELECT 1 FROM (SELECT * FROM asof_corr_l WHERE asof_corr_l.id = o.k) AS l ASOF JOIN asof_corr_r r MATCH_CONDITION (l.ts >= r.ts)); +---- +1 + +# So is a correlated filter above the ASOF join. +query TT +EXPLAIN SELECT o.k, s.rts FROM asof_corr_o o, LATERAL (SELECT r.ts AS rts FROM asof_corr_l l ASOF JOIN asof_corr_r r MATCH_CONDITION (l.ts >= r.ts) WHERE r.val = o.k) AS s ORDER BY o.k; +---- +logical_plan +01)Sort: o.k ASC NULLS LAST +02)--Projection: o.k, s.rts +03)----Inner Join: o.k = s.val +04)------SubqueryAlias: o +05)--------TableScan: asof_corr_o projection=[k] +06)------SubqueryAlias: s +07)--------Projection: r.ts AS rts, r.val +08)----------AsOf Join: match=[l.ts >= r.ts], constraint=On +09)------------SubqueryAlias: l +10)--------------TableScan: asof_corr_l projection=[ts] +11)------------SubqueryAlias: r +12)--------------TableScan: asof_corr_r projection=[ts, val] + +query II +SELECT o.k, s.rts FROM asof_corr_o o, LATERAL (SELECT r.ts AS rts FROM asof_corr_l l ASOF JOIN asof_corr_r r MATCH_CONDITION (l.ts >= r.ts) WHERE r.val = o.k) AS s ORDER BY o.k; +---- +2 8 + +statement ok +RESET datafusion.explain.logical_plan_only; + +statement ok +DROP TABLE asof_corr_o; + +statement ok +DROP TABLE asof_corr_l; + +statement ok +DROP TABLE asof_corr_r; From 8dceb7c7965a65cb64848f1048ed0639f4de944f Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Mon, 28 Sep 2026 10:48:49 -0500 Subject: [PATCH 4/8] test: cover a correlated filter below an Unnest The subquery stays correlated instead of failing the optimizer with a schema error. Co-Authored-By: Claude Opus 5.5 --- .../sqllogictest/test_files/subquery.slt | 44 +++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index 714ebb6de2593..fa2f59421ee67 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -3098,3 +3098,47 @@ DROP TABLE win_outer; statement ok DROP TABLE win_inner; + +# A correlated filter below an Unnest is not pulled above the Unnest. The pull +# up does not carry the columns of the filter through the Unnest, so it failed +# with a schema error. +statement ok +CREATE TABLE unnest_outer(k INT) AS VALUES (1), (2); + +statement ok +CREATE TABLE unnest_inner(id INT, arr INT[]) AS VALUES (1, [3, 4]), (2, [5]); + +statement ok +set datafusion.explain.logical_plan_only = true; + +query TT +EXPLAIN SELECT unnest_outer.k FROM unnest_outer WHERE EXISTS (SELECT 1 FROM (SELECT unnest(t.arr) AS u FROM (SELECT * FROM unnest_inner WHERE unnest_inner.id = unnest_outer.k) AS t) AS x WHERE x.u = 3); +---- +logical_plan +01)Filter: EXISTS () +02)--Subquery: +03)----Projection: Int64(1) +04)------SubqueryAlias: x +05)--------Projection: __unnest_placeholder(t.arr,depth=1) AS UNNEST(t.arr) AS u +06)----------Filter: __unnest_placeholder(t.arr,depth=1) = Int32(3) +07)------------Unnest: lists[__unnest_placeholder(t.arr)|depth=1] structs[] +08)--------------Projection: t.arr AS __unnest_placeholder(t.arr) +09)----------------SubqueryAlias: t +10)------------------Projection: unnest_inner.id, unnest_inner.arr +11)--------------------Filter: unnest_inner.id = outer_ref(unnest_outer.k) +12)----------------------TableScan: unnest_inner +13)--TableScan: unnest_outer projection=[k] + +# The subquery stays correlated, which the physical planner does not support. +# The right answer is `1`. +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression Exists +SELECT unnest_outer.k FROM unnest_outer WHERE EXISTS (SELECT 1 FROM (SELECT unnest(t.arr) AS u FROM (SELECT * FROM unnest_inner WHERE unnest_inner.id = unnest_outer.k) AS t) AS x WHERE x.u = 3); + +statement ok +RESET datafusion.explain.logical_plan_only; + +statement ok +DROP TABLE unnest_outer; + +statement ok +DROP TABLE unnest_inner; From f31fc8cb9f76529b08ada604b90f32bd8f88bb12 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 30 Sep 2026 02:42:55 +0000 Subject: [PATCH 5/8] fix: pull correlated filters through Window and Unnest when safe Address review feedback: - Window: pull a correlated filter above the window when it reads only plain PARTITION BY columns of every window function, so it keeps or drops whole partitions. - Unnest: recompute the Unnest schema so it carries the columns the pull up adds below it. Stop when the filter reads an unnested column. - Stop at a Window or Unnest over an aggregate with count bug handling. - Use holds_outer_reference for Union, Sort, Extension and Limit. - Note that DISTINCT shares the Aggregate exemption and link #25808. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01NZEssuMQANC3qLy1k2zUus --- datafusion/optimizer/src/decorrelate.rs | 148 +++++++++++++++--- .../src/decorrelate_predicate_subquery.rs | 38 +++++ .../sqllogictest/test_files/lateral_join.slt | 51 ++++++ .../sqllogictest/test_files/subquery.slt | 66 +++++--- 4 files changed, 264 insertions(+), 39 deletions(-) diff --git a/datafusion/optimizer/src/decorrelate.rs b/datafusion/optimizer/src/decorrelate.rs index d50072d0b9872..f98bd7dd9d7c5 100644 --- a/datafusion/optimizer/src/decorrelate.rs +++ b/datafusion/optimizer/src/decorrelate.rs @@ -178,18 +178,34 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { LogicalPlan::AsOfJoin(ref join) if holds_outer_reference(&join.right) => { Ok(self.stop_pull_up(plan)) } + // An Unnest replaces the column it unnests with the values of that + // column, so a filter on it means something else above the + // Unnest. The output column loses its qualifier, so compare names. + LogicalPlan::Unnest(ref unnest) + if correlated_filter_columns(&unnest.input).iter().any(|col| { + unnest + .exec_columns + .iter() + .any(|exec_col| exec_col.name == col.name) + }) => + { + Ok(self.stop_pull_up(plan)) + } + // A correlated filter cannot move out of one input of a Union, and + // `f_up` does not add its columns to a Sort or an Extension, so no + // correlated filter may be below these nodes. LogicalPlan::Union(_) | LogicalPlan::Sort(_) | LogicalPlan::Extension(_) => { - let plan_hold_outer = !plan.all_out_ref_exprs().is_empty(); - if plan_hold_outer { - // the unsupported case + if holds_outer_reference(&plan) { Ok(self.stop_pull_up(plan)) } else { Ok(Transformed::no(plan)) } } + // `f_up` drops the Limit of an EXISTS subquery, which does not + // change whether the subquery returns a row. Any other Limit keeps + // the correlated filters below it. LogicalPlan::Limit(_) => { - let plan_hold_outer = !plan.all_out_ref_exprs().is_empty(); - match (self.exists_sub_query, plan_hold_outer) { + match (self.exists_sub_query, holds_outer_reference(&plan)) { (false, true) => { // the unsupported case Ok(self.stop_pull_up(plan)) @@ -197,16 +213,29 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { _ => Ok(Transformed::no(plan)), } } - // A correlated filter below these nodes can move above them. `f_up` - // adds the columns the pulled up filter needs to a Projection or an - // Aggregate, and checks that the filter can move above an - // Aggregate. A `DISTINCT` gives the same rows when a filter runs - // before it or after it, and its output takes the columns the node - // below it adds. The node itself must not hold an outer reference, - // because only a Filter gives its outer references to the join. + // A correlated filter below these nodes can move above them. The + // node itself must not hold an outer reference, because only a + // Filter gives its outer references to the join. + // - `f_up` adds the columns the pulled up filter needs to a + // Projection or an Aggregate, and checks that the filter can move + // above an Aggregate. + // - A `DISTINCT` shares the exemption of an Aggregate without + // aggregate functions: its output takes the columns the node + // below it adds, so they become part of its keys. This is only + // right when the pulled up filter fixes those columns for each + // outer row. A non-equality filter can return duplicates, see + // https://github.com/apache/datafusion/issues/25808. + // - A Window keeps or drops whole partitions when the filter reads + // only its `PARTITION BY` columns. `f_up` checks this. + // - An Unnest maps each input row to its own output rows, so a + // filter on a column it passes through keeps the same rows. + // The arm above checks that the filter does not read an unnested + // column. LogicalPlan::Projection(_) | LogicalPlan::Aggregate(_) | LogicalPlan::Distinct(Distinct::All(_)) + | LogicalPlan::Window(_) + | LogicalPlan::Unnest(_) | LogicalPlan::Join(_) | LogicalPlan::AsOfJoin(_) | LogicalPlan::Repartition(_) @@ -223,17 +252,11 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { } } // A correlated filter below these nodes cannot move above them: - // - A Window function reads the rows of its partition, so the - // filter changes the rows each function sees. // - `DISTINCT ON` keeps the first row of each group, so the filter // changes which row is first. - // - The pull up does not carry the columns of the pulled up filter - // through an Unnest, and fails with a schema error. // - A RecursiveQuery repeats its recursive term. // - The other nodes are not queries and are not in a subquery. - LogicalPlan::Window(_) - | LogicalPlan::Distinct(Distinct::On(_)) - | LogicalPlan::Unnest(_) + LogicalPlan::Distinct(Distinct::On(_)) | LogicalPlan::RecursiveQuery(_) | LogicalPlan::Statement(_) | LogicalPlan::Explain(_) @@ -528,6 +551,30 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { } Ok(new_plan) } + LogicalPlan::Window(window) => { + let mut local_correlated_cols = BTreeSet::new(); + collect_local_correlated_cols( + &plan, + &self.correlated_subquery_cols_map, + &mut local_correlated_cols, + ); + // A filter that reads only `PARTITION BY` columns keeps or drops + // whole partitions, so each window function sees the same rows + // above it or below it. A filter on any other column changes the + // rows of a partition. + if !local_correlated_cols.is_empty() + && !window + .window_expr + .iter() + .all(|expr| partitions_by_all(expr, &local_correlated_cols)) + { + self.can_pull_up = false; + } + self.pass_through_pulled_up_cols(plan) + } + // `f_down` checked that the pulled up filters do not read an + // unnested column. + LogicalPlan::Unnest(_) => self.pass_through_pulled_up_cols(plan), _ => Ok(Transformed::no(plan)), } } @@ -541,6 +588,32 @@ impl PullUpCorrelatedExpr { Transformed::new(plan, false, TreeNodeRecursion::Jump) } + /// Rebuilds a node whose output holds every column of its input, so that + /// it outputs the columns the pull up added below it. + /// + /// The count bug handling of a Projection or a Filter above reads the + /// values an aggregate below gives on an empty input. Those values do not + /// hold once a Window or an Unnest runs over the aggregate, so the pull up + /// stops in that case. + fn pass_through_pulled_up_cols( + &mut self, + plan: LogicalPlan, + ) -> Result> { + if plan + .inputs() + .iter() + .any(|input| self.collected_count_expr_map.contains_key(*input)) + { + self.can_pull_up = false; + } + let new_plan = plan.clone().recompute_schema()?; + if new_plan.schema() == plan.schema() { + Ok(Transformed::no(plan)) + } else { + Ok(Transformed::yes(new_plan)) + } + } + /// Whether the pull up can add its columns to `group_expr` without changing /// what the aggregate returns. /// @@ -683,6 +756,43 @@ fn holds_outer_reference(plan: &LogicalPlan) -> bool { found } +/// The columns that the correlated conjuncts of the Filters in `plan` read, +/// in the scope being decorrelated. Like [`holds_outer_reference`], this does +/// not descend into a [`LogicalPlan::Subquery`]. +fn correlated_filter_columns(plan: &LogicalPlan) -> Vec { + let mut cols = vec![]; + plan.apply(|node| { + Ok(match node { + LogicalPlan::Subquery(_) => TreeNodeRecursion::Jump, + LogicalPlan::Filter(filter) => { + for expr in split_conjunction(&filter.predicate) { + if expr.contains_outer() { + cols.extend(expr.column_refs().into_iter().cloned()); + } + } + TreeNodeRecursion::Continue + } + _ => TreeNodeRecursion::Continue, + }) + }) + .expect("apply closure is infallible"); + cols +} + +/// Whether the window function `expr` lists each of `cols` as a plain +/// `PARTITION BY` column. +fn partitions_by_all(expr: &Expr, cols: &BTreeSet) -> bool { + let Expr::WindowFunction(window_fun) = expr.clone().unalias_nested().data else { + return false; + }; + let partition_by = &window_fun.params.partition_by; + cols.iter().all(|col| { + partition_by + .iter() + .any(|expr| matches!(expr, Expr::Column(c) if c == col)) + }) +} + fn can_pullup_over_aggregation(expr: &Expr) -> bool { if let Expr::BinaryExpr(BinaryExpr { left, diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index e2f4aa40b519e..884f0e2164e1c 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -3077,4 +3077,42 @@ mod tests { " ) } + + /// An Unnest replaces the column it unnests with the values of that + /// column. A correlated filter on that column below the Unnest compares + /// the whole list, so it cannot move above the Unnest. + #[test] + fn exists_subquery_filter_on_unnested_column() -> Result<()> { + let list_type = + DataType::List(Arc::new(Field::new_list_field(DataType::Int32, true))); + let schema = Schema::new(vec![ + Field::new("id", DataType::Int32, true), + Field::new("arr", list_type.clone(), true), + ]); + let subquery = Arc::new( + table_scan(Some("sq"), &schema, None)? + .filter(col("sq.arr").eq(out_ref_col(list_type, "outer_t.arr")))? + .unnest_column("arr")? + .project(vec![col("sq.id")])? + .build()?, + ); + let plan = table_scan(Some("outer_t"), &schema, None)? + .filter(exists(subquery))? + .project(vec![col("outer_t.id")])? + .build()?; + + assert_optimized_plan_equal!( + plan, + @r" + Projection: outer_t.id [id:Int32;N] + Filter: EXISTS () [id:Int32;N, arr:List(Int32);N] + Subquery: [id:Int32;N] + Projection: sq.id [id:Int32;N] + Unnest: lists[sq.arr|depth=1] structs[] [id:Int32;N, arr:Int32;N] + Filter: sq.arr = outer_ref(outer_t.arr) [id:Int32;N, arr:List(Int32);N] + TableScan: sq [id:Int32;N, arr:List(Int32);N] + TableScan: outer_t [id:Int32;N, arr:List(Int32);N] + " + ) + } } diff --git a/datafusion/sqllogictest/test_files/lateral_join.slt b/datafusion/sqllogictest/test_files/lateral_join.slt index 1129e22b9f17e..0e0ceb9cfa4fa 100644 --- a/datafusion/sqllogictest/test_files/lateral_join.slt +++ b/datafusion/sqllogictest/test_files/lateral_join.slt @@ -810,6 +810,57 @@ ORDER BY t1.id, sub.id; 1 20 2 2 30 3 +# A correlated filter below the window that reads only a `PARTITION BY` column +# keeps or drops whole partitions, so it is pulled above the window. +query TT +EXPLAIN SELECT t1.id, sub.id, sub.rn +FROM t1, LATERAL ( + SELECT t2.id, row_number() OVER (PARTITION BY t2.t1_id ORDER BY t2.id) AS rn + FROM t2 WHERE t2.t1_id = t1.id +) AS sub +ORDER BY t1.id, sub.id; +---- +logical_plan +01)Sort: t1.id ASC NULLS LAST, sub.id ASC NULLS LAST +02)--Projection: t1.id, sub.id, sub.rn +03)----Inner Join: t1.id = sub.t1_id +04)------TableScan: t1 projection=[id] +05)------SubqueryAlias: sub +06)--------Projection: t2.id, row_number() PARTITION BY [t2.t1_id] ORDER BY [t2.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS rn, t2.t1_id +07)----------WindowAggr: windowExpr=[[row_number() PARTITION BY [t2.t1_id] ORDER BY [t2.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] +08)------------TableScan: t2 projection=[id, t1_id] +physical_plan +01)SortExec: expr=[id@0 ASC NULLS LAST, id@1 ASC NULLS LAST], preserve_partitioning=[false] +02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(t1_id@2, id@0)], projection=[id@3, id@0, rn@1] +03)----ProjectionExec: expr=[id@0 as id, row_number() PARTITION BY [t2.t1_id] ORDER BY [t2.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as rn, t1_id@1 as t1_id] +04)------BoundedWindowAggExec: wdw=[row_number() PARTITION BY [t2.t1_id] ORDER BY [t2.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [t2.t1_id] ORDER BY [t2.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] +05)--------SortExec: expr=[t1_id@1 ASC NULLS LAST, id@0 ASC NULLS LAST], preserve_partitioning=[false] +06)----------DataSourceExec: partitions=1, partition_sizes=[1] +07)----DataSourceExec: partitions=1, partition_sizes=[1] + +query III +SELECT t1.id, sub.id, sub.rn +FROM t1, LATERAL ( + SELECT t2.id, row_number() OVER (PARTITION BY t2.t1_id ORDER BY t2.id) AS rn + FROM t2 WHERE t2.t1_id = t1.id +) AS sub +ORDER BY t1.id, sub.id; +---- +1 10 1 +1 20 2 +2 30 1 + +# The filter must read a `PARTITION BY` column of every window function. +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression OuterReferenceColumn +SELECT t1.id, sub.id, sub.rn, sub.c +FROM t1, LATERAL ( + SELECT t2.id, + row_number() OVER (PARTITION BY t2.t1_id ORDER BY t2.id) AS rn, + count(*) OVER (PARTITION BY t2.data) AS c + FROM t2 WHERE t2.t1_id = t1.id +) AS sub +ORDER BY t1.id, sub.id; + # --- Implicit LATERAL (comma without LATERAL keyword) --- # DuckDB auto-detects lateral correlation; DataFusion requires # the explicit LATERAL keyword. diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index 0d4af0f6d10c8..d2a90b0516489 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -3108,9 +3108,10 @@ DROP TABLE win_outer; statement ok DROP TABLE win_inner; -# A correlated filter below an Unnest is not pulled above the Unnest. The pull -# up does not carry the columns of the filter through the Unnest, so it failed -# with a schema error. +# A correlated filter below an Unnest is pulled above the Unnest. An Unnest maps +# each input row to its own output rows, so a filter on a column it passes +# through keeps the same rows. The Unnest must output the columns the pull up +# adds below it, which failed with a schema error before. statement ok CREATE TABLE unnest_outer(k INT) AS VALUES (1), (2); @@ -3124,24 +3125,49 @@ query TT EXPLAIN SELECT unnest_outer.k FROM unnest_outer WHERE EXISTS (SELECT 1 FROM (SELECT unnest(t.arr) AS u FROM (SELECT * FROM unnest_inner WHERE unnest_inner.id = unnest_outer.k) AS t) AS x WHERE x.u = 3); ---- logical_plan -01)Filter: EXISTS () -02)--Subquery: -03)----Projection: Int64(1) -04)------SubqueryAlias: x -05)--------Projection: __unnest_placeholder(t.arr,depth=1) AS UNNEST(t.arr) AS u -06)----------Filter: __unnest_placeholder(t.arr,depth=1) = Int32(3) -07)------------Unnest: lists[__unnest_placeholder(t.arr)|depth=1] structs[] -08)--------------Projection: t.arr AS __unnest_placeholder(t.arr) -09)----------------SubqueryAlias: t -10)------------------Projection: unnest_inner.id, unnest_inner.arr -11)--------------------Filter: unnest_inner.id = outer_ref(unnest_outer.k) -12)----------------------TableScan: unnest_inner -13)--TableScan: unnest_outer projection=[k] - -# The subquery stays correlated, which the physical planner does not support. -# The right answer is `1`. -statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression Exists +01)LeftSemi Join: unnest_outer.k = __correlated_sq_1.id +02)--TableScan: unnest_outer projection=[k] +03)--SubqueryAlias: __correlated_sq_1 +04)----SubqueryAlias: x +05)------Projection: t.id +06)--------Filter: __unnest_placeholder(t.arr,depth=1) = Int32(3) +07)----------Unnest: lists[__unnest_placeholder(t.arr)|depth=1] structs[] +08)------------Projection: t.arr AS __unnest_placeholder(t.arr), t.id +09)--------------SubqueryAlias: t +10)----------------TableScan: unnest_inner projection=[id, arr] + +query I SELECT unnest_outer.k FROM unnest_outer WHERE EXISTS (SELECT 1 FROM (SELECT unnest(t.arr) AS u FROM (SELECT * FROM unnest_inner WHERE unnest_inner.id = unnest_outer.k) AS t) AS x WHERE x.u = 3); +---- +1 + +query I +SELECT unnest_outer.k FROM unnest_outer WHERE NOT EXISTS (SELECT 1 FROM (SELECT unnest(t.arr) AS u FROM (SELECT * FROM unnest_inner WHERE unnest_inner.id = unnest_outer.k) AS t) AS x WHERE x.u = 3); +---- +2 + +# The correlated column is selected next to the unnested one. +query TT +EXPLAIN SELECT unnest_outer.k, s.u FROM unnest_outer, LATERAL (SELECT i.id, unnest(i.arr) AS u FROM unnest_inner AS i WHERE i.id = unnest_outer.k) AS s ORDER BY unnest_outer.k, s.u; +---- +logical_plan +01)Sort: unnest_outer.k ASC NULLS LAST, s.u ASC NULLS LAST +02)--Projection: unnest_outer.k, s.u +03)----Inner Join: unnest_outer.k = s.id +04)------TableScan: unnest_outer projection=[k] +05)------SubqueryAlias: s +06)--------Projection: i.id, __unnest_placeholder(i.arr,depth=1) AS UNNEST(i.arr) AS u +07)----------Unnest: lists[__unnest_placeholder(i.arr)|depth=1] structs[] +08)------------Projection: i.id, i.arr AS __unnest_placeholder(i.arr) +09)--------------SubqueryAlias: i +10)----------------TableScan: unnest_inner projection=[id, arr] + +query II +SELECT unnest_outer.k, s.u FROM unnest_outer, LATERAL (SELECT i.id, unnest(i.arr) AS u FROM unnest_inner AS i WHERE i.id = unnest_outer.k) AS s ORDER BY unnest_outer.k, s.u; +---- +1 3 +1 4 +2 5 statement ok RESET datafusion.explain.logical_plan_only; From decd3ecfa64469f2f3d7b3e821fdbef76bb5e08a Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 30 Sep 2026 02:48:56 +0000 Subject: [PATCH 6/8] test: cover more correlated filters through Window and Unnest Add the cases from review: a non-equality filter on a partition column, IN and scalar subqueries through a Window, a PARTITION BY expression, a Window over a count bug aggregate, LEFT JOIN LATERAL, and a scalar subquery over an Unnest. Clarify that the Unnest name check only applies to plans that unnest a column in place, and avoid cloning in partitions_by_all and pass_through_pulled_up_cols. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01NZEssuMQANC3qLy1k2zUus --- datafusion/optimizer/src/decorrelate.rs | 16 ++++--- .../sqllogictest/test_files/subquery.slt | 46 +++++++++++++++++++ 2 files changed, 55 insertions(+), 7 deletions(-) diff --git a/datafusion/optimizer/src/decorrelate.rs b/datafusion/optimizer/src/decorrelate.rs index f98bd7dd9d7c5..3b11237fa34ab 100644 --- a/datafusion/optimizer/src/decorrelate.rs +++ b/datafusion/optimizer/src/decorrelate.rs @@ -181,6 +181,9 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { // An Unnest replaces the column it unnests with the values of that // column, so a filter on it means something else above the // Unnest. The output column loses its qualifier, so compare names. + // The SQL planner unnests a new `__unnest_placeholder` column and + // passes the original column through, so this only stops plans + // built with `unnest_column` and similar. LogicalPlan::Unnest(ref unnest) if correlated_filter_columns(&unnest.input).iter().any(|col| { unnest @@ -606,12 +609,7 @@ impl PullUpCorrelatedExpr { { self.can_pull_up = false; } - let new_plan = plan.clone().recompute_schema()?; - if new_plan.schema() == plan.schema() { - Ok(Transformed::no(plan)) - } else { - Ok(Transformed::yes(new_plan)) - } + Ok(Transformed::yes(plan.recompute_schema()?)) } /// Whether the pull up can add its columns to `group_expr` without changing @@ -782,7 +780,11 @@ fn correlated_filter_columns(plan: &LogicalPlan) -> Vec { /// Whether the window function `expr` lists each of `cols` as a plain /// `PARTITION BY` column. fn partitions_by_all(expr: &Expr, cols: &BTreeSet) -> bool { - let Expr::WindowFunction(window_fun) = expr.clone().unalias_nested().data else { + let mut expr = expr; + while let Expr::Alias(Alias { expr: inner, .. }) = expr { + expr = inner; + } + let Expr::WindowFunction(window_fun) = expr else { return false; }; let partition_by = &window_fun.params.partition_by; diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index d2a90b0516489..40326bebe9b06 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -3099,6 +3099,45 @@ SELECT win_outer.k FROM win_outer WHERE EXISTS (SELECT 1 FROM (SELECT win_inner. ---- 2 +# A correlated filter below the window that reads only a `PARTITION BY` column +# keeps or drops whole partitions, so it is pulled above the window. This holds +# for a non-equality filter too. The right answer is `1`. +query I +SELECT win_outer.k FROM win_outer WHERE EXISTS (SELECT 1 FROM (SELECT row_number() OVER (PARTITION BY win_inner.y ORDER BY win_inner.id) AS rn FROM win_inner WHERE win_inner.y > win_outer.k) AS w WHERE w.rn = 1) ORDER BY win_outer.k; +---- +1 + +# IN. The right answer is `1, 2`. +query I +SELECT win_outer.k FROM win_outer WHERE 1 IN (SELECT row_number() OVER (PARTITION BY win_inner.y ORDER BY win_inner.id) FROM win_inner WHERE win_inner.y = win_outer.k) ORDER BY win_outer.k; +---- +1 +2 + +# A scalar subquery. The right answer is `(1, 1), (2, 1), (5, NULL)`. +query II +SELECT win_outer.k, (SELECT max(w.rn) FROM (SELECT row_number() OVER (PARTITION BY win_inner.y ORDER BY win_inner.id) AS rn FROM win_inner WHERE win_inner.y = win_outer.k) AS w) AS m FROM win_outer ORDER BY win_outer.k; +---- +1 1 +2 1 +5 NULL + +# A window that partitions by an expression of the column does not count. +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression ScalarSubquery +SELECT win_outer.k, (SELECT max(w.rn) FROM (SELECT row_number() OVER (PARTITION BY win_inner.y + 0 ORDER BY win_inner.id) AS rn FROM win_inner WHERE win_inner.y = win_outer.k) AS w) AS m FROM win_outer ORDER BY win_outer.k; + +# A window over an aggregate with count bug handling stays correlated. +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression ScalarSubquery +SELECT win_outer.k, (SELECT count(*) OVER () FROM (SELECT count(*) AS c FROM win_inner WHERE win_inner.y = win_outer.k) AS a) AS m FROM win_outer ORDER BY win_outer.k; + +# LEFT JOIN LATERAL keeps the outer rows without a match. +query III +SELECT win_outer.k, s.id, s.rn FROM win_outer LEFT JOIN LATERAL (SELECT win_inner.id, row_number() OVER (PARTITION BY win_inner.y ORDER BY win_inner.id) AS rn FROM win_inner WHERE win_inner.y = win_outer.k) AS s ON true ORDER BY win_outer.k; +---- +1 1 1 +2 2 1 +5 NULL NULL + statement ok RESET datafusion.explain.logical_plan_only; @@ -3169,6 +3208,13 @@ SELECT unnest_outer.k, s.u FROM unnest_outer, LATERAL (SELECT i.id, unnest(i.arr 1 4 2 5 +# A scalar subquery over an Unnest. The right answer is `(1, 2), (2, 1)`. +query II +SELECT unnest_outer.k, (SELECT count(x.u) FROM (SELECT unnest(t.arr) AS u FROM unnest_inner AS t WHERE t.id = unnest_outer.k) AS x) AS c FROM unnest_outer ORDER BY unnest_outer.k; +---- +1 2 +2 1 + statement ok RESET datafusion.explain.logical_plan_only; From 1a2ad03ab0c444c1853162652f423d5cc7f808ed Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 30 Sep 2026 02:54:27 +0000 Subject: [PATCH 7/8] feat: decorrelate a window below a correlated equality filter When a correlated filter below a Window fixes the value of each column it reads for each outer row (`inner.y = outer.k`), add those columns to the PARTITION BY of every window function and pull the filter above the window. This is what Spark and DuckDB do, and it decorrelates #25792. A Projection on top keeps the old window column names. Other filters on non-partition columns still keep the subquery correlated. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01NZEssuMQANC3qLy1k2zUus --- datafusion/optimizer/src/decorrelate.rs | 73 +++++++++++++++++-- .../sqllogictest/test_files/lateral_join.slt | 31 ++++++-- .../sqllogictest/test_files/subquery.slt | 70 +++++++++++------- 3 files changed, 135 insertions(+), 39 deletions(-) diff --git a/datafusion/optimizer/src/decorrelate.rs b/datafusion/optimizer/src/decorrelate.rs index 3b11237fa34ab..6b03fc5122063 100644 --- a/datafusion/optimizer/src/decorrelate.rs +++ b/datafusion/optimizer/src/decorrelate.rs @@ -37,7 +37,7 @@ use datafusion_expr::utils::{ }; use datafusion_expr::{ BinaryExpr, Cast, Distinct, EmptyRelation, Expr, ExprSchemable, FetchType, - LogicalPlan, LogicalPlanBuilder, Operator, expr, lit, + LogicalPlan, LogicalPlanBuilder, Operator, Window, expr, lit, }; /// This struct rewrite the sub query plan by pull up the correlated @@ -563,17 +563,30 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { ); // A filter that reads only `PARTITION BY` columns keeps or drops // whole partitions, so each window function sees the same rows - // above it or below it. A filter on any other column changes the - // rows of a partition. - if !local_correlated_cols.is_empty() - && !window + // above it or below it. + if local_correlated_cols.is_empty() + || window .window_expr .iter() .all(|expr| partitions_by_all(expr, &local_correlated_cols)) + { + return self.pass_through_pulled_up_cols(plan); + } + // A filter that fixes the value of each column it reads for each + // outer row, like `inner.y = outer.k`, keeps the rows of one + // value of those columns. Partitioning by those columns as well + // gives each window function the same rows once the filter is + // above it. This is what the pull up does for an Aggregate too. + if !self.can_pull_over_aggregation + || self.collected_count_expr_map.contains_key(&*window.input) { self.can_pull_up = false; + return self.pass_through_pulled_up_cols(plan); } - self.pass_through_pulled_up_cols(plan) + Ok(Transformed::yes(partition_window_by( + window, + &local_correlated_cols, + )?)) } // `f_down` checked that the pulled up filters do not read an // unnested column. @@ -777,6 +790,54 @@ fn correlated_filter_columns(plan: &LogicalPlan) -> Vec { cols } +/// Adds `cols` to the `PARTITION BY` of every window function of `window`. +/// That changes the names of the window function columns, so a Projection on +/// top gives them their old names back. +fn partition_window_by(window: &Window, cols: &BTreeSet) -> Result { + let new_window_expr = window + .window_expr + .iter() + .map(|expr| add_partition_by(expr, cols)) + .collect::>(); + let mut proj_exprs = window + .input + .schema() + .columns() + .into_iter() + .map(Expr::Column) + .collect::>(); + for (old, new) in window.window_expr.iter().zip(&new_window_expr) { + let new_col = Expr::Column(Column::from_name(new.schema_name().to_string())); + proj_exprs.push(new_col.alias(old.schema_name().to_string())); + } + LogicalPlanBuilder::from(LogicalPlan::Window(Window::try_new( + new_window_expr, + Arc::clone(&window.input), + )?)) + .project(proj_exprs)? + .build() +} + +fn add_partition_by(expr: &Expr, cols: &BTreeSet) -> Expr { + match expr { + Expr::Alias(alias) => Expr::Alias(Alias { + expr: Box::new(add_partition_by(&alias.expr, cols)), + ..alias.clone() + }), + Expr::WindowFunction(window_fun) => { + let mut window_fun = window_fun.clone(); + for col in cols { + let col = Expr::Column(col.clone()); + if !window_fun.params.partition_by.contains(&col) { + window_fun.params.partition_by.push(col); + } + } + Expr::WindowFunction(window_fun) + } + _ => expr.clone(), + } +} + /// Whether the window function `expr` lists each of `cols` as a plain /// `PARTITION BY` column. fn partitions_by_all(expr: &Expr, cols: &BTreeSet) -> bool { diff --git a/datafusion/sqllogictest/test_files/lateral_join.slt b/datafusion/sqllogictest/test_files/lateral_join.slt index 0e0ceb9cfa4fa..b531df54f0ff1 100644 --- a/datafusion/sqllogictest/test_files/lateral_join.slt +++ b/datafusion/sqllogictest/test_files/lateral_join.slt @@ -755,16 +755,20 @@ ORDER BY t1.id, sub.x; # --- Correlated filter below a window function --- # https://github.com/apache/datafusion/issues/25792 -# Pulling `t2.t1_id = t1.id` above the window would number the t2 rows of all -# t1 rows together instead of the t2 rows of each t1 row. -# Expected result once supported: (1, 10, 1), (1, 20, 2), (2, 30, 1). -statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression OuterReferenceColumn +# Pulling `t2.t1_id = t1.id` above the window as it is would number the t2 rows +# of all t1 rows together. The filter fixes `t2.t1_id` for each t1 row, so the +# pull up adds `t2.t1_id` to the `PARTITION BY` of the window instead. +query III SELECT t1.id, sub.id, sub.rn FROM t1, LATERAL ( SELECT t2.id, row_number() OVER (ORDER BY t2.id) AS rn FROM (SELECT * FROM t2 WHERE t2.t1_id = t1.id) AS t2 ) AS sub ORDER BY t1.id, sub.id; +---- +1 10 1 +1 20 2 +2 30 1 # A correlated filter above the window is still decorrelated. The window # numbers all t2 rows, then the filter keeps the rows of each t1 row. @@ -850,8 +854,8 @@ ORDER BY t1.id, sub.id; 1 20 2 2 30 1 -# The filter must read a `PARTITION BY` column of every window function. -statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression OuterReferenceColumn +# Every window function is partitioned by the columns of the filter. +query IIII SELECT t1.id, sub.id, sub.rn, sub.c FROM t1, LATERAL ( SELECT t2.id, @@ -860,6 +864,21 @@ FROM t1, LATERAL ( FROM t2 WHERE t2.t1_id = t1.id ) AS sub ORDER BY t1.id, sub.id; +---- +1 10 1 1 +1 20 2 1 +2 30 1 1 + +# A non-equality filter does not fix the value of its column for each t1 row, +# so partitioning by the column would split the rows of one t1 row. The +# subquery stays correlated. +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression OuterReferenceColumn +SELECT t1.id, sub.id, sub.rn +FROM t1, LATERAL ( + SELECT t2.id, row_number() OVER (ORDER BY t2.id) AS rn + FROM t2 WHERE t2.t1_id >= t1.id +) AS sub +ORDER BY t1.id, sub.id; # --- Implicit LATERAL (comma without LATERAL keyword) --- # DuckDB auto-detects lateral correlation; DataFusion requires diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index 40326bebe9b06..cefa32b24891b 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -3032,10 +3032,11 @@ DROP TABLE oj_a; statement ok DROP TABLE oj_b; -# Regression test for #25792: a correlated filter below a window function must -# not be pulled above the window. The filter selects the rows the window -# function sees for each outer row, so once it is above the window the window -# function runs over all rows and computes different values. +# Regression test for #25792: a correlated filter below a window function +# selects the rows the window function sees for each outer row. Pulled above +# the window as it is, the window function would run over all rows and compute +# different values. The filter `y = k` fixes `y` for each outer row, so the pull +# up adds `y` to the `PARTITION BY` of the window, which keeps the values. statement ok CREATE TABLE win_outer(k INT) AS VALUES (1), (2), (5); @@ -3045,39 +3046,49 @@ CREATE TABLE win_inner(id INT, y INT) AS VALUES (1, 1), (2, 2); statement ok set datafusion.explain.logical_plan_only = true; -# The correlated filter is below the window, so the subquery stays correlated. +# The correlated filter is below the window, so the window is partitioned by b.y. query TT EXPLAIN SELECT win_outer.k FROM win_outer WHERE EXISTS (SELECT 1 FROM (SELECT b.id, row_number() OVER (ORDER BY b.id) AS rn FROM (SELECT * FROM win_inner WHERE win_inner.y = win_outer.k) AS b) AS w WHERE w.rn = 1); ---- logical_plan -01)Filter: EXISTS () -02)--Subquery: -03)----Projection: Int64(1) -04)------SubqueryAlias: w -05)--------Projection: b.id, row_number() ORDER BY [b.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS rn -06)----------Filter: row_number() ORDER BY [b.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW = UInt64(1) -07)------------WindowAggr: windowExpr=[[row_number() ORDER BY [b.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] -08)--------------SubqueryAlias: b -09)----------------Projection: win_inner.id, win_inner.y -10)------------------Filter: win_inner.y = outer_ref(win_outer.k) -11)--------------------TableScan: win_inner -12)--TableScan: win_outer projection=[k] - -# It errors rather than returning wrong results. The right answer is `1, 2`. -statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression Exists +01)LeftSemi Join: win_outer.k = __correlated_sq_1.y +02)--TableScan: win_outer projection=[k] +03)--SubqueryAlias: __correlated_sq_1 +04)----SubqueryAlias: w +05)------Projection: b.y +06)--------Filter: row_number() PARTITION BY [b.y] ORDER BY [b.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW = UInt64(1) +07)----------Projection: b.y, row_number() PARTITION BY [b.y] ORDER BY [b.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW +08)------------WindowAggr: windowExpr=[[row_number() PARTITION BY [b.y] ORDER BY [b.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] +09)--------------SubqueryAlias: b +10)----------------TableScan: win_inner projection=[id, y] + +# The right answer is `1, 2`. +query I SELECT win_outer.k FROM win_outer WHERE EXISTS (SELECT 1 FROM (SELECT b.id, row_number() OVER (ORDER BY b.id) AS rn FROM (SELECT * FROM win_inner WHERE win_inner.y = win_outer.k) AS b) AS w WHERE w.rn = 1) ORDER BY win_outer.k; +---- +1 +2 # The same for NOT EXISTS. The right answer is `5`. -statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression Exists +query I SELECT win_outer.k FROM win_outer WHERE NOT EXISTS (SELECT 1 FROM (SELECT b.id, row_number() OVER (ORDER BY b.id) AS rn FROM (SELECT * FROM win_inner WHERE win_inner.y = win_outer.k) AS b) AS w WHERE w.rn = 1) ORDER BY win_outer.k; +---- +5 # The same for IN. The right answer is `1, 2`. -statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression InSubquery +query I SELECT win_outer.k FROM win_outer WHERE 1 IN (SELECT row_number() OVER (ORDER BY win_inner.id) FROM win_inner WHERE win_inner.y = win_outer.k) ORDER BY win_outer.k; +---- +1 +2 # And for a scalar subquery. The right answer is `(1, 1), (2, 1), (5, NULL)`. -statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression ScalarSubquery +query II SELECT win_outer.k, (SELECT max(w.rn) FROM (SELECT row_number() OVER (ORDER BY win_inner.id) AS rn FROM win_inner WHERE win_inner.y = win_outer.k) AS w) AS m FROM win_outer ORDER BY win_outer.k; +---- +1 1 +2 1 +5 NULL # A correlated filter above the window is still pulled up, because the window # input does not depend on the outer row. @@ -3100,8 +3111,8 @@ SELECT win_outer.k FROM win_outer WHERE EXISTS (SELECT 1 FROM (SELECT win_inner. 2 # A correlated filter below the window that reads only a `PARTITION BY` column -# keeps or drops whole partitions, so it is pulled above the window. This holds -# for a non-equality filter too. The right answer is `1`. +# keeps or drops whole partitions, so it is pulled above the window as it is. +# This holds for a non-equality filter too. The right answer is `1`. query I SELECT win_outer.k FROM win_outer WHERE EXISTS (SELECT 1 FROM (SELECT row_number() OVER (PARTITION BY win_inner.y ORDER BY win_inner.id) AS rn FROM win_inner WHERE win_inner.y > win_outer.k) AS w WHERE w.rn = 1) ORDER BY win_outer.k; ---- @@ -3122,9 +3133,14 @@ SELECT win_outer.k, (SELECT max(w.rn) FROM (SELECT row_number() OVER (PARTITION 2 1 5 NULL -# A window that partitions by an expression of the column does not count. -statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression ScalarSubquery +# A window that partitions by an expression of the column is partitioned by +# the column as well. +query II SELECT win_outer.k, (SELECT max(w.rn) FROM (SELECT row_number() OVER (PARTITION BY win_inner.y + 0 ORDER BY win_inner.id) AS rn FROM win_inner WHERE win_inner.y = win_outer.k) AS w) AS m FROM win_outer ORDER BY win_outer.k; +---- +1 1 +2 1 +5 NULL # A window over an aggregate with count bug handling stays correlated. statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression ScalarSubquery From 85a3453c88123bb1cca71e4505a602c5d1022aa2 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 30 Sep 2026 02:54:59 +0000 Subject: [PATCH 8/8] Revert "feat: decorrelate a window below a correlated equality filter" Keep this PR to the bug fix so it can be backported. Adding the correlated columns to PARTITION BY is a new optimization and moves to a follow up. This reverts commit 1a2ad03. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01NZEssuMQANC3qLy1k2zUus --- datafusion/optimizer/src/decorrelate.rs | 73 ++----------------- .../sqllogictest/test_files/lateral_join.slt | 31 ++------ .../sqllogictest/test_files/subquery.slt | 70 +++++++----------- 3 files changed, 39 insertions(+), 135 deletions(-) diff --git a/datafusion/optimizer/src/decorrelate.rs b/datafusion/optimizer/src/decorrelate.rs index 6b03fc5122063..3b11237fa34ab 100644 --- a/datafusion/optimizer/src/decorrelate.rs +++ b/datafusion/optimizer/src/decorrelate.rs @@ -37,7 +37,7 @@ use datafusion_expr::utils::{ }; use datafusion_expr::{ BinaryExpr, Cast, Distinct, EmptyRelation, Expr, ExprSchemable, FetchType, - LogicalPlan, LogicalPlanBuilder, Operator, Window, expr, lit, + LogicalPlan, LogicalPlanBuilder, Operator, expr, lit, }; /// This struct rewrite the sub query plan by pull up the correlated @@ -563,30 +563,17 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { ); // A filter that reads only `PARTITION BY` columns keeps or drops // whole partitions, so each window function sees the same rows - // above it or below it. - if local_correlated_cols.is_empty() - || window + // above it or below it. A filter on any other column changes the + // rows of a partition. + if !local_correlated_cols.is_empty() + && !window .window_expr .iter() .all(|expr| partitions_by_all(expr, &local_correlated_cols)) - { - return self.pass_through_pulled_up_cols(plan); - } - // A filter that fixes the value of each column it reads for each - // outer row, like `inner.y = outer.k`, keeps the rows of one - // value of those columns. Partitioning by those columns as well - // gives each window function the same rows once the filter is - // above it. This is what the pull up does for an Aggregate too. - if !self.can_pull_over_aggregation - || self.collected_count_expr_map.contains_key(&*window.input) { self.can_pull_up = false; - return self.pass_through_pulled_up_cols(plan); } - Ok(Transformed::yes(partition_window_by( - window, - &local_correlated_cols, - )?)) + self.pass_through_pulled_up_cols(plan) } // `f_down` checked that the pulled up filters do not read an // unnested column. @@ -790,54 +777,6 @@ fn correlated_filter_columns(plan: &LogicalPlan) -> Vec { cols } -/// Adds `cols` to the `PARTITION BY` of every window function of `window`. -/// That changes the names of the window function columns, so a Projection on -/// top gives them their old names back. -fn partition_window_by(window: &Window, cols: &BTreeSet) -> Result { - let new_window_expr = window - .window_expr - .iter() - .map(|expr| add_partition_by(expr, cols)) - .collect::>(); - let mut proj_exprs = window - .input - .schema() - .columns() - .into_iter() - .map(Expr::Column) - .collect::>(); - for (old, new) in window.window_expr.iter().zip(&new_window_expr) { - let new_col = Expr::Column(Column::from_name(new.schema_name().to_string())); - proj_exprs.push(new_col.alias(old.schema_name().to_string())); - } - LogicalPlanBuilder::from(LogicalPlan::Window(Window::try_new( - new_window_expr, - Arc::clone(&window.input), - )?)) - .project(proj_exprs)? - .build() -} - -fn add_partition_by(expr: &Expr, cols: &BTreeSet) -> Expr { - match expr { - Expr::Alias(alias) => Expr::Alias(Alias { - expr: Box::new(add_partition_by(&alias.expr, cols)), - ..alias.clone() - }), - Expr::WindowFunction(window_fun) => { - let mut window_fun = window_fun.clone(); - for col in cols { - let col = Expr::Column(col.clone()); - if !window_fun.params.partition_by.contains(&col) { - window_fun.params.partition_by.push(col); - } - } - Expr::WindowFunction(window_fun) - } - _ => expr.clone(), - } -} - /// Whether the window function `expr` lists each of `cols` as a plain /// `PARTITION BY` column. fn partitions_by_all(expr: &Expr, cols: &BTreeSet) -> bool { diff --git a/datafusion/sqllogictest/test_files/lateral_join.slt b/datafusion/sqllogictest/test_files/lateral_join.slt index b531df54f0ff1..0e0ceb9cfa4fa 100644 --- a/datafusion/sqllogictest/test_files/lateral_join.slt +++ b/datafusion/sqllogictest/test_files/lateral_join.slt @@ -755,20 +755,16 @@ ORDER BY t1.id, sub.x; # --- Correlated filter below a window function --- # https://github.com/apache/datafusion/issues/25792 -# Pulling `t2.t1_id = t1.id` above the window as it is would number the t2 rows -# of all t1 rows together. The filter fixes `t2.t1_id` for each t1 row, so the -# pull up adds `t2.t1_id` to the `PARTITION BY` of the window instead. -query III +# Pulling `t2.t1_id = t1.id` above the window would number the t2 rows of all +# t1 rows together instead of the t2 rows of each t1 row. +# Expected result once supported: (1, 10, 1), (1, 20, 2), (2, 30, 1). +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression OuterReferenceColumn SELECT t1.id, sub.id, sub.rn FROM t1, LATERAL ( SELECT t2.id, row_number() OVER (ORDER BY t2.id) AS rn FROM (SELECT * FROM t2 WHERE t2.t1_id = t1.id) AS t2 ) AS sub ORDER BY t1.id, sub.id; ----- -1 10 1 -1 20 2 -2 30 1 # A correlated filter above the window is still decorrelated. The window # numbers all t2 rows, then the filter keeps the rows of each t1 row. @@ -854,8 +850,8 @@ ORDER BY t1.id, sub.id; 1 20 2 2 30 1 -# Every window function is partitioned by the columns of the filter. -query IIII +# The filter must read a `PARTITION BY` column of every window function. +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression OuterReferenceColumn SELECT t1.id, sub.id, sub.rn, sub.c FROM t1, LATERAL ( SELECT t2.id, @@ -864,21 +860,6 @@ FROM t1, LATERAL ( FROM t2 WHERE t2.t1_id = t1.id ) AS sub ORDER BY t1.id, sub.id; ----- -1 10 1 1 -1 20 2 1 -2 30 1 1 - -# A non-equality filter does not fix the value of its column for each t1 row, -# so partitioning by the column would split the rows of one t1 row. The -# subquery stays correlated. -statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression OuterReferenceColumn -SELECT t1.id, sub.id, sub.rn -FROM t1, LATERAL ( - SELECT t2.id, row_number() OVER (ORDER BY t2.id) AS rn - FROM t2 WHERE t2.t1_id >= t1.id -) AS sub -ORDER BY t1.id, sub.id; # --- Implicit LATERAL (comma without LATERAL keyword) --- # DuckDB auto-detects lateral correlation; DataFusion requires diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index cefa32b24891b..40326bebe9b06 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -3032,11 +3032,10 @@ DROP TABLE oj_a; statement ok DROP TABLE oj_b; -# Regression test for #25792: a correlated filter below a window function -# selects the rows the window function sees for each outer row. Pulled above -# the window as it is, the window function would run over all rows and compute -# different values. The filter `y = k` fixes `y` for each outer row, so the pull -# up adds `y` to the `PARTITION BY` of the window, which keeps the values. +# Regression test for #25792: a correlated filter below a window function must +# not be pulled above the window. The filter selects the rows the window +# function sees for each outer row, so once it is above the window the window +# function runs over all rows and computes different values. statement ok CREATE TABLE win_outer(k INT) AS VALUES (1), (2), (5); @@ -3046,49 +3045,39 @@ CREATE TABLE win_inner(id INT, y INT) AS VALUES (1, 1), (2, 2); statement ok set datafusion.explain.logical_plan_only = true; -# The correlated filter is below the window, so the window is partitioned by b.y. +# The correlated filter is below the window, so the subquery stays correlated. query TT EXPLAIN SELECT win_outer.k FROM win_outer WHERE EXISTS (SELECT 1 FROM (SELECT b.id, row_number() OVER (ORDER BY b.id) AS rn FROM (SELECT * FROM win_inner WHERE win_inner.y = win_outer.k) AS b) AS w WHERE w.rn = 1); ---- logical_plan -01)LeftSemi Join: win_outer.k = __correlated_sq_1.y -02)--TableScan: win_outer projection=[k] -03)--SubqueryAlias: __correlated_sq_1 -04)----SubqueryAlias: w -05)------Projection: b.y -06)--------Filter: row_number() PARTITION BY [b.y] ORDER BY [b.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW = UInt64(1) -07)----------Projection: b.y, row_number() PARTITION BY [b.y] ORDER BY [b.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW -08)------------WindowAggr: windowExpr=[[row_number() PARTITION BY [b.y] ORDER BY [b.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] -09)--------------SubqueryAlias: b -10)----------------TableScan: win_inner projection=[id, y] - -# The right answer is `1, 2`. -query I +01)Filter: EXISTS () +02)--Subquery: +03)----Projection: Int64(1) +04)------SubqueryAlias: w +05)--------Projection: b.id, row_number() ORDER BY [b.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS rn +06)----------Filter: row_number() ORDER BY [b.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW = UInt64(1) +07)------------WindowAggr: windowExpr=[[row_number() ORDER BY [b.id ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] +08)--------------SubqueryAlias: b +09)----------------Projection: win_inner.id, win_inner.y +10)------------------Filter: win_inner.y = outer_ref(win_outer.k) +11)--------------------TableScan: win_inner +12)--TableScan: win_outer projection=[k] + +# It errors rather than returning wrong results. The right answer is `1, 2`. +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression Exists SELECT win_outer.k FROM win_outer WHERE EXISTS (SELECT 1 FROM (SELECT b.id, row_number() OVER (ORDER BY b.id) AS rn FROM (SELECT * FROM win_inner WHERE win_inner.y = win_outer.k) AS b) AS w WHERE w.rn = 1) ORDER BY win_outer.k; ----- -1 -2 # The same for NOT EXISTS. The right answer is `5`. -query I +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression Exists SELECT win_outer.k FROM win_outer WHERE NOT EXISTS (SELECT 1 FROM (SELECT b.id, row_number() OVER (ORDER BY b.id) AS rn FROM (SELECT * FROM win_inner WHERE win_inner.y = win_outer.k) AS b) AS w WHERE w.rn = 1) ORDER BY win_outer.k; ----- -5 # The same for IN. The right answer is `1, 2`. -query I +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression InSubquery SELECT win_outer.k FROM win_outer WHERE 1 IN (SELECT row_number() OVER (ORDER BY win_inner.id) FROM win_inner WHERE win_inner.y = win_outer.k) ORDER BY win_outer.k; ----- -1 -2 # And for a scalar subquery. The right answer is `(1, 1), (2, 1), (5, NULL)`. -query II +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression ScalarSubquery SELECT win_outer.k, (SELECT max(w.rn) FROM (SELECT row_number() OVER (ORDER BY win_inner.id) AS rn FROM win_inner WHERE win_inner.y = win_outer.k) AS w) AS m FROM win_outer ORDER BY win_outer.k; ----- -1 1 -2 1 -5 NULL # A correlated filter above the window is still pulled up, because the window # input does not depend on the outer row. @@ -3111,8 +3100,8 @@ SELECT win_outer.k FROM win_outer WHERE EXISTS (SELECT 1 FROM (SELECT win_inner. 2 # A correlated filter below the window that reads only a `PARTITION BY` column -# keeps or drops whole partitions, so it is pulled above the window as it is. -# This holds for a non-equality filter too. The right answer is `1`. +# keeps or drops whole partitions, so it is pulled above the window. This holds +# for a non-equality filter too. The right answer is `1`. query I SELECT win_outer.k FROM win_outer WHERE EXISTS (SELECT 1 FROM (SELECT row_number() OVER (PARTITION BY win_inner.y ORDER BY win_inner.id) AS rn FROM win_inner WHERE win_inner.y > win_outer.k) AS w WHERE w.rn = 1) ORDER BY win_outer.k; ---- @@ -3133,14 +3122,9 @@ SELECT win_outer.k, (SELECT max(w.rn) FROM (SELECT row_number() OVER (PARTITION 2 1 5 NULL -# A window that partitions by an expression of the column is partitioned by -# the column as well. -query II +# A window that partitions by an expression of the column does not count. +statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression ScalarSubquery SELECT win_outer.k, (SELECT max(w.rn) FROM (SELECT row_number() OVER (PARTITION BY win_inner.y + 0 ORDER BY win_inner.id) AS rn FROM win_inner WHERE win_inner.y = win_outer.k) AS w) AS m FROM win_outer ORDER BY win_outer.k; ----- -1 1 -2 1 -5 NULL # A window over an aggregate with count bug handling stays correlated. statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression ScalarSubquery