Bring ordering information for grouped aggregation
- 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?
```
statement ok
create table t(a int, b int) as values (1, 2), (2, 3);
query I
select sum(a) from (select a from t order by a) group by a;
----
1
2
query TT
explain select sum(a) from (select a from t order by a) group by a;
----
logical_plan
01)Projection: sum(t.a)
02)--Aggregate: groupBy=[[t.a]], aggr=[[sum(CAST(t.a AS Int64))]]
03)----TableScan: t projection=[a]
physical_plan
01)ProjectionExec: expr=[sum(t.a)@1 as sum(t.a)]
02)--AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[sum(t.a)]
03)----CoalesceBatchesExec: target_batch_size=8192
04)------RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=4
05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
06)----------AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[sum(t.a)]
07)------------DataSourceExec: partitions=1, partition_sizes=[1]
```
In this query, we know the `a` is ordered, when we do the grouped aggregation, we can run with `GroupOrdering::Full` mode. However, this is not the case now.
This is how we get Ordered Group aggregation
```
statement count 0
CREATE EXTERNAL TABLE t (
a INT NOT NULL,
)
STORED AS CSV
WITH ORDER (a ASC)
LOCATION '../core/tests/data/aggregate_agg_multi_order.csv'
OPTIONS ('format.has_header' 'true');
query TT
explain
select sum(a) from t group by a;
----
logical_plan
01)Projection: sum(t.a)
02)--Aggregate: groupBy=[[t.a]], aggr=[[sum(CAST(t.a AS Int64))]]
03)----TableScan: t projection=[a]
physical_plan
01)ProjectionExec: expr=[sum(t.a)@1 as sum(t.a)]
02)--AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[sum(t.a)], ordering_mode=Sorted
03)----SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
04)------CoalesceBatchesExec: target_batch_size=8192
05)--------RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=4
06)----------AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[sum(t.a)], ordering_mode=Sorted
07)------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
08)--------------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/aggregate_agg_multi_order.csv]]}, projection=[a], output_ordering=[a@0 ASC NULLS LAST], file_type=csv, has_header=true
```
We can indicate the ordering with csv file. Then we could get
```
06)----------AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[sum(t.a)], ordering_mode=Sorted
```
It runs with group ordering mode.
The goal is that this query in memory table should be equivalent to the 2nd example in csv format.
```
select sum(a) from (select a from t order by a) group by a;
```
### Describe the solution you'd like
Find a way to keep ordering information and then leverage on it for optimization.
### Describe alternatives you've considered
_No response_
### Additional context
_No response_
Contributor guide
Research direction
Start by running the provided EXPLAIN cases and compare the in-memory ordered subquery with the CSV table that exposes output_ordering. Trace how ordering reaches AggregateExec and how GroupOrdering::Full or ordering_mode=Sorted is selected. Done means the in-memory query preserves enough ordering information to use the grouped aggregation optimization and matches the ordered CSV example.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- databases
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100