From 41ac8a656e933c282d488215c9aced972e5f7b66 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Wed, 23 Sep 2026 09:41:07 -0400 Subject: [PATCH] Chunk pack: combine a carried run once, when emitted `pack` grew its carry by `combine(&mut acc, next)`, which is linear only if appending leaves `acc` in place. Corgi's combine rebuilds its columns, so a carry absorbing many small chunks recopied itself each time: over scc's churn rounds it recopied 30.7M carried rows to append 2.3M new ones. The carry is now a run of chunks, combined once when emitted; between calls the run goes back to the input uncombined. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_013ABLpEcZUPN1oAUY2usfkx --- .../src/columnar/trace/chunk.rs | 11 +-- differential-dataflow/src/trace/chunk/mod.rs | 80 ++++++++----------- differential-dataflow/src/trace/chunk/vec.rs | 7 +- interactive/src/corgi/chunk.rs | 19 +---- 4 files changed, 46 insertions(+), 71 deletions(-) diff --git a/differential-dataflow/src/columnar/trace/chunk.rs b/differential-dataflow/src/columnar/trace/chunk.rs index 141b9b63b..2cbd6fd8d 100644 --- a/differential-dataflow/src/columnar/trace/chunk.rs +++ b/differential-dataflow/src/columnar/trace/chunk.rs @@ -436,17 +436,18 @@ where U::Time: 'static { } /// Maximal packing via the harness [`pack`](crate::trace::chunk::pack): coalesce by melding - /// the next trie onto the carry (adjacent chunks of a sorted, consolidated + /// each trie of the run onto the first (adjacent chunks of a sorted, consolidated /// chain, so meld's "strictly greater first triple" precondition holds), split /// with [`trie_merger::split_at`], and seal through [`seal_chunk`] (the spill /// point — pages a committed chunk when a spiller is installed). fn settle(input: &mut VecDeque, done: bool, out: &mut VecDeque) { crate::trace::chunk::pack( input, done, out, - |acc, next| { - let mut build = UpdatesBuilder::new_from(into_trie(std::mem::take(acc))); - build.meld(&into_trie(next)); - *acc = ColChunk::Resident(Rc::new(build.done())); + |run| { + let mut run = run.into_iter(); + let mut build = UpdatesBuilder::new_from(into_trie(run.next().unwrap())); + for next in run { build.meld(&into_trie(next)); } + ColChunk::Resident(Rc::new(build.done())) }, |chunk, n| { let (first, rest) = trie_merger::split_at(into_trie(chunk), n); diff --git a/differential-dataflow/src/trace/chunk/mod.rs b/differential-dataflow/src/trace/chunk/mod.rs index 20658cef1..2f73ca3f6 100644 --- a/differential-dataflow/src/trace/chunk/mod.rs +++ b/differential-dataflow/src/trace/chunk/mod.rs @@ -133,69 +133,57 @@ pub trait Chunk: Sized + Clone { /// Maximal-packing driver an implementor's [`Chunk::settle`] may delegate to. /// -/// Holds a `carry` chunk under construction, grown by `combine` until it reaches -/// `TARGET` (then emitted) and emitted early when the next chunk can't be absorbed -/// without exceeding `TARGET`; over-sized chunks are peeled with `split`. Each -/// committed chunk is passed through `seal` (the compress / spill hook — use the -/// identity closure when there's nothing to do). The closures are the only -/// layout-specific pieces: +/// Holds a `carry` run of chunks, grown until it reaches `TARGET` (then emitted as +/// one chunk) and emitted early when the next chunk can't be absorbed without +/// exceeding `TARGET`; over-sized chunks are peeled with `split`. Each committed +/// chunk is passed through `seal` (the compress / spill hook — use the identity +/// closure when there's nothing to do). The closures are the only layout-specific +/// pieces: /// -/// * `combine(&mut acc, next)` — append `next` onto `acc` (caller guarantees their -/// lengths sum to at most `TARGET`, and `next` follows `acc` in one sorted, -/// consolidated chain), so packing a run of small chunks stays linear. +/// * `combine(run)` — one chunk from a run of at least two (caller guarantees their +/// lengths sum to at most `TARGET`, and that they form one sorted, consolidated +/// chain). A run is combined once, when emitted, so packing stays linear even +/// when combining copies its inputs. /// * `split(chunk, n)` — the first `n` updates and the remaining `len - n`. /// * `seal(chunk)` — commit a chunk (e.g. compress or spill); identity to keep it. pub fn pack( input: &mut VecDeque, done: bool, out: &mut VecDeque, - mut combine: impl FnMut(&mut C, C), + mut combine: impl FnMut(Vec) -> C, mut split: impl FnMut(C, usize) -> (C, C), mut seal: impl FnMut(C) -> C, ) { - let mut carry: Option = None; + let (mut carry, mut carried) = (Vec::new(), 0); + let mut fuse = |carry: &mut Vec| if carry.len() == 1 { carry.pop().unwrap() } else { combine(std::mem::take(carry)) }; while let Some(chunk) = input.pop_front() { - match carry.take() { - None => pack_absorb(chunk, &mut carry, out, &mut split, &mut seal), - Some(mut c) if c.len() + chunk.len() <= C::TARGET => { - // Combines into one legal chunk; coalesce in place. - combine(&mut c, chunk); - if c.len() == C::TARGET { out.push_back(seal(c)); } else { carry = Some(c); } - } - Some(c) => { - // `c` is maximal against this neighbour; emit it and absorb afresh. - out.push_back(seal(c)); - pack_absorb(chunk, &mut carry, out, &mut split, &mut seal); - } - } - } - if let Some(c) = carry { - if done { out.push_back(seal(c)); } else { input.push_front(c); } - } -} - -/// Absorb `chunk` into an empty `carry` (a [`pack`] helper): pass a `TARGET` chunk -/// straight through (sealed), hold a smaller one as the new carry, or peel -/// `TARGET`-sized pieces off a larger one and carry the remainder. -fn pack_absorb(chunk: C, carry: &mut Option, out: &mut VecDeque, split: &mut S, seal: &mut L) -where - C: Chunk, - S: FnMut(C, usize) -> (C, C), - L: FnMut(C) -> C, -{ - match chunk.len().cmp(&C::TARGET) { - std::cmp::Ordering::Equal => out.push_back(seal(chunk)), - std::cmp::Ordering::Less => *carry = Some(chunk), - std::cmp::Ordering::Greater => { + if carried + chunk.len() <= C::TARGET { + // Combines into one legal chunk; absorb it. + carried += chunk.len(); + carry.push(chunk); + if carried == C::TARGET { out.push_back(seal(fuse(&mut carry))); carried = 0; } + } else if !carry.is_empty() { + // The carry is maximal against this neighbour; emit it and absorb afresh. + out.push_back(seal(fuse(&mut carry))); + carried = 0; + input.push_front(chunk); + } else { + // Peel `TARGET`-sized pieces off an over-sized chunk and absorb the remainder. let mut rest = chunk; - loop { + while rest.len() > C::TARGET { let (head, tail) = split(rest, C::TARGET); out.push_back(seal(head)); - if tail.len() >= C::TARGET { rest = tail; } - else { if tail.len() > 0 { *carry = Some(tail); } break; } + rest = tail; } + input.push_front(rest); } } + if done { + if !carry.is_empty() { out.push_back(seal(fuse(&mut carry))); } + } else { + // Hand the run back uncombined, to be combined once when it is emitted. + for chunk in carry.into_iter().rev() { input.push_front(chunk); } + } } /// A batch: an ordered [`Chunk`] sequence whose concatenation is its updates. diff --git a/differential-dataflow/src/trace/chunk/vec.rs b/differential-dataflow/src/trace/chunk/vec.rs index 735458f47..4fecc23b2 100644 --- a/differential-dataflow/src/trace/chunk/vec.rs +++ b/differential-dataflow/src/trace/chunk/vec.rs @@ -251,13 +251,12 @@ where K: Ord+Clone+'static, V: Ord+Clone+'static, T: Lattice+Timestamp, R: Semig } /// Maximal packing via the harness [`pack`](super::pack): coalesce by - /// extending the inner `Vec` in place (`make_mut` is free while the carry's - /// `Rc` is unique, so packing a run of small chunks stays linear), split with - /// `split_off`, and seal as a no-op (`Vec` chunks are never paged). + /// concatenating the run's `Vec`s, split with `split_off`, and seal as a + /// no-op (`Vec` chunks are never paged). fn settle(input: &mut VecDeque, done: bool, out: &mut VecDeque) { super::pack( input, done, out, - |acc, next| Rc::make_mut(&mut acc.0).extend(take(next)), + |run| VecChunk(Rc::new(run.into_iter().flat_map(take).collect())), |chunk, n| { let mut rows = take(chunk); let rest = rows.split_off(n); (VecChunk(Rc::new(rows)), VecChunk(Rc::new(rest))) }, |chunk| chunk, ); diff --git a/interactive/src/corgi/chunk.rs b/interactive/src/corgi/chunk.rs index a78a38ad3..db2a7a949 100644 --- a/interactive/src/corgi/chunk.rs +++ b/interactive/src/corgi/chunk.rs @@ -371,22 +371,9 @@ where input, done, out, - |acc, next| { - let (na, nb) = (acc.len_(), next.len_()); - let kvs = [acc.kv(), next.kv()]; - let srcs = [Some(&kvs[0]), Some(&kvs[1])]; - let mut tags = Vec::with_capacity(na + nb); - let mut offs = Vec::with_capacity(na + nb); - for o in 0..na { tags.push(0); offs.push(o); } - for o in 0..nb { tags.push(1); offs.push(o); } - let kv = gather_lanes(&srcs, &tags, &offs); - let mut times = ColTimes::new(); - times.reserve(acc.times().width().max(next.times().width()), na + nb); - times.push_range(acc.times(), 0, na); - times.push_range(next.times(), 0, nb); - let mut diffs = acc.diffs().to_vec(); - diffs.extend_from_slice(next.diffs()); - *acc = Self::from_kv(kv, times, diffs); + |run| { + let (kv, times, diffs) = Self::concat(&run); + Self::from_kv(kv, times, diffs) }, |chunk, m| { let kv = chunk.kv();