compute: barrier discipline for MV sink shared batches - #37617
Draft
antiguru wants to merge 3 commits into
Draft
Conversation
antiguru
force-pushed
the
pr-35411-barrier
branch
3 times, most recently
from
August 26, 2026 08:41
3bded7d to
9bcafc1
Compare
`SharedBatches` hands out a `SharedBatchBuilder` per batch id. Handles for the same id feed one process-global builder task, so the workers running in one process contribute their parts to a single, larger batch instead of each writing its own small one. The last handle to `finish` receives the batch; the others receive `None`. `BatchBuilder::add_part` accepts a pre-encoded part and concatenates parts until they reach the blob target size, which is what lets the shared builder accept work from several workers without re-encoding it. Building a batch now goes through `PersistClient::batch_builder` rather than a `WriteHandle`, so a shared builder needs no writer registration of its own. Co-authored-by: Moritz Hoffmann <mh@materialize.com>
Wire the shared-batch builder into the sync (v2) materialized-view sink so the workers in one process coalesce their parts for a given batch interval into a single, larger batch instead of each writing its own small batch. All workers building an interval share the batch description broadcast by the mint operator, so they share a batch id and their parts land in one process-global shared batch. Only the last worker to finish receives that batch; the rest get nothing, which maps to the empty-batch response the write operator already handles. Every worker still finishes its builder even when it pushed no data, since any one of them may be the last holder responsible for delivering all workers' parts. The behavior is gated behind the new enable_compute_sync_mv_sink_shared_batches dyncfg, default off. The prior per-worker path is retained unchanged for the off case. Measured locally on an 8-worker replica with a churning view: blob PUTs down ~67% and consensus state-diff bytes down ~54%, at unchanged consensus command count. Adds a testdrive test asserting the view equals the equivalent one-shot aggregation across a multi-worker cluster with churn, enables the flag in the CI system-parameter defaults, and registers it with parallel-workload's flag flipper. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Builds on the shared-batches sink work (its PR): best-effort coalescing only merges workers whose builder handles overlap in time, so a process still writes ~1.4 parts per append. Add an opt-in barrier that makes the process-local workers agree on a single batch per interval. Each of the workers_per_process local workers reports exactly once per batch id: a write (SharedBatchBuilder::finish) or, when it supersedes the description without writing, a skip (SharedBatches::note_skip from the Timely thread). The shared batch is finished only after every participant reports, and delivered to the last worker that actually wrote, which always holds an output capability. The wait is unbounded on purpose. There is no cross-worker clock, so a timeout would have nothing to reason about. Completion rests on every process-local worker reporting once per batch id; in steady state every broadcast description is eventually written or superseded by every local worker, so the barrier converges, and teardown is handled by aborting the write tasks. The batch is finished only after all participants have pushed, so no push can race a finished builder. The barrier is per process: SharedBatches is process-local, so a multi-process replica coalesces to one batch per process, which the append operator already combines into a single compare_and_append. Gated behind enable_compute_sync_mv_sink_shared_batches_barrier, default off, and not randomized across the CI suite (the unbounded wait is too risky for that); covered by the shared-batches testdrive (single- and multi-process) and the parallel-workload flag flipper. Measured on an 8-worker replica: parts per append drop from ~1.4 to 1.0, cutting blob PUTs a further ~30%. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
antiguru
force-pushed
the
pr-35411-barrier
branch
from
August 26, 2026 11:21
9bcafc1 to
a5cf162
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stacked on #37591 (shared-batches sink port) — review the last commit only; the earlier commits are #37591 and will drop out once it merges and this rebases onto main.
Best-effort shared batches coalesce only workers whose builder handles overlap in time, so a process still writes ~1.4 parts per append. This adds an opt-in barrier so the process-local workers agree on exactly one batch per interval.
Each of
workers_per_processlocal workers reports once per batch id — a write, or a skip (note_skipfrom the Timely supersede path). The batch finishes only after all report, delivered to the last worker that wrote (always holds a capability). The wait is unbounded, no timeout — there is no cross-worker clock to justify one; completion rests on every local worker reporting once, and teardown aborts the write tasks. Per process, so a multi-process replica coalesces to one batch per process, combined into a singlecompare_and_appendby the append operator.Gated behind
enable_compute_sync_mv_sink_shared_batches_barrier, default off, not randomized across the CI suite (unbounded wait). Measured on an 8-worker replica: parts per append ~1.4 → 1.0, a further ~30% off blob PUTs.An adversarial review found the barrier core sound (exactly-once reporting, deliverer-holds-capability, push-before-finish, lost-wakeup, multi-process all hold); remaining low-severity items (teardown blob leak, stall→OOM amplifier,
pending_skipscleanup) are noted for follow-up and cluster around the no-timeout design.