From 9bce5f3e8e90b65384272df417ef94fdbc649204 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 30 Jul 2026 20:17:17 -0400 Subject: [PATCH 1/2] corgi reduce: seek-vs-scan presentation, decided per retire MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit collect_present was the one non-delta-proportional path left: an O(trace) rescan (plus per-chunk key re-hashing) every retire, ~50% of an incremental round's profile. It now decides per retire, when the sizes are known: a narrow changed set over seekable keys (single-leaf: ids ARE key values, so the ascending changed set converts to a needle column) gallops each chunk once per changed key with find_ranges — no hashing at all; broad churn keeps the scan, whose flat membership test beats marginal seeking (a find_ranges probe is a structurally-dispatched binary search; SEEK_ADVANTAGE=16 measured as a load regression before widening to 64). Hashed ids of structural keys cannot be inverted into needles and always scan. Steady-state (1000 rounds x batch 100, 100k/200k): reach 4.51s -> 2.10s — past the vec backend's 2.51s (1.8x -> 0.84x); scc 794 -> 597ms/round, the residual being retires whose label cascades genuinely broaden the changed set — the windowed-presentation seam's territory, not a threshold's. Load-shaped scc unchanged (3.60s). Co-Authored-By: Claude Fable 5 --- interactive/src/corgi/reduce.rs | 85 +++++++++++++++++++++++++-------- 1 file changed, 66 insertions(+), 19 deletions(-) diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 0017d7c29..d7a263202 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -37,7 +37,7 @@ use differential_dataflow::trace::chunk::ChunkBatch; use differential_dataflow::operators::int_proxy::ProxyBridge; use differential_dataflow::operators::int_proxy::reduce::{ProxyReduceBackend, ReduceInstance, ReduceWindow}; -use corgi::arrange::{gather, gather_lanes, sort_blocks}; +use corgi::arrange::{find_ranges, gather, gather_lanes, sort_blocks}; use corgi::{Bounds, Shape, Value as CValue}; use crate::corgi::col_times::ColTime; @@ -168,6 +168,19 @@ fn concat_columns(blocks: &[CValue]) -> CValue { /// relied upon — so the raw two's-complement `u64` is correct even for negative ints (no swizzle). /// Applied CONSISTENTLY at every id site (both value presentations AND the freshly-produced /// `reduce_brackets` outputs), else `desired − current` nets across mismatched ids for the same value. +/// The `changed` set as a needle column in the chunks' own key shape — possible exactly +/// when `ids` uses key VALUES (a bare `u64` leaf, or a 1-tuple of one); the hashed ids of +/// structural keys cannot be inverted into needles. +fn seek_needles(sample: &CValue, changed: &[u64]) -> Option { + match corgi::shape_of_value(sample) { + Shape::Prim(64) => Some(CValue::u64(changed.to_vec())), + Shape::Prod(ref fs) if fs.len() == 1 && matches!(fs[0], Shape::Prim(64)) => { + Some(CValue::Prod(vec![CValue::u64(changed.to_vec())])) + } + _ => None, + } +} + fn ids(col: &CValue) -> Vec { match corgi::shape_of_value(col) { Shape::Prim(64) => col.clone().into_u64("ids"), @@ -183,33 +196,67 @@ fn ids(col: &CValue) -> Vec { /// `(keys_col, vals_col)` corgi columns plus per-record `(key_hash, time, diff)`. `changed` is the /// ASCENDING set of changed key hashes; a row is kept iff its key hash is in it. /// -/// NB this is a full scan of the presented chunks (incl. `source_batches`, the accumulated trace), -/// and deliberately NOT a `find_ranges` seek of the changed keys: under label-propagation-shaped -/// workloads the changed set is broad (most keys change each retire), so a scan touches ~every row -/// regardless and the per-chunk gallop only adds overhead. The O(history) re-presentation is -/// inherent to broad change sets, not a seekable-few-keys case. +/// Seek-vs-scan, decided per retire, now that the sizes are known: seeking the changed keys +/// (`find_ranges`, O(|changed|·log rows) per chunk, no key hashing at all) wins when the +/// changed set is narrow — the steady incremental case; the full scan (O(rows) per chunk, +/// plus each chunk's key hashes re-derived) wins for broad churn — loads and label-cascade +/// retires, where most keys change and a gallop per key only adds overhead. Seeking requires +/// ids that ARE key values (single-leaf keys, `ids`' fast paths): hashed ids of structural +/// keys cannot be inverted into needles, so those always scan. /// -/// TODO: the scan's per-row work can still batch: `ids` re-derives (and copies) each chunk's key -/// hashes every retire (memoize per chunk, or a stored hash column), the membership test is a -/// per-row `binary_search`, and each hit materializes an owned time (`times().get`); kept RANGES -/// could move via `push_range`. +/// TODO: the scan's per-row work can still batch: `ids` re-derives (and copies) each chunk's +/// key hashes every retire (memoize per chunk, or a stored hash column), and each hit +/// materializes an owned time (`times().get`); kept RANGES could move via `push_range`. fn collect_present(chunks: &[&CorgiChunk], changed: &[u64]) -> (CValue, CValue, Vec, Vec, Vec) where T: ColTime, { + /// Seek only when the changed set is at least this many times narrower than the + /// presented rows: a `find_ranges` probe is a structurally-dispatched binary search + /// (~log(rows) compares, each far costlier than the scan's flat membership test), so + /// marginal seeks LOSE to the scan — measured, not modeled; 16 regressed load-shaped + /// retires before this was widened. + const SEEK_ADVANTAGE: usize = 64; + let key_srcs: Vec> = chunks.iter().map(|c| Some(c.keys())).collect(); let val_srcs: Vec> = chunks.iter().map(|c| Some(c.vals())).collect(); let (mut tags, mut offs) = (Vec::new(), Vec::new()); let (mut khs, mut times, mut diffs) = (Vec::new(), Vec::new(), Vec::new()); - for (ci, ch) in chunks.iter().enumerate() { - let kh = ids(ch.keys()); - for i in 0..kh.len() { - if changed.binary_search(&kh[i]).is_ok() { - tags.push(ci); - offs.push(i); - khs.push(kh[i]); - times.push(ch.times().get(i)); - diffs.push(ch.diffs()[i]); + let total: usize = chunks.iter().map(|c| c.diffs().len()).sum(); + let needles = if changed.len().saturating_mul(SEEK_ADVANTAGE) < total { + chunks.iter().find(|c| c.diffs().len() > 0).and_then(|c| seek_needles(c.keys(), changed)) + } else { + None + }; + if let Some(needles) = needles { + // Narrow changed set over seekable keys: gallop each chunk once per changed key. + // Chunks are key-ordered and `changed` ascends, so emission order matches the scan's. + for (ci, ch) in chunks.iter().enumerate() { + if ch.diffs().is_empty() { + continue; + } + let (lo, hi) = find_ranges(&needles, ch.keys()); + for (j, (&l, &h)) in lo.iter().zip(hi.iter()).enumerate() { + for i in l..h { + tags.push(ci); + offs.push(i); + khs.push(changed[j]); + times.push(ch.times().get(i)); + diffs.push(ch.diffs()[i]); + } + } + } + } else { + for (ci, ch) in chunks.iter().enumerate() { + let kh = ids(ch.keys()); + for i in 0..kh.len() { + if changed.binary_search(&kh[i]).is_ok() { + tags.push(ci); + offs.push(i); + khs.push(kh[i]); + times.push(ch.times().get(i)); + diffs.push(ch.diffs()[i]); + } } } } From 190cad7a372ad2e191f9fce40d08586c5a610232 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 31 Jul 2026 10:56:38 -0400 Subject: [PATCH 2/2] Reattach ids' doc comment (seek_needles had been inserted mid-block) Co-Authored-By: Claude Fable 5 --- interactive/examples/ddir.rs | 10 +++++++++- interactive/src/corgi/reduce.rs | 22 +++++++++++----------- 2 files changed, 20 insertions(+), 12 deletions(-) diff --git a/interactive/examples/ddir.rs b/interactive/examples/ddir.rs index a706bd804..576c176be 100644 --- a/interactive/examples/ddir.rs +++ b/interactive/examples/ddir.rs @@ -14,6 +14,9 @@ //! - `--diag`: serve timely/DD diagnostics on port 51371. //! - `--backend=vec|corgi`: rendering substrate (default `vec`). The corgi //! backend is single-worker (its arrange does not exchange). +//! - `--sync=K`: await completion only every K rounds (default 1), letting K +//! timestamps retire with whatever inter-timestamp concurrency the system +//! finds — the open(er)-loop regime DD adapts into under load. use mimalloc::MiMalloc; @@ -37,6 +40,7 @@ struct Flags { debug_demand: bool, diag: bool, corgi: bool, + sync: u64, } fn run( @@ -184,7 +188,10 @@ fn run( cursor += 1; } for i in inputs.iter_mut() { i.advance_to(time); i.flush(); } - while probe.less_than(&time) { worker.step(); } + let sync = flags.sync.max(1); + if (round + 1) % sync == 0 || round + 1 == limit { + while probe.less_than(&time) { worker.step(); } + } round += 1; if round % 100 == 0 { @@ -209,6 +216,7 @@ fn main() { else if let Some(q) = a.strip_prefix("--query=") { flags.query = Some(q.to_string()); } else if a == "--debug-demand" { flags.debug_demand = true; } else if a == "--diag" { flags.diag = true; } + else if let Some(k) = a.strip_prefix("--sync=") { flags.sync = k.parse().expect("--sync=K"); } else if let Some(b) = a.strip_prefix("--backend=") { flags.corgi = match b { "corgi" => true, "vec" => false, other => panic!("unknown backend {other:?} (vec|corgi)") }; } diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index d7a263202..1435d0a59 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -168,6 +168,17 @@ fn concat_columns(blocks: &[CValue]) -> CValue { /// relied upon — so the raw two's-complement `u64` is correct even for negative ints (no swizzle). /// Applied CONSISTENTLY at every id site (both value presentations AND the freshly-produced /// `reduce_brackets` outputs), else `desired − current` nets across mismatched ids for the same value. +fn ids(col: &CValue) -> Vec { + match corgi::shape_of_value(col) { + Shape::Prim(64) => col.clone().into_u64("ids"), + Shape::Prod(ref fs) if fs.len() == 1 && matches!(fs[0], Shape::Prim(64)) => match col { + CValue::Prod(fields) => fields[0].clone().into_u64("ids"), + _ => unreachable!("shape Prod but value not Prod"), + }, + _ => corgi::hash(col).into_u64("ids"), + } +} + /// The `changed` set as a needle column in the chunks' own key shape — possible exactly /// when `ids` uses key VALUES (a bare `u64` leaf, or a 1-tuple of one); the hashed ids of /// structural keys cannot be inverted into needles. @@ -181,17 +192,6 @@ fn seek_needles(sample: &CValue, changed: &[u64]) -> Option { } } -fn ids(col: &CValue) -> Vec { - match corgi::shape_of_value(col) { - Shape::Prim(64) => col.clone().into_u64("ids"), - Shape::Prod(ref fs) if fs.len() == 1 && matches!(fs[0], Shape::Prim(64)) => match col { - CValue::Prod(fields) => fields[0].clone().into_u64("ids"), - _ => unreachable!("shape Prod but value not Prod"), - }, - _ => corgi::hash(col).into_u64("ids"), - } -} - /// Concatenate the records of the `changed` keys across a run of chunks into parallel /// `(keys_col, vals_col)` corgi columns plus per-record `(key_hash, time, diff)`. `changed` is the /// ASCENDING set of changed key hashes; a row is kept iff its key hash is in it.