enhancement (v3.0.0): rework the query and transaction engine as a streaming, pull-based operator pipeline with bounded memory #432

Summary

Every stage of query evaluation materialises its full input and output in memory, and the only protection is a hard cap that turns a large result into an error. This proposal replaces that with a streaming engine: a pull-based operator pipeline over batches of bindings, bounded memory per operator with spill where an operator must materialise, a cursor API that delivers results incrementally across the FFI, and a transaction path that streams large writes to the WAL. It targets v3.0.0 because the public QueryResult shape and the FFI surface change, and 3.0.0 is the breaking release.

Where the engine materialises today (branch v3 @ 02d66e15, 2026-09-27; identical on main @ dac151c0)

Consequences observed in production use (temporal_reasoning, a 4-9 GB graph of ~4M facts):

• a [:find ?e ?d :where [?e :description ?d]] returned a single 4.7 MB JSON blob; a count over :entity-type with an audit dry run held the server thread for 21+ minutes with RSS growing 4.2 to 5.7 GB and no way to interrupt (temporal_reasoning notes 2026-09-03);
• LIMIT-shaped questions ("first 10 modules matching X") cost the same as the full answer;
• any aggregation over a large intermediate (count, max over a join) is bounded by RAM, not by the size of the answer;
• the sidecar rebuild and the proposed offline migration (enhancement: streaming fact-log iterator with full history (tx, asserted, valid bounds) and coarse filters #430, temporal_reasoning#378) cannot read the graph without going through this path.

Goals

1. Bounded memory per operator. Memory use is a function of the operator's state (a hash table of groups, a sort run), never of the input size, and is capped by a configurable budget with spill to a temp file when exceeded.
2. Incremental delivery. Results reach the caller as batches as soon as the pipeline produces them; LIMIT n terminates early; a caller can stop consuming and the pipeline stops working.
3. Streaming aggregation. count, sum, min, max, avg, and grouped aggregates run in one pass over the input stream; grouped aggregation uses a hash table with a budget and spill, or a sort-based path when the input is already ordered on the group key.
4. Consistent snapshots. A cursor reads at a fixed transaction bound (as-of the 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 on tx, not a copy.
5. 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.
6. 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 one Vec<Fact> first; a WriteTransaction reads 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 EAVT/AEVT (VAET if added) with the :as-of bound and :valid-at/:any-valid-time predicates 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; replaces join_with_pattern.
• HashJoin: when neither side has a usable index; build side under the memory budget, spill by partition (Grace hash join) beyond it.
• Filter, Project, Distinct (hash with spill), Union, NotJoin/anti-join (hash on the right side, budgeted).
• Sort: external merge sort when the run exceeds the budget; feeds Limit with top-k when both are present.
• Limit/Offset: early termination propagates close() upstream.
• Aggregate: streaming for ungrouped; hash-grouped with spill; sort-grouped when the child is ordered on the key.
• Recursive (semi-naive fixpoint per stratum): the delta and the accumulated relation live in a budgeted temp relation (in-memory table that spills to a temp file keyed like the index), replacing FactStorage + BTreeSet seen_facts; max_derived_facts becomes 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.

API.

pub fn query(&self, datalog: &str) -> Result<Cursor, MinigrafError>;
impl Cursor {
    pub fn vars(&self) -> &[String];
    pub fn next_batch(&mut self, max_rows: usize) -> Result<Option<Batch>, MinigrafError>;
    pub fn close(self);
}
// convenience, unchanged signature, implemented as `query(..).collect()` under max_results:
pub fn execute(&self, input: &str) -> Result<QueryResult, MinigrafError>;

OpenOptions gains query_memory_budget: usize (bytes; default a few hundred MB) and spill_dir. max_results keeps its meaning for execute only; a cursor has no result cap. INT-020 is raised only by the recursive operator's accumulated-relation limit.

FFI. A Cursor object with vars(), next_batch(n) -> String (JSON rows) and close(); uniffi cannot carry an iterator, so this is the shape #430 already proposes. Bindings (python, node, wasm, java, swift, c) add a thin iterator over it. The MCP server in temporal_reasoning can then page results to the model and yield its event loop between batches.

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.

Acceptance tests (release gate)

• [:find ?e :where [?e :entity-type :type/function] :limit 10] over 10M facts completes in time and memory proportional to 10, not 10M.
• count over 10M facts runs in constant memory; grouped count with 1M groups completes under a 256 MB budget by spilling.
• A recursive rule deriving 5M facts completes under budget; max_derived_facts still stops a runaway rule.
• A cursor opened before a transact returns the pre-transaction answer to the end.
• Closing a cursor after the first batch stops all upstream work (verify via operator counters).
• A 1M-fact transact never holds more than one chunk of parsed facts.
• 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.
• Existing execute-based tests pass unchanged.

Relationship to the v3.0.0 roadmap

ROADMAP.md scopes v3.0.0 to format v8 and data integrity and explicitly keeps additive features out of it. This item is not additive: the QueryResult shape, the cursor type and the FFI surface are breaking, and 3.0.0 is the only breaking window before the 12-month v2.x support period. Two ways to reconcile:

1. Full engine in 3.0.0 (this proposal as written).
2. API in 3.0.0, operators incrementally. Land query() -> Cursor, the batch type and the FFI cursor in 3.0.0 with the current materialising evaluator behind them (a cursor over a Vec), then replace operators one by one in 3.x without further breaking changes. This keeps the bug: same-transaction multi-valued facts read back as ONE value, differing between EAVT and AEVT; a batched retract can leave retracted values visible #371 fix unblocked and still makes 3.0.0 the release after which no client has to change.

Either way, #430 (fact-log iterator), #429 (read-only open) and #431 (raw writer) are the Scan operator and its write-side counterpart and should be designed as part of this rather than separately. #379 (O(live) point queries) and #394 (performance envelope) are the measurements this must not regress.