From ec26d717e8a090d36915c5dd472f9b4311e5c8a7 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Mon, 2 Nov 2026 18:06:53 -0500 Subject: [PATCH 1/4] test: an in-place batch update writes only the updated row --- packages/rust/src/db.rs | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/packages/rust/src/db.rs b/packages/rust/src/db.rs index 40c17807..cf950bde 100644 --- a/packages/rust/src/db.rs +++ b/packages/rust/src/db.rs @@ -4176,6 +4176,25 @@ mod tests { assert_eq!(changes.added.len(), 1, "got: {:?}", changes.added); } + #[test] + fn reconcile_batch_rows_writes_only_the_update_when_rows_change_in_place() { + let db = experiments_db(); + let next = shaped( + &db, + "experiments", + serde_json::json!([ + {"folder_name":"a","name":"Alpha"}, + {"folder_name":"b","name":"Renamed"} + ]), + ); + let before = db.conn().total_changes(); + + db.reconcile_batch_rows("experiments", "", &next).unwrap(); + + assert_eq!(db.conn().total_changes() - before, 1); + assert_eq!(range_rows(&db, "experiments"), vec![(String::new(), 1, 2)]); + } + #[test] fn reconcile_batch_rows_writes_nothing_for_an_unchanged_keyed_table() { let db = experiments_db(); From 900d89821767c1af609f1c1b2f21bf87c97c2332 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Mon, 2 Nov 2026 18:25:00 -0500 Subject: [PATCH 2/4] perf: cut the SQLite work in a per-table rerun's reconcile Read a table's key candidates once per schema version, read its columns through a cached statement, skip the range refit when every change is an in-place update, and fit the range from the rowids the reconcile placed rather than a min/max query. --- ...-11-02-faster-per-table-rerun-reconcile.md | 1 + ...-11-02-faster-per-table-rerun-reconcile.md | 30 ++++ packages/rust/src/db.rs | 128 ++++++++++++++---- 3 files changed, 136 insertions(+), 23 deletions(-) create mode 100644 packages/rust/changelog.d/2026-11-02-faster-per-table-rerun-reconcile.md create mode 100644 packages/rust/migrations.d/2026-11-02-faster-per-table-rerun-reconcile.md diff --git a/packages/rust/changelog.d/2026-11-02-faster-per-table-rerun-reconcile.md b/packages/rust/changelog.d/2026-11-02-faster-per-table-rerun-reconcile.md new file mode 100644 index 00000000..c0679d39 --- /dev/null +++ b/packages/rust/changelog.d/2026-11-02-faster-per-table-rerun-reconcile.md @@ -0,0 +1 @@ +**Changed** A watched change to a table filled by a per-table `on-file` command reaches `/events` sooner: the table's key and columns are read from the schema once per schema change rather than on every rerun, and a rerun that only updates rows in place no longer rewrites the table's row range. diff --git a/packages/rust/migrations.d/2026-11-02-faster-per-table-rerun-reconcile.md b/packages/rust/migrations.d/2026-11-02-faster-per-table-rerun-reconcile.md new file mode 100644 index 00000000..ce2adb7f --- /dev/null +++ b/packages/rust/migrations.d/2026-11-02-faster-per-table-rerun-reconcile.md @@ -0,0 +1,30 @@ +### Faster per-table rerun reconcile + +#### Summary + +A per-table `on-file` rerun after a watched change does less SQLite work before its events are sent. Nothing breaks. + +#### Required changes + +_None._ + +#### Deprecations removed + +_None._ + +#### Behavior changes without code changes + +_None._ + +#### Verification + +```bash +mkdir -p d +echo '{"v": 0}' > d/f.jl +printf '[[table]]\nname = "rows"\nglob = "**/*.jl"\nddl = "CREATE TABLE rows (v INTEGER)"\non-file = "cat"\n' > .dirsql.toml +dirsql server -c .dirsql.toml --port 7117 & +curl -sN http://localhost:7117/events & +echo '{"v": 1}' > d/f.jl +``` + +Prints a `rows` event carrying `"row":{"v":1}`. diff --git a/packages/rust/src/db.rs b/packages/rust/src/db.rs index cf950bde..d79d9042 100644 --- a/packages/rust/src/db.rs +++ b/packages/rust/src/db.rs @@ -1,5 +1,6 @@ use rusqlite::Connection; use rusqlite::types::{ToSql, ToSqlOutput, ValueRef}; +use std::cell::RefCell; use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -468,8 +469,15 @@ pub struct Db { /// The declared functions that batch. Every statement opens a collect /// pass on each of them; empty, a statement runs exactly once. batched: Vec>, + /// Each table's key candidates (see [`Self::row_key`]) as of a schema + /// version: reading them takes three pragma queries, and a watched + /// change would otherwise pay for them every time. + keys: RefCell>, } +/// A table's `PRIMARY KEY` column names, then each plain `UNIQUE` index's. +type KeyCandidates = Vec>>; + /// Drops every batching worker's gathered replies when the statement that /// gathered them is done, whichever way it left. struct ClearOnDrop<'a>(&'a [Arc]); @@ -521,6 +529,7 @@ impl Db { path_table_parser: None, calls: CallReporter::new(), batched: Vec::new(), + keys: RefCell::default(), }) } @@ -546,6 +555,7 @@ impl Db { path_table_parser: None, calls: CallReporter::new(), batched: Vec::new(), + keys: RefCell::default(), }) } @@ -657,12 +667,11 @@ impl Db { /// Return the column names declared in `table`'s DDL. pub fn get_table_columns(&self, table: &str) -> Result> { validate_identifier(table)?; - let mut stmt = self.conn.prepare(&format!( - "PRAGMA table_info({})", - crate::sql_literal::quote_identifier(table) - ))?; + let mut stmt = self + .conn + .prepare_cached("SELECT name FROM pragma_table_info(?1) ORDER BY cid")?; let columns: Vec = stmt - .query_map([], |row| row.get::<_, String>(1))? + .query_map([table], |row| row.get::<_, String>(0))? .filter_map(|r| r.ok()) .collect(); Ok(columns) @@ -902,20 +911,19 @@ impl Db { &mut stored, )?; let key = self.row_key(table, rows)?; - let (changes, _) = self.reconcile_rows(table, &stored, rows, key.as_deref())?; - if changes.is_empty() { + let (changes, placed) = self.reconcile_rows(table, &stored, rows, key.as_deref())?; + // An update keeps its rowid, so only a removal or an addition can move + // the owner's range. + if changes.removed.is_empty() && changes.added.is_empty() { return Ok(changes); } - self.conn.execute( - "DELETE FROM _dirsql_internal_ranges WHERE table_name = ?1 AND file_path = ?2", - rusqlite::params![table, owner], - )?; - let (first, last): (Option, Option) = self.conn.query_row( - &format!("SELECT min(rowid), max(rowid) FROM \"{table}\""), - [], - |r| Ok((r.get(0)?, r.get(1)?)), - )?; - if let (Some(first), Some(last)) = (first, last) { + self.conn + .prepare_cached( + "DELETE FROM _dirsql_internal_ranges WHERE table_name = ?1 AND file_path = ?2", + )? + .execute(rusqlite::params![table, owner])?; + // `placed` holds every row the table keeps. + if let (Some(&first), Some(&last)) = (placed.iter().min(), placed.iter().max()) { self.conn .prepare_cached(RANGE_INSERT)? .execute(rusqlite::params![table, owner, first, last])?; @@ -1135,12 +1143,10 @@ impl Db { Ok((changes, placed)) } - /// The positions in `rows.columns` of `table`'s key: its `PRIMARY KEY`, - /// else the first `UNIQUE` index whose columns `rows` emits non-NULL in - /// every row. A key the rows leave unset (a generated `INTEGER PRIMARY - /// KEY`) cannot match anything, so it is passed over. - fn row_key(&self, table: &str, rows: &ShapedRows) -> Result>> { - let mut candidates: Vec>> = vec![ + /// `table`'s `PRIMARY KEY` column names, then each plain `UNIQUE` + /// index's, as the schema declares them now. + fn key_candidates(&self, table: &str) -> Result { + let mut candidates: KeyCandidates = vec![ self.conn .prepare_cached("SELECT name FROM pragma_table_info(?1) WHERE pk > 0 ORDER BY pk")? .query_map([table], |r| r.get(0))? @@ -1164,6 +1170,34 @@ impl Db { .collect::>()?, ); } + Ok(candidates) + } + + /// The positions in `rows.columns` of `table`'s key: its `PRIMARY KEY`, + /// else the first `UNIQUE` index whose columns `rows` emits non-NULL in + /// every row. A key the rows leave unset (a generated `INTEGER PRIMARY + /// KEY`) cannot match anything, so it is passed over. + fn row_key(&self, table: &str, rows: &ShapedRows) -> Result>> { + let version: i64 = self + .conn + .prepare_cached("PRAGMA schema_version")? + .query_row([], |r| r.get(0))?; + let cached = self + .keys + .borrow() + .get(table) + .filter(|(at, _)| *at == version) + .map(|(_, candidates)| candidates.clone()); + let candidates = match cached { + Some(candidates) => candidates, + None => { + let candidates = self.key_candidates(table)?; + self.keys + .borrow_mut() + .insert(table.to_string(), (version, candidates.clone())); + candidates + } + }; Ok(candidates.into_iter().find_map(|names| { let at = names .iter() @@ -4195,6 +4229,54 @@ mod tests { assert_eq!(range_rows(&db, "experiments"), vec![(String::new(), 1, 2)]); } + #[test] + fn reconcile_batch_rows_keys_by_an_index_created_after_an_earlier_reconcile() { + let changes = |db: &Db, v: &str| { + let next = shaped(db, "t", serde_json::json!([{"k":"a","v":v}])); + db.reconcile_batch_rows("t", "", &next).unwrap() + }; + let db = Db::new().unwrap(); + db.create_table("t", "CREATE TABLE t (k TEXT, v TEXT)") + .unwrap(); + db.insert_rows( + "t", + &shaped(&db, "t", serde_json::json!([{"k":"a","v":"0"}])), + "", + ) + .unwrap(); + assert!(!changes(&db, "1").keyed); + + db.conn() + .execute_batch("CREATE UNIQUE INDEX t_k ON t(k)") + .unwrap(); + + assert!(changes(&db, "2").keyed); + } + + #[test] + fn reconcile_batch_rows_reuses_the_key_read_at_the_same_schema_version() { + let db = experiments_db(); + let version: i64 = db + .conn() + .query_row("PRAGMA schema_version", [], |r| r.get(0)) + .unwrap(); + db.keys + .borrow_mut() + .insert("experiments".into(), (version, Vec::new())); + let next = shaped( + &db, + "experiments", + serde_json::json!([ + {"folder_name":"a","name":"Alpha"}, + {"folder_name":"b","name":"Renamed"} + ]), + ); + + let changes = db.reconcile_batch_rows("experiments", "", &next).unwrap(); + + assert!(!changes.keyed); + } + #[test] fn reconcile_batch_rows_writes_nothing_for_an_unchanged_keyed_table() { let db = experiments_db(); From 958013077e5930f6f7feb0a079c69d27be7c98e8 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Mon, 2 Nov 2026 18:31:53 -0500 Subject: [PATCH 3/4] e2e attestation for bot/1562-watch-speed-1102 --- .../python/e2e-attestations/bot-1562-watch-speed-1102.json | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 packages/python/e2e-attestations/bot-1562-watch-speed-1102.json diff --git a/packages/python/e2e-attestations/bot-1562-watch-speed-1102.json b/packages/python/e2e-attestations/bot-1562-watch-speed-1102.json new file mode 100644 index 00000000..741b6f82 --- /dev/null +++ b/packages/python/e2e-attestations/bot-1562-watch-speed-1102.json @@ -0,0 +1,7 @@ +{ + "command": "uv run python -m pytest tests/e2e/on_file_query_test.py tests/e2e/query_contract_test.py tests/e2e/declared_table_name_test.py tests/e2e/fanout_test.py tests/e2e/use_case_dossier_test.py tests/e2e/use_case_join_test.py tests/e2e/use_case_notebook_test.py tests/e2e/use_case_transcripts_test.py tests/e2e/watch_speed_test.py -q -p no:cacheprovider", + "ran_at": 1793662313, + "exit_code": 0, + "commit": "900d89821767c1af609f1c1b2f21bf87c97c2332", + "branch": "bot/1562-watch-speed-1102" +} From 97776daea103a35501079b9461b9d98e7d04a648 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Mon, 2 Nov 2026 18:33:00 -0500 Subject: [PATCH 4/4] e2e attestation for bot/1562-watch-speed-1102 --- .../ts/e2e-attestations/bot-1562-watch-speed-1102.json | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 packages/ts/e2e-attestations/bot-1562-watch-speed-1102.json diff --git a/packages/ts/e2e-attestations/bot-1562-watch-speed-1102.json b/packages/ts/e2e-attestations/bot-1562-watch-speed-1102.json new file mode 100644 index 00000000..e0d0b16e --- /dev/null +++ b/packages/ts/e2e-attestations/bot-1562-watch-speed-1102.json @@ -0,0 +1,7 @@ +{ + "command": "pnpm test:e2e", + "ran_at": 1793662380, + "exit_code": 0, + "commit": "958013077e5930f6f7feb0a079c69d27be7c98e8", + "branch": "bot/1562-watch-speed-1102" +}