[azimafroozeh]

What & why

A load csv overwrite parses its captured source with one Arrow reader and extracts what its checks consult (ids, each uniqueness check's keys, cardinality sources) in one pass over the held delta after the parse ends. On SF30 the parse is 190 s of the 413 s load (#1121's phase table) and the validation 175 s (the width-1 run below), both on one thread, while the planner already prints max_partitions on the source and the sink writes with eight writers. This PR cuts the capture on record ends into 8 MiB slices in the resolve pass that already walks it, runs the source as up to stage_write_concurrency (default 8) partitions that each claim the next slice from a shared counter and tag their batches with its index, and has the validation drain every partition at once, sweeping each one's batches into a partial as they arrive, then sort the held batches by slice and merge the partials (ids unioned, each uniqueness check's first holders sharded by key hash and merged shard by shard, each shard on its own worker) before the checks judge once as before. A cut lands only after a complete record, so the slices concatenate to the single reader's rows in its order; every extracted fact carries its (slice, row) position, so the merged image and the first violation are the single pass's; and a refused slice stops the claiming while the slices in flight finish, so the lowest failed slice's record is the refusal on every run. SF1 loads in 8.6 s instead of 13.2, SF30 in 261 s instead of 413.

Backing issue / RFC

• Implements the parallel source of the bulk-load RFC, docs/rfcs/2026-10-15-bulk-load-as-a-query-plan.md (feat(load): plan CSV overwrites and reuse validation reads #1026, perf(load): check edge endpoints in an unchained integer key set #1085, feat(explain): cut pipelines at breakers on every route and measure a load's operators #1099, perf(load): fan out a csv overwrite's Lance writers over file-sized chunks #1121). The RFC's workers read ranges of the source; here they claim slices of the whole capture the load already holds, so every cut is proved before a partition starts and the declared single_scan fallback never fires. D11 (each phase holds its whole input) is unchanged. That and perf(load): fan out a csv overwrite's Lance writers over file-sized chunks #1121's writer fan-out owe the RFC a sentence.

Checklist

• Focused: one logical change, the source's partitions and the per-partition sweep that consumes them
• Tests: Rust, since a GQT fixture cannot reach a second 8 MiB slice and the hook with_csv_slice_bytes cuts at 64 bytes: the cut (record ends numbered from the source, no header, CRLF cut after the line feed, never inside a quoted field, a header alone is one empty slice, a slice's parse error names the source line), the source (three partitions claim every slice once and tag their batches; a refused slice stops the claims), the breaker (partitions drain back into source order; the lowest failed slice names the refusal; the first repeated key in source order is the refusal from one and from three partitions), the planner's partition count (8 MiB is 1, 8 MiB + 1 is 2, capped by the load workers), the end-to-end load_csv_overwrite_from_many_slices_keeps_source_order (2,000 rows, every slice in the receipt, keys read back in order), and the gqt-core load pipeline … partitions <n> round trip. The GQT corpus (412 cases) is unchanged: every fixture is under 8 MiB, and every case's refusal equals the sequential sweep's.
• Public docs: docs/user/queries/explain-load.md, docs/user/queries/explain.md, docs/user/queries/settings.md, docs/user/mutations/index.md, crates/omnigraph-gqt/README.md; docs/dev/execution.md and docs/dev/writes.md for the mechanism
• Release note: changelog.d/load-csv-parse-partitions.performance.md
• Invariants: reviewed against docs/dev/invariants.md, no Hard Invariant weakened, no deny-list item hit; nearest one touched is 8 (integrity failures are loud): a refusal still names a record, the lowest failed slice's

Local verification

• LDBC SNB SF1, 31 load csv … mode overwrite statements through omnigraph mutate --json, release builds of main (888223b7) and this branch at the default width, 2 rounds interleaved, phases read from executed.operators as in feat(explain): cut pipelines at breakers on every route and measure a load's operators #1099: load 13.19 → 8.64 s; Comment 2.63 → 1.15 s (parse 1.31 → 0.43 s, validate 0.81 → 0.25 s), Post 1.26 → 0.61 s, HasMember 1.19 → 0.65 s; the 16 statements under 8 MiB and the write phase unchanged.
• LDBC SNB SF30, the same statements on this branch, one round: 261.4 s, against the 412.9 s perf(load): fan out a csv overwrite's Lance writers over file-sized chunks #1121 measured for its branch the same way; the parse half alone measured 295.8 s, the per-partition sweep took validate 167 → 136 s (Comment 43 → 17 s, Post 9 → 3 s); the 23 edge statements' validate is flat at 116 s, their endpoint checks still run on one thread.
• SF30, one binary of the parse half, OMNIGRAPH_LOAD_CONCURRENCY 1 vs 8, one round each: parse 168.6 → 48.4 s (3.5×; HasMember, LikeComment, LikePost 4.1×), validate 175.4 → 171.1 s; width 1 also serializes perf(load): fan out a csv overwrite's Lance writers over file-sized chunks #1121's writers, so the load total (481 → 296 s) overstates the parse gain.

Notes for reviewers

• Engine v2's first operator with more than one output partition. The shape helpers are general (Source::with_partitions with per-partition metrics, the breaker draining N inputs, partitions on the planner's pipelines), but only the load's source is wide; every query operator and the sink still run one stream.
• No format change: an existing store reads unchanged; the rows reach the sink in source order, so the committed files are those of a single reader.
• A default changes: stage_write_concurrency (OMNIGRAPH_LOAD_CONCURRENCY, default 8) now also caps the source's partitions, one per 8 MiB of source; 1 restores the single reader, and a source under 8 MiB plans one partition at any width.
• openapi.json changes shape, additive: LoadOperatorExecutedOutput gains partitions and slices (LoadSliceExecutedOutput: the slice, the partition that claimed it, its rows and timings); the explain's pipelines entries gain partitions under the same explain_version.
• The one visible change in refusals: typed conversion is judged per slice, so two bad values on different properties in different slices name the earlier record where the single reader named the earlier property; no corpus case has both, and one needs two slices to reach it.
• Memory: the breaker still holds the whole delta; the parse's decoding scratch is admitted per slice per partition instead of once for the whole capture.

[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	✅ Completed 2026-10-22T09:57:46.139764Z	7f504e3	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.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 7f504e3eda

ℹ️ 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".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".