xudong963 opened a new issue, #27114: URL: https://github.com/apache/datafusion/issues/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)`](https://github.com/apache/datafusion/blob/70f2ed01b8392127fa0eb4f1fb5e57869c42ecfa/datafusion/physical-plan/src/aggregates/aggregate_stream.rs#L494) - grouped hash aggregation (since #24869): [`filter_record_batch(batch, selection)`](https://github.com/apache/datafusion/blob/70f2ed01b8392127fa0eb4f1fb5e57869c42ecfa/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs#L753) in `evaluate_compacted_args` - skip-partial state conversion: [`argument.evaluate_selection(batch, selection)`](https://github.com/apache/datafusion/blob/70f2ed01b8392127fa0eb4f1fb5e57869c42ecfa/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs#L798), whose default implementation [filters the whole batch](https://github.com/apache/datafusion/blob/70f2ed01b8392127fa0eb4f1fb5e57869c42ecfa/datafusion/physical-expr-common/src/physical_expr.rs#L143) 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 (70f2ed01b): ```sql -- 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: | Query | `FILTER` | `CASE WHEN` | |---|---|---| | query above | 50 ms | 38 ms | | same, `GROUP BY quantity % 8` | 68 ms | 49 ms | `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: | Query | main | prototype | |---|---|---| | query above | 50 ms | **28 ms** (1.8x) | | same, `GROUP BY quantity % 8` | 68 ms | **43 ms** (1.6x) | | `arguments_time` sum, query above | 207–236 ms | 37–39 ms | | `CASE WHEN` variants (no `FILTER`) | 38 / 49 ms | 38 / 49 ms | 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, 290ad15cc). 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 #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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
