Skip to main content

pond_rs/
rowwise.rs

1//! Row-by-row operations: `diff`, `rate`, `pct_change`, `shift`,
2//! `cumulative` and `fill`. Each reads one row against its neighbours in
3//! storage order, as pond-ts does.
4//!
5//! On a series holding several entities (hosts, devices) the "previous row"
6//! may belong to another entity; pond-ts scopes these per entity with
7//! `partitionBy`, which pond-rs does not have yet.
8
9use std::ops::Range;
10
11use arrow_buffer::i256;
12
13use crate::column::{
14    ArrayColumn, BooleanColumn, Cell, ColumnData, DecimalColumn, Scalar, StringColumn, Value,
15};
16use crate::duration::parse_duration;
17use crate::error::{PondError, Result};
18use crate::reducer::Reducer;
19use crate::series::{Column, TimeSeries};
20
21/// Whether [`TimeSeries::diff`], [`TimeSeries::rate`] and
22/// [`TimeSeries::pct_change`] keep the first row, which has no row before
23/// it (pond-ts `{ drop }`).
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
25pub enum FirstRow {
26    /// Keep it, with a gap in each changed column (pond-ts default).
27    #[default]
28    Keep,
29    /// Remove it from the result (pond-ts `{ drop: true }`).
30    Drop,
31}
32
33/// Options for [`TimeSeries::diff_with`], [`TimeSeries::rate_with`] and
34/// [`TimeSeries::pct_change_with`]. `Change::default()` is pond-ts's
35/// default: the first row kept, each named column replaced by its change.
36///
37/// ```
38/// # use pond_rs::{Change, Column, FirstRow, TimeSeries};
39/// let s = TimeSeries::new("s", vec![0, 1000, 2000],
40///     vec![Column::required("v", vec![1.0, 4.0, 3.0])])?;
41/// let d = s.diff_with(&["v"], Change::default().output("dv").first(FirstRow::Drop))?;
42/// assert_eq!(d.column("v")?.numbers()?.get(0), Some(4.0)); // source kept
43/// assert_eq!(d.column("dv")?.numbers()?.get(0), Some(3.0));
44/// # Ok::<(), pond_rs::PondError>(())
45/// ```
46#[derive(Debug, Clone, Default, PartialEq, Eq)]
47pub struct Change {
48    first: FirstRow,
49    output: Option<String>,
50}
51
52impl Change {
53    /// Keep or drop the first row (pond-ts `{ drop }`).
54    pub fn first(mut self, first: FirstRow) -> Self {
55        self.first = first;
56        self
57    }
58
59    /// Put the result in a new optional column of this name and keep the
60    /// source column as it was (pond-ts `{ output }`). Takes exactly one
61    /// source column; the name must not already exist. The new column has
62    /// the kind the in-place result would: a decimal source stays an exact
63    /// decimal for `diff`, and gives numbers for `rate` and `pct_change`.
64    pub fn output(mut self, name: &str) -> Self {
65        self.output = Some(name.to_string());
66        self
67    }
68}
69
70/// How [`TimeSeries::fill`] fills a run of gaps.
71#[derive(Debug, Clone, PartialEq)]
72pub enum Fill {
73    /// The value before the run (forward fill). A run at the start stays.
74    Hold,
75    /// The value after the run (backward fill). A run at the end stays.
76    Bfill,
77    /// A straight line between the values on either side, by time. Number
78    /// columns only; a run at either end stays.
79    Linear,
80    /// Zero. Number and decimal columns only.
81    Zero,
82    /// This value. Its kind must match the column's.
83    Value(Value),
84}
85
86/// Caps on which runs of gaps [`TimeSeries::fill`] fills. A run that breaks
87/// a cap is left entirely unfilled, never partly (pond-ts `{ limit, maxGap
88/// }`).
89///
90/// ```
91/// # use pond_rs::FillLimits;
92/// let caps = FillLimits::none().limit(3).max_gap("5m");
93/// ```
94#[derive(Debug, Clone, Default, PartialEq, Eq)]
95pub struct FillLimits {
96    limit: Option<usize>,
97    max_gap: Option<String>,
98}
99
100impl FillLimits {
101    /// No caps: every run that can be filled is.
102    pub fn none() -> Self {
103        Self::default()
104    }
105
106    /// Fill only runs of at most `rows` gaps.
107    pub fn limit(mut self, rows: usize) -> Self {
108        self.limit = Some(rows);
109        self
110    }
111
112    /// Fill only runs spanning at most this duration (`"5m"`), measured
113    /// from the value before the run to the value after it. A run at the
114    /// start is measured from its first row to the value after; a run at
115    /// the end from the value before to its last row (as pond-ts does).
116    pub fn max_gap(mut self, duration: &str) -> Self {
117        self.max_gap = Some(duration.to_string());
118        self
119    }
120}
121
122#[derive(Clone, Copy, PartialEq, Eq)]
123enum Delta {
124    Diff,
125    Rate,
126    Pct,
127}
128
129impl Delta {
130    fn call(self) -> &'static str {
131        match self {
132            Delta::Diff => "diff",
133            Delta::Rate => "rate",
134            Delta::Pct => "pct_change",
135        }
136    }
137}
138
139/// A number result, with non-finite values as gaps (`-0.0` kept).
140fn finite_or_gap(v: f64) -> f64 {
141    if v.is_finite() { v } else { f64::NAN }
142}
143
144impl TimeSeries {
145    /// Each row's value minus the row before it, for each named column
146    /// (pond-ts `diff`). The first row, and any row where it or the row
147    /// before is a gap, gets a gap: the row before is the one directly
148    /// above, not the last row with a value. Other columns pass through.
149    ///
150    /// Number columns give numbers. Decimal columns give exact decimals
151    /// (precision 38; an error if a difference has more digits). Other
152    /// kinds are an error. Changed columns may have gaps afterwards.
153    ///
154    /// ```
155    /// # use pond_rs::{Column, TimeSeries};
156    /// let s = TimeSeries::new("s", vec![0, 1000, 2000],
157    ///     vec![Column::required("v", vec![1.0, 4.0, 3.0])])?;
158    /// let d = s.diff(&["v"])?;
159    /// let got: Vec<_> = d.column("v")?.numbers()?.iter().collect();
160    /// assert_eq!(got, [None, Some(3.0), Some(-1.0)]);
161    /// # Ok::<(), pond_rs::PondError>(())
162    /// ```
163    ///
164    /// The first row is kept and each named column is replaced; see
165    /// [`TimeSeries::diff_with`] to drop the first row or keep the source.
166    pub fn diff(&self, columns: &[&str]) -> Result<TimeSeries> {
167        self.change(Delta::Diff, columns, &Change::default())
168    }
169
170    /// [`TimeSeries::diff`] with [`Change`] options: drop the first row, or
171    /// put the result in a new column and keep the source (pond-ts
172    /// `diff(column, { drop, output })`).
173    pub fn diff_with(&self, columns: &[&str], options: Change) -> Result<TimeSeries> {
174        self.change(Delta::Diff, columns, &options)
175    }
176
177    /// Each row's change from the row before, per second of key time:
178    /// `(value - previous) / ((begin - previous begin) / 1000)` (pond-ts
179    /// `rate`). Gaps as for [`TimeSeries::diff`]; two rows with the same
180    /// begin give a gap. Range and interval keys use their begin. Number
181    /// and decimal columns give numbers (decimals via the nearest `f64`).
182    pub fn rate(&self, columns: &[&str]) -> Result<TimeSeries> {
183        self.change(Delta::Rate, columns, &Change::default())
184    }
185
186    /// [`TimeSeries::rate`] with [`Change`] options, as
187    /// [`TimeSeries::diff_with`].
188    pub fn rate_with(&self, columns: &[&str], options: Change) -> Result<TimeSeries> {
189        self.change(Delta::Rate, columns, &options)
190    }
191
192    /// Each row's change as a fraction of the row before:
193    /// `(value - previous) / previous` (pond-ts `pctChange`; not multiplied
194    /// by 100). Gaps as for [`TimeSeries::diff`]; a previous value of 0
195    /// gives a gap. Number and decimal columns give numbers.
196    pub fn pct_change(&self, columns: &[&str]) -> Result<TimeSeries> {
197        self.change(Delta::Pct, columns, &Change::default())
198    }
199
200    /// [`TimeSeries::pct_change`] with [`Change`] options, as
201    /// [`TimeSeries::diff_with`].
202    pub fn pct_change_with(&self, columns: &[&str], options: Change) -> Result<TimeSeries> {
203        self.change(Delta::Pct, columns, &options)
204    }
205
206    fn change(&self, change: Delta, columns: &[&str], options: &Change) -> Result<TimeSeries> {
207        let call = change.call();
208        let targets = self.targets(call, columns)?;
209        if let Some(output) = &options.output {
210            if targets.len() != 1 {
211                return Err(PondError::InvalidData {
212                    call,
213                    message: format!(
214                        "output needs exactly one source column, got {}",
215                        targets.len()
216                    ),
217                });
218            }
219            if output == self.key_name() || self.columns().iter().any(|c| c.name() == output) {
220                return Err(PondError::InvalidData {
221                    call,
222                    message: format!(
223                        "output column '{output}' already exists; leave output unset to replace the source column"
224                    ),
225                });
226            }
227        }
228        let begin = self.key().begin();
229        let n = self.len();
230        // Seconds between each row and the one before, as pond-ts computes
231        // them: one subtraction, one division.
232        let seconds = |t: i64, t0: i64| (t - t0) as f64 / 1000.0;
233        // No per-row gap test: a gap is non-finite, so `cur - prev` is
234        // non-finite when either side is a gap, and so is every result
235        // built on it (`pct_change` divides it by `prev`, `rate` by a
236        // finite `dt`); `finite_or_gap` turns that into a gap. Dividing by
237        // a zero `dt` or `prev` gives an infinity or NaN, which becomes a
238        // gap the same way. Without the
239        // branch the compiler can work on several rows at once (the diff loop
240        // on 1M rows with 5% gaps: about 3 ms with the branch, 0.65 ms
241        // without).
242        let floats = |values: &[f64]| -> Vec<f64> {
243            let mut out = Vec::with_capacity(n);
244            if n == 0 {
245                return out;
246            }
247            out.push(f64::NAN);
248            let pairs = values[1..].iter().zip(&values[..n - 1]);
249            match change {
250                Delta::Diff => out.extend(pairs.map(|(&cur, &prev)| finite_or_gap(cur - prev))),
251                Delta::Rate => {
252                    let steps = begin[1..n].iter().zip(&begin[..n - 1]);
253                    out.extend(pairs.zip(steps).map(|((&cur, &prev), (&t, &t0))| {
254                        finite_or_gap((cur - prev) / seconds(t, t0))
255                    }))
256                }
257                Delta::Pct => {
258                    out.extend(pairs.map(|(&cur, &prev)| finite_or_gap((cur - prev) / prev)))
259                }
260            }
261            out
262        };
263        let mut out = self.columns().to_vec();
264        for &c in &targets {
265            let column = &self.columns()[c];
266            let data: ColumnData = match column.data() {
267                ColumnData::Number(f) => floats(f.values()).into(),
268                ColumnData::Decimal(d) if change == Delta::Diff => {
269                    let mut values = vec![None; n];
270                    for (i, slot) in values.iter_mut().enumerate().skip(1) {
271                        if let (Some(prev), Some(cur)) = (d.get(i - 1), d.get(i)) {
272                            *slot = Some(decimal_fit(
273                                call,
274                                column.name(),
275                                i,
276                                cur.value.checked_sub(prev.value),
277                            )?);
278                        }
279                    }
280                    d.rebuild(values, 38).into()
281                }
282                ColumnData::Decimal(d) => floats(&d.to_f64s()).into(),
283                other => {
284                    return Err(wrong_kind(
285                        call,
286                        column,
287                        other,
288                        "needs a number or decimal column",
289                    ));
290                }
291            };
292            match &options.output {
293                Some(output) => out.push(Column::from_data(output.clone(), data, false)),
294                None => out[c] = Column::from_data(column.name().to_string(), data, false),
295            }
296        }
297        let result = self.with_value_columns(out);
298        Ok(match options.first {
299            FirstRow::Drop if n > 0 => result.row_range(1..n),
300            _ => result,
301        })
302    }
303
304    /// Move each named column's values down `n` rows (a lag; `n` negative
305    /// moves them up, a lead), leaving gaps in the rows moved away from
306    /// (pond-ts `shift`). Keys and other columns stay put. Works on every
307    /// column kind (pond-ts: numbers only); the columns may have gaps
308    /// afterwards.
309    pub fn shift(&self, columns: &[&str], n: i64) -> Result<TimeSeries> {
310        let targets = self.targets("shift", columns)?;
311        let len = self.len() as i64;
312        // Row numbers for the other kinds, built only if one is shifted.
313        let rows = std::cell::OnceCell::new();
314        let rows = || {
315            rows.get_or_init(|| {
316                (0..len)
317                    .map(|i| {
318                        let from = i.checked_sub(n)?;
319                        (0..len).contains(&from).then_some(from as usize)
320                    })
321                    .collect::<Vec<Option<usize>>>()
322            })
323        };
324        let mut out = self.columns().to_vec();
325        for &c in &targets {
326            let column = &self.columns()[c];
327            let data = match column.data() {
328                // Numbers: one block copy instead of a lookup per row.
329                ColumnData::Number(f) => shift_numbers(f.values(), n).into(),
330                other => other.take(rows()),
331            };
332            out[c] = Column::from_data(column.name().to_string(), data, false);
333        }
334        Ok(self.with_value_columns(out))
335    }
336
337    /// Running totals down each named column (pond-ts `cumulative`):
338    /// `Reducer::Sum`, `Max`, `Min` or `Count` (of rows with a value). A gap
339    /// keeps the running value; rows before the first value are gaps.
340    /// `sum` adds left to right from `+0.0`; once it overflows, the rest of
341    /// the column is gaps. Number columns give numbers; decimal columns give
342    /// exact decimals (`count` gives numbers; `sum` at precision 38, an
343    /// error past it). Other kinds, or other reducers, are an error.
344    ///
345    /// ```
346    /// # use pond_rs::{Column, Reducer, TimeSeries};
347    /// let s = TimeSeries::new("s", vec![0, 1000, 2000],
348    ///     vec![Column::optional("v", vec![Some(1.0), None, Some(2.0)])])?;
349    /// let c = s.cumulative(&[("v", Reducer::Sum)])?;
350    /// let got: Vec<_> = c.column("v")?.numbers()?.iter().collect();
351    /// assert_eq!(got, [Some(1.0), Some(1.0), Some(3.0)]);
352    /// # Ok::<(), pond_rs::PondError>(())
353    /// ```
354    pub fn cumulative(&self, spec: &[(&str, Reducer)]) -> Result<TimeSeries> {
355        const CALL: &str = "cumulative";
356        let names: Vec<&str> = spec.iter().map(|&(name, _)| name).collect();
357        let targets = self.targets(CALL, &names)?;
358        let mut out = self.columns().to_vec();
359        for (i, (&c, &(name, reducer))) in targets.iter().zip(spec).enumerate() {
360            named_once(CALL, names[..i].iter().copied(), name)?;
361            if !matches!(
362                reducer,
363                Reducer::Sum | Reducer::Max | Reducer::Min | Reducer::Count
364            ) {
365                return Err(PondError::InvalidData {
366                    call: CALL,
367                    message: format!(
368                        "column '{name}': cumulative takes sum, max, min or count, not {reducer:?}"
369                    ),
370                });
371            }
372            let column = &self.columns()[c];
373            let data: ColumnData = match column.data() {
374                ColumnData::Number(f) => running(f.values(), reducer).into(),
375                ColumnData::Decimal(d) if reducer == Reducer::Count => {
376                    running(&d.to_f64s(), reducer).into()
377                }
378                ColumnData::Decimal(d) => running_decimal(column.name(), d, reducer)?.into(),
379                other => {
380                    return Err(wrong_kind(
381                        CALL,
382                        column,
383                        other,
384                        "needs a number or decimal column",
385                    ));
386                }
387            };
388            out[c] = Column::from_data(column.name().to_string(), data, false);
389        }
390        Ok(self.with_value_columns(out))
391    }
392
393    /// A running fold of your own down one number column (pond-ts
394    /// `cumulative({ col: (acc, v) => … })`): the first value starts the
395    /// running value as it is, then each later value `v` makes it
396    /// `f(acc, v)`. Gaps keep the running value; rows before the first
397    /// value are gaps, and a non-finite running value is written as a gap
398    /// (but still passed to `f` next).
399    ///
400    /// ```
401    /// # use pond_rs::{Column, TimeSeries};
402    /// let s = TimeSeries::new("s", vec![0, 1, 2],
403    ///     vec![Column::required("v", vec![2.0, 3.0, 4.0])])?;
404    /// let product = s.cumulative_with("v", |acc, v| acc * v)?;
405    /// assert_eq!(product.column("v")?.numbers()?.get(2), Some(24.0));
406    /// # Ok::<(), pond_rs::PondError>(())
407    /// ```
408    pub fn cumulative_with(
409        &self,
410        column: &str,
411        mut f: impl FnMut(f64, f64) -> f64,
412    ) -> Result<TimeSeries> {
413        const CALL: &str = "cumulative_with";
414        let c = self.column_index(CALL, column)?;
415        let source = &self.columns()[c];
416        let values = match source.data() {
417            ColumnData::Number(v) => v.values(),
418            other => return Err(wrong_kind(CALL, source, other, "needs a number column")),
419        };
420        let mut acc: Option<f64> = None;
421        let folded: Vec<f64> = values
422            .iter()
423            .map(|&v| {
424                if v.is_finite() {
425                    acc = Some(acc.map_or(v, |a| f(a, v)));
426                }
427                acc.map_or(f64::NAN, finite_or_gap)
428            })
429            .collect();
430        let mut out = self.columns().to_vec();
431        out[c] = Column::from_data(source.name().to_string(), ColumnData::from(folded), false);
432        Ok(self.with_value_columns(out))
433    }
434
435    /// Fill runs of gaps in the named columns (pond-ts `fill({ col:
436    /// strategy })`). Other columns are left as they are, and each column
437    /// keeps its kind and its required flag.
438    ///
439    /// A run that a strategy cannot fill (nothing before it for `Hold`,
440    /// nothing after for `Bfill`, either for `Linear`) or that breaks a cap
441    /// in `limits` stays entirely unfilled. `Linear` interpolates by begin
442    /// time: `before + (after - before) * ((t - t_before) / (t_after -
443    /// t_before))`, or `before` when the two times are equal.
444    ///
445    /// Errors, where pond-ts is silent or errs only on some data: an
446    /// unknown column, `Linear` on a column that is not numbers, `Zero` on
447    /// one that is not numbers or decimals, and a `Value` of the wrong
448    /// kind (or a decimal of another scale).
449    ///
450    /// ```
451    /// # use pond_rs::{Column, Fill, FillLimits, TimeSeries};
452    /// let s = TimeSeries::new("s", vec![0, 1000, 2000],
453    ///     vec![Column::optional("v", vec![Some(1.0), None, Some(3.0)])])?;
454    /// let f = s.fill(&[("v", Fill::Linear)])?;
455    /// assert_eq!(f.column("v")?.numbers()?.get(1), Some(2.0));
456    /// # Ok::<(), pond_rs::PondError>(())
457    /// ```
458    pub fn fill(&self, spec: &[(&str, Fill)]) -> Result<TimeSeries> {
459        self.fill_with(spec, &FillLimits::none())
460    }
461
462    /// [`TimeSeries::fill`] with limits on how far a fill reaches
463    /// ([`FillLimits`]: rows per run, longest gap in time), as pond-ts
464    /// `fill(spec, { limit, maxGap })`.
465    pub fn fill_with(&self, spec: &[(&str, Fill)], limits: &FillLimits) -> Result<TimeSeries> {
466        let mut plan = Vec::with_capacity(spec.len());
467        for (i, (name, fill)) in spec.iter().enumerate() {
468            let c = self.column_index("fill", name)?;
469            named_once("fill", spec[..i].iter().map(|(n, _)| *n), name)?;
470            check_fill(&self.columns()[c], fill)?;
471            plan.push((c, fill));
472        }
473        self.fill_plan(&plan, limits)
474    }
475
476    /// Fill runs of gaps in every value column with one strategy (pond-ts
477    /// `fill('hold')`). As in pond-ts, `Linear` skips columns that are not
478    /// numbers and `Zero` skips columns that are not numbers or decimals.
479    /// A `Value` is an error here; give values per column with
480    /// [`TimeSeries::fill`].
481    pub fn fill_all(&self, fill: Fill) -> Result<TimeSeries> {
482        self.fill_all_with(fill, &FillLimits::none())
483    }
484
485    /// [`TimeSeries::fill_all`] with [`FillLimits`], as
486    /// [`TimeSeries::fill_with`].
487    pub fn fill_all_with(&self, fill: Fill, limits: &FillLimits) -> Result<TimeSeries> {
488        if let Fill::Value(_) = fill {
489            return Err(PondError::InvalidData {
490                call: "fill_all",
491                message: "takes Hold, Bfill, Linear or Zero; give a value per column with fill"
492                    .into(),
493            });
494        }
495        let plan: Vec<(usize, &Fill)> = (0..self.columns().len())
496            .filter(|&c| check_fill(&self.columns()[c], &fill).is_ok())
497            .map(|c| (c, &fill))
498            .collect();
499        self.fill_plan(&plan, limits)
500    }
501
502    fn fill_plan(&self, plan: &[(usize, &Fill)], limits: &FillLimits) -> Result<TimeSeries> {
503        let max_gap = limits.max_gap.as_deref().map(parse_duration).transpose()?;
504        let times = self.key().begin();
505        let mut out = self.columns().to_vec();
506        for &(c, fill) in plan {
507            let column = &self.columns()[c];
508            let data = column.data();
509            let n = data.len();
510            let keep = |run: &Range<usize>| fillable(run, n, fill, limits.limit, max_gap, times);
511            let runs: Vec<Range<usize>> = match data {
512                // Numbers: test the raw values, not the column kind per row.
513                ColumnData::Number(f) => {
514                    let v = f.values();
515                    gap_runs(n, |i| v[i].is_finite()).filter(keep).collect()
516                }
517                _ => gap_runs(n, |i| data.is_present(i)).filter(keep).collect(),
518            };
519            if runs.is_empty() {
520                continue;
521            }
522            let filled: ColumnData = match data {
523                ColumnData::Number(f) => {
524                    let mut v = f.values().to_vec();
525                    for run in &runs {
526                        for j in run.clone() {
527                            v[j] = match fill {
528                                Fill::Hold => v[run.start - 1],
529                                Fill::Bfill => v[run.end],
530                                Fill::Zero => 0.0,
531                                Fill::Value(Value::Number(x)) => *x,
532                                Fill::Linear => {
533                                    let (t0, t1) = (times[run.start - 1], times[run.end]);
534                                    let (before, after) = (v[run.start - 1], v[run.end]);
535                                    let span = (t1 - t0) as f64;
536                                    if span == 0.0 {
537                                        before
538                                    } else {
539                                        before + (after - before) * ((times[j] - t0) as f64 / span)
540                                    }
541                                }
542                                _ => unreachable!("checked by check_fill"),
543                            };
544                        }
545                    }
546                    ColumnData::from(v)
547                }
548                ColumnData::Decimal(d) => {
549                    let mut v: Vec<Option<i128>> = d.iter().map(|x| x.map(|x| x.value)).collect();
550                    for run in &runs {
551                        let value = match fill {
552                            Fill::Hold => v[run.start - 1],
553                            Fill::Bfill => v[run.end],
554                            Fill::Zero => Some(0),
555                            Fill::Value(Value::Decimal(x)) => Some(x.value),
556                            _ => unreachable!("checked by check_fill"),
557                        };
558                        v[run.clone()].fill(value);
559                    }
560                    d.rebuild(v, d.precision()).into()
561                }
562                ColumnData::String(s) => {
563                    let mut v: Vec<Option<&str>> = s.iter().collect();
564                    for run in &runs {
565                        let value = match fill {
566                            Fill::Hold => v[run.start - 1],
567                            Fill::Bfill => v[run.end],
568                            Fill::Value(Value::String(x)) => Some(x.as_str()),
569                            _ => unreachable!("checked by check_fill"),
570                        };
571                        v[run.clone()].fill(value);
572                    }
573                    StringColumn::from_options(v).into()
574                }
575                ColumnData::Boolean(b) => {
576                    let mut v: Vec<Option<bool>> = b.iter().collect();
577                    for run in &runs {
578                        let value = match fill {
579                            Fill::Hold => v[run.start - 1],
580                            Fill::Bfill => v[run.end],
581                            Fill::Value(Value::Boolean(x)) => Some(*x),
582                            _ => unreachable!("checked by check_fill"),
583                        };
584                        v[run.clone()].fill(value);
585                    }
586                    BooleanColumn::from_options(v).into()
587                }
588                ColumnData::Array(a) => {
589                    let mut v: Vec<Option<Cell>> = (0..n).map(|i| a.cell(i).cloned()).collect();
590                    for run in &runs {
591                        let value = match fill {
592                            Fill::Hold => v[run.start - 1].clone(),
593                            Fill::Bfill => v[run.end].clone(),
594                            Fill::Value(Value::Array(x)) => Some(Cell::from(x.as_slice())),
595                            _ => unreachable!("checked by check_fill"),
596                        };
597                        v[run.clone()].fill(value);
598                    }
599                    ArrayColumn::from_cells(v).into()
600                }
601            };
602            out[c] = Column::from_data(column.name().to_string(), filled, column.is_required());
603        }
604        Ok(self.with_value_columns(out))
605    }
606
607    /// The value columns named, as indexes; an error for none or an unknown
608    /// name.
609    fn targets(&self, call: &'static str, names: &[&str]) -> Result<Vec<usize>> {
610        if names.is_empty() {
611            return Err(PondError::InvalidData {
612                call,
613                message: "needs at least one column".into(),
614            });
615        }
616        names
617            .iter()
618            .map(|name| self.column_index(call, name))
619            .collect()
620    }
621
622    /// This series with its value columns replaced.
623    pub(crate) fn with_value_columns(&self, columns: Vec<Column>) -> TimeSeries {
624        TimeSeries::from_parts(
625            self.name().to_string(),
626            self.key_name().to_string(),
627            self.shared_key(),
628            columns,
629        )
630    }
631
632    /// The rows in `range`, key and columns together. Value columns share
633    /// their buffers with this series; the key is copied.
634    pub(crate) fn row_range(&self, range: Range<usize>) -> TimeSeries {
635        let columns = self
636            .columns()
637            .iter()
638            .map(|c| {
639                Column::from_data(
640                    c.name().to_string(),
641                    c.data().slice(range.clone()),
642                    c.is_required(),
643                )
644            })
645            .collect();
646        TimeSeries::from_parts(
647            self.name().to_string(),
648            self.key_name().to_string(),
649            self.key().rows(range),
650            columns,
651        )
652    }
653}
654
655/// An error if `name` is among the earlier entries of a spec.
656fn named_once<'a>(
657    call: &'static str,
658    mut earlier: impl Iterator<Item = &'a str>,
659    name: &str,
660) -> Result<()> {
661    if earlier.any(|e| e == name) {
662        return Err(PondError::InvalidData {
663            call,
664            message: format!("column '{name}' is named twice"),
665        });
666    }
667    Ok(())
668}
669
670fn wrong_kind(call: &'static str, column: &Column, data: &ColumnData, reason: &str) -> PondError {
671    PondError::WrongKind {
672        call,
673        column: column.name().to_string(),
674        kind: data.kind().as_str(),
675        reason: reason.to_string(),
676    }
677}
678
679/// A decimal result as a 38-digit raw value, or an error naming the row.
680fn decimal_fit(call: &'static str, column: &str, row: usize, value: Option<i128>) -> Result<i128> {
681    const MAX: u128 = 10u128.pow(38) - 1;
682    value
683        .filter(|v| v.unsigned_abs() <= MAX)
684        .ok_or_else(|| PondError::InvalidData {
685            call,
686            message: format!("decimal result in '{column}' at row {row} has more than 38 digits"),
687        })
688}
689
690/// pond-ts's running sum / max / min / count over `f64` values.
691fn running(values: &[f64], reducer: Reducer) -> Vec<f64> {
692    let mut acc = 0.0f64;
693    let mut seen = false;
694    values
695        .iter()
696        .map(|&v| {
697            if v.is_finite() {
698                acc = match reducer {
699                    Reducer::Sum => acc + v,
700                    Reducer::Count => acc + 1.0,
701                    // Strict comparisons: on a tie the earlier value stays.
702                    Reducer::Max => {
703                        if !seen || v > acc {
704                            v
705                        } else {
706                            acc
707                        }
708                    }
709                    Reducer::Min => {
710                        if !seen || v < acc {
711                            v
712                        } else {
713                            acc
714                        }
715                    }
716                    _ => unreachable!("checked by cumulative"),
717                };
718                seen = true;
719            }
720            if seen { finite_or_gap(acc) } else { f64::NAN }
721        })
722        .collect()
723}
724
725/// Exact running sum / max / min of a decimal column.
726fn running_decimal(name: &str, d: &DecimalColumn, reducer: Reducer) -> Result<DecimalColumn> {
727    let mut sum = i256::ZERO;
728    let mut best: Option<i128> = None;
729    let mut values = Vec::with_capacity(d.len());
730    for (row, v) in d.iter().enumerate() {
731        if let Some(v) = v {
732            match reducer {
733                Reducer::Sum => {
734                    sum = sum.wrapping_add(i256::from_i128(v.value));
735                    best = Some(decimal_fit("cumulative", name, row, sum.to_i128())?);
736                }
737                Reducer::Max => best = Some(best.map_or(v.value, |b| b.max(v.value))),
738                Reducer::Min => best = Some(best.map_or(v.value, |b| b.min(v.value))),
739                _ => unreachable!("checked by cumulative"),
740            }
741        }
742        values.push(best);
743    }
744    let precision = if reducer == Reducer::Sum {
745        38
746    } else {
747        d.precision()
748    };
749    Ok(d.rebuild(values, precision))
750}
751
752/// `values` moved down `n` rows (up when negative), gaps in the rows moved
753/// away from: the same as `take` with the shifted row numbers.
754fn shift_numbers(values: &[f64], n: i64) -> Vec<f64> {
755    let len = values.len();
756    let k = usize::try_from(n.unsigned_abs()).map_or(len, |k| k.min(len));
757    let mut out = vec![f64::NAN; len];
758    if n >= 0 {
759        out[k..].copy_from_slice(&values[..len - k]);
760    } else {
761        out[..len - k].copy_from_slice(&values[k..]);
762    }
763    out
764}
765
766/// The maximal runs of rows where `present` is false.
767fn gap_runs(n: usize, present: impl Fn(usize) -> bool) -> impl Iterator<Item = Range<usize>> {
768    let mut i = 0;
769    std::iter::from_fn(move || {
770        while i < n && present(i) {
771            i += 1;
772        }
773        if i == n {
774            return None;
775        }
776        let start = i;
777        while i < n && !present(i) {
778            i += 1;
779        }
780        Some(start..i)
781    })
782}
783
784/// pond-ts's checks for one run, in its order: the strategy's neighbours,
785/// then `limit`, then `max_gap`.
786fn fillable(
787    run: &Range<usize>,
788    n: usize,
789    fill: &Fill,
790    limit: Option<usize>,
791    max_gap: Option<i64>,
792    times: &[i64],
793) -> bool {
794    let (has_prev, has_next) = (run.start > 0, run.end < n);
795    let neighbours = match fill {
796        Fill::Linear => has_prev && has_next,
797        Fill::Hold => has_prev,
798        Fill::Bfill => has_next,
799        Fill::Zero | Fill::Value(_) => true,
800    };
801    if !neighbours || limit.is_some_and(|l| run.len() > l) {
802        return false;
803    }
804    let Some(max_gap) = max_gap else {
805        return true;
806    };
807    let span = match (has_prev, has_next) {
808        (true, true) => times[run.end] - times[run.start - 1],
809        (true, false) => times[run.end - 1] - times[run.start - 1],
810        (false, true) => times[run.end] - times[run.start],
811        (false, false) => 0,
812    };
813    span <= max_gap
814}
815
816/// Whether `fill` can apply to this column's kind.
817fn check_fill(column: &Column, fill: &Fill) -> Result<()> {
818    let data = column.data();
819    let ok = match (fill, data) {
820        (Fill::Hold | Fill::Bfill, _) => true,
821        (Fill::Linear, ColumnData::Number(_)) => true,
822        (Fill::Zero, ColumnData::Number(_) | ColumnData::Decimal(_)) => true,
823        (Fill::Value(Value::Number(_)), ColumnData::Number(_)) => true,
824        (Fill::Value(Value::Decimal(x)), ColumnData::Decimal(d)) => {
825            if x.scale != d.scale() {
826                return Err(PondError::InvalidData {
827                    call: "fill",
828                    message: format!(
829                        "column '{}' has scale {}, but the fill value {x} has scale {}",
830                        column.name(),
831                        d.scale(),
832                        x.scale
833                    ),
834                });
835            }
836            let max = 10u128.pow(u32::from(d.precision())) - 1;
837            if x.value.unsigned_abs() > max {
838                return Err(PondError::InvalidData {
839                    call: "fill",
840                    message: format!(
841                        "fill value {x} for column '{}' has more than its {} digits",
842                        column.name(),
843                        d.precision()
844                    ),
845                });
846            }
847            true
848        }
849        (Fill::Value(Value::String(_)), ColumnData::String(_)) => true,
850        (Fill::Value(Value::Boolean(_)), ColumnData::Boolean(_)) => true,
851        (Fill::Value(Value::Array(x)), ColumnData::Array(_)) => {
852            if let Some(e) = x
853                .iter()
854                .position(|e| matches!(e, Scalar::Number(v) if !v.is_finite()))
855            {
856                return Err(PondError::InvalidData {
857                    call: "fill",
858                    message: format!(
859                        "fill value for column '{}': array element {e} must be a finite number, string, or boolean",
860                        column.name()
861                    ),
862                });
863            }
864            true
865        }
866        _ => false,
867    };
868    if ok {
869        Ok(())
870    } else {
871        Err(wrong_kind(
872            "fill",
873            column,
874            data,
875            &format!(
876                "{fill:?} does not apply to a {} column",
877                data.kind().as_str()
878            ),
879        ))
880    }
881}
882
883#[cfg(test)]
884mod tests {
885    use super::*;
886    use crate::{Decimal, IntervalKeys, Labels, Sequence};
887
888    fn decimals(values: Vec<Option<i128>>) -> TimeSeries {
889        let times = (0..values.len() as i64).map(|i| i * 1000).collect();
890        TimeSeries::new(
891            "s",
892            times,
893            vec![
894                Column::optional("d", DecimalColumn::from_options(values, 2).unwrap()),
895                Column::optional("host", [Some("a"), None, Some("c"), None]),
896                Column::optional("up", [None, Some(true), None, Some(false)]),
897            ],
898        )
899        .unwrap()
900    }
901
902    fn text(s: &TimeSeries, name: &str) -> Vec<Option<String>> {
903        let c = s.column(name).unwrap();
904        match c.data() {
905            ColumnData::Decimal(d) => d.iter().map(|v| v.map(|v| v.to_string())).collect(),
906            ColumnData::String(t) => t.iter().map(|v| v.map(str::to_string)).collect(),
907            ColumnData::Boolean(b) => b.iter().map(|v| v.map(|v| v.to_string())).collect(),
908            ColumnData::Number(f) => f.iter().map(|v| v.map(|v| v.to_string())).collect(),
909            ColumnData::Array(a) => a.iter().map(|v| v.map(|v| format!("{v:?}"))).collect(),
910        }
911    }
912
913    fn strs(v: &[Option<&str>]) -> Vec<Option<String>> {
914        v.iter().map(|s| s.map(str::to_string)).collect()
915    }
916
917    /// Raw values that hit every edge of the branch-free number loops:
918    /// gaps stored as NaN and as infinities, zeros of both signs, values
919    /// whose difference overflows, and repeated times (a zero `dt`).
920    fn edge_numbers(n: usize, seed: u64) -> TimeSeries {
921        let mut x = seed;
922        let mut next = move || {
923            x ^= x << 13;
924            x ^= x >> 7;
925            x ^= x << 17;
926            x
927        };
928        let picks = [
929            f64::NAN,
930            f64::INFINITY,
931            f64::NEG_INFINITY,
932            0.0,
933            -0.0,
934            f64::MAX,
935            -f64::MAX,
936            1e-310,
937            1.5,
938            -2.25,
939        ];
940        let mut t = 0i64;
941        let mut times = Vec::with_capacity(n);
942        let mut values = Vec::with_capacity(n);
943        for _ in 0..n {
944            let r = next();
945            t += [0, 1, 1000, 1500][(r % 4) as usize];
946            times.push(t);
947            values.push(if r % 3 == 0 {
948                picks[(r / 3 % picks.len() as u64) as usize]
949            } else {
950                (r % 10_000) as f64 / 7.0 - 700.0
951            });
952        }
953        TimeSeries::new("s", times, vec![Column::optional_raw("v", values)]).unwrap()
954    }
955
956    fn raw_bits(s: &TimeSeries) -> Vec<u64> {
957        let v = s.column("v").unwrap().numbers().unwrap().values();
958        v.iter().map(|x| x.to_bits()).collect()
959    }
960
961    #[test]
962    fn number_changes_match_the_row_by_row_loop_bit_for_bit() {
963        // The loop `change` used before it went branch-free.
964        fn reference(begin: &[i64], values: &[f64], change: Delta) -> Vec<f64> {
965            let n = values.len();
966            let mut out = vec![f64::NAN; n];
967            for i in 1..n {
968                let (prev, cur) = (values[i - 1], values[i]);
969                if !prev.is_finite() || !cur.is_finite() {
970                    continue;
971                }
972                let delta = cur - prev;
973                out[i] = finite_or_gap(match change {
974                    Delta::Diff => delta,
975                    Delta::Rate => {
976                        let dt = (begin[i] - begin[i - 1]) as f64 / 1000.0;
977                        if dt == 0.0 { f64::NAN } else { delta / dt }
978                    }
979                    Delta::Pct => {
980                        if prev == 0.0 {
981                            f64::NAN
982                        } else {
983                            delta / prev
984                        }
985                    }
986                });
987            }
988            out
989        }
990        for (n, seed) in [(0, 1), (1, 2), (2, 3), (500, 4), (5000, 5)] {
991            let s = edge_numbers(n, seed);
992            let values = s.column("v").unwrap().numbers().unwrap().values().to_vec();
993            for (change, got) in [
994                (Delta::Diff, s.diff(&["v"])),
995                (Delta::Rate, s.rate(&["v"])),
996                (Delta::Pct, s.pct_change(&["v"])),
997            ] {
998                let want: Vec<u64> = reference(s.key().begin(), &values, change)
999                    .iter()
1000                    .map(|x| x.to_bits())
1001                    .collect();
1002                assert_eq!(raw_bits(&got.unwrap()), want, "{}, {n} rows", change.call());
1003            }
1004        }
1005    }
1006
1007    #[test]
1008    fn number_shift_matches_a_row_lookup() {
1009        for (n, seed) in [(0, 6), (1, 7), (7, 8), (300, 9)] {
1010            let s = edge_numbers(n, seed);
1011            let values = s.column("v").unwrap().numbers().unwrap().values().to_vec();
1012            let len = n as i64;
1013            for by in [
1014                0,
1015                1,
1016                -1,
1017                3,
1018                -3,
1019                len - 1,
1020                1 - len,
1021                len,
1022                -len,
1023                len + 5,
1024                -len - 5,
1025                i64::MAX,
1026                i64::MIN,
1027            ] {
1028                let want: Vec<u64> = (0..len)
1029                    .map(|i| match i.checked_sub(by) {
1030                        Some(from) if (0..len).contains(&from) => values[from as usize],
1031                        _ => f64::NAN,
1032                    })
1033                    .map(f64::to_bits)
1034                    .collect();
1035                assert_eq!(
1036                    raw_bits(&s.shift(&["v"], by).unwrap()),
1037                    want,
1038                    "{n} rows by {by}"
1039                );
1040            }
1041        }
1042    }
1043
1044    #[test]
1045    fn decimal_diff_and_cumulative_are_exact() {
1046        // 0.10, 0.30, gap, 0.20
1047        let s = decimals(vec![Some(10), Some(30), None, Some(20)]);
1048        let d = s.diff(&["d"]).unwrap();
1049        assert_eq!(text(&d, "d"), strs(&[None, Some("0.20"), None, None]));
1050        let c = s.cumulative(&[("d", Reducer::Sum)]).unwrap();
1051        assert_eq!(
1052            text(&c, "d"),
1053            strs(&[Some("0.10"), Some("0.40"), Some("0.40"), Some("0.60")])
1054        );
1055        let c = s.cumulative(&[("d", Reducer::Max)]).unwrap();
1056        assert_eq!(
1057            text(&c, "d"),
1058            strs(&[Some("0.10"), Some("0.30"), Some("0.30"), Some("0.30")])
1059        );
1060        let c = s.cumulative(&[("d", Reducer::Count)]).unwrap();
1061        assert_eq!(
1062            text(&c, "d"),
1063            strs(&[Some("1"), Some("2"), Some("2"), Some("3")])
1064        );
1065        // rate and pct_change give numbers, via f64, so the same bits as a
1066        // number column of 0.1 and 0.3 (not exactly 2).
1067        let r = s.pct_change(&["d"]).unwrap();
1068        let got = r.column("d").unwrap().numbers().unwrap().get(1).unwrap();
1069        assert_eq!(got.to_bits(), ((0.3f64 - 0.1) / 0.1).to_bits());
1070    }
1071
1072    #[test]
1073    fn decimal_results_past_38_digits_are_errors() {
1074        let max = 10i128.pow(38) - 1;
1075        let s = decimals(vec![Some(max), Some(-max), Some(1), Some(1)]);
1076        let e = s.diff(&["d"]).unwrap_err().to_string();
1077        assert!(e.contains("'d' at row 1 has more than 38 digits"), "{e}");
1078        let s = decimals(vec![Some(max), Some(1), None, None]);
1079        let e = s
1080            .cumulative(&[("d", Reducer::Sum)])
1081            .unwrap_err()
1082            .to_string();
1083        assert!(e.contains("at row 1"), "{e}");
1084    }
1085
1086    #[test]
1087    fn shift_and_fill_work_on_every_kind() {
1088        let s = decimals(vec![Some(10), None, Some(20), None]);
1089        let sh = s.shift(&["d", "host", "up"], 1).unwrap();
1090        assert_eq!(
1091            text(&sh, "d"),
1092            strs(&[None, Some("0.10"), None, Some("0.20")])
1093        );
1094        assert_eq!(text(&sh, "host"), strs(&[None, Some("a"), None, Some("c")]));
1095        assert_eq!(text(&sh, "up"), strs(&[None, None, Some("true"), None]));
1096        let lead = s.shift(&["host"], -3).unwrap();
1097        assert_eq!(text(&lead, "host"), strs(&[None, None, None, None]));
1098
1099        let f = s
1100            .fill(&[
1101                (
1102                    "d",
1103                    Fill::Value(Value::Decimal(Decimal {
1104                        value: -5,
1105                        scale: 2,
1106                    })),
1107                ),
1108                ("host", Fill::Hold),
1109                ("up", Fill::Bfill),
1110            ])
1111            .unwrap();
1112        assert_eq!(
1113            text(&f, "d"),
1114            strs(&[Some("0.10"), Some("-0.05"), Some("0.20"), Some("-0.05")])
1115        );
1116        assert_eq!(
1117            text(&f, "host"),
1118            strs(&[Some("a"), Some("a"), Some("c"), Some("c")])
1119        );
1120        assert_eq!(
1121            text(&f, "up"),
1122            strs(&[Some("true"), Some("true"), Some("false"), Some("false")])
1123        );
1124
1125        // fill_all: Zero fills decimals and skips text and true/false.
1126        let z = s.fill_all(Fill::Zero).unwrap();
1127        assert_eq!(
1128            text(&z, "d"),
1129            strs(&[Some("0.10"), Some("0.00"), Some("0.20"), Some("0.00")])
1130        );
1131        assert_eq!(text(&z, "host"), text(&s, "host"));
1132    }
1133
1134    #[test]
1135    fn fill_refuses_what_cannot_apply() {
1136        let s = decimals(vec![Some(10), None, Some(20), None]);
1137        let none = FillLimits::none();
1138        for (name, fill, want) in [
1139            ("nope", Fill::Hold, "no column named 'nope'"),
1140            (
1141                "d",
1142                Fill::Linear,
1143                "Linear does not apply to a decimal column",
1144            ),
1145            ("host", Fill::Zero, "Zero does not apply to a string column"),
1146            (
1147                "up",
1148                Fill::Value(Value::Number(1.0)),
1149                "does not apply to a boolean",
1150            ),
1151            (
1152                "d",
1153                Fill::Value(Value::Decimal(Decimal { value: 1, scale: 3 })),
1154                "scale 2",
1155            ),
1156        ] {
1157            let e = s.fill_with(&[(name, fill)], &none).unwrap_err().to_string();
1158            assert!(e.contains(want), "{e}");
1159        }
1160        let twice = [("host", Fill::Hold), ("host", Fill::Bfill)];
1161        let e = s.fill_with(&twice, &none).unwrap_err().to_string();
1162        assert!(e.contains("'host' is named twice"), "{e}");
1163        let e = s
1164            .fill_all_with(Fill::Value(Value::Boolean(true)), &none)
1165            .unwrap_err();
1166        assert!(e.to_string().contains("per column"), "{e}");
1167        let e = s
1168            .fill_all_with(Fill::Hold, &FillLimits::none().max_gap("2 s"))
1169            .unwrap_err();
1170        assert!(matches!(e, PondError::InvalidDuration { .. }), "{e}");
1171    }
1172
1173    #[test]
1174    fn other_errors_name_the_problem() {
1175        let s = decimals(vec![Some(1), Some(2), Some(3), Some(4)]);
1176        let e = s.diff(&[]).unwrap_err().to_string();
1177        assert!(e.contains("diff: needs at least one column"), "{e}");
1178        let e = s.rate(&["host"]).unwrap_err().to_string();
1179        assert!(
1180            e.contains("'host'") && e.contains("number or decimal"),
1181            "{e}"
1182        );
1183        let e = s.shift(&["time"], 1).unwrap_err().to_string();
1184        assert!(e.contains("no column named 'time'"), "{e}");
1185        let e = s
1186            .cumulative(&[("d", Reducer::Avg)])
1187            .unwrap_err()
1188            .to_string();
1189        assert!(e.contains("sum, max, min or count"), "{e}");
1190        let e = s
1191            .cumulative(&[("d", Reducer::Sum), ("d", Reducer::Max)])
1192            .unwrap_err()
1193            .to_string();
1194        assert!(e.contains("'d' is named twice"), "{e}");
1195        let e = s
1196            .cumulative_with("host", |a, b| a + b)
1197            .unwrap_err()
1198            .to_string();
1199        assert!(e.contains("needs a number column"), "{e}");
1200    }
1201
1202    #[test]
1203    fn cumulative_with_folds_from_the_first_value() {
1204        let s = TimeSeries::new(
1205            "s",
1206            vec![0, 1, 2, 3],
1207            vec![Column::optional(
1208                "v",
1209                vec![None, Some(2.0), None, Some(5.0)],
1210            )],
1211        )
1212        .unwrap();
1213        let mut calls = 0;
1214        let r = s
1215            .cumulative_with("v", |a, b| {
1216                calls += 1;
1217                a * 10.0 + b
1218            })
1219            .unwrap();
1220        assert_eq!(calls, 1);
1221        let got: Vec<_> = r.column("v").unwrap().numbers().unwrap().iter().collect();
1222        assert_eq!(got, [None, Some(2.0), Some(2.0), Some(25.0)]);
1223        // A non-finite running value is a gap but still carried.
1224        let r = s.cumulative_with("v", |_, _| f64::INFINITY).unwrap();
1225        assert_eq!(r.column("v").unwrap().numbers().unwrap().get(3), None);
1226    }
1227
1228    #[test]
1229    fn drop_keeps_interval_labels_in_step() {
1230        let keys = IntervalKeys {
1231            label: Labels::from(vec!["mon", "tue", "wed"]),
1232            begin: vec![0, 10, 20],
1233            end: vec![10, 20, 30],
1234        };
1235        let s =
1236            TimeSeries::from_intervals("i", keys, vec![Column::required("v", vec![1.0, 2.0, 4.0])])
1237                .unwrap();
1238        let d = s
1239            .diff_with(&["v"], Change::default().first(FirstRow::Drop))
1240            .unwrap();
1241        let crate::Key::Interval(k) = d.key() else {
1242            panic!()
1243        };
1244        let labels: Vec<_> = k.label.strings().unwrap().iter().collect();
1245        assert_eq!(labels, [Some("tue"), Some("wed")]);
1246        assert_eq!(k.begin, [10, 20]);
1247        let got: Vec<_> = d.column("v").unwrap().numbers().unwrap().iter().collect();
1248        assert_eq!(got, [Some(1.0), Some(2.0)]);
1249        // Number labels too, and an empty series drops nothing.
1250        let empty = s.row_range(0..0);
1251        assert_eq!(
1252            empty
1253                .diff_with(&["v"], Change::default().first(FirstRow::Drop))
1254                .unwrap()
1255                .len(),
1256            0
1257        );
1258        let agg = TimeSeries::new(
1259            "t",
1260            vec![0, 1000],
1261            vec![Column::required("v", vec![1.0, 2.0])],
1262        )
1263        .unwrap()
1264        .aggregate(
1265            &Sequence::every("1s").unwrap(),
1266            &[crate::Agg::of("v", Reducer::Sum)],
1267        )
1268        .unwrap();
1269        let d = agg
1270            .rate_with(&["v"], Change::default().first(FirstRow::Drop))
1271            .unwrap();
1272        let crate::Key::Interval(k) = d.key() else {
1273            panic!()
1274        };
1275        assert_eq!(k.label.numbers().unwrap(), [1000.0]);
1276        assert_eq!(d.column("v").unwrap().numbers().unwrap().get(0), Some(1.0));
1277    }
1278
1279    #[test]
1280    fn a_fill_value_must_fit_the_column_precision() {
1281        // A Decimal128(5, 2) column, as read from a file: 999.99 at most.
1282        let narrow =
1283            DecimalColumn::with_type(vec![Some(1i128), None].into_iter().collect(), 5, 2).unwrap();
1284        let s = TimeSeries::new("s", vec![0, 1], vec![Column::optional("d", narrow)]).unwrap();
1285        let big = Fill::Value(Value::Decimal(Decimal {
1286            value: 123_456_789,
1287            scale: 2,
1288        }));
1289        let e = s.fill(&[("d", big)]).unwrap_err().to_string();
1290        assert!(e.contains("more than its 5 digits"), "{e}");
1291        let ok = Fill::Value(Value::Decimal(Decimal {
1292            value: 99_999,
1293            scale: 2,
1294        }));
1295        let f = s.fill(&[("d", ok)]).unwrap();
1296        let mut bytes = Vec::new();
1297        f.write_ipc(&mut bytes).unwrap();
1298        assert!(TimeSeries::from_ipc_bytes("s", bytes, "time").is_ok());
1299    }
1300
1301    #[test]
1302    fn sliced_rows_share_buffers_and_behave_like_copies() {
1303        use arrow_array::{DictionaryArray, types::Int32Type};
1304        let words: DictionaryArray<Int32Type> = vec![Some("x"), None, Some("y"), Some("x"), None]
1305            .into_iter()
1306            .collect();
1307        let s = TimeSeries::new(
1308            "s",
1309            vec![0, 1000, 2000, 3000, 4000],
1310            vec![
1311                Column::optional("n", vec![Some(1.0), None, Some(3.0), Some(4.0), None]),
1312                Column::optional(
1313                    "d",
1314                    DecimalColumn::from_options([Some(1), Some(2), None, Some(4), Some(5)], 1)
1315                        .unwrap(),
1316                ),
1317                Column::optional("t", StringColumn::dictionary(words).unwrap()),
1318                Column::optional("u", [Some("a"), Some("b"), None, Some("d"), Some("e")]),
1319                Column::optional("b", [Some(true), None, Some(false), Some(true), None]),
1320            ],
1321        )
1322        .unwrap();
1323        let cut = s.row_range(1..4);
1324        // The same as copying rows 1..4 one by one.
1325        let rows: Vec<Option<usize>> = (1..4).map(Some).collect();
1326        for (sliced, full) in cut.columns().iter().zip(s.columns()) {
1327            assert_eq!(sliced.data(), &full.data().take(&rows), "{}", sliced.name());
1328        }
1329        assert!(cut.column("t").unwrap().strings().unwrap().is_dictionary());
1330        // Survives Arrow out and back, and feeds aggregate.
1331        let mut bytes = Vec::new();
1332        cut.write_ipc(&mut bytes).unwrap();
1333        let back = TimeSeries::from_ipc_bytes("s", bytes, "time").unwrap();
1334        assert_eq!(back.columns(), cut.columns());
1335        let aggs = [
1336            crate::Agg::new("n", "n", Reducer::Sum),
1337            crate::Agg::new("d", "d", Reducer::Sum),
1338            crate::Agg::new("dmax", "d", Reducer::Max),
1339            crate::Agg::new("t", "t", Reducer::Last),
1340            crate::Agg::new("b", "b", Reducer::First),
1341        ];
1342        let m = cut
1343            .aggregate(&Sequence::every("1m").unwrap(), &aggs)
1344            .unwrap();
1345        assert_eq!(m.column("n").unwrap().numbers().unwrap().get(0), Some(7.0));
1346        assert_eq!(text(&m, "d"), strs(&[Some("0.6")]));
1347        assert_eq!(text(&m, "dmax"), strs(&[Some("0.4")]));
1348        assert_eq!(text(&m, "t"), strs(&[Some("x")]));
1349        assert_eq!(text(&m, "b"), strs(&[Some("false")]));
1350    }
1351
1352    #[test]
1353    fn number_columns_with_gaps_compare_equal() {
1354        let a = Column::optional("v", vec![Some(1.0), None]);
1355        assert_eq!(a, a.clone());
1356        // A gap stored as an infinity equals one stored as NaN.
1357        assert_eq!(a, Column::optional_raw("v", vec![1.0, f64::INFINITY]));
1358        assert_ne!(
1359            Column::required("v", vec![0.0]),
1360            Column::required("v", vec![-0.0])
1361        );
1362    }
1363}