nordisk/knut

Þund — incremental streaming state (watermarks + checkpoints) — design

Class: design · Function axis: f3 (the NativeBackend execution engine). Strict: this file states only the shape we build — the watermark model, the checkpoint state store, and the additive mode switch. Every why, the landscape it is drawn from (Arroyo/Flink/RisingWave), and the honest status live in thund-reasoning.md; the enclosing engine shape is in thund-design.md. Reuse: the state store reuses graphar's object_store + parquet-rs substrate (graphar/src/storage.rs:ChunkStore, graphar/src/io/parquet.rs).

What this adds, and the one hard constraint

The native backend already runs unbounded flows as recompute micro-batch streaming (backend/native.rs:run_streaming_flow / run_cdc_flow): each trigger pulls the next micro-batch into an accumulated MemTable and re-runs the flow's SQL over all rows seen. That path is correct, deterministic, and at-least-once, but it recomputes per trigger and maintains no operator state, so the doc-comment goals — event-time watermarks, window closing, barrier checkpoints, exactly-once-ish recovery — are unimplemented, and NativeBackend::state_root (backend/native.rs:74) is a dead field (with_state_root sets it; nothing reads it).

This design makes those goals real as a new, opt-in incremental mode alongside the recompute path. The hard constraint, stated once and honoured by every phase below:

Recompute is the default and the fallback, and is never removed. The existing run_flow / run_streaming_flow / run_cdc_flow code path is untouched. Incremental streaming is selected explicitly; when it cannot run (no state_root, an unsupported query shape, a non-replayable source) it degrades to recompute with an honest RunEvent, never a silent failure.

The additive switch

A new mode selector on the backend, defaulting to today's behaviour:

/// How the native backend drives unbounded flows.
pub enum StreamingMode {
    /// Re-run the flow's query over accumulated state each trigger
    /// (`run_streaming_flow`). At-least-once, no operator state. DEFAULT.
    Recompute,
    /// Event-time watermark tracker + windowed/keyed operator state +
    /// barrier checkpoints under `state_root`. Opt-in.
    Incremental,
}

impl NativeBackend {
    pub fn with_streaming_mode(mut self, mode: StreamingMode) -> Self { … }
}

StreamingMode::default() = Recompute. NativeBackend::default() is unchanged, so every existing caller and test keeps the recompute path with no edit. The mode is read only inside the streaming/CDC dispatch in run_flow; batch flows ignore it entirely.

Parity in the capability report: Capabilities gains one field, incremental_state: bool (backend/mod.rs:Capabilities). NativeBackend reports true only when mode == Incremental and state_root.is_some(); SparkBackend reports false (a one-line parity edit in spark.rs:capabilities, mirroring how it already reports event_time: false). The default check is unchanged — incremental_state is a capability advertised, not a gate; a pipeline never fails check for lacking it, it just runs on recompute.

The watermark model (event-time, allowed-lateness, window close)

Consumes the IR's existing Watermark { event_time_column, allowed_lateness_ms, idle_timeout_ms } and WindowSpec { Tumbling | Hopping | Session } (ir/streaming.rs) — no IR change. Active only in Incremental mode; in Recompute mode a watermark on the flow is documented as advisory (recompute sees every accumulated row, as today).

Tracking. A WatermarkTracker holds, per input source, the maximum event_time_column value observed. The combined watermark is

W = min over non-idle inputs of ( max_event_time(input) − allowed_lateness_ms )

the minimum so the slowest input gates progress (Flink/Arroyo consensus). An input that has produced no rows for longer than idle_timeout_ms stops gating (drops out of the min) so one quiet partition cannot freeze the stream; idle_timeout_ms == 0 means never idle. W advances monotonically — a trigger that would lower it leaves it unchanged.

Late-row dropping. After W advances on a trigger, rows whose event_time_column < W are late and are dropped before the operator sees them. The count is recorded on a RunEvent and as a functional-status / bench metric (late_rows_dropped) — Arroyo drops silently; Þund reports, honestly, exactly how many. Rows within [W, max_event_time] are on-time.

Window close. A window (tumbling/hopping by size/slide, session by inactivity gap, all keyed on event_time_column) is closeable once W ≥ window.end + allowed_lateness_ms: its aggregate is final, so it is emitted once (Append) and its keyed state is evicted from the operator store. This is what bounds memory: closed windows leave the working set, unlike recompute, whose accumulated MemTable only grows. Windows still open at end-of-stream (or AvailableNow) are flushed on the final pass so the result matches the batch path over the same rows.

The checkpoint state store (the resurrected state_root)

Barrier/epoch checkpointing that finally uses NativeBackend::state_root. The substrate is the one knut already writes Parquet through: parquet-rs ArrowWriter over object_store (s3://…, file://…), exactly graphar's ChunkStore (graphar/src/storage.rs) — no RocksDB, so workers stay stateless (the Arroyo/RisingWave "disaggregated state on object storage" position).

Layout under state_root

{state_root}/{pipeline}/{flow_target}/
  epoch=000007/
    operator-{op_id}.parquet     ← keyed/windowed operator state (Arrow → Parquet)
    _thund_checkpoint.json       ← the epoch manifest (the commit point)
  epoch=000008/…
  _thund_latest.json             ← pointer to the newest COMMITTED epoch

What is persisted per epoch

Barrier / epoch semantics

One epoch = one trigger. A barrier is injected at each trigger boundary. The native engine is single-node and processes one operator chain sequentially, so barrier alignment is trivial (there is no fan-in to align) — a deliberately simpler Chandy–Lamport than a distributed engine needs. On the barrier:

1. Each stateful operator snapshots its state to epoch=N/operator-*.parquet.
2. The source's consumed offset and the current W are gathered.
3. The manifest is written last, and its write is the atomic commit point: _thund_checkpoint.json with status: COMMITTED, then _thund_latest.json is flipped to N. A crash before the manifest lands leaves a partial epoch=N/ dir that recovery ignores (latest still points at N-1) — write-ahead, commit-by-manifest.

Recovery on restart

On run with Incremental mode and a populated state_root: read _thund_latest.json → load that epoch's manifest → restore each operator's state from its Parquet files, restore W, and resume the source from the committed offset. The uncommitted trigger (if the crash was mid-epoch) is re-read from the source and re-processed. A cold state_root (first run) starts from empty state — identical to today.

Exactly-once-ish — the guarantees and their honest limits

• State is exactly-once w.r.t. the checkpoint store. Because the manifest commit is atomic and operators restore from the last committed epoch, operator state after recovery is exactly the state as of that epoch — no double-counting of already-checkpointed input.
• Output is at-least-once with idempotent replay. On recovery the uncommitted trigger's output is re-emitted. A sink that MERGEs by key (the nordisk _gar_id MERGE law, knut-bifrost / FalkorDbSink) absorbs the replay idempotently → effectively exactly-once. A plain append sink may see the replayed trigger's rows twice → at-least-once.
• Limits, stated plainly. Single-node sequential engine: no distributed barrier alignment, no multi-worker rescale (the state layout permits it later; the engine does not do it now). No two-phase commit to external sinks — exactly-once end-to-end holds only for idempotent/keyed sinks. Recovery requires a replayable source (Kafka offsets, or a seeded with_stream_input stream); a non-replayable source loses in-flight, uncheckpointed rows on crash. Late rows past allowed_lateness are dropped by design, not recovered.

Compared to the recompute default (at-least-once, no recovery, unbounded working set), incremental mode adds bounded (window-evicting) memory, durable recovery, and idempotent-sink exactly-once — at the cost of only running the query shapes the incremental operators support (windowed/keyed aggregation and row-wise maps); anything else falls back to recompute.

Phased implementation plan

Each phase is small, lands independently green, is additive (recompute untouched, nothing removed), carries a bencher arm wired into the nornir suite (xtask/examples/nornir-bench.rs, register_bench!; runs persist to the workspace bench_runs Iceberg table via nornir bench run knut — benching-benchmarks.md), and a test-matrix functional_status emit point (lib.rs:functional_status, feature testmatrix). The HEAVY bench arms all compare recompute vs incremental and record throughput (rows/s) + peak RSS + host cores + host memory in the BenchResult metrics map (three new keys: peak_rss_bytes, host_cores, host_mem_bytes).

Green invariant across all phases: the incremental result equals the recompute result equals the batch path over the same rows (the equality already asserted in tests/streaming_microbatch.rs), extended to the incremental path so recompute remains the correctness oracle.

Landed status (phases 3-4) — checkpoint store + crash recovery

Phases 3 and 4 landed together (a checkpoint you cannot recover from is half a feature) in backend/checkpoint.rs + backend/native.rs, additively — the recompute default and the phase-0/1/2 paths are byte-for-byte unchanged.

The store (backend/checkpoint.rs::ParquetStateStore). Rooted at {state_root}/{pipeline}/{flow_target}/, over datafusion::object_store (the same object_store 0.13 graphar's ChunkStore uses, version-matched to datafusion's Arrow via its re-exports — no second Arrow tree, no RocksDB). commit(manifest, state) writes in the order operator-state Parquet → manifest JSON (_thund_checkpoint.json, stamped COMMITTED) → _thund_latest.json: the manifest is the write-ahead commit point and the latest-pointer flip is the publish. latest() reads the pointer → the epoch manifest and returns it only if COMMITTED, so a crash between any two writes leaves a partial epoch=N/ that recovery ignores (latest still names N-1). One epoch == one trigger.

What each path checkpoints. The windowed operator's state is the buffered open-window rows (post-evict, so it stays bounded); the non-windowed watermark path's state is the accumulated kept rows. Both persist, in the manifest, the combined watermark W, the per-input maxima (to rebuild the tracker exactly), the consumed offset (= trigger index + 1), the running late-drop + windows-closed counters, and the emitted-window tags (so a resumed run never re-emits a window that closed pre-crash).

Recovery (phase 4). On a run whose state_root already holds a committed epoch, the operator restores its state, WatermarkTracker::restores W + the maxima, seeds its counters + emitted set, and resumes from the committed offset — the already-committed triggers are not re-read, so there is no double-work and, because the on-time rows of still-open windows were checkpointed, no lost (non-late) event. The restart boundary is NativeBackend::with_max_triggers(n): a run stops after n triggers (checkpointing each) — a simulated crash — and a second run over the same source

• state_root recovers and finishes. tests/streaming_microbatch.rs proves the resumed result equals a single uninterrupted run (windowed and non-windowed), which is the phase-4 correctness oracle. NativeRun::checkpoints_committed / resumed_from_epoch expose the barrier count and the epoch resumed from.

Refinement to the "re-emit the uncommitted trigger" sketch above. Because the engine commits per trigger (one epoch = one trigger), a clean restart resumes exactly at the committed offset with no re-emit. Re-emission only arises if a crash interrupts a trigger mid-epoch (its manifest never lands), in which case that whole trigger is re-read from the source and re-processed — the at-least-once replay the idempotent _gar_id MERGE sink absorbs. This is the honest, stronger guarantee the per-trigger barrier buys.

Folded-in lessons (Flink / Arroyo / Kafka Streams / RisingWave). The manifest-as-commit-point + write-ahead epoch dir is Flink's checkpoint barrier → _metadata file made single-node (no barrier alignment: one sequential operator chain, so a barrier is just a trigger boundary — a deliberately trivial Chandy–Lamport). Committing the source offset inside the same checkpoint as the operator state is the Kafka-Streams/Flink invariant that makes state-and-progress advance atomically (no offset drift on replay). State on object storage, not embedded RocksDB is the Arroyo/RisingWave disaggregated-state position, which is why recovery needs no local disk and a worker stays stateless. Restoring the watermark monotonically (never below the committed W) preserves the event-time-frontier invariant across the restart, so late-vs-on-time classification is identical before and after a crash.

Landed status (phase 5) — incremental aggregation (no per-window recompute)

The incremental-aggregation operator (backend/native.rs::exec::agg) is the new default for a supported windowed aggregate in Incremental mode: instead of buffering a window's rows and re-running its SQL on close (the phase-2 behaviour, still the fallback), it maintains one accumulator per (window, group) — updated as each kept row arrives — and emits a window's aggregate directly from those accumulators on close, then evicts them. Raw rows are never buffered, so the working set is bounded by the group cardinality × concurrently-open windows, not the row count.

Detected + supported shape (else honest fallback to recompute-per-window). Detection is by DataFusion LogicalPlan introspection (AggPlan::detect): the plan must be an Aggregate directly over a plain TableScan (no WHERE/join), optionally under the SELECT projection; group columns must be bare Int64 / Utf8 columns; aggregates must be COUNT(*) / COUNT(col) / SUM / MIN / MAX / AVG over Int64; the window must be Tumbling or Hopping; and the query's output schema must be exactly [group cols, agg cols] with each agg column typed as its aggregate emits (Int64 for COUNT/SUM/MIN/MAX, Float64 for AVG). The top projection must be a pure pass-through (column refs / bare aggregate exprs, optionally aliased): a computing projection such as SUM(x)+1, a CAST, or a scalar function is rejected by is_passthrough_top (else the operator would emit the raw aggregate, not the computed value). Anything outside this — float source columns, a WHERE, a Session window, a computing/reordering projection, DISTINCT, a filtered aggregate — fails detection and falls through to the recompute-per-window path, so nothing regresses and no query silently returns a wrong answer.

AVG is kept exactly as two flat i64 partials per (window, group) — the running sum and non-null count — divided to Float64 only on window emit (an all-null group emits SQL NULL), so no float rounding accumulates across a window's arrivals; the accumulator layout generalises to a per-aggregate partial width (AVG = 2 slots, the rest = 1) addressed by a precomputed offset.

Equivalence (the green oracle). The pre-existing windowed COUNT(*) tests (windowed_tumbling / windowed_hopping / windowed_late_after_close) now run through this operator and still assert byte-identical window aggregates — they are the recompute-equivalence proof. incremental_agg_grouped_matches_ground_truth adds a grouped COUNT/SUM/MIN/MAX corpus checked against an independent hand-computed ground truth; incremental_agg_avg_matches_ground_truth checks a mixed COUNT/AVG/SUM query (multi-slot offsets + a Float64 emit column) against an independent sum/count ground truth; incremental_agg_falls_back_for_unsupported_shapes proves a computing projection (SUM(x)+1) and a WHERE degrade to recompute (used_incremental_agg == false); and incremental_agg_recovers_across_a_crash (plus incremental_agg_avg_recovers_across_a_crash for the two-slot AVG state) proves the accumulator state itself checkpoints + recovers (phases 3-4 substrate) with a resumed result equal to a single run. NativeRun::used_incremental_agg exposes which path ran.

Checkpointed state. Every partial slot is an i64 (COUNT/SUM/MIN/MAX one each; AVG's sum + count), so the accumulator state serialises to a fixed schema [__ws, __we, __wc, <group cols>, <one Int64 per partial slot>] — one row per open (window, group) — and the phase-3/4 store snapshots + restores it losslessly with no bespoke encoding, AVG included.

Honest limits. SUM/AVG are not incrementalised for float source columns or when exact large-integer precision beyond i64 matters (a very large running sum divided to Float64 loses precision past 2^53 — the recompute path stays correct there); Session windows (whose spans merge as events arrive) are not incrementalised; the operator is single-node and processes one chain sequentially. Every one of these is a fallback, not a failure: the recompute path runs and stays correct.

See also

• thund-design.md — the enclosing engine (IR + ExecBackend

  • the two backends).

• thund-reasoning.md — the landscape, the why, and the honest status this design turns green.

• bifrost-design.md — the MERGE-by-_gar_id idempotent sink that makes incremental mode's replay effectively exactly-once.

• benching-benchmarks.md — the nornir bench contract the phase arms plug into.