You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/03 20:13:22 UTC
(datafusion) 01/01: fix: pull a correlated filter only through plan nodes known to be safe (#25838)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25838-14b8de8b76f5a5ff6ae113eb554be581e1dc2b94
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit b2cfc23ad4cab1fd00b6e56d5a1dee84f70d111d
Author: Adrian Garcia Badaracco <17...@users.noreply.github.com>
AuthorDate: Sat Oct 3 20:12:59 2026 +0000
fix: pull a correlated filter only through plan nodes known to be safe (#25838)
## Which issue does this PR close?
- Closes https://github.com/apache/datafusion/issues/25834.
- Closes https://github.com/apache/datafusion/issues/25792.
This PR also contains the code change of
https://github.com/apache/datafusion/pull/25810 (`Window`) and its tests
(the test commit is the same). #25810 has no reviews yet, so I suggest
to review this PR instead and close #25810 when this PR merges.
Related: https://github.com/apache/datafusion/issues/25837 (found while
writing the tests: the `LATERAL` test uses the table name as the alias
of the derived table because of it),
https://github.com/apache/datafusion/issues/25507,
https://github.com/apache/datafusion/issues/25792,
https://github.com/apache/datafusion/issues/25808,
https://github.com/apache/datafusion/issues/25283.
## Rationale for this change
A correlated subquery that has an ASOF join returns wrong rows when the
correlated filter is on the right side of the join.
```sql
CREATE TABLE o(k INT) AS VALUES (1), (2);
CREATE TABLE l(id INT, ts INT) AS VALUES (1, 10);
CREATE TABLE r(ts INT, val INT) AS VALUES (5, 1), (8, 2);
SELECT o.k, s.rts
FROM o, LATERAL (
SELECT r.ts AS rts
FROM l ASOF JOIN (SELECT * FROM r WHERE r.val = o.k) AS r
MATCH_CONDITION (l.ts >= r.ts)
) AS s
ORDER BY o.k;
```
| | Result |
|---|---|
| Correct | `(1, 5), (2, 8)` |
| `main` | `(2, 8)` |
| This PR | `This feature is not implemented: Physical plan does not
support logical expression OuterReferenceColumn(...)` |
The filter `r.val = o.k` selects the rows the ASOF join can match. The
decorrelation moves it above the join. There, the join has already
matched `l` to `r.ts = 8` for every outer row, and the filter removes
the row for `o.k = 1`.
The root cause is not specific to the ASOF join. `PullUpCorrelatedExpr`
moves a correlated filter above every plan node that it does not block
explicitly. Each node it does not know is a possible wrong result.
https://github.com/apache/datafusion/issues/25507 (outer join) and
https://github.com/apache/datafusion/issues/25792 (window) are two more
cases of the same problem. `PushDownFilter` asks the same question in
the other direction, and there the default is safe: a filter stays where
it is unless the rule knows the node.
## What changes are included in this PR?
`PullUpCorrelatedExpr::f_down` now matches every `LogicalPlan` variant,
with no `_` arm. A new variant does not compile until somebody puts it
in a group.
| Group | Variants | What `f_down` does |
|---|---|---|
| Special arms (no change) | `Filter`, `Subquery`, `Join`, `Union`,
`Sort`, `Extension`, `Limit` | Same as before |
| Pass-through | `Projection`, `Aggregate`, `Distinct::All`, `Join`
(preserved sides), `AsOfJoin` (left side), `Repartition`,
`SubqueryAlias`, `TableScan`, `EmptyRelation`, `Values` | Continue. Stop
when the node's own expressions hold an outer reference (same as
before). |
| Stop when the input holds an outer reference | `AsOfJoin` (right side)
| New |
| Stop when the input holds an outer reference | `Window` | New (same as
https://github.com/apache/datafusion/pull/25810) |
| Stop when the input holds an outer reference | `Distinct::On`,
`Unnest`, `RecursiveQuery`, `Statement`, `Explain`, `Analyze`, `Dml`,
`Ddl`, `Copy`, `DescribeTable` | New |
"Stop" means the subquery stays correlated. The check uses
`holds_outer_reference`, which does not go into a nested `Subquery`.
The effect for each new stop:
| Node | Before | After |
|---|---|---|
| `AsOfJoin`, filter on the right side | Wrong rows | Not-implemented
error |
| `Window` | Wrong rows | Not-implemented error |
| `Unnest` | `Optimizer rule 'decorrelate_predicate_subquery' failed:
Schema error: No field named t.id` | Not-implemented error |
| `Distinct::On`, `RecursiveQuery` | Not reachable in the default
optimizer (`ReplaceDistinctWithAggregate` runs first) or not seen in
tests | Stop |
| `Statement`, `Explain`, `Analyze`, `Dml`, `Ddl`, `Copy`,
`DescribeTable` | Not in a subquery | Stop |
## What is the testing strategy for this PR?
New `sqllogictest` cases:
| File | Case | Without the fix |
|---|---|---|
| `asof_join.slt` | LATERAL, filter on the right side: `EXPLAIN` shows
the subquery stays correlated, and the query errors | Wrong rows `(2,
8)` |
| `asof_join.slt` | EXISTS, filter on the right side: the query errors |
Wrong result (no rows, correct is `1`) |
| `asof_join.slt` | EXISTS, filter on the left side: `EXPLAIN` and
result | Same plan and result |
| `asof_join.slt` | LATERAL, filter above the join: `EXPLAIN` and result
| Same plan and result |
| `subquery.slt` | EXISTS, filter below an `Unnest`: `EXPLAIN` and the
error | Schema error |
| `subquery.slt`, `lateral_join.slt` | Window cases from
https://github.com/apache/datafusion/pull/25810 | Wrong rows |
All new negative cases fail when `decorrelate.rs` is reverted to `main`.
The full `sqllogictest` suite (including the TPC-H plan files) and
`cargo test -p datafusion-optimizer` pass with no change to an existing
test.
## Are there any user-facing changes?
Queries that returned wrong rows (ASOF join, window) or failed with a
schema error (unnest) now fail with a not-implemented error. No API
change.
Out of scope: the `Aggregate` and `Limit` arms still have exemptions
that are too broad (https://github.com/apache/datafusion/issues/25808,
https://github.com/apache/datafusion/issues/25283). This PR does not
change them. `Distinct::All` is a pass-through, with the same behavior
as an `Aggregate` without aggregate expressions.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <no...@anthropic.com>
---
datafusion/optimizer/src/decorrelate.rs | 211 ++++++++++++++++++---
.../src/decorrelate_predicate_subquery.rs | 38 ++++
datafusion/sqllogictest/test_files/asof_join.slt | 103 ++++++++++
.../sqllogictest/test_files/lateral_join.slt | 108 +++++++++++
datafusion/sqllogictest/test_files/subquery.slt | 192 +++++++++++++++++++
5 files changed, 630 insertions(+), 22 deletions(-)
diff --git a/datafusion/optimizer/src/decorrelate.rs b/datafusion/optimizer/src/decorrelate.rs
index 186d7a9bbe..e4577d6eb0 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, SkipType, expr, lit,
+ BinaryExpr, Cast, Distinct, EmptyRelation, Expr, ExprSchemable, FetchType,
+ LogicalPlan, LogicalPlanBuilder, Operator, SkipType, expr, lit,
};
/// This struct rewrite the sub query plan by pull up the correlated
@@ -135,13 +135,6 @@ impl PullUpCorrelatedExpr {
self.exists_sub_query = exists_sub_query;
self
}
-
- /// Mark the plan as one whose correlated expressions cannot be pulled up
- /// and stop descending into it
- fn unsupported(&mut self, plan: LogicalPlan) -> Result<Transformed<LogicalPlan>> {
- self.can_pull_up = false;
- Ok(Transformed::new(plan, false, TreeNodeRecursion::Jump))
- }
}
/// Used to indicate the unmatched rows from the inner(subquery) table after the left out Join
@@ -159,6 +152,9 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr {
type Node = LogicalPlan;
fn f_down(&mut self, plan: LogicalPlan) -> Result<Transformed<LogicalPlan>> {
+ // 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
@@ -174,24 +170,49 @@ 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))
+ }
+ // 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
+ .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 {
- self.unsupported(plan)
+ if holds_outer_reference(&plan) {
+ Ok(self.stop_pull_up(plan))
} else {
Ok(Transformed::no(plan))
}
}
+ // For EXISTS only emptiness matters, so `f_down` removes or
+ // replaces a Limit over a correlated filter where it can. Any other
+ // Limit keeps the correlated filters below it.
LogicalPlan::Limit(ref limit) => {
- if plan.all_out_ref_exprs().is_empty() {
+ if !holds_outer_reference(&plan) {
return Ok(Transformed::no(plan));
}
if !self.exists_sub_query {
- return self.unsupported(plan);
+ return Ok(self.stop_pull_up(plan));
}
// Only emptiness matters for EXISTS, so remove a limit that
// cannot make its input empty and replace one that always does
@@ -213,14 +234,67 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr {
t.transformed = true;
Ok(t)
}
- _ => self.unsupported(plan),
+ _ => Ok(self.stop_pull_up(plan)),
}
}
- _ if plan.contains_outer_reference() => {
- // the unsupported cases, the plan expressions contain out reference columns(like window expressions)
- self.unsupported(plan)
+ // 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(_)
+ | 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:
+ // - `DISTINCT ON` keeps the first row of each group, so the filter
+ // changes which row is first.
+ // - A RecursiveQuery repeats its recursive term.
+ // - The other nodes are not queries and are not in a subquery.
+ LogicalPlan::Distinct(Distinct::On(_))
+ | 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)),
}
}
@@ -486,12 +560,64 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr {
}
Ok(Transformed::no(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)),
}
}
}
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<LogicalPlan> {
+ self.can_pull_up = false;
+ 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<Transformed<LogicalPlan>> {
+ if plan
+ .inputs()
+ .iter()
+ .any(|input| self.collected_count_expr_map.contains_key(*input))
+ {
+ self.can_pull_up = false;
+ }
+ Ok(Transformed::yes(plan.recompute_schema()?))
+ }
+
/// Whether the pull up can add its columns to `group_expr` without changing
/// what the aggregate returns.
///
@@ -634,6 +760,47 @@ 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<Column> {
+ 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<Column>) -> bool {
+ 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;
+ 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 4c6bbdc83a..88a929654b 100644
--- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs
+++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs
@@ -3072,4 +3072,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 (<subquery>) [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/asof_join.slt b/datafusion/sqllogictest/test_files/asof_join.slt
index 2e5160f4e5..55fb3b52aa 100644
--- a/datafusion/sqllogictest/test_files/asof_join.slt
+++ b/datafusion/sqllogictest/test_files/asof_join.slt
@@ -551,3 +551,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;
diff --git a/datafusion/sqllogictest/test_files/lateral_join.slt b/datafusion/sqllogictest/test_files/lateral_join.slt
index 02d4d9f0ea..0e0ceb9cfa 100644
--- a/datafusion/sqllogictest/test_files/lateral_join.slt
+++ b/datafusion/sqllogictest/test_files/lateral_join.slt
@@ -753,6 +753,114 @@ 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
+
+# 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 8ab3d518fa..c1fd856497 100644
--- a/datafusion/sqllogictest/test_files/subquery.slt
+++ b/datafusion/sqllogictest/test_files/subquery.slt
@@ -3187,3 +3187,195 @@ 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 (<subquery>)
+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
+
+# 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;
+
+statement ok
+DROP TABLE win_outer;
+
+statement ok
+DROP TABLE win_inner;
+
+# 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);
+
+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)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
+
+# 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;
+
+statement ok
+DROP TABLE unnest_outer;
+
+statement ok
+DROP TABLE unnest_inner;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org