diff --git a/interactive/src/backend/vec.rs b/interactive/src/backend/vec.rs index 8ef431415..66454178f 100644 --- a/interactive/src/backend/vec.rs +++ b/interactive/src/backend/vec.rs @@ -190,6 +190,24 @@ impl Backend for VecBackend { for (v, d) in vals { for _ in 0..(*d).max(0) { items.push((*v).clone()); } } output.push((Value::List(items), 1)); }), + // The term reads the key and its values with their diffs, in value order (DD hands + // them sorted, and only those with non-zero diffs), and lists the output with diffs. + Reducer::Expr(term) => { + let term = term.clone(); + Arc::new(move |key, vals, output| { + let vals = vals.iter().map(|(v, d)| Value::Tuple(vec![(*v).clone(), Value::Int(*d)])).collect(); + match eval(&term, &mut vec![key.clone(), Value::List(vals)]) { + Value::List(outs) => output.extend(outs.into_iter().map(|out| match out { + Value::Tuple(mut pair) if pair.len() == 2 && matches!(pair[1], Value::Int(_)) => { + let diff = pair.pop().unwrap().as_int(); + (pair.pop().unwrap(), diff) + } + other => panic!("reduce: expected an output (val, diff), got {other:?}"), + })), + other => panic!("reduce: expected a List, got {other:?}"), + } + }) + } }; // The same reduction over either arrangement; its output is always local. macro_rules! reduce { diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index 960ae02fd..c69a4df75 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -789,6 +789,16 @@ pub fn compile_flatmap(list_term: &Term, kshape: &Shape, vshape: &Shape) -> Res< } } +/// Compile a `reduce(term)` term over `Var(0)=key` (`kshape`) and `Var(1)` = a list of +/// `(val, diff)` (`vshape` the values') → a corgi `List` column of `(val, diff)`, one list per key. +pub fn compile_reduce(term: &Term, kshape: &Shape, vshape: &Shape) -> Res> { + let list = Shape::List(Box::new(Shape::Prod(vec![vshape.clone(), Shape::Int]))); + match lower(&[term], &[kshape.clone(), list])? { + (g, Shape::List(e)) if matches!(&*e, Shape::Prod(fs) if fs.len() == 2 && fs[1] == Shape::Int) => Ok(g), + (_, other) => Err(format!("reduce output is not a list of (val, diff): {other}")), + } +} + /// Compile a scalar term (`EnterAt`'s delay field) → an `Int` column; a non-integer term is the /// type error (the delay is read as one integer per row). pub fn compile_scalar(term: &Term, kshape: &Shape, vshape: &Shape) -> Res> { diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 378acf212..3776278e1 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -15,7 +15,8 @@ //! records the row holding its key, which is what `emit` reads. //! * the value callback — `reduce_brackets` runs ONE crossing per wave over every `(key, time)` //! bracket, building the output value COLUMNS directly (Count → an `Int` column, Distinct → a -//! `Unit`, Min → the chosen input rows, Collect → a `List`), never through DDIR rows. +//! `Unit`, Min → the chosen input rows, Collect → a `List`, a term → its graph's output over +//! every bracket's key and `(val, diff)` list), never through DDIR rows. //! * materialize — gather the emitted keys and values from the pools' columns and seal a //! `CorgiChunk` batch column-natively. //! @@ -36,11 +37,12 @@ use differential_dataflow::operators::int_proxy::{KeyPosition, ProxyBridge}; use differential_dataflow::operators::int_proxy::reduce::{ProxyReduceBackend, ReduceInstance, ReduceWindow}; use corgi::arrange::{compare_at, gather_lanes, sort_blocks}; -use corgi::{Bounds, Value as CValue}; +use corgi::{Bounds, Graph, NumOp, Shape, Value as CValue}; use crate::corgi::col_times::{ColTime, ColTimes}; use crate::corgi::search::matching_ranges; -use crate::corgi::chunk::{columns_to_batch, ident, key_ids, key_is_hashed, CorgiChunk}; +use crate::corgi::chunk::{columns_to_batch, ident, key_ids, key_is_hashed, recover_key, CorgiChunk}; +use crate::corgi::logic::compile_reduce; use crate::ir::{Diff, Reducer}; type CBatch = Rc>>; @@ -77,6 +79,8 @@ type Matches = Vec)>>; /// differs from what it has, and `emit` records the difference; finally `finish` builds the batch. pub struct CorgiReduceBackend { reducer: Reducer, + /// A `Reducer::Expr`'s graph, compiled on first use at the key and value shapes it reads. + graph: Option<(Shape, Shape, Graph)>, /// The current window's input values: its chunks' value columns. input: Pool, /// The current retire's output values: its output chunks' value columns, then those built here. @@ -145,6 +149,7 @@ impl CorgiReduceBackend { pub fn with_window(reducer: Reducer, window_size: usize) -> Self { CorgiReduceBackend { reducer, + graph: None, input: Pool::default(), output: Pool::default(), output_keys: Vec::new(), @@ -377,6 +382,7 @@ where }; (self.input.gather(&min_reps), out_ends) } + Reducer::Expr(_) => unreachable!("a term's brackets are `reduce_term`'s"), Reducer::Collect => { // One row per bracket: the values sorted in DDIR observable order, // each repeated by its diff, as a `List`. One `sort_blocks` orders every bracket's @@ -427,6 +433,67 @@ where } } } + + /// As [`Self::reduce_brackets`] for `Reducer::Expr`, whose output rows carry diffs: every + /// bracket's key and its values, as a list of `(val, diff)`, through the term's graph at once. + /// Returns the output values, their diffs, and each bracket's end among them. + /// + /// The harness presents each bracket consolidated, so its diffs are non-zero and its value ids + /// ascend, which is value order: `present` numbers each key's values in sorted order. The + /// brackets are presented as they are. One with no input has no output, and the term does not + /// see it. + fn reduce_term(&mut self, ends: &[usize], input: &[(u64, Diff)]) -> (CValue, Vec, Vec) { + // Per bracket with input: its index, its end in `input`, and the row holding its key. + let (mut blocks, mut key_refs) = (Vec::new(), Vec::new()); + let mut start = 0; + for (bi, &end) in ends.iter().enumerate() { + if end > start { + blocks.push((bi, end)); + key_refs.push(self.input.refs[input[start].0 as usize]); + } + start = end; + } + if blocks.is_empty() { + return (CValue::Unit(0), Vec::new(), vec![0; ends.len()]); + } + let vals = self.input.gather(&input.iter().map(|&(id, _)| id).collect::>()); + let diffs = CValue::i64(input.iter().map(|&(_, diff)| diff).collect()); + if cfg!(debug_assertions) { + let mut start = 0; + for &(_, end) in &blocks { + for i in start + 1..end { + assert!(compare_at(&vals, i - 1, &vals, i).is_lt(), "reduce: a bracket's values are not strictly ascending"); + } + start = end; + } + assert!(input.iter().all(|&(_, diff)| diff != 0), "reduce: a bracket holds a zero diff"); + } + let keys = recover_key(&gather_refs(&self.key_columns, key_refs.into_iter())); + let (kshape, vshape) = (corgi::shape_of_value(&keys), corgi::shape_of_value(&vals)); + let lists = CValue::List(Bounds::offsets(blocks.iter().map(|b| b.1).collect()), Box::new(CValue::Prod(vec![vals, diffs]))); + + if self.graph.is_none() { + let Reducer::Expr(term) = &self.reducer else { unreachable!("only a term compiles a graph") }; + let g = compile_reduce(term, &kshape, &vshape) + .unwrap_or_else(|e| panic!("reduce: type error at shapes ({kshape}, {vshape}): {e}")); + self.graph = Some((kshape.clone(), vshape.clone(), g)); + } + let (k, v, graph) = self.graph.as_ref().unwrap(); + assert!(*k == kshape && *v == vshape, "reduce: input of shape ({kshape}, {vshape}) reached an operator pinned at ({k}, {v})"); + let (bounds, elems) = corgi::eval_graph(graph, CValue::Prod(vec![keys, lists])).into_list("reduce output").unwrap(); + let mut fields = elems.into_prod("reduce output").unwrap(); + let out_diffs = fields.pop().unwrap().into_i64("reduce output diffs").unwrap().to_vec(); + let out_vals = fields.pop().unwrap(); + + // Each bracket's end among the output rows; brackets without input add none. + let list_ends = bounds.to_vec(); + let (mut out_ends, mut blocks, mut emitted) = (Vec::with_capacity(ends.len()), blocks.iter().zip(list_ends).peekable(), 0); + for bi in 0..ends.len() { + if let Some((_, end)) = blocks.next_if(|((b, _), _)| *b == bi) { emitted = end; } + out_ends.push(emitted); + } + (out_vals, out_diffs, out_ends) + } } impl ProxyReduceBackend, CBatch> for CorgiReduceBackend @@ -507,7 +574,8 @@ where } fn reduce_corrections(&mut self, keys: &[i64], in_ends: &[usize], input: &[(u64, Diff)], out_ends: &[usize], output: &[(u64, Diff)]) -> (Vec<(u64, Diff)>, Vec) { - // Each key wants its rows of `desired`, once each, and has `(id, diff)`s of output. The + // Each key wants its rows of `desired`, each with its diff (once each, but for a term's), and + // has `(id, diff)`s of output. The // correction nets the two by value: a wanted row equal to an output id's value counts // toward that id, and one equal to none gets a new id. Keys have a handful of each. // @@ -534,7 +602,11 @@ where } } let (in_ends, input) = if split { (&sub_ends[..], &sub_input[..]) } else { (in_ends, input) }; - let (desired, desired_ends) = self.reduce_brackets(in_ends, input); + // The desired rows, and each one's diff where it is not one copy. + let (desired, desired_diffs, desired_ends) = match self.reducer { + Reducer::Expr(_) => { let (d, diffs, e) = self.reduce_term(in_ends, input); (d, Some(diffs), e) } + _ => { let (d, e) = self.reduce_brackets(in_ends, input); (d, None, e) } + }; let fresh = self.output.columns.len(); if !desired.is_empty() { self.output.columns.push(desired); } let Pool { columns, refs } = &mut self.output; @@ -552,6 +624,7 @@ where let first = if s == 0 { 0 } else { in_ends[s - 1] }; let (kc, kr) = self.input.refs[input[first].0 as usize]; for row in ds..de { + let diff = desired_diffs.as_ref().map_or(1, |diffs| diffs[row]); let equal = |&(id, _): &(u64, Diff)| { let (column, r) = refs[id as usize]; compare_at(&columns[fresh], row, &columns[column], r).is_eq() && (!checked || { @@ -560,9 +633,9 @@ where }) }; match net.iter().position(equal) { - Some(at) => net[at].1 += 1, + Some(at) => net[at].1 += diff, None => { - net.push((refs.len() as u64, 1)); + net.push((refs.len() as u64, diff)); refs.push((fresh, row)); output_keys.push((kc, kr)); } @@ -811,12 +884,24 @@ mod tests { let early: Vec<_> = early.into_iter().filter(|r| r.2.outer < 2).collect(); let stamp = |outer| Time::new(outer, PointStamp::new(std::iter::empty().collect())); - for reducer in [Reducer::Count, Reducer::Distinct, Reducer::Min, Reducer::Collect] { + // A term: the weighted sum of the values' first fields, once, and how many values, twice negated. + let term = { + use crate::ir::{BinOp, Term as Tm, UnOp}; + let b = Box::new; + let weighted = Tm::Binary(BinOp::Mul, b(Tm::Proj(b(Tm::Proj(b(Tm::Bound(0)), 0)), 0)), b(Tm::Proj(b(Tm::Bound(0)), 1))); + let sum = Tm::Fold { list: b(Tm::Var(1)), init: b(Tm::Int(0)), step: b(Tm::Binary(BinOp::Add, b(Tm::Bound(1)), b(weighted))) }; + Tm::List(vec![ + Tm::Tuple(vec![Tm::Tuple(vec![sum]), Tm::Int(1)]), + Tm::Tuple(vec![Tm::Tuple(vec![Tm::Unary(UnOp::Len, b(Tm::Var(1)))]), Tm::Int(-2)]), + ]) + }; + for reducer in [Reducer::Count, Reducer::Distinct, Reducer::Min, Reducer::Collect, Reducer::Expr(term)] { let out_shape = match reducer { Reducer::Count => Shape::Prod(vec![Shape::Int]), Reducer::Distinct => Shape::Unit, Reducer::Min => shape.clone(), Reducer::Collect => Shape::List(Box::new(shape.clone())), + Reducer::Expr(_) => Shape::Prod(vec![Shape::Int]), }; // Keys `(a, 7a)`, identified by their hash or by `a % 2`. let run = |forged: bool, window_size: usize| { diff --git a/interactive/src/ir.rs b/interactive/src/ir.rs index 2a529ecae..3a719a3ee 100644 --- a/interactive/src/ir.rs +++ b/interactive/src/ir.rs @@ -245,6 +245,11 @@ pub enum Reducer { /// (`Value: Ord`) order — so deterministic, and position-ordered when the /// values are `tuple(pos, …)` as `flatmap` emits. The inverse of `FlatMap`. Collect, + /// A key's output as a term: `Var(0)` = key, `Var(1)` = the key's values with + /// non-zero accumulated diffs, as a `List` of `tuple(val, diff)` in the values' + /// (`Value: Ord`) order. The term gives a `List` of `tuple(val, diff)`: the key's + /// output values and their diffs. A key with no such values has no output. + Expr(Term), } /// An individual step within a Linear node. diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs index a611a9113..dc03b0d3b 100644 --- a/interactive/src/parse/pipe.rs +++ b/interactive/src/parse/pipe.rs @@ -24,6 +24,9 @@ //! - `| join(other, (k… ; v…))` — equijoin on the key. //! - `| min` / `| distinct` / `| count` / `| collect` — reduce; `collect` is //! NEST (gather a key's values into a `List`). +//! - `| reduce(term)` — reduce by a term: `$0` is the key and `$1` the key's +//! values with their non-zero diffs, a `List` of `tuple(val, diff)` in value +//! order; the term gives the key's output, a `List` of `tuple(val, diff)`. //! - `| filter(term)` — keep rows where `term` is truthy (a nonzero `Int`). //! - `| flatmap(term)` — UNNEST: explode a `List`-valued `term` into one row //! per element, value `tuple(pos, element)`. @@ -98,7 +101,8 @@ //! Any slot may instead be a corgi lambda, `pat -> body`, in corgi's ML syntax: it runs //! `let pat = input in body` with `input` the tuple of the slot's rows, `(key, val)` for a linear //! op and `(key, left val, right val)` for a join. In `key`, `map`, and `join` the body gives the -//! pair `(key, val)`; in `value`, `filter`, `flatmap`, and `enter_at` it gives the one value. For +//! pair `(key, val)`; in `value`, `filter`, `flatmap`, and `enter_at` it gives the one value; in +//! `reduce` its input is `(key, vals)` and it gives the output list. For //! example, `key(((src, dst), ()) -> ((src), (dst)))` is `key($0[0] ; $0[1])`. use super::*; @@ -532,6 +536,11 @@ impl Parser { Token::Distinct => { self.next(); Expr::Reduce(Box::new(lhs), Reducer::Distinct) }, Token::Count => { self.next(); Expr::Reduce(Box::new(lhs), Reducer::Count) }, Token::Collect => { self.next(); Expr::Reduce(Box::new(lhs), Reducer::Collect) }, + Token::Ident(name) if name == "reduce" => { + self.next(); self.expect(&Token::LParen); + let term = self.parse_slot_term(); self.expect(&Token::RParen); + Expr::Reduce(Box::new(lhs), Reducer::Expr(term)) + }, Token::FlatMap => { self.next(); self.expect(&Token::LParen); let l = self.parse_slot_term(); self.expect(&Token::RParen); Expr::FlatMap(Box::new(lhs), l) }, Token::Arrange => { self.next(); Expr::Arrange(Box::new(lhs)) }, Token::Negate => { self.next(); Expr::Negate(Box::new(lhs)) }, diff --git a/interactive/src/shapes.rs b/interactive/src/shapes.rs index 1a5f5fbee..750948dc0 100644 --- a/interactive/src/shapes.rs +++ b/interactive/src/shapes.rs @@ -19,7 +19,7 @@ use corgi::Shape; use crate::corgi::logic::shape_of_term; -use crate::ir::{LinearOp, Reducer}; +use crate::ir::{LinearOp, Reducer, Term}; use crate::scope_ir as st; /// The shape of a collection's updates. @@ -184,6 +184,19 @@ pub fn linear_shapes(ops: &[LinearOp], key: Shape, val: Shape) -> Result Result { + let env = [key.clone(), Shape::List(Box::new(Shape::Prod(vec![val.clone(), Shape::Int])))]; + match shape_of_term(term, &env, None).map_err(|e| format!("reduce: {e}"))? { + Shape::List(e) => match *e { + Shape::Prod(mut fs) if fs.len() == 2 && fs[1] == Shape::Int => Ok(fs.swap_remove(0)), + other => Err(format!("reduce: output elements of shape {other}, not (val, diff)")), + }, + other => Err(format!("reduce: output of shape {other}, not a list of (val, diff)")), + } +} + fn infer_node(node: &st::Node, shapes: &ScopeShapes, depth: usize, at: &str, problems: &mut Problems) -> Option { let mut fail = |e: String| { problems.conflicts.push(format!("{at}: {e}")); None }; match node { @@ -218,6 +231,10 @@ fn infer_node(node: &st::Node, shapes: &ScopeShapes, depth: usize, at: &str, pro Reducer::Distinct => Shape::Unit, Reducer::Count => Shape::Prod(vec![Shape::Int]), Reducer::Collect => Shape::List(Box::new(c.val)), + Reducer::Expr(t) => match reduce_shape(t, &c.key, &c.val) { + Ok(v) => v, + Err(e) => return fail(e), + }, }; Some(CollShape::new(c.key, v, depth)) } diff --git a/interactive/tests/corgi_backend.rs b/interactive/tests/corgi_backend.rs index ed03d423e..571d66753 100644 --- a/interactive/tests/corgi_backend.rs +++ b/interactive/tests/corgi_backend.rs @@ -39,6 +39,8 @@ fn inputs_for(prog: &str) -> Vec> { // scalar_ops: (key, a, b) triples; a values straddle the `> 2` and `= -5` tests. "scalar_ops" => vec![rows(&[&[1, 1, 9], &[1, 4, 8], &[2, 3, 7], &[3, -5, 6], &[3, 2, 5]])], "sum_ops" => vec![rows(&[&[1, 10], &[2, 20], &[2, 21]])], + // reduce_term: keys with one value, several, and a repeated one. + "reduce_term" => vec![rows(&[&[1, 10], &[2, 20], &[2, -21], &[2, 20], &[3, 30], &[3, 5]])], // empty_batch: only key 1 has the two values the filter keeps, so at several workers at // least one gets a batch that arrives non-empty and leaves empty. "empty_batch" => vec![rows(&[ @@ -144,6 +146,7 @@ fn serializing(n: usize) -> timely::Config { #[test] fn signed_min() { assert_backends_agree("signed_min"); } #[test] fn spread_values() { assert_backends_agree("spread_values"); } #[test] fn f64_math() { assert_backends_agree("f64_math"); } +#[test] fn reduce_term() { assert_backends_agree("reduce_term"); } #[test] fn registered() { assert_backends_agree("registered"); // And the program does what it says: the refinement from 1 and 3 reaches every diff --git a/interactive/tests/explain.rs b/interactive/tests/explain.rs index ecdce57d4..7c74dbf8e 100644 --- a/interactive/tests/explain.rs +++ b/interactive/tests/explain.rs @@ -399,6 +399,18 @@ fn count_duplicate_inputs_sufficient() { ); } +/// A reduction by a term: the sum of each node's in-neighbors. As for `count`, the keyed lookup +/// demands every input row at the key, which is what the term reads. +#[test] +fn reduce_term_explanations_sufficient_small() { + const SUM_ROW: &str = r#" + let edges = input 0 : ((int, int) ; ()) | key($0[0] ; $0[1]); + export "result" = edges | key($1 ; $0) | reduce(list(tuple(tuple(fold($1, 0, ^1 + ^0[0][0] * ^0[1])), 1))); + "#; + let sizes = assert_all_rows_sufficient(SUM_ROW, &[gen_edges(20, 22)]); + assert!(!sizes.is_empty()); +} + /// Two queries seeded together (distinct q ids): the union demand-set must /// regenerate both queried rows. Exercises the q-id plumbing through every /// reverse rule — a mismatched q in a lookup would silently drop demand. diff --git a/interactive/tests/programs/reduce_term.ddp b/interactive/tests/programs/reduce_term.ddp new file mode 100644 index 000000000..b9f345d25 --- /dev/null +++ b/interactive/tests/programs/reduce_term.ddp @@ -0,0 +1,11 @@ +-- Reductions written as terms, checked against vec: the term reads the key and a list of +-- `(val, diff)`, and lists the output with diffs. Weighted sums and maxima fold the list; `$1` +-- alone is the identity; an output diff need not be one, and may be negative. +let pairs = input 0 : ((int, int) ; ()) | key($0[0] ; $0[1]); + +export "sum" = pairs | reduce(list(tuple(tuple(fold($1, 0, ^1 + ^0[0][0] * ^0[1])), 1))); +export "max" = pairs | reduce(list(tuple(tuple(fold($1, 0 - 1000000, if(^0[0][0] > ^1, ^0[0][0], ^1))), 1))); +export "same" = pairs | reduce($1); +export "weights" = pairs | reduce(list(tuple(tuple(len($1)), 0 - 2), tuple(tuple(fold($1, 0, ^1 + ^0[1])), 3))); +-- A two-field key, which the arrangement identifies by its hash. +export "pair_keys" = pairs | key($0, 7 ; $1) | reduce(list(tuple(tuple(fold($1, 0, ^1 + ^0[1])), 1))); diff --git a/interactive/tests/reduce_term.rs b/interactive/tests/reduce_term.rs new file mode 100644 index 000000000..388e02288 --- /dev/null +++ b/interactive/tests/reduce_term.rs @@ -0,0 +1,73 @@ +//! `reduce(term)`: the four prefab reducers written as terms reduce as the prefabs do, and a term +//! and the same term as a corgi lambda agree, on both backends, across workers, as inputs are added +//! and retracted. + +use interactive::ir::Value; +use interactive::server::{RenderBackend, Server}; +use interactive::{lower, parse}; + +const PROGRAM: &str = r#" + let pairs = input 0 : ((int, int) ; ()) | key($0[0] ; $0[1]); + export "count" = pairs | count; + export "count_term" = pairs | reduce(list(tuple(tuple(fold($1, 0, ^1 + ^0[1])), 1))); + export "distinct" = pairs | distinct; + export "distinct_term" = pairs | reduce(list(tuple(tuple(), 1))); + export "min" = pairs | min; + export "min_term" = pairs | reduce(list(tuple(tuple(fold($1, 1000000, if(^0[0][0] < ^1, ^0[0][0], ^1))), 1))); + export "sum_term" = pairs | reduce(list(tuple(tuple(fold($1, 0, ^1 + ^0[0][0] * ^0[1])), 1))); + export "sum_lambda" = pairs | reduce((_, vals) -> ((vals map (p -> (p.0.0, p.1) mul) fold_add), 1) enlist); +"#; + +/// The rows fed at each tick, as `(key, val, diff)`: additions, repeats, retractions, a key +/// retracted entirely, and a key brought back. +fn feeds(tick: i64) -> Vec<(i64, i64, i64)> { + match tick { + 0 => vec![(1, 10, 1), (1, 20, 1), (2, 5, 1), (2, 5, 1), (3, -7, 1), (4, 1, 1)], + 1 => vec![(1, 20, -1), (1, 3, 1), (2, 6, 1), (3, -8, 1), (5, 50, 1)], + 2 => vec![(2, 5, -2), (2, 6, -1), (3, -7, -1), (4, 2, 1)], + _ => vec![(2, 9, 1), (1, 10, -1), (1, -4, 1)], + } +} + +/// Every export, read after each tick. +fn run(backend: RenderBackend, workers: usize) -> Vec> { + let names = ["count", "count_term", "distinct", "distinct_term", "min", "min_term", "sum_term", "sum_lambda"]; + let guards = timely::execute(timely::Config::process(workers), move |worker| { + let mut server = Server::with_backend(backend); + let mut program = lower::lower_tree(parse::pipe::parse(PROGRAM)); + program.optimize(); + server.install(worker, "p", &program).unwrap(); + let mut seen = Vec::new(); + for tick in 0..4 { + if worker.index() == 0 { + for (k, v, d) in feeds(tick) { + let row = Value::Tuple(vec![Value::Int(k), Value::Int(v)]); + server.feed("p", 0, row, Value::unit(), None, d).unwrap(); + } + } + server.tick(worker); + for name in names { + seen.push(server.snapshot(worker, name).unwrap()); + } + } + seen + }) + .unwrap(); + guards.join().into_iter().next().unwrap().unwrap() +} + +#[test] +fn terms_reduce_as_the_prefabs_do() { + let want = run(RenderBackend::Vec, 1); + for tick in want.chunks(8) { + assert!(tick.iter().all(|rows| !rows.is_empty()), "every export has rows: {tick:?}"); + assert_eq!(tick[0], tick[1], "count"); + assert_eq!(tick[2], tick[3], "distinct"); + assert_eq!(tick[4], tick[5], "min"); + assert_eq!(tick[6], tick[7], "a term and its lambda"); + } + for workers in [1, 3] { + assert_eq!(run(RenderBackend::Vec, workers), want, "vec, {workers} workers"); + assert_eq!(run(RenderBackend::Corgi, workers), want, "corgi, {workers} workers"); + } +}