diff --git a/.openspec/adr/0040-streaming-execution-with-batch-as-wrapper.md b/.openspec/adr/0040-streaming-execution-with-batch-as-wrapper.md new file mode 100644 index 00000000..5691822d --- /dev/null +++ b/.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>`. 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, ExecError>` the only shape, which + forces every consumer through `collect::>()` 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)` + 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. diff --git a/.openspec/adr/index.md b/.openspec/adr/index.md index 2b4d7e6c..986af00b 100644 --- a/.openspec/adr/index.md +++ b/.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 | diff --git a/Cargo.toml b/Cargo.toml index 40e704c9..99457437 100644 --- a/Cargo.toml +++ b/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" diff --git a/src/vdbe.rs b/src/vdbe.rs index 39e3dad1..ee6da5c0 100644 --- a/src/vdbe.rs +++ b/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}; diff --git a/src/vdbe/exec.rs b/src/vdbe/exec.rs index 8cef6e72..951a5c15 100644 --- a/src/vdbe/exec.rs +++ b/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>, 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>, +} + +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>, 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>, 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)] diff --git a/tests/fuzz/fuzz_targets/vdbe_exec.rs b/tests/fuzz/fuzz_targets/vdbe_exec.rs index bbe65243..cd3f1803 100644 --- a/tests/fuzz/fuzz_targets/vdbe_exec.rs +++ b/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] = &[ diff --git a/tests/fuzz/seeds/vdbe_exec/result_row_emit_loop b/tests/fuzz/seeds/vdbe_exec/result_row_emit_loop new file mode 100644 index 00000000..f0b1efe7 Binary files /dev/null and b/tests/fuzz/seeds/vdbe_exec/result_row_emit_loop differ diff --git a/tests/unit/vdbe_streaming_execution_test.rs b/tests/unit/vdbe_streaming_execution_test.rs new file mode 100644 index 00000000..a56f8c44 --- /dev/null +++ b/tests/unit/vdbe_streaming_execution_test.rs @@ -0,0 +1,352 @@ +// Copyright 2026 Schuberg Philis +// SPDX-License-Identifier: Apache-2.0 +//! Acceptance for the streaming execution primitive (#683, ADR-0038): +//! `Execution::next_row` must deliver exactly the rows +//! `execute_transaction_step` delivers, in exactly the same order, +//! without buffering them all first. +//! +//! `run()` is implemented as a wrapper over `next_row`, so equivalence +//! is structural for anything the batch path can reach. What these tests +//! guard is the part that is *not* structural: the `pending` FIFO, whose +//! necessity depends on how many rows a single dispatch can emit, and +//! the terminal-state handling a streaming caller can observe but a +//! batch caller cannot. + +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::indexing_slicing +)] + +use std::cell::RefCell; +use std::path::Path; +use std::rc::Rc; + +use sqlite_rs::btree::TableCursor; +use sqlite_rs::codegen::{ + compile_select_with_catalog, compile_statement, resolve_from_table_schema, +}; +use sqlite_rs::header::DatabaseHeader; +use sqlite_rs::pager::Pager; +use sqlite_rs::parser::{parse_select, ParseOutcome}; +use sqlite_rs::record::Value; +use sqlite_rs::schema::{read_schema, read_views}; +use sqlite_rs::vdbe::{execute_transaction_step, Execution, Vm}; +use sqlite_rs::vfs::{MemoryVfs, PageSource, Vfs}; + +fn empty_db(page_size: u32) -> (MemoryVfs, DatabaseHeader) { + let mut page1 = vec![0u8; page_size as usize]; + page1[0..16].copy_from_slice(b"SQLite format 3\0"); + page1[16..18].copy_from_slice(&u16::try_from(page_size).unwrap().to_be_bytes()); + page1[18] = 1; + page1[19] = 1; + page1[28..32].copy_from_slice(&1u32.to_be_bytes()); + page1[56..60].copy_from_slice(&1u32.to_be_bytes()); + page1[100] = 0x0D; + page1[105..107].copy_from_slice(&u16::try_from(page_size).unwrap().to_be_bytes()); + + let mut header_bytes = [0u8; 100]; + header_bytes.copy_from_slice(&page1[..100]); + let header = DatabaseHeader::parse(&header_bytes).unwrap(); + + let mut vfs = MemoryVfs::new(); + vfs.insert("/test.db", page1); + (vfs, header) +} + +struct Db { + vfs: MemoryVfs, + pager: Rc>, + header: DatabaseHeader, + autocommit: bool, +} + +impl Db { + fn new() -> Self { + let page_size = 4096; + let (vfs, header) = empty_db(page_size); + let pager = Pager::open(&vfs, Path::new("/test.db"), page_size).unwrap(); + Self { + vfs, + pager: Rc::new(RefCell::new(pager)), + header, + autocommit: true, + } + } + + /// Replaces `index`'s root page with a structurally valid but *empty* + /// index leaf, so `PRAGMA integrity_check` reports an entry-count + /// mismatch for it. + /// + /// This is the only way to reach the multi-row-per-dispatch path: + /// `pragma::integrity_check` is the sole `Vm::emit_row` caller that + /// emits more than one row from a single dispatch, and a clean + /// database reports exactly one row (`"ok"`). Two details were + /// measured rather than assumed: zeroing the page is no good (a + /// malformed page errors instead of producing problem rows), and + /// emptying *one* index yields only one problem row — the count + /// mismatch — so several indexes have to be emptied to get several + /// rows out of the single dispatch. + fn empty_out_index(&mut self, index: &str) { + let page_size = 4096u32; + let root = { + let borrowed = self.pager.borrow(); + let mut cursor = TableCursor::new(&*borrowed, &self.header, 1); + let schemas = read_schema(&mut cursor, self.header.text_encoding).unwrap(); + schemas + .iter() + .flat_map(|t| t.indexes.iter()) + .find(|i| i.name == index) + .expect("index must exist") + .root_page + }; + + let mut page = vec![0u8; page_size as usize]; + page[0] = 0x0A; // leaf index b-tree page + page[1..3].copy_from_slice(&0u16.to_be_bytes()); // no freeblocks + page[3..5].copy_from_slice(&0u16.to_be_bytes()); // zero cells + page[5..7].copy_from_slice(&u16::try_from(page_size).unwrap().to_be_bytes()); + page[7] = 0; // no fragmented bytes + + // Page numbers are 1-based; `saturating_sub`/`checked_mul` keep + // the crate's `arithmetic_side_effects` lint satisfied without + // pretending a root page of 0 is reachable here. + let offset = u64::from(root.saturating_sub(1)) + .checked_mul(u64::from(page_size)) + .expect("page offset must fit in u64"); + let file = self.vfs.open_write(Path::new("/test.db")).unwrap(); + file.write_at(&page, offset).unwrap(); + file.sync().unwrap(); + + // The live pager has the old page cached; a fresh one is the + // simplest way to read the corrupted file. + let pager = Pager::open(&self.vfs, Path::new("/test.db"), page_size).unwrap(); + self.pager = Rc::new(RefCell::new(pager)); + self.autocommit = true; + } + + /// `compile_statement` deliberately does not handle `SELECT` — + /// query compilation needs the resolved `FROM` table, and the + /// richer join/compound/stats dispatch lives in the CLI. Single-table + /// `SELECT`s are compiled the way `examples/query.rs` does it; every + /// other statement goes through the ordinary dispatcher. + fn compile(&self, sql: &str) -> sqlite_rs::vdbe::Program { + let (schemas, views) = { + let borrowed = self.pager.borrow(); + let mut schema_cursor = TableCursor::new(&*borrowed, &self.header, 1); + let schemas = read_schema(&mut schema_cursor, self.header.text_encoding).unwrap(); + let mut view_cursor = TableCursor::new(&*borrowed, &self.header, 1); + let views = read_views(&mut view_cursor, self.header.text_encoding).unwrap(); + (schemas, views) + }; + if sql.trim_start().to_ascii_uppercase().starts_with("SELECT") { + let select = match parse_select(sql) { + ParseOutcome::Accepted(select) => *select, + other => panic!("failed to parse {sql}: {other:?}"), + }; + let from = select.from.as_ref().expect("test SELECTs all have a FROM"); + let table = resolve_from_table_schema(&from.first, &schemas).unwrap(); + return compile_select_with_catalog(&select, &table, &schemas).unwrap(); + } + compile_statement(sql, &schemas, &views).unwrap() + } + + /// The batch path, as any existing caller uses it. + fn exec(&mut self, sql: &str) -> Vec> { + let program = self.compile(sql); + let (rows, autocommit) = execute_transaction_step( + &program, + Rc::clone(&self.pager), + self.header, + self.autocommit, + ) + .unwrap(); + self.autocommit = autocommit; + rows + } + + /// The streaming path, pulled one row at a time. + /// + /// Read-only (`Vm::with_db`): a streaming caller cannot thread the + /// autocommit flag from outside the crate, and does not need to — + /// #683 scopes streaming to result-producing statements. Write + /// equivalence is structural, since `run` is a wrapper over + /// `next_row`, and is covered by the existing suite. + fn stream(&self, sql: &str) -> Vec> { + let program = self.compile(sql); + let source: Rc = Rc::clone(&self.pager) as Rc; + let mut execution = Execution::new(Vm::with_db(source, self.header), &program); + let mut rows = Vec::new(); + while let Some(row) = execution.next_row().unwrap() { + rows.push(row); + } + rows + } + + fn seed(&mut self, rows: usize) { + self.exec("CREATE TABLE t(a INTEGER, b TEXT)"); + for i in 0..rows { + self.exec(&format!("INSERT INTO t VALUES ({i}, 'row{i}')")); + } + } +} + +fn text_rows(rows: &[Vec]) -> Vec { + rows.iter() + .map(|r| match &r[0] { + Value::Text(s) => s.to_string(), + other => panic!("expected TEXT row, got {other:?}"), + }) + .collect() +} + +/// The core claim: same rows, same order, for a range of result shapes. +#[test] +fn streaming_matches_batch_for_every_result_shape() { + let mut db = Db::new(); + db.seed(25); + + for sql in [ + "SELECT a, b FROM t", + "SELECT a FROM t WHERE a > 20", + "SELECT a FROM t LIMIT 3", + "SELECT count(*) FROM t", + "SELECT a FROM t ORDER BY a DESC", + "SELECT a FROM t WHERE a > 9999", // empty result + ] { + let batch = db.exec(sql); + let streamed = db.stream(sql); + assert_eq!(batch, streamed, "streaming diverged from batch for: {sql}"); + } +} + +/// The regression guard the `pending` FIFO exists for. +/// +/// `Vm::emit_row` has three callers, and `pragma::integrity_check` emits +/// one row *per problem found* from a single dispatch. A drain that +/// assumed one row per step — popping off the back of `Vm`'s row vector +/// rather than through a queue — would reverse this output silently, +/// because every other opcode emits at most one row and looks correct +/// either way. +/// +/// A clean database reports a single `"ok"` row, which would pass any +/// ordering regardless, so this deliberately corrupts an index first to +/// force a genuinely multi-row result. No other test in the suite +/// reaches this path. +#[test] +fn streaming_preserves_multi_row_single_dispatch_order() { + let mut db = Db::new(); + db.exec("CREATE TABLE t(a INTEGER, b TEXT)"); + db.exec("CREATE INDEX t_a ON t(a)"); + db.exec("CREATE INDEX t_b ON t(b)"); + db.exec("CREATE INDEX t_ab ON t(a, b)"); + for i in 0..12 { + db.exec(&format!("INSERT INTO t VALUES ({i}, 'row{i}')")); + } + assert_eq!( + text_rows(&db.exec("PRAGMA integrity_check")), + vec!["ok"], + "database should start clean" + ); + + for index in ["t_a", "t_b", "t_ab"] { + db.empty_out_index(index); + } + + let batch = text_rows(&db.exec("PRAGMA integrity_check")); + let streamed = text_rows(&db.stream("PRAGMA integrity_check")); + + // The point of the whole test: more than one row from one dispatch. + assert!( + batch.len() > 1, + "corruption should yield several problem rows, got {batch:?}" + ); + assert_eq!(batch.len(), 3, "expected one problem per emptied index"); + assert_eq!( + batch, streamed, + "streaming reordered a multi-row single-dispatch result" + ); +} + +/// A wide `SELECT` over many rows: the ordinary one-row-per-dispatch +/// path, asserted positionally rather than just by length, so a +/// reversal or an off-by-one would fail rather than pass on count. +#[test] +fn streaming_yields_rows_in_emission_order() { + let mut db = Db::new(); + db.seed(50); + + let streamed = db.stream("SELECT a FROM t"); + assert_eq!(streamed.len(), 50); + for (i, row) in streamed.iter().enumerate() { + assert_eq!( + row[0], + Value::Integer(i as i64), + "row {i} out of order or wrong" + ); + } +} + +/// Halting is terminal. A batch caller can never observe this, because +/// it never holds an `Execution` after `run` returns; a streaming caller +/// can, and must not be able to re-enter a finished program. +#[test] +fn polling_past_the_end_keeps_returning_none() { + let mut db = Db::new(); + db.seed(3); + let program = db.compile("SELECT a FROM t"); + let source: Rc = Rc::clone(&db.pager) as Rc; + let mut execution = Execution::new(Vm::with_db(source, db.header), &program); + + let mut seen = 0; + while execution.next_row().unwrap().is_some() { + seen += 1; + } + assert_eq!(seen, 3); + + for _ in 0..3 { + assert!( + execution.next_row().unwrap().is_none(), + "a halted program must stay halted" + ); + } +} + +/// Abandoning a stream part-way must not disturb the database or the +/// next statement — the property a `Statement` handle dropped mid-result +/// depends on. +#[test] +fn abandoning_a_stream_leaves_the_database_usable() { + let mut db = Db::new(); + db.seed(20); + + { + let program = db.compile("SELECT a FROM t"); + let source: Rc = Rc::clone(&db.pager) as Rc; + let mut execution = Execution::new(Vm::with_db(source, db.header), &program); + assert!(execution.next_row().unwrap().is_some()); + // dropped here with 19 rows unread + } + + assert_eq!(db.exec("SELECT count(*) FROM t")[0][0], Value::Integer(20)); + db.exec("INSERT INTO t VALUES (999, 'after')"); + assert_eq!(db.exec("SELECT count(*) FROM t")[0][0], Value::Integer(21)); +} + +/// `autocommit` is what `execute_transaction_step` threads between +/// statements, so the streaming primitive has to report it truthfully +/// for a future facade to chain statements on one pager. +#[test] +fn autocommit_is_reported_after_a_read_only_statement() { + let mut db = Db::new(); + db.seed(2); + let program = db.compile("SELECT a FROM t"); + let source: Rc = Rc::clone(&db.pager) as Rc; + let mut execution = Execution::new(Vm::with_db(source, db.header), &program); + while execution.next_row().unwrap().is_some() {} + assert!( + execution.autocommit(), + "a plain SELECT must leave autocommit on" + ); +}