From 8eff54398179966acb5ccd0ac26901aa97170561 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sun, 12 Jul 2026 15:51:54 -0400 Subject: [PATCH] MergeBatcher: extract-first seal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit seal was merge-then-extract: every chain was drained into one before the frontier split, so updates parked beyond 'upper' (e.g. delay/enter_at bands, future iterations) were recopied on EVERY seal. Instrumenting seal showed drained-over-shipped ratios of 35-38x on the enter_at-band arrangements of an SCC benchmark (10.3M rows drained per seal stream to ship 293k), against 1.0-1.8x for operators without future-dated residue. seal now extracts each chain in place and merges only the shipped parts — delta-sized, and the merge is where cross-chain consolidation happens (the ship/keep split is by time, so no equal (data, time) group can straddle it). Kept chains re-enter the ladder as-is, sorted by weight with the geometric invariant restored, so residue is copied only at doubling events. Measured on the SCC suite: DDIR-corgi -15%, DDIR-vec -11%, compiled native (whose extract still copies rows; only the drain removal applies) -2.5%. Co-Authored-By: Claude Fable 5 --- .../trace/implementations/merge_batcher.rs | 59 +++++++++++++------ 1 file changed, 42 insertions(+), 17 deletions(-) diff --git a/differential-dataflow/src/trace/implementations/merge_batcher.rs b/differential-dataflow/src/trace/implementations/merge_batcher.rs index df09fb41a..00975fb25 100644 --- a/differential-dataflow/src/trace/implementations/merge_batcher.rs +++ b/differential-dataflow/src/trace/implementations/merge_batcher.rs @@ -63,25 +63,50 @@ where // which we call `lower`, by assumption that after sealing a batcher we receive no more // updates with times not greater or equal to `upper`. fn seal(&mut self, upper: Antichain) -> (Vec, Description) { - // Merge all remaining chains into a single chain. - while self.chains.len() > 1 { - let list1 = self.chain_pop().unwrap(); - let list2 = self.chain_pop().unwrap(); - let merged = self.merge_by(list1, list2); - self.chain_push(merged); - } - let merged = self.chain_pop().unwrap_or_default(); - - // Extract readied data. - let mut kept = Vec::new(); - let mut readied = Vec::new(); + // Split each chain by `upper` without draining the ladder: only the shipped parts + // need linear (merged, consolidated) form for the builder. Kept updates (e.g. + // delayed far beyond `upper`) stay in their chains, and are copied only when the + // ladder's geometric maintenance merges them. self.frontier.clear(); - - self.merger.extract(merged, upper.borrow(), &mut self.frontier, &mut readied, &mut kept, &mut self.stash); - - if !kept.is_empty() { - self.chain_push(kept); + let mut ship_chains: Vec> = Vec::new(); + let mut kept_chains: Vec> = Vec::new(); + while let Some(chain) = self.chain_pop() { + let mut kept = Vec::new(); + let mut ship = Vec::new(); + self.merger.extract(chain, upper.borrow(), &mut self.frontier, &mut ship, &mut kept, &mut self.stash); + if !kept.is_empty() { + kept_chains.push(kept); + } + if !ship.is_empty() { + ship_chains.push(ship); + } + } + // Re-ladder the residue through the standard insertion, largest first: each + // `insert_chain` cascades tail merges until the geometric invariant holds, so + // inserting into the (currently empty) chain list maintains it inductively. + kept_chains.sort_by_key(|c| std::cmp::Reverse(c.iter().map(M::len).sum::())); + for kept in kept_chains { + self.insert_chain(kept); + } + // Merge the shipped chains smallest pair first, so total copying stays near-linear + // in shipped rows even when the readied set is large (pending updates eventually + // ship, and one frontier advance can release a whole residue). These merges also + // provide cross-chain consolidation of shipped updates; equal (data, time) updates + // cannot straddle the ship/keep split (it is by time), so this suffices. + let mut ship_chains: Vec<(usize, Vec)> = ship_chains + .into_iter() + .map(|c| (c.iter().map(M::len).sum::(), c)) + .collect(); + ship_chains.sort_by_key(|&(w, _)| std::cmp::Reverse(w)); + while ship_chains.len() > 1 { + let (_, a) = ship_chains.pop().unwrap(); + let (_, b) = ship_chains.pop().unwrap(); + let merged = self.merge_by(a, b); + let w = merged.iter().map(M::len).sum::(); + let pos = ship_chains.partition_point(|&(cw, _)| cw > w); + ship_chains.insert(pos, (w, merged)); } + let readied = ship_chains.pop().map(|(_, c)| c).unwrap_or_default(); self.stash.clear();