feat(zset): the circuit, incremental operators, traces and resumable rebuild (improvise-bw4p.1)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze
This commit is contained in:
Claude (dbsp-circuit)
committed by
Edward Langley
Edward Langley
parent
0e78a97e99
commit
42ee9bac98
@@ -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<T>(Rc<RefCell<T>>);
|
||||
|
||||
impl<T> Clone for Stream<T> {
|
||||
fn clone(&self) -> Self {
|
||||
Stream(Rc::clone(&self.0))
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Clone> Stream<T> {
|
||||
/// 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<K>(Rc<RefCell<ZSet<K>>>);
|
||||
|
||||
impl<K> Clone for Input<K> {
|
||||
fn clone(&self) -> Self {
|
||||
Input(Rc::clone(&self.0))
|
||||
}
|
||||
}
|
||||
|
||||
impl<K: Ord> Input<K> {
|
||||
pub fn push(&self, delta: ZSet<K>) {
|
||||
*self.0.borrow_mut() += delta;
|
||||
}
|
||||
}
|
||||
|
||||
/// A sink that sums every step's delta until it is taken.
|
||||
pub struct Output<K>(Rc<RefCell<ZSet<K>>>);
|
||||
|
||||
impl<K: Ord> Output<K> {
|
||||
pub fn take(&self) -> ZSet<K> {
|
||||
std::mem::take(&mut *self.0.borrow_mut())
|
||||
}
|
||||
}
|
||||
|
||||
trait Node {
|
||||
fn eval(&mut self);
|
||||
}
|
||||
|
||||
struct Unary<I, O, F> {
|
||||
src: Stream<I>,
|
||||
out: Stream<O>,
|
||||
f: F,
|
||||
}
|
||||
|
||||
impl<I, O, F: FnMut(&I) -> O> Node for Unary<I, O, F> {
|
||||
fn eval(&mut self) {
|
||||
let v = (self.f)(&self.src.0.borrow());
|
||||
*self.out.0.borrow_mut() = v;
|
||||
}
|
||||
}
|
||||
|
||||
struct Binary<A, B, O, F> {
|
||||
a: Stream<A>,
|
||||
b: Stream<B>,
|
||||
out: Stream<O>,
|
||||
f: F,
|
||||
}
|
||||
|
||||
impl<A, B, O, F: FnMut(&A, &B) -> O> Node for Binary<A, B, O, F> {
|
||||
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<K> {
|
||||
pending: Rc<RefCell<ZSet<K>>>,
|
||||
out: Stream<ZSet<K>>,
|
||||
}
|
||||
|
||||
impl<K: Ord> Node for Source<K> {
|
||||
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<Box<dyn Node>>,
|
||||
}
|
||||
|
||||
type Z<K> = ZSet<K>;
|
||||
|
||||
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<I: 'static, O: Default + 'static>(
|
||||
&mut self,
|
||||
src: &Stream<I>,
|
||||
f: impl FnMut(&I) -> O + 'static,
|
||||
) -> Stream<O> {
|
||||
let out = Stream(Rc::default());
|
||||
self.nodes.push(Box::new(Unary {
|
||||
src: src.clone(),
|
||||
out: out.clone(),
|
||||
f,
|
||||
}));
|
||||
out
|
||||
}
|
||||
|
||||
fn binary<A: 'static, B: 'static, O: Default + 'static>(
|
||||
&mut self,
|
||||
a: &Stream<A>,
|
||||
b: &Stream<B>,
|
||||
f: impl FnMut(&A, &B) -> O + 'static,
|
||||
) -> Stream<O> {
|
||||
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<K: Ord + 'static>(&mut self) -> (Input<K>, Stream<Z<K>>) {
|
||||
let pending: Rc<RefCell<Z<K>>> = 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<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Output<K> {
|
||||
let sink = Output(Rc::default());
|
||||
let acc = Rc::clone(&sink.0);
|
||||
self.unary(s, move |d: &Z<K>| {
|
||||
*acc.borrow_mut() += d;
|
||||
});
|
||||
sink
|
||||
}
|
||||
|
||||
/// out(t) = in(t - 1); out(0) is zero.
|
||||
pub fn delay<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Stream<Z<K>> {
|
||||
let mut held = ZSet::new();
|
||||
self.unary(s, move |d: &Z<K>| std::mem::replace(&mut held, d.clone()))
|
||||
}
|
||||
|
||||
/// out(t) = in(0) + .. + in(t).
|
||||
pub fn integrate<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Stream<Z<K>> {
|
||||
let mut acc = ZSet::new();
|
||||
self.unary(s, move |d: &Z<K>| {
|
||||
acc += d;
|
||||
acc.clone()
|
||||
})
|
||||
}
|
||||
|
||||
/// out(t) = in(t) - in(t - 1).
|
||||
pub fn differentiate<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Stream<Z<K>> {
|
||||
let prev = self.delay(s);
|
||||
self.binary(s, &prev, |a: &Z<K>, b: &Z<K>| a - b)
|
||||
}
|
||||
|
||||
pub fn plus<K: Ord + Clone + 'static>(
|
||||
&mut self,
|
||||
a: &Stream<Z<K>>,
|
||||
b: &Stream<Z<K>>,
|
||||
) -> Stream<Z<K>> {
|
||||
self.binary(a, b, |a: &Z<K>, b: &Z<K>| a + b)
|
||||
}
|
||||
|
||||
pub fn map<K: Ord + 'static, L: Ord + 'static>(
|
||||
&mut self,
|
||||
s: &Stream<Z<K>>,
|
||||
mut f: impl FnMut(&K) -> L + 'static,
|
||||
) -> Stream<Z<L>> {
|
||||
self.unary(s, move |d: &Z<K>| d.map(&mut f))
|
||||
}
|
||||
|
||||
pub fn filter<K: Ord + Clone + 'static>(
|
||||
&mut self,
|
||||
s: &Stream<Z<K>>,
|
||||
mut keep: impl FnMut(&K) -> bool + 'static,
|
||||
) -> Stream<Z<K>> {
|
||||
self.unary(s, move |d: &Z<K>| d.filter(&mut keep))
|
||||
}
|
||||
|
||||
/// Each key becomes any number of images, each with the key's weight.
|
||||
pub fn flat_map<K: Ord + 'static, L: Ord + 'static, I: IntoIterator<Item = L>>(
|
||||
&mut self,
|
||||
s: &Stream<Z<K>>,
|
||||
mut f: impl FnMut(&K) -> I + 'static,
|
||||
) -> Stream<Z<L>> {
|
||||
self.unary(s, move |d: &Z<K>| d.flat_map(&mut f))
|
||||
}
|
||||
|
||||
/// Keys each element by `key`, giving the `(key, element)` pairs a join reads.
|
||||
pub fn index_by<K: Ord + Clone + 'static, G: Ord + 'static>(
|
||||
&mut self,
|
||||
s: &Stream<Z<K>>,
|
||||
mut key: impl FnMut(&K) -> G + 'static,
|
||||
) -> Stream<Z<(G, K)>> {
|
||||
self.unary(s, move |d: &Z<K>| d.map(|k| (key(k), k.clone())))
|
||||
}
|
||||
|
||||
/// Incremental `distinct`.
|
||||
pub fn distinct<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Stream<Z<K>> {
|
||||
let mut st = DistinctState::new();
|
||||
self.unary(s, move |d: &Z<K>| st.step(d))
|
||||
}
|
||||
|
||||
/// The bilinear delta join over in-memory traces.
|
||||
pub fn join_indexed<K, A, B, O>(
|
||||
&mut self,
|
||||
a: &Stream<Z<(K, A)>>,
|
||||
b: &Stream<Z<(K, B)>>,
|
||||
f: impl FnMut(&K, &A, &B) -> O + 'static,
|
||||
) -> Stream<Z<O>>
|
||||
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<K, A, B, O, TA, TB>(
|
||||
&mut self,
|
||||
a: &Stream<Z<(K, A)>>,
|
||||
b: &Stream<Z<(K, B)>>,
|
||||
left: TA,
|
||||
right: TB,
|
||||
mut f: impl FnMut(&K, &A, &B) -> O + 'static,
|
||||
) -> Stream<Z<O>>
|
||||
where
|
||||
K: Ord + Clone + 'static,
|
||||
A: Ord + Clone + 'static,
|
||||
B: Ord + Clone + 'static,
|
||||
O: Ord + 'static,
|
||||
TA: Trace<K, A> + 'static,
|
||||
TB: Trace<K, B> + '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<K, V, R>(
|
||||
&mut self,
|
||||
s: &Stream<Z<(K, V)>>,
|
||||
f: impl FnMut(&K, &[(V, i64)]) -> Option<R> + 'static,
|
||||
) -> Stream<Z<(K, R)>>
|
||||
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<K, V, R, T>(
|
||||
&mut self,
|
||||
s: &Stream<Z<(K, V)>>,
|
||||
groups: T,
|
||||
mut f: impl FnMut(&K, &[(V, i64)]) -> Option<R> + 'static,
|
||||
) -> Stream<Z<(K, R)>>
|
||||
where
|
||||
K: Ord + Clone + 'static,
|
||||
V: Ord + Clone + 'static,
|
||||
R: Ord + 'static,
|
||||
T: Trace<K, V> + '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<K> {
|
||||
input: Input<K>,
|
||||
rest: std::collections::btree_map::IntoIter<K, i64>,
|
||||
chunk: usize,
|
||||
}
|
||||
|
||||
impl<K: Ord> Rebuild<K> {
|
||||
/// `chunk` keys per circuit step; zero is read as one.
|
||||
pub fn new(input: Input<K>, snapshot: ZSet<K>, 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<K> = 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<K: Ord> ZSet<K> {
|
||||
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<L: Ord, I: IntoIterator<Item = L>>(
|
||||
&self,
|
||||
mut f: impl FnMut(&K) -> I,
|
||||
) -> ZSet<L> {
|
||||
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
|
||||
|
||||
@@ -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<K> {
|
||||
held: ZSet<K>,
|
||||
}
|
||||
|
||||
impl<K: Ord + Clone> DistinctState<K> {
|
||||
pub fn new() -> Self {
|
||||
DistinctState { held: ZSet::new() }
|
||||
}
|
||||
|
||||
pub fn step(&mut self, d: &ZSet<K>) -> ZSet<K> {
|
||||
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<TA, TB> {
|
||||
left: TA,
|
||||
right: TB,
|
||||
}
|
||||
|
||||
impl<TA, TB> JoinState<TA, TB> {
|
||||
pub fn new(left: TA, right: TB) -> Self {
|
||||
JoinState { left, right }
|
||||
}
|
||||
|
||||
pub fn step<K, A, B, O>(
|
||||
&mut self,
|
||||
da: &ZSet<(K, A)>,
|
||||
db: &ZSet<(K, B)>,
|
||||
mut f: impl FnMut(&K, &A, &B) -> O,
|
||||
) -> ZSet<O>
|
||||
where
|
||||
K: Ord + Clone,
|
||||
A: Ord + Clone,
|
||||
B: Ord + Clone,
|
||||
O: Ord,
|
||||
TA: Trace<K, A>,
|
||||
TB: Trace<K, B>,
|
||||
{
|
||||
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<T> {
|
||||
groups: T,
|
||||
}
|
||||
|
||||
impl<T> AggregateState<T> {
|
||||
pub fn new(groups: T) -> Self {
|
||||
AggregateState { groups }
|
||||
}
|
||||
|
||||
pub fn step<K, V, R>(
|
||||
&mut self,
|
||||
d: &ZSet<(K, V)>,
|
||||
mut f: impl FnMut(&K, &[(V, i64)]) -> Option<R>,
|
||||
) -> ZSet<(K, R)>
|
||||
where
|
||||
K: Ord + Clone,
|
||||
V: Ord + Clone,
|
||||
R: Ord,
|
||||
T: Trace<K, V>,
|
||||
{
|
||||
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
|
||||
}
|
||||
}
|
||||
@@ -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<K, V> {
|
||||
/// 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<K, V> {
|
||||
groups: BTreeMap<K, ZSet<V>>,
|
||||
}
|
||||
|
||||
impl<K, V> Default for MemTrace<K, V> {
|
||||
fn default() -> Self {
|
||||
MemTrace {
|
||||
groups: BTreeMap::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<K: Ord, V: Ord> MemTrace<K, V> {
|
||||
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<K: Ord>(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<K: Ord + Clone, V: Ord + Clone> Trace<K, V> for MemTrace<K, V> {
|
||||
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::<K, _>((lo, hi))
|
||||
.flat_map(|(k, g)| g.iter().map(move |(v, w)| (k.clone(), v.clone(), w)))
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
@@ -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<K> = ZSet<K>;
|
||||
|
||||
fn pairs() -> impl Strategy<Value = Vec<(u8, i64)>> {
|
||||
prop::collection::vec((0u8..6, -3i64..=3), 0..6)
|
||||
}
|
||||
|
||||
fn stream() -> impl Strategy<Value = Vec<Z<u8>>> {
|
||||
prop::collection::vec(pairs().prop_map(|p| p.into_iter().collect::<Z<u8>>()), 1..8)
|
||||
}
|
||||
|
||||
fn kv() -> impl Strategy<Value = Vec<Z<(u8, u8)>>> {
|
||||
let one = prop::collection::vec(((0u8..4, 0u8..3), -3i64..=3), 0..6)
|
||||
.prop_map(|p| p.into_iter().collect::<Z<(u8, u8)>>());
|
||||
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<K, L>(
|
||||
steps: &[Z<K>],
|
||||
build: impl FnOnce(&mut Circuit, &Stream<Z<K>>) -> Stream<Z<L>>,
|
||||
reference: impl Fn(&Z<K>) -> Z<L>,
|
||||
) 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::<K>();
|
||||
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<i64>) -> Z<(u8, i64)> {
|
||||
let mut out = Z::new();
|
||||
for k in z
|
||||
.iter()
|
||||
.map(|((k, _), _)| *k)
|
||||
.collect::<std::collections::BTreeSet<_>>()
|
||||
{
|
||||
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<i64> {
|
||||
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<Z<(u8, u8, u8)>> {
|
||||
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<u8>| 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::<u8>();
|
||||
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::<u8>();
|
||||
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::<u8>();
|
||||
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<u8> = snapshot.into_iter().collect();
|
||||
let build = |c: &mut Circuit, s: &Stream<Z<u8>>| {
|
||||
let m = c.map(s, |k| k % 3);
|
||||
c.distinct(&m)
|
||||
};
|
||||
let mut whole = Circuit::new();
|
||||
let (input, s) = whole.input::<u8>();
|
||||
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::<u8>();
|
||||
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::<std::collections::BTreeSet<_>>().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::<u8>();
|
||||
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<u8, u8>, Rc<Cell<usize>>);
|
||||
impl Trace<u8, u8> 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::<u8, u8>::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)]));
|
||||
}
|
||||
Reference in New Issue
Block a user