fix

Author: forsaken628

src/query/service/src/physical_plans/physical_join.rs

@@ -14,12 +14,14 @@
 
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_expression::types::DataType;
 use databend_common_settings::Settings;
 use databend_common_sql::binder::is_range_join_condition;
 use databend_common_sql::optimizer::ir::RelExpr;
 use databend_common_sql::optimizer::ir::SExpr;
 use databend_common_sql::plans::FunctionCall;
 use databend_common_sql::plans::Join;
+use databend_common_sql::plans::JoinEquiCondition;
 use databend_common_sql::plans::JoinType;
 use databend_common_sql::ColumnSet;
 use databend_common_sql::ScalarExpr;
@@ -63,17 +65,9 @@ fn physical_join(join: &Join, s_expr: &SExpr, settings: &Settings) -> Result<Phy
         let conditions = join
             .non_equi_conditions
             .iter()
-            .cloned()
-            .chain(join.equi_conditions.iter().cloned().map(|condition| {
-                FunctionCall {
-                    span: condition.left.span(),
-                    func_name: "eq".to_string(),
-                    params: vec![],
-                    arguments: vec![condition.left, condition.right],
-                }
-                .into()
-            }))
-            .collect();
+            .map(|c| Ok(c.clone()))
+            .chain(join.equi_conditions.iter().map(condition_to_expr))
+            .collect::<Result<_>>()?;
         return Ok(PhysicalJoinType::LoopJoin { conditions });
     };
 
@@ -229,3 +223,28 @@ impl PhysicalPlanBuilder {
         }
     }
 }
+
+fn condition_to_expr(condition: &JoinEquiCondition) -> Result<ScalarExpr> {
+    let left_type = condition.left.data_type()?;
+    let right_type = condition.right.data_type()?;
+
+    let arguments = match (&left_type, &right_type) {
+        (DataType::Nullable(left), right) if **left == *right => vec![
+            condition.left.clone(),
+            condition.right.clone().unify_to_data_type(&left_type),
+        ],
+        (left, DataType::Nullable(right)) if *left == **right => vec![
+            condition.left.clone().unify_to_data_type(&right_type),
+            condition.right.clone(),
+        ],
+        _ => vec![condition.left.clone(), condition.right.clone()],
+    };
+
+    Ok(FunctionCall {
+        span: condition.left.span(),
+        func_name: "eq".to_string(),
+        params: vec![],
+        arguments,
+    }
+    .into())
+}

src/query/service/src/pipelines/processors/transforms/new_hash_join/memory/basic.rs

@@ -60,17 +60,20 @@ impl BasicHashJoin {
         state: Arc<BasicHashJoinState>,
     ) -> Result<Self> {
         let settings = ctx.get_settings();
-        let block_size = settings.get_max_block_size()? as usize;
-        let block_bytes = settings.get_max_block_size()? as usize;
+        let squash_block = SquashBlocks::new(
+            settings.get_max_block_size()? as _,
+            settings.get_max_block_bytes()? as _,
+        );
 
         Ok(BasicHashJoin {
             desc,
             state,
             method,
             function_ctx,
-            squash_block: SquashBlocks::new(block_size, block_bytes),
+            squash_block,
         })
     }
+
     pub(crate) fn add_block(&mut self, mut data: Option<DataBlock>) -> Result<()> {
         let mut squashed_block = match data.take() {
             None => self.squash_block.finalize()?,
@@ -128,6 +131,7 @@ impl BasicHashJoin {
             std::mem::swap(&mut chunks[chunk_index], &mut chunk_block);
         }
 
+        log::info!("build_hash_table chunk_index{chunk_index}");
         self.build_hash_table(keys_block, chunk_index)?;
 
         Ok(Some(ProgressValues {

src/query/service/src/pipelines/processors/transforms/new_hash_join/mod.rs

@@ -26,6 +26,7 @@ mod transform_hash_join;
 pub use grace::GraceHashJoin;
 pub use hash_join_factory::HashJoinFactory;
 pub use join::Join;
+pub use join::JoinStream;
 pub use memory::BasicHashJoinState;
 pub use memory::InnerHashJoin;
 pub use runtime_filter::RuntimeFiltersDesc;

src/query/service/src/pipelines/processors/transforms/transform_loop_join.rs

@@ -18,6 +18,7 @@ use std::sync::Arc;
 use std::sync::Mutex;
 use std::sync::RwLock;
 
+use databend_common_base::base::ProgressValues;
 use databend_common_exception::Result;
 use databend_common_expression::BlockEntry;
 use databend_common_expression::DataBlock;
@@ -28,6 +29,8 @@ use databend_common_pipeline::core::Processor;
 use databend_common_pipeline::sinks::Sink;
 use databend_common_sql::plans::JoinType;
 
+use super::Join;
+use super::JoinStream;
 use crate::physical_plans::NestedLoopJoin;
 use crate::pipelines::executor::WatchNotify;
 use crate::sessions::QueryContext;
@@ -137,7 +140,6 @@ pub struct LoopJoinState {
 
     right_sinker_count: RwLock<usize>,
 
-    #[allow(dead_code)]
     join_type: JoinType,
 }
 
@@ -255,3 +257,49 @@ impl LoopJoinState {
         Ok(iter)
     }
 }
+
+impl Join for LoopJoinState {
+    fn add_block(&mut self, data: Option<DataBlock>) -> Result<()> {
+        let Some(right_block) = data else {
+            return Ok(());
+        };
+
+        let right = if matches!(self.join_type, JoinType::Left | JoinType::Full) {
+            let rows = right_block.num_rows();
+            let entries = right_block
+                .take_columns()
+                .into_iter()
+                .map(|entry| entry.into_nullable())
+                .collect::<Vec<_>>();
+            DataBlock::new(entries, rows)
+        } else {
+            right_block
+        };
+        self.right_table.write()?.push(right);
+        Ok(())
+    }
+
+    fn final_build(&mut self) -> Result<Option<ProgressValues>> {
+        let progress = self.right_table.read()?.iter().fold(
+            ProgressValues::default(),
+            |mut progress, block| {
+                progress.rows += block.num_rows();
+                progress.bytes += block.memory_size();
+                progress
+            },
+        );
+        Ok(Some(progress))
+    }
+
+    fn probe_block(&mut self, data: DataBlock) -> Result<Box<dyn JoinStream + '_>> {
+        todo!();
+    }
+}
+
+struct LoopJoinStream {}
+
+impl JoinStream for LoopJoinStream {
+    fn next(&mut self) -> Result<Option<DataBlock>> {
+        todo!()
+    }
+}

src/query/sql/src/planner/plans/join.rs

@@ -222,7 +222,11 @@ impl JoinEquiCondition {
         left.into_iter()
             .zip(right)
             .enumerate()
-            .map(|(index, (l, r))| JoinEquiCondition::new(l, r, is_null_equal.contains(&index)))
+            .map(|(index, (left, right))| Self {
+                left,
+                right,
+                is_null_equal: is_null_equal.contains(&index),
+            })
             .collect()
     }
 }