Wrong results: a correlated filter under the nullable side of an outer join is pulled above the join #25507

Describe the bug

A correlated filter that sits on the nullable side of an outer join inside a subquery is pulled out of the subquery and attached to the decorrelated join. The filter then applies after the outer join instead of before it, so the rows that the outer join extends with NULLs are different and the query gives wrong results.

This affects EXISTS, IN and NOT IN subqueries. It is not specific to three-valued logic: the EXISTS form below is wrong as well.

To Reproduce

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

The subquery keeps only the rows of b where b.y = o.k, then left joins a to them. For o.k = 5 no row of b remains, so every row of a is unmatched and b.y is NULL for all of them.

EXISTS form

SELECT o.k,
       EXISTS (SELECT 1
               FROM a LEFT JOIN (SELECT * FROM b WHERE b.y = o.k) AS b
                 ON a.id = b.id
               WHERE b.y IS NULL) AS e
FROM o ORDER BY k;

For o.k = 1 only b.id = 1 remains, so a.id = 2 is unmatched and b.y is NULL for it. For o.k = 5 both rows of a are unmatched. The correct answer is true for both rows.

IN form

SELECT o.k,
       o.k IN (SELECT b.y
               FROM a LEFT JOIN (SELECT * FROM b WHERE b.y = o.k) AS b
                 ON a.id = b.id) AS m
FROM o ORDER BY k;

For o.k = 5 the subquery gives {NULL, NULL}, so 5 IN (...) is UNKNOWN.

Expected behavior

The results of DuckDB above.

Additional context

The plan shows the cause. Filter: b.y = o.k is gone from below the Left Join and is now the condition of the mark join:

Projection: o.k, __correlated_sq_1.mark AS e
  LeftMark Join: o.k = __correlated_sq_1.y
    TableScan: o projection=[k]
    SubqueryAlias: __correlated_sq_1
      Filter: b.y IS NULL
        Projection: b.y
          Left Join: a.id = b.id
            TableScan: a projection=[id]
            SubqueryAlias: b
              TableScan: b projection=[id, y]

Filter: b.y IS NULL and o.k = __correlated_sq_1.y cannot both hold, so the mark is always false.

PullUpCorrelatedExpr in datafusion/optimizer/src/decorrelate.rs pulls a correlated Filter up through every node it does not know, and LogicalPlan::Join is one of those nodes. A correlated filter can move above an Inner join without a change of meaning, but not above the nullable side of an outer join.

Found on main at 64871d9. It is independent of #25480 and of #25338.

#25283 is the same theme in a different node: PullUpCorrelatedExpr walks through a plan node that it does not model, there a Limit with an OFFSET, here a Join. A fix can probably reuse the unsupported() helper that #25284 adds.