compute: coalesce MV sink batches across workers by antiguru · Pull Request #37591 · MaterializeInc/materialize

antiguru · 2026-07-12T01:47:35Z

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

def- · 2026-08-26T09:59:21Z

QA LLM Review

1. HIGH -- Cleanup sweep's `Weak::upgrade` can orphan a finished shared batch

src/storage-operators/src/persist.rs:98

The map-cleanup sweep in SharedBatches::builder probes liveness with weak.upgrade(), which takes a temporary strong reference. SharedBatchBuilder::finish decides who owns the batch with Arc::into_inner, which returns None whenever the strong count is above one. If the sweep's probe overlaps the last real handle's finish, every worker returns None, nobody emits the batch, and the append operator advances the shard upper over an interval whose updates were never written.

Details

The window: builder runs the sweep whenever state_map.len() > last_retained_len * 2, which is most calls once last_retained_len settles low, and each sweep upgrades every live entry in the map. A worker entering builder for sink X therefore probes sink Y's in-flight entry while another worker is finishing it. Both operations are unsynchronized with respect to each other: finish never takes self.data's mutex.

What follows from a hit:

Arc::into_inner at src/storage-operators/src/persist.rs:218 sees count 2, returns None, and drops the last real handle.
The sweep's temporary then drops the BatchState, closing tx. The builder task completes normally and produces Some(batch), but its JoinHandle lives in the dropped BatchState, so the return value is discarded. The parts stay in blob until leaked-blob cleanup (with an un-consumed Batch ... dangling blob keys warning).
Every worker's write_shared_batch returned None, so apply_command sends WriteResponse { batch: None } and no worker emits into batches. In append::maybe_append_batches, batch_description is still Some(desc) and batches is empty, so compare_and_append_batch(&mut [], lower, upper, true) seals [lower, upper) with no data.
The corrections are not lost (consolidated_updates_before borrows rather than drains), but the subsequent advance_since(upper) rounds them to upper, so they are re-written at upper. The MV's contents at every timestamp in [lower, upper) are permanently missing those updates. That is observable by AS OF reads, by SUBSCRIBE, and by downstream indexes/MVs that already computed at those times; the self-correcting feedback loop cannot repair a sealed timestamp.

Fix — probe the count instead of materializing a reference:

-            state_map.retain(|_, weak| weak.upgrade().is_some());
+            // Do not `upgrade()` here: the temporary strong reference it creates can make a
+            // concurrent `SharedBatchBuilder::finish` observe a strong count above one and
+            // return `None`, orphaning a batch that nobody then appends.
+            state_map.retain(|_, weak| weak.strong_count() > 0);

Weak::strong_count is racy in the harmless direction only: a stale 0 cannot become live again, and a stale non-zero just keeps a dead entry for one more sweep.

2. MEDIUM -- `add_part` recomputes accumulated part sizes on every call, so a flush window costs O(n²)

src/persist-client/src/batch.rs:673

in_progress_goodbytes walks all of finished_parts and calls Part::goodbytes() on each, and add_part calls it twice per part while finished_parts grows monotonically until the blob target is reached. With the default 128 MiB persist_blob_target_size and the sink's 1 KiB-update parts, a narrow MV accumulates thousands of parts per flush window, so the shared builder task makes on the order of n² Part::goodbytes() calls per window — millions of them — before a single part is flushed.

Details

Part::goodbytes() is not free: it constructs an ArrayOrd for the key and the val on each call (src/persist-types/src/part.rs:43), which allocates a Vec per struct level and clones an ArrayRef per column. This is confined to the shared-batch path — add_part has no other caller, and in the add path finished_parts is always empty — but that is the path this PR introduces, and the cost concentrates exactly where batches are largest: hydration and backfill of a large MV.

Keeping a running byte count alongside the vector removes it:

add a finished_parts_goodbytes: usize field, bump it in finish_part_builder and in add_part's push, and zero it in flush_parts next to finished_parts.clear();
in_progress_goodbytes then becomes self.part_builder.goodbytes() + self.finished_parts_goodbytes.

antiguru · 2026-08-26T12:03:11Z

Both confirmed and fixed in the first commit. Branch is now at ae3b03bc7c (pr-35411) and a5cf162608 (pr-35411-barrier).

1. Cleanup sweep orphaning a finished batch. Confirmed, including the consequence. The sweep holds the registry mutex, finish never takes it, so the two are unsynchronized, and Arc::into_inner returns None on any strong count above one. Corrections survive the empty append because consolidated_updates_before borrows rather than drains, but the following advance_since(upper) rounds them to upper, so the interval's timestamps are permanently missing those updates rather than eventually repaired.

Applied the suggested fix, with the reasoning recorded at the call site:

// Probe the count rather than `upgrade()`: the temporary strong reference an upgrade
// creates can make a concurrent `SharedBatchBuilder::finish` see a strong count above
// one and hand the batch to nobody, which seals the interval with no data. A stale
// count is harmless in both directions, since zero cannot become live again and a
// stale non-zero only keeps a dead entry until the next sweep.
inner.states.retain(|_, weak| weak.strong_count() > 0);

Two other Weak::upgrade sites were checked and left alone. upgrade_or_init materializes a reference for a worker that is a genuine participant, so a finish losing the race to it still delivers the batch through the joining handle. note_skip (in the barrier follow-up) has the same shape but only runs in barrier mode, where finish_barrier resolves through the barrier rather than Arc::into_inner.

2. Quadratic in_progress_goodbytes. Confirmed: Part::goodbytes builds an ArrayOrd for the key and the val on every call, and add_part calls the sum twice per part. Added a finished_parts_goodbytes running counter, maintained in a new push_finished_part helper (used by both finish_part_builder and add_part) and zeroed next to finished_parts.clear() in flush_parts, so in_progress_goodbytes is part_builder.goodbytes() + finished_parts_goodbytes.

Two things also changed since the reviewed revision, both from the same pass:

The v1 sink rewiring is dropped. It was not flag-gated, and enable_compute_sync_mv_sink defaults off, so it would have changed the default sink's behavior. v1 now only carries PersistApi.persist_batches and BatchDescription.shared_id, both read solely by v2.
Dropping it exposed two defects that also affected the v2 shared path. PersistClient::batch_builder hardcoded "peek_stash" as the shard-metrics name, which is first-writer-wins and would have mislabeled a user shard for its lifetime; it now takes a shard_name. And the sink assembled Schemas { id: None, .. }, so parts were written with schema_id: None and could not be decoded through the shard's schema registry; the id now comes from WriteHandle::schema_id().

Reviewed and applied by Claude Code.

`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]>

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]>

antiguru · 2026-10-13T09:01:03Z


+        // Read the shared-batch flag on the Timely thread. `worker_config` is an `Rc<ConfigSet>`
+        // and thus not `Send`, so the value must be captured before it moves into the Tokio task.
+        let shared_batches_enabled = ENABLE_SYNC_MV_SINK_SHARED_BATCHES.get(&worker_config);


You could get a ConfigValHandle here using the handle function.

Done in cd58526. Both sinks pass a ConfigValHandle to SharedBatchWriter, and start reads it once per batch interval, so a flip applies without re-rendering the sink. Workers can disagree on an interval. That is safe, because each worker either contributes to the shared batch or writes its own, and the append operator appends all of them.

In #37617 the barrier mode stays fixed at render time. A worker that writes its own batch never reports to the barrier, so a per-interval read there could wedge it.

Posted by Claude Code.

antiguru · 2026-10-13T09:05:08Z

+                let proto_batch = if batch_writer.shared_batches_enabled {
+                    write_shared_batch(batch_writer, &desc, &mut updates_rx).await
+                } else {
+                    // Per-worker path: each worker writes its own batch for the interval. Create


Should we extract the per-worker path into a separate function, too?

Done in cd58526: write_worker_batch and write_shared_batch in materialized_view_v2.rs. The shared side's part building moved to sink::shared_batch, which the v1 sink now uses too.

Posted by Claude Code.

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]>

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 · 2026-10-13T14:15:17Z

Measurements for the shared-batch path. All runs used a single-process replica with 8 workers, the sync sink, local file blob storage and an optimized local build. The host was shared, so runs were gated on host load, and polluted runs are excluded.

Without a size threshold (`cd58526fb2`)

Routing every interval through the shared builder serializes encoding and upload in one task per process.

	Shared off	Shared on
Hydration of 108M rows / 9.55 GB (mean of 5)	26.2 s	33.1 s
Write phase after the snapshot	5.7 s	10.2 s
5M-row insert, time to propagate (2 runs)	2.65 s	3.6 s
Persist pool CPU during the write phase	~310%	~140%
Blob writes at hydration	80	78

At hydration, blob writes are unchanged because parts already reach the 128 MiB target.

With `compute_mv_sink_shared_batch_max_bytes` (`55efba7ebf`)

Each worker shares only while its updates for the interval stay below the threshold. "On, 0" spills every worker, which should behave like "off".

	Off	On, 1 MiB	On, 0
Hydration, 9.55 GB (mean of 7)	23.7 s	25.0 s	23.5 s
5M-row insert (3 runs)	2.5–3.15 s	3.4 s	3.15 s
1000-row writes at 5/s: data parts per append	7.8	1.05	7.8
1000-row writes: blob write time per 60 s	~70–78 s	8–13 s	~76–87 s
1000-row writes: p50 latency	108–117 ms	101–113 ms	128 ms
100-row writes at 20/s: inline parts per append	7.95	0.30	7.96
100-row writes: blob writes per append	0.63	1.26	0.64
100-row writes: p50 latency	71 ms	81 ms	74 ms
Compare-and-append calls	1 per append	1 per append	1 per append

With the threshold, hydration and large inserts go through per-worker batches and match the off mode. All modes wrote the same 80 blobs. The 1 MiB mean comes from two outlier runs on an identical persist path. Small writes coalesce to about one part per append.

Writes small enough to be inlined change shape. Eight inline parts per append become one shared blob, which costs one blob write and about 10 ms at p50, and keeps the data out of the consensus state.

The compare-and-append count is the same in every mode, because the append operator commits all batches of an interval in one call. What changes is the payload: with sharing, an append carries about one entry per process instead of one per worker. A measurement of consensus bytes per append at 4 processes × 32 workers with a trickle of updates is in progress and will follow.

Posted by Claude Code.

antiguru · 2026-10-13T14:47:50Z

This follow-up measures consensus bytes per append on a 128-worker replica (scale=4,workers=32) under a trickle of updates, without hydration. It ran on 84dcf56517, which includes #37617, with the sync sink, local CockroachDB and local file blob. The MV is SELECT k, v FROM t over an empty table, fed with one insert of N distinct keys per second for 60 s. Metrics are summed over the 4 clusterd processes. "Diff bytes" is mz_persist_shard_diff_size_bytes for the MV shard, which matches the compare-and-append payload to within 0.1%.

128 rows per interval (about one per worker), compaction on

Mode	Runs	Diff bytes / append	Inline parts / append	Inline bytes / append
Off	3	40.5k–41.3k	81.0	18.7k
Best-effort	4	18.2k–19.6k	9.4–13.2	8.4k–9.0k
Barrier (#37617)	3	16.4k–16.7k	4.00	7.7k

32 rows per interval, compaction on

Mode	Runs	Diff bytes / append	Inline parts / append
Off	2	13.5k–13.6k	28.0–28.4
Best-effort	3	6.8k–6.9k	6.9–7.0
Barrier	2	6.0k–6.1k	4.00

Compaction suppressed, one 30 s run each

Here every diff byte comes from compare-and-append.

Rows	Off	Best-effort	Barrier
128	20.9k	9.4k	8.6k
32	7.1k	3.6k	3.1k

Results

Without sharing, every worker that receives a row commits its own inline part, so the consensus payload of an append grows with the number of active workers.
Coalescing per process cuts consensus bytes per append by 2.2–2.5x. Most of that comes from best-effort coalescing alone.
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 best-effort stays within 10–20% of the barrier's bytes.
Compaction roughly doubles consensus bytes in every mode, and its merge count is the same across modes.
Compare-and-append calls per append are the same in all modes. SUBSCRIBE latency at p50 is 100–140 ms everywhere.

Caveats

128 workers on 32 cores is 4x oversubscribed, and straggling depends on timing, so the best-effort spread may differ on dedicated cores. One run on a loaded host is excluded.
At this rate each shared batch is about 1.9 KiB and stays inline. At higher rates it goes to blob, which costs one blob write per process per interval in place of the inline bytes.
The roughly 230 bytes per inline part depend on this row shape (bigint plus 32-character text).

Posted by Claude Code.

mtabebe

Ok, I'm trying to write down my understanding of the change here.

So the problem today is that we have workers that are writing the diffs based on changes to the MV, and these writes need to end up in persist shards. This work is spread across the workers. Consequently, each worker writes their own batch which results in small batches. These batches then result in work for consolidation and reads then need to contend with these small batches until consolidation happens.

This PR basically directs these writes to the a new shared buffer that effectively packs them into a single batch. This trades off some latency on the write becuase we need to wait for the interval to pass. But the benefit is less small data in the system to manage downstream (and hence less work on consolidation and reads). Because it also increases the size of writes we also get less inlining.

I had a few small nits, and one request for more tests. Otherwise, this is a good change. Thanks!

mtabebe · 2026-10-13T16:08:52Z

+        // disagree on an interval: each either contributes to the shared batch or writes its own,
+        // and the append operator appends all of them together.
+        if !self.enabled.get() {
+            return None;


I wonder if there is a cleaner way to handle this case, since it seems like now we have several ways of writing a batch and handle the old path.

Can the disabled case simply default to setting max_bytes = 0, and we fall into the block later on by default?

Done in 7728c76, slightly differently. start always returns a write now, and both sinks have a single write path. Setting max_bytes = 0 alone would still register a shared builder and spawn its task every interval, which a kill switch should avoid. So a disabled writer joins no shared batch, and every update goes straight to the worker's own batch.

Posted by Claude Code.

mtabebe · 2026-10-13T16:11:58Z

+            BatchState {
+                desc: desc.clone(),
+                tx,
+                handle: mz_ore::task::spawn(|| task_id, async move {


Does anything govern or bound how many of these tasks we can spawn? Should we have a limitation on this? Otherwise wouldn't it scale with number of sources? Maybe we just rely on the OS for this though

There is no explicit bound. A shared builder lives for one interval of one sink's output shard in one process, so the number of live tasks follows the number of MV sinks on the replica (not sources). Each sink worker writes one interval at a time, so a process holds roughly one live builder per sink. These are tokio tasks, not OS threads. Memory per builder is bounded by the channel capacity of 4 parts and by persist's batch builder limits. Each task replaces one batch builder per worker that the per-worker path would otherwise run.

Posted by Claude Code.

mtabebe · 2026-10-13T16:14:33Z

+//!
+//! One task per process encodes and uploads a shared batch, so routing large intervals through it
+//! serializes work the workers otherwise do in parallel. Hydrating a 9.5 GB MV on 8 workers against a
+//! local blob store took 33 s through the shared batch and 26 s with per-worker batches. Each worker therefore stages


Do numbers need to be in the code?

Removed in 7728c76. The doc states the tradeoff, and the numbers stay in the PR description.

Posted by Claude Code.

mtabebe · 2026-10-13T16:56:53Z

+    /// will include all data written by all workers.
+    /// This means that, for any batch interval that we actually want to append, all workers
+    /// must call finish even if they did not push any batches themselves.
+    pub async fn finish(self) -> Option<Batch<SourceData, (), Timestamp, StorageDiff>> {


Ok so what I think happens here is: every caller needs to call finish, and when that happens, we know
(1) anyone in this batch has the same lower/upper
(2) The callers data is in the channel
(3) We hand the complete batch back to exactly one caller, which is the last finisher
(4) That last finisher must still hold the write capability to make sure the frontier doesn't advance past the lower
(5) The last finisher then has the batch built, which will be written to persist by the append operator?

So we have the same basic properties as to when writes are visible and durable via the CAS.

Do we have tests for failure of the last finisher (either before it actually finishes or after finish but before emission)? Since this all seems to ride on behaviour of the last finisher as its responsibilities are now slightly bigger then a usual worker.

Your summary matches. Every worker holds its capability until its write returns, so the append frontier cannot pass the interval before the last finisher has emitted the batch. Nothing becomes durable until the CaS.

Process failure: the shared batch lives in one process, so a worker panic or crash takes all of the process's contributors down together. The restarted sink rewrites the interval from the shard's upper.

Teardown before finish: dropping the last handle aborts the builder task and delivers the batch to nobody.

A non-last worker dropping out: its pushed parts still reach the last finisher.

Teardown after finish, before the append: this is the same as an unappended per-worker batch today.

7728c76 adds unit tests for the last two cases, test_shared_batch_dropped_handle and test_shared_batch_abandoned. With the flag on by default, the restart and kill workloads in CI also run this path.

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]>

mtabebe

Thanks for the tweaks, new tests, and the answers to my questions 👍

bosconi

Some plausible Claudisms, but I defer to your judgement.

(We should definitely analyze its misses, so see if more background context can improve the situation. I'll start with a scorecard.)

bosconi · 2026-10-14T02:56:20Z

+/// interval, instead of each worker writing its own batch.
+pub const ENABLE_SYNC_MV_SINK_SHARED_BATCHES: Config<bool> = Config::new(
+    "enable_compute_sync_mv_sink_shared_batches",
+    true,


This turns the shared path on for every production MV on merge, including the default v1 sink (enable_compute_sync_mv_sink defaults to false). The repo convention is new flags off in production and on in CI, and the description lists a cost for small writes (an extra blob write per append, about +10 ms p50). You dropped the v1 rewiring in ae3b03b for the same reason.

Since 7728c76 the kill switch is also weaker: with the flag off, both sinks still go through SharedBatchWrite::push and own_builder, not the pre-PR WriteHandle::batch / WriteHandle::builder.

CI coverage doesn't need the production default: mzcompose already sets it to true in get_variable_system_parameters and parallel-workload flips it. false here keeps that coverage and leaves production to a separate parameter flip.

Suggested change

true,

false,

Agreed, the default is off as of c9cedc0, and production gets shared batches through a staged parameter rollout. 14d5cbd renames the flag to enable_compute_mv_sink_shared_batches, since both sinks read it. With the max_runs fix, the flag-off path builds through the same builder_inner/CompactConfig as WriteHandle::builder.

Posted by Claude Code.

Since then, enable_compute_mv_sink_shared_batches is folded into compute_mv_sink_shared_batch_max_bytes, which defaults to 0 (off).

Posted by Claude Code.

bosconi · 2026-10-14T02:56:21Z

+        self.persist_client.batch_builder(
+            self.shard_id,
+            &self.shard_name,
+            self.schemas.clone(),
+            desc.lower.clone(),
+            None,
+        )


A hardcoded None here overrides persist_batch_max_runs (PersistClient::batch_builder copies it into compact_cfg.batch.max_runs), and the shared builder task does the same (persist.rs:133-139). The pre-PR WriteHandle::batch / WriteHandle::builder honored the config. Since this is now also the flag-off path, no MV sink batch honors it. The two agree today only because the default of 1 maps to None anyway (batch.rs:474). Intentional, or should both builders read the config like WriteHandle::builder does?

Not intentional, fixed in 42c6ca5. batch_builder now treats max_runs as an override when Some and otherwise honors persist_batch_max_runs. Peek stash, the only other caller, always passes Some.

Posted by Claude Code.

bosconi · 2026-10-14T02:56:21Z

+                let batches =
+                    write_batches(batch_writer.shared.start(&desc), &mut updates_rx).await;


With the flag on, maybe_start_batch sends a WriteBatch for every ready interval, including empty ones, and this calls start before any data arrives. So every idle interval of every MV sink takes the process-wide SharedBatches mutex on each worker, allocates a PartBuilder, and spawns a builder task that builds nothing. v1 returns early when updates is empty (materialized_view.rs:1189-1194). Calling start on the first chunk would match that, and is safe, since a worker that never registers cannot be the last holder.

Done in 42c6ca5: write_batches starts the write on the first chunk, so idle intervals no longer take the lock or spawn a builder task. In #37617's barrier mode every worker still starts its write, because the barrier waits for every worker's report.

Posted by Claude Code.

bosconi · 2026-10-14T02:56:21Z

+        for _ in 0..5000 {
+            let batch_id = SharedBatchId::new();
+            let first = builder(batch_id);
+            let second = builder(batch_id);
+            first.push(bool_part(&schemas)).await;
+            let first = mz_ore::task::spawn(|| "finish-first", first.finish());
+            let second = mz_ore::task::spawn(|| "finish-second", second.finish());
+            let batches: Vec<_> = [first.await, second.await].into_iter().flatten().collect();
+            assert_eq!(
+                batches.len(),
+                1,
+                "exactly one finishing handle receives the batch"
+            );
+            for batch in batches {
+                batch.into_hollow_batch();
+            }
+        }


This test can't catch a sweep that evicts the live entry: only first pushes, so if second got a fresh state its task returns None and the count is still 1. Having second push on alternate iterations and asserting the update count catches that, and keeps the non-pushing case:

Suggested change

for _ in 0..5000 {

let batch_id = SharedBatchId::new();

let first = builder(batch_id);

let second = builder(batch_id);

first.push(bool_part(&schemas)).await;

let first = mz_ore::task::spawn(|| "finish-first", first.finish());

let second = mz_ore::task::spawn(|| "finish-second", second.finish());

let batches: Vec<_> = [first.await, second.await].into_iter().flatten().collect();

assert_eq!(

batches.len(),

1,

"exactly one finishing handle receives the batch"

);

for batch in batches {

batch.into_hollow_batch();

}

}

for i in 0..5000 {

let batch_id = SharedBatchId::new();

let first = builder(batch_id);

let second = builder(batch_id);

first.push(bool_part(&schemas)).await;

// When both handles push, a sweep that evicted the live entry between the two

// `builder` calls would split the updates across two batches.

let pushes = if i % 2 == 0 {

second.push(bool_part(&schemas)).await;

2

} else {

1

};

let first = mz_ore::task::spawn(|| "finish-first", first.finish());

let second = mz_ore::task::spawn(|| "finish-second", second.finish());

let batches: Vec<_> = [first.await, second.await].into_iter().flatten().collect();

assert_eq!(

batches.len(),

1,

"exactly one finishing handle receives the batch"

);

for batch in batches {

assert_eq!(

batch.into_hollow_batch().len,

pushes,

"the batch holds every handle's updates"

);

}

}

Applied in 42c6ca5.

Posted by Claude Code.

bosconi · 2026-10-14T02:56:21Z

+    /// into one flushed part, so an individual `part` must stay well below persist's
+    /// `KEY_VAL_DATA_MAX_LEN`: a single oversized part is flushed on its own, but nothing splits
+    /// it.
+    pub async fn add_part(&mut self, part: Part) -> Result<Added, InvalidUsage<T>> {


add_part is new public persist API with three untested branches: rejecting a part not beyond lower (UpdateNotBeyondLower), flushing when the accumulated parts would exceed blob_target_size, and flushing an oversized part on its own. The SharedBatches tests push single-update parts and testdrive can't see part counts. The fixtures in this file's mod tests cover all three in a few lines, and the second case also guards the finished_parts_goodbytes accounting.

Added in 42c6ca5: batch_builder_add_part_before_lower and batch_builder_add_part_flushing. The second covers accumulation, the flush before an overflowing part, an oversized part flushed alone, and the size reset, and fails if finished_parts_goodbytes is not reset after a flush.

Posted by Claude Code.

* `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]>

* 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]>

antiguru · 2026-10-14T18:37:59Z

Thanks for the reviews!

antiguru · 2026-10-14T18:38:51Z

And thanks to @bkirwi for starting this work!

Sets the compiled-in default of `compute_mv_sink_shared_batch_max_bytes` to 1 MiB, so MV sinks coalesce small per-worker writes into one batch per process (#37591) without a LaunchDarkly override. CI already runs with 1 MiB, and staging and internal production environments serve it today. Staging A/B results are on [CPU-30](https://linear.app/materializeinc/issue/CPU-30). Merge after the production LaunchDarkly default moves to 1 MiB. Until then `test/launchdarkly-flag-consistency` reports the compiled default as diverging from production. Part of [CPU-30](https://linear.app/materializeinc/issue/CPU-30). 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 <[email protected]>

Release notes for the **v26.47.0** train. One commit was added per snapshot; the `final` snapshot sets the real dates. **All six snapshots (`rc.1`–`rc.5`, `final`) are now applied — this branch is complete and ready for review.** - **Content**: 2 Feature sections (`TRY_CAST`, `EXCLUDE ALL NOT NULL CONSTRAINTS`), 7 Improvements, 13 Bug Fixes — `rc.1` contributed all of it except the top four Bug Fixes: `rc.2` added two, and `rc.3` and `rc.4` one each. `rc.5` added no new entries; it amended `rc.4`'s probe-port Bug Fix instead (see below). `final` adds nothing at all — its tracking issue holds only the unassociated `release: bump to version v26.47.0` commit ([`bed23494d`](bed2349)), and the `v26.47.0-rc.5` and `v26.47.0` tags are 14h28m apart with that bump the only thing between them. Its commit is empty and exists to record on the branch that the snapshot was applied. - **Dates (final, no longer provisional)**: Cloud **2026-10-22**, Self-Managed **2026-10-23**. Every RC estimated this same pair from the `v26.47.0-rc.1` tag, and the real dates confirmed it — the five RCs compressed into one week rather than pushing the train out — so no published date text changed in the `final` commit. ⚠️ One thing worth a last look before merge: **2026-10-22 is a Wednesday**, one day off the usual Thursday Cloud date (2026-10-23 is the normal Friday Self-Managed day). That is the supplied value and it stands, but if the release owner expected Thursday 2026-10-23 / Friday 2026-10-24, this branch is where to correct it. - **Operator compatibility**: row added for `v26.47`, release date `2026-10-23`. - **Branch base**: `main`, whose newest section is `## v26.45.1` — v26.46.0's section is still on its own unmerged branch (#39690). Expect a top-of-list conflict when both merge; `v26.47.0` goes above `v26.46.0`. ✅ **Corrections already on this branch** — a `final` verification pass over the whole assembled section confirmed these and deliberately left them alone: - **`rc.1` named a syntax that does not exist.** It published "`CREATE TABLE ... FROM SOURCE` now accepts an `ALL COLUMNS NULLABLE` option", taking the name from [#39680](#39680 title and body. At the `v26.47.0` tag the option is `EXCLUDE ALL NOT NULL CONSTRAINTS` (`TableFromSourceOptionName::ExcludeAllNotNullConstraints`), and `NULLABLE` is not a lexer keyword at all, so the published syntax would have been a parse error for anyone who copied it. A human editor caught this and replaced the bullet with the `{#v26.47-exclude-not-null-constraints}` feature section now on the branch. The `rc.1` snapshot directory in mz-skills still carries the wrong wording — snapshot directories are the record of what was reviewed at the time and are left untouched. - **`rc.5`'s cross-snapshot rewrite is applied once**, including the "Deployments using password, SASL, or OIDC authentication have no separate internal SQL port and continue to be probed on the external one" clause. - **PR citations are stripped from all entries** as a deliberate house-style edit, matching how published sections read. PR links remain available in the mz-skills snapshot drafts linked below. ⚠️ **Open questions for the release owner** — none of these were resolved by `final`; this branch is the last place to act on them - **Was any `rc.4` artifact handed to a customer or used for an upgrade rehearsal?** This decides a published entry. [#40036](#40036) (published as a Bug Fix by `rc.4`) made the operator probe `environmentd`'s internal SQL port unconditionally, which crashloops deployments using Password, SASL, or OIDC authentication — they have no internal SQL port. [#40046](#40046) fixed that roughly seven hours later, inside this same unreleased train, so no shipped build (`v26.46.1` probes the external port unconditionally) could have hit it. It is therefore recorded as **Borderline** rather than published as a Bug Fix of its own, and its substance folded into `rc.4`'s bullet as a scoping clause. If an `rc.4` build did reach anyone, the crashloop *was* reachable and [#40046](#40046) should be published in its own right. Note the author wrote an explicit positive release-note line for it, which is why this is flagged rather than quietly dropped. - **GA announcements — the largest judgment call in this train.** Three docs PRs remove preview markers: [#39831](#39831) (source versioning, `CREATE TABLE ... FROM SOURCE`), [#39836](#39836) (replacement materialized views), [#39830](#39830) (Kafka sink headers, noted as "GA in v26.47"). None changes code, so all three were held; drafted wording for each is in `omitted.md`. Which are actually GA in v26.47.0 and should be announced? Note v26.46.0 published Kafka sink message attribution as *private preview* — if the header goes GA here, this release should say so rather than repeat the preview framing. - **GCS blob read throughput** ([#39906](#39906), held in `rc.2`): restores the read throughput GCS deployments lost when [#39351](#39351) — already shipped — forced HTTP/2 and collapsed per-request connections into one. It is selected by `persist_blob_http_pool`, verified to default to `off` at the `v26.47.0-rc.2` tag, and the author notes "the default should eventually be 'GCS-only'". **Is `persist_blob_http_pool` being set to `gcs` for v26.47.0?** If so this is a clear win to publish for GCP users, who carry the regression today; drafted wording is in rc.2's `omitted.md`. - **`SUBSCRIBE` soft assertions** ([#37611](#37611), [#39764](#39764), both held): drafted as Bug Fixes, then pulled back. Both failures come from `soft_assert_or_log!`, which panics only where soft assertions are enabled and otherwise logs; neither PR says what a release build does. Are soft assertions on in the environments customers run? If yes, both are publishable as drafted in `omitted.md`. - **Console Freshness page** ([#39340](#39340), held): a complete new Monitoring > Freshness page, gated behind `freshness-page-CNS164`, off in both Cloud and self-managed, and the LaunchDarkly flag is not yet created. If it is turned on for v26.47.0, this likely deserves a Feature heading. - **PostgreSQL sources ingesting during a snapshot** ([#39419](#39419), [#39479](#39479), held): adding a large table to an existing source would no longer stall the tables already hydrated — but `storage_persist_sink_description_lookahead` is `Duration::ZERO` at the tag. Publish if the lookahead is being enabled. - **Coalesced materialized-view sink batches** ([#37591](#37591), held): measures ~67% fewer blob writes and ~54% smaller state diffs, at ~10 ms added p50 latency on small writes, but `compute_mv_sink_shared_batch_max_bytes` is `0` at the tag. Publish (with the latency cost stated) if the staged rollout reaches customers in this release. - **MCP 2026-07-28 conformance checks** ([#39694](#39694), [#39695](#39695), held): `enable_mcp_protocol_2026_07_28` is `false` at the tag. If the revision is being served, these belong together as one Improvement. - **Refresh strategy columns** ([#39597](#39597), published as an Improvement): this renames `aligned_to` → `aligned_to_expression` and `at` → `at_expression` and changes their type. It is a breaking change for anyone selecting them, in a schema documented as unstable — it may warrant more prominence than a bullet. - **Retained input after `ALTER SINK ... SET FROM`** (from `rc.3`'s [#37653](#37653)): the PR records a pre-existing gap it does not fix — after `SET FROM`, the *old* upstream relation stays held, and therefore un-compactable, until the sink is dropped or `environmentd` restarts. It is not something this release changed, so it is out of these notes, but it is a real footgun for the blue/green pattern that `sql/alter-sink.md` recommends `SET FROM` for. Should the `ALTER SINK` page or the blue/green guide call it out? - **Operator upgrade shows pending changes** (from `rc.4`'s [#40036](#40036)): the probe-port change alters the generated `environmentd` StatefulSet, so after upgrading the operator, existing environments report pending changes and pick up the new port on their next rollout — they are **not** restarted on their own. That is an upgrade consequence rather than something the release changed, so it is out of the notes, but a Self-Managed operator who cannot account for the pending-changes state may read it as a problem. The v26.47 upgrade guide looks like the right venue. Snapshot drafts (with PR links) in mz-skills: - rc.1: [release_notes.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.1/release_notes.md) / [omitted.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.1/omitted.md) - rc.2: [release_notes.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.2/release_notes.md) / [omitted.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.2/omitted.md) - rc.3: [release_notes.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.3/release_notes.md) / [omitted.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.3/omitted.md) - rc.4: [release_notes.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.4/release_notes.md) / [omitted.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.4/omitted.md) - rc.5: [release_notes.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.5/release_notes.md) / [omitted.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.5/omitted.md) - final: [release_notes.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/final/release_notes.md) / [omitted.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/final/omitted.md) **Borderline PRs omitted:** rc.1: 14 — [omitted.md#borderline](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.1/omitted.md#borderline), rc.2: 2 — [omitted.md#borderline](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.2/omitted.md#borderline), rc.3: 0 — [omitted.md#borderline](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.3/omitted.md#borderline), rc.4: 0 — [omitted.md#borderline](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.4/omitted.md#borderline), rc.5: 1 — [omitted.md#borderline](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.5/omitted.md#borderline), final: 0 — [omitted.md#borderline](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/final/omitted.md#borderline) 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude <[email protected]> Co-authored-by: Valerie Chiang <[email protected]> Co-authored-by: Pranshu Maheshwari <[email protected]>

antiguru force-pushed the pr-35411 branch 5 times, most recently from b01dfbe to dd6fa4a Compare July 13, 2026 18:50

antiguru mentioned this pull request Jul 13, 2026

compute: barrier discipline for MV sink shared batches #37617

Draft

antiguru marked this pull request as ready for review July 13, 2026 18:57

antiguru requested review from a team as code owners July 13, 2026 18:57

antiguru requested a review from DAlperin July 13, 2026 18:57

antiguru force-pushed the pr-35411 branch 3 times, most recently from afcdb73 to 1580c0f Compare August 26, 2026 08:41

antiguru force-pushed the pr-35411 branch from 1580c0f to ae3b03b Compare August 26, 2026 11:21

antiguru mentioned this pull request Oct 13, 2026

cluster / persist: Shared batches in the sink #35411

Closed

antiguru force-pushed the pr-35411 branch from ae3b03b to 022ee5d Compare October 13, 2026 08:42

bkirwi and others added 2 commits October 13, 2026 10:50

antiguru force-pushed the pr-35411 branch from 022ee5d to 339f0ec Compare October 13, 2026 08:53

antiguru commented Oct 13, 2026

View reviewed changes

antiguru and others added 3 commits October 13, 2026 11:41

compute: enable MV sink shared batches by default

55efba7

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]>

mtabebe reviewed Oct 13, 2026

View reviewed changes

mtabebe approved these changes Oct 13, 2026

View reviewed changes

bosconi approved these changes Oct 14, 2026

View reviewed changes

antiguru and others added 6 commits October 14, 2026 09:47

compute: rename flag to enable_compute_mv_sink_shared_batches

14d5cbd

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]>

compute: tighten MV sink shared batch comments

2ba4c94

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 merged commit 5d70e73 into MaterializeInc:main Oct 14, 2026
86 checks passed

antiguru deleted the pr-35411 branch October 14, 2026 18:38

bosconi mentioned this pull request Oct 16, 2026

v26.47.0 release notes #39891

Merged

antiguru mentioned this pull request Oct 21, 2026

compute: share MV sink batches below 1 MiB by default #40051

Merged

		let batches =
		write_batches(batch_writer.shared.start(&desc), &mut updates_rx).await;

Conversation

antiguru commented Jul 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

def- commented Aug 26, 2026

QA LLM Review

1. HIGH -- Cleanup sweep's Weak::upgrade can orphan a finished shared batch

2. MEDIUM -- add_part recomputes accumulated part sizes on every call, so a flush window costs O(n²)

Uh oh!

antiguru commented Aug 26, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

antiguru commented Oct 13, 2026

Without a size threshold (cd58526fb2)

With compute_mv_sink_shared_batch_max_bytes (55efba7ebf)

Uh oh!

antiguru commented Oct 13, 2026

128 rows per interval (about one per worker), compaction on

32 rows per interval, compaction on

Compaction suppressed, one 30 s run each

Results

Caveats

Uh oh!

mtabebe left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mtabebe left a comment

Choose a reason for hiding this comment

Uh oh!

bosconi left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

antiguru commented Oct 14, 2026

Uh oh!

Uh oh!

antiguru commented Jul 12, 2026 •

edited

Loading

1. HIGH -- Cleanup sweep's `Weak::upgrade` can orphan a finished shared batch

2. MEDIUM -- `add_part` recomputes accumulated part sizes on every call, so a flush window costs O(n²)

Without a size threshold (`cd58526fb2`)

With `compute_mv_sink_shared_batch_max_bytes` (`55efba7ebf`)