harmos_storage/query/transform/
downsample.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Storage Arrow Query Contract
11
12/* === Modules === */
13
14/* === Re-Exports === */
15
16pub use harmos_domain::model::storage::query::transform::downsample::{Algorithm, Downsample};
17
18/* === Imports === */
19
20// Standard Library Imports
21
22// Third Party Imports
23use arrow_array::RecordBatch;
24use arrow_schema::{DataType, Schema, TimeUnit};
25use harmos_domain::model::storage::query::scalar::ScalarValue;
26
27// Local Imports
28use super::{invalid, numeric_type, take};
29use crate::Result;
30use crate::storage::query::transform::FieldPathExecution;
31
32/* === Constants === */
33
34/* === Macros === */
35
36/* === Type Aliases === */
37
38/* === Structs === */
39
40/* === Enums === */
41
42/* === Traits === */
43
44pub trait DownsampleExecution {
45    fn output_schema(&self, input: &Schema) -> Result<Schema>;
46
47    fn apply(&self, input: &RecordBatch, check: &mut impl FnMut() -> Result<()>) -> Result<RecordBatch>;
48}
49
50/* === Implementations === */
51
52impl DownsampleExecution for Downsample {
53    fn output_schema(&self, input: &Schema) -> Result<Schema> {
54        if self.max_rows == 0 {
55            return Err(invalid("downsample max_rows must be positive"));
56        }
57        if !matches!(self.algorithm, Algorithm::Decimate) {
58            let field = self.value.as_ref().ok_or_else(|| invalid("downsampling algorithm requires value field"))?.field(input)?;
59            if !numeric_type(field.data_type()) {
60                return Err(invalid("downsampling requires numeric values"));
61            }
62            let minimum = if matches!(self.algorithm, Algorithm::Extrema) { 2 } else { 3 };
63            if self.max_rows < minimum {
64                return Err(invalid("downsampling output bound is below algorithm minimum"));
65            }
66        }
67        if matches!(self.algorithm, Algorithm::LttbLog) && self.axis.is_none() {
68            return Err(invalid("logarithmic LTTB requires an explicit positive numeric axis"));
69        }
70        if let Some(axis) = &self.axis {
71            let field = axis.field(input)?;
72            if matches!(self.algorithm, Algorithm::LttbLog) && !numeric_type(field.data_type()) {
73                return Err(invalid("logarithmic LTTB requires a numeric axis"));
74            }
75            if !numeric_type(field.data_type()) && !matches!(field.data_type(), DataType::Timestamp(TimeUnit::Nanosecond, _)) {
76                return Err(invalid("downsample axis must be numeric or a timestamp"));
77            }
78        }
79        Ok(input.clone())
80    }
81
82    fn apply(&self, input: &RecordBatch, check: &mut impl FnMut() -> Result<()>) -> Result<RecordBatch> {
83        if input.num_rows() <= self.max_rows {
84            return Ok(input.clone());
85        }
86        let indices = match self.algorithm {
87            Algorithm::Decimate => crate::signal::decimate(input.num_rows(), self.max_rows),
88            Algorithm::Extrema => {
89                let field = self.value.as_ref().ok_or_else(|| invalid("missing extrema value field"))?;
90                let mut values = Vec::with_capacity(input.num_rows());
91                for row in 0..input.num_rows() {
92                    check()?;
93                    let value = number(&field.value(input, row)?)?;
94                    if !value.is_finite() {
95                        return Err(invalid("extrema requires finite non-null values"));
96                    }
97                    values.push(value);
98                }
99                crate::signal::extrema(&values, self.max_rows / 2)
100            }
101            Algorithm::Lttb | Algorithm::LttbLog => {
102                let field = self.value.as_ref().ok_or_else(|| invalid("missing LTTB value field"))?;
103                let mut x = Vec::with_capacity(input.num_rows());
104                let mut y = Vec::with_capacity(input.num_rows());
105                let mut first_time = None;
106                for row in 0..input.num_rows() {
107                    check()?;
108                    let value = number(&field.value(input, row)?)?;
109                    let axis = match &self.axis {
110                        None => row as f64,
111                        Some(axis) => match axis.value(input, row)? {
112                            ScalarValue::Datetime(value) => {
113                                let first = *first_time.get_or_insert(value);
114                                value.checked_sub(first).ok_or_else(|| invalid("LTTB timestamp difference overflow"))? as f64
115                            }
116                            value => number(&value)?,
117                        },
118                    };
119                    if !axis.is_finite() || !value.is_finite() {
120                        return Err(invalid("LTTB requires finite non-null coordinates"));
121                    }
122                    if matches!(self.algorithm, Algorithm::LttbLog) && (axis <= 0.0 || x.last().is_some_and(|previous| *previous > axis)) {
123                        return Err(invalid("logarithmic LTTB requires positive ordered coordinates"));
124                    }
125                    x.push(axis);
126                    y.push(value);
127                }
128                let selected = match self.algorithm {
129                    Algorithm::Lttb => crate::signal::lttb(&x, &y, self.max_rows),
130                    Algorithm::LttbLog => crate::signal::lttb_log(&x, &y, self.max_rows),
131                    _ => unreachable!(),
132                };
133                check()?;
134                selected
135            }
136        };
137        take(input, indices.into_iter().map(|index| index as u64).collect())
138    }
139}
140
141/* === Public Functions === */
142
143/* === Private Functions === */
144
145pub(in crate::query) fn number(value: &ScalarValue) -> Result<f64> {
146    match value {
147        ScalarValue::Float(value) => Ok(*value),
148        value => {
149            let exact = harmos_domain::model::storage::query::scalar::exact(value)
150                .ok_or_else(|| invalid("numeric field has a null or nonnumeric value"))?;
151            let value = exact.numerator as f64 / exact.denominator.get() as f64;
152            Ok(if exact.negative { -value } else { value })
153        }
154    }
155}
156
157/* === Tests === */
harmos_storage/query/transform/downsample.rs

harmos_storage/query/transform/
downsample.rs
