1use anyhow::{anyhow, Result};
4use std::collections::HashMap;
5use std::sync::Arc;
6use tracing::{debug, info};
7
8use crate::data::arithmetic_evaluator::ArithmeticEvaluator;
9use crate::data::datatable::{DataColumn, DataRow, DataTable, DataValue};
10use crate::data::value_comparisons::compare_with_op;
11use crate::sql::parser::ast::{JoinClause, JoinOperator, JoinType};
12use crate::sql::recursive_parser::SqlExpression;
13
14fn canonical_join_key(value: &DataValue, coerce_numeric: bool) -> DataValue {
43 match value {
44 DataValue::String(s) => normalize_join_text(s, coerce_numeric),
45 DataValue::InternedString(s) => normalize_join_text(s.as_str(), coerce_numeric),
46 DataValue::Float(f) => fold_whole_float(*f),
47 other => other.clone(),
48 }
49}
50
51fn normalize_join_text(s: &str, coerce_numeric: bool) -> DataValue {
54 if coerce_numeric {
55 if let Ok(i) = s.parse::<i64>() {
56 return DataValue::Integer(i);
57 }
58 if let Ok(f) = s.parse::<f64>() {
59 if f.is_finite() {
60 return fold_whole_float(f);
61 }
62 }
63 }
64 DataValue::String(s.to_string())
65}
66
67#[derive(PartialEq, Eq)]
70enum KeyKind {
71 Stringy,
72 Numeric,
73 Other,
74}
75
76fn value_kind(value: &DataValue) -> KeyKind {
77 match value {
78 DataValue::String(_) | DataValue::InternedString(_) => KeyKind::Stringy,
79 DataValue::Integer(_) | DataValue::Float(_) => KeyKind::Numeric,
80 _ => KeyKind::Other,
81 }
82}
83
84fn column_key_kind(table: &DataTable, col_idx: usize) -> Option<KeyKind> {
90 table
91 .rows
92 .iter()
93 .filter_map(|r| r.values.get(col_idx))
94 .find(|v| !matches!(v, DataValue::Null))
95 .map(value_kind)
96}
97
98fn join_key_coercion(
104 left_table: &DataTable,
105 left_col_idx: usize,
106 right_table: &DataTable,
107 right_col_idx: usize,
108) -> bool {
109 match (
110 column_key_kind(left_table, left_col_idx),
111 column_key_kind(right_table, right_col_idx),
112 ) {
113 (Some(l), Some(r)) => l != r,
114 _ => true,
115 }
116}
117
118fn fold_whole_float(f: f64) -> DataValue {
121 if f.is_finite() && f.fract() == 0.0 && f >= i64::MIN as f64 && f <= i64::MAX as f64 {
122 DataValue::Integer(f as i64)
123 } else {
124 DataValue::Float(f)
125 }
126}
127
128pub struct HashJoinExecutor {
130 case_insensitive: bool,
131}
132
133impl HashJoinExecutor {
134 pub fn new(case_insensitive: bool) -> Self {
135 Self { case_insensitive }
136 }
137
138 pub fn execute_join(
140 &self,
141 left_table: Arc<DataTable>,
142 join_clause: &JoinClause,
143 right_table: Arc<DataTable>,
144 ) -> Result<DataTable> {
145 info!(
146 "Executing {:?} JOIN: {} rows x {} rows with {} conditions",
147 join_clause.join_type,
148 left_table.row_count(),
149 right_table.row_count(),
150 join_clause.condition.conditions.len()
151 );
152
153 let mut condition_indices = Vec::new();
156 let mut all_equal = true;
157 let mut has_complex_expr = false;
158
159 for single_condition in &join_clause.condition.conditions {
160 let left_col_name = Self::extract_simple_column_name(&single_condition.left_expr);
162 let right_col_name = Self::extract_simple_column_name(&single_condition.right_expr);
163
164 if left_col_name.is_none() || right_col_name.is_none() {
165 has_complex_expr = true;
167 all_equal = false; break;
169 }
170
171 let (left_col_idx, right_col_idx) = self.resolve_join_columns(
172 &left_table,
173 &right_table,
174 &left_col_name.unwrap(),
175 &right_col_name.unwrap(),
176 )?;
177
178 if single_condition.operator != JoinOperator::Equal {
179 all_equal = false;
180 }
181
182 condition_indices.push((
183 left_col_idx,
184 right_col_idx,
185 single_condition.operator.clone(),
186 ));
187 }
188
189 let use_hash_join = all_equal && !has_complex_expr;
193
194 match join_clause.join_type {
196 JoinType::Inner => {
197 if use_hash_join && condition_indices.len() == 1 {
198 let (left_col_idx, right_col_idx, _) = condition_indices[0];
200 let left_col_name = Self::extract_simple_column_name(
201 &join_clause.condition.conditions[0].left_expr,
202 )
203 .expect("left_expr should be a simple column in hash join path");
204 let right_col_name = Self::extract_simple_column_name(
205 &join_clause.condition.conditions[0].right_expr,
206 )
207 .expect("right_expr should be a simple column in hash join path");
208 self.hash_join_inner(
209 left_table,
210 right_table,
211 left_col_idx,
212 right_col_idx,
213 &left_col_name,
214 &right_col_name,
215 &join_clause.alias,
216 )
217 } else {
218 self.nested_loop_join_inner_multi(
220 left_table,
221 right_table,
222 &join_clause.condition.conditions,
223 &join_clause.alias,
224 true, )
226 }
227 }
228 JoinType::Left => {
229 if use_hash_join && condition_indices.len() == 1 {
230 let (left_col_idx, right_col_idx, _) = condition_indices[0];
232 let left_col_name = Self::extract_simple_column_name(
233 &join_clause.condition.conditions[0].left_expr,
234 )
235 .expect("left_expr should be a simple column in hash join path");
236 let right_col_name = Self::extract_simple_column_name(
237 &join_clause.condition.conditions[0].right_expr,
238 )
239 .expect("right_expr should be a simple column in hash join path");
240 self.hash_join_left(
241 left_table,
242 right_table,
243 left_col_idx,
244 right_col_idx,
245 &left_col_name,
246 &right_col_name,
247 &join_clause.alias,
248 )
249 } else {
250 self.nested_loop_join_left_multi(
252 left_table,
253 right_table,
254 &join_clause.condition.conditions,
255 &join_clause.alias,
256 true, )
258 }
259 }
260 JoinType::Right => {
261 let swapped_indices: Vec<(usize, usize, JoinOperator)> = condition_indices
263 .into_iter()
264 .map(|(l, r, op)| (r, l, self.reverse_operator(&op)))
265 .collect();
266
267 if use_hash_join && swapped_indices.len() == 1 {
268 let (right_col_idx, left_col_idx, _) = swapped_indices[0];
270 let left_col_name = Self::extract_simple_column_name(
271 &join_clause.condition.conditions[0].left_expr,
272 )
273 .expect("left_expr should be a simple column in hash join path");
274 let right_col_name = Self::extract_simple_column_name(
275 &join_clause.condition.conditions[0].right_expr,
276 )
277 .expect("right_expr should be a simple column in hash join path");
278 self.hash_join_left(
279 right_table,
280 left_table,
281 right_col_idx,
282 left_col_idx,
283 &right_col_name,
284 &left_col_name,
285 &join_clause.alias,
286 )
287 } else {
288 self.nested_loop_join_right_multi(
297 left_table, right_table, &join_clause.condition.conditions,
300 &join_clause.alias,
301 )
302 }
303 }
304 JoinType::Cross => self.cross_join(left_table, right_table),
305 JoinType::Full => {
306 return Err(anyhow!("FULL OUTER JOIN not yet implemented"));
307 }
308 }
309 }
310
311 fn extract_simple_column_name(expr: &SqlExpression) -> Option<String> {
314 match expr {
315 SqlExpression::Column(col_ref) => {
316 if let Some(table_prefix) = &col_ref.table_prefix {
318 Some(format!("{}.{}", table_prefix, col_ref.name))
319 } else {
320 Some(col_ref.name.clone())
321 }
322 }
323 _ => None, }
325 }
326
327 fn expr_table_prefix(expr: &SqlExpression) -> Option<&str> {
330 match expr {
331 SqlExpression::Column(col) => col.table_prefix.as_deref(),
332 _ => None,
333 }
334 }
335
336 fn operand_uses_right(
348 &self,
349 expr: &SqlExpression,
350 join_alias: &Option<String>,
351 join_alias_is_right: bool,
352 default_is_right: bool,
353 ) -> bool {
354 if let (Some(prefix), Some(alias)) = (Self::expr_table_prefix(expr), join_alias.as_deref())
355 {
356 let matches_join_alias = if self.case_insensitive {
357 prefix.eq_ignore_ascii_case(alias)
358 } else {
359 prefix == alias
360 };
361 return if matches_join_alias {
364 join_alias_is_right
365 } else {
366 !join_alias_is_right
367 };
368 }
369 default_is_right
370 }
371
372 #[allow(clippy::too_many_arguments)]
377 fn eval_join_operand(
378 &self,
379 expr: &SqlExpression,
380 left_evaluator: &mut ArithmeticEvaluator,
381 right_evaluator: &mut ArithmeticEvaluator,
382 left_row_idx: usize,
383 right_row_idx: usize,
384 join_alias: &Option<String>,
385 join_alias_is_right: bool,
386 default_is_right: bool,
387 ) -> Result<DataValue> {
388 if self.operand_uses_right(expr, join_alias, join_alias_is_right, default_is_right) {
389 right_evaluator.evaluate(expr, right_row_idx)
390 } else {
391 left_evaluator.evaluate(expr, left_row_idx)
392 }
393 }
394
395 fn resolve_join_columns(
397 &self,
398 left_table: &DataTable,
399 right_table: &DataTable,
400 left_col_name: &str,
401 right_col_name: &str,
402 ) -> Result<(usize, usize)> {
403 let left_col_idx = if let Ok(idx) = self.find_column_index(left_table, left_col_name) {
405 idx
406 } else if let Ok(_idx) = self.find_column_index(right_table, left_col_name) {
407 return Err(anyhow!(
410 "Column '{}' found in right table but specified as left operand. \
411 Please rewrite the condition with columns in correct positions.",
412 left_col_name
413 ));
414 } else {
415 return Err(anyhow!(
416 "Column '{}' not found in either table",
417 left_col_name
418 ));
419 };
420
421 let right_col_idx = if let Ok(idx) = self.find_column_index(right_table, right_col_name) {
423 idx
424 } else if let Ok(_idx) = self.find_column_index(left_table, right_col_name) {
425 return Err(anyhow!(
428 "Column '{}' found in left table but specified as right operand. \
429 Please rewrite the condition with columns in correct positions.",
430 right_col_name
431 ));
432 } else {
433 return Err(anyhow!(
434 "Column '{}' not found in either table",
435 right_col_name
436 ));
437 };
438
439 Ok((left_col_idx, right_col_idx))
440 }
441
442 fn find_column_index(&self, table: &DataTable, col_name: &str) -> Result<usize> {
444 let col_name = if let Some(dot_pos) = col_name.rfind('.') {
446 &col_name[dot_pos + 1..]
447 } else {
448 col_name
449 };
450
451 debug!(
452 "Looking for column '{}' in table with columns: {:?}",
453 col_name,
454 table.column_names()
455 );
456
457 table
458 .columns
459 .iter()
460 .position(|col| {
461 if self.case_insensitive {
462 col.name.to_lowercase() == col_name.to_lowercase()
463 } else {
464 col.name == col_name
465 }
466 })
467 .ok_or_else(|| anyhow!("Column '{}' not found in table", col_name))
468 }
469
470 fn hash_join_inner(
472 &self,
473 left_table: Arc<DataTable>,
474 right_table: Arc<DataTable>,
475 left_col_idx: usize,
476 right_col_idx: usize,
477 _left_col_name: &str,
478 _right_col_name: &str,
479 join_alias: &Option<String>,
480 ) -> Result<DataTable> {
481 let start = std::time::Instant::now();
482
483 let coerce = join_key_coercion(&left_table, left_col_idx, &right_table, right_col_idx);
487
488 let (build_table, probe_table, build_col_idx, probe_col_idx, build_is_left) =
490 if left_table.row_count() <= right_table.row_count() {
491 (
492 left_table.clone(),
493 right_table.clone(),
494 left_col_idx,
495 right_col_idx,
496 true,
497 )
498 } else {
499 (
500 right_table.clone(),
501 left_table.clone(),
502 right_col_idx,
503 left_col_idx,
504 false,
505 )
506 };
507
508 debug!(
509 "Building hash index on {} table ({} rows)",
510 if build_is_left { "left" } else { "right" },
511 build_table.row_count()
512 );
513
514 let mut hash_index: HashMap<DataValue, Vec<usize>> = HashMap::new();
516 for (row_idx, row) in build_table.rows.iter().enumerate() {
517 let key = canonical_join_key(&row.values[build_col_idx], coerce);
518 hash_index.entry(key).or_default().push(row_idx);
519 }
520
521 debug!(
522 "Hash index built with {} unique keys in {:?}",
523 hash_index.len(),
524 start.elapsed()
525 );
526
527 let mut result = DataTable::new("joined");
529
530 for col in &left_table.columns {
532 result.add_column(DataColumn {
533 name: col.name.clone(),
534 data_type: col.data_type.clone(),
535 nullable: col.nullable,
536 unique_values: col.unique_values,
537 null_count: col.null_count,
538 metadata: col.metadata.clone(),
539 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), });
542 }
543
544 for col in &right_table.columns {
546 if !left_table
548 .columns
549 .iter()
550 .any(|left_col| left_col.name == col.name)
551 {
552 result.add_column(DataColumn {
553 name: col.name.clone(),
554 data_type: col.data_type.clone(),
555 nullable: col.nullable,
556 unique_values: col.unique_values,
557 null_count: col.null_count,
558 metadata: col.metadata.clone(),
559 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), });
562 } else {
563 let (column_name, qualified_name) = if let Some(alias) = join_alias {
565 (
567 format!("{}.{}", alias, col.name),
568 Some(format!("{}.{}", alias, col.name)),
569 )
570 } else {
571 (format!("{}_right", col.name), col.qualified_name.clone())
573 };
574 result.add_column(DataColumn {
575 name: column_name,
576 data_type: col.data_type.clone(),
577 nullable: col.nullable,
578 unique_values: col.unique_values,
579 null_count: col.null_count,
580 metadata: col.metadata.clone(),
581 qualified_name,
582 source_table: join_alias.clone().or_else(|| col.source_table.clone()),
583 });
584 }
585 }
586
587 debug!(
588 "Joined table will have {} columns: {:?}",
589 result.column_count(),
590 result.column_names()
591 );
592
593 let mut match_count = 0;
595 for probe_row in &probe_table.rows {
596 let probe_key = canonical_join_key(&probe_row.values[probe_col_idx], coerce);
597
598 if let Some(matching_indices) = hash_index.get(&probe_key) {
599 for &build_idx in matching_indices {
600 let build_row = &build_table.rows[build_idx];
601
602 let mut joined_row = DataRow { values: Vec::new() };
604
605 if build_is_left {
606 joined_row.values.extend_from_slice(&build_row.values);
608 joined_row.values.extend_from_slice(&probe_row.values);
609 } else {
610 joined_row.values.extend_from_slice(&probe_row.values);
612 joined_row.values.extend_from_slice(&build_row.values);
613 }
614
615 result.add_row(joined_row);
616 match_count += 1;
617 }
618 }
619 }
620
621 let qualified_cols: Vec<String> = result
623 .columns
624 .iter()
625 .filter_map(|c| c.qualified_name.clone())
626 .collect();
627
628 info!(
629 "INNER JOIN complete: {} matches found in {:?}. Result has {} columns ({} qualified: {:?})",
630 match_count,
631 start.elapsed(),
632 result.columns.len(),
633 qualified_cols.len(),
634 qualified_cols
635 );
636
637 Ok(result)
638 }
639
640 fn hash_join_left(
642 &self,
643 left_table: Arc<DataTable>,
644 right_table: Arc<DataTable>,
645 left_col_idx: usize,
646 right_col_idx: usize,
647 _left_col_name: &str,
648 _right_col_name: &str,
649 join_alias: &Option<String>,
650 ) -> Result<DataTable> {
651 let start = std::time::Instant::now();
652
653 let coerce = join_key_coercion(&left_table, left_col_idx, &right_table, right_col_idx);
655
656 debug!(
657 "Building hash index on right table ({} rows)",
658 right_table.row_count()
659 );
660
661 let mut hash_index: HashMap<DataValue, Vec<usize>> = HashMap::new();
663 for (row_idx, row) in right_table.rows.iter().enumerate() {
664 let key = canonical_join_key(&row.values[right_col_idx], coerce);
665 hash_index.entry(key).or_default().push(row_idx);
666 }
667
668 let mut result = DataTable::new("joined");
670
671 for col in &left_table.columns {
673 result.add_column(DataColumn {
674 name: col.name.clone(),
675 data_type: col.data_type.clone(),
676 nullable: col.nullable,
677 unique_values: col.unique_values,
678 null_count: col.null_count,
679 metadata: col.metadata.clone(),
680 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), });
683 }
684
685 for col in &right_table.columns {
687 if !left_table
689 .columns
690 .iter()
691 .any(|left_col| left_col.name == col.name)
692 {
693 result.add_column(DataColumn {
694 name: col.name.clone(),
695 data_type: col.data_type.clone(),
696 nullable: true, unique_values: col.unique_values,
698 null_count: col.null_count,
699 metadata: col.metadata.clone(),
700 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), });
703 } else {
704 let (column_name, qualified_name) = if let Some(alias) = join_alias {
706 (
708 format!("{}.{}", alias, col.name),
709 Some(format!("{}.{}", alias, col.name)),
710 )
711 } else {
712 (format!("{}_right", col.name), col.qualified_name.clone())
714 };
715 result.add_column(DataColumn {
716 name: column_name,
717 data_type: col.data_type.clone(),
718 nullable: true, unique_values: col.unique_values,
720 null_count: col.null_count,
721 metadata: col.metadata.clone(),
722 qualified_name,
723 source_table: join_alias.clone().or_else(|| col.source_table.clone()),
724 });
725 }
726 }
727
728 debug!(
729 "LEFT JOIN table will have {} columns: {:?}",
730 result.column_count(),
731 result.column_names()
732 );
733
734 let mut match_count = 0;
736 let mut null_count = 0;
737
738 for left_row in &left_table.rows {
739 let left_key = canonical_join_key(&left_row.values[left_col_idx], coerce);
740
741 if let Some(matching_indices) = hash_index.get(&left_key) {
742 for &right_idx in matching_indices {
744 let right_row = &right_table.rows[right_idx];
745
746 let mut joined_row = DataRow { values: Vec::new() };
747 joined_row.values.extend_from_slice(&left_row.values);
748 joined_row.values.extend_from_slice(&right_row.values);
749
750 result.add_row(joined_row);
751 match_count += 1;
752 }
753 } else {
754 let mut joined_row = DataRow { values: Vec::new() };
756 joined_row.values.extend_from_slice(&left_row.values);
757
758 for _ in 0..right_table.column_count() {
760 joined_row.values.push(DataValue::Null);
761 }
762
763 result.add_row(joined_row);
764 null_count += 1;
765 }
766 }
767
768 let qualified_cols: Vec<String> = result
770 .columns
771 .iter()
772 .filter_map(|c| c.qualified_name.clone())
773 .collect();
774
775 info!(
776 "LEFT JOIN complete: {} matches, {} nulls in {:?}. Result has {} columns ({} qualified: {:?})",
777 match_count,
778 null_count,
779 start.elapsed(),
780 result.columns.len(),
781 qualified_cols.len(),
782 qualified_cols
783 );
784
785 Ok(result)
786 }
787
788 fn cross_join(
790 &self,
791 left_table: Arc<DataTable>,
792 right_table: Arc<DataTable>,
793 ) -> Result<DataTable> {
794 let start = std::time::Instant::now();
795
796 let result_rows = left_table.row_count() * right_table.row_count();
798 if result_rows > 1_000_000 {
799 return Err(anyhow!(
800 "CROSS JOIN would produce {} rows, which exceeds the safety limit",
801 result_rows
802 ));
803 }
804
805 let mut result = DataTable::new("joined");
807
808 for col in &left_table.columns {
810 result.add_column(col.clone());
811 }
812 for col in &right_table.columns {
813 result.add_column(col.clone());
814 }
815
816 for left_row in &left_table.rows {
818 for right_row in &right_table.rows {
819 let mut joined_row = DataRow { values: Vec::new() };
820 joined_row.values.extend_from_slice(&left_row.values);
821 joined_row.values.extend_from_slice(&right_row.values);
822 result.add_row(joined_row);
823 }
824 }
825
826 info!(
827 "CROSS JOIN complete: {} rows in {:?}",
828 result.row_count(),
829 start.elapsed()
830 );
831
832 Ok(result)
833 }
834
835 fn qualify_column_name(
837 &self,
838 col_name: &str,
839 table_side: &str,
840 left_join_col: &str,
841 right_join_col: &str,
842 ) -> String {
843 let base_name = if let Some(dot_pos) = col_name.rfind('.') {
845 &col_name[dot_pos + 1..]
846 } else {
847 col_name
848 };
849
850 let left_base = if let Some(dot_pos) = left_join_col.rfind('.') {
851 &left_join_col[dot_pos + 1..]
852 } else {
853 left_join_col
854 };
855
856 let right_base = if let Some(dot_pos) = right_join_col.rfind('.') {
857 &right_join_col[dot_pos + 1..]
858 } else {
859 right_join_col
860 };
861
862 if base_name == left_base || base_name == right_base {
864 format!("{}_{}", table_side, base_name)
865 } else {
866 col_name.to_string()
867 }
868 }
869
870 fn reverse_operator(&self, op: &JoinOperator) -> JoinOperator {
872 match op {
873 JoinOperator::Equal => JoinOperator::Equal,
874 JoinOperator::NotEqual => JoinOperator::NotEqual,
875 JoinOperator::LessThan => JoinOperator::GreaterThan,
876 JoinOperator::GreaterThan => JoinOperator::LessThan,
877 JoinOperator::LessThanOrEqual => JoinOperator::GreaterThanOrEqual,
878 JoinOperator::GreaterThanOrEqual => JoinOperator::LessThanOrEqual,
879 }
880 }
881
882 fn compare_values(&self, left: &DataValue, right: &DataValue, op: &JoinOperator) -> bool {
890 let op_str = match op {
891 JoinOperator::Equal => "=",
892 JoinOperator::NotEqual => "!=",
893 JoinOperator::LessThan => "<",
894 JoinOperator::GreaterThan => ">",
895 JoinOperator::LessThanOrEqual => "<=",
896 JoinOperator::GreaterThanOrEqual => ">=",
897 };
898 compare_with_op(left, right, op_str, self.case_insensitive)
899 }
900
901 fn nested_loop_join_inner(
903 &self,
904 left_table: Arc<DataTable>,
905 right_table: Arc<DataTable>,
906 left_col_idx: usize,
907 right_col_idx: usize,
908 operator: &JoinOperator,
909 join_alias: &Option<String>,
910 ) -> Result<DataTable> {
911 let start = std::time::Instant::now();
912
913 info!(
914 "Executing nested loop INNER JOIN with {:?} operator: {} x {} rows",
915 operator,
916 left_table.row_count(),
917 right_table.row_count()
918 );
919
920 let mut result = DataTable::new("joined");
922
923 for col in &left_table.columns {
925 result.add_column(DataColumn {
926 name: col.name.clone(),
927 data_type: col.data_type.clone(),
928 nullable: col.nullable,
929 unique_values: col.unique_values,
930 null_count: col.null_count,
931 metadata: col.metadata.clone(),
932 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), });
935 }
936
937 for col in &right_table.columns {
939 if !left_table
940 .columns
941 .iter()
942 .any(|left_col| left_col.name == col.name)
943 {
944 result.add_column(DataColumn {
945 name: col.name.clone(),
946 data_type: col.data_type.clone(),
947 nullable: col.nullable,
948 unique_values: col.unique_values,
949 null_count: col.null_count,
950 metadata: col.metadata.clone(),
951 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), });
954 } else {
955 let (column_name, qualified_name) = if let Some(alias) = join_alias {
956 (
958 format!("{}.{}", alias, col.name),
959 Some(format!("{}.{}", alias, col.name)),
960 )
961 } else {
962 (format!("{}_right", col.name), col.qualified_name.clone())
964 };
965 result.add_column(DataColumn {
966 name: column_name,
967 data_type: col.data_type.clone(),
968 nullable: col.nullable,
969 unique_values: col.unique_values,
970 null_count: col.null_count,
971 metadata: col.metadata.clone(),
972 qualified_name,
973 source_table: join_alias.clone().or_else(|| col.source_table.clone()),
974 });
975 }
976 }
977
978 let mut match_count = 0;
980 for left_row in &left_table.rows {
981 let left_value = &left_row.values[left_col_idx];
982
983 for right_row in &right_table.rows {
984 let right_value = &right_row.values[right_col_idx];
985
986 if self.compare_values(left_value, right_value, operator) {
987 let mut joined_row = DataRow { values: Vec::new() };
988 joined_row.values.extend_from_slice(&left_row.values);
989 joined_row.values.extend_from_slice(&right_row.values);
990 result.add_row(joined_row);
991 match_count += 1;
992 }
993 }
994 }
995
996 info!(
997 "Nested loop INNER JOIN complete: {} matches found in {:?}",
998 match_count,
999 start.elapsed()
1000 );
1001
1002 Ok(result)
1003 }
1004
1005 fn nested_loop_join_inner_multi(
1007 &self,
1008 left_table: Arc<DataTable>,
1009 right_table: Arc<DataTable>,
1010 conditions: &[crate::sql::parser::ast::SingleJoinCondition],
1011 join_alias: &Option<String>,
1012 join_alias_is_right: bool,
1013 ) -> Result<DataTable> {
1014 let start = std::time::Instant::now();
1015
1016 info!(
1017 "Executing nested loop INNER JOIN with {} conditions: {} x {} rows",
1018 conditions.len(),
1019 left_table.row_count(),
1020 right_table.row_count()
1021 );
1022
1023 let mut result = DataTable::new("joined");
1025
1026 for col in &left_table.columns {
1028 result.add_column(DataColumn {
1029 name: col.name.clone(),
1030 data_type: col.data_type.clone(),
1031 nullable: col.nullable,
1032 unique_values: col.unique_values,
1033 null_count: col.null_count,
1034 metadata: col.metadata.clone(),
1035 qualified_name: col.qualified_name.clone(),
1036 source_table: col.source_table.clone(),
1037 });
1038 }
1039
1040 for col in &right_table.columns {
1042 if !left_table
1043 .columns
1044 .iter()
1045 .any(|left_col| left_col.name == col.name)
1046 {
1047 result.add_column(DataColumn {
1048 name: col.name.clone(),
1049 data_type: col.data_type.clone(),
1050 nullable: col.nullable,
1051 unique_values: col.unique_values,
1052 null_count: col.null_count,
1053 metadata: col.metadata.clone(),
1054 qualified_name: col.qualified_name.clone(),
1055 source_table: col.source_table.clone(),
1056 });
1057 } else {
1058 let (column_name, qualified_name) = if let Some(alias) = join_alias {
1059 (
1060 format!("{}.{}", alias, col.name),
1061 Some(format!("{}.{}", alias, col.name)),
1062 )
1063 } else {
1064 (format!("{}_right", col.name), col.qualified_name.clone())
1065 };
1066 result.add_column(DataColumn {
1067 name: column_name,
1068 data_type: col.data_type.clone(),
1069 nullable: col.nullable,
1070 unique_values: col.unique_values,
1071 null_count: col.null_count,
1072 metadata: col.metadata.clone(),
1073 qualified_name,
1074 source_table: join_alias.clone().or_else(|| col.source_table.clone()),
1075 });
1076 }
1077 }
1078
1079 let mut left_evaluator = ArithmeticEvaluator::new(&left_table);
1081 let mut right_evaluator = ArithmeticEvaluator::new(&right_table);
1082
1083 let mut match_count = 0;
1085 for (left_row_idx, left_row) in left_table.rows.iter().enumerate() {
1086 for (right_row_idx, right_row) in right_table.rows.iter().enumerate() {
1087 let mut all_conditions_met = true;
1089 for condition in conditions.iter() {
1090 let left_val = self.eval_join_operand(
1095 &condition.left_expr,
1096 &mut left_evaluator,
1097 &mut right_evaluator,
1098 left_row_idx,
1099 right_row_idx,
1100 join_alias,
1101 join_alias_is_right,
1102 false, );
1104 let left_value = match left_val {
1105 Ok(val) => val,
1106 Err(_) => {
1107 all_conditions_met = false;
1108 break;
1109 }
1110 };
1111
1112 let right_val = self.eval_join_operand(
1113 &condition.right_expr,
1114 &mut left_evaluator,
1115 &mut right_evaluator,
1116 left_row_idx,
1117 right_row_idx,
1118 join_alias,
1119 join_alias_is_right,
1120 true, );
1122 let right_value = match right_val {
1123 Ok(val) => val,
1124 Err(_) => {
1125 all_conditions_met = false;
1126 break;
1127 }
1128 };
1129
1130 if !self.compare_values(&left_value, &right_value, &condition.operator) {
1131 all_conditions_met = false;
1132 break;
1133 }
1134 }
1135
1136 if all_conditions_met {
1137 let mut joined_row = DataRow { values: Vec::new() };
1138 joined_row.values.extend_from_slice(&left_row.values);
1139 joined_row.values.extend_from_slice(&right_row.values);
1140 result.add_row(joined_row);
1141 match_count += 1;
1142 }
1143 }
1144 }
1145
1146 info!(
1147 "Nested loop INNER JOIN complete: {} matches found in {:?}",
1148 match_count,
1149 start.elapsed()
1150 );
1151
1152 Ok(result)
1153 }
1154
1155 fn nested_loop_join_left_multi(
1157 &self,
1158 left_table: Arc<DataTable>,
1159 right_table: Arc<DataTable>,
1160 conditions: &[crate::sql::parser::ast::SingleJoinCondition],
1161 join_alias: &Option<String>,
1162 join_alias_is_right: bool,
1163 ) -> Result<DataTable> {
1164 let start = std::time::Instant::now();
1165
1166 info!(
1167 "Executing nested loop LEFT JOIN with {} conditions: {} x {} rows",
1168 conditions.len(),
1169 left_table.row_count(),
1170 right_table.row_count()
1171 );
1172
1173 let mut result = DataTable::new("joined");
1175
1176 for col in &left_table.columns {
1178 result.add_column(DataColumn {
1179 name: col.name.clone(),
1180 data_type: col.data_type.clone(),
1181 nullable: col.nullable,
1182 unique_values: col.unique_values,
1183 null_count: col.null_count,
1184 metadata: col.metadata.clone(),
1185 qualified_name: col.qualified_name.clone(),
1186 source_table: col.source_table.clone(),
1187 });
1188 }
1189
1190 for col in &right_table.columns {
1192 if !left_table
1193 .columns
1194 .iter()
1195 .any(|left_col| left_col.name == col.name)
1196 {
1197 result.add_column(DataColumn {
1198 name: col.name.clone(),
1199 data_type: col.data_type.clone(),
1200 nullable: true, unique_values: col.unique_values,
1202 null_count: col.null_count,
1203 metadata: col.metadata.clone(),
1204 qualified_name: col.qualified_name.clone(),
1205 source_table: col.source_table.clone(),
1206 });
1207 } else {
1208 let (column_name, qualified_name) = if let Some(alias) = join_alias {
1209 (
1210 format!("{}.{}", alias, col.name),
1211 Some(format!("{}.{}", alias, col.name)),
1212 )
1213 } else {
1214 (format!("{}_right", col.name), col.qualified_name.clone())
1215 };
1216 result.add_column(DataColumn {
1217 name: column_name,
1218 data_type: col.data_type.clone(),
1219 nullable: true, unique_values: col.unique_values,
1221 null_count: col.null_count,
1222 metadata: col.metadata.clone(),
1223 qualified_name,
1224 source_table: join_alias.clone().or_else(|| col.source_table.clone()),
1225 });
1226 }
1227 }
1228
1229 let mut left_evaluator = ArithmeticEvaluator::new(&left_table);
1231 let mut right_evaluator = ArithmeticEvaluator::new(&right_table);
1232
1233 let mut match_count = 0;
1235 let mut null_count = 0;
1236
1237 for (left_row_idx, left_row) in left_table.rows.iter().enumerate() {
1238 let mut found_match = false;
1239
1240 for (right_row_idx, right_row) in right_table.rows.iter().enumerate() {
1241 let mut all_conditions_met = true;
1243 for condition in conditions.iter() {
1244 let left_val = self.eval_join_operand(
1247 &condition.left_expr,
1248 &mut left_evaluator,
1249 &mut right_evaluator,
1250 left_row_idx,
1251 right_row_idx,
1252 join_alias,
1253 join_alias_is_right,
1254 false, );
1256 let left_value = match left_val {
1257 Ok(val) => val,
1258 Err(_) => {
1259 all_conditions_met = false;
1260 break;
1261 }
1262 };
1263
1264 let right_val = self.eval_join_operand(
1265 &condition.right_expr,
1266 &mut left_evaluator,
1267 &mut right_evaluator,
1268 left_row_idx,
1269 right_row_idx,
1270 join_alias,
1271 join_alias_is_right,
1272 true, );
1274 let right_value = match right_val {
1275 Ok(val) => val,
1276 Err(_) => {
1277 all_conditions_met = false;
1278 break;
1279 }
1280 };
1281
1282 if !self.compare_values(&left_value, &right_value, &condition.operator) {
1283 all_conditions_met = false;
1284 break;
1285 }
1286 }
1287
1288 if all_conditions_met {
1289 let mut joined_row = DataRow { values: Vec::new() };
1290 joined_row.values.extend_from_slice(&left_row.values);
1291 joined_row.values.extend_from_slice(&right_row.values);
1292 result.add_row(joined_row);
1293 match_count += 1;
1294 found_match = true;
1295 }
1296 }
1297
1298 if !found_match {
1300 let mut joined_row = DataRow { values: Vec::new() };
1301 joined_row.values.extend_from_slice(&left_row.values);
1302 for _ in 0..right_table.column_count() {
1303 joined_row.values.push(DataValue::Null);
1304 }
1305 result.add_row(joined_row);
1306 null_count += 1;
1307 }
1308 }
1309
1310 info!(
1311 "Nested loop LEFT JOIN complete: {} matches, {} nulls in {:?}",
1312 match_count,
1313 null_count,
1314 start.elapsed()
1315 );
1316
1317 Ok(result)
1318 }
1319
1320 fn nested_loop_join_right_multi(
1339 &self,
1340 from_table: Arc<DataTable>,
1341 joined_table: Arc<DataTable>,
1342 conditions: &[crate::sql::parser::ast::SingleJoinCondition],
1343 join_alias: &Option<String>,
1344 ) -> Result<DataTable> {
1345 let start = std::time::Instant::now();
1346
1347 info!(
1348 "Executing nested loop RIGHT JOIN with {} conditions: {} x {} rows",
1349 conditions.len(),
1350 from_table.row_count(),
1351 joined_table.row_count()
1352 );
1353
1354 let mut result = DataTable::new("joined");
1356
1357 for col in &from_table.columns {
1360 result.add_column(DataColumn {
1361 name: col.name.clone(),
1362 data_type: col.data_type.clone(),
1363 nullable: true, unique_values: col.unique_values,
1365 null_count: col.null_count,
1366 metadata: col.metadata.clone(),
1367 qualified_name: col.qualified_name.clone(),
1368 source_table: col.source_table.clone(),
1369 });
1370 }
1371
1372 for col in &joined_table.columns {
1375 if !from_table
1376 .columns
1377 .iter()
1378 .any(|from_col| from_col.name == col.name)
1379 {
1380 result.add_column(DataColumn {
1381 name: col.name.clone(),
1382 data_type: col.data_type.clone(),
1383 nullable: col.nullable,
1384 unique_values: col.unique_values,
1385 null_count: col.null_count,
1386 metadata: col.metadata.clone(),
1387 qualified_name: col.qualified_name.clone(),
1388 source_table: col.source_table.clone(),
1389 });
1390 } else {
1391 let (column_name, qualified_name) = if let Some(alias) = join_alias {
1392 (
1393 format!("{}.{}", alias, col.name),
1394 Some(format!("{}.{}", alias, col.name)),
1395 )
1396 } else {
1397 (format!("{}_right", col.name), col.qualified_name.clone())
1398 };
1399 result.add_column(DataColumn {
1400 name: column_name,
1401 data_type: col.data_type.clone(),
1402 nullable: col.nullable,
1403 unique_values: col.unique_values,
1404 null_count: col.null_count,
1405 metadata: col.metadata.clone(),
1406 qualified_name,
1407 source_table: join_alias.clone().or_else(|| col.source_table.clone()),
1408 });
1409 }
1410 }
1411
1412 let mut from_evaluator = ArithmeticEvaluator::new(&from_table);
1415 let mut joined_evaluator = ArithmeticEvaluator::new(&joined_table);
1416
1417 let mut match_count = 0;
1419 let mut null_count = 0;
1420
1421 for (joined_row_idx, joined_row) in joined_table.rows.iter().enumerate() {
1422 let mut found_match = false;
1423
1424 for (from_row_idx, from_row) in from_table.rows.iter().enumerate() {
1425 let mut all_conditions_met = true;
1427 for condition in conditions.iter() {
1428 let left_val = self.eval_join_operand(
1433 &condition.left_expr,
1434 &mut from_evaluator,
1435 &mut joined_evaluator,
1436 from_row_idx,
1437 joined_row_idx,
1438 join_alias,
1439 true, false, );
1442 let left_value = match left_val {
1443 Ok(val) => val,
1444 Err(_) => {
1445 all_conditions_met = false;
1446 break;
1447 }
1448 };
1449
1450 let right_val = self.eval_join_operand(
1451 &condition.right_expr,
1452 &mut from_evaluator,
1453 &mut joined_evaluator,
1454 from_row_idx,
1455 joined_row_idx,
1456 join_alias,
1457 true, true, );
1460 let right_value = match right_val {
1461 Ok(val) => val,
1462 Err(_) => {
1463 all_conditions_met = false;
1464 break;
1465 }
1466 };
1467
1468 if !self.compare_values(&left_value, &right_value, &condition.operator) {
1469 all_conditions_met = false;
1470 break;
1471 }
1472 }
1473
1474 if all_conditions_met {
1475 let mut joined_result_row = DataRow { values: Vec::new() };
1477 joined_result_row.values.extend_from_slice(&from_row.values);
1478 joined_result_row
1479 .values
1480 .extend_from_slice(&joined_row.values);
1481 result.add_row(joined_result_row);
1482 match_count += 1;
1483 found_match = true;
1484 }
1485 }
1486
1487 if !found_match {
1490 let mut joined_result_row = DataRow { values: Vec::new() };
1491 for _ in 0..from_table.column_count() {
1492 joined_result_row.values.push(DataValue::Null);
1493 }
1494 joined_result_row
1495 .values
1496 .extend_from_slice(&joined_row.values);
1497 result.add_row(joined_result_row);
1498 null_count += 1;
1499 }
1500 }
1501
1502 info!(
1503 "Nested loop RIGHT JOIN complete: {} matches, {} nulls in {:?}",
1504 match_count,
1505 null_count,
1506 start.elapsed()
1507 );
1508
1509 Ok(result)
1510 }
1511
1512 fn nested_loop_join_left(
1514 &self,
1515 left_table: Arc<DataTable>,
1516 right_table: Arc<DataTable>,
1517 left_col_idx: usize,
1518 right_col_idx: usize,
1519 operator: &JoinOperator,
1520 join_alias: &Option<String>,
1521 ) -> Result<DataTable> {
1522 let start = std::time::Instant::now();
1523
1524 info!(
1525 "Executing nested loop LEFT JOIN with {:?} operator: {} x {} rows",
1526 operator,
1527 left_table.row_count(),
1528 right_table.row_count()
1529 );
1530
1531 let mut result = DataTable::new("joined");
1533
1534 for col in &left_table.columns {
1536 result.add_column(DataColumn {
1537 name: col.name.clone(),
1538 data_type: col.data_type.clone(),
1539 nullable: col.nullable,
1540 unique_values: col.unique_values,
1541 null_count: col.null_count,
1542 metadata: col.metadata.clone(),
1543 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), });
1546 }
1547
1548 for col in &right_table.columns {
1550 if !left_table
1551 .columns
1552 .iter()
1553 .any(|left_col| left_col.name == col.name)
1554 {
1555 result.add_column(DataColumn {
1556 name: col.name.clone(),
1557 data_type: col.data_type.clone(),
1558 nullable: true, unique_values: col.unique_values,
1560 null_count: col.null_count,
1561 metadata: col.metadata.clone(),
1562 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), });
1565 } else {
1566 let (column_name, qualified_name) = if let Some(alias) = join_alias {
1567 (
1569 format!("{}.{}", alias, col.name),
1570 Some(format!("{}.{}", alias, col.name)),
1571 )
1572 } else {
1573 (format!("{}_right", col.name), col.qualified_name.clone())
1575 };
1576 result.add_column(DataColumn {
1577 name: column_name,
1578 data_type: col.data_type.clone(),
1579 nullable: true, unique_values: col.unique_values,
1581 null_count: col.null_count,
1582 metadata: col.metadata.clone(),
1583 qualified_name,
1584 source_table: join_alias.clone().or_else(|| col.source_table.clone()),
1585 });
1586 }
1587 }
1588
1589 let mut match_count = 0;
1591 let mut null_count = 0;
1592
1593 for left_row in &left_table.rows {
1594 let left_value = &left_row.values[left_col_idx];
1595 let mut found_match = false;
1596
1597 for right_row in &right_table.rows {
1598 let right_value = &right_row.values[right_col_idx];
1599
1600 if self.compare_values(left_value, right_value, operator) {
1601 let mut joined_row = DataRow { values: Vec::new() };
1602 joined_row.values.extend_from_slice(&left_row.values);
1603 joined_row.values.extend_from_slice(&right_row.values);
1604 result.add_row(joined_row);
1605 match_count += 1;
1606 found_match = true;
1607 }
1608 }
1609
1610 if !found_match {
1612 let mut joined_row = DataRow { values: Vec::new() };
1613 joined_row.values.extend_from_slice(&left_row.values);
1614 for _ in 0..right_table.column_count() {
1615 joined_row.values.push(DataValue::Null);
1616 }
1617 result.add_row(joined_row);
1618 null_count += 1;
1619 }
1620 }
1621
1622 info!(
1623 "Nested loop LEFT JOIN complete: {} matches, {} nulls in {:?}",
1624 match_count,
1625 null_count,
1626 start.elapsed()
1627 );
1628
1629 Ok(result)
1630 }
1631}
1632
1633#[cfg(test)]
1634mod tests {
1635 use super::*;
1636 use std::sync::Arc;
1637
1638 #[test]
1639 fn numeric_string_folds_to_integer_when_coercing() {
1640 assert_eq!(
1643 canonical_join_key(&DataValue::String("220".to_string()), true),
1644 DataValue::Integer(220)
1645 );
1646 assert_eq!(
1647 canonical_join_key(&DataValue::Integer(220), true),
1648 DataValue::Integer(220)
1649 );
1650 assert_eq!(
1651 canonical_join_key(&DataValue::String("220".to_string()), true),
1652 canonical_join_key(&DataValue::Integer(220), true)
1653 );
1654 }
1655
1656 #[test]
1657 fn numeric_strings_stay_distinct_when_not_coercing() {
1658 assert_eq!(
1661 canonical_join_key(&DataValue::String("007".to_string()), false),
1662 DataValue::String("007".to_string())
1663 );
1664 assert_ne!(
1665 canonical_join_key(&DataValue::String("007".to_string()), false),
1666 canonical_join_key(&DataValue::String("7".to_string()), false)
1667 );
1668 assert_ne!(
1670 canonical_join_key(&DataValue::String("7".to_string()), false),
1671 canonical_join_key(&DataValue::Integer(7), false)
1672 );
1673 }
1674
1675 #[test]
1676 fn interned_and_plain_strings_collapse_regardless_of_coercion() {
1677 for coerce in [true, false] {
1678 assert_eq!(
1679 canonical_join_key(
1680 &DataValue::InternedString(Arc::new("North".to_string())),
1681 coerce
1682 ),
1683 canonical_join_key(&DataValue::String("North".to_string()), coerce),
1684 "interned/plain strings must collapse (coerce = {coerce})"
1685 );
1686 }
1687 }
1688
1689 #[test]
1690 fn whole_float_folds_to_integer_when_coercing() {
1691 assert_eq!(
1692 canonical_join_key(&DataValue::Float(220.0), true),
1693 DataValue::Integer(220)
1694 );
1695 assert_eq!(
1696 canonical_join_key(&DataValue::String("220.0".to_string()), true),
1697 DataValue::Integer(220)
1698 );
1699 assert_eq!(
1701 canonical_join_key(&DataValue::Float(220.5), true),
1702 DataValue::Float(220.5)
1703 );
1704 assert_eq!(
1707 canonical_join_key(&DataValue::Float(220.0), false),
1708 DataValue::Integer(220)
1709 );
1710 }
1711
1712 #[test]
1713 fn non_numeric_text_is_preserved() {
1714 assert_eq!(
1715 canonical_join_key(&DataValue::String("North".to_string()), true),
1716 DataValue::String("North".to_string())
1717 );
1718 assert_eq!(
1720 canonical_join_key(&DataValue::String(" 220".to_string()), true),
1721 DataValue::String(" 220".to_string())
1722 );
1723 }
1724
1725 #[test]
1726 fn non_finite_strings_stay_strings() {
1727 assert_eq!(
1728 canonical_join_key(&DataValue::String("inf".to_string()), true),
1729 DataValue::String("inf".to_string())
1730 );
1731 assert_eq!(
1732 canonical_join_key(&DataValue::String("NaN".to_string()), true),
1733 DataValue::String("NaN".to_string())
1734 );
1735 }
1736
1737 #[test]
1738 fn null_is_unchanged() {
1739 assert_eq!(canonical_join_key(&DataValue::Null, true), DataValue::Null);
1740 }
1741
1742 #[test]
1743 fn coercion_enabled_only_for_differing_value_kinds() {
1744 let stringy = single_col_table(DataValue::String("7".to_string()));
1746 let numeric = single_col_table(DataValue::Integer(7));
1747 let stringy2 = single_col_table(DataValue::String("8".to_string()));
1748 let empty = DataTable::new("empty"); assert!(join_key_coercion(&stringy, 0, &numeric, 0));
1752 assert!(!join_key_coercion(&stringy, 0, &stringy2, 0));
1754 assert!(!join_key_coercion(&numeric, 0, &numeric, 0));
1756 assert!(join_key_coercion(&stringy, 0, &empty, 0));
1758 }
1759
1760 fn single_col_table(value: DataValue) -> DataTable {
1761 let mut t = DataTable::new("t");
1762 t.add_column(DataColumn::new("k"));
1763 let _ = t.add_row(DataRow {
1764 values: vec![value],
1765 });
1766 t
1767 }
1768}