Skip to main content

reifydb_engine/flow/compiler/operator/
aggregate.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_core::interface::catalog::flow::FlowNodeId;
5use reifydb_rql::{
6	expression::Expression, flow::node::FlowNodeType::Aggregate, nodes::AggregateNode, query::QueryPlan,
7};
8use reifydb_transaction::transaction::Transaction;
9use reifydb_value::Result;
10
11use crate::flow::{
12	aggregate::AggregateContext,
13	compiler::{CompileOperator, FlowCompiler, operator::aggregate_validation::validate_flow_aggregations},
14};
15
16pub(crate) struct AggregateCompiler {
17	pub input: Box<QueryPlan>,
18	pub by: Vec<Expression>,
19	pub map: Vec<Expression>,
20}
21
22impl From<AggregateNode> for AggregateCompiler {
23	fn from(node: AggregateNode) -> Self {
24		Self {
25			input: node.input,
26			by: node.by,
27			map: node.map,
28		}
29	}
30}
31
32impl CompileOperator for AggregateCompiler {
33	fn compile(self, compiler: &mut FlowCompiler, txn: &mut Transaction<'_>) -> Result<FlowNodeId> {
34		validate_flow_aggregations(&compiler.routines, &self.map, AggregateContext::Grouped)?;
35
36		let input_node = compiler.compile_plan(txn, *self.input)?;
37
38		let node_id = compiler.add_node(
39			txn,
40			Aggregate {
41				by: self.by,
42				map: self.map,
43			},
44		)?;
45
46		compiler.add_edge(txn, &input_node, &node_id)?;
47		Ok(node_id)
48	}
49}