How databases sort, join, and group data that's too big for memory

1 External merge sort — sorting data bigger than RAM

Sorting sounds simple, but a database often must sort a table that's far larger than the memory it's allowed to use. A laptop algorithm like quicksort assumes the whole array fits in RAM. Databases can't assume that. The solution is external merge sort — "external" meaning it uses external storage (disk) as scratch space.

The unit of work: pages and buffers

Recall from earlier sessions that a database reads and writes the disk in fixed-size chunks called pages (also called blocks). The database is given a chunk of memory measured in pages — we'll call the number of memory pages available B (for "buffers"). Let N be the number of pages in the table we want to sort. The whole game is: sort all N pages while only ever holding B of them in memory at once.

Phase 1 — Create sorted "runs"

A run is a chunk of data that is internally sorted. In the first pass we read the table B pages at a time, sort those pages in memory (any in-memory sort works), and write the result back to disk as one sorted run. Each pass over a group of pages produces one run of length B. After reading the whole table we have produced ⌈N / B⌉ sorted runs.

Phase 2 — Merge runs together

Now we repeatedly merge runs into bigger runs. With B memory pages we can merge B − 1 runs at once (each input run needs one page to read from, and we keep one page as the output buffer). This is called a (B−1)-way merge. We look at the smallest unread record across all input pages, copy it to the output, refill pages as they empty, and write out output pages as they fill. Each merge pass reduces the number of runs by a factor of B − 1 until only one sorted run remains: the answer.

The I/O cost formula

Databases measure sort cost in page I/Os (disk reads + writes), because disk is the bottleneck — not CPU. Every pass reads all N pages and writes all N pages, so one pass costs 2N I/Os. The number of passes is: 1 pass to make the initial runs, then merge passes that each shrink the run count by a factor of B − 1.

The logarithm is what makes this practical: because each merge pass merges B − 1 runs at once and B is often large (hundreds or thousands of pages), the number of passes is tiny even for huge tables. In many real systems two or three passes sort terabytes.

Pseudocode

// External merge sort: N pages of input, B memory pages
function externalSort(table, B):
    // --- Phase 1: build initial sorted runs ---
    runs = []
    while not table.atEnd():
        chunk = table.read(B)          // up to B pages
        inMemorySort(chunk)            // e.g. quicksort
        runs.append(writeToDisk(chunk))

    // --- Phase 2: merge until one run remains ---
    while runs.length > 1:
        nextRuns = []
        for group in chunksOf(runs, B - 1):   // (B-1)-way merge
            merged = mergeRuns(group)              // heap of run heads
            nextRuns.append(merged)
        runs = nextRuns
    return runs[0]

The heart of Phase 2: a k-way merge with a min-heap

The merge step needs to repeatedly find the smallest unread record across all k input runs. Scanning every run head each time would cost O(k) per record; a min-heap (priority queue) keyed on the run heads makes it O(log k). Below is the in-memory k-way merge of already-sorted runs — the core inner loop a real engine runs on each group of B − 1 runs. (When data exceeds memory, those input runs are produced by Phase 1, sorting B pages at a time and spilling each sorted chunk to disk; here we keep them in lists to show the merge itself.)

# k-way merge of already-sorted runs using a min-heap.
# Each heap entry is (value, run_index, pos_within_run).
import heapq

def kway_merge(runs):
    # Each run is a list sorted ascending. Returns one merged sorted list.
    heap = []
    for i, run in enumerate(runs):     # seed heap with each run's head
        if run:
            heapq.heappush(heap, (run[0], i, 0))
    out = []
    while heap:
        val, i, pos = heapq.heappop(heap)   # smallest head across all runs
        out.append(val)
        if pos + 1 < len(runs[i]):         # refill from the SAME run
            heapq.heappush(heap, (runs[i][pos + 1], i, pos + 1))
    return out

if __name__ == "__main__":
    runs = [[1, 4, 7], [2, 3, 9], [5, 6, 8]]   # 3 sorted runs
    print(kway_merge(runs))                  # [1, 2, 3, 4, 5, 6, 7, 8, 9]

// k-way merge of already-sorted runs using a min-heap.
// Each heap entry is {value, runIndex, posWithinRun}.
#include <iostream>
#include <vector>
#include <queue>

struct Item {
    int value, run, pos;
    bool operator>(const Item& o) const { return value > o.value; } // min-heap
};

std::vector<int> kwayMerge(const std::vector<std::vector<int>>& runs) {
    // greater<> turns the default max-heap into a min-heap
    std::priority_queue<Item, std::vector<Item>, std::greater<Item>> heap;
    for (int i = 0; i < (int)runs.size(); ++i)   // seed with each run's head
        if (!runs[i].empty()) heap.push({runs[i][0], i, 0});

    std::vector<int> out;
    while (!heap.empty()) {
        Item t = heap.top(); heap.pop();        // smallest head across runs
        out.push_back(t.value);
        if (t.pos + 1 < (int)runs[t.run].size())  // refill from SAME run
            heap.push({runs[t.run][t.pos + 1], t.run, t.pos + 1});
    }
    return out;
}

int main() {
    std::vector<std::vector<int>> runs = {{1,4,7},{2,3,9},{5,6,8}};
    for (int v : kwayMerge(runs)) std::cout << v << ' ';
    std::cout << "\n";                          // 1 2 3 4 5 6 7 8 9
}

// k-way merge of already-sorted runs using a min-heap.
// Each heap entry is (value, runIndex, posWithinRun).
import java.util.*;

public class KWayMerge {
    record Item(int value, int run, int pos) {}

    static List<Integer> kwayMerge(List<List<Integer>> runs) {
        // PriorityQueue is a min-heap; order by value
        PriorityQueue<Item> heap = new PriorityQueue<>(Comparator.comparingInt(Item::value));
        for (int i = 0; i < runs.size(); i++)      // seed with each run's head
            if (!runs.get(i).isEmpty())
                heap.add(new Item(runs.get(i).get(0), i, 0));

        List<Integer> out = new ArrayList<>();
        while (!heap.isEmpty()) {
            Item t = heap.poll();                  // smallest head across runs
            out.add(t.value());
            if (t.pos() + 1 < runs.get(t.run()).size())  // refill from SAME run
                heap.add(new Item(runs.get(t.run()).get(t.pos() + 1), t.run(), t.pos() + 1));
        }
        return out;
    }

    public static void main(String[] args) {
        List<List<Integer>> runs = List.of(List.of(1,4,7),
                                          List.of(2,3,9),
                                          List.of(5,6,8));
        System.out.println(kwayMerge(runs));       // [1, 2, 3, 4, 5, 6, 7, 8, 9]
    }
}

2 Sort-merge join

A join combines rows from two tables that match on a condition, usually equality on a shared key (an equi-join), like orders.customer_id = customers.id. The sort-merge join works in two stages: (1) sort both inputs on the join key, then (2) merge them with the zipper sweep.

Stage 1 — Sort both sides

We sort both tables on the join key, typically using the external merge sort from Topic 1. If a side is already sorted on that key — say it arrives from an index scan, or is the output of an earlier sort — we skip its sort entirely. That's the single biggest reason to choose this join.

Stage 2 — Merge the sorted inputs

We walk both sorted inputs with two cursors. Compare the join keys at the cursors:

• If left.key < right.key → advance the left cursor.
• If left.key > right.key → advance the right cursor.
• If they're equal → output the matching pair(s), then advance.

Handling duplicates

The tricky case is duplicate keys — when many left rows and many right rows share the same key, every left must pair with every right (a small cross-product for that key group). To do this we mark the start of the matching group on the right and, for each left row with that key, replay the whole right group. After the keys differ, we move on. This is why the inner loop below "rewinds" the right cursor for repeated keys.

// Sort-merge join on equality of key k
sort(L) by k;  sort(R) by k
i = 0;  j = 0
while i < L.size and j < R.size:
    if L[i].k < R[j].k:  i += 1
    elif L[i].k > R[j].k:  j += 1
    else:                              // keys match
        jStart = j                     // remember start of R group
        while i < L.size and L[i].k == R[jStart].k:
            j = jStart
            while j < R.size and R[j].k == L[i].k:
                emit(L[i], R[j])       // every L × every R in group
                j += 1
            i += 1

Complexity & cost

If sorting both sides costs Sort(M) and Sort(N) page I/Os, and the merge sweep reads each input once for M + N, the total is:

3 Hash join (and grace / partitioned hash join)

A hash join is usually the fastest way to do an equi-join when neither input is sorted. It relies on a hash table: a structure that turns a key into a slot via a hash function, so you can find matching rows in roughly constant time instead of scanning.

The two phases: Build and Probe

Build phase: scan the smaller input (the build side) and insert every row into a hash table keyed on the join key. We pick the smaller side so the table is more likely to fit in memory. Probe phase: scan the larger input (the probe side); for each row, hash its key, jump to that bucket, and emit a result for every build row in the bucket that actually matches (hashing can put different keys in the same bucket — a collision — so we still compare the real keys).

# In-memory hash join: build on smaller S, probe with larger R.
# Each row is (key, payload). We join on the key.
from collections import defaultdict

def hash_join(build, probe):
    # --- Build phase: hash the SMALLER relation by key ---
    table = defaultdict(list)          # key -> list of build rows
    for key, payload in build:
        table[key].append((key, payload))
    # --- Probe phase: scan the LARGER relation ---
    out = []
    for rkey, rpay in probe:
        for srow in table.get(rkey, ()):   # dict guards collisions for us
            out.append((srow, (rkey, rpay)))
    return out

# --- Tiny demo: customers (small) JOIN orders (big) on customer_id ---
if __name__ == "__main__":
    customers = [(1, "Alice"), (2, "Bob"), (3, "Cara")]
    orders    = [(2, "ord-A"), (2, "ord-B"), (3, "ord-C"), (9, "ord-D")]
    for s, r in hash_join(customers, orders):
        print(s[1], "<->", r[1])
    # Alice never matches; customer 9 has no row => both dropped.
    # Output: Bob <-> ord-A / Bob <-> ord-B / Cara <-> ord-C

// In-memory hash join: build on smaller S, probe with larger R.
// Each row is {key, payload}. We join on the key.
#include <iostream>
#include <string>
#include <vector>
#include <unordered_map>

struct Row { int key; std::string payload; };

std::vector<std::pair<Row, Row>>
hashJoin(const std::vector<Row>& build, const std::vector<Row>& probe) {
    // --- Build phase: hash the SMALLER relation by key ---
    std::unordered_map<int, std::vector<Row>> table;   // key -> build rows
    for (const Row& s : build) table[s.key].push_back(s);

    // --- Probe phase: scan the LARGER relation ---
    std::vector<std::pair<Row, Row>> out;
    for (const Row& r : probe) {
        auto it = table.find(r.key);          // map guards collisions
        if (it == table.end()) continue;
        for (const Row& s : it->second) out.push_back({s, r});
    }
    return out;
}

int main() {
    // customers (small) JOIN orders (big) on customer_id
    std::vector<Row> customers = {{1,"Alice"},{2,"Bob"},{3,"Cara"}};
    std::vector<Row> orders    = {{2,"ord-A"},{2,"ord-B"},{3,"ord-C"},{9,"ord-D"}};
    for (auto& [s, r] : hashJoin(customers, orders))
        std::cout << s.payload << " <-> " << r.payload << "\n";
    // Output: Bob <-> ord-A / Bob <-> ord-B / Cara <-> ord-C
}

// In-memory hash join: build on smaller S, probe with larger R.
// Each row is (key, payload). We join on the key.
import java.util.*;

public class HashJoin {
    record Row(int key, String payload) {}

    static List<Row[]> hashJoin(List<Row> build, List<Row> probe) {
        // --- Build phase: hash the SMALLER relation by key ---
        Map<Integer, List<Row>> table = new HashMap<>();
        for (Row s : build)
            table.computeIfAbsent(s.key(), k -> new ArrayList<>()).add(s);

        // --- Probe phase: scan the LARGER relation ---
        List<Row[]> out = new ArrayList<>();
        for (Row r : probe)
            for (Row s : table.getOrDefault(r.key(), List.of()))  // guards collisions
                out.add(new Row[]{s, r});
        return out;
    }

    public static void main(String[] args) {
        // customers (small) JOIN orders (big) on customer_id
        List<Row> customers = List.of(new Row(1,"Alice"), new Row(2,"Bob"), new Row(3,"Cara"));
        List<Row> orders    = List.of(new Row(2,"ord-A"), new Row(2,"ord-B"),
                                       new Row(3,"ord-C"), new Row(9,"ord-D"));
        for (Row[] pair : hashJoin(customers, orders))
            System.out.println(pair[0].payload() + " <-> " + pair[1].payload());
        // Output: Bob <-> ord-A / Bob <-> ord-B / Cara <-> ord-C
    }
}

Complexity

When the hash table doesn't fit: grace / partitioned hash join

The simple version assumes the build side fits in memory. What if it doesn't? The answer is the grace hash join (also called partitioned hash join), and it leans on the same idea as external sort: use the disk and the hash function to break a big problem into small ones.

Partition phase: apply a hash function on the join key to split both inputs into the same set of k partitions written to disk. Because the same hash sends a given key to the same partition number in both tables, a row in partition i of S can only ever match rows in partition i of R. Join phase: for each partition pair, run the ordinary in-memory hash join (build on the smaller partition, probe with the other). We sized partitions so each build partition fits in memory.

// Grace (partitioned) hash join
// --- Partition phase: split both inputs by hash(key) ---
for s in S:  writeToPartition(Sp[ hashA(s.key) ], s)   // k partitions on disk
for r in R:  writeToPartition(Rp[ hashA(r.key) ], r)
// --- Join phase: matching partitions only ---
for i in 0 .. k-1:
    H = {}
    for s in Sp[i]:  H[ hashB(s.key) ].append(s)  // 2nd hash fn
    for r in Rp[i]:
        for s in H[ hashB(r.key) ]:
            if s.key == r.key:  emit(s, r)

4 Hash aggregation & grouping

An aggregation collapses many rows into summary numbers: COUNT, SUM, AVG, MIN, MAX. With GROUP BY you get one summary row per group (e.g. total sales per region). There are two main strategies to compute GROUP BY: sorting and hashing.

Strategy A — Grouping by sorting

Sort the rows on the grouping key (Topic 1). Now all rows of a group sit next to each other, so a single linear pass can detect each group boundary and emit its aggregate. Cost is dominated by the sort: Sort(N) + N. It's the natural choice if the data is already sorted or if you also need the output sorted.

Strategy B — Grouping by hashing

Scan the rows once. For each row, hash its grouping key to find that group's slot in a hash table, and update a running aggregate stored there. No sorting required — and no need to keep the individual rows, only one small accumulator per distinct group. This is hash aggregation, and it's usually the winner when the number of distinct groups is small enough that all the accumulators fit in memory.

Computing the aggregates incrementally

The clever part: most aggregates can be maintained with a tiny running state, updated one row at a time.

Note that AVG can't be kept as a single number — you must track both the running sum and the count, and divide only at the end.

// Hash aggregation: SELECT region, SUM(amount), COUNT(*)
//                   FROM sales GROUP BY region
H = {}                                  // group key -> running state
for row in sales:
    g = row.region
    if g not in H:
        H[g] = { sum: 0, count: 0 }
    H[g].sum   += row.amount
    H[g].count += 1
for g in H:                            // one output row per group
    emit(g, H[g].sum, H[g].count)

Comparison: choosing a join / aggregation strategy

★ Putting it all together

You just learned the heavy machinery behind ORDER BY, JOIN, and GROUP BY on data that's too big for memory. Here's the one-paragraph story connecting all four topics:

Quick self-check

📚 References & Further Reading

Class material

• 📄 Original course notes / handout (source sheet) — open the shared class material for this session.
• 📘 Class handout: "DBMS Session 10 — Query Execution 2" — covers external sort, sort-merge join, hash join, and aggregation strategies.

Papers, docs & deep dives

• CMU 15-445 Database Systems — the gold-standard course; see the lectures on sorting/aggregation and on join algorithms for slides and recordings of exactly these topics.
• Database System Concepts (Silberschatz, Korth, Sudarshan) — the query-processing chapter walks through external sort-merge, sort-merge join, and hash join with the same cost formulas used here.
• Goetz Graefe, "Query Evaluation Techniques for Large Databases" (ACM Computing Surveys, 1993) — the classic, deeply readable survey of sorting, joins, and hashing in real query engines.
• PostgreSQL Documentation — Planner/Optimizer — see how a production system actually chooses between merge join, hash join, and sort vs. hash aggregation.
• Use The Index, Luke! — a friendly, example-driven guide to how indexes and sorted data make joins and ordering fast.