1use 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#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
25pub enum FirstRow {
26 #[default]
28 Keep,
29 Drop,
31}
32
33#[derive(Debug, Clone, Default, PartialEq, Eq)]
47pub struct Change {
48 first: FirstRow,
49 output: Option<String>,
50}
51
52impl Change {
53 pub fn first(mut self, first: FirstRow) -> Self {
55 self.first = first;
56 self
57 }
58
59 pub fn output(mut self, name: &str) -> Self {
65 self.output = Some(name.to_string());
66 self
67 }
68}
69
70#[derive(Debug, Clone, PartialEq)]
72pub enum Fill {
73 Hold,
75 Bfill,
77 Linear,
80 Zero,
82 Value(Value),
84}
85
86#[derive(Debug, Clone, Default, PartialEq, Eq)]
95pub struct FillLimits {
96 limit: Option<usize>,
97 max_gap: Option<String>,
98}
99
100impl FillLimits {
101 pub fn none() -> Self {
103 Self::default()
104 }
105
106 pub fn limit(mut self, rows: usize) -> Self {
108 self.limit = Some(rows);
109 self
110 }
111
112 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
139fn finite_or_gap(v: f64) -> f64 {
141 if v.is_finite() { v } else { f64::NAN }
142}
143
144impl TimeSeries {
145 pub fn diff(&self, columns: &[&str]) -> Result<TimeSeries> {
167 self.change(Delta::Diff, columns, &Change::default())
168 }
169
170 pub fn diff_with(&self, columns: &[&str], options: Change) -> Result<TimeSeries> {
174 self.change(Delta::Diff, columns, &options)
175 }
176
177 pub fn rate(&self, columns: &[&str]) -> Result<TimeSeries> {
183 self.change(Delta::Rate, columns, &Change::default())
184 }
185
186 pub fn rate_with(&self, columns: &[&str], options: Change) -> Result<TimeSeries> {
189 self.change(Delta::Rate, columns, &options)
190 }
191
192 pub fn pct_change(&self, columns: &[&str]) -> Result<TimeSeries> {
197 self.change(Delta::Pct, columns, &Change::default())
198 }
199
200 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 let seconds = |t: i64, t0: i64| (t - t0) as f64 / 1000.0;
233 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 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 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 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 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 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 pub fn fill(&self, spec: &[(&str, Fill)]) -> Result<TimeSeries> {
459 self.fill_with(spec, &FillLimits::none())
460 }
461
462 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 pub fn fill_all(&self, fill: Fill) -> Result<TimeSeries> {
482 self.fill_all_with(fill, &FillLimits::none())
483 }
484
485 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 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 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 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 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
655fn 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
679fn 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
690fn 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 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
725fn 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
752fn 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
766fn 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
784fn 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
816fn 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 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 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 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 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 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 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 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 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 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 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 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}