adityamukho · Oct 7, 2026 · Oct 6, 2026
@@ -20,6 +20,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Cursor API: `Minigraf::query()` and `PreparedQuery::query()`** (#432). Both return a `Cursor` that hands a query's rows back in batches: `vars()`, `next_batch(max_rows) -> Result<Option<Batch>, _>`, `close()`, and an iterator over rows. The answer is fixed when the cursor opens, so writes that commit while it is open do not change it. A cursor borrows nothing from the database handle: it can outlive it and move to another thread. For now the answer is still computed when the cursor opens, under the same limits as `execute()`; the API is shaped so the engine can later produce rows as they are pulled without breaking callers. `query()` rejects `transact`, `retract` and `rule` with the new `API-012`, and a query with bind slots with `API-010`. `execute()` is unchanged. Rust API only for now; the bindings follow.
 - **`Minigraf::verify()` and `Minigraf::rebuild_indexes()`** (#373). Page checksums turn a torn or rotted page into an error, but they cannot see a page that is valid and says the wrong thing. `verify()` walks every committed page and returns an `IntegrityReport` with one structured error per finding: an index that holds different facts than EAVT or a different count than the meta page (new `STG-038`), keys out of order or a page reached twice (new `STG-039`), dictionary maps that disagree (new `STG-040`), ids with no dictionary entry (`STG-036`), and leaked, doubly used or wrongly free pages (`STG-035`). It keeps no per-fact state: indexes are compared by an order-independent digest. `rebuild_indexes()` rebuilds EAVT, AEVT, AVET and VAET from an intact index that agrees with another one (or the only intact one), derives the free list again, and commits like a checkpoint, including uncheckpointed writes; a crash at any point leaves the previous checkpoint. If no index can serve as the source, or the dictionary is damaged, it fails with new `STG-041` and writes nothing. Both are Rust API only for now; the bindings follow.
 
 ### Fixed

@@ -138,7 +138,8 @@ cargo run < demos/demo_negation.txt
 
 5. **`src/repl.rs`** — Interactive REPL; TTY-aware (suppresses prompts/banner for piped input)
 
-6. **`src/db.rs`** — Public API: `Minigraf::open/execute/prepare/begin_write/checkpoint/save/verify/rebuild_indexes`, `WriteTransaction`, `IntegrityReport`, `OpenOptions::page_cache_size`
+6. **`src/db.rs`** — Public API: `Minigraf::open/execute/query/prepare/begin_write/checkpoint/save/verify/rebuild_indexes`, `WriteTransaction`, `IntegrityReport`, `OpenOptions::page_cache_size`
+   - `src/cursor.rs`: `Cursor` / `Batch` — owned, `Send` cursor returned by `Minigraf::query` and `PreparedQuery::query` (#432); answer computed at open today, shaped for streaming operators
 
 7. **`src/wal.rs`** — Fact-level sidecar WAL, CRC32-protected entries, crash recovery
 
@@ -194,7 +195,7 @@ Auto-migrates v7 → v8 on open (spec §9, with a backup meta page). v1–v6 are
 
 ## Test Coverage
 
-**1267 tests** (1258 passing, 9 ignored; unit + integration + doc).
+**1285 tests** (1276 passing, 9 ignored; unit + integration + doc).
 See `docs/TEST_COVERAGE.md` for the full per-file breakdown.
 
 **Testing conventions** — see the Testing Conventions section below before writing any tests.

@@ -84,6 +84,12 @@ let results = db.execute(r#"
                    [?friend :person/name ?friend-name]])
 "#)?;
 
+// Cursor — read a large answer in batches; the answer is fixed at open
+let mut cursor = db.query("(query [:find ?name :where [?e :person/name ?name]])")?;
+while let Some(batch) = cursor.next_batch(1000)? {
+    for row in batch.rows() { /* row[0] is ?name */ }
+}
+
 // Explicit transaction — all-or-nothing
 let mut tx = db.begin_write()?;
 tx.execute(r#"(transact [[:alice :person/age 31]])"#)?;
@@ -105,7 +111,7 @@ let r2 = pq.execute(&[("tx", BindValue::TxCount(2)), ("entity", BindValue::Entit
 
 ```bash
 cargo run          # interactive Datalog REPL
-cargo test         # run 1267 tests
+cargo test         # run 1285 tests
 cargo run < demos/demo_recursive.txt   # recursive rules demo
 ```
 

@@ -65,6 +65,10 @@ Scope — format and storage integrity:
 - One current valid-time window per `(e, a, v)`: a later assertion replaces the earlier window, and two windows of one fact in one transaction are `API-011` (#435; done on `v3`)
 - `btree_page` fuzz target that reaches node decoding (#375)
 
+Scope — query API:
+
+- Cursor API: `Minigraf::query` and `PreparedQuery::query` return an owned `Cursor` that delivers rows in batches, with the answer fixed at open (#432; Rust API done on `v3`, with the current evaluator computing the answer at open). The binding cursor objects follow. Streaming operators replace the evaluator behind the cursor in 3.x without further API changes; how operators bound their memory (spill, scratch pages in the `.graph` file, or a clear error) is still open.
+
 Scope — release-gate testing and documentation:
 
 - Golden-file compatibility corpus for every format version (#391; done on `v3`: six v7 files from 2.0.3 checked after migration, seven v8 files frozen at `23aa57c`)

@@ -167,6 +167,7 @@ with no `CodedError` anywhere in its chain.
 | API-009 | WAL not initialized | Database API |
 | API-010 | Query with bind slots passed to execute() | Database API |
 | API-011 | Two valid-time windows for one fact in one transaction | Database API |
+| API-012 | Non-query command passed to query() | Database API |
 | INT-000 | Unclassified internal error | Internal |
 | INT-001 | WriteTransaction already in progress on this thread | Internal |
 | INT-002 | Invalid entity (API layer) | Internal |
@@ -2452,6 +2453,25 @@ let result = pq.execute(&[
 
 **Scenario**: A batch import writes the same fact once per period it was true, all in one transaction.
 
+### API-012 Non-query command passed to query()
+
+**Error text**: `only (query ...) commands can be opened as a cursor; got {}`
+
+**Cause**: `db.query()` returns a `Cursor` over the rows of a query. It was given a `transact`, `retract` or `rule` command, which has no rows. The `{}` names the command. Nothing is written or registered.
+
+**Resolution**:
+- Run writes and rule definitions with `db.execute()` (or inside a `WriteTransaction`), and open cursors only for `(query ...)` commands.
+
+```rust
+db.execute(r#"(transact [[:alice :person/name "Alice"]])"#)?;
+let mut cursor = db.query("(query [:find ?name :where [?e :person/name ?name]])")?;
+while let Some(batch) = cursor.next_batch(1000)? {
+    for row in batch.rows() { /* ... */ }
+}
+```
+
+**Scenario**: Code that dispatches every command string through one function switches from `execute()` to `query()` for all of them.
+
 ---
 
 ## INT — Internal Errors

@@ -0,0 +1,89 @@
+# Cursor API (#432, v3.0.0 scope) — design
+
+**Issue:** #432 (streaming, pull-based query engine). **Branch:** `v3`.
+**Scope decision (2026-10-01):** option 2 of the issue — land the public cursor API in
+3.0.0 with the current materialising evaluator behind it, then replace operators one by
+one in 3.x without further breaking changes. The spill/memory-budget design (goal 1)
+is explicitly deferred: the issue comment of 2026-10-03 asks for a revisit (index-ordered
+streaming, scratch pages inside the `.graph` file, or a clear error over budget) and the
+cursor API does not depend on it.
+
+## Goal
+
+Fix the public shape now, so that a client written against 3.0.0 never has to change when
+the engine starts streaming:
+
+- `Minigraf::query(&str) -> Result<Cursor, MinigrafError>`
+- `PreparedQuery::query(&[(&str, BindValue)]) -> Result<Cursor, MinigrafError>`
+- `Cursor::vars()`, `Cursor::next_batch(max_rows) -> Result<Option<Batch>, MinigrafError>`,
+  `Cursor::close(self)`, and `impl Iterator for Cursor` yielding
+  `Result<Vec<Value>, MinigrafError>` rows.
+- `Batch`: `len`, `is_empty`, `rows() -> &[Vec<Value>]`, `into_rows() -> Vec<Vec<Value>>`,
+  `IntoIterator`.
+
+`execute()` keeps its signature and behaviour; every existing test passes unchanged.
+
+## Decisions
+
+1. **`Cursor` is owned, `Send`, and has no lifetime.** A streaming cursor will need read
+   access to the store after `query()` returns. It gets that through the same `Arc`s
+   `PreparedQuery` already holds, never a borrow of `&Minigraf` and never a lock held for
+   the cursor's life: the store is append-only, so a snapshot is a bound on `tx_count`, not
+   a copy or a lock. Fixing this now is the one thing that would otherwise be a breaking
+   change later.
+2. **`next_batch` returns `Result`.** Today it cannot fail after `query()` succeeds, but a
+   streaming engine can fail mid-stream (a damaged page, the recursive operator's
+   `max_derived_facts` limit). Iterator items are `Result` for the same reason.
+3. **The public `Batch` is row-major.** Internal operator batches may become columnar; the
+   conversion happens once, at the cursor boundary, where rows leave the engine anyway.
+   Fields are private so the representation can change behind `rows()`/`into_rows()`.
+4. **`max_rows == 0` is treated as 1**, so a loop of `next_batch(n)` always makes progress
+   and `None` is the only end-of-results signal. Batches are never empty.
+5. **Snapshot.** A cursor's answer is fixed at `query()`: writes that commit while it is
+   open do not change it. Today this holds because the answer is computed at open; the
+   streaming engine must keep it by bounding reads at the `tx_count` current at open.
+   A test pins the guarantee.
+6. **Limits.** `query()` uses the same executor and limits as `execute()`.
+   `max_results` and `max_derived_facts` bound recursive rule evaluation (`INT-020`); they
+   do not cap the rows of a plain query, for either entry point. The issue's "a cursor has
+   no result cap" therefore needs no change now. When the recursive operator is rewritten,
+   `max_derived_facts` stays as its safety limit. `PreparedQuery::query` matches
+   `PreparedQuery::execute`, which does not apply the `OpenOptions` limits (unchanged here).
+7. **Commands other than `(query ...)`.** `query()` rejects `transact`, `retract` and `rule`
+   with new **`API-012`**: "only (query ...) commands can be opened as a cursor; got {}".
+   A query with bind slots fails with `API-010`, as `execute()` does. Called on a thread
+   that holds a `WriteTransaction`, it fails with `INT-001`, as `execute()` does (a cursor
+   there would silently miss the transaction's own writes).
+8. **`close(self)`** drops the cursor. It exists so the API names the early-stop operation
+   the streaming engine will propagate upstream; dropping without `close` is equivalent.
+9. **Out of scope:** `WriteTransaction::query` (additive later), `:limit` (#306, v3.1.0),
+   operator rewrite, spill/memory budget, FFI. The FFI `Cursor` object (`vars()`,
+   `next_batch(n) -> String`, `close()`) lives in the binding repos and gets its own issue;
+   it is additive over the core API.
+
+## Implementation
+
+- New module `src/cursor.rs` with `Cursor` and `Batch`; re-exported from `lib.rs`.
+  `Cursor { vars: Vec<String>, rows: std::vec::IntoIter<Vec<Value>> }`.
+- `Minigraf::query` parses, rejects non-queries (`API-012`) and unbound slots (`API-010`),
+  runs the read path `execute()` uses, and wraps the `QueryResults`.
+- `PreparedQuery::query` calls the existing substitute + execute path and wraps the result.
+- Error code `API-012` registered in `src/error.rs` and `docs/ERROR_REFERENCE.md`.
+
+## Tests (`tests/cursor_test.rs`)
+
+- Batches partition the `execute()` answer: same vars, same multiset of rows, each batch
+  `<= max_rows`, no empty batch, then `None` (and `None` again after the end).
+- `max_rows == 0` makes progress.
+- The iterator yields the same rows as `execute()`.
+- Empty result: `vars` set, first `next_batch` is `None`.
+- Aggregates, rules and `:as-of` through `query()` match `execute()`.
+- Snapshot: a cursor opened before a `transact` (and before a `WriteTransaction` commit)
+  returns the pre-transaction answer to the end; on a file-backed database too.
+- `transact`/`retract`/`rule` → `API-012`; bind slots → `API-010`; inside a
+  `WriteTransaction` thread → `INT-001`; parse error → the parser's code.
+- `PreparedQuery::query` with binds matches `PreparedQuery::execute`.
+- A recursive rule over `max_results` → `INT-020` from `query()`, as from `execute()`.
+- `Cursor` is `Send` and can be consumed on another thread after `db` is dropped.
+- Close after the first batch is accepted (counters for upstream work arrive with the
+  operator rewrite).
@@ -0,0 +1,135 @@
+//! Incremental delivery of query results (#432).
+//!
+//! [`crate::Minigraf::query`] and [`crate::PreparedQuery::query`] return a
+//! [`Cursor`], which hands the answer back in [`Batch`]es of rows. The answer
+//! is fixed when the cursor opens: writes that commit while it is open do not
+//! change it.
+//!
+//! The engine behind the cursor currently computes the whole answer when the
+//! cursor opens, so `max_results` and `max_derived_facts` apply exactly as they
+//! do to [`crate::Minigraf::execute`]. The API is shaped for a streaming engine
+//! that produces rows as they are pulled, which is why a cursor is owned (it
+//! borrows nothing from the database handle) and why
+//! [`Cursor::next_batch`] can fail.
+
+use crate::error::{ErrorCode, MinigrafError, err_coded};
+use crate::graph::types::Value;
+use crate::query::datalog::executor::QueryResult;
+
+/// A forward-only cursor over the rows of one query.
+///
+/// Pull rows with [`Cursor::next_batch`] or iterate over them; each item is one
+/// row aligned with [`Cursor::vars`]. Stop early with [`Cursor::close`] or by
+/// dropping the cursor.
+///
+/// ```
+/// # use minigraf::Minigraf;
+/// let db = Minigraf::in_memory().unwrap();
+/// db.execute(r#"(transact [[:alice :person/name "Alice"] [:bob :person/name "Bob"]])"#)
+///     .unwrap();
+///
+/// let mut cursor = db
+///     .query("(query [:find ?name :where [?e :person/name ?name]])")
+///     .unwrap();
+/// assert_eq!(cursor.vars(), ["?name".to_string()].as_slice());
+/// while let Some(batch) = cursor.next_batch(1000).unwrap() {
+///     for row in batch.rows() {
+///         println!("{:?}", row[0]);
+///     }
+/// }
+/// ```
+#[derive(Debug)]
+pub struct Cursor {
+    vars: Vec<String>,
+    rows: std::vec::IntoIter<Vec<Value>>,
+}
+
+impl Cursor {
+    /// Wrap the answer of a query. Any other result is a write, which the
+    /// callers reject before executing.
+    pub(crate) fn from_result(result: QueryResult) -> anyhow::Result<Self> {
+        match result {
+            QueryResult::QueryResults { vars, results } => Ok(Cursor {
+                vars,
+                rows: results.into_iter(),
+            }),
+            _ => Err(err_coded!(ErrorCode::Api002)),
+        }
+    }
+
+    /// The variable names of the `:find` clause, in order. Every row has one
+    /// value per variable.
+    pub fn vars(&self) -> &[String] {
+        &self.vars
+    }
+
+    /// The next batch of at most `max_rows` rows, or `None` once every row has
+    /// been returned. A batch is never empty; `max_rows == 0` is treated as 1.
+    ///
+    /// # Errors
+    ///
+    /// Reserved for failures while rows are produced. The current engine
+    /// computes the answer when the cursor opens and never fails here.
+    pub fn next_batch(&mut self, max_rows: usize) -> Result<Option<Batch>, MinigrafError> {
+        let rows: Vec<Vec<Value>> = self.rows.by_ref().take(max_rows.max(1)).collect();
+        if rows.is_empty() {
+            Ok(None)
+        } else {
+            Ok(Some(Batch { rows }))
+        }
+    }
+
+    /// Stop reading. Equivalent to dropping the cursor.
+    pub fn close(self) {}
 }
+
+impl Iterator for Cursor {
+    type Item = Result<Vec<Value>, MinigrafError>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.rows.next().map(Ok)
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (0, None)
+    }
+}
+
+/// One batch of rows from a [`Cursor`]. Each row is aligned with
+/// [`Cursor::vars`].
+#[derive(Debug, Clone, PartialEq)]
+pub struct Batch {
+    rows: Vec<Vec<Value>>,
+}
+
+impl Batch {
+    /// The number of rows in this batch.
+    pub fn len(&self) -> usize {
+        self.rows.len()
+    }
+
+    /// Whether this batch has no rows. Batches returned by
+    /// [`Cursor::next_batch`] are never empty.
+    pub fn is_empty(&self) -> bool {
+        self.rows.is_empty()
+    }
+
+    /// The rows of this batch.
+    pub fn rows(&self) -> &[Vec<Value>] {
+        &self.rows
+    }
+
+    /// Take the rows of this batch.
+    pub fn into_rows(self) -> Vec<Vec<Value>> {
+        self.rows
+    }
+}
+
+impl IntoIterator for Batch {
+    type Item = Vec<Value>;
+    type IntoIter = std::vec::IntoIter<Vec<Value>>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.rows.into_iter()
+    }
+}
@@ -15,6 +15,7 @@ use crate::graph::types::{Fact, TxId, VALID_TIME_FOREVER};
 /// in any practical context, avoiding the collision that `0` would have with the Unix
 /// epoch (1970-01-01T00:00:00Z), which is a legitimate `valid_from` value.
 pub(crate) const VALID_FROM_USE_TX_TIME: i64 = i64::MIN;
+use crate::cursor::Cursor;
 use crate::error::{ErrorCode, MinigrafError, bail_coded, err_coded};
 use crate::graph::FactStorage;
 use crate::graph::types::Value;
@@ -674,17 +675,58 @@ impl Minigraf {
             }
         } else {
             // Read-only: no lock needed
-            let mut executor = DatalogExecutor::new_with_rules_and_functions(
-                self.inner.fact_storage.clone(),
-                self.inner.rules.clone(),
-                self.inner.functions.clone(),
-            );
-            executor.set_limits(
-                self.inner.options.max_derived_facts,
-                self.inner.options.max_results,
-            );
-            executor.execute(cmd)
+            self.read_executor().execute(cmd)
+        }
+    }
+
+    /// The executor for a read-only command: no lock, the database's limits.
+    fn read_executor(&self) -> DatalogExecutor {
+        let mut executor = DatalogExecutor::new_with_rules_and_functions(
+            self.inner.fact_storage.clone(),
+            self.inner.rules.clone(),
+            self.inner.functions.clone(),
+        );
+        executor.set_limits(
+            self.inner.options.max_derived_facts,
+            self.inner.options.max_results,
+        );
+        executor
+    }
+
+    /// Run a query and return a [`Cursor`] over its rows.
+    ///
+    /// The cursor's answer is fixed when this returns: writes that commit
+    /// while it is open do not change it. The cursor owns what it needs, so it
+    /// can outlive this handle and move to another thread. While the engine
+    /// computes the whole answer up front, `max_results` and
+    /// `max_derived_facts` apply exactly as they do to [`Minigraf::execute`].
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if:
+    /// - A `WriteTransaction` is active on **this thread** (`INT-001`).
+    /// - Parsing fails.
+    /// - The command is not a `(query ...)` (`API-012`), or it has bind slots
+    ///   (`API-010`; use [`Minigraf::prepare`] and [`PreparedQuery::query`]).
+    /// - Execution fails.
+    ///
+    /// [`PreparedQuery::query`]: crate::PreparedQuery::query
+    pub fn query(&self, input: &str) -> Result<Cursor, MinigrafError> {
+        self.query_inner(input).map_err(MinigrafError::from)
+    }
+
+    fn query_inner(&self, input: &str) -> Result<Cursor> {
+        if is_write_tx_active() {
+            bail_coded!(ErrorCode::Int001);
+        }
+        let cmd = parse_datalog_command(input)?;
+        match &cmd {
+            DatalogCommand::Query(q) => reject_unbound_slots(q)?,
+            DatalogCommand::Transact(_) => bail_coded!(ErrorCode::Api012, "transact"),
+            DatalogCommand::Retract(_) => bail_coded!(ErrorCode::Api012, "retract"),
+            DatalogCommand::Rule(_) => bail_coded!(ErrorCode::Api012, "rule"),
         }
+        Cursor::from_result(self.read_executor().execute(cmd)?)
     }
 
     // ── Explicit transaction ──────────────────────────────────────────────────