Repository navigation
Conversation
|
Thanks for opening this PR! Here are a few tips to help make the review process smooth for everyone. PR title guidelines
Pre-merge checklist
|
Replace `Pipeline` with a chunked round-robin exchange on the `desired_input` in `write_batches`. With many workers, `Pipeline` causes each worker to write its own small batch per timestamp, increasing persist overhead. The chunked exchange concentrates updates onto fewer workers, producing fewer but larger batches. The new `ChunkedRoundRobinDistributor` in `mz-timely-util` sends `chunk_size` full containers to one target before rotating to the next. Different timestamps start at different targets via a rotating seed to avoid permanent skew. Controlled by a new `storage_persist_sink_chunk_size` dyncfg (default 1). Set to 0 to fall back to `Pipeline`. Co-Authored-By: Claude Opus 4.6 <[email protected]>
|
I looked into whether we could do the same in compute, but it's much harder. In compute, we have an operator that accepts data from a dataflow, and (negated) data from persist, and emits corrections according to some batch description. The batches are supposed to bring the persist input to the batche's upper, at which point the operator can advance the since, and potentially consolidate most of its data. The data hash-partitions its inputs to ensure that records that would cancel are at the same worker. This also means that the batches are similarly hash-partitioned, although there is no requirement that they are. We'd like to take the batch and re-shard it to different workers. This is difficult because the contents of the batch are somewhat independent of the frontiers. Each operator knows the persist and desired frontiers, and the batches it's supposed to write. It can write a batch when it brings the persist frontier to the desired frontier, and it knows exactly that it has seen all data from the two inputs. Extracting this into a separate operator is tricky because we need to maintain the same behavior, but lose context if we're not careful. Specifically, we need to indicate the end of updates without releasing a capability, which is at odds with how we usually indicate progress to downstream operators. |
Replace
Pipelinewith a chunked round-robin exchange on thedesired_inputinwrite_batches.With many workers,
Pipelinecauses each worker to write its own small batch per timestamp, increasing persist overhead.The chunked exchange concentrates updates onto fewer workers, producing fewer but larger batches.
The
ChunkedRoundRobinDistributoruses a singleContainerBuilderand shipschunk_sizefull containers to one target before rotating to the next.The seed determines a fixed initial target per source, and on every flush the target resets to it so that all workers make consistent routing decisions regardless of flush count.
Controlled by a new
storage_persist_sink_chunk_sizedyncfg (default 1).Set to 0 to fall back to
Pipeline.Changes
src/timely-util/src/pact.rs—ChunkedRoundRobinDistributorimplementing timely'sDistributor<C>trait, pluschunked_exchangeconvenience constructor.src/storage-types/src/dyncfgs.rs— NewSTORAGE_PERSIST_SINK_CHUNK_SIZEdyncfg.src/storage/src/render/persist_sink.rs— Wire up the exchange inwrite_batches, withPipelinefallback when chunk size is 0.🤖 Generated with Claude Code