diff --git a/crates/zset/src/circuit.rs b/crates/zset/src/circuit.rs new file mode 100644 index 00000000..ab388313 --- /dev/null +++ b/crates/zset/src/circuit.rs @@ -0,0 +1,343 @@ +//! The synchronous, single-threaded circuit. This module is the wiring (`Rc`, `RefCell`, +//! boxed nodes); the operators' logic is in `ops.rs` and `lib.rs`. + +use crate::ops::{AggregateState, DistinctState, JoinState}; +use crate::{MemTrace, Trace, ZSet}; +use std::cell::RefCell; +use std::rc::Rc; + +/// The value an edge carries during the current step. +pub struct Stream(Rc>); + +impl Clone for Stream { + fn clone(&self) -> Self { + Stream(Rc::clone(&self.0)) + } +} + +impl Stream { + /// The value of the last completed step (the zero before the first). + pub fn current(&self) -> T { + self.0.borrow().clone() + } +} + +/// Where a caller feeds deltas; they enter the circuit at the next `step`. +pub struct Input(Rc>>); + +impl Clone for Input { + fn clone(&self) -> Self { + Input(Rc::clone(&self.0)) + } +} + +impl Input { + pub fn push(&self, delta: ZSet) { + *self.0.borrow_mut() += delta; + } +} + +/// A sink that sums every step's delta until it is taken. +pub struct Output(Rc>>); + +impl Output { + pub fn take(&self) -> ZSet { + std::mem::take(&mut *self.0.borrow_mut()) + } +} + +trait Node { + fn eval(&mut self); +} + +struct Unary { + src: Stream, + out: Stream, + f: F, +} + +impl O> Node for Unary { + fn eval(&mut self) { + let v = (self.f)(&self.src.0.borrow()); + *self.out.0.borrow_mut() = v; + } +} + +struct Binary { + a: Stream, + b: Stream, + out: Stream, + f: F, +} + +impl O> Node for Binary { + fn eval(&mut self) { + let v = (self.f)(&self.a.0.borrow(), &self.b.0.borrow()); + *self.out.0.borrow_mut() = v; + } +} + +/// Moves what was pushed into the circuit's stream. +struct Source { + pending: Rc>>, + out: Stream>, +} + +impl Node for Source { + fn eval(&mut self) { + *self.out.0.borrow_mut() = std::mem::take(&mut *self.pending.borrow_mut()); + } +} + +/// Nodes run in creation order, which is a topological order because a node can only +/// be built from streams that already exist. There are no cycles to schedule. +#[derive(Default)] +pub struct Circuit { + nodes: Vec>, +} + +type Z = ZSet; + +impl Circuit { + pub fn new() -> Self { + Self::default() + } + + /// Runs one clock tick: every node, once, in order. + pub fn step(&mut self) { + for n in &mut self.nodes { + n.eval(); + } + } + + fn unary( + &mut self, + src: &Stream, + f: impl FnMut(&I) -> O + 'static, + ) -> Stream { + let out = Stream(Rc::default()); + self.nodes.push(Box::new(Unary { + src: src.clone(), + out: out.clone(), + f, + })); + out + } + + fn binary( + &mut self, + a: &Stream, + b: &Stream, + f: impl FnMut(&A, &B) -> O + 'static, + ) -> Stream { + let out = Stream(Rc::default()); + self.nodes.push(Box::new(Binary { + a: a.clone(), + b: b.clone(), + out: out.clone(), + f, + })); + out + } + + pub fn input(&mut self) -> (Input, Stream>) { + let pending: Rc>> = Rc::default(); + let out = Stream(Rc::default()); + self.nodes.push(Box::new(Source { + pending: Rc::clone(&pending), + out: out.clone(), + })); + (Input(pending), out) + } + + pub fn output(&mut self, s: &Stream>) -> Output { + let sink = Output(Rc::default()); + let acc = Rc::clone(&sink.0); + self.unary(s, move |d: &Z| { + *acc.borrow_mut() += d; + }); + sink + } + + /// out(t) = in(t - 1); out(0) is zero. + pub fn delay(&mut self, s: &Stream>) -> Stream> { + let mut held = ZSet::new(); + self.unary(s, move |d: &Z| std::mem::replace(&mut held, d.clone())) + } + + /// out(t) = in(0) + .. + in(t). + pub fn integrate(&mut self, s: &Stream>) -> Stream> { + let mut acc = ZSet::new(); + self.unary(s, move |d: &Z| { + acc += d; + acc.clone() + }) + } + + /// out(t) = in(t) - in(t - 1). + pub fn differentiate(&mut self, s: &Stream>) -> Stream> { + let prev = self.delay(s); + self.binary(s, &prev, |a: &Z, b: &Z| a - b) + } + + pub fn plus( + &mut self, + a: &Stream>, + b: &Stream>, + ) -> Stream> { + self.binary(a, b, |a: &Z, b: &Z| a + b) + } + + pub fn map( + &mut self, + s: &Stream>, + mut f: impl FnMut(&K) -> L + 'static, + ) -> Stream> { + self.unary(s, move |d: &Z| d.map(&mut f)) + } + + pub fn filter( + &mut self, + s: &Stream>, + mut keep: impl FnMut(&K) -> bool + 'static, + ) -> Stream> { + self.unary(s, move |d: &Z| d.filter(&mut keep)) + } + + /// Each key becomes any number of images, each with the key's weight. + pub fn flat_map>( + &mut self, + s: &Stream>, + mut f: impl FnMut(&K) -> I + 'static, + ) -> Stream> { + self.unary(s, move |d: &Z| d.flat_map(&mut f)) + } + + /// Keys each element by `key`, giving the `(key, element)` pairs a join reads. + pub fn index_by( + &mut self, + s: &Stream>, + mut key: impl FnMut(&K) -> G + 'static, + ) -> Stream> { + self.unary(s, move |d: &Z| d.map(|k| (key(k), k.clone()))) + } + + /// Incremental `distinct`. + pub fn distinct(&mut self, s: &Stream>) -> Stream> { + let mut st = DistinctState::new(); + self.unary(s, move |d: &Z| st.step(d)) + } + + /// The bilinear delta join over in-memory traces. + pub fn join_indexed( + &mut self, + a: &Stream>, + b: &Stream>, + f: impl FnMut(&K, &A, &B) -> O + 'static, + ) -> Stream> + where + K: Ord + Clone + 'static, + A: Ord + Clone + 'static, + B: Ord + Clone + 'static, + O: Ord + 'static, + { + self.join_indexed_in(a, b, MemTrace::new(), MemTrace::new(), f) + } + + /// `join_indexed` over traces the caller chose. + pub fn join_indexed_in( + &mut self, + a: &Stream>, + b: &Stream>, + left: TA, + right: TB, + mut f: impl FnMut(&K, &A, &B) -> O + 'static, + ) -> Stream> + where + K: Ord + Clone + 'static, + A: Ord + Clone + 'static, + B: Ord + Clone + 'static, + O: Ord + 'static, + TA: Trace + 'static, + TB: Trace + 'static, + { + let mut st = JoinState::new(left, right); + self.binary(a, b, move |da: &Z<(K, A)>, db: &Z<(K, B)>| { + st.step(da, db, &mut f) + }) + } + + /// Incremental group-by over in-memory state. + pub fn aggregate( + &mut self, + s: &Stream>, + f: impl FnMut(&K, &[(V, i64)]) -> Option + 'static, + ) -> Stream> + where + K: Ord + Clone + 'static, + V: Ord + Clone + 'static, + R: Ord + 'static, + { + self.aggregate_in(s, MemTrace::new(), f) + } + + /// `aggregate` over a trace the caller chose. + pub fn aggregate_in( + &mut self, + s: &Stream>, + groups: T, + mut f: impl FnMut(&K, &[(V, i64)]) -> Option + 'static, + ) -> Stream> + where + K: Ord + Clone + 'static, + V: Ord + Clone + 'static, + R: Ord + 'static, + T: Trace + 'static, + { + let mut st = AggregateState::new(groups); + self.unary(s, move |d: &Z<(K, V)>| st.step(d, &mut f)) + } +} + +/// Whether a resumable job finished. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Progress { + Pending, + Done, +} + +/// Building from a snapshot as a sliced job: the snapshot enters an input a chunk at a +/// time, one circuit step per chunk. Outputs sum to what one big delta would give. +pub struct Rebuild { + input: Input, + rest: std::collections::btree_map::IntoIter, + chunk: usize, +} + +impl Rebuild { + /// `chunk` keys per circuit step; zero is read as one. + pub fn new(input: Input, snapshot: ZSet, chunk: usize) -> Self { + Rebuild { + input, + rest: snapshot.into_iter(), + chunk: chunk.max(1), + } + } + + /// Runs chunks until the snapshot is consumed or `expired()` says stop. It always + /// runs at least one chunk when any remain, so each call makes progress. `expired` + /// is a predicate, not an `Instant`: `Instant::now()` panics on wasm32. + pub fn step(&mut self, circuit: &mut Circuit, mut expired: impl FnMut() -> bool) -> Progress { + loop { + let chunk: ZSet = self.rest.by_ref().take(self.chunk).collect(); + if chunk.is_zero() { + return Progress::Done; + } + self.input.push(chunk); + circuit.step(); + if expired() { + return Progress::Pending; + } + } + } +} diff --git a/crates/zset/src/lib.rs b/crates/zset/src/lib.rs index 8d60c65a..eef8e806 100644 --- a/crates/zset/src/lib.rs +++ b/crates/zset/src/lib.rs @@ -1,6 +1,14 @@ //! Z-sets: maps from keys to non-zero `i64` weights (ADR 009 clause 1). An abelian group //! wherever no weight leaves `i64`; past that it panics. See `context/code/zset.md`. +mod circuit; +mod ops; +mod trace; + +pub use circuit::{Circuit, Input, Output, Progress, Rebuild, Stream}; +pub use ops::{AggregateState, DistinctState, JoinState}; +pub use trace::{MemTrace, Trace}; + use std::collections::BTreeMap; use std::collections::btree_map::Entry; use std::iter::FromIterator; @@ -116,6 +124,21 @@ impl ZSet { out } + /// Maps every key to any number of images, each carrying the key's weight; the + /// weights of equal images sum. + pub fn flat_map>( + &self, + mut f: impl FnMut(&K) -> I, + ) -> ZSet { + let mut out = ZSet::new(); + for (k, w) in &self.weights { + for image in f(k) { + out.insert(image, *w); + } + } + out + } + /// Keeps the keys `keep` accepts, with their weights. It looks at the key only, /// which is what makes it linear. pub fn filter(&self, mut keep: impl FnMut(&K) -> bool) -> Self diff --git a/crates/zset/src/ops.rs b/crates/zset/src/ops.rs new file mode 100644 index 00000000..7c2caec0 --- /dev/null +++ b/crates/zset/src/ops.rs @@ -0,0 +1,119 @@ +//! The incremental operators' state and step, with no wiring: each `step` takes the +//! delta of its inputs and returns the delta of its output. + +use crate::{Trace, ZSet, plus, times}; +use std::collections::BTreeMap; + +/// Incremental `distinct`: out = distinct(held + d) - distinct(held), per touched key. +#[derive(Clone, Debug, Default)] +pub struct DistinctState { + held: ZSet, +} + +impl DistinctState { + pub fn new() -> Self { + DistinctState { held: ZSet::new() } + } + + pub fn step(&mut self, d: &ZSet) -> ZSet { + let mut out = ZSet::new(); + for (k, w) in d.iter() { + let before = self.held.weight(k); + let after = plus(before, w); + out.insert(k.clone(), i64::from(after > 0) - i64::from(before > 0)); + self.held.insert(k.clone(), w); + } + out + } +} + +/// The bilinear delta join of two keyed streams: +/// out = da x b_prev + a_prev x db + da x db, never a rescan of either side. +#[derive(Clone, Debug, Default)] +pub struct JoinState { + left: TA, + right: TB, +} + +impl JoinState { + pub fn new(left: TA, right: TB) -> Self { + JoinState { left, right } + } + + pub fn step( + &mut self, + da: &ZSet<(K, A)>, + db: &ZSet<(K, B)>, + mut f: impl FnMut(&K, &A, &B) -> O, + ) -> ZSet + where + K: Ord + Clone, + A: Ord + Clone, + B: Ord + Clone, + O: Ord, + TA: Trace, + TB: Trace, + { + let mut out = ZSet::new(); + for ((k, a), wa) in da.iter() { + for (b, wb) in self.right.lookup(k) { + out.insert(f(k, a, &b), times(wa, wb)); + } + } + for ((k, a), wa) in da.iter() { + self.left.apply(k, a, wa); + } + for ((k, b), wb) in db.iter() { + for (a, wa) in self.left.lookup(k) { + out.insert(f(k, &a, b), times(wa, wb)); + } + } + for ((k, b), wb) in db.iter() { + self.right.apply(k, b, wb); + } + out + } +} + +/// Incremental group-by: `f` folds one key's integrated values to at most one result. +/// Only the touched keys are re-folded: out = -(k, f(old)) + (k, f(new)). +#[derive(Clone, Debug, Default)] +pub struct AggregateState { + groups: T, +} + +impl AggregateState { + pub fn new(groups: T) -> Self { + AggregateState { groups } + } + + pub fn step( + &mut self, + d: &ZSet<(K, V)>, + mut f: impl FnMut(&K, &[(V, i64)]) -> Option, + ) -> ZSet<(K, R)> + where + K: Ord + Clone, + V: Ord + Clone, + R: Ord, + T: Trace, + { + let mut touched: BTreeMap<&K, Vec<(&V, i64)>> = BTreeMap::new(); + for ((k, v), w) in d.iter() { + touched.entry(k).or_default().push((v, w)); + } + let mut out = ZSet::new(); + for (k, changes) in touched { + if let Some(r) = f(k, &self.groups.lookup(k)) { + out.insert((k.clone(), r), -1); + } + for (v, w) in changes { + self.groups.apply(k, v, w); + } + if let Some(r) = f(k, &self.groups.lookup(k)) { + out.insert((k.clone(), r), 1); + } + } + out + } +} diff --git a/crates/zset/src/trace.rs b/crates/zset/src/trace.rs new file mode 100644 index 00000000..c8c0a1a5 --- /dev/null +++ b/crates/zset/src/trace.rs @@ -0,0 +1,78 @@ +//! The trace interface: operator state behind look up, read a range, apply a delta. + +use crate::ZSet; +use std::collections::BTreeMap; +use std::ops::Bound; + +/// An integrated `ZSet<(K, V)>`, indexed by `K`. Reads return owned values, so an +/// implementation may page or compress what it holds. See `context/code/zset.md`. +pub trait Trace { + /// Adds `w` to the weight of `(k, v)`. + fn apply(&mut self, k: &K, v: &V, w: i64); + /// The values at `k` with their non-zero weights, ascending by value. + fn lookup(&self, k: &K) -> Vec<(V, i64)>; + /// Every `(key, value, weight)` with a key in the range, ascending. + fn range(&self, lo: Bound<&K>, hi: Bound<&K>) -> Vec<(K, V, i64)>; +} + +/// The in-memory trace: a sorted map from key to a Z-set of values. +#[derive(Clone, Debug)] +pub struct MemTrace { + groups: BTreeMap>, +} + +impl Default for MemTrace { + fn default() -> Self { + MemTrace { + groups: BTreeMap::new(), + } + } +} + +impl MemTrace { + pub fn new() -> Self { + Self::default() + } + + /// The number of keys with at least one value. + pub fn keys(&self) -> usize { + self.groups.len() + } +} + +fn is_empty_range(lo: Bound<&K>, hi: Bound<&K>) -> bool { + match (lo, hi) { + (Bound::Included(a), Bound::Included(b)) => a > b, + (Bound::Included(a) | Bound::Excluded(a), Bound::Included(b) | Bound::Excluded(b)) => { + a >= b + } + _ => false, + } +} + +impl Trace for MemTrace { + fn apply(&mut self, k: &K, v: &V, w: i64) { + let group = self.groups.entry(k.clone()).or_default(); + group.insert(v.clone(), w); + if group.is_zero() { + self.groups.remove(k); + } + } + + fn lookup(&self, k: &K) -> Vec<(V, i64)> { + self.groups + .get(k) + .map(|g| g.iter().map(|(v, w)| (v.clone(), w)).collect()) + .unwrap_or_default() + } + + fn range(&self, lo: Bound<&K>, hi: Bound<&K>) -> Vec<(K, V, i64)> { + if is_empty_range(lo, hi) { + return Vec::new(); + } + self.groups + .range::((lo, hi)) + .flat_map(|(k, g)| g.iter().map(move |(v, w)| (k.clone(), v.clone(), w))) + .collect() + } +} diff --git a/crates/zset/tests/circuit.rs b/crates/zset/tests/circuit.rs new file mode 100644 index 00000000..04eb0aa1 --- /dev/null +++ b/crates/zset/tests/circuit.rs @@ -0,0 +1,380 @@ +//! The circuit's operators against recomputing from scratch, and the DBSP identities. +//! Evidence: `context/test-evidence/zset.md`. + +use proptest::prelude::*; +use std::cell::Cell; +use std::ops::Bound; +use std::rc::Rc; +use zset::{ + AggregateState, Circuit, MemTrace, Progress, Rebuild, Stream, Trace, ZSet, differentiate, + integrate, +}; + +type Z = ZSet; + +fn pairs() -> impl Strategy> { + prop::collection::vec((0u8..6, -3i64..=3), 0..6) +} + +fn stream() -> impl Strategy>> { + prop::collection::vec(pairs().prop_map(|p| p.into_iter().collect::>()), 1..8) +} + +fn kv() -> impl Strategy>> { + let one = prop::collection::vec(((0u8..4, 0u8..3), -3i64..=3), 0..6) + .prop_map(|p| p.into_iter().collect::>()); + prop::collection::vec(one, 1..8) +} + +/// Feeds `steps` to one input, one circuit step each, and checks after every step that +/// the sum of the outputs so far is `reference` of the sum of the inputs so far. +fn matches_recompute( + steps: &[Z], + build: impl FnOnce(&mut Circuit, &Stream>) -> Stream>, + reference: impl Fn(&Z) -> Z, +) where + K: Ord + Clone + std::fmt::Debug + 'static, + L: Ord + Clone + std::fmt::Debug + 'static, +{ + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let out = build(&mut c, &s); + let sink = c.output(&out); + let (mut fed, mut got) = (Z::new(), Z::new()); + for (i, d) in steps.iter().enumerate() { + input.push(d.clone()); + c.step(); + fed += d; + got += sink.take(); + assert_eq!(got, reference(&fed), "[incremental] after step {i}"); + } +} + +fn sum_by_key(z: &Z<(u8, u8)>) -> Z<(u8, i64)> { + group(z, |vals| { + let s: i64 = vals.iter().map(|(v, w)| i64::from(*v) * w).sum(); + (s != 0).then_some(s) + }) +} + +/// The recompute-from-scratch aggregate: `f` over each key's integrated values. +fn group(z: &Z<(u8, u8)>, f: impl Fn(&[(u8, i64)]) -> Option) -> Z<(u8, i64)> { + let mut out = Z::new(); + for k in z + .iter() + .map(|((k, _), _)| *k) + .collect::>() + { + let vals: Vec<(u8, i64)> = z + .iter() + .filter(|((a, _), _)| *a == k) + .map(|((_, v), w)| (*v, w)) + .collect(); + if let Some(r) = f(&vals) { + out.insert((k, r), 1); + } + } + out +} + +fn max_present(vals: &[(u8, i64)]) -> Option { + vals.iter() + .filter(|(_, w)| *w > 0) + .map(|(v, _)| i64::from(*v)) + .max() +} + +fn join_ref(a: &Z<(u8, u8)>, b: &Z<(u8, u8)>) -> Z<(u8, u8, u8)> { + let mut out = Z::new(); + for ((k, x), wa) in a.iter() { + for ((k2, y), wb) in b.iter() { + if k == k2 { + out.insert((*k, *x, *y), wa * wb); + } + } + } + out +} + +fn join_circuit(da: &[Z<(u8, u8)>], db: &[Z<(u8, u8)>]) -> Vec> { + let mut c = Circuit::new(); + let (ia, a) = c.input::<(u8, u8)>(); + let (ib, b) = c.input::<(u8, u8)>(); + let j = c.join_indexed(&a, &b, |k, x, y| (*k, *x, *y)); + let sink = c.output(&j); + let mut deltas = Vec::new(); + for i in 0..da.len().max(db.len()) { + ia.push(da.get(i).cloned().unwrap_or_default()); + ib.push(db.get(i).cloned().unwrap_or_default()); + c.step(); + deltas.push(sink.take()); + } + deltas +} + +proptest! { + #[test] + fn map_matches_recompute(steps in stream()) { + matches_recompute(&steps, |c, s| c.map(s, |k| k % 3), |z| z.map(|k| k % 3)); + } + + #[test] + fn filter_matches_recompute(steps in stream()) { + matches_recompute(&steps, |c, s| c.filter(s, |k| k % 2 == 0), |z| z.filter(|k| k % 2 == 0)); + } + + #[test] + fn flat_map_matches_recompute(steps in stream()) { + let f = |k: &u8| vec![*k % 3, *k % 3 + 10, 20]; + matches_recompute(&steps, |c, s| c.flat_map(s, f), |z| z.flat_map(f)); + } + + #[test] + fn index_by_matches_recompute(steps in stream()) { + matches_recompute(&steps, |c, s| c.index_by(s, |k| k % 2), |z| z.map(|k| (k % 2, *k))); + } + + #[test] + fn distinct_matches_recompute(steps in stream()) { + matches_recompute(&steps, |c, s| c.distinct(s), |z| z.distinct()); + } + + #[test] + fn a_composed_pipeline_matches_recompute(steps in stream()) { + let pipeline = |z: &Z| z.filter(|k| *k != 3).map(|k| k % 3).distinct(); + matches_recompute( + &steps, + |c, s| { + let f = c.filter(s, |k| *k != 3); + let m = c.map(&f, |k| k % 3); + c.distinct(&m) + }, + pipeline, + ); + } + + #[test] + fn aggregate_sum_matches_recompute(steps in kv()) { + matches_recompute( + &steps, + |c, s| c.aggregate(s, |_, vals| { + let t: i64 = vals.iter().map(|(v, w)| i64::from(*v) * w).sum(); + (t != 0).then_some(t) + }), + sum_by_key, + ); + } + + #[test] + fn aggregate_max_matches_recompute(steps in kv()) { + matches_recompute( + &steps, + |c, s| c.aggregate(s, |_, vals| max_present(vals)), + |z| group(z, max_present), + ); + } + + #[test] + fn join_matches_recompute(da in kv(), db in kv()) { + let deltas = join_circuit(&da, &db); + let (mut a, mut b, mut got) = (Z::new(), Z::new(), Z::new()); + for (i, d) in deltas.iter().enumerate() { + a += da.get(i).cloned().unwrap_or_default(); + b += db.get(i).cloned().unwrap_or_default(); + got += d; + prop_assert_eq!(&got, &join_ref(&a, &b), "[join] after step {}", i); + } + } + + #[test] + fn join_is_bilinear(a1 in kv(), a2 in kv(), b1 in kv(), b2 in kv()) { + let total = |s: &[Z<(u8, u8)>]| s.iter().fold(Z::new(), |acc, d| acc + d.clone()); + let (a1, a2, b1, b2) = (total(&a1), total(&a2), total(&b1), total(&b2)); + prop_assert_eq!(join_ref(&(&a1 + &a2), &b1), &join_ref(&a1, &b1) + &join_ref(&a2, &b1), "[bilinear-left]"); + prop_assert_eq!(join_ref(&a1, &(&b1 + &b2)), &join_ref(&a1, &b1) + &join_ref(&a1, &b2), "[bilinear-right]"); + // and the circuit's output is bilinear too: one step of a+a' equals two of a, a'. + let once = join_circuit(&[&a1 + &a2], &[b1.clone()]); + let twice = join_circuit(&[a1.clone(), a2.clone()], &[b1.clone(), Z::new()]); + let sum_once = once.into_iter().fold(Z::new(), |x, d| x + d); + let sum_twice = twice.into_iter().fold(Z::new(), |x, d| x + d); + prop_assert_eq!(sum_once, sum_twice, "[bilinear-circuit]"); + } + + #[test] + fn integrate_after_differentiate_is_the_identity(steps in stream()) { + prop_assert_eq!(integrate(&differentiate(&steps)), steps.clone(), "[slice]"); + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let d = c.differentiate(&s); + let back = c.integrate(&d); + let (sink_s, sink_back) = (c.output(&s), c.output(&back)); + let (mut seen_s, mut seen_back) = (Vec::new(), Vec::new()); + for d in &steps { + input.push(d.clone()); + c.step(); + seen_s.push(s.current()); + seen_back.push(back.current()); + } + let _ = (sink_s, sink_back); + prop_assert_eq!(seen_back, seen_s, "[circuit-i-d]"); + } + + #[test] + fn differentiate_after_integrate_is_the_identity(steps in stream()) { + prop_assert_eq!(differentiate(&integrate(&steps)), steps.clone(), "[slice]"); + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let i = c.integrate(&s); + let back = c.differentiate(&i); + for (n, d) in steps.iter().enumerate() { + input.push(d.clone()); + c.step(); + prop_assert_eq!(back.current(), d.clone(), "[circuit-d-i] step {}", n); + } + } + + #[test] + fn delay_is_the_previous_step(steps in stream()) { + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let d = c.delay(&s); + let mut prev = Z::new(); + for (n, step) in steps.iter().enumerate() { + input.push(step.clone()); + c.step(); + prop_assert_eq!(d.current(), prev.clone(), "[delay] step {}", n); + prev = step.clone(); + } + } + + /// Whatever the deadline does, the rebuilt output is what one delta would give. + #[test] + fn a_resumable_rebuild_equals_an_uninterrupted_step( + snapshot in pairs(), + chunk in 0usize..4, + budget in 1usize..4, + ) { + let snapshot: Z = snapshot.into_iter().collect(); + let build = |c: &mut Circuit, s: &Stream>| { + let m = c.map(s, |k| k % 3); + c.distinct(&m) + }; + let mut whole = Circuit::new(); + let (input, s) = whole.input::(); + let out = build(&mut whole, &s); + let sink = whole.output(&out); + input.push(snapshot.clone()); + whole.step(); + let want = sink.take(); + + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let out = build(&mut c, &s); + let sink = c.output(&out); + let mut job = Rebuild::new(input, snapshot, chunk); + let mut got = Z::new(); + let mut calls = 0; + loop { + calls += 1; + prop_assert!(calls < 100, "[rebuild] never finished"); + let left = Cell::new(budget); + let progress = job.step(&mut c, || { + left.set(left.get() - 1); + left.get() == 0 + }); + got += sink.take(); + if progress == Progress::Done { break; } + } + prop_assert_eq!(got, want, "[rebuild]"); + } + + #[test] + fn a_trace_reads_what_was_applied(ops in prop::collection::vec(((0u8..5, 0u8..3), -3i64..=3), 0..12), lo in 0u8..6, hi in 0u8..6) { + let mut t = MemTrace::new(); + let mut reference: Z<(u8, u8)> = Z::new(); + for ((k, v), w) in &ops { + t.apply(k, v, *w); + reference.insert((*k, *v), *w); + } + for k in 0..5u8 { + let want: Vec<(u8, i64)> = reference.iter().filter(|((a, _), _)| *a == k).map(|((_, v), w)| (*v, w)).collect(); + prop_assert_eq!(t.lookup(&k), want, "[lookup] key {}", k); + } + let want: Vec<(u8, u8, i64)> = reference.iter().filter(|((a, _), _)| *a >= lo && *a < hi).map(|((a, v), w)| (*a, *v, w)).collect(); + prop_assert_eq!(t.range(Bound::Included(&lo), Bound::Excluded(&hi)), want, "[range]"); + let all: Vec<(u8, u8, i64)> = reference.iter().map(|((a, v), w)| (*a, *v, w)).collect(); + prop_assert_eq!(t.range(Bound::Unbounded, Bound::Unbounded), all, "[range-all]"); + prop_assert_eq!(t.keys(), reference.iter().map(|((a, _), _)| *a).collect::>().len()); + } +} + +#[test] +fn a_rebuild_that_is_out_of_time_reports_pending_after_one_chunk() { + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let sink = c.output(&s); + let mut job = Rebuild::new(input, (0..6u8).map(|k| (k, 1)).collect(), 2); + assert_eq!(job.step(&mut c, || true), Progress::Pending); + assert_eq!(sink.take().len(), 2, "exactly one chunk ran"); + assert_eq!(job.step(&mut c, || false), Progress::Done); + assert_eq!(sink.take().len(), 4); + assert_eq!(job.step(&mut c, || true), Progress::Done); +} + +#[test] +fn an_unbounded_range_that_is_empty_reads_nothing() { + let mut t = MemTrace::new(); + t.apply(&1u8, &1u8, 1); + assert!(t.range(Bound::Excluded(&1), Bound::Included(&1)).is_empty()); + assert!(t.range(Bound::Included(&2), Bound::Included(&1)).is_empty()); + assert_eq!(t.range(Bound::Included(&1), Bound::Included(&1)).len(), 1); +} + +#[test] +fn the_join_reads_state_through_the_trace_it_was_given() { + struct Counting(MemTrace, Rc>); + impl Trace for Counting { + fn apply(&mut self, k: &u8, v: &u8, w: i64) { + self.0.apply(k, v, w) + } + fn lookup(&self, k: &u8) -> Vec<(u8, i64)> { + self.1.set(self.1.get() + 1); + self.0.lookup(k) + } + fn range(&self, lo: Bound<&u8>, hi: Bound<&u8>) -> Vec<(u8, u8, i64)> { + self.0.range(lo, hi) + } + } + let looked = Rc::new(Cell::new(0)); + let mut c = Circuit::new(); + let (ia, a) = c.input::<(u8, u8)>(); + let (ib, b) = c.input::<(u8, u8)>(); + let j = c.join_indexed_in( + &a, + &b, + Counting(MemTrace::new(), looked.clone()), + MemTrace::new(), + |k, x, y| (*k, *x, *y), + ); + ia.push(Z::singleton((1, 1), 1)); + ib.push(Z::singleton((1, 2), 1)); + c.step(); + assert_eq!(j.current(), Z::singleton((1, 1, 2), 1)); + assert!(looked.get() > 0); +} + +#[test] +fn aggregate_state_refolds_only_the_touched_keys() { + let mut st = AggregateState::new(MemTrace::::new()); + let folds = Cell::new(0); + let f = |_: &u8, vals: &[(u8, i64)]| { + folds.set(folds.get() + 1); + Some(vals.len() as i64) + }; + let big: Z<(u8, u8)> = (0..50u8).map(|k| ((k, 0), 1)).collect(); + st.step(&big, f); + folds.set(0); + let out = st.step(&Z::singleton((7, 1), 1), f); + assert_eq!(folds.get(), 2, "one key, folded before and after"); + assert_eq!(out, Z::from_iter([((7, 1), -1), ((7, 2), 1)])); +}