refine

Author: forsaken628

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

@@ -240,14 +240,14 @@ impl IPhysicalPlan for HashJoin {
             probe_keys: self.probe_keys.clone(),
             is_null_equal: self.is_null_equal.clone(),
             non_equi_conditions: self.non_equi_conditions.clone(),
-            join_type: self.join_type.clone(),
+            join_type: self.join_type,
             marker_index: self.marker_index,
             from_correlated_subquery: self.from_correlated_subquery,
             probe_to_build: self.probe_to_build.clone(),
             output_schema: self.output_schema.clone(),
             need_hold_hash_table: self.need_hold_hash_table,
             stat_info: self.stat_info.clone(),
-            single_to_inner: self.single_to_inner.clone(),
+            single_to_inner: self.single_to_inner,
             build_side_cache_info: self.build_side_cache_info.clone(),
             runtime_filter: self.runtime_filter.clone(),
             broadcast_id: self.broadcast_id,
@@ -445,7 +445,7 @@ impl HashJoin {
                 build_input.clone(),
                 probe_input.clone(),
                 joined_output.clone(),
-                factory.create_hash_join(self.join_type.clone(), 0)?,
+                factory.create_hash_join(self.join_type, 0)?,
                 stage_sync_barrier.clone(),
                 self.projections.clone(),
                 rf_desc.clone(),
@@ -1210,7 +1210,7 @@ impl PhysicalPlanBuilder {
             probe_projections,
             build: build_side,
             probe: probe_side,
-            join_type: join.join_type.clone(),
+            join_type: join.join_type,
             build_keys: right_join_conditions,
             probe_keys: left_join_conditions,
             is_null_equal,
@@ -1222,7 +1222,7 @@ impl PhysicalPlanBuilder {
             output_schema,
             need_hold_hash_table: join.need_hold_hash_table,
             stat_info: Some(stat_info),
-            single_to_inner: join.single_to_inner.clone(),
+            single_to_inner: join.single_to_inner,
             build_side_cache_info,
             runtime_filter,
             broadcast_id,

src/query/service/src/physical_plans/physical_range_join.rs

@@ -133,7 +133,7 @@ impl IPhysicalPlan for RangeJoin {
             right: right_child,
             conditions: self.conditions.clone(),
             other_conditions: self.other_conditions.clone(),
-            join_type: self.join_type.clone(),
+            join_type: self.join_type,
             range_join_type: self.range_join_type.clone(),
             output_schema: self.output_schema.clone(),
             stat_info: self.stat_info.clone(),
@@ -264,7 +264,7 @@ impl PhysicalPlanBuilder {
                 .iter()
                 .map(|scalar| resolve_scalar(scalar, &merged_schema))
                 .collect::<Result<_>>()?,
-            join_type: join_type.clone(),
+            join_type,
             range_join_type,
             output_schema: Arc::new(DataSchema::new(output_schema)),
             stat_info: Some(self.build_plan_stat_info(s_expr)?),

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

@@ -118,7 +118,7 @@ impl HashJoinDesc {
             .collect();
 
         Ok(HashJoinDesc {
-            join_type: join.join_type.clone(),
+            join_type: join.join_type,
             build_keys,
             probe_keys,
             is_null_equal: join.is_null_equal.clone(),
@@ -128,7 +128,7 @@ impl HashJoinDesc {
                 // marker_index: join.marker_index,
             },
             from_correlated_subquery: join.from_correlated_subquery,
-            single_to_inner: join.single_to_inner.clone(),
+            single_to_inner: join.single_to_inner,
             runtime_filter: (&join.runtime_filter).into(),
             probe_to_build: join.probe_to_build.clone(),
             build_projection: join.build_projections.clone(),

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

@@ -870,7 +870,7 @@ impl HashJoinBuildState {
     }
 
     pub(crate) fn join_type(&self) -> JoinType {
-        self.hash_join_state.hash_join_desc.join_type.clone()
+        self.hash_join_state.hash_join_desc.join_type
     }
 
     pub fn runtime_filter_desc(&self) -> &[RuntimeFilterDesc] {

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

@@ -784,6 +784,6 @@ impl HashJoinProbeState {
     }
 
     pub(crate) fn join_type(&self) -> JoinType {
-        self.hash_join_state.hash_join_desc.join_type.clone()
+        self.hash_join_state.hash_join_desc.join_type
     }
 }

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

@@ -123,7 +123,7 @@ impl HashJoinSpiller {
         }
 
         for data_block in data_blocks {
-            let mut hashes = self.get_hashes(&data_block, &self.join_type)?;
+            let mut hashes = self.get_hashes(&data_block, self.join_type)?;
 
             for hash in hashes.iter_mut() {
                 *hash = Self::get_partition_id(*hash, self.spill_partition_bits as u64);
@@ -143,10 +143,10 @@ impl HashJoinSpiller {
             return Ok(());
         }
 
-        let join_type = self.join_type.clone();
+        let join_type = self.join_type;
         let data_block = DataBlock::concat(&data_blocks)?;
         let partition_data_blocks =
-            self.partition_data_block(&data_block, &join_type, self.spill_partition_bits)?;
+            self.partition_data_block(&data_block, join_type, self.spill_partition_bits)?;
         for (partition_id, data_block) in partition_data_blocks.into_iter().enumerate() {
             if !data_block.is_empty() {
                 self.partition_buffer
@@ -184,7 +184,7 @@ impl HashJoinSpiller {
         let mut partitions_writer = HashMap::<usize, BlocksWriter>::new();
 
         for data_block in data_blocks {
-            let mut hashes = self.get_hashes(&data_block, &self.join_type)?;
+            let mut hashes = self.get_hashes(&data_block, self.join_type)?;
 
             for hash in hashes.iter_mut() {
                 *hash = Self::get_partition_id(*hash, self.spill_partition_bits as u64);
@@ -249,14 +249,14 @@ impl HashJoinSpiller {
         data_blocks: &[DataBlock],
         partition_need_to_spill: Option<&HashSet<usize>>,
     ) -> Result<Vec<DataBlock>> {
-        let join_type = self.join_type.clone();
+        let join_type = self.join_type;
         let mut unspilled_data_blocks = vec![];
 
         let data_block = DataBlock::concat(data_blocks)?;
         let fetch_option =
             PartitionBufferFetchOption::PickPartitionWithThreshold(self.partition_threshold);
         for (partition_id, data_block) in self
-            .partition_data_block(&data_block, &join_type, self.spill_partition_bits)?
+            .partition_data_block(&data_block, join_type, self.spill_partition_bits)?
             .into_iter()
             .enumerate()
         {
@@ -343,10 +343,10 @@ impl HashJoinSpiller {
     fn partition_data_block(
         &mut self,
         data_block: &DataBlock,
-        join_type: &JoinType,
+        join_type: JoinType,
         partition_bits: usize,
     ) -> Result<Vec<DataBlock>> {
-        if join_type == &JoinType::Cross {
+        if join_type == JoinType::Cross {
             Ok(vec![data_block.clone()])
         } else {
             let mut hashes = self.get_hashes(data_block, join_type)?;
@@ -364,7 +364,7 @@ impl HashJoinSpiller {
     }
 
     // Get all hashes for build input data.
-    fn get_hashes(&self, data_block: &DataBlock, join_type: &JoinType) -> Result<Vec<u64>> {
+    fn get_hashes(&self, data_block: &DataBlock, join_type: JoinType) -> Result<Vec<u64>> {
         let mut hashes = Vec::with_capacity(data_block.num_rows());
         get_hashes(
             &self.func_ctx,

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

@@ -236,7 +236,7 @@ impl HashJoinState {
     }
 
     pub fn join_type(&self) -> JoinType {
-        self.hash_join_desc.join_type.clone()
+        self.hash_join_desc.join_type
     }
 
     pub fn need_outer_scan(&self) -> bool {

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

@@ -32,7 +32,7 @@ pub fn get_hashes(
     block: &DataBlock,
     keys: &[Expr],
     method: &HashMethodKind,
-    join_type: &JoinType,
+    join_type: JoinType,
     from_build: bool,
     is_null_equal: &[bool],
     hashes: &mut Vec<u64>,

src/query/service/src/pipelines/processors/transforms/new_hash_join/grace/grace_join.rs

@@ -277,7 +277,7 @@ impl<T: GraceMemoryJoin> GraceHashJoin<T> {
             &data,
             &self.desc.build_keys,
             &self.hash_method_kind,
-            &self.desc.join_type,
+            self.desc.join_type,
             true,
             &self.desc.is_null_equal,
             &mut hashes,
@@ -298,7 +298,7 @@ impl<T: GraceMemoryJoin> GraceHashJoin<T> {
             &data,
             &self.desc.probe_keys,
             &self.hash_method_kind,
-            &self.desc.join_type,
+            self.desc.join_type,
             false,
             &self.desc.is_null_equal,
             &mut hashes,

src/query/service/src/pipelines/processors/transforms/range_join/range_join_state.rs

@@ -99,7 +99,7 @@ impl RangeJoinState {
             finished_tasks: AtomicU64::new(0),
             completed_pair: AtomicU64::new(0),
             ie_join_state,
-            join_type: range_join.join_type.clone(),
+            join_type: range_join.join_type,
             left_match: RwLock::new(MutableBitmap::new()),
             right_match: RwLock::new(MutableBitmap::new()),
             partition_count: AtomicU64::new(0),