DDIR: reduce by a term

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 {

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<Graph<NumOp>> {
+    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<Graph<NumOp>> {

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<T> = Rc<ChunkBatch<CorgiChunk<T, Diff>>>;
@@ -77,6 +79,8 @@ type Matches = Vec<Vec<(usize, Range<usize>)>>;
 /// differs from what it has, and `emit` records the difference; finally `finish` builds the batch.
 pub struct CorgiReduceBackend<T> {
     reducer: Reducer,
+    /// A `Reducer::Expr`'s graph, compiled on first use at the key and value shapes it reads.
+    graph: Option<(Shape, Shape, Graph<NumOp>)>,
     /// 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<T> CorgiReduceBackend<T> {
     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<Diff>, Vec<usize>) {
+        // 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::<Vec<_>>());
+        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<T> ProxyReduceBackend<T, CBatch<T>, CBatch<T>> for CorgiReduceBackend<T>
@@ -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<usize>) {
-        // 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| {

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.

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)) },

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<Vec<(Sh
     Ok(steps)
 }
 
+/// The output value shape of `reduce(term)` over a key of shape `key` with values of shape `val`:
+/// the term reads the key and a `List` of `(val, diff)`, and must give a `List` of `(val, diff)`.
+fn reduce_shape(term: &Term, key: &Shape, val: &Shape) -> Result<Shape, String> {
+    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<CollShape> {
     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))
         }

interactive/tests/corgi_backend.rs

@@ -39,6 +39,8 @@ fn inputs_for(prog: &str) -> Vec<Vec<(Value, Value)>> {
         // 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

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.