OLAP – Phase 6 Hash Aggregation
The scan/filter/project pipeline from Phase 5 processes rows one batch at a time — it’s a streaming operation where each batch flows through independently. Aggregation (GROUP BY with SUM, COUNT, AVG) is fundamentally different: it needs to see all the data before producing any output. This makes it a pipeline breaker — it consumes all input, builds an internal state, then emits results.
This phase implements GROUP BY aggregation using a hash table: the HashAggregateOperator, the GroupByHashTable, and five aggregate functions (COUNT, SUM, AVG, MIN, MAX).
In DuckDB, these are PhysicalHashAggregate (src/execution/operator/aggregate/physical_hash_aggregate.cpp) and GroupedAggregateHashTable (src/execution/aggregate_hashtable.cpp).
Here is the roadmap for the phases to come:
- Phase 6: Hash aggregation
- Phase 7: Hash join
- Phase 8: SQL parser
- Phase 9: Query planner and optimizer
- Phase 10: Sorting, parallel execution, REPL, and server
Full Source Code
The code referenced in this post can be found in https://gitlab.com/kimserey.lam/olap-learn.
Aggregate Functions
Each aggregate function follows a three-phase lifecycle — Initialize, Update (called once per input row), and Finalize (called once to produce the result):
1
2
3
4
5
6
7
8
9
type AggregateState interface {
Update(value any, isNull bool)
Finalize() (any, bool)
}
type AggregateFunction interface {
NewState() AggregateState
ResultType() vector.LogicalType
}
COUNT(*)
Counts all rows, including NULLs:
1
2
3
4
type countStarState struct{ count int64 }
func (s *countStarState) Update(_ any, _ bool) { s.count++ }
func (s *countStarState) Finalize() (any, bool) { return s.count, true }
COUNT(column)
Counts only non-null values:
1
2
3
4
5
6
7
type countState struct{ count int64 }
func (s *countState) Update(_ any, isNull bool) {
if !isNull {
s.count++
}
}
SUM
Accumulates numeric values, returning NULL if no non-null values were seen:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
type sumState struct {
sum float64
hasValue bool
}
func (s *sumState) Update(value any, isNull bool) {
if isNull {
return
}
s.sum += toFloat(value)
s.hasValue = true
}
func (s *sumState) Finalize() (any, bool) {
if !s.hasValue {
return nil, false
}
return s.sum, true
}
AVG
Tracks both sum and count, dividing at finalization:
1
2
3
4
5
6
7
8
9
10
11
type avgState struct {
sum float64
count int64
}
func (s *avgState) Finalize() (any, bool) {
if s.count == 0 {
return nil, false
}
return s.sum / float64(s.count), true
}
MIN / MAX
Track the running minimum or maximum:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
type minMaxState struct {
val any
hasValue bool
isMin bool
}
func (s *minMaxState) Update(value any, isNull bool) {
if isNull {
return
}
if !s.hasValue {
s.val = value
s.hasValue = true
return
}
if s.isMin {
if anyLess(value, s.val) {
s.val = value
}
} else {
if anyLess(s.val, value) {
s.val = value
}
}
}
GroupByHashTable
The hash table maps group keys to aggregate states using open addressing with linear probing:
1
2
3
4
5
6
7
8
9
10
11
12
13
type GroupByHashTable struct {
entries []htEntry
capacity int
count int
funcs []AggregateFunction
}
type htEntry struct {
occupied bool
hash uint64
keys []any
states []AggregateState
}
FindOrCreate
For each row, FindOrCreate looks up the group keys. If the group exists, it returns the existing aggregate states. If not, it creates a new entry with fresh states:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
func (ht *GroupByHashTable) FindOrCreate(keys []any) []AggregateState {
h := hashKeys(keys)
idx := int(h % uint64(ht.capacity))
for {
e := &ht.entries[idx]
if !e.occupied {
// new group
e.occupied = true
e.hash = h
e.keys = copyKeys(keys)
e.states = make([]AggregateState, len(ht.funcs))
for i, f := range ht.funcs {
e.states[i] = f.NewState()
}
ht.count++
if ht.count*2 > ht.capacity {
ht.grow()
}
return e.states
}
if e.hash == h && keysEqual(e.keys, keys) {
return e.states // existing group
}
idx = (idx + 1) % ht.capacity // linear probing
}
}
The cached hash value (e.hash) avoids full key comparison on ~99% of probes — a hash mismatch means the keys can’t be equal.
HashAggregateOperator
The operator has two phases:
- Sink phase: consume all input from the child operator, update hash table entries
- Source phase: scan the hash table, emit result DataChunks
1
2
3
4
5
6
7
8
func (op *HashAggregateOperator) Next() *vector.DataChunk {
if !op.done {
op.sink() // consume all input
op.done = true
op.collectResults()
}
return op.emitNextChunk()
}
Processing Input Chunks
For each input chunk, evaluate the group-by expressions and aggregate input expressions, then update the hash table:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
func (op *HashAggregateOperator) processChunk(chunk *vector.DataChunk) {
// evaluate group-by expressions → groupVecs
// evaluate aggregate inputs → aggVecs
for row := 0; row < count; row++ {
// extract keys for this row
for k, gv := range groupVecs {
gi := vecIdx(gv, row)
if !gv.IsValid(gi) {
keys[k] = nil
} else {
keys[k] = readValue(gv, gi)
}
}
states := op.ht.FindOrCreate(keys)
// update each aggregate with this row's value
for a, av := range aggVecs {
if av == nil {
states[a].Update(nil, false) // COUNT(*)
} else {
ai := vecIdx(av, row)
if !av.IsValid(ai) {
states[a].Update(nil, true)
} else {
states[a].Update(readValue(av, ai), false)
}
}
}
}
}
Ungrouped Aggregation
A query like SELECT SUM(amount) FROM orders (no GROUP BY) has one implicit group containing all rows. If no rows match the filter, the hash table is empty, so we create a single entry with fresh (unfed) states — this correctly returns SUM = NULL:
1
2
3
4
5
6
7
8
9
10
11
12
13
func (op *HashAggregateOperator) collectResults() {
if len(op.groupByExprs) == 0 && op.ht.Count() == 0 {
states := make([]AggregateState, len(op.aggregates))
for i, agg := range op.aggregates {
states[i] = agg.Function.NewState()
}
op.results = append(op.results, htResult{keys: nil, states: states})
return
}
op.ht.Scan(func(keys []any, states []AggregateState) {
op.results = append(op.results, htResult{keys: keys, states: states})
})
}
Example
1
SELECT region, SUM(amount), COUNT(*) FROM orders GROUP BY region;
Input:
1
2
3
4
5
6
region | amount
"west" | 100
"east" | 200
"west" | 150
"east" | 300
"west" | 50
After processing all rows, the hash table contains:
| Group Key | SUM state | COUNT state |
|---|---|---|
| “west” | sum=300 | count=3 |
| “east” | sum=500 | count=2 |
Output after finalization:
1
2
3
region | SUM(amount) | COUNT(*)
"west" | 300.00 | 3
"east" | 500.00 | 2
Summary
Hash aggregation is the first pipeline breaker — it must see all input before producing any output. The hash table maps group keys to aggregate states using open addressing with linear probing. Each aggregate function follows an Initialize/Update/Finalize lifecycle that handles NULL propagation correctly. The operator is the bridge between the streaming scan/filter pipeline and the next phase: hash join.
