go-incremental

An incremental computation engine for Go, descended from Jane Street's Incremental. You build a DAG of nodes, Observe the ones you care about, and call Stabilize whenever inputs change; only the nodes actually affected by the change recompute.

Two things set it apart from every incremental/IVM engine we know of:

• The graph's SHAPE responds to data. Bind rebuilds subgraphs when their driving value changes, so a ruleset, a config, or a plan can be an input - change it and only the affected computations restructure and recompute.
• Scenarios: N live what-if worlds on one graph, one stabilization. Fork the computation into overlay worlds that each pin some inputs to hypothetical values while tracking every other input live. A round updates the base world and every scenario together; a world costs memory and compute proportional to how far it actually diverges, never O(graph). No shipping IVM engine (Materialize, Feldera, RisingWave) or branching database (Neon, Dolt) offers this: they either maintain one world incrementally or fork storage and recompute each branch from scratch.

Scenarios

sc := incremental.NewScenario(s)      // a live overlay world
incremental.SetIn(sc, price, 130.0)   // pin one input in that world only
_ = s.StabilizeAndWait()              // one round: base + every scenario
v, _ := incremental.ValueIn(sc, obs)  // any observer, as sc sees it
sc.Drop()                             // reclaim everything the world touched

Semantics:

• Live tracking. Inputs a scenario does not pin are read live: a base Set flows into every scenario's derived values on the next round. Pins hold until the scenario is dropped.
• Collections fork too. SetKeyIn/DeleteKeyIn pin single keys of a MapVar; the whole operator suite (filter, group aggregates, joins) maintains per-scenario views, and MapObserver.GetIn/SnapshotIn read them. A scenario's footprint on a collection node is its divergence, not a copy of the map.
• Rulesets as scenarios. A Bind whose left-hand side is pinned builds that world its own right-hand side subgraph - spliced in as real graph nodes, scheduled and torn down like any bind refire. Deploy a candidate ruleset as a scenario against live traffic and diff the decisions.

Answering N what-ifs as overlay worlds shares one wavefront; the alternative

• perturb, stabilize, read, revert, stabilize, per scenario - pays the full round trip N times (Apple M3 Pro, 512-input pricing graph, one pinned input per scenario):

live scenarios	one overlay round	sequential set/read/revert
10	55 us	303 us
100	0.92 ms	2.96 ms
500	4.8 ms	12.6 ms

Known costs, current honesty: a scenario that touches a collection key keeps that key masked until dropped (no automatic un-masking without a cutoff); taint tracking uses a 64-bit mask, so past 64 live scenarios rounds fall back to scanning all of them; a bind's per-scenario subgraph also computes base values it never reads.

Divergence from the OCaml original

Feature set is 1:1 parity: Var, Map/MapN, Bind, Join, IfThenElse, Freeze, Both, ArrayFold, UnorderedArrayFold, ReduceBalanced, cutoffs, At/AtIntervals/BeforeOrAfter, Snapshot, StepFunction, Expert nodes, and debug/introspection tooling (ForAnalyzer, dot-graph export).

The execution engine is not a port - it's a redesign. The OCaml original recomputes on a single thread via a global height-ordered heap. This port replaces that with a concurrent, non-blocking wavefront scheduler (see below). Node identity uses Go generics (Node[T]) instead of OCaml's existential wrapping.

Concurrency model

Every node's compute function is assumed pure: same inputs, same output, no side effects other than the returned value. This is a hard contract the engine relies on for correctness, not just performance.

Stabilization is seed-driven and costs O(affected subgraph), not O(graph):

• every node carries an exact necessary refcount, maintained incrementally on observe/unobserve and edge changes (a node is necessary iff an observer transitively reads it);
• between rounds the State accumulates dirty seeds - Var.Sets, clock-driven staleness, nodes that just became necessary - and a round's work set is the upward closure of those seeds through necessary parents;
• a per-round coordinator goroutine owns all round bookkeeping and feeds ready nodes to workers over channels; a node recomputes only if one of its in-round children actually changed (cutoffs stop propagation for free);
• a round is handed to the worker pool only when it is worth it: the engine measures what recomputing a node actually costs and dispatches only if spreading the round beats running it on one goroutine. Cheap graphs stay serial on their own, so NewState(0) is a safe default rather than a gamble; pass NewState(1) to rule out concurrency entirely.

Nodes travel to workers in batches, since a rendezvous each way costs far more than a cheap node's recompute. On an Apple M3 Pro that puts the crossover near 0.3 us of work per node: below it rounds stay serial, and a wide round of 12 us nodes runs 7.5x faster on twelve workers. Numbers and method: benchmarks/README.md.

Because compute functions are pure, a round's result is identical to what a serial, height-ordered recompute would have produced, regardless of how the scheduler interleaved work. Height is still tracked and used to correctly reshape the graph under Bind, it just no longer drives recompute order.

Concurrency follows an actor model rather than shared locks: one scheduler goroutine (the graph actor) owns roots, seeds, necessity cascades, and topology changes; round coordinators own round state; workers only compute and message. State.Stabilize() queues a round and returns a *Future immediately; State.StabilizeAndWait() is the blocking convenience wrapper. Var.Set never blocks: it hands the change to the actor and returns, with channel FIFO ordering guaranteeing a subsequent stabilization sees it.

Usage

s := incremental.NewState(0) // 0 -> GOMAXPROCS workers; 1 -> always serial
defer s.Close()

x := incremental.NewVar(s, 1)
y := incremental.NewVar(s, 2)
sum := incremental.Map2(x.Watch(), y.Watch(), func(a, b int) int { return a + b }, incremental.NeverCutoff[int]())

obs := incremental.Observe(s, sum)
defer obs.Unobserve()

if err := s.StabilizeAndWait(); err != nil {
    // handle err
}
v, _ := obs.Value() // 3

x.Set(10)
_ = s.StabilizeAndWait()
v, _ = obs.Value() // 12

Collections (IncrMap)

Scalar nodes are the wrong shape for relational data. The collection layer propagates per-key diffs between map-valued nodes - O(keys changed), never O(rows):

orders := incremental.NewMapVar[int, Order](s)
big := incremental.MapFilter(incremental.MapNode[int, Order](orders),
	func(_ int, o Order) bool { return o.Amount >= 100 })
byRegion := incremental.MapGroupAggregate(big,
	func(_ int, o Order) string { return o.Region },
	0.0,
	func(a float64, o Order) float64 { return a + o.Amount },
	func(a float64, o Order) float64 { return a - o.Amount },
)
obs := incremental.ObserveMap(s, byRegion)
obs.OnDiffs(func(diffs []incremental.MapDiff[string, float64]) { /* push to clients */ })

100k rows through filter + group-sum: one changed row stabilizes in ~7 us (8.7x faster than a full recompute).

The operator set covers MapFilter, MapValues, three flavours of grouped aggregate, and equi-joins:

operator	aggregate shape	cost per changed source entry
MapGroupAggregate	invertible (sum, count)	O(1)
MapGroupMinBy / MapGroupMaxBy	extremes over an ordered projection	O(1), except a removal of the current extreme (one pass over the group's distinct values)
MapGroupReduce	any associative+commutative fold, no inverse	O(1) insert; a removal or in-place change refolds that group once per round

Joins come in three shapes, all costing O(keys changed) rather than O(inputs), and all emitting a single diff for a key that moves on both sides in one round:

operator	join	keyed by
MapJoin	inner equi-join on the key	the shared key
MapLeftJoin	left outer equi-join on the key	the left key
MapLookupJoin	foreign-key lookup (stream/table enrichment)	the source key

MapLookupJoin is the stream/table shape: source entries carry a foreign key, one table entry may serve many of them. It keeps a reverse index from table key to referring source keys, so a table change revisits only the rows that actually reference it - O(source keys changed + rows referencing changed table keys).

Windowing

There is deliberately no window operator. Windowing is a retention policy plus time semantics - which time (event or processing), what watermark, what to do with late data, when to emit - and those are choices a general-purpose library should not make for you. Neither the OCaml original nor Salsa has them either.

It composes from what is here:

• the event buffer is a MapVar keyed by event id;
• eviction is Delete on entries that fell out of the window, driven either by your ingest loop or by an AtIntervals alarm on a Clock - and the clock is logical, advanced by AdvanceClock, so feed timestamps can drive it as a watermark and tests stay deterministic;
• the aggregate is any of the grouped operators above, over that buffer;
• tumbling or sliding buckets fall out of the group key: G is any comparable, so struct{Bucket int64; Key K} works.

Eviction is the operation a window does constantly, which is why MapGroupMinBy/MapGroupMaxBy treat it as the fast path: removing a member that is not the current extreme is O(1).

Benchmarks

benchmarks/ compares this library against naive full recompute, hand-rolled dirty-flag memoization, and RxGo v2 on shared, correctness-gated scenarios. Headline (portfolio pricing: instrument prices -> sector sums -> total, Apple M3 Pro, Go 1.26, median of 3):

scenario, 1 changed input	incremental	naive full recompute
10k instruments, 101 outputs observed	12.9 us	8.9 us
100k instruments, total observed	10.3 us	113 us

Incremental's cost is flat in graph size (the 100k-input graph stabilizes a sparse change faster than the 10k one); naive's is linear. With trivially cheap per-node compute the break-even sits around ten thousand inputs on this machine - real per-node work moves it sharply lower. When most inputs change per round, full recompute wins instead. Full tables (including the cases this library loses), methodology, and fairness notes: benchmarks/README.md.

Development

go test -race ./...
go vet ./...
gofmt -l .

License

Apache License 2.0 - see LICENSE. The design follows Jane Street's Incremental (see NOTICE); the implementation is original.