reifydb_engine/flow/compiler/operator/
window.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_core::{common::WindowKind, interface::catalog::flow::FlowNodeId};
5use reifydb_rql::{expression::Expression, flow::node::FlowNodeType::Window, nodes::WindowNode, query::QueryPlan};
6use reifydb_transaction::transaction::Transaction;
7use reifydb_value::{Result, value::duration::Duration};
8
9use crate::flow::{
10	aggregate::AggregateContext,
11	compiler::{CompileOperator, FlowCompiler, operator::aggregate_validation::validate_flow_aggregations},
12};
13
14pub(crate) struct WindowCompiler {
15	pub input: Option<Box<QueryPlan>>,
16	pub kind: WindowKind,
17	pub group_by: Vec<Expression>,
18	pub aggregations: Vec<Expression>,
19	pub ts: Option<String>,
20	pub lateness: Option<Duration>,
21}
22
23impl From<WindowNode> for WindowCompiler {
24	fn from(node: WindowNode) -> Self {
25		Self {
26			input: node.input,
27			kind: node.kind,
28			group_by: node.group_by,
29			aggregations: node.aggregations,
30			ts: node.ts,
31			lateness: node.lateness,
32		}
33	}
34}
35
36impl CompileOperator for WindowCompiler {
37	fn compile(self, compiler: &mut FlowCompiler, txn: &mut Transaction<'_>) -> Result<FlowNodeId> {
38		validate_flow_aggregations(&compiler.routines, &self.aggregations, AggregateContext::Windowed)?;
39
40		let input_node = if let Some(input) = self.input {
41			Some(compiler.compile_plan(txn, *input)?)
42		} else {
43			None
44		};
45
46		let node_id = compiler.add_node(
47			txn,
48			Window {
49				kind: self.kind,
50				group_by: self.group_by,
51				aggregations: self.aggregations,
52				ts: self.ts,
53				lateness: self.lateness,
54			},
55		)?;
56
57		if let Some(input_node) = input_node {
58			compiler.add_edge(txn, &input_node, &node_id)?;
59		}
60
61		Ok(node_id)
62	}
63}
reifydb_engine/flow/compiler/operator/window.rs

reifydb_engine/flow/compiler/operator/
window.rs
