[mdibaiee]

Description:

The preview-next fixture feeder parsed every streamed line into a serde_json::Value, cloned it, rebuilt it as a HeapNode, then archived, validated and keyed it, all on one task. Benchmark runs were bound by it: a TPC-DS scale 1 fixture through a no-op materialization ran at 84K docs/s with the runtime and connector mostly idle.

• Document lines of a streamed fixture are built into shuffle::log_request::Appends on scoped worker threads, as the live slice builds them from journal reads: parsed once with simd-doc straight into a HeapNode, stamped with the synthetic UUID, archived, validated and keyed. The Appends are then routed and pushed in fixture order, so blocks and document clocks match a serial feeder.
• runtime_local::segments splits that work into build_append (no transaction state, safe to run in parallel) and push_append (routing and serial block and frontier accounting). push_binding keeps its signature and uses both, so eager fixtures and catalog tests share one implementation.
• --fixture-validate=false flags streamed documents valid unchecked, for generated fixtures known to be valid.

Same TPC-DS run on a c3-standard-22: 84K → 142K docs/s.

Workflow steps:

flowctl raw preview-next --fixture <fifo> [--fixture-validate=false]. Output is unchanged.

Documentation links affected:

None.

Notes for reviewers:

• built_appends_match_the_value_path checks that the simd-doc path builds the same Appends as the serde_json::Value path; batches_write_the_same_log_on_any_number_of_workers checks that 1, 4 and 7 workers write identical segments and frontiers.
• Past the feeder, the next limit in these runs was the shard's connector request queue (runtime-next: queue a shard's connector requests in a VecDeque #3638).

The base branch was changed.

[jgraettinger]

Please hold off on merging this, until this code refactor move lands? #3416

[mdibaiee]

@jgraettinger okay 👍🏽

[strix-security]

Strix Security Review

✅ No open security findings.

PR summary: This PR speeds up the streaming `--fixture` feeder of `flowctl raw preview-next` by parsing each fixture line once straight into a `HeapNode` and preparing it into shuffle-log `Append`s on worker threads (then pushing them in fixture order). It also splits the segment writer into a state-free `build_append` and a stateful `push_append`, and adds a `--fixture-validate` flag (default on) that lets an operator skip schema validation for fixtures already known to be valid.

Review summary

Reviewed all three changed files in the diff: the new fixture_validate CLI flag and its plumbing in crates/flowctl/src/raw/preview_next/mod.rs, the split of the segment writer into build_append / push_append (plus TxnState::take_doc_seconds) in crates/runtime-local/src/segments.rs, and the batched, worker-thread Append builder in crates/flowctl/src/raw/preview_next/fixture.rs, including the new equivalence tests. Two suspicions were raised and independently validated: whether --fixture-validate=false lets unvalidated documents carry the schema-valid bit into a trust boundary, and whether the reused arena plus unchecked archive access in the parallel builder could be unsound. Both were disproved — the flag is a documented, local-operator-only opt-out confined to the streaming fixture path whose shuffle log lives in a per-invocation temporary directory, and the archive is produced by the serializer over a live HeapNode while all Append payloads are copied out of the bump arena before it is reset, with no unsafe added. No security issues were found in the changed code. One non-blocking documentation note for the author: schema validation is also where redact annotations are applied, so --fixture-validate=false additionally skips redaction of the operator's own fixture documents; a sentence in the flag's help text would make that explicit.

Ruled out (3)

• Can `--fixture-validate=false` let unvalidated documents pass a trust boundary by carrying the schema-valid flag?
  The flag is a plain CLI argument on a locally-invoked developer command, only applies to the streaming fixture path whose shuffle log is written to a per-invocation temporary directory, and the same operator already supplies both the fixture documents and the flag, so no boundary is crossed.
• Is the reused bump arena and unchecked archive access in the parallel Append builder a use-after-reset or data race?
  `Allocator::reset` needs exclusive access that the `'alloc` lifetimes enforce, `simd_doc::Parser::parse_one` returns a real `HeapNode` whose archive is emitted by rkyv itself, and every `Append` payload is copied out of the arena before reset while each worker owns its own builder.
• Can a malformed fixture line trigger out-of-bounds reads through the archived document view or key extraction?
  The archive passed to `ArchivedNode::from_archive` is serialized by rkyv from a live `HeapNode`, so alignment and internal offsets come from the serializer rather than from document content.

Review history (3)

• Review 3 · 2966dfa · Oct 21 12:34 UTC · Nothing new
• Review 2 · c6276ff · Oct 21 12:27 UTC · Nothing new
• Review 1 · 6d54b29 · Oct 20 18:56 UTC · Nothing new

_{Review 3 on this PR · 2966dfa · requested Oct 21 12:31 UTC on a new commit}

---

Reviewed by Strix
_{Re-run review · Configure security review settings}

[mdibaiee]

This PR has been rebased on latest changes and is ready for review again now that #3416 has been merged. Only affects runtime-local and preview_next

[jacobmarble]

I'm learning a little about Rust here. Does this hold a tokio worker without letting other tasks schedule, even when the task blocks on I/O?

OTOH, this seems to be CPU bound, so probably better to let it burn without interruption?