Speed: ./**/*.md at or under fd

packages/python/e2e-attestations/bot-1474-vtab-speed-1014.json

@@ -0,0 +1,7 @@
+{
+  "command": "uv run python -m pytest tests/e2e/ --ignore=tests/e2e/watch_speed_test.py -q",
+  "ran_at": 1792075222,
+  "exit_code": 0,
+  "commit": "111db17ede0ace14d6a6b62eea7561c8d5270277",
+  "branch": "bot/1474-vtab-speed-1014"
+}

packages/python/tests/e2e/glob_speed_recursive_test.py

@@ -15,6 +15,7 @@
 import pytest
 
 from .speed_of_light import (
+    FASTEST,
     Native,
     agreed_rows,
     assert_speed_of_light,
@@ -53,8 +54,8 @@
     *suite_only(globstar("./a/b/**/*.md")),
 ]
 CASES = [
-    ("./**/*.md", "./**/*.md", ALL_NATIVE, ""),
-    ("./a/b/**/*.md", "./a/b/**/*.md", PREFIX_NATIVE, "a/b"),
+    ("./**/*.md", "./**/*.md", ALL_NATIVE, "", FASTEST),
+    ("./a/b/**/*.md", "./a/b/**/*.md", PREFIX_NATIVE, "a/b", ""),
 ]
 FANOUT = 8
 
@@ -113,8 +114,8 @@ def tree(tmp_path_factory):
         finally:
             shutil.rmtree(root, ignore_errors=True)
 
-    @pytest.mark.parametrize(("name", "glob", "specs", "prefix"), CASES)
-    def it_matches_native_files_within_the_bar(tree, name, glob, specs, prefix):
+    @pytest.mark.parametrize(("name", "glob", "specs", "prefix", "bar"), CASES)
+    def it_matches_native_files_within_the_bar(tree, name, glob, specs, prefix, bar):
         root, n, startup = tree
 
         def dirsql(timeout):
@@ -125,7 +126,7 @@ def dirsql(timeout):
             )
             return sorted(row for (row,) in dirsql_rows(proc, ("path",))), seconds
 
-        result = paired(shell_natives(root, specs), dirsql, startup)
+        result = paired(shell_natives(root, specs, bar=bar), dirsql, startup)
         expected, actual = agreed_rows(result), result.dirsql_rows
         assert len(expected) == matched(n, prefix)
 

packages/rust/changelog.d/2026-10-15-path-rows-stream-to-the-cursor.md

@@ -0,0 +1,3 @@
+**Changed**
+
+- **A path table hands its rows to the query while the walk is still running.** `SELECT path FROM './**/*.md'` over a large tree now runs at or under `fd`. Output is unchanged.

packages/rust/migrations.d/2026-10-15-path-rows-stream-to-the-cursor.md

@@ -0,0 +1,25 @@
+### Path rows stream to the query
+
+#### Summary
+
+A path table hands rows to the query in chunks as the walk finds them, instead of after the walk ends, and passes path text to SQLite without copying it. Query results and their order are unchanged; nothing breaks.
+
+#### Required changes
+
+_None._
+
+#### Deprecations removed
+
+_None._
+
+#### Behavior changes without code changes
+
+_None._
+
+#### Verification
+
+```bash
+dirsql query "SELECT path FROM './**/*.md'"
+```
+
+Prints the same rows as before, sooner.

packages/rust/src/lib.rs

@@ -40,6 +40,7 @@ mod posix_class;
 pub mod progress;
 #[doc(hidden)]
 pub mod row_event_flat;
+mod row_stream;
 #[doc(hidden)]
 pub mod scanner;
 #[doc(hidden)]

packages/rust/src/parsed_vtab.rs

@@ -28,6 +28,7 @@ use crate::Value;
 use crate::infer::{JsonRow, cell, declared_schema, infer_schema};
 use crate::matcher::TableMatcher;
 use crate::on_file;
+use crate::row_stream::RowStream;
 use crate::scanner::{PathGlob, scan_glob_checking_root};
 use crate::vtab_scaffold::{self, StatementScope, TableSource};
 
@@ -172,10 +173,10 @@ impl TableSource for ParsedTable {
         Ok((schema, table))
     }
 
-    fn rows(&self) -> Arc<Vec<JsonRow>> {
+    fn rows(&self, rows: &Arc<RowStream<JsonRow>>) {
         // Materialized once at CREATE, so every statement reads the same rows
         // rather than re-spawning a parser per file.
-        Arc::clone(&self.rows)
+        rows.push(Arc::clone(&self.rows));
     }
 
     fn column(&self, row: &JsonRow, ctx: &mut Context, i: c_int) -> Result<()> {
@@ -199,6 +200,17 @@ mod tests {
         assert_eq!(MODULE_NAME, "dirsql_parsed");
     }
 
+    #[test]
+    fn a_statement_reads_the_rows_materialized_at_create() {
+        let table = ParsedTable {
+            column_names: Vec::new(),
+            rows: Arc::new(vec![JsonRow(Vec::new())]),
+        };
+        let stream = Arc::new(RowStream::default());
+        table.rows(&stream);
+        assert!(Arc::ptr_eq(&stream.chunk(0).unwrap(), &table.rows));
+    }
+
     /// The three fixed arguments SQLite prepends before the module's own.
     fn args_with<'a>(user: &[&'a [u8]]) -> Vec<&'a [u8]> {
         let mut all: Vec<&'a [u8]> = vec![

packages/rust/src/row_stream.rs

@@ -0,0 +1,192 @@
+//! One statement's rows, handed over in chunks as a scan makes them, so a
+//! cursor reads the first rows while the walk is still finding the rest.
+
+use std::sync::{Arc, Condvar, Mutex, MutexGuard};
+
+pub struct RowStream<R> {
+    state: Mutex<Streamed<R>>,
+    grown: Condvar,
+}
+
+struct Streamed<R> {
+    chunks: Vec<Arc<Vec<R>>>,
+    done: bool,
+    failed: bool,
+}
+
+impl<R> Default for RowStream<R> {
+    fn default() -> Self {
+        Self {
+            state: Mutex::new(Streamed {
+                chunks: Vec::new(),
+                done: false,
+                failed: false,
+            }),
+            grown: Condvar::new(),
+        }
+    }
+}
+
+impl<R> RowStream<R> {
+    fn lock(&self) -> MutexGuard<'_, Streamed<R>> {
+        self.state
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+    }
+
+    /// Append `chunk`; an empty one is dropped, so a chunk a reader is handed
+    /// always holds a row.
+    pub fn push(&self, chunk: Arc<Vec<R>>) {
+        if chunk.is_empty() {
+            return;
+        }
+        self.lock().chunks.push(chunk);
+        self.grown.notify_all();
+    }
+
+    /// Mark the stream complete; `failed` when the scan did not finish.
+    pub fn finish(&self, failed: bool) {
+        let mut state = self.lock();
+        state.done = true;
+        state.failed = failed;
+        drop(state);
+        self.grown.notify_all();
+    }
+
+    fn settled(&self) -> MutexGuard<'_, Streamed<R>> {
+        let state = self.lock();
+        let state = self
+            .grown
+            .wait_while(state, |state| !state.done)
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        assert!(!state.failed, "the scan feeding this table panicked");
+        state
+    }
+
+    /// Chunk `n`, waiting for the scan to make it; `None` once the scan
+    /// finished short of it.
+    pub fn chunk(&self, n: usize) -> Option<Arc<Vec<R>>> {
+        let state = self.lock();
+        let state = self
+            .grown
+            .wait_while(state, |state| !state.done && state.chunks.len() <= n)
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        assert!(!state.failed, "the scan feeding this table panicked");
+        state.chunks.get(n).cloned()
+    }
+
+    /// Every row, once the scan is through.
+    pub fn all(&self) -> Settled<R> {
+        Settled::new(self.settled().chunks.clone())
+    }
+
+    /// The rows made so far.
+    pub fn len(&self) -> usize {
+        self.lock().chunks.iter().map(|chunk| chunk.len()).sum()
+    }
+}
+
+/// A finished stream's rows, addressable by position.
+pub struct Settled<R> {
+    chunks: Vec<Arc<Vec<R>>>,
+    /// The position of each chunk's first row.
+    starts: Vec<usize>,
+}
+
+impl<R> Settled<R> {
+    fn new(chunks: Vec<Arc<Vec<R>>>) -> Self {
+        let starts = chunks
+            .iter()
+            .scan(0, |next, chunk| {
+                let start = *next;
+                *next += chunk.len();
+                Some(start)
+            })
+            .collect();
+        Self { chunks, starts }
+    }
+
+    pub fn get(&self, position: usize) -> Option<&R> {
+        let n = self
+            .starts
+            .partition_point(|&start| start <= position)
+            .checked_sub(1)?;
+        self.chunks[n].get(position - self.starts[n])
+    }
+
+    pub fn iter(&self) -> impl Iterator<Item = &R> {
+        self.chunks.iter().flat_map(|chunk| chunk.iter())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn stream_of(chunks: &[&[u32]]) -> RowStream<u32> {
+        let stream = RowStream::default();
+        for chunk in chunks {
+            stream.push(Arc::new(chunk.to_vec()));
+        }
+        stream.finish(false);
+        stream
+    }
+
+    #[test]
+    fn chunks_come_back_in_push_order_then_none() {
+        let stream = stream_of(&[&[1, 2], &[3]]);
+        assert_eq!(*stream.chunk(0).unwrap(), vec![1, 2]);
+        assert_eq!(*stream.chunk(1).unwrap(), vec![3]);
+        assert!(stream.chunk(2).is_none());
+    }
+
+    #[test]
+    fn an_empty_chunk_is_never_handed_out() {
+        let stream = stream_of(&[&[], &[1], &[]]);
+        assert_eq!(*stream.chunk(0).unwrap(), vec![1]);
+        assert!(stream.chunk(1).is_none());
+    }
+
+    #[test]
+    fn a_made_chunk_is_read_before_the_scan_finishes() {
+        let stream = RowStream::default();
+        stream.push(Arc::new(vec![7]));
+        assert_eq!(*stream.chunk(0).unwrap(), vec![7]);
+    }
+
+    #[test]
+    fn a_failed_scan_fails_its_readers() {
+        let stream = RowStream::<u32>::default();
+        stream.finish(true);
+        let read = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| stream.chunk(0)));
+        assert!(read.is_err());
+        let read =
+            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| stream.all().iter().count()));
+        assert!(read.is_err());
+    }
+
+    #[test]
+    fn len_counts_the_rows_made_so_far() {
+        let stream = RowStream::default();
+        stream.push(Arc::new(vec![1, 2]));
+        assert_eq!(stream.len(), 2);
+        stream.push(Arc::new(vec![3]));
+        assert_eq!(stream.len(), 3);
+    }
+
+    #[test]
+    fn settled_rows_are_found_by_position_across_chunks() {
+        let all = stream_of(&[&[10, 11], &[12], &[13, 14]]).all();
+        let found: Vec<_> = (0..6).map(|p| all.get(p).copied()).collect();
+        assert_eq!(
+            found,
+            vec![Some(10), Some(11), Some(12), Some(13), Some(14), None]
+        );
+        assert_eq!(all.iter().count(), 5);
+    }
+
+    #[test]
+    fn nothing_is_found_in_an_empty_stream() {
+        assert!(stream_of(&[]).all().get(0).is_none());
+    }
+}