1use 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
36pub struct NestedLoopJoinOperator {
41 left: Box<dyn Operator>,
43 right: Box<dyn Operator>,
44
45 join_type: JoinType,
47 condition: Option<Expression>,
48
49 filter: Option<JoinFilter>,
51
52 schema: Vec<ColumnInfo>,
54 left_col_count: usize,
55 right_col_count: usize,
56 projection: Option<JoinProjection>,
57
58 right_rows: Vec<Row>,
60
61 current_left_row: Option<Row>,
63 current_right_idx: usize,
64 left_had_match: bool,
65
66 right_matched: Vec<bool>,
68
69 returning_unmatched_right: bool,
71 unmatched_right_idx: usize,
72
73 cached_null_right: Vec<Value>,
75 cached_null_left: Vec<Value>,
76
77 opened: bool,
79 left_exhausted: bool,
80}
81
82impl NestedLoopJoinOperator {
83 pub fn new(
91 left: Box<dyn Operator>,
92 right: Box<dyn Operator>,
93 join_type: JoinType,
94 condition: Option<Expression>,
95 ) -> Self {
96 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(), cached_null_left: Vec::new(), opened: false,
124 left_exhausted: false,
125 }
126 }
127
128 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 #[inline]
146 fn null_left_row(&self) -> Row {
147 Row::from_values(self.cached_null_left.clone())
148 }
149
150 #[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 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 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 #[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 #[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 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 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 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 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 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 if matches!(self.join_type, JoinType::Right | JoinType::Full) {
275 self.right_matched = vec![false; self.right_rows.len()];
276 }
277
278 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 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 loop {
328 if self.left_exhausted {
330 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 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 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 let matches = if let Some(ref filter) = self.filter {
367 filter.matches_checked(left_row, right_row)?
368 } else {
369 is_cross || self.condition.is_none()
371 };
372
373 if matches {
374 self.left_had_match = true;
375
376 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 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 Row::from_combined_owned(left_row, null_right)
397 };
398 return Ok(Some(RowRef::Owned(combined)));
399 }
400
401 if !self.advance_left()? {
403 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, 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 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 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 assert_eq!(results.len(), 3);
556
557 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 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}