fix(incremental): pushdown takes the source alias, refuses cell edits, shares Eager's reads and pins stem order and refresh snapshots (improvise-rnak)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze

crates/improvise-incremental/src/eager.rs

@@ -1,6 +1,6 @@
 //! Today's evaluator behind the trait. Temporary (ADR 009 clause 2).
 
-use crate::{Derived, Edit, Maintained, Reading};
+use crate::{Derived, Edit, Maintained, Reading, reads};
 use improvise_core::model::cell::{CellKey, CellValue};
 use improvise_core::model::{Aggregation, Model};
 
@@ -36,27 +36,19 @@ impl Eager {
 
 impl Derived for Eager {
     fn value(&self, key: &CellKey, agg: Aggregation) -> Option<CellValue> {
-        self.model.evaluate_aggregated(key, agg)
+        reads::value(&self.model, key, agg)
     }
 
     fn raw_fold(&self, partial: &CellKey) -> Option<CellValue> {
-        match self.model.formula_for(partial) {
-            Some(_) => None,
-            None => self
-                .model
-                .evaluate_aggregated(partial, Aggregation::OverHidden),
-        }
+        reads::raw_fold(&self.model, partial)
     }
 
     fn covered(&self, key: &CellKey, reading: &Reading) -> Option<CellValue> {
-        let covers = |cat: &str, item: &str| reading.covers(cat, item);
-        self.model.evaluate_covering(key, reading.agg, &covers)
+        reads::covered(&self.model, key, reading)
     }
 
     fn total(&self, key: &CellKey, over: &[String], reading: &Reading) -> Option<CellValue> {
-        let covers = |cat: &str, item: &str| reading.covers(cat, item);
-        self.model
-            .evaluate_spanning(key, over, reading.agg, &covers)
+        reads::total(&self.model, key, over, reading)
     }
 
     fn occupancy(&self, partial: &CellKey, reading: &Reading) -> usize {

crates/improvise-incremental/src/lib.rs

@@ -11,6 +11,7 @@ mod harness;
 mod incremental;
 mod pushdown;
 mod reading;
+mod reads;
 mod shapes;
 
 pub use delta::{CellDelta, DeltaValue, cell_delta, moved_records};

crates/improvise-incremental/src/pushdown.rs

@@ -1,11 +1,10 @@
 //! Derived values over a source that answers aggregate queries and reports no changes.
 //! Reasoning: `context/code/improvise-incremental.md`, step 7.
 
-use crate::{Derived, Edit, Reading};
+use crate::{Derived, Edit, Reading, reads};
 use improvise_core::model::cell::{CellKey, CellValue};
-use improvise_core::model::source::{CellSource, Select};
+use improvise_core::model::source::{Select, SharedSource};
 use improvise_core::model::{Aggregation, Model};
-use std::sync::Arc;
 
 /// A model whose cells are in `source`: a fold is a query, the formula cells are computed in RAM
 /// by the model's own evaluator from those folds, and nothing is kept per shape or per change.
@@ -16,11 +15,7 @@ pub struct Pushdown {
 
 impl Pushdown {
     /// `definition` supplies the categories, formulas and aggregations; its own store is not read.
-    pub fn new<Q: CellSource + Send + Sync + 'static>(
-        definition: Model,
-        source: Arc<Q>,
-        none_cats: Vec<String>,
-    ) -> Self {
+    pub fn new(definition: Model, source: SharedSource, none_cats: Vec<String>) -> Self {
         let mut backend = Pushdown {
             model: definition.over(source),
             none_cats,
@@ -33,14 +28,19 @@ impl Pushdown {
         &self.model
     }
 
-    /// Computes the formula cells again from the source as it now stands. A read of a stored
-    /// cell or a fold is a query and always current; the formula cache is the one thing held.
+    /// Computes the formula cells again from the source as it now stands; only they are held.
+    /// The source should serve a refresh from one snapshot: improvise-incremental.md, step 7.
     pub fn refresh(&mut self) {
         self.model.recompute_formulas(&self.none_cats);
     }
 
     /// A change to the definition: a formula or an aggregation. Cells belong to the source.
     pub fn define(&mut self, edit: &Edit) {
+        let refused = edit.key().is_some();
+        assert!(
+            !refused,
+            "a pushdown is read-only: cell edits go through the source"
+        );
         edit.apply(&mut self.model);
         self.refresh();
     }
@@ -64,25 +64,19 @@ impl Pushdown {
 
 impl Derived for Pushdown {
     fn value(&self, key: &CellKey, agg: Aggregation) -> Option<CellValue> {
-        self.model.evaluate_aggregated(key, agg)
+        reads::value(&self.model, key, agg)
     }
 
     fn raw_fold(&self, partial: &CellKey) -> Option<CellValue> {
-        match self.model.formula_for(partial) {
-            Some(_) => None,
-            None => self.value(partial, Aggregation::OverHidden),
-        }
+        reads::raw_fold(&self.model, partial)
     }
 
     fn covered(&self, key: &CellKey, reading: &Reading) -> Option<CellValue> {
-        let covers = |cat: &str, item: &str| reading.covers(cat, item);
-        self.model.evaluate_covering(key, reading.agg, &covers)
+        reads::covered(&self.model, key, reading)
     }
 
     fn total(&self, key: &CellKey, over: &[String], reading: &Reading) -> Option<CellValue> {
-        let covers = |cat: &str, item: &str| reading.covers(cat, item);
-        self.model
-            .evaluate_spanning(key, over, reading.agg, &covers)
+        reads::total(&self.model, key, over, reading)
     }
 
     fn occupancy(&self, partial: &CellKey, reading: &Reading) -> usize {

crates/improvise-incremental/src/reads.rs

@@ -0,0 +1,33 @@
+//! The reads a model answers for a view, shared by every backend that leaves them to the model.
+//! Reasoning: `context/code/improvise-incremental.md`, step 7.
+
+use crate::Reading;
+use improvise_core::model::cell::{CellKey, CellValue};
+use improvise_core::model::{Aggregation, Model};
+
+pub fn value(model: &Model, key: &CellKey, agg: Aggregation) -> Option<CellValue> {
+    model.evaluate_aggregated(key, agg)
+}
+
+/// Never a formula: `None` for a partial naming a formula target.
+pub fn raw_fold(model: &Model, partial: &CellKey) -> Option<CellValue> {
+    match model.formula_for(partial) {
+        Some(_) => None,
+        None => value(model, partial, Aggregation::OverHidden),
+    }
+}
+
+pub fn covered(model: &Model, key: &CellKey, reading: &Reading) -> Option<CellValue> {
+    let covers = |cat: &str, item: &str| reading.covers(cat, item);
+    model.evaluate_covering(key, reading.agg, &covers)
+}
+
+pub fn total(
+    model: &Model,
+    key: &CellKey,
+    over: &[String],
+    reading: &Reading,
+) -> Option<CellValue> {
+    let covers = |cat: &str, item: &str| reading.covers(cat, item);
+    model.evaluate_spanning(key, over, reading.agg, &covers)
+}

crates/improvise-incremental/tests/pushdown.rs

@@ -4,11 +4,18 @@
 mod support;
 
 use improvise_core::formula::ast::AggFunc;
+use improvise_core::formula::parse_formula;
 use improvise_core::model::cell::{CellKey, CellValue};
+use improvise_core::model::source::SharedSource;
 use improvise_core::model::{Aggregation, Model};
 use improvise_core::view::View;
 use improvise_incremental::{Derived, Edit, Probe, Reading, differential_in};
-use support::remote::{IgnoresHidden, Pushed};
+use improvise_incremental::{Eager, Pushdown};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use support::remote::{
+    ChangesAfterAFold, ChangesAfterStems, Hooked, IgnoresHidden, Pushed, Remote, SortsStems,
+};
 use support::{
     corpus, edits_for, grouped_model, hand_model, hidden_counted_workbook, hidden_workbook, key,
     none_of, ratio_workbook, wide_workbook,
@@ -195,3 +202,154 @@ fn a_source_that_forgets_the_hidden_items_is_caught_as_a_divergence() {
     let sound = Pushed::new(&wb.model, none.clone());
     assert!(differential_in(&wb.model, &views, &none, sound, &[]).is_ok());
 }
+
+/// Eighteen products stored in chain order and named to sort the other way, and a formula that
+/// reads the product before it: one pass along the chain, one per link against it.
+fn chain() -> Model {
+    let name = |k: usize| format!("n{}", 100 - k);
+    let mut model = Model::new("chain");
+    model.add_category("Product").unwrap();
+    model.category_mut("_Measure").unwrap().add_item("Revenue");
+    for k in 1..=18 {
+        model.category_mut("Product").unwrap().add_item(name(k));
+        let cell = CellKey::new(vec![
+            ("Product".into(), name(k)),
+            ("_Measure".into(), "Revenue".into()),
+        ]);
+        model.set_cell(cell, CellValue::Number(k as f64));
+    }
+    let mut text = "0".to_string();
+    for k in (2..=18).rev() {
+        text = format!(
+            "IF((Product IS {}) > 0, {} + 1, {text})",
+            name(k),
+            name(k - 1)
+        );
+    }
+    text = format!("IF((Product IS {}) > 0, 1, {text})", name(1));
+    model.add_formula(parse_formula(&format!("Chain = {text}"), "_Measure").unwrap());
+    model
+}
+
+#[test]
+fn the_order_stems_are_listed_in_decides_a_result_and_a_source_must_keep_it() {
+    let model = chain();
+    let last = CellKey::new(vec![
+        ("Product".into(), "n82".into()),
+        ("_Measure".into(), "Chain".into()),
+    ]);
+    let eager = Eager::new(model.clone(), vec![]);
+    let want = eager.value(&last, Aggregation::OverHidden);
+    assert_eq!(
+        want,
+        Some(CellValue::Number(18.0)),
+        "along the chain it settles"
+    );
+    let sound = Pushed::new(&model, vec![]);
+    assert_eq!(sound.pd.value(&last, Aggregation::OverHidden), want);
+    let ran = differential_in(&model, &[], &[], sound, &[]);
+    assert!(ran.is_ok(), "{}", ran.unwrap_err());
+    let sorted = Pushed::through(&model, vec![], |remote| Hooked(remote, SortsStems));
+    let got = sorted.pd.value(&last, Aggregation::OverHidden);
+    assert_eq!(
+        got,
+        Some(CellValue::Error("circular".into())),
+        "against it, the bound of 16 passes"
+    );
+    let ran = differential_in(&model, &[], &[], sorted, &[]);
+    assert!(ran.is_err(), "a sorted stem list must be a divergence");
+}
+
+type Change = Box<dyn Fn(&Remote) + Send + Sync>;
+
+/// Another client's writes: Product A's revenue and cost move, and Product C appears.
+fn another_clients_writes() -> Change {
+    Box::new(|remote| {
+        remote.write(key("East", "A", "Revenue"), Some(CellValue::Number(50.0)));
+        remote.write(key("East", "A", "Cost"), Some(CellValue::Number(0.0)));
+        remote.write(key("East", "C", "Revenue"), Some(CellValue::Number(7.0)));
+        remote.write(key("East", "C", "Cost"), Some(CellValue::Number(2.0)));
+    })
+}
+
+fn profit_of(product: &str) -> CellKey {
+    CellKey::new(vec![
+        ("Product".into(), product.into()),
+        ("_Measure".into(), "Profit".into()),
+    ])
+}
+
+#[test]
+fn writes_between_two_queries_of_a_refresh_are_healed_by_the_fixed_point() {
+    let none = vec!["Region".to_string()];
+    let armed = Arc::new(AtomicBool::new(false));
+    let hook = ChangesAfterAFold {
+        armed: armed.clone(),
+        change: another_clients_writes(),
+    };
+    let none_cats = vec!["Region".to_string()];
+    let mut pushed = Pushed::through(&hand_model(), none_cats, |r| Hooked(r, hook));
+    assert_eq!(
+        pushed.pd.value(&profit_of("A"), Aggregation::OverHidden),
+        Some(CellValue::Number(9.0))
+    );
+    armed.store(true, Ordering::SeqCst);
+    pushed.pd.refresh();
+    assert!(
+        !armed.load(Ordering::SeqCst),
+        "the change happened mid-refresh"
+    );
+    let mut after = hand_model();
+    after.set_cell(key("East", "A", "Revenue"), CellValue::Number(50.0));
+    after.set_cell(key("East", "A", "Cost"), CellValue::Number(0.0));
+    let truth = Eager::new(after, none);
+    let want = truth.value(&profit_of("A"), Aggregation::OverHidden);
+    assert_eq!(want, Some(CellValue::Number(53.0)));
+    let got = pushed.pd.value(&profit_of("A"), Aggregation::OverHidden);
+    assert_eq!(
+        got, want,
+        "the last pass read the new state: 55 - 2, and not 13"
+    );
+}
+
+#[test]
+fn a_stem_that_appears_between_the_stem_list_and_the_folds_waits_for_the_next_refresh() {
+    let armed = Arc::new(AtomicBool::new(false));
+    let hook = ChangesAfterStems {
+        armed: armed.clone(),
+        change: another_clients_writes(),
+    };
+    let none_cats = vec!["Region".to_string()];
+    let mut pushed = Pushed::through(&hand_model(), none_cats, |r| Hooked(r, hook));
+    armed.store(true, Ordering::SeqCst);
+    pushed.pd.refresh();
+    assert_eq!(
+        pushed.pd.value(&profit_of("C"), Aggregation::OverHidden),
+        None,
+        "the stem list was taken before Product C had cells"
+    );
+    pushed.pd.refresh();
+    assert_eq!(
+        pushed.pd.value(&profit_of("C"), Aggregation::OverHidden),
+        Some(CellValue::Number(5.0)),
+        "and the next refresh has it"
+    );
+}
+
+#[test]
+#[should_panic(expected = "cell edits go through the source")]
+fn a_cell_edit_on_a_pushdown_is_refused() {
+    let mut pushed = Pushed::new(&hand_model(), vec![]);
+    let edit = Edit::SetCell(key("East", "A", "Revenue"), CellValue::Number(1.0));
+    pushed.pd.define(&edit);
+}
+
+#[test]
+fn a_pushdown_can_be_built_over_the_exported_source_alias() {
+    let model = hand_model();
+    let rows = model.data.iter_cells().map(|(k, v)| (k, v.clone()));
+    let source: SharedSource = Arc::new(Remote::of(rows));
+    let pd = Pushdown::new(model, source, vec![]);
+    let revenue = CellKey::new(vec![("_Measure".into(), "Revenue".into())]);
+    assert_eq!(pd.raw_fold(&revenue), Some(CellValue::Number(23.0)));
+}

crates/improvise-incremental/tests/support/remote.rs

@@ -6,6 +6,7 @@ use improvise_core::model::cell::{CellKey, CellValue};
 use improvise_core::model::source::{CellSource, Select, Summary, TableSource};
 use improvise_core::model::{Aggregation, Model};
 use improvise_incremental::{Derived, Edit, Maintained, Pushdown, Reading};
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
 
 /// What the engine was asked, and how many rows it sent back.
@@ -17,7 +18,7 @@ pub struct Asked {
     pub stem_lists: u64,
     pub key_lists: u64,
     pub cell_lists: u64,
-    /// Keys and cells sent back by `keys` and `cells`: rows, not aggregates.
+    /// Keys, cells and stems sent back by `keys`, `cells` and `distinct_without`: rows, not aggregates.
     pub rows_sent: u64,
 }
 
@@ -151,6 +152,94 @@ impl CellSource for IgnoresHidden {
     }
 }
 
+/// What a source does besides answering: reorders the stems it lists, or acts after a query.
+pub trait Hook: Send + Sync {
+    fn stems(&self, stems: Vec<CellKey>, _remote: &Remote) -> Vec<CellKey> {
+        stems
+    }
+
+    fn after_summary(&self, _remote: &Remote) {}
+}
+
+/// A source that answers from `remote` and consults `hook`.
+pub struct Hooked<H>(pub Arc<Remote>, pub H);
+
+impl<H: Hook> CellSource for Hooked<H> {
+    fn value_at(&self, key: &CellKey) -> Option<CellValue> {
+        self.0.value_at(key)
+    }
+
+    fn summarize(&self, select: &Select) -> Summary {
+        let summary = self.0.summarize(select);
+        self.1.after_summary(&self.0);
+        summary
+    }
+
+    fn keys<'a>(
+        &'a self,
+        partial: &'a [(String, String)],
+    ) -> Box<dyn Iterator<Item = CellKey> + 'a> {
+        self.0.keys(partial)
+    }
+
+    fn cells<'a>(
+        &'a self,
+        partial: &'a [(String, String)],
+    ) -> Box<dyn Iterator<Item = (CellKey, CellValue)> + 'a> {
+        self.0.cells(partial)
+    }
+
+    fn count(&self, select: &Select, excluding: &[(String, String)]) -> usize {
+        self.0.count(select, excluding)
+    }
+
+    fn distinct_without(&self, without: &[String]) -> Vec<CellKey> {
+        let stems = self.0.distinct_without(without);
+        self.1.stems(stems, &self.0)
+    }
+}
+
+/// Lists the stems sorted, not in the order the cells were first seen.
+pub struct SortsStems;
+
+impl Hook for SortsStems {
+    fn stems(&self, mut stems: Vec<CellKey>, _remote: &Remote) -> Vec<CellKey> {
+        stems.sort();
+        stems
+    }
+}
+
+type Change = Box<dyn Fn(&Remote) + Send + Sync>;
+
+/// Changes the table once, when `armed`, right after the stem list is sent.
+pub struct ChangesAfterStems {
+    pub armed: Arc<AtomicBool>,
+    pub change: Change,
+}
+
+impl Hook for ChangesAfterStems {
+    fn stems(&self, stems: Vec<CellKey>, remote: &Remote) -> Vec<CellKey> {
+        if self.armed.swap(false, Ordering::SeqCst) {
+            (self.change)(remote);
+        }
+        stems
+    }
+}
+
+/// Changes the table once, when `armed`, right after the first fold is answered.
+pub struct ChangesAfterAFold {
+    pub armed: Arc<AtomicBool>,
+    pub change: Change,
+}
+
+impl Hook for ChangesAfterAFold {
+    fn after_summary(&self, remote: &Remote) {
+        if self.armed.swap(false, Ordering::SeqCst) {
+            (self.change)(remote);
+        }
+    }
+}
+
 /// A `Pushdown` over a remote table that holds the model's cells; the model keeps no store.
 /// A cell edit is a write to the table, as another client's would be, then a refresh.
 pub struct Pushed {