akar_processor/physical/order_aggregate/
aggregate.rs

1//! Auto-extracted from physical_operator.rs
2use crate::physical::order_aggregate::AggregateHashTable;
3use crate::physical::types::{OperatorResult, PhysicalOperatorExec};
4use akar_common::vector::DataChunk;
5use akar_function::AggregateFunction;
6
7// ==================== Aggregate ====================
8
9/// Helper: parse an aggregate function name string into an AggregateFunction enum.
10pub fn parse_aggregate_function(name: &str) -> AggregateFunction {
11    match name.to_uppercase().as_str() {
12        "COUNT" => AggregateFunction::Count,
13        "COUNT(*)" => AggregateFunction::CountStar,
14        "SUM" => AggregateFunction::Sum,
15        "AVG" => AggregateFunction::Avg,
16        "MIN" => AggregateFunction::Min,
17        "MAX" => AggregateFunction::Max,
18        "COLLECT" => AggregateFunction::Collect,
19        "STDDEV" => AggregateFunction::StdDev,
20        "VARIANCE" => AggregateFunction::Variance,
21        "PERCENTILE_DISC" => AggregateFunction::PercentileDisc { percentile: 0.5 },
22        "PERCENTILE_CONT" => AggregateFunction::PercentileCont { percentile: 0.5 },
23        _ => AggregateFunction::Count,
24    }
25}
26
27pub struct PhysicalAggregate {
28    pub group_by_cols: Vec<u32>,
29    pub aggregate_functions: Vec<String>,
30}
31
32impl PhysicalOperatorExec for PhysicalAggregate {
33    fn operator_type(&self) -> &str {
34        "aggregate"
35    }
36
37    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
38        let funcs: Vec<AggregateFunction> = self
39            .aggregate_functions
40            .iter()
41            .map(|name| parse_aggregate_function(name))
42            .collect();
43
44        let table = AggregateHashTable::new(funcs, self.group_by_cols.clone(), Vec::new());
45        table.aggregate(&input)
46    }
47}
akar_processor/physical/order_aggregate/aggregate.rs

akar_processor/physical/order_aggregate/
aggregate.rs
