x

Author: forsaken628

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

@@ -1184,82 +1184,6 @@ 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(
-        &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,
-        }))
-    }
-
     pub async fn build_hash_join(
         &mut self,
         join: &Join,
@@ -1345,23 +1269,38 @@ impl PhysicalPlanBuilder {
         .await?;
 
         // Step 12: Create and return the HashJoin
-        self.create_hash_join(
-            s_expr,
-            join,
-            probe_side,
-            build_side,
-            projections,
-            probe_projections,
-            build_projections,
-            left_join_conditions,
-            right_join_conditions,
-            is_null_equal,
-            non_equi_conditions,
-            probe_to_build,
-            output_schema,
-            build_side_cache_info,
-            runtime_filter,
-            stat_info,
-        )
+        {
+            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: projections,
+                build_projections: build_projections,
+                probe_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: is_null_equal,
+                non_equi_conditions: non_equi_conditions,
+                marker_index: join.marker_index,
+                meta: PhysicalPlanMeta::new("HashJoin"),
+                from_correlated_subquery: join.from_correlated_subquery,
+                probe_to_build: probe_to_build,
+                output_schema: 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: build_side_cache_info,
+                runtime_filter: runtime_filter,
+                broadcast_id,
+            }))
+        }
     }
 }