Skip to content

compute: barrier discipline for MV sink shared batches - #37617

Draft
antiguru wants to merge 14 commits into
MaterializeInc:mainfrom
antiguru:pr-35411-barrier
Draft

antiguru wants to merge 14 commits into
MaterializeInc:mainfrom
antiguru:pr-35411-barrier

Conversation

@antiguru

@antiguru antiguru commented Jul 13, 2026 •

Copy link
Copy Markdown
Member

Stacked on #37591. Review the last commit only. The earlier commits belong to #37591 and drop out once it merges.

Best-effort shared batches only coalesce workers whose writes overlap in time, so a process still writes about 1.4 parts per append. This adds an opt-in barrier, with which the workers of a process agree on exactly one batch per interval. On an 8-worker replica, parts per append drop to 1.0 and blob PUTs fall by a further ~30%.

Each local worker reports once per batch interval, either by writing or by skipping a description it superseded. The batch finishes once every worker has reported, and goes to the last worker that wrote, which always holds a capability to emit it. A multi-process replica writes one batch per process, and the append operator combines them into a single compare_and_append. The barrier applies to the sync sink only. The async sink skips empty and superseded descriptions without reporting them, so it stays best-effort.

The wait has no timeout, because there is no cross-worker clock to justify one. Completion relies on every local worker reporting exactly once, and teardown aborts the waiting write tasks. For this reason, enable_compute_sync_mv_sink_shared_batches_barrier defaults off and is not randomized across CI. A dedicated testdrive test and the parallel-workload flag flipper cover it.

Known follow-ups, all tied to the no-timeout design: parts uploaded before teardown wait for leaked-blob cleanup, a stalled worker holds the batch open and can amplify memory pressure, and pending_skips lacks cleanup.

🤖 Generated with Claude Code

bkirwi and others added 2 commits October 13, 2026 10:50
`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 <[email protected]>
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) <[email protected]>
antiguru and others added 2 commits October 13, 2026 11:26
The async (v1) MV sink now writes through the shared batches behind the same
`enable_compute_sync_mv_sink_shared_batches` flag as the sync sink. Both sinks
share the batch-writing code in `sink::shared_batch`, and the testdrive test
covers both.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Both MV sinks read `enable_compute_sync_mv_sink_shared_batches` through a
config handle when they start a batch, so a flip applies without re-rendering
the sink. The sync sink's shared and per-worker write paths are separate
functions.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
antiguru and others added 2 commits October 13, 2026 13:32
A single task per process encodes and uploads a shared batch, which
serializes work the workers otherwise do in parallel and slowed hydration
and large writes. Each worker now stages its updates for an interval and
hands them to the shared batch only while they stay below
`compute_mv_sink_shared_batch_max_bytes`. Past the threshold, the worker
writes its own batch, so an interval can yield a shared batch and
per-worker batches, which the append operator appends together.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
The flag now serves as a kill switch. Both sinks read it per batch
interval, so turning it off takes effect without re-rendering.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
@antiguru

Copy link
Copy Markdown
Member Author

This PR's barrier was measured against best-effort coalescing on a 128-worker replica (scale=4,workers=32) with a trickle of updates. Full tables are in #37591 (comment) and the comment after it.

Rows per interval Mode Parts per append Diff bytes per append
128 Best-effort 9.4–13.2 18.2k–19.6k
128 Barrier 4.00 16.4k–16.7k
32 Best-effort 6.9–7.0 6.8k–6.9k
32 Barrier 4.00 6.0k–6.1k

The barrier delivers exactly one batch per process in every run. At 32 workers per process, best-effort produces 2.3–3.3 batches per process, up from 1.4 at 8 workers. These straggler batches are small, so the barrier saves only 10–20% of consensus bytes over best-effort. Without coalescing, the same load commits 40.5k–41.3k bytes per append at 128 rows.

Posted by Claude Code.

With shared batches disabled, `SharedBatchWriter::start` now returns a write
that joins no shared batch and sends every update to the worker's own batch,
instead of returning `None` and leaving each sink with a separate per-worker
path. Both sinks have a single write path as a result.

Add unit tests for a shared batch whose handles drop without finishing, and a
testdrive phase that runs both sinks with sharing disabled.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
antiguru and others added 3 commits October 14, 2026 09:47
* `PersistClient::batch_builder` honors `persist_batch_max_runs` when
  `max_runs` is `None`, so the shared and per-worker MV sink builders
  follow the config like `WriteHandle::builder` does.
* The sync sink starts a write only once an interval has updates, so idle
  intervals no longer take the `SharedBatches` lock or spawn a builder
  task.
* Test `BatchBuilder::add_part`'s lower check and flushing, and make the
  sweep race test detect an evicted live entry.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Production picks up shared batches through a staged parameter rollout
instead of on merge. CI keeps covering the path, since mzcompose sets the
flag to true and parallel workload toggles it.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
The flag gates both the async and the sync MV sink, so its name should not
refer to the sync sink only.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Drop unverified qualifiers and contrasts, and keep the requirement that
every handle finishes in one place, `SharedBatchBuilder::finish`.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
antiguru and others added 3 commits October 14, 2026 17:52
* A zero `compute_mv_sink_shared_batch_max_bytes` joins no shared batch,
  matching the "never shares" setting CI exercises.
* Test a write that crosses the threshold, a write that does not share,
  and conflicting descriptions for one shared batch.
* The async sink drops the write handle it opens only for the schema id.
* Name the shared batch channel capacity, include the shared id in
  `BatchDescription`'s debug output, and correct several comments.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Remove `enable_compute_mv_sink_shared_batches`. A zero
`compute_mv_sink_shared_batch_max_bytes` already disables sharing, so the
threshold now defaults to zero and is the only knob. This removes the
combination of an enabled flag with a zero threshold.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
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, either by a write (SharedBatchBuilder::finish) or, when it supersedes the
description without writing, by a skip (SharedBatches::note_skip from the
Timely thread). The shared batch is finished only after every participant
reports, and is delivered to the last worker that pushed data, which always
holds an output capability.

The wait has no timeout, since there is no cross-worker clock to base one on.
In steady state every local worker eventually writes or supersedes each
broadcast description, so the barrier completes. Teardown aborts 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, because SharedBatches is process-local. A
multi-process replica coalesces to one batch per process, which the append
operator combines into a single compare_and_append.

The barrier is gated behind enable_compute_sync_mv_sink_shared_batches_barrier,
default off, and not randomized across the CI suite, because a missed report
would hang the sink. The shared-batches testdrive (single- and multi-process)
and the parallel-workload flag flipper cover it. 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) <[email protected]>
antiguru added a commit that referenced this pull request Oct 14, 2026
Continues #35411 (draft by @bkirwi), rebased onto `main`.

With many workers each MV sink worker writes its own small batch per
interval, so the sink emits many small parts and a large consensus state
diff. This coalesces the parts the workers in one process write for a
batch interval into a single, larger shared batch.

`SharedBatches` hands out a `SharedBatchBuilder` per batch id. Handles
for the same id feed one process-global builder task, and the last
handle to finish receives the batch. Both MV sinks use it, best-effort.
On an 8-worker replica this cut persist blob PUTs ~67% and consensus
state-diff bytes ~54%, at unchanged consensus command count.

The single builder task serializes encoding and upload, which slowed
hydration of a 9.5 GB MV from 26 s to 33 s. Each worker therefore stages
its updates for an interval and only contributes them to the shared
batch while they stay below `compute_mv_sink_shared_batch_max_bytes` (1
MiB in these measurements). Past that, the worker writes its own batch,
and the append operator appends both kinds together. With the threshold,
hydration and 5M-row inserts match the per-worker path, while 1000-row
writes drop from ~7.8 to ~1.05 data parts per append. Writes small
enough for persist's inline path now cost one blob write per append
instead of none, at about +10 ms p50 latency and 2.5x fewer compaction
parts.

`compute_mv_sink_shared_batch_max_bytes` is the only knob. It defaults
to 0, which disables sharing, and production enables it through a staged
parameter rollout. CI runs it at several values, including 0. Both sinks
read it per batch interval, so a change applies without re-rendering the
sink.

Best-effort means a worker only merges into the shared batch when its
write overlaps the others still building in the same interval, so parts
per append land around 1.4, where the ideal is 1.0. The barrier that
closes that gap is split into the follow-up #37617.

The async (v1) and sync (v2) sinks share the batch-writing code in
`sink::shared_batch`. Unlike @bkirwi's draft, the v1 path is behind the
same flag, since v1 is the default sink.

Original authorship by @bkirwi is preserved in the first commit.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Ben Kirwin <[email protected]>
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants