From 4eaa7f0c26fa324fdb9b161e90fdb3fe22b8774f Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Fri, 16 Oct 2026 09:40:25 -0400 Subject: [PATCH 1/5] perf: shape per-table hook rows beside their insert Shapes each chunk on its own thread, settles early tables' file lists on the grouping thread, and defers the scan's ScannedFile copy there too. --- .../2026-10-16-shape-beside-insert.md | 3 + .../2026-10-16-shape-beside-insert.md | 25 +++ packages/rust/src/db.rs | 106 ++++++----- packages/rust/src/lib.rs | 165 ++++++++++++------ 4 files changed, 202 insertions(+), 97 deletions(-) create mode 100644 packages/rust/changelog.d/2026-10-16-shape-beside-insert.md create mode 100644 packages/rust/migrations.d/2026-10-16-shape-beside-insert.md diff --git a/packages/rust/changelog.d/2026-10-16-shape-beside-insert.md b/packages/rust/changelog.d/2026-10-16-shape-beside-insert.md new file mode 100644 index 000000000..8e23a7290 --- /dev/null +++ b/packages/rust/changelog.d/2026-10-16-shape-beside-insert.md @@ -0,0 +1,3 @@ +**Changed** + +- **A per-table `on-file` hook's rows are shaped on their own thread while the previous chunk is inserted.** Startup over a large tree, such as a frontmatter table over 131k notes, finishes about 4% sooner. Query results are unchanged. diff --git a/packages/rust/migrations.d/2026-10-16-shape-beside-insert.md b/packages/rust/migrations.d/2026-10-16-shape-beside-insert.md new file mode 100644 index 000000000..5dec05f6b --- /dev/null +++ b/packages/rust/migrations.d/2026-10-16-shape-beside-insert.md @@ -0,0 +1,25 @@ +### Shape beside insert + +#### Summary + +Startup shapes a per-table hook's rows on a separate thread from the one inserting them. No public API changes. + +#### Required changes + +_None._ + +#### Deprecations removed + +_None._ + +#### Behavior changes without code changes + +_None._ + +#### Verification + +```bash +dirsql query "SELECT status, count(*) FROM notes GROUP BY status" -c .dirsql.toml +``` + +Prints the same rows as before, sooner. diff --git a/packages/rust/src/db.rs b/packages/rust/src/db.rs index d2d51a16e..8a63f3311 100644 --- a/packages/rust/src/db.rs +++ b/packages/rust/src/db.rs @@ -312,6 +312,63 @@ impl ShapedRows { } } +/// [`Db::shape_rows`] against `columns`, the table's columns in DDL order, +/// so rows can be shaped off the connection's thread. +pub(crate) fn shape_json_rows( + table: &str, + columns: Vec, + rows: Vec, + strict: bool, +) -> Result { + 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) + .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 }) +} + fn json_into_value(value: serde_json::Value) -> Value { match value { serde_json::Value::String(s) => Value::Text(s), @@ -714,54 +771,7 @@ impl Db { /// 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) - .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 }) + shape_json_rows(table, self.get_table_columns(table)?, rows, strict) } /// Insert every row of `rows` under `file_path`, in order, through one diff --git a/packages/rust/src/lib.rs b/packages/rust/src/lib.rs index f16289c8d..1a79b9ee4 100644 --- a/packages/rust/src/lib.rs +++ b/packages/rust/src/lib.rs @@ -1288,20 +1288,11 @@ impl DirSQL { // (and that pass the racy-window check) are trusted instead of // re-parsed. let (scanned_files, deleted) = match &persist_ctx { - None => { - let files = scanned - .into_iter() - .map(|found| ScannedFile { - abs_path: found.path, - rel_path: found.rel_path, - table_name: found.table_name, - stat: None, - trusted: false, - }) - .collect(); - (files, Vec::new()) + None => (Scanned::Found(scanned), Vec::new()), + Some(ctx) => { + let (files, deleted) = reconcile_scan(scanned, ctx, &RealFs)?; + (Scanned::Files(files), deleted) } - Some(ctx) => reconcile_scan(scanned, ctx, &RealFs)?, }; Ok(PreparedBuild { @@ -1482,17 +1473,22 @@ impl DirSQL { let mut batch_files: HashMap> = HashMap::new(); // Grouping walks every scanned file, so it runs beside the early - // hooks' inserts instead of ahead of them. + // hooks' inserts instead of ahead of them, and settles the early + // tables' file lists there too. + let early_tables: HashSet = early.keys().cloned().collect(); let (grouped, streamed) = std::thread::scope(|scope| { - let grouping = scope.spawn(|| group_scanned(scanned_files, &on_file_map)); + let grouping = scope.spawn(|| { + group_scanned(scanned_files.into_files(), &on_file_map) + .map(|(per_file, batches)| (per_file, split_streamed(batches, &early_tables))) + }); let streamed = stream_early(&db, early, &strict_map); ( grouping.join().expect("grouping scanned files panicked"), streamed, ) }); - let streamed = streamed?; - let (per_file, batches) = grouped?; + streamed?; + let (per_file, (batches, streamed)) = grouped?; // The phase a user actually waits on: one `on_file` round trip per // file or per table, plus whatever the table's DDL triggers on insert. @@ -1501,8 +1497,14 @@ impl DirSQL { + batches .iter() .map(|(_, _, files)| files.len()) - .sum::()) as u64; + .sum::() + + streamed.iter().map(|(_, rel)| rel.len()).sum::()) + as u64; let mut done: u64 = 0; + for (table_name, rel_paths) in streamed { + done += rel_paths.len() as u64; + batch_files.insert(table_name, rel_paths); + } for ( ScannedFile { rel_path, @@ -1580,7 +1582,7 @@ impl DirSQL { let current = persist_ready.as_ref().is_some_and(|(deleted, _, _)| { files.iter().all(|f| f.trusted) && !deleted.iter().any(|(_, t)| *t == table_name) }); - if !current && !streamed.contains(&table_name) { + if !current { let table_root = anchors.of(&table_name); let production = start_hook(hook, abs_paths.into_iter()); let outcome = stream_table_rows(&db, &table_name, production, strict) @@ -2004,6 +2006,33 @@ pub struct ScannedFile { pub trusted: bool, } +/// The files the startup scan found: as the walk found them, or already +/// reconciled against the persistent cache. +enum Scanned { + Found(Vec), + Files(Vec), +} + +impl Scanned { + /// Every file as a [`ScannedFile`]. A fresh walk's are made here rather + /// than in prepare, so the copy runs off the thread that inserts rows. + fn into_files(self) -> Vec { + match self { + Self::Found(found) => found + .into_iter() + .map(|found| ScannedFile { + abs_path: found.path, + rel_path: found.rel_path, + table_name: found.table_name, + stat: None, + trusted: false, + }) + .collect(), + Self::Files(files) => files, + } + } +} + /// Intermediate state produced by [`DirSQL::prepare_resolved`] and consumed /// by [`DirSQL::finish_build`]. Opaque on purpose; fields are only visible /// to the in-workspace bindings. @@ -2020,7 +2049,7 @@ pub struct PreparedBuild { /// The configured skip rules, carried through so path-table scans apply /// the same ones declared tables do. ignore: Vec, - scanned_files: Vec, + scanned_files: Scanned, /// Per-table hooks already running over the paths the scan found, by table. early: HashMap, persist: Option, @@ -2437,9 +2466,10 @@ fn collect_table_rows( /// Replace a per-table hook's rows under [`BATCH_OWNER`], storing each chunk /// the hook emits while the hook, on its own thread, goes on producing the -/// next. The caller's open transaction keeps a failure from being seen: the -/// outer `Err` is SQLite's, the inner one the hook's or a row's. A hook -/// failure outranks a row that failed to shape. +/// next. Chunks are shaped on a thread of their own, in order, so shaping one +/// overlaps inserting the last. The caller's open transaction keeps a failure +/// from being seen: the outer `Err` is SQLite's, the inner one the hook's or a +/// row's. A hook failure outranks a row that failed to shape. fn stream_table_rows( db: &Db, table: &str, @@ -2447,17 +2477,32 @@ fn stream_table_rows( strict: bool, ) -> db::Result> { db.delete_rows_by_file(table, BATCH_OWNER)?; - let mut shaped = Ok(()); - for chunk in production.chunks { - if shaped.is_err() { - continue; + let columns = db.get_table_columns(table)?; + let Production { chunks, handle } = production; + let (shaped_tx, shaped_rx) = std::sync::mpsc::channel(); + let shaper = std::thread::spawn({ + let table = table.to_string(); + move || { + for chunk in chunks { + let shaped = db::shape_json_rows(&table, columns.clone(), chunk, strict); + if shaped_tx.send(shaped).is_err() { + return; + } + } } - match db.shape_rows(table, chunk, strict) { + }); + let mut shaped = Ok(()); + for rows in shaped_rx { + match rows { Ok(rows) => db.insert_rows(table, &rows, BATCH_OWNER)?, - Err(e) => shaped = Err(e.to_string()), + Err(e) => { + shaped = Err(e.to_string()); + break; + } } } - let produced = production.handle.join().expect("per-table hook panicked"); + shaper.join().expect("shaping a hook's rows panicked"); + let produced = handle.join().expect("per-table hook panicked"); Ok(produced.and(shaped)) } @@ -2591,14 +2636,12 @@ fn group_scanned( Ok((per_file, batches)) } -/// Insert the rows of every hook started before the scan, returning the -/// tables streamed. +/// Insert the rows of every hook started before the scan. fn stream_early( db: &Db, early: HashMap, strict_map: &HashMap, -) -> Result> { - let mut streamed = HashSet::new(); +) -> Result<()> { for (table_name, production) in early { let strict = *strict_map.get(&table_name).unwrap_or(&false); if let Err(message) = @@ -2609,9 +2652,30 @@ fn stream_early( message, }); } - streamed.insert(table_name); } - Ok(streamed) + Ok(()) +} + +/// Split off the batches of tables whose hooks already ran, keeping only +/// each one's relative paths in scan order. +#[allow(clippy::type_complexity)] +fn split_streamed( + batches: Vec<(String, Arc, Vec)>, + streamed: &HashSet, +) -> ( + Vec<(String, Arc, Vec)>, + Vec<(String, Vec)>, +) { + let mut rest = Vec::new(); + let mut done = Vec::new(); + for (table_name, hook, mut files) in batches { + if streamed.contains(&table_name) { + done.push((table_name, batch_paths(&mut files, false).0)); + } else { + rest.push((table_name, hook, files)); + } + } + (rest, done) } fn batch_paths(files: &mut [ScannedFile], keep_rel: bool) -> (Vec, Vec) { @@ -3154,7 +3218,7 @@ mod internal_tests { 0, "zero tables can match nothing, so the startup walk is pure waste" ); - assert!(prepared.scanned_files.is_empty()); + assert!(prepared.scanned_files.into_files().is_empty()); } /// The skip is narrow: one declared table still walks, and prepare still @@ -3175,10 +3239,11 @@ mod internal_tests { .unwrap(); assert_eq!(scan.calls.get(), 1); - let found: Vec<&str> = prepared + let found: Vec = prepared .scanned_files - .iter() - .map(|f| f.rel_path.as_str()) + .into_files() + .into_iter() + .map(|f| f.rel_path) .collect(); assert_eq!(found, vec!["a.txt"]); } @@ -3295,13 +3360,13 @@ mod internal_tests { functions: Vec::new(), groups: Vec::new(), anchors: TableAnchors::at_root(dir.path()), - scanned_files: vec![ScannedFile { + scanned_files: Scanned::Files(vec![ScannedFile { abs_path: PathBuf::new(), rel_path: "ghost.txt".into(), table_name: "ghost".into(), stat: None, trusted: false, - }], + }]), early: HashMap::new(), poll_interval: DEFAULT_POLL_INTERVAL, persist: None, @@ -5367,13 +5432,15 @@ mod internal_tests { scanned_files: Vec, persist: Option, ) -> PreparedBuild { - let scanned_files = scanned_files - .into_iter() - .map(|mut file| { - file.abs_path = root.join(&file.rel_path); - file - }) - .collect(); + let scanned_files = Scanned::Files( + scanned_files + .into_iter() + .map(|mut file| { + file.abs_path = root.join(&file.rel_path); + file + }) + .collect(), + ); PreparedBuild { ignore: Vec::new(), root: root.to_path_buf(), From 0a2890d6a76b9fc650854796bc4375220fbbd792 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Fri, 16 Oct 2026 09:47:53 -0400 Subject: [PATCH 2/5] e2e attestation for bot/1534-fm-native-1016 --- .../python/e2e-attestations/bot-1534-fm-native-1016.json | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 packages/python/e2e-attestations/bot-1534-fm-native-1016.json diff --git a/packages/python/e2e-attestations/bot-1534-fm-native-1016.json b/packages/python/e2e-attestations/bot-1534-fm-native-1016.json new file mode 100644 index 000000000..0e92621b2 --- /dev/null +++ b/packages/python/e2e-attestations/bot-1534-fm-native-1016.json @@ -0,0 +1,7 @@ +{ + "command": "uv run python -m pytest tests/e2e/frontmatter_parse_speed_test.py tests/e2e/jsonl_parse_speed_test.py tests/e2e/on_file_query_test.py tests/e2e/fanout_test.py tests/e2e/declared_table_name_test.py tests/e2e/query_contract_test.py tests/e2e/use_case_notebook_test.py tests/e2e/use_case_transcripts_test.py tests/e2e/sigint_interrupt_test.py -q -s", + "ran_at": 1792158472, + "exit_code": 0, + "commit": "4eaa7f0c26fa324fdb9b161e90fdb3fe22b8774f", + "branch": "bot/1534-fm-native-1016" +} From 55d3e7cd1bfc6882caad2f841d6c5206ffc48b46 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Fri, 16 Oct 2026 09:49:31 -0400 Subject: [PATCH 3/5] e2e attestation for bot/1534-fm-native-1016 --- packages/ts/e2e-attestations/bot-1534-fm-native-1016.json | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 packages/ts/e2e-attestations/bot-1534-fm-native-1016.json diff --git a/packages/ts/e2e-attestations/bot-1534-fm-native-1016.json b/packages/ts/e2e-attestations/bot-1534-fm-native-1016.json new file mode 100644 index 000000000..51a2a0fb2 --- /dev/null +++ b/packages/ts/e2e-attestations/bot-1534-fm-native-1016.json @@ -0,0 +1,7 @@ +{ + "command": "pnpm test:e2e", + "ran_at": 1792158571, + "exit_code": 0, + "commit": "0a2890d6a76b9fc650854796bc4375220fbbd792", + "branch": "bot/1534-fm-native-1016" +} From e35b7c8d31005eb9e632fa53c12726297d9efd4f Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Fri, 16 Oct 2026 09:54:04 -0400 Subject: [PATCH 4/5] fix: report early tables' ingest progress and quote the table name read for columns --- packages/rust/src/db.rs | 16 +++++++++++++--- packages/rust/src/lib.rs | 1 + 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/packages/rust/src/db.rs b/packages/rust/src/db.rs index 8a63f3311..7215dbd08 100644 --- a/packages/rust/src/db.rs +++ b/packages/rust/src/db.rs @@ -623,9 +623,10 @@ 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({})", table))?; + let mut stmt = self.conn.prepare(&format!( + "PRAGMA table_info({})", + crate::sql_literal::quote_identifier(table) + ))?; let columns: Vec = stmt .query_map([], |row| row.get::<_, String>(1))? .filter_map(|r| r.ok()) @@ -2486,6 +2487,15 @@ mod tests { assert!(result.is_err()); } + #[test] + fn get_table_columns_reads_a_table_named_by_a_keyword() { + let db = Db::new().unwrap(); + db.create_table("order", "CREATE TABLE \"order\" (name TEXT)") + .unwrap(); + let cols = db.get_table_columns("order").unwrap(); + assert!(cols.contains(&"name".to_string()), "{cols:?}"); + } + #[test] fn get_table_columns_nonexistent_table_returns_empty() { let db = Db::new().unwrap(); diff --git a/packages/rust/src/lib.rs b/packages/rust/src/lib.rs index 1a79b9ee4..0253910a8 100644 --- a/packages/rust/src/lib.rs +++ b/packages/rust/src/lib.rs @@ -1502,6 +1502,7 @@ impl DirSQL { as u64; let mut done: u64 = 0; for (table_name, rel_paths) in streamed { + progress.update(done, Some(total_files)); done += rel_paths.len() as u64; batch_files.insert(table_name, rel_paths); } From 217e3280f6dcb131402f398404a3ddeb7d6357e4 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Fri, 16 Oct 2026 10:19:42 -0400 Subject: [PATCH 5/5] test: an early hook's files count toward the ingest total --- packages/rust/src/lib.rs | 42 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/packages/rust/src/lib.rs b/packages/rust/src/lib.rs index 0253910a8..8b0f3db9e 100644 --- a/packages/rust/src/lib.rs +++ b/packages/rust/src/lib.rs @@ -5778,6 +5778,48 @@ mod internal_tests { } } + #[test] + fn finish_build_counts_the_files_of_a_hook_started_before_the_scan_toward_the_ingest_total() { + let dir = TempDir::new().unwrap(); + let out = Captured::default(); + let progress = Progress::new( + "indexing", + "indexed", + "files", + Box::new(out.clone()), + Box::new(progress::SystemClock), + progress::Mode::Always, + false, + ); + let notes = Table::new("notes", "CREATE TABLE notes (name TEXT)", "*.md", |_| { + Vec::new() + }); + let (items, _) = recording_per_table("items", "*.txt"); + let (feeds, early) = start_per_table_hooks(std::slice::from_ref(&items)); + for name in ["a.txt", "b.txt"] { + feeds["items"].send(dir.path().join(name)).unwrap(); + } + drop(feeds); + let scanned = vec![ + scanned("a.txt", "items", false), + scanned("b.txt", "items", false), + scanned("n.md", "notes", false), + ]; + let mut prepared = prepared_build(dir.path(), vec![notes, items], scanned, None); + prepared.early = early; + + DirSQL::finish_build_with_fs(prepared, Arc::new(FakeFs::default()), progress).unwrap(); + + let text = out.text(); + for line in [ + "dirsql: indexing 0/3 files (0%)", + "dirsql: indexing 2/3 files (66%)", + "dirsql: indexed 3 files in", + ] { + assert!(text.contains(line), "missing {line:?} in {text:?}"); + } + } + #[test] fn finish_build_skips_a_per_file_hook_for_a_file_the_cache_holds() { let dir = TempDir::new().unwrap();