apache / apache/datafusion

Enable pushdown of (non-negative-cost) scalar_fn in aggregations into `TableScan` nodes

Open
#25,106 2 comments 0 reactions 0 assignees View on GitHub
enhancement
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.