radixdb_executor/operators/
nested_loop_join.rs

1// Copyright 2026 RadixDB Contributors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Nested Loop Join Operator.
16//!
17//! This operator implements the classic nested loop join with O(N*M) complexity.
18//! It's the fallback algorithm used when:
19//! - No equality join keys exist (non-equi joins)
20//! - Join condition uses complex expressions
21//! - CROSS JOIN is requested
22//!
23//! Despite its higher complexity, it supports all join conditions and types.
24
25use crate::context::check_current_query_cancelled;
26use crate::expression::JoinFilter;
27use crate::operator::{ColumnInfo, ColumnSource, JoinProjection, Operator, RowRef};
28use radixdb_core::value::NULL_VALUE;
29use radixdb_core::CompactVec;
30use radixdb_core::{Result, Row, Value};
31use radixdb_functions::registry::global_registry;
32use radixdb_sql::ast::Expression;
33
34use super::hash_join::JoinType;
35
36/// Nested Loop Join Operator.
37///
38/// For each row in the outer (left) input, scans all rows in the inner (right)
39/// input and emits matches based on the join condition.
40pub struct NestedLoopJoinOperator {
41    // Input operators
42    left: Box<dyn Operator>,
43    right: Box<dyn Operator>,
44
45    // Join configuration
46    join_type: JoinType,
47    condition: Option<Expression>,
48
49    // Compiled filter (created in open())
50    filter: Option<JoinFilter>,
51
52    // Output schema
53    schema: Vec<ColumnInfo>,
54    left_col_count: usize,
55    right_col_count: usize,
56    projection: Option<JoinProjection>,
57
58    // Materialized right side (inner loop)
59    right_rows: Vec<Row>,
60
61    // Current state
62    current_left_row: Option<Row>,
63    current_right_idx: usize,
64    left_had_match: bool,
65
66    // Track matched right rows for RIGHT/FULL OUTER
67    right_matched: Vec<bool>,
68
69    // Phase for returning unmatched right rows
70    returning_unmatched_right: bool,
71    unmatched_right_idx: usize,
72
73    // Cached null rows for OUTER joins (avoid repeated allocation)
74    cached_null_right: Vec<Value>,
75    cached_null_left: Vec<Value>,
76
77    // State tracking
78    opened: bool,
79    left_exhausted: bool,
80}
81
82impl NestedLoopJoinOperator {
83    /// Create a new nested loop join operator.
84    ///
85    /// # Arguments
86    /// * `left` - Left (outer) input operator
87    /// * `right` - Right (inner) input operator
88    /// * `join_type` - Type of join (INNER, LEFT, RIGHT, FULL, CROSS)
89    /// * `condition` - Join condition (None for CROSS JOIN)
90    pub fn new(
91        left: Box<dyn Operator>,
92        right: Box<dyn Operator>,
93        join_type: JoinType,
94        condition: Option<Expression>,
95    ) -> Self {
96        // Build combined schema
97        let mut schema = Vec::new();
98        schema.extend(left.schema().iter().cloned());
99        schema.extend(right.schema().iter().cloned());
100
101        let left_col_count = left.schema().len();
102        let right_col_count = right.schema().len();
103
104        Self {
105            left,
106            right,
107            join_type,
108            condition,
109            filter: None,
110            schema,
111            left_col_count,
112            right_col_count,
113            projection: None,
114            right_rows: Vec::new(),
115            current_left_row: None,
116            current_right_idx: 0,
117            left_had_match: false,
118            right_matched: Vec::new(),
119            returning_unmatched_right: false,
120            unmatched_right_idx: 0,
121            cached_null_right: Vec::new(), // Initialized in open()
122            cached_null_left: Vec::new(),  // Initialized in open()
123            opened: false,
124            left_exhausted: false,
125        }
126    }
127
128    /// Set fused output projection.
129    ///
130    /// Nested-loop still evaluates the join predicate against the full logical
131    /// left/right rows. After that predicate boundary, projected output can be
132    /// built directly, and unmatched OUTER rows can use sparse NULLs only for
133    /// requested columns.
134    pub fn with_projection(
135        mut self,
136        columns: Vec<ColumnSource>,
137        projected_schema: Vec<ColumnInfo>,
138    ) -> Self {
139        self.projection = Some(JoinProjection { columns });
140        self.schema = projected_schema;
141        self
142    }
143
144    /// Create a NULL row for the left side (uses cached values).
145    #[inline]
146    fn null_left_row(&self) -> Row {
147        Row::from_values(self.cached_null_left.clone())
148    }
149
150    /// Create a NULL row for the right side (uses cached values).
151    #[inline]
152    fn null_right_row(&self) -> Row {
153        Row::from_values(self.cached_null_right.clone())
154    }
155
156    #[inline]
157    fn project_rows(&self, left: Option<&Row>, right: Option<&Row>) -> Row {
158        let Some(proj) = self.projection.as_ref() else {
159            match (left, right) {
160                (Some(left), Some(right)) => return Row::from_combined(left, right),
161                _ => {
162                    return Row::from_values(Vec::new());
163                }
164            }
165        };
166
167        let mut values = CompactVec::with_capacity(proj.columns.len());
168
169        for col_source in &proj.columns {
170            match col_source {
171                ColumnSource::Outer(idx) => match left {
172                    Some(row) => {
173                        // Indices are produced by compute_join_projection()
174                        // against the logical left schema before construction.
175                        values.push(row.as_slice()[*idx].clone());
176                    }
177                    None => values.push(NULL_VALUE),
178                },
179                ColumnSource::Inner(idx) => match right {
180                    Some(row) => {
181                        // Indices are produced by compute_join_projection()
182                        // against the logical right schema before construction.
183                        values.push(row.as_slice()[*idx].clone());
184                    }
185                    None => values.push(NULL_VALUE),
186                },
187            }
188        }
189
190        Row::from_compact_vec(values)
191    }
192
193    /// Combine left and right rows into output row.
194    #[inline]
195    fn combine(&self, left: &Row, right: &Row) -> Row {
196        if self.projection.is_some() {
197            return self.project_rows(Some(left), Some(right));
198        }
199        Row::from_combined(left, right)
200    }
201
202    /// Get the next left row from the outer input.
203    #[inline]
204    fn advance_left(&mut self) -> Result<bool> {
205        match self.left.next()? {
206            Some(row_ref) => {
207                self.current_left_row = Some(row_ref.into_owned());
208                self.current_right_idx = 0;
209                self.left_had_match = false;
210                Ok(true)
211            }
212            None => {
213                self.left_exhausted = true;
214                Ok(false)
215            }
216        }
217    }
218}
219
220impl Operator for NestedLoopJoinOperator {
221    fn open(&mut self) -> Result<()> {
222        if let Some(projection) = &self.projection {
223            projection.validate(self.left_col_count, self.right_col_count, self.schema.len())?;
224        }
225        // Open both inputs
226        if let Err(error) = self.left.open() {
227            let _ = self.left.close();
228            return Err(error);
229        }
230        if let Err(error) = self.right.open() {
231            let _ = self.right.close();
232            let _ = self.left.close();
233            return Err(error);
234        }
235
236        let open_result = (|| {
237            check_current_query_cancelled()?;
238
239            // Pre-cache null rows for OUTER joins (avoids repeated allocation)
240            // NULL_VALUE is a static constant, cloning Vec is just memcpy
241            if matches!(
242                self.join_type,
243                JoinType::Left | JoinType::Right | JoinType::Full
244            ) {
245                self.cached_null_right = vec![NULL_VALUE; self.right_col_count];
246                self.cached_null_left = vec![NULL_VALUE; self.left_col_count];
247            }
248
249            // Build column names for filter compilation
250            let left_cols: Vec<String> =
251                self.left.schema().iter().map(|c| c.name.clone()).collect();
252            let right_cols: Vec<String> =
253                self.right.schema().iter().map(|c| c.name.clone()).collect();
254
255            // Compile join filter if condition exists
256            if let Some(ref cond) = self.condition {
257                self.filter = Some(JoinFilter::new(
258                    cond,
259                    &left_cols,
260                    &right_cols,
261                    global_registry(),
262                )?);
263            }
264
265            // Materialize right side (inner loop must be restarted for each left row)
266            while let Some(row_ref) = self.right.next()? {
267                if self.right_rows.len() & 0xff == 0 {
268                    check_current_query_cancelled()?;
269                }
270                self.right_rows.push(row_ref.into_owned());
271            }
272
273            // Initialize matched tracking for RIGHT/FULL OUTER
274            if matches!(self.join_type, JoinType::Right | JoinType::Full) {
275                self.right_matched = vec![false; self.right_rows.len()];
276            }
277
278            // Get first left row
279            self.advance_left()?;
280
281            self.opened = true;
282            Ok(())
283        })();
284        if let Err(error) = open_result {
285            let _ = self.close();
286            return Err(error);
287        }
288        Ok(())
289    }
290
291    fn next(&mut self) -> Result<Option<RowRef>> {
292        check_current_query_cancelled()?;
293        if !self.opened {
294            return Err(radixdb_core::Error::internal(
295                "NestedLoopJoinOperator::next called before open",
296            ));
297        }
298
299        let is_left_outer = matches!(self.join_type, JoinType::Left | JoinType::Full);
300        let is_right_outer = matches!(self.join_type, JoinType::Right | JoinType::Full);
301        let is_cross = matches!(self.join_type, JoinType::Cross);
302
303        // Phase 2: Return unmatched right rows (for RIGHT/FULL OUTER)
304        if self.returning_unmatched_right {
305            while self.unmatched_right_idx < self.right_rows.len() {
306                if self.unmatched_right_idx & 0xff == 0 {
307                    check_current_query_cancelled()?;
308                }
309                let idx = self.unmatched_right_idx;
310                self.unmatched_right_idx += 1;
311
312                if !self.right_matched[idx] {
313                    let right_row = &self.right_rows[idx];
314                    let combined = if self.projection.is_some() {
315                        self.project_rows(None, Some(right_row))
316                    } else {
317                        let null_left = self.null_left_row();
318                        self.combine(&null_left, right_row)
319                    };
320                    return Ok(Some(RowRef::Owned(combined)));
321                }
322            }
323            return Ok(None);
324        }
325
326        // Phase 1: Nested loop join
327        loop {
328            // Check if left is exhausted
329            if self.left_exhausted {
330                // Switch to returning unmatched right rows if needed
331                if is_right_outer {
332                    self.returning_unmatched_right = true;
333                    self.unmatched_right_idx = 0;
334                    return self.next();
335                }
336                return Ok(None);
337            }
338
339            let left_row = match &self.current_left_row {
340                Some(row) => row,
341                None => {
342                    // Try to get next left row
343                    if !self.advance_left()? {
344                        if is_right_outer {
345                            self.returning_unmatched_right = true;
346                            self.unmatched_right_idx = 0;
347                            return self.next();
348                        }
349                        return Ok(None);
350                    }
351                    self.current_left_row.as_ref().unwrap()
352                }
353            };
354
355            // Try to find a match in right rows
356            while self.current_right_idx < self.right_rows.len() {
357                if self.current_right_idx & 0xff == 0 {
358                    check_current_query_cancelled()?;
359                }
360                let right_idx = self.current_right_idx;
361                self.current_right_idx += 1;
362
363                let right_row = &self.right_rows[right_idx];
364
365                // Check join condition
366                let matches = if let Some(ref filter) = self.filter {
367                    filter.matches_checked(left_row, right_row)?
368                } else {
369                    // CROSS JOIN or no condition
370                    is_cross || self.condition.is_none()
371                };
372
373                if matches {
374                    self.left_had_match = true;
375
376                    // Mark right row as matched for OUTER joins
377                    if is_right_outer {
378                        self.right_matched[right_idx] = true;
379                    }
380
381                    let combined = self.combine(left_row, right_row);
382                    return Ok(Some(RowRef::Owned(combined)));
383                }
384            }
385
386            // Exhausted right side for current left row
387            // Handle LEFT/FULL OUTER: emit left row with NULLs if no match
388            if is_left_outer && !self.left_had_match {
389                let left_row = self.current_left_row.take().unwrap();
390                self.advance_left()?;
391                let combined = if self.projection.is_some() {
392                    self.project_rows(Some(&left_row), None)
393                } else {
394                    let null_right = self.null_right_row();
395                    // Use owned variant - both rows are owned and won't be used again
396                    Row::from_combined_owned(left_row, null_right)
397                };
398                return Ok(Some(RowRef::Owned(combined)));
399            }
400
401            // Move to next left row
402            if !self.advance_left()? {
403                // Left exhausted - handle unmatched right rows
404                if is_right_outer {
405                    self.returning_unmatched_right = true;
406                    self.unmatched_right_idx = 0;
407                    return self.next();
408                }
409                return Ok(None);
410            }
411        }
412    }
413
414    fn close(&mut self) -> Result<()> {
415        let left = self.left.close();
416        let right = self.right.close();
417        left.and(right)
418    }
419
420    fn schema(&self) -> &[ColumnInfo] {
421        &self.schema
422    }
423
424    fn estimated_rows(&self) -> Option<usize> {
425        let left_est = self.left.estimated_rows()?;
426        let right_est = self.right.estimated_rows()?;
427
428        Some(match self.join_type {
429            JoinType::Inner => (left_est * right_est) / 10, // Assume 10% selectivity
430            JoinType::Left => left_est,
431            JoinType::Right => right_est,
432            JoinType::Full => left_est + right_est,
433            JoinType::Cross => left_est * right_est,
434            JoinType::Semi => left_est.min(right_est),
435            JoinType::Anti => left_est,
436        })
437    }
438
439    fn name(&self) -> &str {
440        match self.join_type {
441            JoinType::Inner => "NestedLoop (INNER)",
442            JoinType::Left => "NestedLoop (LEFT)",
443            JoinType::Right => "NestedLoop (RIGHT)",
444            JoinType::Full => "NestedLoop (FULL)",
445            JoinType::Cross => "NestedLoop (CROSS)",
446            JoinType::Semi => "NestedLoop (SEMI)",
447            JoinType::Anti => "NestedLoop (ANTI)",
448        }
449    }
450}
451
452#[cfg(test)]
453mod tests {
454    use super::*;
455    use crate::operator::MaterializedOperator;
456    use radixdb_sql::ast::{Identifier, InfixExpression};
457    use radixdb_sql::token::{Position, Token, TokenType};
458
459    fn make_rows(data: Vec<Vec<i64>>) -> Vec<Row> {
460        data.into_iter()
461            .map(|vals| Row::from_values(vals.into_iter().map(Value::integer).collect()))
462            .collect()
463    }
464
465    fn make_operator(data: Vec<Vec<i64>>, cols: Vec<&str>) -> Box<dyn Operator> {
466        let rows = make_rows(data);
467        let schema = cols.into_iter().map(ColumnInfo::new).collect();
468        Box::new(MaterializedOperator::new(rows, schema))
469    }
470
471    fn collect_results(op: &mut dyn Operator) -> Result<Vec<Row>> {
472        let mut results = Vec::new();
473        op.open()?;
474        while let Some(row_ref) = op.next()? {
475            results.push(row_ref.into_owned());
476        }
477        op.close()?;
478        Ok(results)
479    }
480
481    fn make_eq_condition(left_col: &str, right_col: &str) -> Expression {
482        Expression::Infix(InfixExpression::new(
483            Token::new(TokenType::Operator, "=", Position::default()),
484            Box::new(Expression::Identifier(Identifier::new(
485                Token::new(TokenType::Identifier, left_col, Position::default()),
486                left_col.to_string(),
487            ))),
488            "=".to_string(),
489            Box::new(Expression::Identifier(Identifier::new(
490                Token::new(TokenType::Identifier, right_col, Position::default()),
491                right_col.to_string(),
492            ))),
493        ))
494    }
495
496    #[test]
497    fn test_inner_nested_loop() {
498        let left = make_operator(
499            vec![vec![1, 10], vec![2, 20], vec![3, 30]],
500            vec!["left_id", "value"],
501        );
502        let right = make_operator(vec![vec![1, 100], vec![3, 300]], vec!["right_id", "data"]);
503
504        let condition = make_eq_condition("left_id", "right_id");
505
506        let mut join = NestedLoopJoinOperator::new(left, right, JoinType::Inner, Some(condition));
507
508        let results = collect_results(&mut join).unwrap();
509
510        // Should have 2 matches: id=1 and id=3
511        assert_eq!(results.len(), 2);
512    }
513
514    #[test]
515    fn public_nested_loop_rejects_invalid_projection_before_reading_rows() {
516        let left = make_operator(vec![vec![1]], vec!["left_id"]);
517        let right = make_operator(vec![vec![1]], vec!["right_id"]);
518        let mut join = NestedLoopJoinOperator::new(left, right, JoinType::Cross, None)
519            .with_projection(
520                vec![ColumnSource::Outer(1)],
521                vec![ColumnInfo::new("invalid")],
522            );
523
524        assert!(join.open().is_err());
525    }
526
527    #[test]
528    fn test_cross_join() {
529        let left = make_operator(vec![vec![1], vec![2]], vec!["a"]);
530        let right = make_operator(vec![vec![10], vec![20]], vec!["b"]);
531
532        let mut join = NestedLoopJoinOperator::new(left, right, JoinType::Cross, None);
533
534        let results = collect_results(&mut join).unwrap();
535
536        // 2 x 2 = 4 rows
537        assert_eq!(results.len(), 4);
538    }
539
540    #[test]
541    fn test_left_nested_loop() {
542        let left = make_operator(
543            vec![vec![1, 10], vec![2, 20], vec![3, 30]],
544            vec!["left_id", "value"],
545        );
546        let right = make_operator(vec![vec![1, 100]], vec!["right_id", "data"]);
547
548        let condition = make_eq_condition("left_id", "right_id");
549
550        let mut join = NestedLoopJoinOperator::new(left, right, JoinType::Left, Some(condition));
551
552        let results = collect_results(&mut join).unwrap();
553
554        // All 3 left rows preserved
555        assert_eq!(results.len(), 3);
556
557        // id=2 and id=3 should have NULLs
558        let row2 = results
559            .iter()
560            .find(|r| r.get(0) == Some(&Value::integer(2)))
561            .unwrap();
562        assert!(row2.get(2).unwrap().is_null());
563    }
564
565    #[test]
566    fn test_nested_loop_projection_returns_only_requested_columns() {
567        let left = make_operator(vec![vec![1, 10], vec![2, 20]], vec!["left_id", "value"]);
568        let right = make_operator(vec![vec![1, 100], vec![2, 200]], vec!["right_id", "data"]);
569
570        let condition = make_eq_condition("left_id", "right_id");
571        let mut join = NestedLoopJoinOperator::new(left, right, JoinType::Inner, Some(condition))
572            .with_projection(
573                vec![ColumnSource::Inner(1), ColumnSource::Outer(1)],
574                vec![ColumnInfo::new("data"), ColumnInfo::new("value")],
575            );
576
577        let results = collect_results(&mut join).unwrap();
578
579        assert_eq!(results.len(), 2);
580        assert_eq!(results[0].len(), 2);
581        assert_eq!(results[0].get(0), Some(&Value::integer(100)));
582        assert_eq!(results[0].get(1), Some(&Value::integer(10)));
583        assert_eq!(join.schema().len(), 2);
584    }
585
586    #[test]
587    fn test_nested_loop_left_projection_uses_sparse_null_right_side() {
588        let left = make_operator(vec![vec![1, 10], vec![2, 20]], vec!["left_id", "value"]);
589        let right = make_operator(vec![vec![1, 100]], vec!["right_id", "data"]);
590
591        let condition = make_eq_condition("left_id", "right_id");
592        let mut join = NestedLoopJoinOperator::new(left, right, JoinType::Left, Some(condition))
593            .with_projection(
594                vec![ColumnSource::Outer(1), ColumnSource::Inner(1)],
595                vec![ColumnInfo::new("value"), ColumnInfo::new("data")],
596            );
597
598        let results = collect_results(&mut join).unwrap();
599
600        assert_eq!(results.len(), 2);
601        let unmatched = results
602            .iter()
603            .find(|row| row.get(0) == Some(&Value::integer(20)))
604            .unwrap();
605        assert_eq!(unmatched.len(), 2);
606        assert!(unmatched.get(1).unwrap().is_null());
607    }
608
609    #[test]
610    fn test_right_nested_loop() {
611        let left = make_operator(vec![vec![1, 10]], vec!["left_id", "value"]);
612        let right = make_operator(
613            vec![vec![1, 100], vec![2, 200], vec![3, 300]],
614            vec!["right_id", "data"],
615        );
616
617        let condition = make_eq_condition("left_id", "right_id");
618
619        let mut join = NestedLoopJoinOperator::new(left, right, JoinType::Right, Some(condition));
620
621        let results = collect_results(&mut join).unwrap();
622
623        // All 3 right rows preserved
624        assert_eq!(results.len(), 3);
625    }
626
627    #[test]
628    fn test_nested_loop_right_projection_uses_sparse_null_left_side() {
629        let left = make_operator(vec![vec![1, 10]], vec!["left_id", "value"]);
630        let right = make_operator(vec![vec![1, 100], vec![2, 200]], vec!["right_id", "data"]);
631
632        let condition = make_eq_condition("left_id", "right_id");
633        let mut join = NestedLoopJoinOperator::new(left, right, JoinType::Right, Some(condition))
634            .with_projection(
635                vec![ColumnSource::Outer(1), ColumnSource::Inner(1)],
636                vec![ColumnInfo::new("value"), ColumnInfo::new("data")],
637            );
638
639        let results = collect_results(&mut join).unwrap();
640
641        assert_eq!(results.len(), 2);
642        let unmatched = results
643            .iter()
644            .find(|row| row.get(1) == Some(&Value::integer(200)))
645            .unwrap();
646        assert_eq!(unmatched.len(), 2);
647        assert!(unmatched.get(0).unwrap().is_null());
648    }
649}
radixdb_executor/operators/nested_loop_join.rs

radixdb_executor/operators/
nested_loop_join.rs
