Stream Joins
Joining two unbounded inputs means holding both sides in state until a time bound says you may stop holding them. Without that bound it is not a join — it is a memory leak with a schema.
Who needs this, what one row is, and why the obvious build breaks
Every lesson starts from the consumer, because designing from the source outward is this domain's characteristic mistake.
An orders stream and a payments stream arrive independently. What has to be true before they can be joined, and what are you agreeing to throw away when you set the bound?
A payments reconciliation view, a fraud rule that needs the order and the authorisation together, and a revenue metric that only counts an order once money moved. All three need the *pair*, and all three will read whatever the join emits without being told which orders were dropped for arriving too far apart (Missing Rows).
One output row is one matched pair within the bound — one order joined to one payment, not one order. That is the sentence to write down, because the moment a second payment matches the same order the grain becomes one payment per order and every downstream sum doubles (Grain: What Does One Row Represent?).
Write the join the way you would in the warehouse: FROM orders JOIN payments USING (order_id). In SQL over two tables this is correct and finished. Streaming engines accept almost the same text, which is exactly why the mistake is so easy to make — nothing rejects it and it produces correct output all week.
The engine must retain every order it has ever seen, forever, in case a payment for it arrives tomorrow — and every payment, forever, in case its order is late. State grows monotonically with the *lifetime volume* of both streams and the job dies of a restore it can no longer complete (Streaming State).
- The engine must retain every order it has ever seen, forever, in case a payment for it arrives tomorrow — and every payment, forever, in case its order is late. State grows monotonically with the *lifetime volume* of both streams and the job dies of a restore it can no longer complete (Streaming State).
- An order arrives before its payment, so the join emits nothing; the payment arrives four seconds later and the join emits then. Whether a row appears at all depends on arrival interleaving, so a replay that reorders arrivals produces a different result set from the same input (Deterministic Replay: Making the Schedule Reproducible).
- A retried producer emits the payment twice. The join emits two rows for one order, the revenue metric doubles for that order, and no test on either input stream fails because each stream individually looks fine (Duplicate Rows).
- The order is enriched with the customer's tier by joining to a customer stream. The customer upgraded at noon; the order was placed at 09:00; the join used whatever tier had arrived last, so the order is attributed to a tier the customer did not have when they bought (Slowly Changing Dimensions).
- Payments for cancelled orders never match. Their buffered entries are the only ones that are never cleared, so the unmatched side of the state grows fastest precisely where the business logic says nothing will ever arrive (Streaming State).
- One
order_idis a test fixture reused by a load generator. Its buffer entry matches ten thousand payments and one task holds the entire join's state and runtime (Data Skew).
What is actually happening
- A join over bounded inputs can read both sides to completion and then decide. A join over unbounded inputs cannot, so it does the only thing available: it buffers each side in keyed state and emits when a counterpart arrives (Stateful Stream Processing).
- That buffer must be bounded by something, and the only honest bound is time. An interval join says a payment may match an order whose event time is within a stated window of it; entries outside that window are expired from state and can never match again (Event Time).
- The expiry is driven by the watermark, not the wall clock, which is what makes the result reproducible: replay the same events and the same entries expire at the same points, so the same pairs emit (Watermarks).
- A stream-table join — enrichment — is a different mechanism wearing the same word. One side is a stream of facts, the other is a keyed view of current-or-historical state; only the stream side triggers output, and the table side is looked up rather than buffered (Dimension Tables).
- When the table side is itself built from a stream — a compacted changelog of customers, prices or exchange rates — a temporal join looks up the version of the key that was valid at the fact's event time, which requires retaining versions rather than only the latest value (SCD Type 2 in Practice).
- Both sides are partitioned by the join key before the operator runs, so the two streams must be co-partitioned on that key. A join key that is not the partition key forces a shuffle, and a partition-count change on either side reshuffles the whole join (Event Keys and Partition Assignment).
- The unmatched side is where the semantics live. An inner interval join simply never emits the lonely row; an outer one emits it with nulls only once the bound has expired, which means an outer stream-stream join is always delayed by the full bound (Late Events).
Three different mechanisms wearing the same word
Ask what a join is in a streaming job and you will get three answers, all correct, describing three operators with different state behaviour, different failure modes and different costs. Conflating them is the single most productive source of streaming incidents, because the query text barely changes between them.
The distinction that matters is which side is buffered. In a stream-stream join both sides are held, so state grows with event volume and time. In a stream-table join only the keyed table is held, so state grows with the number of distinct keys and is otherwise flat. That one difference decides whether the job survives a year.
The second distinction is what triggers output. A stream-stream join emits when a counterpart arrives, from either side. A stream-table join emits only when the stream side arrives; an update to the table side changes future output and, crucially, does not retract past output. A consumer reading a metric built on enrichment is reading attributions made at many different moments in the dimension's history, and nothing in the table says so.
| Join | What is buffered | State grows with | What triggers output | How it goes wrong quietly |
|---|---|---|---|---|
| Stream-stream, bounded (interval / windowed) | Both sides, for the bound | Arrival rate x bound x row size | Either side arriving with a counterpart in the window | Pairs outside the bound are excluded by definition and counted nowhere |
| Stream-stream, unbounded | Both sides, forever | Lifetime volume of both streams | Either side, ever | Nothing fails for months; then a restore does not finish |
| Stream-table, current state | The keyed table only | Distinct keys | The stream side only | A lagging changelog enriches with old values at full speed |
| Stream-table, temporal (as-of event time) | Versioned table history | Distinct keys x versions retained | The stream side, once the version is known | A fact processed before its version arrives silently takes an earlier one |
| Table-table (changelog to changelog) | Both changelogs, latest per key | Distinct keys on both sides | Either side changing | Emits a retraction-and-update pair that a naive sink appends instead of merging |
The bound is the join
Walk one order through an interval join with a thirty-minute bound on event time. The order is buffered when it arrives; any payment whose event time falls inside the bound and which arrives before the watermark passes the bound's end will match it. Everything else will not, and the word for what happens to those is not "late" — it is excluded, because the join was defined to exclude them.
That distinction matters operationally. A late event is a record that missed a deadline and can be counted, side-outputted and corrected. An excluded event is a record the specification says does not belong, and by default nothing anywhere counts it. The most valuable line of code in a streaming join is usually the one that increments a counter when a buffer entry expires unmatched.
Notice P-4 below. Its payment happened thirty-eight minutes after the order — a genuine settlement, a real customer, real money — and it lands outside the bound. It will never join, the revenue metric will never include it, and the only evidence will be a slow divergence between the streaming number and the batch number computed over retained raw data.
| Event | Happened | Arrived | Lands in |
|---|---|---|---|
| O-1 order | 10:02 | 10:02 | buffered, retained until watermark passes 10:32 The order is the driving side. Buffering starts the moment it arrives, not when a payment is expected. |
| P-1 payment | 10:04 | 10:04 | matched to O-1, emitted immediately The healthy case: both sides inside the bound and inside the buffer, so the pair emits with no added latency. |
| O-2 order | 10:05 | 10:05 | buffered, retained until watermark passes 10:35 |
| P-2 payment | 10:06 | 10:21 | matched to O-2, emitted at arrival Fifteen minutes late in arrival but well inside the bound in event time. Arrival lateness and bound membership are different questions and only the second one is the join's business. |
| O-3 order | 10:09 | 10:09 | buffered, retained until watermark passes 10:39 |
| P-3 payment | 10:33 | 10:33 | matched to O-3, emitted at arrival Twenty-four minutes after its order — near the tail, still inside the bound. Widening or narrowing the bound moves this row in or out and nothing else changes. |
| P-4 payment | 10:40 | 10:40 | excluded — O-1 expired, and 10:40 is outside every open window A real payment for a real order, silently absent from the output. This is the row a match-rate trend exists to find. |
| O-4 order | 10:31 | 10:31 | buffered, still open — no payment yet An inner join emits nothing for this. An outer join emits it with nulls only once the watermark passes 11:01, so the absence of a payment costs the full bound in latency. |
Every clock label here is a teaching position on an event-time axis, not a measurement. What transfers is the rule: membership is decided by event time against the bound, expiry is decided by the watermark, and the two are separate mechanisms that fail separately.
The dimension as of when?
Enrichment looks like the easy join, and it is — right up to the question "as of when?". An order placed at 09:00 belongs to a customer who had a tier at 09:00, a price that was in force at 09:00 and an exchange rate quoted at 09:00. A lookup against current state answers a different question: what is that customer's tier *now*, at the moment the record happened to be processed.
In a warehouse this is the slowly-changing-dimension problem and it has a settled answer: keep validity intervals and join on them (SCD Type 2 in Practice). In streaming the same problem acquires a second half, because the dimension is not a table you can read at leisure — it is itself a stream, arriving at its own pace, through its own consumer, with its own lag. The version you need may not have arrived yet.
That is the coupling to be explicit about: a temporal join is correct only when the dimension stream is ahead of the fact stream in event time. Engines that support it enforce this by holding facts until the versioned side's watermark passes the fact's event time — which is why a stalled changelog stops a job that appears to have nothing to do with it, and why replaying a temporal join means replaying the dimension first.
When none of that is affordable, the honest fallback is to stop pretending. Enrich at the producer so the tier travels inside the order event and is true by construction, or accept a current-state lookup and *document it in the model* so the consumer knows the attribution is as-of-processing rather than as-of-event.
An order event needs the customer tier that applied when the order was placed. Where does that value come from?
when The producing service already knows the value at write time, and the value is part of what the business means by the event.
cost The event schema grows and the value is frozen. A later correction to the tier can only be applied by reprocessing every affected event (Reprocessing vs Retrying).
when The dimension changes rarely relative to the fact rate, or the consumer genuinely wants current attributes — an operational alerting view rather than a historical metric.
cost Attribution is as-of-processing. Replays produce different answers than the original run, and a lagging changelog degrades correctness with no drop in throughput (Deterministic Replay: Making the Schedule Reproducible).
when Historical attribution must be stable — revenue by tier, price at time of sale, rate at time of settlement.
cost Version retention on the dimension side, and a hard dependency on the dimension stream being ahead of the facts. A stalled changelog stalls the join (Watermarks).
when The dimension history already lives in the warehouse as an SCD2 table and the freshness requirement is hours rather than seconds.
cost Two systems now compute the same metric and must agree. The streaming output is less useful on its own, and the warehouse join becomes the authoritative one whether or not that was decided (Batch and Streaming Unification).
when Several consumers need different vintages of the same attribute, which is more common than it sounds.
cost The lookup problem is pushed to every consumer, and they will solve it differently. Expect the same metric to have several values (Two Dashboards, Two Numbers).
1-- 1. Current state. Fast, cheap, and answers "what is the tier now".2-- The tier attached to a 09:00 order is whatever had arrived by processing time.3SELECT o.order_id, o.order_time, c.tier4FROM orders AS o5JOIN customers_latest AS c ON o.customer_id = c.customer_id;6 7-- 2. Temporal join. Answers "what was the tier at the order's event time".8-- Requires versioned state on the customers side, and the customers9-- watermark must have passed o.order_time before the row can emit.10SELECT o.order_id, o.order_time, c.tier11FROM orders AS o12JOIN customers FOR SYSTEM_TIME AS OF o.order_time AS c13 ON o.customer_id = c.customer_id;14 15-- 3. Interval join between two fact streams. The predicate IS the bound;16-- remove it and this becomes an unbounded join that never expires state.17SELECT o.order_id, o.amount_minor, p.payment_id18FROM orders AS o, payments AS p19WHERE o.order_id = p.order_id20 AND p.pay_time BETWEEN o.order_time AND o.order_time + INTERVAL '30' MINUTE;Queries 1 and 2 differ by one clause and produce different history. Query 3 differs from an unbounded join by one predicate and differs from it in whether the job survives the year. In all three cases the dangerous edit is a deletion.
Engine support for versioned-table joins, state time-to-live and side outputs has changed materially between major versions of every engine named here. Confirm the semantics of your version — in particular whether unmatched entries are expired by watermark or by processing-time TTL — against current documentation before relying on them.
How to build it
Most important first.
- Bound every stream-stream join in event time, explicitly, in the query — an interval predicate, a windowed join, or a state time-to-live. Treat an unbounded one as a bug the same way you would treat a query with no
WHEREon a petabyte table (Stateful Stream Processing). - Derive the bound from the measured distribution of the gap between the two events, not from a round number. The right question is "how long after an order does a payment normally settle, and what fraction settles later than that" (Percentiles: Which One, and How Many Users Is That?).
- Count and route the expired-unmatched rows to a side output rather than dropping them. The count is the only evidence you will ever have that the bound is too tight, and the rows themselves are what a batch correction reprocesses (Reprocessing vs Retrying).
- Prefer a stream-table enrichment join to a stream-stream join wherever one side is genuinely slow-changing reference data. It holds bounded state proportional to the number of keys rather than to the volume of events (Data Marts).
- When enriching, state which time the lookup uses — processing time or event time — in the model's documentation, because the two produce different numbers and neither the schema nor the row count reveals which one ran (Semantic Changes).
- Deduplicate each input on its own business key *before* the join. A duplicate on either side becomes a fan-out through the join, and fan-out is far harder to detect downstream than a duplicate is upstream (Deduplication).
- Co-partition both inputs on the join key at the producer where you can. It removes a shuffle, and it removes the class of incident where a partition-count change on one topic silently redistributes one side of the join (Topics and Partitions).
- For anything requiring an unbounded lookback — lifetime attribution, first-touch, subscription history — accept that streaming is the wrong tool and compute it in batch against retained history (Batch vs Streaming Ingestion).
What this actually promises
Naming the guarantee you do not have is worth more than naming the one you do — everything downstream inherits the weakest promise in the chain.
- An interval join guarantees that every pair whose event times fall within the stated bound, and whose records both arrive before the watermark expires their buffers, is emitted. It guarantees nothing whatsoever about pairs outside the bound — they are not late, they are *excluded by definition*.
- Output ordering is per key, not global. Two orders processed by different tasks emit in whatever order those tasks make progress (Processing Time).
- A temporal join guarantees the versioned lookup was correct as of the event time provided the version history was already in state when the fact was processed. If the changelog is behind, the lookup silently uses an older version (CDC Ordering and Transaction Boundaries).
- Nothing here guarantees cardinality. The engine will happily emit ten rows for one order if ten payments matched, and the schema of the output is identical either way (Grain: What Does One Row Represent?).
- What is explicitly not guaranteed: completeness of the pairing. A join over two at-least-once streams is a fan-out risk on both sides and produces at-least-once output pairs unless you deduplicate (At-Least-Once Delivery).
Can I trust it?
A green pipeline is evidence that code ran. These four fields are the evidence that the data is right.
- The check that matters is a match-rate trend: matched pairs as a fraction of the driving side, per hour, per source. A join that starts silently excluding a segment shows up here first and nowhere else (Distribution Tests).
- Add a uniqueness assertion on the output at its declared grain. It is the only thing standing between a fan-out on one input and a doubled metric downstream (Data Tests).
- Both miss the case where the join is matching the *wrong* counterpart — a payment matched to a re-used order id, or an enrichment that used the current dimension value instead of the historical one. Those reconcile perfectly on count and are wrong on value (Reconciliation).
- Reconciling the streaming join against a batch recomputation over the same retained range is the only check that quantifies exclusion by the bound, because the batch job has no bound (Batch and Streaming Unification).
- An inner interval join is as fresh as the *later* of the two records. It can emit the instant the second side arrives, so a well-matched pair costs nothing beyond transport.
- An outer join, or any join whose downstream needs to know that no match occurred, is delayed by the full bound plus the watermark allowance. The absence of a payment cannot be observed until the time in which one could have arrived has passed (Watermarks).
- A stream-table enrichment join is as fresh as the stream side and as *correct* as the table side. If the changelog feeding the table lags, output continues at full freshness carrying stale attributes, which is the worst of the two failure shapes because nothing slows down (Stale Dashboards).
- Widening the bound to capture more pairs delays nothing for matched rows and delays everything for unmatched ones. The two halves of the output have different freshness and consumers rarely know that.
- Changing the interval bound changes which historical pairs would have been included, with no schema change and no marker in the output. It is a redefinition of the metric presented as a tuning parameter (Semantic Changes).
- Adding a field to either input is usually harmless; changing the *type* of the join key is not, because keyed state is stored under the serialised key and existing entries become unreachable rather than wrong (Schema Evolution).
- Switching an enrichment join from a current-state lookup to a temporal one changes every historical attribution the job will produce from that point forward, so the series has a discontinuity that looks like a business event (Two Dashboards, Two Numbers).
- Changing the partition count of either input topic re-keys the shuffle; existing keyed state was built under the old assignment and the job must be restarted from a rebuilt state or an earlier offset (Topics and Partitions).
- Join state is checkpointed with the rest of the operator state, so a restart resumes with the buffers intact and the pairing continues. This is the entire reason checkpointing exists here (Checkpointing).
- Recovering pairs lost to a too-tight bound is not a streaming operation. Reprocess the affected range in batch from retained raw data with a wider bound, and publish the corrections through the same merge path the streaming job writes to (Upserts and Merges).
- Make the output idempotent on the pair key — order id plus payment id — so a replay of the join re-emits rows that overwrite rather than accumulate (Idempotent Data Pipelines).
- A replay from the log reproduces the join only if watermarks are regenerated from event time and the enrichment side is replayed *ahead of* the fact side. Replaying both from the same offset usually enriches early facts with an empty table (Replay from the Log).
What can go wrong
- The unbounded join: no time predicate, state grows with lifetime volume, and the symptom is a checkpoint duration that creeps upward for months before anything fails (Streaming State).
- Silent exclusion: the bound is tighter than the real settlement tail, a stable few per cent of pairs never emit, and the number looks like a plausible business figure (Missing Rows).
- Fan-out: a duplicate on one input multiplies through the join, and the doubled metric is discovered by a finance team rather than a monitor (Duplicate Rows).
- Enrichment with the wrong version: the join used current state where event-time state was meant, so every historical attribution drifts as the dimension changes underneath it (SCD Type 2 in Practice).
- Key skew: one join key holding most of the buffered state, so one task decides checkpoint duration, restore time and end-to-end latency for the whole job (Data Skew).
- The mitigation failing: widening the bound to stop excluding pairs multiplies retained state, which lengthens checkpoints and restores, which turns a correctness problem into an availability one (Checkpointing).
- Replay ordering: the changelog side replayed at the same rate as the fact side, so the first hours of a reprocess enrich against a table that is still filling (Reprocessing vs Retrying).
- "A streaming join is a SQL join that runs continuously." A SQL join can see both inputs in full. A streaming join sees a moving window of each and its result is a function of that window — the same query text means a different thing (Stream Processing).
- "The join is not emitting, so the data must be missing." Far more often the counterpart arrived outside the bound, or the watermark has not advanced enough to release the unmatched side. Check the expired counter before you go looking upstream (Watermarks).
- "Enrichment joins are cheap, so use them everywhere." They are cheap in state and expensive in coupling: the enriched output is only as correct as the changelog that fed the table, and that dependency is invisible in the output schema (Data Lineage).
- "We joined on the primary key, so it is one-to-one." It is one-to-one in the source database. In a stream of *changes* to that database, one key produces many records, and a join against it fans out unless you reduce to the latest version per key first (What a CDC Event Contains).
- "Setting a state time-to-live makes the join safe." It makes it survivable. It also silently changes the join's semantics from "matched within a bound" to "matched before an unrelated expiry", which is much harder to explain to a consumer (Streaming State).
Operating it
- Buffered entries and state size per join operator, split by side. The asymmetry is the diagnosis — a growing left buffer with a stable right one says the right side stopped arriving (Pipeline Metrics).
- Match rate and expired-unmatched count per side, per unit time. These two together tell you whether the bound is right; either alone tells you very little (Pipeline Observability).
- Output rows per driving-side row. A ratio above one is fan-out and it should be alerted on directly rather than inferred from a downstream metric (Quality Alerting).
- The distribution of the event-time gap between matched pairs, retained as a histogram. It is the evidence that sets the bound and the evidence that the bound is drifting (Percentiles: Which One, and How Many Users Is That?).
- Per-key state size for the top keys, which is how skew in a join is found before it becomes a restore failure (Data Skew).
- At 10x event volume with the same bound, join state grows roughly 10x. The bound, not the volume, is the lever that is actually under your control.
- At 10x key cardinality the state grows with distinct keys held in the window and the per-key overhead starts to dominate for small rows — a join over user ids behaves very differently from a join over country codes (Partition Cardinality).
- At 100x, an interval join over two high-volume streams usually stops being the right shape entirely. The common resolution is to enrich as close to the producer as possible so that only one stream reaches the platform (The Event-Driven Data Platform).
- Adding a third stream to the join multiplies buffering rather than adding to it, because each pairwise stage holds its own state and the intermediate result is itself a stream that must be buffered (Stateful Stream Processing).
- The dominant cost is retained state: roughly the arrival rate of each side multiplied by the bound, multiplied by the serialised row size. Every one of those three is a design decision (Streaming State).
- The second cost is the shuffle that co-partitions both inputs on the join key. It is paid on every record on both sides, continuously, for the life of the job (The Shuffle).
- Checkpoint cost scales with state, so a wider bound is paid twice — once in memory or on disk, and again in every checkpoint and every restore (Checkpointing).
- A stream-table join is dramatically cheaper in state terms because the table side is bounded by key count rather than by event volume, and that is usually the strongest argument for restructuring the problem (Data Marts).
- A tight bound buys small state, fast checkpoints and quick restores, and costs completeness — silently, on the tail, for the segment with the slowest settlement.
- A wide bound buys completeness and costs state, checkpoint duration, restore time and the latency of every unmatched decision. There is no setting that avoids both costs.
- A temporal join buys historically correct attribution and costs version retention on the dimension side plus a strict requirement that the changelog is ahead of the facts — a coupling that turns one job's lag into another job's wrong answer.
- Enriching at the producer buys the cheapest possible join and costs flexibility: the enrichment is baked into the event, so a corrected dimension value can never be applied retroactively without reprocessing (Reprocessing vs Retrying).
Dataset review questions
This lesson uses the shared review exercise.
Where this applies
Almost nothing here is universal. These labels say what each claim is specific to, and where a different engine, format, warehouse or scale would differ.
- GENERALThe requirement that an unbounded-to-unbounded join carry an explicit time bound follows from the inputs being unbounded, not from any engine: a system with no bound must retain every record forever to remain correct, which is a statement about the problem rather than the implementation.
- ENGINE-SPECIFICFlink SQL expresses this as an interval join with a time predicate, a windowed join, or a temporal join via FOR SYSTEM_TIME AS OF, and expires state by watermark; Kafka Streams offers a windowed stream-stream join plus KTable lookups whose semantics depend on the co-partitioning it does not verify for you; Spark Structured Streaming requires a watermark on both sides before it will accept a stateful stream-stream join at all. The bound is universal; where you declare it and what happens if you forget are not.
- SIMPLIFIEDThe timeline below treats each event as arriving once and in a single partition. Real joins run across many partitions whose watermarks advance independently, so the effective expiry is the minimum across inputs and a single lagging partition delays expiry for every key in the operator.
Where the depth lives
This domain teaches how data moves and how you know it arrived intact. It hands the rest off by name.
- — Distributed Systems owns why the two streams cannot simply be synchronised: there is no shared clock and no way for a receiver to distinguish a slow producer from a silent one, so the bound is an engineering choice rather than a derivable constant.
- — DevOps / Production Engineering owns the deployment side of changing a join bound — the state migration, the parallel run against the old semantics, and the fact that a query change here silently redefines a published metric.