improv_engine/
lib.rs

1//! Improv computation engine.
2//!
3//! Phase 1. Before building the full formula compiler on differential dataflow,
4//! this module contains the **DD-viability spike** (see AGENT_STEERING.md
5//! Constraint 1): proof that a `Coordinate`-keyed, `f64`-valued incremental
6//! computation (`Revenue = Price * Quantity`) works on differential dataflow,
7//! including a live delta after the initial run.
8//!
9//! Encoding at the dataflow boundary (timely 0.13's exchange bound is serde
10//! `Serialize + Deserialize` plus `Ord + Hash` for keys; a `Coordinate`'s
11//! `BTreeMap` and a raw `f64` are encoded rather than used directly):
12//! * DD *key* = the coordinate as a sorted `Vec<(u32, u32)>` (category, item).
13//!   Deterministic and `Ord + Hash`.
14//! * cell *value* = [`CellValue`]: a number as `f64::to_bits()` (so it is
15//!   `Eq + Hash`), or a boolean, text, date (epoch ms) or error.
16//! * diff `R` stays `isize` multiplicity; a `reduce` collapses to one value/key.
17//!
18//! `core_model` types convert to/from these encodings at the edge.
19#![cfg_attr(coverage_nightly, feature(coverage_attribute))]
20// The attribute is only used inside #[cfg(test)] modules, so non-test
21// coverage builds would otherwise warn that the feature is unused.
22#![cfg_attr(coverage_nightly, allow(unused_features))]
23
24pub mod builtins;
25pub mod compiler;
26pub mod dataflow;
27pub mod external;
28mod hosteval;
29pub mod plan;
30pub mod reduce;
31pub mod scalar;
32pub mod session;
33pub mod typed;
34
35use improv_core_model::{CategoryId, Coordinate, ItemId, Value};
36
37/// The dataflow encoding of a coordinate: sorted `(category_id, item_id)` pairs.
38pub type CoordKey = Vec<(u32, u32)>;
39
40/// The dataflow encoding of a cell value.
41///
42/// Differential-dataflow data must be `Ord + Eq + Hash + Clone + 'static` and
43/// exchange-safe. `f64` is none of the first three, so numbers ride as their
44/// bit pattern; the other variants make Text / Boolean / Error first-class in
45/// the dataflow (the non-numeric value lane). Numbers stay the fast path.
46#[derive(
47    Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, serde::Serialize, serde::Deserialize,
48)]
49pub enum CellValue {
50    /// A number, as `f64::to_bits()` (total order via `f64::total_cmp` at the
51    /// boundary; bit order is fine for equality/hashing/exchange).
52    Num(u64),
53    Bool(bool),
54    Text(String),
55    /// A UTC timestamp as milliseconds since the Unix epoch. `i64` keeps the
56    /// value totally-ordered/`Hash`/`Ord` for the DD lane (unlike a chrono
57    /// `DateTime`, which is fine but this avoids a chrono dep in the key type).
58    Date(i64),
59    /// An error value; the `u8` is the `ValueError` kind discriminant so error
60    /// cells are distinguishable and propagate through operators.
61    Err(u8),
62}
63
64impl CellValue {
65    /// Wrap an `f64`.
66    pub fn num(n: f64) -> Self {
67        CellValue::Num(n.to_bits())
68    }
69    /// The number this holds, if it is `Num`.
70    pub fn as_num(&self) -> Option<f64> {
71        match self {
72            CellValue::Num(b) => Some(f64::from_bits(*b)),
73            _ => None,
74        }
75    }
76    /// Encode a boolean as the numeric-lane 1.0/0.0 (back-compat with the
77    /// comparison/logical convention) — used where downstream expects a number.
78    pub fn from_bool_numeric(b: bool) -> Self {
79        CellValue::num(if b { 1.0 } else { 0.0 })
80    }
81    /// Convert a model `Value` into a `CellValue` for feeding inputs.
82    pub fn from_model_value(v: &Value) -> Option<CellValue> {
83        match v {
84            Value::Number(n) => Some(CellValue::num(*n)),
85            Value::Boolean(b) => Some(CellValue::Bool(*b)),
86            Value::Text(t) => Some(CellValue::Text(t.clone())),
87            Value::Enum(e) => Some(CellValue::num(*e as f64)),
88            Value::DateTime(dt) => Some(CellValue::Date(dt.timestamp_millis())),
89            Value::Error(e) => Some(CellValue::Err(e.kind.error_type().unwrap_or(0))),
90        }
91    }
92}
93
94impl std::fmt::Display for CellValue {
95    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
96        match self {
97            CellValue::Num(bits) => f.write_str(&display_number(f64::from_bits(*bits))),
98            CellValue::Bool(b) => write!(f, "{b}"),
99            CellValue::Text(t) => write!(f, "{t}"),
100            CellValue::Date(ms) => {
101                // Render as an RFC3339-ish UTC timestamp.
102                match chrono::DateTime::<chrono::Utc>::from_timestamp_millis(*ms) {
103                    Some(dt) => write!(f, "{}", dt.to_rfc3339()),
104                    None => write!(f, "date({ms})"),
105                }
106            }
107            CellValue::Err(c) => f.write_str(
108                improv_core_model::ValueErrorKind::from_error_type(*c).map_or("#ERR", |k| k.code()),
109            ),
110        }
111    }
112}
113
114/// A number as people read it: at most 15 significant digits, the precision
115/// an `f64` always carries through decimal, so binary noise like
116/// `58.650000000000006` (from `1173 * 0.05`) shows as `58.65`. The value
117/// itself is untouched; exports (CSV, SQL, Postgres) keep every bit.
118pub fn display_number(n: f64) -> String {
119    if !n.is_finite() || n == 0.0 {
120        return if n == 0.0 { "0".into() } else { n.to_string() };
121    }
122    // `{:.14e}` rounds to 15 significant digits; parsing it back and printing
123    // with `{}` gives the shortest decimal for that rounded value.
124    let rounded: f64 = format!("{n:.14e}").parse().unwrap_or(n);
125    rounded.to_string()
126}
127
128/// Encode a `Coordinate` as its dataflow key (BTreeMap already iterates sorted).
129pub fn encode_coord(c: &Coordinate) -> CoordKey {
130    c.dims.iter().map(|(cat, it)| (cat.0, it.0)).collect()
131}
132
133/// Decode a dataflow key back into a `Coordinate`.
134pub fn decode_coord(k: &CoordKey) -> Coordinate {
135    Coordinate::from_pairs(k.iter().map(|(c, i)| (CategoryId(*c), ItemId(*i))))
136}
137
138/// Project a coord key down to a subset of category ids (join key / broadcast).
139/// Exactly the pairs whose category is kept, in order (proven in
140/// [`improv_verified::keys::project`]).
141pub fn project_key(k: &CoordKey, keep: &[u32]) -> CoordKey {
142    improv_verified::keys::project(k, keep)
143}
144
145#[cfg(test)]
146#[cfg_attr(coverage_nightly, coverage(off))]
147mod spike {
148    //! The DD-viability spike. If this compiles and passes, differential
149    //! dataflow is a viable engine substrate for Improv's cube model.
150
151    use super::*;
152    use differential_dataflow::input::InputSession;
153    use differential_dataflow::operators::{Join, Reduce};
154    use std::sync::{Arc, Mutex};
155
156    const TIME: u32 = 1;
157    const PRODUCT: u32 = 2;
158
159    fn ck(pairs: &[(u32, u32)]) -> CoordKey {
160        let c = Coordinate::from_pairs(
161            pairs
162                .iter()
163                .map(|(cat, it)| (CategoryId(*cat), ItemId(*it))),
164        );
165        encode_coord(&c)
166    }
167
168    /// Compute `Revenue[Time,Product] = Price[Product] * Quantity[Time,Product]`
169    /// incrementally on DD, then apply a delta and confirm the affected cell
170    /// updates while the unaffected cell does not recompute.
171    #[test]
172    fn dd_revenue_is_incremental() {
173        let out: Arc<Mutex<Vec<(u64, CoordKey, f64)>>> = Arc::new(Mutex::new(Vec::new()));
174        let out2 = out.clone();
175
176        timely::execute::execute_directly(move |worker| {
177            let mut price_in: InputSession<u64, (CoordKey, u64), isize> = InputSession::new();
178            let mut qty_in: InputSession<u64, (CoordKey, u64), isize> = InputSession::new();
179            let probe_out = out2.clone();
180
181            worker.dataflow(|scope| {
182                let price = price_in.to_collection(scope);
183                let qty = qty_in.to_collection(scope);
184
185                // Key Price by Product; carry the full coord + qty so we can
186                // rebuild the Revenue key after the join broadcasts over Time.
187                let price_by_product = price.map(|(k, v)| (project_key(&k, &[PRODUCT]), v));
188                let qty_by_product = qty.map(|(k, v)| (project_key(&k, &[PRODUCT]), (k, v)));
189
190                let revenue = qty_by_product
191                    .join(&price_by_product)
192                    .map(|(_p, ((full, qbits), pbits))| {
193                        let rev = f64::from_bits(pbits) * f64::from_bits(qbits);
194                        (full, rev.to_bits())
195                    })
196                    .reduce(|_key, input, output| {
197                        for (bits, mult) in input {
198                            if *mult > 0 {
199                                output.push((**bits, 1isize));
200                            }
201                        }
202                    });
203
204                revenue.inspect(move |((k, bits), time, _diff)| {
205                    probe_out
206                        .lock()
207                        .unwrap()
208                        .push((*time, k.clone(), f64::from_bits(*bits)));
209                });
210            });
211
212            // Round 0: initial data.
213            price_in.advance_to(0);
214            price_in.insert((ck(&[(PRODUCT, 20)]), 10.0f64.to_bits())); // Widget A = 10
215            price_in.insert((ck(&[(PRODUCT, 21)]), 20.0f64.to_bits())); // Widget B = 20
216            qty_in.advance_to(0);
217            qty_in.insert((ck(&[(TIME, 10), (PRODUCT, 20)]), 100.0f64.to_bits()));
218            qty_in.insert((ck(&[(TIME, 10), (PRODUCT, 21)]), 50.0f64.to_bits()));
219            price_in.advance_to(1);
220            qty_in.advance_to(1);
221            price_in.flush();
222            qty_in.flush();
223            while price_in.time() < &1 || qty_in.time() < &1 {
224                worker.step();
225            }
226
227            // Round 1: delta -- Quantity[2025,WidgetA] 100 -> 120.
228            qty_in.remove((ck(&[(TIME, 10), (PRODUCT, 20)]), 100.0f64.to_bits()));
229            qty_in.insert((ck(&[(TIME, 10), (PRODUCT, 20)]), 120.0f64.to_bits()));
230            price_in.advance_to(2);
231            qty_in.advance_to(2);
232            price_in.flush();
233            qty_in.flush();
234            while qty_in.time() < &2 {
235                worker.step();
236            }
237        });
238
239        let events = out.lock().unwrap();
240        let a = ck(&[(TIME, 10), (PRODUCT, 20)]);
241        let b = ck(&[(TIME, 10), (PRODUCT, 21)]);
242
243        assert_eq!(final_value(&events, &b), 1000.0, "B = 20*50, unchanged");
244        assert_eq!(final_value(&events, &a), 1200.0, "A updated to 10*120");
245
246        // Incrementality: B must not re-emit at round >= 2 (only A changed).
247        let b_recomputed = events.iter().any(|(t, k, _)| *t >= 2 && k == &b);
248        assert!(!b_recomputed, "unaffected cell B must not recompute");
249    }
250
251    fn final_value(events: &[(u64, CoordKey, f64)], target: &CoordKey) -> f64 {
252        events
253            .iter()
254            .rfind(|(_, k, _)| k == target)
255            .map(|(_, _, v)| *v)
256            .unwrap_or(0.0)
257    }
258
259    #[test]
260    fn date_value_enters_the_dd_lane() {
261        // A model Date value maps to the Date(i64 millis) lane and displays as
262        // an RFC3339 timestamp.
263        let dt = chrono::DateTime::<chrono::Utc>::from_timestamp(1_700_000_000, 0).unwrap();
264        let cv = CellValue::from_model_value(&Value::DateTime(dt)).expect("date -> cell");
265        assert_eq!(cv, CellValue::Date(1_700_000_000_000));
266        assert!(format!("{cv}").starts_with("2023-11-"));
267    }
268}
269
270#[cfg(test)]
271#[cfg_attr(coverage_nightly, coverage(off))]
272mod tests {
273    #[test]
274    fn display_number_hides_binary_noise_not_digits() {
275        use super::display_number as d;
276        assert_eq!(d(1173.0 * 0.05), "58.65");
277        assert_eq!(d(936.0 * 0.05), "46.8");
278        assert_eq!(d(0.1 + 0.2), "0.3");
279        assert_eq!(d(1000.0), "1000");
280        assert_eq!(d(-2.5), "-2.5");
281        assert_eq!(d(0.0), "0");
282        assert_eq!(d(-0.0), "0");
283        assert_eq!(d(1.0 / 3.0), "0.333333333333333");
284        assert_eq!(d(123_456_789_012_345.0), "123456789012345");
285        assert_eq!(d(1e300), 1e300.to_string(), "no exponent form, as before");
286        assert_eq!(d(1.5e-7), "0.00000015");
287        assert_eq!(d(f64::NAN), "NaN");
288        assert_eq!(d(f64::INFINITY), "inf");
289        assert_eq!(super::CellValue::num(1173.0 * 0.05).to_string(), "58.65");
290    }
291
292    use super::*;
293    use improv_core_model::{ValueError, ValueErrorKind};
294
295    #[test]
296    fn cell_value_conversions_and_display() {
297        assert_eq!(CellValue::from_bool_numeric(true), CellValue::num(1.0));
298        assert_eq!(CellValue::from_bool_numeric(false), CellValue::num(0.0));
299        assert_eq!(CellValue::Bool(true).as_num(), None);
300        let cv = |v: Value| CellValue::from_model_value(&v).unwrap();
301        assert_eq!(cv(Value::Boolean(true)), CellValue::Bool(true));
302        assert_eq!(cv(Value::Text("t".into())), CellValue::Text("t".into()));
303        assert_eq!(cv(Value::Enum(3)), CellValue::num(3.0));
304        assert_eq!(
305            cv(Value::Error(ValueError::new(ValueErrorKind::MissingInput))),
306            CellValue::Err(0)
307        );
308        let shown: Vec<String> = [
309            CellValue::num(1.5),
310            CellValue::Bool(false),
311            CellValue::Text("hi".into()),
312            CellValue::Date(0),
313            CellValue::Date(i64::MAX),
314            CellValue::Err(1),
315        ]
316        .iter()
317        .map(ToString::to_string)
318        .collect();
319        assert_eq!(
320            shown[..4],
321            ["1.5", "false", "hi", "1970-01-01T00:00:00+00:00"]
322        );
323        assert_eq!(shown[4], format!("date({})", i64::MAX));
324        assert_eq!(shown[5], "#NULL!");
325        assert_eq!(CellValue::Err(0).to_string(), "#ERR");
326        assert_eq!(CellValue::Err(2).to_string(), "#DIV/0!");
327        assert_eq!(
328            cv(Value::Error(ValueError::new(ValueErrorKind::Na))),
329            CellValue::Err(7)
330        );
331    }
332
333    #[test]
334    fn coord_keys_round_trip_and_project() {
335        let c = Coordinate::from_pairs([(CategoryId(2), ItemId(20)), (CategoryId(1), ItemId(10))]);
336        let k = encode_coord(&c);
337        assert_eq!(k, vec![(1, 10), (2, 20)]);
338        assert_eq!(decode_coord(&k), c);
339        assert_eq!(project_key(&k, &[2]), vec![(2, 20)]);
340    }
341}
improv_engine/lib.rs

improv_engine/
lib.rs
