Streaming Merge Join with Missing Keys

Worked Solution

How to Think About It: This is a classic stream-processing / merge-join problem -- the kind you would face building a real-time data pipeline for a trading floor (think merging a market-data feed with an order feed). The core tension is between *completeness* and *latency*: you want to join on $(t, \text{loc})$, but one side might be missing or late. The offline formulation here fixes the fill strategy: last-value-carried-forward (LVCF) per location.

The contract precisely. The output has one record per distinct (t, loc) key appearing in either stream. For each key and each side, the value is the most recent event at that location with event_t <= t (ties on the same timestamp resolved by input order -- the last event wins), or None if no such event exists. Records are sorted by (loc, t). Note the value is carried forward *within a side* too: if a location has temperature events at t=1 and t=3, then a humidity-only key at t=2 reports the t=1 temperature.

Algorithm:

1. Index each stream as loc -> list of (event_t, order, value) and sort each list by (event_t, order). Recording the arrival order makes tie-breaking (last write wins at a timestamp) deterministic.
2. Collect the set of output keys as the union of (loc, t) across both streams.
3. For each key (loc, t), scan that location's sorted list and take the value of the latest entry with event_t <= t (LVCF); this naturally yields the last-in-input value at the maximal qualifying timestamp. Missing side -> None.
4. Emit [t, loc, temp, humid] for every key, sorted by (loc, t).

Code:

```python def merge_join(temp_events, humid_events): # Join two event streams on (t, loc) with last-value-carried-forward fill. # Output keys = union of (t, loc) across both streams (deduplicated). # For each key, each side's value is the most recent event at that loc with # event_t <= t (ties broken by input order: the last-arriving event wins), # or None if no such event exists. Records sorted by (loc, t).

def build(events): # loc -> list of (event_t, order, value), sorted by (event_t, order) d = {} for order, (t, loc, val) in enumerate(events): d.setdefault(loc, []).append((t, order, val)) for loc in d: d[loc].sort() return d

temp_d = build(temp_events) humid_d = build(humid_events)

keys = set() for t, loc, _ in temp_events: keys.add((loc, t)) for t, loc, _ in humid_events: keys.add((loc, t))

def lvcf(d, loc, t): lst = d.get(loc) if not lst: return None best = None for event_t, _order, val in lst: # sorted ascending by (event_t, order) if event_t <= t: best = val else: break return best

result = [] for loc, t in sorted(keys): tv = lvcf(temp_d, loc, t) hv = lvcf(humid_d, loc, t) result.append([t, loc, tv, hv]) return result

```

Practical Considerations:

• Watermarking (streaming variant): in a true streaming system you would advance a watermark $W$ = min of the two streams' latest timestamps minus a slack factor, emit joins for keys with $t \le W$, and evict state older than $W - \text{grace}$. Flink and Kafka Streams call this "event-time processing with allowed lateness." The offline function above assumes all events are already in hand, so it can compute every LVCF fill exactly.
• Ties / duplicates: multiple events at the same (t, loc) are resolved last-write-wins via the recorded input order.
• LVCF vs. interpolation: LVCF is simple and conservative. For continuous signals, linear interpolation between the two nearest timestamps may be more accurate but adds complexity.

Complexity:

• Building and sorting the indexes: $O(N \log N)$ where $N$ is the number of events.
• The LVCF scan shown is linear per key for clarity; with bisect on the sorted per-location list it is $O(\log n)$ per lookup, giving $O((K) \log n)$ over $K$ output keys.
• Space: $O(N)$ for the buffered per-location state.