ddb_core/functions/
aggregate.rs

1// Aggregate functions (these operate on multiple rows)
2use super::Value;
3use crate::error::Result;
4use rayon::prelude::*;
5use std::collections::HashSet;
6
7/// Threshold for using parallel aggregation (1000 rows)
8const PARALLEL_THRESHOLD: usize = 1000;
9
10/// COUNT(*) / COUNT(expr) - Count rows or non-null values
11pub fn count(values: &[Value], count_nulls: bool) -> Result<Value> {
12    if count_nulls {
13        Ok(Value::Integer(values.len() as i64))
14    } else {
15        let count = values.iter().filter(|v| !v.is_null()).count();
16        Ok(Value::Integer(count as i64))
17    }
18}
19
20/// COUNT(DISTINCT expr) - Count distinct non-null values
21pub fn count_distinct(values: &[Value]) -> Result<Value> {
22    let mut seen = HashSet::new();
23    let count = values
24        .iter()
25        .filter(|v| !v.is_null())
26        .filter(|v| seen.insert(v.to_string()))
27        .count();
28    Ok(Value::Integer(count as i64))
29}
30
31/// SUM(expr) - Sum of values with automatic parallelization for large datasets
32pub fn sum(values: &[Value]) -> Result<Value> {
33    if values.len() >= PARALLEL_THRESHOLD {
34        // Parallel sum for large datasets
35        let (int_sum, float_sum, has_float) = values
36            .par_iter()
37            .filter(|v| !v.is_null())
38            .fold(
39                || (0i64, 0.0f64, false),
40                |(mut i_acc, mut f_acc, mut has_f), value| {
41                    match value {
42                        Value::Integer(i) => i_acc += i,
43                        Value::Float(f) => {
44                            f_acc += f;
45                            has_f = true;
46                        }
47                        _ => {
48                            if let Ok(f) = value.as_f64() {
49                                f_acc += f;
50                                has_f = true;
51                            }
52                        }
53                    }
54                    (i_acc, f_acc, has_f)
55                },
56            )
57            .reduce(
58                || (0i64, 0.0f64, false),
59                |(i1, f1, h1), (i2, f2, h2)| (i1 + i2, f1 + f2, h1 || h2),
60            );
61
62        if has_float {
63            Ok(Value::Float(float_sum + int_sum as f64))
64        } else {
65            Ok(Value::Integer(int_sum))
66        }
67    } else {
68        // Sequential sum for small datasets
69        let mut int_sum: i64 = 0;
70        let mut float_sum: f64 = 0.0;
71        let mut has_float = false;
72
73        for value in values {
74            if value.is_null() {
75                continue;
76            }
77
78            match value {
79                Value::Integer(i) => int_sum += i,
80                Value::Float(f) => {
81                    float_sum += f;
82                    has_float = true;
83                }
84                _ => {
85                    if let Ok(f) = value.as_f64() {
86                        float_sum += f;
87                        has_float = true;
88                    }
89                }
90            }
91        }
92
93        if has_float {
94            Ok(Value::Float(float_sum + int_sum as f64))
95        } else {
96            Ok(Value::Integer(int_sum))
97        }
98    }
99}
100
101/// AVG(expr) - Average of values
102pub fn avg(values: &[Value]) -> Result<Value> {
103    let non_null: Vec<_> = values.iter().filter(|v| !v.is_null()).collect();
104
105    if non_null.is_empty() {
106        return Ok(Value::Null);
107    }
108
109    let sum_val = sum(&non_null.iter().map(|v| (*v).clone()).collect::<Vec<_>>())?;
110    let count = non_null.len() as f64;
111
112    match sum_val {
113        Value::Integer(i) => Ok(Value::Float(i as f64 / count)),
114        Value::Float(f) => Ok(Value::Float(f / count)),
115        _ => Ok(Value::Null),
116    }
117}
118
119/// MIN(expr) - Minimum value
120pub fn min(values: &[Value]) -> Result<Value> {
121    values
122        .iter()
123        .filter(|v| !v.is_null())
124        .min_by(|a, b| {
125            // Compare values
126            match (a, b) {
127                (Value::Integer(x), Value::Integer(y)) => x.cmp(y),
128                (Value::Float(x), Value::Float(y)) => {
129                    x.partial_cmp(y).unwrap_or(std::cmp::Ordering::Equal)
130                }
131                (Value::String(x), Value::String(y)) => x.cmp(y),
132                (Value::Integer(x), Value::Float(y)) => {
133                    (*x as f64).partial_cmp(y).unwrap_or(std::cmp::Ordering::Equal)
134                }
135                (Value::Float(x), Value::Integer(y)) => {
136                    x.partial_cmp(&(*y as f64)).unwrap_or(std::cmp::Ordering::Equal)
137                }
138                _ => std::cmp::Ordering::Equal,
139            }
140        })
141        .cloned()
142        .ok_or_else(|| crate::error::DdbError::FunctionError("No values for MIN".to_string()))
143}
144
145/// MAX(expr) - Maximum value
146pub fn max(values: &[Value]) -> Result<Value> {
147    values
148        .iter()
149        .filter(|v| !v.is_null())
150        .max_by(|a, b| {
151            // Compare values
152            match (a, b) {
153                (Value::Integer(x), Value::Integer(y)) => x.cmp(y),
154                (Value::Float(x), Value::Float(y)) => {
155                    x.partial_cmp(y).unwrap_or(std::cmp::Ordering::Equal)
156                }
157                (Value::String(x), Value::String(y)) => x.cmp(y),
158                (Value::Integer(x), Value::Float(y)) => {
159                    (*x as f64).partial_cmp(y).unwrap_or(std::cmp::Ordering::Equal)
160                }
161                (Value::Float(x), Value::Integer(y)) => {
162                    x.partial_cmp(&(*y as f64)).unwrap_or(std::cmp::Ordering::Equal)
163                }
164                _ => std::cmp::Ordering::Equal,
165            }
166        })
167        .cloned()
168        .ok_or_else(|| crate::error::DdbError::FunctionError("No values for MAX".to_string()))
169}
170
171/// GROUP_CONCAT(expr, separator) - Concatenate strings from group
172pub fn group_concat(values: &[Value], separator: Option<&str>) -> Result<Value> {
173    let sep = separator.unwrap_or(",");
174    let result: Vec<String> = values
175        .iter()
176        .filter(|v| !v.is_null())
177        .map(|v| v.to_string())
178        .collect();
179
180    Ok(Value::String(result.join(sep)))
181}
182
183/// STDDEV(expr) / STDDEV_POP(expr) - Standard deviation (population) with parallel support
184pub fn stddev_pop(values: &[Value]) -> Result<Value> {
185    let non_null: Vec<_> = values.iter().filter(|v| !v.is_null()).collect();
186
187    if non_null.is_empty() {
188        return Ok(Value::Null);
189    }
190
191    // Calculate mean
192    let mean = match avg(&non_null.iter().map(|v| (*v).clone()).collect::<Vec<_>>())? {
193        Value::Float(f) => f,
194        Value::Integer(i) => i as f64,
195        _ => return Ok(Value::Null),
196    };
197
198    // Calculate variance (parallelized for large datasets)
199    let variance: f64 = if non_null.len() >= PARALLEL_THRESHOLD {
200        non_null
201            .par_iter()
202            .map(|v| {
203                let val = match v {
204                    Value::Integer(i) => *i as f64,
205                    Value::Float(f) => *f,
206                    _ => v.as_f64().unwrap_or(0.0),
207                };
208                let diff = val - mean;
209                diff * diff
210            })
211            .sum::<f64>()
212            / non_null.len() as f64
213    } else {
214        non_null
215            .iter()
216            .map(|v| {
217                let val = match v {
218                    Value::Integer(i) => *i as f64,
219                    Value::Float(f) => *f,
220                    _ => v.as_f64().unwrap_or(0.0),
221                };
222                let diff = val - mean;
223                diff * diff
224            })
225            .sum::<f64>()
226            / non_null.len() as f64
227    };
228
229    Ok(Value::Float(variance.sqrt()))
230}
231
232/// VARIANCE(expr) / VAR_POP(expr) - Variance (population) with parallel support
233pub fn var_pop(values: &[Value]) -> Result<Value> {
234    let non_null: Vec<_> = values.iter().filter(|v| !v.is_null()).collect();
235
236    if non_null.is_empty() {
237        return Ok(Value::Null);
238    }
239
240    // Calculate mean
241    let mean = match avg(&non_null.iter().map(|v| (*v).clone()).collect::<Vec<_>>())? {
242        Value::Float(f) => f,
243        Value::Integer(i) => i as f64,
244        _ => return Ok(Value::Null),
245    };
246
247    // Calculate variance (parallelized for large datasets)
248    let variance: f64 = if non_null.len() >= PARALLEL_THRESHOLD {
249        non_null
250            .par_iter()
251            .map(|v| {
252                let val = match v {
253                    Value::Integer(i) => *i as f64,
254                    Value::Float(f) => *f,
255                    _ => v.as_f64().unwrap_or(0.0),
256                };
257                let diff = val - mean;
258                diff * diff
259            })
260            .sum::<f64>()
261            / non_null.len() as f64
262    } else {
263        non_null
264            .iter()
265            .map(|v| {
266                let val = match v {
267                    Value::Integer(i) => *i as f64,
268                    Value::Float(f) => *f,
269                    _ => v.as_f64().unwrap_or(0.0),
270                };
271                let diff = val - mean;
272                diff * diff
273            })
274            .sum::<f64>()
275            / non_null.len() as f64
276    };
277
278    Ok(Value::Float(variance))
279}
280
281#[cfg(test)]
282mod tests {
283    use super::*;
284
285    #[test]
286    fn test_count() {
287        let values = vec![Value::Integer(1), Value::Integer(2), Value::Null, Value::Integer(3)];
288        assert_eq!(count(&values, true).unwrap(), Value::Integer(4));
289        assert_eq!(count(&values, false).unwrap(), Value::Integer(3));
290    }
291
292    #[test]
293    fn test_sum() {
294        let values = vec![Value::Integer(1), Value::Integer(2), Value::Integer(3)];
295        assert_eq!(sum(&values).unwrap(), Value::Integer(6));
296
297        let float_values = vec![Value::Float(1.5), Value::Float(2.5)];
298        assert_eq!(sum(&float_values).unwrap(), Value::Float(4.0));
299    }
300
301    #[test]
302    fn test_avg() {
303        let values = vec![Value::Integer(2), Value::Integer(4), Value::Integer(6)];
304        assert_eq!(avg(&values).unwrap(), Value::Float(4.0));
305    }
306
307    #[test]
308    fn test_min_max() {
309        let values = vec![Value::Integer(3), Value::Integer(1), Value::Integer(2)];
310        assert_eq!(min(&values).unwrap(), Value::Integer(1));
311        assert_eq!(max(&values).unwrap(), Value::Integer(3));
312    }
313
314    #[test]
315    fn test_group_concat() {
316        let values = vec![Value::from("a"), Value::from("b"), Value::from("c")];
317        assert_eq!(group_concat(&values, None).unwrap(), Value::from("a,b,c"));
318        assert_eq!(group_concat(&values, Some("|")).unwrap(), Value::from("a|b|c"));
319    }
320}
ddb_core/functions/aggregate.rs

ddb_core/functions/
aggregate.rs
