xx

Author: forsaken628

src/query/service/src/physical_plans/physical_hash_join.rs

@@ -150,7 +150,7 @@ pub struct HashJoin {
 
     pub runtime_filter: PhysicalRuntimeFilters,
     pub broadcast_id: Option<u32>,
-    pub nested_loop_filter: NestedLoopFilterInfo,
+    pub nested_loop_filter: Option<NestedLoopFilterInfo>,
 }
 
 #[typetag::serde]
@@ -1202,7 +1202,11 @@ impl PhysicalPlanBuilder {
         probe_schema: &DataSchema,
         build_schema: &DataSchema,
         target_schema: &DataSchema,
-    ) -> Result<NestedLoopFilterInfo> {
+    ) -> Result<Option<NestedLoopFilterInfo>> {
+        if matches!(join.join_type, JoinType::Inner) {
+            return Ok(None);
+        }
+
         let merged = DataSchema::new(
             probe_schema
                 .fields
@@ -1212,19 +1216,40 @@ impl PhysicalPlanBuilder {
                 .collect(),
         );
 
-        let predicates = join
-            .non_equi_conditions
-            .iter()
-            .map(|c| Ok(c.clone()))
-            .chain(join.equi_conditions.iter().map(condition_to_expr))
-            .map(|scalar| resolve_scalar(&scalar?, &merged))
-            .collect::<Result<Vec<_>>>()
-            .map_err(|err| {
-                err.add_message(format!(
+        let mut predicates =
+            Vec::with_capacity(join.equi_conditions.len() + join.non_equi_conditions.len());
+
+        for condition in &join.equi_conditions {
+            let scalar = condition_to_expr(condition)?;
+            match resolve_scalar(&scalar, &merged) {
+                Ok(expr) => predicates.push(expr),
+                Err(_)
+                    if condition
+                        .left
+                        .data_type()
+                        .map(|data_type| data_type.remove_nullable().is_bitmap())
+                        .unwrap_or_default() =>
+                {
+                    // no function matches signature `eq(Bitmap NULL, Bitmap NULL)
+                    return Ok(None);
+                }
+                Err(err) => {
+                    return Err(err.add_message(format!(
                     "Failed build nested loop filter schema: {merged:#?} equi_conditions: {:#?}",
                     join.equi_conditions
+                )))
+                }
+            }
+        }
+
+        for scalar in &join.non_equi_conditions {
+            predicates.push(resolve_scalar(scalar, &merged).map_err(|err|{
+                err.add_message(format!(
+                    "Failed build nested loop filter schema: {merged:#?} non_equi_conditions: {:#?}",
+                    join.non_equi_conditions
                 ))
-            })?;
+            })?);
+        }
 
         let projection = target_schema
             .fields
@@ -1237,10 +1262,10 @@ impl PhysicalPlanBuilder {
                 ))
             })?;
 
-        Ok(NestedLoopFilterInfo {
+        Ok(Some(NestedLoopFilterInfo {
             predicates,
             projection,
-        })
+        }))
     }
 
     pub async fn build_hash_join(
@@ -1372,11 +1397,11 @@ fn condition_to_expr(condition: &JoinEquiCondition) -> Result<ScalarExpr> {
     let right_type = condition.right.data_type()?;
 
     let arguments = match (&left_type, &right_type) {
-        (DataType::Nullable(left), right) if **left == *right => vec![
+        (DataType::Nullable(box 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![
+        (left, DataType::Nullable(box right)) if left == right => vec![
             condition.left.clone().unify_to_data_type(&right_type),
             condition.right.clone(),
         ],

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

@@ -66,7 +66,7 @@ pub struct HashJoinDesc {
     pub(crate) probe_projection: ColumnSet,
     pub(crate) probe_to_build: Vec<(usize, (bool, bool))>,
     pub(crate) build_schema: DataSchemaRef,
-    pub(crate) nested_loop_filter: NestedLoopFilterInfo,
+    pub(crate) nested_loop_filter: Option<NestedLoopFilterInfo>,
 }
 
 #[derive(Debug, Clone)]
@@ -270,9 +270,15 @@ impl HashJoinDesc {
         function_ctx: &FunctionContext,
         block_size: usize,
     ) -> Result<Option<(FilterExecutor, Option<Vec<FieldIndex>>)>> {
-        let predicates = self
-            .nested_loop_filter
-            .predicates
+        let Some(NestedLoopFilterInfo {
+            predicates,
+            projection,
+        }) = &self.nested_loop_filter
+        else {
+            return Ok(None);
+        };
+
+        let predicates = predicates
             .iter()
             .map(|x| Ok(x.as_expr(&BUILTIN_FUNCTIONS)))
             .reduce(|lhs, rhs| {
@@ -283,7 +289,6 @@ impl HashJoinDesc {
             return Ok(None);
         };
 
-        let projection = &self.nested_loop_filter.projection;
         let reorder = if !projection.is_sorted() {
             let mut mapper = projection.iter().cloned().enumerate().collect::<Vec<_>>();
             mapper.sort_by_key(|(_, field)| *field);