[azimafroozeh]

Since #1158 a load csv overwrite parses its source by partition and checks uniqueness as the rows drain, but its write is still a pipeline breaker: the sink materializes every validated batch and starts its Lance writers only after the validate operator's verdict on the whole delta, a serial tail of 45 s in SF30's 104 s load, with the delta held twice until then.

The sink now writes as the rows drain: each parse partition fills fragments of up to 1 Mi rows, and a full fragment goes to one of stage_write_concurrency Lance writers while the next rows arrive. The planner decides per statement whether the validate operator streams (every check swept as the rows pass and settled in the judgment after the drain) or stays a breaker (a retained-edge check, a judged endpoint check or a removal scan reads the whole delta), so a streaming load plans as one pipeline from the source through the write. The guard is the refusal path: nothing commits before the statement's one publish, and a refusal that arrives after fragments were written deletes them and reports the validate operator's verdict, so a refused load leaves no fragment behind and the table version never moves.

LDBC SF30, release builds of 2af30259 and this branch, two interleaved rounds, medians: load 104.5 → 87.2 s, the write phase after the last row 45.4 → 3.6 s, the writers' remaining work now inside the parse and validate walls. A sink test counts Lance's data/ files across a refused run that had written fragments, and two corpus cases pin, at 10-row fragments, that a refusal past written fragments keeps the committed rows and reports the validate operator's text and value.

For review: the decision to challenge is the validate operator's shape as a plan decision, streaming whenever no check reads the whole delta after the drain, with lowering refusing an operator whose emission disagrees with the plan. Least sure is refusal precedence in the streaming shape: the sink's keyed-write check can meet a key repeated inside one full fragment before the verdict, so the staging drains its input to that verdict before deleting; it is pinned at 10-row fragments and not run at the default 1 Mi. A delta parsed in several partitions now lands as one fragment per partition where it was one per 1 Mi rows; the store format and the overwrite transaction are unchanged. A crash between the first fragment write and the publish leaves unreferenced files under data/ for the cleanup that already reclaims that class, over a longer window than before.

[chatgpt-codex-connector]

Codex Review Summary

This comment shows the latest Codex review activity on this pull request.

Review	Status	Commit	Review trigger
📝 Code Review	⚠️ Failed 2026-10-26T10:06:02.454216Z We couldn’t complete this review. Comment @codex review to try again.	eeb614b	PR opened

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review" or "@codex security review".

Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings.

[ragnorc]

Recommendation: revise.

The published graph stays one overwrite and one publish. Revise the written memory contract so it matches that code. One blocking finding. Three optional findings.

Why

A load csv overwrite parses by partition and checks uniqueness as the rows drain. The sink still waited for the verdict on the whole delta, then started the Lance writers. A large load held every typed batch for that whole wait.

What

Each parse partition now fills a fragment of up to 1,048,576 rows while the rows drain. A full fragment goes to a Lance writer. At most stage_write_concurrency writers run at once (default 8). The planner keeps one pipeline when every check can judge from the rows as they pass: value, unique, cardinality, and endpoint checks that sweep. A retained-edge check or a removal scan stays a pipeline breaker and still sorts by slice before the sink runs.

How the overwrite stays one publication

InsertBuilder::execute_uncommitted returns a transaction and leaves the table uncommitted. stage_overwrite_stream joins those fragments in (partition, index) order into one Operation::Overwrite. Fragment ids in that transaction stay 0 until Lance assigns them at commit. Stable row ids stay unset, so the overwrite allocates fresh ids above the high-water mark. WriteParams match the previous path. The dataset keeps its storage version (V2_2 on current tables). The table commit stays detached. The graph still publishes once, through __manifest.

A producer error sets producer_refused. The sink then reads the producer through to its end, so Validate can park its verdict. The sink waits for the writers that are already in flight, deletes the data files those writers finished, and returns the input error ahead of a staging error. The table version stays on the pinned version.

A crash before publication leaves the graph unchanged. A crash after the manifest commit keeps the publication. load_csv_statement_dst_failure_keeps_writing passes on this commit for seeds 0 and 42.

Parse partitions own contiguous source ranges, so (partition, index) order is source order. A cut also falls on a partition boundary. The same file can therefore receive different fragment boundaries when stage_write_concurrency changes. Row order and the fresh stable row ids follow the source. Fragment layout is derived state. An overwrite still drops indexes.

Impact

The working set of a streaming overwrite is the open chunks plus the writers in flight. The pool charges an open chunk through WorkMemory::hold. The sink pulls the next chunk only while fewer than writers calls are in flight. A load whose full typed delta is larger than query_memory_bytes can succeed. A hold that fails on a later chunk is a producer error after some fragment files exist. Discard deletes those .lance files. The table version stays put.

Blocking

docs/user/mutations/index.md lines 273–275:

The new sentences above say a streaming overwrite writes a fragment per parse partition as the rows drain, and that its working set is the open fragments plus the writers in flight. These two sentences still say a query_memory_bytes refusal happens before any fragment write, and that a CSV load needs the source plus twice its typed input.

That pair is now false. WorkMemory::hold runs as a chunk opens (fragment_sink.rs). A full chunk can already sit in a Lance writer before a later row is judged. A hold that fails on a later chunk is a producer error after files exist. discard deletes those .lance files, and the table version stays on the pinned version. A load whose full typed delta is larger than the pool can succeed.

docs/user/queries/settings.md (query_memory_bytes, the long cell) still says the staging working set is reserved before the first fragment write. It repeats the refusal-before-any-fragment-write rule and the "source plus twice its typed input" peak. docs/dev/writes.md still says stage_overwrite_batches cuts the batches at the file target, and that ValidateExec drains every partition, sorts, and only then reaches the sink (about lines 414–425). The same page says refusal precedes this operation's fragment staging (about lines 481–483). docs/dev/execution.md matches this pull request.

Please make these three pages describe one contract: when the first fragment file appears, what a query_memory_bytes refusal does, and which bytes a streaming CSV overwrite holds.

Optional

1. crates/omnigraph/src/table_store.rs delete_overwrite_files (the loop at line 5790):

This loop deletes fragment.files only. Lance cleanup_data_fragments (lance-13.0.0 dataset/write.rs) deletes the data file and then remove_dir_alls data/<stem>/, the directory of blob v2 sidecars data/<stem>/<id>.blob.

CSV load rejects Blob and Vector (unsupported_kind in loader/csv.rs), so a CSV overwrite writes no sidecar. stage_overwrite_batches now reaches this deleter, and a JSONL overwrite can write those sidecars. The .lance file goes away on refusal. The sidecar directory stays until the orphan sweep (UNVERIFIED_THRESHOLD_DAYS, 7 days). orphan_key maps data/<stem>/<id>.blob to data/<stem>.lance, so the bytes stay unreachable.

a_refused_input_leaves_no_fragment_file_behind counts .lance files whose parent directory is data. A leftover sidecar stays invisible to that test.

FLAG_MANAGED_BLOBS is off on this Lance pin. A later Lance that keeps blob objects after the data file is gone would make a .lance-only delete incomplete for those objects. The matching cleanup is the one Lance already uses on its own write error: delete the data file, then remove the stem directory.

2. WrittenChunks::discard takes the file list, then awaits the delete. A cancel during that await drops the list. Drop then sees an empty list and starts no second delete. The 7-day orphan sweep is the backstop, the same backstop a crash between the first fragment file and publish already uses.

3. changelog.d/load-csv-streaming-overwrite-sink.performance.md line 17:

This 1.2× figure is a local measurement. This review has no rerun of the SF30 load. The baseline in the pull request text is 2af30259 (#1158). Three commits sit between that commit and the merge base da1c46368c33876d61615add0ceb6778f9ce2b58: 1f996c30 (csv row estimate), 507f98f5 (MCP tools), and da1c4636 (Studio). Please name 2af30259 in this note. No checked-in instrument records the run.

Tradeoff and liability

This removes the need to hold the whole typed delta before the first Lance call. It adds one owner for uncommitted overwrite files (WrittenChunks inside stage_overwrite_stream), the task-local overwrite_chunk_rows pin, and the planner predicate validate_streams. stage_overwrite_batches calls the new stream, so CSV and JSONL share one overwrite door. That door is the right owner. A later streaming write should call it, so a second deleter does not drift away from this one.

The new state is the chunk order, the in-flight bound, and the file list a refusal deletes. Each one earns its place. The liability that stays open is the written limit. An operator who sizes query_memory_bytes from the current user guide will plan for a refusal the code no longer returns, and for a peak the code no longer holds.

Alignment

#1174 (open) sizes unique-holder tables with grow_if_free on the first batch. This sink still calls size_unique_holders from the sweep, before the batch is sent. Streaming can leave one partition's chunks in the sink while another partition sizes its holders. The race #1174 describes can show up on that schedule. Both changes edit validate.rs. Leave the holder-sizing fix on #1174.

#1158 is the parent of this design. This change is the sink that #1158 left until after the verdict.

#1156, #1171, #1168, #1177, and #818 need no change here. An overwrite still drops indexes. Fragment identity stays the identity of the fragment files. Cleanup already owns unreferenced objects under data/. The commit cache is unchanged.

Lance 13.0.0 (the pinned crate, unmodified) matches this use. execute_uncommitted writes files and returns a transaction. The commit is a separate manifest put. An omitted storage version follows the existing dataset. FLAG_MANAGED_BLOBS is off. A future Lance that keeps blob objects after the data file is gone would make a delete of only .lance files incomplete for those objects.

Validation

Head c976c592ead4c5f8ca6e25f7e5a34c9eb1a5b702. Merge base da1c46368c33876d61615add0ceb6778f9ce2b58. CARGO_TARGET_DIR=/workspace/target.

On the head, from /tmp/pr1178:

• cargo test -p omnigraph-engine --lib --locked -- --test-threads=8 stage_overwrite_stream_joins_chunks_in_partition_then_index_order a_refused_input_leaves_no_fragment_file_behind a_fence_refusal_yields_to_the_inputs_own_verdict partitions_stream_their_rows_tagged_by_slice — 4 passed.
• cargo test -p omnigraph-planner --test integration --locked load_plan:: -- --test-threads=8 — 8 passed, including per_row_checks_plan_one_pipeline_through_the_write.
• cargo test -p omnigraph-gqt --test gq_logic_tests --locked load_csv_overwrite_refusal -- --test-threads=2 — both refusal cases passed (0.74 s). The precedence case expects Error while parsing value 'abc' as type 'Int64' for column 2 at line 95 when an earlier fragment can also hit a repeated key.
• cargo test -p omnigraph-gqt --test gq_logic_tests --locked load_csv_statement_dst_failure_keeps_writing -- --test-threads=1 — passed (1.96 s). Seeds 0 and 42. Replay matched.

On the merge base, after cargo clean -p omnigraph-engine -p omnigraph-core -p omnigraph-planner -p omnigraph-gqt -p omnigraph-gqt-core, with sources at da1c4636. The base copies of the two new cases omit the overwrite_chunk_rows header, because that key is new on this branch. The plan assertion does not use the 10-row pin.

• The same load_csv_overwrite_refusal filter. The precedence case passed, with the same rows and the same parse error. The late case failed at step 8 only. The executed plan was load pipeline 0: CsvSource => Validate partitions 2 and load pipeline 1: Validate -> FragmentWrite => consumer after 0, with load executed: partitions 2 rows 100. Exit 101. That is the breaker this change removes. The engine crate in that run compiled from /workspace/crates/omnigraph at the merge base.

Limitations

SF30, S3, Azure, the full workspace test graph, clippy, and lance_surface_guards are outside this run. The extended DST case ran on the head only. Its new steps use seams that already exist. CI on this commit was still in progress at review time. a_refused_input_leaves_no_fragment_file_behind counts .lance files whose parent directory is named data.