harmos_storage/query/
accumulate.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::{cmp::Ordering, num::NonZeroU128};
18
19// Third Party Imports
20
21// Local Imports
22use super::order_stats::Values;
23use super::scalar::{self, Agg, ExactNumber, ScalarKind, ScalarValue};
24use crate::aggregate;
25use crate::signal::Summary;
26use crate::{Aggregate, Error, Result};
27
28/* === Constants === */
29
30/* === Macros === */
31
32/* === Type Aliases === */
33
34/* === Structs === */
35
36#[derive(Clone, Debug)]
37pub struct Accumulator {
38    agg: Agg,
39    rows: u64,
40    count: u64,
41    kind: Option<ScalarKind>,
42    selected: Option<(u64, ScalarValue)>,
43    total: ExactNumber,
44    running: Summary,
45    scratch: Values,
46}
47
48/* === Enums === */
49
50/* === Traits === */
51
52/* === Implementations === */
53
54impl Accumulator {
55    pub fn new(agg: Agg) -> Self {
56        Self {
57            agg,
58            rows: 0,
59            count: 0,
60            kind: None,
61            selected: None,
62            total: ExactNumber::new(false, 0, NonZeroU128::MIN),
63            running: Summary::new(),
64            scratch: Values::default(),
65        }
66    }
67
68    pub fn push(&mut self, capture: u64, value: &ScalarValue) -> Result<()> {
69        value.check()?;
70        self.rows = self.rows.checked_add(1).ok_or(invalid("row count overflow"))?;
71        if self.agg == Agg::Last && !matches!(value, ScalarValue::Missing) {
72            let later = self.selected.as_ref().is_none_or(|(previous, _)| capture > *previous);
73            if later {
74                self.selected = Some((capture, value.clone()));
75            }
76        }
77        if matches!(value, ScalarValue::Missing | ScalarValue::Null) {
78            return Ok(());
79        }
80        self.count = self.count.checked_add(1).ok_or(invalid("value count overflow"))?;
81        let Agg::Of(statistic) = self.agg else {
82            return Ok(());
83        };
84        if statistic == Aggregate::Count {
85            return Ok(());
86        }
87        let value_kind = value.kind().expect("checked non-missing scalar has a kind");
88        match self.kind {
89            Some(previous) if previous != value_kind && !(exact_kind(previous) && exact_kind(value_kind)) => {
90                return Err(invalid("mixed scalar aggregate types"));
91            }
92            Some(_) => {}
93            None => self.kind = Some(value_kind),
94        }
95        match statistic {
96            Aggregate::Min | Aggregate::Max => {
97                let replace = match &self.selected {
98                    None => true,
99                    Some((_, previous)) => {
100                        let order = value.compare(previous)?;
101                        if statistic == Aggregate::Min { order.is_lt() } else { order.is_gt() }
102                    }
103                };
104                if replace {
105                    self.selected = Some((capture, value.clone()));
106                }
107            }
108            Aggregate::Sum | Aggregate::Mean => {
109                if let ScalarValue::Float(value) = value {
110                    self.running.push(*value);
111                    let answered = aggregate::of_summary(&self.running, statistic);
112                    if !answered.is_some_and(f64::is_finite) {
113                        return Err(invalid("aggregate exceeds finite range"));
114                    }
115                } else {
116                    let exact = scalar::exact(value).ok_or(invalid("aggregate requires numbers"))?;
117                    self.total = self.total.checked_add(exact)?;
118                }
119            }
120            Aggregate::Rms | Aggregate::Std => {
121                self.running.push(approximate(value).ok_or(invalid("aggregate requires numbers"))?);
122            }
123            Aggregate::Median | Aggregate::Percentile(_) => {
124                if !numeric_kind(value_kind) {
125                    return Err(invalid("order statistic requires numbers"));
126                }
127                self.scratch.push(value)?;
128            }
129            Aggregate::Count => unreachable!("handled before type checks"),
130        }
131        Ok(())
132    }
133
134    // Exact Disk Merge Passes Poll Cancellation Without Retaining The History.
135    fn ordered(&mut self, statistic: Aggregate, check: &mut impl FnMut() -> Result<()>) -> Result<ScalarValue> {
136        let rank = match statistic {
137            Aggregate::Percentile(rank) => f64::from(rank),
138            _ => 50.0,
139        };
140        let position = rank / 100.0 * (self.scratch.length - 1) as f64;
141        let (lower, upper) = self.scratch.ranks(
142            position.floor() as usize,
143            position.ceil() as usize,
144            |a, b| Ok(approximate(a).unwrap().total_cmp(&approximate(b).unwrap())),
145            check,
146        )?;
147        let (low, high) = (approximate(&lower).unwrap(), approximate(&upper).unwrap());
148        let fraction = position - position.floor();
149        let value = if fraction == 0.0 || low == high {
150            low
151        } else if low.signum() == high.signum() {
152            low + (high - low) * fraction
153        } else {
154            low * (1.0 - fraction) + high * fraction
155        };
156        Ok(ScalarValue::Float(finite(Some(value))?))
157    }
158
159    pub fn finish(mut self, check: &mut impl FnMut() -> Result<()>) -> Result<ScalarValue> {
160        let statistic = match self.agg {
161            Agg::Len => return Ok(ScalarValue::Unsigned(self.rows)),
162            Agg::Last => {
163                return Ok(self.selected.map_or(ScalarValue::Missing, |(_, value)| value));
164            }
165            Agg::Of(statistic) => statistic,
166        };
167        let floating = self.kind == Some(ScalarKind::Float);
168        Ok(match statistic {
169            Aggregate::Count => ScalarValue::Unsigned(self.count),
170            Aggregate::Min | Aggregate::Max => self.selected.map_or(ScalarValue::Missing, |(_, value)| value),
171            // The empty sum is zero, as `Summary` answers it.
172            Aggregate::Sum if self.count == 0 => ScalarValue::Unsigned(0),
173            Aggregate::Sum | Aggregate::Mean if floating => ScalarValue::Float(finite(aggregate::of_summary(&self.running, statistic))?),
174            Aggregate::Sum => integral(self.total, self.kind),
175            Aggregate::Mean if self.count == 0 => ScalarValue::Missing,
176            Aggregate::Mean => integral(self.total.divide(u128::from(self.count))?, self.kind),
177            Aggregate::Rms | Aggregate::Std if self.count == 0 => ScalarValue::Missing,
178            Aggregate::Rms | Aggregate::Std => ScalarValue::Float(finite(aggregate::of_summary(&self.running, statistic))?),
179            Aggregate::Median | Aggregate::Percentile(_) if self.count == 0 => ScalarValue::Missing,
180            Aggregate::Percentile(_) => self.ordered(statistic, check)?,
181            Aggregate::Median if floating => self.ordered(statistic, check)?,
182            Aggregate::Median => {
183                let (lower, upper) =
184                    self.scratch.ranks((self.scratch.length - 1) / 2, self.scratch.length / 2, |a, b| Ok(a.compare(b)?), check)?;
185                let midpoint = scalar::exact(&lower)
186                    .expect("median values are numeric")
187                    .checked_add(scalar::exact(&upper).expect("median values are numeric"))?
188                    .divide(2)?;
189                integral(midpoint, self.kind)
190            }
191        })
192    }
193}
194
195/* === Public Functions === */
196
197// Fallible In-Place Heapsort Polls Original Cancellation During Comparisons.
198pub fn sort_values<T: Copy>(order: &mut [T], mut compare: impl FnMut(T, T) -> Result<Ordering>) -> Result<()> {
199    fn sift<T: Copy>(order: &mut [T], mut root: usize, end: usize, compare: &mut impl FnMut(T, T) -> Result<Ordering>) -> Result<()> {
200        while root < end / 2 {
201            let mut child = root * 2 + 1;
202            if child + 1 < end && compare(order[child], order[child + 1])?.is_lt() {
203                child += 1;
204            }
205            if !compare(order[root], order[child])?.is_lt() {
206                break;
207            }
208            order.swap(root, child);
209            root = child;
210        }
211        Ok(())
212    }
213    for root in (0..order.len() / 2).rev() {
214        sift(order, root, order.len(), &mut compare)?;
215    }
216    for end in (1..order.len()).rev() {
217        order.swap(0, end);
218        sift(order, 0, end, &mut compare)?;
219    }
220    Ok(())
221}
222
223/* === Private Functions === */
224
225fn numeric_kind(kind: ScalarKind) -> bool {
226    matches!(kind, ScalarKind::Signed | ScalarKind::Unsigned | ScalarKind::Float | ScalarKind::Exact)
227}
228// The nearest float to a numeric scalar, for statistics that are not exact.
229fn approximate(value: &ScalarValue) -> Option<f64> {
230    if let ScalarValue::Float(value) = value {
231        return Some(*value);
232    }
233    let exact = scalar::exact(value)?;
234    let magnitude = exact.numerator as f64 / exact.denominator.get() as f64;
235    Some(if exact.negative { -magnitude } else { magnitude })
236}
237fn finite(answer: Option<f64>) -> Result<f64> {
238    answer.filter(|value| value.is_finite()).ok_or(invalid("aggregate exceeds finite range"))
239}
240fn exact_kind(kind: ScalarKind) -> bool {
241    matches!(kind, ScalarKind::Signed | ScalarKind::Unsigned | ScalarKind::Exact)
242}
243fn integral(value: ExactNumber, kind: Option<ScalarKind>) -> ScalarValue {
244    if value.denominator.get() == 1 {
245        if kind == Some(ScalarKind::Signed) {
246            let signed = i128::try_from(value.numerator).ok().and_then(|n| i64::try_from(if value.negative { -n } else { n }).ok());
247            if let Some(signed) = signed {
248                return ScalarValue::Signed(signed);
249            }
250        } else if !value.negative
251            && let Ok(unsigned) = u64::try_from(value.numerator)
252        {
253            return ScalarValue::Unsigned(unsigned);
254        }
255    }
256    ScalarValue::Exact(value)
257}
258pub(super) fn invalid(reason: &'static str) -> Error {
259    Error::InvalidInput { field: "storage reduce", reason }
260}
261
262/* === Tests === */
263
264#[cfg(test)]
265mod tests {
266    use super::*;
267
268    fn fold(agg: Agg, values: &[ScalarValue]) -> ScalarValue {
269        let mut accumulator = Accumulator::new(agg);
270        for (capture, value) in values.iter().enumerate() {
271            accumulator.push(capture as u64, value).unwrap();
272        }
273        accumulator.finish(&mut || Ok(())).unwrap()
274    }
275
276    fn ratio(negative: bool, numerator: u128, denominator: u128) -> ScalarValue {
277        ScalarValue::Exact(ExactNumber::new(negative, numerator, NonZeroU128::new(denominator).unwrap()))
278    }
279
280    #[test]
281    fn public_scalar_fold() {
282        let unsigned = [1u64, 2, 3, 4].map(ScalarValue::Unsigned);
283        assert_eq!(fold(Agg::Of(Aggregate::Sum), &unsigned), ScalarValue::Unsigned(10));
284        assert_eq!(fold(Agg::Of(Aggregate::Mean), &unsigned), ratio(false, 5, 2));
285        assert_eq!(fold(Agg::Of(Aggregate::Median), &unsigned), ratio(false, 5, 2));
286        let signed = [ScalarValue::Signed(-3), ScalarValue::Signed(1)];
287        assert_eq!(fold(Agg::Of(Aggregate::Mean), &signed), ScalarValue::Signed(-1));
288        let text = ["b", "c", "a"].map(|value| ScalarValue::Text(value.into()));
289        assert_eq!(fold(Agg::Of(Aggregate::Min), &text), ScalarValue::Text("a".into()));
290        assert_eq!(fold(Agg::Of(Aggregate::Max), &text), ScalarValue::Text("c".into()));
291        assert_eq!(fold(Agg::Last, &text), ScalarValue::Text("a".into()));
292        let sparse = [ScalarValue::Unsigned(1), ScalarValue::Null, ScalarValue::Missing];
293        assert_eq!(fold(Agg::Len, &sparse), ScalarValue::Unsigned(3));
294        assert_eq!(fold(Agg::Of(Aggregate::Count), &sparse), ScalarValue::Unsigned(1));
295        assert_eq!(fold(Agg::Of(Aggregate::Mean), &[ScalarValue::Null]), ScalarValue::Missing);
296
297        // Floating Statistics Answer Through The Signal Kernels: Population Std.
298        let values = [1.0, 2.0, 4.0, 8.0].map(ScalarValue::Float);
299        let float = |agg| match fold(Agg::Of(agg), &values) {
300            ScalarValue::Float(value) => value,
301            other => panic!("floating statistic answered {other:?}"),
302        };
303        assert!((float(Aggregate::Std) - 7.1875f64.sqrt()).abs() < 1e-12);
304        assert!((float(Aggregate::Rms) - 21.25f64.sqrt()).abs() < 1e-12);
305        assert_eq!(fold(Agg::Of(Aggregate::Median), &values), ScalarValue::Float(3.0));
306        let mut accumulator = Accumulator::new(Agg::Of(Aggregate::Sum));
307        accumulator.push(0, &ScalarValue::Unsigned(1)).unwrap();
308        assert!(accumulator.push(1, &ScalarValue::Text("x".into())).is_err());
309        let mut median = Accumulator::new(Agg::Of(Aggregate::Median));
310        median.push(0, &ScalarValue::Unsigned(1)).unwrap();
311    }
312    #[test]
313    fn spilled_statistics_preserve_exact_values_percentiles_and_clone_branches() {
314        let mut median = Accumulator::new(Agg::Of(Aggregate::Median));
315        median.push(0, &ScalarValue::Unsigned(9_007_199_254_740_993)).unwrap();
316        median.push(1, &ScalarValue::Unsigned(9_007_199_254_740_995)).unwrap();
317        let mut branch = median.clone();
318        median.push(2, &ScalarValue::Unsigned(9_007_199_254_740_997)).unwrap();
319        branch.push(2, &ScalarValue::Unsigned(9_007_199_254_740_991)).unwrap();
320        assert_eq!(median.finish(&mut || Ok(())).unwrap(), ScalarValue::Unsigned(9_007_199_254_740_995));
321        assert_eq!(branch.finish(&mut || Ok(())).unwrap(), ScalarValue::Unsigned(9_007_199_254_740_993));
322        let values = (0..7 * 24 * 60).map(|i| ((i * 19 % 101) as f64 - 50.0).sin()).collect::<Vec<_>>();
323        let series = crate::signal::Series::new(&values);
324        for rank in [0, 13, 50, 77, 100] {
325            let mut accumulator = Accumulator::new(Agg::Of(Aggregate::Percentile(rank)));
326            for (index, value) in values.iter().enumerate() {
327                accumulator.push(index as u64, &ScalarValue::Float(*value)).unwrap();
328            }
329            assert_eq!(accumulator.finish(&mut || Ok(())).unwrap(), ScalarValue::Float(series.percentile(f64::from(rank)).unwrap()));
330        }
331    }
332}
harmos_storage/query/accumulate.rs

harmos_storage/query/
accumulate.rs
