py-20261008-temporal-join-watermark-retraction-late-data

Answer

Done. The verified, self-contained deliverable is at ~/solution/SOLUTION.md (with runnable interval_join.py and test_interval_join.py). Summary:

Title

Event-Time Interval Join with Watermarks, Retractions, Bounded State, and Exactly-Once Recovery.

Root-cause analysis

A naive implementation fails for five interacting reasons: 1. Global/arrival-time watermark → head-of-line blocking across keys/partitions and impossible late matches. Fix: per-(partition,key) max event time, per-key watermark = min across live partitions, idle partitions excluded. 2. Unbounded state → OOM. Fix: retain right events only until wm > r.ts, left until wm > l.ts + W; shed events older than lateness. The subtle leak found by testing: too-late events must not be written to the dedup table, and eviction must still run on the late path. 3. No retractions → stale aggregates. Fix: retract(old) + insert(new) on every change. 4. Duplicate delivery → double counting / spoofed watermark jumps. Fix: dedup on (side, eid) before advancing clocks; idempotent sink. 5. Non-idempotent recovery → duplicated output after restore. Fix: deterministic output IDs (epoch, seq) regenerated byte-identically on replay from the checkpoint offset. Plus: watermarks must be monotonic, so each key watermark is clamped to a non-decreasing floor (otherwise a late-joining lagging partition silently invalidates already-shed data).

Exact fix

interval_join.py implements the operator (IntervalJoin), idempotent Sink, and a brute_force oracle. test_interval_join.py contains the verification suite. Run:

cd ~/solution && python3 test_interval_join.py

Verification (actual output)

test_random_matches_bruteforce: OK
test_duplicate_delivery: OK
test_retraction_protocol: OK
test_per_key_watermark_isolation: OK
test_partition_skew_and_idleness: OK
test_checkpoint_restore_exactly_once: OK (deduped 4)
test_bounded_state_adversarial: OK (buffered=6, seen=6, dropped=4994)
test_bound_guard_trips: OK
test_memory_bound_scales: OK (peak=31 <= bound=44)
test_duplicate_cannot_advance_watermark: OK
test_checkpoint_restore_matrix: OK (69 crash scenarios)

ALL TESTS PASSED

The tests prove order-independence vs. brute force, duplicate/idempotency, exact retraction sequences, per-key watermark isolation, partition skew + idle + monotonic regression guard, 69 crash/checkpoint recovery scenarios, and hard bounded memory under 5000 adversarial decreasing timestamps.

Adversarial complexity

• Space: O(P·λ·(L+W) + P·K) (retained time span is L+W); max_buffer is a fail-fast guard, not a spill.
• Time: O(N·λ·(L+W)) worst case per hot key, Θ(J) output for J matches. The genuine worst case (N×N events in one window) is Θ(N²) work/output — inherent to a materialized interval join; the recommended upgrade is a per-key timestamp index for O(N log N + J).
• Adversarial cases (decreasing timestamps, all-late flood, duplicate flood, stalled/resuming partitions, crash-at-any-offset) are tabulated with their bounds in §5 of SOLUTION.md.