fiddlerwoaroof/improvise
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'gh/main' into adr009/cell-delta

Browse Source
This commit is contained in:
Claude (adr009-step3)
2026-10-10 13:18:48 -07:00
committed by Edward Langley
parent 437f748a50 802db7e483
commit 70fd5ec9e0
10 changed files with 1617 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
context/code/zset.md
+82 -1
View File
@@ -38,7 +38,7 @@ and which a `HashMap` would give up (`context/code/core.md`, gotcha 18's argumen
The ADR is one paragraph on the crate, and says more about what it is for than what it is.
These are the readings this step takes.
1. **Clause 1 lists a `Circuit` and more operators; this step builds the pure part.** The
1. **Clause 1 lists a `Circuit` and more operators; step 1 built the pure part, and "The circuit" below built the rest (improvise-bw4p.1).** The
ADR's list is `delay`, `integrate`, `differentiate`, `map`, `filter`, `flat_map`,
`index_by`, `join_indexed`, `distinct` and `aggregate`, run by a synchronous `Circuit`,
plus resumable rebuilds and a trace interface. Step 1 as ticketed is the group and the
@@ -108,3 +108,84 @@ Weights come from `-3..=3`, an explicit zero, and `±2^40`, so sums of three can
The overflow policy is not a property over the generators (they cannot reach it by design)
and is tested by example in `src/lib.rs`, with `should_panic(expected = ..)`.
## The circuit (improvise-bw4p.1)
Clause 1's deferred scope, built without ADR 008. The ADR is not edited; this is how the crate reads it.
| item | what it is |
|---|---|
| `Circuit` | creation-order list of boxed nodes; `step()` runs each once |
| `Input<K>`, `Stream<ZSet<K>>`, `Output<K>` | push a delta in; the value on an edge; a sink that sums every step's delta until `take()` |
| `delay`, `integrate`, `differentiate`, `plus` | the stream operators as nodes |
| `map`, `filter`, `flat_map`, `index_by` | linear, per-element |
| `distinct` | incremental, state is the integrated input (`DistinctState`) |
| `join_indexed`, `join_indexed_in` | bilinear delta join of two `ZSet<(K, _)>` (`JoinState`) |
| `aggregate`, `aggregate_in` | group-by, re-folds only touched keys (`AggregateState`) |
| `Trace`, `MemTrace` | operator state behind lookup, range, apply |
| `Rebuild`, `Progress` | a snapshot loaded as a resumable job |
Decisions, where the ADR left a choice:
1. **Wiring is apart from logic.** `circuit.rs` is the only file with `Rc`, `RefCell` and
`Box<dyn Node>`. `ops.rs` holds each stateful operator as a plain struct with a pure
`step(delta) -> delta`, testable and usable without a circuit.
2. **Scheduling is creation order.** A node is built from streams that already exist, so
creation order is a topological order, and the circuit has no cycles: the ADR has no
recursion operator, and `integrate` is its own stateful node, not a delayed feedback loop.
`delay` is a node with one held Z-set. Not `Send`; one thread per circuit (clause 1).
3. **Incremental means the integral of the output is the operator of the integral of the
input.** That is the property every operator is tested against, over random delta streams.
`distinct` and `aggregate` are not linear, so they keep the integrated input and emit the
difference of the operator at the touched keys only.
4. **`index_by` is `map` to `(key, element)`.** A keyed stream is a `ZSet<(K, V)>`; there is no
separate indexed collection type, so the join and the aggregate read the same shape.
5. **The join** is `da x b_prev + a_prev x db + da x db`, computed as `da` against the right
trace, `da` applied to the left, then `db` against the updated left (which holds the cross
term), then `db` applied. Output weights are products and panic on `i64` overflow like
every weight (interpretation 5). `f(&K, &A, &B) -> O` builds the output key.
6. **`aggregate`'s fold gets the key and the group's `(value, weight)` pairs** (owned, because
a paged trace cannot lend references) and returns `Option<R>`; `None` means the group has no
row (an empty group, or a `Min` of nothing). Sum, count, min and max are closures; text
unanimity and the ADR's `ExactSum` belong to the caller (clause 4, 8), not this crate.
7. **The trace interface** is `apply(k, v, w)`, `lookup(k)` and `range(lo, hi)`, returning
owned vectors. `MemTrace` is a `BTreeMap<K, ZSet<V>>`; the `_in` constructors take any
`Trace`, which is the seam for paging or compression. An empty or inverted range reads
nothing (`BTreeMap::range` would panic).
8. **The resumable step takes a predicate, not an `Instant`, and a load is one logical tick.**
`Rebuild::step(circuit, expired)` loads `chunk` keys at a time and stops when `expired()` is
true, always running at least one chunk first, so every call progresses. It reports `Done`
as soon as the snapshot is exhausted, whatever `expired()` would say. `Instant::now()`
panics on wasm32 (`web-time` is a dependency this crate refuses), so the host closes over
its own clock. The first chunk is `Circuit::step`; the rest are `step_continue`, which does
not advance the clock. A circuit tick that arrives in pieces is exact for the incremental
operators because their outputs are deltas. For the temporal ones the pieces are shaped so
the outputs still sum to the one-step output: a continued `delay` emits nothing and adds to
what it holds, a continued `integrate` emits only its piece. An empty snapshot still takes
its one tick; a finished job takes none. Deltas pushed to the same input between calls
are absorbed into the next piece. Buffering live deltas during a rebuild (clause 7) is the
host's.
**`current()` is the whole tick.** A stream keeps the latest piece, which downstream
operators read, and the whole value of the tick, which `Stream::current` reads: pieces
within one tick are absorbed into it, so `integrate.current()` after a chunked load is the
full integral and a continued `delay` still reads the previous tick's input. `Output`
sums pieces as before.
**A pending rebuild owns the clock.** `Circuit::step` counts ticks, and a `Rebuild` records
the count after its first chunk. If a `Circuit::step` runs before the job is Done (the host
stepped between calls, or two jobs interleaved), the next `Rebuild::step` panics with
"the rebuild was interrupted"; `Rebuild::interrupted` asks without panicking, so a host
can refuse first. `step_continue` calls made by another job are not counted and are not
detected. The state a load leaves for the next tick is tested by pushing another delta
after Done and stepping twice.
9. **Not built:** `Circuit` does not own a job runner (ADR 008's), and nothing here is
serialised (clause 9, `improvise-protocol`).
The tests are `crates/zset/tests/circuit.rs`; the rivals and where they failed are in
`context/test-evidence/zset.md`.
A panic inside `step` (a weight overflow in an operator, say) leaves the circuit unusable:
nodes ahead of the panic have updated their state and the rest have not, and there is no
rollback. Drop the circuit and build another. ADR 009 clause 11's per-command snapshot and
restore is the host's job, not this crate's; `zset-ffi` poisons the circuit for this reason.
context/test-evidence/zset.md
+151
View File
@@ -278,3 +278,154 @@ the result must equal the exact value when it fits `i64`, and the operation must
does not. **Injected**, one at a time, `wrapping_add`, `wrapping_sub`, `wrapping_mul` and
`wrapping_neg` in place of the checked calls: this test failed under every one, and the
example tests named above failed under the matching one.
## `crates/zset/tests/circuit.rs`: the circuit
Each operator is checked against recomputing from scratch: a random stream of deltas goes in,
one circuit step each, and after every step the sum of the outputs so far must equal the
operator applied to the sum of the inputs so far. The failure message carries `[incremental]`,
`[join]`, `[rebuild]` and so on. Keys come from a range of six (four by three for the keyed
ones) and weights from `-3..=3`, so cancelling deltas and re-insertions are the common case.
**Injected in the source**, one at a time, each reverted and the suite rerun. The failing
tests, by name (a build that printed `Running` was confirmed each time; the first attempt at
the `delay` fault did not compile and was redone):
- `DistinctState` does not update its held set: `distinct_matches_recompute`,
`a_composed_pipeline_matches_recompute`, `a_resumable_rebuild_equals_an_uninterrupted_step`;
- the join applies `da` to the left trace after reading it for `db` (drops the `da x db`
term): `join_matches_recompute`, `join_is_bilinear`,
`the_join_reads_state_through_the_trace_it_was_given`;
- the join drops `da x b_prev`: `join_matches_recompute`, `join_is_bilinear`;
- `aggregate` does not retract the old result: `aggregate_sum_matches_recompute`,
`aggregate_max_matches_recompute`, `aggregate_state_refolds_only_the_touched_keys`;
- `delay` passes the current input: `delay_is_the_previous_step`, and both identities;
- `integrate` forgets history (`acc = d`): `integrate_after_differentiate_is_the_identity`,
`differentiate_after_integrate_is_the_identity`;
- `Rebuild` stops after the first chunk: `a_rebuild_that_is_out_of_time_reports_pending_after_one_chunk`,
`a_resumable_rebuild_equals_an_uninterrupted_step`;
- `Rebuild` checks the deadline before running any chunk (no progress is possible):
the same two;
- `MemTrace::range` ignores the upper bound, and `MemTrace` keeps an emptied key:
`a_trace_reads_what_was_applied`;
- `flat_map` gives every image weight 1: `flat_map_matches_recompute`.
### map_matches_recompute
`map`, `filter`, `index_by` and the pipeline test below are linear or compositions of
linear and `distinct`; they pin the circuit's wiring (the input node, the output sink,
creation-order scheduling) as much as the operators, which are `ZSet` methods tested in
`properties.rs`. No fault in them separates from the wiring faults above, and none was
needed.
### join_is_bilinear
Three tags. `[bilinear-left]` and `[bilinear-right]` check the reference join is bilinear, so
the oracle itself is a legitimate one; `[bilinear-circuit]` checks the incremental join: one
step of `a1 + a2` and two steps `a1` then `a2` (same `b`) sum to the same output. A join that
drops the `da x db` term fails the first-step, one-delta case, which is why the cross term is
exercised at all.
### a_resumable_rebuild_equals_an_uninterrupted_step
The snapshot enters a pipeline (`map` then `distinct`) in chunks of 1 to 3, and the deadline
predicate expires after 1 to 3 chunks per call, so the job is interrupted at varied points.
The sum of the outputs must equal one uninterrupted step over the whole snapshot. The loop is
bounded at 100 calls, so a job that never finishes is the named failure `[rebuild]`, not a hang.
### a_rebuild_that_is_out_of_time_reports_pending_after_one_chunk
A predicate that is always expired still lets exactly one chunk through (two of six keys),
then `Pending`; the rest completes on the next call; an exhausted job answers `Done` even
when out of time. This is the progress guarantee the property cannot see, because its budget
is never zero.
### aggregate_state_refolds_only_the_touched_keys
Fifty groups are loaded, then one delta touches one key: `f` is called twice (before and
after), not once per group. It is the O(delta) claim, in the one place a count can show it.
### an_unbounded_range_that_is_empty_reads_nothing
`range(Excluded(1), Included(1))` and an inverted range are empty. `BTreeMap::range` panics
on those; the trace guards it first. Not from the property, whose `lo > hi` draws are
covered by the same guard, but the equal-bounds corner is rare there.
### the_join_reads_state_through_the_trace_it_was_given
A counting `Trace` handed to `join_indexed_in` is looked up, so the join's state really is
behind the trace interface and not a private map beside it.
## `crates/zset/tests/circuit.rs`: review of #345
The Opus review of PR #345 at ba17f92b found four defects; each test below failed before its fix.
**Injected afterwards**, one at a time: `Rebuild` calling `step` for every chunk fails
`a_chunked_rebuild_is_one_logical_step_for_delay_integrate_and_differentiate` and
`a_chunked_temporal_rebuild_equals_one_step`; a continued `integrate` re-emitting its integral
fails the same two; a continued `delay` emitting what it holds fails the first; folding empty
groups (before or after) fails the two aggregate tests below; asking the deadline before
checking the job is done fails `the_last_chunk_is_done_even_when_the_deadline_has_expired`.
### a_chunked_rebuild_is_one_logical_step_for_delay_integrate_and_differentiate
Snapshot `{1, 2}` after one prior tick `{9}`, loaded in chunks of one with a deadline always
expired, against chunks of five. The three sinks (delay, integrate, differentiate) must sum
equal. Before the fix delay gave `{1:1}` where `{9:1}` was due, integrate `{1:2, 2:1}`.
### a_chunked_temporal_rebuild_equals_one_step
The same property over random prior ticks, snapshots, chunk sizes and budgets, through a
pipeline that combines all three temporal operators and ends in `distinct`.
### an_empty_snapshot_is_still_one_tick
An empty snapshot ticks the clock once (a `delay` shows the prior input), and a finished job
does not tick again.
### the_last_chunk_is_done_even_when_the_deadline_has_expired
One chunk exhausts the snapshot with an always-expired deadline: `Done`, not `Pending`.
### a_group_that_empties_leaves_no_phantom_row_for_a_count
A count closure, insert then delete then insert another: before the fix the delete step gave
phantom `(7, 0)` rows for the empty group, e.g. `{(7,0):-1, (7,1):1}` where `{(7,1):1}` was due.
### an_aggregate_that_returns_some_on_empty_input_matches_recompute
The same closure through the circuit against recompute-from-scratch. The earlier aggregate
properties used closures that return `None` on empty input, which could not see the defect.
`aggregate_state_refolds_only_the_touched_keys` now asserts both steps' fold counts and outputs.
## `crates/zset/tests/rebuild.rs`: delta review of #345 (0849981a)
**Injected**, one at a time: a continued `delay` that replaces what it holds fails
`every_operator_chunked_equals_one_step_and_so_do_the_ticks_after`; a continued `integrate`
that does not accumulate fails that and `a_chunked_temporal_rebuild_equals_one_step` (which
now pushes another delta after Done and steps twice, so the state left for later ticks is
compared); a `current()` that reads the last piece fails
`current_after_a_chunked_rebuild_is_the_whole_ticks_value`; an undetected split tick, or a
`step` that does not count ticks, fails both should-panic tests below. Before these, the
first two faults passed the whole suite.
### every_operator_chunked_equals_one_step_and_so_do_the_ticks_after
One pipeline with every public operator (delay, integrate, differentiate, index_by, join,
aggregate, distinct, flat_map), a prior tick, a chunked load, then two more ticks. The
load tick `[rebuild-tick]`, the next tick `[next-tick]` and the one after `[tick-after]` must
equal the single-step run. From the reviewer's property.
### current_after_a_chunked_rebuild_is_the_whole_ticks_value
`integrate.current()` after a load in chunks of one equals the one-chunk load.
### a_host_step_between_rebuild_calls_is_refused_loudly
A `Circuit::step` between two `Rebuild::step` calls makes the second panic.
### two_interleaved_rebuilds_are_refused_loudly
A second job's first chunk is a tick, so the first job's next call panics.
### a_finished_rebuild_and_an_interrupted_check_do_not_trip_on_later_ticks
A finished job is never interrupted, and steps afterwards do not make it panic.
crates/zset/src/circuit.rs
+476
View File
@@ -0,0 +1,476 @@
//! 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;
/// What an edge carries: the latest piece, and the whole value of the current tick.
/// They differ only while a tick arrives in pieces (`Circuit::step_continue`).
pub(crate) struct Slot<T> {
piece: T,
whole: T,
}
impl<T: Default> Default for Slot<T> {
fn default() -> Self {
Slot {
piece: T::default(),
whole: T::default(),
}
}
}
/// Values whose pieces of one tick add up to the tick's value.
pub trait Piecewise: Default + Clone {
fn absorb(&mut self, piece: &Self);
}
impl Piecewise for () {
fn absorb(&mut self, _: &Self) {}
}
impl<K: Ord + Clone> Piecewise for ZSet<K> {
fn absorb(&mut self, piece: &Self) {
*self += piece;
}
}
/// The value an edge carries during the current step.
pub struct Stream<T>(Rc<RefCell<Slot<T>>>);
impl<T: Piecewise> Stream<T> {
fn piece(&self) -> T {
self.0.borrow().piece.clone()
}
fn publish(&self, v: T, cont: bool) {
let mut slot = self.0.borrow_mut();
if cont {
slot.whole.absorb(&v);
} else {
slot.whole = v.clone();
}
slot.piece = v;
}
}
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 tick, all its pieces together (the zero before the first).
pub fn current(&self) -> T {
self.0.borrow().whole.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 {
/// `cont` is true for a later piece of a tick that has already begun.
fn eval(&mut self, cont: bool);
}
struct Unary<I, O, F> {
src: Stream<I>,
out: Stream<O>,
f: F,
}
impl<I: Piecewise, O: Piecewise, F: FnMut(&I) -> O> Node for Unary<I, O, F> {
fn eval(&mut self, cont: bool) {
let v = (self.f)(&self.src.piece());
self.out.publish(v, cont);
}
}
struct Binary<A, B, O, F> {
a: Stream<A>,
b: Stream<B>,
out: Stream<O>,
f: F,
}
impl<A: Piecewise, B: Piecewise, O: Piecewise, F: FnMut(&A, &B) -> O> Node for Binary<A, B, O, F> {
fn eval(&mut self, cont: bool) {
let v = (self.f)(&self.a.piece(), &self.b.piece());
self.out.publish(v, cont);
}
}
/// out(t) = in(t - 1). A continued piece adds to what is held and emits nothing.
struct Delay<K> {
src: Stream<ZSet<K>>,
out: Stream<ZSet<K>>,
held: ZSet<K>,
}
impl<K: Ord + Clone> Node for Delay<K> {
fn eval(&mut self, cont: bool) {
let d = self.src.piece();
let v = if cont {
self.held += &d;
ZSet::new()
} else {
std::mem::replace(&mut self.held, d)
};
self.out.publish(v, cont);
}
}
/// out(t) = in(0) + .. + in(t). A continued piece emits only its own input.
struct Integrate<K> {
src: Stream<ZSet<K>>,
out: Stream<ZSet<K>>,
acc: ZSet<K>,
}
impl<K: Ord + Clone> Node for Integrate<K> {
fn eval(&mut self, cont: bool) {
let d = self.src.piece();
self.acc += &d;
let v = if cont { d } else { self.acc.clone() };
self.out.publish(v, cont);
}
}
/// Moves what was pushed into the circuit's stream.
struct Source<K> {
pending: Rc<RefCell<ZSet<K>>>,
out: Stream<ZSet<K>>,
}
impl<K: Ord + Clone> Node for Source<K> {
fn eval(&mut self, cont: bool) {
let v = std::mem::take(&mut *self.pending.borrow_mut());
self.out.publish(v, cont);
}
}
/// 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>>,
ticks: u64,
}
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) {
self.ticks += 1;
self.run(false);
}
/// How many ticks `step` has begun. A `Rebuild` uses it to see a tick split.
pub fn ticks(&self) -> u64 {
self.ticks
}
/// Runs another piece of the tick the last `step` began; the clock does not advance.
/// Pieces sum to one `step` of their combined input (`context/code/zset.md`).
pub fn step_continue(&mut self) {
self.run(true);
}
fn run(&mut self, cont: bool) {
for n in &mut self.nodes {
n.eval(cont);
}
}
fn unary<I: Piecewise + 'static, O: Piecewise + '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: Piecewise + 'static, B: Piecewise + 'static, O: Piecewise + '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 + Clone + '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 out = Stream(Rc::default());
self.nodes.push(Box::new(Delay {
src: s.clone(),
out: out.clone(),
held: ZSet::new(),
}));
out
}
/// out(t) = in(0) + .. + in(t).
pub fn integrate<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Stream<Z<K>> {
let out = Stream(Rc::default());
self.nodes.push(Box::new(Integrate {
src: s.clone(),
out: out.clone(),
acc: ZSet::new(),
}));
out
}
/// 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 + Clone + 'static, L: Ord + Clone + '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 + Clone + 'static,
L: Ord + Clone + '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 + Clone + '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 + Clone + '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 + Clone + '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 + Clone + '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 + Clone + '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,
}
/// A snapshot load as a sliced job: one logical tick (`context/code/zset.md`). A pending
/// job owns the clock: a `Circuit::step` before it is Done makes `step` panic.
pub struct Rebuild<K> {
input: Input<K>,
rest: std::collections::btree_map::IntoIter<K, i64>,
chunk: usize,
/// The circuit's tick count right after this job's first chunk, once it has run one.
began_at: Option<u64>,
}
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),
began_at: None,
}
}
/// Whether a `Circuit::step` since this job began has split its tick. A finished job
/// cannot be interrupted.
pub fn interrupted(&self, circuit: &Circuit) -> bool {
self.rest.len() > 0 && self.began_at.is_some_and(|t| t != circuit.ticks())
}
/// Runs chunks until done or `expired()`, at least one per call. A predicate, not an
/// `Instant` (wasm32). An empty snapshot takes its one tick; a finished job none.
pub fn step(&mut self, circuit: &mut Circuit, mut expired: impl FnMut() -> bool) -> Progress {
assert!(
!self.interrupted(circuit),
"zset: the rebuild was interrupted by a Circuit::step before it finished"
);
loop {
let chunk: ZSet<K> = self.rest.by_ref().take(self.chunk).collect();
if chunk.is_zero() && self.began_at.is_some() {
return Progress::Done;
}
self.input.push(chunk);
if self.began_at.is_some() {
circuit.step_continue();
} else {
circuit.step();
self.began_at = Some(circuit.ticks());
}
if self.rest.len() == 0 {
return Progress::Done;
}
if expired() {
return Progress::Pending;
}
}
}
}
crates/zset/src/lib.rs
+23
View File
@@ -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
crates/zset/src/ops.rs
+124
View File
@@ -0,0 +1,124 @@
//! 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.
/// An empty group has no row, and `f` is not asked about it. Only 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 {
let before = self.groups.lookup(k);
if let Some(r) = Some(&before)
.filter(|g| !g.is_empty())
.and_then(|g| f(k, g))
{
out.insert((k.clone(), r), -1);
}
for (v, w) in changes {
self.groups.apply(k, v, w);
}
let after = self.groups.lookup(k);
if let Some(r) = Some(&after).filter(|g| !g.is_empty()).and_then(|g| f(k, g)) {
out.insert((k.clone(), r), 1);
}
}
out
}
}
crates/zset/src/trace.rs
+78
View File
@@ -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()
}
}
crates/zset/tests/circuit.proptest-regressions
+7
View File
@@ -0,0 +1,7 @@
# Seeds for failure cases proptest has generated in the past. It is
# automatically read and these particular cases re-run before any
# novel cases are generated.
#
# It is recommended to check this file in to source control so that
# everyone who runs the test benefits from these saved cases.
cc da6d39e3f2d54dc796842f642b0fc17d09a1491a9937c3d3f95d453d2d9487ed # shrinks to prior = [(3, -1)], snapshot = [(3, 2), (5, -2), (0, 1), (1, 1)], chunk = 0, budget = 1
crates/zset/tests/circuit.rs
+513
View File
@@ -0,0 +1,513 @@
//! 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], std::slice::from_ref(&b1));
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)]));
folds.set(0);
let out = st.step(&Z::singleton((7, 1), -1), f);
assert_eq!(folds.get(), 2, "the second step also folds only its key");
assert_eq!(out, Z::from_iter([((7, 2), -1), ((7, 1), 1)]));
}
fn temporal(c: &mut Circuit, s: &Stream<Z<u8>>) -> Stream<Z<u8>> {
let d = c.delay(s);
let i = c.integrate(s);
let df = c.differentiate(s);
let a = c.plus(&d, &i);
let b = c.plus(&a, &df);
c.distinct(&b)
}
/// The sums of three sinks over a snapshot loaded after one prior tick.
fn loaded(snapshot: &Z<u8>, chunk: usize, budget: usize) -> (Z<u8>, Z<u8>, Z<u8>) {
let mut c = Circuit::new();
let (input, s) = c.input::<u8>();
let (d, i, df) = (c.delay(&s), c.integrate(&s), c.differentiate(&s));
let sinks = (c.output(&d), c.output(&i), c.output(&df));
input.push(Z::singleton(9, 1));
c.step();
let _ = (sinks.0.take(), sinks.1.take(), sinks.2.take());
let mut job = Rebuild::new(input, snapshot.clone(), chunk);
for _ in 0..50 {
let left = Cell::new(budget);
let p = job.step(&mut c, || {
left.set(left.get() - 1);
left.get() == 0
});
if p == Progress::Done {
break;
}
}
(sinks.0.take(), sinks.1.take(), sinks.2.take())
}
#[test]
fn a_chunked_rebuild_is_one_logical_step_for_delay_integrate_and_differentiate() {
let snap: Z<u8> = [(1, 1), (2, 1)].into_iter().collect();
let want = loaded(&snap, 5, 5);
assert_eq!(
want.0,
Z::singleton(9, 1),
"delay shows the previous tick once"
);
assert_eq!(loaded(&snap, 1, 1), want, "[temporal-chunk]");
}
#[test]
fn an_empty_snapshot_is_still_one_tick() {
let mut c = Circuit::new();
let (input, s) = c.input::<u8>();
let d = c.delay(&s);
input.push(Z::singleton(1, 1));
c.step();
let mut job = Rebuild::new(input, Z::new(), 3);
assert_eq!(job.step(&mut c, || false), Progress::Done);
assert_eq!(d.current().len(), 1, "the empty load ticked the clock once");
assert_eq!(job.step(&mut c, || false), Progress::Done);
assert_eq!(d.current().len(), 1, "a finished job does not tick again");
}
#[test]
fn the_last_chunk_is_done_even_when_the_deadline_has_expired() {
let mut c = Circuit::new();
let (input, _s) = c.input::<u8>();
let mut job = Rebuild::new(input, Z::singleton(1, 1), 4);
assert_eq!(job.step(&mut c, || true), Progress::Done);
}
proptest! {
#[test]
fn a_chunked_temporal_rebuild_equals_one_step(
prior in pairs(), snapshot in pairs(), chunk in 0usize..4, budget in 1usize..4,
) {
let (prior, snapshot): (Z<u8>, Z<u8>) = (prior.into_iter().collect(), snapshot.into_iter().collect());
let run = |chunk: usize, budget: usize| {
let mut c = Circuit::new();
let (input, s) = c.input::<u8>();
let out = temporal(&mut c, &s);
let sink = c.output(&out);
input.push(prior.clone());
c.step();
let _ = sink.take();
let input_again = input.clone();
let mut job = Rebuild::new(input, snapshot.clone(), chunk);
for _ in 0..50 {
let left = Cell::new(budget);
if job.step(&mut c, || { left.set(left.get() - 1); left.get() == 0 }) == Progress::Done { break; }
}
let at = sink.take();
input_again.push(Z::singleton(5, 1));
c.step();
let next = sink.take();
c.step();
(at, next, sink.take())
};
prop_assert_eq!(run(chunk, budget), run(100, 100), "[temporal-rebuild]");
}
}
#[test]
fn a_group_that_empties_leaves_no_phantom_row_for_a_count() {
let mut st = AggregateState::new(MemTrace::<u8, u8>::new());
let count = |_: &u8, vals: &[(u8, i64)]| Some(vals.len() as i64);
assert_eq!(
st.step(&Z::singleton((7, 1), 1), count),
Z::singleton((7, 1), 1)
);
assert_eq!(
st.step(&Z::singleton((7, 1), -1), count),
Z::singleton((7, 1), -1)
);
assert_eq!(
st.step(&Z::singleton((7, 2), 1), count),
Z::singleton((7, 1), 1)
);
}
#[test]
fn an_aggregate_that_returns_some_on_empty_input_matches_recompute() {
let steps = vec![
Z::singleton((7u8, 1u8), 1),
Z::singleton((7, 1), -1),
Z::singleton((7, 2), 1),
];
matches_recompute(
&steps,
|c, s| c.aggregate(s, |_, vals| Some(vals.len() as i64)),
|z| group(z, |vals| Some(vals.len() as i64)),
);
}
crates/zset/tests/rebuild.proptest-regressions
+7
View File
@@ -0,0 +1,7 @@
# Seeds for failure cases proptest has generated in the past. It is
# automatically read and these particular cases re-run before any
# novel cases are generated.
#
# It is recommended to check this file in to source control so that
# everyone who runs the test benefits from these saved cases.
cc b276367684808122e5f3fedb8e87a6ab970895f3efce9e6b98a40ade2cdb7227 # shrinks to prior = ZSet { weights: {} }, snap = ZSet { weights: {0: -1, 1: -1} }, post = ZSet { weights: {} }, chunk = 0, budget = 1
crates/zset/tests/rebuild.rs
+156
View File
@@ -0,0 +1,156 @@
//! A rebuild across every public operator: the tick it loads, the ticks after it, and `current()`.
//! Evidence: `context/test-evidence/zset.md`.
use proptest::prelude::*;
use std::cell::Cell;
use zset::{Circuit, Input, Progress, Rebuild, ZSet};
type Z<K> = ZSet<K>;
fn pairs() -> impl Strategy<Value = Z<u8>> {
prop::collection::vec((0u8..6, -3i64..=3), 0..6).prop_map(|p| p.into_iter().collect())
}
/// A pipeline using every operator; returns the input and a closure that drains all sinks.
fn build(c: &mut Circuit) -> (Input<u8>, Box<dyn Fn() -> String>) {
let (input, s) = c.input::<u8>();
let d = c.delay(&s);
let i = c.integrate(&s);
let df = c.differentiate(&s);
let a = c.index_by(&s, |k| k % 2);
let b = c.index_by(&i, |k| k % 2);
let j = c.join_indexed(&a, &b, |k, x, y| (*k, *x, *y));
let ij = c.integrate(&j);
let g = c.index_by(&df, |k| k % 3);
let agg = c.aggregate(&g, |_, vals| {
Some(vals.iter().map(|(_, w)| *w).sum::<i64>())
});
let dagg = c.delay(&agg);
let dist = c.distinct(&i);
let ddist = c.differentiate(&dist);
let fm = c.flat_map(&d, |k| vec![*k, k + 1]);
let sinks = (
c.output(&d),
c.output(&i),
c.output(&df),
c.output(&j),
c.output(&ij),
c.output(&agg),
c.output(&dagg),
c.output(&ddist),
c.output(&fm),
);
let drain = move || {
format!(
"d={:?} i={:?} df={:?} j={:?} ij={:?} agg={:?} dagg={:?} ddist={:?} fm={:?}",
sinks.0.take(),
sinks.1.take(),
sinks.2.take(),
sinks.3.take(),
sinks.4.take(),
sinks.5.take(),
sinks.6.take(),
sinks.7.take(),
sinks.8.take()
)
};
(input, Box::new(drain))
}
fn run(
prior: &Z<u8>,
snap: &Z<u8>,
post: &Z<u8>,
chunk: usize,
budget: usize,
) -> (String, String, String) {
let mut c = Circuit::new();
let (input, drain) = build(&mut c);
input.push(prior.clone());
c.step();
let _ = drain();
let mut job = Rebuild::new(input.clone(), snap.clone(), chunk);
let mut done = false;
for _ in 0..50 {
let left = Cell::new(budget);
if job.step(&mut c, || {
left.set(left.get() - 1);
left.get() == 0
}) == Progress::Done
{
done = true;
break;
}
}
assert!(done, "[rebuild] never finished");
let at = drain();
input.push(post.clone());
c.step();
let next = drain();
c.step();
(at, next, drain())
}
proptest! {
#[test]
fn every_operator_chunked_equals_one_step_and_so_do_the_ticks_after(
prior in pairs(), snap in pairs(), post in pairs(), chunk in 0usize..4, budget in 1usize..4,
) {
let want = run(&prior, &snap, &post, 100, 100);
let got = run(&prior, &snap, &post, chunk, budget);
prop_assert_eq!(&got.0, &want.0, "[rebuild-tick]");
prop_assert_eq!(&got.1, &want.1, "[next-tick]");
prop_assert_eq!(&got.2, &want.2, "[tick-after]");
}
}
#[test]
fn current_after_a_chunked_rebuild_is_the_whole_ticks_value() {
let snap: Z<u8> = [(1, 1), (2, 1)].into_iter().collect();
let cur = |chunk| {
let mut c = Circuit::new();
let (input, s) = c.input::<u8>();
let i = c.integrate(&s);
let mut job = Rebuild::new(input, snap.clone(), chunk);
while job.step(&mut c, || false) == Progress::Pending {}
i.current()
};
assert_eq!(cur(1), cur(10), "[current]");
}
#[test]
#[should_panic(expected = "rebuild was interrupted")]
fn a_host_step_between_rebuild_calls_is_refused_loudly() {
let snap: Z<u8> = [(1, 1), (2, 1)].into_iter().collect();
let mut c = Circuit::new();
let (input, _s) = c.input::<u8>();
let mut job = Rebuild::new(input, snap, 1);
assert_eq!(job.step(&mut c, || true), Progress::Pending);
c.step();
job.step(&mut c, || false);
}
#[test]
#[should_panic(expected = "rebuild was interrupted")]
fn two_interleaved_rebuilds_are_refused_loudly() {
let snap: Z<u8> = [(1, 1), (2, 1)].into_iter().collect();
let mut c = Circuit::new();
let (i1, _s1) = c.input::<u8>();
let (i2, _s2) = c.input::<u8>();
let (mut a, mut b) = (Rebuild::new(i1, snap.clone(), 1), Rebuild::new(i2, snap, 1));
assert_eq!(a.step(&mut c, || true), Progress::Pending);
assert_eq!(b.step(&mut c, || true), Progress::Pending);
a.step(&mut c, || false);
}
#[test]
fn a_finished_rebuild_and_an_interrupted_check_do_not_trip_on_later_ticks() {
let snap: Z<u8> = [(1, 1)].into_iter().collect();
let mut c = Circuit::new();
let (input, _s) = c.input::<u8>();
let mut job = Rebuild::new(input, snap, 1);
assert_eq!(job.step(&mut c, || true), Progress::Done);
c.step();
assert!(!job.interrupted(&c));
assert_eq!(job.step(&mut c, || true), Progress::Done);
}