Count event-time windows with late arrivals

A baseline groups by arrival time and immediately emits on the local clock's ten-unit boundary. A delayed timestamp 2 arriving after timestamp 10 then enters the wrong window or is lost prematurely. Another tempting baseline closes by maximum observed event time: one far-future outlier can close valid earlier windows.

Store counts keyed by window start. On add, derive start with floor division and compare its close deadline against the current watermark before incrementing. On watermark advancement, validate monotonicity, find active windows whose deadline is reached, sort them by start, return their counts, and remove them from state. Missing/empty windows are not materialized or emitted.

stored windows have not yet closed under the latest watermark, and their counts equal accepted arrivals assigned to those exact half-open intervals. Because the watermark never decreases, a removed window can never be recreated: the add-time boundary test rejects it without keeping unbounded tombstones. This is local single-run finality, not durable delivery or deduplication.

Add costs expected O(1). With A active windows and F closing windows, advancement costs O(A + F log(F+1)) time and O(F) auxiliary/output space. State is O(A), which is not automatically bounded by lateness: arbitrary future timestamps or a stalled watermark can retain arbitrarily many windows. A heap can reduce closure scans, but introduces additional indexing and cleanup work.

Follow-up 1 — emit early and accept corrections. Predict the new output if count 1 was emitted at watermark 10 and event 2 arrives afterward. Send a versioned replacement count 2 or delta +1, then a final marker at 15. Consumers must know which update model applies and deduplicate/reconcile retries.

Follow-up 2 — multiple input partitions. A global watermark normally depends on each active partition's progress, often their minimum. Idle partitions need an explicit exclusion/reactivation policy; taking the maximum closes windows ahead of slow partitions. Future-time skew limits and backpressure can bound retained state.

Senior depth tests exact boundaries and arrival permutations. Lead depth names watermark authority, idle-source policy, restart state, and output retry semantics.

Reference: solution.py (download file, source below); tests cover negative timestamps, event order, late rejection, duplicate counts, empty windows, and monotonic progress.

"""Explicit-watermark tumbling counts with a fixed allowed-lateness duration."""


class TumblingCounter:
    def __init__(self, width, allowed_lateness=0):
        if not isinstance(width, int) or width <= 0:
            raise ValueError("positive integer width required")
        if not isinstance(allowed_lateness, int) or allowed_lateness < 0:
            raise ValueError("nonnegative integer lateness required")
        self.width, self.allowed_lateness = width, allowed_lateness
        self.watermark = None
        self._counts = {}

    def add(self, timestamp):
        if not isinstance(timestamp, int):
            raise ValueError("integer event timestamp required")
        start = (timestamp // self.width) * self.width
        close_at = start + self.width + self.allowed_lateness
        if self.watermark is not None and close_at <= self.watermark:
            return False
        self._counts[start] = self._counts.get(start, 0) + 1
        return True

    def advance_watermark(self, watermark):
        if not isinstance(watermark, int):
            raise ValueError("integer watermark required")
        if self.watermark is not None and watermark < self.watermark:
            raise ValueError("watermark cannot decrease")
        self.watermark = watermark
        closed = sorted(start for start in self._counts
                        if start + self.width + self.allowed_lateness <= watermark)
        return [(start, start + self.width, self._counts.pop(start)) for start in closed]

python -m unittest discover -s curriculum/04-scale-and-evolution/01-data-at-scale/problems/40-event-time-windows -p 'test_*.py'