Wrong results: a correlated filter under a window function is pulled above the window

[adriangb]

Describe the bug

A correlated filter that sits below a window function inside a subquery is pulled out of the subquery and attached to the decorrelated join. The window function then runs over all rows of the inner table instead of only the rows that match the outer row, so row_number(), rank(), lag() and similar functions give different values and the query gives wrong results.

This affects EXISTS and LATERAL subqueries.

To Reproduce

CREATE TABLE o(k INT) AS VALUES (1), (2), (5);
CREATE TABLE b(id INT, y INT) AS VALUES (1, 1), (2, 2);

For each row of o, the subquery keeps only the rows of b where b.y = o.k, then numbers them. Each k matches at most one row of b, so that row always gets rn = 1.

EXISTS form

SELECT o.k FROM o
WHERE EXISTS (
  SELECT 1
  FROM (SELECT b.id, row_number() OVER (ORDER BY b.id) AS rn
        FROM (SELECT * FROM b WHERE b.y = o.k) AS b) AS w
  WHERE w.rn = 1)
ORDER BY o.k;

DataFusion	DuckDB 1.5.2	PostgreSQL 17.11
1	1, 2	1, 2

LATERAL form

SELECT o.k, w.id, w.rn
FROM o, LATERAL (
  SELECT b.id, row_number() OVER (ORDER BY b.id) AS rn
  FROM (SELECT * FROM b WHERE b.y = o.k) AS b) AS w
ORDER BY o.k;

k	id	DataFusion rn	DuckDB 1.5.2 rn	PostgreSQL 17.11 rn
1	1	1	1	1
2	2	2	1	1

Expected behavior

The results of DuckDB and PostgreSQL above.

Additional context

The plan shows the cause. Filter: b.y = o.k is gone from below the WindowAggr and is now the condition of the semi join, so row_number() numbers both rows of b:

LeftSemi Join: o.k = __correlated_sq_1.y
  TableScan: o projection=[k]
  SubqueryAlias: __correlated_sq_1
    SubqueryAlias: w
      Projection: b.y
        Filter: row_number() ORDER BY [b.id ASC NULLS LAST] ... = UInt64(1)
          Projection: b.y, row_number() ORDER BY [b.id ASC NULLS LAST] ...
            WindowAggr: windowExpr=[[row_number() ORDER BY [b.id ASC NULLS LAST] ...]]
              SubqueryAlias: b
                TableScan: b projection=[id, y]

PullUpCorrelatedExpr in datafusion/optimizer/src/decorrelate.rs pulls a correlated Filter up through every node that it does not model. LogicalPlan::Window is one of those nodes. A filter can only move above a window if it reads only the PARTITION BY columns, which is not the case here (the window has no PARTITION BY).

Found on main at 6a792c6.

This is the same class of bug as these, each for a different node:

• Wrong results: a correlated filter under the nullable side of an outer join is pulled above the join #25507 (Join, the nullable side of an outer join), fixed by fix: keep a correlated filter below the nullable side of an outer join #25764
• Correlated EXISTS subquery with OFFSET returns wrong results #25283 (Limit with an OFFSET), fixed by fix: keep the OFFSET of a correlated EXISTS subquery #25284

A fix can do the same as those PRs: mark the subquery as not pull-up-able when a Window has outer references below it, so the query fails with a not-implemented error instead of returning wrong results.

[mohitgurav20]

I took a deep dive into PullUpCorrelatedExpr in datafusion/optimizer/src/decorrelate.rs to trace why correlated filters under window functions get pulled up into decorrelated joins.

Root Cause

During f_up traversal in PullUpCorrelatedExpr:

1. LogicalPlan::Filter extracts the outer-referencing predicates into self.join_filters.
2. When bottom-up rewriting reaches LogicalPlan::Window, the match statement in f_up currently lacks an arm for Window. It falls through to _ => Ok(Transformed::no(plan)) while leaving can_pull_up = true.
3. The correlated predicate gets pulled past the window operator into the generated Join. Consequently, window evaluation runs over the entire inner relation prior to join filtering, distorting rank/row count/aggregation results.

Proposed Fix

We should handle LogicalPlan::Window in f_up:

• Check if join filters or IN-predicates were pulled up from below (!self.join_filters.is_empty() || self.in_predicate_opt.is_some()).
• Check whether the window's PARTITION BY expressions cover all inner columns referenced in the pulled-up join filters.
• If the correlated columns are not covered by PARTITION BY (or if PARTITION BY is empty), set self.can_pull_up = false. This gracefully prevents invalid decorrelation and falls back to scalar/unsupported subquery handling.

I can open a PR with the fix and regression test cases in sqllogictest if you'd like to assign this to me!