1use crate::column::{ArrayColumn, BooleanColumn, ColumnData, Scalar, StringColumn};
12use crate::error::{PondError, Result};
13use crate::reducer::{Reducer, clamp_at_zero, percentile_of_sorted};
14use crate::series::{Column, ColumnKind, TimeSeries};
15
16#[derive(Debug, Clone, Default, PartialEq, Eq)]
30pub struct ArrayOptions {
31 pub output: Option<String>,
35 pub kind: Option<ColumnKind>,
38}
39
40impl TimeSeries {
41 pub fn array_contains(&self, column: &str, value: impl Into<Scalar>) -> Result<TimeSeries> {
61 let value = value.into();
62 self.keep_arrays("TimeSeries::array_contains", column, |cell| {
63 cell.contains(&value)
64 })
65 }
66
67 pub fn array_contains_all(&self, column: &str, values: &[Scalar]) -> Result<TimeSeries> {
72 self.keep_arrays("TimeSeries::array_contains_all", column, |cell| {
73 values.iter().all(|v| cell.contains(v))
74 })
75 }
76
77 pub fn array_contains_any(&self, column: &str, values: &[Scalar]) -> Result<TimeSeries> {
81 self.keep_arrays("TimeSeries::array_contains_any", column, |cell| {
82 values.iter().any(|v| cell.contains(v))
83 })
84 }
85
86 fn keep_arrays(
87 &self,
88 call: &'static str,
89 column: &str,
90 keep: impl Fn(&[Scalar]) -> bool,
91 ) -> Result<TimeSeries> {
92 let arrays = self.array_column(call, column)?;
93 let rows: Vec<usize> = (0..self.len())
94 .filter(|&i| arrays.get(i).is_some_and(&keep))
95 .collect();
96 Ok(self.take_rows(&rows))
97 }
98
99 pub fn array_aggregate(
149 &self,
150 column: &str,
151 reducer: Reducer,
152 options: &ArrayOptions,
153 ) -> Result<TimeSeries> {
154 const CALL: &str = "TimeSeries::array_aggregate";
155 reducer.check(CALL)?;
156 let arrays = self.array_column(CALL, column)?;
157 let picks = matches!(reducer, Reducer::First | Reducer::Last | Reducer::Keep);
158 if reducer.gives_list() {
159 if let Some(k) = options.kind.filter(|&k| k != ColumnKind::Array) {
160 return Err(PondError::InvalidData {
161 call: CALL,
162 message: format!(
163 "{reducer:?} gives lists, so the output kind must be array, not {}",
164 k.as_str()
165 ),
166 });
167 }
168 let ranges: Vec<_> = (0..arrays.len()).map(|r| r..r + 1).collect();
169 let lists = crate::elements::Elements::new(&ColumnData::Array(arrays.clone()))
170 .reduce_ranges(reducer, &ranges);
171 let rows: Vec<Option<usize>> = (0..arrays.len())
173 .map(|r| arrays.get(r).map(|_| r))
174 .collect();
175 let data = ColumnData::Array(lists.take(&rows));
176 return self.put_output(CALL, column, options, |name| {
177 Column::from_data(name, data, false)
178 });
179 }
180 if !picks && options.kind.is_some_and(|k| k != ColumnKind::Number) {
181 return Err(PondError::InvalidData {
182 call: CALL,
183 message: format!(
184 "{reducer:?} gives numbers, so the output kind must be number, not {}",
185 options.kind.expect("checked").as_str()
186 ),
187 });
188 }
189 let data: ColumnData = if picks {
190 let picked: Vec<Option<&Scalar>> = arrays
191 .iter()
192 .map(|cell| {
193 let cell = cell?;
194 match reducer {
195 Reducer::First => cell.first(),
196 Reducer::Last => cell.last(),
197 _ => cell.first().filter(|f| cell.iter().all(|e| e == *f)),
200 }
201 })
202 .collect();
203 scalar_column(CALL, &format!("{reducer:?}"), &picked, options.kind)?
204 } else {
205 arrays
206 .iter()
207 .map(|cell| cell.and_then(|cell| reduce_elements(reducer, cell)))
208 .collect::<Vec<Option<f64>>>()
209 .into()
210 };
211 self.put_output(CALL, column, options, |name| {
212 Column::from_data(name, data, false)
213 })
214 }
215
216 pub fn array_explode(&self, column: &str, options: &ArrayOptions) -> Result<TimeSeries> {
244 const CALL: &str = "TimeSeries::array_explode";
245 let arrays = self.array_column(CALL, column)?;
246 let mut rows = Vec::new();
247 let mut elements = Vec::new();
248 for (i, cell) in arrays.iter().enumerate() {
249 for e in cell.unwrap_or(&[]) {
250 rows.push(i);
251 elements.push(Some(e));
252 }
253 }
254 let data = scalar_column(CALL, "element", &elements, options.kind)?;
255 self.take_rows(&rows)
256 .put_output(CALL, column, options, |name| {
257 Column::from_data(name, data, false)
258 })
259 }
260
261 fn array_column(&self, call: &'static str, name: &str) -> Result<&ArrayColumn> {
263 let c = &self.columns()[self.column_index(call, name)?];
264 match c.data() {
265 ColumnData::Array(a) => Ok(a),
266 other => Err(PondError::WrongKind {
267 call,
268 column: name.to_string(),
269 kind: other.kind().as_str(),
270 reason: "this call needs an array column".into(),
271 }),
272 }
273 }
274
275 fn put_output(
278 &self,
279 call: &'static str,
280 source: &str,
281 options: &ArrayOptions,
282 build: impl FnOnce(String) -> Column,
283 ) -> Result<TimeSeries> {
284 let c = self.column_index(call, source)?;
285 let mut columns = self.columns().to_vec();
286 match options.output.as_deref() {
287 None => columns[c] = build(source.to_string()),
288 Some(name) if name == source => columns[c] = build(source.to_string()),
289 Some(name) => {
290 if name == self.key_name() || columns.iter().any(|c| c.name() == name) {
291 return Err(PondError::InvalidData {
292 call,
293 message: format!("output column '{name}' already exists"),
294 });
295 }
296 columns.push(build(name.to_string()));
297 }
298 }
299 Ok(self.with_value_columns(columns))
300 }
301}
302
303fn scalar_column(
307 call: &'static str,
308 what: &str,
309 values: &[Option<&Scalar>],
310 kind: Option<ColumnKind>,
311) -> Result<ColumnData> {
312 let kind = match kind {
313 Some(k @ (ColumnKind::Number | ColumnKind::String | ColumnKind::Boolean)) => k,
314 Some(k) => {
315 return Err(PondError::InvalidData {
316 call,
317 message: format!(
318 "the output kind must be number, string or boolean, not {}",
319 k.as_str()
320 ),
321 });
322 }
323 None => values
324 .iter()
325 .flatten()
326 .next()
327 .map_or(ColumnKind::String, |e| e.kind()),
328 };
329 if let Some((row, e)) = values
330 .iter()
331 .enumerate()
332 .find_map(|(i, v)| v.filter(|e| e.kind() != kind).map(|e| (i, e)))
333 {
334 return Err(PondError::InvalidData {
335 call,
336 message: format!(
337 "output row {row}: {what} is a {}, but the output column is {}; a column holds one kind",
338 e.kind().as_str(),
339 kind.as_str(),
340 ),
341 });
342 }
343 Ok(match kind {
344 ColumnKind::Number => values
345 .iter()
346 .map(|v| v.and_then(Scalar::as_f64))
347 .collect::<Vec<Option<f64>>>()
348 .into(),
349 ColumnKind::String => {
350 StringColumn::from_options(values.iter().map(|v| v.and_then(Scalar::as_str))).into()
351 }
352 _ => BooleanColumn::from_options(values.iter().map(|v| v.and_then(Scalar::as_bool))).into(),
353 })
354}
355
356fn reduce_elements(reducer: Reducer, cell: &[Scalar]) -> Option<f64> {
360 let numeric: Vec<f64> = cell.iter().filter_map(Scalar::as_f64).collect();
361 let out = match reducer {
362 Reducer::Count => Some(cell.len() as f64),
363 Reducer::Sum => Some(numeric.iter().fold(0.0, |s, v| s + v)),
364 Reducer::Avg => (!numeric.is_empty())
365 .then(|| numeric.iter().fold(0.0, |s, v| s + v) / numeric.len() as f64),
366 Reducer::Min => numeric
368 .iter()
369 .copied()
370 .reduce(|a, b| if a <= b { a } else { b }),
371 Reducer::Max => numeric
372 .iter()
373 .copied()
374 .reduce(|a, b| if a >= b { a } else { b }),
375 Reducer::Median => sorted_percentile(numeric, 50.0),
376 Reducer::Percentile(q) => sorted_percentile(numeric, q),
377 Reducer::Difference => {
380 let (mut lo, mut hi) = (*numeric.first()?, *numeric.first()?);
381 for &v in &numeric[1..] {
382 if v < lo {
383 lo = v;
384 }
385 if v > hi {
386 hi = v;
387 }
388 }
389 Some(hi - lo)
390 }
391 Reducer::Stdev => {
392 let (mut n, mut mean, mut m2) = (0.0f64, 0.0f64, 0.0f64);
394 for &v in &numeric {
395 n += 1.0;
396 let delta = v - mean;
397 mean += delta / n;
398 m2 += delta * (v - mean);
399 }
400 (n != 0.0).then(|| clamp_at_zero(m2 / n).sqrt())
401 }
402 Reducer::First | Reducer::Last | Reducer::Keep => unreachable!("picked, not reduced"),
403 Reducer::Unique | Reducer::Top(_) | Reducer::Samples => unreachable!("lists, not reduced"),
404 };
405 out.filter(|v| v.is_finite())
406}
407
408fn sorted_percentile(mut numeric: Vec<f64>, q: f64) -> Option<f64> {
411 if numeric.is_empty() {
412 return None;
413 }
414 numeric.sort_by(|a, b| a.partial_cmp(b).expect("elements are finite"));
415 Some(percentile_of_sorted(&numeric, q))
416}
417
418#[cfg(test)]
419mod tests {
420 use super::*;
421 use crate::{Agg, Fill, Value};
422
423 fn series() -> TimeSeries {
424 TimeSeries::new(
425 "s",
426 vec![0, 1, 2, 3],
427 vec![
428 Column::optional(
429 "tags",
430 vec![
431 Some(vec![Scalar::from("a"), "b".into()]),
432 Some(vec![]),
433 None,
434 Some(vec![Scalar::from(-0.0), true.into()]),
435 ],
436 ),
437 Column::required("v", vec![1.0, 2.0, 3.0, 4.0]),
438 ],
439 )
440 .unwrap()
441 }
442
443 fn err(r: Result<TimeSeries>) -> String {
444 r.unwrap_err().to_string()
445 }
446
447 #[test]
448 fn wrong_columns_are_errors_not_empty_results() {
449 let s = series();
450 let e = err(s.array_contains("v", 1.0));
451 assert!(e.contains("column 'v' is a number column"), "{e}");
452 assert!(err(s.array_contains("nope", 1.0)).contains("no column named 'nope'"));
453 let e = err(s.array_aggregate("v", Reducer::Count, &ArrayOptions::default()));
454 assert!(e.contains("needs an array column"), "{e}");
455 assert!(s.array_explode("v", &ArrayOptions::default()).is_err());
456 }
457
458 #[test]
459 fn contains_matches_zero_of_either_sign_and_keeps_the_schema() {
460 let s = series();
461 let got = s.array_contains("tags", 0.0).unwrap();
462 assert_eq!(got.key().begin(), &[3]);
463 assert_eq!(got.schema(), s.schema());
464 assert_eq!(s.array_contains("tags", "a").unwrap().len(), 1);
465 assert_eq!(s.array_contains("tags", 1.0).unwrap().len(), 0);
466 assert_eq!(s.array_contains_all("tags", &[]).unwrap().len(), 3);
467 assert_eq!(s.array_contains_any("tags", &[]).unwrap().len(), 0);
468 }
469
470 #[test]
471 fn first_and_explode_take_the_elements_kind() {
472 let s = TimeSeries::new(
473 "s",
474 vec![0, 1],
475 vec![Column::required(
476 "flags",
477 vec![vec![Scalar::from(true), false.into()], vec![false.into()]],
478 )],
479 )
480 .unwrap();
481 let first = s
482 .array_aggregate("flags", Reducer::First, &ArrayOptions::default())
483 .unwrap();
484 let got: Vec<_> = first
485 .column("flags")
486 .unwrap()
487 .booleans()
488 .unwrap()
489 .iter()
490 .collect();
491 assert_eq!(got, [Some(true), Some(false)]);
492 assert!(!first.column("flags").unwrap().is_required());
493 let rows = s.array_explode("flags", &ArrayOptions::default()).unwrap();
494 assert_eq!(rows.key().begin(), &[0, 0, 1]);
495 assert_eq!(rows.column("flags").unwrap().kind(), ColumnKind::Boolean);
496
497 let e = err(series().array_explode("tags", &ArrayOptions::default()));
499 assert!(
500 e.contains("output row 2: element is a number, but the output column is string"),
501 "{e}"
502 );
503 let as_number = ArrayOptions {
505 kind: Some(ColumnKind::Number),
506 ..ArrayOptions::default()
507 };
508 let e = err(series().array_aggregate("tags", Reducer::First, &as_number));
509 assert!(e.contains("output row 0: First is a string"), "{e}");
510 let gaps = series().slice(1..3);
512 let first = gaps
513 .array_aggregate("tags", Reducer::First, &ArrayOptions::default())
514 .unwrap();
515 assert_eq!(first.column("tags").unwrap().kind(), ColumnKind::String);
516 }
517
518 #[test]
519 fn options_are_checked_up_front() {
520 let s = series();
521 let text = ArrayOptions {
522 kind: Some(ColumnKind::String),
523 ..ArrayOptions::default()
524 };
525 assert!(s.array_aggregate("tags", Reducer::Count, &text).is_err());
526 let array = ArrayOptions {
527 kind: Some(ColumnKind::Array),
528 ..ArrayOptions::default()
529 };
530 let e = err(s.array_explode("tags", &array));
531 assert!(e.contains("must be number, string or boolean"), "{e}");
532 for taken in ["v", "time"] {
533 let o = ArrayOptions {
534 output: Some(taken.into()),
535 ..ArrayOptions::default()
536 };
537 let e = err(s.array_aggregate("tags", Reducer::Count, &o));
538 assert!(e.contains("already exists"), "{e}");
539 }
540 assert!(
541 s.array_aggregate("tags", Reducer::Percentile(200.0), &ArrayOptions::default())
542 .is_err()
543 );
544 }
545
546 #[test]
547 fn arrays_through_other_calls() {
548 let s = series();
549 let bad = TimeSeries::new(
551 "s",
552 vec![0],
553 vec![Column::required("a", vec![vec![Scalar::from(f64::NAN)]])],
554 );
555 let e = bad.unwrap_err().to_string();
556 assert!(
557 e.contains("column 'a' row 0: array element 0 must be"),
558 "{e}"
559 );
560 let e = s.partition_by(&["tags"]).unwrap_err().to_string();
562 assert!(e.contains("partitioning by an array column"), "{e}");
563 let hosts = s
565 .with_column(Column::required("host", vec!["a", "b", "a", "b"]))
566 .unwrap();
567 let p = hosts.partition_by(&["host"]).unwrap();
568 let swapped = p.map(|g| g.select(&["tags"])?.rename(&[("tags", "host")]));
569 let e = swapped.unwrap_err().to_string();
570 assert!(e.contains("partitioning by an array column"), "{e}");
571 let seq = crate::Sequence::every("10s").unwrap();
572 let e = p
573 .aggregate(&seq, &[Agg::new("host", "tags", Reducer::Last)])
574 .unwrap_err()
575 .to_string();
576 assert!(e.contains("partitioning by an array column"), "{e}");
577 let e = s.to_arrow().unwrap_err().to_string();
579 assert!(
580 e.contains("column 'tags' is an array column; row 3 element 0 is a number"),
581 "{e}"
582 );
583 let row = s.row(0).unwrap();
585 assert_eq!(row.array("tags").unwrap().map(<[Scalar]>::len), Some(2));
586 assert!(row.array("v").is_err());
587 let r = s.reduce(&[Agg::of("tags", Reducer::Last)]).unwrap();
588 assert_eq!(r.array("tags").unwrap().map(<[Scalar]>::len), Some(2));
589 assert!(matches!(r.get("tags").unwrap(), Some(Value::Array(_))));
590 assert!(s.reduce(&[Agg::of("tags", Reducer::Sum)]).is_err());
592 let f = s
594 .fill(&[("tags", Fill::Value(Value::Array(vec!["x".into()])))])
595 .unwrap();
596 assert_eq!(
597 f.column("tags").unwrap().arrays().unwrap().get(2),
598 Some(&["x".into()][..])
599 );
600 let nan = Fill::Value(Value::Array(vec![f64::INFINITY.into()]));
601 assert!(s.fill(&[("tags", nan)]).is_err());
602 let e = s
604 .map_arrays("tags", |_| vec![f64::NAN.into()])
605 .unwrap_err()
606 .to_string();
607 assert!(e.contains("row 0 of 'tags'"), "{e}");
608 let lens = s
609 .map_arrays("tags", |a| vec![(a.len() as f64).into()])
610 .unwrap();
611 let tags = lens.column("tags").unwrap().arrays().unwrap();
612 assert_eq!(tags.get(1), Some(&[Scalar::from(0.0)][..]));
613 assert_eq!(tags.get(2), None);
614 }
615
616 #[test]
617 fn equal_columns_compare_zeros_by_sign() {
618 let a = ArrayColumn::from_options([Some(vec![Scalar::from(0.0)])]);
619 let b = ArrayColumn::from_options([Some(vec![Scalar::from(-0.0)])]);
620 assert_ne!(a, b);
621 assert_eq!(a, a.clone());
622 let c = ColumnData::from(ArrayColumn::from_options([None, Some(vec![]), None]));
624 let tail = c.slice(1..3);
625 assert_eq!(tail.len(), 2);
626 assert!(tail.is_present(0) && !tail.is_present(1));
627 assert_eq!(tail.missing_count(), 1);
628 }
629
630 fn s(f: f64) -> Scalar {
631 Scalar::Number(f)
632 }
633
634 #[test]
635 fn reducers_follow_pond_ts_row_path() {
636 let mixed = [s(-0.0), s(0.0), s(2.0), s(3.5)];
637 assert_eq!(reduce_elements(Reducer::Count, &mixed), Some(4.0));
638 assert_eq!(reduce_elements(Reducer::Sum, &mixed), Some(5.5));
639 assert_eq!(reduce_elements(Reducer::Avg, &mixed), Some(1.375));
640 let min = reduce_elements(Reducer::Min, &mixed).unwrap();
641 assert!(min == 0.0 && min.is_sign_negative());
642 let min = reduce_elements(Reducer::Min, &[s(0.0), s(-0.0)]).unwrap();
643 assert!(min.is_sign_positive());
644 assert_eq!(
645 reduce_elements(Reducer::Stdev, &mixed),
646 Some(1.4737282653189494)
647 );
648 let text = [Scalar::from("a")];
649 assert_eq!(reduce_elements(Reducer::Sum, &text), Some(0.0));
650 assert_eq!(reduce_elements(Reducer::Avg, &text), None);
651 assert_eq!(reduce_elements(Reducer::Count, &[]), Some(0.0));
652 assert_eq!(
654 reduce_elements(Reducer::Sum, &[s(f64::MAX), s(f64::MAX)]),
655 None
656 );
657 }
658}