diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 8b72364a3..c1cb9544a 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -202,7 +202,7 @@ The `dirsql` Rust crate handles all heavy lifting: ### `db` -- Ephemeral SQLite -Wraps `rusqlite` with an anonymous disk-backed temp database (#402) — ephemeral like `:memory:`, but index pages spill to disk so resident memory does not scale with the corpus. Handles DDL execution (run verbatim -- no injected columns, epic #358), row insertion with per-file ownership recorded in the internal `_dirsql_internal_rows` table, querying, and row deletion by file path. The internal bookkeeping tables (`_dirsql_internal_rows`, `_dirsql_files`, `_dirsql_meta`) are a private surface: a SQLite authorizer installed on the `query()` path denies any read (or schema `PRAGMA`) targeting the reserved `_dirsql_*` namespace, so they are unreachable through the public query surface (issue #378) while the engine still writes them in the same transaction as the user rows. +Wraps `rusqlite` with an anonymous disk-backed temp database (#402) — ephemeral like `:memory:`, but index pages spill to disk so resident memory does not scale with the corpus. Handles DDL execution (run verbatim -- no injected columns, epic #358), row insertion with per-file ownership recorded in the internal `_dirsql_internal_rows` table (a row at a time) or `_dirsql_internal_ranges` (a per-table hook's batch, as one rowid range), querying, and row deletion by file path. The internal bookkeeping tables (`_dirsql_internal_rows`, `_dirsql_internal_ranges`, `_dirsql_files`, `_dirsql_meta`) are a private surface: a SQLite authorizer installed on the `query()` path denies any read (or schema `PRAGMA`) targeting the reserved `_dirsql_*` namespace, so they are unreachable through the public query surface (issue #378) while the engine still writes them in the same transaction as the user rows. **A table's name is declared, never derived.** Both `[[table]]` and the SDK `Table` take a required `name`; nothing tokenizes the `CREATE TABLE` head of diff --git a/PARITY.md b/PARITY.md index dd53c90e8..137d59235 100644 --- a/PARITY.md +++ b/PARITY.md @@ -216,7 +216,8 @@ at once with no binding changes and no public-surface change. **Verbatim table columns (#361, epic #358) — parity by construction, no drift.** User tables carry exactly the columns declared in their DDL (row ownership is -tracked in the internal `_dirsql_internal_rows` table, not injected columns). +tracked in the internal `_dirsql_internal_rows` and `_dirsql_internal_ranges` +tables, not injected columns). This lives entirely in the shared Rust core (`create_table` runs DDL verbatim; `query` returns vanilla rows), so `PRAGMA table_info` and `SELECT *` report the same user-only columns across all three SDKs. @@ -232,7 +233,7 @@ together. **Internal tables unreachable through `query()` (#378, epic #358) — parity by construction, no drift.** The internal bookkeeping tables (`_dirsql_internal_rows`, -`_dirsql_files`, `_dirsql_meta`) are denied on the `query()` path by a SQLite +`_dirsql_internal_ranges`, `_dirsql_files`, `_dirsql_meta`) are denied on the `query()` path by a SQLite authorizer in the shared Rust core (`db::query`), so a read of any `_dirsql_*` table fails identically across all three SDKs (and the CLI's `POST /query`) with a "not authorized" error — no per-binding surface. diff --git a/packages/python/e2e-attestations/claude-1243-ingest-perf.json b/packages/python/e2e-attestations/claude-1243-ingest-perf.json new file mode 100644 index 000000000..0d00b26d5 --- /dev/null +++ b/packages/python/e2e-attestations/claude-1243-ingest-perf.json @@ -0,0 +1,7 @@ +{ + "command": "just test-e2e", + "ran_at": 1790978388, + "exit_code": 0, + "commit": "cb8a9e14d99d059576c313c41cac87b9d865f020", + "branch": "claude/1243-ingest-perf" +} diff --git a/packages/rust/changelog.d/2026-10-02-batch-rows-owned-by-range.md b/packages/rust/changelog.d/2026-10-02-batch-rows-owned-by-range.md new file mode 100644 index 000000000..b964a5bf9 --- /dev/null +++ b/packages/rust/changelog.d/2026-10-02-batch-rows-owned-by-range.md @@ -0,0 +1 @@ +**Changed** A per-table `on-file` hook's rows are owned as one rowid range in the new internal `_dirsql_internal_ranges` table instead of one `_dirsql_internal_rows` entry per row, and are laid out in DDL column order once rather than rebuilt per row. Ingesting the transcripts fixture's 273k rows drops from 0.99 s to 0.58 s; the full run from 2.65 s to 2.17 s. Refresh and watch keep their delete-before-reingest semantics; rows, order and query results are unchanged. diff --git a/packages/rust/migrations.d/2026-10-02-batch-rows-owned-by-range.md b/packages/rust/migrations.d/2026-10-02-batch-rows-owned-by-range.md new file mode 100644 index 000000000..52b038d3e --- /dev/null +++ b/packages/rust/migrations.d/2026-10-02-batch-rows-owned-by-range.md @@ -0,0 +1,45 @@ +### Core: a per-table hook's rows are owned as one rowid range + +#### Summary + +A table fed by an `on-file` command no longer records one +`_dirsql_internal_rows` entry per ingested row. The batch is owned as a single +`(table, first_rowid, last_rowid)` range in a new internal table, +`_dirsql_internal_ranges`, created alongside `_dirsql_internal_rows` on every +connection, including an existing persisted cache. No signature, config key +or CLI flag changes. + +#### Required changes + +_None._ + +#### Deprecations removed + +_None._ + +#### Behavior changes without code changes + +- `_dirsql_internal_ranges` joins the reserved `_dirsql_*` namespace: a + `query()` that reads it is denied like the other internal tables. +- A persisted cache gains the table on next open; a cold rebuild clears it + with the rest of the bookkeeping. +- Rows, their order, `rowid`s and query results are unchanged. + +#### Verification + +```bash +cd "$(mktemp -d)" +printf 'a\n' > a.txt +printf 'b\n' > b.txt +cat > .dirsql.toml <<'TOML' +[[table]] +name = "items" +ddl = "CREATE TABLE items (name TEXT)" +glob = "*.txt" +on-file = "sh -c 'printf \"[\"; sep=\"\"; for f in \"$@\"; do printf \"%s{\\\"name\\\":\\\"%s\\\"}\" \"$sep\" \"$(basename \"$f\")\"; sep=\",\"; done; printf \"]\\n\"' sh" +TOML +dirsql query "SELECT name FROM items ORDER BY name" -c .dirsql.toml +# expected: [{"name":"a.txt"},{"name":"b.txt"}] +dirsql query "SELECT * FROM _dirsql_internal_ranges" -c .dirsql.toml; echo "exit: $?" +# expected: a "not authorized" error and a non-zero exit +``` diff --git a/packages/rust/src/db.rs b/packages/rust/src/db.rs index 2d9c2f2cb..ba75e65b7 100644 --- a/packages/rust/src/db.rs +++ b/packages/rust/src/db.rs @@ -5,6 +5,7 @@ use std::sync::Arc; use thiserror::Error; use crate::functions::{self, CallReporter, ResolvedFunction, Worker}; +use crate::infer::JsonRow; use crate::parsed_vtab; use crate::path_table::{self, PathTable, Resolution}; use crate::scanner; @@ -219,20 +220,33 @@ pub type Result = std::result::Result; const MAPPING_INSERT: &str = "INSERT INTO _dirsql_internal_rows \ (table_name, file_path, row_index, rowid_ref) VALUES (?1, ?2, ?3, ?4)"; +const RANGE_INSERT: &str = "INSERT INTO _dirsql_internal_ranges \ + (table_name, file_path, first_rowid, last_rowid) VALUES (?1, ?2, ?3, ?4)"; + +const RANGES_BY_FILE: &str = "SELECT first_rowid, last_rowid FROM _dirsql_internal_ranges \ + WHERE table_name = ?1 AND file_path = ?2 ORDER BY first_rowid"; + /// Name of the internal row-bookkeeping table. /// -/// The sole record of row ownership: it maps every inserted user row back to -/// the file that produced it — `(table_name, file_path, row_index, rowid_ref)` -/// — keyed on the row's rowid. Written in the same SQLite transaction as each -/// row insert/delete, so it can never diverge from the rows it describes. +/// Together with `_dirsql_internal_ranges` it is the sole record of row +/// ownership. A row inserted one at a time maps back to the file that +/// produced it here — `(table_name, file_path, row_index, rowid_ref)` — keyed +/// on the row's rowid; a batch inserted at once is owned as one +/// `(table_name, file_path, first_rowid, last_rowid)` range. Both are written +/// in the same SQLite transaction as the row insert/delete, so neither can +/// diverge from the rows it describes. pub const INTERNAL_ROWS_TABLE: &str = "_dirsql_internal_rows"; -/// Create the internal `_dirsql_internal_rows` bookkeeping table and its -/// by-file index if they don't already exist. Idempotent. +/// Name of the internal batch-ownership table. +pub const INTERNAL_RANGES_TABLE: &str = "_dirsql_internal_ranges"; + +/// Create the internal row-ownership tables (`_dirsql_internal_rows` and +/// `_dirsql_internal_ranges`) and their by-file indexes if they don't already +/// exist. Idempotent. /// -/// A **real** table (not virtual): it lives in the persisted cache and is -/// written inside the same transaction as the row inserts/deletes it -/// describes, which is what gives the mapping crash-atomicity with the user +/// **Real** tables (not virtual): they live in the persisted cache and are +/// written inside the same transaction as the row inserts/deletes they +/// describe, which is what gives ownership crash-atomicity with the user /// rows. pub fn ensure_internal_rows_table(conn: &Connection) -> rusqlite::Result<()> { conn.execute_batch( @@ -243,10 +257,49 @@ pub fn ensure_internal_rows_table(conn: &Connection) -> rusqlite::Result<()> { rowid_ref INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS _dirsql_internal_rows_by_file - ON _dirsql_internal_rows(table_name, file_path);", + ON _dirsql_internal_rows(table_name, file_path); + CREATE TABLE IF NOT EXISTS _dirsql_internal_ranges ( + table_name TEXT NOT NULL, + file_path TEXT NOT NULL, + first_rowid INTEGER NOT NULL, + last_rowid INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS _dirsql_internal_ranges_by_file + ON _dirsql_internal_ranges(table_name, file_path);", ) } +/// A batch of rows laid out in a table's DDL column order, so each row binds +/// positionally with no per-cell lookup. +#[derive(Debug, Clone, PartialEq)] +pub struct ShapedRows { + pub columns: Vec, + pub cells: Vec>, +} + +impl ShapedRows { + /// The rows as column-name maps, for callers that compare rows by name. + pub fn to_maps(&self) -> Vec> { + self.cells + .iter() + .map(|row| { + self.columns + .iter() + .cloned() + .zip(row.iter().cloned()) + .collect() + }) + .collect() + } +} + +fn json_into_value(value: serde_json::Value) -> Value { + match value { + serde_json::Value::String(s) => Value::Text(s), + other => crate::json_to_value(&other), + } +} + /// One query's rows plus the column order the SELECT list asked for. #[derive(Debug, Clone, PartialEq)] pub struct QueryResult { @@ -598,18 +651,76 @@ impl Db { Ok(()) } + /// Lay `rows` out in `table`'s DDL column order. + /// + /// In relaxed mode (strict=false): a key the table lacks is dropped and a + /// column the row lacks is NULL. In strict mode (strict=true): every key + /// is validated as a safe SQL identifier, then any extra or missing key + /// produces a SchemaMismatch error. The identifier check runs *first* so a + /// malformed key reports as [`DbError::InvalidIdentifier`] rather than as + /// a less-actionable "extra columns" mismatch. Every row is shaped before + /// any is returned, so a hook that got one row wrong fails as a unit. + pub fn shape_rows(&self, table: &str, rows: Vec, strict: bool) -> Result { + let columns = self.get_table_columns(table)?; + let mut cells = Vec::with_capacity(rows.len()); + for row in rows { + if strict { + for (key, _) in &row.0 { + validate_identifier(key)?; + } + let extra: Vec<&str> = row + .0 + .iter() + .map(|(key, _)| key.as_str()) + .filter(|key| !columns.iter().any(|c| c == key)) + .collect(); + if !extra.is_empty() { + return Err(DbError::SchemaMismatch(format!( + "extra columns not in table {}: {}", + table, + extra.join(", ") + ))); + } + let missing: Vec<&str> = columns + .iter() + .map(String::as_str) + .filter(|c| !row.0.iter().any(|(key, _)| key == c)) + .collect(); + if !missing.is_empty() { + return Err(DbError::SchemaMismatch(format!( + "missing columns for table {}: {}", + table, + missing.join(", ") + ))); + } + } + let in_order = row.0.len() == columns.len() + && row.0.iter().zip(&columns).all(|((key, _), c)| key == c); + if in_order { + cells.push(row.0.into_iter().map(|(_, v)| json_into_value(v)).collect()); + continue; + } + let mut shaped = vec![Value::Null; columns.len()]; + for (key, value) in row.0 { + if let Some(i) = columns.iter().position(|c| *c == key) { + shaped[i] = json_into_value(value); + } + } + cells.push(shaped); + } + Ok(ShapedRows { columns, cells }) + } + /// Insert every row of `rows` under `file_path`, in order, through one - /// prepared statement pair. Rows are bound by the table's columns, so a - /// key the table lacks is dropped and a column the row lacks is NULL; - /// callers normalize first when they need strictness. - pub fn insert_rows( - &self, - table: &str, - rows: &[HashMap], - file_path: &str, - ) -> Result<()> { + /// prepared statement, and own the batch as one rowid range rather than + /// a mapping row each. + /// + /// The range spans the smallest to the largest rowid the batch received, + /// so a table whose DDL aliases rowid (`INTEGER PRIMARY KEY`) must not + /// interleave batches of different owners. + pub fn insert_rows(&self, table: &str, rows: &ShapedRows, file_path: &str) -> Result<()> { validate_identifier(table)?; - let columns = self.get_table_columns(table)?; + let columns = &rows.columns; let sql = if columns.is_empty() { format!("INSERT INTO \"{table}\" DEFAULT VALUES") } else { @@ -628,18 +739,22 @@ impl Db { }; { let mut insert = self.conn.prepare_cached(&sql)?; - let mut map = self.conn.prepare_cached(MAPPING_INSERT)?; - for (row_index, row) in rows.iter().enumerate() { - for (i, column) in columns.iter().enumerate() { - insert.raw_bind_parameter(i + 1, row.get(column).unwrap_or(&Value::Null))?; + let mut range: Option<(i64, i64)> = None; + for row in &rows.cells { + for (i, value) in row.iter().enumerate() { + insert.raw_bind_parameter(i + 1, value)?; } insert.raw_execute()?; - map.execute(rusqlite::params![ - table, - file_path, - i64::try_from(row_index).expect("row index fits in i64"), - self.conn.last_insert_rowid() - ])?; + let rowid = self.conn.last_insert_rowid(); + range = Some(match range { + None => (rowid, rowid), + Some((first, last)) => (first.min(rowid), last.max(rowid)), + }); + } + if let Some((first, last)) = range { + self.conn + .prepare_cached(RANGE_INSERT)? + .execute(rusqlite::params![table, file_path, first, last])?; } } if let Some(tx) = tx { @@ -721,10 +836,11 @@ impl Db { Ok(()) } - /// Read back the rows a given file produced for `table`, ordered by row - /// index. Ownership and ordering come from the `_dirsql_internal_rows` - /// mapping (joined on `rowid`); user columns are qualified with the table - /// alias so a user column named like a mapping column stays unambiguous. + /// Read back the rows a given file produced for `table`: the rows it + /// inserted one at a time, ordered by row index, then the batches it + /// inserted at once, in rowid order. User columns are qualified with the + /// table alias so a user column named like a mapping column stays + /// unambiguous. /// /// A row read back compares equal to the normalized row that was inserted /// only when the on-file callback's value types match the declared column @@ -747,33 +863,59 @@ impl Db { // let SQLite report "no such table". col_list = "1".to_string(); } - let sql = format!( - "SELECT {col_list} FROM \"{table}\" AS t \ - JOIN _dirsql_internal_rows AS m ON m.rowid_ref = t.rowid \ - WHERE m.table_name = ?1 AND m.file_path = ?2 ORDER BY m.row_index" - ); - let mut stmt = self.conn.prepare(&sql)?; - let rows = stmt.query_map(rusqlite::params![table, file_path], |row| { + let read_row = |row: &rusqlite::Row<'_>| { let mut map = HashMap::new(); for (i, name) in user_columns.iter().enumerate() { let v: rusqlite::types::Value = row.get(i)?; map.insert(name.clone(), Value::from(v)); } Ok(map) - })?; - let mut out = Vec::new(); - for row in rows { - out.push(row?); + }; + let mapped = format!( + "SELECT {col_list} FROM \"{table}\" AS t \ + JOIN _dirsql_internal_rows AS m ON m.rowid_ref = t.rowid \ + WHERE m.table_name = ?1 AND m.file_path = ?2 ORDER BY m.row_index" + ); + let mut out = self + .conn + .prepare(&mapped)? + .query_map(rusqlite::params![table, file_path], &read_row)? + .collect::>>()?; + let ranged = format!( + "SELECT {col_list} FROM \"{table}\" AS t \ + WHERE t.rowid BETWEEN ?1 AND ?2 ORDER BY t.rowid" + ); + let mut stmt = self.conn.prepare(&ranged)?; + for (first, last) in Self::ranges_by_file(&self.conn, table, file_path)? { + out.extend( + stmt.query_map([first, last], &read_row)? + .collect::>>()?, + ); } Ok(out) } + fn ranges_by_file( + conn: &Connection, + table: &str, + file_path: &str, + ) -> rusqlite::Result> { + conn.prepare_cached(RANGES_BY_FILE)? + .query_map(rusqlite::params![table, file_path], |r| { + Ok((r.get(0)?, r.get(1)?)) + })? + .collect() + } + /// Every distinct `(table_name, file_path)` whose file sits beneath the /// directory `dir`, ordered by table then path. pub fn files_under(&self, dir: &str) -> Result> { let prefix = format!("{dir}/"); let mut stmt = self.conn.prepare( - "SELECT DISTINCT table_name, file_path FROM _dirsql_internal_rows \ + "SELECT DISTINCT table_name, file_path FROM ( \ + SELECT table_name, file_path FROM _dirsql_internal_rows \ + UNION ALL \ + SELECT table_name, file_path FROM _dirsql_internal_ranges) \ WHERE substr(file_path, 1, length(?1)) = ?1 \ ORDER BY table_name, file_path", )?; @@ -785,26 +927,34 @@ impl Db { /// Called from delete_rows_by_file() either inside a caller-supplied /// transaction or inside a transaction opened by delete_rows_by_file(). fn delete_rows_by_file_stmts(conn: &Connection, table: &str, file_path: &str) -> Result { - let sql = format!( - "DELETE FROM {} WHERE rowid IN \ + let mapped = format!( + "DELETE FROM \"{table}\" WHERE rowid IN \ (SELECT rowid_ref FROM _dirsql_internal_rows \ - WHERE table_name = ?1 AND file_path = ?2)", - table + WHERE table_name = ?1 AND file_path = ?2)" ); - let count = conn.execute(&sql, rusqlite::params![table, file_path])?; + let mut count = conn.execute(&mapped, rusqlite::params![table, file_path])?; conn.execute( "DELETE FROM _dirsql_internal_rows WHERE table_name = ?1 AND file_path = ?2", rusqlite::params![table, file_path], )?; + let ranged = format!("DELETE FROM \"{table}\" WHERE rowid BETWEEN ?1 AND ?2"); + for (first, last) in Self::ranges_by_file(conn, table, file_path)? { + count += conn.execute(&ranged, [first, last])?; + } + conn.execute( + "DELETE FROM _dirsql_internal_ranges WHERE table_name = ?1 AND file_path = ?2", + rusqlite::params![table, file_path], + )?; Ok(count) } /// Delete all rows that were produced by a given file path. Row ownership - /// is resolved through the `_dirsql_internal_rows` mapping. + /// is resolved through the `_dirsql_internal_rows` mapping and the + /// `_dirsql_internal_ranges` batches. /// - /// The user-row deletes and the matching mapping deletes commit in ONE + /// The user-row deletes and the matching ownership deletes commit in ONE /// transaction (either via a transaction opened here in autocommit mode, - /// or via the caller's already-open transaction), so the mapping never + /// or via the caller's already-open transaction), so the ownership never /// outlives the rows it describes. pub fn delete_rows_by_file(&self, table: &str, file_path: &str) -> Result { validate_identifier(table)?; @@ -2405,19 +2555,148 @@ mod tests { assert_eq!(rows[0].get("id").unwrap(), &Value::Text("a".into())); } + fn range_rows(db: &Db, table: &str) -> Vec<(String, i64, i64)> { + let mut stmt = db + .conn + .prepare( + "SELECT file_path, first_rowid, last_rowid FROM _dirsql_internal_ranges \ + WHERE table_name = ?1 ORDER BY first_rowid", + ) + .unwrap(); + stmt.query_map([table], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?))) + .unwrap() + .map(|r| r.unwrap()) + .collect() + } + + fn json_rows(payload: serde_json::Value) -> Vec { + let serde_json::Value::Array(rows) = payload else { + panic!("payload must be an array"); + }; + rows.into_iter() + .map(|row| { + let serde_json::Value::Object(map) = row else { + panic!("row must be an object"); + }; + JsonRow(map.into_iter().collect()) + }) + .collect() + } + + fn shaped(db: &Db, table: &str, payload: serde_json::Value) -> ShapedRows { + db.shape_rows(table, json_rows(payload), false).unwrap() + } + #[test] - fn insert_rows_maps_every_row_to_its_file_in_order() { + fn shape_rows_lays_cells_out_in_ddl_order_whatever_the_key_order() { let db = Db::new().unwrap(); db.create_table("t", "CREATE TABLE t (id TEXT, n INTEGER)") .unwrap(); - let rows = vec![ - HashMap::from([("id".into(), Value::Text("a".into()))]), - HashMap::from([ - ("id".into(), Value::Text("b".into())), - ("n".into(), Value::Integer(2)), - ("extra".into(), Value::Integer(9)), - ]), - ]; + let out = shaped( + &db, + "t", + serde_json::json!([{"id":"a","n":1},{"n":2,"id":"b"}]), + ); + assert_eq!(out.columns, vec!["id", "n"]); + assert_eq!( + out.cells, + vec![ + vec![Value::Text("a".into()), Value::Integer(1)], + vec![Value::Text("b".into()), Value::Integer(2)], + ] + ); + } + + #[test] + fn shape_rows_relaxed_drops_extra_keys_and_nulls_missing_columns() { + let db = Db::new().unwrap(); + db.create_table("t", "CREATE TABLE t (name TEXT, n INTEGER)") + .unwrap(); + let out = shaped(&db, "t", serde_json::json!([{"name":"a"},{"n":1,"x":null}])).to_maps(); + assert_eq!(out[0]["name"], Value::Text("a".into())); + assert_eq!(out[0]["n"], Value::Null); + assert_eq!(out[1]["name"], Value::Null); + assert_eq!(out[1]["n"], Value::Integer(1)); + assert!(!out[1].contains_key("x")); + } + + #[test] + fn shape_rows_strict_accepts_every_column_in_any_order() { + let db = Db::new().unwrap(); + db.create_table("t", "CREATE TABLE t (id TEXT, n INTEGER)") + .unwrap(); + let out = db + .shape_rows("t", json_rows(serde_json::json!([{"n":1,"id":"a"}])), true) + .unwrap(); + assert_eq!( + out.cells, + vec![vec![Value::Text("a".into()), Value::Integer(1)]] + ); + } + + #[test] + fn shape_rows_strict_names_the_extra_columns() { + let db = Db::new().unwrap(); + db.create_table("t", "CREATE TABLE t (name TEXT)").unwrap(); + let err = db + .shape_rows( + "t", + json_rows(serde_json::json!([{"name":"ok"},{"name":"x","color":"red"}])), + true, + ) + .unwrap_err(); + assert!( + matches!(&err, DbError::SchemaMismatch(m) if m == "extra columns not in table t: color"), + "got: {err}" + ); + } + + #[test] + fn shape_rows_strict_names_the_missing_columns() { + let db = Db::new().unwrap(); + db.create_table("t", "CREATE TABLE t (name TEXT, n INTEGER)") + .unwrap(); + let err = db + .shape_rows("t", json_rows(serde_json::json!([{"name":"ok"}])), true) + .unwrap_err(); + assert!( + matches!(&err, DbError::SchemaMismatch(m) if m == "missing columns for table t: n"), + "got: {err}" + ); + } + + #[test] + fn shape_rows_strict_rejects_an_unsafe_key_before_judging_the_mismatch() { + let db = Db::new().unwrap(); + db.create_table("t", "CREATE TABLE t (name TEXT)").unwrap(); + let err = db + .shape_rows( + "t", + json_rows(serde_json::json!([{"name); DROP TABLE t; --":"x"}])), + true, + ) + .unwrap_err(); + assert!(matches!(err, DbError::InvalidIdentifier(_)), "got: {err}"); + let relaxed = db + .shape_rows( + "t", + json_rows(serde_json::json!([{"name); DROP TABLE t; --":"x"}])), + false, + ) + .unwrap(); + assert_eq!(relaxed.cells, vec![vec![Value::Null]]); + } + + #[test] + fn insert_rows_owns_the_batch_as_one_rowid_range() { + let db = Db::new().unwrap(); + db.create_table("t", "CREATE TABLE t (id TEXT, n INTEGER)") + .unwrap(); + let rows = shaped( + &db, + "t", + serde_json::json!([{"id":"a"},{"id":"b","n":2,"extra":9}]), + ); db.insert_rows("t", &rows, "x.json").unwrap(); let read = db.get_rows_by_file("t", "x.json").unwrap(); @@ -2426,29 +2705,191 @@ mod tests { assert_eq!(read[0]["n"], Value::Null); assert_eq!(read[1]["n"], Value::Integer(2)); assert!(!read[1].contains_key("extra")); - let mapping = mapping_rows(&db, "t"); - assert_eq!(mapping[0].0, "x.json"); - assert_eq!((mapping[0].1, mapping[1].1), (0, 1)); + assert!(mapping_rows(&db, "t").is_empty()); + assert_eq!(range_rows(&db, "t"), vec![("x.json".to_string(), 1, 2)]); + } + + #[test] + fn insert_rows_records_no_range_for_an_empty_batch() { + let db = Db::new().unwrap(); + db.create_table("t", "CREATE TABLE t (id TEXT)").unwrap(); + db.insert_rows("t", &shaped(&db, "t", serde_json::json!([])), "x.json") + .unwrap(); + assert!(range_rows(&db, "t").is_empty()); + assert!(db.get_rows_by_file("t", "x.json").unwrap().is_empty()); + } + + #[test] + fn insert_rows_range_spans_a_user_declared_rowid_alias() { + let db = Db::new().unwrap(); + db.create_table("t", "CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)") + .unwrap(); + let rows = shaped( + &db, + "t", + serde_json::json!([{"id":7,"name":"g"},{"id":3,"name":"c"}]), + ); + db.insert_rows("t", &rows, "x.json").unwrap(); + + assert_eq!(range_rows(&db, "t"), vec![("x.json".to_string(), 3, 7)]); + let names: Vec = db + .get_rows_by_file("t", "x.json") + .unwrap() + .into_iter() + .map(|r| r["name"].clone()) + .collect(); + assert_eq!( + names, + vec![Value::Text("c".into()), Value::Text("g".into())] + ); + assert_eq!(db.delete_rows_by_file("t", "x.json").unwrap(), 2); + } + + #[test] + fn insert_rows_binds_an_empty_column_list_by_default_values() { + let db = Db::new().unwrap(); + db.create_table("t", "CREATE TABLE t (id TEXT)").unwrap(); + let rows = ShapedRows { + columns: vec![], + cells: vec![vec![], vec![]], + }; + db.insert_rows("t", &rows, "x.json").unwrap(); + assert_eq!(db.get_rows_by_file("t", "x.json").unwrap().len(), 2); + } + + #[test] + fn get_rows_by_file_reads_mapped_rows_before_ranged_rows() { + let db = Db::new().unwrap(); + db.create_table("t", "CREATE TABLE t (id TEXT)").unwrap(); + db.insert_rows( + "t", + &shaped(&db, "t", serde_json::json!([{"id":"r1"},{"id":"r2"}])), + "x.json", + ) + .unwrap(); + db.insert_row( + "t", + &HashMap::from([("id".into(), Value::Text("m".into()))]), + "x.json", + 0, + ) + .unwrap(); + db.insert_rows( + "t", + &shaped(&db, "t", serde_json::json!([{"id":"other"}])), + "y.json", + ) + .unwrap(); + + let ids: Vec = db + .get_rows_by_file("t", "x.json") + .unwrap() + .into_iter() + .map(|r| r["id"].clone()) + .collect(); + assert_eq!( + ids, + vec![ + Value::Text("m".into()), + Value::Text("r1".into()), + Value::Text("r2".into()) + ] + ); + } + + #[test] + fn delete_rows_by_file_removes_a_range_with_its_rows_and_leaves_other_owners() { + let db = Db::new().unwrap(); + db.create_table("t", "CREATE TABLE t (id TEXT)").unwrap(); + db.insert_rows( + "t", + &shaped(&db, "t", serde_json::json!([{"id":"a"},{"id":"b"}])), + "x.json", + ) + .unwrap(); + db.insert_rows( + "t", + &shaped(&db, "t", serde_json::json!([{"id":"c"}])), + "y.json", + ) + .unwrap(); + db.insert_row( + "t", + &HashMap::from([("id".into(), Value::Text("m".into()))]), + "x.json", + 0, + ) + .unwrap(); + + assert_eq!(db.delete_rows_by_file("t", "x.json").unwrap(), 3); + + assert!(db.get_rows_by_file("t", "x.json").unwrap().is_empty()); + assert!(mapping_rows(&db, "t").is_empty()); + assert_eq!(range_rows(&db, "t"), vec![("y.json".to_string(), 3, 3)]); + let left: i64 = db + .conn + .query_row("SELECT COUNT(*) FROM t", [], |r| r.get(0)) + .unwrap(); + assert_eq!(left, 1); + } + + #[test] + fn files_under_lists_range_owners_too() { + let db = Db::new().unwrap(); + db.create_table("t", "CREATE TABLE t (id TEXT)").unwrap(); + db.insert_rows( + "t", + &shaped(&db, "t", serde_json::json!([{"id":"a"}])), + "dir/a.json", + ) + .unwrap(); + db.insert_rows( + "t", + &shaped(&db, "t", serde_json::json!([{"id":"b"}])), + "dir/a.json", + ) + .unwrap(); + db.insert_rows("t", &shaped(&db, "t", serde_json::json!([{"id":"c"}])), "") + .unwrap(); + db.insert_row( + "t", + &HashMap::from([("id".into(), Value::Text("m".into()))]), + "dir/b.json", + 0, + ) + .unwrap(); + + assert_eq!( + db.files_under("dir").unwrap(), + vec![ + ("t".to_string(), "dir/a.json".to_string()), + ("t".to_string(), "dir/b.json".to_string()), + ] + ); } #[test] fn insert_rows_joins_callers_open_transaction() { let db = Db::new().unwrap(); db.create_table("t", "CREATE TABLE t (id TEXT)").unwrap(); - let rows = vec![HashMap::from([("id".into(), Value::Text("a".into()))])]; + let rows = shaped(&db, "t", serde_json::json!([{"id":"a"}])); let tx = db.conn.unchecked_transaction().unwrap(); db.insert_rows("t", &rows, "a.json").unwrap(); drop(tx); assert!(db.get_rows_by_file("t", "a.json").unwrap().is_empty()); - assert!(mapping_rows(&db, "t").is_empty()); + assert!(range_rows(&db, "t").is_empty()); } #[test] fn insert_rows_rejects_an_unsafe_table_name() { let db = Db::new().unwrap(); - let err = db.insert_rows("bad name", &[], "a.json").unwrap_err(); + let rows = ShapedRows { + columns: vec![], + cells: vec![], + }; + let err = db.insert_rows("bad name", &rows, "a.json").unwrap_err(); assert!(matches!(err, DbError::InvalidIdentifier(_)), "got: {err}"); } diff --git a/packages/rust/src/lib.rs b/packages/rust/src/lib.rs index 1ec42d824..489ff2f9c 100644 --- a/packages/rust/src/lib.rs +++ b/packages/rust/src/lib.rs @@ -52,7 +52,7 @@ pub mod watcher; pub mod cli; use crate::config::Source; -use crate::db::Db; +use crate::db::{Db, ShapedRows}; use crate::functions::ResolvedFunction; use crate::matcher::TableMatcher; use crate::persist::{ @@ -97,8 +97,10 @@ pub const DEFAULT_CONFIG_TOML: &str = include_str!("default_config.toml"); type BoxError = Box; type OnFileFn = dyn Fn(&str) -> std::result::Result, BoxError> + Send + Sync + 'static; -type OnFilesFn = - dyn Fn(&[PathBuf]) -> std::result::Result, BoxError> + Send + Sync + 'static; +type OnFilesFn = dyn Fn(&[PathBuf]) -> std::result::Result, BoxError> + + Send + + Sync + + 'static; /// How a table turns matched files into rows: a programmatic callback runs /// once per file; a configured `on-file` command runs once per table over @@ -333,7 +335,10 @@ impl Table { on_files: F, ) -> Self where - F: Fn(&[PathBuf]) -> std::result::Result, BoxError> + Send + Sync + 'static, + F: Fn(&[PathBuf]) -> std::result::Result, BoxError> + + Send + + Sync + + 'static, { Self { name: name.into(), @@ -685,7 +690,7 @@ impl DirSQL { Ok(g) => g, Err(e) => return vec![error_event(Some(table), trigger, e.to_string())], }; - let new_rows = match normalize_rows(&db, table, raw_rows, strict) { + let new_rows = match shape_rows(&db, table, raw_rows, strict) { Ok(rows) => rows, Err(message) => return vec![error_event(Some(table), trigger, message)], }; @@ -703,7 +708,7 @@ impl DirSQL { if let Err(e) = _tx.commit() { return vec![error_event(Some(table), trigger, e.to_string())]; } - (old_rows, new_rows) + (old_rows, new_rows.to_maps()) }; if let Ok(mut files) = self.inner.batch_files.lock() { files.insert(table.to_string(), rel_paths); @@ -1288,7 +1293,7 @@ impl DirSQL { let abs_paths: Vec = rel_paths.iter().map(|r| root.join(r)).collect(); let outcome = hook(&abs_paths) .map_err(|e| e.to_string()) - .and_then(|raw| normalize_rows(&db, &table_name, raw, strict)); + .and_then(|raw| shape_rows(&db, &table_name, raw, strict)); match outcome { Ok(rows) => { replace_table_rows(&db, &table_name, &rows).map_err(map_db_error)?; @@ -2023,8 +2028,20 @@ fn normalize_rows( .map_err(|e| e.to_string()) } +/// Shape every row before any is inserted, so a hook that got one row wrong +/// fails as a unit rather than leaving half its rows behind. +fn shape_rows( + db: &Db, + table: &str, + raw_rows: Vec, + strict: bool, +) -> std::result::Result { + db.shape_rows(table, raw_rows, strict) + .map_err(|e| e.to_string()) +} + /// Replace a per-table hook's rows wholesale under [`BATCH_OWNER`]. -fn replace_table_rows(db: &Db, table: &str, rows: &[Row]) -> db::Result<()> { +fn replace_table_rows(db: &Db, table: &str, rows: &ShapedRows) -> db::Result<()> { db.delete_rows_by_file(table, BATCH_OWNER)?; db.insert_rows(table, rows, BATCH_OWNER) } @@ -2159,23 +2176,8 @@ fn run_on_files( paths: &[PathBuf], config_dir: &Path, root: &Path, -) -> std::result::Result, BoxError> { - let rows = on_file::run(command, config_dir, root, paths)?; - Ok(rows.into_iter().map(json_row_into_row).collect()) -} - -fn json_row_into_row(row: infer::JsonRow) -> Row { - row.0 - .into_iter() - .map(|(key, value)| (key, json_into_value(value))) - .collect() -} - -fn json_into_value(value: serde_json::Value) -> Value { - match value { - serde_json::Value::String(s) => Value::Text(s), - other => json_to_value(&other), - } +) -> std::result::Result, BoxError> { + Ok(on_file::run(command, config_dir, root, paths)?) } /// Map a JSON value to a SQLite [`Value`]: `null` → `Null`; `bool` → `Integer` @@ -4448,7 +4450,7 @@ mod internal_tests { ) .expect("a well-formed payload parses"); assert_eq!(rows.len(), 1); - assert_eq!(rows[0]["n"], Value::Integer(1)); + assert_eq!(rows[0].get("n"), Some(&serde_json::json!(1))); } /// `{abspath}` is not in the substitution table: it is left literal like any @@ -4464,7 +4466,7 @@ mod internal_tests { ) .expect("a well-formed payload parses"); assert_eq!(rows.len(), 1); - assert_eq!(rows[0]["q"], Value::Text("{abspath}".into())); + assert_eq!(rows[0].get("q"), Some(&serde_json::json!("{abspath}"))); } /// Every matched file's **absolute** path arrives as a trailing argument, @@ -4667,6 +4669,10 @@ mod internal_tests { Row::from_iter([("name".to_string(), Value::Text(name.into()))]) } + fn name_json_row(name: &str) -> infer::JsonRow { + infer::JsonRow(vec![("name".to_string(), serde_json::json!(name))]) + } + type Calls = Arc>>>; /// A per-table hook that records every argument list it is handed and @@ -4682,7 +4688,7 @@ mod internal_tests { seen.lock().unwrap().push(paths.to_vec()); Ok(paths .iter() - .map(|p| name_row(&p.file_name().unwrap().to_string_lossy())) + .map(|p| name_json_row(&p.file_name().unwrap().to_string_lossy())) .collect()) }, ); @@ -4716,6 +4722,37 @@ mod internal_tests { assert!(db.scan_failures().is_empty(), "{:?}", db.scan_failures()); } + #[test] + fn finish_build_owns_a_per_table_hooks_rows_as_one_range_not_a_mapping_row_each() { + let dir = TempDir::new().unwrap(); + let (items, _) = recording_per_table("items", "*.txt"); + let scanned = vec![ + scanned("a.txt", "items", false), + scanned("b.txt", "items", false), + scanned("c.txt", "items", false), + ]; + + let db = build(prepared_build(dir.path(), vec![items], scanned, None)); + + assert_eq!(row_names(&db), vec!["a.txt", "b.txt", "c.txt"]); + let mapped: i64 = db + .inner + .db + .lock() + .unwrap() + .conn() + .query_row( + "SELECT COUNT(*) FROM _dirsql_internal_rows WHERE table_name = 'items'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!( + mapped, 0, + "a per-table hook's rows are owned by one rowid range, not a mapping row each" + ); + } + /// A `Write` the test reads back, standing in for stderr. #[derive(Clone, Default)] struct Captured(Arc>>); @@ -4921,11 +4958,15 @@ mod command_rows_tests { use super::*; fn rows(payload: &str) -> Vec { - infer::parse_rows(payload) - .unwrap() - .into_iter() - .map(json_row_into_row) - .collect() + let parsed = infer::parse_rows(payload).unwrap(); + let columns: Vec = parsed + .first() + .map(|row| row.0.iter().map(|(key, _)| format!("\"{key}\"")).collect()) + .unwrap_or_else(|| vec!["x".to_string()]); + let db = Db::new().unwrap(); + db.create_table("t", &format!("CREATE TABLE t ({})", columns.join(", "))) + .unwrap(); + db.shape_rows("t", parsed, false).unwrap().to_maps() } #[test] diff --git a/packages/rust/src/persist.rs b/packages/rust/src/persist.rs index ac1c408bc..a450c511b 100644 --- a/packages/rust/src/persist.rs +++ b/packages/rust/src/persist.rs @@ -425,9 +425,10 @@ pub fn drop_user_tables(conn: &Connection) -> rusqlite::Result<()> { conn.execute(&format!("DROP TABLE IF EXISTS \"{name}\""), [])?; } conn.execute("DELETE FROM _dirsql_files", [])?; - // Wipe the row mapping too: a cold rebuild re-ingests every file and - // `insert_row` repopulates it; stale rows would duplicate/orphan state. + // Wipe the row ownership too: a cold rebuild re-ingests every file and + // repopulates it; stale rows would duplicate/orphan state. conn.execute("DELETE FROM _dirsql_internal_rows", [])?; + conn.execute("DELETE FROM _dirsql_internal_ranges", [])?; // The parsed path-table rows are keyed by their own identity hash, not by // the meta the caller just rejected, but they are cached data in a cache // being discarded: keep the file one thing, wholly stale or wholly warm. @@ -606,6 +607,14 @@ mod tests { assert!(read_cached_files(&conn).unwrap().is_empty()); } + /// These tests drive a raw connection, so `Db::open` never creates these. + const INTERNAL_ROWS_DDL: &str = "CREATE TABLE _dirsql_internal_rows ( + table_name TEXT NOT NULL, file_path TEXT NOT NULL, + row_index INTEGER NOT NULL, rowid_ref INTEGER NOT NULL); + CREATE TABLE _dirsql_internal_ranges ( + table_name TEXT NOT NULL, file_path TEXT NOT NULL, + first_rowid INTEGER NOT NULL, last_rowid INTEGER NOT NULL);"; + /// A `ddl` batch may leave a virtual table and its shadow tables in the /// cache. The sweep must clear all of it: the shadows go with the virtual /// table, and dropping one out from under the other is what poisons the @@ -614,13 +623,7 @@ mod tests { fn drop_user_tables_clears_virtual_tables_and_their_shadows() { let conn = Connection::open_in_memory().unwrap(); create_sidecar_tables(&conn).unwrap(); - conn.execute( - "CREATE TABLE _dirsql_internal_rows ( - table_name TEXT NOT NULL, file_path TEXT NOT NULL, - row_index INTEGER NOT NULL, rowid_ref INTEGER NOT NULL)", - [], - ) - .unwrap(); + conn.execute_batch(INTERNAL_ROWS_DDL).unwrap(); conn.execute_batch( "CREATE TABLE notes (body TEXT);\n\ CREATE VIRTUAL TABLE notes_fts USING fts5(body, content='notes');", @@ -649,17 +652,9 @@ mod tests { fn drop_user_tables_clears_user_data_and_files_index() { let conn = Connection::open_in_memory().unwrap(); create_sidecar_tables(&conn).unwrap(); - // `_dirsql_internal_rows` is created by `Db::open` in production; - // declared inline here since this test uses a raw connection. - conn.execute( - "CREATE TABLE _dirsql_internal_rows ( - table_name TEXT NOT NULL, file_path TEXT NOT NULL, - row_index INTEGER NOT NULL, rowid_ref INTEGER NOT NULL)", - [], - ) - .unwrap(); + conn.execute_batch(INTERNAL_ROWS_DDL).unwrap(); conn.execute("CREATE TABLE rows (x TEXT)", []).unwrap(); - conn.execute("INSERT INTO rows (x) VALUES ('a')", []) + conn.execute("INSERT INTO rows (x) VALUES ('a'), ('b')", []) .unwrap(); conn.execute( "INSERT INTO _dirsql_internal_rows (table_name, file_path, row_index, rowid_ref) \ @@ -667,6 +662,12 @@ mod tests { [], ) .unwrap(); + conn.execute( + "INSERT INTO _dirsql_internal_ranges (table_name, file_path, first_rowid, last_rowid) \ + VALUES ('rows', '', 2, 2)", + [], + ) + .unwrap(); let stat = FileStat { size: 1, mtime_ns: 1, @@ -694,6 +695,12 @@ mod tests { }) .unwrap(); assert_eq!(mapping_count, 0); + let range_count: i64 = conn + .query_row("SELECT COUNT(*) FROM _dirsql_internal_ranges", [], |r| { + r.get(0) + }) + .unwrap(); + assert_eq!(range_count, 0); // Meta is preserved; callers replace it explicitly. let m = read_meta(&conn).unwrap(); assert_eq!(m.get("x"), Some(&"y".to_string())); diff --git a/packages/rust/tests/internal_rows.rs b/packages/rust/tests/internal_rows.rs index 7a9760860..001835497 100644 --- a/packages/rust/tests/internal_rows.rs +++ b/packages/rust/tests/internal_rows.rs @@ -4,7 +4,7 @@ //! consistent with the live user rows across create / modify / delete and //! round-trips the persisted cache. -use dirsql::db::{Db, INTERNAL_ROWS_TABLE}; +use dirsql::db::{Db, INTERNAL_RANGES_TABLE, INTERNAL_ROWS_TABLE}; use dirsql::{DirSQL, Table, Value}; use rusqlite::Connection; use std::collections::HashMap; @@ -111,6 +111,10 @@ fn mapping_table_is_a_durable_sidecar() { table_exists(&conn, INTERNAL_ROWS_TABLE), "{INTERNAL_ROWS_TABLE} must persist in the cache" ); + assert!( + table_exists(&conn, INTERNAL_RANGES_TABLE), + "{INTERNAL_RANGES_TABLE} must persist in the cache" + ); } fn table_exists(conn: &Connection, name: &str) -> bool { diff --git a/packages/rust/tests/internal_table_auth.rs b/packages/rust/tests/internal_table_auth.rs index bc49322f3..72b3dccdd 100644 --- a/packages/rust/tests/internal_table_auth.rs +++ b/packages/rust/tests/internal_table_auth.rs @@ -63,6 +63,13 @@ fn query_rejects_internal_rows_table() { assert_rejected(&db, "_dirsql_internal_rows"); } +#[test] +fn query_rejects_internal_ranges_table() { + let root = TempDir::new().unwrap(); + let db = persisted_db(root.path()); + assert_rejected(&db, "_dirsql_internal_ranges"); +} + #[test] fn query_rejects_files_table() { let root = TempDir::new().unwrap(); diff --git a/packages/ts/e2e-attestations/claude-1243-ingest-perf.json b/packages/ts/e2e-attestations/claude-1243-ingest-perf.json new file mode 100644 index 000000000..47a87ccfc --- /dev/null +++ b/packages/ts/e2e-attestations/claude-1243-ingest-perf.json @@ -0,0 +1,7 @@ +{ + "command": "pnpm test:e2e", + "ran_at": 1790978416, + "exit_code": 0, + "commit": "a259a6ab12807856f8d5f81fc3068b5e53a2a99a", + "branch": "claude/1243-ingest-perf" +}