Interval Joins Over Event Time

Joining two unbounded streams is the operation that most clearly exposes the cost of event time, because a join is inherently a comparison against a set of past records and a stream has no past unless something retains it. A relational join over two tables can scan; a stream join over two event streams must hold one side long enough for the other to arrive, and the length of “long enough” is a statement about time, not about counts. Every stream join is therefore a buffer with an eviction rule, and the rule is derived from a time bound supplied by the query.

The bound is part of the query

An interval join matches records whose event times are within a stated distance, typically an asymmetric range: an order event matches a shipment event that occurred between five minutes before and one minute after it. The range is the query, not a tuning parameter, because it defines which pairs are supposed to match. Widening it changes the answer.

Given the range [t - lower, t + upper] for a record at t on the left side, a left record can only match a right record whose event time lies in that interval. Once the right stream’s watermark passes t + upper, no further right record can match, so the left record’s buffered copy can be released. The eviction time is computed per record and stored with it, and a timer fires at that event time to remove it.

That is the mechanism in one sentence: buffer each record until the opposite stream’s watermark proves it cannot match anything more. Everything else is bookkeeping about which side is buffered, how the state is keyed, and what happens to records that never match.

Both sides must be buffered

A join cannot know whether a left record will match until right records arrive, and it cannot know whether a right record will match until left records arrive. So both sides are retained for the window of possible overlap. The state is roughly proportional to the event rate on each side multiplied by the overlap duration.

This is the number that decides feasibility. Two streams at ten thousand events per second each, with a one-minute overlap, hold about 1.2 million records at steady state. At a hundred bytes per buffered record that is on the order of a hundred megabytes, which fits in memory. Multiply the rate by ten or the window by sixty and the same job needs tens of gigabytes, and the join becomes a disk-backed state problem.

The asymmetry of the range matters for the buffer. A range of [-5 min, +1 min] means left records are held for six minutes of right-stream progress while right records are held for six minutes of left-stream progress, but the number of records held depends on each side’s rate over its own retention interval. A fast left stream and a slow right stream puts the large buffer on the left, and the join’s memory is dominated by whichever side is fast.

Keying, and the watermark gate

An interval join is almost always keyed: the pair must share an entity, or the join degenerates into a cross product over the time window. The keyed form partitions both streams by the same key function and joins within each key, which confines the state per subtask and makes the join parallelizable.

The watermark that gates eviction is the minimum of the two inputs’ watermarks, for the reason given in Watermarks as Completeness Signals: a record on the left can only be released when the right side is complete past its range, and the right side’s completeness is exactly its watermark. An operator that used the maximum, or its own clock, would release records that later matches could still need.

The practical consequence is that a join between a fast stream and a slow one runs at the slow stream’s pace. If the right side is a daily batch feed, the left side’s records accumulate for a day and the join’s output lags by a day. That is not an implementation deficiency; it is what “match against records that may still arrive” costs. A pipeline that needs same-second output from a join between a real-time stream and a batch feed is asking for a different operation, usually a lookup against a materialized table rather than a stream join.

-- Interval join in continuous SQL: each order matches shipments
-- within the stated event-time range on the same key.
SELECT o.id, s.id, o.amount
FROM orders AS o
JOIN shipments AS s
  ON o.order_id = s.order_id
 AND s.event_time BETWEEN o.event_time - INTERVAL '5' MINUTE
                      AND o.event_time + INTERVAL '1' MINUTE;

The BETWEEN clause is evaluated per pair, which is why the engine cannot use an index in the ordinary sense: it has to consider every retained record in the range, and the range query over the buffered side is what the state backend’s ordered iteration provides.

Unmatched records and outer joins

An inner join emits only pairs, and a record that never matches is silently dropped when its timer fires. For a left outer join, the operator must instead emit the left record with nulls at the moment it becomes certain that no match can arrive — which is the same timer, evaluated against the right watermark. The timer’s job changes from “delete” to “emit and delete”, and the emission carries a null right side.

The subtlety is that a left record may match several right records, and the outer-join semantics decide whether the null row is emitted for a left record with zero matches only. Emitting a null row for a record that later gets a match produces a spurious unmatched result that must be retracted, which pushes the correction problem into the join’s output. Joins with a lateness allowance therefore emit and retract, and the retraction contract propagates to every downstream consumer.

A second subtlety is the duplicate. Two records that match produce one output row, and if one of them is reprocessed — because a checkpoint was restored, or because the source replayed — the pair is produced again. Exactly-once output from a join depends on the sink’s transactionality, not on the join. Two-Phase Commit and Exactly-Once Sinks is where that guarantee actually lives.

Deduplication is the degenerate join

A deduplication operator is a join of a stream with itself, matching a record against any prior record with the same key. Its state is one entry per key, holding either the last seen record or a marker, and its eviction is the TTL from Keyed State Layout and the Cost of Keeping It. The interesting case is the one where the key space is unbounded — a deduplication keyed by message identifier over all history grows without limit, which is the same failure as unbounded key cardinality in an aggregation.

The versioned-table pattern is the other degenerate case: a stream of changes joined against a stream of lookups, where the lookup side is retained in full as a table and the change side is matched against the table’s state at the change’s event time. That requires retaining versions of the table keyed by time, which is a join whose buffer is the table’s history rather than a fixed interval, and whose state is bounded by how far back the query needs to see.

The eviction timer is the correctness boundary

The timer that releases a buffered record is where the join’s assumptions become visible. If it fires too early — because a watermark advanced on a stream that later produced an older record — the match is lost, and the loss is not recoverable by the join itself. If it fires too late, the buffer grows past the point where the state backend can hold it, and the operator fails.

Both failures come from the same place: the watermark is an assertion, and the join trusts it. A join with an allowed lateness is a join that keeps the timer’s state beyond the eviction point, at the cost of the extra retention, so that a late record can still be matched against records that would otherwise have been released. The setting is per join, and it is chosen from the disorder measurement in Measuring Disorder to Size a Lateness Budget, because the join’s correctness is exactly as good as that measurement.