# DDIA Ch. 12: a naive stream-table join time-travels — replay changes the answer ## 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. 1. **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. 2. **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. 3. **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. 4. **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 Deterministic, so the totals reproduce exactly: - 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. ## Steps ### 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? **Observe.** ``` 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? **Observe.** ``` 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? **Observe.** ``` 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 1. **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. 2. **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. 3. **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.