common_function/scalars/anomaly/
mad.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! `anomaly_score_mad` window function — MAD-based anomaly scoring.
16//!
17//! Algorithm: `score = |x - median(window)| / (MAD * 1.4826)`
18//! where `MAD = median(|xi - median(window)|)`
19//!
20//! When MAD = 0 (majority-constant window), returns 0.0 if value equals
21//! median, or +inf otherwise.
22
23use std::fmt::Debug;
24use std::ops::Range;
25use std::sync::Arc;
26
27use arrow::array::{Array, ArrayRef, Float64Array};
28use arrow::datatypes::{DataType, Field, FieldRef};
29use datafusion_common::{DataFusionError, Result, ScalarValue};
30use datafusion_expr::{PartitionEvaluator, Signature, Volatility, WindowUDFImpl};
31use datafusion_functions_window_common::field::WindowUDFFieldArgs;
32use datafusion_functions_window_common::partition::PartitionEvaluatorArgs;
33
34use crate::helper::NUMERICS;
35use crate::scalars::anomaly::utils::{
36    anomaly_ratio, cast_to_f64, collect_window_values, median_f64,
37};
38
39/// Minimum valid samples for MAD (n <= 2 makes MAD almost always 0, yielding spurious +inf).
40const MIN_SAMPLES: usize = 3;
41
42/// MAD consistency constant for normal distribution: `1 / Φ⁻¹(3/4) ≈ 1.4826`
43const MAD_CONSISTENCY_CONSTANT: f64 = 1.4826;
44
45#[derive(Debug, Clone, PartialEq, Eq, Hash)]
46pub struct AnomalyScoreMad {
47    signature: Signature,
48}
49
50impl AnomalyScoreMad {
51    pub fn new() -> Self {
52        Self {
53            signature: Signature::uniform(1, NUMERICS.to_vec(), Volatility::Immutable),
54        }
55    }
56}
57
58impl WindowUDFImpl for AnomalyScoreMad {
59    fn name(&self) -> &str {
60        "anomaly_score_mad"
61    }
62
63    fn signature(&self) -> &Signature {
64        &self.signature
65    }
66
67    fn field(&self, field_args: WindowUDFFieldArgs) -> Result<FieldRef> {
68        Ok(Arc::new(Field::new(
69            field_args.name(),
70            DataType::Float64,
71            true, // nullable
72        )))
73    }
74
75    fn partition_evaluator(
76        &self,
77        _partition_evaluator_args: PartitionEvaluatorArgs,
78    ) -> Result<Box<dyn PartitionEvaluator>> {
79        Ok(Box::new(AnomalyScoreMadEvaluator { current_row: 0 }))
80    }
81}
82
83#[derive(Debug)]
84struct AnomalyScoreMadEvaluator {
85    /// Tracks the current row index within the partition.
86    /// DataFusion calls `evaluate()` sequentially for row 0, 1, 2, …
87    /// and creates a fresh evaluator per partition.
88    current_row: usize,
89}
90
91impl PartitionEvaluator for AnomalyScoreMadEvaluator {
92    fn uses_window_frame(&self) -> bool {
93        true
94    }
95
96    fn supports_bounded_execution(&self) -> bool {
97        false
98    }
99
100    fn evaluate(&mut self, values: &[ArrayRef], range: &Range<usize>) -> Result<ScalarValue> {
101        let values_f64 = cast_to_f64(&values[0])?;
102        let array = values_f64
103            .as_any()
104            .downcast_ref::<Float64Array>()
105            .ok_or_else(|| {
106                DataFusionError::Internal(format!(
107                    "Expected Float64Array, got: {:?}",
108                    values_f64.data_type()
109                ))
110            })?;
111
112        // Use the tracked current row index — correct for any window frame
113        // (trailing, leading, centered, unbounded).
114        let current_idx = self.current_row;
115        self.current_row += 1;
116
117        if current_idx >= array.len()
118            || !array.is_valid(current_idx)
119            || !array.value(current_idx).is_finite()
120        {
121            return Ok(ScalarValue::Float64(None));
122        }
123        let current_value = array.value(current_idx);
124
125        let mut window_values = collect_window_values(array, range);
126        if window_values.len() < MIN_SAMPLES {
127            return Ok(ScalarValue::Float64(None));
128        }
129
130        // Compute median of window
131        let median = match median_f64(&mut window_values) {
132            Some(m) => m,
133            None => return Ok(ScalarValue::Float64(None)),
134        };
135
136        // Compute MAD = median(|xi - median|)
137        let mut abs_deviations: Vec<f64> =
138            window_values.iter().map(|x| (x - median).abs()).collect();
139        let mad = match median_f64(&mut abs_deviations) {
140            Some(m) => m,
141            None => return Ok(ScalarValue::Float64(None)),
142        };
143
144        let distance = (current_value - median).abs();
145        let scale = mad * MAD_CONSISTENCY_CONSTANT;
146        let score = anomaly_ratio(distance, scale);
147        Ok(ScalarValue::Float64(Some(score)))
148    }
149}
150
151#[cfg(test)]
152mod tests {
153    use std::sync::Arc;
154
155    use arrow::array::Float64Array;
156    use datafusion_expr::WindowUDF;
157
158    use super::*;
159
160    fn eval_mad(values: &[Option<f64>], range: Range<usize>) -> ScalarValue {
161        let array = Arc::new(Float64Array::from(values.to_vec())) as ArrayRef;
162        // current_row is the last index in the range — the row being scored
163        let current_row = range.end.saturating_sub(1);
164        let mut evaluator = AnomalyScoreMadEvaluator { current_row };
165        evaluator.evaluate(&[array], &range).unwrap()
166    }
167
168    #[test]
169    fn test_basic_outlier() {
170        // Normal data with one outlier (100.0)
171        let values: Vec<Option<f64>> = vec![
172            Some(1.0),
173            Some(2.0),
174            Some(1.5),
175            Some(2.5),
176            Some(1.0),
177            Some(100.0),
178        ];
179        let result = eval_mad(&values, 0..6);
180        match result {
181            ScalarValue::Float64(Some(score)) => assert!(score > 3.0, "score={score}"),
182            other => panic!("expected Some(score), got {other:?}"),
183        }
184    }
185
186    #[test]
187    fn test_constant_sequence() {
188        let values: Vec<Option<f64>> = vec![Some(5.0); 10];
189        // All values are the same -> zero spread and zero distance => 0.0
190        let result = eval_mad(&values, 0..10);
191        match result {
192            ScalarValue::Float64(Some(score)) => assert_eq!(score, 0.0),
193            other => panic!("expected Some(0.0), got {other:?}"),
194        }
195    }
196
197    #[test]
198    fn test_mad_zero_non_median() {
199        // More than 50% identical and current differs from median -> non-zero / 0 => +inf
200        let values: Vec<Option<f64>> = vec![Some(1.0), Some(1.0), Some(1.0), Some(1.0), Some(5.0)];
201        let result = eval_mad(&values, 0..5);
202        match result {
203            ScalarValue::Float64(Some(score)) => {
204                assert!(score.is_infinite() && score.is_sign_positive())
205            }
206            other => panic!("expected Some(+inf), got {other:?}"),
207        }
208    }
209
210    #[test]
211    fn test_all_null() {
212        let values: Vec<Option<f64>> = vec![None, None, None, None];
213        let result = eval_mad(&values, 0..4);
214        assert_eq!(result, ScalarValue::Float64(None));
215    }
216
217    #[test]
218    fn test_insufficient_samples() {
219        let values: Vec<Option<f64>> = vec![Some(1.0), Some(2.0)];
220        let result = eval_mad(&values, 0..2);
221        assert_eq!(result, ScalarValue::Float64(None));
222    }
223
224    #[test]
225    fn test_nan_inf_skipped() {
226        let values: Vec<Option<f64>> = vec![
227            Some(1.0),
228            Some(f64::NAN),
229            Some(f64::INFINITY),
230            Some(2.0),
231            Some(3.0),
232        ];
233        let result = eval_mad(&values, 0..5);
234        // Should compute using [1.0, 2.0, 3.0] only
235        match result {
236            ScalarValue::Float64(Some(_)) => {}
237            other => panic!("expected Some(score), got {other:?}"),
238        }
239    }
240
241    #[test]
242    fn test_current_row_null() {
243        let values: Vec<Option<f64>> = vec![Some(1.0), Some(2.0), Some(3.0), Some(4.0), None];
244        let result = eval_mad(&values, 0..5);
245        assert_eq!(result, ScalarValue::Float64(None));
246    }
247
248    #[test]
249    fn test_udwf_creation() {
250        let udwf = WindowUDF::from(AnomalyScoreMad::new());
251        assert_eq!(udwf.name(), "anomaly_score_mad");
252    }
253}
common_function/scalars/anomaly/mad.rs

common_function/scalars/anomaly/
mad.rs
