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

Merge pull request #360 from fiddlerwoaroof/dbsp/pushdown

Browse Source 
feat(incremental): derived values over a source that answers aggregate queries (improvise-rnak)
This commit is contained in:
Ed Langley
2026-10-10 20:14:26 -07:00
committed by GitHub
parent a329a57159 f9b1e891df
commit 5f76bfaea1
17 changed files with 1193 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.nix
+5
View File
@@ -3095,6 +3095,11 @@ rec {
}
];
devDependencies = [
{
name = "improvise-core";
packageId = "improvise-core";
features = [ "test-support" ];
}
{
name = "improvise-io";
packageId = "improvise-io";
context/code/improvise-incremental.md
+73
View File
@@ -431,3 +431,76 @@ from the source's folds, never pushed down; `Derived` stays in this crate.
`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`.
### Query pushdown (`Pushdown`)
For a source that answers aggregate queries and reports nothing. `Pushdown::new(definition, source,
none_cats)` puts the model over the source (`Model::over`) and keeps nothing else: no shapes, no
change log, no stems. These are the readings taken.
1. **`AggregateQuery` is `CellSource`; there are two traits, not three.** The brief's `AggregateQuery`
(`fold`, `count`) is the trait core already has: `summarize` is the fold, `count` the count. `Derived`
stays here (user ruling) and stays a different thing: the values a view draws, which a backend
builds from a source's aggregates and its own state. A source implements `CellSource` and a backend
implements `Derived`.
2. **Folds are queries, formula cells are computed in RAM.** `value`, `raw_fold`, `covered` and `total` are
the model's evaluator over `Model::source()`, so a fold is one `summarize` and the rest is arithmetic
on the answers. The formula cache is the one thing `Pushdown` holds. That is the user's ruling
(2026-10-10), and it makes the evaluator hook the only change to core's evaluation.
3. **Staleness is `refresh()`.** A stored cell or a fold is current at the moment of the read. A formula
cell is the cache's value until `refresh` recomputes it from the source as it now stands, which
costs one `distinct_without` per formula and the folds each formula reads. When to refresh is the
host's (ADR 003's "polled, triggered, or explicitly stale"); nothing here schedules it. A
delivery layer that wants notices from a pushdown-backed shape has no moved keys to name, so it
marks them "correct" (ADR 009 clause 9), which is a bead (below).
4. **Occupancy is one `count`.** A group in the partial becomes a span over its own name and its
members; the view's hidden pairs go as `excluding`. `Eager::occupancy` lists every key to say the
same thing.
5. **`define(edit)` takes a formula or an aggregation edit and refreshes.** A cell edit is the source's:
`define` refuses it with an `assert!` of its own in every build (`a_cell_edit_on_a_pushdown_is_refused`).
6. **What the differential proves, and what it does not.** `value`, `raw_fold`, `covered` and `total` are
the same four functions for `Eager` and `Pushdown` (`reads.rs`; they were copies, and are shared now),
so a Pushdown-against-Eager differential cannot see a mistake in them. `Pushdown::members` repeats
`Eager::group_holds` in a different form (a span list against a predicate), and that repetition
is compared. What the differential proves is the **source's** semantics: that `summarize`, `count`
and `distinct_without` give, for every select the evaluator and the views make, the answers a
`DataStore` gives. The evaluator over a source is proven equal to the evaluator over the owned
store in `improvise-core` (`over_tests`). Nothing else is claimed.
7. **A refresh wants one snapshot.** `refresh` asks `distinct_without` and then folds, so a source that
changes between the queries can serve them from different revisions. The passes heal what they
can: a write between two folds is read again by the next pass, and the cache ends equal to the new
state's values (`writes_between_two_queries_of_a_refresh_are_healed_by_the_fixed_point`). A stem
that appears after the stem list was taken has no cell until the next refresh
(`a_stem_that_appears_between_the_stem_list_and_the_folds_waits_for_the_next_refresh`). A real source
should serve a refresh from one read transaction; nothing here can ask it to.
8. **Known evaluator defect: the result can depend on the order of the stems** (improvise-bw4p.30). The
passes of the fixed point update the cache in place, visiting stems in the order `distinct_without`
lists them, and stop at sixteen. A document whose stems read each other settles in one pass listed
along its dependency chain and takes one pass per link listed against it, so an eighteen-link chain
is `circular` the second way. That is the evaluator's fault, not an obligation on a source: a SQL
source has no order to keep, and bw4p.30 is to make the fixed point independent of it. Until then
the first-seen order (`DataStore`'s insertion order) is what the oracle uses, and
`the_order_stems_are_listed_in_decides_a_result_today` pins today's behaviour: along the chain 18,
sorted `circular`, and a source listing its stems sorted is a divergence. No fixture before it had a
result that depended on the order. When bw4p.30 lands the test's sorted half must change to expect
agreement, and this item goes.
9. **What is and is not proven.** The tests run against `Remote` (`tests/support/remote.rs`), a scanning
in-memory table with counters, that shares no evaluator with `Eager` and answers queries over rows.
Without a real SQL engine only the **contract** is proven: that a source answering these six
queries as specified gives `Eager`'s answers, and what each read costs in queries. Nothing here
shows that SQLite can answer `summarize` exactly (an exact sum is not what its `SUM` returns:
`context/code/source.md`), or what a query costs. The first real source is improvise-rucf, still
deferred.
### What step 7 did not predict
- **The hook came before the change feed**, not after (a fed backend's formulas are evaluated over the
source; see `context/code/source.md`).
- **`TableSource` and `LiveTable` found nothing in `Incremental`**; the corpus and the random edits pass
over a source on the first run. What the tests found was in the fixtures: a rival that read the owned
store for the stored fallback of a filtered formula passed until a cell was stored where the `WHERE`
fails, and a rival that forgot a group's own name as an item passed until a cell was stored under
one.
- **The two-second limit decided the shape of the tests** more than the design did: the corpus
differential over a scanning source is 2.5s untrimmed, so it follows `corpus_eager`'s trimming, and
each random test is its own binary.
context/code/source.md
+20
View File
@@ -14,6 +14,8 @@ they leave a choice.
| `summarize(select)` | the summary of the cells a `Select` names: the aggregate, not the rows |
| `keys(partial)` | the keys of the cells containing `partial`: stems, and the records behind an aggregate |
| `cells(partial)` | the same with values: records mode and a drill |
| `count(select, excluding)` | how many cells a select names, leaving out those holding a hidden pair: occupancy |
| `distinct_without(cats)` | the cell keys with `cats` removed, each once, first seen first: the stems formulas are evaluated at |
1. **Aggregation is the interface.** ADR 003 clause 1: `GridLayout` asks for a value, not for the
rows behind it, and the aggregates are SQL's. So the one read the evaluator makes in bulk,
@@ -99,6 +101,24 @@ 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.
## Pushdown needs two more queries
`count` and `distinct_without` joined the trait for `Pushdown` (improvise-rnak). Both are what a query
engine answers without sending rows (`COUNT(*) ... WHERE`, `SELECT DISTINCT`), and each replaces a
read that would have transferred every key: `Eager::occupancy` lists all keys, and
`recompute_formulas` walked every cell to find its stems. `recompute_formulas` now asks
`distinct_without(target and None categories)` once per formula, and the order it returns is
the order the cells were first seen in, so the stems are inserted in the order they always were.
That order is not a contract a source can be held to: the passes of the fixed point visit stems in
it and a chain of stems can settle or not by it. That is a known evaluator defect, improvise-bw4p.30
(`context/code/improvise-incremental.md`, step 7 item 8), and until it is fixed a source with no order
of its own should list stems in a stable one.
`Select::admits` is the predicate of a select over one key, for a source that must test keys itself
(`DataStore::count`). `TableSource` does not use it: its own `holds` is the independent second reading.
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).
context/test-evidence/core.md
+21
View File
@@ -2387,6 +2387,27 @@ branch reads the owned store; it fails here.
`set_cell` on a model over a source panics in a debug build, naming the reason. Without the assert the
write goes to the owned store and nothing reads it.
### model::source::tests::the_store_and_a_scanned_table_count_every_select_alike
improvise-rnak. `count(select, excluding)` on the store and on a scanning table, over every select of
the fixture, at both widths, hiding none, each of four items and each pair of them: 31,680 counts, 3,370
of them not zero (a count that was zero on both sides would pass). Rivals, each failing it: the count
ignores the excluded pairs (also fails `hiding_an_item_removes_exactly_the_cells_that_hold_it`), ignores the
spans, and ignores `exact`.
### model::source::tests::hiding_an_item_removes_exactly_the_cells_that_hold_it
The hand-checked half: all rows, all but those holding `R=E`, and a select naming a hidden item is
zero. A count that agreed with a wrong scan on both sides would pass the test above and fail this.
### model::source::tests::the_store_and_a_scanned_table_list_the_same_stems
`distinct_without` on the store and on a scanning table, for every subset of three categories and one
that does not exist, equal and in the same order, each stem once. Rivals, each failing it by name:
the stems not deduplicated, sorted instead of first seen (this is the only test that sees it: the
evaluator's results do not depend on stem order for the fixtures), and the dropped categories kept (also
fails several evaluator tests).
### a_summary_built_from_an_engines_parts_reduces_as_one_built_from_the_values
Review of #356 (Codex): an engine outside core could not build a `Summary` without replaying rows. The
context/test-evidence/improvise-incremental.md
+90
View File
@@ -843,6 +843,63 @@ the model not over the feed fails it with the others.
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_pushdown_matches_eager_over_the_shipped_documents
improvise-rnak. `Pushdown` over `Remote` (a scanning table that counts its queries; the model keeps no
cells) against `Eager`, over the shipped documents, their views and `edits_for` (trimmed as
`corpus_eager` is). Rivals, each failing it by name: the occupancy forgetting the hidden items; the
model not over the source; a raw fold answering for a formula target; a covered read covering every
member; a total covering every member; the engine ignoring `exact`; a refresh that does nothing; a
definition edit that does not refresh.
### a_pushdown_matches_eager_on_hidden_items_and_collapsed_groups_under_every_aggregation
The hidden and the counted fixtures, drawn, under each of five aggregations, with a NaN, a `-0.0` and
a text written to a hidden cell and a cell stored under the group's own name. The last is why the
rival "occupancy forgets the group's own name" is caught: before that edit, it passed everywhere.
Rivals: the occupancy treating a group as an item; the same forgetting the group's name; the occupancy
forgetting the hidden items; covered and total covering everything.
### a_pushdown_matches_eager_on_formulas_groups_and_wide_keys
The ratio fixture (a formula over two measures), the hand model under every aggregation of two
measures, a grouped model and the six-category wide model. Rivals: a refresh that does nothing, a
definition edit that does not refresh, a raw fold of a formula target, the engine ignoring `exact`.
### a_fold_is_one_query_and_sends_no_row
A raw fold, a collapsed group's covered read and a column total are three `summarize` calls between
them and no row sent back by the table (`rows_sent` unchanged); an exact read is one lookup. Rival:
the model not over the source (the reads then never reach the table).
### occupancy_is_one_count_whatever_the_groups_and_hidden_items
The Coastal-by-A occupancy of the hidden fixture (East hidden, West holding A) is 1, in one `count`,
with no row sent. Rivals: the hidden items forgotten, a group treated as an item.
### formula_cells_are_computed_from_the_distinct_stems_not_from_every_cell
Refreshing the hand model's one formula asks for stems once and gets two rows back (Product A and B)
from twelve cells, lists no keys and no cells, and still queries the folds the formula reads.
Rivals: a refresh that does nothing, the model not over the source.
### a_stored_cell_is_current_and_a_formula_cell_waits_for_a_refresh
The staleness contract of `Pushdown`. A write by another client is seen by the next exact read (50) and
not by the cached formula cell (9) until `refresh` (49). Rival: a refresh that does nothing.
### a_source_that_forgets_the_hidden_items_is_caught_as_a_divergence
The check of the checker, the user's ruling about faults: `IgnoresHidden` is a source whose `count`
drops the hidden pairs. The differential must report a `Divergence` and the probe must be an
occupancy; a sound source over the same fixture must pass.
### a_pushdown_matches_eager_over_random_formula_and_cell_edits
`Pushdown` over `Remote` against `Eager` over random formula and cell edits on the formula fixture
(cycles, text, NaN and `-0.0` among the values), in its own binary. Rival: a refresh that does
nothing.
### a_backend_can_be_built_over_the_exported_feed_alias
Review of #358: `over<S>(Arc<S>)` could not take the `SharedFeed` the same PR exports. The test builds
@@ -867,3 +924,36 @@ The same for `ClearCell`.
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.
### the_order_stems_are_listed_in_decides_a_result_today
Review of #360: stem order was invisible to every differential (a sorted `distinct_without` went
uncaught). It pins today's behaviour of a known evaluator defect (improvise-bw4p.30), not a contract. Eighteen products are stored in chain order with names that sort the other way, and one nested
`IF` formula reads the product before it. Along the chain it settles in one pass (the last link is 18);
sorted it needs one pass per link and the bound of sixteen leaves it `circular`. The test asserts both
values, that a sound source passes the differential and that a source listing the stems sorted is a
divergence. Rival: `DataStore::distinct_without` sorting fails it (the oracle then gives `circular`, so
the first assertion fails). Before it, a source sorting its stems passed every differential; only the core stems test (which compares two stores) saw a sort.
### writes_between_two_queries_of_a_refresh_are_healed_by_the_fixed_point
Review of #360: a source that changes between the queries of one refresh. Another client's writes land
right after the first fold; the cached Profit of Product A ends at 53, the new state's value, and not at
13 (the old revenue less the new cost). The documented guard is one snapshot per refresh. Rival:
none; this pins a behaviour that the fixed point gives and the note says may not be relied on.
### a_stem_that_appears_between_the_stem_list_and_the_folds_waits_for_the_next_refresh
The stem list is taken, then Product C is written: it has no formula cell until the next refresh,
which has it (5).
### a_cell_edit_on_a_pushdown_is_refused
Review of #360: `define` applied a cell edit to the model, whose guard was a `debug_assert`; now an
`assert!` of its own, with its own message. Rival: the assert removed fails it (the model's message
differs).
### a_pushdown_can_be_built_over_the_exported_source_alias
`Pushdown::new` took a generic `Arc<Q>` that could not take a `SharedSource`; the test builds one from
the alias and did not compile against the generic signature.
crates/improvise-core/src/model/source.rs
+139
View File
@@ -24,6 +24,16 @@ impl<'a> Select<'a> {
}
}
/// Whether the cell at `key` is one this select names.
pub fn admits(&self, key: &CellKey) -> bool {
let spanned = |(cat, items): &(&str, Vec<&str>)| {
key.get(cat).is_some_and(|held| items.contains(&held))
};
key.matches_partial(self.base)
&& self.spans.iter().all(spanned)
&& (!self.exact || key.0.len() == self.base.len() + self.spans.len())
}
/// Every cell containing `base` and one listed item of each span.
pub fn spanning(
base: &'a [(String, String)],
@@ -243,6 +253,20 @@ pub trait CellSource {
&'a self,
partial: &'a [(String, String)],
) -> Box<dyn Iterator<Item = (CellKey, CellValue)> + 'a>;
/// How many cells `select` names, leaving out those holding any of the `excluding` pairs.
fn count(&self, select: &Select, excluding: &[(String, String)]) -> usize;
/// The key of every cell with the categories in `without` removed, each once, in the order
/// first seen: the stems a formula is evaluated at, as `SELECT DISTINCT` gives them.
fn distinct_without(&self, without: &[String]) -> Vec<CellKey>;
}
/// Whether the cell at `key` holds any of the `pairs`.
fn holds_any(key: &CellKey, pairs: &[(String, String)]) -> bool {
pairs
.iter()
.any(|(cat, item)| key.get(cat) == Some(item.as_str()))
}
impl CellSource for DataStore {
@@ -282,6 +306,19 @@ impl CellSource for DataStore {
),
}
}
fn count(&self, select: &Select, excluding: &[(String, String)]) -> usize {
let named = self.keys(select.base).filter(|key| select.admits(key));
named.filter(|key| !holds_any(key, excluding)).count()
}
fn distinct_without(&self, without: &[String]) -> Vec<CellKey> {
let mut seen = std::collections::HashSet::new();
let stems = self
.iter_cells()
.map(|(key, _)| without.iter().fold(key, |k, cat| k.without(cat)));
stems.filter(|stem| seen.insert(stem.clone())).collect()
}
}
/// A table of rows read by scanning, the way a query engine with no index would. A second
@@ -337,6 +374,28 @@ impl CellSource for TableSource {
.cloned(),
)
}
fn count(&self, select: &Select, excluding: &[(String, String)]) -> usize {
let has = |key: &CellKey, (cat, item): &(String, String)| {
key.0.iter().any(|(c, i)| c == cat && i == item)
};
let named = self.rows.iter().filter(|(key, _)| Self::holds(key, select));
named
.filter(|(key, _)| !excluding.iter().any(|pair| has(key, pair)))
.count()
}
fn distinct_without(&self, without: &[String]) -> Vec<CellKey> {
let mut stems: Vec<CellKey> = Vec::new();
for (key, _) in &self.rows {
let kept = key.0.iter().filter(|(c, _)| !without.contains(c));
let stem = CellKey::new(kept.cloned().collect());
if !stems.contains(&stem) {
stems.push(stem);
}
}
stems
}
}
#[cfg(test)]
@@ -641,6 +700,86 @@ mod tests {
);
}
/// Every way to hide items of the fixture: none, each item of each category, and two at once.
fn hidden_sets() -> Vec<Vec<(String, String)>> {
let items = [("R", "E"), ("R", "W"), ("P", "A"), ("_Measure", "Note")];
let mut sets = vec![vec![]];
for (i, a) in items.iter().enumerate() {
sets.push(vec![pair(a.0, a.1)]);
for b in &items[i + 1..] {
sets.push(vec![pair(a.0, a.1), pair(b.0, b.1)]);
}
}
sets
}
#[test]
fn the_store_and_a_scanned_table_count_every_select_alike() {
let rows = rows();
let (store, table) = (store_of(&rows), TableSource { rows: rows.clone() });
let (mut asked, mut nonzero) = (0, 0);
for (base, spans) in selects(&rows) {
let spans: Vec<(&str, Vec<&str>)> = spans
.iter()
.map(|(c, i)| (c.as_str(), i.iter().map(String::as_str).collect()))
.collect();
for exact in [false, true] {
let select = Select::spanning(&base, &spans, exact);
for hidden in hidden_sets() {
let (a, b) = (store.count(&select, &hidden), table.count(&select, &hidden));
assert_eq!(a, b, "{base:?} {spans:?} exact={exact} hiding {hidden:?}");
asked += 1;
nonzero += usize::from(a > 0);
}
}
}
assert!(asked > 20_000, "the enumeration shrank to {asked}");
assert!(nonzero > 2_000, "{nonzero} of {asked} counts were not zero");
}
#[test]
fn hiding_an_item_removes_exactly_the_cells_that_hold_it() {
let rows = rows();
let store = store_of(&rows);
let none = Select::matching(&[]);
assert_eq!(store.count(&none, &[]), rows.len());
let east = [pair("R", "E")];
let held = rows.iter().filter(|(k, _)| k.get("R") == Some("E")).count();
assert_eq!(store.count(&none, &east), rows.len() - held);
assert_eq!(
store.count(&Select::matching(&east), &east),
0,
"a named hidden item matches nothing"
);
}
#[test]
fn the_store_and_a_scanned_table_list_the_same_stems() {
let rows = rows();
let (store, table) = (store_of(&rows), TableSource { rows: rows.clone() });
let cats = ["R", "P", "_Measure", "Nowhere"].map(String::from);
let mut asked = 0;
for mask in 0..(1u32 << cats.len()) {
let without: Vec<String> = cats
.iter()
.enumerate()
.filter(|(i, _)| mask >> i & 1 == 1)
.map(|(_, c)| c.clone())
.collect();
let (a, b) = (
store.distinct_without(&without),
table.distinct_without(&without),
);
assert_eq!(a, b, "without {without:?}, in first-seen order");
let mut sorted = a.clone();
sorted.sort();
sorted.dedup();
assert_eq!(sorted.len(), a.len(), "each stem once");
asked += a.len();
}
assert!(asked > 40, "only {asked} stems were compared");
}
/// The parts an engine would compute for `values`, found without `Summary::of`.
fn parts_of(values: &[CellValue]) -> Summary {
let numbers: Vec<f64> = values.iter().filter_map(|v| v.as_f64()).collect();
crates/improvise-core/src/model/types.rs
+3 -9
View File
@@ -1036,15 +1036,9 @@ impl Model {
// (improvise-6os replaced an O(n^2) Vec::contains).
let mut stems: IndexSet<CellKey> = IndexSet::new();
for (target_cat, _) in &formula_cats {
for key in self.source().keys(&[]) {
let stem = key.without(target_cat);
// Strip none_cats too: the grid queries without them.
let mut stripped = stem;
for nc in none_cats {
stripped = stripped.without(nc);
}
stems.insert(stripped);
}
// The target and the none_cats go: the grid queries without them.
let without = [std::slice::from_ref(target_cat), none_cats].concat();
stems.extend(self.source().distinct_without(&without));
}
// Fixed-point iteration.
crates/improvise-incremental/Cargo.toml
+2
View File
@@ -13,5 +13,7 @@ numtower = { path = "../numtower" }
zset = { path = "../zset" }
[dev-dependencies]
# `TableSource`, the scanning source the pushdown tests are built on.
improvise-core = { path = "../improvise-core", features = ["test-support"] }
improvise-io = { path = "../improvise-io" }
proptest = "1"
crates/improvise-incremental/src/eager.rs
+5 -13
View File
@@ -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
+3
View File
@@ -9,7 +9,9 @@ mod formulas;
mod group;
mod harness;
mod incremental;
mod pushdown;
mod reading;
mod reads;
mod shapes;
pub use delta::{CellDelta, DeltaValue, cell_delta, moved_records};
@@ -20,6 +22,7 @@ pub use harness::{
drawn_probes, edit_probes, probes, same_value,
};
pub use incremental::{Evaluation, Incremental};
pub use pushdown::Pushdown;
pub use reading::Reading;
pub use shapes::Meter;
crates/improvise-incremental/src/pushdown.rs
+98
View File
@@ -0,0 +1,98 @@
//! 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, reads};
use improvise_core::model::cell::{CellKey, CellValue};
use improvise_core::model::source::{Select, SharedSource};
use improvise_core::model::{Aggregation, Model};
/// 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.
pub struct Pushdown {
model: Model,
none_cats: Vec<String>,
}
impl Pushdown {
/// `definition` supplies the categories, formulas and aggregations; its own store is not read.
pub fn new(definition: Model, source: SharedSource, none_cats: Vec<String>) -> Self {
let mut backend = Pushdown {
model: definition.over(source),
none_cats,
};
backend.refresh();
backend
}
pub fn model(&self) -> &Model {
&self.model
}
/// 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();
}
/// The items a stored cell may hold in `cat` to be named by `item`, when `item` names a group.
fn members(&self, cat: &str, item: &str) -> Vec<String> {
let group = self
.model
.category(cat)
.filter(|c| !c.items.contains_key(item));
let members = group.map(|c| c.items_in_group(item)).unwrap_or_default();
match members.is_empty() {
true => Vec::new(),
false => std::iter::once(item)
.chain(members)
.map(String::from)
.collect(),
}
}
}
impl Derived for Pushdown {
fn value(&self, key: &CellKey, agg: Aggregation) -> Option<CellValue> {
reads::value(&self.model, key, agg)
}
fn raw_fold(&self, partial: &CellKey) -> Option<CellValue> {
reads::raw_fold(&self.model, partial)
}
fn covered(&self, key: &CellKey, reading: &Reading) -> Option<CellValue> {
reads::covered(&self.model, key, reading)
}
fn total(&self, key: &CellKey, over: &[String], reading: &Reading) -> Option<CellValue> {
reads::total(&self.model, key, over, reading)
}
fn occupancy(&self, partial: &CellKey, reading: &Reading) -> usize {
let (mut base, mut held) = (Vec::new(), Vec::new());
for (cat, item) in &partial.0 {
match self.members(cat, item) {
members if members.is_empty() => base.push((cat.clone(), item.clone())),
members => held.push((cat.clone(), members)),
}
}
let spans: Vec<(&str, Vec<&str>)> = held
.iter()
.map(|(cat, items)| (cat.as_str(), items.iter().map(String::as_str).collect()))
.collect();
let hidden: Vec<(String, String)> = reading.hidden().cloned().collect();
let select = Select::spanning(&base, &spans, false);
self.model.source().count(&select, &hidden)
}
}
crates/improvise-incremental/src/reads.rs
+33
View File
@@ -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
+355
View File
@@ -0,0 +1,355 @@
//! `Pushdown` over a remote table against `Eager`: values, hidden items, groups, formulas.
//! Without a real SQL engine only the contract is proven: `context/code/improvise-incremental.md`.
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 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,
};
fn run(model: &Model, views: &[View], none: &[String], edits: &[Edit]) -> usize {
let candidate = Pushed::new(model, none.to_vec());
match differential_in(model, views, none, candidate, edits) {
Ok(report) => report.probes,
Err(d) => panic!("{d}"),
}
}
fn agg_edits(measure: &str) -> Vec<Edit> {
let all = [
AggFunc::Count,
AggFunc::Min,
AggFunc::Max,
AggFunc::Avg,
AggFunc::Sum,
];
all.into_iter()
.map(|func| Edit::SetMeasureAgg {
measure: measure.into(),
func,
})
.collect()
}
#[test]
fn a_pushdown_matches_eager_over_the_shipped_documents() {
let (mut compared, mut files_done) = (0, Vec::<String>::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();
}
compared += run(&model, &views, &none, &edits);
}
assert!(compared > 10_000, "only {compared} answers were compared");
}
#[test]
fn a_pushdown_matches_eager_on_hidden_items_and_collapsed_groups_under_every_aggregation() {
for wb in [hidden_workbook(), hidden_counted_workbook()] {
let views: Vec<View> = wb.views.values().cloned().collect();
let none = none_of(&wb);
let mut edits = edits_for(&wb.model);
edits.extend(agg_edits("Revenue"));
let hidden = key("East", "A", "Revenue");
edits.push(Edit::SetCell(hidden.clone(), CellValue::Number(f64::NAN)));
edits.push(Edit::SetCell(hidden.clone(), CellValue::Number(-0.0)));
edits.push(Edit::SetCell(hidden, CellValue::Text("t".into())));
// A cell stored under the group's own name is a cell its fold and its count both see.
let named = key("Coastal", "B", "Revenue");
edits.push(Edit::SetCell(named, CellValue::Number(4.0)));
assert!(run(&wb.model, &views, &none, &edits) > 1_000);
}
}
#[test]
fn a_pushdown_matches_eager_on_formulas_groups_and_wide_keys() {
let ratio = ratio_workbook();
let views: Vec<View> = ratio.views.values().cloned().collect();
let mut edits = edits_for(&ratio.model);
edits.extend(agg_edits("Revenue"));
run(&ratio.model, &views, &none_of(&ratio), &edits);
let hand = hand_model();
let mut edits = agg_edits("Revenue");
edits.extend(agg_edits("Note"));
run(&hand, &[], &["Region".to_string()], &edits);
run(&grouped_model(), &[], &[], &edits_for(&grouped_model()));
let wide = wide_workbook();
let views: Vec<View> = wide.views.values().cloned().collect();
run(
&wide.model,
&views,
&none_of(&wide),
&edits_for(&wide.model),
);
}
#[test]
fn a_fold_is_one_query_and_sends_no_row() {
let wb = hidden_counted_workbook();
let none = none_of(&wb);
let pushed = Pushed::new(&wb.model, none);
let at = |name: &str| key("Inland", name, "Revenue");
let before = pushed.remote.asked();
pushed.pd.raw_fold(&CellKey::new(vec![
("Product".into(), "A".into()),
("_Measure".into(), "Revenue".into()),
]));
let reading = Reading::of_view(wb.active_view(), &none_of(&wb));
let coastal = key("Coastal", "A", "Revenue");
pushed.pd.covered(&coastal, &reading);
let total = key("Inland", "A", "Revenue").without("Region");
pushed.pd.total(&total, &["Region".to_string()], &reading);
let spent = pushed.remote.asked();
assert_eq!(spent.summaries - before.summaries, 3, "one query per fold");
assert_eq!(spent.rows_sent, before.rows_sent, "and no row sent back");
let one = pushed.pd.value(&at("A"), Aggregation::Exact);
assert_eq!(one, Some(CellValue::Number(2.0)));
assert_eq!(pushed.remote.asked().lookups - spent.lookups, 1);
}
#[test]
fn occupancy_is_one_count_whatever_the_groups_and_hidden_items() {
let wb = hidden_workbook();
let pushed = Pushed::new(&wb.model, none_of(&wb));
let reading = Reading::of_view(wb.active_view(), &none_of(&wb));
let before = pushed.remote.asked();
let coastal = CellKey::new(vec![
("Region".into(), "Coastal".into()),
("Product".into(), "A".into()),
]);
let seen = pushed.pd.occupancy(&coastal, &reading);
let spent = pushed.remote.asked();
assert_eq!(seen, 1, "East is hidden, West holds A");
assert_eq!(spent.counts - before.counts, 1);
assert_eq!(spent.rows_sent, before.rows_sent);
}
#[test]
fn formula_cells_are_computed_from_the_distinct_stems_not_from_every_cell() {
let model = hand_model();
let cells = model.data.iter_cells().count() as u64;
let pushed = Pushed::new(&model, vec!["Region".to_string()]);
let asked = pushed.remote.asked();
assert_eq!(asked.stem_lists, 1, "one list for the one formula");
assert_eq!(
asked.rows_sent, 2,
"Product A and Product B, of {cells} cells"
);
assert_eq!((asked.key_lists, asked.cell_lists), (0, 0));
assert!(
asked.summaries > 0,
"and the folds the formula reads are queries"
);
}
#[test]
fn a_stored_cell_is_current_and_a_formula_cell_waits_for_a_refresh() {
let mut pushed = Pushed::new(&hand_model(), vec!["Region".to_string()]);
let revenue = key("East", "A", "Revenue");
let profit = CellKey::new(vec![
("Product".into(), "A".into()),
("_Measure".into(), "Profit".into()),
]);
let fold = |p: &Pushed| p.pd.value(&profit, Aggregation::OverHidden);
assert_eq!(fold(&pushed), Some(CellValue::Number(9.0)));
pushed
.remote
.write(revenue.clone(), Some(CellValue::Number(50.0)));
let now = pushed.pd.value(&revenue, Aggregation::Exact);
assert_eq!(now, Some(CellValue::Number(50.0)), "a read is a query");
assert_eq!(
fold(&pushed),
Some(CellValue::Number(9.0)),
"the cache is stale"
);
pushed.pd.refresh();
assert_eq!(fold(&pushed), Some(CellValue::Number(49.0)));
}
#[test]
fn a_source_that_forgets_the_hidden_items_is_caught_as_a_divergence() {
let wb = hidden_workbook();
let views: Vec<View> = wb.views.values().cloned().collect();
let none = none_of(&wb);
let faulty = Pushed::through(&wb.model, none.clone(), IgnoresHidden);
let ran = differential_in(&wb.model, &views, &none, faulty, &[]);
let divergence = ran.expect_err("a wrong hidden filter must not pass");
assert!(
matches!(*divergence.probe, Probe::Occupancy(..)),
"{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_today() {
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/random_pushdown.rs
+22
View File
@@ -0,0 +1,22 @@
//! `Pushdown` over a remote table against `Eager` over random formula and cell edits (NaN,
//! `-0.0`, cycles and text among them). Evidence: `context/test-evidence/improvise-incremental.md`.
mod support;
use improvise_incremental::differential_in;
use proptest::prelude::*;
use support::remote::Pushed;
use support::{any_formula_edit, formula_workbook};
proptest! {
#[test]
fn a_pushdown_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 = Pushed::new(&wb.model, none.clone());
let ran = differential_in(&wb.model, &[], &none, candidate, &edits);
prop_assert!(ran.is_ok(), "{}", ran.unwrap_err());
}
}
crates/improvise-incremental/tests/support/live.rs
+8
View File
@@ -88,6 +88,14 @@ impl CellSource for LiveTable {
}
Box::new(cells.into_iter())
}
fn count(&self, select: &Select, excluding: &[(String, String)]) -> usize {
self.state.lock().unwrap().visible.count(select, excluding)
}
fn distinct_without(&self, without: &[String]) -> Vec<CellKey> {
self.state.lock().unwrap().visible.distinct_without(without)
}
}
impl ChangeSource for LiveTable {
crates/improvise-incremental/tests/support/mod.rs
+1
View File
@@ -2,6 +2,7 @@
#![allow(dead_code)]
pub mod live;
pub mod remote;
use improvise_core::formula::ast::AggFunc;
use improvise_core::formula::parse_formula;
crates/improvise-incremental/tests/support/remote.rs
+315
View File
@@ -0,0 +1,315 @@
//! A remote table: an in-memory stand-in for a query engine, read by scanning and counted.
//! It is not `Eager` and shares no evaluator with it; it answers queries over rows.
#![allow(dead_code)]
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.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Asked {
pub lookups: u64,
pub summaries: u64,
pub counts: u64,
pub stem_lists: u64,
pub key_lists: u64,
pub cell_lists: u64,
/// Keys, cells and stems sent back by `keys`, `cells` and `distinct_without`: rows, not aggregates.
pub rows_sent: u64,
}
#[derive(Default)]
pub struct Remote {
table: Mutex<TableSource>,
asked: Mutex<Asked>,
}
impl Remote {
pub fn of(rows: impl IntoIterator<Item = (CellKey, CellValue)>) -> Self {
let table = TableSource {
rows: rows.into_iter().collect(),
};
Remote {
table: Mutex::new(table),
asked: Mutex::default(),
}
}
/// Writes as another client of the database would: in place, or at the end, or removed.
pub fn write(&self, key: CellKey, value: Option<CellValue>) {
let mut table = self.table.lock().unwrap();
let at = table.rows.iter().position(|(k, _)| *k == key);
match (at, value) {
(Some(i), Some(v)) => table.rows[i].1 = v,
(None, Some(v)) => table.rows.push((key, v)),
(Some(i), None) => {
table.rows.remove(i);
}
(None, None) => {}
}
}
pub fn asked(&self) -> Asked {
*self.asked.lock().unwrap()
}
fn ask(&self, f: impl FnOnce(&mut Asked)) {
f(&mut self.asked.lock().unwrap());
}
fn table(&self) -> std::sync::MutexGuard<'_, TableSource> {
self.table.lock().unwrap()
}
}
impl CellSource for Remote {
fn value_at(&self, key: &CellKey) -> Option<CellValue> {
self.ask(|a| a.lookups += 1);
self.table().value_at(key)
}
fn summarize(&self, select: &Select) -> Summary {
self.ask(|a| a.summaries += 1);
self.table().summarize(select)
}
fn keys<'a>(
&'a self,
partial: &'a [(String, String)],
) -> Box<dyn Iterator<Item = CellKey> + 'a> {
let keys: Vec<CellKey> = self.table().keys(partial).collect();
self.ask(|a| {
a.key_lists += 1;
a.rows_sent += keys.len() as u64;
});
Box::new(keys.into_iter())
}
fn cells<'a>(
&'a self,
partial: &'a [(String, String)],
) -> Box<dyn Iterator<Item = (CellKey, CellValue)> + 'a> {
let cells: Vec<_> = self.table().cells(partial).collect();
self.ask(|a| {
a.cell_lists += 1;
a.rows_sent += cells.len() as u64;
});
Box::new(cells.into_iter())
}
fn count(&self, select: &Select, excluding: &[(String, String)]) -> usize {
self.ask(|a| a.counts += 1);
self.table().count(select, excluding)
}
fn distinct_without(&self, without: &[String]) -> Vec<CellKey> {
let stems = self.table().distinct_without(without);
self.ask(|a| {
a.stem_lists += 1;
a.rows_sent += stems.len() as u64;
});
stems
}
}
/// A source with a fault, for checking that a differential can see it: it forgets the hidden
/// items a count leaves out.
pub struct IgnoresHidden(pub Arc<Remote>);
impl CellSource for IgnoresHidden {
fn value_at(&self, key: &CellKey) -> Option<CellValue> {
self.0.value_at(key)
}
fn summarize(&self, select: &Select) -> Summary {
self.0.summarize(select)
}
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, &[])
}
fn distinct_without(&self, without: &[String]) -> Vec<CellKey> {
self.0.distinct_without(without)
}
}
/// 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 {
pub pd: Pushdown,
pub remote: Arc<Remote>,
}
impl Pushed {
pub fn new(model: &Model, none_cats: Vec<String>) -> Pushed {
let remote = Arc::new(Remote::of(rows_of(model)));
let pd = Pushdown::new(bare(model), remote.clone(), none_cats);
Pushed { pd, remote }
}
/// The same, reading the table through `wrap`.
pub fn through<Q: CellSource + Send + Sync + 'static>(
model: &Model,
none_cats: Vec<String>,
wrap: impl FnOnce(Arc<Remote>) -> Q,
) -> Pushed {
let remote = Arc::new(Remote::of(rows_of(model)));
let pd = Pushdown::new(bare(model), Arc::new(wrap(remote.clone())), none_cats);
Pushed { pd, remote }
}
}
fn rows_of(model: &Model) -> Vec<(CellKey, CellValue)> {
model
.data
.iter_cells()
.map(|(k, v)| (k, v.clone()))
.collect()
}
/// `model` with no cells of its own.
fn bare(model: &Model) -> Model {
let mut bare = model.clone();
bare.data = improvise_core::model::cell::DataStore::new();
bare
}
impl Derived for Pushed {
fn value(&self, key: &CellKey, agg: Aggregation) -> Option<CellValue> {
self.pd.value(key, agg)
}
fn raw_fold(&self, partial: &CellKey) -> Option<CellValue> {
self.pd.raw_fold(partial)
}
fn covered(&self, key: &CellKey, reading: &Reading) -> Option<CellValue> {
self.pd.covered(key, reading)
}
fn total(&self, key: &CellKey, over: &[String], reading: &Reading) -> Option<CellValue> {
self.pd.total(key, over, reading)
}
fn occupancy(&self, partial: &CellKey, reading: &Reading) -> usize {
self.pd.occupancy(partial, reading)
}
}
impl Maintained for Pushed {
fn apply(&mut self, edit: &Edit) {
match edit {
Edit::SetCell(key, value) => self.remote.write(key.clone(), Some(value.clone())),
Edit::ClearCell(key) => self.remote.write(key.clone(), None),
_ => return self.pd.define(edit),
}
self.pd.refresh();
}
}