reifydb_engine/flow/compiler/operator/
join.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_core::{
5	common::JoinType::{self, Inner, Left},
6	interface::catalog::flow::FlowNodeId,
7	row::JoinTtl,
8};
9use reifydb_rql::{
10	expression::Expression,
11	flow::node::FlowNodeType,
12	nodes::{JoinInnerNode, JoinLeftNode, JoinNaturalNode},
13	query::QueryPlan,
14};
15use reifydb_transaction::transaction::Transaction;
16use reifydb_value::Result;
17
18use crate::flow::compiler::{CompileOperator, FlowCompiler};
19
20pub(crate) struct JoinCompiler {
21	pub join_type: JoinType,
22	pub left: Box<QueryPlan>,
23	pub right: Box<QueryPlan>,
24	pub on: Vec<Expression>,
25	pub alias: Option<String>,
26	pub ttl: Option<JoinTtl>,
27	pub snapshot: bool,
28	pub natural: bool,
29	pub latest: bool,
30}
31
32impl From<JoinInnerNode> for JoinCompiler {
33	fn from(node: JoinInnerNode) -> Self {
34		Self {
35			join_type: Inner,
36			left: node.left,
37			right: node.right,
38			on: node.on,
39			alias: node.alias.map(|f| f.text().to_string()),
40			ttl: node.ttl,
41			snapshot: node.snapshot,
42			natural: false,
43			latest: node.latest,
44		}
45	}
46}
47
48impl From<JoinLeftNode> for JoinCompiler {
49	fn from(node: JoinLeftNode) -> Self {
50		Self {
51			join_type: Left,
52			left: node.left,
53			right: node.right,
54			on: node.on,
55			alias: node.alias.map(|f| f.text().to_string()),
56			ttl: node.ttl,
57			snapshot: node.snapshot,
58			natural: false,
59			latest: node.latest,
60		}
61	}
62}
63
64impl From<JoinNaturalNode> for JoinCompiler {
65	fn from(node: JoinNaturalNode) -> Self {
66		Self {
67			join_type: node.join_type,
68			left: node.left,
69			right: node.right,
70			on: Vec::new(),
71			alias: node.alias.map(|f| f.text().to_string()),
72			ttl: node.ttl,
73			snapshot: node.snapshot,
74			natural: true,
75			latest: node.latest,
76		}
77	}
78}
79
80fn extract_source_name(plan: &QueryPlan) -> Option<String> {
81	match plan {
82		QueryPlan::TableScan(node) => Some(node.source.def().name.clone()),
83		QueryPlan::ViewScan(node) => Some(node.source.def().name().to_string()),
84		QueryPlan::RingBufferScan(node) => Some(node.source.def().name.clone()),
85		QueryPlan::DictionaryScan(node) => Some(node.source.def().name.clone()),
86
87		QueryPlan::Filter(node) => extract_source_name(&node.input),
88		QueryPlan::Map(node) => node.input.as_ref().and_then(|p| extract_source_name(p)),
89		QueryPlan::Take(node) => extract_source_name(&node.input),
90		_ => None,
91	}
92}
93
94fn collect_equal_conditions(expr: &Expression, out: &mut Vec<Expression>) {
95	match expr {
96		Expression::And(and) => {
97			collect_equal_conditions(&and.left, out);
98			collect_equal_conditions(&and.right, out);
99		}
100		other => out.push(other.clone()),
101	}
102}
103
104fn extract_join_keys(conditions: &[Expression]) -> (Vec<Expression>, Vec<Expression>) {
105	let mut left_keys = Vec::new();
106	let mut right_keys = Vec::new();
107
108	let mut flat = Vec::new();
109	for condition in conditions {
110		collect_equal_conditions(condition, &mut flat);
111	}
112
113	for condition in flat {
114		match condition {
115			Expression::Equal(eq) => {
116				left_keys.push(*eq.left.clone());
117				right_keys.push(*eq.right.clone());
118			}
119			_ => {
120				left_keys.push(condition.clone());
121				right_keys.push(condition.clone());
122			}
123		}
124	}
125
126	(left_keys, right_keys)
127}
128
129impl CompileOperator for JoinCompiler {
130	fn compile(self, compiler: &mut FlowCompiler, txn: &mut Transaction<'_>) -> Result<FlowNodeId> {
131		let source_name = extract_source_name(&self.right);
132
133		let left_node = compiler.compile_plan(txn, *self.left)?;
134		let right_node = compiler.compile_plan(txn, *self.right)?;
135
136		let (left_keys, right_keys) = extract_join_keys(&self.on);
137
138		let effective_alias = self.alias.or(source_name).or_else(|| Some("other".to_string()));
139
140		let node_id = compiler.add_node(
141			txn,
142			FlowNodeType::Join {
143				join_type: self.join_type,
144				left: left_keys,
145				right: right_keys,
146				alias: effective_alias,
147				snapshot: self.snapshot,
148				natural: self.natural,
149				latest: self.latest,
150			},
151		)?;
152
153		compiler.write_operator_settings_join(txn, node_id, self.ttl)?;
154
155		compiler.add_edge(txn, &left_node, &node_id)?;
156		compiler.add_edge(txn, &right_node, &node_id)?;
157
158		Ok(node_id)
159	}
160}
reifydb_engine/flow/compiler/operator/join.rs

reifydb_engine/flow/compiler/operator/
join.rs
