Wrong results: a correlated filter under the right side of an ASOF JOIN is pulled above the join #25834

Describe the bug

A correlated filter that sits below the right side of an ASOF JOIN inside a subquery is pulled out of the subquery and attached to the decorrelated join. The ASOF JOIN then picks the nearest row from all rows of the right input, instead of from the rows that match the outer row, so the query gives wrong results.

This affects LATERAL and EXISTS subqueries.

To Reproduce

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);

For each row of o, the subquery keeps only the rows of r where r.val = o.k, then finds the latest of those rows with r.ts <= l.ts. For o.k = 1 that is ts = 5, and for o.k = 2 it is ts = 8.

LATERAL form

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;

EXISTS form

SELECT o.k FROM o
WHERE EXISTS (
  SELECT 1
  FROM l ASOF JOIN (SELECT * FROM r WHERE r.val = o.k) AS r
    MATCH_CONDITION (l.ts >= r.ts)
  WHERE r.ts IS NOT NULL)
ORDER BY o.k;

The same query without the correlation gives the correct row. With r.val = 1 instead of r.val = o.k, DataFusion returns ts = 5.

Reference results. PostgreSQL has no ASOF JOIN. This equivalent query, with ORDER BY ... LIMIT 1 in place of the ASOF JOIN, returns (1, 5), (2, 8) on PostgreSQL 17.11 and on DuckDB 1.5.2:

SELECT o.k, s.rts FROM o, LATERAL (
  SELECT m.ts AS rts FROM l LEFT JOIN LATERAL (
    SELECT r2.ts FROM (SELECT * FROM r WHERE r.val = o.k) r2
    WHERE l.ts >= r2.ts ORDER BY r2.ts DESC LIMIT 1) m ON true) s
ORDER BY o.k;

Note: DuckDB 1.5.2 gives the same wrong result as DataFusion for the ASOF JOIN form (l ASOF JOIN ... ON l.ts >= r.ts). It gives the correct result for the uncorrelated form.

Expected behavior

The correct results above: (1, 5), (2, 8) for the LATERAL form and 1, 2 for the EXISTS form. If DataFusion cannot decorrelate the subquery, it should fail with a not-implemented error. It must not return wrong rows.

Additional context

The plan shows the cause. r.val = o.k is gone from below the AsOf Join and is now the condition of the join that replaces the LATERAL subquery:

Projection: o.k, s.rts
  Inner Join: o.k = s.val
    TableScan: o projection=[k]
    SubqueryAlias: s
      Projection: r.ts AS rts, r.val
        AsOf Join: match=[l.ts >= r.ts], constraint=On
          TableScan: l projection=[ts]
          SubqueryAlias: r
            TableScan: r projection=[ts, val]

PullUpCorrelatedExpr::f_down in datafusion/optimizer/src/decorrelate.rs has no arm for LogicalPlan::AsOfJoin, so the default arm pulls the filter through it. push_down_filter handles AsOfJoin correctly in the other direction: it moves only left-only predicates below the join.

Found on main at 871058c.

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

• Wrong results: a correlated filter under the nullable side of an outer join is pulled above the join #25507 (Join, fixed by fix: keep a correlated filter below the nullable side of an outer join #25764)
• Wrong results: a correlated filter under a window function is pulled above the window #25792 (Window, fix: keep a subquery correlated when its filter is below a window function #25810)
• Wrong results: LATERAL with SELECT DISTINCT and a non-equality correlated filter returns duplicate rows #25808 (Aggregate used as DISTINCT)
• Correlated EXISTS subquery with OFFSET returns wrong results #25283 (Limit with an OFFSET)

The default arm of PullUpCorrelatedExpr::f_down pulls a correlated filter through every plan node that the rule does not know. The default in push_down_filter is the opposite: the filter stays where it is. A change that makes PullUpCorrelatedExpr allow only the nodes it knows are safe would fix this bug and prevent the next one of this class.