Stream-table joins and reference versions

Last updated: 6 Oct 20265 min read

tutorial

A stream-table join enriches each event with reference data, but correctness depends on which version of that data the event is allowed to see.

Choose a time rule before joining

A current-value lookup answers what the merchant category is now. A temporal lookup answers what it was when a payment occurred. Both are valid for different products. For historical risk analysis, use event time and a version interval; otherwise a merchant recategorization can rewrite past cohorts during replay. An as-of join has the same requirement in a warehouse.

Make update order explicit

The event stream and reference change stream can arrive in either order. A payment at hour 47 may be processed before the category update effective at hour 46 reaches this worker. Choose to wait for reference completeness, emit unknown and repair later, or use a bounded temporal store with delayed finalization. The choice changes latency, state and historical accuracy.

Reject overlapping versions

For one merchant, version intervals should be half-open and non-overlapping. If two rows match one event time, enrichment must fail or quarantine the key, never arbitrarily choose whichever arrived last. Preserve source namespace and stable merchant identity. Conformed identity prevents two systems with the same numeric ID from merging.

Control fanout and missing matches

A join intended to yield one output per payment can multiply records when a lookup key has duplicate active rows. Assert output count equals input count and record the unresolved-reference rate. A valid but late dimension can use an explicit unknown member and later repair. Invalid keys should be quarantined, not guessed from a display name.

Rehearse a historical correction

Deliver category A, a payment, then category B with an effective time before the payment. Under a temporal contract, a repair may revise the payment enrichment; under an immutable-at-first-seen contract, it should not. Publish the selected rule and its audit trail so readers can explain why a cohort moved.

Implementation

versions = [
    {"merchant": "m-47", "start": 40, "end": 48, "category": "retail"},
    {"merchant": "m-47", "start": 48, "end": None, "category": "marketplace"},
]
payments = [{"id": "pay-47", "merchant": "m-47", "event_hour": 47}]

def category_at_time(payment, reference_rows):
    matches = [row["category"] for row in reference_rows
               if row["merchant"] == payment["merchant"]
               and row["start"] <= payment["event_hour"]
               and (row["end"] is None or payment["event_hour"] < row["end"])]
    if len(matches) != 1:
        raise ValueError("reference version is missing or overlaps")
    return matches[0]

assert category_at_time(payments[0], versions) == "retail"

Performance and operating cost

The reference scan is O(V) per event for V versions; an indexed temporal store reduces lookup work but stores each retained version until its supported replay horizon ends. Stream output still needs a one-to-one cardinality check. Waiting for reference completeness costs latency; optimistic output costs repair work and versioned publication.

Common Mistakes

Do not use the current reference row when historical truth is required.
Do not let duplicate active versions multiply events.
Do not hide a missing reference behind a guessed category.