Topic 27 — Streaming & Incremental View Maintenance

Why this matters: recomputing from scratch is the enemy. A standing query over a changing graph should cost per-change, not per-database. Differential dataflow and DBSP made that rigorous — and FalkorDB’s delta matrices (topic 20) are already halfway there conceptually: DP/DM are positive and negative Z-sets waiting for an algebra.

The problem, measured (bench lane 1, provided — runs today)

cargo run --release --bin ivm_bench — 50 000 nodes / 500 000 edges, then 10 batches of +90/−10 edges, recomputing each view from scratch per batch:

view                        full recompute per batch
triangles                              141.6 ms
wedge join                            1111.0 ms
reachability (re-BFS)                   31.2 ms

One hundred changes to half a million edges, and the wedge join spends 1.1 seconds re-deriving an answer that barely moved. The batch is 0.02% of the graph. The recompute is 100% of the work, every time.

That ratio — change size against work size — is the only motivation incremental view maintenance needs, and stating it in milliseconds first is what keeps the rest of the topic honest. The three views are deliberately different shapes: reachability is cheap to redo (31 ms) so incrementalization has little room and mostly has to avoid regressions; the wedge join is expensive and local, so delta rules should win enormously; triangles sit between them and are where the bookkeeping cost of maintaining state starts to show against the recompute it saves.

Predict the speedup for each of the three before implementing, and predict which one’s incremental version is slower to initialize than a full recompute — that one is the interesting result, not the 100× one.

Our motivation numbers first (Apple M3 Pro, 50K nodes / 500K edges, batches of 100 changes)

The gap is 3-5 orders of magnitude, and none of it requires cleverness — just refusing to touch data that didn’t change.

The one algebraic idea

Changes are Z-sets: collections with i64 weights (+1 insert, −1 delete). Operators split into two classes:

  LINEAR (stateless to incrementalize)       NONLINEAR (need state)
  map, filter, flat_map, union               distinct, count, sum, top-k,
    op(ΔA) = Δop(A) — deltas stream through    min/max — deleting the last
                                               copy must RETRACT the output,
  BILINEAR (need arranged inputs)              which requires knowing how
  join:  Δ(A⋈B) = ΔA⋈B + A⋈ΔB + ΔA⋈ΔB          many copies existed: state

That table is the topic. DBSP’s contribution: any query built from these pieces auto-incrementalizes by circuit rewriting; the state each nonlinear operator needs is exactly an integral (z^-1 feedback) of its input. Differential’s contribution: it also works inside recursion, with deltas indexed by (iteration, input-version) lattice timestamps.

flowchart LR
    subgraph "DBSP incrementalization"
    dI["ΔIN (z-set/tick)"] --> I1["I (integrate)"] --> Q["Q (the query)"] --> D1["D (differentiate)"] --> dO["ΔOUT"]
    end

The chain I→Q→D is the specification; the engineering is pushing I and D through Q’s structure until only nonlinear operators keep integrals — those integrals are Materialize’s arrangements (shared, indexed, compacted update logs).

Timestamps, watermarks, and why “when” is half the problem

Timely’s insight (Naiad): every message carries a logical timestamp; the scheduler broadcasts progress (“no more messages ≤ t will ever arrive” — a frontier/watermark, MutableAntichain timely frontier.rs:380). Only when the frontier passes t may a nonlinear operator emit finalized output for t. That single mechanism subsumes: batch boundaries, out-of-order data, iteration rounds (timestamps extend to (epoch, round) pairs), and exactly-once output (emit per closed timestamp).

RisingWave makes the same call with different machinery: barriers flow through the dataflow (Chandy-Lamport style), every operator checkpoints its state to S3 at barrier alignment — its Op enum (stream_chunk.rs:45: Insert/Delete/UpdateDelete/UpdateInsert) is a Z-set weight wearing protocol clothing.

The systems, placed

The stubs (experiments/)

Provided: zset.rs (consolidation, merge, the distinct-is-not-linear test), graph.rs (churn generator + all three full-recompute oracles), ivm_bench (prices the enemy even before the stubs exist).

Deliberate scope cut: SemiNaiveReach is insert-only. Deleting an edge from a reachability result is the problem that needs differential’s timestamp machinery — see reading-differential-dataflow.md §4 for why.

Reading guides

• reading-naiad-timely.md — Naiad: the clock that unified batch, streaming, and iteration
• reading-differential-dataflow.md — Differential dataflow: retractions that survive recursion
• reading-dbsp.md — DBSP: incremental view maintenance as a calculus
• reading-materialize-risingwave.md — Materialize vs RisingWave: two production IVM bets
• reading-kafka-log.md — Kafka: the log is the database

Further references: “MillWheel” (VLDB 2013) — where watermarks (low watermarks over event time) entered production streaming; the heuristic ancestor of timely’s proof-carrying frontiers, and the lineage behind Google Dataflow/Beam and Flink’s model.

Cross-topic links

• Topic 20: FalkorDB delta matrices — DP=+Δ, DM=−Δ, wait = integrate; what’s missing vs DBSP is pushing queries through the deltas instead of forcing a merge first. That gap is exactly M27.
• Topic 4: an arrangement’s batch/spine/compaction IS an LSM over update triples — merging batches consolidates weights like compaction drops tombstones.
• Topic 8: retractions are the MVCC intuition inverted — instead of versions hiding rows from the past, negative weights erase rows from derived futures.
• Topic 24: semi-naive frontier = delta-stepping’s bucket discipline; both refuse to re-derive settled facts.
• Topic 5/15: Kafka = the WAL promoted to the database; Materialize’s persist and RisingWave’s Hummock both re-derive state from a shared log.