diff --git a/interactive/Cargo.toml b/interactive/Cargo.toml index 8ada1d0aa..c31ee2212 100644 --- a/interactive/Cargo.toml +++ b/interactive/Cargo.toml @@ -13,6 +13,8 @@ workspace = true [dependencies] columnar = { workspace = true } +# The columnar kernels for the interpreted backend, pinned by git rev. +corgi = { git = "https://github.com/frankmcsherry/wip", rev = "1301b281501d5e70ab63f9405412770a3250d985" } differential-dataflow = { workspace = true } mimalloc = "0.1.48" serde = { version = "1.0", features = ["derive"] } diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs new file mode 100644 index 000000000..ef228c546 --- /dev/null +++ b/interactive/src/backend/corgi.rs @@ -0,0 +1,402 @@ +//! The corgi rendering substrate: corgi columns are the native representation on dataflow edges, +//! arrangements are chains of sorted columnar chunks (`ChunkSpine`, cursor-less), and +//! scalar logic runs columnar via `eval_graph`. Parallels the row-wise `backend::vec`, which stays +//! the correctness reference. +//! +//! All `Backend` methods are corgi-native: `linear` folds a `LinearOp` chain over each container +//! ([`apply_ops`], columnar fast paths with row-wise fallbacks); `arrange` ingests columns without +//! a row round-trip; `join`/`reduce` run through whole-chunk tactics ([`CorgiJoinTactic`], the +//! rank/int-proxy reduce backends) over the columnar chunks. + +use timely::dataflow::Scope; +use timely::dataflow::channels::pact::Pipeline; +use timely::dataflow::operators::generic::Operator; + +use differential_dataflow::AsCollection; +use differential_dataflow::Collection; +use differential_dataflow::operators::join::join_with_tactic; +use differential_dataflow::operators::reduce::reduce_with_tactic; +use differential_dataflow::operators::arrange::arrangement::arrange_core; +use differential_dataflow::operators::arrange::{Arranged, TraceAgent}; +use differential_dataflow::trace::chunk::{ChunkBatcher, ChunkBuilder}; + +use corgi::arrange::gather; +use corgi::Value as CValue; + +use crate::backend::Backend; +use crate::corgi::chunk::{chunks_to_columns, CorgiChunk, CorgiChunker}; +use crate::corgi::container::CorgiContainer; +use crate::corgi::join::CorgiJoinTactic; +use crate::corgi::reduce::CorgiReduceBackend; +use differential_dataflow::operators::int_proxy::ProxyReduceTactic; +use crate::corgi::logic::{compilable, compile_predicate, compile_projection}; +use crate::ir::{Diff, LinearOp, Time, Value as DValue}; +use crate::parse::{Projection, Reducer}; +use crate::scope_ir as st; + +/// A DDIR row, an update, the corgi container on dataflow edges, and the columnar trace — +/// the shorthands the `Backend` methods below are written in terms of. +type Row = DValue; +type Upd = ((Row, Row), Time, Diff); +type CC = CorgiContainer; +type CTrace = differential_dataflow::trace::chunk::ChunkSpine>; + +/// Apply a `LinearOp` chain to one corgi container (the corgi-native row-wise compute per batch). +/// Project = corgi `eval_graph`; Filter = corgi mask + `gather`; Negate = Rust — all columnar. +/// The time/list-shaping ops (EnterAt/LiftIter/FlatMap) take a correctness-first row-wise path +/// (untranscode → vec-style transform → `from_updates`), matching `backend::vec` exactly; a columnar +/// fast-path is future work. `level` is the scope depth (locates the iteration coordinate). +fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize) -> CC { + use timely::order::Product; + use differential_dataflow::lattice::Lattice; + use differential_dataflow::dynamic::pointstamp::PointStamp; + + for op in ops { + c = match op { + LinearOp::Project(p) if compilable(&p.key) && compilable(&p.val) => { + let (kshape, vshape) = (corgi::shape_of_value(&c.keys), corgi::shape_of_value(&c.vals)); + let g = compile_projection(&p.key, &p.val, &kshape, &vshape); + let mut cols = corgi::eval_graph(&g, CValue::Prod(vec![c.keys, c.vals])).into_prod("linear project"); + let vals = cols.pop().unwrap(); + let keys = cols.pop().unwrap(); + CorgiContainer { keys, vals, times: c.times, diffs: c.diffs } + } + LinearOp::Filter(cond) if compilable(cond) => { + let (kshape, vshape) = (corgi::shape_of_value(&c.keys), corgi::shape_of_value(&c.vals)); + let g = compile_predicate(cond, &kshape, &vshape); + let mask = corgi::eval_graph(&g, CValue::Prod(vec![c.keys.clone(), c.vals.clone()])).into_u64("filter mask"); + let keep: Vec = (0..mask.len()).filter(|&i| mask[i] != 0).collect(); + let keys = gather(&c.keys, &keep); + let vals = gather(&c.vals, &keep); + let times = keep.iter().map(|&i| c.times[i].clone()).collect(); + let diffs = keep.iter().map(|&i| c.diffs[i]).collect(); + CorgiContainer { keys, vals, times, diffs } + } + // Row-wise fallback for projections/predicates the corgi compiler can't lower (List, + // Case/Inject, Unary, Hash) — `ir::eval`, parity with `backend::vec`. + LinearOp::Project(p) => { + let mut out: Vec = Vec::new(); + for ((k, v), t, d) in c.into_updates() { + let mut env = vec![k, v]; + let nk = crate::ir::eval(&p.key, &mut env); + let nv = crate::ir::eval(&p.val, &mut env); + out.push(((nk, nv), t, d)); + } + CorgiContainer::from_updates(out) + } + LinearOp::Filter(cond) => { + let mut out: Vec = Vec::new(); + for ((k, v), t, d) in c.into_updates() { + let keep = { let mut env = vec![k.clone(), v.clone()]; crate::ir::eval(cond, &mut env).truthy() }; + if keep { out.push(((k, v), t, d)); } + } + CorgiContainer::from_updates(out) + } + LinearOp::Negate => { + for d in c.diffs.iter_mut() { + *d = -*d; + } + c + } + // Row-wise ops (parity with `backend::vec::render_linear`). + LinearOp::EnterAt(field) => { + let mut out: Vec = Vec::new(); + for ((k, v), t, d) in c.into_updates() { + let delay = { + let mut env = vec![k.clone(), v.clone()]; + let raw = crate::ir::eval(field, &mut env).as_int() as u64; + 256 * (64 - raw.leading_zeros() as u64) + }; + let mut coords = smallvec::SmallVec::<[u64; 1]>::new(); + for _ in 0..level.saturating_sub(1) { coords.push(0); } + coords.push(delay); + let delta = Product::new(0u64, PointStamp::new(coords)); + out.push(((k, v), t.join(&delta), d)); + } + CorgiContainer::from_updates(out) + } + LinearOp::LiftIter => { + let mut out: Vec = Vec::new(); + for ((k, v), t, d) in c.into_updates() { + let iter = level + .checked_sub(1) + .and_then(|idx| t.inner.get(idx).copied()) + .unwrap_or(0) as i64; + out.push(((k, append_iter(v, iter)), t, d)); + } + CorgiContainer::from_updates(out) + } + LinearOp::FlatMap(list_term) => { + let mut out: Vec = Vec::new(); + for ((k, v), t, d) in c.into_updates() { + let elems = { + let mut env = vec![k.clone(), v.clone()]; + match crate::ir::eval(list_term, &mut env) { + DValue::List(xs) => xs, + other => panic!("flatmap: expected a List, got {other:?}"), + } + }; + for (pos, elem) in elems.into_iter().enumerate() { + out.push(((k.clone(), DValue::Tuple(vec![DValue::Int(pos as i64), elem])), t.clone(), d)); + } + } + CorgiContainer::from_updates(out) + } + }; + } + c +} + +/// Append the user-iter coordinate to a value (mirrors `backend::vec::append_iter`): extend a `Tuple`, +/// or wrap any other value as `(value, iter)`. +fn append_iter(val: DValue, iter: i64) -> DValue { + match val { + DValue::Tuple(mut xs) => { xs.push(DValue::Int(iter)); DValue::Tuple(xs) } + other => DValue::Tuple(vec![other, DValue::Int(iter)]), + } +} + + +/// The corgi rendering substrate. An uninhabited type used only as a type-level tag: it +/// carries the [`Backend`] impl (a namespace of rendering functions selected by type) and is +/// never a value — rendering goes through `render_tree::`. The empty enum (vs a +/// unit struct) makes constructing one impossible, signalling "type only". Mirrors `VecBackend`. +pub enum CorgiBackend {} + +impl Backend for CorgiBackend { + type Container = CC; + type Arr<'scope> = Arranged<'scope, TraceAgent>; + + fn linear<'s>(c: Collection<'s, Time, CC>, ops: Vec, level: usize) -> Collection<'s, Time, CC> { + // Container-level: fold the LinearOp chain over each corgi batch (no inter-op transcode). + // `level` is the scope depth (locates the iteration coordinate for LiftIter/EnterAt). + c.inner + .unary(Pipeline, "CorgiLinear", move |_, _| { + move |input, output| { + input.for_each(|cap, data| { + let mut out = apply_ops(std::mem::take(data), &ops, level); + output.session(&cap).give_container(&mut out); + }); + } + }) + .as_collection() + } + + fn arrange<'s>(c: Collection<'s, Time, CC>) -> Self::Arr<'s> { + // Column-native ingest: `CorgiChunker` sort-consolidates each input `CorgiContainer`'s + // columns straight into a `CorgiChunk` (no drain-to-rows), then the standard chunk batcher + + // builder. No columns→rows→columns round-trip at the arrangement boundary. + arrange_core::<_, CC, CorgiChunker, ChunkBatcher>, ChunkBuilder>, CTrace>( + c.inner, + Pipeline, + "CorgiArrange", + ) + } + + fn as_collection<'s>(a: Self::Arr<'s>) -> Collection<'s, Time, CC> { + // Cursor-free AND transcode-free: the arrangement's batches already hold corgi columns, so + // concatenate each batch's chunk columns straight into a `CorgiContainer` — no columns→rows→ + // columns round-trip (the old `batch_to_rows` + `from_updates` path). + a.stream + .unary(Pipeline, "CorgiAsCollection", |_, _| { + |input, output| { + input.for_each(|cap, data| { + let mut chunks: Vec> = Vec::new(); + for batch in data.iter() { + chunks.extend(batch.chunks.iter().cloned()); + } + let (keys, vals, times, diffs) = chunks_to_columns(&chunks); + let mut c = CorgiContainer { keys, vals, times, diffs }; + output.session(&cap).give_container(&mut c); + }); + } + }) + .as_collection() + } + + fn join<'s>(l: Self::Arr<'s>, r: Self::Arr<'s>, projection: &Projection) -> Collection<'s, Time, CC> { + // The tactic compiles the projection per work-unit (shape-directed, for `Spread`) and emits + // corgi columns directly into a `CorgiContainer` (via `give_container`) — the output stream is + // column-native, so there is no row round-trip / no `JoinToCorgi` unary. + let tactic = CorgiJoinTactic::new(projection.key.clone(), projection.val.clone()); + join_with_tactic::<_, _, _, CC>(l, r, tactic).as_collection() + } + + fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer) -> Self::Arr<'s> { + reduce_with_tactic::<_, CTrace, _>(a, "CorgiReduce", ProxyReduceTactic::new(CorgiReduceBackend::new(reducer.clone()))) + } + + fn inspect<'s>(c: Collection<'s, Time, CC>, label: String) -> Collection<'s, Time, CC> { + c.inner + .unary(Pipeline, "CorgiInspect", move |_, _| { + move |input, output| { + input.for_each(|cap, data| { + let mut cont = std::mem::take(data); + for ((k, v), t, d) in cont.clone().into_updates() { + eprintln!(" [{label}] (({k:?}, {v:?}), {t:?}, {d})"); + } + output.session(&cap).give_container(&mut cont); + }); + } + }) + .as_collection() + } + + fn leave_dynamic<'s>(c: Collection<'s, Time, CC>, level: usize) -> Collection<'s, Time, CC> { + // Mirror DD's `Collection::leave_dynamic` (dynamic/mod.rs:40), but over a `CorgiContainer`: + // strip all but `level-1` PointStamp coordinates from the capability AND from each row's time + // (stored columnar in `CorgiContainer.times`, not inline in the data tuples). The input + // connection summary advertises the `retain` so timely's progress tracking stays correct. + use timely::dataflow::operators::generic::{builder_rc::OperatorBuilder, OutputBuilder}; + use timely::order::Product; + use timely::progress::Antichain; + use differential_dataflow::dynamic::pointstamp::{PointStamp, PointStampSummary}; + + let mut builder = OperatorBuilder::new("CorgiLeaveDynamic".to_string(), c.inner.scope()); + let (output, stream) = builder.new_output(); + let mut output = OutputBuilder::from(output); + let summary = Product { outer: Default::default(), inner: PointStampSummary { retain: Some(level - 1), actions: Vec::new() } }; + let mut input = builder.new_input_connection(c.inner, Pipeline, [(0, Antichain::from_elem(summary))]); + + builder.build(move |_capability| move |_frontier| { + let mut output = output.activate(); + input.for_each(|cap, data| { + let mut new_time = cap.time().clone(); + let mut v = std::mem::take(&mut new_time.inner).into_inner(); + v.truncate(level - 1); + new_time.inner = PointStamp::new(v); + let new_cap = cap.delayed(&new_time, 0); + for t in data.times.iter_mut() { + let mut v = std::mem::take(&mut t.inner).into_inner(); + v.truncate(level - 1); + t.inner = PointStamp::new(v); + } + output.session(&new_cap).give_container(data); + }); + }); + + stream.as_collection() + } +} + +/// Render `s` with the corgi substrate. See [`crate::backend::render_tree`]. +pub fn render_tree<'s>( + s: &st::Scope, + scope: Scope<'s, Time>, + depth: usize, + imports: Vec>, +) -> Vec> { + crate::backend::render_tree::(s, scope, depth, imports) +} + +/// Evaluate `program` on explicit inputs via the **corgi** backend (mirrors [`crate::backend::vec::evaluate`]). +/// +/// Inputs/exports cross the iterative-scope boundary as Vec rows (which support refinement +/// enter/leave); corgi containers exist only INSIDE the dynamic scope, where `Enter`/`Leave` are the +/// same-Time identity. The `ToCorgi`/`FromCorgi` unaries are the only row↔corgi conversions. +pub fn evaluate( + program: &st::Program, + inputs: &[Vec<(Row, Row)>], +) -> std::collections::BTreeMap> { + use std::collections::BTreeMap; + use std::sync::mpsc::channel; + use timely::dataflow::operators::core::capture::{Capture, Event}; + use timely::dataflow::operators::generic::Operator; + use differential_dataflow::input::Input; + use differential_dataflow::AsCollection; + use differential_dataflow::dynamic::pointstamp::PointStamp; + use crate::corgi::container::CorgiContainer; + + let names: Vec = program.root.exports.iter().map(|e| e.name.clone()).collect(); + let mut txs = Vec::new(); + let mut rxs = Vec::new(); + for _ in &names { + let (tx, rx) = channel::>>(); + txs.push(tx); + rxs.push(rx); + } + + let program = program.clone(); + let inputs: Vec> = inputs.to_vec(); + timely::execute_directly(move |worker| { + let mut handles = worker.dataflow::(|scope| { + let mut handles = Vec::new(); + let mut collections = Vec::new(); + for _ in 0..inputs.len() { + let (h, c) = scope.new_collection::<(Row, Row), Diff>(); + handles.push(h); + collections.push(c); + } + let exports = scope.iterative::, _, _>(|inner| { + // Enter Vec collections (refinement), then convert each to a corgi container. + let mut corgi_imports = Vec::new(); + for c in collections.iter().map(|c| c.clone().enter(inner)) { + let cs = c + .inner + .unary(Pipeline, "ToCorgi", |_, _| { + |input, output| { + input.for_each(|cap, data| { + let mut cc = CorgiContainer::from_updates(std::mem::take(data)); + output.session(&cap).give_container(&mut cc); + }); + } + }) + .as_collection(); + corgi_imports.push(cs); + } + let root_imports: Vec<_> = program + .root + .imports + .iter() + .map(|imp| match &imp.from { + st::Source::Input(n) => corgi_imports[*n].clone(), + other => panic!("corgi evaluate: unsupported source {other:?}"), + }) + .collect(); + let exports = render_tree(&program.root, inner.clone(), 0, root_imports); + // Convert corgi exports back to Vec rows, then leave the scope. + let mut leaved = Vec::new(); + for c in exports { + let rows = c + .inner + .unary(Pipeline, "FromCorgi", |_, _| { + |input, output| { + input.for_each(|cap, data| { + let mut rows = std::mem::take(data).into_updates(); + output.session(&cap).give_container(&mut rows); + }); + } + }) + .as_collection(); + leaved.push(rows.leave(scope)); + } + leaved + }); + for (col, tx) in exports.into_iter().zip(txs) { + col.inner.capture_into(tx); + } + handles + }); + for (i, rows) in inputs.iter().enumerate() { + for r in rows { + handles[i].update(r.clone(), 1); + } + } + }); + + names + .into_iter() + .zip(rxs) + .map(|(name, rx)| { + let mut acc: BTreeMap<(Row, Row), Diff> = BTreeMap::new(); + for event in rx { + if let Event::Messages(_, data) = event { + for ((k, v), _, d) in data { + *acc.entry((k, v)).or_insert(0) += d; + } + } + } + (name, acc.into_iter().filter(|(_, d)| *d != 0).collect()) + }) + .collect() +} diff --git a/interactive/src/backend/mod.rs b/interactive/src/backend/mod.rs index d1bbf5fa6..df67d70c9 100644 --- a/interactive/src/backend/mod.rs +++ b/interactive/src/backend/mod.rs @@ -11,6 +11,7 @@ //! that pick a backend and call [`render_tree`]. pub mod vec; +pub mod corgi; // The columnar substrate is deferred on the Value-model port (it needs a // `Columnar`/columnar-storage story for `Value`); see `backend::vec`. // pub mod col; diff --git a/interactive/src/corgi/chunk.rs b/interactive/src/corgi/chunk.rs new file mode 100644 index 000000000..40792e975 --- /dev/null +++ b/interactive/src/corgi/chunk.rs @@ -0,0 +1,712 @@ +//! `CorgiChunk`: a [`Chunk`](differential_dataflow::trace::chunk::Chunk) whose key/val payload is a +//! pair of corgi columns, with per-tuple times held columnar in a [`ColTimes`] (SoA over +//! `::Container`, killing the per-row `PointStamp` allocation); diffs stay a `Vec`. +//! It is a **`Chunk` but NOT `NavigableChunk`**: +//! it exposes no `Ord` key, no cursor, no sorted-trie layout — the merge/advance/settle transducers +//! drive everything through corgi's own structural order (`compare_at`) and gather primitives +//! (`gather`, `gather_lanes`). Consumption is the tactics' job (they read the columns in bulk), which +//! is exactly why cursor-less `Chunk` suffices here. +//! +//! This is a faithful port of the reference [`VecChunk`](differential_dataflow::trace::chunk::vec): +//! same resumable merge→advance→settle pipeline and grade-at-yield invariant, with the flat +//! `Rc>` swapped for corgi columns. Adopting the `Chunk` framework gives us the fueled, +//! graded `ChunkBatchMerger` for free — replacing the eager whole-trace `CorgiMerger`. +//! +//! Order: `(key, val)` by corgi structural order (`compare_at` over `Prod([keys, vals])`), then `time` +//! by `Ord`. Any consistent total order is fine — correctness compares multisets, not DDIR's `Ord`. +//! +//! Simplification vs `VecChunk`: `merge` processes only the two front chunks per call (no mid-merge +//! refill), so `gather_lanes` source indices stay valid for the whole call. The `Chunk` contract +//! permits this — "consume at least one input; the harness may re-invoke." + +use std::collections::VecDeque; +use std::rc::Rc; + +use timely::progress::Antichain; +use timely::progress::frontier::AntichainRef; + +use differential_dataflow::difference::Semigroup; +use differential_dataflow::trace::chunk::{merge_chains, pack, Chunk, ChunkBatch}; +use differential_dataflow::trace::Description; + +use corgi::arrange::{compare_at, compare_idx, find_ranges, gather, gather_lanes, group_bounds, sort_perm}; +use corgi::Value as CValue; + +use columnar::Columnar; + +use crate::corgi::col_times::{ColTime, ColTimes}; +use crate::ir::Value as DValue; + +use std::cmp::Ordering; + +/// A DDIR row update: `((key, val), time, diff)`. +pub type Upd = ((DValue, DValue), T, R); + +/// The grading target (also the merge/advance emit-chunk size). Larger than `VecChunk`'s 8192 to +/// amortize corgi's per-chunk columnar set-up (each chunk boundary costs a `gather` materialization); +/// bigger chunks mean fewer boundaries. Bounded so the fueled merger still yields. +const TARGET: usize = 1 << 18; + +/// Shared, immutable chunk contents. `Clone` of a `CorgiChunk` is an `Rc` bump. +struct Inner { + /// Key column (corgi), aligned with `vals`/`times`/`diffs`, sorted by `(key, val, time)`. + keys: CValue, + /// Val column (corgi). + vals: CValue, + /// Per-update times, SoA-columnar (the lattice algebra lives here; corgi never sees time). + times: ColTimes, + /// Per-update diffs. + diffs: Vec, +} + +/// A sorted, consolidated run of `((key, val), time, diff)` with corgi-columnar key/val, shared via `Rc`. +pub struct CorgiChunk(Rc>); + +impl Clone for CorgiChunk { + fn clone(&self) -> Self { CorgiChunk(Rc::clone(&self.0)) } +} + +impl Default for CorgiChunk { + fn default() -> Self { + CorgiChunk(Rc::new(Inner { keys: CValue::Unit(0), vals: CValue::Unit(0), times: ColTimes::new(), diffs: Vec::new() })) + } +} + +/// Split a `Prod([keys, vals])` corgi value into its two columns. +fn split_kv(kv: CValue) -> (CValue, CValue) { + let mut cols = kv.into_prod("corgi chunk kv"); + let vals = cols.pop().unwrap(); + let keys = cols.pop().unwrap(); + (keys, vals) +} + +impl CorgiChunk { + fn from_parts(keys: CValue, vals: CValue, times: ColTimes, diffs: Vec) -> Self { + CorgiChunk(Rc::new(Inner { keys, vals, times, diffs })) + } + /// The `(key, val)` sort payload as one corgi `Prod` column (cheap `Arc` bumps). + fn kv(&self) -> CValue { CValue::Prod(vec![self.0.keys.clone(), self.0.vals.clone()]) } + fn from_kv(kv: CValue, times: ColTimes, diffs: Vec) -> Self { + let (keys, vals) = split_kv(kv); + Self::from_parts(keys, vals, times, diffs) + } + pub fn keys(&self) -> &CValue { &self.0.keys } + pub fn vals(&self) -> &CValue { &self.0.vals } + pub fn times(&self) -> &ColTimes { &self.0.times } + pub fn diffs(&self) -> &[R] { &self.0.diffs } +} + +impl CorgiChunk +where + T: ColTime, + R: Semigroup + Clone + 'static, +{ + /// Materialize `[start, end)` of accumulated `(tag → src, off, time, diff)` into `TARGET`-sized + /// output chunks. `srcs` are the (stable) source kv columns the tags/offs index into. + fn emit( + srcs: &[Option<&CValue>], + tags: &[usize], + offs: &[usize], + times: &ColTimes, + diffs: &[R], + out: &mut VecDeque, + ) { + let n = times.len(); + let mut s = 0; + while s < n { + let e = (s + TARGET).min(n); + let kv = gather_lanes(srcs, &tags[s..e], &offs[s..e]); + let mut t = ColTimes::new(); + t.push_range(times, s, e); + out.push_back(Self::from_kv(kv, t, diffs[s..e].to_vec())); + s = e; + } + } + + /// Concatenate a run of (globally-sorted) chunks into one combined `(kv, times, diffs)`. + fn concat(chunks: &[Self]) -> (CValue, ColTimes, Vec) { + let kvs: Vec = chunks.iter().map(Self::kv).collect(); + let srcs: Vec> = kvs.iter().map(Some).collect(); + let total: usize = chunks.iter().map(Self::len_).sum(); + let (mut tags, mut offs) = (Vec::with_capacity(total), Vec::with_capacity(total)); + let (mut times, mut diffs) = (ColTimes::new(), Vec::with_capacity(total)); + for (ti, ch) in chunks.iter().enumerate() { + for o in 0..ch.len_() { tags.push(ti); offs.push(o); } + times.push_range(ch.times(), 0, ch.len_()); + diffs.extend_from_slice(ch.diffs()); + } + let kv = if total == 0 { CValue::Unit(0) } else { gather_lanes(&srcs, &tags, &offs) }; + (kv, times, diffs) + } + + fn len_(&self) -> usize { self.0.times.len() } +} + +impl Chunk for CorgiChunk +where + T: ColTime, + R: Semigroup + Clone + 'static, +{ + type Time = T; + const TARGET: usize = TARGET; + + fn len(&self) -> usize { self.0.times.len() } + + /// Two-pointer merge of the two front chunks through their shared horizon, FULLY consolidating + /// equal `(key, val, time)` triples and pushing back the survivor's suffix (the fueled-merger + /// contract). Reverted from a `survey`-based merge: survey aligns only `(key, val)` (corgi owns no + /// time), so its positional `Both` under-consolidates cross-side times, and — consuming both chunks + /// with no push-back — bloats the multi-chunk arrangement, which the reduce then re-presents each + /// retire → super-linear past the chunk boundary (n≳131k). A survey merge needs a group-RANGE + /// `Both` (both sides' full equal-`(key,val)` group) to time-merge correctly; the corgi agent has + /// that flagged. `group_bounds` in `advance` is kept (it doesn't touch consolidation). + fn merge(in1: &mut VecDeque, in2: &mut VecDeque, out: &mut VecDeque) { + let c1 = in1.pop_front().unwrap(); + let c2 = in2.pop_front().unwrap(); + let (kv1, kv2) = (c1.kv(), c2.kv()); + let (n1, n2) = (c1.len_(), c2.len_()); + let (t1, d1) = (c1.times(), c1.diffs()); + let (t2, d2) = (c2.times(), c2.diffs()); + + let (mut tags, mut offs) = (Vec::new(), Vec::new()); + let (mut times, mut diffs) = (ColTimes::new(), Vec::new()); + let (mut p1, mut p2) = (0usize, 0usize); + while p1 < n1 && p2 < n2 { + // `(key, val)` structurally, then `time` in place via the columnar `Ref: Ord`. + let ord = compare_at(&kv1, p1, &kv2, p2).then_with(|| t1.cmp_cross(p1, t2, p2)); + match ord { + Ordering::Less => { tags.push(0); offs.push(p1); times.push_ref(t1, p1); diffs.push(d1[p1].clone()); p1 += 1; } + Ordering::Greater => { tags.push(1); offs.push(p2); times.push_ref(t2, p2); diffs.push(d2[p2].clone()); p2 += 1; } + Ordering::Equal => { + let mut d = d1[p1].clone(); + d.plus_equals(&d2[p2]); + if !d.is_zero() { tags.push(0); offs.push(p1); times.push_ref(t1, p1); diffs.push(d); } + p1 += 1; p2 += 1; + } + } + } + + let srcs = [Some(&kv1), Some(&kv2)]; + Self::emit(&srcs, &tags, &offs, ×, &diffs, out); + + // Push back the survivor's unconsumed suffix (all `>` the horizon), ahead of its deque. + if p1 < n1 { + let idx: Vec = (p1..n1).collect(); + let mut t = ColTimes::new(); + t.push_range(t1, p1, n1); + in1.push_front(Self::from_kv(gather(&kv1, &idx), t, d1[p1..].to_vec())); + } + if p2 < n2 { + let idx: Vec = (p2..n2).collect(); + let mut t = ColTimes::new(); + t.push_range(t2, p2, n2); + in2.push_front(Self::from_kv(gather(&kv2, &idx), t, d2[p2..].to_vec())); + } + } + + fn extract( + input: &mut VecDeque, + frontier: AntichainRef, + residual: &mut Antichain, + keep: &mut VecDeque, + ship: &mut VecDeque, + ) { + // One input chunk per call: partition into keep (`>= frontier`) and ship pieces via `gather`. + let Some(chunk) = input.pop_front() else { return }; + let kv = chunk.kv(); + let (times, diffs) = (chunk.times(), chunk.diffs()); + let (mut ki, mut si) = (Vec::new(), Vec::new()); + for i in 0..chunk.len_() { + let ti = times.get(i); + if frontier.less_equal(&ti) { residual.insert_ref(&ti); ki.push(i); } else { si.push(i); } + } + if !ki.is_empty() { + let mut t = ColTimes::new(); + for &i in &ki { t.push_ref(times, i); } + let d: Vec = ki.iter().map(|&i| diffs[i].clone()).collect(); + keep.push_back(Self::from_kv(gather(&kv, &ki), t, d)); + } + if !si.is_empty() { + let mut t = ColTimes::new(); + for &i in &si { t.push_ref(times, i); } + let d: Vec = si.iter().map(|&i| diffs[i].clone()).collect(); + ship.push_back(Self::from_kv(gather(&kv, &si), t, d)); + } + } + + fn advance( + input: &mut VecDeque, + frontier: AntichainRef, + done: bool, + out: &mut VecDeque, + ) { + // Concatenate the pushed-back carry with the newly-arrived chunks, then advance/consolidate + // each *complete* `(key, val)` group; withhold the last group as the carry unless `done`. + if input.is_empty() { return; } + let chunks: Vec = input.drain(..).collect(); + let (ckv, ctimes, cdiffs) = Self::concat(&chunks); + let n = ctimes.len(); + if n == 0 { return; } + + // Group boundaries in ONE pass (corgi `group_bounds`) instead of a per-row `compare_at` scan. + let bounds = group_bounds(&ckv); // exclusive group ends, ascending, `bounds.last() == n` + + // Giant-key case: a single group spanning the whole buffer → no group provably complete. + if !done && bounds.len() == 1 { + input.push_front(Self::from_kv(ckv, ctimes, cdiffs)); + return; + } + + // Withhold the trailing group as the carry unless `done` (its start = the 2nd-to-last end). + let end = if done { n } else { bounds[bounds.len() - 2] }; + if end < n { + let idx: Vec = (end..n).collect(); + let mut ct = ColTimes::new(); + ct.push_range(&ctimes, end, n); + input.push_front(Self::from_kv(gather(&ckv, &idx), ct, cdiffs[end..].to_vec())); + } + + // Advance + consolidate each complete group; emit `TARGET`-sized chunks. All rows of a group + // share `(key, val)`, so one representative offset materializes each output row's kv. Times are + // materialized here (owned `T`) because `advance_by` mutates and the tiebreak re-sort is a Rust + // sort — the compaction path, not the merge hot path. + let srcs = [Some(&ckv)]; + let (mut tags, mut offs) = (Vec::new(), Vec::new()); + let (mut otimes, mut odiffs): (ColTimes, Vec) = (ColTimes::new(), Vec::new()); + let mut i = 0; + for &g_end in &bounds { + if g_end > end { break; } + let mut pairs: Vec<(T, R)> = (i..g_end) + .map(|k| { let mut t = ctimes.get(k); t.advance_by(frontier); (t, cdiffs[k].clone()) }) + .collect(); + pairs.sort_by(|a, b| a.0.cmp(&b.0)); + let mut k = 0; + while k < pairs.len() { + let t = pairs[k].0.clone(); + let mut d = pairs[k].1.clone(); + k += 1; + while k < pairs.len() && pairs[k].0 == t { d.plus_equals(&pairs[k].1); k += 1; } + if !d.is_zero() { + tags.push(0); offs.push(i); otimes.push(&t); odiffs.push(d); + if otimes.len() >= TARGET { + Self::emit(&srcs, &tags, &offs, &otimes, &odiffs, out); + tags.clear(); offs.clear(); otimes.clear(); odiffs.clear(); + } + } + } + i = g_end; + } + if !otimes.is_empty() { Self::emit(&srcs, &tags, &offs, &otimes, &odiffs, out); } + } + + /// Maximal packing via the harness [`pack`]: coalesce by concatenating columns (`gather_lanes`), + /// split with `gather`, seal as a no-op (corgi chunks are never paged here). + fn settle(input: &mut VecDeque, done: bool, out: &mut VecDeque) { + pack( + input, + done, + out, + |acc, next| { + let (na, nb) = (acc.len_(), next.len_()); + let kvs = [acc.kv(), next.kv()]; + let srcs = [Some(&kvs[0]), Some(&kvs[1])]; + let mut tags = Vec::with_capacity(na + nb); + let mut offs = Vec::with_capacity(na + nb); + for o in 0..na { tags.push(0); offs.push(o); } + for o in 0..nb { tags.push(1); offs.push(o); } + let kv = gather_lanes(&srcs, &tags, &offs); + let mut times = ColTimes::new(); + times.push_range(acc.times(), 0, na); + times.push_range(next.times(), 0, nb); + let mut diffs = acc.diffs().to_vec(); + diffs.extend_from_slice(next.diffs()); + *acc = Self::from_kv(kv, times, diffs); + }, + |chunk, m| { + let kv = chunk.kv(); + let n = chunk.len_(); + let left: Vec = (0..m).collect(); + let right: Vec = (m..n).collect(); + let (mut lt, mut rt) = (ColTimes::new(), ColTimes::new()); + lt.push_range(chunk.times(), 0, m); + rt.push_range(chunk.times(), m, n); + let l = Self::from_kv(gather(&kv, &left), lt, chunk.diffs()[..m].to_vec()); + let r = Self::from_kv(gather(&kv, &right), rt, chunk.diffs()[m..].to_vec()); + (l, r) + }, + |chunk| chunk, + ); + } +} + +/// Sort parallel columns by `(key, val, time)` and consolidate exact `(key, val, time)` triples +/// (summing diffs, dropping zeros). Returns a sorted+consolidated `(keys, vals, times, diffs)`. +/// +/// Multi-record: one columnar `sort_perm` (discrimination sort) orders by `(key, val)`, one batched +/// `compare_idx` flags adjacent-equal runs; only the small per-run *time* tiebreak is a Rust sort +/// (time is not a corgi type). No per-pair `compare_at`. +fn sort_consolidate(keys: CValue, vals: CValue, times: Vec, diffs: Vec) -> (CValue, CValue, Vec, Vec) +where + T: Ord + Clone + Columnar, + R: Semigroup + Clone, +{ + let n = times.len(); + if n == 0 { + return (keys, vals, times, diffs); + } + let kv = CValue::Prod(vec![keys, vals]); + // Batched argsort by (key, val); reorder the parallel Rust columns by the same permutation. + let perm = sort_perm(&kv); + let kv_s = gather(&kv, &perm); + let times_s: Vec = perm.iter().map(|&i| times[i].clone()).collect(); + let diffs_s: Vec = perm.iter().map(|&i| diffs[i].clone()).collect(); + // Batched adjacent-equality over the kv-sorted column: `adj[m] == 0` iff `kv_s[m] == kv_s[m+1]`. + let adj: Vec = if n > 1 { + let left: Vec = (0..n - 1).collect(); + let right: Vec = (1..n).collect(); + compare_idx(&kv_s, &kv_s, &left, &right) + } else { + Vec::new() + }; + + // Walk maximal equal-`(key,val)` runs; within each, order by time and consolidate equal times. + let (mut keep, mut ot, mut od) = (Vec::new(), Vec::new(), Vec::new()); + let mut i = 0; + while i < n { + let mut j = i + 1; + while j < n && adj[j - 1] == 0 { + j += 1; + } + let mut run: Vec = (i..j).collect(); + run.sort_by(|&a, &b| times_s[a].cmp(×_s[b])); + let mut k = 0; + while k < run.len() { + let rep = run[k]; + let t = times_s[rep].clone(); + let mut d = diffs_s[rep].clone(); + k += 1; + while k < run.len() && times_s[run[k]] == t { + d.plus_equals(&diffs_s[run[k]]); + k += 1; + } + if !d.is_zero() { + keep.push(rep); + ot.push(t); + od.push(d); + } + } + i = j; + } + let (keys, vals) = split_kv(gather(&kv_s, &keep)); + (keys, vals, ot, od) +} + +impl CorgiChunk +where + T: ColTime, + R: Semigroup + Clone + 'static, +{ + /// One sorted+consolidated chunk from columns already in corgi form (the column-native arrange + /// ingest — no transcode). + pub fn from_columns(keys: CValue, vals: CValue, times: Vec, diffs: Vec) -> Self { + let (keys, vals, times, diffs) = sort_consolidate(keys, vals, times, diffs); + Self::from_parts(keys, vals, ColTimes::from_iter(times), diffs) + } + +} + +/// Concatenate chunks' columns into flat `(keys, vals, times, diffs)` with **no transcode** — for +/// reading an arrangement back column-natively (e.g. `Backend::as_collection` straight into a +/// `CorgiContainer`), instead of untranscoding to rows and re-transcoding. +pub fn chunks_to_columns(chunks: &[CorgiChunk]) -> (CValue, CValue, Vec, Vec) +where + T: ColTime, + R: Semigroup + Clone + 'static, +{ + if chunks.iter().all(|c| c.len_() == 0) { + return (CValue::Unit(0), CValue::Unit(0), Vec::new(), Vec::new()); + } + let (kv, times, diffs) = CorgiChunk::concat(chunks); + let (keys, vals) = split_kv(kv); + (keys, vals, times.to_vec(), diffs) +} + +/// Build a `ChunkBatch` from corgi key/val COLUMNS directly (no transcode): sort + +/// consolidate into one chunk, then `settle`. The column-native egress the reduce backend seals its +/// output with (it resolves proxy ids to real columns by `gather` and hands them here). +pub fn columns_to_batch(keys: CValue, vals: CValue, times: Vec, diffs: Vec, description: Description) -> ChunkBatch> +where + T: ColTime, + R: Semigroup + Clone + 'static, +{ + let chunk = CorgiChunk::from_columns(keys, vals, times, diffs); + settle_one(chunk, description) +} + +/// Grade one chunk into a `ChunkBatch` (shared tail of `rows_to_batch`/`columns_to_batch`). +fn settle_one(chunk: CorgiChunk, description: Description) -> ChunkBatch> +where + T: ColTime, + R: Semigroup + Clone + 'static, +{ + let mut input = VecDeque::new(); + if chunk.len_() > 0 { input.push_back(chunk); } + let mut output = VecDeque::new(); + CorgiChunk::settle(&mut input, true, &mut output); + ChunkBatch::new(output.into(), description) +} + +/// A column-native arrange **chunker**: turns input `CorgiContainer`s into sorted+consolidated +/// `CorgiChunk`s WITHOUT `ContainerChunker`'s drain-to-rows (which untranscodes). Paired with the +/// standard `ChunkBatcher`/`ChunkBuilder`, the arrange ingest stays column-native — no +/// columns→rows→columns round-trip at the arrangement boundary. +/// +/// Crucially it **accumulates to `TARGET`** before consolidating (like `ContainerChunker`), so it +/// emits few large chunks rather than one tiny chunk per input container — otherwise the columnar +/// per-chunk set-up (`gather`/`sort_perm`) dominates when input arrives as many small batches. +pub struct CorgiChunker { + /// Un-consolidated key/val column blocks (one per absorbed container), flat time/diff. + k_blocks: Vec, + v_blocks: Vec, + times: Vec, + diffs: Vec, + ready: VecDeque>, + current: Option>, +} + +impl Default for CorgiChunker { + fn default() -> Self { + CorgiChunker { k_blocks: Vec::new(), v_blocks: Vec::new(), times: Vec::new(), diffs: Vec::new(), ready: VecDeque::new(), current: None } + } +} + +/// Concatenate column blocks into one column (multi-source `gather_lanes`, no sort). +fn concat_blocks(blocks: &[CValue]) -> CValue { + if blocks.len() == 1 { + return blocks[0].clone(); + } + let srcs: Vec> = blocks.iter().map(Some).collect(); + let (mut tags, mut offs) = (Vec::new(), Vec::new()); + for (ti, b) in blocks.iter().enumerate() { + for o in 0..b.len() { tags.push(ti); offs.push(o); } + } + gather_lanes(&srcs, &tags, &offs) +} + +impl CorgiChunker +where + T: ColTime, + R: Semigroup + Clone + 'static, +{ + /// Consolidate the accumulated blocks into one graded chunk (concat columns, then sort+consolidate). + fn flush(&mut self) { + if self.times.is_empty() { + return; + } + let keys = concat_blocks(&self.k_blocks); + let vals = concat_blocks(&self.v_blocks); + self.k_blocks.clear(); + self.v_blocks.clear(); + let times = std::mem::take(&mut self.times); + let diffs = std::mem::take(&mut self.diffs); + let chunk = CorgiChunk::from_columns(keys, vals, times, diffs); + if chunk.len_() > 0 { + self.ready.push_back(chunk); + } + } +} + +impl timely::container::PushInto<&mut crate::corgi::container::CorgiContainer> for CorgiChunker +where + T: ColTime, + R: Semigroup + Clone + 'static, +{ + fn push_into(&mut self, c: &mut crate::corgi::container::CorgiContainer) { + if c.times.is_empty() { + return; + } + self.k_blocks.push(std::mem::replace(&mut c.keys, CValue::Unit(0))); + self.v_blocks.push(std::mem::replace(&mut c.vals, CValue::Unit(0))); + self.times.append(&mut c.times); + self.diffs.append(&mut c.diffs); + if self.times.len() >= TARGET { + self.flush(); + } + } +} + +impl timely::container::ContainerBuilder for CorgiChunker +where + T: ColTime, + R: Semigroup + Clone + 'static, +{ + type Container = CorgiChunk; + // `extract` ships ready chunks, leaving the sub-TARGET remainder to accumulate further. + fn extract(&mut self) -> Option<&mut Self::Container> { + self.current = self.ready.pop_front(); + self.current.as_mut() + } + // `finish` also flushes the remainder (called until it returns None). + fn finish(&mut self) -> Option<&mut Self::Container> { + self.flush(); + self.extract() + } +} + +/// A single sorted+consolidated run over corgi columns — the join tactic's per-side input, produced +/// by merging a batch list [`flatten_batches`]. Separate `keys`/`vals` columns so the merge-join can +/// compare by key (`compare_at`) and `gather` matched runs. +pub struct SortedRun { + pub keys: CValue, + pub vals: CValue, + pub times: Vec, + pub diffs: Vec, +} + +/// Flatten a list of batches into one sorted+consolidated run over columns (no row round-trip, no +/// re-sort). Each batch's `.chunks` is already a sorted, consolidated chain; we **merge** those chains +/// (reusing their order via `merge_chains` → `CorgiChunk::merge`) into one, then concatenate. `None` +/// if empty. This replaces an earlier concat+full-sort that dominated recursive (reach) cost. +pub fn flatten_batches(batches: &[Rc>>]) -> Option> +where + T: ColTime, + R: Semigroup + Clone + 'static, +{ + let mut merged: VecDeque> = VecDeque::new(); + for b in batches { + if b.chunks.is_empty() { continue; } + if merged.is_empty() { + merged.extend(b.chunks.iter().cloned()); + } else { + let chain: Vec> = merged.drain(..).collect(); + merge_chains(chain, b.chunks.clone(), &mut merged); + } + } + if merged.is_empty() { return None; } + // The merged chain is globally sorted+consolidated: concatenate (no sort) into one flat run. + let chunks: Vec> = merged.into(); + let (kv, times, diffs) = CorgiChunk::concat(&chunks); + let (keys, vals) = split_kv(kv); + Some(SortedRun { keys, vals, times: times.to_vec(), diffs }) +} + +/// A DELTA-PROPORTIONAL flatten of the accumulated side of a bilinear join: instead of merging the +/// whole trace, probe each accumulated chunk with the fresh side's `needle` keys (`find_ranges`, a +/// batched equal-range seek — each chunk is structurally sorted by key) and gather ONLY the matched +/// records, then sort+consolidate the (small) matched set into one run. Keys absent from `needles` +/// can never join, so dropping them is exact. Cost tracks the fresh key set + matches, never the +/// trace — replacing the O(trace)-per-work-unit `flatten_batches` that dominated recursive joins. +pub fn flatten_restricted(acc: &[Rc>>], needles: &CValue) -> Option> +where + T: ColTime, + R: Semigroup + Clone + 'static, +{ + let (mut kblocks, mut vblocks): (Vec, Vec) = (Vec::new(), Vec::new()); + let (mut times, mut diffs): (Vec, Vec) = (Vec::new(), Vec::new()); + for b in acc { + for ch in &b.chunks { + if ch.len_() == 0 { continue; } + // Per needle key, its equal-range in this chunk's (key-sorted) key column. + let (lo, hi) = find_ranges(needles, ch.keys()); + let mut idx: Vec = Vec::new(); + for i in 0..lo.len() { idx.extend(lo[i]..hi[i]); } + idx.sort_unstable(); + idx.dedup(); + if idx.is_empty() { continue; } + kblocks.push(gather(ch.keys(), &idx)); + vblocks.push(gather(ch.vals(), &idx)); + for &j in &idx { + times.push(ch.times().get(j)); + diffs.push(ch.diffs()[j].clone()); + } + } + } + if times.is_empty() { return None; } + let keys = concat_blocks(&kblocks); + let vals = concat_blocks(&vblocks); + let (keys, vals, times, diffs) = sort_consolidate(keys, vals, times, diffs); + Some(SortedRun { keys, vals, times, diffs }) +} + +#[cfg(test)] +mod test { + use super::*; + use differential_dataflow::trace::chunk::{ChunkBatchMerger, is_graded}; + use differential_dataflow::trace::{Description, Merger}; + use std::collections::BTreeMap; + + fn xorshift(s: &mut u64) -> u64 { *s ^= *s << 13; *s ^= *s >> 7; *s ^= *s << 17; *s } + + /// Build a single sorted+consolidated CorgiChunk from u64 (key,val,time,diff) rows. + fn chunk(rows: &[((u64, u64), u64, i64)]) -> CorgiChunk { + // Sort + consolidate by ((k,v),t) so a chunk is a legal sorted run. + let mut m: BTreeMap<((u64, u64), u64), i64> = BTreeMap::new(); + for &(kv, t, d) in rows { *m.entry((kv, t)).or_insert(0) += d; } + m.retain(|_, d| *d != 0); + let keys = CValue::u64(m.keys().map(|((k, _), _)| *k).collect()); + let vals = CValue::u64(m.keys().map(|((_, v), _)| *v).collect()); + let times: ColTimes = m.keys().map(|(_, t)| *t).collect(); + let diffs = m.values().copied().collect(); + CorgiChunk::from_parts(keys, vals, times, diffs) + } + + fn read_batch(b: &ChunkBatch>) -> BTreeMap<((u64, u64), u64), i64> { + let mut m = BTreeMap::new(); + for ch in &b.chunks { + let ks = ch.keys().clone().into_u64("k"); + let vs = ch.vals().clone().into_u64("v"); + for i in 0..ch.len_() { *m.entry(((ks[i], vs[i]), ch.times().get(i))).or_insert(0) += ch.diffs()[i]; } + } + m.retain(|_, d| *d != 0); + m + } + + fn reference(u1: &[((u64, u64), u64, i64)], u2: &[((u64, u64), u64, i64)], f: u64) -> BTreeMap<((u64, u64), u64), i64> { + let mut m = BTreeMap::new(); + for u in u1.iter().chain(u2) { *m.entry((u.0, u.1.max(f))).or_insert(0) += u.2; } // advance_by on u64 = max + m.retain(|_, d| *d != 0); + m + } + + /// Cut a consolidated set into a batch of small chunks (globally sorted; groups straddle). + fn batch(rows: &[((u64, u64), u64, i64)], sz: usize) -> ChunkBatch> { + let mut m: BTreeMap<((u64, u64), u64), i64> = BTreeMap::new(); + for &(kv, t, d) in rows { *m.entry((kv, t)).or_insert(0) += d; } + m.retain(|_, d| *d != 0); + let all: Vec<((u64, u64), u64, i64)> = m.into_iter().map(|((kv, t), d)| (kv, t, d)).collect(); + let chunks: Vec<_> = all.chunks(sz.max(1)).map(chunk).collect(); + let desc = Description::new(Antichain::from_elem(0u64), Antichain::from_elem(10u64), Antichain::from_elem(0u64)); + ChunkBatch::new(chunks, desc) + } + + #[test] + fn batch_merger_resumable_matches_reference() { + let mut seed = 0x9E3779B97F4A7C15u64; + for _ in 0..200 { + let gen = |seed: &mut u64| -> Vec<((u64, u64), u64, i64)> { + let n = (xorshift(seed) % 40) as usize + 1; + (0..n).map(|_| { + let k = xorshift(seed) % 10; let v = xorshift(seed) % 3; let t = xorshift(seed) % 6; + let d = if xorshift(seed) % 4 == 0 { -1 } else { 1 }; + ((k, v), t, d) + }).collect() + }; + let u1 = gen(&mut seed); + let u2 = gen(&mut seed); + let sz = (xorshift(&mut seed) % 4) as usize + 1; + let f = xorshift(&mut seed) % 6; + let (s1, s2) = (batch(&u1, sz), batch(&u2, sz)); + let frontier = Antichain::from_elem(f); + + let mut merger = ChunkBatchMerger::new(&s1, &s2, frontier.borrow()); + loop { + let mut fuel = 1isize; // tiny → many yields, each settling + merger.work(&s1, &s2, &mut fuel); + if fuel > 0 { break; } + } + let result = merger.done(); + assert!(is_graded(&result.chunks), "ungraded: {:?}", result.chunks.iter().map(Chunk::len).collect::>()); + assert_eq!(read_batch(&result), reference(&u1, &u2, f), "u1={u1:?}\nu2={u2:?}\nf={f}"); + } + } +} diff --git a/interactive/src/corgi/col_times.rs b/interactive/src/corgi/col_times.rs new file mode 100644 index 000000000..01d3cb313 --- /dev/null +++ b/interactive/src/corgi/col_times.rs @@ -0,0 +1,149 @@ +//! Columnar per-tuple times for `CorgiChunk`. +//! +//! `Inner.times` was `Vec` — one heap-ish `PointStamp` (`SmallVec`) per row, the dominant SCC +//! allocation (~40% of the profile). Since our `T = Product>` already derives +//! `columnar::Columnar` (Product, PointStamp, and u64 all do), the same times live in SoA form in +//! `::Container` — one pair of allocations (offsets + values) for the whole column +//! instead of `n` `SmallVec`s. +//! +//! Times are compared IN PLACE via the container's derived `Ord` on `Ref` (both `Product` and +//! `PointStamp` carry `#[columnar(derive(Ord, PartialOrd))]`), so merge/sort never materialize a +//! `T`. An owned `T` is reconstructed (`get`) only where a `Lattice` op is unavoidable — `join` in +//! the join cross-product, `advance_by` in compaction — or at the emit boundary handing `T` back to +//! DD. Range copies (`emit`/`concat`) push `Ref`s straight across (`push_ref`), also no `T`. +//! +//! This is the O(data) time store; DD's `Chunk` boundary only ever sees whole chunks + frontier +//! antichains (control complexity), so this stays entirely inside the backend — no DD change. + +use std::cmp::Ordering; + +use columnar::{Borrow, Clear, Columnar, Index, Len, Push}; + +use differential_dataflow::lattice::Lattice; +use timely::progress::Timestamp; + +/// A timestamp usable as a columnar time column: `Timestamp + Lattice` (DD's algebra) plus +/// `Columnar` with an *ordered* `Ref` (so times compare in their SoA form). Our +/// `Product>` satisfies it — every layer derives `Columnar` with +/// `#[columnar(derive(Ord, PartialOrd))]`. +/// +/// The `Ref: Ord` requirement is an HRTB on a projection (`for<'a> Ref<'a, Self>: Ord`), which Rust +/// does NOT imply from a plain `T: ColTime` bound at use sites — so it would otherwise go viral +/// across every fn driving the `Chunk` methods. We discharge it ONCE here, in the blanket impl, +/// behind [`ColTime::cmp_refs`]; downstream code compares times through that method and needs only +/// `T: ColTime`. +pub trait ColTime: Timestamp + Lattice + Columnar { + /// Order two SoA time references (via the derived `Ref: Ord`, lifetimes reborrowed to unify). + fn cmp_refs(a: columnar::Ref<'_, Self>, b: columnar::Ref<'_, Self>) -> Ordering; +} + +impl ColTime for T +where + T: Timestamp + Lattice + Columnar, + for<'a> columnar::Ref<'a, T>: Ord, +{ + #[inline] + fn cmp_refs(a: columnar::Ref<'_, T>, b: columnar::Ref<'_, T>) -> Ordering { + let a = <::Container as Borrow>::reborrow_ref(a); + let b = <::Container as Borrow>::reborrow_ref(b); + a.cmp(&b) + } +} + +/// SoA column of per-tuple times, backed by `::Container`. +#[derive(Default)] +pub struct ColTimes { + store: ::Container, +} + +impl ColTimes { + #[inline] + pub fn new() -> Self { + ColTimes { store: Default::default() } + } + + #[inline] + pub fn len(&self) -> usize { + self.store.borrow().len() + } + + #[inline] + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// Empty the column, retaining allocation (the emit-flush reuse in `advance`). + #[inline] + pub fn clear(&mut self) { + self.store.clear(); + } + + /// Append an owned time (columnar `Push<&T>`); the value is stored SoA, not cloned whole. + #[inline] + pub fn push(&mut self, t: &T) { + self.store.push(t); + } + + /// Append `other`'s row `i` by pushing its `Ref` straight across — no `T` materialized. Used by + /// the range copies in `emit`/`concat` (`Container: Push`). + #[inline] + pub fn push_ref(&mut self, other: &ColTimes, i: usize) { + self.store.push(other.store.borrow().get(i)); + } + + /// The owned time at row `i` — materializes a `T`. Reserve for `Lattice` ops and the emit + /// boundary; use `cmp`/`cmp_cross` for ordering. + #[inline] + pub fn get(&self, i: usize) -> T { + ::into_owned(self.store.borrow().get(i)) + } + + /// Append rows `[s, e)` of `other`, pushing `Ref`s straight across — no `T` materialized. The + /// range copy used by `emit`/`concat`/merge-suffix. + #[inline] + pub fn push_range(&mut self, other: &ColTimes, s: usize, e: usize) { + let b = other.store.borrow(); + for i in s..e { + self.store.push(b.get(i)); + } + } + + /// Materialize the whole column to `Vec` — the egress boundary (`SortedRun` for the join, + /// `CorgiContainer` for `as_collection`), where owned `T` is wanted anyway. + pub fn to_vec(&self) -> Vec { + let b = self.store.borrow(); + (0..b.len()).map(|i| ::into_owned(b.get(i))).collect() + } + + /// Order rows `i` and `j` within this column, in place via [`ColTime::cmp_refs`] (no `T`). + #[inline] + pub fn cmp(&self, i: usize, j: usize) -> Ordering + where + T: ColTime, + { + let b = self.store.borrow(); + T::cmp_refs(b.get(i), b.get(j)) + } + + /// Order this column's row `i` against `other`'s row `j` (the two-pointer merge compare). + #[inline] + pub fn cmp_cross(&self, i: usize, other: &ColTimes, j: usize) -> Ordering + where + T: ColTime, + { + let ba = self.store.borrow(); + let bb = other.store.borrow(); + T::cmp_refs(ba.get(i), bb.get(j)) + } +} + +/// Build a column from an iterator of owned times (the `FromIterator` path used at construction). +impl FromIterator for ColTimes { + fn from_iter>(iter: I) -> Self { + let mut store: ::Container = Default::default(); + for t in iter { + store.push(&t); + } + ColTimes { store } + } +} diff --git a/interactive/src/corgi/container.rs b/interactive/src/corgi/container.rs new file mode 100644 index 000000000..fe753670a --- /dev/null +++ b/interactive/src/corgi/container.rs @@ -0,0 +1,149 @@ +//! Phase 2 — the corgi-native container: corgi columns for the (key,val) payload, plain Rust +//! Vecs for time/diff (corgi never touches the lattice). This is what flows on dataflow edges in +//! the corgi backend; operators transform block→block via `eval_graph` with NO per-op transcode. +//! Conversion to/from DDIR rows happens only at I/O boundaries (`from_updates`/`into_updates`). +//! +//! Minimal trait surface for a pipelined (single-worker) dataflow: `Accountable + Default + Clone` +//! (= `timely::Container`). `Negate`/`Enter`/`Leave`/`ResultsIn` (iterative scopes) come with M3. + +use timely::Accountable; +use timely::progress::{PathSummary, Timestamp}; + +use differential_dataflow::collection::containers::{Enter, Leave, Negate, ResultsIn}; +use differential_dataflow::difference::Abelian; + +use crate::corgi::logic::{infer_shape_cols, transcode, untranscode}; +use crate::ir::Value as DValue; + +type Row = DValue; + +/// A batch of `(key, val, time, diff)` updates: payload columnar (corgi), time/diff native Rust. +pub struct CorgiContainer { + /// Key column (corgi columnar `Value`). + pub keys: corgi::Value, + /// Val column (corgi columnar `Value`). + pub vals: corgi::Value, + /// Per-update times (corgi never reads these; the Rust side keeps the lattice algebra). + pub times: Vec, + /// Per-update diffs. + pub diffs: Vec, +} + +impl Default for CorgiContainer { + fn default() -> Self { + // Empty sentinel columns (shape-agnostic length-0 unit columns). + CorgiContainer { keys: corgi::Value::Unit(0), vals: corgi::Value::Unit(0), times: Vec::new(), diffs: Vec::new() } + } +} + +impl Clone for CorgiContainer { + fn clone(&self) -> Self { + // corgi `Value` clone is an Arc bump on the leaf buffers — columns are shared, not copied. + CorgiContainer { keys: self.keys.clone(), vals: self.vals.clone(), times: self.times.clone(), diffs: self.diffs.clone() } + } +} + +impl Accountable for CorgiContainer { + #[inline] + fn record_count(&self) -> i64 { + self.times.len() as i64 + } +} + +/// Draining a `CorgiContainer` yields its DDIR row updates (untranscode) — lets the standard +/// `ContainerChunker>` chunk a corgi-container stream into row chains for the +/// reused `MergeBatcher`, which the `CorgiBatchBuilder` then transcodes back to corgi columns at +/// arrangement build (the one ingest-boundary round-trip; reduce/join read corgi columns). +impl timely::container::DrainContainer for CorgiContainer { + type Item<'a> = ((Row, Row), T, R) where Self: 'a; + type DrainIter<'a> = std::vec::IntoIter<((Row, Row), T, R)> where Self: 'a; + fn drain(&mut self) -> Self::DrainIter<'_> { + std::mem::take(self).into_updates().into_iter() + } +} + +impl CorgiContainer { + /// Build a container from DDIR row updates — the **ingest boundary** transcode (once per batch). + /// Shapes are inferred by scanning the whole column ([`infer_shape_cols`]) — required so a + /// `Variant` column discovers all its arms (a single sample shows only one tag). + pub fn from_updates(updates: Vec<((Row, Row), T, R)>) -> Self { + if updates.is_empty() { + return Self::default(); + } + let keys_rows: Vec = updates.iter().map(|u| u.0 .0.clone()).collect(); + let vals_rows: Vec = updates.iter().map(|u| u.0 .1.clone()).collect(); + let kshape = infer_shape_cols(&keys_rows); + let vshape = infer_shape_cols(&vals_rows); + let times = updates.iter().map(|u| u.1.clone()).collect(); + let diffs = updates.iter().map(|u| u.2.clone()).collect(); + CorgiContainer { keys: transcode(&keys_rows, &kshape), vals: transcode(&vals_rows, &vshape), times, diffs } + } + + /// Read the container back to DDIR row updates — the **egress boundary** transcode (once). + /// corgi `Value` is self-describing, so shapes come from `shape_of_value`. + pub fn into_updates(self) -> Vec<((Row, Row), T, R)> { + if self.times.is_empty() { + return Vec::new(); + } + let kshape = corgi::shape_of_value(&self.keys); + let vshape = corgi::shape_of_value(&self.vals); + let keys_rows = untranscode(self.keys, &kshape); + let vals_rows = untranscode(self.vals, &vshape); + keys_rows + .into_iter() + .zip(vals_rows) + .zip(self.times) + .zip(self.diffs) + .map(|(((k, v), t), d)| ((k, v), t, d)) + .collect() + } +} + +// --- Container traits required by the `Backend` bound + iterative scopes --- +// time/diff live in Rust, so these are plain Rust passes; the corgi key/val columns only move +// (`gather`) when `ResultsIn` drops rows. `Enter`/`Leave` are identity for DDIR's same-Time dynamic +// timestamp model (region entry doesn't change the time type; `leave_dynamic` pops the coord). + +impl Negate for CorgiContainer { + fn negate(mut self) -> Self { + for d in self.diffs.iter_mut() { + d.negate(); + } + self + } +} + +impl Enter for CorgiContainer { + type InnerContainer = Self; + fn enter(self) -> Self { + self + } +} + +impl Leave for CorgiContainer { + type OuterContainer = Self; + fn leave(self) -> Self { + self + } +} + +impl ResultsIn for CorgiContainer { + fn results_in(self, step: &T::Summary) -> Self { + let n = self.times.len(); + let mut keep = Vec::with_capacity(n); + let mut new_times = Vec::with_capacity(n); + for (i, t) in self.times.iter().enumerate() { + if let Some(nt) = step.results_in(t) { + keep.push(i); + new_times.push(nt); + } + } + if keep.len() == n { + return CorgiContainer { keys: self.keys, vals: self.vals, times: new_times, diffs: self.diffs }; + } + let keys = corgi::arrange::gather(&self.keys, &keep); + let vals = corgi::arrange::gather(&self.vals, &keep); + let diffs = keep.iter().map(|&i| self.diffs[i].clone()).collect(); + CorgiContainer { keys, vals, times: new_times, diffs } + } +} diff --git a/interactive/src/corgi/join.rs b/interactive/src/corgi/join.rs new file mode 100644 index 000000000..f8589840f --- /dev/null +++ b/interactive/src/corgi/join.rs @@ -0,0 +1,120 @@ +//! Phase 2 / M2(b)-4 — the cursor-less corgi JOIN tactic (Route B; reduce `retire` is Frank's). +//! +//! `join_with_tactic` needs only `TraceReader` (not `Navigable`), so a cursor-less `CorgiBatch` +//! arrangement drives a custom `JoinTactic`: `defer` queues bilinear units (fresh batch list × +//! accumulated other side, keyed by `Fresh`); `work` (NEXT ITERATION) merges each side's list to one +//! run, merge-joins by key over corgi columns (`compare_at`), cross-products matched val runs with +//! lattice time-join + diff-multiply, runs the projection corgi program, and emits via the session. +//! +//! THIS ITERATION: struct + `defer` + a stub `work`, compiling as a valid `JoinTactic` (validates the +//! generic-bound wiring). + +use std::rc::Rc; + + +use differential_dataflow::operators::join::{Fresh, JoinTactic}; + +use corgi::arrange::{find_ranges, gather}; +use corgi::Value as CValue; + +use differential_dataflow::trace::chunk::{Chunk, ChunkBatch}; + +use crate::corgi::container::CorgiContainer; +use crate::corgi::col_times::ColTime; +use crate::corgi::chunk::{flatten_batches, flatten_restricted, CorgiChunk}; +use crate::corgi::logic::compile_join_projection; +use crate::ir::Diff; +use crate::parse::Term; +type CBatch = Rc>>; + +/// Cursor-less corgi join tactic. Holds the projection `Term`s (`Var(0)=key`, `Var(1)=val0`, +/// `Var(2)=val1`); compiled to a corgi graph per work-unit using the matched columns' shapes (so +/// `Spread` resolves against the actual key/val arities). +pub struct CorgiJoinTactic { + key: Term, + val: Term, + _t: std::marker::PhantomData, +} + +impl CorgiJoinTactic { + pub fn new(key: Term, val: Term) -> Self { + CorgiJoinTactic { key, val, _t: std::marker::PhantomData } + } +} + +impl JoinTactic, CBatch, CorgiContainer> for CorgiJoinTactic +where + T: ColTime, +{ + /// One bilinear unit → its output container. The operator (`join_with_tactic`) drives capabilities + /// and fuel by draining the returned iterator; `meet` (compaction) is ignored (correctness-first, + /// like the reference identity backend — output times are just less compact, never wrong). + fn prep(&mut self, input0: Vec>, input1: Vec>, fresh: Fresh, _meet: T) -> Box>> { + Box::new(run_unit(input0, input1, fresh, &self.key, &self.val).into_iter()) + } +} + +/// Join one deferred bilinear unit: merge-join the two runs by key over corgi columns, cross-product +/// matched val runs (lattice time-join + diff-multiply), project via corgi, emit at the capability. +fn run_unit(left_b: Vec>, right_b: Vec>, fresh: Fresh, key: &Term, val: &Term) -> Option> +where + T: ColTime, +{ + // Materialize the FRESH (delta-sized) side whole; then present the ACCUMULATED side. When it is + // meaningfully larger than the fresh delta (recursive joins against a built-up trace), probe it + // with the fresh keys and gather only matches (delta-proportional, not O(trace)); otherwise + // (non-recursive / comparable sizes) a plain flatten is cheaper than the probe + reconsolidate. + // Roles (left/right) are preserved for the projection. + let accumulated = |acc: &[CBatch], fresh: &crate::corgi::chunk::SortedRun| { + let acc_len: usize = acc.iter().flat_map(|b| b.chunks.iter()).map(|c| c.len()).sum(); + if acc_len > 2 * fresh.times.len() { + flatten_restricted(acc, &fresh.keys) + } else { + flatten_batches(acc) + } + }; + let (left, right) = match fresh { + Fresh::Input0 => { + let left = flatten_batches(&left_b)?; + let right = accumulated(&right_b, &left)?; + (left, right) + } + Fresh::Input1 => { + let right = flatten_batches(&right_b)?; + let left = accumulated(&left_b, &right)?; + (left, right) + } + }; + let nl = left.times.len(); + + // Multi-record merge-join: one batched `find` gives, per left row, the equal-range of its key in + // the sorted right keys. Cross-product each left row with its matched right rows (lattice + // time-join + diff-multiply). Replaces the per-pair `compare_at` scan. + let (lo, hi) = find_ranges(&left.keys, &right.keys); + let (mut li, mut ri): (Vec, Vec) = (Vec::new(), Vec::new()); + let (mut ot, mut od): (Vec, Vec) = (Vec::new(), Vec::new()); + for a in 0..nl { + for b in lo[a]..hi[a] { + li.push(a); + ri.push(b); + ot.push(left.times[a].join(&right.times[b])); + od.push(left.diffs[a] * right.diffs[b]); + } + } + if li.is_empty() { + return None; + } + + // gather matched (key, val0, val1) columns, run the projection corgi program, untranscode. + let kc = gather(&left.keys, &li); + let v0 = gather(&left.vals, &li); + let v1 = gather(&right.vals, &ri); + // Compile the projection against the matched columns' shapes (so `Spread` resolves correctly). + let proj = compile_join_projection(key, val, &corgi::shape_of_value(&kc), &corgi::shape_of_value(&v0), &corgi::shape_of_value(&v1)); + let projected = corgi::eval_graph(&proj, CValue::Prod(vec![kc, v0, v1])); + let mut cols = projected.into_prod("corgi join projection"); + let nv = cols.pop().unwrap(); + let nk = cols.pop().unwrap(); + // The projection's corgi columns directly as a column-native CorgiContainer (no untranscode). + Some(CorgiContainer { keys: nk, vals: nv, times: ot, diffs: od }) +} diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs new file mode 100644 index 000000000..3132aa93e --- /dev/null +++ b/interactive/src/corgi/logic.rs @@ -0,0 +1,481 @@ +//! corgi as DDIR's columnar scalar logic: compile a `Term` to a corgi `Graph`, +//! transcode DDIR rows (`ir::Value`) to/from corgi columnar `Value`, directed by a `Shape` +//! inferred from the data (the dynamic-typing primitive). See SPIKE.md. +//! +//! The compiler (`compile`) covers Var/Bound/Int/Tuple(+Spread)/Proj/Binary/If/Fold over non-negative +//! ints; terms it can't lower (List, Case/Inject, Unary, Hash — see `compilable`) fall back to row-wise +//! `ir::eval` in the backend. The transcode layer is total over `Shape` (Prim/Unit/Prod/List/Sum), so a +//! `Variant` column round-trips via corgi `Sum` (see `infer_shape_cols` for the all-rows arm scan). + +use crate::ir::Value as DValue; +use crate::parse::{BinOp, Term}; + +use corgi::{ArithOp, BinOp as CBinOp, Builder, CmpOp, Graph, Kind, NumOp, Op, Pred, Shape, Value as CValue}; + +/// Dynamic typing: form a corgi `Shape` by observing a sample row. +pub fn infer_shape(sample: &DValue) -> Shape { + match sample { + DValue::Int(_) => Shape::Prim(64), + DValue::Tuple(xs) if xs.is_empty() => Shape::Unit, // DDIR unit ↔ corgi length-carrying Unit + DValue::Tuple(xs) => Shape::Prod(xs.iter().map(infer_shape).collect()), + DValue::List(xs) => Shape::List(Box::new(infer_shape(xs.first().expect("nonempty list")))), + DValue::Variant(..) => panic!("variant shape inference is a later rung"), + } +} + +/// Dynamic typing over a whole COLUMN: infer a `Shape` by scanning every row, not just a sample. +/// Required for sum types — a `Variant` column's shape is the union of all arms that appear, which a +/// single sample can't reveal (it shows only one tag). Non-sum shapes match [`infer_shape`] but recurse +/// column-wise so nested variants are covered too. (`from_updates` uses this; `infer_shape` stays for +/// the single-sample callers that never carry variants.) +pub fn infer_shape_cols(rows: &[DValue]) -> Shape { + let Some(first) = rows.first() else { return Shape::Unit }; + match first { + DValue::Int(_) => Shape::Prim(64), + DValue::Tuple(xs) if xs.is_empty() => Shape::Unit, + DValue::Tuple(xs) => { + let n = xs.len(); + Shape::Prod( + (0..n) + .map(|i| { + let col: Vec = rows + .iter() + .map(|r| match r { + DValue::Tuple(f) => f[i].clone(), + other => panic!("infer_shape_cols: expected Tuple, got {other:?}"), + }) + .collect(); + infer_shape_cols(&col) + }) + .collect(), + ) + } + DValue::List(_) => { + let flat: Vec = rows + .iter() + .flat_map(|r| match r { + DValue::List(xs) => xs.clone(), + other => panic!("infer_shape_cols: expected List, got {other:?}"), + }) + .collect(); + Shape::List(Box::new(infer_shape_cols(&flat))) + } + DValue::Variant(..) => { + // One lane per variant 0..=max_tag; a lane present in the data gets its arm's shape (from + // that arm's rows), an absent tag stays `None` (⊥, uncommitted — adopts a sibling on join). + let max_tag = rows + .iter() + .map(|r| match r { + DValue::Variant(t, _) => *t as usize, + other => panic!("infer_shape_cols: expected Variant, got {other:?}"), + }) + .max() + .unwrap(); + let lanes = (0..=max_tag) + .map(|tag| { + let payloads: Vec = rows + .iter() + .filter_map(|r| match r { + DValue::Variant(t, p) if *t as usize == tag => Some((**p).clone()), + _ => None, + }) + .collect(); + (!payloads.is_empty()).then(|| infer_shape_cols(&payloads)) + }) + .collect(); + Shape::Sum(lanes) + } + } +} + +/// AoS rows -> SoA corgi columns, directed by `shape`. +pub fn transcode(rows: &[DValue], shape: &Shape) -> CValue { + match shape { + Shape::Prim(_) => CValue::u64(rows.iter().map(|r| r.as_int() as u64).collect()), + Shape::Unit => CValue::Unit(rows.len()), + Shape::Prod(fs) => CValue::Prod( + fs.iter() + .enumerate() + .map(|(i, fsi)| { + let sub: Vec = rows + .iter() + .map(|r| match r { + DValue::Tuple(xs) => xs[i].clone(), + other => panic!("transcode: expected Tuple, got {other:?}"), + }) + .collect(); + transcode(&sub, fsi) + }) + .collect(), + ), + Shape::List(elem) => { + // List column = per-row END offsets + a flattened element column. + let mut ends = Vec::with_capacity(rows.len()); + let mut flat: Vec = Vec::new(); + let mut acc = 0usize; + for r in rows { + match r { + DValue::List(xs) => { + acc += xs.len(); + ends.push(acc); + flat.extend(xs.iter().cloned()); + } + other => panic!("transcode: expected List, got {other:?}"), + } + } + CValue::List(ends.into(), Box::new(transcode(&flat, elem))) + } + Shape::Sum(lanes) => { + // Per-row tag, plus one packed lane per committed variant (its arm's rows in row order; + // `Value::sum_opt` derives each row's within-lane offset). Absent arms stay `⊥` (None). + let tags: Vec = rows + .iter() + .map(|r| match r { + DValue::Variant(t, _) => *t as usize, + other => panic!("transcode: expected Variant, got {other:?}"), + }) + .collect(); + let lane_vals: Vec> = lanes + .iter() + .enumerate() + .map(|(tag, ls)| { + ls.as_ref().map(|lshape| { + let payloads: Vec = rows + .iter() + .filter_map(|r| match r { + DValue::Variant(t, p) if *t as usize == tag => Some((**p).clone()), + _ => None, + }) + .collect(); + transcode(&payloads, lshape) + }) + }) + .collect(); + CValue::sum_opt(tags, lane_vals) + } + } +} + +/// SoA corgi columns -> AoS rows, directed by `shape`. Inverse of [`transcode`]. +pub fn untranscode(col: CValue, shape: &Shape) -> Vec { + match shape { + Shape::Prim(_) => col.into_u64("untranscode").into_iter().map(|x| DValue::Int(x as i64)).collect(), + Shape::Unit => vec![DValue::unit(); col.len()], + Shape::Prod(fs) => { + let cols = col.into_prod("untranscode"); + let n = if cols.is_empty() { 0 } else { cols[0].len() }; + let per_field: Vec> = + cols.into_iter().zip(fs.iter()).map(|(c, fsi)| untranscode(c, fsi)).collect(); + (0..n).map(|i| DValue::Tuple(per_field.iter().map(|f| f[i].clone()).collect())).collect() + } + Shape::List(elem) => { + // Inverse of transcode's List: per-row END offsets + a flattened element column → one + // `List` per row, slicing the untranscoded flat column by each row's span. + let (bounds, vals) = match col { + CValue::List(b, vals) => (b, *vals), + other => panic!("untranscode: expected List, got {other:?}"), + }; + let flat = untranscode(vals, elem); + let ends: Vec = match &bounds { + corgi::Bounds::Offsets(v) => v.clone(), + corgi::Bounds::Stride(k, rows) => (1..=*rows).map(|i| i * k).collect(), + }; + let mut out = Vec::with_capacity(ends.len()); + let mut start = 0usize; + for end in ends { + out.push(DValue::List(flat[start..end].to_vec())); + start = end; + } + out + } + Shape::Sum(lanes) => { + // Inverse of transcode's Sum: untranscode each committed lane, then for each row pull its + // payload from its lane at the recorded within-lane OFFSET (robust to row reordering from + // a prior gather/merge — not a sequential cursor). + let (tags, offsets, variant_vals) = col.into_sum("untranscode"); + let lane_rows: Vec>> = variant_vals + .into_iter() + .zip(lanes.iter()) + .map(|(v, ls)| match (v, ls) { + (Some(cv), Some(lshape)) => Some(untranscode(cv, lshape)), + _ => None, + }) + .collect(); + tags.iter() + .zip(&offsets) + .map(|(&tag, &off)| { + let payload = lane_rows[tag].as_ref().expect("untranscode: committed lane")[off].clone(); + DValue::Variant(tag as u32, Box::new(payload)) + }) + .collect() + } + } +} + +/// Structural shape of a "place" Term (Var/Proj chain) given the env vars' shapes — used to resolve +/// `Spread`'s arity (how many fields to splice) and (later) Proj-on-list vs Proj-on-tuple. +pub fn shape_of_place(t: &Term, env_shapes: &[Shape]) -> Shape { + match t { + Term::Var(i) => env_shapes[*i].clone(), + Term::Proj(inner, i) => match shape_of_place(inner, env_shapes) { + Shape::Prod(fs) => fs[*i].clone(), + Shape::List(e) => *e, + other => panic!("shape_of_place: Proj on non-aggregate {other:?}"), + }, + other => panic!("shape_of_place: unsupported place {other:?}"), + } +} + +/// Best-effort structural `Shape` of a (non-place) compiled `Term`, used to decide cross-shape +/// `Eq`/`Ne`: DDIR compares `Value`s structurally, so e.g. `Tuple != Int` is *always* true (the +/// variants differ) — but corgi's `CmpOp::Rel` requires matching shapes. When the operand shapes +/// differ we fold the comparison to a constant instead of emitting a (panicking) `Rel`. +fn infer_term_shape(t: &Term, env_shapes: &[Shape]) -> Shape { + match t { + Term::Var(i) => env_shapes[*i].clone(), + Term::Int(_) => Shape::Prim(64), + Term::Proj(inner, i) => match infer_term_shape(inner, env_shapes) { + Shape::Prod(fs) => fs[*i].clone(), + Shape::List(e) => *e, + other => panic!("infer_term_shape: Proj on non-aggregate {other:?}"), + }, + Term::Tuple(fields) => { + let mut fs = Vec::new(); + for f in fields { + match f { + Term::Spread(inner) => match infer_term_shape(inner, env_shapes) { + Shape::Prod(inner_fs) => fs.extend(inner_fs), + Shape::Unit => {} + other => fs.push(other), + }, + _ => fs.push(infer_term_shape(f, env_shapes)), + } + } + if fs.is_empty() { Shape::Unit } else { Shape::Prod(fs) } + } + Term::If { then, .. } => infer_term_shape(then, env_shapes), + // Arithmetic, comparisons, and anything else scalar-ish reduce to a primitive column. + _ => Shape::Prim(64), + } +} + +/// Whether [`compile`] can lower this term to a corgi graph. Terms that use features the corgi +/// compiler doesn't model yet — `List`, `Inject`/`Case` (sum types), `Unary`, `Hash` — return false, +/// and the backend falls back to row-wise `ir::eval` (parity with `backend::vec`). +pub fn compilable(t: &Term) -> bool { + match t { + Term::Var(_) | Term::Bound(_) | Term::Int(_) => true, + Term::Proj(inner, _) | Term::Spread(inner) => compilable(inner), + Term::Tuple(fs) => fs.iter().all(compilable), + Term::Binary(_, l, r) => compilable(l) && compilable(r), + Term::If { cond, then, els } => compilable(cond) && compilable(then) && compilable(els), + Term::Fold { list, init, step } => compilable(list) && compilable(init) && compilable(step), + _ => false, // List, Inject, Case, Unary, Hash — row-wise fallback. + } +} + +/// Compile a `Term` to a corgi node. `env[i]` = node for `Var(i)`; `env_shapes[i]` = its shape +/// (for `Spread`). Binders push on top (read by `Bound(k)`). `anchor` sizes `Lit` broadcasts. +pub fn compile(term: &Term, b: &mut Builder, env: &[usize], env_shapes: &[Shape], anchor: usize) -> usize { + match term { + Term::Var(i) => env[*i], + Term::Bound(k) => env[env.len() - 1 - *k], + Term::Int(n) => b.add(Op::Lit(CValue::u64(vec![*n as u64])), vec![anchor]), + Term::Tuple(fields) => { + // A `Spread(place)` child splices the place's `Prod` fields in place (the flat-row model). + let mut ids: Vec = Vec::new(); + for f in fields { + match f { + Term::Spread(inner) => { + let node = compile(inner, b, env, env_shapes, anchor); + match shape_of_place(inner, env_shapes) { + Shape::Prod(fs) => { + for i in 0..fs.len() { + ids.push(b.add(Op::Field(i), vec![node])); + } + } + Shape::Unit => {} // unit splices nothing + _ => ids.push(node), // scalar: splice the value itself + } + } + _ => ids.push(compile(f, b, env, env_shapes, anchor)), + } + } + // An empty field list is DDIR unit: emit a length-carrying `Unit` column over the anchor, + // NOT `Prod([])` (an empty product has no rows to count, so the row count would be lost). + if ids.is_empty() { + b.add(Op::Unit, vec![anchor]) + } else { + b.tuple(ids) + } + } + Term::Proj(t, i) => { + let id = compile(t, b, env, env_shapes, anchor); + b.add(Op::Field(*i), vec![id]) + } + Term::Binary(op, l, r) => { + let lid = compile(l, b, env, env_shapes, anchor); + let rid = compile(r, b, env, env_shapes, anchor); + let pair = |b: &mut Builder, x, y| b.tuple(vec![x, y]); + match op { + BinOp::Add => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Add, Kind::U, 64), vec![p]) } + BinOp::Sub => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Sub, Kind::U, 64), vec![p]) } + BinOp::Mul => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Mul, Kind::U, 64), vec![p]) } + BinOp::Eq | BinOp::Ne => { + // Cross-shape structural compare folds to a constant (Eq→0, Ne→1) over `anchor`; + // same-shape emits a real corgi `Rel`. + if infer_term_shape(l, env_shapes) != infer_term_shape(r, env_shapes) { + let v = if matches!(op, BinOp::Ne) { 1u64 } else { 0u64 }; + b.add(Op::Lit(CValue::u64(vec![v])), vec![anchor]) + } else { + let pred = if matches!(op, BinOp::Eq) { Pred::Eq } else { Pred::Ne }; + let p = pair(b, lid, rid); + b.add(CmpOp::Rel(pred), vec![p]) + } + } + BinOp::Lt => { let p = pair(b, lid, rid); b.add(CmpOp::Rel(Pred::Lt), vec![p]) } + BinOp::Le => { let p = pair(b, lid, rid); b.add(CmpOp::Rel(Pred::Le), vec![p]) } + BinOp::Gt => { let p = pair(b, rid, lid); b.add(CmpOp::Rel(Pred::Lt), vec![p]) } + BinOp::Ge => { let p = pair(b, rid, lid); b.add(CmpOp::Rel(Pred::Le), vec![p]) } + BinOp::And => { let p = pair(b, lid, rid); b.add(CmpOp::Min, vec![p]) } + BinOp::Or => { let p = pair(b, lid, rid); b.add(CmpOp::Max, vec![p]) } + } + } + Term::If { cond, then, els } => { + let c = compile(cond, b, env, env_shapes, anchor); + let t = compile(then, b, env, env_shapes, anchor); + let e = compile(els, b, env, env_shapes, anchor); + let sel = b.tuple(vec![c, t, e]); + b.add(Op::Select, vec![sel]) + } + // Fold over a List. corgi `Op::Fold` consumes `Prod([seed, List])` and folds each row's + // list; its body is a closed sub-graph over `Prod([acc, elem])`. DDIR's step sees + // elem=Bound(0), acc=Bound(1), so the body env is [acc, elem] (Bound counts from the top). + // Restriction (this rung): the step references only its binders (monoid-style), not outer + // Vars — corgi closes the body; an outer reference would need CapList capture. + Term::Fold { list, init, step } => { + let init_id = compile(init, b, env, env_shapes, anchor); + let list_id = compile(list, b, env, env_shapes, anchor); + let pair = b.tuple(vec![init_id, list_id]); + let body = compile_fold_body(step); + b.add(Op::Fold(Box::new(body)), vec![pair]) + } + other => panic!("compile: unsupported Term in this rung: {other:?}"), + } +} + +/// Compile a `Fold` step into a closed corgi sub-graph over `Prod([acc, elem])`. +/// Env `[acc, elem]` so `Bound(0)`=elem (top), `Bound(1)`=acc — matching `ir::eval`'s Fold. +fn compile_fold_body(step: &Term) -> Graph { + let mut bb = Builder::::default(); + let inp = bb.input(); + let acc = bb.add(Op::Field(0), vec![inp]); + let elem = bb.add(Op::Field(1), vec![inp]); + // Monoid fold bodies use only binders (no Spread/Proj-on-list), so no env shapes are needed. + let out = compile(step, &mut bb, &[acc, elem], &[], inp); + bb.finish(out) +} + +/// Compile a single `Term` whose `Var(0)` is the whole input row/column. (Spread-free terms only — +/// the bench/chain/fold examples — so no env shape is needed.) +pub fn compile_term_single(term: &Term) -> Graph { + let mut b = Builder::::default(); + let input = b.input(); + let out = compile(term, &mut b, &[input], &[], input); + b.finish(out) +} + +/// Compile a `Filter` predicate over `Var(0)=key` (shape `kshape`), `Var(1)=val` (`vshape`) → mask. +pub fn compile_predicate(cond: &Term, kshape: &Shape, vshape: &Shape) -> Graph { + let mut b = Builder::::default(); + let input = b.input(); + let var_k = b.add(Op::Field(0), vec![input]); + let var_v = b.add(Op::Field(1), vec![input]); + let out = compile(cond, &mut b, &[var_k, var_v], &[kshape.clone(), vshape.clone()], input); + b.finish(out) +} + +/// Compile a join projection: key/val Terms over `Var(0)=key`, `Var(1)=val0`, `Var(2)=val1` (with +/// their shapes for `Spread`). Input `Prod([key, val0, val1])`; output `Prod([newkey, newval])`. +pub fn compile_join_projection(key: &Term, val: &Term, kshape: &Shape, v0shape: &Shape, v1shape: &Shape) -> Graph { + let mut b = Builder::::default(); + let input = b.input(); + let var_k = b.add(Op::Field(0), vec![input]); + let var_0 = b.add(Op::Field(1), vec![input]); + let var_1 = b.add(Op::Field(2), vec![input]); + let env = [var_k, var_0, var_1]; + let shapes = [kshape.clone(), v0shape.clone(), v1shape.clone()]; + let nk = compile(key, &mut b, &env, &shapes, input); + let nv = compile(val, &mut b, &env, &shapes, input); + let out = b.tuple(vec![nk, nv]); + b.finish(out) +} + +/// Compile a DDIR `Projection` over `Var(0)=key` (`kshape`), `Var(1)=val` (`vshape`). +/// Input `Prod([key, val])`; output `Prod([newkey, newval])`. +pub fn compile_projection(key: &Term, val: &Term, kshape: &Shape, vshape: &Shape) -> Graph { + let mut b = Builder::::default(); + let input = b.input(); + let var_k = b.add(Op::Field(0), vec![input]); + let var_v = b.add(Op::Field(1), vec![input]); + let env = [var_k, var_v]; + let shapes = [kshape.clone(), vshape.clone()]; + let nk = compile(key, &mut b, &env, &shapes, input); + let nv = compile(val, &mut b, &env, &shapes, input); + let out = b.tuple(vec![nk, nv]); + b.finish(out) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::ir::Value as V; + + /// Round-trip a column of rows through infer_shape_cols → transcode → untranscode. + fn roundtrip(rows: Vec) { + let shape = infer_shape_cols(&rows); + let col = transcode(&rows, &shape); + let back = untranscode(col, &shape); + assert_eq!(back, rows, "roundtrip mismatch (shape {shape:?})"); + } + + #[test] + fn roundtrip_variant_single_arm() { + // binders-style: a single constructor wrapping a list. + roundtrip(vec![ + V::Variant(0, Box::new(V::List(vec![V::Int(1), V::Int(2)]))), + V::Variant(0, Box::new(V::List(vec![V::Int(3)]))), + V::Variant(0, Box::new(V::List(vec![]))), + ]); + } + + #[test] + fn roundtrip_variant_multi_arm() { + // adt-style: two arms, interleaved; payloads of different shape per arm. + roundtrip(vec![ + V::Variant(0, Box::new(V::Int(10))), + V::Variant(1, Box::new(V::Tuple(vec![V::Int(1), V::Int(2)]))), + V::Variant(0, Box::new(V::Int(20))), + V::Variant(1, Box::new(V::Tuple(vec![V::Int(3), V::Int(4)]))), + V::Variant(0, Box::new(V::Int(30))), + ]); + } + + #[test] + fn roundtrip_variant_absent_arm_is_bottom() { + // tags {0, 2} present, arm 1 absent → a ⊥ lane; round-trip must still reconstruct. + roundtrip(vec![ + V::Variant(0, Box::new(V::Int(1))), + V::Variant(2, Box::new(V::Int(2))), + V::Variant(0, Box::new(V::Int(3))), + ]); + } + + #[test] + fn roundtrip_nested_variant_in_tuple() { + roundtrip(vec![ + V::Tuple(vec![V::Int(1), V::Variant(0, Box::new(V::Int(7)))]), + V::Tuple(vec![V::Int(2), V::Variant(1, Box::new(V::unit()))]), + ]); + } +} diff --git a/interactive/src/corgi/mod.rs b/interactive/src/corgi/mod.rs new file mode 100644 index 000000000..56fc484f5 --- /dev/null +++ b/interactive/src/corgi/mod.rs @@ -0,0 +1,12 @@ +//! The corgi rendering substrate's internals: the columnar container on dataflow +//! edges ([`container`]), the columnar chunk and its arrangement plumbing ([`chunk`]), +//! the DDIR-term compiler ([`logic`]), and the join/reduce tactic bindings ([`join`], +//! [`reduce`]). The `Backend` impl wiring these into rendering is +//! [`crate::backend::corgi`]. + +pub mod chunk; +pub mod col_times; +pub mod container; +pub mod join; +pub mod logic; +pub mod reduce; diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs new file mode 100644 index 000000000..1f89d4cce --- /dev/null +++ b/interactive/src/corgi/reduce.rs @@ -0,0 +1,503 @@ +//! The corgi `ProxyReduceBackend`: the value semantics for the DD `ProxyReduceTactic`. +//! +//! The tactic (differential's `operators::int_proxy::reduce`) owns ALL time/lattice logic over +//! integer proxies `(key_hash, value_id, time, diff)`; this backend supplies only: +//! +//! * ids — `key_hash`/`value_id` are value-as-id for primitive columns (the value IS the id) and +//! the canonical native `corgi::hash` for compound columns (columnar, content-addressed, so ids +//! coincide across the output→input boundary); DD never hashes. +//! * the value callback — `reduce_many` runs ONE crossing per retire over every `(key, time)` +//! bracket, building the output value COLUMNS directly (Count → a `u64` prim, Distinct → a +//! `Unit`, Min → the chosen input rows, Collect → a `List`), never through DDIR rows. +//! * materialize — resolve proxy ids back to real columns by `gather` from per-retire pools and +//! seal a `CorgiChunk` batch column-natively. +//! +//! Transcode-free: the real keys/values never leave corgi columns. Ids are resolved to rows by +//! integer index (`key_index`/`val_index` → offsets into the concatenated `key_blocks`/`val_blocks` +//! pools), not by carrying `DValue`s. Min/Collect's ordering is corgi's own — one `sort_blocks` per +//! retire orders every bracket's candidates (Min = each block's first, Collect = each block's sorted +//! run, expanded by diff). This uses corgi's STRUCTURAL order, which equals DDIR `Ord` for the +//! non-negative scalar/tuple values these reductions see (all 6 canonical programs); it diverges only +//! for negative ints (corgi's leaf compare is unsigned) and list-valued compares (corgi lists order +//! length-first) — neither arises here. A signed/​list-general order would need a corgi order fix +//! (offset-binary leaf or lex-first lists), not a change here. +//! +//! The changed-key restriction is honored by presenting only the changed keys: novel batches are +//! read whole (delta-sized), the accumulated history is scanned and filtered to the changed hashes +//! (a columnar semijoin — matching the row-wise tactic's read). + +use std::collections::HashMap; +use std::hash::{BuildHasherDefault, Hasher}; +use std::rc::Rc; + + +use differential_dataflow::consolidation::consolidate_updates; +use differential_dataflow::trace::Description; +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::{Bounds, Shape, Value as CValue}; + +use crate::corgi::col_times::ColTime; +use crate::corgi::chunk::{columns_to_batch, CorgiChunk}; +use crate::ir::Diff; +use crate::parse::Reducer; + +type CBatch = Rc>>; + +/// An identity `Hasher` for the id-index maps: their keys are already well-distributed 64-bit +/// content hashes (`hash_rows`), so passing the id straight through avoids re-hashing it (siphash +/// on `register_keys`/lookups was ~7% of the reduce in profiling). Only `write_u64` is used. +#[derive(Default)] +struct IdHasher(u64); +impl Hasher for IdHasher { + #[inline] + fn write_u64(&mut self, i: u64) { self.0 = i; } + #[inline] + fn write(&mut self, _: &[u8]) { unreachable!("IdMap keys are u64") } + #[inline] + fn finish(&self) -> u64 { self.0 } +} +/// `key_hash`/`value_id` → row index, hashed by identity. +type IdMap = HashMap>; + +/// A corgi reduce backend for a single `Reducer`. All per-retire scratch is corgi columns + integer +/// id→row-index maps; nothing carries a `DValue`. +pub struct CorgiReduceBackend { + reducer: Reducer, + /// Input value column for the current window, indexed by `in_index` (for Min/Collect resolution). + in_vals: CValue, + /// Input `value_id → row` in `in_vals` for the current window (reduce-time resolution; first row + /// wins, so equal values — which share a content-hash `value_id` — resolve to one representative). + in_index: IdMap, + /// Output tiling for `begin`/`emit`/`finish`: the tile descriptions, and per-tile accumulated + /// output rows `(key row, value row, time, diff)` (pool indices, gathered into columns at `finish`). + tiles: Vec>, + tile_rows: Vec<(Vec, Vec, Vec, Vec)>, + /// Key-resolution pool for the current retire: `key_hash → row index` into the concatenation of + /// `key_blocks` (representative keys from the input + output presentations). + key_index: IdMap, + key_blocks: Vec, + key_len: usize, + /// Value-resolution pool for the current retire: `value_id → row index` into the concatenation of + /// `val_blocks` (output-history values + values minted by `reduce_many`). + val_index: IdMap, + val_blocks: Vec, + val_len: usize, + _t: std::marker::PhantomData, +} + +impl CorgiReduceBackend { + pub fn new(reducer: Reducer) -> Self { + CorgiReduceBackend { + reducer, + in_vals: CValue::Unit(0), + in_index: IdMap::default(), + tiles: Vec::new(), + tile_rows: Vec::new(), + key_index: IdMap::default(), + key_blocks: Vec::new(), + key_len: 0, + val_index: IdMap::default(), + val_blocks: Vec::new(), + val_len: 0, + _t: std::marker::PhantomData, + } + } + + /// Clear the resolution pools at the start of a retire (called from `next_window`'s first call). + fn reset_pools(&mut self) { + self.key_index.clear(); + self.key_blocks.clear(); + self.key_len = 0; + self.val_index.clear(); + self.val_blocks.clear(); + self.val_len = 0; + } + + /// Add representative key rows (aligned with `ids`) to the key pool; first id wins. + fn register_keys(&mut self, col: CValue, ids: &[u64]) { + for (i, &id) in ids.iter().enumerate() { + self.key_index.entry(id).or_insert(self.key_len + i); + } + self.key_len += col.len(); + self.key_blocks.push(col); + } + + /// Add value rows (aligned with `ids`) to the val pool; first id wins. + fn register_vals(&mut self, col: CValue, ids: &[u64]) { + for (i, &id) in ids.iter().enumerate() { + self.val_index.entry(id).or_insert(self.val_len + i); + } + self.val_len += col.len(); + self.val_blocks.push(col); + } +} + +/// Concatenate corgi columns (skipping empties, which contribute no rows and so don't shift the +/// pool offsets accounted at registration). One `gather_lanes` over the non-empty blocks. +fn concat_columns(blocks: &[CValue]) -> CValue { + let non_empty: Vec<&CValue> = blocks.iter().filter(|b| b.len() > 0).collect(); + match non_empty.len() { + 0 => CValue::Unit(0), + 1 => non_empty[0].clone(), + _ => { + let srcs: Vec> = non_empty.iter().map(|b| Some(*b)).collect(); + let (mut tags, mut offs) = (Vec::new(), Vec::new()); + for (ti, b) in non_empty.iter().enumerate() { + for o in 0..b.len() { + tags.push(ti); + offs.push(o); + } + } + gather_lanes(&srcs, &tags, &offs) + } + } +} + +/// Id column for a key/value column. For a PRIMITIVE column — a bare 64-bit `Prim`, or a 1-field +/// `Prod([Prim(64)])` — the value itself is already a collision-free id (`i64 as u64` is a bijection), +/// so pass it straight through and skip the content hash. Compound shapes (Unit / List / Sum / +/// multi-field `Prod`) hash via the CANONICAL native `corgi::hash` (the designed boundary-id fold, +/// width-blind and consistent-with-equality) — not the branch-local `arrange::hash_rows`; DDIR +/// transcodes every leaf to `u64`, so width-blindness is a no-op for us and there is no cross-path +/// hash comparison (value-as-id and native hash are never used for the same value: shape is uniform +/// per column). The id is used ONLY as an identity for netting/dedup — its numeric order is never +/// 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"), + } +} + +/// 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. +/// +/// NB this is a full scan of the presented chunks (incl. `source_batches`, the accumulated trace). +/// A `find_ranges` seek of the changed keys was tried (delta-proportional in principle) but REGRESSED +/// SCC (1.62x→2.16x): SCC's changed set is broad (label propagation touches most keys each retire), so +/// the scan already touches ~every row while the per-chunk gallop only adds overhead. The O(history) +/// re-presentation is inherent to SCC here, not a seekable-few-keys case. +fn collect_present(chunks: &[&CorgiChunk], changed: &[u64]) -> (CValue, CValue, Vec, Vec, Vec) +where + T: ColTime, +{ + 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]); + } + } + } + if tags.is_empty() { + return (CValue::Unit(0), CValue::Unit(0), khs, times, diffs); + } + let keys_col = gather_lanes(&key_srcs, &tags, &offs); + let vals_col = gather_lanes(&val_srcs, &tags, &offs); + (keys_col, vals_col, khs, times, diffs) +} + +/// All chunks of a batch list, flattened (empty chunks included — `hash_rows` yields nothing for them). +fn chunks_of(batches: &[CBatch]) -> Vec<&CorgiChunk> +where + T: ColTime, +{ + batches.iter().flat_map(|b| b.chunks.iter()).collect() +} + +impl CorgiReduceBackend +where + T: ColTime + Ord, +{ + /// The one value crossing for a retire: every `(key, time)` bracket at once. Builds the output + /// value COLUMN directly per reducer, registers it (id → row) into the val pool, and returns the + /// proxy `(value_id, diff)` deltas with per-bracket ends. `input[k] = (rep index into the input + /// presentation, accumulated diff)`; the bracket `i` is `input[ends[i-1]..ends[i]]`, non-empty. + fn reduce_brackets(&mut self, ends: &[usize], input: &[(usize, Diff)]) -> (Vec<(u64, Diff)>, Vec) { + let mut out_diffs: Vec = Vec::new(); + let mut out_ends: Vec = Vec::with_capacity(ends.len()); + let out_ids: Vec; + + match self.reducer { + Reducer::Count => { + // Per-bracket sum of diffs; survivors become a `Tuple([Int(sum)])` = corgi `Prod([u64])`. + let mut sums: Vec = Vec::new(); + let mut start = 0; + for &end in ends { + let c: Diff = input[start..end].iter().map(|&(_, d)| d).sum(); + if c > 0 { + sums.push(c as u64); + out_diffs.push(1); + } + out_ends.push(out_diffs.len()); + start = end; + } + if sums.is_empty() { + return (Vec::new(), out_ends); + } + let col = CValue::Prod(vec![CValue::u64(sums)]); + out_ids = ids(&col); + self.register_vals(col, &out_ids); + } + Reducer::Distinct => { + // Present iff any value has positive net; output value is unit (a `Unit` column). + let mut present = 0usize; + let mut start = 0; + for &end in ends { + if input[start..end].iter().any(|&(_, d)| d > 0) { + present += 1; + out_diffs.push(1); + } + out_ends.push(out_diffs.len()); + start = end; + } + if present == 0 { + return (Vec::new(), out_ends); + } + let col = CValue::Unit(present); + out_ids = ids(&col); // all equal (unit content hash) + self.register_vals(col, &out_ids); + } + Reducer::Min => { + // The DDIR `min` over the positive-diff values, in corgi's structural order (== DDIR + // `Ord` for the non-negative scalar/tuple values these reductions see; see module doc). + // Gather all positive-diff candidates across brackets into one column, segment by + // bracket, and one corgi `sort_blocks` gives every bracket's argmin at once + // (`perm[block_start]`). The winning ROW is taken columnar and reuses its input value id. + let mut cand_reps: Vec = Vec::new(); // input presentation rep index per candidate + let mut labels: Vec = Vec::new(); // dense segment id per candidate + let mut block_starts: Vec = Vec::new(); // per emitted bracket: start offset in cand_reps + let mut start = 0; + for &end in ends { + let lo = cand_reps.len(); + let seg = block_starts.len() as u64; + for k in start..end { + if input[k].1 > 0 { + cand_reps.push(input[k].0); + labels.push(seg); + } + } + if cand_reps.len() > lo { + block_starts.push(lo); + out_diffs.push(1); + } + out_ends.push(out_diffs.len()); + start = end; + } + if cand_reps.is_empty() { + return (Vec::new(), out_ends); + } + let cand_col = gather(&self.in_vals, &cand_reps); + let (perm, _) = sort_blocks(&labels, &cand_col); + let min_reps: Vec = block_starts.iter().map(|&lo| cand_reps[perm[lo]]).collect(); + let col = gather(&self.in_vals, &min_reps); + out_ids = ids(&col); + self.register_vals(col, &out_ids); + } + Reducer::Collect => { + // One row per bracket: the values sorted in corgi structural order (== DDIR `Ord` here), + // each repeated by its diff, as a `List`. One `sort_blocks` orders every bracket's + // entries at once; element rows are then taken columnar. Every bracket emits (empty + // list if all diffs ≤ 0), matching the row reducer. + let mut entry_reps: Vec = Vec::new(); + let mut entry_diffs: Vec = Vec::new(); + let mut labels: Vec = Vec::new(); + let mut blocks: Vec<(usize, usize)> = Vec::with_capacity(ends.len()); + let mut start = 0; + for (bi, &end) in ends.iter().enumerate() { + let lo = entry_reps.len(); + for k in start..end { + entry_reps.push(input[k].0); + entry_diffs.push(input[k].1); + labels.push(bi as u64); + } + blocks.push((lo, entry_reps.len())); + out_diffs.push(1); + out_ends.push(out_diffs.len()); + start = end; + } + let perm = if entry_reps.is_empty() { + Vec::new() + } else { + sort_blocks(&labels, &gather(&self.in_vals, &entry_reps)).0 + }; + // Expand each bracket's sorted entries by their diff (max(0, ·) copies). + let mut elem_reps: Vec = Vec::new(); + let mut bracket_ends: Vec = Vec::with_capacity(ends.len()); + for (lo, hi) in blocks { + for &e in &perm[lo..hi] { + for _ in 0..entry_diffs[e].max(0) { + elem_reps.push(entry_reps[e]); + } + } + bracket_ends.push(elem_reps.len()); + } + let elems = if elem_reps.is_empty() { CValue::Unit(0) } else { gather(&self.in_vals, &elem_reps) }; + let col = CValue::List(Bounds::Offsets(bracket_ends), Box::new(elems)); + out_ids = ids(&col); + self.register_vals(col, &out_ids); + } + } + + let outs = out_ids.into_iter().zip(out_diffs).collect(); + (outs, out_ends) + } +} + +impl ProxyReduceBackend, CBatch> for CorgiReduceBackend +where + T: ColTime + Ord, +{ + type RIn = Diff; + type ROut = Diff; + + fn seed_times(&self, instance: &ReduceInstance<'_, CBatch, CBatch>) -> Vec<(u64, T)> { + // The batch's raw (key_hash, time) support — hash the novel KEY columns only, one entry per + // record, sorted by key_hash. Seeds may over-derive (a non-changing seed yields a zero delta), + // so this superset of b.support suffices; `instance.lower` is not applied (see ReduceInstance). + let mut out: Vec<(u64, T)> = Vec::new(); + for ch in chunks_of(instance.input_batches) { + let kh = ids(ch.keys()); + let times = ch.times(); + for (i, h) in kh.into_iter().enumerate() { + out.push((h, times.get(i))); + } + } + out.sort_by_key(|(k, _)| *k); + out + } + + fn begin(&mut self, tiles: &[Description]) { + // Open a tiled output session for this retire; reset the per-retire resolution pools. + self.reset_pools(); + self.tiles = tiles.to_vec(); + self.tile_rows = (0..tiles.len()).map(|_| (Vec::new(), Vec::new(), Vec::new(), Vec::new())).collect(); + } + + fn next_window(&mut self, instance: &ReduceInstance<'_, CBatch, CBatch>, changed: &[u64], cursor: &mut usize) -> Option> { + // Single window: present ALL remaining changed keys at once (bounded-memory windowing is a + // later refinement). `changed` is ascending, so `binary_search` is the changed-key filter. + if *cursor >= changed.len() { + return None; + } + let keys: Vec = changed[*cursor..].to_vec(); + *cursor = changed.len(); + let present = |chunks: &[&CorgiChunk]| collect_present(chunks, &keys); + + // Input presentation: accumulated history ∪ novel delta, restricted to the window's keys. + // value_id = content hash of the value (equal values share an id → the tactic nets them); + // `in_index` resolves an id back to a representative in_vals row for `reduce_corrections`. + let mut in_chunks = chunks_of(instance.input_batches); + in_chunks.extend(chunks_of(instance.source_batches)); + let (in_keys, in_vals, in_khs, in_times, in_diffs) = present(&in_chunks); + self.in_index = IdMap::default(); + let input: ProxyBridge = if in_khs.is_empty() { + self.in_vals = CValue::Unit(0); + Vec::new() + } else { + let vids = ids(&in_vals); + for (r, &vid) in vids.iter().enumerate() { self.in_index.entry(vid).or_insert(r); } + self.in_vals = in_vals; + self.register_keys(in_keys, &in_khs); + let mut b: ProxyBridge = + (0..in_khs.len()).map(|i| ((in_khs[i], vids[i]), in_times[i].clone(), in_diffs[i])).collect(); + consolidate_updates(&mut b); + b + }; + + // Output-history presentation, same keys (register keys + values for correction resolution). + let (o_keys, o_vals, o_khs, o_times, o_diffs) = present(&chunks_of(instance.output_batches)); + let output: ProxyBridge = if o_khs.is_empty() { + Vec::new() + } else { + let vids = ids(&o_vals); + self.register_keys(o_keys, &o_khs); + self.register_vals(o_vals, &vids); + let mut b: ProxyBridge = + (0..o_khs.len()).map(|i| ((o_khs[i], vids[i]), o_times[i].clone(), o_diffs[i])).collect(); + consolidate_updates(&mut b); + b + }; + + Some(ReduceWindow { keys, input, output }) + } + + fn reduce_corrections(&mut self, keys: &[u64], in_ends: &[usize], input: &[(u64, Diff)], out_ends: &[usize], output: &[(u64, Diff)]) -> (Vec<(u64, Diff)>, Vec) { + // Resolve input value_ids to `in_vals` rows, reduce (desired output), then difference the + // desired against the presented current output per key: correction = desired − current. + let in_rows: Vec<(usize, Diff)> = input.iter() + .map(|&(vid, d)| (*self.in_index.get(&vid).expect("input value_id presented this window"), d)) + .collect(); + let (desired, desired_ends) = self.reduce_brackets(in_ends, &in_rows); + + let mut corr: Vec<(u64, Diff)> = Vec::new(); + let mut corr_ends: Vec = Vec::with_capacity(keys.len()); + let (mut ds, mut os) = (0usize, 0usize); + for i in 0..keys.len() { + let (de, oe) = (desired_ends[i], out_ends[i]); + // Net by value_id: desired (+) minus current output (−); keep non-zero, in first-seen order. + let mut net: HashMap> = Default::default(); + let mut order: Vec = Vec::new(); + for &(vid, d) in &desired[ds..de] { + if let Some(x) = net.get_mut(&vid) { *x += d; } else { net.insert(vid, d); order.push(vid); } + } + for &(vid, d) in &output[os..oe] { + if let Some(x) = net.get_mut(&vid) { *x -= d; } else { net.insert(vid, -d); order.push(vid); } + } + for vid in order { + let d = net[&vid]; + if d != 0 { corr.push((vid, d)); } + } + corr_ends.push(corr.len()); + ds = de; + os = oe; + } + (corr, corr_ends) + } + + fn emit(&mut self, tile: usize, records: &[((u64, u64), T, Diff)]) { + // Resolve each correction's key/value proxies to pool rows and accumulate into the tile. + for rec in records { + let ((kh, vid), t, d) = (rec.0, &rec.1, rec.2); + let kr = *self.key_index.get(&kh).expect("key resolvable this retire"); + let vr = *self.val_index.get(&vid).expect("value resolvable this retire"); + let (krows, vrows, times, diffs) = &mut self.tile_rows[tile]; + krows.push(kr); + vrows.push(vr); + times.push(t.clone()); + diffs.push(d); + } + } + + fn finish(&mut self) -> Vec> { + // Seal each tile: gather its accumulated (key, val) pool rows into columns, one CorgiChunk batch. + let key_pool = concat_columns(&self.key_blocks); + let val_pool = concat_columns(&self.val_blocks); + let tiles = std::mem::take(&mut self.tiles); + let tile_rows = std::mem::take(&mut self.tile_rows); + tiles.into_iter().zip(tile_rows).map(|(desc, (krows, vrows, times, diffs))| { + let keys = gather(&key_pool, &krows); + let vals = gather(&val_pool, &vrows); + Rc::new(columns_to_batch(keys, vals, times, diffs, desc)) + }).collect() + } +} diff --git a/interactive/src/lib.rs b/interactive/src/lib.rs index 1146d8c4a..ba3b3a95b 100644 --- a/interactive/src/lib.rs +++ b/interactive/src/lib.rs @@ -1,5 +1,6 @@ pub mod parse; pub mod ir; +pub mod corgi; pub mod lower; pub mod scope_ir; pub mod backend; diff --git a/interactive/tests/corgi_backend.rs b/interactive/tests/corgi_backend.rs new file mode 100644 index 000000000..d3faad679 --- /dev/null +++ b/interactive/tests/corgi_backend.rs @@ -0,0 +1,49 @@ +//! The corgi backend's correctness gate: each canonical `.ddp` program must evaluate +//! identically through the corgi backend and the reference vec backend. + +use interactive::backend::{corgi, vec}; +use interactive::ir::Value; +use interactive::{lower, parse}; + +fn tup(fields: &[i64]) -> Value { + Value::Tuple(fields.iter().map(|&n| Value::Int(n)).collect()) +} +fn rows(rs: &[&[i64]]) -> Vec<(Value, Value)> { + rs.iter().map(|f| (tup(f), Value::unit())).collect() +} + +/// Per-program inputs (arity matches each `.ddp`'s `input N` usage). +fn inputs_for(prog: &str) -> Vec> { + let edges = rows(&[&[1, 2], &[2, 3], &[3, 4], &[5, 6], &[4, 2]]); + match prog { + "reach" => vec![edges, rows(&[&[1]])], + "scc" => vec![edges], + // stable: edges (l_node, l_pref, r_node, r_pref) + "stable" => vec![rows(&[&[1, 1, 10, 1], &[1, 2, 11, 1], &[2, 1, 10, 2], &[2, 2, 11, 2]])], + "unnest" => vec![rows(&[&[1, 2], &[3, 4]])], + "adt" => vec![edges], + "binders" => vec![rows(&[&[1, 2], &[3, 4]])], + other => panic!("no inputs configured for {other}"), + } +} + +/// Evaluate `prog` through both backends and assert the outputs match. +fn assert_backends_agree(prog: &str) { + let path = format!("{}/examples/programs/{prog}.ddp", env!("CARGO_MANIFEST_DIR")); + let src = interactive::load_program(&path); + let mut tree = lower::lower_tree(parse::pipe::parse(&src)); + tree.optimize(); + let inputs = inputs_for(prog); + assert_eq!( + corgi::evaluate(&tree, &inputs), + vec::evaluate(&tree, &inputs), + "corgi backend disagrees with the vec backend on {prog}", + ); +} + +#[test] fn reach() { assert_backends_agree("reach"); } +#[test] fn scc() { assert_backends_agree("scc"); } +#[test] fn stable() { assert_backends_agree("stable"); } +#[test] fn unnest() { assert_backends_agree("unnest"); } +#[test] fn adt() { assert_backends_agree("adt"); } +#[test] fn binders() { assert_backends_agree("binders"); }