1use std::borrow::Cow;
8use std::collections::HashMap;
9use std::str::FromStr;
10
11use crate::column::ColumnData;
12use crate::error::PondError;
13use crate::series::Column;
14
15#[derive(Debug, Clone, Copy, PartialEq)]
17pub enum Reducer {
18 Count,
20 Sum,
22 Avg,
25 Min,
27 Max,
29 First,
31 Last,
33 Median,
35 Stdev,
37 Percentile(f64),
40 Difference,
44 Keep,
53 Unique,
60 Top(usize),
64 Samples,
67}
68
69impl Reducer {
70 pub(crate) fn check(self, call: &'static str) -> crate::error::Result<()> {
74 match self {
75 Reducer::Percentile(q) if !(0.0..=100.0).contains(&q) => Err(PondError::InvalidData {
76 call,
77 message: format!("percentile must be in [0, 100], got {q}"),
78 }),
79 Reducer::Top(0) => Err(PondError::InvalidData {
80 call,
81 message: "top needs N of at least 1".into(),
82 }),
83 _ => Ok(()),
84 }
85 }
86 pub fn parse(name: &str) -> Result<Self, PondError> {
90 name.parse()
91 }
92
93 pub(crate) fn gives_list(self) -> bool {
95 matches!(self, Reducer::Unique | Reducer::Top(_) | Reducer::Samples)
96 }
97}
98
99impl FromStr for Reducer {
100 type Err = PondError;
101
102 fn from_str(name: &str) -> Result<Self, PondError> {
103 Ok(match name {
104 "count" => Reducer::Count,
105 "sum" => Reducer::Sum,
106 "avg" | "mean" => Reducer::Avg,
107 "min" => Reducer::Min,
108 "max" => Reducer::Max,
109 "first" => Reducer::First,
110 "last" => Reducer::Last,
111 "median" => Reducer::Median,
112 "stdev" => Reducer::Stdev,
113 "difference" => Reducer::Difference,
114 "keep" => Reducer::Keep,
115 "unique" => Reducer::Unique,
116 "samples" => Reducer::Samples,
117 _ => {
118 if let Some(q) = parse_percentile(name) {
119 Reducer::Percentile(q)
120 } else if let Some(n) = parse_top_n(name) {
121 Reducer::Top(n)
122 } else {
123 return Err(PondError::UnknownReducer {
124 name: name.to_string(),
125 });
126 }
127 }
128 })
129 }
130}
131
132pub(crate) fn clamp_at_zero(x: f64) -> f64 {
136 if x.is_nan() || x > 0.0 { x } else { 0.0 }
137}
138
139fn parse_percentile(name: &str) -> Option<f64> {
143 let rest = name.strip_prefix('p')?;
144 if rest.is_empty() {
145 return None;
146 }
147 let q: f64 = rest.parse().ok()?;
148 (0.0..=100.0).contains(&q).then_some(q)
149}
150
151fn parse_top_n(name: &str) -> Option<usize> {
154 let digits = name.strip_prefix("top")?;
155 if digits.is_empty() || !digits.bytes().all(|b| b.is_ascii_digit()) {
156 return None;
157 }
158 let digits = digits.trim_start_matches('0');
159 if digits.is_empty() {
160 return None;
161 }
162 Some(digits.parse().unwrap_or(usize::MAX))
163}
164
165#[derive(Debug, Clone, PartialEq)]
167pub struct Agg {
168 pub(crate) output: String,
169 pub(crate) source: String,
170 pub(crate) reducer: Reducer,
171 pub(crate) shorthand: bool,
174}
175
176impl Agg {
177 pub fn of(source: impl Into<String>, reducer: Reducer) -> Self {
184 let source = source.into();
185 Self {
186 output: source.clone(),
187 source,
188 reducer,
189 shorthand: true,
190 }
191 }
192
193 pub fn new(output: impl Into<String>, source: impl Into<String>, reducer: Reducer) -> Self {
196 Self {
197 output: output.into(),
198 source: source.into(),
199 reducer,
200 shorthand: false,
201 }
202 }
203
204 pub fn output(&self) -> &str {
206 &self.output
207 }
208
209 pub fn source(&self) -> &str {
211 &self.source
212 }
213
214 pub fn reducer(&self) -> Reducer {
216 self.reducer
217 }
218}
219
220const BLOCKED_MIN: usize = 32;
227
228#[inline]
229fn finite(v: f64) -> bool {
230 v.is_finite()
231}
232
233fn blocked_sum_finite(values: &[f64]) -> f64 {
237 let mut a = [0.0f64; 8];
238 let lim = values.len() / 8 * 8;
239 for chunk in values.as_chunks::<8>().0 {
240 for (lane, &v) in a.iter_mut().zip(chunk) {
241 *lane += if finite(v) { v } else { 0.0 };
242 }
243 }
244 let mut s = a[0] + a[1] + a[2] + a[3] + a[4] + a[5] + a[6] + a[7];
245 for &v in &values[lim..] {
246 if finite(v) {
247 s += v;
248 }
249 }
250 s
251}
252
253pub(crate) fn sum_range(values: &[f64]) -> f64 {
256 if values.len() >= BLOCKED_MIN {
257 return blocked_sum_finite(values);
258 }
259 let mut s = 0.0;
260 for &v in values {
261 if finite(v) {
262 s += v;
263 }
264 }
265 s
266}
267
268fn count_finite(values: &[f64]) -> usize {
269 values.iter().filter(|v| v.is_finite()).count()
270}
271
272pub(crate) fn avg_range(values: &[f64]) -> Option<f64> {
274 if values.is_empty() {
275 return None;
276 }
277 if values.len() >= BLOCKED_MIN {
278 let n = count_finite(values);
279 return (n != 0).then(|| blocked_sum_finite(values) / n as f64);
280 }
281 let mut s = 0.0;
282 let mut n = 0usize;
283 for &v in values {
284 if finite(v) {
285 s += v;
286 n += 1;
287 }
288 }
289 (n != 0).then(|| s / n as f64)
290}
291
292pub(crate) fn min_range(values: &[f64]) -> Option<f64> {
294 let mut lo: Option<f64> = None;
295 for &v in values {
296 if finite(v) && lo.is_none_or(|lo| v < lo) {
297 lo = Some(v);
298 }
299 }
300 lo
301}
302
303pub(crate) fn max_range(values: &[f64]) -> Option<f64> {
305 let mut hi: Option<f64> = None;
306 for &v in values {
307 if finite(v) && hi.is_none_or(|hi| v > hi) {
308 hi = Some(v);
309 }
310 }
311 hi
312}
313
314pub(crate) fn stdev_range(values: &[f64]) -> Option<f64> {
316 let mut n = 0.0f64;
317 let mut mean = 0.0f64;
318 let mut m2 = 0.0f64;
319 for &v in values {
320 if !finite(v) {
321 continue;
322 }
323 n += 1.0;
324 let delta = v - mean;
325 mean += delta / n;
326 m2 += delta * (v - mean);
327 }
328 (n != 0.0).then(|| clamp_at_zero(m2 / n).sqrt())
329}
330
331pub(crate) fn percentile_of_sorted(sorted: &[f64], q: f64) -> f64 {
333 let rank = (q / 100.0) * (sorted.len() - 1) as f64;
334 let lo = rank.floor();
335 let hi = rank.ceil();
336 let (l, h) = (sorted[lo as usize], sorted[hi as usize]);
337 if lo == hi {
338 return l;
339 }
340 l + (h - l) * (rank - lo)
341}
342
343const SELECT_INSERTION_THRESHOLD: usize = 16;
345
346fn quickselect(a: &mut [f64], k: usize, lo: usize, hi: usize) -> f64 {
349 let mut left = lo as isize;
351 let mut right = hi as isize;
352 let k = k as isize;
353 let at = |a: &[f64], i: isize| a[i as usize];
354 while left < right {
355 if ((right - left) as usize) < SELECT_INSERTION_THRESHOLD {
356 for i in (left + 1)..=right {
357 let v = at(a, i);
358 let mut j = i - 1;
359 while j >= left && at(a, j) > v {
360 a[(j + 1) as usize] = at(a, j);
361 j -= 1;
362 }
363 a[(j + 1) as usize] = v;
364 }
365 return at(a, k);
366 }
367 let mid = (left + right) >> 1;
368 if at(a, mid) < at(a, left) {
369 a.swap(mid as usize, left as usize);
370 }
371 if at(a, right) < at(a, left) {
372 a.swap(right as usize, left as usize);
373 }
374 if at(a, right) < at(a, mid) {
375 a.swap(right as usize, mid as usize);
376 }
377 let pivot = at(a, mid);
378 let mut i = left;
379 let mut j = right;
380 while i <= j {
381 while at(a, i) < pivot {
382 i += 1;
383 }
384 while at(a, j) > pivot {
385 j -= 1;
386 }
387 if i <= j {
388 a.swap(i as usize, j as usize);
389 i += 1;
390 j -= 1;
391 }
392 }
393 if k <= j {
394 right = j;
395 } else if k >= i {
396 left = i;
397 } else {
398 return at(a, k);
399 }
400 }
401 at(a, k)
402}
403
404pub(crate) fn percentile_range(values: &[f64], q: f64) -> Option<f64> {
406 let mut dense: Vec<f64> = values.iter().copied().filter(|v| v.is_finite()).collect();
407 let k = dense.len();
408 if k == 0 {
409 return None;
410 }
411 let rank = (q / 100.0) * (k - 1) as f64;
412 let lo = rank.floor();
413 let hi = rank.ceil();
414 if lo == hi {
415 return Some(quickselect(&mut dense, lo as usize, 0, k - 1));
416 }
417 let v_hi = quickselect(&mut dense, hi as usize, 0, k - 1);
418 let v_lo = quickselect(&mut dense, lo as usize, 0, hi as usize);
419 Some(v_lo + (v_hi - v_lo) * (rank - lo))
420}
421
422pub(crate) fn first_range(values: &[f64]) -> Option<f64> {
424 values.iter().copied().find(|v| v.is_finite())
425}
426
427pub(crate) fn last_range(values: &[f64]) -> Option<f64> {
429 values.iter().rev().copied().find(|v| v.is_finite())
430}
431
432pub(crate) fn keep_range(values: &[f64]) -> Option<f64> {
435 let mut present = values.iter().copied().filter(|v| v.is_finite());
436 let first = present.next()?;
437 present.all(|v| v == first).then_some(first)
438}
439
440pub(crate) fn difference_range(values: &[f64]) -> Option<f64> {
443 Some(max_range(values)? - min_range(values)?)
444}
445
446pub(crate) fn reduce_range(reducer: Reducer, values: &[f64]) -> Option<f64> {
449 let out = match reducer {
450 Reducer::Count => Some(count_finite(values) as f64),
451 Reducer::Sum => Some(sum_range(values)),
452 Reducer::Avg => avg_range(values),
453 Reducer::Min => min_range(values),
454 Reducer::Max => max_range(values),
455 Reducer::First => first_range(values),
456 Reducer::Last => last_range(values),
457 Reducer::Median => percentile_range(values, 50.0),
458 Reducer::Stdev => stdev_range(values),
459 Reducer::Percentile(q) => percentile_range(values, q),
460 Reducer::Difference => difference_range(values),
461 Reducer::Keep => keep_range(values),
462 Reducer::Unique | Reducer::Top(_) | Reducer::Samples => {
463 unreachable!("list reducers run in elements.rs")
464 }
465 };
466 out.filter(|v| v.is_finite())
467}
468
469pub(crate) fn check_source_kind(
483 call: &'static str,
484 column: &Column,
485 reducer: Reducer,
486) -> Result<(), PondError> {
487 match (column.data(), reducer) {
488 (ColumnData::Number(_) | ColumnData::Decimal(_), _)
489 | (_, Reducer::Count | Reducer::First | Reducer::Last | Reducer::Keep) => Ok(()),
490 (_, r) if r.gives_list() => Ok(()),
491 (data, _) => Err(PondError::WrongKind {
492 call,
493 column: column.name().to_string(),
494 kind: data.kind().as_str(),
495 reason: format!(
496 "{} needs a number or decimal column; string, boolean and array columns take count, first, last, keep, unique, top<N> or samples",
497 reducer_name(reducer)
498 ),
499 }),
500 }
501}
502
503fn reducer_name(reducer: Reducer) -> String {
505 match reducer {
506 Reducer::Count => "count".into(),
507 Reducer::Sum => "sum".into(),
508 Reducer::Avg => "avg".into(),
509 Reducer::Min => "min".into(),
510 Reducer::Max => "max".into(),
511 Reducer::First => "first".into(),
512 Reducer::Last => "last".into(),
513 Reducer::Median => "median".into(),
514 Reducer::Stdev => "stdev".into(),
515 Reducer::Percentile(q) => format!("p{q}"),
516 Reducer::Difference => "difference".into(),
517 Reducer::Keep => "keep".into(),
518 Reducer::Unique => "unique".into(),
519 Reducer::Top(n) => format!("top{n}"),
520 Reducer::Samples => "samples".into(),
521 }
522}
523
524#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
526pub(crate) enum InputKind {
527 Values,
529 Rows,
533 Ranks,
539 Floats,
542 ExactSum,
544 Ids,
547}
548
549pub(crate) fn input_kind(data: &ColumnData, reducer: Reducer) -> InputKind {
550 match (data, reducer) {
551 (ColumnData::Number(_), _) => InputKind::Values,
552 (_, Reducer::Count | Reducer::First | Reducer::Last) => InputKind::Rows,
553 (
554 ColumnData::Decimal(_),
555 Reducer::Min | Reducer::Max | Reducer::Difference | Reducer::Keep,
556 ) => InputKind::Ranks,
557 (ColumnData::Decimal(_), Reducer::Sum) => InputKind::ExactSum,
558 (ColumnData::Decimal(_), _) => InputKind::Floats,
559 (_, Reducer::Keep) => InputKind::Ids,
560 (_, _) => InputKind::Rows,
562 }
563}
564
565pub(crate) struct Input<'a> {
569 pub(crate) values: Cow<'a, [f64]>,
570 distinct: Option<Vec<i128>>,
571 id_rows: Vec<usize>,
572}
573
574impl Input<'_> {
575 pub(crate) fn raw_decimal(&self, v: f64) -> i128 {
577 match &self.distinct {
578 Some(distinct) => distinct[v as usize],
579 None => v as i128,
580 }
581 }
582}
583
584pub(crate) struct Inputs<'a>(HashMap<(usize, InputKind), Input<'a>>);
587
588impl<'a> Inputs<'a> {
589 pub(crate) fn build(columns: &'a [Column], used: &[(usize, Reducer)]) -> Self {
590 let mut map = HashMap::new();
591 for &(c, reducer) in used {
592 if reducer.gives_list() {
593 continue;
594 }
595 let data = columns[c].data();
596 let kind = input_kind(data, reducer);
597 map.entry((c, kind))
598 .or_insert_with(|| build_input(data, kind));
599 }
600 Self(map)
601 }
602
603 pub(crate) fn get(&self, column: usize, kind: InputKind) -> &Input<'a> {
604 &self.0[&(column, kind)]
605 }
606}
607
608fn build_input(data: &ColumnData, kind: InputKind) -> Input<'_> {
609 let plain = |values| Input {
610 values,
611 distinct: None,
612 id_rows: Vec::new(),
613 };
614 match (data, kind) {
615 (ColumnData::Number(c), _) => plain(Cow::Borrowed(c.values())),
616 (ColumnData::Decimal(c), InputKind::Floats) => plain(Cow::Owned(c.to_f64s())),
617 (ColumnData::Decimal(c), InputKind::Ranks) => {
618 let present = |i: usize| c.nulls().is_none_or(|n| n.is_valid(i));
619 let raw = c.raw();
620 const EXACT: u128 = 1 << 53;
621 if (0..raw.len()).all(|i| !present(i) || raw[i].unsigned_abs() <= EXACT) {
623 return plain(Cow::Owned(
624 (0..raw.len())
625 .map(|i| if present(i) { raw[i] as f64 } else { f64::NAN })
626 .collect(),
627 ));
628 }
629 let mut distinct: Vec<i128> = (0..raw.len())
630 .filter(|&i| present(i))
631 .map(|i| raw[i])
632 .collect();
633 distinct.sort_unstable();
634 distinct.dedup();
635 let values = (0..raw.len())
636 .map(|i| {
637 if present(i) {
638 distinct.binary_search(&raw[i]).expect("listed") as f64
639 } else {
640 f64::NAN
641 }
642 })
643 .collect();
644 Input {
645 values: Cow::Owned(values),
646 distinct: Some(distinct),
647 id_rows: Vec::new(),
648 }
649 }
650 (ColumnData::Decimal(_), InputKind::ExactSum) => plain(Cow::Borrowed(&[])),
651 (other, InputKind::Ids) => {
652 let (values, id_rows) = ids(other);
653 Input {
654 values: Cow::Owned(values),
655 distinct: None,
656 id_rows,
657 }
658 }
659 (other, _) => plain(Cow::Owned(match other.arrow_nulls() {
660 None => (0..other.len()).map(|i| i as f64).collect(),
661 Some(nulls) => nulls
662 .iter()
663 .enumerate()
664 .map(|(i, valid)| if valid { i as f64 } else { f64::NAN })
665 .collect(),
666 })),
667 }
668}
669
670fn ids(data: &ColumnData) -> (Vec<f64>, Vec<usize>) {
676 fn dense<K: Eq + std::hash::Hash + Clone>(
677 cells: impl Iterator<Item = Option<K>>,
678 ) -> (Vec<f64>, Vec<usize>) {
679 let mut seen: HashMap<K, usize> = HashMap::new();
680 let mut id_rows = Vec::new();
681 let mut last: Option<(K, f64)> = None;
684 let values = cells
685 .enumerate()
686 .map(|(row, cell)| {
687 let Some(k) = cell else { return f64::NAN };
688 if let Some((prev, id)) = &last
689 && *prev == k
690 {
691 return *id;
692 }
693 let id = *seen.entry(k.clone()).or_insert_with(|| {
694 id_rows.push(row);
695 id_rows.len() - 1
696 }) as f64;
697 last = Some((k, id));
698 id
699 })
700 .collect();
701 (values, id_rows)
702 }
703 match data {
704 ColumnData::String(c) => dense(c.iter()),
705 ColumnData::Boolean(c) => dense(c.iter()),
706 ColumnData::Array(c) => dense(c.iter().map(|cell| cell.map(CellKey))),
707 ColumnData::Number(_) | ColumnData::Decimal(_) => unreachable!("read as values"),
708 }
709}
710
711#[derive(Clone)]
715struct CellKey<'a>(&'a [crate::column::Scalar]);
716
717impl PartialEq for CellKey<'_> {
718 fn eq(&self, other: &Self) -> bool {
719 self.0.len() == other.0.len() && self.0.iter().zip(other.0).all(|(a, b)| a.same(b))
720 }
721}
722
723impl Eq for CellKey<'_> {}
724
725impl std::hash::Hash for CellKey<'_> {
726 fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
727 use crate::column::Scalar;
728 state.write_usize(self.0.len());
729 for e in self.0 {
730 match e {
731 Scalar::Number(v) => (0u8, v.to_bits()).hash(state),
732 Scalar::String(s) => (1u8, s).hash(state),
733 Scalar::Boolean(b) => (2u8, b).hash(state),
734 }
735 }
736 }
737}
738
739pub(crate) fn output_column(
743 name: String,
744 source: &ColumnData,
745 reducer: Reducer,
746 input: &Input<'_>,
747 out: Vec<f64>,
748) -> Column {
749 let picked = || -> Vec<Option<usize>> {
750 out.iter()
751 .map(|&v| v.is_finite().then_some(v as usize))
752 .collect()
753 };
754 match (source, input_kind(source, reducer)) {
755 (ColumnData::Number(_), _) | (_, InputKind::Floats) => Column::optional_raw(name, out),
756 (_, _) if reducer == Reducer::Count => Column::optional_raw(name, out),
757 (ColumnData::Decimal(c), InputKind::Ranks) => {
758 let values = out
759 .iter()
760 .map(|&v| v.is_finite().then(|| input.raw_decimal(v)));
761 Column::from_data(name, c.rebuild(values, c.precision()), false)
762 }
763 (data, InputKind::Ids) => {
764 let rows: Vec<Option<usize>> = out
765 .iter()
766 .map(|&v| v.is_finite().then(|| input.id_rows[v as usize]))
767 .collect();
768 Column::from_data(name, data.take(&rows), false)
769 }
770 (ColumnData::Decimal(c), _) => Column::from_data(name, c.take(&picked()), false),
771 (ColumnData::String(c), _) => Column::from_data(name, c.take(&picked()), false),
772 (ColumnData::Boolean(c), _) => Column::from_data(name, c.take(&picked()), false),
773 (ColumnData::Array(c), _) => Column::from_data(name, c.take(&picked()), false),
774 }
775}
776
777#[cfg(test)]
778mod tests {
779 #[test]
780 fn clamp_at_zero_is_js_math_max() {
781 assert!(clamp_at_zero(f64::NAN).is_nan());
782 assert_eq!(clamp_at_zero(-0.0).to_bits(), 0.0f64.to_bits());
783 assert_eq!(clamp_at_zero(-1.0), 0.0);
784 assert_eq!(clamp_at_zero(2.0), 2.0);
785 }
786
787 #[test]
788 fn percentile_out_of_range_is_an_error_not_a_panic() {
789 use crate::{Agg, Column, Sequence, TimeSeries};
790 let s =
791 TimeSeries::new("s", vec![0, 1], vec![Column::required("a", vec![1.0, 2.0])]).unwrap();
792 for q in [150.0, -10.0, f64::NAN] {
793 let aggs = [Agg::of("a", Reducer::Percentile(q))];
794 assert!(s.rolling("1s", &aggs).is_err(), "rolling p{q}");
795 assert!(
796 s.aggregate(&Sequence::every("1s").unwrap(), &aggs).is_err(),
797 "aggregate p{q}"
798 );
799 }
800 }
801
802 use super::*;
803
804 #[test]
805 fn parses_names() {
806 assert_eq!(Reducer::parse("mean"), Ok(Reducer::Avg));
807 assert_eq!(Reducer::parse("p95"), Ok(Reducer::Percentile(95.0)));
808 assert_eq!(Reducer::parse("p0"), Ok(Reducer::Percentile(0.0)));
809 assert!(matches!(
810 Reducer::parse("p101"),
811 Err(PondError::UnknownReducer { .. })
812 ));
813 assert!(matches!(
814 Reducer::parse("p"),
815 Err(PondError::UnknownReducer { .. })
816 ));
817 assert!(matches!(
818 Reducer::parse("bogus"),
819 Err(PondError::UnknownReducer { .. })
820 ));
821 assert_eq!(Reducer::parse("top3"), Ok(Reducer::Top(3)));
822 assert_eq!(Reducer::parse("top007"), Ok(Reducer::Top(7)));
823 assert_eq!(
824 Reducer::parse("top99999999999999999999999"),
825 Ok(Reducer::Top(usize::MAX))
826 );
827 for bad in ["top0", "top", "top-1", "top1.5", "top 2"] {
828 assert!(
829 matches!(Reducer::parse(bad), Err(PondError::UnknownReducer { .. })),
830 "{bad}"
831 );
832 }
833 assert_eq!(Reducer::parse("unique"), Ok(Reducer::Unique));
834 assert_eq!(Reducer::parse("samples"), Ok(Reducer::Samples));
835 assert_eq!(Reducer::parse("keep"), Ok(Reducer::Keep));
836 assert_eq!(Reducer::parse("difference"), Ok(Reducer::Difference));
837 }
838
839 fn col<'a>(s: &'a crate::TimeSeries, name: &str) -> &'a crate::ColumnData {
840 s.column(name).unwrap().data()
841 }
842
843 #[test]
844 fn keep_and_difference_on_decimals_are_exact() {
845 use crate::{Agg, Column, Decimal, DecimalColumn, Sequence, TimeSeries};
846 let big = 10i128.pow(30);
848 let s = TimeSeries::new(
849 "s",
850 vec![0, 1000, 2000, 3000, 4000],
851 vec![Column::optional(
852 "d",
853 DecimalColumn::from_options(
854 vec![Some(big + 1), Some(big + 1), None, Some(-big), Some(7)],
855 2,
856 )
857 .unwrap(),
858 )],
859 )
860 .unwrap();
861 let aggs = [
862 Agg::new("keep", "d", Reducer::Keep),
863 Agg::new("diff", "d", Reducer::Difference),
864 ];
865 let m = s.aggregate(&Sequence::every("2s").unwrap(), &aggs).unwrap();
866 let dec = |v| Some(Decimal { value: v, scale: 2 });
867 let get = |s: &TimeSeries, name: &str| -> Vec<Option<Decimal>> {
868 s.column(name).unwrap().decimals().unwrap().iter().collect()
869 };
870 assert_eq!(get(&m, "keep"), [dec(big + 1), dec(-big), dec(7)]);
871 assert_eq!(get(&m, "diff"), [dec(0), dec(0), dec(0)]);
872 let r = s.rolling("4s", &aggs).unwrap();
873 assert_eq!(
874 get(&r, "keep"),
875 [dec(big + 1), dec(big + 1), dec(big + 1), None, None]
876 );
877 assert_eq!(
878 get(&r, "diff"),
879 [dec(0), dec(0), dec(0), dec(2 * big + 1), dec(2 * big + 1)]
880 );
881 let all = s.reduce(&aggs).unwrap();
882 assert_eq!(all.decimal("keep").unwrap(), None);
883 assert_eq!(all.decimal("diff").unwrap(), dec(2 * big + 1));
884 let wide = TimeSeries::new(
886 "s",
887 vec![0, 1],
888 vec![Column::required(
889 "d",
890 DecimalColumn::from_values(vec![10i128.pow(38) - 1, 1 - 10i128.pow(38)], 0)
891 .unwrap(),
892 )],
893 )
894 .unwrap();
895 let e = wide.rolling("1s", &aggs[1..]).unwrap_err().to_string();
896 assert!(e.contains("decimal difference 'diff' at row 1"), "{e}");
897 }
898
899 #[test]
900 fn list_reducers_take_any_kind() {
901 use crate::{Agg, ArrayOptions, Column, ColumnKind, DecimalColumn, Scalar, TimeSeries};
902 let s = TimeSeries::new(
903 "s",
904 vec![0, 1000, 2000],
905 vec![
906 Column::optional(
907 "d",
908 DecimalColumn::from_options(vec![Some(150), None, Some(-25)], 2).unwrap(),
909 ),
910 Column::required(
911 "tags",
912 vec![
913 vec![Scalar::from("b"), "a".into(), "b".into()],
914 vec![],
915 vec![Scalar::from(1.0)],
916 ],
917 ),
918 ],
919 )
920 .unwrap();
921 let r = s
923 .reduce(&[
924 Agg::new("u", "d", Reducer::Unique),
925 Agg::new("t", "tags", Reducer::Top(1)),
926 ])
927 .unwrap();
928 let n = |v: f64| Scalar::Number(v);
929 assert_eq!(r.array("u").unwrap(), Some(&[n(-0.25), n(1.5)][..]));
930 assert_eq!(r.array("t").unwrap(), Some(&[Scalar::from("b")][..]));
931 let a = s
933 .array_aggregate("tags", Reducer::Samples, &ArrayOptions::default())
934 .unwrap();
935 let col = a.column("tags").unwrap().arrays().unwrap();
936 assert_eq!(col.get(1), Some(&[][..]));
937 let wrong = ArrayOptions {
938 kind: Some(ColumnKind::String),
939 ..ArrayOptions::default()
940 };
941 let e = s
942 .array_aggregate("tags", Reducer::Unique, &wrong)
943 .unwrap_err()
944 .to_string();
945 assert!(e.contains("output kind must be array"), "{e}");
946 let e = s
947 .reduce(&[Agg::of("d", Reducer::Top(0))])
948 .unwrap_err()
949 .to_string();
950 assert!(e.contains("at least 1"), "{e}");
951 }
952
953 #[test]
954 fn keep_takes_any_kind_and_difference_numbers_only() {
955 use crate::{Agg, Column, Scalar, Sequence, TimeSeries};
956 let s = TimeSeries::new(
957 "s",
958 vec![0, 1000, 2000, 3000],
959 vec![
960 Column::optional("t", [Some("a"), Some("a"), None, Some("b")]),
961 Column::optional("b", [Some(true), None, Some(true), Some(false)]),
962 Column::optional(
965 "arr",
966 [
967 Some(vec![Scalar::from(0.0), "x".into()]),
968 Some(vec![Scalar::from(0.0), "x".into()]),
969 None,
970 Some(vec![Scalar::from(1.0)]),
971 ],
972 ),
973 ],
974 )
975 .unwrap();
976 let aggs = [
977 Agg::of("t", Reducer::Keep),
978 Agg::of("b", Reducer::Keep),
979 Agg::of("arr", Reducer::Keep),
980 ];
981 let m = s.aggregate(&Sequence::every("2s").unwrap(), &aggs).unwrap();
982 assert_eq!(
983 m.column("t")
984 .unwrap()
985 .strings()
986 .unwrap()
987 .iter()
988 .collect::<Vec<_>>(),
989 [Some("a"), Some("b")]
990 );
991 assert_eq!(
992 m.column("b")
993 .unwrap()
994 .booleans()
995 .unwrap()
996 .iter()
997 .collect::<Vec<_>>(),
998 [Some(true), None]
999 );
1000 let arr = m.column("arr").unwrap().arrays().unwrap();
1001 assert_eq!(arr.get(0), Some(&[Scalar::from(0.0), "x".into()][..]));
1002 assert_eq!(arr.get(1), Some(&[Scalar::from(1.0)][..]));
1003 let zeros = TimeSeries::new(
1006 "z",
1007 vec![0, 1000, 2000, 3000],
1008 vec![Column::required(
1009 "a",
1010 vec![
1011 vec![Scalar::from(0.0)],
1012 vec![Scalar::from("x")],
1013 vec![Scalar::from(-0.0)],
1014 vec![Scalar::from(0.0)],
1015 ],
1016 )],
1017 )
1018 .unwrap();
1019 let one = [Agg::of("a", Reducer::Keep)];
1020 let by_second = zeros
1021 .aggregate(&Sequence::every("1s").unwrap(), &one)
1022 .unwrap();
1023 let cell =
1024 |s: &TimeSeries, i| s.column("a").unwrap().arrays().unwrap().get(i).unwrap()[0].clone();
1025 assert!(cell(&by_second, 2).same(&Scalar::from(-0.0)));
1026 let rolled = zeros.rolling("1s", &one).unwrap();
1027 assert!(cell(&rolled, 2).same(&Scalar::from(-0.0)));
1028 let pair = zeros.rolling("2s", &one).unwrap();
1029 assert_eq!(pair.column("a").unwrap().arrays().unwrap().get(3), None);
1030 let r = s.rolling("2s", &aggs).unwrap();
1031 assert_eq!(
1032 r.column("t")
1033 .unwrap()
1034 .strings()
1035 .unwrap()
1036 .iter()
1037 .collect::<Vec<_>>(),
1038 [Some("a"), Some("a"), Some("a"), Some("b")]
1039 );
1040 assert!(matches!(col(&r, "arr"), crate::ColumnData::Array(_)));
1041 for result in [
1042 s.reduce(&[Agg::of("t", Reducer::Difference)]).map(|_| ()),
1043 s.rolling("1s", &[Agg::of("b", Reducer::Difference)])
1044 .map(|_| ()),
1045 ] {
1046 let e = result.unwrap_err().to_string();
1047 assert!(e.contains("count, first, last, keep"), "{e}");
1048 }
1049 }
1050
1051 #[test]
1052 fn percentile_matches_a_full_sort() {
1053 let mut seed = 7u64;
1055 for n in 1..80usize {
1056 let values: Vec<f64> = (0..n)
1057 .map(|_| {
1058 seed = seed
1059 .wrapping_mul(6364136223846793005)
1060 .wrapping_add(1442695040888963407);
1061 ((seed >> 33) % 50) as f64 - 25.0
1062 })
1063 .collect();
1064 let mut sorted = values.clone();
1065 sorted.sort_by(f64::total_cmp);
1066 for q in [0.0, 10.0, 50.0, 90.0, 95.0, 100.0] {
1067 assert_eq!(
1068 percentile_range(&values, q),
1069 Some(percentile_of_sorted(&sorted, q)),
1070 "n={n} q={q}"
1071 );
1072 }
1073 }
1074 }
1075
1076 #[test]
1077 fn blocked_sum_adds_lanes_in_order() {
1078 let mut values = vec![1.0; 40];
1081 values[0] = 1e16;
1082 let lanes: f64 = {
1083 let mut a = [0.0f64; 8];
1084 for (i, v) in values.iter().enumerate() {
1085 a[i % 8] += v;
1086 }
1087 a.iter().sum()
1088 };
1089 assert_eq!(sum_range(&values), lanes);
1090 }
1091}