From 74db2a0f140bfc7760193f77b9d163205cf738e0 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 22 Sep 2026 19:07:12 -0400 Subject: [PATCH] Corgi reduce: present in bounded windows Cover a retire's keys in windows of `window_size` input records (`1 << 12` by default, as `VecReduceBackend` uses; `with_window` to choose), as the `next_window` contract intends, rather than presenting every key at once. The first window searches the input chunks for all of the retire's keys and counts the records each holds; each window then takes the next keys until it holds the budget, never splitting a key, and hands the presentations its share of the matches. Seeds are recorded per window, and the input id pool is cleared per window, as no input id outlives its window. Same-plan harness, 1 worker, median of 3 (before -> after): count 1M rows, 1000 x 500 load 81.6 -> 80.0 ms churn 3.44 -> 3.48 ms/round min 1M rows, 1000 x 500 load 81.8 -> 79.1 ms churn 2.60 -> 2.68 distinct 1M rows, 1000 x 500 load 81.4 -> 78.7 ms churn 2.34 -> 2.40 count2 1M rows, 10000 x 60 load 87.1 -> 85.0 ms churn 21.7 -> 21.7 min2 1M rows, 10000 x 60 load 95.6 -> 95.0 ms churn 20.1 -> 20.0 reach 2M edges, 1000 x 100 load 674 -> 683 ms churn 75.5 -> 76.6 scc 200k edges, 100 x 50 load 667 -> 673 ms churn 34.7 -> 35.1 Peak RSS for a 4M-row load: count 805 -> 646 MiB, count2 846 -> 770 MiB, distinct 804 -> 645 MiB; reach 638 -> 630 MiB. Forced to a single window, this code is +0.8% to +2.0% on churn against before, so most of the churn cost is the per-retire bookkeeping rather than the number of windows. The comment this replaces recorded bounded windows as measured and rejected (scc server session at 1 << 14: 84.4s against 63.7s). That session now runs 3.57s before and 3.60s after, at 1 << 12, 1 << 14, or one window alike. Its peak RSS (284 MiB before; 313, 300, 305 MiB after) moves as much between builds with no memory change intended, so I don't read it as signal. Co-Authored-By: Claude Opus 5.5 (1M context) --- interactive/src/corgi/reduce.rs | 289 ++++++++++++++++++++++++-------- 1 file changed, 220 insertions(+), 69 deletions(-) diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 618c4e1f8..ae5f324f3 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -28,6 +28,7 @@ use std::cmp::Reverse; use std::collections::BinaryHeap; use std::collections::HashMap; use std::hash::{BuildHasherDefault, Hasher}; +use std::ops::Range; use std::rc::Rc; use differential_dataflow::consolidation::{consolidate, consolidate_updates}; @@ -198,16 +199,40 @@ pub struct CorgiReduceBackend { vals: IdPool, /// Output IDs, times, and diffs accumulated until `finish`. rows: (Vec, Vec, ColTimes, Vec), + /// Input records per window, bounding what the presentations cost at once. + window_size: usize, + /// The retire in progress, as its first window found it. + retire: Retire, +} + +/// A retire's keys and where its input chunks hold them: searched once, by the first window, +/// and presented a window at a time, each window a prefix of the keys not yet presented. +#[derive(Default)] +struct Retire { + /// The retire's keys, ascending. + keys: Vec, + /// The input records each key holds, novel and prior. + held: Vec, + /// Each input chunk's `(key index, rows)` matches, in key order; prior chunks first. + input: Vec)>>, + /// The first key not yet presented. + next: usize, } impl CorgiReduceBackend { - pub fn new(reducer: Reducer) -> Self { + /// A backend covering the key space in windows of `1 << 12` input records. + pub fn new(reducer: Reducer) -> Self { Self::with_window(reducer, 1 << 12) } + + /// A backend with an explicit window budget, in presented input records. + pub fn with_window(reducer: Reducer, window_size: usize) -> Self { CorgiReduceBackend { reducer, input: IdPool::default(), keys: IdPool::default(), vals: IdPool::default(), rows: (Vec::new(), Vec::new(), ColTimes::default(), Vec::new()), + window_size: window_size.max(1), + retire: Retire::default(), } } } @@ -272,6 +297,10 @@ fn leaf_fields<'a>(col: &'a CValue, into: &mut Vec<&'a [u64]>) -> bool { } /// Each chunk's `(key index, rows)` matches of the ascending `keys`, in key order. +/// +/// Both the keys and stored identifier lane are sorted. Match them with +/// monotone positions, galloping over long gaps and stepping through adjacent +/// keys. The same compiled search covers narrow updates and broad cascades. fn search(chunks: &[&CorgiChunk], keys: &[u64]) -> Vec)>> { chunks.iter().map(|chunk| { if chunk.diffs().is_empty() { return Vec::new(); } @@ -282,12 +311,8 @@ fn search(chunks: &[&CorgiChunk], keys: &[u64]) -> Vec(chunks: &[&CorgiChunk], changed: &[u64]) -> (CValue, CValue, Vec, Vec, Vec, Vec) +/// ASCENDING set of changed key ids, and `matches` their rows in each chunk, as [`search`] finds. +fn collect_present(chunks: &[&CorgiChunk], changed: &[u64], matches: &[Vec<(usize, Range)>]) -> (CValue, CValue, Vec, Vec, Vec, Vec) where T: ColTime, { @@ -298,12 +323,7 @@ where let mut run_ends = Vec::new(); for (ci, ch) in chunks.iter().enumerate() { let before = khs.len(); - if ch.diffs().is_empty() { - continue; - } - let lane = key_lane(ch.keys()); - let kh = corgi::arrange::leaf_slice(lane).expect("the identifier lane is a u64 leaf"); - for (j, range) in MatchingRanges::new(changed, kh) { + for (j, range) in matches[ci].iter().cloned() { for i in range { tags.push(ci); offs.push(i); @@ -420,6 +440,32 @@ where batches.iter().flat_map(|b| b.chunks.iter()).collect() } +/// A retire's keys: the hashes the novel batches touch, merged with the `changed` set the harness +/// supplies. +fn retire_keys(novel_chunks: &[&CorgiChunk], changed: &[u64]) -> Vec { + let mut keys: Vec = novel_chunks.iter().flat_map(|ch| key_ids(ch.keys())).collect(); + keys.sort_unstable(); + keys.dedup(); + if !changed.is_empty() { + // Both sides ascend, so this is a merge. + let mut merged: Vec = Vec::with_capacity(keys.len() + changed.len()); + let (mut a, mut b) = (0usize, 0usize); + while a < keys.len() || b < changed.len() { + let key = match (keys.get(a), changed.get(b)) { + (Some(x), Some(y)) => *x.min(y), + (Some(x), None) => *x, + (None, Some(y)) => *y, + (None, None) => unreachable!("loop condition ensures one is present"), + }; + if keys.get(a) == Some(&key) { a += 1; } + if changed.get(b) == Some(&key) { b += 1; } + merged.push(key); + } + keys = merged; + } + keys +} + impl CorgiReduceBackend where T: ColTime + Ord, @@ -434,9 +480,10 @@ where &mut self, chunks: &[&CorgiChunk], keys: &[u64], + matches: &[Vec<(usize, Range)>], bridge: &mut ProxyBridge, ) { - let (p_keys, p_vals, khs, mut times, diffs, run_ends) = collect_present(chunks, keys); + let (p_keys, p_vals, khs, mut times, diffs, run_ends) = collect_present(chunks, keys, matches); if khs.is_empty() { return; } @@ -462,6 +509,7 @@ where &mut self, chunks: &[&CorgiChunk], keys: &[u64], + matches: &[Vec<(usize, Range)>], bridge: &mut ProxyBridge, ) -> bool { let Some(first) = chunks.iter().find(|chunk| !chunk.diffs().is_empty()) else { return true }; @@ -490,7 +538,6 @@ where let order = |(ca, ra): (usize, usize), (cb, rb): (usize, usize)| { (0..width).map(|f| leaves[ca][f][ra].cmp(&leaves[cb][f][rb])).find(|o| o.is_ne()).unwrap_or(std::cmp::Ordering::Equal) }; - let matches = search(chunks, keys); // At most one record per matched row, as in `merge_present`. bridge.reserve(matches.iter().flatten().map(|(_, rows)| rows.len()).sum()); let mut cursors = vec![0; chunks.len()]; @@ -707,72 +754,67 @@ where } fn next_window(&mut self, instance: &ReduceInstance<'_, T, CBatch, CBatch>, changed: &[u64], from: &mut KeyPosition, window: &mut ReduceWindow) { - // Single window: present the WHOLE key space at once, and report it covered. This is NOT a - // deferred refinement — bounded windows were measured and rejected: at WINDOW = 1<<14, scc - // (100 rounds x batch 100) cost 84.4s against 63.7s, a 33% regression, while peak RSS - // fell only 356MB -> 340MB. Two reasons: the per-window, per-chunk seek setup is a - // fixed cost that multiplies by the window count, and the presentation is not the - // memory peak in the first place (the trace is). + // Bounded windows: each takes the retire's next keys until it holds `window_size` input + // records, never splitting a key, as its presentations are all live at once. if *from == KeyPosition::End { return; } - *from = KeyPosition::End; - - // The window's keys: the hashes the novel batches touch, merged with the `changed` set the - // harness supplies. The novel hashes come from the scan the presentation needs anyway — the - // separate seeding pass this replaced read the delta a second time to derive them. let novel_chunks = chunks_of(instance.input_batches); - let mut keys: Vec = Vec::new(); - // The seeds are the novel batches' RAW (key_hash, time) support, recorded here — before the - // merged presentation below, whose consolidation may net a novel record away entirely. The - // key hashes come from the scan the key list needs anyway. - let mut seeds: Vec<(u64, T)> = Vec::with_capacity(novel_chunks.iter().map(|c| c.diffs().len()).sum()); - for ch in novel_chunks.iter() { - let khs = key_ids(ch.keys()); - let times = ch.times(); - for (i, kh) in khs.iter().enumerate() { - seeds.push((*kh, times.get(i))); - } - keys.extend(khs); + let mut in_chunks = chunks_of(instance.source_batches); + let prior = in_chunks.len(); + in_chunks.extend(novel_chunks.iter().copied()); + if *from == KeyPosition::Start { + let keys = retire_keys(&novel_chunks, changed); + let input = search(&in_chunks, &keys); + let mut held = vec![0; keys.len()]; + for (index, rows) in input.iter().flatten() { held[*index] += rows.len(); } + self.retire = Retire { keys, held, input, next: 0 }; } - seeds.sort_unstable_by(|a, b| a.cmp(b)); - seeds.dedup(); - window.seeds = seeds; - keys.sort_unstable(); - keys.dedup(); - if !changed.is_empty() { - // Both sides ascend, so this is a merge. - let mut merged: Vec = Vec::with_capacity(keys.len() + changed.len()); - let (mut a, mut b) = (0usize, 0usize); - while a < keys.len() || b < changed.len() { - let key = match (keys.get(a), changed.get(b)) { - (Some(x), Some(y)) => *x.min(y), - (Some(x), None) => *x, - (None, Some(y)) => *y, - (None, None) => unreachable!("loop condition ensures one is present"), - }; - if keys.get(a) == Some(&key) { a += 1; } - if changed.get(b) == Some(&key) { b += 1; } - merged.push(key); - } - keys = merged; + let mut retire = std::mem::take(&mut self.retire); + let (start, mut stop, mut records) = (retire.next, retire.next, 0); + while stop < retire.keys.len() && (records < self.window_size || retire.held[stop] == 0) { + records += retire.held[stop]; + stop += 1; } - if keys.is_empty() { + if stop == start { + *from = KeyPosition::End; return; } + let keys = &retire.keys[start..stop]; + // Each chunk's matches within the window, indexed from its first key: all of them, when the + // window is the whole retire. + let matches: Vec> = if keys.len() == retire.keys.len() { std::mem::take(&mut retire.input) } else { + retire.input.iter().map(|list| { + let (lo, hi) = (list.partition_point(|m| m.0 < start), list.partition_point(|m| m.0 < stop)); + list[lo..hi].iter().map(|(index, rows)| (index - start, rows.clone())).collect() + }).collect() + }; + + // The seeds are the novel batches' RAW (key_hash, time) support, recorded here — before the + // merged presentation below, whose consolidation may net a novel record away entirely. The + // rows come from the retire's search, which the presentation needs anyway. + window.seeds.reserve(matches[prior..].iter().flatten().map(|(_, rows)| rows.len()).sum()); + for (chunk, found) in in_chunks[prior..].iter().zip(&matches[prior..]) { + let times = chunk.times(); + for (index, rows) in found { + window.seeds.extend(rows.clone().map(|row| (keys[*index], times.get(row)))); + } + } + window.seeds.sort_unstable(); + window.seeds.dedup(); // ONE merged input presentation: novel and prior together, netted by the consolidation — // equal values share an id, so an exactly cancelling pair vanishes here, and // its time survives in `window.seeds` above. The input pool resolves values - // needed by Min and Collect. - let mut in_chunks = chunks_of(instance.source_batches); - in_chunks.extend(novel_chunks.iter().copied()); - if !self.present_input_merged(&in_chunks, &keys, &mut window.input) { - self.present_input(&in_chunks, &keys, &mut window.input); + // needed by Min and Collect, for this window only: no input id outlives its window. + self.input.clear(); + if !self.present_input_merged(&in_chunks, keys, &matches, &mut window.input) { + self.present_input(&in_chunks, keys, &matches, &mut window.input); } // Output-history presentation, same keys (register keys + values for correction resolution). - let (o_keys, o_vals, o_khs, mut o_times, o_diffs, o_run_ends) = collect_present(&chunks_of(instance.output_batches), &keys); + let out_chunks = chunks_of(instance.output_batches); + let (o_keys, o_vals, o_khs, mut o_times, o_diffs, o_run_ends) = collect_present(&out_chunks, keys, &search(&out_chunks, keys)); if !o_khs.is_empty() { let vids = ids(&o_vals); let merged = merge_present(&o_keys, &o_vals, &o_khs, &vids, &mut o_times, &o_diffs, &o_run_ends, &mut window.output); @@ -783,6 +825,11 @@ where consolidate_updates(&mut window.output); } } + + *from = retire.keys.get(stop).map_or(KeyPosition::End, |key| KeyPosition::At(*key)); + if stop < retire.keys.len() { + self.retire = Retire { next: stop, ..retire }; + } } fn reduce_corrections(&mut self, keys: &[u64], in_ends: &[usize], input: &[(u64, Diff)], out_ends: &[usize], output: &[(u64, Diff)]) -> (Vec<(u64, Diff)>, Vec) { @@ -1004,8 +1051,9 @@ mod tests { let chunks: Vec<_> = chunks.iter().collect(); let (mut merged, mut hashed) = (CorgiReduceBackend::