xx

Author: forsaken628

src/query/service/src/physical_plans/mod.rs

@@ -82,7 +82,7 @@ pub use physical_exchange::Exchange;
 pub use physical_exchange_sink::ExchangeSink;
 pub use physical_exchange_source::ExchangeSource;
 pub use physical_filter::Filter;
-pub use physical_hash_join::HashJoin;
+pub use physical_hash_join::*;
 pub use physical_limit::Limit;
 pub use physical_materialized_cte::*;
 pub use physical_multi_table_insert::*;

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

@@ -35,7 +35,9 @@ use databend_common_pipeline::core::Pipe;
 use databend_common_pipeline::core::PipeItem;
 use databend_common_pipeline::core::ProcessorPtr;
 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::ColumnEntry;
 use databend_common_sql::ColumnSet;
@@ -52,6 +54,7 @@ use crate::physical_plans::format::PhysicalFormat;
 use crate::physical_plans::physical_plan::IPhysicalPlan;
 use crate::physical_plans::physical_plan::PhysicalPlan;
 use crate::physical_plans::physical_plan::PhysicalPlanMeta;
+use crate::physical_plans::resolve_scalar;
 use crate::physical_plans::runtime_filter::build_runtime_filter;
 use crate::physical_plans::Exchange;
 use crate::physical_plans::PhysicalPlanBuilder;
@@ -99,6 +102,12 @@ type MergedFieldsResult = (
     Vec<(usize, (bool, bool))>,
 );
 
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub struct NestedLoopFilterInfo {
+    pub predicates: Vec<RemoteExpr>,
+    pub projection: Vec<usize>,
+}
+
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct HashJoin {
     pub meta: PhysicalPlanMeta,
@@ -140,6 +149,7 @@ pub struct HashJoin {
 
     pub runtime_filter: PhysicalRuntimeFilters,
     pub broadcast_id: Option<u32>,
+    pub nested_loop_filter: NestedLoopFilterInfo,
 }
 
 #[typetag::serde]
@@ -261,6 +271,7 @@ impl IPhysicalPlan for HashJoin {
             build_side_cache_info: self.build_side_cache_info.clone(),
             runtime_filter: self.runtime_filter.clone(),
             broadcast_id: self.broadcast_id,
+            nested_loop_filter: self.nested_loop_filter.clone(),
         })
     }
 
@@ -1184,6 +1195,25 @@ impl PhysicalPlanBuilder {
             .collect::<Result<_>>()
     }
 
+    fn build_nested_loop_filter_info(
+        &self,
+        join: &Join,
+        merged_schema: &DataSchemaRef,
+    ) -> Result<NestedLoopFilterInfo> {
+        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_schema))
+            .collect::<Result<_>>()?;
+
+        Ok(NestedLoopFilterInfo {
+            predicates,
+            projection: vec![],
+        })
+    }
+
     pub async fn build_hash_join(
         &mut self,
         join: &Join,
@@ -1256,6 +1286,8 @@ impl PhysicalPlanBuilder {
         // Step 10: Process non-equi conditions
         let non_equi_conditions = self.process_non_equi_conditions(join, &merged_schema)?;
 
+        let nested_loop_filter = self.build_nested_loop_filter_info(join, &merged_schema)?;
+
         // Step 11: Build runtime filter
         let runtime_filter = build_runtime_filter(
             self.ctx.clone(),
@@ -1300,6 +1332,32 @@ impl PhysicalPlanBuilder {
             build_side_cache_info,
             runtime_filter,
             broadcast_id,
+            nested_loop_filter,
         }))
     }
 }
+
+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/hash_join/desc.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 use databend_common_column::bitmap::Bitmap;
-use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::arrow::and_validities;
 use databend_common_expression::type_check::check_function;
@@ -32,6 +31,7 @@ use databend_common_sql::ColumnSet;
 use parking_lot::RwLock;
 
 use crate::physical_plans::HashJoin;
+use crate::physical_plans::NestedLoopFilterInfo;
 use crate::physical_plans::PhysicalRuntimeFilter;
 use crate::physical_plans::PhysicalRuntimeFilters;
 use crate::pipelines::processors::transforms::wrap_true_validity;
@@ -65,7 +65,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) probe_schema: DataSchemaRef,
+    pub(crate) nested_loop_filter: NestedLoopFilterInfo,
 }
 
 #[derive(Debug, Clone)]
@@ -141,7 +141,7 @@ impl HashJoinDesc {
             build_projection: join.build_projections.clone(),
             probe_projection: join.probe_projections.clone(),
             build_schema: join.build.output_schema()?,
-            probe_schema: join.probe.output_schema()?,
+            nested_loop_filter: join.nested_loop_filter.clone(),
         })
     }
 
@@ -269,47 +269,21 @@ impl HashJoinDesc {
         function_ctx: &FunctionContext,
         block_size: usize,
     ) -> Result<FilterExecutor> {
-        let probe_len = self.probe_schema.num_fields();
-
-        let projection = self
-            .probe_projection
-            .iter()
-            .copied()
-            .chain(self.build_projection.iter().map(|idx| probe_len + *idx))
-            .collect::<Vec<_>>();
-
-        let eq_predicates = self
-            .probe_keys
+        let predicates = self
+            .nested_loop_filter
+            .predicates
             .iter()
-            .zip(&self.build_keys)
-            .map(|(probe, build)| {
-                let build = build.project_column_ref(|old| Ok(old + probe_len))?;
-                check_function(None, "eq", &[], &[probe.clone(), build], &BUILTIN_FUNCTIONS)
-            });
-
-        let other_predicate = self.other_predicate.as_ref().map(|expr| {
-            expr.project_column_ref(|index| {
-                projection.get(*index).copied().ok_or_else(|| {
-                    ErrorCode::Internal(format!(
-                        "Invalid column index {} for nested loop predicate projection",
-                        index
-                    ))
-                })
-            })
-        });
-
-        let predicate = eq_predicates
-            .chain(other_predicate)
+            .map(|x| Ok(x.as_expr(&BUILTIN_FUNCTIONS)))
             .reduce(|lhs, rhs| {
                 check_function(None, "and_filters", &[], &[lhs?, rhs?], &BUILTIN_FUNCTIONS)
             })
             .unwrap()?;
 
         Ok(FilterExecutor::new(
-            predicate,
+            predicates,
             function_ctx.clone(),
             block_size,
-            Some(projection.into_iter().collect()),
+            None, // Some(self.nested_loop_filter.projection.iter().copied().collect()),
             &BUILTIN_FUNCTIONS,
             false,
         ))