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.
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 whentarget_partitions > 1. When the sorted aggregation is chosen, after thePartialaggregate, results are hash-repartitioned by the group key, which scatters the key order, so aSortExecmust be inserted to restore order before theFinalPartitioned(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 sourceadvertises 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:Parallel (
target_partitions = 8) — sorted is consistently slower:Expected behavior
The planner should compare sorted vs hash aggregation for the partitioned/final stage (accounting for the mandatory post-repartition
SortExecand estimated group cardinality), rather than always preferring sorted whenever the input happens to be ordered by the group key.