feat(vdbe): seek the inner side of an INNER rowid-join with a live cursor (B5b-2a)

The VDBE two-table INNER equi-join no longer materializes and scans the whole
inner table when the `ON` binds the inner table's INTEGER PRIMARY KEY
(`… JOIN t ON o.x = t.<ipk>`). Instead it scans only the outer table and, per
outer row, seeks the single matching inner row through a live b-tree cursor
(`read_row` → `TableCursor::seek`), reusing the tree-walker's own
value→rowid coercion (`eval::to_i64`) so no new coercion semantics are
introduced. The full `ON` is re-evaluated against the assembled row (the
superset invariant), so every rowid-coercion corner (`= 2.5`, text/blob keys,
`NULL`) filters exactly as the materialized cross-product would.

Routing is parity-gated: any shape outside the guard (NATURAL/USING, non-base
sources, non-ipk key, ambiguous columns, or a projection the single-cursor
compiler can't take) falls through to the existing materialized path, and the
VDBE returns `Unsupported` → tree-walker fallback, so correctness cannot
regress. Verified VDBE-vs-tree-walker and vs sqlite3 3.50.4
(`tests/vdbe_live_cursor.rs`).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: MagicalTux

ROADMAP.md

@@ -1169,12 +1169,30 @@ gated on VDBE-vs-tree-walker parity, so it can't regress correctness):
   - **B5c-2b** — compile `Subquery`/`Exists`/`InSelect` that read an outer column.
 - **B5b-2 — seek-driven inner cursor over real storage (the big B8 step).** The
   nested-loop join currently materializes each table's rows into in-memory
-  row-sets. This step gives the VDBE interpreter **live storage cursors**
-  (`OpenRead` + a b-tree `TableCursor`) so the inner side is *seeked* by
-  rowid/PK/index instead of materialized — mirroring the tree-walker's inner-join
-  seeks. B5b-1's multi-cursor opcodes (`RewindC`/`ColumnC`/`NextC`) are the
-  foundation. This is the largest remaining VDBE piece and the prerequisite that
-  makes correlated-subquery (B5c-2) and window streaming worthwhile.
+  row-sets. This step gives the join a **live storage cursor** so the inner side
+  is *seeked* by rowid/PK/index instead of materialized — mirroring the
+  tree-walker's inner-join seeks. This is the largest remaining VDBE piece and the
+  prerequisite that makes correlated-subquery (B5c-2) and window streaming
+  worthwhile.
+  - **B5b-2a — INNER-join inner rowid seek (INTEGER PRIMARY KEY). DONE.** A
+    two-table `… JOIN t ON o.x = t.<ipk>` now seeks the single matching inner row
+    with a *live* b-tree cursor (`read_row` → `TableCursor::seek`) instead of
+    scanning + materializing the whole inner table; only the outer table is
+    scanned. Modelled on the outer-join *row-assembly* path (build the joined rows
+    in Rust, then reuse `compile_table_select`/`run_rows` for projection / WHERE /
+    ORDER BY / GROUP BY / DISTINCT / LIMIT). Correctness rides the superset
+    invariant: after the seek the full `ON` is re-evaluated against the assembled
+    row, so every rowid-coercion corner (`= 2.5`, text/blob keys, `NULL`, duplicate
+    outer keys) is filtered exactly as the materialized cross-product would.
+    Narrowly routed (single INNER `Eq`-on-ipk, both plain base tables, ipk by its
+    declared column name — a bare `rowid` alias / compound `ON` / derived source
+    defers); any unhandled projection falls through to the materialized path.
+    Verified VDBE-vs-tree-walker and vs sqlite3 3.50.4 (`tests/vdbe_live_cursor.rs`).
+  - **Remaining:** in-*interpreter* `OpenRead`/`SeekRowid` opcodes over B5b-1's
+    multi-cursor foundation (so the seek lives in bytecode, not row-assembly);
+    seek by a **secondary index** / `WITHOUT ROWID` PK; the bare-`rowid`-alias `ON`
+    spelling (blocked on a pre-existing divergence — `t.rowid` in a join projection
+    already errors on the tree-walker); N-table chains.
 - **B1c — RIGHT/FULL join inner seeks.** INNER/LEFT already seek; RIGHT/FULL still
   materialize the inner table.
 
@@ -1452,7 +1470,10 @@ reasonable order:
 1. **B5b-2 — live storage cursors on the VDBE.** The largest remaining VDBE piece;
    it turns the materialized inner join into a seek-driven one and is the
    prerequisite for streaming correlated subqueries and windows. Perf/coverage,
-   parity-gated, low risk.
+   parity-gated, low risk. **B5b-2a landed** (INNER-join inner rowid seek on an
+   INTEGER PRIMARY KEY — live `TableCursor::seek`, no inner materialization);
+   remaining sub-steps (in-interpreter seek opcodes, secondary-index / WITHOUT
+   ROWID seeks, N-table chains) tracked under §4's B5b-2 entry.
 2. **B5c-2 — correlated subqueries on the VDBE**, once B5b-2 lands the live-cursor
    machinery.
 3. **C9a → C9d — the concurrency story** (persistent read locks in `src/pager/`,

src/exec/mod.rs

@@ -1989,6 +1989,236 @@ impl Connection {
         // the `WHERE`, and reuse the single-cursor scan compiler. Every join must
         // be a plain `INNER`/`CROSS`/comma join (no `NATURAL`/`USING`/outer).
         if !from.joins.is_empty() {
+            // B5b-2 (live inner cursor): a two-table INNER equi-join whose `ON`
+            // binds the inner table's INTEGER PRIMARY KEY (`… JOIN t ON o.x =
+            // t.<ipk>`) seeks the single matching inner row with a *live* b-tree
+            // cursor (`read_row` → `TableCursor::seek`) instead of materializing
+            // and scanning the whole inner table. Only the outer table is scanned;
+            // for each outer row the inner row is fetched by rowid. Correctness
+            // rides the superset invariant: after the seek the full `ON` is
+            // re-evaluated against the assembled row, so every rowid-coercion
+            // corner (`= 2.5`, text/blob keys, `NULL`) is filtered exactly as the
+            // materialized cross-product would. Any shape outside this narrow
+            // window — or a projection the single-cursor compiler can't take —
+            // breaks out and falls through to the materialized inner-join path.
+            'seek: {
+                if !(from.joins.len() == 1
+                    && from.joins[0].kind == sql::ast::JoinKind::Inner
+                    && !from.joins[0].natural
+                    && from.joins[0].using.is_empty())
+                {
+                    break 'seek;
+                }
+                let j = &from.joins[0];
+                // Both sides must be plain base tables: the inner side needs a live
+                // rowid b-tree, and a CTE/view/subquery/TVF has none.
+                let plain = |tr: &sql::ast::TableRef| -> bool {
+                    tr.subquery.is_none()
+                        && tr.tvf_args.is_none()
+                        && tr.schema.is_none()
+                        && !self.is_bare_tvf(tr)
+                };
+                if !plain(&from.first) || !plain(&j.table) {
+                    break 'seek;
+                }
+                // The inner name must resolve to a real rowid table with an
+                // INTEGER PRIMARY KEY (a same-named CTE/view shadows a base table).
+                if self.is_view(&j.table.name)
+                    || sel
+                        .ctes
+                        .iter()
+                        .any(|c| c.name.eq_ignore_ascii_case(&j.table.name))
+                {
+                    break 'seek;
+                }
+                let inner_meta = match self.table_meta(&j.table.name, j.table.alias.as_deref()) {
+                    Ok(m) => m,
+                    Err(_) => break 'seek,
+                };
+                if inner_meta.without_rowid {
+                    break 'seek;
+                }
+                let Some(ipk) = inner_meta.ipk else {
+                    break 'seek;
+                };
+                let Some(on) = &j.on else { break 'seek };
+                // Outer schema + rows (the inner side is NOT scanned).
+                let (o_cols, o_tables, o_aff, o_coll, o_rows, _o_ids) = scan_one(&from.first)?;
+                // Inner schema — column metadata only, no row scan.
+                let inner_qual = j
+                    .table
+                    .alias
+                    .clone()
+                    .unwrap_or_else(|| j.table.name.clone());
+                let i_cols: Vec<String> =
+                    inner_meta.columns.iter().map(|c| c.name.clone()).collect();
+                let i_coll: Vec<crate::value::Collation> =
+                    inner_meta.columns.iter().map(|c| c.collation).collect();
+                let i_aff: Vec<eval::Affinity> =
+                    inner_meta.columns.iter().map(|c| c.affinity).collect();
+                let i_tables: Vec<String> = inner_meta
+                    .columns
+                    .iter()
+                    .map(|_| inner_qual.clone())
+                    .collect();
+                let ipk_name = inner_meta.columns[ipk].name.clone();
+                // The `ON` must be a single `Eq` (parens stripped) binding the
+                // inner ipk column on one side and a plain outer column on the
+                // other.
+                let mut on_expr = on;
+                while let sql::ast::Expr::Paren(inner) = on_expr {
+                    on_expr = inner;
+                }
+                let (l, r) = match on_expr {
+                    sql::ast::Expr::Binary {
+                        op: sql::ast::BinaryOp::Eq,
+                        left,
+                        right,
+                    } => (left.as_ref(), right.as_ref()),
+                    _ => break 'seek,
+                };
+                // A plain (optionally parenthesized) unqualified-schema column ref.
+                fn col_ref(mut e: &sql::ast::Expr) -> Option<(Option<&str>, &str)> {
+                    while let sql::ast::Expr::Paren(i) = e {
+                        e = i;
+                    }
+                    match e {
+                        sql::ast::Expr::Column {
+                            schema: None,
+                            table,
+                            column,
+                            ..
+                        } => Some((table.as_deref(), column.as_str())),
+                        _ => None,
+                    }
+                }
+                // Is `e` the inner ipk column, named by its declared column name?
+                // (A bare `rowid`/`_rowid_`/`oid` alias defers — the combined
+                // schema exposes the ipk under its declared name, so the `ON`
+                // re-check below could not resolve the alias.) Qualified with the
+                // inner qualifier, or bare and not also an outer column name.
+                let is_inner_ipk = |e: &sql::ast::Expr| -> bool {
+                    let Some((q, name)) = col_ref(e) else {
+                        return false;
+                    };
+                    if !name.eq_ignore_ascii_case(&ipk_name) {
+                        return false;
+                    }
+                    match q {
+                        Some(q) => q.eq_ignore_ascii_case(&inner_qual),
+                        None => !o_cols.iter().any(|c| c.eq_ignore_ascii_case(name)),
+                    }
+                };
+                // Resolve `e` to a single OUTER column index (qualified to the
+                // outer table, or bare-unambiguous across both tables).
+                let outer_qual = from
+                    .first
+                    .alias
+                    .clone()
+                    .unwrap_or_else(|| from.first.name.clone());
+                let outer_col_index = |e: &sql::ast::Expr| -> Option<usize> {
+                    let (q, name) = col_ref(e)?;
+                    if let Some(q) = q {
+                        if !q.eq_ignore_ascii_case(&outer_qual) {
+                            return None;
+                        }
+                    }
+                    let matches: Vec<usize> = o_cols
+                        .iter()
+                        .enumerate()
+                        .filter(|(_, c)| c.eq_ignore_ascii_case(name))
+                        .map(|(i, _)| i)
+                        .collect();
+                    if matches.len() != 1 {
+                        return None;
+                    }
+                    // A bare name owned by both tables is ambiguous → defer.
+                    if q.is_none() && i_cols.iter().any(|c| c.eq_ignore_ascii_case(name)) {
+                        return None;
+                    }
+                    Some(matches[0])
+                };
+                let oc = if is_inner_ipk(l) {
+                    match outer_col_index(r) {
+                        Some(oc) => oc,
+                        None => break 'seek,
+                    }
+                } else if is_inner_ipk(r) {
+                    match outer_col_index(l) {
+                        Some(oc) => oc,
+                        None => break 'seek,
+                    }
+                } else {
+                    break 'seek;
+                };
+                // Combined schema (outer cols ++ inner cols), leftmost outermost.
+                let mut c_cols = o_cols.clone();
+                c_cols.extend(i_cols.iter().cloned());
+                let mut c_tables = o_tables.clone();
+                c_tables.extend(i_tables.iter().cloned());
+                let mut c_aff = o_aff.clone();
+                c_aff.extend(i_aff.iter().copied());
+                let mut c_coll = o_coll.clone();
+                c_coll.extend(i_coll.iter().copied());
+                let join_cols: Vec<ColumnInfo> = (0..c_cols.len())
+                    .map(|i| ColumnInfo {
+                        name: c_cols[i].clone(),
+                        table: c_tables[i].clone(),
+                        affinity: c_aff[i],
+                        collation: c_coll[i],
+                        schema: None,
+                        hidden: false,
+                    })
+                    .collect();
+                // An ambiguous bare column in the projection/WHERE defers so the
+                // tree-walker can reject it with "ambiguous column name".
+                if validate_unambiguous_columns(sel, &join_cols, &|t| t.into()).is_err() {
+                    break 'seek;
+                }
+                // Compile the projection/WHERE/ORDER BY/… over the assembled join
+                // rows. The `ON` is applied during assembly, so the original `sel`
+                // (not an ON-merged WHERE) is compiled — exactly like the
+                // outer-join assembly path. A shape the single-cursor compiler
+                // can't take falls through to the materialized path.
+                let prog = match vdbe::compile_table_select(
+                    sel, &c_cols, &c_tables, &c_aff, &c_coll, false,
+                ) {
+                    Ok(p) => p,
+                    Err(Error::Unsupported(_)) => break 'seek,
+                    Err(e) => return Err(e),
+                };
+                // Assemble the joined rows by seeking the inner table live, per
+                // outer row: coerce the outer key to a rowid (the same `to_i64`
+                // the tree-walker's rowid seek uses), seek it, then re-check the
+                // full `ON` so the result is a subset of the true cross-product.
+                let on_params = eval::Params::default();
+                let mut rows: Vec<Vec<Value>> = Vec::new();
+                for orow in &o_rows {
+                    let kv = &orow[oc];
+                    if matches!(kv, Value::Null) {
+                        continue;
+                    }
+                    let rid = eval::to_i64(kv);
+                    let Some(ivals) = self.read_row(&inner_meta, rid)? else {
+                        continue;
+                    };
+                    let mut row = orow.clone();
+                    row.extend(ivals.iter().cloned());
+                    let ir = InputRow {
+                        values: row.clone(),
+                        rowid: None,
+                    };
+                    let ctx = ir.ctx(&join_cols, &on_params).with_subqueries(self);
+                    if eval::truth(&eval::eval(on_expr, &ctx)?) == Some(true) {
+                        rows.push(row);
+                    }
+                }
+                let result = vdbe::run_rows(&prog, &rows)?;
+                return Ok(QueryResult {
+                    columns: prog.columns,
+                    rows: result,
+                });
+            }
             // A single two-table LEFT/RIGHT/FULL JOIN routes to the null-padding
             // nested loop below; otherwise only plain INNER joins are handled here
             // (NATURAL/USING fall back to the tree-walker).

tests/vdbe_live_cursor.rs

@@ -0,0 +1,99 @@
+//! B5b-2 (live inner cursor): a two-table INNER equi-join whose `ON` binds the
+//! inner table's INTEGER PRIMARY KEY (`… JOIN t ON o.x = t.<ipk>`) seeks the
+//! single matching inner row with a *live* b-tree cursor (`TableCursor::seek`)
+//! instead of materializing and scanning the whole inner table. Only the outer
+//! table is scanned; each outer row fetches its one inner row by rowid.
+//!
+//! `query_vdbe` errors on any fallback to the tree-walker, so a passing
+//! `query_vdbe` proves the query routed onto this VDBE path. Correctness rides
+//! the superset invariant: after the seek the full `ON` is re-evaluated against
+//! the assembled row, so every rowid-coercion corner (`= 2.5`, text/blob keys,
+//! `NULL`) is filtered exactly as the materialized cross-product would. Each
+//! query is checked three ways: it routes on the VDBE, it equals the tree-walker
+//! (`set_use_vdbe(false)`), and it equals the sqlite3 CLI.
+
+#![cfg(feature = "std")]
+
+use graphitesql::Connection;
+use std::process::Command;
+
+fn sqlite3_available() -> bool {
+    Command::new("sqlite3").arg("--version").output().is_ok()
+}
+
+const SCHEMA: &str = "CREATE TABLE o(x, y TEXT);\
+     INSERT INTO o VALUES(1,'a'),(2,'b'),(3,'c'),(NULL,'n'),(2,'b2'),(2.0,'real'),\
+       ('2','txt'),('2abc','junk'),(x'32','blob'),(2.5,'half'),(99,'miss'),(-1,'neg');\
+     CREATE TABLE t(id INTEGER PRIMARY KEY, name TEXT);\
+     INSERT INTO t VALUES(1,'one'),(2,'two'),(5,'five');";
+
+/// Queries that must route onto the live-inner-cursor path (`o JOIN t ON … =
+/// t.id`, with the inner table's INTEGER PRIMARY KEY on one side).
+const VDBE_QUERIES: &[&str] = &[
+    "SELECT o.x, t.name FROM o JOIN t ON o.x = t.id ORDER BY o.y",
+    "SELECT o.x, t.name FROM o JOIN t ON t.id = o.x ORDER BY o.y", // ipk on the left
+    "SELECT o.y, t.name FROM o JOIN t ON o.x = t.id WHERE t.name <> 'one' ORDER BY o.y",
+    "SELECT * FROM o JOIN t ON o.x = t.id ORDER BY o.y",
+    "SELECT o.x * 100 + t.id AS s FROM o JOIN t ON o.x = t.id ORDER BY s",
+    "SELECT DISTINCT t.name FROM o JOIN t ON o.x = t.id ORDER BY t.name",
+    "SELECT o.x, t.name FROM o JOIN t ON o.x = t.id ORDER BY o.y LIMIT 2 OFFSET 1",
+    "SELECT count(*), sum(t.id) FROM o JOIN t ON o.x = t.id",
+    "SELECT t.name, count(*) FROM o JOIN t ON o.x = t.id GROUP BY t.name ORDER BY t.name",
+    "SELECT o.y FROM o JOIN t ON (o.x) = (t.id) ORDER BY o.y", // parenthesized ON
+];
+
+fn setup() -> Connection {
+    let mut c = Connection::open_memory().unwrap();
+    for stmt in SCHEMA.split(';') {
+        if !stmt.trim().is_empty() {
+            c.execute(stmt).unwrap();
+        }
+    }
+    c
+}
+
+#[test]
+fn live_inner_cursor_routes_and_matches_tree_walker() {
+    let c = setup();
+    for q in VDBE_QUERIES {
+        // Routes onto the VDBE (errors on any fallback).
+        let vdbe = c
+            .query_vdbe(q)
+            .unwrap_or_else(|e| panic!("expected VDBE routing for `{q}`: {e:?}"));
+        // Tree-walker (source of truth).
+        c.set_use_vdbe(false);
+        let tw = c.query(q).unwrap();
+        c.set_use_vdbe(true);
+        assert_eq!(vdbe.rows, tw.rows, "VDBE vs tree-walker rows for `{q}`");
+        assert_eq!(vdbe.columns, tw.columns, "columns for `{q}`");
+    }
+}
+
+fn cli_rows(bin: &str, sql: &str) -> String {
+    let full = format!("{SCHEMA} {sql}");
+    let out = Command::new(bin)
+        .arg(":memory:")
+        .arg(&full)
+        .output()
+        .unwrap();
+    String::from_utf8_lossy(&out.stdout).into_owned()
+}
+
+#[test]
+fn live_inner_cursor_matches_sqlite3() {
+    if !sqlite3_available() {
+        eprintln!("sqlite3 CLI not found; skipping");
+        return;
+    }
+    // Compare the graphite CLI binary (VDBE enabled by default, so the seek path
+    // runs) against sqlite3 end-to-end — float/blob rendering included, so no
+    // hand-rolled value formatting can mask a divergence.
+    let g = env!("CARGO_BIN_EXE_graphitesql");
+    for q in VDBE_QUERIES {
+        assert_eq!(
+            cli_rows("sqlite3", q),
+            cli_rows(g, q),
+            "sqlite3 vs graphite `{q}`"
+        );
+    }
+}