Skip to content

Time, watermarks, windows and CEP

The event-time subsystem: how the engine tracks event-time progress through watermarks, fires keyed windows, and matches patterns over per-key event sequences.

Overview

clink processes records on the event-time axis: the time at which an event occurred in the source system, not when the engine sees it. Event time is carried per record as a 64-bit millisecond count since the Unix epoch (EventTime). A Watermark is the engine's lower bound on event-time progress on a stream, and it is what lets stateful operators decide that "no further records earlier than t will arrive", so a window can fire or an expired pattern partial can be pruned. Watermarks are assigned by a dedicated operator from a pluggable strategy, then travel in-band alongside data through the operator DAG. Window operators and the CEP operator consume those watermarks to drive their firing and eviction logic.

This page covers the watermark machinery, the four window operators (tumbling, sliding, session, evicting), the trigger and evictor abstractions, late-data handling, and the NFA-based complex-event-processing operator.

Where it lives

Path What
include/clink/time/event_time.hpp EventTime (millis-since-epoch wrapper)
include/clink/time/watermark.hpp Watermark (timestamp + idle flag)
include/clink/time/watermark_strategy.hpp WatermarkStrategy<T> and the built-in strategies
include/clink/time/alignment_group.hpp AlignmentGroup + registry for watermark alignment
include/clink/operators/watermark_assigner_operator.hpp WatermarkAssignerOperator<T>
include/clink/runtime/multi_input_alignment.hpp MultiInputAlignment (watermark min + barrier alignment across N inputs)
include/clink/operators/window_trigger.hpp Trigger, TriggerContext, TimeWindow, built-in triggers
include/clink/operators/window_state.hpp WindowEntry, SessionRow, BufferEntry and their codecs
include/clink/operators/window_evictor.hpp Evictor, CountEvictor, TimeEvictor
include/clink/operators/tumbling_window_operator.hpp TumblingWindowOperator<Key,Value,Agg>
include/clink/operators/sliding_window_operator.hpp SlidingWindowOperator<Key,Value,Agg>
include/clink/operators/session_window_operator.hpp SessionWindowOperator<Key,Value,Agg>
include/clink/operators/evicting_tumbling_window_operator.hpp EvictingTumblingWindowOperator<Key,Value,Out>
include/clink/operators/async_window_operator.hpp async-state variants (tumbling/sliding) on the disaggregated path
include/clink/core/pane_info.hpp PaneInfo (per-emission timing metadata)
include/clink/cep/pattern.hpp Pattern<T> DSL, Step, EdgeKind, SkipStrategy
include/clink/cep/cep_operator.hpp CepOperator<T,U> (NFA matcher)
include/clink/cep/cep.hpp, include/clink/cep/pattern_stream.hpp cep::pattern(...) entry points and PatternStream<T>

How it works

Event time and watermarks

A record may carry an EventTime. The WatermarkAssignerOperator<T> is a pass-through operator inserted into the stream that does two things per data element: it stamps event time on any record that lacks one (via a user-supplied extractor) and it feeds every record to a WatermarkStrategy<T>. The strategy is purely observational: on_record(...) updates its internal state and current_watermark() reports the current watermark if it advanced since the last query, otherwise std::nullopt. The assigner is the only component that actually emits watermarks; in the current cadence it queries the strategy once per batch and emits a watermark element downstream if one is available.

flowchart LR
  SRC["source"] --> WA["WatermarkAssignerOperator<br/>extractor sets EventTime;<br/>strategy.on_record(r);<br/>strategy.current_watermark()"]
  WA -->|"record batch"| DS["downstream"]
  WA -->|"watermark (if advanced)"| DS

Three strategies ship in watermark_strategy.hpp:

  • MonotonicWatermarkStrategy<T>: watermark = max event time observed. Out-of-order arrivals are tolerated but never regress the watermark.
  • BoundedOutOfOrdernessStrategy<T>: watermark = (max event time observed) minus a configured bound in milliseconds. Events arriving up to bound late are still in-window.
  • PartitionAwareBoundedOutOfOrdernessStrategy<T>: tracks the max event time per Record::source_partition, then emits (min across partitions) minus bound, so the watermark advances only as fast as the slowest partition. Records with no partition fold into one global bucket, making it byte-identical to the bounded strategy for non-partitioned sources. Its documented v1 limitation: a partition that goes quiet mid-stream keeps pinning the min low, because its per-partition max never advances.

The assigner also forwards (but does not re-feed) upstream watermarks: deriving event-time progress at this boundary is the assigner's responsibility. Its flush() emits the strategy's final watermark at end of stream.

Idleness

A Watermark carries an idle flag in addition to its timestamp. Calling .with_idleness(duration) on the assigner arms a processing-time probe (registered on open() and re-scheduled at half the idleness interval) so the assigner can detect that no record has arrived for duration even when zero records, watermarks, or barriers are flowing through process(). When that happens it emits Watermark::idle(...). The timestamp on an idle marker is informational; downstream alignment keys off is_idle(). When a record next arrives, the operator transitions back to active and the next watermark is non-idle.

Multi-input alignment

When an operator has N inputs (a union or a join), MultiInputAlignment tracks per-input watermark state and computes the downstream watermark as the running min of all inputs. Idle inputs are special-cased: an idle input's watermark is set to Watermark::max() so it cannot hold the min back, and an input transitioning from idle back to active is clamped to at least the currently-emitted global watermark so the global watermark never regresses. If every alive input is idle, a single idle marker is forwarded; if all inputs close, the min naturally becomes max() (end of time). The same class also drives Chandy-Lamport barrier alignment across the inputs (see ./checkpointing.md).

Watermark alignment groups

Separate from multi-input alignment, AlignmentGroup lets several WatermarkAssignerOperators (typically one per source) coordinate so the fastest is back-pressured when it runs too far ahead of the slowest. Enrol via .with_watermark_alignment(group_name, max_drift, max_wait). Each member publishes its watermark into a shared, process-wide AlignmentGroupRegistry-resolved group; before processing a data element the assigner blocks (bounded by max_wait) while its own watermark exceeds group_min + max_drift. Blocking at the assigner back-pressures the upstream source through the channel. The drift computation guards the Watermark::min() (INT64_MIN) sentinel against integer overflow by treating it as max drift.

Window operators

The aggregate window operators (TumblingWindowOperator, SlidingWindowOperator, SessionWindowOperator) share a common shape: they are keyed (Operator<pair<Key,Value>, pair<Key,Agg>>), they fold each value into a per-window aggregate using a user initial() + combiner(), and they fire windows when the watermark crosses the window end. TimeWindow is a half-open interval [start, end) on the event-time axis (window_trigger.hpp).

Window assignment differs per operator:

  • Tumbling: one window, start = floor(ts / size) * size.
  • Sliding: a record fans out to up to size/slide overlapping windows (window_starts_for_). The constructor validates that size is a positive multiple of a positive slide.

Every kind computes that assignment through one shared header, include/clink/time/window_arithmetic.hpp (floor_div, window_start_for, window_end_for, hop_window_starts_for, sat_add, sat_sub). It is shared because the same two lines had been written independently at ten sites, and nine of them used C++ integer division, which truncates toward zero rather than flooring. The two agree for every non-negative timestamp, so the divergence only showed up before the Unix epoch: a record at ts = -15000 in a 10s tumbling window landed in [-10000, 0), which does not contain it, and a hopping window - which additionally clamped its first pane to zero - discarded such records outright. The helpers also compute every intermediate in 128 bits and clamp once, because start + size, a floored start near INT64_MIN, and a hop's ts - size + slide all overflow at the extremes of int64 - and INT64_MIN / INT64_MAX are exactly what an uninitialised or sentinel timestamp column carries. Saturation leaves the topmost window closed at INT64_MAX rather than half-open, since no larger bound is representable. - Session: each record establishes a provisional window [ts, ts + gap]; overlapping sessions for the same key merge (min start, max end, aggregates combined via a third user function merger). Sessions are kept in a per-key sorted std::map<start, Session>.

Each operator's process() dispatches on element kind: data records go through ingest_one_ (window assignment + late-data routing), watermarks call on_watermark_advance_ / fire_due_sessions, and barriers go through on_barrier. The flush() hook synthesises a max watermark at end of stream so any still-open windows fire before the job ends.

Firing, panes and the earliest-window-end fast path

On a watermark advance, the tumbling and sliding operators walk their active windows, and for each unfired window whose end the watermark has crossed they dispatch the trigger's on_event_time and, if it fires, emit the aggregate. Each emission carries a PaneInfo describing timing (Early / OnTime / Late), a monotonic pane_index, and is_first / is_last flags. On-time fires are accumulated into a single batch before being pushed downstream to avoid one channel lock per fired window.

The tumbling operator keeps an earliest_window_end_ watermark over its active state. When the incoming watermark is below it, no window can fire or purge yet, so the full state scan is skipped. This turns the common "watermark advanced but crossed no window boundary" tick into a single comparison; it is recomputed after a batch of fires/purges.

Triggers

When (not just whether) a window emits is decided by a Trigger<T, TimeWindow>. A trigger returns a TriggerResult: Continue, Fire, Purge, or FireAndPurge (helpers should_fire / should_purge). The default is EventTimeTrigger, which fires when the watermark crosses window.end and deliberately returns Fire (not FireAndPurge) so the operator's allowed-lateness logic owns purging. Built-ins also include ProcessingTimeTrigger (fires on wall-clock crossing window.end) and CountTrigger (fires after N records). Triggers see the engine's time signals through a TriggerContext (current_watermark(), current_processing_time()); the operator-side implementation is detail::OperatorTriggerContext. The tumbling and sliding operators expose .with_trigger(...). The session operator has no pluggable trigger (it always fires on watermark crossing session end).

Stateful triggers (e.g. CountTrigger) hold per-window progress that must survive checkpoint/restore: they override is_stateful(), snapshot_state(), and restore_state(). The window operator persists the blob under a fixed key in a trigger_state slot at each barrier (persist_trigger_state_) and feeds it back on open() (restore_trigger_state_). Stateless triggers leave all of this as zero-cost no-ops. Trigger state is operator-level (one blob, not per-user-key), so on rescale it lands in a single key-group and only one post-rescale subtask restores it; a same-parallelism restart restores it fully.

Allowed lateness and the late side output

.allowed_lateness(duration) configures how long after window.end state is retained to accept late records. The window-entry fired flag (in WindowEntry<Agg>) distinguishes the on-time fire from late re-fires:

  • A record arriving within [window.end, window.end + allowed_lateness] updates the aggregate and immediately re-emits a Late pane.
  • A window is purged when the watermark reaches window.end + allowed_lateness, or when a trigger returns Purge / FireAndPurge. The lateness deadline overrides the trigger (the cleanup wins).

.late_output_tag(tag) opts in to routing records that arrive after the purge deadline to a side output (typed on Value, preserving event time), instead of the historic behaviour of silently creating a fresh bucket. Records still within the lateness band are not routed to the side output; only records past the deadline are. The default (no tag) is unchanged for back-compatibility. Side outputs are obtained from RuntimeContext::side_output<Value>(tag).

Evicting windows and evictors

EvictingTumblingWindowOperator<Key,Value,Out> buffers raw records per (window, key) so an Evictor<Value, TimeWindow> can pre-filter them before a user ProcessFn runs over the surviving records. The interaction order is: on Fire/FireAndPurge, run evict_before, then the process function, then evict_after, then emit; on Purge, drop the buffer without running the process function. Built-in evictors are CountEvictor (keep the most recent N records) and TimeEvictor (keep records within max_age of the most recent record). Evictors exist because you cannot un-combine an aggregate to drop an old record, so the evicting operator is a separate type that retains the raw buffer. Evictors are not supported on the aggregate window operators.

Durability of window state

Each operator has an in-memory constructor and a persistent constructor that additionally takes codecs. In persistent mode, window state is mirrored to KeyedState:

  • Tumbling: KeyedState<pair<window_start,Key>, WindowEntry<Agg>> in slot "windows".
  • Sliding: KeyedState<pair<window_start,Key>, WindowEntry<Agg>> in slot "sliding_windows" (write-through; the hot path reads/writes the backend directly).
  • Session: one KeyedState<Key, vector<SessionRow<Agg>>> row per key in slot "session_windows", so a merge is a single atomic put of the whole vector.
  • Evicting: KeyedState<pair<window_start,Key>, BufferEntry<Value>> in slot "evicting_buffers".

The codecs (window_entry_codec, session_row_codec, buffer_entry_codec, record_codec) live in window_state.hpp; each encodes the fired flag and next_pane_index so a restored window does not re-emit its on-time pane and late re-fires keep consistent pane numbering. The tumbling, session, and evicting operators treat an in-memory map as the authoritative hot path and rehydrate it from the backend on open() via scan(...). They support a write-back cache mode: with CLINK_WB_STATE_CACHE=1, per-record puts are skipped and the in-memory working set is flushed to the backend once per barrier in on_barrier, before the barrier is forwarded; in strict mode (the default) each record is mirrored through immediately. Because these are stateful operators, a stable .uid() must be pinned so state realigns on restore (see ./state-and-backends.md). The async-state variants in async_window_operator.hpp keep accumulators in KeyedState and ride event-time timers in the framework TimerService for firing; see ./async-state-execution.md.

Columnar ingest fast path

For int64 key + int64 value instantiations only, the tumbling, sliding, and session operators expose supports_columnar() + process_columnar(). When a columnar batch carrying the 3-column {event_time, key, value} Arrow sidecar arrives, the operator folds the columns straight into the per-window accumulators by calling the same ingest_one_ the row path uses, skipping only the per-record decode, so triggers, late panes, allowed-lateness, and watermark firing stay byte-identical to the row path. All schema guards are checked before any ingest_one_ runs, so a false return never half-processes a batch and the runner can safely fall back to process(). The WatermarkAssignerOperator has its own columnar fast path (with_columnar_event_times / with_columnar_partitions) that reads event times (and optionally partitions) from the sidecar and forwards the batch unchanged. See ./columnar-execution.md.

Event-time timers

Beyond windows, operators can register event-time and processing-time timers through RuntimeContext::timer_service() (register_event_time_timer(ts, key) / register_processing_time_timer(ts, key)). The base Operator::on_watermark default fires every event-time timer whose timestamp is at most the watermark via fire_due_event_time_timers, then forwards the watermark.

Because the runner drives that call on every watermark it forwards, whether or not a timer is due, the TimerService is also where event-time progress is recorded: TimerService::current_watermark() returns the highest watermark the operator has been driven with, surfaced to operator code as ProcessFunctionContext::current_watermark() and AsyncKeyedProcessContext::current_watermark(). An operator that keeps its own ordering - a reordering buffer, a custom join, anything that must decide whether a record can still be placed against records yet to come - needs to read it: at or below the watermark, it cannot. The window operators express the same judgement as allowed_lateness plus a late output tag. It is not part of the timer snapshot, so after a restore it is empty until the next watermark arrives; an operator whose correctness depends on it across a restore should keep its own high-water mark in keyed state. Note also that watermarks are emitted per source batch rather than per record, so the granularity at which the question can be answered is one batch of event time. Window operators override on_watermark and own their firing directly rather than relying on per-window timers; the assigner's idleness probe uses a processing-time timer.

Complex event processing

CEP matches a Pattern<T> against the per-key event sequence and emits a user value per completed match. Patterns are built fluently from Pattern<T>::begin(name) and a chain of steps. Each step carries a name, a predicate (.where(...), simple or iterative), and an EdgeKind describing how it connects to the previous step:

  • Begin: the first step (always present, exactly once).
  • Next: strict contiguity. The very next event must satisfy the predicate or the partial dies.
  • FollowedBy: relaxed contiguity. Skip non-matching events until one satisfies the step.
  • NotNext: strict negation. The very next event must not match; a match kills the partial, a non-match advances past the negative step.
  • NotFollowedBy: relaxed negation. While waiting for the next positive step, no event may match.

Quantifiers apply to a step after .where(...): .times(n), .times(min,max), .one_or_more(), .optional(), with .lazy() to prefer the shortest match (greedy is the default). .within(duration) bounds the event-time span between a partial's first and last events. .after_match_skip(strategy) controls what happens to other in-flight partials when one completes (NoSkip, SkipPastLastEvent, SkipToNext, SkipToFirst(name), SkipToLast(name)). Pattern::validate() runs at operator construction and rejects malformed patterns (empty, misplaced Begin, quantifier on a negative step, trailing NotFollowedBy without within, duplicate step names).

The matcher

CepOperator<T,U> is an NFA-style matcher. Partial matches are stored per key in KeyedState<int64_t, vector<PartialMatch<T>>> under slot "__cep_partials__" (one row per user key, keeping routing-by-key correct under parallelism). A PartialMatch<T> tracks the captured events, a parallel list of which step captured each event, per-event timestamps, the current step index, and the capture count within the current step (for quantifiers). The matcher consumes events in ARRIVAL order - it does not buffer and re-sort by event time (event time governs within bounds and watermark eviction, not step sequencing). A stream whose arrival order can invert the steps of a pattern - out-of-orderness comparable to the event-time gap between consecutive pattern steps - needs a reordering stage upstream of the operator, or pattern gaps wider than the arrival skew.

flowchart TD
  REC["record (key k)"] --> LP["load_partials_(k)"]
  LP --> AO["advance_one_(p, rec) per existing partial"]
  LP --> SP["spawn a fresh partial from step 0"]
  AO --> RES{"Keep / Advanced / Completed / Died"}
  RES -->|Completed| EM["emit_match_ to select / flat_select"]
  AO --> SV["save_partials_(k, survivors)"]

advance_one_ walks steps from the partial's current step, evaluating the predicate (iterative predicates receive a PatternMatch<T> view of events captured so far). A positive match captures the event up to max_count and auto-advances when the count is met; a non-match either advances (if the quantifier minimum is met) or dies (strict Next) or waits (FollowedBy). Negative steps do not capture: a match kills the partial, a non-match advances past it. A trailing NotFollowedBy stays pending until its within window closes, at which point prune_expired_ completes it as a success (emitted on the operator's main output with the watermark as event time) rather than timing it out.

On watermark advance, prune_expired_ evicts partials whose start_ts is older than (watermark - within). When a timed-out side output is wired (with_timed_out_output / with_timed_out_flat_output on the operator, or PatternStream::select_with_timed_out on the fluent surface, whose side stream is obtained with the standard .side_output<U>(tag) idiom), each pruned partial that is not a pending trailing-negative success is emitted on that tag before being dropped; otherwise it is silently dropped. The within bound equally holds at match time: advance_for_key_ expires a partial - routing it exactly as pruning would (trailing-negative success on the main output, else the timed-out tag) - when an incoming record's event time is already past (start_ts + within), so a record arriving between watermarks can never extend or complete a match whose event-time span would exceed within. flush() drops incomplete partials at end of stream.

Entry points and SQL

cep::pattern(stream, pattern, codec) has keyed and non-keyed overloads (cep.hpp). The keyed form looks up the typed key extractor registered by .key_by(...) so the planner uses hash routing into the operator; the non-keyed form routes all events through sentinel key 0. The returned PatternStream<T> materialises the operator on .select<U>(fn) (one record per match) or .flat_select<U>(fn) (zero or more records per match). CEP is also reachable from SQL MATCH_RECOGNIZE: the planner lowers it to a match_recognize_row operator (one row per match, append-only) that rebuilds a Pattern<Row> from the parsed steps/defines/measures and drives a CepOperator<Row,Row>; PARTITION BY columns become a row_compute_key keyer with hash routing (src/sql/physical_plan.cpp). See ./sql-frontend.md.

Key types and APIs

Type / function Responsibility
EventTime 64-bit millis-since-epoch event timestamp
Watermark event-time lower bound with an idle flag; idle(), min(), max()
WatermarkStrategy<T> on_record + current_watermark; observes records, reports the current watermark
MonotonicWatermarkStrategy / BoundedOutOfOrdernessStrategy / PartitionAwareBoundedOutOfOrdernessStrategy the three built-in strategies
WatermarkAssignerOperator<T> stamps event time and emits watermarks; .with_idleness, .with_watermark_alignment, columnar fast path
AlignmentGroup / AlignmentGroupRegistry cross-source watermark alignment with bounded drift
MultiInputAlignment per-input watermark min + barrier alignment across N inputs
TimeWindow half-open [start, end) event-time interval
Trigger<T,Window> / TriggerContext<Window> decide when a window emits; durability hooks for stateful triggers
EventTimeTrigger / ProcessingTimeTrigger / CountTrigger built-in triggers
Evictor<T,Window> / CountEvictor / TimeEvictor pre-filter buffered records in evicting windows
WindowEntry<Agg> / SessionRow<Agg> / BufferEntry<Value> per-window durable state + codecs
PaneInfo per-emission timing (Early/OnTime/Late), pane index, first/last flags
TumblingWindowOperator / SlidingWindowOperator / SessionWindowOperator aggregate window operators
EvictingTumblingWindowOperator raw-buffer window with an evictor + process function
Pattern<T> / Step / EdgeKind / SkipStrategy CEP pattern DSL and compiled IR
CepOperator<T,U> NFA matcher; per-key partials in KeyedState, timed-out side output
cep::pattern / PatternStream<T> fluent CEP entry points and .select / .flat_select

Configuration and knobs

  • WatermarkAssignerOperator::with_idleness(duration): emit an idle watermark when no record arrives for duration. Default 0 (disabled). The idle probe fires at half the idleness interval.
  • WatermarkAssignerOperator::with_watermark_alignment(group, max_drift, max_wait): bound cross-source drift; max_wait default 1 second per process() call.
  • BoundedOutOfOrdernessStrategy / PartitionAwareBoundedOutOfOrdernessStrategy: bound (milliseconds) constructor argument.
  • Window operators: .allowed_lateness(duration) (default 0, drop late records), .late_output_tag(tag) (default unset, late-late records silently dropped), .with_trigger(...) (tumbling/sliding; default EventTimeTrigger).
  • CLINK_WB_STATE_CACHE=1 (env var): write-back cache mode for the tumbling, session, and evicting operators; defers per-record state puts and flushes once per barrier. Default unset (strict per-record mirroring).
  • Columnar ingest fast path: set_columnar_enabled(bool) on the window operators (default on; only effective for int64/int64 instantiations and only under CLINK_HAS_ARROW).
  • Pattern<T>: .within(duration) (default none, partials live indefinitely), .after_match_skip(strategy) (default NoSkip), quantifier modifiers, .lazy().

Guarantees and caveats

  • Watermarks are monotonic by contract: an operator forwarding a regressing watermark is a bug, and downstream operators may ignore it. Idle and idle-to-active transitions are clamped in MultiInputAlignment so the global watermark never regresses.
  • The three watermark strategies are exactly those shipped (monotonic, bounded out-of-orderness, partition-aware bounded). The partition-aware strategy now supports per-partition idleness: the WatermarkAssignerOperator tracks each partition's last-activity wall-clock and, on its processing-time probe, calls set_idle_partitions so a partition quiet for longer than the idleness duration is excluded from the min (it re-enters the instant it produces again). This is enabled via with_idleness(...) on the assigner, reachable from SQL by the source-table idle_timeout_ms option, and complements the assigner's existing stream-level idle marker (whole-source quiet) for multi-input alignment.
  • Tumbling, sliding, and session windows support allowed_lateness and late_output_tag. Tumbling and sliding support pluggable triggers; the session operator does not (it fires on watermark crossing session end). Evictors are supported only on the evicting window operator, not on the aggregate operators.
  • The SQL windowed-aggregate operators (WindowRowOp for TUMBLE/HOP/CUMULATE and SessionWindowRowOp, in src/sql/install.cpp) are a separate, thinner implementation from the typed operators above. They now drop late records: a record whose window has already fired and been purged is dropped instead of re-creating the window and emitting a duplicate, partial row. For WindowRowOp the cutoff is "window end at or under the current watermark (less allowed_lateness_ms_, default 0)", captured at ingest on every path - sync, columnar, and the async/disaggregated coroutine path (where the fold is deferred, so the ingest-time watermark, not the fold-time one, is the correct cutoff). For SessionWindowRowOp (sync + columnar; no async path) the guard is applied only when a record would OPEN a fresh session: if that single-event session is already purged (ts + gap at/under the cutoff) it is dropped, whereas a record that MERGES into a still-live session legitimately extends it and always proceeds. Idle-source handling is wired through SQL via the source-table idle_timeout_ms option (per-partition + stream-level idleness on the assigner).
  • Allowed lateness (fire-once grace band) and the late side output are wired on the SQL window path via two source-table options, threaded to the window op by the planner (first_source_property in src/sql/physical_plan.cpp):
  • allowed_lateness_ms (default 0) holds a fired window/session open for a grace band: the fire/purge threshold becomes window_end + allowed_lateness_ms (sessions: end + gap + allowed_lateness_ms), so a late-but-within-band record still folds into the window before its single emit. The semantics are fire-once after the grace band, not re-fire-per-late-record: the window emits exactly one row when the watermark passes the band, with every within-band record already included. This keeps the codec and all three WindowRowOp paths (sync, columnar-WS3, async) byte-identical - the async path registers its event-time timer at window_end + allowed_lateness_ms and maps it back to window_end on fire. The cutoff is still frozen at ingest (current_wm_ - allowed_lateness_ms).
  • late_records_to_dlq (default off) routes a record dropped as fully-late to the ambient dead-letter channel (RuntimeContext::report_bad_record, connector = sql_window / sql_session_window, direction = operator) instead of dropping it silently. Best-effort and not barrier-tied (a replayed late record is re-reported), consistent with the DLQ contract. The drop is detected where the fold sees zero live windows for the record (WindowRowOp::fold_record_into_ folded-count, plus the async pre-filter short-circuit) and, for sessions, on the new-session branch. Gap: the fully-vectorised WS6 columnar session fold (fold_session_columnar_) drops late records silently - it has no per-record Row to hand the DLQ - so late_records_to_dlq does not cover that one path.
  • The SQL window operators read the event-time column exactly (event_time_of), not through a double. A double carries 53 bits of mantissa, so a timestamp above about 9.0e15 was previously rounded before window assignment: epoch milliseconds (~1.7e12) survive that, epoch nanoseconds (~1.7e18) do not.
  • The SQL window operators are durable on a synchronous backend. WindowRowOp and SessionWindowRowOp keep their hot state in in-memory maps unless the backend defers reads, and flush it in snapshot_timers (the runner's pre-snapshot hook) to the same slot their async paths use - win for the fixed windows, session for sessions - reloading it in open(). This is the shape AggregateRowOp uses for agg. Before 2026-07-28 neither flushed, so a restore brought back an operator with no open windows: the source resumed at the checkpointed offset, the records already folded into those windows were never replayed, and their contribution was silently lost. WindowRowOp was fixed first, SessionWindowRowOp a day later, and TopNPerKeyRowOp (the ROW_NUMBER top-N / dedup changelog operator, slot topn) the day after that - three separate discoveries of one defect class, each fix written against one class while its siblings stayed broken. For the changelog operator the symptom is sharper than lost output: restored empty, a displacing arrival emits no delete for the row it displaces, so the stale row stays live in an upsert sink forever. Its codec also round-trips each partition's vector ORDER, because ROW_NUMBER ties break by arrival order and that order exists nowhere but the vector; TopNStateLifecycle in tests/test_sql_runtime.cpp pins both properties. An inventory of every stateful Row-channel operator now lives with the pending-gaps list: joins, distinct and set-ops hold state per-record in KeyedState (durable by construction on a synchronous backend). LastNAggRowOp (bounded ROWS OVER frames, slot lastn) joined the durable family on 2026-07-29 - the fifth member, found by that inventory rather than by a production surprise; its codec serialises both the frame window (order included - eviction pops the front) and prior_emitted, the row its changelog last declared live, without which a drained window emits nothing instead of its delete. LastNAggStateLifecycle pins both. OverAggregateRowOp's sync path - the inventory's last unverified entry - was then VERIFIED BY TEST to have the same defect (a pending pre-checkpoint row lost outright, every post-restore running value computed from an empty fold) and fixed the same day, closing the family at six: it flushes its sync partitions to the over slot the async path already checkpoints through, and persists the sync tie-break counter in the over.wm blob (9 -> 17 bytes, trailing-bytes-tolerant; an old blob falls back to a hydrate-time max over restored pending seqs) so a restarted-to-zero counter cannot collide a new (ts, seq) key with a restored one - map::emplace on a duplicate silently drops the record. OverAggStateLifecycle pins both hazards. Unlike TopN/LastN there is no deferring-backend gap here: on a deferring backend the async path owns this state per-record. The two file-materialising changelog SINKS (FileJsonUpsertSink, slot upsert; ChangelogNetSink, slot net) followed on 2026-07-29: both write the file only at end-of-stream, so mid-job the whole compacted view lived in memory and a restore lost every row whose last change predated the checkpoint - a data-LOSS class, not the duplicate-emission class the gap ledger first recorded it as. Their persistence is WRITE-THROUGH at on_data (dirty keys coalesced per batch), deliberately not on_barrier: the sink runner snapshots the backend BEFORE delivering the barrier to the sink, so a barrier-time flush would be one checkpoint stale, forever. Deletes write through too - UpsertSinkLifecycle's double-failover test resurrects a zombie row if they do not - and the netting sink round-trips its MULTIPLICITY, since a count of 2 restored as less nets to empty one delete early.
  • Memory per (key, window). The unit to reason about is the pane, not the group: a HOP of size S and slide s holds ceil(S / s) buckets per key (a 10s/2s hop, five) and a CUMULATE holds S / step. Measured 2026-07-28 for the cheapest realistic shape, an int64 group key and a COUNT(*):
serialised, per (key, window) what it is
TUMBLE 127 B window_start, the group values as a JSON object string, and one encoded AggState
HOP 116 B as above, stored once per overlapping pane
CUMULATE 117 B as above, once per cumulative slice
SESSION 131 B as above plus the session's explicit end

Against 16 bytes of actual information. Instrumenting encode_bucket attributes it (row counts per pane, same run):

component B/pane share
the encoded AggState 75.0 70%
group values, as a JSON object string 19.4 18%
window_start + window_end 16.0 15%
aggregate-count prefix 4.0 4%

So the encoded AggState is the dominant term, not the group values. An earlier revision of this page said the opposite; that was arithmetic rather than measurement, and it was wrong. It matters because it points at a different fix. Of the 75 bytes an AggState writes for a COUNT(*), only the 8-byte running_count carries information: the rest is unconditional scaffolding for aggregate families the query does not use - two null JSON values for MIN/MAX at 8 bytes each, four zero container counts, an empty exact-decimal string, a null UDAF accumulator, five flags. This is the same accretion that took sizeof(AggState) to 264 bytes in memory; that was fixed, the serialised form was not.

Resident memory is higher again: sizeof(WindowBucket) is 56 bytes and sizeof(Session) 64 on libc++, on top of one std::map node per window, one vector allocation per bucket's AggStates (104 bytes each), and one for the group values.

Two guards, because the two costs need different instruments. static_asserts at the definitions in src/sql/install.cpp bound sizeof for AggState, WindowBucket and Session, written in terms of the members rather than byte literals so they hold on both libc++ and libstdc++; what they forbid is a second container going inline, which is how AggState reached 264 bytes. WindowMemory.SerialisedStatePerOpenWindowStaysWithinItsBound gates the serialised figure at 160 B per pane, which is a regression gate and not a target. Resident memory is not deterministic enough to assert on and stays with benchmarks/clink_window_state_bench, which differences peak RSS between many groups and one, in child processes because ru_maxrss is a high-water mark. - Window state redistributes correctly across a parallelism change. Both operators key their state by the group key, so KeyedState writes it under that key's key-group prefix byte and each new subtask restores exactly the slice key_group_range_for_subtask assigns it. Scale-up filters one parent snapshot per new subtask, scale-down merges several through combine_snapshots before restoring; in both directions the union over the new subtasks is the same set of panes the old parallelism would have fired. WindowRescale in tests/test_sql_runtime.cpp pins this for all four kinds at 1 -> 2, 1 -> 4, 4 -> 2 and 4 -> 1, on both the row and columnar ingest paths. - Stateful-trigger state and window state are durable in the persistent (codec-bearing) constructors and restore fully on a same-parallelism restart. Operator-level trigger state lands in a single key-group on rescale, so only one post-rescale subtask restores it; full rescale routing of operator-level trigger state (and of the async window operators' timers and "wm" slot) is a deferred follow-on. Window operators are stateful and require a pinned .uid(). - The evicting window re-puts the whole record vector per mutation in strict mode (write amplification); CLINK_WB_STATE_CACHE=1 amortises this to one put per barrier. A true incremental list-state is the longer-term home. - CEP is a v1 subset: linear patterns, greedy quantifiers by default (lazy available), strict/relaxed contiguity, and the listed negation and skip strategies. Documented gaps (in pattern.hpp) include group patterns as addressable units (groups are flattened at construction), group-level repetition, and lazy controls beyond .lazy(). A trailing NotFollowedBy requires .within(...) or the pattern is rejected at validation. CEP partials are durable in KeyedState only when a state backend is present; without one the operator falls back to an in-memory shadow that is lost on restart (used by unit tests, not the default cluster wiring). - The inline fluent CEP .select / .flat_select capture the user function into the process-wide runner registry, so submitting such a job to a remote Worker that lacks the registration fails there; package the operator as a real plugin for cross-process CEP jobs. - The columnar ingest fast path is limited to int64 key + int64 value and the 3-column {event_time, key, value} Arrow sidecar; all other instantiations and non-columnar batches take the row path, which is the semantic reference.