linkrs_query/executor/expression/functions/builtin/
aggregate.rs

1//! Implementation of aggregate functions
2
3use crate::executor::expression::ExpressionError;
4use linkrs_core::types::operators::AggregateFunction;
5use linkrs_core::value::list::List;
6use linkrs_core::value::NullType;
7use linkrs_core::Expression;
8use linkrs_core::Value;
9use serde::{Deserialize, Serialize};
10
11/// Create an AggregateFunction from a string name
12pub fn aggregate_from_str(func_name: &str) -> Result<AggregateFunction, ExpressionError> {
13    match func_name.to_uppercase().as_str() {
14        "COUNT" | "COUNT_DISTINCT" | "DISTINCT" => Ok(AggregateFunction::Count),
15        "SUM" => Ok(AggregateFunction::Sum),
16        "AVG" => Ok(AggregateFunction::Avg),
17        "MIN" => Ok(AggregateFunction::Min),
18        "MAX" => Ok(AggregateFunction::Max),
19        "COLLECT" => Ok(AggregateFunction::Collect),
20        "PERCENTILE" => Ok(AggregateFunction::Percentile),
21        "STDDEV_POP" => Ok(AggregateFunction::StddevPop),
22        "STDDEV_SAMP" => Ok(AggregateFunction::StddevSamp),
23        "PRODUCT" => Ok(AggregateFunction::Product),
24        "PERCENTILE_CONT" => Ok(AggregateFunction::PercentileCont),
25        "VARIANCE" => Ok(AggregateFunction::Variance),
26        "MEDIAN" => Ok(AggregateFunction::Median),
27        "MODE" => Ok(AggregateFunction::Mode),
28        "BOOL_AND" => Ok(AggregateFunction::BoolAnd),
29        "BOOL_OR" => Ok(AggregateFunction::BoolOr),
30        _ => Err(ExpressionError::function_error(format!(
31            "Unknown aggregate function: {}",
32            func_name
33        ))),
34    }
35}
36
37/// Creating aggregate functions from strings and parameters
38pub fn aggregate_from_str_with_args(
39    func_name: &str,
40    args: &[String],
41) -> Result<AggregateFunction, ExpressionError> {
42    match func_name.to_uppercase().as_str() {
43        "COUNT" | "COUNT_DISTINCT" | "DISTINCT" => Ok(AggregateFunction::Count),
44        "SUM" => {
45            if args.is_empty() {
46                return Err(ExpressionError::function_error(
47                    "SUM function requires a field name".to_string(),
48                ));
49            }
50            Ok(AggregateFunction::Sum)
51        }
52        "AVG" => {
53            if args.is_empty() {
54                return Err(ExpressionError::function_error(
55                    "AVG function requires a field name".to_string(),
56                ));
57            }
58            Ok(AggregateFunction::Avg)
59        }
60        "MIN" => {
61            if args.is_empty() {
62                return Err(ExpressionError::function_error(
63                    "MIN function requires a field name".to_string(),
64                ));
65            }
66            Ok(AggregateFunction::Min)
67        }
68        "MAX" => {
69            if args.is_empty() {
70                return Err(ExpressionError::function_error(
71                    "MAX function requires a field name".to_string(),
72                ));
73            }
74            Ok(AggregateFunction::Max)
75        }
76        "COLLECT" => {
77            if args.is_empty() {
78                return Err(ExpressionError::function_error(
79                    "COLLECT function requires a field name".to_string(),
80                ));
81            }
82            Ok(AggregateFunction::Collect)
83        }
84        "PERCENTILE" => {
85            if args.len() < 2 {
86                return Err(ExpressionError::function_error(
87                    "PERCENTILE function requires a field name and percentile value".to_string(),
88                ));
89            }
90            Ok(AggregateFunction::Percentile)
91        }
92        "PERCENTILE_CONT" => {
93            if args.len() < 2 {
94                return Err(ExpressionError::function_error(
95                    "PERCENTILE_CONT function requires a field name and percentile value"
96                        .to_string(),
97                ));
98            }
99            Ok(AggregateFunction::PercentileCont)
100        }
101        "VEC_SUM" => {
102            if args.is_empty() {
103                return Err(ExpressionError::function_error(
104                    "VEC_SUM function requires a field name".to_string(),
105                ));
106            }
107            Ok(AggregateFunction::VecSum)
108        }
109        "VEC_AVG" => {
110            if args.is_empty() {
111                return Err(ExpressionError::function_error(
112                    "VEC_AVG function requires a field name".to_string(),
113                ));
114            }
115            Ok(AggregateFunction::VecAvg)
116        }
117        "STDDEV_POP" => {
118            if args.is_empty() {
119                return Err(ExpressionError::function_error(
120                    "STDDEV_POP function requires a field name".to_string(),
121                ));
122            }
123            Ok(AggregateFunction::StddevPop)
124        }
125        "STDDEV_SAMP" => {
126            if args.is_empty() {
127                return Err(ExpressionError::function_error(
128                    "STDDEV_SAMP function requires a field name".to_string(),
129                ));
130            }
131            Ok(AggregateFunction::StddevSamp)
132        }
133        "PRODUCT" => {
134            if args.is_empty() {
135                return Err(ExpressionError::function_error(
136                    "PRODUCT function requires a field name".to_string(),
137                ));
138            }
139            Ok(AggregateFunction::Product)
140        }
141        "VARIANCE" => {
142            if args.is_empty() {
143                return Err(ExpressionError::function_error(
144                    "VARIANCE function requires a field name".to_string(),
145                ));
146            }
147            Ok(AggregateFunction::Variance)
148        }
149        "MEDIAN" => {
150            if args.is_empty() {
151                return Err(ExpressionError::function_error(
152                    "MEDIAN function requires a field name".to_string(),
153                ));
154            }
155            Ok(AggregateFunction::Median)
156        }
157        "MODE" => {
158            if args.is_empty() {
159                return Err(ExpressionError::function_error(
160                    "MODE function requires a field name".to_string(),
161                ));
162            }
163            Ok(AggregateFunction::Mode)
164        }
165        "BOOL_AND" => {
166            if args.is_empty() {
167                return Err(ExpressionError::function_error(
168                    "BOOL_AND function requires a field name".to_string(),
169                ));
170            }
171            Ok(AggregateFunction::BoolAnd)
172        }
173        "BOOL_OR" => {
174            if args.is_empty() {
175                return Err(ExpressionError::function_error(
176                    "BOOL_OR function requires a field name".to_string(),
177                ));
178            }
179            Ok(AggregateFunction::BoolOr)
180        }
181        _ => Err(ExpressionError::function_error(format!(
182            "Unknown aggregate function: {}",
183            func_name
184        ))),
185    }
186}
187
188/// Aggregate expressions
189#[derive(Debug, Clone, Serialize, Deserialize)]
190pub struct AggregateExpression {
191    pub function: AggregateFunction,
192    pub argument: Box<Expression>,
193    pub distinct: bool,
194}
195
196impl AggregateExpression {
197    pub fn new(function: AggregateFunction, argument: Expression, distinct: bool) -> Self {
198        Self {
199            function,
200            argument: Box::new(argument),
201            distinct,
202        }
203    }
204
205    /// Calculate the value of the aggregate expression.
206    pub fn evaluate<C: crate::executor::expression::ExpressionContext>(
207        &self,
208        context: &mut C,
209        state: &mut AggregateState,
210    ) -> Result<Value, ExpressionError> {
211        let arg_value =
212            crate::executor::expression::evaluator::expression_evaluator::ExpressionEvaluator::evaluate(
213                &self.argument,
214                context,
215            )
216            .map_err(|e| ExpressionError::function_error(e.to_string()))?;
217
218        // Update the aggregation status.
219        state.update(&self.function, &arg_value, self.distinct);
220
221        // Return the aggregated results of the current state.
222        match &self.function {
223            AggregateFunction::Count => Ok(Value::BigInt(state.count)),
224            AggregateFunction::Sum => Ok(state.sum.clone()),
225            AggregateFunction::Min => Ok(state
226                .min
227                .clone()
228                .unwrap_or(Value::Null(linkrs_core::value::NullType::Null))),
229            AggregateFunction::Max => Ok(state
230                .max
231                .clone()
232                .unwrap_or(Value::Null(linkrs_core::value::NullType::Null))),
233            AggregateFunction::Avg => {
234                if state.count > 0 {
235                    match &state.sum {
236                        Value::SmallInt(i) => Ok(Value::Double(*i as f64 / state.count as f64)),
237                        Value::Int(i) => Ok(Value::Double(*i as f64 / state.count as f64)),
238                        Value::BigInt(i) => Ok(Value::Double(*i as f64 / state.count as f64)),
239                        Value::Float(f) => Ok(Value::Double(*f as f64 / state.count as f64)),
240                        Value::Double(f) => Ok(Value::Double(*f / state.count as f64)),
241                        _ => Ok(Value::Double(0.0)),
242                    }
243                } else {
244                    Ok(Value::Double(0.0))
245                }
246            }
247            AggregateFunction::Collect => Ok(Value::list(List::from(state.values.clone()))),
248            AggregateFunction::CollectSet => Ok(Value::set(
249                state
250                    .values
251                    .iter()
252                    .cloned()
253                    .collect::<std::collections::HashSet<_>>(),
254            )),
255            AggregateFunction::Percentile => state.calculate_percentile(50.0),
256            AggregateFunction::PercentileCont => state.calculate_percentile(50.0),
257            AggregateFunction::Std => state.calculate_std(),
258            AggregateFunction::StddevPop => state.calculate_stddev_pop(),
259            AggregateFunction::StddevSamp => state.calculate_stddev_samp(),
260            AggregateFunction::Product => state.calculate_product(),
261            AggregateFunction::Variance => state.calculate_variance(),
262            AggregateFunction::Median => state.calculate_median(),
263            AggregateFunction::Mode => state.calculate_mode(),
264            AggregateFunction::BitAnd => state.calculate_bit_and(),
265            AggregateFunction::BitOr => state.calculate_bit_or(),
266            AggregateFunction::BoolAnd => state.calculate_bool_and(),
267            AggregateFunction::BoolOr => state.calculate_bool_or(),
268            AggregateFunction::GroupConcat | AggregateFunction::GroupConcatWithOrder => {
269                state.calculate_group_concat()
270            }
271            AggregateFunction::VecSum => Ok(state.vec_sum.clone()),
272            AggregateFunction::VecAvg => {
273                if state.count > 0 {
274                    Ok(state.vec_avg.clone())
275                } else {
276                    Ok(Value::Null(linkrs_core::value::NullType::NaN))
277                }
278            }
279        }
280    }
281}
282
283/// Aggregation status, used to accumulate the intermediate results of aggregate functions.
284#[derive(Debug, Clone)]
285pub struct AggregateState {
286    pub count: i64,
287    pub sum: Value,
288    pub min: Option<Value>,
289    pub max: Option<Value>,
290    pub values: Vec<Value>,
291    pub distinct_values: std::collections::HashSet<String>,
292    pub percentile_values: Vec<f64>,
293    pub std_values: Vec<f64>,
294    pub variance_values: Vec<f64>,
295    pub median_values: Vec<f64>,
296    pub mode_values: Vec<Value>,
297    pub bit_and_value: Option<i64>,
298    pub bit_or_value: Option<i64>,
299    pub bool_and_value: Option<bool>,
300    pub bool_or_value: Option<bool>,
301    pub group_concat_values: Vec<Value>,
302    /// Vector sum for VEC_SUM
303    pub vec_sum: Value,
304    /// Vector average for VEC_AVG
305    pub vec_avg: Value,
306}
307
308impl Default for AggregateState {
309    fn default() -> Self {
310        Self::new()
311    }
312}
313
314impl AggregateState {
315    pub fn new() -> Self {
316        Self {
317            count: 0,
318            sum: Value::Int(0),
319            min: None,
320            max: None,
321            values: Vec::new(),
322            distinct_values: std::collections::HashSet::new(),
323            percentile_values: Vec::new(),
324            std_values: Vec::new(),
325            variance_values: Vec::new(),
326            median_values: Vec::new(),
327            mode_values: Vec::new(),
328            bit_and_value: None,
329            bit_or_value: None,
330            bool_and_value: None,
331            bool_or_value: None,
332            group_concat_values: Vec::new(),
333            vec_sum: Value::Null(linkrs_core::value::NullType::NaN),
334            vec_avg: Value::Null(linkrs_core::value::NullType::NaN),
335        }
336    }
337
338    pub fn reset(&mut self) {
339        self.count = 0;
340        self.sum = Value::Int(0);
341        self.min = None;
342        self.max = None;
343        self.values.clear();
344        self.distinct_values.clear();
345        self.percentile_values.clear();
346        self.std_values.clear();
347        self.variance_values.clear();
348        self.median_values.clear();
349        self.mode_values.clear();
350        self.bit_and_value = None;
351        self.bit_or_value = None;
352        self.bool_and_value = None;
353        self.bool_or_value = None;
354        self.group_concat_values.clear();
355        self.vec_sum = Value::Null(linkrs_core::value::NullType::NaN);
356        self.vec_avg = Value::Null(linkrs_core::value::NullType::NaN);
357    }
358
359    /// Update the aggregation status.
360    pub fn update(&mut self, function: &AggregateFunction, value: &Value, distinct: bool) {
361        let value_str = format!("{}", value);
362
363        // If `distinct` is enabled, check whether it already exists.
364        if distinct && self.distinct_values.contains(&value_str) {
365            return;
366        }
367
368        // The recorded values are used for deduplication (i.e., to remove duplicate entries).
369        if distinct {
370            self.distinct_values.insert(value_str);
371        }
372
373        self.count += 1;
374        self.values.push(value.clone());
375
376        // Special processing is performed depending on the type of aggregate function.
377        match function {
378            AggregateFunction::Percentile | AggregateFunction::PercentileCont => {
379                // Special handling of the PERCENTILE function: Collecting numerical values
380                match value {
381                    Value::SmallInt(v) => self.percentile_values.push(*v as f64),
382                    Value::Int(v) => self.percentile_values.push(*v as f64),
383                    Value::BigInt(v) => self.percentile_values.push(*v as f64),
384                    Value::Float(v) => self.percentile_values.push(*v as f64),
385                    Value::Double(v) => self.percentile_values.push(*v),
386                    _ => {}
387                }
388            }
389            AggregateFunction::Std
390            | AggregateFunction::StddevPop
391            | AggregateFunction::StddevSamp
392            | AggregateFunction::Variance => {
393                // Special handling of the STD and VARIANCE functions: Collecting numerical values
394                match value {
395                    Value::SmallInt(v) => self.std_values.push(*v as f64),
396                    Value::Int(v) => self.std_values.push(*v as f64),
397                    Value::BigInt(v) => self.std_values.push(*v as f64),
398                    Value::Float(v) => self.std_values.push(*v as f64),
399                    Value::Double(v) => self.std_values.push(*v),
400                    _ => {}
401                }
402            }
403            AggregateFunction::Median => {
404                // Special handling of the MEDIAN function: Collecting numerical values
405                match value {
406                    Value::SmallInt(v) => self.median_values.push(*v as f64),
407                    Value::Int(v) => self.median_values.push(*v as f64),
408                    Value::BigInt(v) => self.median_values.push(*v as f64),
409                    Value::Float(v) => self.median_values.push(*v as f64),
410                    Value::Double(v) => self.median_values.push(*v),
411                    _ => {}
412                }
413            }
414            AggregateFunction::Mode => {
415                self.mode_values.push(value.clone());
416            }
417            AggregateFunction::BitAnd => {
418                // Special handling of the BIT_AND function
419                if let Value::BigInt(v) = value {
420                    if let Some(current) = self.bit_and_value {
421                        self.bit_and_value = Some(current & v);
422                    } else {
423                        self.bit_and_value = Some(*v);
424                    }
425                }
426            }
427            AggregateFunction::BitOr => {
428                // Special handling of the BIT_OR function
429                if let Value::BigInt(v) = value {
430                    if let Some(current) = self.bit_or_value {
431                        self.bit_or_value = Some(current | v);
432                    } else {
433                        self.bit_or_value = Some(*v);
434                    }
435                }
436            }
437            AggregateFunction::BoolAnd => {
438                if let Value::Bool(b) = value {
439                    if let Some(current) = self.bool_and_value {
440                        self.bool_and_value = Some(current && *b);
441                    } else {
442                        self.bool_and_value = Some(*b);
443                    }
444                }
445            }
446            AggregateFunction::BoolOr => {
447                if let Value::Bool(b) = value {
448                    if let Some(current) = self.bool_or_value {
449                        self.bool_or_value = Some(current || *b);
450                    } else {
451                        self.bool_or_value = Some(*b);
452                    }
453                }
454            }
455            AggregateFunction::GroupConcat | AggregateFunction::GroupConcatWithOrder => {
456                // Special handling of the GROUP_CONCAT function
457                self.group_concat_values.push(value.clone());
458            }
459            AggregateFunction::VecSum => {
460                // Special handling for VEC_SUM function
461                if matches!(value, Value::Vector(_)) {
462                    if self.vec_sum.is_null() {
463                        self.vec_sum = value.clone();
464                    } else if let (Value::Vector(sum_vec), Value::Vector(input_vec)) =
465                        (&mut self.vec_sum, value)
466                    {
467                        let sum_data = sum_vec.to_dense();
468                        let input_data = input_vec.to_dense();
469
470                        if sum_data.len() == input_data.len() {
471                            let new_data: Vec<f32> = sum_data
472                                .iter()
473                                .zip(input_data.iter())
474                                .map(|(&a, &b)| a + b)
475                                .collect();
476                            self.vec_sum = Value::vector(new_data);
477                        }
478                    }
479                }
480            }
481            AggregateFunction::VecAvg => {
482                // Special handling for VEC_AVG function
483                if matches!(value, Value::Vector(_)) {
484                    if self.vec_avg.is_null() {
485                        self.vec_avg = value.clone();
486                    } else if let (Value::Vector(avg_vec), Value::Vector(input_vec)) =
487                        (&mut self.vec_avg, value)
488                    {
489                        let avg_data = avg_vec.to_dense();
490                        let input_data = input_vec.to_dense();
491
492                        if avg_data.len() == input_data.len() {
493                            // Incremental average calculation
494                            let new_avg: Vec<f32> = avg_data
495                                .iter()
496                                .zip(input_data.iter())
497                                .map(|(&avg, &input)| avg + (input - avg) / self.count as f32)
498                                .collect();
499                            self.vec_avg = Value::vector(new_avg);
500                        }
501                    }
502                }
503            }
504            AggregateFunction::Product => {
505                // Product accumulation - multiply values
506                if !value.is_null() && !value.is_empty() {
507                    if self.sum == Value::Int(0) {
508                        self.sum = value.clone();
509                    } else {
510                        self.sum = match (&self.sum, value) {
511                            (Value::SmallInt(a), Value::SmallInt(b)) => Value::SmallInt(a * b),
512                            (Value::SmallInt(a), Value::Int(b)) => Value::Int(*a as i32 * b),
513                            (Value::Int(a), Value::Int(b)) => Value::Int(a * b),
514                            (Value::Int(a), Value::SmallInt(b)) => Value::Int(a * *b as i32),
515                            (Value::BigInt(a), Value::BigInt(b)) => Value::BigInt(a * b),
516                            (Value::Float(a), Value::Float(b)) => Value::Float(a * b),
517                            (Value::Double(a), Value::Double(b)) => Value::Double(a * b),
518                            _ => Value::Null(NullType::BadData),
519                        };
520                    }
521                }
522            }
523            _ => {
524                // General handling of other aggregate functions
525                // Update the minimum value
526                if self.min.as_ref().is_none_or(|min_val| value < min_val) {
527                    self.min = Some(value.clone());
528                }
529
530                // Update the maximum value
531                if self.max.as_ref().is_none_or(|max_val| value > max_val) {
532                    self.max = Some(value.clone());
533                }
534
535                // Update Total
536                match (&mut self.sum, value) {
537                    (Value::SmallInt(ref mut sum_int), Value::SmallInt(val_int)) => {
538                        *sum_int += *val_int;
539                    }
540                    (Value::Int(ref mut sum_int), Value::Int(val_int)) => {
541                        *sum_int += *val_int;
542                    }
543                    (Value::BigInt(ref mut sum_int), Value::BigInt(val_int)) => {
544                        *sum_int += *val_int;
545                    }
546                    (Value::Float(ref mut sum_float), Value::Float(val_float)) => {
547                        *sum_float += *val_float;
548                    }
549                    (Value::Double(ref mut sum_float), Value::Double(val_float)) => {
550                        *sum_float += *val_float;
551                    }
552                    (Value::SmallInt(ref mut sum_int), Value::Int(val_int)) => {
553                        self.sum = Value::Int(*sum_int as i32 + *val_int);
554                    }
555                    (Value::Int(ref mut sum_int), Value::SmallInt(val_int)) => {
556                        *sum_int += *val_int as i32;
557                    }
558                    (Value::Int(ref mut sum_int), Value::BigInt(val_int)) => {
559                        self.sum = Value::BigInt(*sum_int as i64 + *val_int);
560                    }
561                    (Value::BigInt(ref mut sum_int), Value::Int(val_int)) => {
562                        *sum_int += *val_int as i64;
563                    }
564                    (Value::Float(ref mut sum_float), Value::Double(val_float)) => {
565                        self.sum = Value::Double(*sum_float as f64 + *val_float);
566                    }
567                    (Value::Double(ref mut sum_float), Value::Float(val_float)) => {
568                        *sum_float += *val_float as f64;
569                    }
570                    _ => {}
571                }
572            }
573        }
574    }
575
576    /// Calculating percentiles
577    pub fn calculate_percentile(&self, percentile: f64) -> Result<Value, ExpressionError> {
578        if self.percentile_values.is_empty() {
579            return Ok(Value::Null(linkrs_core::value::NullType::Null));
580        }
581
582        if !(0.0..=100.0).contains(&percentile) {
583            return Err(ExpressionError::function_error(
584                "Percentile must be between 0 and 100".to_string(),
585            ));
586        }
587
588        let mut sorted_values = self.percentile_values.clone();
589        sorted_values.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
590
591        let index = (percentile / 100.0) * (sorted_values.len() - 1) as f64;
592        let lower_index = index.floor() as usize;
593        let upper_index = index.ceil() as usize;
594
595        if lower_index == upper_index {
596            Ok(Value::Double(sorted_values[lower_index]))
597        } else {
598            let lower_value = sorted_values[lower_index];
599            let upper_value = sorted_values[upper_index];
600            let weight = index - lower_index as f64;
601            let interpolated = lower_value + weight * (upper_value - lower_value);
602            Ok(Value::Double(interpolated))
603        }
604    }
605
606    /// Calculate the standard deviation
607    pub fn calculate_std(&self) -> Result<Value, ExpressionError> {
608        if self.std_values.is_empty() {
609            return Ok(Value::Null(linkrs_core::value::NullType::Null));
610        }
611
612        let n = self.std_values.len() as f64;
613        let mean: f64 = self.std_values.iter().sum::<f64>() / n;
614        let variance: f64 = self
615            .std_values
616            .iter()
617            .map(|value| (value - mean).powi(2))
618            .sum::<f64>()
619            / n;
620        let std_dev = variance.sqrt();
621
622        Ok(Value::Double(std_dev))
623    }
624
625    /// Performing a bitwise AND operation
626    pub fn calculate_bit_and(&self) -> Result<Value, ExpressionError> {
627        if let Some(value) = self.bit_and_value {
628            Ok(Value::BigInt(value))
629        } else {
630            Ok(Value::Null(linkrs_core::value::NullType::Null))
631        }
632    }
633
634    /// Performing a bitwise OR operation
635    pub fn calculate_bit_or(&self) -> Result<Value, ExpressionError> {
636        if let Some(value) = self.bit_or_value {
637            Ok(Value::BigInt(value))
638        } else {
639            Ok(Value::Null(linkrs_core::value::NullType::Null))
640        }
641    }
642
643    /// Computing group joins
644    pub fn calculate_group_concat(&self) -> Result<Value, ExpressionError> {
645        if self.group_concat_values.is_empty() {
646            return Ok(Value::string(String::new()));
647        }
648
649        let result: Vec<String> = self
650            .group_concat_values
651            .iter()
652            .map(|v| format!("{}", v))
653            .collect();
654        Ok(Value::string(result.join(",")))
655    }
656
657    /// Calculate the variance
658    pub fn calculate_variance(&self) -> Result<Value, ExpressionError> {
659        if self.std_values.is_empty() {
660            return Ok(Value::Null(linkrs_core::value::NullType::Null));
661        }
662
663        let n = self.std_values.len() as f64;
664        let mean: f64 = self.std_values.iter().sum::<f64>() / n;
665        let variance: f64 = self
666            .std_values
667            .iter()
668            .map(|value| (value - mean).powi(2))
669            .sum::<f64>()
670            / n;
671
672        Ok(Value::Double(variance))
673    }
674
675    /// Calculate the median
676    pub fn calculate_median(&self) -> Result<Value, ExpressionError> {
677        if self.median_values.is_empty() {
678            return Ok(Value::Null(linkrs_core::value::NullType::Null));
679        }
680
681        let mut sorted_values = self.median_values.clone();
682        sorted_values.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
683
684        let len = sorted_values.len();
685        if len.is_multiple_of(2) {
686            let mid = len / 2;
687            Ok(Value::Double(
688                (sorted_values[mid - 1] + sorted_values[mid]) / 2.0,
689            ))
690        } else {
691            Ok(Value::Double(sorted_values[len / 2]))
692        }
693    }
694
695    /// Calculate the mode
696    pub fn calculate_mode(&self) -> Result<Value, ExpressionError> {
697        if self.mode_values.is_empty() {
698            return Ok(Value::Null(linkrs_core::value::NullType::Null));
699        }
700
701        let mut frequency_map = std::collections::HashMap::new();
702        for value in &self.mode_values {
703            let key = format!("{}", value);
704            *frequency_map.entry(key).or_insert(0usize) += 1;
705        }
706
707        let mode = frequency_map
708            .into_iter()
709            .max_by_key(|(_, count)| *count)
710            .map(|(key, _)| key);
711
712        match mode {
713            Some(mode_str) => {
714                if let Ok(int_val) = mode_str.parse::<i32>() {
715                    Ok(Value::Int(int_val))
716                } else if let Ok(float_val) = mode_str.parse::<f64>() {
717                    Ok(Value::Double(float_val))
718                } else if mode_str == "true" {
719                    Ok(Value::Bool(true))
720                } else if mode_str == "false" {
721                    Ok(Value::Bool(false))
722                } else {
723                    Ok(Value::string(mode_str))
724                }
725            }
726            None => Ok(Value::Null(linkrs_core::value::NullType::Null)),
727        }
728    }
729
730    /// Performing a logical AND operation
731    pub fn calculate_bool_and(&self) -> Result<Value, ExpressionError> {
732        if let Some(value) = self.bool_and_value {
733            Ok(Value::Bool(value))
734        } else {
735            Ok(Value::Null(linkrs_core::value::NullType::Null))
736        }
737    }
738
739    /// Performing a logical OR operation
740    pub fn calculate_bool_or(&self) -> Result<Value, ExpressionError> {
741        if let Some(value) = self.bool_or_value {
742            Ok(Value::Bool(value))
743        } else {
744            Ok(Value::Null(linkrs_core::value::NullType::Null))
745        }
746    }
747
748    /// Calculate population standard deviation (divide by n)
749    pub fn calculate_stddev_pop(&self) -> Result<Value, ExpressionError> {
750        if self.std_values.is_empty() {
751            return Ok(Value::Null(linkrs_core::value::NullType::Null));
752        }
753        let n = self.std_values.len() as f64;
754        let mean: f64 = self.std_values.iter().sum::<f64>() / n;
755        let variance: f64 = self
756            .std_values
757            .iter()
758            .map(|value| (value - mean).powi(2))
759            .sum::<f64>()
760            / n;
761        Ok(Value::Double(variance.sqrt()))
762    }
763
764    /// Calculate sample standard deviation (divide by n-1)
765    pub fn calculate_stddev_samp(&self) -> Result<Value, ExpressionError> {
766        let n = self.std_values.len() as f64;
767        if n < 2.0 {
768            return Ok(Value::Null(linkrs_core::value::NullType::Null));
769        }
770        let mean: f64 = self.std_values.iter().sum::<f64>() / n;
771        let variance: f64 = self
772            .std_values
773            .iter()
774            .map(|value| (value - mean).powi(2))
775            .sum::<f64>()
776            / (n - 1.0);
777        Ok(Value::Double(variance.sqrt()))
778    }
779
780    /// Calculate product of values
781    pub fn calculate_product(&self) -> Result<Value, ExpressionError> {
782        Ok(self.sum.clone())
783    }
784}
785
786#[cfg(test)]
787mod tests {
788    use super::*;
789
790    #[test]
791    fn test_unified_aggregate_function() {
792        // The test involves creating objects from strings.
793        let func = aggregate_from_str("COUNT").expect("from_str should succeed");
794        assert!(matches!(func, AggregateFunction::Count));
795
796        let func = aggregate_from_str("SUM").expect("from_str should succeed");
797        assert!(matches!(func, AggregateFunction::Sum));
798
799        let sum_func = aggregate_from_str_with_args("SUM", &["field".to_string()])
800            .expect("from_str_with_args should succeed");
801        assert!(sum_func.is_numeric());
802        assert!(!sum_func.is_collection());
803
804        let collect_func = aggregate_from_str_with_args("COLLECT", &["field".to_string()])
805            .expect("from_str_with_args should succeed");
806        assert!(!collect_func.is_numeric());
807        assert!(collect_func.is_collection());
808    }
809}
linkrs_query/executor/expression/functions/builtin/aggregate.rs

linkrs_query/executor/expression/functions/builtin/
aggregate.rs
