diff --git a/docs/reference/config.md b/docs/reference/config.md index 71589f020..0605b047e 100644 --- a/docs/reference/config.md +++ b/docs/reference/config.md @@ -236,6 +236,11 @@ rows — filtered to the DDL — are the table. For plain stat columns with no command, query the path directly with a [path-table](./path-tables.md) instead of declaring a table. +Tables' `on-file` commands run one table at a time, in declaration order: +a table's command finishes before the next table's starts, whatever order +their files sort in. That holds at startup and when a change under watch +re-runs several tables at once. + ```toml [[table]] name = "comments" @@ -369,6 +374,8 @@ The configs load and merge in **argv order**: that config file's own directory** — so a relative command like `on-file = "sh ./extract.sh"` resolves against the config that declared it, wherever it lives. +- **`on-file` commands run in argv order**: every table of the first config, + in declaration order, before any table of the second. - Each config is **validated on its own** (the [parse errors](#parse-errors) below apply per file). There is no cross-file merge validation, with two structural exceptions: **two configs defining a table of the same name is diff --git a/packages/python/e2e-attestations/bot-1538-onfile-order-1019.json b/packages/python/e2e-attestations/bot-1538-onfile-order-1019.json new file mode 100644 index 000000000..c73fbecc0 --- /dev/null +++ b/packages/python/e2e-attestations/bot-1538-onfile-order-1019.json @@ -0,0 +1,7 @@ +{ + "command": "uv run python -m pytest tests/e2e/on_file_query_test.py tests/e2e/declared_table_name_test.py tests/e2e/fanout_test.py tests/e2e/query_contract_test.py tests/e2e/sigint_interrupt_test.py tests/e2e/use_case_notebook_test.py tests/e2e/use_case_transcripts_test.py tests/e2e/plugin_discovery_test.py tests/e2e/frontmatter_parse_speed_test.py tests/e2e/jsonl_parse_speed_test.py tests/e2e/watch_speed_test.py tests/e2e/server_speed_test.py -q -p no:randomly", + "ran_at": 1792408377, + "exit_code": 0, + "commit": "33defae577621505df735a61e99e1286a3dc96e1", + "branch": "bot/1538-onfile-order-1019" +} diff --git a/packages/rust/changelog.d/2026-10-19-on-file-declaration-order.md b/packages/rust/changelog.d/2026-10-19-on-file-declaration-order.md new file mode 100644 index 000000000..ff8fb2294 --- /dev/null +++ b/packages/rust/changelog.d/2026-10-19-on-file-declaration-order.md @@ -0,0 +1,3 @@ +**Changed** + +- **Config tables' `on-file` commands run in declaration order.** One table's command finishes before the next table's starts, at startup and when a watched change re-runs several tables; across repeated `-c`, the configs' argv order. Commands of different tables previously overlapped at startup, or ran in the order their files sorted. diff --git a/packages/rust/migrations.d/2026-10-19-on-file-declaration-order.md b/packages/rust/migrations.d/2026-10-19-on-file-declaration-order.md new file mode 100644 index 000000000..ed4e5211c --- /dev/null +++ b/packages/rust/migrations.d/2026-10-19-on-file-declaration-order.md @@ -0,0 +1,25 @@ +### `on-file` declaration order + +#### Summary + +Config tables' `on-file` commands run one table at a time, in declaration order (argv order across repeated `-c`). Nothing breaks. + +#### Required changes + +_None._ + +#### Deprecations removed + +_None._ + +#### Behavior changes without code changes + +Two tables' commands no longer run at the same time; a table's command starts only after the table declared before it has finished. + +#### Verification + +```bash +dirsql query "SELECT 1" -c .dirsql.toml +``` + +Each table's command starts after the previous table's has exited. diff --git a/packages/rust/src/lib.rs b/packages/rust/src/lib.rs index 8b0f3db9e..b8e636798 100644 --- a/packages/rust/src/lib.rs +++ b/packages/rust/src/lib.rs @@ -418,6 +418,8 @@ struct DirSqlInner { groups: Vec, on_file_map: HashMap, strict_map: HashMap, + /// Every table's name in declaration order: per-table hooks run in it. + declared: Vec, /// The root-relative paths each per-table hook was last run over. Rows /// are recorded under [`BATCH_OWNER`], not their files, so this is how a /// deleted directory is traced back to the tables it fed. @@ -887,8 +889,10 @@ impl DirSQL { /// Re-run every per-table hook `pending` names, once each, over the files /// its glob matches now, and replace that table's rows. fn refresh_pending(&self, pending: PendingRefresh) -> Vec { + let mut pending = pending.0; + pending.sort_by_key(|(table, _)| declared_position(&self.inner.declared, table)); let mut events = Vec::new(); - for (table, trigger) in pending.0 { + for (table, trigger) in pending { events.extend(self.refresh_table(&table, &trigger)); } events @@ -1263,7 +1267,7 @@ impl DirSQL { let (feeds, early) = if persist_ctx.is_none() { start_per_table_hooks(&tables) } else { - (HashMap::new(), HashMap::new()) + (HashMap::new(), Vec::new()) }; let scanned = if table_names.is_empty() { Vec::new() @@ -1417,6 +1421,7 @@ impl DirSQL { let mut on_file_map: HashMap = HashMap::new(); let mut strict_map: HashMap = HashMap::new(); let mut ddl_map: HashMap = HashMap::new(); + let declared: Vec = tables.iter().map(|table| table.name.clone()).collect(); for table in tables { let table_name = table.name.clone(); @@ -1475,7 +1480,7 @@ impl DirSQL { // Grouping walks every scanned file, so it runs beside the early // 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 early_tables: HashSet = early.iter().map(|(name, _)| name.clone()).collect(); let (grouped, streamed) = std::thread::scope(|scope| { let grouping = scope.spawn(|| { group_scanned(scanned_files.into_files(), &on_file_map) @@ -1488,7 +1493,8 @@ impl DirSQL { ) }); streamed?; - let (per_file, (batches, streamed)) = grouped?; + let (per_file, (mut batches, streamed)) = grouped?; + batches.sort_by_key(|(table_name, _, _)| declared_position(&declared, table_name)); // 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. @@ -1585,7 +1591,7 @@ impl DirSQL { }); if !current { let table_root = anchors.of(&table_name); - let production = start_hook(hook, abs_paths.into_iter()); + let production = start_hook(hook, abs_paths.into_iter(), Turn::default()); let outcome = stream_table_rows(&db, &table_name, production, strict) .map_err(map_db_error)?; match outcome { @@ -1653,6 +1659,7 @@ impl DirSQL { groups, on_file_map, strict_map, + declared, batch_files: Mutex::new(batch_files), scan_failures: on_file_failures, watcher: Mutex::new(None), @@ -2052,7 +2059,7 @@ pub struct PreparedBuild { ignore: Vec, scanned_files: Scanned, /// Per-table hooks already running over the paths the scan found, by table. - early: HashMap, + early: Vec<(String, Production)>, persist: Option, poll_interval: Duration, hint_legacy_files_table: bool, @@ -2513,12 +2520,25 @@ struct Production { handle: std::thread::JoinHandle>, } +/// A per-table hook's place in declaration order: it waits for the hook +/// before it to finish, and releases the next by dropping `done` as it ends. +#[derive(Default)] +struct Turn { + after: Option>, + done: Option>, +} + fn start_hook( hook: Arc, paths: impl Iterator + Send + 'static, + turn: Turn, ) -> Production { let (sender, chunks) = std::sync::mpsc::channel(); let handle = std::thread::spawn(move || { + let Turn { after, done: _done } = turn; + if let Some(after) = after { + let _ = after.recv(); + } let mut paths = paths.peekable(); if paths.peek().is_none() { return Ok(()); @@ -2531,24 +2551,32 @@ fn start_hook( Production { chunks, handle } } -/// Start every per-table hook before the scan, so each runs over its paths as -/// the scan finds them: the senders feed the hooks, and closing them ends the -/// paths. +/// Start every per-table hook before the scan, so the first runs over its +/// paths as the scan finds them and each later one runs as soon as the one +/// declared before it finishes: the senders feed the hooks, and closing them +/// ends the paths. +#[allow(clippy::type_complexity)] fn start_per_table_hooks( tables: &[Table], ) -> ( HashMap>, - HashMap, + Vec<(String, Production)>, ) { let mut feeds = HashMap::new(); - let mut running = HashMap::new(); + let mut running = Vec::new(); + let mut after = None; for table in tables { if let Hook::PerTable(hook) = &table.hook { let (feed, paths) = std::sync::mpsc::channel(); - running.insert( + let (done, next) = std::sync::mpsc::channel(); + let turn = Turn { + after: after.replace(next), + done: Some(done), + }; + running.push(( table.name.clone(), - start_hook(Arc::clone(hook), paths.into_iter()), - ); + start_hook(Arc::clone(hook), paths.into_iter(), turn), + )); feeds.insert(table.name.clone(), feed); } } @@ -2637,10 +2665,11 @@ fn group_scanned( Ok((per_file, batches)) } -/// Insert the rows of every hook started before the scan. +/// Insert the rows of every hook started before the scan, in the order the +/// hooks run. fn stream_early( db: &Db, - early: HashMap, + early: Vec<(String, Production)>, strict_map: &HashMap, ) -> Result<()> { for (table_name, production) in early { @@ -2657,6 +2686,10 @@ fn stream_early( Ok(()) } +fn declared_position(declared: &[String], table: &str) -> Option { + declared.iter().position(|name| name == table) +} + /// Split off the batches of tables whose hooks already ran, keeping only /// each one's relative paths in scan order. #[allow(clippy::type_complexity)] @@ -3368,7 +3401,7 @@ mod internal_tests { stat: None, trusted: false, }]), - early: HashMap::new(), + early: Vec::new(), poll_interval: DEFAULT_POLL_INTERVAL, persist: None, hint_legacy_files_table: false, @@ -5451,7 +5484,7 @@ mod internal_tests { groups: Vec::new(), anchors: TableAnchors::at_root(root), scanned_files, - early: HashMap::new(), + early: Vec::new(), poll_interval: DEFAULT_POLL_INTERVAL, persist, hint_legacy_files_table: false, @@ -5579,6 +5612,14 @@ mod internal_tests { .unwrap() } + #[test] + fn declared_position_is_a_tables_index_in_declaration_order() { + let declared = ["zeta".to_string(), "alpha".to_string()]; + assert_eq!(declared_position(&declared, "zeta"), Some(0)); + assert_eq!(declared_position(&declared, "alpha"), Some(1)); + assert_eq!(declared_position(&declared, "beta"), None); + } + #[test] fn finish_build_runs_a_per_table_hook_once_over_its_files_in_scan_order() { let dir = TempDir::new().unwrap(); diff --git a/packages/rust/tests/on_file_order.rs b/packages/rust/tests/on_file_order.rs new file mode 100644 index 000000000..b9ca84a33 --- /dev/null +++ b/packages/rust/tests/on_file_order.rs @@ -0,0 +1,127 @@ +//! Config tables' `on-file` commands run in declaration order: one table's +//! command finishes before the next table's starts, whatever order the walk +//! finds their files in. Across repeated configs, the order is the configs'. +//! +//! Unix-only: the fixtures shell out to `sh`. +#![cfg(unix)] + +use std::fs; +use std::path::{Path, PathBuf}; + +use dirsql::DirSQL; +use dirsql::watcher::FileEvent; +use tempfile::TempDir; + +/// Each run logs its start, lingers long enough for an overlapping run to +/// start too, then logs its end. +const RECORD: &str = "#!/bin/sh\nname=$1\nprintf 'start %s\\n' \"$name\" >> \"$LOG\"\nsleep 0.3\nprintf 'end %s\\n' \"$name\" >> \"$LOG\"\n"; + +/// A config whose tables are declared in `names` order, each over +/// `data//*.txt`, logging to `log`. +fn write_config(dir: &Path, names: &[&str], log: &Path) -> PathBuf { + fs::write(dir.join("record.sh"), RECORD).unwrap(); + let mut toml = String::new(); + for name in names { + let data = dir.join("data").join(name); + fs::create_dir_all(&data).unwrap(); + fs::write(data.join("one.txt"), "x\n").unwrap(); + toml.push_str(&format!( + "[[table]]\nname = \"{name}\"\nddl = \"CREATE TABLE {name} (v TEXT)\"\nglob = \"data/{name}/*.txt\"\non-file = \"env LOG='{}' sh record.sh {name}\"\n\n", + log.display() + )); + } + let path = dir.join(".dirsql.toml"); + fs::write(&path, toml).unwrap(); + path +} + +fn runs(log: &Path) -> Vec { + fs::read_to_string(log) + .unwrap_or_default() + .lines() + .map(str::to_string) + .collect() +} + +fn in_order(names: &[&str]) -> Vec { + names + .iter() + .flat_map(|name| [format!("start {name}"), format!("end {name}")]) + .collect() +} + +#[test] +fn startup_runs_commands_in_declaration_order_not_walk_order() { + let dir = TempDir::new().unwrap(); + let log = dir.path().join("order.log"); + let config = write_config(dir.path(), &["zeta", "alpha"], &log); + + let db = DirSQL::builder() + .root(dir.path()) + .config(&config) + .build() + .unwrap(); + assert!(db.scan_failures().is_empty(), "{:?}", db.scan_failures()); + + assert_eq!(runs(&log), in_order(&["zeta", "alpha"])); +} + +#[test] +fn startup_with_a_cache_runs_commands_in_declaration_order() { + let dir = TempDir::new().unwrap(); + let log = dir.path().join("order.log"); + let config = write_config(dir.path(), &["zeta", "alpha"], &log); + let cache = TempDir::new().unwrap(); + + DirSQL::builder() + .root(dir.path()) + .config(&config) + .persist(Some(cache.path().join("cache.db"))) + .build() + .unwrap(); + + assert_eq!(runs(&log), in_order(&["zeta", "alpha"])); +} + +#[test] +fn repeated_configs_run_commands_in_argv_order() { + let first = TempDir::new().unwrap(); + let second = TempDir::new().unwrap(); + let log_dir = TempDir::new().unwrap(); + let log = log_dir.path().join("order.log"); + let first_config = write_config(first.path(), &["zeta"], &log); + let second_config = write_config(second.path(), &["alpha"], &log); + + DirSQL::builder() + .root(first.path()) + .config(&first_config) + .config(&second_config) + .build() + .unwrap(); + + assert_eq!(runs(&log), in_order(&["zeta", "alpha"])); +} + +#[test] +fn a_watch_refresh_over_several_tables_runs_commands_in_declaration_order() { + let dir = TempDir::new().unwrap(); + let log = dir.path().join("order.log"); + let config = write_config(dir.path(), &["zeta", "alpha"], &log); + let db = DirSQL::builder() + .root(dir.path()) + .config(&config) + .build() + .unwrap(); + fs::write(&log, "").unwrap(); + + let alpha_file = dir.path().join("data/alpha/one.txt"); + let zeta_file = dir.path().join("data/zeta/one.txt"); + fs::write(&alpha_file, "y\n").unwrap(); + fs::write(&zeta_file, "y\n").unwrap(); + db.apply_file_events(vec![ + FileEvent::Modified(alpha_file), + FileEvent::Modified(zeta_file), + ]); + + assert_eq!(runs(&log), in_order(&["zeta", "alpha"])); +} diff --git a/packages/ts/e2e-attestations/bot-1538-onfile-order-1019.json b/packages/ts/e2e-attestations/bot-1538-onfile-order-1019.json new file mode 100644 index 000000000..be3efdcd5 --- /dev/null +++ b/packages/ts/e2e-attestations/bot-1538-onfile-order-1019.json @@ -0,0 +1,7 @@ +{ + "command": "pnpm test:e2e", + "ran_at": 1792408488, + "exit_code": 0, + "commit": "b0c1059f4dfcd0651d83f2a73298c98007eb2e61", + "branch": "bot/1538-onfile-order-1019" +}