perf(load): make the csv overwrite's fragment write a streaming operator

changelog.d/load-csv-streaming-overwrite-sink.performance.md

@@ -0,0 +1,18 @@
+- A `load csv` overwrite now writes its Lance fragments while the rows
+  drain instead of after the whole delta is validated: each parse
+  partition's rows fill a fragment of up to 1 Mi rows and a full fragment
+  goes to a writer, `stage_write_concurrency` of them at once, as the next
+  rows arrive. A validation with no check that reads the whole delta after
+  the rows are in (value, uniqueness and cardinality checks and endpoint
+  checks swept against integer keys stream; a retained-edge check or a
+  removal scan does not) streams each batch on and the load plans as one
+  pipeline from the source through the write; the explain's `pipelines`
+  shows it, two pipelines otherwise. Its defects are swept as the rows pass
+  and refused in the judgment after the drain, ranked across the parse
+  partitions as before, and that refusal is the one the statement reports
+  even when the sink's own keyed-write check met the same repeated key first.
+  A refusal that reaches the sink after fragments were written deletes them,
+  so a refused load leaves no fragment behind and the table version never
+  moves. Nothing commits before the statement's one publish, as before. LDBC SF30 loads
+  1.2× faster (104 s to 87 s; the write phase after the last row 45 s to
+  4 s, the rest of its work now overlapping the parse).

crates/omnigraph-core/src/instrumentation.rs

@@ -416,6 +416,26 @@ pub fn csv_slice_bytes() -> Option<usize> {
     CSV_SLICE_BYTES.try_with(|bytes| *bytes).ok()
 }
 
+tokio::task_local! {
+    static OVERWRITE_CHUNK_ROWS: usize;
+}
+
+/// Run `fut` with every `load csv` overwrite chunk holding `rows` rows before
+/// its Lance writer takes it, instead of Lance's file target (1 Mi rows).
+/// Test-only entry point; nothing in production sets it.
+pub async fn with_overwrite_chunk_rows<F>(rows: usize, fut: F) -> F::Output
+where
+    F: std::future::Future,
+{
+    OVERWRITE_CHUNK_ROWS.scope(rows, fut).await
+}
+
+/// The chunk row count a test installed for this task, if any; `None` in
+/// production.
+pub fn overwrite_chunk_rows() -> Option<usize> {
+    OVERWRITE_CHUNK_ROWS.try_with(|rows| *rows).ok()
+}
+
 /// Reservations and execution metrics observed by v2 acceptance tests.
 /// Holding a pool here keeps the observer alive without retaining reservations.
 #[derive(Clone, Debug, Default)]

crates/omnigraph-gqt-core/src/lib.rs

@@ -79,6 +79,9 @@ pub struct Case {
     /// The `# csv_slice_bytes:` pin: the bytes one `load csv` parse partition
     /// owns and the slice it reads at once, so a small source cuts partitions.
     csv_slice_bytes: Option<usize>,
+    /// The `# overwrite_chunk_rows:` pin: the rows one `load csv` overwrite
+    /// chunk holds before its writer takes it.
+    overwrite_chunk_rows: Option<usize>,
     pub items: Vec<Item>,
     pub needs_indices: bool,
     /// `--- csv <name>` sections: the CSV text a `load csv "<name>"` step
@@ -575,6 +578,7 @@ struct Header {
     issue: IssueRef,
     traversal: Option<&'static str>,
     csv_slice_bytes: Option<usize>,
+    overwrite_chunk_rows: Option<usize>,
 }
 
 #[derive(Debug, PartialEq)]
@@ -583,8 +587,15 @@ enum IssueRef {
     Num(u64),
 }
 
-/// The five header keys, in the spelling the canonical form requires.
-const HEADER_KEYS: [&str; 5] = ["issue", "red_on", "notes", "traversal", "csv_slice_bytes"];
+/// The six header keys, in the spelling the canonical form requires.
+const HEADER_KEYS: [&str; 6] = [
+    "issue",
+    "red_on",
+    "notes",
+    "traversal",
+    "csv_slice_bytes",
+    "overwrite_chunk_rows",
+];
 
 /// Canonical spelling for one header key and its trimmed value.
 fn canonical_header_line(key: &str, value: &str) -> String {
@@ -623,6 +634,7 @@ fn parse_header(lines: &[&str]) -> Result<Header, String> {
     let mut red_on = false;
     let mut traversal: Option<&'static str> = None;
     let mut csv_slice_bytes: Option<usize> = None;
+    let mut overwrite_chunk_rows: Option<usize> = None;
     for (idx, line) in lines.iter().enumerate() {
         if line.trim().is_empty() {
             continue;
@@ -639,6 +651,7 @@ fn parse_header(lines: &[&str]) -> Result<Header, String> {
             "red_on" => red_on,
             "traversal" => traversal.is_some(),
             "csv_slice_bytes" => csv_slice_bytes.is_some(),
+            "overwrite_chunk_rows" => overwrite_chunk_rows.is_some(),
             _ => false,
         };
         if duplicate {
@@ -689,6 +702,19 @@ fn parse_header(lines: &[&str]) -> Result<Header, String> {
                     })?;
                 csv_slice_bytes = Some(bytes);
             }
+            "overwrite_chunk_rows" => {
+                let rows = value
+                    .parse::<usize>()
+                    .ok()
+                    .filter(|rows| *rows > 0 && value == rows.to_string())
+                    .ok_or_else(|| {
+                        format!(
+                            "line {}: `# overwrite_chunk_rows:` takes a positive row count without sign or leading zeros, got `{value}`",
+                            idx + 1
+                        )
+                    })?;
+                overwrite_chunk_rows = Some(rows);
+            }
             other => unreachable!("split_header_line admits only HEADER_KEYS, got `{other}`"),
         }
     }
@@ -702,6 +728,7 @@ fn parse_header(lines: &[&str]) -> Result<Header, String> {
         issue,
         traversal,
         csv_slice_bytes,
+        overwrite_chunk_rows,
     })
 }
 
@@ -2848,6 +2875,7 @@ pub fn parse_case(stem: &str, text: &str) -> Result<Case, String> {
         fixture,
         traversal: header.traversal,
         csv_slice_bytes: header.csv_slice_bytes,
+        overwrite_chunk_rows: header.overwrite_chunk_rows,
         items,
         needs_indices,
         csv_sections,
@@ -4312,9 +4340,13 @@ pub fn execute_steps<'a, H: ExecutionHost>(
     host: &'a H,
 ) -> futures::future::BoxFuture<'a, Result<Session, String>> {
     let steps = execute_steps_inner(case, path, bless, session, uri, storage, host);
-    match case.csv_slice_bytes {
+    let steps = match case.csv_slice_bytes {
         Some(bytes) => omnigraph::instrumentation::with_csv_slice_bytes(bytes, steps).boxed(),
         None => steps.boxed(),
+    };
+    match case.overwrite_chunk_rows {
+        Some(rows) => omnigraph::instrumentation::with_overwrite_chunk_rows(rows, steps).boxed(),
+        None => steps,
     }
 }
 

crates/omnigraph-gqt-core/src/served.rs

@@ -70,6 +70,9 @@ pub fn admit_served(case: &Case) -> Result<(), String> {
     if case.csv_slice_bytes.is_some() {
         refused.push("the `# csv_slice_bytes:` pin".to_string());
     }
+    if case.overwrite_chunk_rows.is_some() {
+        refused.push("the `# overwrite_chunk_rows:` pin".to_string());
+    }
     if case.needs_indices {
         refused.push("queries that need indices (search, fuzzy, nearest, rrf)".to_string());
     }

crates/omnigraph-gqt/README.md

@@ -607,6 +607,12 @@ the header, so a case of a few KiB cuts two or three partitions and pins what
 happens at their boundaries; the header's head range keeps its 64 KiB floor.
 An in-process pin: a server target refuses it.
 
+The optional `# overwrite_chunk_rows: <n>` header sets the rows one `load csv`
+overwrite chunk holds before a Lance writer takes it, Lance's file target
+(1 Mi rows) without the header, so a case of a few hundred records writes
+fragments before its last record is judged and pins what a late refusal leaves
+behind. An in-process pin: a server target refuses it.
+
 An `expand $src <Edge> $dst:` line selects every matching named-edge physical `Expand` between
 those bindings over that edge type and claims `mode csr` or `mode
 indexed_scan`, the traversal mode the planner recorded (pass `expand_mode`