Keeping a join alive, one row at a time

TPC-H Query 10 asks a reporting question. For one quarter of 1993, which customers cost the most in returned goods?

select c_custkey, c_name, sum(l_extendedprice * (1 - l_discount)) as revenue, ...
from customer, orders, lineitem, nation
where c_custkey = o_custkey
  and l_orderkey = o_orderkey
  and o_orderdate >= date '1993-10-01'
  and o_orderdate < date '1993-10-01' + interval '3' month
  and l_returnflag = 'R'
  and c_nationkey = n_nationkey
group by c_custkey, c_name, c_acctbal, c_phone, n_name, c_address, c_comment
order by revenue desc
limit 20;

Against a finished database this is an ordinary four-way join. The interesting version is the other one: rows are still arriving, one at a time, and the answer has to be correct after every single arrival.

Re-running the join per arrival is out — that is millions of rows of work to account for one new order. The obvious repair, “join just the new row against everything else”, sounds cheap and is not. If the new row is a nation, it joins with a sizeable fraction of the database.

What follows is the way out taken by Cquirrel (SIGMOD 2020), narrowed to this one query. There is no mathematics in it. The whole thing rests on a single property of a row, and everything else is bookkeeping to keep that property true. Every figure below is driven by a working implementation of the algorithm rather than by hand-drawn frames, and the last one hands you the buttons.

The chain has a direction

Before anything can be maintained, the four relations have to be put in some order. The order is not the one most people reach for.

So the shape for the rest of this essay is a vertical chain, with lineitem on top and nation at the bottom. Read each step downward as “references”. Root and leaf follow the arrows, not the row counts: lineitem is the root although there are six million of them, and nation is the leaf although there are twenty-five.

A row is live when it can reach the bottom

Here is the entire idea.

A row is live when it can join with one row in every relation below it, all the way down to the leaf. A leaf row is live for nothing, because it has no reference that could fail.

That is the only definition in this essay. A stored row that is not live is dormant, and dormant is neither an error nor a rejection. The row is present, indexed, and perfectly good. It just cannot reach the bottom yet, so it cannot contribute to the answer yet.

Watch what that buys. Below, four rows arrive in the worst order imaginable: the lineitem first, its nation last.

Nothing there required a sorted stream, or a lineitem to be held back until its order showed up, or any row to be read twice. Each arrival asked one question — is the row I reference live? — and then either joined the answer or lay down and waited to be woken.

Now the same four rows in the order you would have picked yourself.

Same final picture, same +$2,000. This is worth saying plainly, because it is what makes the approach usable at all: the state after a set of updates does not depend on the order the updates arrived in. Order changes how much work happens and when, never the answer.

Notice too where the number came from. Both times it left at the lineitem, and that is not an accident of the example. A customer becoming live does not change the query’s answer by itself — it changes the answer only by allowing lineitems to become live. So the root is the one place output is produced, and every relation below it exists to push liveness upward toward it.

Nothing is stored that cannot matter

Two of Q10’s predicates look at a single row: the three-month order window and l_returnflag = 'R'. Those run before anything is stored.

A rejected row is not stored dormant, waiting to be re-examined. It is dropped, and the picture does not move.

That has a consequence worth stating out loud, because it looks like a bug the first time you see it. Inside this maintained state, foreign keys do not hold. An order can reference a customer that is not there — because the customer has not arrived yet, or because the order passed a filter that its neighbours failed. Dangling references are the normal condition here, not corruption.

What one relation remembers

A relation in the middle of the chain keeps three things, and it is worth watching all three move.

The third structure is the one that does the real work. When a customer becomes live, nobody scans the orders table looking for affected rows: the customer key is looked up in that index, and it hands back exactly the orders that reference it. Membership in L or N then says whether each of those orders actually crossed a boundary, so a row that was already live is silently skipped and the cascade stops early.

The paper gives every row a counter, s(t), holding the number of child relations it is currently live on; a row is live when that number reaches the number of children it has. On a chain every relation has exactly one child, so the counter can only ever be zero or one — which is to say, it is already recorded by which of the two maps the row is sitting in. The Scala carries no counter at all:

private val liveByPrimaryKey = mutable.HashMap.empty[PK, Value]
private val nonLiveByPrimaryKey = mutable.HashMap.empty[PK, Value]
private val parentKeysByChildKey =
  mutable.HashMap.empty[ChildPK, mutable.HashSet[PK]]

Deleting is not inserting backwards

Deletion looks like it should be insertion with the signs flipped, and for the state it very nearly is: rows slide from L back to N, and the deltas come out negative. Step through it, and then look closely at the last panel of the first update.

The two negative deltas say ALGERIA. By the time they were built, no index in the system still held that name — the nation row was removed in the first frame, several frames before the root was reached.

This is the one place where a plausible implementation is quietly wrong. On the way up, a lineitem that has just become live can rebuild its own join by looking downward, because everything below it is live by definition:

private def assembleFromLive(lineitem: Lineitem, sign: Int): Option[JoinDelta] =
  for
    order <- ordersState.live(lineitem.orderkey)
    customer <- customerState.live(order.custkey)
    nation <- nationState.live(customer.nationkey)
  yield Aju.toDelta(lineitem, customer, nation, sign)

On the way down, that same lookup returns nothing, because the deleted row is already gone. So the demotion path does not look anything up. It carries the values it is about to lose down the chain as arguments, gathering one more at each level, until the root has everything a delta needs:

private def demoteLineitems(
    order: Orders,
    customer: Customer,
    nation: Nation,
): List[JoinDelta] =
  lineitemState.childBecameNonLive(order.orderkey).toList.map { lineitem =>
    Aju.toDelta(lineitem, customer, nation, sign = -1)
  }

The second half of that player is the other thing to take from this section. Putting the nation back restores the entire join, and not one order or lineitem had to be re-read to do it. Dormant rows never left their indexes. They were waiting the whole time.

Summing without forgetting how many

The root emits signed money attached to a customer key. The last stage adds it up, and has exactly one problem worth solving.

A running sum cannot tell “this customer’s returns cancelled out to zero” from “this customer has no returns and should not be in the answer”. Both are zero. So each group carries the number of live joins standing behind it, and it is that count reaching zero — not the money reaching zero — that removes a customer from the result.

Q10 finishes by ranking these groups and keeping the top twenty. That part is not built yet in the implementation these figures come from.

Drive it yourself

Every row button below is a real insert or delete against a real instance of the algorithm. The random-order button is a different kind of aid: it shuffles the accepted rows, then inserts them one at a time so you can watch the exact sequence of clicks and the cascade each click causes. It is still the same instance and the same insert operation — the only script is the pause between clicks.

Run it a few times, then compare the final picture: the order changes the route through dormant state, never the answer. Or make your own order and delete a row in the middle of a live chain — a customer, say — to watch a single removal put four relations to sleep and pull the money back out of the answer.

The two rows with red outlines can never be stored. One is a lineitem that was never returned, the other an order dated outside the window.

What it costs

The usual summary of this algorithm is “constant time per update”, and that is not quite what is true. The bound is amortised, and the cost of a single update is not bounded at all: inserting one nation row can wake every customer in that nation, every one of their in-window orders, and every returned lineitem on those orders. You saw a four-row version of that cascade; there is nothing stopping it from being a two-hundred-thousand-row version.

What is genuinely constant is the work per row that actually changes status. No row is ever scanned speculatively, nothing is recomputed to check whether it changed, and each transition is a handful of hash-map operations. The cost is proportional to how much of the answer really moved, which is the honest thing to want from an incremental system.

Two limits are worth knowing. This works because Q10’s relations form a chain — one child each. When the foreign keys form a branching graph, “live on every child” stops being sufficient, because two children can each be live through paths that reach different rows further down, and the algorithm needs extra state to notice. And every row here has an identity: an update is a delete followed by an insert, never an edit in place. That is what lets a delta be signed, and a signed delta is what lets the answer be maintained instead of recomputed.