perf(load): stream a csv overwrite's source by partition and check uniqueness as the rows drain

changelog.d/load-csv-partitioned-read.performance.md

@@ -0,0 +1,12 @@
+- A `load csv` overwrite now plans from the object's metadata and each of its
+  parse partitions reads its own byte range in 8 MiB pieces, pinned to the
+  object's version, instead of one capture of the whole source followed by a
+  serial pass that cut it into slices: no whole-object buffer, no serial
+  resolve pass, and the memory held for the source is one piece per
+  partition, admitted before the first partition is polled. An overwrite whose source's first record is longer than the head
+  range the header is read from (8 MiB) is refused. The receipt's source
+  `slices` gain `read_ns`, the wait for the slice's bytes. The uniqueness
+  check's first holders are one set of maps shared by every partition and
+  sized once from the first slice's rows per byte, so nothing merges after
+  the last batch. LDBC SF30 loads 1.4× faster (155 s to 110 s; the Comment
+  file 2.5×).

crates/omnigraph-api-types/src/lib.rs

@@ -799,6 +799,9 @@ pub struct LoadSliceExecutedOutput {
     pub rows: u64,
     /// Nanoseconds the slice's parse and normalization took.
     pub elapsed_compute_ns: u64,
+    /// Nanoseconds the slice's partition waited for the source's bytes.
+    #[serde(default)]
+    pub read_ns: u64,
     /// Nanoseconds after the source's first poll at which the slice was
     /// claimed.
     pub start_ns: u64,
@@ -2983,6 +2986,7 @@ pub fn load_executed_output(executed: &LoadExecuted) -> LoadExecutedOutput {
                             partition: slice.partition,
                             rows: slice.rows,
                             elapsed_compute_ns: slice.elapsed_compute_ns,
+                            read_ns: slice.read_ns,
                             start_ns: slice.start_ns,
                             end_ns: slice.end_ns,
                         })
@@ -3957,6 +3961,7 @@ mod tests {
                     partition: 0,
                     rows: 2,
                     elapsed_compute_ns: 10,
+                    read_ns: 0,
                     start_ns: 1,
                     end_ns: 30,
                 }],
@@ -3971,7 +3976,8 @@ mod tests {
                                "rows": 2, "elapsed_compute_ns": 12, "wall_ns": 34,
                                "partitions": 1,
                                "slices": [{"slice": 0, "partition": 0, "rows": 2,
-                                           "elapsed_compute_ns": 10, "start_ns": 1, "end_ns": 30}]}]
+                                           "elapsed_compute_ns": 10, "read_ns": 0,
+                                           "start_ns": 1, "end_ns": 30}]}]
             }
         });
         let decoded: ChangeOutput = serde_json::from_value(change.clone()).unwrap();

crates/omnigraph-core/src/instrumentation.rs

@@ -803,6 +803,68 @@ pub async fn external_payload_read_stall() {
     }
 }
 
+/// The task-local seams an external object payload read answers to: the
+/// stall, the hold and the merge-write probes. A load plan's source
+/// partitions read on blocking workers under tasks DataFusion spawns, none of
+/// which inherits a task local, so the statement's task captures these when
+/// it builds its query resources and the worker reinstalls them with
+/// [`with_external_read_seams`]. Production captures three unset lookups.
+#[derive(Clone, Default)]
+pub struct ExternalReadSeams {
+    stalled: bool,
+    hold: Option<ExternalPayloadReadHold>,
+    probes: Option<MergeWriteProbes>,
+}
+
+impl std::fmt::Debug for ExternalReadSeams {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ExternalReadSeams")
+            .field("stalled", &self.stalled)
+            .field("hold", &self.hold.is_some())
+            .field("probes", &self.probes.is_some())
+            .finish()
+    }
+}
+
+/// Capture the external read seams of the current task.
+pub fn capture_external_read_seams() -> ExternalReadSeams {
+    ExternalReadSeams {
+        stalled: EXTERNAL_PAYLOAD_READ_STALL.try_with(|_| ()).is_ok(),
+        hold: EXTERNAL_PAYLOAD_READ_HOLD.try_with(Clone::clone).ok(),
+        probes: MERGE_WRITE_PROBES.try_with(Clone::clone).ok(),
+    }
+}
+
+/// Run `fut` under the seams `capture_external_read_seams` took on another
+/// task; an unset seam installs nothing.
+pub async fn with_external_read_seams<F>(seams: ExternalReadSeams, fut: F) -> F::Output
+where
+    F: std::future::Future,
+{
+    let ExternalReadSeams {
+        stalled,
+        hold,
+        probes,
+    } = seams;
+    let probed = async move {
+        match probes {
+            Some(probes) => MERGE_WRITE_PROBES.scope(probes, fut).await,
+            None => fut.await,
+        }
+    };
+    let held = async move {
+        match hold {
+            Some(hold) => EXTERNAL_PAYLOAD_READ_HOLD.scope(hold, probed).await,
+            None => probed.await,
+        }
+    };
+    if stalled {
+        EXTERNAL_PAYLOAD_READ_STALL.scope((), held).await
+    } else {
+        held.await
+    }
+}
+
 /// Deterministic probe for the number of table-fragment staging futures that
 /// are inside their storage call at once.
 ///

crates/omnigraph-dst/cost_table.txt

@@ -91,18 +91,18 @@ OverwriteWriter128KiB        l.put        calls=6
 OverwriteWriter32MiB         l.get        calls=13
 OverwriteWriter32MiB         l.list       calls=1
 OverwriteWriter32MiB         l.put        calls=6
-OverwriteRemovalScan         l.get        calls=18
+OverwriteRemovalScan         l.get        calls=19
 OverwriteRemovalScan         l.list       calls=1
 OverwriteRemovalScan         l.put        calls=10
-OverwriteRemovalNone         l.get        calls=10
+OverwriteRemovalNone         l.get        calls=11
 OverwriteRemovalNone         l.list       calls=1
 OverwriteRemovalNone         l.put        calls=10
-OverwriteRemovalNoneEmptyTarget l.get        calls=10
+OverwriteRemovalNoneEmptyTarget l.get        calls=11
 OverwriteRemovalNoneEmptyTarget l.list       calls=1
 OverwriteRemovalNoneEmptyTarget l.put        calls=10
-EndpointProbeSwept           l.get        calls=15
+EndpointProbeSwept           l.get        calls=16
 EndpointProbeSwept           l.list       calls=1
 EndpointProbeSwept           l.put        calls=6
-EndpointProbeJudged          l.get        calls=13
+EndpointProbeJudged          l.get        calls=14
 EndpointProbeJudged          l.list       calls=1
 EndpointProbeJudged          l.put        calls=6

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

@@ -76,6 +76,9 @@ pub struct Case {
     /// The harness-only traversal setting; any header requests index preparation.
     /// `auto` retains cost selection, while `indexed` and `csr` check the forced path.
     traversal: Option<&'static str>,
+    /// 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>,
     pub items: Vec<Item>,
     pub needs_indices: bool,
     /// `--- csv <name>` sections: the CSV text a `load csv "<name>"` step
@@ -571,6 +574,7 @@ struct BodySpan {
 struct Header {
     issue: IssueRef,
     traversal: Option<&'static str>,
+    csv_slice_bytes: Option<usize>,
 }
 
 #[derive(Debug, PartialEq)]
@@ -579,8 +583,8 @@ enum IssueRef {
     Num(u64),
 }
 
-/// The four header keys, in the spelling the canonical form requires.
-const HEADER_KEYS: [&str; 4] = ["issue", "red_on", "notes", "traversal"];
+/// The five header keys, in the spelling the canonical form requires.
+const HEADER_KEYS: [&str; 5] = ["issue", "red_on", "notes", "traversal", "csv_slice_bytes"];
 
 /// Canonical spelling for one header key and its trimmed value.
 fn canonical_header_line(key: &str, value: &str) -> String {
@@ -618,6 +622,7 @@ fn parse_header(lines: &[&str]) -> Result<Header, String> {
     let mut issue: Option<IssueRef> = None;
     let mut red_on = false;
     let mut traversal: Option<&'static str> = None;
+    let mut csv_slice_bytes: Option<usize> = None;
     for (idx, line) in lines.iter().enumerate() {
         if line.trim().is_empty() {
             continue;
@@ -633,6 +638,7 @@ fn parse_header(lines: &[&str]) -> Result<Header, String> {
             "issue" => issue.is_some(),
             "red_on" => red_on,
             "traversal" => traversal.is_some(),
+            "csv_slice_bytes" => csv_slice_bytes.is_some(),
             _ => false,
         };
         if duplicate {
@@ -670,6 +676,19 @@ fn parse_header(lines: &[&str]) -> Result<Header, String> {
                     }
                 });
             }
+            "csv_slice_bytes" => {
+                let bytes = value
+                    .parse::<usize>()
+                    .ok()
+                    .filter(|bytes| *bytes > 0 && value == bytes.to_string())
+                    .ok_or_else(|| {
+                        format!(
+                            "line {}: `# csv_slice_bytes:` takes a positive byte count without sign or leading zeros, got `{value}`",
+                            idx + 1
+                        )
+                    })?;
+                csv_slice_bytes = Some(bytes);
+            }
             other => unreachable!("split_header_line admits only HEADER_KEYS, got `{other}`"),
         }
     }
@@ -679,7 +698,11 @@ fn parse_header(lines: &[&str]) -> Result<Header, String> {
     if matches!(issue, IssueRef::Num(_)) && !red_on {
         return Err("`# red_on:` is required when `# issue:` names a number".into());
     }
-    Ok(Header { issue, traversal })
+    Ok(Header {
+        issue,
+        traversal,
+        csv_slice_bytes,
+    })
 }
 
 fn check_file_name(stem: &str, issue: &IssueRef) -> Result<(), String> {
@@ -2824,6 +2847,7 @@ pub fn parse_case(stem: &str, text: &str) -> Result<Case, String> {
         source_lines,
         fixture,
         traversal: header.traversal,
+        csv_slice_bytes: header.csv_slice_bytes,
         items,
         needs_indices,
         csv_sections,
@@ -4287,7 +4311,11 @@ pub fn execute_steps<'a, H: ExecutionHost>(
     storage: Option<Arc<dyn StorageAdapter>>,
     host: &'a H,
 ) -> futures::future::BoxFuture<'a, Result<Session, String>> {
-    execute_steps_inner(case, path, bless, session, uri, storage, host).boxed()
+    let steps = execute_steps_inner(case, path, bless, session, uri, storage, host);
+    match case.csv_slice_bytes {
+        Some(bytes) => omnigraph::instrumentation::with_csv_slice_bytes(bytes, steps).boxed(),
+        None => steps.boxed(),
+    }
 }
 
 async fn execute_steps_inner<H: ExecutionHost>(

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

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

crates/omnigraph-gqt/README.md

@@ -601,6 +601,12 @@ selection to the cost model. Without a traversal header or another index
 requirement, a fixture does not build indexes. `indexed` and `csr` prepare
 indexes and force their respective paths.
 
+The optional `# csv_slice_bytes: <n>` header sets the source bytes one
+`load csv` parse partition owns and the slice it reads at once, 8 MiB without
+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.
+
 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`