Aggregates with FILTER (WHERE ...) copy every input column once per aggregate #27114

Is your feature request related to a problem or challenge?

An aggregate with a FILTER clause evaluates its arguments only on the rows that the filter selects, so that SUM(1 / x) FILTER (WHERE x <> 0) cannot fail on the rejected rows (#24443). The selected rows are produced by filtering the whole input batch of the aggregate, once per filtered aggregate:

• ungrouped AggregateStream: batch_filter(batch, filter)
• grouped hash aggregation (since perf: compact inputs for grouped aggregate filter #24869): filter_record_batch(batch, selection) in evaluate_compacted_args
• skip-partial state conversion: argument.evaluate_selection(batch, selection), whose default implementation filters the whole batch once per argument

The input batch of an aggregate holds the group keys and every column read by any of its aggregates, while one aggregate usually reads one or two of them. The copying therefore grows with (number of filtered aggregates) × (number of input columns), and queries with many filtered aggregates spend most of their aggregation time on it. Typical shapes are pivot-style reports and scalar subqueries merged into one aggregate (TPC-DS q9, as Spark's MergeScalarSubqueries does).

Reproducer with datafusion-cli built from main (70f2ed0):

-- 20M rows
COPY (
  SELECT
    value % 100 + 1 AS quantity,
    arrow_cast((value % 9973) / 100.0, 'Decimal128(7, 2)') AS ext_tax,
    arrow_cast((value % 99991) / 10.0, 'Decimal128(7, 2)') AS net_paid
  FROM generate_series(0, 19999999)
) TO 'sales.parquet' STORED AS PARQUET;

CREATE EXTERNAL TABLE sales STORED AS PARQUET LOCATION 'sales.parquet';

-- TPC-DS q9 shape: 15 filtered aggregates over 3 columns
SELECT
  count(*) FILTER (WHERE quantity BETWEEN 1 AND 20) AS c1,
  avg(ext_tax) FILTER (WHERE quantity BETWEEN 1 AND 20) AS t1,
  avg(net_paid) FILTER (WHERE quantity BETWEEN 1 AND 20) AS p1,
  count(*) FILTER (WHERE quantity BETWEEN 21 AND 40) AS c2,
  avg(ext_tax) FILTER (WHERE quantity BETWEEN 21 AND 40) AS t2,
  avg(net_paid) FILTER (WHERE quantity BETWEEN 21 AND 40) AS p2,
  count(*) FILTER (WHERE quantity BETWEEN 41 AND 60) AS c3,
  avg(ext_tax) FILTER (WHERE quantity BETWEEN 41 AND 60) AS t3,
  avg(net_paid) FILTER (WHERE quantity BETWEEN 41 AND 60) AS p3,
  count(*) FILTER (WHERE quantity BETWEEN 61 AND 80) AS c4,
  avg(ext_tax) FILTER (WHERE quantity BETWEEN 61 AND 80) AS t4,
  avg(net_paid) FILTER (WHERE quantity BETWEEN 61 AND 80) AS p4,
  count(*) FILTER (WHERE quantity BETWEEN 81 AND 100) AS c5,
  avg(ext_tax) FILTER (WHERE quantity BETWEEN 81 AND 100) AS t5,
  avg(net_paid) FILTER (WHERE quantity BETWEEN 81 AND 100) AS p5
FROM sales;

Apple M4 Pro (12 cores), median of 21 warm runs. The second column is the same query with each agg(x) FILTER (WHERE c) written as agg(CASE WHEN c THEN x END) (and count(*) as count(CASE WHEN c THEN 1 END)), which gives identical results:

EXPLAIN ANALYZE shows where the time goes. In the mode=Partial AggregateExec of the query above, the agg_expr_*_arguments_time metrics add up to 207–236 ms of its 219–255 ms elapsed_compute.

Describe the solution you'd like

Filter only the columns that the aggregate's arguments read, including the ORDER BY expressions of ordered aggregates such as first_value(x ORDER BY y). The other columns can be replaced by data-free NullArray placeholders, so column indices in the argument expressions stay valid, and arguments are still evaluated only on the selected rows. The skip-partial path can use the same filtered batch, once per aggregate instead of once per argument.

With a local prototype of this, the results are identical:

On TPC-DS SF1, q9 with its 15 scalar subqueries merged by hand into one aggregate like the query above went from 93 ms to 25 ms (measured on an earlier main, 290ad15).

When grouping dominates, the copy is a small share of the work: with 1.7M groups from 20M rows, the prototype made no measurable difference.

Describe alternatives you've considered

• Evaluating arguments on all rows and passing the filter to the accumulator, as GroupedHashAggregateStream (enable_migration_aggregate = false) still does. It copies nothing but brings back Grouped aggregate FILTER evaluates arguments for rejected rows #24443.
• Rewriting FILTER into CASE WHEN in the optimizer. That is only equivalent for aggregates that ignore NULLs (not count(*), first_value, array_agg, ...), and it is still slower than filtering only the needed columns.

Additional context

cc @lyne7-sc, who wrote #24444 and #24869.

Willingness to contribute

I plan to contribute to this feature independently.