Keeping state alive: streaming joins without recomputing everything

Suppose you need to maintain the live results of TPC-H Query 10. It joins customer, orders, lineitem, and nation, then ranks customers by revenue lost to returned goods.

In a traditional data warehouse, this is a routine batch job. But what if rows are streaming in live—via change-data-capture (CDC) or event streams—and the query results must be accurate after every single insert or delete?

You immediately run into two bad choices:

1. Re-run the query on every row arrival. Scanning millions of records just because one new order arrived will quickly overwhelm the database.
2. Join the newly arrived row against the existing tables. This sounds cheap until you insert into a parent table. Adding a single nation row would force the engine to join it against thousands of related customers, orders, and lineitems, causing a massive intermediate fan-out.

How do streaming systems maintain multi-way joins without drowning in redundant computation or state explosion?

In 2020, a SIGMOD paper introduced Cquirrel. Instead of buffering massive intermediate join trees or blindly recomputing joins, Cquirrel organizes the relations into a directional dependency chain and tracks a simple local state: row liveness.

Every figure below is powered by a live implementation of this algorithm, allowing you to inspect state transitions and trigger updates step by step.

The dependency chain: Consumers and Providers

Before maintaining join state incrementally, we need to clarify how the four relations depend on each other.

In traditional database discussions, talking about “parent” and “child” tables gets confusing fast—foreign keys point one way in schema design, while query execution trees often point the other. Spatial metaphors like “top” and “bottom” only add to the ambiguity.

Because we are processing a continuous stream of updates, it is much clearer to use the language of dataflow: Consumers and Providers.

• Consumer (The depender): A relation that holds a foreign key. It cannot complete a join on its own—it must consume matching data from the relation it references. For example, orders is a consumer of customer.
• Provider (The dependee): A relation being referenced. It provides the identity and attributes required by its consumers. For example, customer is a provider to orders.

If you trace the foreign-key equalities in Query 10, the four relations line up into a single dependency chain where each relation consumes the one below it:

Notice the two ends of this chain:

1. The Leaf Provider (nation): At the end of the chain, nation consumes nothing. Because it has no foreign-key references to satisfy, a nation row is unconditionally self-sufficient.
2. The Root Consumer (lineitem): At the beginning of the chain, lineitem is consumed by nothing. It is the fact table holding the actual business measurements (revenue). This is where the query’s aggregate output will be emitted.

This single mental model defines the entire algorithm:

• Foreign-key dependencies point downstream: Consumer -> Provider.
• Activation signals bubble upstream: When a Provider becomes ready, it notifies its dependent Consumers.

Defining “liveness”: the state of a row

In a traditional batch query, a join is an operation. In Cquirrel, a join is maintained as a state machine where every row has a status:

• Live: A Consumer row is live when its referenced Provider is also live—recursively all the way down to the leaf Provider (nation). A leaf Provider is unconditionally live because it has no external dependencies to satisfy.
• Dormant: A stored row whose referenced Provider is not yet available or not yet live. Being dormant is neither an error nor a rejection. The row is safely indexed in local memory—it simply cannot contribute to the query answer until its Provider becomes live.

Handling out-of-order stream arrivals

The primary benefit of this state model is how easily it absorbs out-of-order data.

Below, four rows arrive in the worst possible order: the Root Consumer (lineitem) arrives first, and the Leaf Provider (nation) arrives last.

Notice how cleanly the state absorbs these updates:

• The stream does not need to be sorted or pre-buffered.
• The engine does not stall the incoming lineitem waiting for its order.
• No table is ever scanned twice.

Whenever a row arrives, it asks a single local question: is the Provider I reference already live? If yes, the row becomes live immediately. If not, it indexes itself as dormant and waits for an activation signal from below.

Now watch what happens when the exact same four rows arrive in reverse order—from Leaf Provider to Root Consumer:

The final result is identical: +$2,000 emitted for customer 10.

This demonstrates the core architectural guarantee: the final query state is independent of stream arrival order. Arrival order only changes when the intermediate activations happen, never the final answer.

Notice also where the output delta is emitted: in both scenarios, the +$2,000 delta is produced at the Root Consumer (lineitem). A customer or order becoming live does not directly change the query answer; it only unblocks upstream lineitems to complete their join chain and emit revenue.

Filtering before storing: single-row predicates

Two of Query 10’s conditions apply to individual rows without requiring any joins: the three-month date window on orders, and l_returnflag = 'R' on lineitem.

These filter predicates run before any record enters memory:

If an incoming row fails a predicate, it is immediately discarded. It is not stored as dormant, because it can never contribute to a valid join.

This brings up an important contrast with static relational databases: streaming state routinely contains dangling foreign keys. An order row may pass its date filter and be stored, even if its referenced customer was discarded or has not arrived yet. In an incremental streaming engine, dangling references are normal operating conditions, not data corruption.

What an intermediate relation stores

A relation in the middle of the chain (such as orders or customer) maintains three internal data structures:

1. liveRecords (L): A hash map of rows whose referenced Provider is confirmed live.
2. dormantRecords (N): A hash map of rows whose referenced Provider is not currently live (or has not yet arrived).
3. consumersByProviderKey: A reverse index mapping each Provider ID to the set of Consumers that reference it (for example, mapping each c_custkey to all orderkeys belonging to that customer).

Why the reverse index is an unconditional waitlist

Notice a subtle but fundamental property of out-of-order streams: a consumer registers itself in the reverse index before it even knows whether its provider exists.

When order 100 arrives with foreign key c_custkey = 10, the engine does not wait for customer 10 to show up. It unconditionally appends order 100 into consumersByProviderKey(c10) as a waiting list. Then, it checks downstream:

• Because customer 10 has not arrived yet, order 100 enters the dormant set N(orders).
• In the reverse index on the right, the key c10 is displayed in a dashed, dormant outline: the waitlist is recorded, but the provider is not yet live.

When customer 10 finally arrives and becomes live, notice how the state reacts:

• The key c10 on the right immediately turns solid teal (Live).
• orders is notified once by key: it performs an $O(1)$ lookup in consumersByProviderKey(c10), instantly retrieves [order 100, order 101], and moves them from dormantRecords into liveRecords.
• The engine never scans N(orders).

Even if customer 10 is later deleted, the orders slide back to dormantRecords, but their mapping in consumersByProviderKey stays put. When the customer is re-inserted later, the join is restored instantly without rescanning anything.

On a linear chain, each relation has exactly one Provider. A row is therefore either dormant or live. No richer state is needed.

Handling deletions and retractions

At first glance, a deletion seems like an insertion in reverse: rows transition from liveRecords back to dormantRecords, and the engine emits negative deltas (-$2,000).

Step through the deletion trace below, and watch what happens when a nation row is removed:

Notice the negative delta emitted at the root: it says ALGERIA. But by the time that delta was constructed at the root, the nation row for Algeria had already been deleted from memory in the very first step.

Insertion and deletion differ in one important way:

• On insertion, every live Provider is still present. The root can read back down the chain to assemble the result.
• On deletion, the first Provider is already gone. Its attributes must travel upstream with the suspension signal so the root can emit the retraction.

The second half of the trace above shows another critical property: re-inserting the deleted nation restores the entire join immediately, without re-reading a single order or lineitem from disk. The dormant rows never left their hash maps; they were simply waiting to be reactivated.

Aggregation: distinguishing zero from empty

The Root Consumer (lineitem) emits signed revenue changes attached to a customer key (such as +2,000 or -2,000). The aggregation stage sums these values per customer, but faces one practical challenge:

A naive running sum cannot distinguish between two very different business states:

• A customer whose returns cancelled out to $0.00.
• A customer who has no returned items at all (and should not appear in the final query result).

Both evaluate to a sum of zero. To solve this, each aggregation group tracks both the running revenue and an active reference count (the number of live lineitems currently backing that customer). Only when the active count reaches zero is the customer completely evicted from the result set.

Drive it yourself

Use the interactive sandbox below to test how the algorithm handles arbitrary sequences of inserts and deletes.

The Random order button shuffles the valid records and inserts them one by one, allowing you to observe how different arrival sequences converge to the exact same final state:

Rows outlined in red are rejected by filter predicates (such as a lineitem that was never returned, or an order outside the target date window) and can never enter the state store.

Try deleting a row in the middle of a live chain—such as a customer—to watch how a single deletion cascades upstream, moving dependent orders and lineitems into dormant state and retracting revenue from the answer.

One chain, several machines

The single-process intuition was simple: store now, wake later. A dormant lineitem records which order it needs. When that order becomes live, the reverse index delivers the wake-up.

Sharding adds one constraint: the waiter and the wake-up must reach the same machine. If an order lands on shard 0 and its lineitem lands on shard 1, both rows are stored correctly. The lineitem still waits forever because its order is not local.

Watch the edge break

Start with no key and step through the stream. Then switch to orderkey without changing the current step.

The important unit is not an individual row. It is the dependency edge between a Consumer and its Provider. Routing works when that edge stays local.

Pick a key that keeps the wake-up path local

There are two useful choices for this query:

• Route facts by orderkey, keeping each lineitem beside its order. Copy customer and nation to every shard. A customer’s orders may still need a final aggregate merge.
• Route the whole customer subtree by custkey. Aggregation becomes local, but each lineitem must learn its customer’s key first.

That second option has an extra step because lineitem carries l_orderkey, not custkey. Toggle enrichment off to expose the missing link.

Now compare the complete plans. The table reports what each one copies, routes, re-keys, or leaves stranded.

The state machine inside each shard has not changed. It still uses L, N, the reverse index, and signed deltas. Routing only decides whether a Provider’s wake-up can reach its Consumers.

Keep one picture in mind

• A row passes its local filter, enters an index, and asks whether its Provider is live.
• If the Provider is missing, the row waits. It does not need to be retried or rescanned.
• Provider changes travel through the reverse index. They wake or suspend only the Consumers that reference that key.
• The Root Consumer emits the signed delta. Aggregation keeps the running value and a count of what still supports it.
• On several machines, route each dependency edge so the waiter and its wake-up meet on the same shard.

That is the algorithm: indexed waiting lists plus wake-up signals.