reifydb_engine/flow/compiler/operator/
join.rs

1// SPDX-License-Identifier: AGPL-3.0-or-later
2// Copyright (c) 2025 ReifyDB
3
4use reifydb_core::{
5	common::JoinType::{self, Inner, Left},
6	interface::catalog::flow::FlowNodeId,
7};
8use reifydb_rql::{
9	expression::Expression,
10	flow::node::FlowNodeType,
11	nodes::{JoinInnerNode, JoinLeftNode},
12	query::QueryPlan,
13};
14use reifydb_transaction::transaction::admin::AdminTransaction;
15use reifydb_type::Result;
16
17use crate::flow::compiler::{CompileOperator, FlowCompiler};
18
19pub(crate) struct JoinCompiler {
20	pub join_type: JoinType,
21	pub left: Box<QueryPlan>,
22	pub right: Box<QueryPlan>,
23	pub on: Vec<Expression>,
24	pub alias: Option<String>,
25}
26
27impl From<JoinInnerNode> for JoinCompiler {
28	fn from(node: JoinInnerNode) -> Self {
29		Self {
30			join_type: Inner,
31			left: node.left,
32			right: node.right,
33			on: node.on,
34			alias: node.alias.map(|f| f.text().to_string()),
35		}
36	}
37}
38
39impl From<JoinLeftNode> for JoinCompiler {
40	fn from(node: JoinLeftNode) -> Self {
41		Self {
42			join_type: Left,
43			left: node.left,
44			right: node.right,
45			on: node.on,
46			alias: node.alias.map(|f| f.text().to_string()),
47		}
48	}
49}
50
51// Extract the source name from a query plan if it's a scan node
52fn extract_source_name(plan: &QueryPlan) -> Option<String> {
53	match plan {
54		QueryPlan::TableScan(node) => Some(node.source.def().name.clone()),
55		QueryPlan::ViewScan(node) => Some(node.source.def().name.clone()),
56		QueryPlan::RingBufferScan(node) => Some(node.source.def().name.clone()),
57		QueryPlan::DictionaryScan(node) => Some(node.source.def().name.clone()),
58		// For other node types, try to recursively find the source
59		QueryPlan::Filter(node) => extract_source_name(&node.input),
60		QueryPlan::Map(node) => node.input.as_ref().and_then(|p| extract_source_name(p)),
61		QueryPlan::Take(node) => extract_source_name(&node.input),
62		_ => None,
63	}
64}
65
66// Extract the left and right column references from join conditions
67fn extract_join_keys(conditions: &[Expression]) -> (Vec<Expression>, Vec<Expression>) {
68	let mut left_keys = Vec::new();
69	let mut right_keys = Vec::new();
70
71	for condition in conditions {
72		match condition {
73			Expression::Equal(eq) => {
74				// For equality conditions, extract the left and
75				// right expressions
76				left_keys.push(*eq.left.clone());
77				right_keys.push(*eq.right.clone());
78			}
79			// For now, we only support simple equality joins
80			// More complex conditions could be added later
81			_ => {
82				// If it's not an equality, we'll add the whole
83				// condition to both sides This maintains
84				// backwards compatibility but may not work
85				// correctly
86				left_keys.push(condition.clone());
87				right_keys.push(condition.clone());
88			}
89		}
90	}
91
92	(left_keys, right_keys)
93}
94
95impl CompileOperator for JoinCompiler {
96	fn compile(self, compiler: &mut FlowCompiler, txn: &mut AdminTransaction) -> Result<FlowNodeId> {
97		// Extract source name from right plan for fallback alias
98		let source_name = extract_source_name(&self.right);
99
100		let left_node = compiler.compile_plan(txn, *self.left)?;
101		let right_node = compiler.compile_plan(txn, *self.right)?;
102
103		let (left_keys, right_keys) = extract_join_keys(&self.on);
104
105		// Use explicit alias, or fall back to extracted source name, or use "other"
106		let effective_alias = self.alias.or(source_name).or_else(|| Some("other".to_string()));
107
108		let node_id = compiler.add_node(
109			txn,
110			FlowNodeType::Join {
111				join_type: self.join_type,
112				left: left_keys,
113				right: right_keys,
114				alias: effective_alias,
115			},
116		)?;
117
118		compiler.add_edge(txn, &left_node, &node_id)?;
119		compiler.add_edge(txn, &right_node, &node_id)?;
120
121		Ok(node_id)
122	}
123}
reifydb_engine/flow/compiler/operator/join.rs

reifydb_engine/flow/compiler/operator/
join.rs
