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	pub state_cache_size: Option<usize>,
22	pub internal_state_cache_size: Option<usize>,
23}
24
25impl From<WindowNode> for WindowCompiler {
26	fn from(node: WindowNode) -> Self {
27		Self {
28			input: node.input,
29			kind: node.kind,
30			group_by: node.group_by,
31			aggregations: node.aggregations,
32			ts: node.ts,
33			lateness: node.lateness,
34			state_cache_size: node.state_cache_size,
35			internal_state_cache_size: node.internal_state_cache_size,
36		}
37	}
38}
39
40impl CompileOperator for WindowCompiler {
41	fn compile(self, compiler: &mut FlowCompiler, txn: &mut Transaction<'_>) -> Result<FlowNodeId> {
42		validate_flow_aggregations(&compiler.routines, &self.aggregations, AggregateContext::Windowed)?;
43
44		let input_node = if let Some(input) = self.input {
45			Some(compiler.compile_plan(txn, *input)?)
46		} else {
47			None
48		};
49
50		let node_id = compiler.add_node(
51			txn,
52			Window {
53				kind: self.kind,
54				group_by: self.group_by,
55				aggregations: self.aggregations,
56				ts: self.ts,
57				lateness: self.lateness,
58				state_cache_size: self.state_cache_size,
59				internal_state_cache_size: self.internal_state_cache_size,
60			},
61		)?;
62
63		if let Some(input_node) = input_node {
64			compiler.add_edge(txn, &input_node, &node_id)?;
65		}
66
67		Ok(node_id)
68	}
69}
reifydb_engine/flow/compiler/operator/window.rs

reifydb_engine/flow/compiler/operator/
window.rs
