From 7f48393d3fd8b3cdd0797863276c18aac0c7d06c Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 30 Oct 2026 12:01:28 -0400 Subject: [PATCH 1/4] DDIR: project join values before assigning tokens Consolidate values that agree on the fields the join reads before matching. Keep the existing merge for whole values and leading leaf projections; sort other projections within each real key. Unread values share one token. Collision checks retain the original keys. Adapt the half join's staging API without changing its demand yet. Implemented by GPT-6 Astra (Codex); reviewed by GPT-6.1 Sol (Codex) and Claude. Co-Authored-By: GPT-6 Astra (Codex) Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o --- interactive/src/runtime/half_join/backend.rs | 7 +- interactive/src/runtime/join.rs | 229 ++++++++++++++++--- 2 files changed, 195 insertions(+), 41 deletions(-) diff --git a/interactive/src/runtime/half_join/backend.rs b/interactive/src/runtime/half_join/backend.rs index 8fecdef9d..e47a3f530 100644 --- a/interactive/src/runtime/half_join/backend.rs +++ b/interactive/src/runtime/half_join/backend.rs @@ -23,7 +23,7 @@ use corgi::{shape_of_value, Demand, Graph, NumOp, Shape, Value as CValue}; use crate::runtime::chunk::{colliding_keys, ident, key_is_hashed, recover_key_ref, rewrap, CorgiChunk}; use crate::runtime::col_times::{ColTimes, Lanes}; use crate::runtime::container::CorgiContainer; -use crate::runtime::join::{demanded, leaf_valued, primitive_values, stage_collision, Probe, Reads, RunRef, SideScratch, COORD_BITS}; +use crate::runtime::join::{demanded, primitive_values, stage_collision, Probe, Reads, RunRef, SideProj, SideScratch, COORD_BITS}; use crate::lambda::compile_projection; use crate::ir::{Diff, Lambda, Time}; @@ -173,10 +173,11 @@ impl ProxyHalfJoinBackend for CorgiHalfJoinBackend { let chunks = arranged_chunks(&instance.batches); // An empty arrangement meets nothing; the stream rows are presented and skipped. if chunks.is_empty() { return; } - let leaf_vals = leaf_valued(&chunks); + // The arranged values are compared whole here: the half join's `cross` gathers them whole. + let proj = SideProj::new(&chunks, &corgi::Demand::All); let probes: Vec> = chunks.iter().enumerate() .filter(|(_, c)| { let ids = ident(c.keys()); ids[0] <= last && ids[ids.len() - 1] >= first }) - .map(|(cid, c)| Probe::new(c, cid, &needles, leaf_vals)) + .map(|(cid, c)| Probe::new(c, cid, &needles, &proj)) .collect(); let matched: Vec = (0..needles.len()).map(|n| probes.iter().any(|p| p.lo[n] < p.hi[n])).collect(); diff --git a/interactive/src/runtime/join.rs b/interactive/src/runtime/join.rs index c852e1386..f2ee6661a 100644 --- a/interactive/src/runtime/join.rs +++ b/interactive/src/runtime/join.rs @@ -14,13 +14,14 @@ //! A hash collision remains under that token through proxy matching; only then does `cross` //! compare the recorded real-key coordinates and discard unequal pairs. //! -//! * With primitive keys, a primitive **value token** is the value itself (including +//! * Values are compared after projection to the fields the join reads. Unread values share +//! one token per real key. With primitive keys, a read primitive **value token** is the value itself (including //! singleton-product wrappers). Equal values consolidate without choosing a representative, //! and matched columns are reconstructed from IDs. Other values use a *canonical coordinate*: //! `(chunk << 48) | row` of the value's first occurrence among its side's chunks. Coordinates redeem against the instance //! alone (no backend state, no reliance on call adjacency), and value columns are never //! copied into the presentation — they are gathered once, straight from chunk storage -//! into the projection input, and only for rows that matched. Assigning *equal* values +//! into the projection input, and only for rows that matched. Assigning *equal projected* values //! (found by the per-key merge across chunks) the *same* canonical coordinate is what //! lets bridge consolidation cancel cross-batch churn: tokens are the unit of //! cancellation, and location-distinct tokens for equal data would silently opt out. @@ -147,6 +148,11 @@ impl ProxyJoinBackend, CBatch> for CorgiJoinBackend< // leaf walk applies to every key shape. A hash collision is retained in proxy space and // recorded here; `cross` then filters only that token's unequal real-key pairs. Restarting // through a whole-key walk is not valid after an earlier block has already retired. + // Tokens are minted on values projected to what the projection reads, so values that + // differ only in unread fields share a token and consolidate before matching. `cross` + // reads the projected columns at the token's coordinate, which any row of the token serves. + let proj0 = SideProj::new(&chunks0, &self.reads.columns[1]); + let proj1 = SideProj::new(&chunks1, &self.reads.columns[2]); advance_leaf( &chunks0, &chunks1, @@ -155,6 +161,7 @@ impl ProxyJoinBackend, CBatch> for CorgiJoinBackend< bridge0, bridge1, &mut self.colliding, + (&proj0, &proj1), ); *from = next.map_or(KeyPosition::End, KeyPosition::At); } @@ -328,29 +335,114 @@ pub(super) fn leaf_valued(chunks: &[&CorgiChunk]) -> bool { && chunks.iter().filter(|c| c.len() > 0).all(|c| leaf_lanes(c.vals()).is_some()) } +/// How one side's values are compared when minting tokens: projected to what the projection +/// reads of them. Equal projected values share one token, so values that differ only in unread +/// fields consolidate before matching. +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub(super) enum ProjKind { + /// The projection reads the whole value: tokens as before. + Whole, + /// The projection reads nothing of the value: one token per key. + Collapse, + /// It reads leading leaf lanes, so the stored order still groups equal projected values. + Grouped, + /// It reads fields the stored order does not group by: sort to find equal projected values. + Ungrouped, +} + +/// One side's projection for tokens, over its chunks. +pub(super) struct SideProj { + pub(super) kind: ProjKind, + /// Whether the values are leaf-laned (and not primitive tokens), so lanes are borrowed. + pub(super) leaf: bool, + /// The kept leaf lanes, in `leaf_lanes` order, when `leaf`. + keep: Vec, + /// Per chunk, the projected column for structural compares, when not `leaf`. + cols: Vec, +} + +impl SideProj { + /// For `demand` on the values of `chunks` (in coordinate order). + pub(super) fn new(chunks: &[&CorgiChunk], demand: &Demand) -> Self { + let leaf = !chunks.is_empty() && leaf_valued(chunks); + let whole = SideProj { kind: ProjKind::Whole, leaf, keep: Vec::new(), cols: Vec::new() }; + let Some(first) = chunks.first() else { return whole }; + if *demand == Demand::All { return whole; } + if *demand == Demand::None || matches!(demand.project(first.vals().clone()), Ok(CValue::Unit(_))) { + return SideProj { kind: ProjKind::Collapse, leaf, keep: Vec::new(), cols: Vec::new() }; + } + if primitive_values(first) { return whole; } + if leaf { + let mut keep = Vec::new(); + demanded_lanes(first.vals(), demand, &mut 0, &mut keep); + let total = leaf_lanes(first.vals()).map_or(0, |l| l.len()); + if keep.is_empty() { return SideProj { kind: ProjKind::Collapse, leaf, keep, cols: Vec::new() }; } + if keep.len() == total { return whole; } + let grouped = keep.iter().enumerate().all(|(i, &l)| i == l); + SideProj { kind: if grouped { ProjKind::Grouped } else { ProjKind::Ungrouped }, leaf, keep, cols: Vec::new() } + } else { + let cols = chunks.iter().map(|c| demand.project(c.vals().clone()).expect("the join value's demand")).collect(); + SideProj { kind: ProjKind::Ungrouped, leaf, keep: Vec::new(), cols } + } + } + /// The comparison lanes of `chunk`: its kept leaf lanes. + fn lanes<'a, T: ColTime>(&self, chunk: &'a CorgiChunk) -> Option> { + if !self.leaf { return None; } + let lanes = leaf_lanes(chunk.vals()).expect("a leaf-laned side"); + Some(match self.kind { + ProjKind::Whole => lanes, + ProjKind::Collapse => Vec::new(), + _ => self.keep.iter().map(|&i| lanes[i]).collect(), + }) + } + /// The column of chunk `cid` to compare structurally, when projected. + fn pcol(&self, cid: usize) -> Option<&CValue> { + self.cols.get(cid) + } +} + +/// The leaf lanes of `col`, in `leaf_lanes` order, that `demand` keeps. +fn demanded_lanes(col: &CValue, demand: &Demand, next: &mut usize, keep: &mut Vec) { + match col { + CValue::Prim(_) => { if *demand != Demand::None { keep.push(*next); } *next += 1; } + CValue::Prod(fields) => match demand { + Demand::Prod(needs) => for (f, d) in fields.iter().zip(needs) { demanded_lanes(f, d, next, keep) }, + d => for f in fields { demanded_lanes(f, d, next, keep) }, + }, + CValue::Unit(_) => {} + _ => unreachable!("leaf-laned values are products of integer leaves"), + } +} + /// One key's records in one chunk: absolute rows `[s, e)`. When the vals are leaf-laned, -/// `vals` borrows the chunk's full leaf lanes, indexed by absolute row; `None` -/// falls back to structural compares against the chunk itself. +/// `vals` borrows the chunk's kept leaf lanes, indexed by absolute row; `None` +/// falls back to structural compares against `pcol`, or the chunk itself. pub(super) struct RunRef<'a, T: ColTime> { pub(super) chunk: &'a CorgiChunk, pub(super) cid: usize, pub(super) s: usize, pub(super) e: usize, pub(super) vals: Option<&'a [&'a [i64]]>, + pub(super) pcol: Option<&'a CValue>, + pub(super) kind: ProjKind, } impl<'a, T: ColTime> RunRef<'a, T> { - fn val_less(&self, row: usize, other: &Self, orow: usize) -> bool { + fn col(&self) -> &'a CValue { + self.pcol.unwrap_or(self.chunk.vals()) + } + fn val_cmp(&self, row: usize, other: &Self, orow: usize) -> Ordering { + if self.kind == ProjKind::Collapse { return Ordering::Equal; } match (self.vals, other.vals) { - (Some(a), Some(b)) => a.iter().zip(b).map(|(la, lb)| (la[row], lb[orow])).find(|(x, y)| x != y).is_some_and(|(x, y)| x < y), - _ => compare_at(self.chunk.vals(), row, other.chunk.vals(), orow) == Ordering::Less, + (Some(a), Some(b)) => a.iter().zip(b).map(|(la, lb)| la[row].cmp(&lb[orow])).find(|o| *o != Ordering::Equal).unwrap_or(Ordering::Equal), + _ => compare_at(self.col(), row, other.col(), orow), } } + fn val_less(&self, row: usize, other: &Self, orow: usize) -> bool { + self.val_cmp(row, other, orow) == Ordering::Less + } fn val_eq(&self, row: usize, other: &Self, orow: usize) -> bool { - match (self.vals, other.vals) { - (Some(a), Some(b)) => a.iter().zip(b).all(|(la, lb)| la[row] == lb[orow]), - _ => compare_at(self.chunk.vals(), row, other.chunk.vals(), orow) == Ordering::Equal, - } + self.val_cmp(row, other, orow) == Ordering::Equal } /// One past the end of the value run starting at `row`. fn val_run_end(&self, row: usize) -> usize { @@ -388,17 +480,49 @@ impl SideScratch { /// Entries end sorted by `(token, time)`. pub(super) fn stage_runs(&mut self, runs: &[RunRef<'_, T>], lower: &T) { self.entries.clear(); - if runs.first().is_some_and(|run| primitive_values(run.chunk)) { + let Some(first) = runs.first() else { return }; + if primitive_values(first.chunk) { for run in runs { let ids = int_lane(run.chunk.vals()).unwrap(); + // An unread primitive value is one token (`cross` never reads it back). + let collapse = run.kind == ProjKind::Collapse; self.entries.extend((run.s..run.e).map(|row| { let time = run.chunk.times().get_joined(row, lower); - (ids[row] as u64, time, run.chunk.diffs()[row]) + (if collapse { 0 } else { ids[row] as u64 }, time, run.chunk.diffs()[row]) })); } differential_dataflow::consolidation::consolidate_updates(&mut self.entries); return; } + if first.kind == ProjKind::Ungrouped && (runs.len() != 1 || first.e - first.s > 1) { + // The stored order does not group equal projected values: sort the key's rows by + // projected value, then consolidate each value's times. The canonical coordinate + // is the least contributing (chunk, row). + let mut idx: Vec<(u32, u32)> = Vec::new(); + for (i, run) in runs.iter().enumerate() { idx.extend((run.s..run.e).map(|row| (i as u32, row as u32))); } + idx.sort_unstable_by(|a, b| runs[a.0 as usize].val_cmp(a.1 as usize, &runs[b.0 as usize], b.1 as usize)); + let mut start = 0; + while start < idx.len() { + let (ri, row) = (idx[start].0 as usize, idx[start].1 as usize); + let mut end = start + 1; + while end < idx.len() && runs[idx[end].0 as usize].val_eq(idx[end].1 as usize, &runs[ri], row) { end += 1; } + let mut coord = u64::MAX; + self.tds.clear(); + for &(i, r) in &idx[start..end] { + let run = &runs[i as usize]; + coord = coord.min(((run.cid as u64) << COORD_BITS) | r as u64); + self.tds.push((run.chunk.times().get_joined(r as usize, lower), run.chunk.diffs()[r as usize])); + } + self.tds.sort_by(|a, b| a.0.cmp(&b.0)); + let tds = std::mem::take(&mut self.tds); + for (time, diff) in tds.iter() { self.push(coord, time.clone(), *diff); } + self.tds = tds; + start = end; + } + self.entries.sort_by(|a, b| (a.0, &a.1).cmp(&(b.0, &b.1))); + self.entries.retain(|(_, _, d)| *d != 0); + return; + } if let [run] = runs { // Single-chunk run: values are grouped and coordinates of successive value // runs ascend. Joining product times need not preserve their total order, @@ -539,8 +663,11 @@ struct LeafView<'a, T: ColTime> { /// Absolute row of `keys[0]`. base: usize, keys: &'a [i64], - /// The chunk's full borrowed value lanes; `None` when vals are structured. + /// The chunk's borrowed comparison lanes (the kept leaf lanes); `None` when vals are structured. vals: Option>, + /// The chunk's projected column, for structural compares of projected values. + pcol: Option<&'a CValue>, + kind: ProjKind, /// Cursor within `keys`. cur: usize, } @@ -549,9 +676,9 @@ struct LeafView<'a, T: ColTime> { const PULL: usize = 1 << 14; impl<'a, T: ColTime> LeafView<'a, T> { - fn new(chunk: &'a CorgiChunk, cid: usize, start: usize, end: usize, leaf_vals: bool) -> Self { - let vals = (leaf_vals && end > start).then(|| leaf_lanes(chunk.vals()).unwrap()); - LeafView { chunk, cid, base: start, keys: &ident(chunk)[start..end], vals, cur: 0 } + fn new(chunk: &'a CorgiChunk, cid: usize, start: usize, end: usize, proj: &'a SideProj) -> Self { + let vals = if end > start { proj.lanes(chunk) } else { None }; + LeafView { chunk, cid, base: start, keys: &ident(chunk)[start..end], vals, pcol: proj.pcol(cid), kind: proj.kind, cur: 0 } } /// The key under the cursor, if any remains in this block. fn cur_key(&self) -> Option { @@ -573,6 +700,8 @@ impl<'a, T: ColTime> LeafView<'a, T> { s, e, vals: self.vals.as_deref(), + pcol: self.pcol, + kind: self.kind, } } } @@ -586,10 +715,12 @@ pub(super) struct Probe<'a, T: ColTime> { pub(super) lo: Vec, pub(super) hi: Vec, vals: Option>, + pcol: Option<&'a CValue>, + kind: ProjKind, } impl<'a, T: ColTime> Probe<'a, T> { - pub(super) fn new(chunk: &'a CorgiChunk, cid: usize, needles: &[i64], leaf_vals: bool) -> Self { + pub(super) fn new(chunk: &'a CorgiChunk, cid: usize, needles: &[i64], proj: &'a SideProj) -> Self { let keys = ident(chunk); let (mut lo, mut hi) = (vec![0; needles.len()], vec![0; needles.len()]); let mut found = Vec::new(); @@ -598,8 +729,7 @@ impl<'a, T: ColTime> Probe<'a, T> { lo[j] = range.start; hi[j] = range.end; } - let vals = leaf_vals.then(|| leaf_lanes(chunk.vals()).unwrap()); - Probe { chunk, cid, lo, hi, vals } + Probe { chunk, cid, lo, hi, vals: proj.lanes(chunk), pcol: proj.pcol(cid), kind: proj.kind } } /// The run of driver key `j` in this chunk, if any. pub(super) fn run_ref(&self, j: usize) -> Option> { @@ -611,6 +741,8 @@ impl<'a, T: ColTime> Probe<'a, T> { s, e, vals: self.vals.as_deref(), + pcol: self.pcol, + kind: self.kind, }) } } @@ -654,8 +786,7 @@ pub(super) fn stage_collision( compare_at(run.chunk.keys(), candidate, reference, row) != Ordering::Equal }) .unwrap_or(run.e - start); - let vals = run.vals; - equal_runs.push(RunRef { chunk: run.chunk, cid: run.cid, s: start, e: end, vals }); + equal_runs.push(RunRef { chunk: run.chunk, cid: run.cid, s: start, e: end, vals: run.vals, pcol: run.pcol, kind: run.kind }); positions[index] = end; } scratch.stage_runs(&equal_runs, lower); @@ -681,6 +812,7 @@ fn advance_leaf( bridge0: &mut ProxyBridge, bridge1: &mut ProxyBridge, colliding: &mut Vec, + (proj0, proj1): (&SideProj, &SideProj), ) { let start = from.expect("advance called on an exhausted unit"); let hashed = key_is_hashed(chunks0[0].keys()); @@ -695,28 +827,27 @@ fn advance_leaf( chunks.iter().zip(starts).map(|(c, &s)| c.len() - s).sum() }; let (r0, r1) = (remaining(chunks0, &start0), remaining(chunks1, &start1)); - fn views<'a, T: ColTime>(chunks: &[&'a CorgiChunk], starts: &[usize], ends: &[usize]) -> Vec> { - let leaf_vals = leaf_valued(chunks); + fn views<'a, T: ColTime>(chunks: &[&'a CorgiChunk], starts: &[usize], ends: &[usize], proj: &'a SideProj) -> Vec> { chunks.iter().enumerate() .filter(|(cid, _)| ends[*cid] > starts[*cid]) - .map(|(cid, c)| LeafView::new(c, cid, starts[cid], ends[cid], leaf_vals)) + .map(|(cid, c)| LeafView::new(c, cid, starts[cid], ends[cid], proj)) .collect() } if r0.max(r1) >= 2 * r0.min(r1) { // Lopsided: only the driver is read in bulk, so only the driver bounds the block. let drive0 = r0 <= r1; - let (dchunks, dstarts, pchunks) = if drive0 { (chunks0, &start0, chunks1) } else { (chunks1, &start1, chunks0) }; + let (dchunks, dstarts, pchunks, dproj, pproj) = if drive0 { (chunks0, &start0, chunks1, proj0, proj1) } else { (chunks1, &start1, chunks0, proj1, proj0) }; let h = block_horizon(dchunks, dstarts, PULL); - let dviews = views(dchunks, dstarts, &block_ends(dchunks, h)); + let dviews = views(dchunks, dstarts, &block_ends(dchunks, h), dproj); let (bd, bp) = if drive0 { (bridge0, bridge1) } else { (bridge1, bridge0) }; - leaf_probe(dviews, pchunks, lower, h, from, bd, bp, hashed, colliding) + leaf_probe(dviews, pchunks, pproj, lower, h, from, bd, bp, hashed, colliding) } else { // Symmetric: both sides are read, so both bound the block. let h = block_horizon(chunks0, &start0, PULL).into_iter() .chain(block_horizon(chunks1, &start1, PULL)) .min(); - let views0 = views(chunks0, &start0, &block_ends(chunks0, h)); - let views1 = views(chunks1, &start1, &block_ends(chunks1, h)); + let views0 = views(chunks0, &start0, &block_ends(chunks0, h), proj0); + let views1 = views(chunks1, &start1, &block_ends(chunks1, h), proj1); leaf_merge(views0, views1, lower, h, from, bridge0, bridge1, hashed, colliding) } } @@ -726,6 +857,7 @@ fn advance_leaf( fn leaf_probe<'a, T: ColTime>( mut dviews: Vec>, pchunks: &[&CorgiChunk], + pproj: &SideProj, lower: &T, h: Option, from: &mut Option, @@ -756,10 +888,9 @@ fn leaf_probe<'a, T: ColTime>( } // One batched probe of every probee chunk at the driver's keys. - let pvleaf = leaf_valued(pchunks); let probes: Vec> = pchunks.iter().enumerate() .filter(|(_, c)| c.len() > 0) - .map(|(cid, c)| Probe::new(c, cid, &keyset, pvleaf)) + .map(|(cid, c)| Probe::new(c, cid, &keyset, pproj)) .collect(); // Check the whole block's matched keys before staging any one key. Unmatched driver keys @@ -978,8 +1109,11 @@ mod tests { } let mut output = Vec::new(); backend.cross(&instance, &mut matches, &mut output); - assert_eq!(output.iter().map(|c| c.diffs.len()).sum::(), if hashed { 2 } else { 4 }); - output.into_iter().map(|c| (c.keys, c.vals, c.times.to_vec(), c.diffs)).collect::>() + // Tokens minted on projected values can merge rows before `cross`: compare + // the consolidated updates, not the column layout. + let mut rows: Vec<_> = output.into_iter().flat_map(|c| c.into_updates()).collect(); + differential_dataflow::consolidation::consolidate_updates(&mut rows); + rows }; assert_eq!(run(&mut narrowed), run(&mut original), "hashed={hashed}, body={body}"); } @@ -1007,6 +1141,24 @@ mod tests { output.iter().map(|container| container.diffs.len()).sum() } + /// Unread-only changes cancel before any matching, including a side read only for existence. + #[test] + fn unread_changes_cancel_before_matching() { + for body in ["(k, ())", "(k, a)", "(k, c)"] { + let make = |noise, diff| Rc::new(ChunkBatch::new(vec![CorgiChunk::from_columns( + ints(vec![7]), CValue::Prod(vec![ints(vec![1]), ints(vec![noise]), ints(vec![3])]), + [0].into_iter().collect(), vec![diff], + )])); + let instance = JoinInstance { batches0: vec![make(10, -1), make(11, 1)], batches1: vec![make(12, 1)], lower: 0 }; + let lambda = Lambda::new(format!("let (k, (a, _, c), _) = input in {body}"), &corgi::Scope::default()); + let shape = Shape::Prod(vec![Shape::Int; 3]); + let mut backend = CorgiJoinBackend::new(&lambda, [Shape::Int, shape.clone(), shape]); + let (mut left, mut right) = (Vec::new(), Vec::new()); + backend.advance(&instance, &mut KeyPosition::Start, &mut left, &mut right); + assert!(left.is_empty() && right.is_empty(), "{body}"); + } + } + #[test] fn primitive_tokens_cancel_across_chunks_and_reconstruct_mixed_shapes() { for depth in 0..3 { @@ -1130,12 +1282,13 @@ mod tests { }; let left = make(vec![0, 0, 1, 2, 2, 2, 4, 4, 5], 1); let right = make(vec![0, 1, 1, 4, 5], 2); - let mut view = LeafView::new(&left, 0, 2, 8, true); + let (lproj, rproj) = (SideProj::new(&[&left], &Demand::All), SideProj::new(&[&right], &Demand::All)); + let mut view = LeafView::new(&left, 0, 2, 8, &lproj); assert_eq!(view.take_run(1), Some((2, 3))); assert_eq!(view.take_run(2), Some((3, 6))); assert_eq!(view.take_run(4), Some((6, 8))); assert_eq!(view.cur_key(), None); - let probe = Probe::new(&right, 1, &[0, 1, 3, 5], true); + let probe = Probe::new(&right, 1, &[0, 1, 3, 5], &rproj); assert!(probe.run_ref(2).is_none()); for (s, e) in [(2, 3), (3, 6), (6, 8)] { let a = view.run_ref(s, e); @@ -1163,7 +1316,7 @@ mod tests { ); let mut scratch = SideScratch::new(); scratch.stage_runs( - &[RunRef { chunk: &chunk, cid: 0, s: 0, e: 3, vals: None }], + &[RunRef { chunk: &chunk, cid: 0, s: 0, e: 3, vals: None, pcol: None, kind: ProjKind::Whole }], &t(1, 0), ); assert_eq!(scratch.entries, vec![(11, t(1, 0), 1)]); @@ -1180,7 +1333,7 @@ mod tests { ); let mut scratch = SideScratch::new(); scratch.stage_runs( - &[RunRef { chunk: &chunk, cid: 0, s: 0, e: 3, vals: None }], + &[RunRef { chunk: &chunk, cid: 0, s: 0, e: 3, vals: None, pcol: None, kind: ProjKind::Whole }], &t(1, 0), ); assert_eq!(scratch.entries, vec![(0, t(1, 0), 1)]); From 207a0ac7ee5b122da49334ebcd1fd46e78e9d71c Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 30 Oct 2026 12:01:28 -0400 Subject: [PATCH 2/4] DDIR: project both half-join inputs before matching Use the join's projected-value staging for the arranged side. Consolidate released stream rows on their projected values within each block, preserving real keys and both update and data times. Original coordinates still serve output gathering and collision checks. Keep the whole-read stream path. Implemented by GPT-6 Astra (Codex); reviewed by GPT-6.1 Sol (Codex) and Claude. Co-Authored-By: GPT-6 Astra (Codex) Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o --- interactive/src/runtime/half_join/backend.rs | 70 +++++++++++++++++--- 1 file changed, 61 insertions(+), 9 deletions(-) diff --git a/interactive/src/runtime/half_join/backend.rs b/interactive/src/runtime/half_join/backend.rs index e47a3f530..4bae7be5e 100644 --- a/interactive/src/runtime/half_join/backend.rs +++ b/interactive/src/runtime/half_join/backend.rs @@ -4,7 +4,7 @@ //! their data times evaluated lane by lane over the presented times. //! //! The arranged side is staged as the join backend stages its sides: one key's rows across -//! chunks are merged by value, equal values share one token (the value itself for integer +//! chunks are merged by projected value, equal projections share one token (the value itself for read integer //! values under integer keys, else the coordinate of its first occurrence), and the records are //! consolidated. Identifiers that hold more than one real key are recorded, and only pairs under //! those have their real keys compared. @@ -17,7 +17,7 @@ use differential_dataflow::operators::int_proxy::{KeyPosition, ProxyBridge}; use differential_dogs3::operators::half_join::proxy::{data_time, HalfJoinInstance, HalfJoinMatches, HalfJoinStream, Lifted, ProxyHalfJoinBackend}; use differential_dataflow::trace::chunk::Chunk; -use corgi::arrange::{equal_idx, gather_lanes}; +use corgi::arrange::{compare_at, equal_idx, gather_lanes}; use corgi::{shape_of_value, Demand, Graph, NumOp, Shape, Value as CValue}; use crate::runtime::chunk::{colliding_keys, ident, key_is_hashed, recover_key_ref, rewrap, CorgiChunk}; @@ -44,6 +44,8 @@ pub(crate) struct CorgiHalfJoinBackend { graph: Graph, /// What the graph reads of the key, the update's value, and the arranged value. reads: Reads, + /// Whether the stream value loses fields, and can coalesce previously distinct rows. + project_stream: bool, /// Identifiers in the current block that cover more than one real key, ascending. `advance` /// writes this and the immediately following `cross` reads it; only matches under these /// identifiers need a real-key comparison. @@ -60,7 +62,10 @@ impl CorgiHalfJoinBackend { let input = Shape::Prod(vec![k, v0, v1]); let narrowed = corgi::compile(g, &input).and_then(|compiled| corgi::narrow(compiled, &Demand::All)) .unwrap_or_else(|e| panic!("half_join: narrowing a compiled projection at {input}: {e}")); - CorgiHalfJoinBackend { width, graph: narrowed.graph, reads: Reads::new(narrowed.input, &input), colliding: Vec::new() } + let reads = Reads::new(narrowed.input, &input); + let Shape::Prod(parts) = &input else { unreachable!() }; + let project_stream = reads.columns[1].projected_shape(&parts[1]).expect("the stream value's demand") != parts[1]; + CorgiHalfJoinBackend { width, graph: narrowed.graph, reads, project_stream, colliding: Vec::new() } } } @@ -154,16 +159,40 @@ impl ProxyHalfJoinBackend for CorgiHalfJoinBackend { let ids = ident(run.keys()); rows.extend((starts[r]..ends[r]).map(|u| (ids[u], r as u32, u as u32))); } - rows.sort_unstable(); let lanes: Vec>> = runs.iter().map(|run| data_lanes(run, self.width)).collect(); - for &(id, r, u) in &rows { + let mut push = |(id, r, u): (i64, u32, u32), diff| { + if diff == 0 { return; } let (r, u) = (r as usize, u as usize); let lifted = Lifted { time: runs[r].times().get(u), data: Time::from_coordinates(lanes[r].iter().map(|lane| lane[u] as u64)), }; - stream.push(((id, ((r as u64) << COORD_BITS) | u as u64), lifted, runs[r].diffs()[u])); + stream.push(((id, ((r as u64) << COORD_BITS) | u as u64), lifted, diff)); + }; + if self.project_stream { + // The released runs are only in identifier order. Sort on the projected value, + // retaining both times and the real key: only rows that agree on all can cancel. + // A representative coordinate still names the original columns for `cross`. + let vals = demanded(&self.reads.columns[1], runs.iter().map(|r| &run_fields(r)[0])); + let hashed = key_is_hashed(runs[0].keys()); + let cmp = |&(id0, r0, u0): &(i64, u32, u32), &(id1, r1, u1): &(i64, u32, u32)| { + let (r0, u0, r1, u1) = (r0 as usize, u0 as usize, r1 as usize, u1 as usize); + id0.cmp(&id1) + .then_with(|| if hashed { compare_at(runs[r0].keys(), u0, runs[r1].keys(), u1) } else { std::cmp::Ordering::Equal }) + .then_with(|| if vals.is_empty() { std::cmp::Ordering::Equal } else { compare_at(&vals[r0], u0, &vals[r1], u1) }) + .then_with(|| runs[r0].times().cmp_cross(u0, runs[r1].times(), u1)) + .then_with(|| lanes[r0].iter().map(|l| l[u0]).cmp(lanes[r1].iter().map(|l| l[u1]))) + }; + rows.sort_unstable_by(cmp); + for group in rows.chunk_by(|a, b| cmp(a, b).is_eq()) { + let diff = group.iter().map(|&(_, r, u)| runs[r as usize].diffs()[u as usize]).sum(); + push(group[0], diff); + } + } else { + rows.sort_unstable(); + for &row in &rows { push(row, runs[row.1 as usize].diffs()[row.2 as usize]); } } + if stream.is_empty() { return; } // The arranged rows of the block's identifiers: one batched probe of each chunk that // can hold any of them. @@ -173,8 +202,8 @@ impl ProxyHalfJoinBackend for CorgiHalfJoinBackend { let chunks = arranged_chunks(&instance.batches); // An empty arrangement meets nothing; the stream rows are presented and skipped. if chunks.is_empty() { return; } - // The arranged values are compared whole here: the half join's `cross` gathers them whole. - let proj = SideProj::new(&chunks, &corgi::Demand::All); + // Values equal on the read fields share a token before matching. + let proj = SideProj::new(&chunks, &self.reads.columns[2]); let probes: Vec> = chunks.iter().enumerate() .filter(|(_, c)| { let ids = ident(c.keys()); ids[0] <= last && ids[ids.len() - 1] >= first }) .map(|(cid, c)| Probe::new(c, cid, &needles, &proj)) @@ -368,6 +397,29 @@ mod tests { ProxyHalfJoinTactic::new(CorgiHalfJoinBackend::new(&pairing(), 2, shapes), strict) } + /// Projection cancels unread churn, but neither of a lifted update's times may be merged. + #[test] + fn projected_churn_keeps_both_stream_times() { + for change_data in [false, true] { + let arranged = Rc::new(columns_to_batch( + present_key(ints(&[1, 1])), CValue::Prod(vec![ints(&[8, 8]), ints(&[20, 21])]), + [t(0, 0), t(0, 0)].into_iter().collect(), vec![1, -1], + )); + let run = Run::new(ints(&[1; 4]), CValue::Prod(vec![ + CValue::Prod(vec![ints(&[7; 4]), ints(&[10, 11, 12, 13])]), + lifted(&[(3, 0), (3, 0), (4, 0), if change_data { (5, 0) } else { (4, 0) }]), + ]), [t(2, 0), t(2, 0), t(2, 0), if change_data { t(2, 0) } else { t(3, 0) }].into_iter().collect(), vec![1, -1, 1, -1]); + let pair = Shape::Prod(vec![Shape::Int, Shape::Int]); + let lambda = Lambda::new("let (k, (a, _), (b, _)) = input in (k, (a, b))".into(), &corgi::Scope::default()); + let mut backend = CorgiHalfJoinBackend::new(&lambda, 2, [Shape::Int, pair.clone(), pair]); + let instance = HalfJoinInstance { stream: vec![run], batches: vec![arranged] }; + let (mut stream, mut arranged) = (Vec::new(), Vec::new()); + backend.advance(&instance, &mut KeyPosition::Start, &mut stream, &mut arranged); + assert_eq!(stream.len(), 2); + assert!(arranged.is_empty()); + } + } + /// An update meets the arranged rows before it in the total order on times, strictly or not: /// never a later one, even one beyond it in a later coordinate alone. Its data time joins the /// arranged row's, coordinate by coordinate, and every output row is covered by `lower`. @@ -589,7 +641,7 @@ mod tests { let lambda = Lambda::new(format!("let (k, v0, v1) = input in let (a, b, c, d) = v0 in let (e, f, g, h) = v1 in {body}"), &corgi::Scope::default()); let narrowed = CorgiHalfJoinBackend::new(&lambda, 2, shapes.clone()); let original = CorgiHalfJoinBackend { - width: 2, graph: compile_projection(&lambda, &shapes).unwrap(), + width: 2, project_stream: false, graph: compile_projection(&lambda, &shapes).unwrap(), reads: Reads::new(Demand::All, &Shape::Prod(shapes.to_vec())), colliding: Vec::new(), }; let rows = |backend: CorgiHalfJoinBackend| { From 3282feb6ae23226c2f4ad0a2c5ed179325618949 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 30 Oct 2026 12:01:28 -0400 Subject: [PATCH 3/4] DDIR: project reduce identifiers when weighted sums permit it Port the compiled-graph analysis for a diff-weighted map and total sum, requiring empty-group suppression independent of the key. Use projected columns in input presentation only with this fact and without a value limit. Keep output identity and raw novel time seeds. Other reducers retain whole values: in particular, signed projected extrema can change when values cancel. Implemented by GPT-6 Astra (Codex); reviewed by GPT-6.1 Sol (Codex) and Claude. Co-Authored-By: GPT-6 Astra (Codex) Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o --- interactive/src/passes/mod.rs | 1 + interactive/src/passes/reduce_demand.rs | 120 ++++++++++++++++++++++++ interactive/src/runtime/reduce.rs | 52 ++++++++-- 3 files changed, 166 insertions(+), 7 deletions(-) create mode 100644 interactive/src/passes/reduce_demand.rs diff --git a/interactive/src/passes/mod.rs b/interactive/src/passes/mod.rs index 5e9c73f27..175057817 100644 --- a/interactive/src/passes/mod.rs +++ b/interactive/src/passes/mod.rs @@ -5,6 +5,7 @@ mod dead_code; mod arrangements; mod canonical; pub mod projection; +pub(crate) mod reduce_demand; mod as_of; mod timing; pub use timing::{Measurement, Pass, Timings}; diff --git a/interactive/src/passes/reduce_demand.rs b/interactive/src/passes/reduce_demand.rs new file mode 100644 index 000000000..10e978d09 --- /dev/null +++ b/interactive/src/passes/reduce_demand.rs @@ -0,0 +1,120 @@ +//! Prove when coalescing equal projected values preserves a reducer's answer. +//! This is an analysis only: it never changes a graph or a scope. + +use corgi::{ArithOp, BinOp, Bounds, Demand, Graph, NodeRef, NumOp, Op, Red, Shape, Value}; + +#[derive(Clone, Copy, PartialEq, Eq)] +enum Term { Neutral, Diff, Linear, Bad } + +/// A map element must be `diff * g(value)` (with `g = 1` for the bare diff). +/// Each node is examined once, including tuple aliases. +fn weighted(body: &Graph) -> bool { + let nodes: Vec<_> = body.nodes().collect(); + let mut terms = Vec::with_capacity(nodes.len()); + for node in &nodes { + let term = match node { + NodeRef::Input => Term::Bad, + NodeRef::Tuple(fields) => { + if fields.iter().any(|&i| terms[i] != Term::Neutral) { Term::Bad } else { Term::Neutral } + } + NodeRef::Op(NumOp::Core(Op::Field(field)), i) => match &nodes[*i] { + NodeRef::Input => match field { 0 => Term::Neutral, 1 => Term::Diff, _ => Term::Bad }, + NodeRef::Tuple(fields) => fields.get(*field).map(|&j| terms[j]).unwrap_or(Term::Bad), + _ if terms[*i] == Term::Neutral => Term::Neutral, + _ => Term::Bad, + }, + NodeRef::Op(NumOp::Arith(ArithOp::Bin(BinOp::Mul)), i) => match &nodes[*i] { + NodeRef::Input => Term::Linear, + NodeRef::Tuple(fields) if fields.len() == 2 => match (terms[fields[0]], terms[fields[1]]) { + (Term::Diff | Term::Linear, Term::Neutral) | (Term::Neutral, Term::Diff | Term::Linear) => Term::Linear, + _ => Term::Bad, + }, + _ => Term::Bad, + }, + NodeRef::Op(NumOp::Core(Op::Unit | Op::Lit(_)), _) => Term::Neutral, + NodeRef::Op(_, i) => if terms[*i] == Term::Neutral { Term::Neutral } else { Term::Bad }, + }; + terms.push(term); + } + matches!(terms[body.output()], Term::Diff | Term::Linear) +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum ListUse { Input, Vals, Mapped, Other } + +/// Prove that the reducer reads values only through diff-weighted sums. +/// Return its value demand, or `None` when an entry boundary can matter. +/// The compiled input must be `(key, List<(value, Int-diff)>)`. +pub(crate) fn projection(compiled: &corgi::Compiled) -> Option { + let input = compiled.input(); + let Shape::Prod(root_shape) = input else { return None }; + let [key, Shape::List(element)] = root_shape.as_slice() else { return None }; + let Shape::Prod(pair_shape) = &**element else { return None }; + let [val, Shape::Int] = pair_shape.as_slice() else { return None }; + let graph = compiled.graph(); + let nodes: Vec<_> = graph.nodes().collect(); + let mut uses = Vec::with_capacity(nodes.len()); + let mut maps = 0; + let mut sums = 0; + for node in &nodes { + let use_ = match node { + NodeRef::Input => ListUse::Input, + NodeRef::Tuple(fields) => { + if fields.iter().any(|&i| uses[i] != ListUse::Other) { return None; } + ListUse::Other + } + NodeRef::Op(NumOp::Core(Op::Field(field)), i) if uses[*i] == ListUse::Input => { + match field { 0 => ListUse::Other, 1 => ListUse::Vals, _ => return None } + } + NodeRef::Op(NumOp::Core(Op::MapList(body)), i) if uses[*i] == ListUse::Vals => { + if !weighted(body) { return None; } + maps += 1; + ListUse::Mapped + } + NodeRef::Op(NumOp::Arith(ArithOp::Reduce(Red::Add)), i) if uses[*i] == ListUse::Mapped => { + sums += 1; + ListUse::Other + } + NodeRef::Op(NumOp::Core(Op::Unit | Op::Lit(_)), _) => ListUse::Other, + NodeRef::Op(_, i) => { + if uses[*i] != ListUse::Other { return None; } + ListUse::Other + } + }; + uses.push(use_); + } + if maps != 1 || sums != 1 || uses[graph.output()] != ListUse::Other { return None; } + + // A projected group can disappear when all its values cancel. The reducer must + // emit nothing for an empty list, or coalescing would change key presence. + let empty = Value::List(Bounds::offsets(vec![0]), Box::new(Value::empty(&Shape::Prod(vec![val.clone(), Shape::Int])))); + let argument = Value::Prod(vec![zero_row(key)?, empty]); + let empty_result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + corgi::eval_graph(graph, argument).into_list("empty reduce").ok().map(|(bounds, _)| bounds.to_vec()) + })).ok()??; + if empty_result != [0] { return None; } + + let analysis = corgi::analyze_demand(graph, input, &Demand::All).ok()?; + let Demand::Prod(root) = analysis.input_demand() else { return None }; + // The empty-list check uses one representative key. No key read means that + // result is the same for every key, including after a whole group cancels. + if root[0] != Demand::None { return None; } + let Demand::List(element) = &root[1] else { return None }; + let Demand::Prod(pair) = &**element else { return None }; + Some(pair[0].clone()) +} + +fn zero_row(shape: &Shape) -> Option { + Some(match shape { + Shape::Unit => Value::Unit(1), + Shape::Int => Value::i64(vec![0]), + Shape::Float => Value::f64(vec![0.0]), + Shape::Prod(parts) => Value::Prod(parts.iter().map(zero_row).collect::>()?), + Shape::List(element) => Value::List(Bounds::offsets(vec![0]), Box::new(Value::empty(element))), + Shape::Sum(parts) if !parts.is_empty() => { + let lanes = parts.iter().enumerate().map(|(i, p)| if i == 0 { zero_row(p) } else { Some(Value::empty(p)) }).collect::>()?; + Value::sum(vec![0], lanes) + } + _ => return None, + }) +} diff --git a/interactive/src/runtime/reduce.rs b/interactive/src/runtime/reduce.rs index 98238f5ca..d5c666d6a 100644 --- a/interactive/src/runtime/reduce.rs +++ b/interactive/src/runtime/reduce.rs @@ -6,7 +6,8 @@ //! * ids — a key's id is the identifier its arrangement is sorted by (the key itself when it is a //! primitive integer, else its hash lane). A value's id names a row in a pool of columns: the //! chunks' own value columns, and the output columns this backend builds. Within a window, a -//! key's equal input values share one id, found by corgi's segmented sort; output values are +//! key's equal input values share one id, found by corgi's segmented sort. A proven weighted +//! sum compares projected values here; other reducers keep whole-value identity. Output values are //! matched to the current output by structural comparison. Nothing hashes values. //! * collisions — a hashed key's identifier may hold several real keys. Chunks sort each //! identifier's rows by real key, so a window finds them with two comparisons per run. Their @@ -138,6 +139,8 @@ pub(crate) struct CorgiReduceBackend { graph: Graph, /// What the graph reads of its input. reads: Reads, + /// The reducer is invariant under consolidation of equal projected values. + project_values: bool, /// How many of each key's first values the graph is handed, if not all. limit: Option, /// The current window's input values: its chunks' value columns, projected by `reads.value`. @@ -214,11 +217,17 @@ impl CorgiReduceBackend { .unwrap_or_else(|e| panic!("reduce: type error at shapes ({kshape}, {vshape}): {e}")); // The graph rebuilt to read only what it reads, and the demand saying what that is. let input = Shape::Prod(vec![kshape, Shape::List(Box::new(Shape::Prod(vec![vshape, Shape::Int])))]); - let narrowed = corgi::compile(graph, &input).and_then(|compiled| corgi::narrow(compiled, &Demand::All)) + let compiled = corgi::compile(graph, &input) + .unwrap_or_else(|e| panic!("reduce: compiling at {input}: {e}")); + // A limit observes whole-value order and entry boundaries. Demand alone does not prove + // that merging those entries is valid; the sum analysis also checks disappearing groups. + let project_values = limit.is_none() && crate::passes::reduce_demand::projection(&compiled).is_some(); + let narrowed = corgi::narrow(compiled, &Demand::All) .unwrap_or_else(|e| panic!("reduce: narrowing a compiled graph at {input}: {e}")); CorgiReduceBackend { graph: narrowed.graph, reads: Reads::new(narrowed.input, &input), + project_values, limit, input: Pool::default(), output: Pool::default(), @@ -258,6 +267,7 @@ fn present( keys: &[i64], matches: &Matches, colliding: &[bool], + projected: Option<&[CValue]>, refs: &mut Vec<(usize, usize)>, bridge: &mut ProxyBridge, ) { @@ -280,8 +290,13 @@ fn present( (labels, rows, label_keys) = split_real_keys(chunks, colliding, labels, rows); } let key_of = |label: u64| label_keys.as_ref().map_or(label as usize, |keys: &Vec| keys[label as usize]); - let vals: Vec = chunks.iter().map(|chunk| chunk.vals().clone()).collect(); - let (sorted, groups) = sort_blocks(&labels, &gather_refs(&vals, rows.iter().copied())); + // Output and unproven input reducers retain whole-value identity and order. + let whole; + let vals = if let Some(projected) = projected { projected } else { + whole = chunks.iter().map(|chunk| chunk.vals().clone()).collect::>(); + &whole + }; + let (sorted, groups) = sort_blocks(&labels, &gather_refs(vals, rows.iter().copied())); let update = |(chunk, row): (usize, usize)| (chunks[chunk].times().get(row), chunks[chunk].diffs()[row]); let (mut start, mut updates) = (0, Vec::new()); for group in groups.chunk_by(|a, b| a == b) { @@ -484,11 +499,12 @@ where // equal values share an id, so an exactly cancelling pair vanishes here, and its time // survives in `window.seeds` above. Input ids name rows for this window only. self.input.refs.clear(); - present(&in_chunks, keys, &matches, colliding, &mut self.input.refs, &mut window.input); + let projected = (self.project_values && self.reads.value != Demand::All).then_some(&self.input.columns[..]); + present(&in_chunks, keys, &matches, colliding, projected, &mut self.input.refs, &mut window.input); // The output history, same keys. Output ids name rows for the whole retire, and the rows // that hold their keys are at the same positions of the output chunks' key columns. let fresh = self.output.refs.len(); - present(&out_chunks, keys, &out_matches, colliding, &mut self.output.refs, &mut window.output); + present(&out_chunks, keys, &out_matches, colliding, None, &mut self.output.refs, &mut window.output); self.output_keys.extend(self.output.refs[fresh..].iter().map(|&(chunk, row)| (in_chunks.len() + chunk, row))); *from = retire.keys.get(retire.window.end).map_or(KeyPosition::End, |key| KeyPosition::At(*key)); @@ -698,6 +714,28 @@ mod tests { use crate::runtime::chunk::present_key; use crate::ir::Time; + /// Demand is insufficient: entry counts, limits, extrema on signed lists and zero-emitting + /// sums must keep whole-value identity. A weighted sum that suppresses zero may consolidate. + #[test] + fn projected_values_require_a_mergeable_reducer() { + let shape = (Shape::Int, Shape::Prod(vec![Shape::Int, Shape::Int])); + let sum = "let n = vals map (((v, _), d) -> (v, d) mul) fold_add in (n, (n, 1)) enlist filter"; + for (body, limit, expected) in [ + (sum, None, true), (sum, Some(1), false), + ("(vals map ((_, d) -> d) fold_add, 1) enlist", None, false), + ("(vals len, 1) enlist", None, false), + ("(vals head.0.0, 1) enlist", Some(1), false), + ("(vals map (((v, _), _) -> v) fold_min, 1) enlist", None, false), + ("(vals map (((v, _), _) -> v) fold_max, 1) enlist", None, false), + ] { + let lambda = crate::ir::Lambda::new(format!("let (_, vals) = input in {body}"), &corgi::Scope::default()); + let backend = CorgiReduceBackend::