Enable pushdown of (non-negative-cost) scalar_fn in aggregations into `TableScan` nodes
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
### Is your feature request related to a problem or challenge?
We want to allow projection pushdown of compute inside `aggr_expr` into lower nodes, ideally a table scan. Currently, this is done for `Projection` nodes, but not for aggregates with compute inside them.
Consider
```
Aggregate: groupBy=[[hits.CounterID]], aggr=[[avg(CAST(octet_length(hits.URL) AS Float64)), count(Int64(1))]]
SubqueryAlias: hits
Filter: hits_raw.URL != Utf8View("")
TableScan: hits_raw projection=[CounterID, URL], partial_filters=[hits_raw.URL != Utf8View("")]
```
### Describe the solution you'd like
The possible solution I would like is to extract each the scalar compute into a Projection.
Before:
```
Aggregate: groupBy=[[hits.CounterID]], aggr=[[avg(CAST(octet_length(hits.URL) AS Float64)), count(Int64(1))]]
SubqueryAlias: hits
Filter: hits_raw.URL != Utf8View("")
TableScan: hits_raw projection=[CounterID, URL], partial_filters=[hits_raw.URL != Utf8View("")]
```
After:
```
Aggregate: groupBy=[[hits.CounterID]], aggr=[[avg(__aggregate_arg_1) AS avg(octet_length(hits.URL)), count(Int64(1))]]
Projection: hits.CounterID, CAST(octet_length(hits.URL) AS Float64) AS __aggregate_arg_1
SubqueryAlias: hits
Filter: hits_raw.URL != Utf8View("")
TableScan: hits_raw projection=[CounterID, URL], partial_filters=[hits_raw.URL != Utf8View("")]
```
This is a cost based change, but as a start its always beneficial to do this if that expr is the only consumer of a column [keeping the scheme len non increasing].
Then existing `TableScan` can try and accept the projection expr. If the table scan doesn't accept this then there is not extra compute.
We cannot do this if the Agg contains a local `Filter`.
### Describe alternatives you've considered
1. Use placement API [doesn't work for costly compute], done in: https://github.com/apache/datafusion/pull/25025
2. Add a pass to try and push down Aggregate directly into a TableScan [likely duplicates code from the Projection Pass when we don't need to].
### Additional context
Us at vortex would like to compute complex projection pushdown over vortex-compressed data
Contributor guide
Research direction
Start by reading the existing Projection pushdown pass and the placement API referenced in the issue, then trace how Aggregate, Projection, and TableScan plans exchange expressions. Verify the shown aggregate shape, including the restriction on local filters, and check how a TableScan handles accepted projection expressions. Done means eligible scalar computations are extracted without increasing the projection scheme, while scans that reject them retain the existing behavior.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- data-engineering, performance
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100