Stream-table joins and enrichment

Which enrichment option is correct, and why does correctness need a temporal join on a dimension stream?

Stream-table joins: enrichment and async I/O

Enrichment means adding fields from a slowly changing reference table (here, restaurant data) to each stream record. It is a different problem from joining two streams. There is a clear ladder of implementations.

Option 1: per-record remote lookup

The simplest option: for each record, call a remote service or database. It caps throughput at concurrency ÷ RTT. That is Little's Law in disguise.

The fix is async I/O: allow N requests in flight at the same time. Throughput is multiplied by N until the remote service becomes the bottleneck. With N = 50: 50 ÷ 0.005 = 10,000 records/s per task.

Async brings a choice:

Unordered completion gives maximum throughput, and breaks per-key ordering: a later record may be emitted before an earlier one of the same key.
Ordered completion keeps order by buffering finished results until all earlier ones finish. That brings back head-of-line blocking (one slow item blocks all items behind it), and the in-flight buffer becomes checkpointed state.

Further points:

A cache in front of the lookup adds the problem of cache coherence, including stale reads.
Any external lookup is non-deterministic under replay. The remote table may have changed between the first run and the replay. So this option is incompatible with exactly-once state semantics (Exactly-once).

Option 2: broadcast the table

Send the whole table into every instance's operator state. This is deterministic and has no RTT. The memory cost is bounded by table size × parallelism. It is fine for a few MB, and useless at a few GB.

Option 3: co-partition the table as a keyed changelog stream

Consume the table's CDC feed (every insert, update and delete) with the same key as the fact stream, and hold it as keyed state. This scales and is deterministic. It forces the question that everyone skips:

Joining against whatever is currently in state is a processing-time join. It is non-reproducible: replay the same input tomorrow and you may get different output, because the dimension rows changed in between. That is Ana's third symptom (the re-run with different numbers). Her first run used the commission rate that was current at that moment. The re-run used today's rates.

A temporal join uses the dimension version that was valid at the fact's event time. This requires:

retaining multiple versions per key in state, and
a watermark on the dimension stream too, so the job knows when it has seen all versions up to the fact's time.

This is the bitemporal model (data has both a valid time and a recorded time) materialized as operator state. It is the correct but expensive option.

Two side effects of the correct option

The dimension stream's watermark now gates the fact stream. An idle dimension topic (restaurants change rarely!) stalls the whole job, unless idleness is handled explicitly (idle channels).
Point-in-time correctness here is the same mechanism as preventing training-serving skew in machine-learning feature stores. (That is the problem where a model is trained on feature values as they were, but served with values as they are now, or the reverse.) Same problem, different vocabulary.

The ladder in one table

Option	Throughput limit	State cost	Deterministic on replay?	Main danger
Per-record lookup	concurrency ÷ RTT	none	no	RTT cap, non-determinism
Async I/O + cache	N × (1 ÷ RTT), remote capacity	in-flight buffer	no	ordering choice, staleness
Broadcast table	local	table size × parallelism	yes	too big for large tables
Changelog as keyed state (processing-time)	local	table size	no (version is "now")	non-reproducible
Temporal join	local	many versions per key	yes	second watermark, idle-source stalls

Case, resolvedConstructed case · Ana at Ridelo

2% of deliveries missing from the join

Joining orders with courier pings silently dropped 2% of matches, a per-record lookup capped throughput at 200 records a second, and replays produced different billing numbers. Where it started: Joins and enrichment →

The 2% shortfall. She measures the real delay between order time and the courier ping that matched, finds that 2% of pings arrive more than the 10-minute bound, and has two choices: widen the interval (more state, per the rate × time arithmetic) or accept the loss and document it. She widens to 15 minutes, costing 50% more join state, and adds a counter of "records dropped because the other side was evicted" so the loss becomes visible.
The 200 records/s. She switches the lookup to async I/O with 50 requests in flight and unordered completion, since downstream only aggregates per restaurant and does not need per-key order. Throughput per task rises to about 10,000 records/s. She keeps this only for data that can tolerate non-determinism, and routes the billing path through option 3.
The different numbers. For billing she consumes the restaurant CDC feed as keyed state, uses a temporal join on event time, and enables idleness handling on the restaurant topic. The re-run of last month now reproduces the original result.