diff --git a/changelog.d/load-csv-streaming-overwrite-sink.performance.md b/changelog.d/load-csv-streaming-overwrite-sink.performance.md
new file mode 100644
index 000000000..cb233a2e9
--- /dev/null
+++ b/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).
diff --git a/crates/omnigraph-core/src/instrumentation.rs b/crates/omnigraph-core/src/instrumentation.rs
index 528daa307..f77d9a905 100644
--- a/crates/omnigraph-core/src/instrumentation.rs
+++ b/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)]
diff --git a/crates/omnigraph-gqt-core/src/lib.rs b/crates/omnigraph-gqt-core/src/lib.rs
index 69f117dc1..9f5d9f91f 100644
--- a/crates/omnigraph-gqt-core/src/lib.rs
+++ b/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,
     }
 }
 
diff --git a/crates/omnigraph-gqt-core/src/served.rs b/crates/omnigraph-gqt-core/src/served.rs
index 753799ded..63edc5942 100644
--- a/crates/omnigraph-gqt-core/src/served.rs
+++ b/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());
     }
diff --git a/crates/omnigraph-gqt/README.md b/crates/omnigraph-gqt/README.md
index 259c89c32..87455657b 100644
--- a/crates/omnigraph-gqt/README.md
+++ b/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`
diff --git a/crates/omnigraph-gqt/cases/load_csv_overwrite_refusal_late_in_the_source_stages_nothing.gqt b/crates/omnigraph-gqt/cases/load_csv_overwrite_refusal_late_in_the_source_stages_nothing.gqt
new file mode 100644
index 000000000..2eb84e273
--- /dev/null
+++ b/crates/omnigraph-gqt/cases/load_csv_overwrite_refusal_late_in_the_source_stages_nothing.gqt
@@ -0,0 +1,415 @@
+# issue: none
+# csv_slice_bytes: 4096
+# overwrite_chunk_rows: 10
+# notes: A node table with no incoming edge type sweeps every check as the rows pass and
+# notes: refuses after the drain: one pipeline, a fragment written every 10 rows (the pin
+# notes: above; 1 Mi rows without it). A refusal past written fragments deletes them and the
+# notes: committed rows stay, with the validator's text and value even where the sink's
+# notes: keyed-write check met a key repeated inside one fragment first.
+
+--- runner
+timeout_ms: 10000
+environments:
+  - target: omnigraph-engine
+    storage: local-filesystem
+
+--- schema
+node Item {
+    id: I64 @key
+    note: String
+    n: I64
+}
+
+--- seed
+{"type":"Item","data":{"id":1,"note":"a","n":10}}
+{"type":"Item","data":{"id":2,"note":"b","n":20}}
+{"type":"Item","data":{"id":3,"note":"c","n":30}}
+
+--- csv bad.csv
+id,note,n
+1000,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,0
+1001,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,1
+1002,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,2
+1003,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,3
+1004,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,4
+1005,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,5
+1006,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,6
+1007,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,7
+1008,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,8
+1009,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,9
+1010,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,10
+1011,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,11
+1012,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,12
+1013,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,13
+1014,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,14
+1015,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,15
+1016,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,16
+1017,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,17
+1018,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,18
+1019,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,19
+1020,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,20
+1021,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,21
+1022,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,22
+1023,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,23
+1024,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,24
+1025,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,25
+1026,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,26
+1027,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,27
+1028,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,28
+1029,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,29
+1030,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,30
+1031,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,31
+1032,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,32
+1033,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,33
+1034,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,34
+1035,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,35
+1036,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,36
+1037,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,37
+1038,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,38
+1039,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,39
+1040,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,40
+1041,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,41
+1042,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,42
+1043,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,43
+1044,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,44
+1045,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,45
+1046,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,46
+1047,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,47
+1048,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,48
+1049,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,49
+1050,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,50
+1051,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,51
+1052,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,52
+1053,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,53
+1054,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,54
+1055,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,55
+1056,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,56
+1057,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,57
+1058,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,58
+1059,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,59
+1060,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,60
+1061,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,61
+1062,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,62
+1063,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,63
+1064,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,64
+1065,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,65
+1066,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,66
+1067,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,67
+1068,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,68
+1069,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,69
+1070,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,70
+1071,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,71
+1072,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,72
+1073,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,73
+1074,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,74
+1075,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,75
+1076,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,76
+1077,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,77
+1078,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,78
+1079,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,79
+1080,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,80
+1081,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,81
+1082,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,82
+1083,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,83
+1084,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,84
+1085,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,85
+1086,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,86
+1087,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,87
+1088,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,88
+1089,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,89
+1090,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,90
+1091,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,91
+1092,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,92
+1093,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,93
+1094,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,94
+1095,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,95
+1096,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,96
+1097,,97
+1098,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,98
+1099,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,99
+
+--- mutate
+load csv "bad.csv" into node:Item mode overwrite header
+
+--- expect error: load csv into node:Item: record 98 has an empty field for non-nullable property 'note'
+
+--- query
+query committed_rows_stay() {
+    match { $i: Item }
+    return { count($i) as c, sum($i.n) as s }
+}
+
+--- expect unordered
+{"c":3,"s":60}
+
+--- expect shape
+c: I64
+s: F64
+
+--- csv repeat_inside.csv
+id,note,n
+1000,a,0
+1001,a,1
+1002,a,2
+1003,a,3
+1003,a,4
+1005,a,5
+1006,a,6
+1007,a,7
+1008,a,8
+1009,a,9
+1010,a,10
+1011,a,11
+1012,a,12
+1013,a,13
+1014,a,14
+1015,a,15
+1016,a,16
+1017,a,17
+1018,a,18
+1019,a,19
+1020,a,20
+1021,a,21
+1022,a,22
+1023,a,23
+1024,a,24
+1025,a,25
+1026,a,26
+1027,a,27
+1028,a,28
+1029,a,29
+
+--- mutate
+load csv "repeat_inside.csv" into node:Item mode overwrite header
+
+--- expect error: @unique violation on Item.id: value '1003'
+
+--- query
+query committed_rows_stay_after_the_in_fragment_repeat() {
+    match { $i: Item }
+    return { count($i) as c, sum($i.n) as s }
+}
+
+--- expect unordered
+{"c":3,"s":60}
+
+--- expect shape
+c: I64
+s: F64
+
+--- csv repeat_closing.csv
+id,note,n
+1000,a,0
+1001,a,1
+1002,a,2
+1003,a,3
+1004,a,4
+1005,a,5
+1006,a,6
+1007,a,7
+1008,a,8
+1008,a,9
+1010,a,10
+1011,a,11
+1012,a,12
+1013,a,13
+1014,a,14
+1015,a,15
+1016,a,16
+1017,a,17
+1018,a,18
+1019,a,19
+1020,a,20
+1021,a,21
+1022,a,22
+1023,a,23
+1024,a,24
+1025,a,25
+1026,a,26
+1027,a,27
+1028,a,28
+1029,a,29
+
+--- mutate
+load csv "repeat_closing.csv" into node:Item mode overwrite header
+
+--- expect error: @unique violation on Item.id: value '1008'
+
+--- csv repeat_straddling.csv
+id,note,n
+1000,a,0
+1001,a,1
+1002,a,2
+1003,a,3
+1004,a,4
+1005,a,5
+1006,a,6
+1007,a,7
+1008,a,8
+1009,a,9
+1009,a,10
+1011,a,11
+1012,a,12
+1013,a,13
+1014,a,14
+1015,a,15
+1016,a,16
+1017,a,17
+1018,a,18
+1019,a,19
+1020,a,20
+1021,a,21
+1022,a,22
+1023,a,23
+1024,a,24
+1025,a,25
+1026,a,26
+1027,a,27
+1028,a,28
+1029,a,29
+
+--- mutate
+load csv "repeat_straddling.csv" into node:Item mode overwrite header
+
+--- expect error: @unique violation on Item.id: value '1009'
+
+--- query
+query committed_rows_stay_after_the_straddling_repeat() {
+    match { $i: Item }
+    return { count($i) as c, sum($i.n) as s }
+}
+
+--- expect unordered
+{"c":3,"s":60}
+
+--- expect shape
+c: I64
+s: F64
+
+--- csv good.csv
+id,note,n
+1000,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,0
+1001,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,1
+1002,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,2
+1003,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,3
+1004,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,4
+1005,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,5
+1006,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,6
+1007,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,7
+1008,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,8
+1009,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,9
+1010,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,10
+1011,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,11
+1012,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,12
+1013,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,13
+1014,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,14
+1015,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,15
+1016,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,16
+1017,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,17
+1018,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,18
+1019,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,19
+1020,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,20
+1021,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,21
+1022,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,22
+1023,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,23
+1024,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,24
+1025,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,25
+1026,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,26
+1027,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,27
+1028,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,28
+1029,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,29
+1030,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,30
+1031,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,31
+1032,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,32
+1033,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,33
+1034,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,34
+1035,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,35
+1036,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,36
+1037,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,37
+1038,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,38
+1039,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,39
+1040,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,40
+1041,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,41
+1042,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,42
+1043,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,43
+1044,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,44
+1045,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,45
+1046,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,46
+1047,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,47
+1048,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,48
+1049,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,49
+1050,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,50
+1051,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,51
+1052,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,52
+1053,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,53
+1054,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,54
+1055,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,55
+1056,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,56
+1057,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,57
+1058,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,58
+1059,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,59
+1060,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,60
+1061,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,61
+1062,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,62
+1063,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,63
+1064,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,64
+1065,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,65
+1066,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,66
+1067,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,67
+1068,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,68
+1069,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,69
+1070,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,70
+1071,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,71
+1072,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,72
+1073,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,73
+1074,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,74
+1075,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,75
+1076,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,76
+1077,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,77
+1078,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,78
+1079,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,79
+1080,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,80
+1081,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,81
+1082,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,82
+1083,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,83
+1084,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,84
+1085,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,85
+1086,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,86
+1087,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,87
+1088,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,88
+1089,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,89
+1090,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,90
+1091,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,91
+1092,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,92
+1093,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,93
+1094,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,94
+1095,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,95
+1096,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,96
+1097,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,97
+1098,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,98
+1099,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,99
+
+--- mutate
+load csv "good.csv" into node:Item mode overwrite header
+
+--- expect ok
+
+--- expect plan
+load route: plan
+load validate removal node:Item: none
+load write: overwrite fragments
+load pipeline 0: CsvSource -> Validate -> FragmentWrite => consumer partitions 2
+load executed: partitions 2 rows 100
+
+--- query
+query loaded_rows() {
+    match { $i: Item }
+    return { count($i) as c, sum($i.n) as s }
+}
+
+--- expect unordered
+{"c":100,"s":4950}
+
+--- expect shape
+c: I64
+s: F64
diff --git a/crates/omnigraph-gqt/cases/load_csv_overwrite_refusal_precedence_holds_while_fragments_stream.gqt b/crates/omnigraph-gqt/cases/load_csv_overwrite_refusal_precedence_holds_while_fragments_stream.gqt
new file mode 100644
index 000000000..614d1a621
--- /dev/null
+++ b/crates/omnigraph-gqt/cases/load_csv_overwrite_refusal_precedence_holds_while_fragments_stream.gqt
@@ -0,0 +1,307 @@
+# issue: none
+# csv_slice_bytes: 4096
+# overwrite_chunk_rows: 10
+# notes: Partition 0 carries a key repeated inside its first fragment (records 3 and 5),
+# notes: partition 1 a bad integer at record 95; the refusal is the validate operator's,
+# notes: ranked over every partition after the drain, not the sink's check that met the
+# notes: repeat first; the committed rows stay.
+
+--- runner
+timeout_ms: 10000
+environments:
+  - target: omnigraph-engine
+    storage: local-filesystem
+
+--- schema
+node Item {
+    id: I64 @key
+    note: String
+    n: I64
+}
+
+--- seed
+{"type":"Item","data":{"id":1,"note":"s1","n":10}}
+{"type":"Item","data":{"id":2,"note":"s2","n":20}}
+{"type":"Item","data":{"id":3,"note":"s3","n":30}}
+
+--- csv mixed.csv
+id,note,n
+1000,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,0
+1001,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,1
+1002,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,2
+1003,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,3
+1002,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,4
+1005,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,5
+1006,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,6
+1007,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,7
+1008,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,8
+1009,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,9
+1010,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,10
+1011,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,11
+1012,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,12
+1013,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,13
+1014,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,14
+1015,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,15
+1016,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,16
+1017,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,17
+1018,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,18
+1019,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,19
+1020,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,20
+1021,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,21
+1022,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,22
+1023,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,23
+1024,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,24
+1025,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,25
+1026,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,26
+1027,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,27
+1028,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,28
+1029,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,29
+1030,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,30
+1031,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,31
+1032,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,32
+1033,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,33
+1034,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,34
+1035,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,35
+1036,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,36
+1037,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,37
+1038,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,38
+1039,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,39
+1040,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,40
+1041,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,41
+1042,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,42
+1043,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,43
+1044,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,44
+1045,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,45
+1046,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,46
+1047,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,47
+1048,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,48
+1049,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,49
+1050,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,50
+1051,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,51
+1052,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,52
+1053,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,53
+1054,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,54
+1055,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,55
+1056,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,56
+1057,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,57
+1058,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,58
+1059,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,59
+1060,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,60
+1061,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,61
+1062,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,62
+1063,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,63
+1064,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,64
+1065,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,65
+1066,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,66
+1067,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,67
+1068,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,68
+1069,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,69
+1070,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,70
+1071,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,71
+1072,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,72
+1073,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,73
+1074,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,74
+1075,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,75
+1076,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,76
+1077,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,77
+1078,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,78
+1079,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,79
+1080,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,80
+1081,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,81
+1082,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,82
+1083,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,83
+1084,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,84
+1085,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,85
+1086,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,86
+1087,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,87
+1088,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,88
+1089,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,89
+1090,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,90
+1091,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,91
+1092,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,92
+1093,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,93
+1094,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,abc
+1095,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,95
+1096,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,96
+1097,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,97
+1098,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,98
+1099,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,99
+1100,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,100
+1101,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,101
+1102,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,102
+1103,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,103
+1104,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,104
+1105,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,105
+1106,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,106
+1107,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,107
+1108,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,108
+1109,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,109
+1110,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,110
+1111,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,111
+1112,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,112
+1113,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,113
+1114,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,114
+1115,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,115
+1116,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,116
+1117,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,117
+1118,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,118
+1119,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,119
+
+--- mutate
+load csv "mixed.csv" into node:Item mode overwrite header
+
+--- expect error: Error while parsing value 'abc' as type 'Int64' for column 2 at line 95.
+
+--- query
+query after_mixed() {
+    match { $i: Item }
+    return { count($i) as c, sum($i.n) as s }
+}
+
+--- expect unordered
+{"c":3,"s":60}
+
+--- expect shape
+c: I64
+s: F64
+
+--- csv clean.csv
+id,note,n
+1000,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,0
+1001,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,1
+1002,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,2
+1003,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,3
+1004,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,4
+1005,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,5
+1006,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,6
+1007,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,7
+1008,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,8
+1009,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,9
+1010,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,10
+1011,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,11
+1012,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,12
+1013,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,13
+1014,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,14
+1015,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,15
+1016,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,16
+1017,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,17
+1018,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,18
+1019,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,19
+1020,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,20
+1021,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,21
+1022,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,22
+1023,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,23
+1024,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,24
+1025,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,25
+1026,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,26
+1027,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,27
+1028,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,28
+1029,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,29
+1030,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,30
+1031,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,31
+1032,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,32
+1033,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,33
+1034,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,34
+1035,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,35
+1036,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,36
+1037,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,37
+1038,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,38
+1039,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,39
+1040,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,40
+1041,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,41
+1042,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,42
+1043,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,43
+1044,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,44
+1045,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,45
+1046,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,46
+1047,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,47
+1048,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,48
+1049,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,49
+1050,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,50
+1051,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,51
+1052,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,52
+1053,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,53
+1054,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,54
+1055,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,55
+1056,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,56
+1057,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,57
+1058,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,58
+1059,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,59
+1060,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,60
+1061,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,61
+1062,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,62
+1063,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,63
+1064,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,64
+1065,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,65
+1066,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,66
+1067,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,67
+1068,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,68
+1069,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,69
+1070,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,70
+1071,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,71
+1072,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,72
+1073,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,73
+1074,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,74
+1075,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,75
+1076,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,76
+1077,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,77
+1078,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,78
+1079,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,79
+1080,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,80
+1081,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,81
+1082,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,82
+1083,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,83
+1084,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,84
+1085,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,85
+1086,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,86
+1087,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,87
+1088,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,88
+1089,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,89
+1090,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,90
+1091,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,91
+1092,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,92
+1093,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,93
+1094,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,94
+1095,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,95
+1096,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,96
+1097,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,97
+1098,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,98
+1099,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,99
+1100,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,100
+1101,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,101
+1102,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,102
+1103,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,103
+1104,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,104
+1105,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,105
+1106,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,106
+1107,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,107
+1108,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,108
+1109,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,109
+1110,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,110
+1111,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,111
+1112,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,112
+1113,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,113
+1114,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,114
+1115,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,115
+1116,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,116
+1117,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,117
+1118,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,118
+1119,xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,119
+
+--- mutate
+load csv "clean.csv" into node:Item mode overwrite header
+
+--- expect ok
+
+--- query
+query reloaded() {
+    match { $i: Item }
+    return { count($i) as c, sum($i.n) as s }
+}
+
+--- expect unordered
+{"c":120,"s":7140}
+
+--- expect shape
+c: I64
+s: F64
diff --git a/crates/omnigraph-gqt/cases/load_csv_statement_dst_failure_keeps_writing.gqt b/crates/omnigraph-gqt/cases/load_csv_statement_dst_failure_keeps_writing.gqt
index 1e807c8c3..1d931495c 100644
--- a/crates/omnigraph-gqt/cases/load_csv_statement_dst_failure_keeps_writing.gqt
+++ b/crates/omnigraph-gqt/cases/load_csv_statement_dst_failure_keeps_writing.gqt
@@ -1,5 +1,5 @@
 # issue: none
-# notes: CSV fixtures keep their source policy before and after a DST restart; a failed publication keeps the graph unchanged.
+# notes: CSV fixtures keep their source policy before and after a DST restart; a failure before the publication point keeps the graph unchanged, one past it leaves the publication durable.
 
 --- runner
 timeout_ms: 10000
@@ -88,6 +88,84 @@ query unchanged_before_publish() { match { $p: Person } return { $p.name } }
 --- expect shape
 p.name: String
 
+--- seam
+at: mutation.post_table_commit
+occurrence: 1
+action: fail
+scope: next_step
+
+--- mutate
+load csv "replacement.csv" into node:Person mode overwrite header
+--- expect error: injected failpoint triggered: mutation.post_table_commit
+
+--- restart
+--- query
+query unchanged_after_table_commit() { match { $p: Person } return { $p.name } }
+--- expect unordered
+{"p.name":"alice"}
+{"p.name":"bob"}
+--- expect shape
+p.name: String
+
+--- seam
+at: publish.pre_merge
+occurrence: 1
+action: fail
+scope: next_step
+
+--- mutate
+load csv "replacement.csv" into node:Person mode overwrite header
+--- expect error: injected failpoint triggered: publish.pre_merge
+
+--- restart
+--- query
+query unchanged_before_merge() { match { $p: Person } return { $p.name } }
+--- expect unordered
+{"p.name":"alice"}
+{"p.name":"bob"}
+--- expect shape
+p.name: String
+
+--- seam
+at: publish.post_merge_pre_ack
+occurrence: 1
+action: fail
+scope: next_step
+
+--- mutate
+load csv "replacement.csv" into node:Person mode overwrite header
+--- expect ok
+
+--- restart
+--- query
+query durable_past_the_merge() { match { $p: Person } return { $p.name } }
+--- expect unordered
+{"p.name":"carol"}
+--- expect shape
+p.name: String
+
+--- mutate
+load csv "people.csv" into node:Person mode overwrite header
+--- expect ok
+
+--- seam
+at: graph_publish.after_manifest_commit
+occurrence: 1
+action: fail
+scope: next_step
+
+--- mutate
+load csv "replacement.csv" into node:Person mode overwrite header
+--- expect error: injected failpoint triggered: graph_publish.after_manifest_commit
+
+--- restart
+--- query
+query durable_past_the_manifest_commit() { match { $p: Person } return { $p.name } }
+--- expect unordered
+{"p.name":"carol"}
+--- expect shape
+p.name: String
+
 --- mutate
 load csv "replacement.csv" into node:Person mode overwrite header
 --- expect ok
diff --git a/crates/omnigraph-planner/src/lib.rs b/crates/omnigraph-planner/src/lib.rs
index 5df37b0d9..e38cef28d 100644
--- a/crates/omnigraph-planner/src/lib.rs
+++ b/crates/omnigraph-planner/src/lib.rs
@@ -80,6 +80,7 @@ pub use physical::{
     PhysicalNode, PhysicalNodeKind, PhysicalPlan, PipelineShape, PipelineSink, PipelineSource,
     PipelineSpec, Prefilter, PrefilterMode, Properties, ROW_ADDRESS_PREFIX, RankArm, RankKind,
     RankScope, RankedAccess, ScanInput, SideChain, StatisticSource, operator_rows,
+    validate_streams,
 };
 pub use registry::{Coverage, Entry, Route, Shape};
 pub use route::RouteOverride;
diff --git a/crates/omnigraph-planner/src/physical.rs b/crates/omnigraph-planner/src/physical.rs
index ced13521b..0f959feb8 100644
--- a/crates/omnigraph-planner/src/physical.rs
+++ b/crates/omnigraph-planner/src/physical.rs
@@ -888,9 +888,9 @@ physical_nodes! {
         max_partitions: u32,
         max_workers: u32,
     } => "CsvSource" [];
-    /// The one validation over the whole delta, a pipeline breaker: every
-    /// check extracted in one pass per batch, each check's `access` saying how
-    /// its committed image is read, `removal` whether removed ids are found.
+    /// The one validation of the delta, each check's `access` saying how its
+    /// committed image is read and `removal` whether removed ids are found;
+    /// a pipeline breaker only while [`validate_streams`] is false.
     Validate {
         input: NodeId,
         table: String,
@@ -922,20 +922,46 @@ pub enum PipelineShape {
     Breaker,
 }
 
+/// Whether a validation judges every row as it passes, so its operator
+/// streams: a value, unique or cardinality check reads only what the sweep
+/// extracted, a swept endpoint check probes each row against sets built before
+/// the drain. A judged endpoint check, a retained-edge check or a removal scan
+/// reads the whole delta after the drain and makes the operator a breaker.
+pub fn validate_streams(checks: &[ValidateSpec], removal: ValidateRemoval) -> bool {
+    use omnigraph_compiler::ir::ValidateCheck;
+    removal == ValidateRemoval::None
+        && checks.iter().all(|spec| match &spec.check {
+            ValidateCheck::Value { .. }
+            | ValidateCheck::Unique { .. }
+            | ValidateCheck::Cardinality { .. } => true,
+            ValidateCheck::EdgeRi { .. } => spec.probe == Some(crate::cost::EndpointProbe::Swept),
+            ValidateCheck::RetainedEdges { .. } => false,
+        })
+}
+
 impl PhysicalNode {
     /// The node's [`PipelineShape`] over its streamed input (the one
     /// [`Self::collected_sides`] does not name). A source has no input to
     /// hold and a `FragmentWrite` is the pipeline's end, whose emission is
     /// its input's; neither is a cut. An `Expand` holds its paths only as a
     /// costed multi-hop traversal: one hop streams its pairs and a budgeted
-    /// traversal streams its frontier.
+    /// traversal streams its frontier. A `Validate` holds the delta only
+    /// while a check reads it whole after the drain ([`validate_streams`]).
     pub fn shape(&self) -> PipelineShape {
         match self {
             Self::Sort { .. }
             | Self::Aggregate { .. }
             | Self::AntiJoin { .. }
-            | Self::RankFuse { .. }
-            | Self::Validate { .. } => PipelineShape::Breaker,
+            | Self::RankFuse { .. } => PipelineShape::Breaker,
+            Self::Validate {
+                checks, removal, ..
+            } => {
+                if validate_streams(checks, *removal) {
+                    PipelineShape::Streaming
+                } else {
+                    PipelineShape::Breaker
+                }
+            }
             Self::Expand {
                 max_hops, policy, ..
             } => {
diff --git a/crates/omnigraph-planner/tests/integration/load_plan.rs b/crates/omnigraph-planner/tests/integration/load_plan.rs
index 2b76434de..fc8f2e8fe 100644
--- a/crates/omnigraph-planner/tests/integration/load_plan.rs
+++ b/crates/omnigraph-planner/tests/integration/load_plan.rs
@@ -219,6 +219,78 @@ fn node_overwrite_resolves_to_source_validate_chain_and_write() {
     assert_eq!(pipelines[1]["depends_on"], serde_json::json!([0]));
 }
 
+/// A node target with an integer key and no incoming edge type carries only
+/// checks judged as the rows pass and no removal scan: its validation streams
+/// and the load is one pipeline; the whole-input validation above stays two.
+#[test]
+fn per_row_checks_plan_one_pipeline_through_the_write() {
+    use omnigraph_planner::{PipelineShape, validate_streams};
+
+    let schema = parse_schema(
+        r#"
+node Item {
+    id: I64 @key
+    note: String
+}
+"#,
+    )
+    .expect("schema parses");
+    let catalog = build_catalog(&schema).expect("catalog builds");
+    let stmt = LoadStmt {
+        source: "s3://bucket/items.csv".to_string(),
+        target: LoadTarget::Node("Item".into()),
+        mode: LoadStmtMode::Overwrite,
+        delimiter: b',',
+        header: true,
+        columns: None,
+    };
+    let op = Operation::Load {
+        ir: Box::new(lower_load(&catalog, &stmt)),
+        tables: vec![side("node:Item", 7)],
+        source_bytes: 3 * SOURCE_PARTITION_BYTES,
+    };
+    let Decision::Engine { plan, explain, .. } =
+        route(&op, &source(), RouteOverride::Registry, &BOUNDS)
+    else {
+        panic!("an overwrite load routes to the engine");
+    };
+    let validations: Vec<&PhysicalNode> = chain(&plan)
+        .into_iter()
+        .filter(|node| matches!(node, PhysicalNode::Validate { .. }))
+        .collect();
+    let [validation] = &validations[..] else {
+        panic!("one validation per load plan, found {}", validations.len());
+    };
+    let PhysicalNode::Validate {
+        checks, removal, ..
+    } = validation
+    else {
+        unreachable!();
+    };
+    assert!(validate_streams(checks, *removal), "{checks:?} {removal:?}");
+    assert_eq!(validation.shape(), PipelineShape::Streaming);
+    let document = explain.to_value();
+    let pipelines = document["pipelines"]
+        .as_array()
+        .expect("a load plan has pipelines");
+    assert_eq!(
+        pipelines.len(),
+        1,
+        "a validation judged per row cuts nothing: {pipelines:?}"
+    );
+    assert_eq!(pipelines[0]["source"]["node"], "CsvSource");
+    assert_eq!(pipelines[0]["partitions"], 3);
+    let operators: Vec<&str> = pipelines[0]["operators"]
+        .as_array()
+        .expect("operators")
+        .iter()
+        .map(|operator| operator["name"].as_str().expect("a name"))
+        .collect();
+    assert_eq!(operators, ["Validate", "FragmentWrite"]);
+    assert_eq!(pipelines[0]["sink"]["name"], "consumer");
+    assert_eq!(pipelines[0]["depends_on"], serde_json::json!([]));
+}
+
 /// A node target whose manifest count is zero removes nothing, so pass 9 plans
 /// neither the removal scan nor the retained-edge checks; a populated or
 /// uncounted target keeps both; an edge overwrite never scans.
diff --git a/crates/omnigraph/src/engine/lower.rs b/crates/omnigraph/src/engine/lower.rs
index 860474b62..70f305b7a 100644
--- a/crates/omnigraph/src/engine/lower.rs
+++ b/crates/omnigraph/src/engine/lower.rs
@@ -475,9 +475,9 @@ impl<'l, 'a> Walk<'l, 'a> {
             .into()
     }
 
-    /// The one operator of node `id`; DataFusion's `AggregateExec`, the one
-    /// operator not built through a shape, has its emission compared against
-    /// the plan's cut (`Final` while its input order mode is linear).
+    /// The one operator of node `id`; `AggregateExec` (not built through a
+    /// shape; `Final` while its input order mode is linear) and `ValidateExec`
+    /// (its shape chosen by the plan's checks) have their emission checked.
     fn built(&mut self, id: NodeId, operator: impl ExecutionPlan) -> Lowers<Plan> {
         let plan: Plan = Arc::new(operator);
         let node = self.lowering.node(id)?;
@@ -491,6 +491,21 @@ impl<'l, 'a> Walk<'l, 'a> {
             ))
             .into());
         }
+        if matches!(node, PhysicalNode::Validate { .. }) {
+            let declared = match node.shape() {
+                omnigraph_planner::PipelineShape::Breaker => EmissionType::Final,
+                omnigraph_planner::PipelineShape::Streaming => EmissionType::Incremental,
+            };
+            if plan.properties().emission_type != declared {
+                return Err(OmniError::manifest_internal(format!(
+                    "node {id} `Validate` is {:?} in the plan's cut but `{}` emits {:?}",
+                    node.shape(),
+                    plan.name(),
+                    plan.properties().emission_type,
+                ))
+                .into());
+            }
+        }
         if matches!(
             node,
             PhysicalNode::Projection { .. }
@@ -1296,7 +1311,8 @@ impl Lower for Walk<'_, '_> {
             snapshot: self.lowering.snapshot("load validation")?.clone(),
             catalog: Arc::clone(self.lowering.catalog),
         };
-        self.built(id, ValidateExec::new(input, judge))
+        let streams = omnigraph_planner::validate_streams(fields.checks, fields.removal);
+        self.built(id, ValidateExec::new(input, judge, streams))
     }
 
     fn fragment_write(
diff --git a/crates/omnigraph/src/engine/operators/fragment_sink.rs b/crates/omnigraph/src/engine/operators/fragment_sink.rs
index 25c4638b7..c35bf014a 100644
--- a/crates/omnigraph/src/engine/operators/fragment_sink.rs
+++ b/crates/omnigraph/src/engine/operators/fragment_sink.rs
@@ -1,15 +1,21 @@
 //! A load's write: the validated delta staged as Lance fragments of the
-//! target under the open transaction, never committed here. DataFusion's
+//! target under the open transaction, never committed here, written while
+//! the delta drains: each source partition's rows fill one chunk at a time
+//! and a full chunk goes to a Lance writer as the next rows arrive. DataFusion's
 //! `DataSinkExec` hosts the sink and emits the one `count` row; the driver
 //! takes the staged transaction from the slot and commits it after the plan.
+//! A refusal reaching the sink deletes every fragment already written.
 
+use std::collections::BTreeMap;
 use std::fmt;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 
+use arrow_array::RecordBatch;
 use arrow_schema::SchemaRef;
 use async_trait::async_trait;
 use datafusion::common::Result as DfResult;
 use datafusion::datasource::sink::DataSink;
+use datafusion::error::DataFusionError;
 use datafusion::execution::TaskContext;
 use datafusion::physical_plan::metrics::{
     BaselineMetrics, ExecutionPlanMetricsSet, MetricBuilder, MetricsSet,
@@ -17,14 +23,15 @@ use datafusion::physical_plan::metrics::{
 use datafusion::physical_plan::{DisplayAs, DisplayFormatType, SendableRecordBatchStream};
 use futures::StreamExt;
 
+use super::csv_source::slice_of;
 use super::external;
 use super::memory::WorkMemory;
 use crate::db::MutationOpKind;
 use crate::engine::LoadWrite;
 use crate::error::OmniError;
-use crate::exec::staging::{MutationStaging, PendingMode};
+use crate::exec::staging::MutationStaging;
 use crate::loader::STAGED_WRITE_INPUT_BYTES_RESOURCE;
-use crate::table_store::write_batch_bytes;
+use crate::table_store::{OverwriteChunk, overwrite_chunk_rows, write_batch_bytes};
 
 /// The `staged_rows` metric: the rows the sink staged.
 pub(crate) const STAGED_ROWS: &str = "staged_rows";
@@ -84,7 +91,7 @@ impl DataSink for LanceFragmentSink {
 
     async fn write_all(
         &self,
-        mut data: SendableRecordBatchStream,
+        data: SendableRecordBatchStream,
         context: &Arc<TaskContext>,
     ) -> DfResult<u64> {
         let mut work = WorkMemory::new(Arc::clone(context), "LanceFragmentSink")?;
@@ -120,31 +127,90 @@ impl DataSink for LanceFragmentSink {
                 MutationOpKind::SchemaRewrite,
             )
             .map_err(external)?;
-        let mut rows: usize = 0;
-        while let Some(batch) = data.next().await {
-            let batch = batch?;
-            let _timer = baseline.elapsed_compute().timer();
-            let bytes = write_batch_bytes(&batch).map_err(external)?.rows;
-            if bytes > self.max_input_bytes {
-                return Err(external(OmniError::resource_limit(
-                    STAGED_WRITE_INPUT_BYTES_RESOURCE,
-                    self.max_input_bytes,
-                    bytes,
-                )));
+        let input_failure: Arc<Mutex<Option<DataFusionError>>> = Arc::new(Mutex::new(None));
+        let chunker = Chunker::new(
+            Arc::clone(&self.schema),
+            overwrite_chunk_rows(),
+            self.max_input_bytes,
+            work.child("overwrite chunks")?,
+        );
+        let chunks = futures::stream::try_unfold(
+            (data, chunker, Vec::<OverwriteChunk>::new(), false),
+            {
+                let input_failure = Arc::clone(&input_failure);
+                let baseline = baseline.clone();
+                move |(mut data, mut chunker, mut ready, mut ended)| {
+                    let input_failure = Arc::clone(&input_failure);
+                    let baseline = baseline.clone();
+                    async move {
+                        loop {
+                            if let Some(chunk) = ready.pop() {
+                                return Ok(Some((chunk, (data, chunker, ready, ended))));
+                            }
+                            if ended {
+                                return Ok(None);
+                            }
+                            match data.next().await {
+                                Some(Ok(batch)) => {
+                                    let _timer = baseline.elapsed_compute().timer();
+                                    let mut cut = chunker.push(batch)?;
+                                    cut.reverse();
+                                    ready = cut;
+                                }
+                                Some(Err(error)) => {
+                                    *input_failure
+                                        .lock()
+                                        .unwrap_or_else(std::sync::PoisonError::into_inner) =
+                                        Some(error);
+                                    return Err(OmniError::manifest_internal(
+                                        "load csv: the validated delta refused before the write finished",
+                                    ));
+                                }
+                                None => {
+                                    ended = true;
+                                    let mut cut = chunker.flush();
+                                    cut.reverse();
+                                    ready = cut;
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+        );
+        let rows = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let chunks = chunks.inspect({
+            let rows = Arc::clone(&rows);
+            move |chunk| {
+                if let Ok(chunk) = chunk {
+                    let in_chunk: usize = chunk.batches.iter().map(RecordBatch::num_rows).sum();
+                    rows.fetch_add(in_chunk, std::sync::atomic::Ordering::Relaxed);
+                }
             }
-            work.hold(&batch)?;
-            rows += batch.num_rows();
-            staging
-                .append_batch(&self.table, batch.schema(), PendingMode::Overwrite, batch)
-                .map_err(external)?;
-        }
-        let mut staging = Box::pin(staging.stage_all_with_concurrency(db, branch, 1));
+        });
+        let mut staging = Box::pin(staging.stage_overwrite_stream(
+            db,
+            &self.table,
+            chunks,
+            Arc::clone(&self.schema),
+        ));
         let staged = futures::future::poll_fn(|cx| {
             let _timer = baseline.elapsed_compute().timer();
             staging.as_mut().poll(cx)
         })
-        .await
-        .map_err(external)?;
+        .await;
+        drop(staging);
+        let staged = match staged {
+            Ok(staged) => staged,
+            Err(error) => {
+                let refused = input_failure
+                    .lock()
+                    .unwrap_or_else(std::sync::PoisonError::into_inner)
+                    .take();
+                return Err(refused.unwrap_or_else(|| external(error)));
+            }
+        };
+        let rows = rows.load(std::sync::atomic::Ordering::Relaxed);
         *self
             .write
             .staged
@@ -161,6 +227,110 @@ impl DataSink for LanceFragmentSink {
     }
 }
 
+/// One source partition's rows between two chunk cuts: the batches so far,
+/// their count, the chunk's index within the partition, and the memory
+/// they are charged to until their writer returns.
+struct OpenChunk {
+    batches: Vec<RecordBatch>,
+    rows: usize,
+    index: u64,
+    memory: WorkMemory,
+}
+
+/// Cuts the sink's input into [`OverwriteChunk`]s of exactly `chunk_rows`
+/// rows per source partition (the slice tag names it; the last chunk shorter),
+/// each batch bound-checked and rewrapped under the declared schema.
+struct Chunker {
+    schema: SchemaRef,
+    chunk_rows: usize,
+    max_input_bytes: u64,
+    work: WorkMemory,
+    open: BTreeMap<u32, OpenChunk>,
+}
+
+impl Chunker {
+    fn new(schema: SchemaRef, chunk_rows: usize, max_input_bytes: u64, work: WorkMemory) -> Self {
+        Self {
+            schema,
+            chunk_rows: chunk_rows.max(1),
+            max_input_bytes,
+            work,
+            open: BTreeMap::new(),
+        }
+    }
+
+    fn push(&mut self, batch: RecordBatch) -> crate::error::Result<Vec<OverwriteChunk>> {
+        let bytes = write_batch_bytes(&batch)?.rows;
+        if bytes > self.max_input_bytes {
+            return Err(OmniError::resource_limit(
+                STAGED_WRITE_INPUT_BYTES_RESOURCE,
+                self.max_input_bytes,
+                bytes,
+            ));
+        }
+        let partition = u32::try_from(slice_of(&batch) >> 32).unwrap_or(u32::MAX);
+        let mut batch = RecordBatch::try_new(Arc::clone(&self.schema), batch.columns().to_vec())
+            .map_err(OmniError::arrow_internal)?;
+        let mut cut = Vec::new();
+        if batch.num_rows() == 0 {
+            return Ok(cut);
+        }
+        let mut open = match self.open.remove(&partition) {
+            Some(open) => open,
+            None => OpenChunk {
+                batches: Vec::new(),
+                rows: 0,
+                index: 0,
+                memory: self
+                    .work
+                    .child("overwrite chunk")
+                    .map_err(OmniError::datafusion)?,
+            },
+        };
+        while open.rows + batch.num_rows() >= self.chunk_rows {
+            let fill = self.chunk_rows - open.rows;
+            let head = batch.slice(0, fill);
+            open.memory.hold(&head).map_err(OmniError::datafusion)?;
+            open.batches.push(head);
+            let next = OpenChunk {
+                batches: Vec::new(),
+                rows: 0,
+                index: open.index + 1,
+                memory: self
+                    .work
+                    .child("overwrite chunk")
+                    .map_err(OmniError::datafusion)?,
+            };
+            let full = std::mem::replace(&mut open, next);
+            cut.push(OverwriteChunk {
+                order: (partition, full.index),
+                batches: full.batches,
+                lease: Some(Box::new(full.memory)),
+            });
+            batch = batch.slice(fill, batch.num_rows() - fill);
+        }
+        if batch.num_rows() > 0 {
+            open.memory.hold(&batch).map_err(OmniError::datafusion)?;
+            open.rows += batch.num_rows();
+            open.batches.push(batch);
+        }
+        self.open.insert(partition, open);
+        Ok(cut)
+    }
+
+    fn flush(&mut self) -> Vec<OverwriteChunk> {
+        std::mem::take(&mut self.open)
+            .into_iter()
+            .filter(|(_, open)| open.rows > 0)
+            .map(|(partition, open)| OverwriteChunk {
+                order: (partition, open.index),
+                batches: open.batches,
+                lease: Some(Box::new(open.memory)),
+            })
+            .collect()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::Mutex;
@@ -290,6 +460,167 @@ mod tests {
         assert_eq!(person_version(&session).await, before);
     }
 
+    /// The sink over `input`, chunks cut every `chunk_rows` rows.
+    async fn write_all_chunked(
+        db: &Arc<Omnigraph>,
+        input: SendableRecordBatchStream,
+        chunk_rows: usize,
+    ) -> (
+        DfResult<u64>,
+        Arc<Mutex<Option<crate::exec::staging::StagedMutation>>>,
+    ) {
+        let schema = input.schema();
+        let txn = Arc::new(db.open_write_txn(None).await.unwrap());
+        let staged = Arc::new(Mutex::new(None));
+        let write = LoadWrite {
+            db: Arc::clone(db),
+            txn,
+            write_budget: WriteBudget::default(),
+            writers: 2,
+            staged: Arc::clone(&staged),
+        };
+        let sink = LanceFragmentSink::new("node:Person".to_string(), schema, write, u64::MAX);
+        let (_pool, ctx) = fixtures::context(64 << 20, 8192);
+        let written = crate::instrumentation::with_overwrite_chunk_rows(
+            chunk_rows,
+            sink.write_all(input, &ctx),
+        )
+        .await;
+        (written, staged)
+    }
+
+    /// Lance's data files under every `data/` directory below `dir`, the
+    /// committed fragments and whatever a writer left.
+    fn lance_data_files(dir: &std::path::Path) -> usize {
+        let Ok(entries) = std::fs::read_dir(dir) else {
+            return 0;
+        };
+        entries
+            .flatten()
+            .map(|entry| {
+                let path = entry.path();
+                if path.is_dir() {
+                    lance_data_files(&path)
+                } else {
+                    let under_data = path
+                        .parent()
+                        .and_then(std::path::Path::file_name)
+                        .is_some_and(|name| name == "data");
+                    usize::from(under_data && path.extension().is_some_and(|ext| ext == "lance"))
+                }
+            })
+            .sum()
+    }
+
+    /// A chunk the keyed-write fence refuses does not end the reading: the
+    /// sink follows its input to the verdict the validate operator parks and
+    /// returns that, the fence's own text only when the input ends clean.
+    #[tokio::test]
+    async fn a_fence_refusal_yields_to_the_inputs_own_verdict() {
+        let dir = tempfile::tempdir().unwrap();
+        let (db, session) = committed_graph(&dir).await;
+        let before_version = person_version(&session).await;
+        let before_files = lance_data_files(dir.path());
+        let csv = "name,email\nCarol,carol@x\nCarol,carol2@x\nDave,dave@x\nErin,erin@x\n";
+        let batches = person_batches(&session, csv);
+        let schema = batches[0].schema();
+        let judged = futures::stream::iter(batches.into_iter().map(Ok)).chain(
+            futures::stream::once(async {
+                Err(external(OmniError::manifest(
+                    "@unique violation on Person.name: value 'Carol'".to_string(),
+                )))
+            }),
+        );
+        let judged: SendableRecordBatchStream = Box::pin(
+            datafusion::physical_plan::stream::RecordBatchStreamAdapter::new(
+                Arc::clone(&schema),
+                judged,
+            ),
+        );
+        let (written, staged) = write_all_chunked(&db, judged, 2).await;
+        let error = written.unwrap_err().to_string();
+        assert!(
+            error.contains("@unique violation on Person.name: value 'Carol'"),
+            "{error}"
+        );
+        assert!(!error.contains("already has this id"), "{error}");
+        assert!(staged.lock().unwrap().is_none());
+        assert_eq!(lance_data_files(dir.path()), before_files);
+        assert_eq!(person_version(&session).await, before_version);
+
+        let batches = person_batches(&session, csv);
+        let unjudged: SendableRecordBatchStream = Box::pin(
+            datafusion::physical_plan::stream::RecordBatchStreamAdapter::new(
+                schema,
+                futures::stream::iter(batches.into_iter().map(Ok)),
+            ),
+        );
+        let (written, staged) = write_all_chunked(&db, unjudged, 2).await;
+        let error = written.unwrap_err().to_string();
+        assert!(error.contains("already has this id"), "{error}");
+        assert!(staged.lock().unwrap().is_none());
+        assert_eq!(lance_data_files(dir.path()), before_files);
+        assert_eq!(person_version(&session).await, before_version);
+    }
+
+    /// A refusal after chunks were written leaves `data/` as the commit left
+    /// it, nothing staged, the input's own error returned; the same rows clean
+    /// stage one file per chunk, so the refused run had chunks to delete.
+    #[tokio::test]
+    async fn a_refused_input_leaves_no_fragment_file_behind() {
+        let dir = tempfile::tempdir().unwrap();
+        let (db, session) = committed_graph(&dir).await;
+        let before_version = person_version(&session).await;
+        let before_files = lance_data_files(dir.path());
+        let csv =
+            "name,email\nCarol,carol@x\nDave,dave@x\nErin,erin@x\nFrank,frank@x\nGrace,grace@x\n";
+        let batches = person_batches(&session, csv);
+        let schema = batches[0].schema();
+        let refused = futures::stream::iter(batches.into_iter().map(Ok)).chain(
+            futures::stream::once(async {
+                Err(external(OmniError::manifest(
+                    "record 6 has an empty field".to_string(),
+                )))
+            }),
+        );
+        let refused: SendableRecordBatchStream = Box::pin(
+            datafusion::physical_plan::stream::RecordBatchStreamAdapter::new(
+                Arc::clone(&schema),
+                refused,
+            ),
+        );
+        let (written, staged) = write_all_chunked(&db, refused, 2).await;
+        let error = written.unwrap_err();
+        assert!(
+            error.to_string().contains("record 6 has an empty field"),
+            "{error}"
+        );
+        assert!(staged.lock().unwrap().is_none());
+        assert_eq!(
+            lance_data_files(dir.path()),
+            before_files,
+            "the refused run's chunk files were deleted"
+        );
+        assert_eq!(person_version(&session).await, before_version);
+
+        let batches = person_batches(&session, csv);
+        let clean: SendableRecordBatchStream = Box::pin(
+            datafusion::physical_plan::stream::RecordBatchStreamAdapter::new(
+                schema,
+                futures::stream::iter(batches.into_iter().map(Ok)),
+            ),
+        );
+        let (written, staged) = write_all_chunked(&db, clean, 2).await;
+        assert_eq!(written.unwrap(), 5);
+        assert!(staged.lock().unwrap().is_some());
+        assert_eq!(
+            lance_data_files(dir.path()),
+            before_files + 3,
+            "five rows in chunks of two stage three files, uncommitted"
+        );
+        assert_eq!(person_version(&session).await, before_version);
+    }
+
     /// One input over the plan's bound refuses with the loader's resource
     /// text before anything is staged.
     #[tokio::test]
diff --git a/crates/omnigraph/src/engine/operators/shape.rs b/crates/omnigraph/src/engine/operators/shape.rs
index f1a81ed3d..5a29c13c4 100644
--- a/crates/omnigraph/src/engine/operators/shape.rs
+++ b/crates/omnigraph/src/engine/operators/shape.rs
@@ -304,6 +304,26 @@ impl Streaming {
         )
     }
 
+    /// The produced stream of an operator that executes its inputs itself
+    /// (every partition of a load's source): `body` runs on the producer's
+    /// owned worker with the sender and sends each batch as it passes, so the
+    /// first output cannot wait for the input's end.
+    pub(super) fn drive<F>(
+        &self,
+        memory: Arc<WorkMemory>,
+        metrics: &ExecutionPlanMetricsSet,
+        body: impl FnOnce(Arc<WorkMemory>, BatchSender) -> F + Send + 'static,
+    ) -> SendableRecordBatchStream
+    where
+        F: Future<Output = DfResult<()>> + Send + 'static,
+    {
+        let schema = self.properties.eq_properties.schema().clone();
+        polled(
+            metrics,
+            producer_stream(schema, memory, Some(metrics), body),
+        )
+    }
+
     /// The produced stream of a per-batch operator: on the producer's owned
     /// worker the shape opens `body`, executes `input` (so a side the body
     /// collected or a filter it filled is in place first), and drives the
diff --git a/crates/omnigraph/src/engine/operators/validate.rs b/crates/omnigraph/src/engine/operators/validate.rs
index 3e89eb78b..ab4aead05 100644
--- a/crates/omnigraph/src/engine/operators/validate.rs
+++ b/crates/omnigraph/src/engine/operators/validate.rs
@@ -1,7 +1,9 @@
-//! A load's validation: a pipeline breaker that holds the whole delta from
-//! every partition of its input, restores source order by slice, judges
-//! every check of the plan in one pass over it, and streams the held batches
-//! on only when the delta is admitted.
+//! A load's validation over every partition of its input, every check of the
+//! plan extracted in one pass per batch. Judged per row (the planner's
+//! `validate_streams`), it sends each batch on as it passes and the judgment
+//! after the drain is the stream's last word; otherwise it is a pipeline
+//! breaker that holds the whole delta, restores source order by slice, and
+//! streams the held batches on only when the delta is admitted.
 
 use std::fmt;
 use std::sync::Arc;
@@ -24,7 +26,7 @@ use tokio::sync::watch;
 use super::csv_source::{CsvSourceExec, FailedSlice, slice_of};
 use super::external;
 use super::memory::WorkMemory;
-use super::shape::{Breaker, work};
+use super::shape::{BatchSender, Breaker, Streaming, work};
 use crate::db::{Omnigraph, Snapshot};
 use crate::error::OmniError;
 use crate::loader::LoadMode;
@@ -57,20 +59,50 @@ pub(crate) struct LoadJudge {
     pub(crate) catalog: Arc<Catalog>,
 }
 
+/// The shape the plan's cut declares for the validation, the same fact the
+/// planner's `validate_streams` reads off the node.
+enum ValidateShape {
+    Streaming(Streaming),
+    Breaker(Breaker),
+}
+
+impl ValidateShape {
+    fn new(schema: arrow_schema::SchemaRef, streams: bool) -> Self {
+        if streams {
+            Self::Streaming(Streaming::new(schema))
+        } else {
+            Self::Breaker(Breaker::new(schema))
+        }
+    }
+
+    fn properties(&self) -> &Arc<PlanProperties> {
+        match self {
+            Self::Streaming(shape) => shape.properties(),
+            Self::Breaker(shape) => shape.properties(),
+        }
+    }
+}
+
+/// The partitions' way of handing batches on: the shared sender of the
+/// streaming shape, each batch sent as it passes; nothing for the breaker,
+/// which holds them.
+type Forward = Option<Arc<tokio::sync::Mutex<BatchSender>>>;
+
 pub(crate) struct ValidateExec {
     input: Arc<dyn ExecutionPlan>,
     judge: Arc<LoadJudge>,
-    shape: Breaker,
+    shape: ValidateShape,
     metrics: ExecutionPlanMetricsSet,
 }
 
 impl ValidateExec {
-    pub(crate) fn new(input: Arc<dyn ExecutionPlan>, judge: LoadJudge) -> Self {
+    /// `streams` is the planner's `validate_streams` for the node's checks.
+    pub(crate) fn new(input: Arc<dyn ExecutionPlan>, judge: LoadJudge, streams: bool) -> Self {
         let schema = input.schema();
         Self {
             input,
             judge: Arc::new(judge),
-            shape: Breaker::new(schema),
+            shape: ValidateShape::new(schema, streams),
             metrics: ExecutionPlanMetricsSet::new(),
         }
     }
@@ -125,8 +157,9 @@ impl ExecutionPlan for ValidateExec {
     ) -> DfResult<Arc<dyn ExecutionPlan>> {
         assert_eq!(children.len(), 1, "ValidateExec has one child");
         let input = children.remove(0);
+        let streams = matches!(self.shape, ValidateShape::Streaming(_));
         Ok(Arc::new(Self {
-            shape: Breaker::new(input.schema()),
+            shape: ValidateShape::new(input.schema(), streams),
             input,
             judge: Arc::clone(&self.judge),
             metrics: ExecutionPlanMetricsSet::new(),
@@ -156,115 +189,181 @@ impl ExecutionPlan for ValidateExec {
             judge.mode == LoadMode::Overwrite,
             partitions,
         ));
-        Ok(self.shape.held(
-            memory,
-            &self.metrics,
-            move |memory| async move {
-                let admitted = admit_sets(&memory, &judge, &plan).await?;
-                let inputs = (0..partitions)
-                    .map(|partition| source.execute(partition, Arc::clone(&ctx)))
-                    .collect::<DfResult<Vec<_>>>()?;
-                let (publish, published) = watch::channel(None::<Arc<SweptSets>>);
-                let (drained, set_guards) = futures::join!(
-                    drain_and_sweep(
-                        inputs,
+        match &self.shape {
+            ValidateShape::Breaker(breaker) => Ok(breaker.held(
+                memory,
+                &self.metrics,
+                move |memory| async move {
+                    let drained = drain(
+                        partitions, ctx, &memory, &source, &schema, &plan, &judge, None,
+                    )
+                    .await?;
+                    let batches = drained.batches;
+                    judge_after_drain(
                         &memory,
-                        &source,
-                        &schema,
-                        &plan,
                         &judge,
-                        published
-                    ),
-                    load_sets(Arc::clone(&memory), Arc::clone(&judge), admitted, publish)
-                );
-                let (_retained_sets, transient) = set_guards?;
-                let (batches, partials, facts) = drained?;
-                drop(transient);
-                let rows: usize = batches.iter().map(RecordBatch::num_rows).sum();
-                let extracted = Extracted::prepare(&plan, partials);
-                {
-                    let charged = memory.child("load removed and retained rows")?;
-                    let mut admit_phase =
-                        |LoadPhase::RemovedInput: LoadPhase| -> crate::error::Result<Box<dyn Send>> {
-                            let phase_memory = memory
-                                .child("load removed input")
-                                .map_err(|error| memory.error(error))?;
-                            phase_memory
-                                .entries::<&str>(rows)
-                                .map_err(|error| memory.error(error))?;
-                            Ok(Box::new(phase_memory))
-                        };
-                    let mut admit_bytes = |bytes: usize| -> crate::error::Result<Box<dyn Send>> {
-                        let allocation = memory
-                            .child("load integer endpoint keys")
-                            .map_err(|error| memory.error(error))?;
-                        allocation
-                            .grow(bytes)
-                            .map_err(|error| memory.error(error))?;
-                        Ok(Box::new(allocation))
-                    };
-                    let mut admit_reuse = || {
-                        memory.admit_read_reuse().and_then(|admitted| {
-                            LoadReadReuse::new(&judge.db, &judge.snapshot, admitted)
-                        })
-                    };
-                    let mut deleted = DeletedIdBudget::new(judge.write_budget);
-                    let mut admit_removed = |id: &str| {
-                        deleted.retain(id)?;
-                        let reserve = || {
-                            charged.entries::<String>(1)?;
-                            charged.entries::<Expr>(2)?;
-                            charged.entries::<&String>(2)?;
-                            charged.string(id.len().saturating_mul(3))
-                        };
-                        reserve().map_err(|error| charged.error(error))
-                    };
-                    let mut admit_retained = |batch: &RecordBatch| {
-                        let reserve = || {
-                            charged.hold(batch)?;
-                            charged.entries::<(String, String, String)>(1)?;
-                            charged.grow(batch.get_array_memory_size().saturating_mul(2))
-                        };
-                        reserve().map_err(|error| charged.error(error))
-                    };
-                    judge_load_swept(
-                        &judge.table,
-                        &judge.checks,
-                        judge.removal,
-                        &judge.catalog,
-                        &judge.snapshot,
-                        judge.mode,
+                        &plan,
                         &batches,
-                        extracted,
-                        &mut LoadAdmission {
-                            phase: Some(&mut admit_phase),
-                            bytes: Some(&mut admit_bytes),
-                            reuse: Some(&mut admit_reuse),
-                            removed: &mut admit_removed,
-                            retained: &mut admit_retained,
-                        },
+                        drained.rows,
+                        drained.partials,
                     )
-                    .await
-                    .map_err(external)?;
-                }
-                release_holders(&memory, &plan).await?;
-                drop(facts);
-                memory.metric(VALIDATED_ROWS, rows);
-                Ok(batches)
-            },
-            move |batches, memory, sender| async move {
-                for batch in batches {
-                    sender.send(batch, Arc::clone(&memory)).await?;
-                }
-                Ok(())
+                    .await?;
+                    drop(drained.facts);
+                    Ok(batches)
+                },
+                move |batches, memory, sender| async move {
+                    for batch in batches {
+                        sender.send(batch, Arc::clone(&memory)).await?;
+                    }
+                    Ok(())
+                },
+            )),
+            ValidateShape::Streaming(streaming) => Ok(streaming.drive(
+                memory,
+                &self.metrics,
+                move |memory, sender| async move {
+                    let forward = Some(Arc::new(tokio::sync::Mutex::new(sender)));
+                    let drained = drain(
+                        partitions, ctx, &memory, &source, &schema, &plan, &judge, forward,
+                    )
+                    .await?;
+                    judge_after_drain(&memory, &judge, &plan, &[], drained.rows, drained.partials)
+                        .await?;
+                    drop(drained.facts);
+                    Ok(())
+                },
+            )),
+        }
+    }
+}
+
+/// What the drain left for the judgment: the held batches in source order
+/// (none when the partitions forwarded them), the rows that passed, each
+/// partition's facts and the memory that holds them.
+struct Drained {
+    batches: Vec<RecordBatch>,
+    rows: usize,
+    partials: Vec<SweepPartial>,
+    facts: Vec<WorkMemory>,
+}
+
+/// The swept sets admitted, then the source's `partitions` executed and every
+/// one drained beside the set load; the loaded sets' transient guards are
+/// released once the drain ends.
+#[allow(clippy::too_many_arguments)]
+async fn drain(
+    partitions: usize,
+    ctx: Arc<TaskContext>,
+    memory: &Arc<WorkMemory>,
+    source: &Arc<dyn ExecutionPlan>,
+    schema: &arrow_schema::SchemaRef,
+    plan: &Arc<SweepPlan>,
+    judge: &Arc<LoadJudge>,
+    forward: Forward,
+) -> DfResult<Drained> {
+    let admitted = admit_sets(memory, judge, plan).await?;
+    let inputs = (0..partitions)
+        .map(|partition| source.execute(partition, Arc::clone(&ctx)))
+        .collect::<DfResult<Vec<_>>>()?;
+    let (publish, published) = watch::channel(None::<Arc<SweptSets>>);
+    let (drained, set_guards) = futures::join!(
+        drain_and_sweep(
+            inputs, memory, source, schema, plan, judge, published, forward
+        ),
+        load_sets(Arc::clone(memory), Arc::clone(judge), admitted, publish)
+    );
+    let (_retained_sets, transient) = set_guards?;
+    let drained = drained?;
+    drop(transient);
+    Ok(drained)
+}
+
+/// The checks judged over what the sweep extracted and, for the breaker,
+/// the held batches; the unique holders freed after the verdict.
+async fn judge_after_drain(
+    memory: &Arc<WorkMemory>,
+    judge: &Arc<LoadJudge>,
+    plan: &Arc<SweepPlan>,
+    batches: &[RecordBatch],
+    rows: usize,
+    partials: Vec<SweepPartial>,
+) -> DfResult<()> {
+    let extracted = Extracted::prepare(plan, partials);
+    {
+        let charged = memory.child("load removed and retained rows")?;
+        let mut admit_phase =
+            |LoadPhase::RemovedInput: LoadPhase| -> crate::error::Result<Box<dyn Send>> {
+                let phase_memory = memory
+                    .child("load removed input")
+                    .map_err(|error| memory.error(error))?;
+                phase_memory
+                    .entries::<&str>(rows)
+                    .map_err(|error| memory.error(error))?;
+                Ok(Box::new(phase_memory))
+            };
+        let mut admit_bytes = |bytes: usize| -> crate::error::Result<Box<dyn Send>> {
+            let allocation = memory
+                .child("load integer endpoint keys")
+                .map_err(|error| memory.error(error))?;
+            allocation
+                .grow(bytes)
+                .map_err(|error| memory.error(error))?;
+            Ok(Box::new(allocation))
+        };
+        let mut admit_reuse = || {
+            memory
+                .admit_read_reuse()
+                .and_then(|admitted| LoadReadReuse::new(&judge.db, &judge.snapshot, admitted))
+        };
+        let mut deleted = DeletedIdBudget::new(judge.write_budget);
+        let mut admit_removed = |id: &str| {
+            deleted.retain(id)?;
+            let reserve = || {
+                charged.entries::<String>(1)?;
+                charged.entries::<Expr>(2)?;
+                charged.entries::<&String>(2)?;
+                charged.string(id.len().saturating_mul(3))
+            };
+            reserve().map_err(|error| charged.error(error))
+        };
+        let mut admit_retained = |batch: &RecordBatch| {
+            let reserve = || {
+                charged.hold(batch)?;
+                charged.entries::<(String, String, String)>(1)?;
+                charged.grow(batch.get_array_memory_size().saturating_mul(2))
+            };
+            reserve().map_err(|error| charged.error(error))
+        };
+        judge_load_swept(
+            &judge.table,
+            &judge.checks,
+            judge.removal,
+            &judge.catalog,
+            &judge.snapshot,
+            judge.mode,
+            batches,
+            rows,
+            extracted,
+            &mut LoadAdmission {
+                phase: Some(&mut admit_phase),
+                bytes: Some(&mut admit_bytes),
+                reuse: Some(&mut admit_reuse),
+                removed: &mut admit_removed,
+                retained: &mut admit_retained,
             },
-        ))
+        )
+        .await
+        .map_err(external)?;
     }
+    release_holders(memory, plan).await?;
+    memory.metric(VALIDATED_ROWS, rows);
+    Ok(())
 }
 
 /// Every input partition drained and swept at once ([`sweep_partition`]), the
 /// held batches sorted back into slice order under `schema`; the refusal is a
 /// sliceless failure first, then the best [`FailedSlice::rank`], any partition.
+#[allow(clippy::too_many_arguments)]
 async fn drain_and_sweep(
     inputs: Vec<SendableRecordBatchStream>,
     memory: &Arc<WorkMemory>,
@@ -273,26 +372,31 @@ async fn drain_and_sweep(
     plan: &Arc<SweepPlan>,
     judge: &Arc<LoadJudge>,
     published: PublishedSets,
-) -> DfResult<(Vec<RecordBatch>, Vec<SweepPartial>, Vec<WorkMemory>)> {
+    forward: Forward,
+) -> DfResult<Drained> {
     let drained = futures::future::join_all(inputs.into_iter().map(|input| {
         let plan = Arc::clone(plan);
         let judge = Arc::clone(judge);
         let source = Arc::clone(source);
         let published = published.clone();
-        memory
-            .blocking(move |memory| sweep_partition(input, memory, source, plan, judge, published))
+        let forward = forward.clone();
+        memory.blocking(move |memory| {
+            sweep_partition(input, memory, source, plan, judge, published, forward)
+        })
     }))
     .await;
     let mut batches = Vec::new();
+    let mut rows = 0;
     let mut partials = Vec::new();
     let mut facts = Vec::new();
     let mut failures = Vec::new();
     for (partition, result) in drained.into_iter().enumerate() {
         match result {
-            Ok((held, partial, held_facts)) => {
-                batches.extend(held);
-                partials.push(partial);
-                facts.push(held_facts);
+            Ok(swept) => {
+                batches.extend(swept.held);
+                rows += swept.rows;
+                partials.push(swept.partial);
+                facts.push(swept.facts);
             }
             Err(error) => failures.push((partition, error)),
         }
@@ -330,7 +434,12 @@ async fn drain_and_sweep(
                 .map_err(|error| external(OmniError::arrow_internal(error)))
         })
         .collect::<DfResult<Vec<_>>>()?;
-    Ok((batches, partials, facts))
+    Ok(Drained {
+        batches,
+        rows,
+        partials,
+        facts,
+    })
 }
 
 /// The swept endpoint checks' key sets admitted on a worker before the source
@@ -390,9 +499,18 @@ async fn load_sets(
     Ok((retained, transient))
 }
 
-/// One partition's input held and swept as it arrives: every batch is held
-/// under `memory`, its facts extracted at its position, probed against the
-/// swept sets once they arrive; the unique holders sized once `source` can.
+/// One partition's drain: what it held (nothing when it forwarded), the rows
+/// it saw, its facts and the memory that holds them.
+struct Swept {
+    held: Vec<RecordBatch>,
+    rows: usize,
+    partial: SweepPartial,
+    facts: WorkMemory,
+}
+
+/// One partition's input swept as it arrives (facts at its position, probed
+/// once the swept sets land), then held under `memory` or sent on through
+/// `forward` once probed, a batch from before the sets waiting in order.
 async fn sweep_partition(
     mut input: SendableRecordBatchStream,
     memory: Arc<WorkMemory>,
@@ -400,10 +518,13 @@ async fn sweep_partition(
     plan: Arc<SweepPlan>,
     judge: Arc<LoadJudge>,
     mut published: PublishedSets,
-) -> DfResult<(Vec<RecordBatch>, SweepPartial, WorkMemory)> {
+    forward: Forward,
+) -> DfResult<Swept> {
     let facts = memory.child("load validation facts")?;
+    let waiting = memory.child("load batches before the sets")?;
     let mut partial = SweepPartial::new(&plan);
     let mut held = Vec::new();
+    let mut rows = 0;
     let mut slice = usize::MAX;
     let mut offset = 0_u64;
     let system_columns = judge.catalog.system_columns;
@@ -414,8 +535,11 @@ async fn sweep_partition(
     while let Some(batch) = input.next().await {
         let batch = batch?;
         memory.metric("input_rows", batch.num_rows());
-        memory.hold(&batch)?;
-        memory.entries::<RecordBatch>(1)?;
+        rows += batch.num_rows();
+        if forward.is_none() {
+            memory.hold(&batch)?;
+            memory.entries::<RecordBatch>(1)?;
+        }
         let tagged = slice_of(&batch);
         if tagged != slice {
             slice = tagged;
@@ -429,26 +553,51 @@ async fn sweep_partition(
         partial
             .observe(&plan, &judge.catalog, &batch, system_columns, base)
             .map_err(external)?;
+        offset += batch.num_rows() as u64;
         if plan.sweeps_endpoints() {
             if sets.is_none() {
                 sets = published.borrow().clone();
             }
             match &sets {
-                Some(sets) => probe_batch(
-                    &facts,
-                    &mut partial,
-                    &plan,
-                    &batch,
-                    system_columns,
-                    base,
-                    sets,
-                    &mut scratch_rows,
-                )?,
-                None => unprobed.push((held.len(), base)),
+                Some(sets) => {
+                    if !unprobed.is_empty() {
+                        probe_held(
+                            &facts,
+                            &mut partial,
+                            &plan,
+                            &held,
+                            &mut unprobed,
+                            system_columns,
+                            sets,
+                            &mut scratch_rows,
+                        )?;
+                        send_held(&forward, &memory, &waiting, &mut held).await?;
+                    }
+                    probe_batch(
+                        &facts,
+                        &mut partial,
+                        &plan,
+                        &batch,
+                        system_columns,
+                        base,
+                        sets,
+                        &mut scratch_rows,
+                    )?;
+                }
+                None => {
+                    if forward.is_some() {
+                        waiting.hold(&batch)?;
+                    }
+                    unprobed.push((held.len(), base));
+                    held.push(batch);
+                    continue;
+                }
             }
         }
-        offset += batch.num_rows() as u64;
-        held.push(batch);
+        match &forward {
+            Some(sender) => sender.lock().await.send(batch, Arc::clone(&memory)).await?,
+            None => held.push(batch),
+        }
     }
     if !unprobed.is_empty() {
         let sets = match sets {
@@ -464,20 +613,70 @@ async fn sweep_partition(
                 })?;
             },
         };
-        for (index, base) in unprobed {
-            probe_batch(
-                &facts,
-                &mut partial,
-                &plan,
-                &held[index],
-                system_columns,
-                base,
-                &sets,
-                &mut scratch_rows,
-            )?;
-        }
+        probe_held(
+            &facts,
+            &mut partial,
+            &plan,
+            &held,
+            &mut unprobed,
+            system_columns,
+            &sets,
+            &mut scratch_rows,
+        )?;
+        send_held(&forward, &memory, &waiting, &mut held).await?;
+    }
+    Ok(Swept {
+        held,
+        rows,
+        partial,
+        facts,
+    })
+}
+
+/// Probe the held batches that arrived before the sets, in order.
+#[allow(clippy::too_many_arguments)]
+fn probe_held(
+    facts: &WorkMemory,
+    partial: &mut SweepPartial,
+    plan: &SweepPlan,
+    held: &[RecordBatch],
+    unprobed: &mut Vec<(usize, u64)>,
+    system_columns: SystemColumns,
+    sets: &SweptSets,
+    scratch_rows: &mut usize,
+) -> DfResult<()> {
+    for (index, base) in unprobed.drain(..) {
+        probe_batch(
+            facts,
+            partial,
+            plan,
+            &held[index],
+            system_columns,
+            base,
+            sets,
+            scratch_rows,
+        )?;
     }
-    Ok((held, partial, facts))
+    Ok(())
+}
+
+/// Send what a forwarding partition held while it waited for the sets, in
+/// order, and release `waiting`'s charge for them; a holding partition keeps
+/// them.
+async fn send_held(
+    forward: &Forward,
+    memory: &Arc<WorkMemory>,
+    waiting: &WorkMemory,
+    held: &mut Vec<RecordBatch>,
+) -> DfResult<()> {
+    let Some(sender) = forward else {
+        return Ok(());
+    };
+    for batch in held.drain(..) {
+        sender.lock().await.send(batch, Arc::clone(memory)).await?;
+    }
+    waiting.release_work();
+    Ok(())
 }
 
 /// The shared unique holders sized from the source's row estimate once it has
@@ -806,6 +1005,17 @@ mod tests {
         schema: &str,
         source: &str,
         partitions: usize,
+    ) -> (Arc<CsvSourceExec>, DfResult<Vec<RecordBatch>>) {
+        validate_as(schema, source, partitions, false).await
+    }
+
+    /// [`validate`] with the shape chosen: the breaker under a removal scan,
+    /// the streaming operator under `removal: None`.
+    async fn validate_as(
+        schema: &str,
+        source: &str,
+        partitions: usize,
+        streams: bool,
     ) -> (Arc<CsvSourceExec>, DfResult<Vec<RecordBatch>>) {
         let dir = tempfile::tempdir().unwrap();
         let db = crate::Session::from_defaults(
@@ -852,13 +1062,18 @@ mod tests {
             db: Arc::clone(db.db()),
             table: "node:Person".to_string(),
             checks,
-            removal: ValidateRemoval::Scan,
+            removal: if streams {
+                ValidateRemoval::None
+            } else {
+                ValidateRemoval::Scan
+            },
             mode: LoadMode::Overwrite,
             write_budget: WriteBudget::from_settings(db.settings()),
             snapshot: db.snapshot().await,
             catalog,
         };
-        let validate = ValidateExec::new(Arc::clone(&exec) as Arc<dyn ExecutionPlan>, judge);
+        let validate =
+            ValidateExec::new(Arc::clone(&exec) as Arc<dyn ExecutionPlan>, judge, streams);
         let (_pool, ctx) = fixtures::context(64 << 20, CSV_READ_BATCH_ROWS);
         let judged = validate
             .execute(0, ctx)
@@ -886,6 +1101,40 @@ mod tests {
             .collect()
     }
 
+    /// The streaming shape sends each partition's batches on as they pass,
+    /// slice tag kept for the sink's ordering: every row once, each partition
+    /// in source order; a case sees the loaded rows, never the batches.
+    #[tokio::test]
+    async fn partitions_stream_their_rows_tagged_by_slice() {
+        let mut source = String::from("key,score\n");
+        for row in 0..1000 {
+            writeln!(&mut source, "{row},1.5").unwrap();
+        }
+        let (exec, streamed) = validate_as(PERSON, &source, 3, true).await;
+        let batches = streamed.unwrap();
+        assert!(
+            batches
+                .iter()
+                .all(|batch| batch.schema().metadata().contains_key(CSV_SLICE_METADATA))
+        );
+        let mut per_partition: std::collections::BTreeMap<usize, Vec<i64>> =
+            std::collections::BTreeMap::new();
+        for batch in &batches {
+            per_partition
+                .entry(slice_of(batch) >> 32)
+                .or_default()
+                .extend(keys(std::slice::from_ref(batch)));
+        }
+        assert_eq!(per_partition.len(), 3, "{per_partition:?}");
+        for rows in per_partition.values() {
+            assert!(rows.windows(2).all(|pair| pair[0] < pair[1]), "{rows:?}");
+        }
+        let mut all: Vec<i64> = per_partition.into_values().flatten().collect();
+        all.sort_unstable();
+        assert_eq!(all, (0..1000).collect::<Vec<i64>>());
+        assert!(exec.failed_slices().is_empty());
+    }
+
     /// Whatever partition claimed each slice, the breaker hands the held rows
     /// on in source order under the plain schema; a case sees the loaded
     /// rows, never the batches or the claim.
@@ -1233,6 +1482,7 @@ mod tests {
                                 plan,
                                 judge,
                                 published,
+                                None,
                             )
                         })
                         .await
@@ -1250,10 +1500,10 @@ mod tests {
                 sweep(earlier_slice_missing_src_and_dst_at_row_2.clone(), ended_b),
                 publisher
             );
-            let (held_a, partial_a, _facts_a) = a.unwrap();
-            let (held_b, partial_b, _facts_b) = b.unwrap();
-            let extracted = Extracted::prepare(&plan, vec![partial_a, partial_b]);
-            let held: Vec<RecordBatch> = held_b.into_iter().chain(held_a).collect();
+            let a = a.unwrap();
+            let b = b.unwrap();
+            let extracted = Extracted::prepare(&plan, vec![a.partial, b.partial]);
+            let held: Vec<RecordBatch> = b.held.into_iter().chain(a.held).collect();
             let refused = judge_load_swept(
                 "edge:Knows",
                 &checks,
@@ -1262,6 +1512,7 @@ mod tests {
                 &snapshot,
                 LoadMode::Overwrite,
                 &held,
+                a.rows + b.rows,
                 extracted,
                 &mut LoadAdmission {
                     phase: None,
@@ -1310,6 +1561,7 @@ mod tests {
         let validate = ValidateExec::new(
             Arc::clone(&exec) as Arc<dyn ExecutionPlan>,
             knows_judge(&db, &checks, &snapshot),
+            false,
         );
         let (_pool, ctx) = fixtures::context(whole_set - 1, CSV_READ_BATCH_ROWS);
         let refused = validate
@@ -1444,6 +1696,7 @@ mod tests {
         let validate = ValidateExec::new(
             Arc::clone(&exec) as Arc<dyn ExecutionPlan>,
             knows_judge(&db, &checks, &snapshot),
+            false,
         );
         let ledger = Arc::new(ChargeLedger::new(64 << 20));
         let ctx = fixtures::context_over(
diff --git a/crates/omnigraph/src/exec/staging.rs b/crates/omnigraph/src/exec/staging.rs
index 1e7e4e2ed..e6724063e 100644
--- a/crates/omnigraph/src/exec/staging.rs
+++ b/crates/omnigraph/src/exec/staging.rs
@@ -39,7 +39,7 @@ use omnigraph_compiler::SystemColumns;
 use crate::db::{DatasetUpdate, MutationOpKind};
 use crate::error::{OmniError, Result};
 use crate::seams::{catalog, decide_seam, fail};
-use crate::table_store::ExternalBlobUriCollector;
+use crate::table_store::{ExternalBlobUriCollector, OverwriteChunk};
 
 /// Logical semantics for one accumulated table write.
 ///
@@ -683,24 +683,7 @@ impl MutationStaging {
             }
         }
 
-        let expected_versions = expected_versions
-            .iter()
-            .map(|(table_key, table_version)| {
-                let path = paths.get(table_key).ok_or_else(|| {
-                    OmniError::manifest_internal(format!(
-                        "MutationStaging::stage_all: missing path for table '{table_key}'"
-                    ))
-                })?;
-                Ok((
-                    path.identity,
-                    crate::db::manifest::TableVersionExpectation {
-                        table_key: table_key.clone(),
-                        table_version: *table_version,
-                        native_ref: path.pinned_native_ref.clone(),
-                    },
-                ))
-            })
-            .collect::<Result<crate::db::manifest::ExpectedTableVersions>>()?;
+        let expected_versions = expected_table_versions(&expected_versions, &paths)?;
 
         Ok(StagedMutation {
             staged: staged_entries,
@@ -709,6 +692,127 @@ impl MutationStaging {
     }
 }
 
+impl MutationStaging {
+    /// Stage one table's overwrite from chunks as the load's sink cuts them,
+    /// the table registered by [`Self::ensure_path`] and nothing else
+    /// pending: every batch passes the keyed-write contract and its external
+    /// Blob references are authorized per chunk before that chunk's writer
+    /// starts, then `TableStorage::stage_overwrite_stream` writes it. The
+    /// result is the [`StagedMutation`] `stage_all` would build for the table.
+    pub(crate) async fn stage_overwrite_stream<S>(
+        self,
+        db: &crate::db::Omnigraph,
+        table_key: &str,
+        chunks: S,
+        schema: SchemaRef,
+    ) -> Result<StagedMutation>
+    where
+        S: futures::Stream<Item = Result<OverwriteChunk>> + Send,
+    {
+        let MutationStaging {
+            expected_versions,
+            paths,
+            pending,
+            write_budget,
+            overwrite_writers,
+            ..
+        } = self;
+        if !pending.is_empty() {
+            return Err(OmniError::manifest_internal(
+                "MutationStaging::stage_overwrite_stream: the staging also holds pending batches",
+            ));
+        }
+        let path = paths.get(table_key).cloned().ok_or_else(|| {
+            OmniError::manifest_internal(format!(
+                "MutationStaging::stage_overwrite_stream: missing path for table '{table_key}'"
+            ))
+        })?;
+        let expected = *expected_versions.get(table_key).ok_or_else(|| {
+            OmniError::manifest_internal(format!(
+                "MutationStaging::stage_overwrite_stream: missing expected version for table '{table_key}'"
+            ))
+        })?;
+        let ds = db
+            .open_pinned_for_write(&path.full_path, &path.entry)
+            .await?;
+        let system_columns = db.catalog().system_columns;
+        let prepared = chunks.then(|chunk| async move {
+            let mut chunk = chunk?;
+            let mut uris = ExternalBlobUriCollector::default();
+            for batch in &chunk.batches {
+                db.storage()
+                    .validate_keyed_write_batch(table_key, batch, system_columns)?;
+                uris.include_batch(batch)?;
+            }
+            if !uris.as_slice().is_empty() {
+                let preflight = db
+                    .storage()
+                    .preflight_external_blob_uris(uris.as_slice())
+                    .await?;
+                for batch in chunk.batches.iter_mut() {
+                    *batch = db
+                        .storage()
+                        .prepare_overwrite_blob_references_with_preflight(
+                            table_key,
+                            batch.clone(),
+                            &preflight,
+                            system_columns,
+                        )?;
+                }
+            }
+            Ok(chunk)
+        });
+        let staged = db
+            .storage()
+            .stage_overwrite_stream(
+                &ds,
+                Box::pin(prepared),
+                Some(schema),
+                write_budget,
+                overwrite_writers,
+            )
+            .await?;
+        let staged_entries = vec![StagedTableEntry {
+            table_key: table_key.to_string(),
+            path,
+            expected_version: expected,
+            dataset: ds,
+            staged_write: Some(staged),
+        }];
+        let expected_versions = expected_table_versions(&expected_versions, &paths)?;
+        Ok(StagedMutation {
+            staged: staged_entries,
+            expected_versions,
+        })
+    }
+}
+
+/// The publisher's CAS fence from the staging's pins: every touched table's
+/// identity with the version and native ref it was read at.
+fn expected_table_versions(
+    expected_versions: &HashMap<String, u64>,
+    paths: &HashMap<String, StagedTablePath>,
+) -> Result<crate::db::manifest::ExpectedTableVersions> {
+    expected_versions
+        .iter()
+        .map(|(table_key, table_version)| {
+            let path = paths.get(table_key).ok_or_else(|| {
+                OmniError::manifest_internal(format!(
+                    "MutationStaging::stage_all: missing path for table '{table_key}'"
+                ))
+            })?;
+            Ok((
+                path.identity,
+                crate::db::manifest::TableVersionExpectation {
+                    table_key: table_key.clone(),
+                    table_version: *table_version,
+                    native_ref: path.pinned_native_ref.clone(),
+                },
+            ))
+        })
+        .collect()
+}
+
 /// One table's input as its Lance writer consumes it: a keyed adapter takes
 /// one batch, an overwrite takes the batches as they were appended.
 enum PreparedPendingTable {
diff --git a/crates/omnigraph/src/instrumentation.rs b/crates/omnigraph/src/instrumentation.rs
index 1e9a3c6bc..0b0bdc8a3 100644
--- a/crates/omnigraph/src/instrumentation.rs
+++ b/crates/omnigraph/src/instrumentation.rs
@@ -9,8 +9,8 @@ pub use omnigraph_core::instrumentation::{
     QueryMemoryProbes, RrfGateFallback, RrfGatePlan, RrfGateVerdict, StageWriteProbes,
     StorageReadCounts, capture_query_io_probes, with_csv_slice_bytes,
     with_external_payload_read_held, with_external_payload_read_stalled, with_merge_write_probes,
-    with_query_io_probes, with_query_memory_limit, with_query_memory_probes,
-    with_stage_write_probes,
+    with_overwrite_chunk_rows, with_query_io_probes, with_query_memory_limit,
+    with_query_memory_probes, with_stage_write_probes,
 };
 
 // Keep this list sorted. Benchmark admission independently derives the same
diff --git a/crates/omnigraph/src/storage_layer.rs b/crates/omnigraph/src/storage_layer.rs
index 9cb5cb3c3..640fb98cc 100644
--- a/crates/omnigraph/src/storage_layer.rs
+++ b/crates/omnigraph/src/storage_layer.rs
@@ -56,8 +56,8 @@ use omnigraph_compiler::SystemColumns;
 use crate::db::{DatasetEntry, Snapshot};
 use crate::error::{OmniError, Result};
 use crate::table_store::{
-    ExternalBlobPreflight, ID_SCAN_BATCH_BYTES, ID_SCAN_BATCH_ROWS, StagedTransactionIdentity,
-    StagedWrite, TableState, TableStore,
+    ExternalBlobPreflight, ID_SCAN_BATCH_BYTES, ID_SCAN_BATCH_ROWS, OverwriteChunk,
+    StagedTransactionIdentity, StagedWrite, TableState, TableStore,
 };
 
 /// One fenced merge chunk is bounded in both rows and materialized Arrow
@@ -841,6 +841,18 @@ pub trait TableStorage: sealed::Sealed + Send + Sync + Debug {
         writers: usize,
     ) -> Result<StagedHandle>;
 
+    /// Stage an overwrite from chunks as a producer cuts them, each written
+    /// as it arrives, the fragments joined in source order at the end; a
+    /// refusal deletes what was written. See `TableStore::stage_overwrite_stream`.
+    async fn stage_overwrite_stream(
+        &self,
+        snapshot: &SnapshotHandle,
+        chunks: futures::stream::BoxStream<'_, Result<OverwriteChunk>>,
+        schema: Option<SchemaRef>,
+        write_budget: WriteBudget,
+        writers: usize,
+    ) -> Result<StagedHandle>;
+
     /// Stage a rename-only column alteration; see
     /// `TableStore::stage_rename_columns`. Committed through `commit_staged_exact`.
     async fn stage_rename_columns(
@@ -1425,6 +1437,26 @@ impl TableStorage for TableStore {
         .map(StagedHandle::new)
     }
 
+    async fn stage_overwrite_stream(
+        &self,
+        snapshot: &SnapshotHandle,
+        chunks: futures::stream::BoxStream<'_, Result<OverwriteChunk>>,
+        schema: Option<SchemaRef>,
+        write_budget: WriteBudget,
+        writers: usize,
+    ) -> Result<StagedHandle> {
+        TableStore::stage_overwrite_stream(
+            self,
+            snapshot.dataset(),
+            chunks,
+            schema,
+            write_budget,
+            writers,
+        )
+        .await
+        .map(StagedHandle::new)
+    }
+
     async fn stage_rename_columns(
         &self,
         snapshot: &SnapshotHandle,
diff --git a/crates/omnigraph/src/table_store.rs b/crates/omnigraph/src/table_store.rs
index dad9e5fb0..a36b5da14 100644
--- a/crates/omnigraph/src/table_store.rs
+++ b/crates/omnigraph/src/table_store.rs
@@ -4804,10 +4804,6 @@ impl TableStore {
         writers: usize,
         chunk_rows: usize,
     ) -> Result<StagedWrite> {
-        // Existing graph datasets retain stable row IDs through Overwrite;
-        // keep the flag explicit at this write site too. This is not the
-        // separate Lance migration API for legacy datasets without stable IDs.
-        // See `stage_overwrite_preserves_stable_row_ids` and docs/dev/lance.md.
         let schema = batches.first().map(RecordBatch::schema);
         let mut filled = batches.iter().filter(|batch| batch.num_rows() > 0);
         if let Some(first) = filled.next()
@@ -4818,84 +4814,182 @@ impl TableStore {
                 first_schema_difference(&first.schema(), &other.schema())
             )));
         }
-        let chunks = overwrite_chunks(batches, chunk_rows);
-        let (transaction, mut new_fragments) = if chunks.is_empty() {
-            let schema = match schema {
-                Some(schema) => {
-                    LanceSchema::try_from(schema.as_ref()).map_err(OmniError::lance_internal)?
+        let chunks = overwrite_chunks(batches, chunk_rows)
+            .into_iter()
+            .enumerate()
+            .map(|(index, batches)| {
+                Ok(OverwriteChunk {
+                    order: (0, index as u64),
+                    batches,
+                    lease: None,
+                })
+            });
+        self.stage_overwrite_stream(
+            ds,
+            futures::stream::iter(chunks),
+            schema,
+            write_budget,
+            writers,
+        )
+        .await
+    }
+
+    /// [`Self::stage_overwrite_batches`] over chunks as a producer cuts them:
+    /// each chunk goes to a Lance writer as it arrives, `writers` in flight,
+    /// and pulling pauses while they are; the finished calls' fragments are
+    /// joined in `order` into one `Operation::Overwrite`, whatever order the
+    /// writers returned in. A refusal from a writer waits for the writers in
+    /// flight, deletes every file the finished calls wrote, and is returned; a
+    /// producer's refusal first follows the producer to its end, dropping every
+    /// later chunk unwritten, so the operator that owns the refused row parks
+    /// its own verdict for the caller before the same deletion and return. A
+    /// drop mid-stream deletes them off the runtime. `schema` is the
+    /// transaction's when no chunk carries a row.
+    pub(crate) async fn stage_overwrite_stream(
+        &self,
+        ds: &Dataset,
+        chunks: impl futures::Stream<Item = Result<OverwriteChunk>> + Send,
+        schema: Option<SchemaRef>,
+        write_budget: WriteBudget,
+        writers: usize,
+    ) -> Result<StagedWrite> {
+        let params = Arc::new(WriteParams {
+            mode: WriteMode::Overwrite,
+            file_writer_options: Some(budgeted_file_writer_options(write_budget)),
+            enable_stable_row_ids: true,
+            allow_external_blob_outside_bases: true,
+            auto_cleanup: None,
+            skip_auto_cleanup: true,
+            ..Default::default()
+        });
+        let writers = writers.max(1);
+        let ds = Arc::new(ds.clone());
+        let mut written = WrittenChunks::new(Arc::clone(&ds));
+        let mut in_flight = futures::stream::FuturesUnordered::new();
+        let mut chunks = std::pin::pin!(chunks);
+        let mut first_schema: Option<SchemaRef> = None;
+        let mut failure: Option<OmniError> = None;
+        let mut producer_refused = false;
+        while failure.is_none() {
+            while in_flight.len() >= writers {
+                match in_flight.next().await {
+                    Some(Ok(finished)) => written.push(finished),
+                    Some(Err(error)) => {
+                        failure = Some(error);
+                        break;
+                    }
+                    None => break,
                 }
-                None => ds.schema().clone(),
+            }
+            if failure.is_some() {
+                break;
+            }
+            let chunk = match chunks.next().await {
+                Some(Ok(chunk)) => chunk,
+                Some(Err(error)) => {
+                    failure = Some(error);
+                    producer_refused = true;
+                    break;
+                }
+                None => break,
             };
-            let transaction = TransactionBuilder::new(
-                ds.manifest.version,
-                Operation::Overwrite {
-                    fragments: Vec::new(),
-                    schema,
-                    config_upsert_values: None,
-                    initial_bases: None,
-                },
-            )
-            .build();
-            (transaction, Vec::new())
-        } else {
-            let params = WriteParams {
-                mode: WriteMode::Overwrite,
-                file_writer_options: Some(budgeted_file_writer_options(write_budget)),
-                enable_stable_row_ids: true,
-                allow_external_blob_outside_bases: true,
-                auto_cleanup: None,
-                skip_auto_cleanup: true,
-                ..Default::default()
+            let OverwriteChunk {
+                order,
+                mut batches,
+                lease,
+            } = chunk;
+            batches.retain(|batch| batch.num_rows() > 0);
+            let Some(first) = batches.first() else {
+                continue;
             };
-            let writers = writers.clamp(1, chunks.len());
-            let ds = Arc::new(ds.clone());
-            let transactions = futures::stream::iter(chunks)
-                .map(|chunk| {
-                    let ds = Arc::clone(&ds);
-                    let params = &params;
-                    async move {
-                        InsertBuilder::new(ds)
-                            .with_params(params)
-                            .execute_uncommitted(chunk)
-                            .await
-                            .map_err(|error| OmniError::storage(error).unpublished())
-                    }
-                })
-                .buffered(writers)
-                .try_collect::<Vec<Transaction>>()
-                .await?;
-            let mut transactions = transactions.into_iter();
-            let mut transaction = transactions.next().ok_or_else(|| {
-                OmniError::manifest_internal("stage_overwrite_batches: no chunk was written")
-            })?;
-            let fragments = match &mut transaction.operation {
-                Operation::Overwrite { fragments, .. } => fragments,
-                other => {
-                    return Err(OmniError::manifest_internal(format!(
-                        "stage_overwrite_batches: unexpected Lance operation {:?}",
-                        std::mem::discriminant(other)
-                    )));
+            let seen = first_schema.get_or_insert_with(|| first.schema());
+            if let Some(other) = batches.iter().find(|batch| batch.schema() != *seen) {
+                failure = Some(OmniError::manifest_internal(format!(
+                    "stage_overwrite_batches: all batches must share one schema, {}",
+                    first_schema_difference(seen, &other.schema())
+                )));
+                break;
+            }
+            let ds = Arc::clone(&ds);
+            let params = Arc::clone(&params);
+            in_flight.push(async move {
+                let transaction = InsertBuilder::new(ds)
+                    .with_params(&params)
+                    .execute_uncommitted(batches)
+                    .await
+                    .map_err(|error| OmniError::storage(error).unpublished())?;
+                drop(lease);
+                Ok::<_, OmniError>((order, transaction))
+            });
+        }
+        while let Some(result) = in_flight.next().await {
+            match result {
+                Ok(finished) => written.push(finished),
+                Err(error) => {
+                    failure.get_or_insert(error);
                 }
-            };
-            for later in transactions {
-                match later.operation {
+            }
+        }
+        if let Some(error) = failure {
+            if producer_refused {
+                while chunks.next().await.is_some() {}
+            }
+            written.discard().await;
+            return Err(error);
+        }
+        let mut transactions = written.take();
+        transactions.sort_by_key(|(order, _)| *order);
+        let mut transactions = transactions.into_iter().map(|(_, transaction)| transaction);
+        let (transaction, mut new_fragments) = match transactions.next() {
+            None => {
+                let schema = match schema.or(first_schema) {
+                    Some(schema) => {
+                        LanceSchema::try_from(schema.as_ref()).map_err(OmniError::lance_internal)?
+                    }
+                    None => ds.schema().clone(),
+                };
+                let transaction = TransactionBuilder::new(
+                    ds.manifest.version,
                     Operation::Overwrite {
-                        fragments: chunk_fragments,
-                        ..
-                    } => fragments.extend(chunk_fragments),
+                        fragments: Vec::new(),
+                        schema,
+                        config_upsert_values: None,
+                        initial_bases: None,
+                    },
+                )
+                .build();
+                (transaction, Vec::new())
+            }
+            Some(mut transaction) => {
+                let fragments = match &mut transaction.operation {
+                    Operation::Overwrite { fragments, .. } => fragments,
                     other => {
                         return Err(OmniError::manifest_internal(format!(
                             "stage_overwrite_batches: unexpected Lance operation {:?}",
-                            std::mem::discriminant(&other)
+                            std::mem::discriminant(other)
                         )));
                     }
+                };
+                for later in transactions {
+                    match later.operation {
+                        Operation::Overwrite {
+                            fragments: chunk_fragments,
+                            ..
+                        } => fragments.extend(chunk_fragments),
+                        other => {
+                            return Err(OmniError::manifest_internal(format!(
+                                "stage_overwrite_batches: unexpected Lance operation {:?}",
+                                std::mem::discriminant(&other)
+                            )));
+                        }
+                    }
                 }
+                let mut new_fragments = fragments.clone();
+                new_fragments
+                    .iter_mut()
+                    .for_each(|fragment| fragment.id = 0);
+                (transaction, new_fragments)
             }
-            let mut new_fragments = fragments.clone();
-            new_fragments
-                .iter_mut()
-                .for_each(|fragment| fragment.id = 0);
-            (transaction, new_fragments)
         };
         // These IDs belong only to the pre-commit scan view. All committed
         // fragments are removed below, so provisional fragment IDs starting
@@ -5616,9 +5710,90 @@ fn prior_stages_fragment_count(prior_stages: &[StagedWrite]) -> u64 {
 }
 
 /// The rows one overwrite chunk holds: Lance's file target, so a chunk is
-/// the file one call would cut there.
-fn overwrite_chunk_rows() -> usize {
-    WriteParams::default().max_rows_per_file
+/// the file one call would cut there; a test pins it lower through
+/// `with_overwrite_chunk_rows`.
+pub(crate) fn overwrite_chunk_rows() -> usize {
+    crate::instrumentation::overwrite_chunk_rows()
+        .unwrap_or_else(|| WriteParams::default().max_rows_per_file)
+}
+
+/// One overwrite chunk as a producer hands it to
+/// [`TableStore::stage_overwrite_stream`]: the batches one Lance call writes,
+/// their place in the source (`order`: the source partition, then the chunk's
+/// index within it), and whatever the producer holds for them until the
+/// writer returns.
+pub(crate) struct OverwriteChunk {
+    pub(crate) order: (u32, u64),
+    pub(crate) batches: Vec<RecordBatch>,
+    pub(crate) lease: Option<Box<dyn Send>>,
+}
+
+/// The overwrite chunks whose Lance calls returned, with the files they wrote:
+/// taken whole by the join, deleted by a refusal, and deleted off the runtime
+/// when the staging is dropped before either.
+struct WrittenChunks {
+    ds: Arc<Dataset>,
+    transactions: Vec<((u32, u64), Transaction)>,
+}
+
+impl WrittenChunks {
+    fn new(ds: Arc<Dataset>) -> Self {
+        Self {
+            ds,
+            transactions: Vec::new(),
+        }
+    }
+
+    fn push(&mut self, finished: ((u32, u64), Transaction)) {
+        self.transactions.push(finished);
+    }
+
+    fn take(&mut self) -> Vec<((u32, u64), Transaction)> {
+        std::mem::take(&mut self.transactions)
+    }
+
+    async fn discard(&mut self) {
+        let transactions = self.take();
+        delete_overwrite_files(&self.ds, &transactions).await;
+    }
+}
+
+impl Drop for WrittenChunks {
+    fn drop(&mut self) {
+        if self.transactions.is_empty() {
+            return;
+        }
+        let transactions = self.take();
+        let ds = Arc::clone(&self.ds);
+        if let Ok(runtime) = tokio::runtime::Handle::try_current() {
+            runtime.spawn(async move { delete_overwrite_files(&ds, &transactions).await });
+        }
+    }
+}
+
+/// Delete the data files the finished overwrite calls wrote under the
+/// dataset's `data/`; a file that stays is logged and left to cleanup, the
+/// refusal that brought us here still wins.
+async fn delete_overwrite_files(ds: &Dataset, transactions: &[((u32, u64), Transaction)]) {
+    let store = match ds.object_store(None).await {
+        Ok(store) => store,
+        Err(error) => {
+            tracing::warn!(%error, "a refused overwrite's fragment files stayed; cleanup reclaims them");
+            return;
+        }
+    };
+    let data_dir = ds.data_dir();
+    for (_, transaction) in transactions {
+        let Operation::Overwrite { fragments, .. } = &transaction.operation else {
+            continue;
+        };
+        for file in fragments.iter().flat_map(|fragment| fragment.files.iter()) {
+            let path = data_dir.clone().join(file.path.as_str());
+            if let Err(error) = store.delete(&path).await {
+                tracing::warn!(%path, %error, "a refused overwrite's fragment file stayed; cleanup reclaims it");
+            }
+        }
+    }
 }
 
 /// Cut `batches` into contiguous runs of exactly `chunk_rows` rows, the last
diff --git a/crates/omnigraph/src/table_store/staged_tests.rs b/crates/omnigraph/src/table_store/staged_tests.rs
index 025d49a24..8ebcff7d5 100644
--- a/crates/omnigraph/src/table_store/staged_tests.rs
+++ b/crates/omnigraph/src/table_store/staged_tests.rs
@@ -2748,6 +2748,69 @@ async fn stage_overwrite_batches_fans_out_writers_in_chunk_order() {
     assert_eq!(ids_in_order(&after), expected);
 }
 
+/// Chunks handed over out of source order, two writers in flight, are joined
+/// in `(partition, index)` order: one file per chunk, read back in source
+/// order before and after the commit.
+#[tokio::test]
+async fn stage_overwrite_stream_joins_chunks_in_partition_then_index_order() {
+    let dir = tempfile::tempdir().unwrap();
+    let uri = format!("{}/people.lance", dir.path().to_str().unwrap());
+    let store = TableStore::new(dir.path().to_str().unwrap(), test_session());
+    let ds = TableStore::write_dataset(&uri, person_batch(&[("alice", Some(30))]))
+        .await
+        .unwrap();
+    let pre_version = ds.version().version;
+    let chunk = |order: (u32, u64), ids: &[&str]| -> Result<super::OverwriteChunk, OmniError> {
+        let rows: Vec<(&str, Option<i32>)> = ids.iter().map(|id| (*id, Some(1))).collect();
+        Ok(super::OverwriteChunk {
+            order,
+            batches: vec![person_batch(&rows)],
+            lease: None,
+        })
+    };
+    let arrival = vec![
+        chunk((1, 0), &["d"]),
+        chunk((0, 1), &["c"]),
+        chunk((1, 1), &["e", "f"]),
+        chunk((0, 0), &["a", "b"]),
+    ];
+    let expected = ["a", "b", "c", "d", "e", "f"];
+
+    let staged = store
+        .stage_overwrite_stream(
+            &ds,
+            futures::stream::iter(arrival),
+            None,
+            WriteBudget::default(),
+            2,
+        )
+        .await
+        .unwrap();
+    assert_eq!(
+        staged
+            .new_fragments()
+            .iter()
+            .map(|fragment| fragment.physical_rows.unwrap_or(0))
+            .collect::<Vec<_>>(),
+        vec![2, 1, 1, 2],
+        "one file per chunk, in (partition, index) order"
+    );
+    let before_commit = store
+        .scan_with_staged(&ds, std::slice::from_ref(&staged), None, None)
+        .await
+        .unwrap();
+    assert_eq!(ids_in_order(&before_commit), expected);
+    assert_eq!(ds.version().version, pre_version);
+
+    let new_ds = store
+        .commit_staged(Arc::new(ds.clone()), staged)
+        .await
+        .unwrap();
+    assert_eq!(new_ds.version().version, pre_version + 1);
+    let after = store.scan_batches(&new_ds).await.unwrap();
+    assert_eq!(ids_in_order(&after), expected);
+}
+
 /// Batches of two schemas are refused before any writer starts, as one Lance
 /// call refuses them (no data file written, HEAD stays), where Lance's own
 /// overwrite would publish a mixed table; a zero-row batch of another schema is dropped.
diff --git a/crates/omnigraph/src/validate/load.rs b/crates/omnigraph/src/validate/load.rs
index 1f8b2845b..2092514bd 100644
--- a/crates/omnigraph/src/validate/load.rs
+++ b/crates/omnigraph/src/validate/load.rs
@@ -89,6 +89,7 @@ pub(crate) async fn judge_load(
         base,
         mode,
         batches,
+        batches.iter().map(RecordBatch::num_rows).sum(),
         extracted,
         admission,
     )
@@ -101,6 +102,7 @@ pub(crate) async fn judge_load(
 /// the plan's word on the ids an overwrite removes: under `Scan` they are
 /// found before the checks run, under `None` the plan carries no check that
 /// reads them, which the committed table's row count must bear out.
+#[allow(clippy::too_many_arguments)]
 pub(crate) async fn judge_load_swept(
     table: &str,
     checks: &[ValidateSpec],
@@ -109,6 +111,7 @@ pub(crate) async fn judge_load_swept(
     base: &Snapshot,
     mode: LoadMode,
     batches: &[RecordBatch],
+    delta_rows: usize,
     mut extracted: Extracted,
     admission: &mut LoadAdmission<'_>,
 ) -> Result<()> {
@@ -186,7 +189,7 @@ pub(crate) async fn judge_load_swept(
                 }
             }
             ValidateCheck::EdgeRi { .. } => {
-                if edge_ri_judged || batches.iter().all(|batch| batch.num_rows() == 0) {
+                if edge_ri_judged || delta_rows == 0 {
                     continue;
                 }
                 edge_ri_judged = true;
@@ -1435,6 +1438,7 @@ mod tests {
             &snapshot,
             LoadMode::Overwrite,
             &batches,
+            3,
             Extracted::sweep("node:Person", &checks, &catalog, &batches, sc, true).unwrap(),
             &mut LoadAdmission {
                 phase: None,
@@ -1466,6 +1470,7 @@ mod tests {
             &snapshot,
             LoadMode::Overwrite,
             &batches,
+            3,
             Extracted::sweep("node:Person", &pruned, &catalog, &batches, sc, true).unwrap(),
             &mut LoadAdmission {
                 phase: None,
diff --git a/crates/omnigraph/tests/forbidden_apis.rs b/crates/omnigraph/tests/forbidden_apis.rs
index 0b73be466..288590fd1 100644
--- a/crates/omnigraph/tests/forbidden_apis.rs
+++ b/crates/omnigraph/tests/forbidden_apis.rs
@@ -864,7 +864,7 @@ gateway_surfaces! {
     ],
     "storage_layer.rs" => "TableStorage" => GatewayDisposition::StageOnly => [
         "stage_create", "stage_keyed_write", "stage_proven_strict_insert", "stage_overwrite",
-        "stage_overwrite_batches",
+        "stage_overwrite_batches", "stage_overwrite_stream",
         "stage_rename_columns", "stage_schema_evolution", "stage_delete", "stage_create_indices",
         "stage_compaction", "stage_index_fold",
     ],
@@ -912,7 +912,7 @@ gateway_surfaces! {
     ],
     "table_store.rs" => "TableStore" => GatewayDisposition::StageOnly => [
         "stage_create", "stage_keyed_write", "stage_proven_strict_insert", "stage_overwrite",
-        "stage_overwrite_batches",
+        "stage_overwrite_batches", "stage_overwrite_stream",
         "stage_rename_columns", "renamed_schema", "stage_schema_evolution",
         "plan_schema_evolution", "stage_delete", "stage_create_indices", "stage_compaction",
         "stage_index_fold", "stage_full_text_declarations",
@@ -1061,7 +1061,7 @@ durable_calls! {
     ("storage_layer.rs", ".commit_staged_exact(", 1, WriteProtocol::Exact("sealed TableStorage forwarding")),
     ("storage_layer.rs", ".commit_staged_detached(", 1, WriteProtocol::Exact("sealed TableStorage forwarding")),
     ("db/omnigraph/promotion.rs", "SnapshotHandle::new(", 1, WriteProtocol::ReadOnlyAccess),
-    ("storage_layer.rs", ".dataset()", 33, WriteProtocol::Composed("sealed TableStorage forwarding")),
+    ("storage_layer.rs", ".dataset()", 34, WriteProtocol::Composed("sealed TableStorage forwarding")),
     ("storage_layer.rs", ".into_arc()", 6, WriteProtocol::Composed("sealed TableStorage forwarding")),
     ("storage_layer.rs", "SnapshotHandle::new(", 4, WriteProtocol::Composed("sealed TableStorage forwarding")),
     ("table_store.rs", ".raw_dataset_append(", 1, WriteProtocol::EphemeralScratch),
@@ -1072,6 +1072,7 @@ durable_calls! {
     ("table_store.rs", "CommitBuilder::new(", 3, WriteProtocol::Composed("staged commit primitive")),
     ("table_store.rs", ".create_index_builder(", 6, WriteProtocol::Composed("staged index primitive, RFC 0067 whole-rebuild fold and untrained full-text declaration")),
     ("table_store.rs", ".execute_uncommitted(", 11, WriteProtocol::Composed("staged physical primitive")),
+    ("table_store.rs", ".delete(", 1, WriteProtocol::Composed("discard of a refused overwrite's own uncommitted fragment files under the dataset's `data/`, named by the finished writers' transactions and referenced by no manifest; a file that stays is logged and left to cleanup, the refusal still wins")),
     ("db/omnigraph/system_column_upgrade.rs", ".commit_staged_detached(", 1, WriteProtocol::Exact("RFC 0067 detached rename-only system-column effect")),
     ("db/omnigraph/system_column_upgrade.rs", ".commit_changes_with_intent_and_expected(", 1, SYSTEM_COLUMNS_V9),
     ("db/omnigraph/system_column_upgrade.rs", ".dataset()", 1, SYSTEM_COLUMNS_V9),
@@ -2443,6 +2444,12 @@ const MUTABLE_KEY_REQUESTS: &[(&str, &str, usize, &str)] = &[
         2,
         "init claim release after the attempt's last schema mutation, and removal of a create-if-absent probe under a fresh ULID name; a late claim delete can only reopen initialization, where the manifest's put-if-absent still admits one initializer",
     ),
+    (
+        "table_store.rs",
+        ".delete(",
+        1,
+        "removal of a refused overwrite's own uncommitted fragment files, each under a fresh Lance data-file name the same call's writer minted and no manifest references; a late delete can only remove that file, and a file that stays is cleanup's",
+    ),
     (
         "db/omnigraph/collector.rs",
         ".delete(",
@@ -3376,6 +3383,7 @@ const SPLIT_CRATE_REEXPORTS: &[&str] = &[
     "instrumentation.rs: omnigraph_core::instrumentation::with_external_payload_read_held",
     "instrumentation.rs: omnigraph_core::instrumentation::with_external_payload_read_stalled",
     "instrumentation.rs: omnigraph_core::instrumentation::with_merge_write_probes",
+    "instrumentation.rs: omnigraph_core::instrumentation::with_overwrite_chunk_rows",
     "instrumentation.rs: omnigraph_core::instrumentation::with_query_io_probes",
     "instrumentation.rs: omnigraph_core::instrumentation::with_query_memory_limit",
     "instrumentation.rs: omnigraph_core::instrumentation::with_query_memory_probes",
diff --git a/docs/dev/execution.md b/docs/dev/execution.md
index 01ca75acf..f425b6cb6 100644
--- a/docs/dev/execution.md
+++ b/docs/dev/execution.md
@@ -1023,9 +1023,15 @@ A `load csv` overwrite is the first write plan: the planner routes
 `Operation::Load` to the engine, which lowers `CsvSource → Validate →
 FragmentWrite` to `CsvSourceExec` (up to `max_partitions` readers over the
 captured source, each claiming the next record-aligned 8 MiB slice and tagging
-its batches with the slice index), `ValidateExec` (a pipeline breaker that
-drains every input partition and sorts the held batches back into slice
-order: one pass over the held delta extracts what
+its batches with the slice index), `ValidateExec` (every input partition
+drained at once; a streaming operator when the planner's `validate_streams`
+says no check reads the whole delta after the rows are in, each batch sent on
+to the sink right after its sweep with its slice tag, the judgment after the
+drain the stream's last word and the refusal the statement reports even when
+the sink's keyed-write check met the same repeated key in a fragment first
+(the sink follows its input to that verdict before deleting what it wrote);
+otherwise a pipeline breaker that holds the batches and
+sorts them back into slice order: one pass over the delta extracts what
 the value, uniqueness and cardinality checks need; the endpoint checks probe
 the batches' endpoint columns against each endpoint table's committed key set,
 on the parse partitions as the batches arrive when the plan marks them
diff --git a/docs/user/mutations/index.md b/docs/user/mutations/index.md
index f7c3c846a..4b819fb2b 100644
--- a/docs/user/mutations/index.md
+++ b/docs/user/mutations/index.md
@@ -261,12 +261,15 @@ A CSV load has no whole-statement input check.
 It applies `write_max_bytes` to each staged-write input instead.
 An overwrite stages the reader's 8,192-row batches, and a batch over the
 allowance refuses as `staged_write_input_bytes`.
+An overwrite writes a fragment per parse partition, up to 1 Mi rows each, as
+the rows drain; nothing commits before the publish, a refusal deletes them all.
 An append or merge keeps the keyed adapter's ceilings
 (`retained keyed batch bytes per operation`, `keyed entities for <table>`,
 `keyed entity bytes for <table>`).
 The [`query_memory_bytes`](../queries/settings.md) setting bounds one
 `load csv` statement's tracked memory as a whole: the retained source, every
-decoded batch, validation state and the staging working set.
+decoded batch, validation state and the staging working set; a streaming
+overwrite's working set is its open fragments and the writers in flight.
 A statement the allowance cannot hold refuses with the `query_memory_bytes`
 resource error before any fragment write.
 A CSV load needs roughly the source plus twice its typed input under that
diff --git a/docs/user/queries/explain-load.md b/docs/user/queries/explain-load.md
index 9588cc135..09db7e7a4 100644
--- a/docs/user/queries/explain-load.md
+++ b/docs/user/queries/explain-load.md
@@ -20,7 +20,7 @@ root down:
 | `node` | `detail` fields |
 |---|---|
 | `FragmentWrite` | `table`, `version`, `mode`, `adapter`; the physical row adds `max_input_bytes`, `max_workers` |
-| `Validate` | `table` and `checks`, every check of the load in the order it runs, which is the refusal precedence; the physical row adds `removal` and `max_workers` |
+| `Validate` | `table` and `checks`, every check of the load in the order it runs, which is the refusal precedence; the physical row adds `removal` and `max_workers`. The node is a pipeline breaker only while a check reads the whole delta after the rows are in (`retained_edges`, a `judged` `edge_ri`, a `scan` removal); otherwise every check is swept as the rows pass and settled in the judgment after the drain, the node streams, and `pipelines` holds one pipeline from the source through the write instead of two |
 | `CsvSource` | `format`, `source`, `source_bytes`, `estimated_rows`, `delimiter`, `header`, `columns`, `table`, `version`; the physical row adds `max_partitions` (the streams that parse the source: one per 8 MiB of it, at most `stage_write_concurrency`), `max_workers`, `fallback` |
 
 Each entry of `checks` carries `check` (`value`, `unique`, `edge_ri`,