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_left_multi(
293 right_table,
294 left_table,
295 &join_clause.condition.conditions,
296 &join_clause.alias,
297 false, )
299 }
300 }
301 JoinType::Cross => self.cross_join(left_table, right_table),
302 JoinType::Full => {
303 return Err(anyhow!("FULL OUTER JOIN not yet implemented"));
304 }
305 }
306 }
307
308 fn extract_simple_column_name(expr: &SqlExpression) -> Option<String> {
311 match expr {
312 SqlExpression::Column(col_ref) => {
313 if let Some(table_prefix) = &col_ref.table_prefix {
315 Some(format!("{}.{}", table_prefix, col_ref.name))
316 } else {
317 Some(col_ref.name.clone())
318 }
319 }
320 _ => None, }
322 }
323
324 fn expr_table_prefix(expr: &SqlExpression) -> Option<&str> {
327 match expr {
328 SqlExpression::Column(col) => col.table_prefix.as_deref(),
329 _ => None,
330 }
331 }
332
333 fn operand_uses_right(
345 &self,
346 expr: &SqlExpression,
347 join_alias: &Option<String>,
348 join_alias_is_right: bool,
349 default_is_right: bool,
350 ) -> bool {
351 if let (Some(prefix), Some(alias)) = (Self::expr_table_prefix(expr), join_alias.as_deref())
352 {
353 let matches_join_alias = if self.case_insensitive {
354 prefix.eq_ignore_ascii_case(alias)
355 } else {
356 prefix == alias
357 };
358 return if matches_join_alias {
361 join_alias_is_right
362 } else {
363 !join_alias_is_right
364 };
365 }
366 default_is_right
367 }
368
369 #[allow(clippy::too_many_arguments)]
374 fn eval_join_operand(
375 &self,
376 expr: &SqlExpression,
377 left_evaluator: &mut ArithmeticEvaluator,
378 right_evaluator: &mut ArithmeticEvaluator,
379 left_row_idx: usize,
380 right_row_idx: usize,
381 join_alias: &Option<String>,
382 join_alias_is_right: bool,
383 default_is_right: bool,
384 ) -> Result<DataValue> {
385 if self.operand_uses_right(expr, join_alias, join_alias_is_right, default_is_right) {
386 right_evaluator.evaluate(expr, right_row_idx)
387 } else {
388 left_evaluator.evaluate(expr, left_row_idx)
389 }
390 }
391
392 fn resolve_join_columns(
394 &self,
395 left_table: &DataTable,
396 right_table: &DataTable,
397 left_col_name: &str,
398 right_col_name: &str,
399 ) -> Result<(usize, usize)> {
400 let left_col_idx = if let Ok(idx) = self.find_column_index(left_table, left_col_name) {
402 idx
403 } else if let Ok(_idx) = self.find_column_index(right_table, left_col_name) {
404 return Err(anyhow!(
407 "Column '{}' found in right table but specified as left operand. \
408 Please rewrite the condition with columns in correct positions.",
409 left_col_name
410 ));
411 } else {
412 return Err(anyhow!(
413 "Column '{}' not found in either table",
414 left_col_name
415 ));
416 };
417
418 let right_col_idx = if let Ok(idx) = self.find_column_index(right_table, right_col_name) {
420 idx
421 } else if let Ok(_idx) = self.find_column_index(left_table, right_col_name) {
422 return Err(anyhow!(
425 "Column '{}' found in left table but specified as right operand. \
426 Please rewrite the condition with columns in correct positions.",
427 right_col_name
428 ));
429 } else {
430 return Err(anyhow!(
431 "Column '{}' not found in either table",
432 right_col_name
433 ));
434 };
435
436 Ok((left_col_idx, right_col_idx))
437 }
438
439 fn find_column_index(&self, table: &DataTable, col_name: &str) -> Result<usize> {
441 let col_name = if let Some(dot_pos) = col_name.rfind('.') {
443 &col_name[dot_pos + 1..]
444 } else {
445 col_name
446 };
447
448 debug!(
449 "Looking for column '{}' in table with columns: {:?}",
450 col_name,
451 table.column_names()
452 );
453
454 table
455 .columns
456 .iter()
457 .position(|col| {
458 if self.case_insensitive {
459 col.name.to_lowercase() == col_name.to_lowercase()
460 } else {
461 col.name == col_name
462 }
463 })
464 .ok_or_else(|| anyhow!("Column '{}' not found in table", col_name))
465 }
466
467 fn hash_join_inner(
469 &self,
470 left_table: Arc<DataTable>,
471 right_table: Arc<DataTable>,
472 left_col_idx: usize,
473 right_col_idx: usize,
474 _left_col_name: &str,
475 _right_col_name: &str,
476 join_alias: &Option<String>,
477 ) -> Result<DataTable> {
478 let start = std::time::Instant::now();
479
480 let coerce = join_key_coercion(&left_table, left_col_idx, &right_table, right_col_idx);
484
485 let (build_table, probe_table, build_col_idx, probe_col_idx, build_is_left) =
487 if left_table.row_count() <= right_table.row_count() {
488 (
489 left_table.clone(),
490 right_table.clone(),
491 left_col_idx,
492 right_col_idx,
493 true,
494 )
495 } else {
496 (
497 right_table.clone(),
498 left_table.clone(),
499 right_col_idx,
500 left_col_idx,
501 false,
502 )
503 };
504
505 debug!(
506 "Building hash index on {} table ({} rows)",
507 if build_is_left { "left" } else { "right" },
508 build_table.row_count()
509 );
510
511 let mut hash_index: HashMap<DataValue, Vec<usize>> = HashMap::new();
513 for (row_idx, row) in build_table.rows.iter().enumerate() {
514 let key = canonical_join_key(&row.values[build_col_idx], coerce);
515 hash_index.entry(key).or_default().push(row_idx);
516 }
517
518 debug!(
519 "Hash index built with {} unique keys in {:?}",
520 hash_index.len(),
521 start.elapsed()
522 );
523
524 let mut result = DataTable::new("joined");
526
527 for col in &left_table.columns {
529 result.add_column(DataColumn {
530 name: col.name.clone(),
531 data_type: col.data_type.clone(),
532 nullable: col.nullable,
533 unique_values: col.unique_values,
534 null_count: col.null_count,
535 metadata: col.metadata.clone(),
536 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), });
539 }
540
541 for col in &right_table.columns {
543 if !left_table
545 .columns
546 .iter()
547 .any(|left_col| left_col.name == col.name)
548 {
549 result.add_column(DataColumn {
550 name: col.name.clone(),
551 data_type: col.data_type.clone(),
552 nullable: col.nullable,
553 unique_values: col.unique_values,
554 null_count: col.null_count,
555 metadata: col.metadata.clone(),
556 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), });
559 } else {
560 let (column_name, qualified_name) = if let Some(alias) = join_alias {
562 (
564 format!("{}.{}", alias, col.name),
565 Some(format!("{}.{}", alias, col.name)),
566 )
567 } else {
568 (format!("{}_right", col.name), col.qualified_name.clone())
570 };
571 result.add_column(DataColumn {
572 name: column_name,
573 data_type: col.data_type.clone(),
574 nullable: col.nullable,
575 unique_values: col.unique_values,
576 null_count: col.null_count,
577 metadata: col.metadata.clone(),
578 qualified_name,
579 source_table: join_alias.clone().or_else(|| col.source_table.clone()),
580 });
581 }
582 }
583
584 debug!(
585 "Joined table will have {} columns: {:?}",
586 result.column_count(),
587 result.column_names()
588 );
589
590 let mut match_count = 0;
592 for probe_row in &probe_table.rows {
593 let probe_key = canonical_join_key(&probe_row.values[probe_col_idx], coerce);
594
595 if let Some(matching_indices) = hash_index.get(&probe_key) {
596 for &build_idx in matching_indices {
597 let build_row = &build_table.rows[build_idx];
598
599 let mut joined_row = DataRow { values: Vec::new() };
601
602 if build_is_left {
603 joined_row.values.extend_from_slice(&build_row.values);
605 joined_row.values.extend_from_slice(&probe_row.values);
606 } else {
607 joined_row.values.extend_from_slice(&probe_row.values);
609 joined_row.values.extend_from_slice(&build_row.values);
610 }
611
612 result.add_row(joined_row);
613 match_count += 1;
614 }
615 }
616 }
617
618 let qualified_cols: Vec<String> = result
620 .columns
621 .iter()
622 .filter_map(|c| c.qualified_name.clone())
623 .collect();
624
625 info!(
626 "INNER JOIN complete: {} matches found in {:?}. Result has {} columns ({} qualified: {:?})",
627 match_count,
628 start.elapsed(),
629 result.columns.len(),
630 qualified_cols.len(),
631 qualified_cols
632 );
633
634 Ok(result)
635 }
636
637 fn hash_join_left(
639 &self,
640 left_table: Arc<DataTable>,
641 right_table: Arc<DataTable>,
642 left_col_idx: usize,
643 right_col_idx: usize,
644 _left_col_name: &str,
645 _right_col_name: &str,
646 join_alias: &Option<String>,
647 ) -> Result<DataTable> {
648 let start = std::time::Instant::now();
649
650 let coerce = join_key_coercion(&left_table, left_col_idx, &right_table, right_col_idx);
652
653 debug!(
654 "Building hash index on right table ({} rows)",
655 right_table.row_count()
656 );
657
658 let mut hash_index: HashMap<DataValue, Vec<usize>> = HashMap::new();
660 for (row_idx, row) in right_table.rows.iter().enumerate() {
661 let key = canonical_join_key(&row.values[right_col_idx], coerce);
662 hash_index.entry(key).or_default().push(row_idx);
663 }
664
665 let mut result = DataTable::new("joined");
667
668 for col in &left_table.columns {
670 result.add_column(DataColumn {
671 name: col.name.clone(),
672 data_type: col.data_type.clone(),
673 nullable: col.nullable,
674 unique_values: col.unique_values,
675 null_count: col.null_count,
676 metadata: col.metadata.clone(),
677 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), });
680 }
681
682 for col in &right_table.columns {
684 if !left_table
686 .columns
687 .iter()
688 .any(|left_col| left_col.name == col.name)
689 {
690 result.add_column(DataColumn {
691 name: col.name.clone(),
692 data_type: col.data_type.clone(),
693 nullable: true, unique_values: col.unique_values,
695 null_count: col.null_count,
696 metadata: col.metadata.clone(),
697 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), });
700 } else {
701 let (column_name, qualified_name) = if let Some(alias) = join_alias {
703 (
705 format!("{}.{}", alias, col.name),
706 Some(format!("{}.{}", alias, col.name)),
707 )
708 } else {
709 (format!("{}_right", col.name), col.qualified_name.clone())
711 };
712 result.add_column(DataColumn {
713 name: column_name,
714 data_type: col.data_type.clone(),
715 nullable: true, unique_values: col.unique_values,
717 null_count: col.null_count,
718 metadata: col.metadata.clone(),
719 qualified_name,
720 source_table: join_alias.clone().or_else(|| col.source_table.clone()),
721 });
722 }
723 }
724
725 debug!(
726 "LEFT JOIN table will have {} columns: {:?}",
727 result.column_count(),
728 result.column_names()
729 );
730
731 let mut match_count = 0;
733 let mut null_count = 0;
734
735 for left_row in &left_table.rows {
736 let left_key = canonical_join_key(&left_row.values[left_col_idx], coerce);
737
738 if let Some(matching_indices) = hash_index.get(&left_key) {
739 for &right_idx in matching_indices {
741 let right_row = &right_table.rows[right_idx];
742
743 let mut joined_row = DataRow { values: Vec::new() };
744 joined_row.values.extend_from_slice(&left_row.values);
745 joined_row.values.extend_from_slice(&right_row.values);
746
747 result.add_row(joined_row);
748 match_count += 1;
749 }
750 } else {
751 let mut joined_row = DataRow { values: Vec::new() };
753 joined_row.values.extend_from_slice(&left_row.values);
754
755 for _ in 0..right_table.column_count() {
757 joined_row.values.push(DataValue::Null);
758 }
759
760 result.add_row(joined_row);
761 null_count += 1;
762 }
763 }
764
765 let qualified_cols: Vec<String> = result
767 .columns
768 .iter()
769 .filter_map(|c| c.qualified_name.clone())
770 .collect();
771
772 info!(
773 "LEFT JOIN complete: {} matches, {} nulls in {:?}. Result has {} columns ({} qualified: {:?})",
774 match_count,
775 null_count,
776 start.elapsed(),
777 result.columns.len(),
778 qualified_cols.len(),
779 qualified_cols
780 );
781
782 Ok(result)
783 }
784
785 fn cross_join(
787 &self,
788 left_table: Arc<DataTable>,
789 right_table: Arc<DataTable>,
790 ) -> Result<DataTable> {
791 let start = std::time::Instant::now();
792
793 let result_rows = left_table.row_count() * right_table.row_count();
795 if result_rows > 1_000_000 {
796 return Err(anyhow!(
797 "CROSS JOIN would produce {} rows, which exceeds the safety limit",
798 result_rows
799 ));
800 }
801
802 let mut result = DataTable::new("joined");
804
805 for col in &left_table.columns {
807 result.add_column(col.clone());
808 }
809 for col in &right_table.columns {
810 result.add_column(col.clone());
811 }
812
813 for left_row in &left_table.rows {
815 for right_row in &right_table.rows {
816 let mut joined_row = DataRow { values: Vec::new() };
817 joined_row.values.extend_from_slice(&left_row.values);
818 joined_row.values.extend_from_slice(&right_row.values);
819 result.add_row(joined_row);
820 }
821 }
822
823 info!(
824 "CROSS JOIN complete: {} rows in {:?}",
825 result.row_count(),
826 start.elapsed()
827 );
828
829 Ok(result)
830 }
831
832 fn qualify_column_name(
834 &self,
835 col_name: &str,
836 table_side: &str,
837 left_join_col: &str,
838 right_join_col: &str,
839 ) -> String {
840 let base_name = if let Some(dot_pos) = col_name.rfind('.') {
842 &col_name[dot_pos + 1..]
843 } else {
844 col_name
845 };
846
847 let left_base = if let Some(dot_pos) = left_join_col.rfind('.') {
848 &left_join_col[dot_pos + 1..]
849 } else {
850 left_join_col
851 };
852
853 let right_base = if let Some(dot_pos) = right_join_col.rfind('.') {
854 &right_join_col[dot_pos + 1..]
855 } else {
856 right_join_col
857 };
858
859 if base_name == left_base || base_name == right_base {
861 format!("{}_{}", table_side, base_name)
862 } else {
863 col_name.to_string()
864 }
865 }
866
867 fn reverse_operator(&self, op: &JoinOperator) -> JoinOperator {
869 match op {
870 JoinOperator::Equal => JoinOperator::Equal,
871 JoinOperator::NotEqual => JoinOperator::NotEqual,
872 JoinOperator::LessThan => JoinOperator::GreaterThan,
873 JoinOperator::GreaterThan => JoinOperator::LessThan,
874 JoinOperator::LessThanOrEqual => JoinOperator::GreaterThanOrEqual,
875 JoinOperator::GreaterThanOrEqual => JoinOperator::LessThanOrEqual,
876 }
877 }
878
879 fn compare_values(&self, left: &DataValue, right: &DataValue, op: &JoinOperator) -> bool {
887 let op_str = match op {
888 JoinOperator::Equal => "=",
889 JoinOperator::NotEqual => "!=",
890 JoinOperator::LessThan => "<",
891 JoinOperator::GreaterThan => ">",
892 JoinOperator::LessThanOrEqual => "<=",
893 JoinOperator::GreaterThanOrEqual => ">=",
894 };
895 compare_with_op(left, right, op_str, self.case_insensitive)
896 }
897
898 fn nested_loop_join_inner(
900 &self,
901 left_table: Arc<DataTable>,
902 right_table: Arc<DataTable>,
903 left_col_idx: usize,
904 right_col_idx: usize,
905 operator: &JoinOperator,
906 join_alias: &Option<String>,
907 ) -> Result<DataTable> {
908 let start = std::time::Instant::now();
909
910 info!(
911 "Executing nested loop INNER JOIN with {:?} operator: {} x {} rows",
912 operator,
913 left_table.row_count(),
914 right_table.row_count()
915 );
916
917 let mut result = DataTable::new("joined");
919
920 for col in &left_table.columns {
922 result.add_column(DataColumn {
923 name: col.name.clone(),
924 data_type: col.data_type.clone(),
925 nullable: col.nullable,
926 unique_values: col.unique_values,
927 null_count: col.null_count,
928 metadata: col.metadata.clone(),
929 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), });
932 }
933
934 for col in &right_table.columns {
936 if !left_table
937 .columns
938 .iter()
939 .any(|left_col| left_col.name == col.name)
940 {
941 result.add_column(DataColumn {
942 name: col.name.clone(),
943 data_type: col.data_type.clone(),
944 nullable: col.nullable,
945 unique_values: col.unique_values,
946 null_count: col.null_count,
947 metadata: col.metadata.clone(),
948 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), });
951 } else {
952 let (column_name, qualified_name) = if let Some(alias) = join_alias {
953 (
955 format!("{}.{}", alias, col.name),
956 Some(format!("{}.{}", alias, col.name)),
957 )
958 } else {
959 (format!("{}_right", col.name), col.qualified_name.clone())
961 };
962 result.add_column(DataColumn {
963 name: column_name,
964 data_type: col.data_type.clone(),
965 nullable: col.nullable,
966 unique_values: col.unique_values,
967 null_count: col.null_count,
968 metadata: col.metadata.clone(),
969 qualified_name,
970 source_table: join_alias.clone().or_else(|| col.source_table.clone()),
971 });
972 }
973 }
974
975 let mut match_count = 0;
977 for left_row in &left_table.rows {
978 let left_value = &left_row.values[left_col_idx];
979
980 for right_row in &right_table.rows {
981 let right_value = &right_row.values[right_col_idx];
982
983 if self.compare_values(left_value, right_value, operator) {
984 let mut joined_row = DataRow { values: Vec::new() };
985 joined_row.values.extend_from_slice(&left_row.values);
986 joined_row.values.extend_from_slice(&right_row.values);
987 result.add_row(joined_row);
988 match_count += 1;
989 }
990 }
991 }
992
993 info!(
994 "Nested loop INNER JOIN complete: {} matches found in {:?}",
995 match_count,
996 start.elapsed()
997 );
998
999 Ok(result)
1000 }
1001
1002 fn nested_loop_join_inner_multi(
1004 &self,
1005 left_table: Arc<DataTable>,
1006 right_table: Arc<DataTable>,
1007 conditions: &[crate::sql::parser::ast::SingleJoinCondition],
1008 join_alias: &Option<String>,
1009 join_alias_is_right: bool,
1010 ) -> Result<DataTable> {
1011 let start = std::time::Instant::now();
1012
1013 info!(
1014 "Executing nested loop INNER JOIN with {} conditions: {} x {} rows",
1015 conditions.len(),
1016 left_table.row_count(),
1017 right_table.row_count()
1018 );
1019
1020 let mut result = DataTable::new("joined");
1022
1023 for col in &left_table.columns {
1025 result.add_column(DataColumn {
1026 name: col.name.clone(),
1027 data_type: col.data_type.clone(),
1028 nullable: col.nullable,
1029 unique_values: col.unique_values,
1030 null_count: col.null_count,
1031 metadata: col.metadata.clone(),
1032 qualified_name: col.qualified_name.clone(),
1033 source_table: col.source_table.clone(),
1034 });
1035 }
1036
1037 for col in &right_table.columns {
1039 if !left_table
1040 .columns
1041 .iter()
1042 .any(|left_col| left_col.name == col.name)
1043 {
1044 result.add_column(DataColumn {
1045 name: col.name.clone(),
1046 data_type: col.data_type.clone(),
1047 nullable: col.nullable,
1048 unique_values: col.unique_values,
1049 null_count: col.null_count,
1050 metadata: col.metadata.clone(),
1051 qualified_name: col.qualified_name.clone(),
1052 source_table: col.source_table.clone(),
1053 });
1054 } else {
1055 let (column_name, qualified_name) = if let Some(alias) = join_alias {
1056 (
1057 format!("{}.{}", alias, col.name),
1058 Some(format!("{}.{}", alias, col.name)),
1059 )
1060 } else {
1061 (format!("{}_right", col.name), col.qualified_name.clone())
1062 };
1063 result.add_column(DataColumn {
1064 name: column_name,
1065 data_type: col.data_type.clone(),
1066 nullable: col.nullable,
1067 unique_values: col.unique_values,
1068 null_count: col.null_count,
1069 metadata: col.metadata.clone(),
1070 qualified_name,
1071 source_table: join_alias.clone().or_else(|| col.source_table.clone()),
1072 });
1073 }
1074 }
1075
1076 let mut left_evaluator = ArithmeticEvaluator::new(&left_table);
1078 let mut right_evaluator = ArithmeticEvaluator::new(&right_table);
1079
1080 let mut match_count = 0;
1082 for (left_row_idx, left_row) in left_table.rows.iter().enumerate() {
1083 for (right_row_idx, right_row) in right_table.rows.iter().enumerate() {
1084 let mut all_conditions_met = true;
1086 for condition in conditions.iter() {
1087 let left_val = self.eval_join_operand(
1092 &condition.left_expr,
1093 &mut left_evaluator,
1094 &mut right_evaluator,
1095 left_row_idx,
1096 right_row_idx,
1097 join_alias,
1098 join_alias_is_right,
1099 false, );
1101 let left_value = match left_val {
1102 Ok(val) => val,
1103 Err(_) => {
1104 all_conditions_met = false;
1105 break;
1106 }
1107 };
1108
1109 let right_val = self.eval_join_operand(
1110 &condition.right_expr,
1111 &mut left_evaluator,
1112 &mut right_evaluator,
1113 left_row_idx,
1114 right_row_idx,
1115 join_alias,
1116 join_alias_is_right,
1117 true, );
1119 let right_value = match right_val {
1120 Ok(val) => val,
1121 Err(_) => {
1122 all_conditions_met = false;
1123 break;
1124 }
1125 };
1126
1127 if !self.compare_values(&left_value, &right_value, &condition.operator) {
1128 all_conditions_met = false;
1129 break;
1130 }
1131 }
1132
1133 if all_conditions_met {
1134 let mut joined_row = DataRow { values: Vec::new() };
1135 joined_row.values.extend_from_slice(&left_row.values);
1136 joined_row.values.extend_from_slice(&right_row.values);
1137 result.add_row(joined_row);
1138 match_count += 1;
1139 }
1140 }
1141 }
1142
1143 info!(
1144 "Nested loop INNER JOIN complete: {} matches found in {:?}",
1145 match_count,
1146 start.elapsed()
1147 );
1148
1149 Ok(result)
1150 }
1151
1152 fn nested_loop_join_left_multi(
1154 &self,
1155 left_table: Arc<DataTable>,
1156 right_table: Arc<DataTable>,
1157 conditions: &[crate::sql::parser::ast::SingleJoinCondition],
1158 join_alias: &Option<String>,
1159 join_alias_is_right: bool,
1160 ) -> Result<DataTable> {
1161 let start = std::time::Instant::now();
1162
1163 info!(
1164 "Executing nested loop LEFT JOIN with {} conditions: {} x {} rows",
1165 conditions.len(),
1166 left_table.row_count(),
1167 right_table.row_count()
1168 );
1169
1170 let mut result = DataTable::new("joined");
1172
1173 for col in &left_table.columns {
1175 result.add_column(DataColumn {
1176 name: col.name.clone(),
1177 data_type: col.data_type.clone(),
1178 nullable: col.nullable,
1179 unique_values: col.unique_values,
1180 null_count: col.null_count,
1181 metadata: col.metadata.clone(),
1182 qualified_name: col.qualified_name.clone(),
1183 source_table: col.source_table.clone(),
1184 });
1185 }
1186
1187 for col in &right_table.columns {
1189 if !left_table
1190 .columns
1191 .iter()
1192 .any(|left_col| left_col.name == col.name)
1193 {
1194 result.add_column(DataColumn {
1195 name: col.name.clone(),
1196 data_type: col.data_type.clone(),
1197 nullable: true, unique_values: col.unique_values,
1199 null_count: col.null_count,
1200 metadata: col.metadata.clone(),
1201 qualified_name: col.qualified_name.clone(),
1202 source_table: col.source_table.clone(),
1203 });
1204 } else {
1205 let (column_name, qualified_name) = if let Some(alias) = join_alias {
1206 (
1207 format!("{}.{}", alias, col.name),
1208 Some(format!("{}.{}", alias, col.name)),
1209 )
1210 } else {
1211 (format!("{}_right", col.name), col.qualified_name.clone())
1212 };
1213 result.add_column(DataColumn {
1214 name: column_name,
1215 data_type: col.data_type.clone(),
1216 nullable: true, unique_values: col.unique_values,
1218 null_count: col.null_count,
1219 metadata: col.metadata.clone(),
1220 qualified_name,
1221 source_table: join_alias.clone().or_else(|| col.source_table.clone()),
1222 });
1223 }
1224 }
1225
1226 let mut left_evaluator = ArithmeticEvaluator::new(&left_table);
1228 let mut right_evaluator = ArithmeticEvaluator::new(&right_table);
1229
1230 let mut match_count = 0;
1232 let mut null_count = 0;
1233
1234 for (left_row_idx, left_row) in left_table.rows.iter().enumerate() {
1235 let mut found_match = false;
1236
1237 for (right_row_idx, right_row) in right_table.rows.iter().enumerate() {
1238 let mut all_conditions_met = true;
1240 for condition in conditions.iter() {
1241 let left_val = self.eval_join_operand(
1244 &condition.left_expr,
1245 &mut left_evaluator,
1246 &mut right_evaluator,
1247 left_row_idx,
1248 right_row_idx,
1249 join_alias,
1250 join_alias_is_right,
1251 false, );
1253 let left_value = match left_val {
1254 Ok(val) => val,
1255 Err(_) => {
1256 all_conditions_met = false;
1257 break;
1258 }
1259 };
1260
1261 let right_val = self.eval_join_operand(
1262 &condition.right_expr,
1263 &mut left_evaluator,
1264 &mut right_evaluator,
1265 left_row_idx,
1266 right_row_idx,
1267 join_alias,
1268 join_alias_is_right,
1269 true, );
1271 let right_value = match right_val {
1272 Ok(val) => val,
1273 Err(_) => {
1274 all_conditions_met = false;
1275 break;
1276 }
1277 };
1278
1279 if !self.compare_values(&left_value, &right_value, &condition.operator) {
1280 all_conditions_met = false;
1281 break;
1282 }
1283 }
1284
1285 if all_conditions_met {
1286 let mut joined_row = DataRow { values: Vec::new() };
1287 joined_row.values.extend_from_slice(&left_row.values);
1288 joined_row.values.extend_from_slice(&right_row.values);
1289 result.add_row(joined_row);
1290 match_count += 1;
1291 found_match = true;
1292 }
1293 }
1294
1295 if !found_match {
1297 let mut joined_row = DataRow { values: Vec::new() };
1298 joined_row.values.extend_from_slice(&left_row.values);
1299 for _ in 0..right_table.column_count() {
1300 joined_row.values.push(DataValue::Null);
1301 }
1302 result.add_row(joined_row);
1303 null_count += 1;
1304 }
1305 }
1306
1307 info!(
1308 "Nested loop LEFT JOIN complete: {} matches, {} nulls in {:?}",
1309 match_count,
1310 null_count,
1311 start.elapsed()
1312 );
1313
1314 Ok(result)
1315 }
1316
1317 fn nested_loop_join_left(
1319 &self,
1320 left_table: Arc<DataTable>,
1321 right_table: Arc<DataTable>,
1322 left_col_idx: usize,
1323 right_col_idx: usize,
1324 operator: &JoinOperator,
1325 join_alias: &Option<String>,
1326 ) -> Result<DataTable> {
1327 let start = std::time::Instant::now();
1328
1329 info!(
1330 "Executing nested loop LEFT JOIN with {:?} operator: {} x {} rows",
1331 operator,
1332 left_table.row_count(),
1333 right_table.row_count()
1334 );
1335
1336 let mut result = DataTable::new("joined");
1338
1339 for col in &left_table.columns {
1341 result.add_column(DataColumn {
1342 name: col.name.clone(),
1343 data_type: col.data_type.clone(),
1344 nullable: col.nullable,
1345 unique_values: col.unique_values,
1346 null_count: col.null_count,
1347 metadata: col.metadata.clone(),
1348 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), });
1351 }
1352
1353 for col in &right_table.columns {
1355 if !left_table
1356 .columns
1357 .iter()
1358 .any(|left_col| left_col.name == col.name)
1359 {
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(), source_table: col.source_table.clone(), });
1370 } else {
1371 let (column_name, qualified_name) = if let Some(alias) = join_alias {
1372 (
1374 format!("{}.{}", alias, col.name),
1375 Some(format!("{}.{}", alias, col.name)),
1376 )
1377 } else {
1378 (format!("{}_right", col.name), col.qualified_name.clone())
1380 };
1381 result.add_column(DataColumn {
1382 name: column_name,
1383 data_type: col.data_type.clone(),
1384 nullable: true, unique_values: col.unique_values,
1386 null_count: col.null_count,
1387 metadata: col.metadata.clone(),
1388 qualified_name,
1389 source_table: join_alias.clone().or_else(|| col.source_table.clone()),
1390 });
1391 }
1392 }
1393
1394 let mut match_count = 0;
1396 let mut null_count = 0;
1397
1398 for left_row in &left_table.rows {
1399 let left_value = &left_row.values[left_col_idx];
1400 let mut found_match = false;
1401
1402 for right_row in &right_table.rows {
1403 let right_value = &right_row.values[right_col_idx];
1404
1405 if self.compare_values(left_value, right_value, operator) {
1406 let mut joined_row = DataRow { values: Vec::new() };
1407 joined_row.values.extend_from_slice(&left_row.values);
1408 joined_row.values.extend_from_slice(&right_row.values);
1409 result.add_row(joined_row);
1410 match_count += 1;
1411 found_match = true;
1412 }
1413 }
1414
1415 if !found_match {
1417 let mut joined_row = DataRow { values: Vec::new() };
1418 joined_row.values.extend_from_slice(&left_row.values);
1419 for _ in 0..right_table.column_count() {
1420 joined_row.values.push(DataValue::Null);
1421 }
1422 result.add_row(joined_row);
1423 null_count += 1;
1424 }
1425 }
1426
1427 info!(
1428 "Nested loop LEFT JOIN complete: {} matches, {} nulls in {:?}",
1429 match_count,
1430 null_count,
1431 start.elapsed()
1432 );
1433
1434 Ok(result)
1435 }
1436}
1437
1438#[cfg(test)]
1439mod tests {
1440 use super::*;
1441 use std::sync::Arc;
1442
1443 #[test]
1444 fn numeric_string_folds_to_integer_when_coercing() {
1445 assert_eq!(
1448 canonical_join_key(&DataValue::String("220".to_string()), true),
1449 DataValue::Integer(220)
1450 );
1451 assert_eq!(
1452 canonical_join_key(&DataValue::Integer(220), true),
1453 DataValue::Integer(220)
1454 );
1455 assert_eq!(
1456 canonical_join_key(&DataValue::String("220".to_string()), true),
1457 canonical_join_key(&DataValue::Integer(220), true)
1458 );
1459 }
1460
1461 #[test]
1462 fn numeric_strings_stay_distinct_when_not_coercing() {
1463 assert_eq!(
1466 canonical_join_key(&DataValue::String("007".to_string()), false),
1467 DataValue::String("007".to_string())
1468 );
1469 assert_ne!(
1470 canonical_join_key(&DataValue::String("007".to_string()), false),
1471 canonical_join_key(&DataValue::String("7".to_string()), false)
1472 );
1473 assert_ne!(
1475 canonical_join_key(&DataValue::String("7".to_string()), false),
1476 canonical_join_key(&DataValue::Integer(7), false)
1477 );
1478 }
1479
1480 #[test]
1481 fn interned_and_plain_strings_collapse_regardless_of_coercion() {
1482 for coerce in [true, false] {
1483 assert_eq!(
1484 canonical_join_key(
1485 &DataValue::InternedString(Arc::new("North".to_string())),
1486 coerce
1487 ),
1488 canonical_join_key(&DataValue::String("North".to_string()), coerce),
1489 "interned/plain strings must collapse (coerce = {coerce})"
1490 );
1491 }
1492 }
1493
1494 #[test]
1495 fn whole_float_folds_to_integer_when_coercing() {
1496 assert_eq!(
1497 canonical_join_key(&DataValue::Float(220.0), true),
1498 DataValue::Integer(220)
1499 );
1500 assert_eq!(
1501 canonical_join_key(&DataValue::String("220.0".to_string()), true),
1502 DataValue::Integer(220)
1503 );
1504 assert_eq!(
1506 canonical_join_key(&DataValue::Float(220.5), true),
1507 DataValue::Float(220.5)
1508 );
1509 assert_eq!(
1512 canonical_join_key(&DataValue::Float(220.0), false),
1513 DataValue::Integer(220)
1514 );
1515 }
1516
1517 #[test]
1518 fn non_numeric_text_is_preserved() {
1519 assert_eq!(
1520 canonical_join_key(&DataValue::String("North".to_string()), true),
1521 DataValue::String("North".to_string())
1522 );
1523 assert_eq!(
1525 canonical_join_key(&DataValue::String(" 220".to_string()), true),
1526 DataValue::String(" 220".to_string())
1527 );
1528 }
1529
1530 #[test]
1531 fn non_finite_strings_stay_strings() {
1532 assert_eq!(
1533 canonical_join_key(&DataValue::String("inf".to_string()), true),
1534 DataValue::String("inf".to_string())
1535 );
1536 assert_eq!(
1537 canonical_join_key(&DataValue::String("NaN".to_string()), true),
1538 DataValue::String("NaN".to_string())
1539 );
1540 }
1541
1542 #[test]
1543 fn null_is_unchanged() {
1544 assert_eq!(canonical_join_key(&DataValue::Null, true), DataValue::Null);
1545 }
1546
1547 #[test]
1548 fn coercion_enabled_only_for_differing_value_kinds() {
1549 let stringy = single_col_table(DataValue::String("7".to_string()));
1551 let numeric = single_col_table(DataValue::Integer(7));
1552 let stringy2 = single_col_table(DataValue::String("8".to_string()));
1553 let empty = DataTable::new("empty"); assert!(join_key_coercion(&stringy, 0, &numeric, 0));
1557 assert!(!join_key_coercion(&stringy, 0, &stringy2, 0));
1559 assert!(!join_key_coercion(&numeric, 0, &numeric, 0));
1561 assert!(join_key_coercion(&stringy, 0, &empty, 0));
1563 }
1564
1565 fn single_col_table(value: DataValue) -> DataTable {
1566 let mut t = DataTable::new("t");
1567 t.add_column(DataColumn::new("k"));
1568 let _ = t.add_row(DataRow {
1569 values: vec![value],
1570 });
1571 t
1572 }
1573}