apache / apache/datafusion

Perf: Streaming Aggregation should not be chosen unconditionally when the input is sorted

Open
#23,338 3 comments 1 reaction 0 assignees View on GitHub
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

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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.