Nested loops, hash tables, and merging

A join's definition is one line: every pair of rows where the condition holds. There are three classic ways to compute it, and the difference between them is the difference between a query that finishes in milliseconds and one that runs until somebody kills it.

Everything so far has read one table. Real queries read several and stitch them together: customers with their orders, orders with their products. In the iterator model a join is just another operator, one with two children instead of one. It still has to answer the same question as everything else: when someone calls Next(), what row do I return?

Our running example is the toy tables:

SELECT c.id, c.name, o.id, o.customer_id
FROM customers c JOIN orders o ON o.customer_id = c.id

Ada has two orders, Cy and Di have one each, and Bo has none. Order 14 belongs to a customer 9 who doesn't exist. Those two oddities are there for outer joins later.

The definition, executed literally: nested loop join

"Every pair of rows where the condition holds" translates straight into code: for each row on the left, look at every row on the right.

struct NLJoin { left, right, pred, cur }     -- cur: the current outer row
  method Next()
    loop
      if cur is null then
        (cur, ok) = left.Next()
        if not ok then return (nil, false)
        right.Rewind()                        -- start the inner side over
      (r, ok) = right.Next()
      if not ok then
        cur = null                            -- inner exhausted: move to next outer row
        continue
      joined = concat(cur, r)
      if pred(joined) == true then return (joined, true)

Note the shape: the operator keeps cur, the outer row it's working on, as a field. Each call to Next() picks up where the last one left off, partway through the inner loop. It's the same trick as Limit's counter from part 2: the "loop" isn't a loop on the stack, it's state on the heap that each call advances.

The cost is what it looks like: |L| × |R| evaluations of the predicate. Four customers times five orders is 20. A thousand customers times a hundred thousand orders is a hundred million. Step through it below, then switch to Hash and compare the counter.

Nested loops have one big thing going for them: they work for any condition. A join on o.qty > c.age / 10, or ST_Contains(region, point), or a.ts BETWEEN b.start AND b.end has no single key to hash or sort on. The nested loop doesn't care; it just evaluates the predicate on every pair. Every engine keeps one around as the fallback for whatever the other algorithms can't handle.

It's also the right choice more often than its complexity suggests. If the outer side has three rows, a nested loop is three passes over the inner side with no hash table to build. And if the inner side has an index on the join column, each "pass" becomes a lookup instead of a scan. That is the index nested-loop join, which is part 5's topic and the workhorse of OLTP databases.

Hash join: do the matching work once

Most joins in practice are equi-joins: the condition is left.key = right.key. For an equi-join, a nested loop throws away information. When it compares Ada (id 1) with every order, it learns which orders have customer_id 1, and then it forgets all of that and starts over for Bo.

A hash join remembers. It reads one side, the build side, once, and files every row in a hash table under its join key. Then it reads the other side, the probe side, and for each row looks up only the bucket where matches could be:

struct HashJoin { probe, build, probeKey, buildKey, table, cur, matches, i }
  method Open()
    probe.Open(); build.Open()
    table = new HashMap(key -> [rows])        -- build phase: drain the build side
    loop
      (r, ok) = build.Next()
      if not ok then break
      k = buildKey(r)
      if k is null then continue              -- NULL never equals anything
      append(table[k], r)
  method Next()
    loop
      if i < len(matches) then               -- still emitting matches for cur
        i = i + 1
        return (concat(cur, matches[i - 1]), true)
      (cur, ok) = probe.Next()                -- probe phase: one probe row at a time
      if not ok then return (nil, false)
      k = probeKey(cur)
      matches = if k is null then [] else table[k]
      i = 0

In the figure the "hash table" is four buckets keyed by customer_id mod 4, so you can see collisions happen. Customer 1 and the nonexistent customer 9 land in the same bucket, so probing for Ada finds order 14 too, and the key comparison rejects it. A real hash table has many more buckets and a better hash function, but it still compares keys after hashing, because hashing only narrows down where to look.

Costs: every build row is hashed and inserted once, and every probe row is hashed and looked up once. That's roughly |L| + |R| work instead of |L| × |R|, and for big inputs that's the whole ballgame. The price is memory: the entire build side has to sit in the hash table before the first output row comes out. Which side to build on is the engine's choice, since an inner join is symmetric. The rule is to build on the smaller side. Get that wrong and you either use far more memory than needed or spill to disk (more on spilling in part 4). Knowing which side is smaller before running the query is a job for statistics, in part 6.

In iterator terms, the build is a pipeline breaker and the probe isn't. Building drains one child completely during Open(); probing streams the other child one row at a time. So a plan like ((A ⋈ B) ⋈ C) can probe with A's rows all the way up through two hash tables without ever materializing A.

Sort-merge join: line both sides up

The third classic algorithm sorts both inputs by the join key, then walks them in lockstep with one cursor each, like the merge step of merge sort. At each step, compare the two keys. If the left one is smaller, no remaining right row can match it, so advance left. If the right one is smaller, advance right. If they're equal, emit the match.

function mergeJoin(L, R)                    -- both sorted by key
  i = 0; j = 0
  while i < len(L) and j < len(R)
    if L[i].key < R[j].key then i = i + 1
    elif L[i].key > R[j].key then j = j + 1
    else
      mark = j                                -- start of the run of equal keys
      while j < len(R) and R[j].key == L[i].key
        emit(concat(L[i], R[j])); j = j + 1
      i = i + 1
      if i < len(L) and L[i].key == L[i - 1].key then
        j = mark                              -- duplicate on the left: replay the run

The fiddly part is duplicates. Ada has two orders, so the right side has a run of equal keys. If the left side had duplicates too, every left row in its run would have to pair with every right row in the other run, which means rewinding the right cursor to the start of the run. That's the mark/restore in the pseudocode.

Why would you ever do this instead of hashing, given that sorting costs n log n? There are three reasons:

• The inputs are already sorted. If both sides come out of a B+tree index on the join key (part 5), or one side was sorted for an ORDER BY anyway, the sort is free and the merge is a single linear pass with no hash table at all. Tick "already sorted" in the figure below to see the difference.
• Memory. A merge join holds only the current run of equal keys. Sorting data that doesn't fit in memory is a solved problem (external merge sort, next post), and so is merging two sorted files. A hash join whose build side doesn't fit has to partition both sides to disk first, which is more complicated.
• The output is sorted. The rows come out ordered by the join key, which a later ORDER BY, merge join, or GROUP BY can use for free.

How they scale

The toy tables make all three look about equally cheap. Drag the sizes up:

The nested loop's bar grows with the product of the sizes, while the hash join's grows with the sum. Make both sides a million rows and the difference is a trillion operations, which is hours, against two million, which is milliseconds. That's why the planner's most important job, by far, is not picking a nested loop when it shouldn't.

Outer joins

An inner join drops rows that don't match: Bo, who has no orders, disappears, and so does order 14, whose customer doesn't exist. A left outer join keeps every row from the left side. If a left row matched nothing, it's emitted once, padded with NULLs for the right side's columns. Switch the first figure to "Left outer" and watch Bo appear with NULLs.

Implementing this means tracking one bit per left row: did anything match?

-- inside the probe loop of a hash join, per probe row:
matches = table[probeKey(cur)]
if len(matches) == 0 and kind == LEFT then
  return (concat(cur, nulls(len(build.Schema()))), true)

For a nested loop, it's a matched flag reset per outer row and checked when the inner side runs out. Either way, the preserved side has to be the outer or probe side, because that's the side where you know, row by row, when matching is finished. A right outer join is a left outer join with the inputs swapped. A full outer join needs the other direction too: mark every build row that ever matched, and after the probe side is exhausted, emit the unmarked build rows (order 14 would show up there) padded with NULLs.

Outer joins also restrict the planner. A JOIN B JOIN C can be evaluated in any order, because inner joins are associative and commutative. A LEFT JOIN B can't simply be flipped. The integrated engine supports only inner joins, so part 6's planner can reorder freely; the lesson code implements left joins so you can see the mechanics.

Choosing

That table is the start of a cost model. Every row of it depends on knowing |L| and |R|, and usually not the base table sizes but how many rows survive the filters below the join. Those numbers aren't known until the query runs. Estimating them anyway is part 6.