Filter on a volatile GROUP BY key returns rows that do not satisfy the filter

[adriangb]

Describe the bug

A filter on a grouping key that contains a volatile function (for example random()) gives incorrect results. The query returns rows that do not satisfy the WHERE clause.

The optimizer moves the filter below the aggregate. It then replaces the key column with the key expression. Thus DataFusion calculates random() two times for each row: one time in the filter and one time in the GROUP BY. The two values are not the same, so the filter and the groups do not agree.

To Reproduce

Use datafusion-cli:

CREATE TABLE v AS SELECT value AS a FROM generate_series(1, 10000);

SELECT k, c
FROM (SELECT random() < 0.5 AS k, count(*) AS c FROM v GROUP BY random() < 0.5)
WHERE k;

Actual result (DataFusion 54.0.0 and main at 4e90755):

+-------+------+
| k     | c    |
+-------+------+
| false | 2447 |
| true  | 2474 |
+-------+------+

The query returns a row with k = false, but the filter is WHERE k. Also, the total count is approximately 5000, not 10000, because the filter removes approximately half of the rows before the aggregate.

The same problem occurs with GROUP BY k and with WHERE k = false.

EXPLAIN shows that the filter is below the aggregate:

ProjectionExec: c, k
  AggregateExec: mode=FinalPartitioned, group_by=[random() < 0.5]
    RepartitionExec: Hash(...)
      AggregateExec: mode=Partial, group_by=[random() < 0.5]
        FilterExec: random() < 0.5          <-- second calculation of random()
          RepartitionExec: RoundRobinBatch
            DataSourceExec

Expected behavior

The query must return only one row, with k = true. The count must be approximately 5000.

PostgreSQL 17.11 gives the expected result. It keeps the filter above the aggregate:

CREATE TABLE v AS SELECT value AS a FROM generate_series(1, 10000) AS value;
SELECT k, c FROM (SELECT random() < 0.5 AS k, count(*) AS c FROM v GROUP BY random() < 0.5) AS s WHERE k;

 k |  c
---+------
 t | 4949
(1 row)

 Subquery Scan on s
   Filter: s.k
   ->  HashAggregate
         Group Key: (random() < '0.5'::double precision)
         ->  Seq Scan on v

Note: DuckDB 1.5.2 has the same bug. It returns both false and true rows for this query, and its plan also has FILTER (random() < 0.5) below HASH_GROUP_BY. Thus DuckDB is not a correct reference for this query.

Additional context

The cause is in PushDownFilter, in the LogicalPlan::Aggregate branch (push_down_filter.rs). This branch pushes a predicate below the aggregate when all the columns of the predicate are grouping columns. It does not examine whether the grouping expression is volatile. The LogicalPlan::Projection branch (rewrite_projection) already keeps predicates on volatile expressions above the projection. The Aggregate branch must do the same.

Found while investigating #25329 and #25388.