perf(rust): read a path table's content ahead, in parallel, when a statement names it

PARITY.md

@@ -305,6 +305,14 @@ names come from the SQLite authorizer at the final prepare, so the fallback
 loop above stays the only thing that registers a table. `StatementScope::warm`
 lives in the shared core; no SDK signature changes.
 
+**Content read ahead in parallel (#1241) — Rust-crate-only surface, no SDK
+drift.** A statement that names a path table's `content` has the column read
+for every row it selects at once, eight files at a time, before SQLite steps
+the first row; a statement that does not name it reads no file, as before. The
+read-ahead and the planner cost that keeps a content-bearing table on the
+scanned side of a join live in the shared core, so all three SDKs and the CLI
+inherit it with no per-binding surface and no signature changes.
+
 ## AsyncDirSQL
 
 | API                        | Python                                | Rust                                   |

docs/reference/path-tables.md

@@ -180,12 +180,14 @@ A path-table is scanned when the statement runs, so it always reflects the
 filesystem as it is *now* — unlike declared tables, which are indexed on build
 and updated by the watcher. A file created a moment ago shows up immediately.
 
-The scan is live all the way down to `content`: a file's body is read when the
-query names the `content` column, not when the row is discovered. A file
-deleted *after* the scan finds it but *before* its `content` is read yields
-`NULL` content — the same NULL an unreadable or non-UTF-8 file gives — rather
-than failing the query. This is an accepted consequence of reading live, not a
-bug to design around.
+The scan is live all the way down to `content`: a file's body is read only
+when the query names the `content` column, not when the row is discovered. A
+query that names it has the bodies of the rows it selects read all at once,
+several files at a time, before the first row comes back. A file deleted
+*after* the scan finds it but *before* its `content` is read yields `NULL`
+content — the same NULL an unreadable or non-UTF-8 file gives — rather than
+failing the query. This is an accepted consequence of reading live, not a bug
+to design around.
 
 The table itself is per-connection: it lives in `temp`, so it cannot leak into
 `sqlite_master` or survive a restart. Under `--persist` a *parsed* table's rows

packages/python/e2e-attestations/claude-1241-content-prefetch.json

@@ -0,0 +1,7 @@
+{
+  "command": "just test-e2e",
+  "ran_at": 1791162240,
+  "exit_code": 0,
+  "commit": "4c645eb195b0c6a04d18c2f1064157adb275c8a5",
+  "branch": "claude/1241-content-prefetch"
+}

packages/rust/changelog.d/2026-10-02-path-table-content-prefetch.md

@@ -0,0 +1 @@
+**Changed** A statement that names a path table's `content` column -- in its projection, its `WHERE`, or across a join -- has the file bodies of the rows it selects read ahead, eight files at a time, before SQLite steps the first row, instead of one file at a time as each row is emitted. Each read is also one stat cheaper: the buffer is sized from the size the scan already recorded. A statement that does not name `content` still reads no file, `SELECT *` still excludes it, and an unreadable or non-UTF-8 file is still a `NULL` cell. In a join, the planner now scans the table whose `content` is named and probes the other, so the read-ahead covers the whole column rather than one probe at a time.

packages/rust/migrations.d/2026-10-02-path-table-content-prefetch.md

@@ -0,0 +1,37 @@
+### Core: a path table's `content` is read ahead when a statement names it
+
+#### Summary
+
+No API changes. A statement that names `content` has the bodies of the rows it
+selects read together, several files at a time, before its first row comes
+back; one that does not name it reads no file, as before.
+
+#### Required changes
+
+_None._
+
+#### Deprecations removed
+
+_None._
+
+#### Behavior changes without code changes
+
+- A file deleted after a statement's scan found it is still readable to that
+  statement if the read-ahead reached it first; it used to yield `NULL`
+  whenever the deletion came before the row was emitted. Either way the next
+  statement sees the file gone.
+- A statement that names `content` holds the text of every row it selects
+  for its own duration, where before it held one row's text at a time.
+- In a join, the planner scans the path table whose `content` is named and
+  probes the other, where before it probed whichever came second.
+
+#### Verification
+
+```bash
+cd "$(mktemp -d)"
+mkdir -p p1 p2
+printf 'one\n' > p1/title.md
+printf 'two\n' > p2/title.md
+dirsql query "SELECT content FROM './*/title.md' ORDER BY path"
+# expected: [{"content":"one\n"},{"content":"two\n"}] -- both bodies, read together
+```

packages/rust/src/vtab.rs

@@ -1,7 +1,9 @@
 use std::ffi::c_int;
-use std::fs;
+use std::fs::{self, File};
+use std::io::{self, Read};
 use std::path::{Path, PathBuf};
-use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, OnceLock};
 use std::thread;
 use std::time::{SystemTime, UNIX_EPOCH};
 
@@ -65,13 +67,71 @@ fn column_type(column: &str) -> &'static str {
     }
 }
 
+/// Files read at once when `content` is read ahead. Each read is a few
+/// syscalls the kernel answers independently of the last, so they overlap
+/// almost perfectly; past eight the box, not the count, sets the pace.
+const READERS: usize = 8;
+
 /// Read `path` as text, yielding `None` when it is unreadable or not valid
 /// UTF-8. A file that cannot be read is a NULL cell, never a failed row: the
 /// filesystem is allowed to be messy and a query over it should still return.
-fn read_text(path: &Path) -> Option<String> {
-    std::fs::read(path)
-        .ok()
-        .and_then(|b| String::from_utf8(b).ok())
+/// `size` is what the scan's stat reported, so the buffer is sized without
+/// stat'ing again; a file that has grown since is still read whole.
+fn read_text(path: &Path, size: Option<i64>) -> Option<String> {
+    let mut file = File::open(path).ok()?;
+    let bytes = read_all(&mut file, presize(size)).ok()?;
+    String::from_utf8(bytes).ok()
+}
+
+/// One past the size the scan saw: a file unchanged since fills the buffer in
+/// one read and reports its end on the next, with no stat in between.
+fn presize(size: Option<i64>) -> usize {
+    usize::try_from(size.unwrap_or(0))
+        .unwrap_or(0)
+        .saturating_add(1)
+}
+
+/// Every byte `reader` yields, into a buffer of `expected` bytes that grows
+/// if the reader has more.
+fn read_all(reader: &mut impl Read, expected: usize) -> io::Result<Vec<u8>> {
+    let mut buf = Vec::new();
+    buf.try_reserve_exact(expected.max(1))?;
+    buf.resize(expected.max(1), 0);
+    let mut filled = 0;
+    loop {
+        if filled == buf.len() {
+            buf.resize(buf.len() * 2, 0);
+        }
+        match reader.read(&mut buf[filled..]) {
+            Ok(0) => break,
+            Ok(n) => filled += n,
+            Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
+            Err(e) => return Err(e),
+        }
+    }
+    buf.truncate(filled);
+    Ok(buf)
+}
+
+/// Read `content` for every row in `rows` that has none yet, [`READERS`]
+/// files at a time, each reader taking the next unread row. `read` is
+/// injected so the sharing-out is testable without a filesystem.
+fn read_contents(rows: &[&FileRow], read: &(dyn Fn(&FileRow) -> Option<String> + Sync)) {
+    let unread: Vec<&FileRow> = rows
+        .iter()
+        .copied()
+        .filter(|row| row.content.get().is_none())
+        .collect();
+    let next = AtomicUsize::new(0);
+    thread::scope(|scope| {
+        for _ in 0..READERS.min(unread.len()) {
+            scope.spawn(|| {
+                while let Some(row) = unread.get(next.fetch_add(1, Ordering::Relaxed)) {
+                    row.content.get_or_init(|| read(row));
+                }
+            });
+        }
+    });
 }
 
 /// Everything a scan needs: where to walk, what to match, what to call the
@@ -97,6 +157,10 @@ struct FileRow {
     path: String,
     spans: PathSpans,
     facts: StatFacts,
+    /// The file's text once a statement has asked for it, read at most once
+    /// per statement: ahead for every row when the statement names the
+    /// column, or on demand for the row being emitted.
+    content: OnceLock<Option<String>>,
 }
 
 impl FileRow {
@@ -108,6 +172,7 @@ impl FileRow {
             path,
             spans,
             facts,
+            content: OnceLock::new(),
         }
     }
 
@@ -217,8 +282,8 @@ fn integer(n: Option<i64>) -> ValueRef<'static> {
     n.map_or(ValueRef::Null, ValueRef::Integer)
 }
 
-/// The cell a row holds for column `i`, `None` for the one it does not hold:
-/// `content` is read from the file when asked for, never stored.
+/// The cell a row holds for column `i`, `None` for the one it does not hold
+/// until asked: `content` is read from the file, not stat'ed with the rest.
 fn cell(row: &FileRow, i: c_int) -> Option<ValueRef<'_>> {
     Some(match usize::try_from(i).unwrap_or(usize::MAX) {
         PATH_COLUMN => ValueRef::from(row.path.as_str()),
@@ -258,6 +323,12 @@ fn parse_module_args(args: &[&[u8]]) -> Result<ScanSpec> {
     })
 }
 
+impl ScanSpec {
+    fn read(&self, row: &FileRow) -> Option<String> {
+        read_text(&self.root.join(&row.rel), row.facts.size)
+    }
+}
+
 /// The string a matched file is reported under: the relative path as scanned,
 /// under the table's path prefix when it has one.
 fn reported_path(path_prefix: &Path, rel_path: &Path) -> String {
@@ -301,10 +372,16 @@ impl TableSource for ScanSpec {
             Some(value) => ctx.set_result(&ToSqlOutput::Borrowed(value)),
             // The one effectful read, reached only when a query names the
             // column: this is where laziness actually lives.
-            None => ctx.set_result(&read_text(&self.root.join(&row.rel))),
+            None => ctx.set_result(row.content.get_or_init(|| self.read(row))),
         }
     }
 
+    const PREFETCH_COLUMN: Option<usize> = Some(CONTENT_COLUMN);
+
+    fn prefetch(&self, rows: &[&FileRow]) {
+        read_contents(rows, &|row| self.read(row));
+    }
+
     fn lookup_key<'r>(&self, row: &'r FileRow, column: usize) -> Option<&'r str> {
         match column {
             PATH_COLUMN => Some(&row.path),
@@ -629,6 +706,164 @@ mod tests {
         assert!(rows.is_empty());
     }
 
+    #[test]
+    fn presize_is_one_past_the_size_the_scan_saw() {
+        assert_eq!(presize(Some(0)), 1);
+        assert_eq!(presize(Some(41)), 42);
+    }
+
+    #[test]
+    fn presize_is_one_byte_without_a_usable_size() {
+        assert_eq!(presize(None), 1);
+        assert_eq!(presize(Some(-1)), 1);
+    }
+
+    /// A reader that yields `chunks` one call at a time, `interrupted` times
+    /// interrupted first.
+    struct Chunked {
+        chunks: Vec<Vec<u8>>,
+        interrupted: usize,
+    }
+
+    impl Read for Chunked {
+        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+            if self.interrupted > 0 {
+                self.interrupted -= 1;
+                return Err(io::Error::from(io::ErrorKind::Interrupted));
+            }
+            if self.chunks.is_empty() {
+                return Ok(0);
+            }
+            let chunk = self.chunks.remove(0);
+            let n = chunk.len().min(buf.len());
+            buf[..n].copy_from_slice(&chunk[..n]);
+            if n < chunk.len() {
+                self.chunks.insert(0, chunk[n..].to_vec());
+            }
+            Ok(n)
+        }
+    }
+
+    #[test]
+    fn read_all_takes_every_byte_a_presized_read_yields() {
+        let mut reader = Chunked {
+            chunks: vec![b"hello".to_vec()],
+            interrupted: 0,
+        };
+        assert_eq!(read_all(&mut reader, 6).unwrap(), b"hello");
+    }
+
+    #[test]
+    fn read_all_grows_past_a_size_the_file_has_outgrown() {
+        let mut reader = Chunked {
+            chunks: vec![b"hello".to_vec(), b" world".to_vec()],
+            interrupted: 0,
+        };
+        assert_eq!(read_all(&mut reader, 2).unwrap(), b"hello world");
+        let mut reader = Chunked {
+            chunks: vec![b"hello".to_vec()],
+            interrupted: 0,
+        };
+        assert_eq!(read_all(&mut reader, 0).unwrap(), b"hello");
+    }
+
+    struct Counting<R>(R, usize);
+
+    impl<R: Read> Read for Counting<R> {
+        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+            self.1 += 1;
+            self.0.read(buf)
+        }
+    }
+
+    #[test]
+    fn read_all_grows_geometrically_when_a_file_has_outgrown_its_size() {
+        let chunked = Chunked {
+            chunks: vec![vec![b'x'; 4096]],
+            interrupted: 0,
+        };
+        let mut reader = Counting(chunked, 0);
+        assert_eq!(read_all(&mut reader, 1).unwrap().len(), 4096);
+        assert!(reader.1 <= 16, "{} reads", reader.1);
+    }
+
+    #[test]
+    fn read_all_retries_an_interrupted_read() {
+        let mut reader = Chunked {
+            chunks: vec![b"hi".to_vec()],
+            interrupted: 2,
+        };
+        assert_eq!(read_all(&mut reader, 3).unwrap(), b"hi");
+    }
+
+    #[test]
+    fn read_all_surfaces_a_failed_read() {
+        struct Broken;
+        impl Read for Broken {
+            fn read(&mut self, _buf: &mut [u8]) -> io::Result<usize> {
+                Err(io::Error::from(io::ErrorKind::PermissionDenied))
+            }
+        }
+        let err = read_all(&mut Broken, 4).unwrap_err();
+        assert_eq!(err.kind(), io::ErrorKind::PermissionDenied);
+    }
+
+    #[test]
+    fn read_all_yields_nothing_for_an_empty_reader() {
+        let mut reader = Chunked {
+            chunks: Vec::new(),
+            interrupted: 0,
+        };
+        assert_eq!(read_all(&mut reader, 1).unwrap(), b"");
+    }
+
+    #[test]
+    fn read_contents_reads_each_unread_row_once() {
+        let rows: Vec<FileRow> = (0..40).map(|n| row_for("", &format!("f{n}.md"))).collect();
+        rows[3].content.set(Some("already".to_string())).unwrap();
+        let reads = std::sync::Mutex::new(Vec::new());
+        let read = |row: &FileRow| {
+            reads.lock().unwrap().push(row.path.clone());
+            Some(format!("body of {}", row.path))
+        };
+
+        read_contents(&rows.iter().collect::<Vec<_>>(), &read);
+
+        let mut reads = reads.into_inner().unwrap();
+        reads.sort();
+        let mut expected: Vec<String> = (0..40)
+            .filter(|&n| n != 3)
+            .map(|n| format!("f{n}.md"))
+            .collect();
+        expected.sort();
+        assert_eq!(
+            reads, expected,
+            "every row but the one already read, once each"
+        );
+        assert_eq!(rows[3].content.get().unwrap().as_deref(), Some("already"));
+        assert_eq!(
+            rows[7].content.get().unwrap().as_deref(),
+            Some("body of f7.md")
+        );
+    }
+
+    #[test]
+    fn read_contents_keeps_a_null_read() {
+        let rows = [row_for("", "a.md")];
+        read_contents(&[&rows[0]], &|_| None);
+        assert_eq!(rows[0].content.get(), Some(&None));
+    }
+
+    #[test]
+    fn read_contents_reads_nothing_when_every_row_is_read() {
+        let rows = [row_for("", "a.md")];
+        rows[0].content.set(None).unwrap();
+        read_contents(&[&rows[0]], &|_| {
+            panic!("a row read already is not read again")
+        });
+        read_contents(&[], &|_| panic!("nothing to read"));
+    }
+
     #[test]
     fn a_nested_path_splits_into_dir_basename_and_ext() {
         let row = row_for("", "nested/sub.txt");