From 9bce5f3e8e90b65384272df417ef94fdbc649204 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 30 Jul 2026 20:17:17 -0400 Subject: [PATCH 1/4] corgi reduce: seek-vs-scan presentation, decided per retire MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit collect_present was the one non-delta-proportional path left: an O(trace) rescan (plus per-chunk key re-hashing) every retire, ~50% of an incremental round's profile. It now decides per retire, when the sizes are known: a narrow changed set over seekable keys (single-leaf: ids ARE key values, so the ascending changed set converts to a needle column) gallops each chunk once per changed key with find_ranges — no hashing at all; broad churn keeps the scan, whose flat membership test beats marginal seeking (a find_ranges probe is a structurally-dispatched binary search; SEEK_ADVANTAGE=16 measured as a load regression before widening to 64). Hashed ids of structural keys cannot be inverted into needles and always scan. Steady-state (1000 rounds x batch 100, 100k/200k): reach 4.51s -> 2.10s — past the vec backend's 2.51s (1.8x -> 0.84x); scc 794 -> 597ms/round, the residual being retires whose label cascades genuinely broaden the changed set — the windowed-presentation seam's territory, not a threshold's. Load-shaped scc unchanged (3.60s). Co-Authored-By: Claude Fable 5 --- interactive/src/corgi/reduce.rs | 85 +++++++++++++++++++++++++-------- 1 file changed, 66 insertions(+), 19 deletions(-) diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 0017d7c29..d7a263202 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -37,7 +37,7 @@ use differential_dataflow::trace::chunk::ChunkBatch; use differential_dataflow::operators::int_proxy::ProxyBridge; use differential_dataflow::operators::int_proxy::reduce::{ProxyReduceBackend, ReduceInstance, ReduceWindow}; -use corgi::arrange::{gather, gather_lanes, sort_blocks}; +use corgi::arrange::{find_ranges, gather, gather_lanes, sort_blocks}; use corgi::{Bounds, Shape, Value as CValue}; use crate::corgi::col_times::ColTime; @@ -168,6 +168,19 @@ fn concat_columns(blocks: &[CValue]) -> CValue { /// relied upon — so the raw two's-complement `u64` is correct even for negative ints (no swizzle). /// Applied CONSISTENTLY at every id site (both value presentations AND the freshly-produced /// `reduce_brackets` outputs), else `desired − current` nets across mismatched ids for the same value. +/// The `changed` set as a needle column in the chunks' own key shape — possible exactly +/// when `ids` uses key VALUES (a bare `u64` leaf, or a 1-tuple of one); the hashed ids of +/// structural keys cannot be inverted into needles. +fn seek_needles(sample: &CValue, changed: &[u64]) -> Option { + match corgi::shape_of_value(sample) { + Shape::Prim(64) => Some(CValue::u64(changed.to_vec())), + Shape::Prod(ref fs) if fs.len() == 1 && matches!(fs[0], Shape::Prim(64)) => { + Some(CValue::Prod(vec![CValue::u64(changed.to_vec())])) + } + _ => None, + } +} + fn ids(col: &CValue) -> Vec { match corgi::shape_of_value(col) { Shape::Prim(64) => col.clone().into_u64("ids"), @@ -183,33 +196,67 @@ fn ids(col: &CValue) -> Vec { /// `(keys_col, vals_col)` corgi columns plus per-record `(key_hash, time, diff)`. `changed` is the /// ASCENDING set of changed key hashes; a row is kept iff its key hash is in it. /// -/// NB this is a full scan of the presented chunks (incl. `source_batches`, the accumulated trace), -/// and deliberately NOT a `find_ranges` seek of the changed keys: under label-propagation-shaped -/// workloads the changed set is broad (most keys change each retire), so a scan touches ~every row -/// regardless and the per-chunk gallop only adds overhead. The O(history) re-presentation is -/// inherent to broad change sets, not a seekable-few-keys case. +/// Seek-vs-scan, decided per retire, now that the sizes are known: seeking the changed keys +/// (`find_ranges`, O(|changed|·log rows) per chunk, no key hashing at all) wins when the +/// changed set is narrow — the steady incremental case; the full scan (O(rows) per chunk, +/// plus each chunk's key hashes re-derived) wins for broad churn — loads and label-cascade +/// retires, where most keys change and a gallop per key only adds overhead. Seeking requires +/// ids that ARE key values (single-leaf keys, `ids`' fast paths): hashed ids of structural +/// keys cannot be inverted into needles, so those always scan. /// -/// TODO: the scan's per-row work can still batch: `ids` re-derives (and copies) each chunk's key -/// hashes every retire (memoize per chunk, or a stored hash column), the membership test is a -/// per-row `binary_search`, and each hit materializes an owned time (`times().get`); kept RANGES -/// could move via `push_range`. +/// TODO: the scan's per-row work can still batch: `ids` re-derives (and copies) each chunk's +/// key hashes every retire (memoize per chunk, or a stored hash column), and each hit +/// materializes an owned time (`times().get`); kept RANGES could move via `push_range`. fn collect_present(chunks: &[&CorgiChunk], changed: &[u64]) -> (CValue, CValue, Vec, Vec, Vec) where T: ColTime, { + /// Seek only when the changed set is at least this many times narrower than the + /// presented rows: a `find_ranges` probe is a structurally-dispatched binary search + /// (~log(rows) compares, each far costlier than the scan's flat membership test), so + /// marginal seeks LOSE to the scan — measured, not modeled; 16 regressed load-shaped + /// retires before this was widened. + const SEEK_ADVANTAGE: usize = 64; + let key_srcs: Vec> = chunks.iter().map(|c| Some(c.keys())).collect(); let val_srcs: Vec> = chunks.iter().map(|c| Some(c.vals())).collect(); let (mut tags, mut offs) = (Vec::new(), Vec::new()); let (mut khs, mut times, mut diffs) = (Vec::new(), Vec::new(), Vec::new()); - for (ci, ch) in chunks.iter().enumerate() { - let kh = ids(ch.keys()); - for i in 0..kh.len() { - if changed.binary_search(&kh[i]).is_ok() { - tags.push(ci); - offs.push(i); - khs.push(kh[i]); - times.push(ch.times().get(i)); - diffs.push(ch.diffs()[i]); + let total: usize = chunks.iter().map(|c| c.diffs().len()).sum(); + let needles = if changed.len().saturating_mul(SEEK_ADVANTAGE) < total { + chunks.iter().find(|c| c.diffs().len() > 0).and_then(|c| seek_needles(c.keys(), changed)) + } else { + None + }; + if let Some(needles) = needles { + // Narrow changed set over seekable keys: gallop each chunk once per changed key. + // Chunks are key-ordered and `changed` ascends, so emission order matches the scan's. + for (ci, ch) in chunks.iter().enumerate() { + if ch.diffs().is_empty() { + continue; + } + let (lo, hi) = find_ranges(&needles, ch.keys()); + for (j, (&l, &h)) in lo.iter().zip(hi.iter()).enumerate() { + for i in l..h { + tags.push(ci); + offs.push(i); + khs.push(changed[j]); + times.push(ch.times().get(i)); + diffs.push(ch.diffs()[i]); + } + } + } + } else { + for (ci, ch) in chunks.iter().enumerate() { + let kh = ids(ch.keys()); + for i in 0..kh.len() { + if changed.binary_search(&kh[i]).is_ok() { + tags.push(ci); + offs.push(i); + khs.push(kh[i]); + times.push(ch.times().get(i)); + diffs.push(ch.diffs()[i]); + } } } } From 190cad7a372ad2e191f9fce40d08586c5a610232 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 31 Jul 2026 10:56:38 -0400 Subject: [PATCH 2/4] Reattach ids' doc comment (seek_needles had been inserted mid-block) Co-Authored-By: Claude Fable 5 --- interactive/examples/ddir.rs | 10 +++++++++- interactive/src/corgi/reduce.rs | 22 +++++++++++----------- 2 files changed, 20 insertions(+), 12 deletions(-) diff --git a/interactive/examples/ddir.rs b/interactive/examples/ddir.rs index a706bd804..576c176be 100644 --- a/interactive/examples/ddir.rs +++ b/interactive/examples/ddir.rs @@ -14,6 +14,9 @@ //! - `--diag`: serve timely/DD diagnostics on port 51371. //! - `--backend=vec|corgi`: rendering substrate (default `vec`). The corgi //! backend is single-worker (its arrange does not exchange). +//! - `--sync=K`: await completion only every K rounds (default 1), letting K +//! timestamps retire with whatever inter-timestamp concurrency the system +//! finds — the open(er)-loop regime DD adapts into under load. use mimalloc::MiMalloc; @@ -37,6 +40,7 @@ struct Flags { debug_demand: bool, diag: bool, corgi: bool, + sync: u64, } fn run( @@ -184,7 +188,10 @@ fn run( cursor += 1; } for i in inputs.iter_mut() { i.advance_to(time); i.flush(); } - while probe.less_than(&time) { worker.step(); } + let sync = flags.sync.max(1); + if (round + 1) % sync == 0 || round + 1 == limit { + while probe.less_than(&time) { worker.step(); } + } round += 1; if round % 100 == 0 { @@ -209,6 +216,7 @@ fn main() { else if let Some(q) = a.strip_prefix("--query=") { flags.query = Some(q.to_string()); } else if a == "--debug-demand" { flags.debug_demand = true; } else if a == "--diag" { flags.diag = true; } + else if let Some(k) = a.strip_prefix("--sync=") { flags.sync = k.parse().expect("--sync=K"); } else if let Some(b) = a.strip_prefix("--backend=") { flags.corgi = match b { "corgi" => true, "vec" => false, other => panic!("unknown backend {other:?} (vec|corgi)") }; } diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index d7a263202..1435d0a59 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -168,6 +168,17 @@ fn concat_columns(blocks: &[CValue]) -> CValue { /// relied upon — so the raw two's-complement `u64` is correct even for negative ints (no swizzle). /// Applied CONSISTENTLY at every id site (both value presentations AND the freshly-produced /// `reduce_brackets` outputs), else `desired − current` nets across mismatched ids for the same value. +fn ids(col: &CValue) -> Vec { + match corgi::shape_of_value(col) { + Shape::Prim(64) => col.clone().into_u64("ids"), + Shape::Prod(ref fs) if fs.len() == 1 && matches!(fs[0], Shape::Prim(64)) => match col { + CValue::Prod(fields) => fields[0].clone().into_u64("ids"), + _ => unreachable!("shape Prod but value not Prod"), + }, + _ => corgi::hash(col).into_u64("ids"), + } +} + /// The `changed` set as a needle column in the chunks' own key shape — possible exactly /// when `ids` uses key VALUES (a bare `u64` leaf, or a 1-tuple of one); the hashed ids of /// structural keys cannot be inverted into needles. @@ -181,17 +192,6 @@ fn seek_needles(sample: &CValue, changed: &[u64]) -> Option { } } -fn ids(col: &CValue) -> Vec { - match corgi::shape_of_value(col) { - Shape::Prim(64) => col.clone().into_u64("ids"), - Shape::Prod(ref fs) if fs.len() == 1 && matches!(fs[0], Shape::Prim(64)) => match col { - CValue::Prod(fields) => fields[0].clone().into_u64("ids"), - _ => unreachable!("shape Prod but value not Prod"), - }, - _ => corgi::hash(col).into_u64("ids"), - } -} - /// Concatenate the records of the `changed` keys across a run of chunks into parallel /// `(keys_col, vals_col)` corgi columns plus per-record `(key_hash, time, diff)`. `changed` is the /// ASCENDING set of changed key hashes; a row is kept iff its key hash is in it. From f5bf60ed89d7782a28208d8afffe7ba3a51344d2 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 31 Jul 2026 11:43:29 -0400 Subject: [PATCH 3/4] Remove dead infer_shape; one shape-inference function, not two MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit infer_shape (single-sample) had no callers but its own recursion — container.rs imports only infer_shape_cols — and its doc's justification ('stays for the single-sample callers that never carry variants') described callers that no longer exist. It also carried a panic for the Variant case that infer_shape_cols exists precisely to handle. Gate green, debug and release. Co-Authored-By: Claude Fable 5 --- interactive/src/corgi/logic.rs | 16 ++-------------- 1 file changed, 2 insertions(+), 14 deletions(-) diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index 5866d51f4..11bb882a1 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -14,22 +14,10 @@ use crate::parse::{BinOp, Term, UnOp}; 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 unimplemented"), - } -} - /// 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.) +/// single sample can't reveal (it shows only one tag), so the scan is over every row, recursing +/// column-wise to cover nested variants too. pub fn infer_shape_cols(rows: &[DValue]) -> Shape { let Some(first) = rows.first() else { return Shape::Unit }; match first { From 775825584bd9a2254ca9a514ca565dcda03c1485 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 31 Jul 2026 11:47:21 -0400 Subject: [PATCH 4/4] Remove backend/col.rs: superseded by the corgi substrate, not merely parked MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit col.rs was commented out of backend/mod.rs pending 'a Columnar/columnar-storage story for Value'. That story arrived as corgi. The module cannot be revived as written: its Row(pub Vec) and columnar::Vecs, Strides> are the FLAT i64 row model DDIR abandoned for structured ir::Value (Int/Tuple/Variant/List), so reviving it would mean rewriting it — which is what the corgi backend is. 259 lines plus a .disabled example, neither compiled. git history keeps them. Co-Authored-By: Claude Fable 5 --- interactive/examples/ddir_col.rs.disabled | 138 ------------ interactive/src/backend/col.rs | 259 ---------------------- interactive/src/backend/mod.rs | 3 - 3 files changed, 400 deletions(-) delete mode 100644 interactive/examples/ddir_col.rs.disabled delete mode 100644 interactive/src/backend/col.rs diff --git a/interactive/examples/ddir_col.rs.disabled b/interactive/examples/ddir_col.rs.disabled deleted file mode 100644 index de4162508..000000000 --- a/interactive/examples/ddir_col.rs.disabled +++ /dev/null @@ -1,138 +0,0 @@ -//! DD IR columnar driver: parse, lower, render (via `interactive::backend::col`), execute. - -use mimalloc::MiMalloc; - -#[global_allocator] -static GLOBAL: MiMalloc = MiMalloc; - -use differential_dataflow::dynamic::pointstamp::PointStamp; - -use interactive::parse; -use interactive::lower; -use interactive::backend::col::{render_tree, Row, Diff}; - -type DdirOuterUpdate = (Row, Row, u64, Diff); - -fn run(name: &str, stmts: Vec, n_inputs: usize, nodes: u64, edges: u64, arity: usize, batch: u64, rounds: Option) { - let mut tree = lower::lower_tree(stmts); - let ops_before = tree.op_count(); - tree.optimize(); - let tree_export_idx = tree.root.exports.iter().position(|e| e.name == "result").unwrap_or(0); - println!("{}: {} ops before optimize, {} after; driving export {:?}", - name, ops_before, tree.op_count(), tree.root.exports[tree_export_idx].name); - let name = name.to_string(); - - timely::execute_from_args(std::env::args().skip(4), move |worker| { - use timely::dataflow::InputHandle; - use timely::container::PushInto; - use differential_dataflow::columnar::ValColBuilder; - - type OuterBuilder = ValColBuilder; - - let (mut inputs, probe) = worker.dataflow::(|scope| { - let mut handles = Vec::new(); - let mut collections = Vec::new(); - for _ in 0..n_inputs { - let mut h = >::new_with_builder(); - let stream = h.to_stream(scope); - handles.push(h); - collections.push(differential_dataflow::Collection::new(stream)); - } - let mut probe = timely::dataflow::ProbeHandle::new(); - let output = scope.iterative::, _, _>(|inner| { - let entered: Vec<_> = collections.iter().map(|c| c.clone().enter(inner)).collect(); - let root_imports: Vec<_> = tree.root.imports.iter().map(|imp| match &imp.from { - interactive::scope_ir::Source::Input(n) => entered[*n].clone(), - interactive::scope_ir::Source::Trace(name) => panic!("ddir_col: Import {:?} not supported in this harness (no trace registry).", name), - interactive::scope_ir::Source::Parent(_) => unreachable!("root scope cannot import from a parent"), - }).collect(); - let exports = render_tree(&tree.root, inner, 0, root_imports); - exports[tree_export_idx].clone().leave(scope) - }); - output.probe_with(&mut probe); - (handles, probe) - }); - - let index = worker.index(); - let peers = worker.peers(); - - let mut builders: Vec = (0..n_inputs).map(|_| OuterBuilder::default()).collect(); - - let timer = std::time::Instant::now(); - let timer_load = std::time::Instant::now(); - for e in 0..edges { - if (e as usize) % peers == index { - let input_idx = (e as usize) % inputs.len(); - let (key, val) = interactive::gen_row::(e, nodes, arity); - let time = *inputs[input_idx].time(); - builders[input_idx].push_into((key, val, time, 1i64)); - } - } - for (i, h) in inputs.iter_mut().enumerate() { - use timely::container::ContainerBuilder; - while let Some(container) = builders[i].finish() { h.send_batch(container); } - h.advance_to(1); - h.flush(); - } - while probe.less_than(&1u64) { worker.step(); } - println!("worker {}: {} loaded ({} edges, total {:.2?}, load {:.2?})", index, name, edges, timer.elapsed(), timer_load.elapsed()); - - let mut cursor = 0u64; - let mut round = 0u64; - let limit = rounds.unwrap_or(u64::MAX); - while round < limit { - let timer_round = std::time::Instant::now(); - let time = (round + 2) as u64; - for _ in 0..batch { - let remove_idx = cursor; - let add_idx = edges + cursor; - if (remove_idx as usize) % peers == index { - let input_idx = (remove_idx as usize) % inputs.len(); - let (key, val) = interactive::gen_row::(remove_idx, nodes, arity); - builders[input_idx].push_into((key, val, time, -1i64)); - } - if (add_idx as usize) % peers == index { - let input_idx = (add_idx as usize) % inputs.len(); - let (key, val) = interactive::gen_row::(add_idx, nodes, arity); - builders[input_idx].push_into((key, val, time, 1i64)); - } - cursor += 1; - } - for (i, h) in inputs.iter_mut().enumerate() { - use timely::container::ContainerBuilder; - while let Some(container) = builders[i].finish() { h.send_batch(container); } - h.advance_to(time); - h.flush(); - } - while probe.less_than(&time) { worker.step(); } - - round += 1; - if round % 100 == 0 { - println!("worker {}: {} round {} (total {:.2?}, round {:.2?})", index, name, round, timer.elapsed(), timer_round.elapsed()); - } - } - println!("worker {}: {} done ({} rounds, batch {}, total {:.2?})", index, name, round, batch, timer.elapsed()); - }).unwrap(); -} - -fn main() { - let program = std::env::args().nth(1).unwrap_or_else(|| { std::process::exit(0); }); - let arity: usize = std::env::args().nth(2).unwrap_or("2".into()).parse().unwrap(); - let nodes: u64 = std::env::args().nth(3).unwrap_or("10".into()).parse().unwrap(); - let edges: u64 = std::env::args().nth(4).unwrap_or_else(|| (2 * nodes).to_string()).parse().unwrap(); - let batch: u64 = std::env::args().nth(5).unwrap_or("1".into()).parse().unwrap(); - let rounds: Option = std::env::args().nth(6).map(|s| s.parse().unwrap()); - - let source = interactive::load_program(&program); - let stmts = if program.ends_with(".ddp") { - parse::pipe::parse(&source) - } else { - parse::applicative::parse(&source) - }; - let (n_inputs, imports) = interactive::survey_sources(&stmts); - if !imports.is_empty() { - panic!("ddir_col: program references imports {:?} but this harness has no trace registry.", imports); - } - let name = std::path::Path::new(&program).file_stem().map(|s| s.to_string_lossy().into_owned()).unwrap_or(program.clone()); - run(&name, stmts, n_inputs, nodes, edges, arity, batch, rounds); -} diff --git a/interactive/src/backend/col.rs b/interactive/src/backend/col.rs deleted file mode 100644 index 1948c55d1..000000000 --- a/interactive/src/backend/col.rs +++ /dev/null @@ -1,259 +0,0 @@ -//! Columnar rendering substrate. -//! -//! Rows are a stride-encoded `Row(Vec)`; the differential container is -//! columnar `RecordedUpdates`. Supplies the substrate leaf operators (join, -//! reduce, arrange over columnar builders/batchers/spines); the scope-tree walk -//! lives in [`crate::backend::render_tree`]. - -mod types { - /// A row type backed by Vec but using Strides for columnar bounds. - /// This ensures uniform-length rows (common in the IR) get compact - /// stride-based offset encoding rather than per-element u64 bounds. - #[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)] - pub struct Row(pub Vec); - - impl Row { - pub fn new() -> Self { Row(Vec::new()) } - pub fn push(&mut self, v: i64) { self.0.push(v); } - } - - impl std::ops::Deref for Row { - type Target = [i64]; - fn deref(&self) -> &[i64] { &self.0 } - } - - impl std::iter::FromIterator for Row { - fn from_iter>(iter: I) -> Self { - Row(iter.into_iter().collect()) - } - } - - impl<'a> IntoIterator for &'a Row { - type Item = &'a i64; - type IntoIter = std::slice::Iter<'a, i64>; - fn into_iter(self) -> Self::IntoIter { self.0.iter() } - } - - impl IntoIterator for Row { - type Item = i64; - type IntoIter = std::vec::IntoIter; - fn into_iter(self) -> Self::IntoIter { self.0.into_iter() } - } - - impl columnar::Columnar for Row { - type Container = columnar::Vecs, columnar::primitive::offsets::Strides>; - - fn into_owned<'a>(other: columnar::Ref<'a, Self>) -> Self { - Row(other.into_iter().copied().collect()) - } - - fn copy_from<'a>(&mut self, other: columnar::Ref<'a, Self>) { - self.0.clear(); - self.0.extend(other.into_iter().copied()); - } - } - - impl crate::ir::RowLike for Row { - fn new() -> Self { Row::new() } - fn push(&mut self, v: i64) { Row::push(self, v); } - fn as_slice(&self) -> &[i64] { &self.0 } - fn extend_from_slice(&mut self, other: &[i64]) { self.0.extend_from_slice(other); } - } - - pub type Diff = i64; - pub type Time = timely::order::Product>; -} - -use differential_dataflow::columnar as columnar_support; - -mod columnar { - use super::types::*; - - pub use super::columnar_support::*; - pub use super::columnar_support::{ValSpine, ValBatcher, ValBuilder}; - - pub type DdirUpdate = (Row, Row, Time, Diff); - pub type DdirRecordedUpdates = RecordedUpdates; - - pub type ColValSpine = ValSpine; - pub type ColValBatcher = ValBatcher; - pub type ColValBuilder = ValBuilder; - pub type ColValChunker = ValChunker; -} - -mod render { - use std::sync::Arc; - use timely::order::Product; - use differential_dataflow::Collection; - use differential_dataflow::dynamic::pointstamp::PointStamp; - use differential_dataflow::operators::arrange::{Arranged, TraceAgent}; - use columnar::Columnar; - use super::types::*; - use crate::ir::{LinearOp, RowLike, eval_fields, eval_field_into, eval_condition}; - use crate::parse::{Projection, Reducer}; - use crate::backend::Backend; - - use super::columnar::{DdirUpdate, DdirRecordedUpdates}; - use super::columnar::{ColValSpine, ColValBuilder}; - - type ConcreteTime = Product>; - - pub type Col<'scope> = Collection<'scope, ConcreteTime, DdirRecordedUpdates>; - type Arr<'scope> = Arranged<'scope, TraceAgent>>; - - /// Render a Linear chain: one pass applying the ops in sequence. `level` is - /// the op's scope depth — it locates the iteration coord for LiftIter and - /// the coordinate position EnterAt's delay lands in. - fn render_linear<'scope>(c: Col<'scope>, ops: Vec, level: usize) -> Col<'scope> { - super::columnar::join_function(c, move |k, v, t_in, _d| { - use timely::progress::Timestamp; - let k: Row = Columnar::into_owned(k); - let v: Row = Columnar::into_owned(v); - // Materialize input time once so LiftIter can read - // the iter coord at the operator's scope depth. - let t_owned: Time = Columnar::into_owned(t_in); - let iter_at_level: i64 = level - .checked_sub(1) - .and_then(|idx| t_owned.inner.get(idx).copied()) - .unwrap_or(0) as i64; - let mut results: Vec<(Row, Row, Time, Diff)> = vec![(k, v, Time::minimum(), 1i64)]; - for op in &ops { - let mut next = Vec::new(); - for (k, v, t, d) in results { - match op { - LinearOp::Project(proj) => { - let i = [k.as_slice(), v.as_slice()]; - next.push((eval_fields(&proj.key, &i), eval_fields(&proj.val, &i), t, d)); - }, - LinearOp::Filter(cond) => { - let i = [k.as_slice(), v.as_slice()]; - if eval_condition(cond, &i) { next.push((k, v, t, d)); } - }, - LinearOp::Negate => { - next.push((k, v, t, -d)); - }, - LinearOp::EnterAt(field) => { - let delay = { - let mut r = Row::new(); - eval_field_into(field, &[k.as_slice(), v.as_slice()], &mut r); - 256 * (64 - (r.as_slice().first().copied().unwrap_or(0) as u64).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); - next.push((k, v, Product::new(0u64, PointStamp::new(coords)), d)); - }, - LinearOp::LiftIter => { - let mut new_v = v.clone(); - new_v.push(iter_at_level); - next.push((k, new_v, t, d)); - }, - } - } - results = next; - } - results.into_iter() - }) - } - - fn render_join<'scope>(l: Arr<'scope>, r: Arr<'scope>, projection: &Projection) -> Col<'scope> { - let proj = projection.clone(); - use differential_dataflow::operators::join::join_traces; - use differential_dataflow::collection::AsCollection; - use super::columnar::ValColBuilder; - let stream = join_traces::<_, _, _, _, ValColBuilder>(l, r, move |k, v1, v2, t, d1, d2, c| { - use differential_dataflow::difference::Multiply; - let d = d1.clone().multiply(d2); - let i = [k.as_slice(), v1.as_slice(), v2.as_slice()]; - let (k2, v2): (Row, Row) = (eval_fields(&proj.key, &i), eval_fields(&proj.val, &i)); - c.give((k2, v2, t, d)); - }); - stream.as_collection() - } - - fn render_reduce<'scope>(a: Arr<'scope>, reducer: &Reducer) -> Arr<'scope> { - let reducer = reducer.clone(); - type ReduceFn = dyn for<'a> Fn(columnar::Ref<'a, Row>, &[(columnar::Ref<'a, Row>, Diff)], &mut Vec<(Row, Diff)>) + Send + Sync; - let f: Arc = match reducer { - Reducer::Min => Arc::new(|_key, vals, output| { - if let Some(min) = vals.iter().map(|(v, _)| v.as_slice()).min() { - output.push((Row(min.to_vec()), 1)); - } - }), - Reducer::Distinct => Arc::new(|_key, _vals, output| { output.push((Row::new(), 1)); }), - Reducer::Count => Arc::new(|_key, vals, output| { - let count: Diff = vals.iter().map(|(_, d)| *d).sum(); - if count != 0 { let mut r = Row::new(); r.push(count); output.push((r, 1)); } - }), - }; - a.reduce_abelian::<_, ColValBuilder<_,_,_,_>, ColValSpine<_,_,_,_>, _, _>( - "Reduce", - move |k, vals, output| { f(k, vals, output); }, - |col, key, upds| { - use columnar::{Clear, Push}; - col.keys.clear(); - col.vals.clear(); - col.times.clear(); - col.diffs.clear(); - for (val, time, diff) in upds.drain(..) { col.push((key, &val, &time, &diff)); } - // NOTE: required because push above doesn't group by key, val. - *col = std::mem::take(col).consolidate(); - }, - ) - } - - fn render_inspect<'scope>(col: Col<'scope>, label: String) -> Col<'scope> { - col.inspect_container(move |event| { - if let Ok((_time, container)) = event { - for (k, v, t, d) in container.updates.view().iter() { - eprintln!(" [{}] ({:?}, {:?}, {:?}, {:?})", label, ::into_owned(k), ::into_owned(v),