akar_processor/processor/mapper/
map_aggregate.rs

1use super::ExecutionContext;
2use crate::physical::order_aggregate::resolve_group_by_indices;
3use crate::physical_operator::*;
4use akar_common::error::ProcessorError;
5use akar_common::vector::DataChunk;
6use akar_parser::ast::Expression;
7use akar_planner::logical_operator::LogicalOperator;
8
9pub fn map_and_execute_aggregate(
10    op: &LogicalOperator,
11    current_input: Vec<DataChunk>,
12    ctx: &mut ExecutionContext,
13) -> Result<Vec<DataChunk>, ProcessorError> {
14    match op {
15        LogicalOperator::Aggregate(a) => {
16            let funcs: Vec<akar_function::AggregateFunction> = a
17                .aggregates
18                .iter()
19                .map(|(n, args)| {
20                    // Detect COUNT(*) from Star arg: override name to get CountStar
21                    let effective_name = if n == "COUNT" && args.iter().any(|e| matches!(e, Expression::Star)) {
22                        "COUNT(*)"
23                    } else {
24                        n
25                    };
26                    crate::physical::order_aggregate::parse_aggregate_function(effective_name)
27                })
28                .collect();
29            let agg_expressions: Vec<Vec<Expression>> = a.aggregates.iter().map(|(_, args)| args.clone()).collect();
30
31            // Resolve GROUP BY expressions to actual column indices using input field_names
32            let field_names = current_input.first().map(|c| c.field_names.as_slice()).unwrap_or(&[]);
33            let group_by_cols = if a.group_by.is_empty() {
34                Vec::new()
35            } else {
36                resolve_group_by_indices(&a.group_by, field_names)
37            };
38
39            let shared_state = std::sync::Arc::new(crate::physical::order_aggregate::SharedAggregateState::new(
40                funcs,
41                group_by_cols,
42                agg_expressions,
43            ));
44
45            let agg_scan = crate::physical::order_aggregate::PhysicalAggregateScan {
46                shared_state: shared_state.clone(),
47            };
48            let agg_finalize = crate::physical::order_aggregate::PhysicalAggregateFinalize { shared_state };
49
50            // Phase 1: Scan and accumulate (returns empty chunk in sequential push-down)
51            let _ = agg_scan.execute(current_input)?;
52
53            // Phase 2: Finalize and yield grouped chunks
54            let result = agg_finalize.execute(vec![])?;
55
56            Ok(result)
57        }
58        LogicalOperator::CountRelTable(crt) => {
59            let physical = PhysicalCountRelTable {
60                table_name: crt.table_name.clone(),
61                table_id: crt.table_id,
62                table_catalog: ctx.table_catalog.clone(),
63            };
64            let result = physical.execute(vec![])?;
65            Ok(result)
66        }
67        _ => Err(format!("Not an aggregate operator: {:?}", op).into()),
68    }
69}
akar_processor/processor/mapper/map_aggregate.rs

akar_processor/processor/mapper/
map_aggregate.rs
