Skip to main content

vantage_vista/
aggregate.rs

1//! Describing a reduction, so a driver can decide whether it can perform one.
2//!
3//! An aggregation is not a value — it is a **new set**. `SELECT count(*) …`
4//! yields a one-row table, `GROUP BY` yields one row per group, and both are
5//! ordinary sets you can then condition, order or count again. So the verb
6//! this spec feeds ([`TableShell::aggregate_vista`](crate::TableShell::aggregate_vista))
7//! returns a [`Vista`](crate::Vista), not a number, and every consumer keeps
8//! the one shape it already knows.
9
10/// What to reduce, how, and what to call the result.
11///
12/// **Conditions are not here.** Narrow the source vista first, then aggregate
13/// it — the order SQL uses, where the filter belongs to the inner query and
14/// the aggregate selects from the result. Narrowing already reports its own
15/// failure (`add_condition_eq` returns `Err` on a driver that can't express a
16/// term), so a caller knows to fall back *before* it asks for a reduction, and
17/// no driver is ever handed a filter it will silently ignore.
18#[derive(Debug, Clone, PartialEq)]
19pub struct AggregateSpec {
20    /// The reduction: `count`, `sum`, `avg`, `min`, `max`, `distinct`, or
21    /// whatever else a driver understands. A driver that doesn't recognise
22    /// the name answers `None`.
23    pub op: String,
24    /// Column being reduced. `None` for reductions over whole rows (`count`).
25    pub column: Option<String>,
26    /// Column name the result is published under — the single column of a
27    /// scalar aggregate's single row.
28    pub alias: String,
29    /// Group keys. Empty means one row out; otherwise one row per distinct
30    /// combination, carrying the key columns alongside `alias`.
31    pub group_by: Vec<String>,
32}
33
34impl AggregateSpec {
35    /// A scalar reduction over whole rows, e.g. `count`.
36    pub fn new(op: impl Into<String>, alias: impl Into<String>) -> Self {
37        Self {
38            op: op.into(),
39            column: None,
40            alias: alias.into(),
41            group_by: Vec::new(),
42        }
43    }
44
45    /// Reduce `column` rather than whole rows.
46    pub fn column(mut self, column: impl Into<String>) -> Self {
47        self.column = Some(column.into());
48        self
49    }
50
51    /// Emit one row per distinct value of `column`.
52    pub fn group_by(mut self, column: impl Into<String>) -> Self {
53        self.group_by.push(column.into());
54        self
55    }
56
57    /// A stable, unique name for the set this spec derives from a source.
58    ///
59    /// `source_key` identifies the source **as narrowed** — pass
60    /// [`Vista::index_key`](crate::Vista::index_key), which already renders a
61    /// vista's conditions and sort in a canonical, order-independent form.
62    /// That is what keeps two differently-filtered aggregates apart without
63    /// this type knowing anything about conditions.
64    ///
65    /// Two things depend on the result. A derived Dio caches under it, so it
66    /// must not collide with the source or with a differently-shaped aggregate
67    /// over the same source. And the same question asked twice must produce
68    /// the *same* key, so a page showing one number in two places shares one
69    /// engine instead of computing it twice.
70    ///
71    /// `#` separates a source from its derivation: it has no other meaning
72    /// here (dots are column paths, `/` already separates datasource from
73    /// table), and it sorts each derived table next to its origin in a cache
74    /// directory.
75    ///
76    /// ```
77    /// # use vantage_vista::AggregateSpec;
78    /// let spec = AggregateSpec::new("sum", "bytes").column("Size");
79    /// assert_eq!(spec.cache_key("events|c:Event=opened|s:"),
80    ///            "events|c:Event=opened|s:#sum(Size)");
81    /// ```
82    pub fn cache_key(&self, source_key: &str) -> String {
83        let mut key = format!("{source_key}#{}", self.op);
84        if let Some(column) = &self.column {
85            key.push_str(&format!("({column})"));
86        }
87        if !self.group_by.is_empty() {
88            // NOT sorted: grouping by (a, b) is a different table from
89            // (b, a) — the key columns come out in this order.
90            key.push_str(&format!("/by:{}", self.group_by.join(",")));
91        }
92        key
93    }
94}
95
96#[cfg(test)]
97mod tests {
98    use super::*;
99    use crate::mocks::MockShell;
100    use crate::vista::Vista;
101    use ciborium::Value as CborValue;
102
103    fn text(s: &str) -> CborValue {
104        CborValue::Text(s.to_string())
105    }
106
107    /// A narrowed source, keyed the way a caller is expected to key it.
108    fn source_key(conditions: &[(String, CborValue)]) -> String {
109        let vista = Vista::new("email_events", Box::new(MockShell::new()));
110        vista.index_key(conditions, None)
111    }
112
113    #[test]
114    fn a_bare_reduction_keys_on_its_op() {
115        let key = AggregateSpec::new("count", "total").cache_key(&source_key(&[]));
116        assert!(key.ends_with("#count"), "got {key}");
117    }
118
119    #[test]
120    fn the_reduced_column_is_part_of_the_key() {
121        let key = AggregateSpec::new("sum", "bytes")
122            .column("Size")
123            .cache_key(&source_key(&[]));
124        assert!(key.ends_with("#sum(Size)"), "got {key}");
125    }
126
127    /// The point of keying off the narrowed source: two aggregates that differ
128    /// ONLY in their source's conditions must not share a cache table, or one
129    /// would serve the other's number.
130    #[test]
131    fn conditions_on_the_source_separate_two_aggregates() {
132        let spec = AggregateSpec::new("count", "n");
133        let opened = spec.cache_key(&source_key(&[("Event".into(), text("opened"))]));
134        let failed = spec.cache_key(&source_key(&[("Event".into(), text("failed"))]));
135        assert_ne!(opened, failed);
136    }
137
138    /// And the converse: the same question asked twice lands on ONE engine,
139    /// however the source's terms were ordered — `index_key` sorts them.
140    #[test]
141    fn source_term_order_does_not_change_the_key() {
142        let spec = AggregateSpec::new("count", "n");
143        let a = spec.cache_key(&source_key(&[
144            ("Event".into(), text("opened")),
145            ("Ip".into(), text("1.2.3.4")),
146        ]));
147        let b = spec.cache_key(&source_key(&[
148            ("Ip".into(), text("1.2.3.4")),
149            ("Event".into(), text("opened")),
150        ]));
151        assert_eq!(a, b);
152    }
153
154    /// Grouping order IS significant — it decides the output column order.
155    #[test]
156    fn group_order_does_change_the_key() {
157        let key = source_key(&[]);
158        let a = AggregateSpec::new("count", "n")
159            .group_by("a")
160            .group_by("b")
161            .cache_key(&key);
162        let b = AggregateSpec::new("count", "n")
163            .group_by("b")
164            .group_by("a")
165            .cache_key(&key);
166        assert_ne!(a, b);
167    }
168
169    /// The alias names the output column, not the question — two aliases for
170    /// the same reduction are the same computation and should share it.
171    #[test]
172    fn the_alias_does_not_affect_the_key() {
173        let key = source_key(&[]);
174        assert_eq!(
175            AggregateSpec::new("count", "total").cache_key(&key),
176            AggregateSpec::new("count", "observed").cache_key(&key),
177        );
178    }
179
180    #[test]
181    fn a_derived_key_cannot_collide_with_its_source() {
182        let key = source_key(&[]);
183        let derived = AggregateSpec::new("count", "total").cache_key(&key);
184        assert!(derived.starts_with(&key));
185        assert_ne!(derived, key);
186    }
187}