@hecto/dataflow

A differential dataflow engine: build a graph of operators once, then feed it small changes and get small changes back, instead of recomputing a query's whole result on every write. This is the role @electric-sql/d2ts plays under TanStack DB; @hecto/db builds its live queries on this package.

import { createGraph, filter, map, materialize, output } from '@hecto/dataflow'

type Row = { id: string; done: boolean; title: string }

const graph = createGraph()
const rows = graph.newInput<string, Row>({ unique: true })

const { operator, view } = materialize<string, Row>()
rows.pipe(
  filter((row: Row) => !row.done),
  map((row: Row) => ({ ...row, title: row.title.toUpperCase() })),
  operator,
)
graph.finalize()

rows.sendEntries([['a', { id: 'a', done: false, title: 'ship it' }, 1]])
graph.step()

view.list() // the current rows, a stable reference until the next change
view.subscribe((changes) => {
  /* the keyed batch that just published */
})

The multiset model

Every element flowing through the graph is a (key, value, diff) triple. diff is a multiplicity: +1 asserts that a value is present under a key, -1 retracts it, and an update to an already-present key is both — a retraction of the old value and an assertion of the new one in the same batch. Batches are columnar: keys, values and diffs (an Int32Array) live in parallel arrays rather than an array of (value, diff) tuples, so operators run tight loops over parallel arrays and reuse the same output buffer from one step to the next. Resetting a reused buffer clears key and value references from its prior live range while retaining the backing capacity, so a large earlier batch does not keep its rows alive through later small updates.

Operators

| operator | what it does | |---|---| | map | transform each value; key and unique unchanged | | filter | drop values failing a predicate; unique unchanged | | flatMap | expand one value into zero or more; clears unique unless the caller asserts otherwise | | negate | flips the sign of every diff | | concat | merges two streams into one node | | keyBy / rekey | derive a new key from the value | | unkey | project a keyed stream back onto a single key | | tap | side-effecting passthrough, for debugging | | join | inner, left, right, full; equi-join, both sides arranged by key | | reduce | generic fold with retraction; the basis count, sum, avg, min, max are built on | | count, sum, avg, min, max | reduce instances | | distinct | one live value per key, by identity or structural equality | | consolidate | collapses a batch to net diffs per key; a no-op on a unique batch | | topK | bounded, ordered window with limit/offset | | orderBy | order over the whole collection, not just a window | | output | hands a node's batch to a plain callback | | materialize | keyed store behind output: get, list(), subscribe() |

The unique fast path

A batch carries unique: true when every key has at most one live value — which is always true of rows straight out of a db collection, since a collection is itself keyed. join, reduce, distinct and topK check this flag and skip multiplicity bookkeeping entirely on it: no per-key Map<value, diff>, one product per matched pair instead of a cross product, one aggregate state per key instead of a retraction replay. A batch loses uniqueness the moment an operator can no longer prove it holds — flatMap and rekey clear it unless the caller asserts otherwise — and it never comes back downstream.

Differences from d2ts

• Keyed columnar batches, not arrays of tuples. d2ts's MultiSet is a list of [value, multiplicity]. A batch here is { keys, values, diffs, length }, with the key derived once at the edge (hash.ts) and never reconstructed inside an operator.
• No JSON serialisation anywhere. d2ts keys values for consolidation by a structural hash of their serialisation. Consolidation here is a Map keyed by the key itself — a string or number, or a composite string joined with a separator — never JSON.stringify.
• No versions or frontiers by default. d2ts carries a version on every batch and tracks frontiers so a multi-source graph knows when a version is complete, which is what iterate needs. This package's default build has a single version and none of that machinery; it ships behind @hecto/dataflow/versions, gated and unopened, because nothing in this repo receives out-of-order batches.
• Identity-stable output. join memoises products per (leftKey, rightKey) while both sides are unchanged; reduce only emits a new aggregate object when the aggregate actually changed. A row that did not change keeps its reference, so a consumer can diff a materialized list by pointer before diffing rows by pointer.
• Stateless chains fuse into one loop at finalize(). A run of map/filter/flatMap/negate between two stateful operators compiles once, at graph construction, into a single function — not N separate operator hops per batch.

Deliberate omissions

• Fractional indices. d2ts uses them to minimise DOM moves under reordering. Not needed here: the output is a keyed window and materialize keeps row identity, so a UI diffs by key rather than by position.
• SQLite-persisted arrangements. Skipped; the Trace interface leaves the seat for it.
• iterate. Skipped; no recursive query exists in this repo to need it.

Performance

Measured on node 24 and bun 1.4, 100k rows with string ids, medians of 15 after warm-up, every shape's row count asserted against naive recomputation before timing.

| shape, one-row change | d2ts (node / bun) | dataflow (node / bun) | |---|---|---| | topK 50 insert | 23.4 / 11.1 ms | 0.0221 / 0.0258 ms | | topK 50 delete | 22.1 / 11.1 ms | 0.0204 / 0.0239 ms | | topK 50 update | 23.5 / 12.5 ms | 0.0165 / 0.0149 ms | | orderBy full-collection update | cannot run | 0.0055 / 0.0024 ms | | join 100k x 1k update | — | 4.5 to 10x faster than d2ts | | reduce count / sum / distinct | — | 7 to 77x faster than d2ts |

And the counterweight, which belongs next to the table above, not in a footnote:

• d2ts's filter and join seeds run at 0.6 to 1.26x of ours — as fast as or faster than this package. Seeding is this package's weak flank.
• filter's 1k-batch is a wash on node and a clear d2ts win on bun (0.52 to 0.61x).
• A true first load of 100k rows through db measures 82.6 ms on node. That is the slowest operation in either package, and it is what a user feels on first paint.

On orderBy, precisely rather than triumphantly: d2ts cannot sort a whole collection incrementally. Its orderBy is topK followed by an innerJoin back into the source to reattach values, so ordering everything means keying every row under one value — and that join becomes a cross-product over the group. Measured: 2,000 rows produced 4,000,000 output rows in about 4 seconds, and 50,000 rows exhausted node's heap. This is a specific structural limitation of d2ts's orderBy, not a general claim that d2ts is slow.

Size. The entry a where + select -> materialize live query actually pulls in is 4385 B gzipped, against @electric-sql/d2ts at 9856 B for its whole package. That comparison is against the tree-shaken query entry specifically — the full barrel is 12013 B, and quoting the smaller number without saying what it measures would repeat the kind of claim this programme's own verification log has spent its time correcting.