x

Author: forsaken628

src/query/service/src/pipelines/processors/transforms/hash_join/desc.rs

@@ -22,6 +22,7 @@ use databend_common_expression::DataBlock;
 use databend_common_expression::DataSchemaRef;
 use databend_common_expression::Evaluator;
 use databend_common_expression::Expr;
+use databend_common_expression::FilterExecutor;
 use databend_common_expression::FunctionContext;
 use databend_common_expression::RemoteExpr;
 use databend_common_functions::BUILTIN_FUNCTIONS;
@@ -259,4 +260,35 @@ impl HashJoinDesc {
             }
         }
     }
+
+    pub fn create_nested_loop_filter(
+        &self,
+        function_ctx: &FunctionContext,
+        block_size: usize,
+    ) -> Result<FilterExecutor> {
+        let offset = self.probe_projections.len();
+
+        let predicate = self
+            .probe_keys
+            .iter()
+            .zip(&self.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(self.other_predicate.iter().map(|expr| Ok(expr.clone())))
+            .reduce(|lhs, rhs| {
+                check_function(None, "and_filters", &[], &[lhs?, rhs?], &BUILTIN_FUNCTIONS)
+            })
+            .unwrap()?;
+
+        Ok(FilterExecutor::new(
+            predicate,
+            function_ctx.clone(),
+            block_size,
+            None,
+            &BUILTIN_FUNCTIONS,
+            false,
+        ))
+    }
 }

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

@@ -20,15 +20,13 @@ 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;
@@ -79,37 +77,14 @@ impl InnerHashJoin {
             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,
-        ));
-
+        let nested_loop_filter = desc.create_nested_loop_filter(&function_ctx, block_size)?;
         Ok(InnerHashJoin {
             desc,
             basic_hash_join,
             function_ctx,
             basic_state: state,
             performance_context: context,
-            nested_loop_filter,
+            nested_loop_filter: Some(nested_loop_filter),
         })
     }
 }