diff --git a/interactive/src/backend/vec.rs b/interactive/src/backend/vec.rs index 8ef431415..fe9cea42e 100644 --- a/interactive/src/backend/vec.rs +++ b/interactive/src/backend/vec.rs @@ -182,7 +182,8 @@ impl Backend for VecBackend { Reducer::Distinct => Arc::new(|_key, _vals, output| { output.push((Value::unit(), 1)); }), // Count yields a one-field tuple `(count)`, keeping the convention // that a value is a tuple (so `$1[0]` and the explain envelope work). - Reducer::Count => Arc::new(|_key, vals, output| { let count: Diff = vals.iter().map(|(_, d)| *d).sum(); if count > 0 { output.push((Value::Tuple(vec![Value::Int(count)]), 1)); } }), + // The count is the sum of the diffs, of either sign, as DD's `count`. + Reducer::Count => Arc::new(|_key, vals, output| { let count: Diff = vals.iter().map(|(_, d)| *d).sum(); if count != 0 { output.push((Value::Tuple(vec![Value::Int(count)]), 1)); } }), // NEST: collect the key's values into a List, in value order (DD // hands them sorted), each repeated per its multiplicity. Reducer::Collect => Arc::new(|_key, vals, output| { diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 378acf212..3dd14a7c2 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -305,12 +305,13 @@ where let mut out_ends: Vec = Vec::with_capacity(ends.len()); match self.reducer { Reducer::Count => { - // Per-bracket sum of diffs; survivors become a `Tuple([Int(sum)])` = corgi `Prod([Int])`. + // Per-bracket sum of diffs, of either sign, as DD's `count`; a non-zero sum becomes a + // `Tuple([Int(sum)])` = corgi `Prod([Int])`. 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 { + if c != 0 { sums.push(c); } out_ends.push(sums.len()); diff --git a/interactive/src/ir.rs b/interactive/src/ir.rs index 2a529ecae..b4b1c153b 100644 --- a/interactive/src/ir.rs +++ b/interactive/src/ir.rs @@ -240,6 +240,8 @@ pub struct Projection { pub key: Term, pub val: Term } pub enum Reducer { Min, Distinct, + /// The sum of a key's diffs, as `(count)`, when it is not zero. It may be negative, as DD's + /// `count` is: a negative count is how a program finds a negative accumulation. Count, /// NEST: collect a key's values into a `Value::List`, in the values' own /// (`Value: Ord`) order — so deterministic, and position-ordered when the diff --git a/interactive/tests/count.rs b/interactive/tests/count.rs new file mode 100644 index 000000000..1aa7b7a66 --- /dev/null +++ b/interactive/tests/count.rs @@ -0,0 +1,39 @@ +//! `count` is the sum of a key's diffs, of either sign, as DD's `count`: a key whose values +//! accumulate negatively has a negative count, and one whose values cancel has none. + +use interactive::ir::Value; +use interactive::server::{evaluate, RenderBackend}; +use interactive::{lower, parse}; + +const PROGRAM: &str = r#" + let pairs = input 0 : ((int, int) ; ()) | key($0[0] ; $0[1]); + let shifted = pairs | key($0 ; $1[0] + 1); + export "positive" = pairs | count; + export "negative" = pairs | negate | count; + export "mixed" = (pairs + pairs - shifted - shifted - shifted) | count; + export "cancel" = (pairs - shifted) | count; +"#; + +fn ints(ns: &[i64]) -> Value { Value::Tuple(ns.iter().map(|&n| Value::Int(n)).collect()) } + +#[test] +fn counts_carry_their_sign() { + let mut program = lower::lower_tree(parse::pipe::parse(PROGRAM)); + program.optimize(); + // Key 1 has two values, key 2 one. + let rows: Vec<(Value, Value)> = [[1, 10], [1, 20], [2, 30]].iter().map(|r| (ints(r), Value::unit())).collect(); + let count = |k: i64, c: i64| ((ints(&[k]), ints(&[c])), 1); + let mut want = std::collections::BTreeMap::new(); + want.insert("positive".to_string(), vec![count(1, 2), count(2, 1)]); + want.insert("negative".to_string(), vec![count(1, -2), count(2, -1)]); + // Each key: twice its values, less three times as many shifted ones. + want.insert("mixed".to_string(), vec![count(1, -2), count(2, -1)]); + // Each key's values cancel against as many shifted ones. + want.insert("cancel".to_string(), vec![]); + for backend in [RenderBackend::Vec, RenderBackend::Corgi] { + for workers in [1, 3] { + let got = evaluate(backend, timely::Config::process(workers), &program, &[rows.clone()]); + assert_eq!(got, want, "{backend:?}, {workers} workers"); + } + } +}