From 9d675c9c33fdf1629d0c6285c4bf54ebf9a34c3a Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-sources)" Date: Sat, 10 Oct 2026 18:12:04 -0700 Subject: [PATCH 1/3] feat(incremental): a backend over a source that reports its changes (improvise-rnak) Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- context/code/improvise-incremental.md | 39 ++++ context/code/source.md | 9 + .../test-evidence/improvise-incremental.md | 41 ++++ crates/improvise-core/src/model/source.rs | 12 +- crates/improvise-incremental/src/eager.rs | 4 +- crates/improvise-incremental/src/formulas.rs | 2 +- .../improvise-incremental/src/incremental.rs | 69 +++++- crates/improvise-incremental/src/shapes.rs | 19 +- .../tests/live_source.rs | 213 ++++++++++++++++++ .../tests/random_live.proptest-regressions | 7 + .../tests/random_live.rs | 22 ++ .../tests/support/live.rs | 156 +++++++++++++ .../tests/support/mod.rs | 2 + 13 files changed, 571 insertions(+), 24 deletions(-) create mode 100644 crates/improvise-incremental/tests/live_source.rs create mode 100644 crates/improvise-incremental/tests/random_live.proptest-regressions create mode 100644 crates/improvise-incremental/tests/random_live.rs create mode 100644 crates/improvise-incremental/tests/support/live.rs diff --git a/context/code/improvise-incremental.md b/context/code/improvise-incremental.md index f3604067..0791e1de 100644 --- a/context/code/improvise-incremental.md +++ b/context/code/improvise-incremental.md @@ -388,3 +388,42 @@ that say otherwise (step 4 items 4 and 10, step 5 item 5). Each is a user-visibl - **Overflow of the exact sum to infinity** follows `ExactSum::value` (round to nearest, so an overflowing total is infinite): correct IEEE behaviour. + +## Step 7: live sources (improvise-rnak) + +ADR 009 clause 2 names two kinds of live source: one that reports its changes, which feeds cell deltas +like `DataStore`, and one that answers aggregate queries but reports none, which implements the derived +value by pushdown. The core half of both is in `context/code/source.md` (`CellSource`, `Model::over`). +The brief ordered this half after the evaluator hook; the hook came first because a source that reports +changes still has its formulas evaluated over it. User ruling 2026-10-10: formulas are computed in RAM +from the source's folds, never pushed down; `Derived` stays in this crate. + +### Change-reporting (`Incremental::over`) + +1. **`ChangeSource` is `CellSource` plus `drain_changes(&self)`, and a source's cells are those as of the + last drain.** A shape is built lazily, by a pass over the source, at its first read. If a source + showed a write before reporting it, a shape built in between would hold the write and the next + drain would add it again. So the contract is that nothing is visible until it is drained, which is + what a database does inside a read transaction and what `DataStore` does already (its log is + drained in the step that made the edit). `&self` because a live source is shared with the model that + reads it and changes behind it by interior mutability of its own. +2. **`Incremental::over(model, feed, none_cats)`**: the model is put over the feed (`Model::over`) and + keeps no cells of its own. Its categories, formulas and aggregations are the definition; the + source holds the data. Shapes and stems are built from the feed as it stands. +3. **`sync()` drains and maintains; `apply_changes(&[CellChange])` maintains changes the source already + reflects**, for a source whose changes reach the backend by another route (a trigger table, a + channel). `Maintained::apply` on a fed backend takes definition edits (a formula, an aggregation) + and drains: a cell edit is the source's to make, and `Model::set_cell` refuses it in a debug build. +4. **Freshness is the source's.** ADR 003 left "polled, triggered, or stale until refreshed" open: `sync` + is the poll, and when it is called is the host's decision. Nothing here schedules it. +5. **What a change costs** is what it costs for the owned store: two group updates and two exact-sum + updates per maintained shape, no pass over the source (`the_cost_of_a_reported_change_...` counts the + rows the source serves for snapshots: zero after the build). Formulas are evaluated at the stems the + change reaches, as in step 5. +6. **Three routes into the store were routed through the source**: `Shape::of` (the build) reads + `cells(&[])`, `Stems::build` reads `keys(&[])`, and `Incremental::read` reads `value_at`. `Eager::occupancy` + reads `keys(&[])` too, so an `Eager` over a source would answer it. + +`tests/support/live.rs` holds `LiveTable`, the scripted source (an indexed store behind a lock, with +a pending list and a counter of snapshot rows) and `Fed`, an `Incremental` over one that a differential +can drive: a cell edit becomes a write to the table followed by `sync`. diff --git a/context/code/source.md b/context/code/source.md index 80a3b3c5..7ed4d1d1 100644 --- a/context/code/source.md +++ b/context/code/source.md @@ -87,3 +87,12 @@ implementation, written as a query engine with no index would answer, sharing on `DataStore`. `the_store_and_a_scanned_table_answer_every_select_alike` asks both every select over a fixture (every subset of every stored key as a base, with no span, one span of each of six item lists, and two), at both widths and for all five functions: over 5,000 triples. + +## `ChangeSource` + +`CellSource` plus `drain_changes(&self) -> Vec`: a source that reports what changed in it. +The contract that matters is that a source's cells are those as of the last drain, so a snapshot read +now plus the changes drained later counts no change twice or not at all. Why, and how `Incremental` +uses it: `context/code/improvise-incremental.md`, step 7. `DataStore` is not a `ChangeSource`: its log +is `&mut` and is drained by the owner in the step that made the edit. A `SharedFeed` is the `Arc` a +backend and a model share. diff --git a/context/test-evidence/improvise-incremental.md b/context/test-evidence/improvise-incremental.md index 55683f27..61011e72 100644 --- a/context/test-evidence/improvise-incremental.md +++ b/context/test-evidence/improvise-incremental.md @@ -801,3 +801,44 @@ User-visible changes this evidence also covers: a cycle that only flips a zero's same cycle over nonzero data always was; a cycle fed a NaN (`A = A + 1`, `A = A` with `A` NaN) settles on NaN instead of `circular`. + +### a_live_source_matches_eager_over_the_corpus_and_its_edits + +improvise-rnak. `Incremental` over a `LiveTable` (the model's cells moved into the table, the model's +own store emptied) against `Eager` over the shipped documents, their views and `edits_for` +(trimmed to three edits above 40 cells, as `corpus_eager` does). Rivals, each failing it by name: `sync` +drains nothing; the model not put over the feed (formulas read an empty store); a shape built from an +empty store; the stems built from the model's own store. + +### changes_named_by_the_caller_are_maintained_as_the_drain_would + +`apply_changes` with the changes the table reports, against `Eager` after each of four edits (an +overwrite, a new record under an item the model lacks, a removal, a text), asking every probe of every +step so far. Rivals: `apply_changes` ignores its changes (only this test and no other fails it); the +stems from the owned store. + +### a_write_is_invisible_until_it_is_drained + +The contract of a `ChangeSource`: a read after a write and before the drain sees the old value, after +`sync` the new one, and a second `sync` changes nothing. Rivals: the table shows a write at once (this +test alone fails it, which is why the contract has a test of its own: the differential cannot see it, +since every shape it asks about was built before the write); `sync` drains nothing. + +### the_cost_of_a_reported_change_does_not_depend_on_the_size_of_the_table + +One overwrite reported by the table, after the shape is built, at 216 and at 864 cells: two group +updates and two sum updates, no build, equal meters, and zero snapshot rows served by the source +for the change. The build itself serves exactly the table (`n * n * 6` rows), so the zero is a count +that could have been positive. Rival: every sync rebuilds the shapes (`builds` and `built_cells` move, +and the table serves the snapshot again). + +### an_overwrite_by_the_same_value_moves_nothing + +A table that reports an overwrite of a cell by its own value leaves the fold where it was (23). Rival: +the model not over the feed fails it with the others. + +### a_live_source_matches_eager_over_random_formula_and_cell_edits + +`Fed` against `Eager` over random formula and cell edits on the formula fixture (no views, so the binary +stays under two seconds). Rivals: `sync` drains nothing, the model not over the feed, a shape built from an +empty store and the stems from the owned store all fail it. diff --git a/crates/improvise-core/src/model/source.rs b/crates/improvise-core/src/model/source.rs index d612761d..5c3fd8a0 100644 --- a/crates/improvise-core/src/model/source.rs +++ b/crates/improvise-core/src/model/source.rs @@ -1,7 +1,7 @@ //! Where cell values are read from: a read-only trait, and the aggregate it answers in. //! Reasoning: `context/code/source.md` (ADR 003, ADR 009 clause 2; improvise-qddo). -use super::cell::{CellKey, CellValue, DataStore}; +use super::cell::{CellChange, CellKey, CellValue, DataStore}; use crate::formula::AggFunc; use numtower::ExactSum; @@ -170,6 +170,16 @@ impl Summary { /// its readers (a database) does so through interior mutability of its own. pub type SharedSource = std::sync::Arc; +/// A source that reports its changes. Its cells are the state after the changes it has drained, +/// so a snapshot read now plus the changes drained later counts no change twice or not at all. +pub trait ChangeSource: CellSource { + /// The changes since the last drain, oldest first; reads see them from now on. + fn drain_changes(&self) -> Vec; +} + +/// A change-reporting source a backend and a model share. +pub type SharedFeed = std::sync::Arc; + /// The read side of cell storage (ADR 003). Nothing here writes, so a source that cannot be /// written to is not one that fails at runtime: it has no method to fail in. pub trait CellSource { diff --git a/crates/improvise-incremental/src/eager.rs b/crates/improvise-incremental/src/eager.rs index 846c950d..7bb674eb 100644 --- a/crates/improvise-incremental/src/eager.rs +++ b/crates/improvise-incremental/src/eager.rs @@ -67,9 +67,9 @@ impl Derived for Eager { held == Some(item.as_str()) || self.group_holds(cat, item, held) }) }; - let cells = self.model.data.iter_cells(); + let cells = self.model.source().keys(&[]); cells - .filter(|(k, _)| named_by(k) && k.0.iter().all(drawn)) + .filter(|k| named_by(k) && k.0.iter().all(drawn)) .count() } } diff --git a/crates/improvise-incremental/src/formulas.rs b/crates/improvise-incremental/src/formulas.rs index f23d4147..2b4b2f73 100644 --- a/crates/improvise-incremental/src/formulas.rs +++ b/crates/improvise-incremental/src/formulas.rs @@ -240,7 +240,7 @@ impl Stems { order: Order::of(model), wide, }; - let cells: Vec = model.data.iter_cells().map(|(k, _)| k).collect(); + let cells: Vec = model.source().keys(&[]).collect(); let presence: Vec<(&CellKey, i64)> = cells.iter().map(|k| (k, 1)).collect(); stems.count(&presence); stems diff --git a/crates/improvise-incremental/src/incremental.rs b/crates/improvise-incremental/src/incremental.rs index 65211405..52b4bdff 100644 --- a/crates/improvise-incremental/src/incremental.rs +++ b/crates/improvise-incremental/src/incremental.rs @@ -9,9 +9,11 @@ use crate::{Derived, Edit, Maintained, Reading, cell_delta}; use improvise_core::formula::ast::AggFunc; use improvise_core::model::category::MEASURE_CATEGORY; use improvise_core::model::cell::{CellChange, CellKey, CellValue}; +use improvise_core::model::source::{ChangeSource, SharedFeed}; use improvise_core::model::types::{Covers, settled}; use improvise_core::model::{Aggregation, Model}; use std::collections::BTreeMap; +use std::sync::Arc; /// How a stem's formulas are evaluated: in dependency order where one pass is exact, or by /// eager's passes always. The two must agree; `Plain` is the reference. @@ -31,6 +33,8 @@ pub struct Incremental { fingerprint: Vec, evaluation: Evaluation, work: Meter, + /// The source the model is over, when its cells are not the model's own. + feed: Option, } static SUM: AggFunc = AggFunc::Sum; @@ -38,6 +42,21 @@ static SUM: AggFunc = AggFunc::Sum; impl Incremental { pub fn new(mut model: Model, none_cats: Vec) -> Self { model.data.record_changes(); + Self::start(model, none_cats, None) + } + + /// A backend over `feed`, which holds the cells and reports their changes; `model`'s own + /// store is not read. Reasoning: `context/code/improvise-incremental.md`, step 7. + pub fn over( + model: Model, + feed: Arc, + none_cats: Vec, + ) -> Self { + let model = model.over(feed.clone()); + Self::start(model, none_cats, Some(feed)) + } + + fn start(model: Model, none_cats: Vec, feed: Option) -> Self { let mut backend = Incremental { model, none_cats, @@ -46,12 +65,37 @@ impl Incremental { fingerprint: Vec::new(), evaluation: Evaluation::default(), work: Meter::default(), + feed, }; backend.rebuild_formulas(); backend.work = Meter::default(); backend } + /// Maintains what the feed has reported since the last drain. + pub fn sync(&mut self) { + let changes = self.drain(); + self.absorb(None, &changes); + } + + /// Maintains `changes`, which the source already reflects. For a source whose changes reach the + /// backend by another route than the drain. + pub fn apply_changes(&mut self, changes: &[CellChange]) { + self.absorb(None, changes); + } + + fn drain(&mut self) -> Vec { + match &self.feed { + Some(feed) => feed.drain_changes(), + None => self.model.data.drain_changes(), + } + } + + fn absorb(&mut self, edit: Option<&Edit>, changes: &[CellChange]) { + self.shapes.apply(&cell_delta(changes), changes); + self.refresh_formulas(edit, changes); + } + pub fn with_evaluation(mut self, evaluation: Evaluation) -> Self { self.evaluation = evaluation; self.rebuild_formulas(); @@ -100,20 +144,24 @@ impl Incremental { let func = Reduce::of(self.func_for(key)); let expansion = Expansion::of(&self.model, key, covers, over); match (expansion.spans.is_empty(), agg) { - (true, Aggregation::Exact) => self.model.data.get(key).cloned(), + (true, Aggregation::Exact) => self.model.source().value_at(key), (true, Aggregation::OverHidden) => self.fold(std::slice::from_ref(key), func), (false, Aggregation::Exact) => { let members = expansion.members(); - let cells = members.iter().filter_map(|m| self.model.data.get(m)); - fold_groups(&[&Group::of(cells)], func) + let cells: Vec = members + .iter() + .filter_map(|m| self.model.source().value_at(m)) + .collect(); + fold_groups(&[&Group::of(&cells)], func) } (false, Aggregation::OverHidden) => self.fold(&expansion.members(), func), } } fn fold(&self, keys: &[CellKey], func: Reduce) -> Option { - self.shapes - .read(&self.model.data, keys, |groups| fold_groups(groups, func)) + self.shapes.read(self.model.source(), keys, |groups| { + fold_groups(groups, func) + }) } /// The items a stored cell may hold in `cat` to be named by `item`: that item, or a member @@ -134,7 +182,7 @@ impl Incremental { fn cells_at(&self, at: &CellKey) -> i64 { let keys = std::slice::from_ref(at); let cells = |groups: &[&Group]| groups.first().map_or(0, |g| g.cells()); - self.shapes.read(&self.model.data, keys, cells) + self.shapes.read(self.model.source(), keys, cells) } /// Cells at or under `at` that draw no hidden coordinate: all of them, less those with a @@ -190,7 +238,7 @@ impl Incremental { } /// Brings the formula cells up to date with an edit and the store changes it made. - fn refresh_formulas(&mut self, edit: &Edit, changes: &[CellChange]) { + fn refresh_formulas(&mut self, edit: Option<&Edit>, changes: &[CellChange]) { if self.fingerprint != self.fingerprint() { return self.rebuild_formulas(); } @@ -203,7 +251,7 @@ impl Incremental { self.clear_stem(stem); } let mut dirty = moved.dirty; - if let Edit::SetMeasureAgg { measure, .. } = edit { + if let Some(Edit::SetMeasureAgg { measure, .. }) = edit { dirty.extend(self.stems.carrying(measure)); if self.stems.reads_name(measure) { dirty.extend(self.stems.live().cloned()); @@ -343,8 +391,7 @@ impl Derived for Incremental { impl Maintained for Incremental { fn apply(&mut self, edit: &Edit) { edit.apply(&mut self.model); - let changes = self.model.data.drain_changes(); - self.shapes.apply(&cell_delta(&changes), &changes); - self.refresh_formulas(edit, &changes); + let changes = self.drain(); + self.absorb(Some(edit), &changes); } } diff --git a/crates/improvise-incremental/src/shapes.rs b/crates/improvise-incremental/src/shapes.rs index 4bad5cd3..5ac700be 100644 --- a/crates/improvise-incremental/src/shapes.rs +++ b/crates/improvise-incremental/src/shapes.rs @@ -3,7 +3,8 @@ use crate::group::Group; use crate::{CellDelta, DeltaValue}; -use improvise_core::model::cell::{CellChange, CellKey, CellValue, DataStore}; +use improvise_core::model::cell::{CellChange, CellKey, CellValue}; +use improvise_core::model::source::CellSource; use std::cell::{Cell, RefCell}; use std::collections::BTreeMap; @@ -111,28 +112,28 @@ impl Shape { (updates, sums) } - /// The shape of `store`'s cells, one pass. - fn of(store: &DataStore, cats: &[String]) -> (Shape, u64) { + /// The shape of `source`'s cells, one pass. + fn of(source: &dyn CellSource, cats: &[String]) -> (Shape, u64) { let mut shape = Shape { groups: BTreeMap::new(), }; let mut visited = 0; - for (key, value) in store.iter_cells() { + for (key, value) in source.cells(&[]) { visited += 1; let Some(at) = project(&key, cats) else { continue; }; let group = shape.groups.entry(at).or_default(); - group.apply(&DeltaValue::new(value), 1); + group.apply(&DeltaValue::new(&value), 1); if let CellValue::Number(x) = value { - group.add_number(*x, 1); + group.add_number(x, 1); } } (shape, visited) } } -/// Every shape a read has asked for. A shape is built from the store on its first read (a +/// Every shape a read has asked for. A shape is built from the source on its first read (a /// full pass) and fed every delta after. #[derive(Default)] pub struct Shapes { @@ -177,7 +178,7 @@ impl Shapes { /// shape first if no read has asked for it. A key with no group is left out. pub fn read( &self, - store: &DataStore, + source: &dyn CellSource, keys: &[CellKey], read: impl FnOnce(&[&Group]) -> R, ) -> R { @@ -192,7 +193,7 @@ impl Shapes { ); let mut built = self.built.borrow_mut(); let shape = built.entry(cats.clone()).or_insert_with(|| { - let (shape, visited) = Shape::of(store, &cats); + let (shape, visited) = Shape::of(source, &cats); self.bump(|m| { m.builds += 1; m.built_cells += visited; diff --git a/crates/improvise-incremental/tests/live_source.rs b/crates/improvise-incremental/tests/live_source.rs new file mode 100644 index 00000000..ba6a622a --- /dev/null +++ b/crates/improvise-incremental/tests/live_source.rs @@ -0,0 +1,213 @@ +//! `Incremental` over a source that reports its own changes (ADR 009 step 7, improvise-rnak). +//! Evidence: `context/test-evidence/improvise-incremental.md`. Counts, never time. + +mod support; + +use improvise_core::formula::ast::AggFunc; +use improvise_core::model::cell::{CellKey, CellValue}; +use improvise_core::model::source::ChangeSource; +use improvise_core::model::{Aggregation, Model}; +use improvise_incremental::{ + Derived, Eager, Edit, Maintained, Meter, differential_in, edit_probes, probes, same_value, +}; +use support::live::Fed; +use support::{corpus, edits_for, hand_model, key}; + +#[test] +fn a_live_source_matches_eager_over_the_corpus_and_its_edits() { + let (mut compared, mut files_done) = (0, Vec::::new()); + for (name, model, none, views) in corpus() { + let file = name.split(':').next().unwrap().to_string(); + if files_done.contains(&file) { + continue; + } + files_done.push(file); + let mut edits = edits_for(&model); + if model.data.iter_cells().count() > 40 { + let picked = [0, 2, 4].iter().filter_map(|&i| edits.get(i).cloned()); + edits = picked.collect(); + } + let candidate = Fed::new(&model, none.clone()); + let ran = differential_in(&model, &views, &none, candidate, &edits); + let report = ran.unwrap_or_else(|d| panic!("{name}: {d}")); + assert_eq!(report.steps, edits.len() + 1, "{name}"); + compared += report.probes; + } + assert!(compared > 10_000, "only {compared} answers were compared"); +} + +#[test] +fn changes_named_by_the_caller_are_maintained_as_the_drain_would() { + let (model, none) = (hand_model(), vec!["Region".to_string()]); + let mut eager = Eager::new(model.clone(), none.clone()); + let mut fed = Fed::new(&model, none); + let edits = [ + Edit::SetCell(key("East", "A", "Revenue"), CellValue::Number(99.0)), + Edit::SetCell(key("North", "A", "Revenue"), CellValue::Number(1.0)), + Edit::ClearCell(key("West", "B", "Cost")), + Edit::SetCell(key("East", "A", "Note"), CellValue::Text("z".into())), + ]; + let mut questions = probes(eager.model()); + let mut asked = 0; + for edit in &edits { + eager.apply(edit); + let (cell, value) = match edit { + Edit::SetCell(cell, value) => (cell, Some(value.clone())), + Edit::ClearCell(cell) => (cell, None), + _ => unreachable!("cell edits only"), + }; + fed.live.write(cell.clone(), value); + let changes = fed.live.drain_changes(); + assert_eq!(changes.len(), 1); + fed.inc.apply_changes(&changes); + questions.extend(probes(eager.model())); + questions.extend(edit_probes(edit)); + for probe in &questions { + let (want, got) = (probe.ask(&eager), probe.ask(&fed)); + assert!( + same_value(&want, &got), + "{probe:?}: eager {want:?} != fed {got:?}" + ); + asked += 1; + } + } + assert!(asked > 500, "only {asked} answers were compared"); +} + +#[test] +fn a_write_is_invisible_until_it_is_drained() { + let model = hand_model(); + let mut fed = Fed::new(&model, vec![]); + let cell = key("East", "A", "Revenue"); + let before = fed.inc.value(&cell, Aggregation::Exact); + assert_eq!(before, Some(CellValue::Number(10.0))); + fed.live.write(cell.clone(), Some(CellValue::Number(50.0))); + assert_eq!( + fed.inc.value(&cell, Aggregation::Exact), + before, + "not yet drained" + ); + fed.inc.sync(); + assert_eq!( + fed.inc.value(&cell, Aggregation::Exact), + Some(CellValue::Number(50.0)) + ); + fed.inc.sync(); + assert_eq!( + fed.inc.value(&cell, Aggregation::Exact), + Some(CellValue::Number(50.0)) + ); +} + +fn row(r: usize, p: usize, m: usize) -> CellKey { + CellKey::new(vec![ + ("Region".into(), format!("r{r}")), + ("Product".into(), format!("p{p}")), + ("Month".into(), format!("m{m}")), + ("_Measure".into(), "Revenue".into()), + ]) +} + +/// `n` regions by `n` products by 6 months of one measure, as a model with a profit-style formula. +fn table(n: usize) -> Model { + let mut model = Model::new("table"); + for (cat, prefix, count) in [("Region", "r", n), ("Product", "p", n), ("Month", "m", 6)] { + model.add_category(cat).unwrap(); + for i in 0..count { + model + .category_mut(cat) + .unwrap() + .add_item(format!("{prefix}{i}")); + } + } + model.category_mut("_Measure").unwrap().add_item("Revenue"); + model.measure_agg.insert("Revenue".into(), AggFunc::Max); + let mut seed = 1u64; + for (r, p, m) in (0..n).flat_map(|r| (0..n).flat_map(move |p| (0..6).map(move |m| (r, p, m)))) { + seed = seed + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + model.set_cell( + row(r, p, m), + CellValue::Number(((seed >> 33) % 1000) as f64), + ); + } + model +} + +fn since(before: Meter, after: Meter) -> Meter { + Meter { + updates: after.updates - before.updates, + sum_updates: after.sum_updates - before.sum_updates, + reads: after.reads - before.reads, + builds: after.builds - before.builds, + built_cells: after.built_cells - before.built_cells, + stems: after.stems - before.stems, + formula_cells: after.formula_cells - before.formula_cells, + rebuilds: after.rebuilds - before.rebuilds, + recomputes: after.recomputes - before.recomputes, + } +} + +fn region_partial() -> CellKey { + CellKey::new(vec![ + ("Region".into(), "r1".into()), + ("_Measure".into(), "Revenue".into()), + ]) +} + +/// What one reported change costs once a shape is built, and what the source served for it. +fn change_cost(n: usize) -> (Meter, u64) { + let mut fed = Fed::new(&table(n), vec![]); + fed.inc.raw_fold(®ion_partial()); + let (before, served) = (fed.inc.meter(), fed.live.snapshot_rows()); + assert_eq!( + served, + (n * n * 6) as u64, + "the build is one pass over the table" + ); + fed.live + .write(row(1, 2, 3), Some(CellValue::Number(2000.0))); + fed.inc.sync(); + let cost = since(before, fed.inc.meter()); + assert_eq!( + fed.inc.raw_fold(®ion_partial()), + Some(CellValue::Number(2000.0)), + "the new maximum comes from the change" + ); + (cost, fed.live.snapshot_rows() - served) +} + +#[test] +fn the_cost_of_a_reported_change_does_not_depend_on_the_size_of_the_table() { + let (small, small_rows) = change_cost(6); + let (large, large_rows) = change_cost(12); + assert_eq!( + (small.updates, small.sum_updates), + (2, 2), + "retract and insert" + ); + assert_eq!(small, large, "216 cells and 864 cells cost the same"); + assert_eq!( + (small.builds, small.built_cells), + (0, 0), + "no shape is rebuilt" + ); + assert_eq!( + (small_rows, large_rows), + (0, 0), + "the source serves no snapshot for a change" + ); +} + +#[test] +fn an_overwrite_by_the_same_value_moves_nothing() { + let mut fed = Fed::new(&hand_model(), vec![]); + let cell = key("East", "A", "Revenue"); + let revenue = CellKey::new(vec![("_Measure".into(), "Revenue".into())]); + let before = fed.inc.raw_fold(&revenue); + fed.live.write(cell, Some(CellValue::Number(10.0))); + fed.inc.sync(); + assert_eq!(fed.inc.raw_fold(&revenue), before); + assert_eq!(before, Some(CellValue::Number(23.0))); +} diff --git a/crates/improvise-incremental/tests/random_live.proptest-regressions b/crates/improvise-incremental/tests/random_live.proptest-regressions new file mode 100644 index 00000000..5698220b --- /dev/null +++ b/crates/improvise-incremental/tests/random_live.proptest-regressions @@ -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 78a76cbc766ceb9d87768ca4fe518dc0773289e681621477b5f2fca6d15e36c9 # shrinks to edits = [SetCell(CellKey([("Product", "A"), ("Region", "East"), ("_Measure", "Revenue")]), Number(0.0))] diff --git a/crates/improvise-incremental/tests/random_live.rs b/crates/improvise-incremental/tests/random_live.rs new file mode 100644 index 00000000..82330922 --- /dev/null +++ b/crates/improvise-incremental/tests/random_live.rs @@ -0,0 +1,22 @@ +//! `Incremental` over a live source against `Eager` over random formula and cell edits. +//! Evidence: `context/test-evidence/improvise-incremental.md`. + +mod support; + +use improvise_incremental::differential_in; +use proptest::prelude::*; +use support::live::Fed; +use support::{any_formula_edit, formula_workbook}; + +proptest! { + #[test] + fn a_live_source_matches_eager_over_random_formula_and_cell_edits( + edits in prop::collection::vec(any_formula_edit(), 0..3), + ) { + let wb = formula_workbook(); + let none = vec!["Region".to_string()]; + let candidate = Fed::new(&wb.model, none.clone()); + let ran = differential_in(&wb.model, &[], &none, candidate, &edits); + prop_assert!(ran.is_ok(), "{}", ran.unwrap_err()); + } +} diff --git a/crates/improvise-incremental/tests/support/live.rs b/crates/improvise-incremental/tests/support/live.rs new file mode 100644 index 00000000..8d6d2259 --- /dev/null +++ b/crates/improvise-incremental/tests/support/live.rs @@ -0,0 +1,156 @@ +//! A live table: a scripted source that reports its changes, the way a database would. +//! Its rows are those as of the last drain; a write is pending until then. +#![allow(dead_code)] + +use improvise_core::model::cell::{CellChange, CellKey, CellValue, DataStore}; +use improvise_core::model::source::{CellSource, ChangeSource, Select, Summary}; +use improvise_core::model::{Aggregation, Model}; +use improvise_incremental::{Derived, Edit, Incremental, Maintained, Reading}; +use std::sync::{Arc, Mutex}; + +#[derive(Default)] +struct State { + /// What a read sees: the rows as of the last drain. + visible: DataStore, + /// The rows with every pending write applied, to say what a write changed. + latest: DataStore, + pending: Vec, + /// Rows handed out by a snapshot read (`cells` with no partial). + snapshot_rows: u64, +} + +#[derive(Default)] +pub struct LiveTable { + state: Mutex, +} + +impl LiveTable { + /// A table already holding `rows`, with nothing pending. + pub fn of(rows: impl IntoIterator) -> Self { + let mut state = State::default(); + for (key, value) in rows { + state.visible.set(key.clone(), value.clone()); + state.latest.set(key, value); + } + LiveTable { + state: Mutex::new(state), + } + } + + /// Writes `value` at `key` (or removes it), as a database user would: visible after the next drain. + pub fn write(&self, key: CellKey, value: Option) { + let mut state = self.state.lock().unwrap(); + let old = state.latest.get(&key).cloned(); + match &value { + Some(v) => state.latest.set(key.clone(), v.clone()), + None if old.is_none() => return, + None => state.latest.remove(&key), + } + state.pending.push(CellChange { + key, + old, + new: value, + }); + } + + /// How many rows snapshot reads have handed out so far. + pub fn snapshot_rows(&self) -> u64 { + self.state.lock().unwrap().snapshot_rows + } +} + +impl CellSource for LiveTable { + fn value_at(&self, key: &CellKey) -> Option { + self.state.lock().unwrap().visible.value_at(key) + } + + fn summarize(&self, select: &Select) -> Summary { + self.state.lock().unwrap().visible.summarize(select) + } + + fn keys<'a>( + &'a self, + partial: &'a [(String, String)], + ) -> Box + 'a> { + let state = self.state.lock().unwrap(); + let keys: Vec = state.visible.keys(partial).collect(); + Box::new(keys.into_iter()) + } + + fn cells<'a>( + &'a self, + partial: &'a [(String, String)], + ) -> Box + 'a> { + let mut state = self.state.lock().unwrap(); + let cells: Vec<_> = state.visible.cells(partial).collect(); + if partial.is_empty() { + state.snapshot_rows += cells.len() as u64; + } + Box::new(cells.into_iter()) + } +} + +impl ChangeSource for LiveTable { + fn drain_changes(&self) -> Vec { + let mut state = self.state.lock().unwrap(); + let changes = std::mem::take(&mut state.pending); + for change in &changes { + match &change.new { + Some(value) => state.visible.set(change.key.clone(), value.clone()), + None => state.visible.remove(&change.key), + } + } + changes + } +} + +/// An `Incremental` over a live table that holds the model's cells; the model keeps no store. +/// A cell edit is a write to the table, the way another client's would be, followed by a drain. +pub struct Fed { + pub inc: Incremental, + pub live: Arc, +} + +impl Fed { + pub fn new(model: &Model, none_cats: Vec) -> Fed { + let rows = model.data.iter_cells().map(|(k, v)| (k, v.clone())); + let live = Arc::new(LiveTable::of(rows)); + let mut bare = model.clone(); + bare.data = DataStore::new(); + let inc = Incremental::over(bare, live.clone(), none_cats); + Fed { inc, live } + } +} + +impl Derived for Fed { + fn value(&self, key: &CellKey, agg: Aggregation) -> Option { + self.inc.value(key, agg) + } + + fn raw_fold(&self, partial: &CellKey) -> Option { + self.inc.raw_fold(partial) + } + + fn covered(&self, key: &CellKey, reading: &Reading) -> Option { + self.inc.covered(key, reading) + } + + fn total(&self, key: &CellKey, over: &[String], reading: &Reading) -> Option { + self.inc.total(key, over, reading) + } + + fn occupancy(&self, partial: &CellKey, reading: &Reading) -> usize { + self.inc.occupancy(partial, reading) + } +} + +impl Maintained for Fed { + fn apply(&mut self, edit: &Edit) { + match edit { + Edit::SetCell(key, value) => self.live.write(key.clone(), Some(value.clone())), + Edit::ClearCell(key) => self.live.write(key.clone(), None), + _ => return self.inc.apply(edit), + } + self.inc.sync(); + } +} diff --git a/crates/improvise-incremental/tests/support/mod.rs b/crates/improvise-incremental/tests/support/mod.rs index 0ba3d060..db732221 100644 --- a/crates/improvise-incremental/tests/support/mod.rs +++ b/crates/improvise-incremental/tests/support/mod.rs @@ -1,6 +1,8 @@ //! Fixtures the integration tests share: the shipped corpus and the hand-built models. #![allow(dead_code)] +pub mod live; + use improvise_core::formula::ast::AggFunc; use improvise_core::formula::parse_formula; use improvise_core::model::Aggregation; From 6bc102f37874cacab105a5f9cb07eaf7de201e06 Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-sources)" Date: Sat, 10 Oct 2026 18:13:15 -0700 Subject: [PATCH 2/3] chore(incremental): drop a proptest seed an injected fault left behind (improvise-rnak) Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- .../tests/random_live.proptest-regressions | 7 ------- 1 file changed, 7 deletions(-) delete mode 100644 crates/improvise-incremental/tests/random_live.proptest-regressions diff --git a/crates/improvise-incremental/tests/random_live.proptest-regressions b/crates/improvise-incremental/tests/random_live.proptest-regressions deleted file mode 100644 index 5698220b..00000000 --- a/crates/improvise-incremental/tests/random_live.proptest-regressions +++ /dev/null @@ -1,7 +0,0 @@ -# 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 78a76cbc766ceb9d87768ca4fe518dc0773289e681621477b5f2fca6d15e36c9 # shrinks to edits = [SetCell(CellKey([("Product", "A"), ("Region", "East"), ("_Measure", "Revenue")]), Number(0.0))] From 82fbe16ea4761240491e71e69c54bf4f4a441cdb Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-sources)" Date: Sat, 10 Oct 2026 19:00:51 -0700 Subject: [PATCH 3/3] fix(incremental): over takes the exported feed alias, drops the model's cells and refuses cell edits in every build (improvise-rnak) Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- context/code/improvise-incremental.md | 10 ++- context/code/source.md | 5 +- .../test-evidence/improvise-incremental.md | 25 ++++++ crates/improvise-core/src/model/source.rs | 4 +- .../improvise-incremental/src/incremental.rs | 17 ++-- .../tests/live_source.rs | 79 ++++++++++++++++++- 6 files changed, 122 insertions(+), 18 deletions(-) diff --git a/context/code/improvise-incremental.md b/context/code/improvise-incremental.md index 0791e1de..ac5d592b 100644 --- a/context/code/improvise-incremental.md +++ b/context/code/improvise-incremental.md @@ -407,13 +407,17 @@ from the source's folds, never pushed down; `Derived` stays in this crate. what a database does inside a read transaction and what `DataStore` does already (its log is drained in the step that made the edit). `&self` because a live source is shared with the model that reads it and changes behind it by interior mutability of its own. -2. **`Incremental::over(model, feed, none_cats)`**: the model is put over the feed (`Model::over`) and - keeps no cells of its own. Its categories, formulas and aggregations are the definition; the +2. **`Incremental::over(model, feed, none_cats)`** takes a `SharedFeed`, the alias core exports (a generic + `Arc` could not take it: `S` is `Sized`). The model is put over the feed (`Model::over`), which + drops the cells it came with, so none is left for `get_cell`, persistence or the harness to read. Its categories, formulas and aggregations are the definition; the source holds the data. Shapes and stems are built from the feed as it stands. 3. **`sync()` drains and maintains; `apply_changes(&[CellChange])` maintains changes the source already reflects**, for a source whose changes reach the backend by another route (a trigger table, a channel). `Maintained::apply` on a fed backend takes definition edits (a formula, an aggregation) - and drains: a cell edit is the source's to make, and `Model::set_cell` refuses it in a debug build. + and drains: a cell edit is the source's to make, and `apply` refuses it with an `assert!` in every + build (a release build lost it silently when only `Model::set_cell`'s `debug_assert` stood in the way). + **A feed has one drainer**: a drain is destructive, so two backends over one feed split its changes + (`context/code/source.md`, `ChangeSource`). 4. **Freshness is the source's.** ADR 003 left "polled, triggered, or stale until refreshed" open: `sync` is the poll, and when it is called is the host's decision. Nothing here schedules it. 5. **What a change costs** is what it costs for the owned store: two group updates and two exact-sum diff --git a/context/code/source.md b/context/code/source.md index 056046c4..f30d8d83 100644 --- a/context/code/source.md +++ b/context/code/source.md @@ -98,7 +98,10 @@ The contract that matters is that a source's cells are those as of the last drai now plus the changes drained later counts no change twice or not at all. Why, and how `Incremental` uses it: `context/code/improvise-incremental.md`, step 7. `DataStore` is not a `ChangeSource`: its log is `&mut` and is drained by the owner in the step that made the edit. A `SharedFeed` is the `Arc` a -backend and a model share. +backend and a model share. **A feed has one drainer**: a drain empties the pending list, so two +backends over one feed split the changes between them (measured in review: the same write gave one +backend a Revenue of 128 and the other 38, and the second then disagreed with its own exact reads). +`a_feed_has_one_drainer` pins the behaviour. Kb-style invariant: `invariant(change_source, one_drainer)`. ## Building a `Summary` from an engine's answer diff --git a/context/test-evidence/improvise-incremental.md b/context/test-evidence/improvise-incremental.md index 61011e72..943bd9a3 100644 --- a/context/test-evidence/improvise-incremental.md +++ b/context/test-evidence/improvise-incremental.md @@ -842,3 +842,28 @@ the model not over the feed fails it with the others. `Fed` against `Eager` over random formula and cell edits on the formula fixture (no views, so the binary stays under two seconds). Rivals: `sync` drains nothing, the model not over the feed, a shape built from an empty store and the stems from the owned store all fail it. + +### a_backend_can_be_built_over_the_exported_feed_alias + +Review of #358: `over(Arc)` could not take the `SharedFeed` the same PR exports. The test builds +a backend from a `SharedFeed`; it did not compile against the generic signature. + +### a_populated_model_put_over_a_feed_keeps_no_cells_of_its_own + +Review of #358: `Fed::new` emptied the store before the call, so no test passed a populated model. This +one does, asserts the owned store is empty afterwards and that every harness probe agrees with +`Eager`. Rival: `Model::over` keeping the cells fails it, and only it. + +### a_cell_edit_on_a_fed_backend_is_refused + +`Maintained::apply` with a `SetCell` panics with the backend's own message in every build. Rival: the +assert removed (the test then sees the model's message, not the backend's, and fails). + +### a_cell_clear_on_a_fed_backend_is_refused + +The same for `ClearCell`. + +### a_feed_has_one_drainer + +Review of #358: two backends over one feed; one write; `a.sync(); b.sync()`. The first sees it, the +second does not. This pins the behaviour as documented rather than as a guarantee: a drain is destructive. diff --git a/crates/improvise-core/src/model/source.rs b/crates/improvise-core/src/model/source.rs index 944ae13f..2cd70a26 100644 --- a/crates/improvise-core/src/model/source.rs +++ b/crates/improvise-core/src/model/source.rs @@ -192,8 +192,8 @@ impl Summary { /// its readers (a database) does so through interior mutability of its own. pub type SharedSource = std::sync::Arc; -/// A source that reports its changes. Its cells are the state after the changes it has drained, -/// so a snapshot read now plus the changes drained later counts no change twice or not at all. +/// A source that reports its changes; its cells are the state as of its last drain. +/// A drain is destructive: a feed has one drainer (`context/code/source.md`). pub trait ChangeSource: CellSource { /// The changes since the last drain, oldest first; reads see them from now on. fn drain_changes(&self) -> Vec; diff --git a/crates/improvise-incremental/src/incremental.rs b/crates/improvise-incremental/src/incremental.rs index 52b4bdff..c83d9746 100644 --- a/crates/improvise-incremental/src/incremental.rs +++ b/crates/improvise-incremental/src/incremental.rs @@ -9,11 +9,10 @@ use crate::{Derived, Edit, Maintained, Reading, cell_delta}; use improvise_core::formula::ast::AggFunc; use improvise_core::model::category::MEASURE_CATEGORY; use improvise_core::model::cell::{CellChange, CellKey, CellValue}; -use improvise_core::model::source::{ChangeSource, SharedFeed}; +use improvise_core::model::source::{SharedFeed, SharedSource}; use improvise_core::model::types::{Covers, settled}; use improvise_core::model::{Aggregation, Model}; use std::collections::BTreeMap; -use std::sync::Arc; /// How a stem's formulas are evaluated: in dependency order where one pass is exact, or by /// eager's passes always. The two must agree; `Plain` is the reference. @@ -47,12 +46,9 @@ impl Incremental { /// A backend over `feed`, which holds the cells and reports their changes; `model`'s own /// store is not read. Reasoning: `context/code/improvise-incremental.md`, step 7. - pub fn over( - model: Model, - feed: Arc, - none_cats: Vec, - ) -> Self { - let model = model.over(feed.clone()); + pub fn over(model: Model, feed: SharedFeed, none_cats: Vec) -> Self { + let source: SharedSource = feed.clone(); + let model = model.over(source); Self::start(model, none_cats, Some(feed)) } @@ -390,6 +386,11 @@ impl Derived for Incremental { impl Maintained for Incremental { fn apply(&mut self, edit: &Edit) { + let refused = self.feed.is_some() && edit.key().is_some(); + assert!( + !refused, + "a fed backend is read-only: cell edits go through the source" + ); edit.apply(&mut self.model); let changes = self.drain(); self.absorb(Some(edit), &changes); diff --git a/crates/improvise-incremental/tests/live_source.rs b/crates/improvise-incremental/tests/live_source.rs index ba6a622a..493c3741 100644 --- a/crates/improvise-incremental/tests/live_source.rs +++ b/crates/improvise-incremental/tests/live_source.rs @@ -5,12 +5,14 @@ mod support; use improvise_core::formula::ast::AggFunc; use improvise_core::model::cell::{CellKey, CellValue}; -use improvise_core::model::source::ChangeSource; +use improvise_core::model::source::{ChangeSource, SharedFeed}; use improvise_core::model::{Aggregation, Model}; use improvise_incremental::{ - Derived, Eager, Edit, Maintained, Meter, differential_in, edit_probes, probes, same_value, + Derived, Eager, Edit, Incremental, Maintained, Meter, differential_in, edit_probes, probes, + same_value, }; -use support::live::Fed; +use std::sync::Arc; +use support::live::{Fed, LiveTable}; use support::{corpus, edits_for, hand_model, key}; #[test] @@ -108,7 +110,7 @@ fn row(r: usize, p: usize, m: usize) -> CellKey { ]) } -/// `n` regions by `n` products by 6 months of one measure, as a model with a profit-style formula. +/// `n` regions by `n` products by 6 months of one measure, with no formula. fn table(n: usize) -> Model { let mut model = Model::new("table"); for (cat, prefix, count) in [("Region", "r", n), ("Product", "p", n), ("Month", "m", 6)] { @@ -211,3 +213,72 @@ fn an_overwrite_by_the_same_value_moves_nothing() { assert_eq!(fed.inc.raw_fold(&revenue), before); assert_eq!(before, Some(CellValue::Number(23.0))); } + +#[test] +fn a_backend_can_be_built_over_the_exported_feed_alias() { + let model = hand_model(); + let rows = model.data.iter_cells().map(|(k, v)| (k, v.clone())); + let feed: SharedFeed = Arc::new(LiveTable::of(rows)); + let mut inc = Incremental::over(model.clone(), feed, vec![]); + let revenue = CellKey::new(vec![("_Measure".into(), "Revenue".into())]); + assert_eq!(inc.raw_fold(&revenue), Some(CellValue::Number(23.0))); + inc.sync(); +} + +#[test] +fn a_populated_model_put_over_a_feed_keeps_no_cells_of_its_own() { + let (model, none) = (hand_model(), vec!["Region".to_string()]); + assert!( + model.data.iter_cells().count() > 0, + "the input is populated" + ); + let rows = model.data.iter_cells().map(|(k, v)| (k, v.clone())); + let live = Arc::new(LiveTable::of(rows)); + let inc = Incremental::over(model.clone(), live, none.clone()); + assert_eq!(inc.model().data.iter_cells().count(), 0, "no stale copy"); + assert_eq!(inc.model().get_cell(&key("East", "A", "Revenue")), None); + let eager = Eager::new(model, none); + for probe in probes(eager.model()) { + let (want, got) = (probe.ask(&eager), probe.ask(&inc)); + assert!( + same_value(&want, &got), + "{probe:?}: eager {want:?} != fed {got:?}" + ); + } +} + +#[test] +#[should_panic(expected = "cell edits go through the source")] +fn a_cell_edit_on_a_fed_backend_is_refused() { + let mut fed = Fed::new(&hand_model(), vec![]); + let edit = Edit::SetCell(key("East", "A", "Revenue"), CellValue::Number(1.0)); + fed.inc.apply(&edit); +} + +#[test] +#[should_panic(expected = "cell edits go through the source")] +fn a_cell_clear_on_a_fed_backend_is_refused() { + let mut fed = Fed::new(&hand_model(), vec![]); + fed.inc.apply(&Edit::ClearCell(key("East", "A", "Revenue"))); +} + +#[test] +fn a_feed_has_one_drainer() { + let model = hand_model(); + let rows = model.data.iter_cells().map(|(k, v)| (k, v.clone())); + let live = Arc::new(LiveTable::of(rows)); + let mut a = Incremental::over(model.clone(), live.clone(), vec![]); + let mut b = Incremental::over(model, live.clone(), vec![]); + let revenue = CellKey::new(vec![("_Measure".into(), "Revenue".into())]); + let (before_a, before_b) = (a.raw_fold(&revenue), b.raw_fold(&revenue)); + assert_eq!(before_a, before_b); + live.write(key("East", "A", "Revenue"), Some(CellValue::Number(100.0))); + a.sync(); + b.sync(); + let (after_a, after_b) = (a.raw_fold(&revenue), b.raw_fold(&revenue)); + assert_ne!(after_a, before_a, "the first drainer saw the write"); + assert_eq!( + after_b, before_b, + "the second saw nothing: a drain is destructive" + ); +}