feat: streaming Execution primitive, with run() as its wrapper (#683)

.openspec/adr/0040-streaming-execution-with-batch-as-wrapper.md

@@ -0,0 +1,105 @@
+# 0040: One streaming execution primitive, with the batch path as its wrapper
+
+Date: 2026-09-01
+
+## Context
+
+#683: every execution entry point (`execute`, `execute_with_params`,
+`execute_with_db`, `execute_with_db_and_params`, `execute_with_writable_db`,
+`execute_transaction_step`) returns a fully materialized
+`Vec<Vec<Value>>`. Reading row 0 of a large result therefore costs building
+row N, and there is no way for a caller to stop early without having already
+paid for everything.
+
+That blocks an embedding API. Spec 013/Req-7 (PR #678) asks for incremental row
+access, and a `Statement::next_row` cannot be layered on top of a function that
+has already collected every row before returning.
+
+It was also not implementable outside the crate. `fn dispatch`
+(`src/vdbe/exec.rs`) and `fn run` are private, and `ResultRow` does not hand a
+row to anyone — it calls `Vm::emit_row`, which pushes into a `Vec` *inside* the
+`Vm`. An external driver had no way to advance the program one instruction at a
+time, and no way to run `run`'s `Halt`-0 arm, which performs the implicit
+commit.
+
+Spike 014 (#682) prototyped and measured the alternatives on a 1,000,000-row
+result:
+
+| | peak heap | time to first row |
+|---|---:|---:|
+| batch (`execute_with_db`) | 137.7 MB | 5.36 ms |
+| streaming | 8.68 MB | 44.7 µs |
+
+Streaming's peak scales 1.03x for 2.5x the rows read, because it is bounded by
+`DEFAULT_PAGE_CACHE_CAPACITY` rather than by result size. Batch scales 2.20x
+and is unbounded in principle.
+
+## Decision
+
+Add one public primitive, `Execution`, holding the program-loop state (`vm`,
+`program`, `pc`, `steps`, `done`, `pending`) and exposing `new`, `next_row` and
+`autocommit`. **Reimplement `run()` as a wrapper that collects `next_row` into
+the same `Vec` it already returned.**
+
+The wrapper is the load-bearing half of this decision. Batch and streaming are
+then literally the same loop, so a behavioural difference between them is a bug
+in one rather than a divergence callers must reason about — and the existing
+suite becomes the equivalence proof: 1562 tests pass with identical per-binary
+counts before and after.
+
+`pending` is a FIFO, not a pop off the back. `Vm::emit_row` has three callers,
+and `pragma::integrity_check` emits *one row per problem found* from a single
+dispatch. Draining from the back would silently reverse
+`PRAGMA integrity_check` output, and every other opcode emits at most one row,
+so nothing else would look wrong. This is not hypothetical: no test in the
+suite reached the multi-row path before #683, which is why
+`vdbe_streaming_execution_test.rs` corrupts three indexes to force it and why
+that test was mutation-checked against a pop-off-the-back implementation.
+
+## Alternatives rejected
+
+- **Two parallel execution paths** — keep `run` as it is and add a separate
+  streaming loop beside it. Rejected: two copies of the step limit, the
+  program-counter bounds check and the `Halt`-0 implicit-commit arm, drifting
+  independently, with no test able to prove they agree. The wrapper shape costs
+  one `Vec` push per row and makes drift impossible.
+- **Making `dispatch` public and letting callers write their own loop.**
+  Rejected: `run`'s `Halt`-0 arm flushes the writer when `autocommit` is set,
+  and `vm.db`/`vm.autocommit` are private, so an external loop would silently
+  skip the implicit commit. The bug would surface as lost writes, not as a
+  compile error.
+- **Returning an `Iterator`** instead of a `next_row` method. Rejected for now:
+  `Iterator::next` cannot return a borrow of the `Execution`, and the error
+  type makes `Item = Result<Vec<Value>, ExecError>` the only shape, which
+  forces every consumer through `collect::<Result<_,_>>()` or a manual loop
+  anyway. `next_row` keeps the error in the caller's control flow. An
+  `Iterator` adapter can be added over this without changing it.
+- **Streaming at the transport level instead** — keep materializing inside the
+  engine and chunk on the way out. Rejected: it cannot fix peak heap, which is
+  the 137.7 MB figure above. The engine has to stop accumulating.
+
+## Consequences
+
+- `Execution` is public API and subject to the crate's stability policy from
+  here. Its scope is deliberately narrow: it does not own the pager, the
+  header, parameter binding or the transaction flag.
+- **Only read-only statements can be streamed from outside the crate today.**
+  `Vm::autocommit` is private and `execute_transaction_step` sets it
+  internally, so an external caller can build a streaming `Vm` via
+  `Vm::with_db` but cannot thread the transaction flag. That is sufficient for
+  result-producing statements, which is Req-7's target, and a
+  transaction-aware constructor is deferred to whichever ticket needs it rather
+  than speculatively added here.
+- **A facade built on this must chunk its transport.** #682 measured a
+  one-row-per-message channel at ~4.5 µs per row — 39.9x slower than batch on a
+  full drain — and chunking at ~1024 rows erasing the penalty entirely (7.30 ms
+  against a worker-batch baseline of 7.43 ms). `Execution` itself has no
+  channel and no chunk size; this constrains the facade, not the primitive.
+- The memory bound is the page cache, not zero. Any acceptance criterion
+  phrased as "without materializing" should be stated as
+  `min(pages_touched, DEFAULT_PAGE_CACHE_CAPACITY) x page_size`, independent of
+  result size.
+- Spec 013's Req-4 rationale needs correcting separately: `Value::Text(Rc<str>)`
+  and `Value::Blob(Rc<[u8]>)` make `Value` itself `!Send`, so a worker-thread
+  design pays an owned copy per text and blob value at the boundary. That is
+  ADR-0034's territory (as filed in PR #678), not this ADR's.

.openspec/adr/index.md

@@ -42,3 +42,4 @@ Specs record what the system must do; ADRs record **why it is shaped this way**
 | [0036](0036-pragma-synchronous-fsync-policy.md) | `PRAGMA synchronous` fsync-skip policy, and why `SynchronousMode` lives in `header.rs` | 2026-08-29 |
 | [0037](0037-macos-plain-fsync-not-fullfsync.md) | On macOS, `Vfs::sync` calls plain `fsync(2)`, not `std`'s `F_FULLFSYNC` | 2026-08-30 |
 | [0038](0038-cargo-registry-opt-in-not-committed.md) | Artifactory Cargo access is opt-in local config, never a committed source replacement | 2026-09-03 |
+| [0040](0040-streaming-execution-with-batch-as-wrapper.md) | One streaming execution primitive, with the batch path as its wrapper | 2026-09-01 |

Cargo.toml

@@ -128,6 +128,10 @@ path = "tests/unit/vdbe_write_opcodes_test.rs"
 name = "vdbe_integrity_check"
 path = "tests/unit/vdbe_integrity_check_test.rs"
 
+[[test]]
+name = "vdbe_streaming_execution"
+path = "tests/unit/vdbe_streaming_execution_test.rs"
+
 [[test]]
 name = "codegen_expr"
 path = "tests/unit/codegen_expr_test.rs"

src/vdbe.rs

@@ -36,7 +36,7 @@ pub use control::{
 };
 pub use exec::{
     execute, execute_transaction_step, execute_with_db, execute_with_db_and_params,
-    execute_with_params, execute_with_writable_db, ExecError, Step, Vm,
+    execute_with_params, execute_with_writable_db, ExecError, Execution, Step, Vm,
 };
 pub use explain::{explain, ExplainRow};
 pub use functions::{call as call_function, like_match, FunctionError};

src/vdbe/exec.rs

@@ -1079,62 +1079,142 @@ pub fn execute_transaction_step(
     run(vm, program)
 }
 
-fn run(mut vm: Vm, program: &Program) -> Result<(Vec<Vec<Value>>, bool), ExecError> {
-    // #509: `steps`/`pc` are both backstops against pathological programs
-    // (a step-limit runaway, a jump target past the end), not values any
-    // real program comes close to overflowing (`MAX_STEPS` is 50_000_000,
-    // program lengths are a handful of instructions per SQL statement) —
-    // `saturating_add` keeps the overflow-safety `arithmetic_side_effects`
-    // lint demands without `checked_add`'s per-step `Option` construction
-    // and `.ok_or` unwrap on this hot loop. (A batched, check-every-N-steps
-    // variant of the `MAX_STEPS` comparison was also tried and measured no
-    // further win beyond this — see the closing comment on #509 — so it
-    // was dropped rather than kept as unjustified complexity.)
-    let mut pc = 0usize;
-    let mut steps = 0u32;
-    loop {
-        steps = steps.saturating_add(1);
-        if steps > MAX_STEPS {
-            return Err(ExecError::StepLimitExceeded);
+/// One statement's execution, driven one result row at a time.
+///
+/// `execute_with_db` and friends materialize every row into a `Vec`
+/// before the caller sees the first one, so reading row 0 of a large
+/// result costs building row N (#683, ADR-0040). Spike 014 (#682)
+/// measured that at 137.7 MB peak heap and 5.36 ms to first row for a
+/// 1,000,000-row result, against 8.68 MB and 44.7 µs here.
+///
+/// Nothing here changes what a program computes: [`run`] is a wrapper
+/// that collects `next_row` into the same `Vec` it always returned, so
+/// the batch path and the streaming path are literally the same loop. A
+/// behaviour difference between them would therefore be a bug in one,
+/// not a divergence callers have to reason about.
+///
+/// The `pending` queue is not redundant. `Vm::emit_row` has three
+/// callers, and `PRAGMA integrity_check`
+/// (`src/vdbe/pragma.rs::integrity_check`) emits *N* rows from a single
+/// dispatch — one per problem found. A drain that assumed one row per
+/// step would silently reverse that output, so rows move through a FIFO
+/// rather than being popped off the back.
+pub struct Execution<'p> {
+    vm: Vm,
+    program: &'p Program,
+    pc: usize,
+    steps: u32,
+    done: bool,
+    pending: std::collections::VecDeque<Vec<Value>>,
+}
+
+impl<'p> Execution<'p> {
+    /// Binds `vm` to `program` without executing anything yet.
+    pub fn new(vm: Vm, program: &'p Program) -> Self {
+        Self {
+            vm,
+            program,
+            pc: 0,
+            steps: 0,
+            done: false,
+            pending: std::collections::VecDeque::new(),
+        }
+    }
+
+    /// Runs until the program produces its next result row, returning
+    /// `None` once it has halted. Errors are terminal: `done` is set
+    /// before returning one, so a caller that keeps polling gets `None`
+    /// rather than re-entering a halted program.
+    pub fn next_row(&mut self) -> Result<Option<Vec<Value>>, ExecError> {
+        if let Some(row) = self.pending.pop_front() {
+            return Ok(Some(row));
+        }
+        if self.done {
+            return Ok(None);
         }
-        let Some(instr) = program.get(pc) else {
-            return Err(ExecError::ProgramCounterOutOfRange { pc });
-        };
-        match dispatch(&mut vm, pc, instr)? {
-            Step::Next => {
-                pc = pc.saturating_add(1);
+        // #509: `steps`/`pc` are both backstops against pathological
+        // programs (a step-limit runaway, a jump target past the end),
+        // not values any real program comes close to overflowing
+        // (`MAX_STEPS` is 50_000_000, program lengths are a handful of
+        // instructions per SQL statement) — `saturating_add` keeps the
+        // overflow-safety `arithmetic_side_effects` lint demands without
+        // `checked_add`'s per-step `Option` construction and `.ok_or`
+        // unwrap on this hot loop. (A batched, check-every-N-steps
+        // variant of the `MAX_STEPS` comparison was also tried and
+        // measured no further win beyond this — see the closing comment
+        // on #509 — so it was dropped rather than kept as unjustified
+        // complexity.)
+        loop {
+            self.steps = self.steps.saturating_add(1);
+            if self.steps > MAX_STEPS {
+                self.done = true;
+                return Err(ExecError::StepLimitExceeded);
             }
-            Step::Jump(target) => pc = target,
-            Step::Halt { code: 0, .. } => {
-                // A program with no explicit `Transaction` (#194's
-                // original behavior, unchanged) treats a successful
-                // `Halt` as an implicit commit, flushing any pending
-                // write-opcode changes before returning. A `Vm::with_db`
-                // (read-only) or a writable `Vm` that never actually
-                // wrote anything both take the cheap
-                // `writer.is_none()`/`dirty.is_empty()` no-op path.
-                //
-                // #360: a program that opened an explicit transaction
-                // (`Transaction` opcode, `vm.autocommit == false`) and
-                // hasn't reached a matching `AutoCommit` yet does
-                // neither — one SQL statement is one `Program`/`Vm`
-                // (see `execute_transaction_step`), so `BEGIN`'s own
-                // `Halt` running with `autocommit == false` is the
-                // normal, expected case: the transaction stays open,
-                // `vm.autocommit` carries that forward to whichever
-                // `Vm` runs the next statement on this same `Pager`.
-                if let Some(db) = &vm.db {
-                    if vm.autocommit {
-                        if let Some(writer) = &db.writer {
-                            writer.borrow_mut().flush()?;
+            let Some(instr) = self.program.get(self.pc) else {
+                self.done = true;
+                return Err(ExecError::ProgramCounterOutOfRange { pc: self.pc });
+            };
+            match dispatch(&mut self.vm, self.pc, instr)? {
+                Step::Next => {
+                    self.pc = self.pc.saturating_add(1);
+                }
+                Step::Jump(target) => self.pc = target,
+                Step::Halt { code: 0, .. } => {
+                    // A program with no explicit `Transaction` (#194's
+                    // original behavior, unchanged) treats a successful
+                    // `Halt` as an implicit commit, flushing any pending
+                    // write-opcode changes before returning. A
+                    // `Vm::with_db` (read-only) or a writable `Vm` that
+                    // never actually wrote anything both take the cheap
+                    // `writer.is_none()`/`dirty.is_empty()` no-op path.
+                    //
+                    // #360: a program that opened an explicit transaction
+                    // (`Transaction` opcode, `vm.autocommit == false`) and
+                    // hasn't reached a matching `AutoCommit` yet does
+                    // neither — one SQL statement is one `Program`/`Vm`
+                    // (see `execute_transaction_step`), so `BEGIN`'s own
+                    // `Halt` running with `autocommit == false` is the
+                    // normal, expected case: the transaction stays open,
+                    // `vm.autocommit` carries that forward to whichever
+                    // `Vm` runs the next statement on this same `Pager`.
+                    if let Some(db) = &self.vm.db {
+                        if self.vm.autocommit {
+                            if let Some(writer) = &db.writer {
+                                writer.borrow_mut().flush()?;
+                            }
                         }
                     }
+                    self.done = true;
+                    return Ok(None);
+                }
+                Step::Halt { code, message } => {
+                    self.done = true;
+                    return Err(ExecError::Halted { code, message });
+                }
+            }
+            if !self.vm.rows.is_empty() {
+                self.pending.extend(self.vm.rows.drain(..));
+                if let Some(row) = self.pending.pop_front() {
+                    return Ok(Some(row));
                 }
-                return Ok((vm.rows, vm.autocommit));
             }
-            Step::Halt { code, message } => return Err(ExecError::Halted { code, message }),
         }
     }
+
+    /// The autocommit flag as it stands, for threading into the next
+    /// statement on the same `Pager` (see [`execute_transaction_step`]).
+    pub fn autocommit(&self) -> bool {
+        self.vm.autocommit
+    }
+}
+
+fn run(vm: Vm, program: &Program) -> Result<(Vec<Vec<Value>>, bool), ExecError> {
+    let mut execution = Execution::new(vm, program);
+    let mut rows = Vec::new();
+    while let Some(row) = execution.next_row()? {
+        rows.push(row);
+    }
+    Ok((rows, execution.autocommit()))
 }
 
 #[cfg(test)]

tests/fuzz/fuzz_targets/vdbe_exec.rs

@@ -4,16 +4,49 @@
 
 use libfuzzer_sys::fuzz_target;
 
-use sqlite_rs::vdbe::{execute, Collation, Instruction, Opcode, Program, P4};
+use sqlite_rs::vdbe::{Collation, Execution, Instruction, Opcode, Program, Vm, P4};
 
-// Discharges spec 009's no-panic-totality obligation (#89): `execute()`
-// must never panic on an arbitrary instruction stream, including
+// Discharges spec 009's no-panic-totality obligation (#89): dispatching
+// an arbitrary instruction stream must never panic, including
 // out-of-range register indices, malformed P4 operands, jumps to
 // nonexistent addresses, and adversarial loops (bounded by `MAX_STEPS`,
 // surfaced as a structured `Err` rather than a hang).
+//
+// Rows are pulled through `Execution` (#683, ADR-0040) and dropped
+// rather than collected by `execute()`, and the pull is capped. Those
+// two bounds are what keep an adversarial program from exhausting
+// memory or time before `MAX_STEPS` can stop it:
+//
+//   * `execute()` hands the caller every row the program emitted, so it
+//     necessarily holds them all. The decoder can build `ResultRow`
+//     with a register range up to `MAX_REGISTERS` jumped back to
+//     forever, which grows that result set without bound — 4.3 GB in
+//     ~1400 rows, the OOM in CI run 33857317005 that seed
+//     `result_row_emit_loop` reproduces. Capping accumulation inside
+//     the engine instead would be wrong: a `SELECT` that legitimately
+//     returns N rows must be allowed to return N rows, and real
+//     programs come from codegen, never from a caller handing the VM
+//     raw bytecode.
+//   * A wide `ResultRow` costs its whole register range per execution,
+//     so the same loop is unbounded in *time* even once rows are
+//     dropped. `MAX_ROWS` bounds the emitted work per input.
+//
+// Dispatch coverage is unchanged by draining rather than collecting:
+// per ADR-0040 `run()` — and so `execute()` — is this same loop plus a
+// `Vec::push`.
+const MAX_ROWS: usize = 64;
+
 fuzz_target!(|data: &[u8]| {
     let program = decode_program(data);
-    let _ = execute(&program);
+    let mut execution = Execution::new(Vm::new(), &program);
+    for _ in 0..MAX_ROWS {
+        // Each row is dropped as it arrives: peak is one row, not the
+        // whole result set.
+        match execution.next_row() {
+            Ok(Some(_row)) => {}
+            Ok(None) | Err(_) => break,
+        }
+    }
 });
 
 const OPCODES: &[Opcode] = &[