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..d25917f81 --- /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]); + // The argument has the compiled input's shape and corgi's ops are total on data, so this + // evaluation cannot fail; a panic here would be a corgi bug, and should not be hidden. + let empty_result = corgi::eval_graph(graph, argument).into_list("empty reduce").ok().map(|(bounds, _)| bounds.to_vec())?; + 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/half_join/backend.rs b/interactive/src/runtime/half_join/backend.rs index 8fecdef9d..ee3b6a7d2 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,13 +17,13 @@ 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}; 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}; @@ -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,29 +159,56 @@ 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. - let mut needles: Vec = rows.iter().map(|row| row.0).collect(); + // Needles come from the consolidated stream, so identifiers whose rows all cancelled + // are neither probed nor staged. The stream is in identifier order. + let mut needles: Vec = stream.iter().map(|((id, _), _, _)| *id).collect(); needles.dedup(); let (first, last) = (needles[0], needles[needles.len() - 1]); 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); + // 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, 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(); @@ -190,8 +222,10 @@ impl ProxyHalfJoinBackend for CorgiHalfJoinBackend { .chain(probes.iter().map(|p| recover_key_ref(p.chunk.keys()))).collect(); let mut cursor = 0usize; let released = rows.iter().filter_map(|&(id, r, u)| { - while needles[cursor] < id { cursor += 1; } - matched[cursor].then_some((cursor, r as usize, u as usize, u as usize + 1)) + while cursor < needles.len() && needles[cursor] < id { cursor += 1; } + // A row whose identifier cancelled entirely is not a needle. + (cursor < needles.len() && needles[cursor] == id && matched[cursor]) + .then_some((cursor, r as usize, u as usize, u as usize + 1)) }); let offset = runs.len(); let probed = probes.iter().enumerate().flat_map(|(pi, p)| { @@ -367,6 +401,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`. @@ -588,7 +645,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| { diff --git a/interactive/src/runtime/join.rs b/interactive/src/runtime/join.rs index c852e1386..850425a0a 100644 --- a/interactive/src/runtime/join.rs +++ b/interactive/src/runtime/join.rs @@ -14,16 +14,18 @@ //! 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 -//! 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 -//! (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. +//! * 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 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. //! //! Times are advanced by `instance.lower` on presentation (consolidate on load), and a key //! whose side nets to zero after consolidation is suppressed entirely (the harness requires @@ -147,6 +149,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 +162,7 @@ impl ProxyJoinBackend, CBatch> for CorgiJoinBackend< bridge0, bridge1, &mut self.colliding, + (&proj0, &proj1), ); *from = next.map_or(KeyPosition::End, KeyPosition::At); } @@ -328,29 +336,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 { @@ -368,11 +461,13 @@ pub(super) struct SideScratch { tds: Vec<(T, Diff)>, /// Cross-chunk value-merge heads, reused across keys. heads: Vec<(usize, usize)>, + /// Row indices sorted by projected value, reused across keys. + idx: Vec<(u32, u32)>, } impl SideScratch { pub(super) fn new() -> Self { - SideScratch { entries: Vec::new(), tds: Vec::new(), heads: Vec::new() } + SideScratch { entries: Vec::new(), tds: Vec::new(), heads: Vec::new(), idx: Vec::new() } } fn push(&mut self, coord: u64, time: T, diff: Diff) { @@ -388,17 +483,52 @@ 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). + // Move the scratch out while `push` borrows `self`, then return its capacity. + let mut idx = std::mem::take(&mut self.idx); + idx.clear(); + 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.idx = idx; + 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 +669,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 +682,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 +706,8 @@ impl<'a, T: ColTime> LeafView<'a, T> { s, e, vals: self.vals.as_deref(), + pcol: self.pcol, + kind: self.kind, } } } @@ -586,10 +721,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 +735,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 +747,8 @@ impl<'a, T: ColTime> Probe<'a, T> { s, e, vals: self.vals.as_deref(), + pcol: self.pcol, + kind: self.kind, }) } } @@ -654,8 +792,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 +818,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 +833,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 +863,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 +894,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 +1115,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 +1147,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 +1288,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 +1322,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 +1339,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)]); 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::