Wrong results: correlated aggregate subquery with a lossy CAST in its filter returns one row per inner value #26014

Describe the bug

A correlated subquery with an aggregate returns wrong results when its correlated filter compares a cast of an inner column with the outer row, and the cast maps several column values to one value. CAST(ts AS DATE) is one example, and CAST(x AS INT) on a DOUBLE column is another. The cast does not have to be written in the query: type coercion adds one for varchar_col = date_col.

A scalar or LATERAL subquery returns several rows for one outer row, each with a partial aggregate. EXISTS ... HAVING and IN over an aggregate drop outer rows.

To Reproduce

CREATE TABLE o(d DATE) AS VALUES (DATE '2024-01-01'), (DATE '2024-01-02');
CREATE TABLE e(ts TIMESTAMP, v INT) AS VALUES
  (TIMESTAMP '2024-01-01 10:00:00', 1),
  (TIMESTAMP '2024-01-01 11:00:00', 2),
  (TIMESTAMP '2024-01-02 09:00:00', 3);

SELECT o.d, (SELECT count(*) FROM e WHERE CAST(e.ts AS DATE) = o.d) AS c
FROM o ORDER BY o.d;

Other forms on the same tables:

-- 1. sum
SELECT o.d, (SELECT sum(e.v) FROM e WHERE CAST(e.ts AS DATE) = o.d) AS s FROM o ORDER BY o.d;
-- 2. LATERAL
SELECT o.d, s.c FROM o, LATERAL (SELECT count(*) AS c FROM e WHERE CAST(e.ts AS DATE) = o.d) AS s ORDER BY o.d;
-- 3. EXISTS with HAVING
SELECT o.d FROM o WHERE EXISTS (SELECT 1 FROM e WHERE CAST(e.ts AS DATE) = o.d HAVING count(*) > 1) ORDER BY o.d;
-- 4. IN over an aggregate
SELECT o.d FROM o WHERE 2 IN (SELECT count(*) FROM e WHERE CAST(e.ts AS DATE) = o.d) ORDER BY o.d;

Without an explicit cast:

CREATE TABLE es(s VARCHAR) AS VALUES ('2024-01-01'), ('2024-01-01T10:00:00'), ('2024-01-02');

SELECT o.d, (SELECT count(*) FROM es WHERE es.s = o.d) AS c FROM o ORDER BY o.d;

Type coercion turns es.s = o.d into CAST(es.s AS Date32) = o.d, and both 2024-01-01 strings cast to the same date. DataFusion returns (2024-01-01, 1) twice and (2024-01-02, 1). DuckDB and the equivalent SELECT o.d, count(es.s) FROM o LEFT JOIN es ON es.s = o.d GROUP BY o.d in DataFusion both return (2024-01-01, 2), (2024-01-02, 1).

A plain col = <outer expr> and a widening cast such as CAST(int_col AS BIGINT) = <outer expr> give correct results.

Expected behavior

The DuckDB results above. The subquery sees both 2024-01-01 rows of e for the outer row 2024-01-01, so count(*) is 2 and there is one result row per outer row.

Additional context

The decorrelated plan groups by e.ts instead of CAST(e.ts AS DATE):

Sort: o.d ASC NULLS LAST
  Projection: o.d, CASE WHEN __scalar_sq_1.__always_true IS NULL THEN Int64(0) ELSE __scalar_sq_1.count(*) END AS c
    Left Join: o.d = CAST(__scalar_sq_1.ts AS Date32)
      TableScan: o projection=[d]
      SubqueryAlias: __scalar_sq_1
        Projection: count(Int64(1)) AS count(*), e.ts, Boolean(true) AS __always_true
          Aggregate: groupBy=[[e.ts]], aggr=[[count(Int64(1))]]
            TableScan: e projection=[ts]

PullUpCorrelatedExpr in datafusion/optimizer/src/decorrelate.rs pulls a correlated filter above an aggregate when can_pullup_over_aggregation accepts every correlated conjunct. That function accepts col = <outer expr>, and also CAST(col AS T) = <outer expr>. The Aggregate arm then adds the plain col to the group keys, so the aggregate makes one group per col value. A lossy cast lets several col values match one outer row, and the join returns one row per group. DataFusion does not check that a scalar subquery returns at most one row (#16134), so the extra rows come out as results instead of an error.

The CAST arms date back to #6084, and #13091 moved them into PullUpCorrelatedExpr.

Possible fixes:

• Accept CAST(col AS T) only when the cast is lossless, such as an integer widening, which is what type coercion adds for int32_col = int64_outer. Keep the subquery correlated for any other cast.
• Or group by the cast expression instead of col, and make the pulled up filter read that group key.

Found on main at b2cfc23. #26013 (for #25897) rejects casts for the same reason when it partitions a window by the correlated column.