x

Author: forsaken628

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

@@ -43,7 +43,7 @@ enum PhysicalJoinType {
 }
 
 // Choose physical join type by join conditions
-fn physical_join(join: &Join, s_expr: &SExpr, settings: &Settings) -> Result<PhysicalJoinType> {
+fn physical_join(join: &Join, s_expr: &SExpr, _settings: &Settings) -> Result<PhysicalJoinType> {
     if join.equi_conditions.is_empty() && join.join_type.is_any_join() {
         return Err(ErrorCode::SemanticError(
             "ANY JOIN only supports equality-based hash joins",
@@ -53,7 +53,7 @@ fn physical_join(join: &Join, s_expr: &SExpr, settings: &Settings) -> Result<Phy
     let left_rel_expr = RelExpr::with_s_expr(s_expr.left_child());
     let right_rel_expr = RelExpr::with_s_expr(s_expr.right_child());
     let right_stat_info = right_rel_expr.derive_cardinality()?;
-    let nested_loop_join_threshold = settings.get_nested_loop_join_threshold()?;
+    let nested_loop_join_threshold = 0;
     if matches!(join.join_type, JoinType::Inner | JoinType::Cross)
         && (right_stat_info
             .statistics

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

@@ -17,7 +17,6 @@ use std::sync::Arc;
 use std::sync::PoisonError;
 
 use databend_common_base::base::ProgressValues;
-use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::Result;
 use databend_common_expression::Column;
 use databend_common_expression::DataBlock;
@@ -27,6 +26,7 @@ use databend_common_expression::HashMethodSerializer;
 use databend_common_expression::HashMethodSingleBinary;
 use databend_common_hashtable::BinaryHashJoinHashMap;
 use databend_common_hashtable::HashJoinHashMap;
+use databend_common_settings::Settings;
 use databend_common_sql::plans::JoinType;
 use ethnum::U256;
 
@@ -40,7 +40,6 @@ use crate::pipelines::processors::transforms::SkipDuplicatesFixedKeyHashJoinHash
 use crate::pipelines::processors::transforms::SkipDuplicatesSerializerHashJoinHashTable;
 use crate::pipelines::processors::transforms::SkipDuplicatesSingleBinaryHashJoinHashTable;
 use crate::pipelines::processors::HashJoinDesc;
-use crate::sessions::QueryContext;
 
 pub struct BasicHashJoin {
     pub(crate) desc: Arc<HashJoinDesc>,
@@ -54,14 +53,13 @@ pub struct BasicHashJoin {
 
 impl BasicHashJoin {
     pub fn create(
-        ctx: &QueryContext,
+        settings: &Settings,
         function_ctx: FunctionContext,
         method: HashMethodKind,
         desc: Arc<HashJoinDesc>,
         state: Arc<BasicHashJoinState>,
+        nested_loop_join_threshold: usize,
     ) -> Result<Self> {
-        let settings = ctx.get_settings();
-        let nested_loop_join_threshold = settings.get_nested_loop_join_threshold()? as _;
         let squash_block = SquashBlocks::new(
             settings.get_max_block_size()? as _,
             settings.get_max_block_bytes()? as _,
@@ -92,7 +90,7 @@ impl BasicHashJoin {
 
     pub(crate) fn final_build(&mut self) -> Result<Option<ProgressValues>> {
         if let Some(true) = self.init_memory_hash_table() {
-            return Ok(Some(self.build_plain()));
+            return Ok(Some(self.build_nested_loop()));
         };
 
         let Some(chunk_index) = self.state.steal_chunk_index() else {
@@ -352,7 +350,7 @@ impl BasicHashJoin {
         Ok(())
     }
 
-    fn build_plain(&self) -> ProgressValues {
+    fn build_nested_loop(&self) -> ProgressValues {
         let mut progress = ProgressValues::default();
         let mut plain = vec![];
         while let Some(chunk_index) = self.state.steal_chunk_index() {
@@ -361,8 +359,6 @@ impl BasicHashJoin {
             let mut chunk_block = DataBlock::empty();
             std::mem::swap(chunk_mut, &mut chunk_block);
 
-            chunk_block = chunk_block.project(&self.desc.build_projection);
-
             progress.rows += chunk_block.num_rows();
             progress.bytes += chunk_block.memory_size();
 

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

@@ -20,13 +20,15 @@ use databend_common_catalog::table_context::TableContext;
 use databend_common_column::bitmap::Bitmap;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_expression::type_check::check_function;
 use databend_common_expression::types::NullableColumn;
 use databend_common_expression::with_join_hash_method;
 use databend_common_expression::BlockEntry;
 use databend_common_expression::DataBlock;
 use databend_common_expression::FilterExecutor;
 use databend_common_expression::FunctionContext;
 use databend_common_expression::HashMethodKind;
+use databend_common_functions::BUILTIN_FUNCTIONS;
 
 use super::basic::BasicHashJoin;
 use super::basic_state::BasicHashJoinState;
@@ -52,6 +54,7 @@ pub struct InnerHashJoin {
     pub(crate) function_ctx: FunctionContext,
     pub(crate) basic_state: Arc<BasicHashJoinState>,
     pub(crate) performance_context: PerformanceContext,
+    nested_loop_filter: Option<FilterExecutor>,
 }
 
 impl InnerHashJoin {
@@ -68,19 +71,45 @@ impl InnerHashJoin {
         let context = PerformanceContext::create(block_size, desc.clone(), function_ctx.clone());
 
         let basic_hash_join = BasicHashJoin::create(
-            ctx,
+            &settings,
             function_ctx.clone(),
             method,
             desc.clone(),
             state.clone(),
+            settings.get_nested_loop_join_threshold()? as _,
         )?;
 
+        let offset = desc.probe_projections.len();
+        let predicate = desc
+            .probe_keys
+            .iter()
+            .zip(&desc.build_keys)
+            .map(|(probe, build)| {
+                let build = build.project_column_ref(|old| Ok(old + offset))?;
+                check_function(None, "eq", &[], &[probe.clone(), build], &BUILTIN_FUNCTIONS)
+            })
+            .chain(desc.other_predicate.iter().map(|expr| Ok(expr.clone())))
+            .reduce(|lhs, rhs| {
+                check_function(None, "and_filters", &[], &[lhs?, rhs?], &BUILTIN_FUNCTIONS)
+            })
+            .unwrap()?;
+
+        let nested_loop_filter = Some(FilterExecutor::new(
+            predicate,
+            function_ctx.clone(),
+            block_size,
+            None,
+            &BUILTIN_FUNCTIONS,
+            false,
+        ));
+
         Ok(InnerHashJoin {
             desc,
             basic_hash_join,
             function_ctx,
             basic_state: state,
             performance_context: context,
+            nested_loop_filter,
         })
     }
 }
@@ -123,7 +152,13 @@ impl Join for InnerHashJoin {
             }
             HashJoinHashTable::NestedLoop(build_blocks) => {
                 let probe_block = data.project(&self.desc.probe_projections);
-                return Ok(Box::new(LoopJoinStream::new(probe_block, build_blocks)));
+                let nested = Box::new(LoopJoinStream::new(probe_block, build_blocks));
+                return match self.nested_loop_filter.as_mut() {
+                    Some(filter_executor) => {
+                        Ok(InnerHashJoinFilterStream::create(nested, filter_executor))
+                    }
+                    None => Ok(nested),
+                };
             }
             _ => (),
         }

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

@@ -68,11 +68,12 @@ impl OuterLeftHashJoin {
         let context = PerformanceContext::create(block_size, desc.clone(), function_ctx.clone());
 
         let basic_hash_join = BasicHashJoin::create(
-            ctx,
+            &settings,
             function_ctx.clone(),
             method,
             desc.clone(),
             state.clone(),
+            0,
         )?;
 
         Ok(OuterLeftHashJoin {

src/query/settings/src/settings_default.rs

@@ -496,7 +496,7 @@ impl DefaultSettings {
                     range: Some(SettingRange::Numeric(0..=u64::MAX)),
                 }),
                 ("nested_loop_join_threshold", DefaultSettingValue {
-                    value: UserSettingValue::UInt64(0),
+                    value: UserSettingValue::UInt64(1024),
                     desc: "Set the threshold for use nested loop join. Setting it to 0 disable nested loop join.",
                     mode: SettingMode::Both,
                     scope: SettingScope::Both,