Grouping, sorting, and running out of memory

GROUP BY and ORDER BY are the operators that have to see everything before they can say anything. This post covers how they work, how to make them stream where possible, what to do when the data doesn't fit in memory, and the cheap trick that makes ORDER BY … LIMIT 10 fast.

Joins (part 3) bring matching rows from two inputs together. GROUP BY does the same thing within one input: bring together all the rows with the same key, then boil each bunch down to one row. Not surprisingly, the two standard ways to do that are the same two ideas as before: hash the key, or sort by it.

Aggregate functions are little state machines

Before grouping anything, look at what an aggregate function actually needs. You can't compute avg(qty) by holding all the quantities and averaging at the end, at least not if a group has a billion rows. Instead, every aggregate is three operations over a small piece of state:

interface Aggregate
  init()             -> state
  step(state, value) -> state      -- fold in one input value
  final(state)       -> value      -- produce the answer

count(*):  init 0         step: s + 1                          final: s
count(x):  init 0         step: if x is null then s else s + 1 final: s
sum(x):    init null      step: if x is null then s else (s ?? 0) + x   final: s
min(x):    init null      step: if x is null then s else min(s ?? x, x) final: s
avg(x):    init (0, 0)    step: if x is null then s else (s.sum + x, s.n + 1)
                          final: if s.n == 0 then null else s.sum / s.n

A few details that matter:

• avg's state isn't the average. It's the pair (sum, count), and the division happens only in final. A running average can't absorb a new value without the count anyway, and keeping the pair is what makes partial aggregates mergeable. Two threads, or two machines, can each aggregate half the data, then add their pairs together.
• Aggregates skip NULLs, except count(*), which counts rows rather than values. sum over zero non-NULL values is NULL, not 0 (SQL's choice, and a frequent surprise), while count is 0.
• With no GROUP BY at all, the whole input is one group, and that group exists even when the input is empty: SELECT count(*) FROM empty_table returns one row, 0. Our aggregate operator handles that case explicitly.

Hash aggregation

Hash aggregation keeps a hash table from group key to aggregate state. For each input row, it finds (or creates) the row's group and steps every aggregate:

struct HashAggregate { child, keys, aggs, groups, order, pos }
  method Open()
    child.Open()
    groups = new HashMap(key -> states); order = []
    loop                                       -- drain the child: pipeline breaker
      (row, ok) = child.Next()
      if not ok then break
      k = [e(row) for e in keys]               -- NULL keys group together
      if k not in groups then
        groups[k] = [a.init() for a in aggs]
        append(order, k)                       -- remember first-seen order
      groups[k] = [a.step(s, a.arg(row)) for (a, s) in zip(aggs, groups[k])]
    if len(keys) == 0 and len(order) == 0 then  -- global aggregate, empty input
      groups[[]] = [a.init() for a in aggs]; append(order, [])
    pos = 0
  method Next()
    if pos >= len(order) then return (nil, false)
    k = order[pos]; pos = pos + 1
    return (k ++ [a.final(s) for (a, s) in zip(aggs, groups[k])], true)

Note the line NULL keys group together. In a join, NULL never equals NULL, so NULL keys never match. In GROUP BY, all the NULLs form one group. SQL is inconsistent here on purpose, and the hash table has to follow suit: the join's build side skips NULL keys while the aggregate treats NULL as an ordinary key.

Step through it, then switch to sort aggregation:

Hash aggregation reads its input once in any order, and its memory is proportional to the number of groups, not the number of rows. Grouping a billion orders by country holds about two hundred accumulators. Grouping them by order id holds a billion. That's the case where it hurts.

Sort aggregation

If the input arrives sorted by the group key, every group's rows are adjacent, and grouping needs no hash table at all. Keep one group's state; when the key changes, that group is finished forever, so emit it and start the next one:

struct SortAggregate { child, keys, aggs, cur, states, pending }
  method Next()
    loop
      (row, ok) = child.Next()
      if not ok then
        if cur is null then return (nil, false)
        out = cur ++ finals(states); cur = null
        return (out, true)                     -- the last group
      k = [e(row) for e in keys]
      if cur is not null and k != cur then
        out = cur ++ finals(states)            -- group boundary: emit the finished group
        cur = k; states = inits(); step(states, row)
        return (out, true)
      if cur is null then cur = k; states = inits()
      step(states, row)

This is a streaming operator, not a pipeline breaker: it emits each group as soon as the group ends, holds one group's state, and a LIMIT above it stops it early. The catch is the precondition. If the input isn't already sorted, you have to sort it first, and sorting is a pipeline breaker. So sort aggregation shines when the order is free: rows coming out of an index on the group key, out of a merge join on the same key, or out of a sort the query needed anyway.

Sorting

ORDER BY in memory is the Sort operator from part 2: drain the child into a buffer, sort it, hand out rows. The interesting part is the comparator, which has to implement SQL ordering exactly:

function compareRows(a, b, keys)              -- keys: [(expr, desc)]
  for (e, desc) in keys
    x = e(a); y = e(b)
    if x is null and y is null then continue
    if x is null then return +1                -- NULLs last, ASC and DESC alike
    if y is null then return -1
    c = compareValues(x, y)
    if c != 0 then return if desc then -c else c
  return 0                                     -- ties keep input order (stable sort)

NULL placement is a per-engine convention: Postgres puts NULLs last in ascending order and first in descending, and offers NULLS FIRST/LAST to override. We chose "always last" for simplicity. The sort is stable, so rows that compare equal keep their input order. That's not required by SQL, but it makes results reproducible, which our tests rely on.

When it doesn't fit: external merge sort

Sorting a hundred gigabytes with ten gigabytes of memory is old, solved, and elegant. It happens in two phases:

1. Run generation. Read as many rows as fit in memory, sort them, write them to a temporary file as a run. Repeat until the input is exhausted. Ten passes of 10 GB gives ten sorted runs.
2. Merge. Open every run and read them in parallel, keeping just the current head of each in a min-heap. Repeatedly pop the smallest head, output it, and refill from that head's run. This is the merge step of merge sort, but k-way.

Two numbers make this practical. First, every row is read and written exactly twice, once per phase, no matter how big the data is relative to memory, as long as the runs can all be merged at once. Second, merging k runs needs only one buffer per run. With 10 GB of memory and 1 MB buffers you can merge 10,000 runs in one pass, so one merge pass covers about 10 GB × 10,000 = 100 TB. Real external sorts almost never need a second merge pass. (In the figure, memory is tiny, so with small M the note tells you how many passes a real implementation would need.)

function externalSort(input, M)
  runs = []
  loop                                          -- phase 1
    chunk = take(input, M)
    if empty(chunk) then break
    append(runs, writeTemp(sort(chunk)))
  heap = minHeap([(first(r), r) for r in runs]) -- phase 2
  while not empty(heap)
    (v, r) = pop(heap)
    emit(v)
    if hasNext(r) then push(heap, (next(r), r))

The same idea rescues the other pipeline breakers. A hash aggregate or hash join whose table won't fit partitions its input to disk by hash of the key, so each partition fits, then processes one partition at a time. It's the hashing version of "sort the pieces, then merge."

Top-N: ORDER BY … LIMIT

"The ten most expensive products" is ORDER BY price DESC LIMIT 10. A literal execution sorts every row and then throws away all but ten. A smarter one notices that it only ever needs to know the ten best rows so far, and keeps them in a heap:

struct TopN { child, n, keys, heap }           -- heap root = worst of the current best n
  method Open()
    loop
      (row, ok) = child.Next()
      if not ok then break
      if len(heap) < n then push(heap, row)
      elif better(row, root(heap)) then replaceRoot(heap, row)   -- evict the worst
      -- else: can't make the top n; one comparison and it's gone
    result = sortDescending(heap)

The win comes on both axes. Memory drops from all rows to n rows. Time drops from N log N to N log n, and in practice it's lower still: once the heap holds good values, most incoming rows lose to the root after a single comparison. For LIMIT 10 over a billion rows, that's the difference between an external sort spilling gigabytes to disk and a ten-element array in cache. Postgres does exactly this ("top-N heapsort" in EXPLAIN ANALYZE). Recognizing Sort under Limit and fusing them into a TopN is a classic planner rewrite.

HAVING, and the shape of a plan

HAVING needs no new machinery. It's a Filter placed above the aggregate, evaluated on the aggregated rows. WHERE filters rows before grouping; HAVING filters groups after. A complete single-table query now looks like this:

Limit(10)
  Sort(revenue DESC)                     -- or TopN(10, revenue DESC)
    Project(city, revenue)
      Filter(count(*) > 2)               -- HAVING
        HashAggregate(keys: city; aggs: count(*), sum(qty * price))
          Filter(age < 30)               -- WHERE
            Scan(customers)

Count the pipelines. Scan → Filter → (aggregate's input) is one: rows stream from the table into the hash table. Then the aggregate's output → Filter → Project → (sort's input) is a second, and the sort's output → Limit is a third. Every pipeline breaker ends one pipeline and starts another. That picture, pipelines joined at breakers, is exactly the structure part 7's compiler will produce.