Repository navigation
enhancement: streaming, pull-based operator pipeline with bounded memory behind the cursor API (follow-up to #432) #473
Description
Follow-up to #432, split off once its v3.0.0 part shipped. This issue is the streaming, pull-based operator pipeline with bounded memory, behind the cursor API that #432 already froze.
What already exists (v3.0.0, on v3)
- Core API (feat: cursor API — Minigraf::query and PreparedQuery::query (#432) #461):
Minigraf::query/PreparedQuery::queryreturn an owned,SendCursor(vars,next_batch(max_rows) -> Result<Option<Batch>>,close, row iterator). A cursor's answer is fixed at open.Batchkeeps its fields private, so it can become columnar here without a break. - Bindings (enhancement (v3.0.0): expose the query cursor in every binding #462, closed): every binding has a cursor object over
next_batch(python#6, java#9, android#6, swift#6, node#5, c#4, wasm#4), plusBrowserDb.query(feat: binding support — API-017/API-018 and BrowserDb cursor (#462, #465, #467) #468). - Scan side (enhancement: read-only open mode (
OpenOptions::read_only) for offline tooling, exposed through the bindings #429/enhancement: streaming fact-log iterator with full history (tx, asserted, valid bounds) and coarse filters #430/enhancement: raw log writer for building a fresh graph from explicit (tx, asserted, valid-time) records #431, closed): read-only open,fact_logandLogWriter.
Behind the API today, the existing evaluator computes the whole answer when the cursor opens (a cursor over a Vec). This issue replaces that one operator at a time, with no change to the public API or the bindings.
Open question first: what happens when an operator's state grows large?
Proposed direction (2026-10-11, see the comment below; to be settled in the spec): no memory budget by default and no spilling: index order first (1), plus optional accounting with a coded error (3). Option 2 is out.
#432 proposed spilling to temp files in a spill_dir. That was set aside on 2026-10-01: separate spill files break the single-file rule (PHILOSOPHY.md), add configuration, have no filesystem in the browser, and suit mobile poorly. Settle the design before building any spilling operator. The candidates:
- Index-ordered streaming: use the covering indexes' order (sort-grouped aggregation, merge joins) so most plans need no large state.
- Scratch pages inside the
.graphfile: spill to free-list pages that are never referenced by a meta page and are reclaimed on close or reopen. - A clear error when over budget: a new coded error naming the operator and the budget, so the caller can narrow the query.
They can be combined, for example 1 where the plan allows and 3 otherwise.
Goals
- Memory follows operator state, not input size. Pipelined operators hold one batch at a time under backpressure. Only the stateful operators (
Distinct,Sort, groupedAggregate, theHashJoinbuild side, theNotJoinright side,Recursive) hold more, and the planner avoids that state where index order allows. What happens when that state grows large is the open question above. - Incremental delivery. Results reach the caller as batches as soon as the pipeline produces them;
LIMIT nterminates early; a caller can stop consuming and the pipeline stops working. - Streaming aggregation.
count,sum,min,max,avg, and grouped aggregates run in one pass over the input stream; grouped aggregation uses a hash table, or a sort-based path when the input is already ordered on the group key. - Consistent snapshots. A cursor reads at a fixed transaction bound (
as-ofthe counter at open), so writes that commit while a cursor is open do not change its answer. The store is append-only and bitemporal, so this is a bound ontx, not a copy. - Cancellation and backpressure. A cursor can be closed early; across the FFI the consumer pulls batches, so a slow client cannot make the server buffer the answer.
- Streaming transactions. A large
transact/retract(ingestion writes tens of thousands of facts per commit) is parsed and appended to the WAL in chunks rather than parsed into oneVec<Fact>first; aWriteTransactionreads its own uncommitted chunks through the same operator interface.
Non-goals: changing Datalog semantics, changing the file format beyond what v8 already does, parallel execution (fits later; the operator model does not preclude it).
Design
Operator model. Pull-based (Volcano) iterators with open() / next_batch() -> Option<Batch> / close(), where a Batch is a fixed-capacity columnar block of bindings (say 1,024 rows) so per-row virtual-call overhead is amortised. Push-based would work too; pull is chosen because LIMIT, early close and cursor semantics fall out of it directly.
Operators.
Scan: an index range on any of the four covering indexes (EAVT, AEVT, AVET, VAET; AVET for attribute+value patterns, see perf: attribute+value patterns scan the whole attribute; AVET index is never read #518, parked onperf/518-avet-lookupwithCommittedReaderreads this operator should use) with the:as-ofbound and:valid-at/:any-valid-timepredicates pushed into the range, yielding facts in index order. This is also the primitive enhancement: streaming fact-log iterator with full history (tx, asserted, valid bounds) and coarse filters #430 asks for, exposed directly.IndexNestedLoopJoin: for each batch of the outer, probe the index for the inner pattern; replacesjoin_with_pattern.HashJoin: when neither side has a usable index; builds a hash table on the smaller side.Filter,Project,Distinct(hash, or skipped when index order makes rows unique),Union,NotJoin/anti-join (hash on the right side).Sort: in-memory sort, avoided when index order already gives the order; feedsLimitwith top-k when both are present.Limit/Offset: early termination propagatesclose()upstream.Aggregate: streaming for ungrouped; hash-grouped; sort-grouped when the child is ordered on the key.Recursive(semi-naive fixpoint per stratum): the delta and the accumulated relation live in an in-memory relation keyed like the index, replacingFactStorage+BTreeSet seen_facts;max_derived_factsbecomes a safety limit on the accumulated relation only. Magic-sets and stratification stay as rewrites in front of the planner.
Planner. The existing optimizer.rs/prepared.rs produce the clause order and index hints today; they become the producer of an operator tree instead of a plan for the matcher. The profiler (#185) gets per-operator counters for free.
Transactions. transact/retract take the EDN through a streaming parser that emits facts in chunks; each chunk is validated and appended to the WAL under the open write lock; the transaction's tx is assigned once at commit. A WriteTransaction's reads see its own chunks via a Scan over the uncommitted tail. Schema validation and the v8 multi-value fix (#371, #287) are unaffected.
Options. No required memory option. If the spec adopts the optional cap, OpenOptions (non-exhaustive) gains it, off by default; it gains no spill_dir. max_results keeps its meaning for execute only, and a cursor has no result cap. INT-020 is raised only by the recursive operator's accumulated-relation limit. Bindings expose the new options through their existing OpenOptions records. In the UniFFI shim counts are signed, and a usize::MAX sentinel needs the same i64::MAX mapping as wal_checkpoint_threshold (see the binding spec, decision 7).
Acceptance tests
Carried over from #432; the two items #432 already met are not repeated.
[:find ?e :where [?e :entity-type :type/function] :limit 10]over 10M facts completes in time and memory proportional to 10, not 10M.:limitis No working result-limiting/pagination mechanism for plain (non-recursive) :find queries #306, tracked in Tracker: production readiness for high-stakes deployments #383. This issue covers the streaming engine only.- Settle the large-state design (open question above) and record it in a spec under
docs/superpowers/specs/. Done: streaming engine spec (docs(spec): streaming query engine design (#473) #552). -
countover 10M facts runs in constant memory; groupedcountwith 1M groups completes with memory proportional to the groups, and with the optional cap set below that, fails with a clear coded error instead of aborting. Done:count_holds_no_rows(docs(551): measurement, acceptance and docs for the executor removal (#551 PR 3) #727; 7 MiB at 10M in the envelope),grouped_count_1m_groups(streaming engine 7/9: memory accounting and optional cap (QRY-012) (#473) #549,QRY-012). - A recursive rule deriving 5M facts completes with memory proportional to the derived relation;
max_derived_factsstill stops a runaway rule. Done:reach_5m(streaming engine 6/9: Recursive operator (#473) #548). - Closing a cursor after the first batch stops all upstream work, verified with operator counters (which also feed the profiler, post-1.0: build query profiler #185). Done:
cursor::tests::first_batch_then_close,close_after_first_batch(streaming engine 2/9: operator core, Scan and oracle harness (#473) #544, enhancement (v3.0.0): rework the query and transaction engine as a streaming, pull-based operator pipeline with bounded memory #432). - A 1M-fact
transactnever holds more than one chunk of parsed facts. Done:one_million_fact_transact_holds_one_chunk(streaming engine 8/9: streaming transactions (#473) #550). - FFI round trip through the Python binding with batches of 1,000; wall time and RSS recorded in the perf: benchmark queries at 100K–10M facts and publish a performance envelope #394 envelope. Done: Python FFI round trip (docs(551): measurement, acceptance and docs for the executor removal (#551 PR 3) #727).
- Existing
execute- and cursor-based tests (tests/cursor_test.rs) and the binding test suites pass unchanged. Done:tests/cursor_test.rspasses onv3(949813c). Its only changes are tests added since enhancement (v3.0.0): rework the query and transaction engine as a streaming, pull-based operator pipeline with bounded memory #432 (streaming engine 1/9: snapshots and page quarantine (#473) #543) and one moved frommax_resultstomax_derived_facts(max_results limits derived facts, not results: merge into max_derived_facts (v3.0.0) #553), which the alias still accepts. All seven bindings'v3branches pin 949813c with their test suites unchanged, and CI is green on each: python#13, node#16, wasm#15, c#9, swift#12, android#13, java#17 (project-minigraf/minigraf-*).
#379 (O(live) point queries) and #394 (performance envelope) are the measurements this must not regress.
Activity
adityamukho commentedon Oct 11, 2026
Data from the #394 envelope for this design: selective joins scale with their widest attribute, because each pattern without a bound entity is fetched in full by attribute before the join (
selective_fact_fetch). At 1M facts[:p83333 :follows ?f] [?f :name ?n](1 row) takes 0.61 s and a 10-row 3-clause join 1.24 s, against 12 µs for a point lookup; at 10M facts, 6.5 s and 13.5 s. Per-rowHashMap<String, Value>bindings with SipHash (insert, lookup, drop) are about 25% of the attribute scan and the 3-clause join.IndexNestedLoopJoinand columnar batches remove both. The envelope (examples/perf_envelope.rs) is the acceptance measurement, and so is the Python round trip (bench/envelope.pyin minigraf-python).adityamukho commentedon Oct 11, 2026
Proposed direction on memory (2026-10-11), to be settled in the spec
No memory budget by default, and no spilling.
Why no budget. With pull-based backpressure, pipelined operators (
Scan,Filter,Project,IndexNestedLoopJoin) hold one batch at a time. A required, tuned budget goes against zero-configuration.Why not "use all memory and fail on OOM". For an embedded library, an out-of-memory condition is never an error the caller can handle. It kills the host process:
handle_alloc_error).memory.growtraps the module.SQLite has
sqlite3_hard_heap_limit64returningSQLITE_NOMEMfor this reason.The stateful operators are more common than they look.
Distinct: (corrected) not needed for plain queries::findreturns a bag today (extract_variableskeeps duplicate rows), and enhancement: streaming, pull-based operator pipeline with bounded memory behind the cursor API (follow-up to #432) #473 keeps Datalog semantics. It's needed only where the engine already deduplicates:orbranches,count-distinct, and the recursive relation.Recursive: the accumulated relation is the worst case today (bug: a rule query fails with INT-020 on any database with 1,000,000 or more live facts #528: 3.2 GiB at 1M facts).HashJoinbuild side, theNotJoinright side,Sortand groupedAggregate.Direction
try_reserve. A real allocation failure then becomes a coded error where the platform reports one (wasm, no-overcommit systems) instead of an abort.max_results/max_derived_facts.These are options 1 and 3 of the open question. Option 2 (scratch pages in the
.graph) is out.Body edits in this pass
Scan: now covers all four indexes, including AVET for perf: attribute+value patterns scan the whole attribute; AVET index is never read #518. The old "EAVT/AEVT (VAET if added)" wording came from enhancement (v3.0.0): rework the query and transaction engine as a streaming, pull-based operator pipeline with bounded memory #432 and was wrong even for v7.Optionsand the acceptance tests.adityamukho commentedon Oct 11, 2026
Spec written:
docs/superpowers/specs/2026-10-11-streaming-engine-design.md(PR above). It settles the open large-state question: no default budget and no spill; byte accounting,try_reserve, and an optionalOpenOptions::query_memory_limitthat raisesQRY-012.Decisions taken in the brainstorm:
API-013is deprecated.Limitoperator is built here. The:limitsyntax and the 10M acceptance test move to No working result-limiting/pagination mechanism for plain (non-recursive) :find queries #306/Feature request: server-side :limit clause for Datalog queries #310.One change to the Transactions section above: a transaction with at least
wal_checkpoint_thresholdfacts writes copy-on-write pages and commits as a checkpoint. It does not append chunks to the WAL. Appending chunks would still leave every fact in the in-memory pending store after commit. Itstx_idis fixed when it switches.Sub-issues, in order: #543 snapshots/quarantine, #544 operator core + Scan + oracle, #545 joins, #546 not/or/Distinct, #547 Aggregate/Sort/Window, #548 Recursive, #549 memory cap, #550 streaming transactions, #551 removal/measurement/docs.
docs(spec): precise wording; snapshots share pending facts; large-tra…
docs(spec): address review of the streaming engine design (#473)
docs(spec): derived-fact limit follows #553 (#473)
Merge pull request #552 from project-minigraf/docs/473-streaming-engi…
fuzz(datalog_eval): skip inputs whose patterns join to a cross product (
ci: temporary binary-size allowance while engine and executor coexist (…
adityamukho commentedon Oct 24, 2026
The streaming engine is done on
v3: sub-issues #543–#551 are all closed, and #551's last PR merged as #727 (949813c). Every acceptance item is ticked with a link, except:limit. That item is #306, which is tracked in #383 and is not engine work.