perf(zql): pull-function protocol for the fetch path - #6523
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
There was a problem hiding this comment.
🟡 Changes recommended
Moderate correctness and cursor-cleanup regressions remain across several pull-stream paths.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
Replaces iterator-based ZQL fetch pipelines with allocation-reducing pull streams across ZQL, SQLite, Zero Cache, and client consumers.
Changes:
- Introduces pull-stream primitives and migrates fetch operators.
- Preserves lazy evaluation, yield propagation, merging, and overlays.
- Updates consumers, tests, and benchmarks for explicit stream closure.
File summaries
| File | Description |
|---|---|
packages/zqlite/src/table-source.ts |
Converts SQLite fetches to pull streams. |
packages/zqlite/src/table-source.test.ts |
Updates SQLite fetch tests. |
packages/zql/src/query/measure-push-operator.ts |
Passes through pull streams. |
packages/zql/src/query/measure-push-operator.test.ts |
Updates stream mocks. |
packages/zql/src/ivm/yield.push.test.ts |
Adapts push-yield tests. |
packages/zql/src/ivm/yield.fetch.test.ts |
Adapts fetch-yield tests. |
packages/zql/src/ivm/view-apply-change.ts |
Pulls relationship children directly. |
packages/zql/src/ivm/view-apply-change.test.ts |
Updates relationship fixtures. |
packages/zql/src/ivm/union-fan-out.ts |
Passes through pull streams. |
packages/zql/src/ivm/union-fan-out.test.ts |
Updates fan-out assertions. |
packages/zql/src/ivm/union-fan-in.ts |
Implements pull-based merging. |
packages/zql/src/ivm/union-fan-in.test.ts |
Updates union tests. |
packages/zql/src/ivm/test/random-yield-source.ts |
Wraps pull streams with random yields. |
packages/zql/src/ivm/test/mode-yield-source.ts |
Converts mode-yield test streams. |
packages/zql/src/ivm/take.ts |
Converts Take fetch scans. |
packages/zql/src/ivm/take.fetch.test.ts |
Updates Take fetch tests. |
packages/zql/src/ivm/stream.ts |
Defines pull primitives and combinators. |
packages/zql/src/ivm/source.test.ts |
Updates source fetch tests. |
packages/zql/src/ivm/snitch.ts |
Converts fetch and filter instrumentation. |
packages/zql/src/ivm/skip.ts |
Converts Skip fetch processing. |
packages/zql/src/ivm/skip-yields.ts |
Reimplements yield filtering. |
packages/zql/src/ivm/push-accumulated.ts |
Uses empty relationship pull streams. |
packages/zql/src/ivm/push-accumulated.test.ts |
Updates relationship tests. |
packages/zql/src/ivm/predicate-pushdown.test.ts |
Updates fetch consumption. |
packages/zql/src/ivm/operator.ts |
Changes the input fetch contract. |
packages/zql/src/ivm/memory-source.test.ts |
Updates memory-source tests. |
packages/zql/src/ivm/join.ts |
Converts join fetch paths. |
packages/zql/src/ivm/join-utils.ts |
Converts join overlay streams. |
packages/zql/src/ivm/join-utils.test.ts |
Updates overlay tests. |
packages/zql/src/ivm/flipped-join.ts |
Converts batched join fetching. |
packages/zql/src/ivm/flipped-join.chunked.test.ts |
Tests chunked pull-stream behavior. |
packages/zql/src/ivm/filter.ts |
Replaces generator filtering. |
packages/zql/src/ivm/filter-operators.ts |
Adds resumable pull filtering. |
packages/zql/src/ivm/filter-operators.test.ts |
Updates filter lifecycle tests. |
packages/zql/src/ivm/fan-out.ts |
Tracks suspended filter branches. |
packages/zql/src/ivm/fan-in.ts |
Passes through filter verdicts. |
packages/zql/src/ivm/exists.ts |
Tracks suspended relationship counts. |
packages/zql/src/ivm/deferred-input.ts |
Converts deferred hydration. |
packages/zql/src/ivm/data.ts |
Defines relationship pull streams. |
packages/zql/src/ivm/catch.ts |
Drains and expands pull streams. |
packages/zql/src/ivm/cap.ts |
Converts capped fetches. |
packages/zql/src/ivm/array-view.ts |
Hydrates views through pull streams. |
packages/zql/src/ivm/array-view.test.ts |
Updates view fixtures. |
packages/zero-client/src/client/ivm-branch.test.ts |
Updates branch fetch tests. |
packages/zero-client/src/client/custom.test.ts |
Updates custom mutation tests. |
packages/zero-cache/src/services/view-syncer/pipeline-driver.ts |
Streams pipeline rows via pull protocol. |
packages/zero-cache/src/services/view-syncer/flipped-exists-fetch-filter.bench.ts |
Updates benchmark traversal. |
packages/zero-cache/src/services/run-ast.ts |
Converts AST result consumption. |
packages/zero-cache/src/auth/write-authorizer.ts |
Explicitly closes authorization fetches. |
packages/shared/src/btree-set.ts |
Adds allocation-free value iteration. |
Review details
Suppressed comments (4)
packages/zql/src/ivm/deferred-input.ts:91
- If
output.pushthrows during attachment, control reachesinput.destroy()without closing this active fetch stream. The formerfor...ofinvoked iterator cleanup on a loop-body exception; with a SQLite-backed pipeline this can leave the hydration cursor open. Wrap the loop intry/finallyand closestream.
const stream = input.fetch({});
for (let node = stream.next(); node !== undefined; node = stream.next()) {
if (node === 'yield') {
continue;
}
consume(output.push(makeAddChange(node), input));
packages/zql/src/ivm/filter-operators.ts:230
- When a filter delegate throws, this ends filter bookkeeping but does not close the input stream. The previous
for...ofclosed its iterator on an exception from the loop body, so this can now leave an upstream SQLite cursor open. Route this path throughclose()before rethrowing.
} catch (e) {
this.#end();
throw e;
packages/zql/src/ivm/filter-operators.ts:245
- If the upstream
close()throws,#end()is skipped, despite this class's guarantee thatendFilter()runs on close. Stateful outputs such asExiststhen retain session state. Usetry/finallyso filter teardown runs even when input cleanup fails.
close(): void {
if (!this.#ended) {
this.#input.close();
this.#end();
}
packages/zql/src/ivm/view-apply-change.ts:680
- An exception while building or recursively initializing a child leaves this relationship stream open. The replaced
for...ofclosed its iterator on abrupt completion; wrap the full loop intry/finallyand callchildren.close().
- Files reviewed: 51/51 changed files
- Comments generated: 11
- Review effort level: Balanced
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| fetch(req: FetchRequest): PullStream<Node | 'yield'> { | ||
| assert(!req.start, 'Cap does not support start'); | ||
| assert(!req.reverse, 'Cap does not support reverse'); | ||
|
|
| #pending: | ||
| | { | ||
| node: Node; | ||
| key: string | undefined; | ||
| count: {stream: PullStream<Node | 'yield'>; size: number} | undefined; | ||
| exists: boolean | undefined; | ||
| } |
| /** Which output suspended on 'yield', so re-entry resumes there. */ | ||
| #filterIndex = 0; |
| return new LazyPullStream(() => { | ||
| this.#output.beginFilter(); | ||
| return new FilterStartPull(this.#input.fetch(childReq), this.#output); | ||
| }); |
| try { | ||
| v = stream.next(); | ||
| } catch (e) { | ||
| // As the generators did: an exception records no state. | ||
| done = true; | ||
| throw e; | ||
| } |
| for (;;) { | ||
| if (this.#priming) { | ||
| while (this.#primeIdx < this.#streams.length) { | ||
| const v = this.#pullOne(this.#primeIdx); |
| for (const s of this.#streams) { | ||
| s.close(); | ||
| } |
| const children = childNodes(change.node, relationship); | ||
| for ( | ||
| let node = children.next(); | ||
| node !== undefined; | ||
| node = children.next() |
| const children = childNodes(node, relationship); | ||
| for ( | ||
| let childNode = children.next(); | ||
| childNode !== undefined; | ||
| childNode = children.next() |
| close(): void { | ||
| if (!this.#done) { | ||
| this.#done = true; | ||
| this.#inner.close(); | ||
| this.#onDone(); | ||
| } | ||
| } |
Chained generators are the most expensive way to move a row on Hermes. A four-deep pipeline over 5000 rows measured 31.9 ms against 16.2 ms for the same chain written as manual iterators and 1.2 ms for a plain loop, and a CPU profile of `hydrate: issues with creator` put 5.4% of the benchmark in generator resume plumbing alone, before counting what hides inside the generator bodies themselves. Three changes, all confined to the fetch path: - `generateRows` delegated with `yield*` to something that is already an `IterableIterator`. That whole generator layer bought nothing and cost a resume per row; it now returns the BTree's iterator directly. Single-use semantics are unchanged -- an `IterableIterator` returns itself from `[Symbol.iterator]()`, exactly as a generator does. - `#fetch` is no longer a generator. Its setup contains no yields, so the three exits become plain returns. The hot path -- no overlay, no start, no filters, which is what a plain scan and every join child-lookup take -- is now `ConstrainedRowIterator` rather than a generator loop. - `generateRows` picked its method with `data[reverse ? a : b]()`. The computed member forces a dynamic lookup Hermes cannot inline-cache; the ternary-of-calls form measured 18% faster on the same object. `#fetch` keeps its laziness. A generator body does not run until the first `next()`, and this one reads `#overlay` and `conn.lastPushedEpoch`, so a caller may fetch and only iterate after a push. `LazyStream` preserves that timing exactly, and propagates `return()` so early termination still closes the underlying scan -- a leaked SQLite cursor makes later writes on the same connection fail. Android emulator, median of 3 runs (rn-bench --repeat 3): hydrate: issues filtered open 87.95 -> 76.46 ms -13.1% hydrate: issues only 77.74 -> 72.64 ms -6.6% hydrate: issues with creator + comments 888.98 -> 849.31 ms -4.5% hydrate: issues with creator 405.42 -> 389.99 ms -3.8% hydrate: issues limit 50 11.72 -> 11.48 ms -2.0% Push is flat, as expected -- it does not run this path. Baseline spreads were 0.3-3.5% except 'add comment' at 6.9%.
Replaces the iterator protocol on the fetch path with a pull protocol:
interface PullStream<T> { next(): T | undefined; close(): void }
`next()` returns the value directly -- `undefined` marks the end -- so there
is no `{done, value}` object per row, and one object per stream rather than
one per stage. It is deliberately not `Iterable`: an adapter back to an
iterable would let any unconverted consumer keep paying the cost this exists
to remove, so there is none.
Chained generators are what this replaces, and they are expensive on both
engines. A four-deep pipeline over 5000 rows measured 31.9 ms of generators
against 16.2 ms of hand-written iterators on Hermes, and 4.0 ms against 1.2 ms
on V8 -- relative to a plain loop, 31x and 74x respectively.
Converted here: MemorySource (including the four overlay generators,
generateWithStart/Constraint/Filter, mergeSortedStreams and #fetchMulti),
join-utils' overlay pair, Join, Take, Cap, Skip, FilterStart/End, Exists,
Snitch, FlippedJoin, UnionFanIn's mergeFetches, DeferredInput, ArrayView,
view-apply-change and skipYields. `Node.relationships` returns a PullStream.
Two behaviours are easy to lose in the translation and are load-bearing:
- `close()` replaces what `for...of` did implicitly. Breaking out of a loop
called `.return()`, which is what closed SQLite cursors; the pull protocol
makes that explicit, so every early exit closes its stream.
- `mergeFetches` advances a branch before emitting the node it selected from
it, while `mergeSortedStreams` emits first and refills after. The two
merges genuinely differ; swapping the order reorders 'yield' markers.
Exists is the one filter that suspends -- counting a relationship pulls a
child stream that can emit 'yield' -- and it now holds that position in
explicit state rather than a parked generator, which let the generator-based
`FilterOutput.filter` API be deleted entirely.
Android/Hermes, median of 3 paired interleaved runs against main:
hydrate: issues filtered open 101.23 -> 68.64 ms -33.3%
hydrate: issues limit 50 13.12 -> 11.27 ms -13.9%
hydrate: issues with creator + comments 998.82 -> 873.14 ms -12.6%
hydrate: issues with creator 454.66 -> 395.81 ms -12.5%
hydrate: issues only 86.74 -> 76.30 ms -12.2%
V8 agrees and gains more (-38.5% to -17.3%). Push is flat except
'add comment (child relation)' at +2.0% (median of 6 paired rounds, range
0.0-4.3%): push does not run the fetch path, so it gets none of the benefit
while still paying the new protocol's per-object cost.
TableSource is a Source, so its fetch has to speak the same protocol as
MemorySource's. A generator structurally satisfies `next()`, which is why
this type-checked before it worked: `next()` returned `{value, done}`
instead of a Node, and there was no `close()`.
`generateWithYields` and `#mapFromSQLiteTypes` become pull streams, the
latter reading the SQLite cursor directly.
`FinallyPull` replaces the generator's `try/finally`, running the cleanup
exactly once on exhaustion, `close()`, or throw. `onDone` is hoisted above
the try so a throw from `debug.initQuery` -- before any row is read -- still
closes the cursor; 'SQLite iterator is closed when an error occurs before
#mapFromSQLiteTypes is iterated' covers precisely that case.
The conversion grew a class per operator, most of them the same pull/check/return loop. 28 pull-stream classes become 15. - `filterPull`, `takeWhilePull` and `mapPull` replace `WithFilter`, `WithConstraint`, `MatchesAllConstraints`, `SkipReverse` and `JoinPull`. `skipYields` falls out as one call, taking skip-yields.ts from 45 lines to 14. - `limitedScan` replaces `TakeInitialPull` and `CapInitialFetch`, which were the same class twice: read up to a limit, record what was seen, and treat early close as a bug. They differ only in what they accumulate. - `PullStreamBase` was an empty abstract class once `[Symbol.iterator]` came off it -- 23 classes extending it and 22 `super()` calls for no behaviour. Classes now say `implements PullStream`. - view-apply-change had its own `ArrayPullStream` and `SkipYieldsPull`, verbatim reimplementations of `pullOf` and `skipYields`. - `EmptyPullStream` was a class for a constant. Also renames the codemod's generated `__pullNNN` variables to what each stream is -- `parents`, `children`, `rows`, `candidates`, `branchRows` -- and drops the 18 bare block scopes that existed only to keep those generated names from colliding. Deliberately left alone: the two overlay classes in join-utils share ~20 lines of queue draining but their step logic is genuinely different, and the two merges are different algorithms -- a heap versus a linear scan with dedup, emitting in different orders relative to refill. Unifying either would trade clarity for line count.
Four consumers read fetch results as iterables. They type-checked against a
generator -- which structurally satisfies `next()` -- but `next()` now returns
the value directly, so they failed at runtime with "res is not iterable".
`write-authorizer`'s permission check ("does any row come back at all"),
`run-ast` and `pipeline-driver`'s scalar-subquery drains, the flipped-exists
bench, `toAdds`, `#streamNodes`, and `QueryFailureLoggingOperator`.
The `finally` blocks are load-bearing, not defensive. `for...of` called
`.return()` when a consumer stopped early, and that is what closed SQLite
cursors; the pull protocol requires an explicit `close()`. Without it an
aborted hydration leaks its cursor and the next write fails with "database
connection is busy executing a query" -- which is exactly what
'abandoned hydration tears down its pipeline' and the view-syncer hydration
timeout tests caught.
`[...input.fetch()]` no longer works now that a fetch returns a PullStream rather than an iterable; `drainPull` reads it to completion.
`filterPull` was named to coexist with the generator-based `FilterOutput.filter`. That method is gone, so the suffix distinguished it from nothing -- and it collided with `filterPull`, the stream combinator, leaving one identifier meaning two unrelated things. Also drops a stale `filter: vi.fn()` left in three test mocks after the generator API was deleted.
Also converts the zql-integration-tests row-collecting runner, which walked relationships with for...of.
1c7b352 to
3431104
Compare
| const withOverlay = generateWithOverlay( | ||
| startAt, | ||
| pkConstraint ? once(rowsIterable) : rowsIterable, | ||
| new RowScan(rowsIterable), |
There was a problem hiding this comment.
this used to do once rather than a full scan if a pkConstraint existed
|
DM to Arv reproduced below: ok. I think we need to manually close all these streams if they return early (either throw or early loop termination). At first I assumed it'd just get GC'ed and we're fine. Problem is that the SQLite backend holds a resource from SQLite (the prepared statement) that needs to be freed as soon as the stream is no longer used or else you can get this error on fetch: --- write with an abandoned cursor open --- This is the fix (by Claude): d3fbd4c and some tests that failed prior to the change: a475ecc I or you should have it add a third commit that reproduces the SQLite failure since that's really the thing we're trying to protect against. |
Replaces the iterator protocol on the fetch path with a pull protocol:
next()returns the value directly —undefinedmarks the end — so there isno
{done, value}object per row, and one object per stream rather than oneper stage.
It is deliberately not
Iterable, and there is no adapter back to one. Anescape hatch would let any unconverted consumer keep paying the exact cost this
exists to remove, and would let the old protocol creep back later.
This now also contains what was #6509 (de-generatoring
MemorySource#fetch),as its first commit. That PR is closed; reviewing this one covers both.
Why
Profiling the ZQL benchmarks on Hermes (harness in #6507) put 5.4% of
hydrate: issues with creatorin generator resume plumbing alone —nextandgeneratorPrototypeResumeframes — before counting what hides inside thegenerator bodies. A four-deep pipeline over 5000 rows, measured in the app's
own runtime:
Relative to a plain loop that is 31x on Hermes and 74x on V8 — V8 optimizes
loops and hand-written iterators hard, and generators are a barrier it will not
cross, so removing a generator layer pays more there. This is not a
Hermes-specific fix.
Results
Android emulator, median of 3 paired interleaved rounds,
main→ this branch:Trust the device numbers over V8's: per-round spreads on iOS are tight
(−30.9, −33.5, −33.3), while V8 on a loaded host produced one +109.8% round on
limit 50that the median absorbs.filtered openis the largest win and the one that required going all the way:at an intermediate stage where
Filterstill reached the source through acompatibility adapter it was −1.7%, because every row still allocated a result
object.
The one regression
push: add comment (child relation)is +2.0% (median of 6 paired rounds,range 0.0–4.3%, 5 of 6 positive). Push does not run the fetch path, so it gets
none of the benefit while still paying the new protocol's per-object cost. A
profile shows the cost is diffuse — push is ~43% BTree-and-view comparison,
which this does not touch — and no converted class appears in its hot frames.
The other five push benchmarks are flat.
Two behaviours that are easy to lose
close()replaces whatfor...ofdid implicitly. Breaking out of a loopcalled
.return(), and that is what closed SQLite cursors. Every early exitnow closes explicitly; the
finallyblocks inzero-cacheare load-bearing,not defensive. Without them an abandoned hydration leaks a cursor and the next
write fails with "database connection is busy executing a query" — caught by
abandoned hydration tears down its pipelineand the view-syncer hydrationtimeout tests.
mergeFetchesadvances a branch beforeemitting the node it selected from it;
mergeSortedStreamsemits first andrefills after. Swapping the order reorders 'yield' markers —
UnionFanIn propagates yieldcatches it.Existsis the one filter that suspends (counting a relationship pulls a childstream that can emit 'yield'). It now holds that position in explicit state
rather than a parked generator, which let the generator-based
FilterOutput.filterAPI be deleted outright.One Hermes-only edit
generateRowspicked its method withdata[reverse ? a : b](). The computedmember forces a dynamic lookup Hermes cannot inline-cache; the ternary-of-calls
form measured 18% faster there. I could not get a trustworthy V8 number for it
in isolation — the computed form timed 1.0 ms for one branch and 19.9 ms for
the other while the ternary sat at ~9.6 ms both ways, stable across two probe
designs. 1.0 ms for 2M allocating calls is 0.5 ns each, so V8's escape analysis
is scalar-replacing the returned object asymmetrically and the microbenchmark
is not measuring the property access. Treat that edit as Hermes-motivated and
V8-unknown; it is subsumed by the end-to-end numbers above.
Commits
Reviewable in dependency order:
MemorySourcefirst, then the protocol, theneach consumer package, then a refactor pass that folds the repeated pull-stream
shapes into combinators (28 classes → 15).
Verification
zql 1448 · shared 774 · zqlite 199 · zero-cache 5084 · zero-client 659 ·
zero-solid 60 · zql-integration-tests 1169. Repo-wide
check-types(44/44),lintandcheck-formatclean, and@rocicorp/zerobuilds.Two failures are not from this branch: zero-client's
logged-out client uses a private storage sentinel for idb namingasserts a hardcoded schema version 52against the current 53 and fails on
maintoo, and thezql-integration-testspg suite intermittently exhausts Postgres connections ("sorry, too many clients
already") — the affected files pass on their own.
The benchmark numbers above were taken before the final refactor, rename and
zero-solidcommits. Those are behaviour-preserving and the suites agree, butthey have not been re-measured.