x

Author: forsaken628

src/query/service/src/lib.rs

@@ -41,6 +41,7 @@
 #![allow(clippy::diverging_sub_expression)]
 #![allow(clippy::arc_with_non_send_sync)]
 #![feature(debug_closure_helpers)]
+#![feature(stmt_expr_attributes)]
 
 extern crate core;
 

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,80 +1195,23 @@ impl PhysicalPlanBuilder {
             .collect::<Result<_>>()
     }
 
-    /// Creates a HashJoin physical plan
-    ///
-    /// # Arguments
-    /// * `join` - Join operation
-    /// * `probe_side` - Probe side physical plan
-    /// * `build_side` - Build side physical plan
-    /// * `is_broadcast` - Whether this is a broadcast join
-    /// * `projections` - Column projections
-    /// * `probe_projections` - Probe side projections
-    /// * `build_projections` - Build side projections
-    /// * `left_join_conditions` - Left join conditions
-    /// * `right_join_conditions` - Right join conditions
-    /// * `is_null_equal` - Null equality flags
-    /// * `non_equi_conditions` - Non-equi conditions
-    /// * `probe_to_build` - Probe to build mapping
-    /// * `output_schema` - Output schema
-    /// * `build_side_cache_info` - Build side cache info
-    /// * `runtime_filter` - Runtime filter
-    /// * `stat_info` - Statistics info
-    ///
-    /// # Returns
-    /// * `Result<PhysicalPlan>` - The HashJoin physical plan
-    #[allow(clippy::too_many_arguments)]
-    fn create_hash_join(
+    fn build_nested_loop_filter_info(
         &self,
-        s_expr: &SExpr,
         join: &Join,
-        probe_side: PhysicalPlan,
-        build_side: PhysicalPlan,
-        projections: ColumnSet,
-        probe_projections: ColumnSet,
-        build_projections: ColumnSet,
-        left_join_conditions: Vec<RemoteExpr>,
-        right_join_conditions: Vec<RemoteExpr>,
-        is_null_equal: Vec<bool>,
-        non_equi_conditions: Vec<RemoteExpr>,
-        probe_to_build: Vec<(usize, (bool, bool))>,
-        output_schema: DataSchemaRef,
-        build_side_cache_info: Option<(usize, HashMap<IndexType, usize>)>,
-        runtime_filter: PhysicalRuntimeFilters,
-        stat_info: PlanStatsInfo,
-    ) -> Result<PhysicalPlan> {
-        let build_side_data_distribution = s_expr.build_side_child().get_data_distribution()?;
-        let broadcast_id = if build_side_data_distribution
-            .as_ref()
-            .is_some_and(|e| matches!(e, databend_common_sql::plans::Exchange::NodeToNodeHash(_)))
-        {
-            Some(self.ctx.get_next_broadcast_id())
-        } else {
-            None
-        };
-        Ok(PhysicalPlan::new(HashJoin {
-            projections,
-            build_projections,
-            probe_projections,
-            build: build_side,
-            probe: probe_side,
-            join_type: join.join_type,
-            build_keys: right_join_conditions,
-            probe_keys: left_join_conditions,
-            is_null_equal,
-            non_equi_conditions,
-            marker_index: join.marker_index,
-            meta: PhysicalPlanMeta::new("HashJoin"),
-            from_correlated_subquery: join.from_correlated_subquery,
-            probe_to_build,
-            output_schema,
-            need_hold_hash_table: join.need_hold_hash_table,
-            stat_info: Some(stat_info),
-            single_to_inner: join.single_to_inner,
-            build_side_cache_info,
-            runtime_filter,
-            broadcast_id,
-        }))
+        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(
@@ -1332,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(),
@@ -1345,23 +1301,63 @@ impl PhysicalPlanBuilder {
         .await?;
 
         // Step 12: Create and return the HashJoin
-        self.create_hash_join(
-            s_expr,
-            join,
-            probe_side,
-            build_side,
+        let build_side_data_distribution = s_expr.build_side_child().get_data_distribution()?;
+        let broadcast_id = if build_side_data_distribution
+            .as_ref()
+            .is_some_and(|e| matches!(e, databend_common_sql::plans::Exchange::NodeToNodeHash(_)))
+        {
+            Some(self.ctx.get_next_broadcast_id())
+        } else {
+            None
+        };
+        Ok(PhysicalPlan::new(HashJoin {
             projections,
-            probe_projections,
             build_projections,
-            left_join_conditions,
-            right_join_conditions,
+            probe_projections,
+            build: build_side,
+            probe: probe_side,
+            join_type: join.join_type,
+            build_keys: right_join_conditions,
+            probe_keys: left_join_conditions,
             is_null_equal,
             non_equi_conditions,
+            marker_index: join.marker_index,
+            meta: PhysicalPlanMeta::new("HashJoin"),
+            from_correlated_subquery: join.from_correlated_subquery,
             probe_to_build,
             output_schema,
+            need_hold_hash_table: join.need_hold_hash_table,
+            stat_info: Some(stat_info),
+            single_to_inner: join.single_to_inner,
             build_side_cache_info,
             runtime_filter,
-            stat_info,
-        )
+            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/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/physical_plans/physical_nested_loop_join.rs

@@ -13,15 +13,17 @@
 // limitations under the License.
 
 use std::any::Any;
-use std::sync::Arc;
 
 use databend_common_exception::Result;
 use databend_common_expression::DataSchemaRef;
 use databend_common_expression::DataSchemaRefExt;
 use databend_common_expression::RemoteExpr;
 use databend_common_functions::BUILTIN_FUNCTIONS;
+use databend_common_pipeline::core::InputPort;
+use databend_common_pipeline::core::OutputPort;
+use databend_common_pipeline::core::Pipe;
+use databend_common_pipeline::core::PipeItem;
 use databend_common_pipeline::core::ProcessorPtr;
-use databend_common_pipeline::sinks::Sinker;
 use databend_common_sql::executor::cast_expr_to_non_null_boolean;
 use databend_common_sql::optimizer::ir::SExpr;
 use databend_common_sql::plans::JoinType;
@@ -38,8 +40,7 @@ use super::PhysicalPlan;
 use super::PhysicalPlanBuilder;
 use super::PhysicalPlanMeta;
 use crate::pipelines::processors::transforms::LoopJoinState;
-use crate::pipelines::processors::transforms::TransformLoopJoinLeft;
-use crate::pipelines::processors::transforms::TransformLoopJoinRight;
+use crate::pipelines::processors::transforms::TransformLoopJoin;
 use crate::pipelines::PipelineBuilder;
 
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
@@ -110,25 +111,56 @@ impl IPhysicalPlan for NestedLoopJoin {
     }
 
     fn build_pipeline2(&self, builder: &mut PipelineBuilder) -> Result<()> {
-        let state = Arc::new(LoopJoinState::new(builder.ctx.clone(), self));
-        self.build_right(state.clone(), builder)?;
-        self.build_left(state, builder)
-    }
-}
+        // Build right side (build side)
+        let right_side_builder = builder.create_sub_pipeline_builder();
+        let mut right_res = right_side_builder.finalize(&self.right)?;
+        let mut build_sinks = right_res.main_pipeline.take_sinks();
+        builder
+            .pipelines
+            .push(right_res.main_pipeline.finalize(None));
+        builder.pipelines.extend(right_res.sources_pipelines);
 
-impl NestedLoopJoin {
-    fn build_left(&self, state: Arc<LoopJoinState>, builder: &mut PipelineBuilder) -> Result<()> {
+        // Build left side (probe side)
         self.left.build_pipeline(builder)?;
 
-        let max_threads = builder.settings.get_max_threads()? as usize;
-        builder.main_pipeline.try_resize(max_threads)?;
-        builder.main_pipeline.add_transform(|input, output| {
-            Ok(ProcessorPtr::create(TransformLoopJoinLeft::create(
-                input,
-                output,
-                state.clone(),
-            )))
-        })?;
+        let output_len = std::cmp::max(build_sinks.len(), builder.main_pipeline.output_len());
+        builder.main_pipeline.resize(output_len, false)?;
+        let probe_sinks = builder.main_pipeline.take_sinks();
+
+        if output_len != build_sinks.len() {
+            builder.main_pipeline.extend_sinks(build_sinks);
+            builder.main_pipeline.resize(output_len, false)?;
+            build_sinks = builder.main_pipeline.take_sinks();
+        }
+
+        debug_assert_eq!(build_sinks.len(), probe_sinks.len());
+
+        let join_pipe_items = build_sinks
+            .into_iter()
+            .zip(probe_sinks)
+            .map(|(build_sink, probe_sink)| {
+                builder.main_pipeline.extend_sinks([build_sink, probe_sink]);
+
+                let build_input = InputPort::create();
+                let probe_input = InputPort::create();
+                let joined_output = OutputPort::create();
+
+                let join_state = LoopJoinState::new(builder.ctx.clone(), self);
+                let loop_join = ProcessorPtr::create(TransformLoopJoin::create(
+                    build_input.clone(),
+                    probe_input.clone(),
+                    joined_output.clone(),
+                    Box::new(join_state),
+                ));
+
+                PipeItem::create(loop_join, vec![build_input, probe_input], vec![
+                    joined_output,
+                ])
+            })
+            .collect();
+
+        let join_pipe = Pipe::create(output_len * 2, output_len, join_pipe_items);
+        builder.main_pipeline.add_pipe(join_pipe);
 
         match self.conditions.len() {
             0 => Ok(()),
@@ -148,24 +180,6 @@ impl NestedLoopJoin {
             }
         }
     }
-
-    fn build_right(&self, state: Arc<LoopJoinState>, builder: &mut PipelineBuilder) -> Result<()> {
-        let right_side_builder = builder.create_sub_pipeline_builder();
-
-        let mut right_res = right_side_builder.finalize(&self.right)?;
-        right_res.main_pipeline.add_sink(|input| {
-            Ok(ProcessorPtr::create(Sinker::create(
-                input,
-                TransformLoopJoinRight::create(state.clone())?,
-            )))
-        })?;
-
-        builder
-            .pipelines
-            .push(right_res.main_pipeline.finalize(None));
-        builder.pipelines.extend(right_res.sources_pipelines);
-        Ok(())
-    }
 }
 
 impl PhysicalPlanBuilder {