From 986b2b782754d1c23cee64e80ea740fcbdc05181 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sun, 21 Jun 2026 12:29:50 -0400 Subject: [PATCH] merge_batcher: weigh the geometric ladder by updates; split Merger::account MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two related changes to the `MergeBatcher`, lifted out of the chunk_basis work so they can land independently of the `Chunk` module that motivates them. Weigh-by-updates. The geometric chain ladder previously compared chains by their *chunk count* (`chain.len()`). That is only a proxy for update count while every chunk is the same size; once a backend regrades — re-melding chunks so size and count decouple — a trickle of single-update chunks re-merges the head chain on every insert. Weigh chains by summed updates instead. A chain is immutable until merged, so its weight is computed once at push and cached alongside it (`chains: Vec<(usize, Vec)>`). Neutral for the existing `vec` backend (uniform chunk sizes make count and update-weight proportional); the behaviour change only bites a regrading backend. Refocus the Merger trait. The bundled `account() -> (records, size, capacity, allocations)` splits into `len() -> usize` (update count — drives the ladder and the logger's `records` field) and a defaulted `allocation() -> (size, capacity, allocations)` for memory telemetry. The logger tuple is reassembled verbatim via a private `record` helper, so `BatcherEvent`'s shape and the emitted figures are unchanged. NOTE: breaking change for out-of-tree `Merger` implementors (e.g. Materialize) — rename `account` -> `len`, optionally override `allocation`. The only in-tree impl (`vec::VecMerger`) is migrated here. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../trace/implementations/merge_batcher.rs | 54 +++++++++++++------ 1 file changed, 39 insertions(+), 15 deletions(-) diff --git a/differential-dataflow/src/trace/implementations/merge_batcher.rs b/differential-dataflow/src/trace/implementations/merge_batcher.rs index 1c7b1ca07..df09fb41a 100644 --- a/differential-dataflow/src/trace/implementations/merge_batcher.rs +++ b/differential-dataflow/src/trace/implementations/merge_batcher.rs @@ -17,10 +17,14 @@ use crate::trace::{Batcher, Description}; /// Creates batches from chunks of sorted, consolidated tuples. pub struct MergeBatcher { - /// A sequence of power-of-two length lists of sorted, consolidated containers. + /// Sorted, consolidated chains, each paired with its cached summed update count. + /// + /// The cached count is the chain's *merge weight*: the geometric ladder weighs + /// chains by updates, not chunk counts, since regrading decouples the two. A + /// chain is immutable until merged, so the weight is computed once at push. /// /// Do not push/pop directly but use the corresponding functions ([`Self::chain_push`]/[`Self::chain_pop`]). - chains: Vec>, + chains: Vec<(usize, Vec)>, /// Stash of empty chunks, recycled through the merging process. stash: Vec, /// Merges consolidated chunks, and extracts the subset of an update chain that lies in an interval of time. @@ -100,12 +104,12 @@ impl PushInto for MergeBatcher { } impl MergeBatcher { - /// Insert a chain and maintain chain properties: Chains are geometrically sized and ordered - /// by decreasing length. + /// Insert a chain and maintain chain properties: Chains are geometrically sized + /// (by summed updates) and ordered by decreasing update weight. fn insert_chain(&mut self, chain: Vec) { if !chain.is_empty() { self.chain_push(chain); - while self.chains.len() > 1 && (self.chains[self.chains.len() - 1].len() >= self.chains[self.chains.len() - 2].len() / 2) { + while self.chains.len() > 1 && (self.chains[self.chains.len() - 1].0 >= self.chains[self.chains.len() - 2].0 / 2) { let list1 = self.chain_pop().unwrap(); let list2 = self.chain_pop().unwrap(); let merged = self.merge_by(list1, list2); @@ -126,16 +130,27 @@ impl MergeBatcher { /// Pop a chain and account size changes. #[inline] fn chain_pop(&mut self) -> Option> { - let chain = self.chains.pop(); - self.account(chain.iter().flatten().map(M::account), -1); - chain + let (_weight, chain) = self.chains.pop()?; + self.account(chain.iter().map(Self::record), -1); + Some(chain) } /// Push a chain and account size changes. + /// + /// Caches the chain's summed update count alongside it for the ladder. #[inline] fn chain_push(&mut self, chain: Vec) { - self.account(chain.iter().map(M::account), 1); - self.chains.push(chain); + let weight = chain.iter().map(M::len).sum(); + self.account(chain.iter().map(Self::record), 1); + self.chains.push((weight, chain)); + } + + /// The `(records, size, capacity, allocations)` logger tuple for one chunk, + /// assembled from the two focused `Merger` methods. + #[inline] + fn record(chunk: &M::Chunk) -> (usize, usize, usize, usize) { + let (size, capacity, allocations) = M::allocation(chunk); + (M::len(chunk), size, capacity, allocations) } /// Account size changes. Only performs work if a logger exists. @@ -189,8 +204,19 @@ pub trait Merger: Default { stash: &mut Vec, ); - /// Account size and allocation changes. Returns a tuple of (records, size, capacity, allocations). - fn account(chunk: &Self::Chunk) -> (usize, usize, usize, usize); + /// The number of updates in a chunk. + /// + /// Drives the geometric ladder (chains are weighed by summed updates, not chunk + /// counts, since regrading decouples the two) and the `records` field of the + /// size logger. + fn len(chunk: &Self::Chunk) -> usize; + + /// Backing-allocation figures for a chunk: `(size, capacity, allocations)`, for + /// the size logger's memory telemetry. + /// + /// Defaults to zero — most chunk types do not track this. Override to report + /// real figures (e.g. Materialize's memory accounting). + fn allocation(_chunk: &Self::Chunk) -> (usize, usize, usize) { (0, 0, 0) } } /// A `Merger` implementation for vector update containers. @@ -351,8 +377,6 @@ pub mod vec { if !ready.is_empty() { ship.push(ready); } } - fn account(chunk: &Vec<(D, T, R)>) -> (usize, usize, usize, usize) { - (chunk.len(), 0, 0, 0) - } + fn len(chunk: &Vec<(D, T, R)>) -> usize { chunk.len() } } }