fix: run config tables' on-file commands in declaration order

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

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"
+}

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.

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.

packages/rust/src/lib.rs

@@ -418,6 +418,8 @@ struct DirSqlInner {
     groups: Vec<AnchorGroup>,
     on_file_map: HashMap<String, Hook>,
     strict_map: HashMap<String, bool>,
+    /// Every table's name in declaration order: per-table hooks run in it.
+    declared: Vec<String>,
     /// 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<RowEvent> {
+        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<String, Hook> = HashMap::new();
         let mut strict_map: HashMap<String, bool> = HashMap::new();
         let mut ddl_map: HashMap<String, String> = HashMap::new();
+        let declared: Vec<String> = 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<String> = early.keys().cloned().collect();
+        let early_tables: HashSet<String> = 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<String>,
     scanned_files: Scanned,
     /// Per-table hooks already running over the paths the scan found, by table.
-    early: HashMap<String, Production>,
+    early: Vec<(String, Production)>,
     persist: Option<PreparedPersist>,
     poll_interval: Duration,
     hint_legacy_files_table: bool,
@@ -2513,12 +2520,25 @@ struct Production {
     handle: std::thread::JoinHandle<std::result::Result<(), String>>,
 }
 
+/// 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<std::sync::mpsc::Receiver<()>>,
+    done: Option<std::sync::mpsc::Sender<()>>,
+}
+
 fn start_hook(
     hook: Arc<OnFilesFn>,
     paths: impl Iterator<Item = PathBuf> + 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<String, std::sync::mpsc::Sender<PathBuf>>,
-    HashMap<String, Production>,
+    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<String, Production>,
+    early: Vec<(String, Production)>,
     strict_map: &HashMap<String, bool>,
 ) -> Result<()> {
     for (table_name, production) in early {
@@ -2657,6 +2686,10 @@ fn stream_early(
     Ok(())
 }
 
+fn declared_position(declared: &[String], table: &str) -> Option<usize> {
+    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();

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/<name>/*.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<String> {
+    fs::read_to_string(log)
+        .unwrap_or_default()
+        .lines()
+        .map(str::to_string)
+        .collect()
+}
+
+fn in_order(names: &[&str]) -> Vec<String> {
+    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"]));
+}