study › DDIA › Chapter 12 · Stream Processing

A Naive Stream-Table Join Time-Travels — Replay Changes the Answer

Join historical sales against a tax-rate table that changed from 8% to 10% on day 100. A naive join reads the table's current rate, so it overcharges old sales — reporting $100.00 where the correct answer is $94.00 — and re-running the same sales on a different day gives $80.00, a $20.00 drift. A point-in-time join pins each sale to the rate in effect at sale time and reports $94.00 on every run.

Concept

Join a stream of historical sales against a tax-rate table that changed over time — 8% before day 100, 10% from day 100 on. A naive stream-table join looks up the table's current state at processing time, so when you reprocess last year's sales it applies this year's rate: old sales are overcharged, and the total comes out $100.00 when the point-in-time-correct answer is $94.00. Worse, the naive job isn't even stable: re-run the identical sales on a day when the current rate was still 8% and it reports $80.00 — same input, same code, a $20.00 drift, purely because of when you ran it. Pin each sale to the rate in effect at sale time (a slowly-changing-dimension lookup) and the total is $94.00 whether you run it today or next year: correct, and deterministic on replay.

Provenance

Designing Data-Intensive Applications, 2nd ed. (Kleppmann & Riccomini, 2025), Chapter 12 — Stream Processing, §"Time dependence of joins" (stream-table joins over slowly-changing dimensions). Paraphrasing the claim under test:

When the table you join against changes over time, the join result depends on which version of the table row you use — and in a data warehouse this versioned dimension is known as a slowly changing dimension (SCD), addressed by giving each version its own identifier.

The book's point is that a stream-table join is only well-defined once you decide which version of the table row a given event should see. Look up the current row and the join silently depends on wall-clock time; look up the version that was in effect at the event's timestamp and it becomes a pure function of the data. Here you watch both.

Prerequisites

The surprise is that a join can quietly read the clock as an input. Each line notes where it shows up; a free resource follows.

Stream-table join (enrichment) — each event in a stream is augmented with a matching row from a table (here: each sale gets its tax rate). It shows up as naive_join / pointintime_join looking a rate up per sale.
Slowly-changing dimension (SCD) & effective-dating — the table row isn't constant; it has versions each valid for a date range (8% from day 0, 10% from day 100). It shows up as the TAX_TABLE list of (effective_from, rate) versions.
Point-in-time / temporal join (carries the result) — the correct lookup returns the version whose effective range contains the event's own timestamp, not the latest version. It shows up as rate_as_of(sale.day) vs current_rate(run_day), and is the entire difference between $94 and the drifting $80/$100.
Determinism under reprocessing — replay/reprocessing is only safe if the job is a pure function of its inputs; a current-state lookup injects run-time state as a hidden input. It shows up as the same sales giving $80 or $100 depending on the run day.

Where to learn the prerequisites Stream-table joins & the time-dependence of joins (#1–#3): DDIA §"Stream joins" / §"Time dependence of joins". Confluent's "Crossing the Streams — Joins in Apache Kafka" post walks stream-table joins concretely, free online. Slowly-changing dimensions (#2): Kimball's SCD write-up (the "Type 2" versioned-row technique) — searchable as "Kimball slowly changing dimension type 2", free. Determinism under reprocessing (#4): DDIA §"Reprocessing data" — replay is the payoff of an append-only log, and it only pays off if the transformation is deterministic. If only one is new, make it #3 — a point-in-time lookup is the whole fix.

Environment these numbers came from Deterministic, so the totals reproduce exactly. Captured on macOS 26.5.2 (Darwin 25.5.0), arm64, Python 3.15.0a8, standard library only — no Docker, no deps. No wall clock, no randomness: the "day the job runs" is passed in explicitly, so both runs reproduce exactly. Money is integer cents, rates are basis points, so every product is exact. Fixed workload: four sales totalling $1,000.00; tax table 8% (day 0), 10% (day 100).

Mental model

A stream-table join enriches each event by reading a row from a table. The only question is which version of the row it reads.

Naive (current-state) join — read the table row live, as it stands when the job runs. If the row is mutable, the answer depends on the wall clock: reprocess the same events on a different day and you get a different result. The run time is a hidden input the events never carried.
Point-in-time (temporal / SCD) join — read the version of the row that was in effect at the event's own timestamp. The events already carry their timestamps, so the lookup is a pure function of the data — no clock, same answer every replay.

Same sales, same table, same two-line lookup — the difference is rate_as_of(day) versus current_rate(run_day). The whole thing is one script, code/stream_table_time_travel.py.

Setup

cd study/ddia/ch12/code
python3 stream_table_time_travel.py

Do not read the output yet — make each prediction first. The table changed from 8% to 10% on day 100; the four sales fall on days 40, 80, 120, 160.

Step 1 · the naive join, run today

Predict. The job runs today (day 300), when the current tax rate is 10%. It joins each of the four sales — $100, $200, $300, $400 — against the table's current rate. What total tax does the naive join report?

What total does the naive join report?

Step 1 NAIVE join, job run on day 300 (today). Looks up the table's CURRENT rate for EVERY sale. ================================================================== current rate on day 300 = 10% (applied to all sales) S1 day 40 amount $100.00 x 10% -> tax $10.00 S2 day 80 amount $200.00 x 10% -> tax $20.00 S3 day 120 amount $300.00 x 10% -> tax $30.00 S4 day 160 amount $400.00 x 10% -> tax $40.00 NAIVE total tax = $100.00

$100.00. Every sale — including the two from before the rate change — got taxed at today's 10%. The two old sales ($100 and $200) are overcharged: they happened while the rate was 8%.

Step 2 · the same job, re-run last year

Predict. Now re-run the exact same job on the exact same sales, but on day 90 — a day when the current rate was still 8% (the change to 10% hasn't happened yet). Same code, same input. Will the total match Step 1's $100.00, or change? To what?

Does the same input give the same total?

Step 2 SAME naive job, re-run on day 90 (last year). Same sales, same code -- only the calendar day differs. ================================================================== current rate on day 90 = 8% (applied to all sales) S1 day 40 amount $100.00 x 8% -> tax $8.00 S2 day 80 amount $200.00 x 8% -> tax $16.00 S3 day 120 amount $300.00 x 8% -> tax $24.00 S4 day 160 amount $400.00 x 8% -> tax $32.00 NAIVE total tax = $80.00 >> Same input, two run days, two answers: $80.00 vs $100.00 >> Reprocessing DRIFT = $20.00 -- the join is NOT deterministic on replay.

$80.00 — a $20.00 drift from Step 1. Nothing about the sales changed; the only difference is the calendar day the job ran. The naive join reads the table's current state, and "current" moved. The job is not a function of its inputs — it smuggles in the wall clock.

Step 3 · the point-in-time join

Predict. Fix it: match each sale to the rate that was in effect at that sale's day — 8% for the day-40 and day-80 sales, 10% for the day-120 and day-160 sales. What total does this report, and will it change between a run on day 90 and a run on day 300?

What total does the point-in-time join report — and is it stable?

Step 3 POINT-IN-TIME join: each sale x the rate effective at ITS day. ================================================================== S1 day 40 amount $100.00 x 8% -> tax $8.00 S2 day 80 amount $200.00 x 8% -> tax $16.00 S3 day 120 amount $300.00 x 10% -> tax $30.00 S4 day 160 amount $400.00 x 10% -> tax $40.00 CORRECT total tax = $94.00 re-run on day 90 -> total tax = $94.00 re-run on day 300 -> total tax = $94.00 >> Both runs IDENTICAL: $94.00 -- deterministic on replay.

$94.00, on both run days. The two old sales use 8%, the two new ones use 10% — each sale sees the rate that was real when it happened. The run day is passed in and ignored, so replay is exact.

What you should see

Naive join, run today (day 300, current rate 10%): $100.00 — old sales overcharged.
Same naive job, re-run last year (day 90, current rate 8%): $80.00 — same input, a $20.00 reprocessing drift; the join is non-deterministic.
Point-in-time join: $94.00, identical on both run days — correct and stable.

Why

Reprocessing — the whole promise of keeping events in a log — is only correct if the job is a deterministic function of its inputs: feed the same events in and you must get the same output out, whenever you run it. A stream-table join reads two things, the event and a table row, and multiplies them. The event is immutable — the sale for $100 on day 40 is fixed forever. The table row is not: the tax rate has versions. So the join's determinism hinges entirely on which version it reads.

The naive join reads the current version — the row as it stands at the instant the job runs. That instant is not part of the input; it's the wall clock. So the naive join is really a function of (events, run-time), and run-time is a hidden input the events never carried. Change the run day and the output changes: $100 today, $80 last year. Same sales, different answer — the definition of non-deterministic replay. The overcharge and the drift are the same bug seen two ways: applying a later version of the row to an earlier event.

The point-in-time join removes the hidden input. Instead of "the current rate" it asks "the rate in effect on the sale's own day" — rate_as_of(sale.day). The sale already carries its day, so the lookup reads nothing but the input: it's a pure function again. Every replay, on any run day, retraces the same versions and lands on $94. That is the SCD fix in one line: don't look up the dimension's present, look up the dimension's history at the event's timestamp — equivalently, denormalize the effective rate (or its version id) into the event at capture time so the join has nothing time-varying left to read.

The boundary — the bug needs a mutable dimension and reprocessing If the tax rate had never changed, current_rate and rate_as_of would return the same number and both joins would agree — a constant dimension makes current-state and point-in-time identical, and there is nothing to drift. Equally, if you only ever processed each event once, at the moment it arrived, "current" would coincide with "at event time" and you'd never notice. The anomaly needs both ingredients: a dimension that changes and events processed (or reprocessed) at a different time than they occurred. That combination is exactly what an event log invites — so this is a hazard you take on the moment you start replaying.

Go deeper

Versioned state stores. Kafka Streams added versioned state stores and temporal stream-table joins precisely for this: the join looks up the table value as of the record's timestamp. Map rate_as_of(day) onto that API and see that this exercise's fix is a first-class feature there.
Bitemporal modeling. There are two times here — when the sale happened and when the rate became effective. Real systems add a third (when a row was recorded, so a late correction can be replayed faithfully). Add a recorded_on axis to the table and watch a backdated rate correction reprocess correctly.
Rebuild the table from a log. Feed the rate changes as their own stream and fold them into the table view with log compaction (the Ch. 11/12 log-as-source-of-truth idea), then run the point-in-time join against the rebuilt table — the join stays correct because the history survives compaction, only superseded-and-shadowed versions are dropped.

Sources: DDIA 2e, Ch. 12, §"Time dependence of joins", §"Stream joins", §"Reprocessing data" · Kimball, "Slowly Changing Dimensions" (Type 2).