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

harmos_storage/query/
accumulate.rs
