From 273792e5d0bc74ebe39b2a87b3b3d3eea9dbcb5b Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Wed, 12 Aug 2026 08:14:28 -0400 Subject: [PATCH 1/2] Stream VecReduceBackend output into open chunks per tile MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The backend staged every retire's output as rows in tile_rows, building chunks only at finish — instrumented at a 336MB high-water against a 2.6MB windowed presentation at 4M keys, the backend's whole memory spike. Emits arrive per tile in disjoint ascending key ranges, each consolidated, so emit can push rows straight into an open chunk and seal at TARGET; finish just wraps the sealed chunks. Also records why merged_run's cross-batch merge must stay: it is load-bearing for ordinal id sharing (netting in the id-keyed accumulations), measured at ~25% of churn when removed. Co-Authored-By: Claude Fable 5 --- .../src/operators/int_proxy/vec_backend.rs | 58 +++++++++++-------- 1 file changed, 34 insertions(+), 24 deletions(-) diff --git a/differential-dataflow/src/operators/int_proxy/vec_backend.rs b/differential-dataflow/src/operators/int_proxy/vec_backend.rs index 426391864..038b82eab 100644 --- a/differential-dataflow/src/operators/int_proxy/vec_backend.rs +++ b/differential-dataflow/src/operators/int_proxy/vec_backend.rs @@ -28,7 +28,7 @@ use std::rc::Rc; use timely::container::PushInto; use timely::progress::Timestamp; -use crate::consolidation::{consolidate, consolidate_updates_from}; +use crate::consolidation::{consolidate, consolidate_updates}; use crate::difference::Semigroup; use crate::lattice::Lattice; use crate::trace::chunk::ChunkBatch; @@ -62,9 +62,17 @@ pub struct VecReduceBackend { /// Lookup-only: the non-determinism of the map's iteration order is never observed. out_ids: HashMap<(K, W), u64>, - /// The retire's output tile descriptions, and the rows accumulated for each. + /// The retire's output tile descriptions, and the chunks accumulated for each: sealed chunks + /// plus the open tail the next `emit` appends to. Emits arrive in ascending key ranges (the + /// windows partition the key space), so chunks fill in batch order and seal for good at + /// `TARGET` — the retire's output never sits staged as rows. (Measured before this change, + /// at 4M keys the row staging was the backend's entire memory spike: a 336MB high-water + /// against a 2.6MB windowed presentation.) tiles: Vec>, - tile_rows: Vec>, + tile_chunks: Vec>>, + /// Scratch for one `emit`'s resolved rows: consolidation restores row order within a hash + /// (vid order need not be row order) before the rows stream into the open chunk. + stage: Vec<((u64, (K, W)), T, R)>, } impl VecReduceBackend { @@ -82,7 +90,8 @@ impl VecReduceBackend { out_pool: Vec::new(), out_ids: HashMap::new(), tiles: Vec::new(), - tile_rows: Vec::new(), + tile_chunks: Vec::new(), + stage: Vec::new(), } } } @@ -103,7 +112,7 @@ where fn begin(&mut self, tiles: &[Description]) { self.keys_stale = true; self.tiles = tiles.to_vec(); - self.tile_rows = (0..tiles.len()).map(|_| Vec::new()).collect(); + self.tile_chunks = (0..tiles.len()).map(|_| Vec::new()).collect(); } #[inline(never)] @@ -298,12 +307,19 @@ where #[inline(never)] fn emit(&mut self, tile: usize, records: &[((u64, u64), T, R)]) { - let mark = self.tile_rows[tile].len(); + self.stage.clear(); for ((h, vid), t, d) in records { let row = self.out_pool[*vid as usize].clone(); - self.tile_rows[tile].push(((*h, row), t.clone(), d.clone())); + self.stage.push(((*h, row), t.clone(), d.clone())); + } + consolidate_updates(&mut self.stage); + let chunks = &mut self.tile_chunks[tile]; + for update in self.stage.drain(..) { + if chunks.last().is_none_or(|c| c.as_slice().len() >= as crate::trace::chunk::Chunk>::TARGET) { + chunks.push(VecChunk::default()); + } + chunks.last_mut().expect("pushed above if absent").push_into(update); } - consolidate_updates_from(&mut self.tile_rows[tile], mark); } #[inline(never)] @@ -312,22 +328,11 @@ where self.out_pool.clear(); self.out_ids.clear(); let tiles = std::mem::take(&mut self.tiles); - let tile_rows = std::mem::take(&mut self.tile_rows); + let tile_chunks = std::mem::take(&mut self.tile_chunks); tiles .into_iter() - .zip(tile_rows) - .map(|(desc, rows)| { - let mut chunks: Vec> = Vec::default(); - let mut iter = rows.into_iter(); - while iter.len() > 0 { - let mut chunk = VecChunk::default(); - for update in (&mut iter).take( as crate::trace::chunk::Chunk>::TARGET) { - chunk.push_into(update); - } - chunks.push(chunk); - } - Rc::new(ChunkBatch::new(chunks, desc)) - }) + .zip(tile_chunks) + .map(|(desc, chunks)| Rc::new(ChunkBatch::new(chunks, desc))) .collect() } } @@ -337,8 +342,13 @@ where /// /// Per hash bracket, the batches' contiguous runs are gathered and sorted by `(payload, time)`, so /// equal payloads meet across batches and each payload's times arrive in order (batch intervals -/// are disjoint, so cross-batch times need ordering but never summing). The walk seeks to the -/// first requested key and stops after the last, so a bounded window pays for its own range. +/// are disjoint, so cross-batch times need ordering but never summing). The merge is load-bearing +/// for the ordinal id scheme, not for the record order (the harness re-sorts each key's records +/// for its own sweep): grouping equal payloads across batches is what lets them share one id, and +/// hence net in the id-keyed accumulations — presented unmerged, each un-netted id pays a value +/// resolution in every correction. (Measured: an unmerged walk cost churn ~25% overall.) The walk +/// seeks to the first requested key and stops after the last, so a bounded window pays for its +/// own range. #[inline(never)] fn merged_run( batches: &[VBatch], From 4c04b779ea7a22dc1cd41e131735c16d83559365 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Wed, 12 Aug 2026 11:36:53 -0400 Subject: [PATCH 2/2] tighten documentation --- .../src/operators/int_proxy/vec_backend.rs | 30 ++++++------------- 1 file changed, 9 insertions(+), 21 deletions(-) diff --git a/differential-dataflow/src/operators/int_proxy/vec_backend.rs b/differential-dataflow/src/operators/int_proxy/vec_backend.rs index 038b82eab..75f1bccfa 100644 --- a/differential-dataflow/src/operators/int_proxy/vec_backend.rs +++ b/differential-dataflow/src/operators/int_proxy/vec_backend.rs @@ -62,16 +62,10 @@ pub struct VecReduceBackend { /// Lookup-only: the non-determinism of the map's iteration order is never observed. out_ids: HashMap<(K, W), u64>, - /// The retire's output tile descriptions, and the chunks accumulated for each: sealed chunks - /// plus the open tail the next `emit` appends to. Emits arrive in ascending key ranges (the - /// windows partition the key space), so chunks fill in batch order and seal for good at - /// `TARGET` — the retire's output never sits staged as rows. (Measured before this change, - /// at 4M keys the row staging was the backend's entire memory spike: a 336MB high-water - /// against a 2.6MB windowed presentation.) + /// The retire's output tile descriptions, and the chunks accumulated for each. tiles: Vec>, tile_chunks: Vec>>, - /// Scratch for one `emit`'s resolved rows: consolidation restores row order within a hash - /// (vid order need not be row order) before the rows stream into the open chunk. + /// Scratch to re-order one `emit`'s output by types, rather than transient identifiers. stage: Vec<((u64, (K, W)), T, R)>, } @@ -312,6 +306,7 @@ where let row = self.out_pool[*vid as usize].clone(); self.stage.push(((*h, row), t.clone(), d.clone())); } + // TODO: could consolidate only within a hash key, rather than the whole chunk. consolidate_updates(&mut self.stage); let chunks = &mut self.tile_chunks[tile]; for update in self.stage.drain(..) { @@ -337,23 +332,16 @@ where } } -/// Walks `batches` restricted to the ascending `keys`, emitting each record as -/// `(hash, &payload, &time, &diff)` in `(hash, payload, time)` order. +/// Merge-walks `batches`, restricted to `keys`, invoking `logic` on each consolidated non-zero update. /// -/// Per hash bracket, the batches' contiguous runs are gathered and sorted by `(payload, time)`, so -/// equal payloads meet across batches and each payload's times arrive in order (batch intervals -/// are disjoint, so cross-batch times need ordering but never summing). The merge is load-bearing -/// for the ordinal id scheme, not for the record order (the harness re-sorts each key's records -/// for its own sweep): grouping equal payloads across batches is what lets them share one id, and -/// hence net in the id-keyed accumulations — presented unmerged, each un-netted id pays a value -/// resolution in every correction. (Measured: an unmerged walk cost churn ~25% overall.) The walk -/// seeks to the first requested key and stops after the last, so a bounded window pays for its -/// own range. +/// The merge-walk is in order of `(hash, data, time)`. +/// +/// TODO: Not actually correct at the moment, in that the consolidation does not yet occur. #[inline(never)] fn merged_run( batches: &[VBatch], keys: &[u64], - mut sink: impl FnMut(u64, &D, &T, &R), + mut logic: impl FnMut(u64, &D, &T, &R), ) where D: Ord + Clone + 'static, T: Lattice + Timestamp, @@ -446,7 +434,7 @@ fn merged_run( { scratch.sort_by(|a, b| (a.0, a.1).cmp(&(b.0, b.1))); for (d, t, r) in scratch.drain(..) { - sink(h, d, t, r); + logic(h, d, t, r); } } }