DDIR: reduce by a term

[frankmcsherry]

| reduce(term) reduces each key by a term, beside the four prefab reducers (which stay; Min's value narrowing is useful to explain).

Semantics

The term reads the key as $0 and the key's values as $1: a List of tuple(val, diff), holding the values with non-zero accumulated diffs, in value order.
It gives the key's output as a List of tuple(val, diff).
A key with no such values has no output, as DD calls a reducer only for a key with input.
Output diffs are used as given, and equal output values add, as DD consolidates a reducer's output.

A slot may also be a corgi lambda, e.g. reduce((_, vals) -> ((vals map (p -> (p.0.0, p.1) mul) fold_add), 1) enlist) for a weighted sum.

Changes

Reducer::Expr(Term), parsed from reduce(slot).
Shape inference checks that the term gives a list of (val, diff); the output value shape is the first field's.

The vec backend evaluates the term per key.

The corgi backend (reduce_term) builds every bracket's key and (val, diff) list as columns, evaluates the term's graph once per window, and returns the output values with their diffs.
reduce_corrections nets those diffs against the current output, where the prefab reducers want one copy of each row.
The graph is compiled once, at the first window's shapes.

Explain needs no change: the keyed lookup already demands all of a key's input for every reducer but Min.

Tests

• tests/programs/reduce_term.ddp in the vec/corgi agreement suite: weighted sums, a fold max, the identity reduce($1), output diffs of -2 and 3, and two-field (hashed) keys.
• tests/reduce_term.rs: count, distinct, and min written as terms equal the prefabs, and a term equals its lambda, over four ticks of additions and retractions (including a key retracted entirely and re-added), on both backends at 1 and 3 workers.
• An explain sufficiency test for a term reduction.
• A term reducer in the colliding-keys unit test.

Reverting the corrections to one copy per row fails the agreement test.

🤖 Generated with Claude Code