[beinan]

Summary

Materializers currently fetch and decode an entire pinned publication before starting Lance file writers. Add an experimental, default-off worker.publication.pipeline_preparation option that feeds bounded Arrow streams into Lance 13 writers as records become available, without payload reserialization.

Both paths preserve the same immutable plan, whole-record partitioning and complete descriptor aggregation. Writers share fetch/decode permits; blocking decode retains its permit and source reservation through cancellation, and all admitted writers finish before a failure is returned. No partial descriptor set can be prepared or published.

This distributes preparation within one worker only. Lance encoder workspace remains outside the source reservation, streamed blob_write timing includes source waiting, and no throughput improvement or end-to-end zero-copy is claimed. A matched short cloud comparison completed full payload and aggregate checks before and after restart in both modes. In the common 60–150 second window, buffered publication was 60.17–60.63 MiB/s and streamed publication 57.23–57.87 MiB/s. Including drain, the respective rates were 52.62–54.15 and 57.33–59.05 MiB/s. These are one-run sampling/clock bounds, not statistical confidence or sustained capacity; initial leaders differed and the streamed run experienced term changes. The results do not establish a steady-window throughput improvement or support enabling the option by default.

Testing

• Workspace formatting, all-target Clippy and Rust tests.
• Python pytest: 229 passed, 2 skipped, 1 xfailed; Ruff formatting/lint and Pyright. The server executable was rebuilt from this source; the unchanged Python native extension was reused after checking its source inputs.
• Service/http-client all-target Clippy and 10 feature tests, including real WAL/Lance recovery in both modes and injected stream failure with observer cancellation.
• Both modes passed release-process qualification: 192 MiB ACKed but unpublished, all nine WAL processes killed, original directories reopened, then one Lance publication with complete payload, aggregate-row, read-only and negative-verifier checks; no producer replay.
• A separate streaming-mode process test replaced one replica directory with an empty directory after the crash. Its group payloads were fetched exclusively from the rebuilt replica and all data checks passed. This is local directory-loss simulation, not a Kubernetes volume-loss qualification.

Refs #378. Built on merged #463 and #465. The streamed materializer implementation matches the qualified candidate; the base additionally contains the engine directory-lock fix #465. A test helper is additionally gated to the features that use it; materializer-only and service/http-client all-target Clippy pass with that correction.

Validation follow-up: the HTTP test publisher helper is compiled only with http-client, matching its callers. Local all-target Clippy passes with worker, materializer, and http-client,materializer separately; production code is unchanged by this test fix. Final-head CI must pass before merge.