Skip to main content

pond_rs/
reducer.rs

1//! Reducers (`avg`, `sum`, …) and the mapping entries that use them.
2//!
3//! The kernels here are ports of pond-ts's, step for step, because parity
4//! with pond-ts is checked bit for bit: the order values are added in decides
5//! the last bits of a sum. Each kernel names the pond-ts function it follows.
6
7use 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/// A built-in reducer.
16#[derive(Debug, Clone, Copy, PartialEq)]
17pub enum Reducer {
18    /// Number of present (non-missing) values.
19    Count,
20    /// Sum of present values; 0 when there are none.
21    Sum,
22    /// Mean of present values; missing when there are none. `"mean"` parses
23    /// to this too.
24    Avg,
25    /// Smallest present value.
26    Min,
27    /// Largest present value.
28    Max,
29    /// First present value.
30    First,
31    /// Last present value.
32    Last,
33    /// 50th percentile, interpolated.
34    Median,
35    /// Population standard deviation (divides by n).
36    Stdev,
37    /// Percentile `q` in `[0, 100]`, linear interpolation between ranks
38    /// (`"p95"`).
39    Percentile(f64),
40    /// Largest present value minus smallest (the spread); missing when
41    /// there are none. Number and decimal columns only. In `rolling` it
42    /// can be `-0` (see [`crate::TimeSeries::rolling`]).
43    Difference,
44    /// The value when every present value is equal, otherwise missing
45    /// (also missing when there are none). Any column kind; the output
46    /// has the source's kind. Numbers compare as JavaScript's `===` does,
47    /// so `0` equals `-0`: `aggregate` and `reduce` give the first one,
48    /// `rolling` gives `0`. Array cells compare by their elements bit for
49    /// bit (so `[0]` and `[-0]` differ), where pond-ts compares the array
50    /// objects (so two equal arrays in different rows are kept here and
51    /// give a gap in pond-ts).
52    Keep,
53    /// The distinct present values as a list (an array column), sorted:
54    /// numbers, then text, then true/false; text in JavaScript's order (by
55    /// UTF-16 code units). An array column's lists are flattened one level
56    /// first. `0` and `-0` are one value, written `0`. Nothing present
57    /// gives an empty list, not a gap. Any column kind; decimals are read
58    /// as their nearest `f64`.
59    Unique,
60    /// The N most frequent present values (`"top3"`) as a list, most
61    /// frequent first, ties in [`Reducer::Unique`]'s order; flattening,
62    /// zeros and empty lists as there. N is at least 1.
63    Top(usize),
64    /// Every present value as a list in arrival order, duplicates and `-0`
65    /// kept; flattening and empty lists as [`Reducer::Unique`].
66    Samples,
67}
68
69impl Reducer {
70    /// Refuse a reducer no result can come from: a percentile outside
71    /// `[0, 100]` (or NaN). [`Reducer::parse`] never builds one, but the
72    /// variant is public.
73    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    /// Parse a pond-ts reducer name: `count`, `sum`, `avg`, `mean`, `min`,
87    /// `max`, `first`, `last`, `median`, `stdev`, `p<q>`, `difference`,
88    /// `keep`, `unique`, `samples`, `top<N>`.
89    pub fn parse(name: &str) -> Result<Self, PondError> {
90        name.parse()
91    }
92
93    /// `unique`, `top<N>` and `samples`: the result is a list.
94    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
132/// JS `Math.max(0, x)`: a NaN stays NaN (Rust's `f64::max` would drop it and
133/// give 0), and `-0` becomes `+0`. A NaN variance then becomes a gap, as in
134/// pond-ts.
135pub(crate) fn clamp_at_zero(x: f64) -> f64 {
136    if x.is_nan() || x > 0.0 { x } else { 0.0 }
137}
138
139/// pond-ts `parsePercentile`: `p` then a number in `[0, 100]`. pond-ts reads
140/// the number with JS `Number`, which also takes surrounding spaces and hex;
141/// those spellings are rejected here.
142fn 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
151/// pond-ts `parseTopN`: `top` then digits, not all zero (`top007` is 7).
152/// A count too large for `usize` keeps every value, as pond-ts's would.
153fn 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/// One mapping entry: write `reducer(source)` to the output column `output`.
166#[derive(Debug, Clone, PartialEq)]
167pub struct Agg {
168    pub(crate) output: String,
169    pub(crate) source: String,
170    pub(crate) reducer: Reducer,
171    /// Built with [`Agg::of`] (output named after its source). Decides
172    /// column order in `rolling`; see [`crate::TimeSeries::rolling`].
173    pub(crate) shorthand: bool,
174}
175
176impl Agg {
177    /// Reduce `source` into a column of the same name: pond-ts's shorthand
178    /// mapping `{ cpu: 'avg' }`.
179    ///
180    /// In `rolling`, a list made only of `Agg::of` entries is put into schema
181    /// order, as pond-ts does; a single [`Agg::new`] entry in the list keeps
182    /// the whole list in the order given.
183    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    /// Reduce `source` into a column named `output`: pond-ts's
194    /// `{ cpu_avg: { from: 'cpu', using: 'avg' } }`.
195    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    /// The output column's name.
205    pub fn output(&self) -> &str {
206        &self.output
207    }
208
209    /// The column read.
210    pub fn source(&self) -> &str {
211        &self.source
212    }
213
214    /// The reducer applied.
215    pub fn reducer(&self) -> Reducer {
216        self.reducer
217    }
218}
219
220// ---------------------------------------------------------------------------
221// Range kernels: reduce `values[start..end]`, skipping gaps. These are the
222// `reduceColumnRange` kernels pond-ts's columnar `aggregate` uses.
223// ---------------------------------------------------------------------------
224
225/// pond-ts `BLOCKED_MIN`: from this many values, sums use eight lanes.
226const BLOCKED_MIN: usize = 32;
227
228#[inline]
229fn finite(v: f64) -> bool {
230    v.is_finite()
231}
232
233/// pond-ts `blockedSumFinite`: eight running sums, combined in order, then
234/// the tail. A gap adds `+0` to its lane (which turns a `-0` lane into `+0`,
235/// exactly as in pond-ts).
236fn 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
253/// pond-ts `sumRange` (the gap-skipping form; with no gaps it gives the same
254/// bits as the plain form).
255pub(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
272/// pond-ts `avgRange`.
273pub(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
292/// pond-ts `minRange`: the first of equal values wins (`v < lo`).
293pub(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
303/// pond-ts `maxRange`.
304pub(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
314/// pond-ts `stdevRange`: Welford's recurrence over present values.
315pub(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
331/// pond-ts `percentileOfSorted`.
332pub(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
343/// pond-ts `SELECT_INSERTION_THRESHOLD`.
344const SELECT_INSERTION_THRESHOLD: usize = 16;
345
346/// pond-ts `quickselect`, line for line: which of two equal values (`0` and
347/// `-0`) ends up at rank `k` depends on the exact swaps.
348fn quickselect(a: &mut [f64], k: usize, lo: usize, hi: usize) -> f64 {
349    // Indices can step one below `left` (j -= 1 past 0), so work in isize.
350    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
404/// pond-ts `reducePercentileColumnRange` / `percentileOfUnsorted`.
405pub(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
422/// First present value (pond-ts aggregate's boundary scan).
423pub(crate) fn first_range(values: &[f64]) -> Option<f64> {
424    values.iter().copied().find(|v| v.is_finite())
425}
426
427/// Last present value.
428pub(crate) fn last_range(values: &[f64]) -> Option<f64> {
429    values.iter().rev().copied().find(|v| v.is_finite())
430}
431
432/// pond-ts `keep`: the first present value when every present value equals
433/// it (`===`, so `0` and `-0` are equal and the first one's sign is kept).
434pub(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
440/// pond-ts `difference`: the largest value minus the smallest, each the
441/// first of equals (so the result is never `-0`).
442pub(crate) fn difference_range(values: &[f64]) -> Option<f64> {
443    Some(max_range(values)? - min_range(values)?)
444}
445
446/// Reduce one range. A non-finite result (an overflowing sum) is a gap, as
447/// in pond-ts (`finiteOrMissing`).
448pub(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
469// ---------------------------------------------------------------------------
470// Text and true/false sources. They take `count`, `first`, `last` and
471// `keep` here, and `unique` / `top<N>` / `samples` in elements.rs (the set
472// pond-ts's types allow). The kernels above run over each row's number where the row has
473// a value and NaN where it has none (`keep`: an id per distinct value);
474// `first` / `last` / `keep` then give the row whose value to take.
475// ---------------------------------------------------------------------------
476
477/// Refuse a reducer that needs numbers on a text, true/false or array
478/// column. pond-ts's types refuse it too; at run time pond-ts answers anyway
479/// (`sum` gives 0, `avg` and `max` give gaps), where pond-rs returns this
480/// error. On an array column each cell is one value: `count` counts cells,
481/// not elements, and `first` / `last` take whole cells.
482pub(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
503/// pond-ts's name for a reducer.
504fn 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/// What the `f64` kernels run over for one source column and reducer.
525#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
526pub(crate) enum InputKind {
527    /// A number column's own values.
528    Values,
529    /// Each row's index where it has a value, NaN where not: `count`,
530    /// `first` and `last` on decimal, text and true/false columns, whose
531    /// results then pick rows.
532    Rows,
533    /// `min`, `max`, `difference` and `keep` on decimals, kept exact: each
534    /// raw whole number as
535    /// an `f64` when all fit within 2^53 (so the conversion is exact),
536    /// otherwise each value's rank among the column's distinct values.
537    /// Either way they compare as the decimals do.
538    Ranks,
539    /// Each decimal as its nearest `f64`: `avg`, `stdev`, `median`,
540    /// percentiles.
541    Floats,
542    /// `sum` of decimals: no kernel, summed exactly (see `decimal.rs`).
543    ExactSum,
544    /// `keep` on text, true/false and array columns: an id per distinct
545    /// value (equal values share one), NaN for a gap.
546    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        // check_source_kind refused the rest.
561        (_, _) => InputKind::Rows,
562    }
563}
564
565/// One kernel input, and for [`InputKind::Ranks`] the distinct values the
566/// ranks index (`None` when the values are the raw numbers themselves), for
567/// [`InputKind::Ids`] the first row holding each id.
568pub(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    /// The raw decimal a [`InputKind::Ranks`] value stands for.
576    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
584/// The kernel inputs for a set of (source column, reducer) pairs, each
585/// built once (a mapping often reduces one column several ways).
586pub(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            // Only present rows count: a gap's raw value can be anything.
622            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
670/// An id per distinct value of a text, true/false or array column (NaN for
671/// a gap), and the first row holding each id. Array cells are equal when
672/// their elements are, bit for bit (`[0]` and `[-0]` differ), so the row
673/// an id names holds exactly the cell of every row with that id; pond-ts
674/// compares the array objects themselves.
675fn 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        // Runs of one value are common (a host name, a status): compare with
682        // the last value before hashing.
683        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/// An array cell as a map key: equal when the elements are the same bit
712/// for bit ([`Scalar::same`](crate::column::Scalar), so `0` and `-0`
713/// differ).
714#[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
739/// The output column for `out`, the kernel results for a `source` column
740/// read through `input`. Always optional (an empty window or bucket is a
741/// gap).
742pub(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        // Past 2^53, so ranks stand in for the raw values.
847        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        // A spread past 38 digits is an error naming the row.
885        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        // Decimals as their nearest f64.
922        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        // An empty list is a value, not a gap; a gap stays a gap.
932        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                // Equal content in separate cells: kept (pond-ts compares
963                // the array objects, so it would give a gap).
964                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        // Each one-row bucket keeps its own cell: `[-0]` is not the `[0]`
1004        // of an earlier row (#797 review), and the two differ.
1005        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        // Many sizes, both sides of the insertion-sort threshold.
1054        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        // 40 values: 5 rounds of 8 lanes. The lanes see 1e16 then 1s; a
1079        // straight left-to-right sum would lose the 1s differently.
1080        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}