Operators that pull one row at a time

A query plan is a tree of operators: scan, filter, project, limit. Each one asks its child for one row, does its bit, and hands one row up. Three methods on an interface are enough to run any plan, and the same interface gives you streaming and early exit for free.

Part 1 gave us a function from (expression, row) to value. A query needs more than that. It has to get rows out of a table, throw away the ones the WHERE rejects, compute the SELECT list, and stop at the LIMIT. Each of those is a separate step, and the steps have to pass rows to each other somehow.

"Somehow" is the design decision in this post. There are two obvious answers, and the gap between them is the gap between a toy and an engine you can point at a billion-row table.

Option one: every step makes a table

The most direct approach treats each step as a function from table to table. The scan produces the whole table. The filter consumes it and produces a smaller one. The projection produces a narrower one, and the limit keeps the first n.

function run(plan)
  match plan
    case Scan(t)          -> return t.rows                           -- all of them
    case Filter(p, child) -> return [r for r in run(child) if eval(p, r) == true]
    case Project(es, child) -> return [[eval(e, r) for e in es] for r in run(child)]
    case Limit(n, child)  -> return run(child)[0 : n]

This is easy to write and easy to debug, because every intermediate result is sitting there to inspect. It also has two problems that get worse with scale. Every intermediate result is fully built in memory, so memory use is the size of the largest intermediate. And LIMIT 10 on a billion-row table still filters a billion rows, then throws away all but ten. The limit knows it only wants ten, but by the time it runs, the work is done.

Option two: every step is an iterator

The alternative, from Goetz Graefe's Volcano system (1994) and still the shape of Postgres's and MySQL's executors, is to make every operator implement the same tiny interface. (SQLite does something different, which is where part 7 ends up.)

interface Operator
  Open()                   -- get ready: open children, allocate state
  Next() -> (row, ok)      -- produce the next row, or ok = false when exhausted
  Close()                  -- release resources, close children
  Schema() -> [names]      -- the columns of the rows this produces

An operator doesn't know or care what its child is. It calls Next() on it. A plan is a tree of these objects, and running the plan means pulling on the root:

function run(root)
  root.Open()
  loop
    (row, ok) = root.Next()
    if not ok then break
    emit(row)                -- send to the client, write to a socket…
  root.Close()

Here are the four simplest operators. They're short, and that's the point:

struct Scan { table, pos }
  method Open()  pos = 0
  method Next()
    if pos >= len(table.rows) then return (nil, false)
    pos = pos + 1
    return (table.rows[pos - 1], true)

struct Filter { pred, child }               -- pred is a compiled closure from part 1
  method Next()
    loop
      (row, ok) = child.Next()
      if not ok then return (nil, false)
      if pred(row) == true then return (row, true)
      -- otherwise: drop it and pull again

struct Project { exprs, child }
  method Next()
    (row, ok) = child.Next()
    if not ok then return (nil, false)
    return ([e(row) for e in exprs], true)

struct Limit { n, count, child }
  method Open()  count = 0; child.Open()
  method Next()
    if count >= n then return (nil, false)  -- stop: don't even ask the child
    (row, ok) = child.Next()
    if not ok then return (nil, false)
    count = count + 1
    return (row, true)

Now run SELECT name FROM customers WHERE city = 'Austin' LIMIT 2 through it, one event at a time. Watch two things: where the Next() calls are at each moment, and how much of the table has been read when the query finishes.

Control flows down and data flows up. Every row the driver receives was produced by Scan at the bottom and passed hand to hand, unchanged or transformed, through every operator above it. At no point does any operator hold more than the one row it's working on.

Look at LIMIT 2 in particular. Scan reads Ada (pass), Bo (dropped), Cy (pass), and then Limit's count is 2. The driver's third call returns done immediately, and Limit never asks Project for another row. Di is never read. On four rows that saves almost nothing. On real data it's the difference between reading a page and reading the whole table:

This is what pipelining means. A pipelined operator can produce output rows as soon as it has consumed enough input to compute them. Scan, Filter, Project, and Limit all need exactly one input row per step, so a chain of them uses constant memory no matter how big the table is, and it stops the moment the consumer stops asking.

The opposite: pipeline breakers

Not every operator can work that way. ORDER BY can't return its first row until it has seen every input row, because the last row read might sort first. The same goes for GROUP BY (the last row might belong to the first group) and for the build side of a hash join (next post). These operators are pipeline breakers. Their Next() still has the same signature, but the first call does a lot of work: it drains the child completely into some internal structure, then starts handing out results.

struct Sort { keys, child, buffer, pos }
  method Next()
    if buffer is null then                -- first call: consume everything
      buffer = []
      loop
        (row, ok) = child.Next()
        if not ok then break
        append(buffer, row)
      sortStable(buffer, keys)
      pos = 0
    if pos >= len(buffer) then return (nil, false)
    pos = pos + 1
    return (buffer[pos - 1], true)

A plan is a series of pipelines joined at pipeline breakers. Within a pipeline, rows stream; at a breaker, everything stops until the input is consumed. Keep that picture in mind for part 7, where the compiler's structure follows directly from it.

Where the state lives

Look at the operator boxes in the first figure again. Each operator's own state is a field on the operator: Scan's pos, Limit's count, Sort's buffer. Those are on the heap. They persist between Next() calls, and they're the only thing that persists.

During a call, there's more. When Filter calls Scan, Filter's stack frame is waiting on the call stack, and the dashed boxes in the figure are exactly those waiting frames. If you wanted to stop in the middle, say right after Scan returns Bo but before Filter has dropped him, that in-progress state is spread across four stack frames. You can't easily save it.

Between calls to the root's Next(), though, the stack is empty. Everything is in operator fields. That means a Volcano plan is naturally resumable at row boundaries: stop calling Next(), come back later, call it again, and the plan continues. That's how database cursors and FETCH 100 work. Whether you can do better than row boundaries, and whether you can save the state somewhere and restore it in another process, is a question for the last part of this series.

What it costs

The iterator model has a well-known price: one virtual call per operator per row, through an interface the CPU can't predict, with little opportunity for the compiler to optimize across operator boundaries. For a plan five operators deep over a hundred million rows, that's half a billion indirect calls just moving rows around, before any real work.

Modern engines attack this cost in two directions. Vectorized engines (MonetDB/X100, DuckDB, most column stores) keep the pull interface but make Next() return a batch of ~1000 rows stored as column arrays, which cuts the call overhead by three orders of magnitude and gives the CPU tight loops it can run fast. Compiled engines (HyPer, Umbra) turn the model inside out: instead of each operator pulling from below, the scan pushes each row up through all the operators of a pipeline in a single generated loop. We'll build a small version of the push model in part 7.