Common subexpression elimination evaluates the arguments of an aggregate with FILTER for rows that the filter rejects #26183

Describe the bug

An aggregate with a FILTER clause evaluates its arguments only for the rows that pass the filter.
Thus a fallible argument, such as a cast or a division, does not fail on a rejected row.

This stops working when two aggregates in the same query have the same argument expression.
The CommonSubexprEliminate optimizer rule moves the shared argument into a projection below the aggregate.
The projection evaluates the expression for every row, and that includes the rows that the filter rejects.
The query then fails with an error from a row that must not take part in the aggregate.

To Reproduce

In datafusion-cli:

CREATE TABLE t AS
SELECT * FROM (VALUES (CAST(-950 AS SMALLINT), 0), (CAST(5 AS SMALLINT), 1)) AS v(x, y);

-- One aggregate: succeeds and returns 5.0.
SELECT avg(arrow_cast(x, 'UInt64')) FILTER (WHERE y > 0) AS a FROM t;

-- Two aggregates with the same argument: fails.
SELECT
    sum(arrow_cast(x, 'UInt64')) FILTER (WHERE y > 0) AS s,
    count(arrow_cast(x, 'UInt64')) FILTER (WHERE y > 0) AS c
FROM t;

Actual result of the second query:

Failed to cast field 'x' from Int16 to UInt64
caused by
Arrow error: Cast error: Can't cast value -950 to type UInt64

EXPLAIN (with SET datafusion.explain.format = 'indent') shows the cast in a projection below the aggregate, as __common_expr_1:

logical_plan
Projection: sum(arrow_cast(t.x,Utf8("UInt64"))) FILTER (WHERE t.y > Int64(0)) AS s, count(arrow_cast(t.x,Utf8("UInt64"))) FILTER (WHERE t.y > Int64(0)) AS c
  Aggregate: groupBy=[[]], aggr=[[sum(__common_expr_1) FILTER (WHERE __common_expr_2) AS sum(arrow_cast(t.x,Utf8("UInt64"))) FILTER (WHERE t.y > Int64(0)), count(__common_expr_1) FILTER (WHERE __common_expr_2) AS count(arrow_cast(t.x,Utf8("UInt64"))) FILTER (WHERE t.y > Int64(0))]]
    Projection: CAST(t.x AS UInt64) AS __common_expr_1, t.y > Int64(0) AS __common_expr_2
      TableScan: t projection=[x, y]
physical_plan
ProjectionExec: expr=[sum(arrow_cast(t.x,Utf8("UInt64"))) FILTER (WHERE t.y > Int64(0))@0 as s, count(arrow_cast(t.x,Utf8("UInt64"))) FILTER (WHERE t.y > Int64(0))@1 as c]
  AggregateExec: mode=Single, gby=[], aggr=[sum(__common_expr_1) FILTER (WHERE __common_expr_2) as ..., count(__common_expr_1) FILTER (WHERE __common_expr_2) as ...]
    ProjectionExec: expr=[CAST(x@0 AS UInt64) as __common_expr_1, y@1 > 0 as __common_expr_2]
      DataSourceExec: partitions=1, partition_sizes=[1]

The same failure occurs with GROUP BY y, and with any other fallible argument, for example sum(10 / v) FILTER (WHERE v <> 0), count(10 / v) FILTER (WHERE v <> 0).

Expected behavior

The second query returns one row:

+---+---+
| s | c |
+---+---+
| 5 | 1 |
+---+---+

A subexpression that occurs only in the arguments of aggregates with a FILTER stays in the aggregates.

Additional context

Versions: datafusion-cli 54.0.0, and main at 334f293 (reports DataFusion CLI v55.1.0).

Suspected cause: CommonSubexprEliminate already does not extract a subexpression that is only conditionally evaluated, such as a CASE branch or the right side of AND / OR.
ExprCSEController::conditional_children in datafusion/optimizer/src/common_subexpr_eliminate.rs makes that decision, and it has no case for Expr::AggregateFunction with a filter.
The rule thus counts the arguments of a filtered aggregate as always evaluated.

The executor side of this contract is deliberate.
The ungrouped aggregate path filters the batch before it evaluates the arguments (aggregate_batch in datafusion/physical-plan/src/aggregates/aggregate_stream.rs).
The grouped path does the same since #24444 (issue #24443), with tests in aggregate_filter_selection.slt.
The optimizer rule undoes that guarantee.

Related: #24507 and #24508 cover the window function executor, which still evaluates the arguments before the FILTER.