Perf: Streaming Aggregation should not be chosen unconditionally when the input is sorted
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
## Describe the issue
When the input is ordered by the group-by key, DataFusion always selects a streaming (`InputOrderMode::Sorted`) aggregation instead of a hash aggregation, without any cost comparison.
This is a clear win if `target_partitions = 1`, but regression is observed when `target_partitions > 1`. When the sorted aggregation is chosen, after the `Partial` aggregate, results are hash-repartitioned by the group key, which scatters the key order, so a `SortExec` must be inserted to restore order before the `FinalPartitioned(Sorted)` aggregate. That re-sort grows with cardinality and negates the streaming benefit, while a hash aggregation just does two hash passes with no sort.
## Benchmark
We benchmarked hash vs sorted aggregation on **identical, already-sorted** input (`SELECT g, sum(v) FROM t GROUP BY g`), sweeping group-key cardinality and parallel degree. Same batches feed both variants; the only difference is whether the source
advertises a sort order on the group key (→ sorted/streaming) or not (→ hash).
**Single partition (`target_partitions = 1`)** — sorted is never slower, up to ~2.8x faster at high cardinality:
| cardinality | hash | sorted |
|---|---|---|
| 512 | 51.9 ms | 52.4 ms (≈) |
| 32,768 | 59.6 ms | 55.2 ms (1.08x faster) |
| 262,144 | 82.7 ms | 53.4 ms (1.55x faster) |
| 2,097,152 | 275.1 ms | 98.7 ms (2.79x faster) |
**Parallel (`target_partitions = 8`)** — sorted is consistently slower:
| cardinality | hash | sorted |
|---|---|---|
| 512 | 18.5 ms | 24.0 ms (1.30x slower) |
| 4,096 | 16.9 ms | 29.6 ms (1.76x slower) |
| 32,768 | 17.9 ms | 32.4 ms (1.81x slower) |
| 262,144 | 39.8 ms | 59.1 ms (1.48x slower) |
| 2,097,152 | 49.1 ms | 109.6 ms (2.23x slower) |
## Expected behavior
The planner should compare sorted vs hash aggregation for the partitioned/final stage (accounting for the mandatory post-repartition `SortExec` and estimated group cardinality), rather than always preferring sorted whenever the input happens to be ordered by the group key.
Contributor guide
Research direction
Start by tracing the planner path that chooses InputOrderMode::Sorted for aggregation, especially the Partial and FinalPartitioned(Sorted) stages and the inserted SortExec after repartitioning. Reproduce the supplied sorted-input benchmark with target_partitions set to 1 and 8; done means the partitioned/final choice accounts for the sort cost and group cardinality instead of unconditionally preferring streaming aggregation.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust, sql
- Domain
- databases, performance
- Issue type
- Bug
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100