common_function/scalars/anomaly/
iqr.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_iqr` window function — IQR-based anomaly scoring.
16//!
17//! Algorithm:
18//! - Compute Q1 (25th percentile) and Q3 (75th percentile)
19//! - IQR = Q3 - Q1
20//! - Lower fence = Q1 - k * IQR, Upper fence = Q3 + k * IQR
21//! - If value is outside fences, score = |distance to nearest fence| / IQR
22//! - Otherwise, score = 0.0
23//!
24//! When IQR = 0 (constant quartiles), returns 0.0 if value is on the fence,
25//! or +inf if value is outside.
26
27use std::fmt::Debug;
28use std::ops::Range;
29use std::sync::Arc;
30
31use arrow::array::{Array, ArrayRef, Float64Array};
32use arrow::datatypes::{DataType, Field, FieldRef};
33use datafusion_common::{DataFusionError, Result, ScalarValue};
34use datafusion_expr::{PartitionEvaluator, Signature, Volatility, WindowUDFImpl};
35use datafusion_functions_window_common::field::WindowUDFFieldArgs;
36use datafusion_functions_window_common::partition::PartitionEvaluatorArgs;
37
38use crate::helper::NUMERICS;
39use crate::scalars::anomaly::utils::{cast_to_f64, collect_window_values, percentile_sorted};
40
41/// Minimum valid samples for IQR (linear-interpolated Q1 != Q3 is possible at n >= 3).
42const MIN_SAMPLES: usize = 3;
43
44#[derive(Debug, Clone, PartialEq, Eq, Hash)]
45pub struct AnomalyScoreIqr {
46    signature: Signature,
47}
48
49impl AnomalyScoreIqr {
50    pub fn new() -> Self {
51        Self {
52            signature: Signature::uniform(2, NUMERICS.to_vec(), Volatility::Immutable),
53        }
54    }
55}
56
57impl WindowUDFImpl for AnomalyScoreIqr {
58    fn name(&self) -> &str {
59        "anomaly_score_iqr"
60    }
61
62    fn signature(&self) -> &Signature {
63        &self.signature
64    }
65
66    fn field(&self, field_args: WindowUDFFieldArgs) -> Result<FieldRef> {
67        Ok(Arc::new(Field::new(
68            field_args.name(),
69            DataType::Float64,
70            true, // nullable
71        )))
72    }
73
74    fn partition_evaluator(
75        &self,
76        _partition_evaluator_args: PartitionEvaluatorArgs,
77    ) -> Result<Box<dyn PartitionEvaluator>> {
78        Ok(Box::new(AnomalyScoreIqrEvaluator { current_row: 0 }))
79    }
80}
81
82#[derive(Debug)]
83struct AnomalyScoreIqrEvaluator {
84    /// Tracks the current row index within the partition.
85    current_row: usize,
86}
87
88impl PartitionEvaluator for AnomalyScoreIqrEvaluator {
89    fn uses_window_frame(&self) -> bool {
90        true
91    }
92
93    fn supports_bounded_execution(&self) -> bool {
94        false
95    }
96
97    fn evaluate(&mut self, values: &[ArrayRef], range: &Range<usize>) -> Result<ScalarValue> {
98        // TODO(dennis): memoize the converted arrays to avoid re-casting inside each row evaluation
99        let values_f64 = cast_to_f64(&values[0])?;
100        let array = values_f64
101            .as_any()
102            .downcast_ref::<Float64Array>()
103            .ok_or_else(|| {
104                DataFusionError::Internal(format!(
105                    "Expected Float64Array, got: {:?}",
106                    values_f64.data_type()
107                ))
108            })?;
109
110        // Extract k from the second argument (constant across the window)
111        let k_f64 = cast_to_f64(&values[1])?;
112        let k_array = k_f64
113            .as_any()
114            .downcast_ref::<Float64Array>()
115            .ok_or_else(|| {
116                DataFusionError::Internal(format!(
117                    "Expected Float64Array for k, got: {:?}",
118                    k_f64.data_type()
119                ))
120            })?;
121
122        // Use the tracked current row index — correct for any window frame.
123        let current_idx = self.current_row;
124        self.current_row += 1;
125
126        // Check bounds and validity of the current row.
127        if current_idx >= array.len()
128            || !array.is_valid(current_idx)
129            || !array.value(current_idx).is_finite()
130        {
131            return Ok(ScalarValue::Float64(None));
132        }
133        let current_value = array.value(current_idx);
134
135        // Get k value (use current row's k)
136        if current_idx >= k_array.len()
137            || !k_array.is_valid(current_idx)
138            || !k_array.value(current_idx).is_finite()
139        {
140            return Ok(ScalarValue::Float64(None));
141        }
142        let k = k_array.value(current_idx);
143        if k < 0.0 {
144            return Ok(ScalarValue::Float64(None));
145        }
146
147        let mut window_values = collect_window_values(array, range);
148        if window_values.len() < MIN_SAMPLES {
149            return Ok(ScalarValue::Float64(None));
150        }
151
152        // Sort for percentile computation
153        window_values.sort_unstable_by(|a, b| a.partial_cmp(b).unwrap());
154
155        let q1 = match percentile_sorted(&window_values, 0.25) {
156            Some(v) => v,
157            None => return Ok(ScalarValue::Float64(None)),
158        };
159        let q3 = match percentile_sorted(&window_values, 0.75) {
160            Some(v) => v,
161            None => return Ok(ScalarValue::Float64(None)),
162        };
163        let iqr = q3 - q1;
164
165        let lower_fence = q1 - k * iqr;
166        let upper_fence = q3 + k * iqr;
167
168        let score = if current_value < lower_fence {
169            (lower_fence - current_value) / iqr
170        } else if current_value > upper_fence {
171            (current_value - upper_fence) / iqr
172        } else {
173            0.0
174        };
175
176        Ok(ScalarValue::Float64(Some(score)))
177    }
178}
179
180#[cfg(test)]
181mod tests {
182    use std::sync::Arc;
183
184    use arrow::array::Float64Array;
185    use datafusion_expr::WindowUDF;
186
187    use super::*;
188
189    fn eval_iqr(values: &[Option<f64>], k: f64, range: Range<usize>) -> ScalarValue {
190        let array = Arc::new(Float64Array::from(values.to_vec())) as ArrayRef;
191        let k_array = Arc::new(Float64Array::from(vec![Some(k); values.len()])) as ArrayRef;
192        let current_row = range.end.saturating_sub(1);
193        let mut evaluator = AnomalyScoreIqrEvaluator { current_row };
194        evaluator.evaluate(&[array, k_array], &range).unwrap()
195    }
196
197    #[test]
198    fn test_basic_outlier() {
199        // Normal range [1..10] with outlier at 100
200        let mut values: Vec<Option<f64>> = (1..=10).map(|x| Some(x as f64)).collect();
201        values.push(Some(100.0));
202        let result = eval_iqr(&values, 1.5, 0..11);
203        match result {
204            ScalarValue::Float64(Some(score)) => assert!(score > 0.0, "score={score}"),
205            other => panic!("expected Some(score), got {other:?}"),
206        }
207    }
208
209    #[test]
210    fn test_value_within_fences() {
211        let values: Vec<Option<f64>> = (1..=10).map(|x| Some(x as f64)).collect();
212        // Last value is 10, which is within fences for k=1.5
213        let result = eval_iqr(&values, 1.5, 0..10);
214        match result {
215            ScalarValue::Float64(Some(score)) => {
216                assert!(score >= 0.0, "score should be non-negative, got {score}");
217            }
218            other => panic!("expected Some(score), got {other:?}"),
219        }
220    }
221
222    #[test]
223    fn test_constant_sequence() {
224        let values: Vec<Option<f64>> = vec![Some(5.0); 10];
225        // IQR=0 and current is on the fence -> 0.0
226        let result = eval_iqr(&values, 1.5, 0..10);
227        assert_eq!(result, ScalarValue::Float64(Some(0.0)));
228    }
229
230    #[test]
231    fn test_zero_iqr_outlier_is_infinite() {
232        // Q1=Q3=1.0 -> IQR=0.0, current value is outside fence.
233        let values: Vec<Option<f64>> =
234            vec![Some(1.0), Some(1.0), Some(1.0), Some(1.0), Some(100.0)];
235        let result = eval_iqr(&values, 1.5, 0..5);
236        match result {
237            ScalarValue::Float64(Some(score)) => {
238                assert!(score.is_infinite() && score.is_sign_positive())
239            }
240            other => panic!("expected Some(+inf), got {other:?}"),
241        }
242    }
243
244    #[test]
245    fn test_all_null() {
246        let values: Vec<Option<f64>> = vec![None; 5];
247        let result = eval_iqr(&values, 1.5, 0..5);
248        assert_eq!(result, ScalarValue::Float64(None));
249    }
250
251    #[test]
252    fn test_insufficient_samples() {
253        // IQR requires min_samples=3; 2 points should return NULL
254        let values: Vec<Option<f64>> = vec![Some(1.0), Some(2.0)];
255        let result = eval_iqr(&values, 1.5, 0..2);
256        assert_eq!(result, ScalarValue::Float64(None));
257    }
258
259    #[test]
260    fn test_negative_k() {
261        let values: Vec<Option<f64>> =
262            vec![Some(48.0), Some(49.0), Some(50.0), Some(51.0), Some(52.0)];
263        let result = eval_iqr(&values, -1.0, 0..5);
264        assert_eq!(result, ScalarValue::Float64(None));
265    }
266
267    #[test]
268    fn test_current_row_null() {
269        let values: Vec<Option<f64>> = vec![Some(1.0), Some(2.0), Some(3.0), None];
270        let result = eval_iqr(&values, 1.5, 0..4);
271        assert_eq!(result, ScalarValue::Float64(None));
272    }
273
274    #[test]
275    fn test_lower_outlier() {
276        // Data mostly around 50, one low value at the end
277        let values: Vec<Option<f64>> = vec![
278            Some(48.0),
279            Some(49.0),
280            Some(50.0),
281            Some(51.0),
282            Some(52.0),
283            Some(-100.0),
284        ];
285        let result = eval_iqr(&values, 1.5, 0..6);
286        match result {
287            ScalarValue::Float64(Some(score)) => assert!(score > 0.0, "score={score}"),
288            other => panic!("expected Some(score), got {other:?}"),
289        }
290
291        // Test that a value in the middle gets 0
292        let values: Vec<Option<f64>> = vec![
293            Some(48.0),
294            Some(49.0),
295            Some(50.0),
296            Some(51.0),
297            Some(52.0),
298            Some(-100.0),
299            Some(50.0),
300        ];
301        let result = eval_iqr(&values, 1.5, 0..7);
302        assert_eq!(result, ScalarValue::Float64(Some(0.0)));
303    }
304
305    #[test]
306    fn test_udwf_creation() {
307        let udwf = WindowUDF::from(AnomalyScoreIqr::new());
308        assert_eq!(udwf.name(), "anomaly_score_iqr");
309    }
310}
common_function/scalars/anomaly/iqr.rs

common_function/scalars/anomaly/
iqr.rs
