apache / apache/datafusion

Bring ordering information for grouped aggregation

Open
#15,818 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?

```
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

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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.