Wrong results: LATERAL with SELECT DISTINCT and a non-equality correlated filter returns duplicate rows #25808

Describe the bug

A LATERAL subquery with SELECT DISTINCT and a correlated filter that is not an equality (for example >) returns duplicate rows. The correlated filter is pulled above the DISTINCT, and the column that the filter reads is added to the DISTINCT keys. Rows that differ only in that column are then not merged.

To Reproduce

CREATE TABLE po(k INT) AS VALUES (0), (5);
CREATE TABLE pb(x INT, y INT) AS VALUES (1, 1), (1, 2), (2, 3);

SELECT po.k, s.x
FROM po, LATERAL (SELECT DISTINCT pb.x FROM pb WHERE pb.y > po.k) AS s
ORDER BY po.k, s.x;

For po.k = 0, all three rows of pb match, and the distinct values of x are 1 and 2.

The same query with GROUP BY instead of DISTINCT does not give wrong results. DataFusion does not decorrelate it and fails with a not-implemented error:

SELECT po.k, s.x, s.n
FROM po, LATERAL (SELECT pb.x, count(*) AS n FROM pb WHERE pb.y > po.k GROUP BY pb.x) AS s;
-- This feature is not implemented: Physical plan does not support logical expression OuterReferenceColumn(...)

With an equality filter (pb.y = po.k) the DISTINCT form gives the correct result.

Expected behavior

The results of DuckDB and PostgreSQL above: each distinct x one time for each po row.

Additional context

The plan shows the cause. pb.y is now a group-by key of the DISTINCT aggregate, and pb.y > po.k is the join filter:

Projection: po.k, s.x
  Inner Join:  Filter: s.y > po.k
    TableScan: po projection=[k]
    SubqueryAlias: s
      Aggregate: groupBy=[[pb.x, pb.y]], aggr=[[]]
        TableScan: pb projection=[x, y]

In PullUpCorrelatedExpr::f_up in datafusion/optimizer/src/decorrelate.rs, the Aggregate arm does the can_pull_over_aggregation check only when the aggregate is not a DISTINCT. That check allows only equality filters to move above an aggregate. For EXISTS and IN, adding a column to the DISTINCT keys does not change the result, because duplicate rows do not change whether a match exists. For LATERAL, every row is output, so the duplicate rows are visible.

A fix can do the can_pull_over_aggregation check for a DISTINCT aggregate as well, except for EXISTS and IN subqueries. The GROUP BY form already gets this check.

Found on main at 6a792c6.

This is the same class of bug as these: PullUpCorrelatedExpr pulls a correlated filter through a plan node where that changes the result.

• Wrong results: a correlated filter under the nullable side of an outer join is pulled above the join #25507 (outer join)
• Wrong results: a correlated filter under a window function is pulled above the window #25792 (window function)
• Correlated EXISTS subquery with OFFSET returns wrong results #25283 (Limit with an OFFSET)