x

Author: forsaken628

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

@@ -62,7 +62,7 @@ pub struct HashJoinDesc {
     pub(crate) runtime_filter: RuntimeFiltersDesc,
 
     pub(crate) build_projection: ColumnSet,
-    pub(crate) probe_projections: ColumnSet,
+    pub(crate) probe_projection: ColumnSet,
     pub(crate) probe_to_build: Vec<(usize, (bool, bool))>,
     pub(crate) build_schema: DataSchemaRef,
     pub(crate) probe_schema: DataSchemaRef,
@@ -139,7 +139,7 @@ impl HashJoinDesc {
             runtime_filter: (&join.runtime_filter).into(),
             probe_to_build: join.probe_to_build.clone(),
             build_projection: join.build_projections.clone(),
-            probe_projections: join.probe_projections.clone(),
+            probe_projection: join.probe_projections.clone(),
             build_schema: join.build.output_schema()?,
             probe_schema: join.probe.output_schema()?,
         })
@@ -268,11 +268,11 @@ impl HashJoinDesc {
         &self,
         function_ctx: &FunctionContext,
         block_size: usize,
-    ) -> Result<Option<FilterExecutor>> {
+    ) -> Result<FilterExecutor> {
         let probe_len = self.probe_schema.num_fields();
 
         let projection = self
-            .probe_projections
+            .probe_projection
             .iter()
             .copied()
             .chain(self.build_projection.iter().map(|idx| probe_len + *idx))
@@ -305,13 +305,13 @@ impl HashJoinDesc {
             })
             .unwrap()?;
 
-        Ok(Some(FilterExecutor::new(
+        Ok(FilterExecutor::new(
             predicate,
             function_ctx.clone(),
             block_size,
-            Some(projection),
+            Some(projection.into_iter().collect()),
             &BUILTIN_FUNCTIONS,
             false,
-        )))
+        ))
     }
 }

src/query/service/src/pipelines/processors/transforms/new_hash_join/memory/inner_join.rs

@@ -52,7 +52,7 @@ pub struct InnerHashJoin {
     pub(crate) function_ctx: FunctionContext,
     pub(crate) basic_state: Arc<BasicHashJoinState>,
     pub(crate) performance_context: PerformanceContext,
-    nested_loop_filter: Option<FilterExecutor>,
+    nested_loop_filter: FilterExecutor,
 }
 
 impl InnerHashJoin {
@@ -84,7 +84,7 @@ impl InnerHashJoin {
             function_ctx,
             basic_state: state,
             performance_context: context,
-            nested_loop_filter: Some(nested_loop_filter),
+            nested_loop_filter,
         })
     }
 }
@@ -127,12 +127,10 @@ impl Join for InnerHashJoin {
             }
             HashJoinHashTable::NestedLoop(build_blocks) => {
                 let nested = Box::new(LoopJoinStream::new(data, build_blocks));
-                return match self.nested_loop_filter.as_mut() {
-                    Some(filter_executor) => {
-                        Ok(InnerHashJoinFilterStream::create(nested, filter_executor))
-                    }
-                    None => Ok(nested),
-                };
+                return Ok(InnerHashJoinFilterStream::create(
+                    nested,
+                    &mut self.nested_loop_filter,
+                ));
             }
             _ => (),
         }
@@ -146,7 +144,7 @@ impl Join for InnerHashJoin {
         };
 
         self.desc.remove_keys_nullable(&mut keys);
-        let probe_block = data.project(&self.desc.probe_projections);
+        let probe_block = data.project(&self.desc.probe_projection);
 
         let joined_stream = with_join_hash_method!(|T| match self.basic_state.hash_table.deref() {
             HashJoinHashTable::T(table) => {
@@ -252,31 +250,27 @@ impl<'a> JoinStream for InnerHashJoinStream<'a> {
                 (None, None) => DataBlock::new(vec![], self.probed_rows.matched_build.len()),
             };
 
-            if !self.desc.probe_to_build.is_empty() {
-                for (index, (is_probe_nullable, is_build_nullable)) in
-                    self.desc.probe_to_build.iter()
-                {
-                    let entry = match (is_probe_nullable, is_build_nullable) {
-                        (true, true) | (false, false) => result_block.get_by_offset(*index).clone(),
-                        (true, false) => {
-                            result_block.get_by_offset(*index).clone().remove_nullable()
-                        }
-                        (false, true) => {
-                            let entry = result_block.get_by_offset(*index);
-                            let col = entry.to_column();
-
-                            match col.is_null() || col.is_nullable() {
-                                true => entry.clone(),
-                                false => BlockEntry::from(NullableColumn::new_column(
-                                    col,
-                                    Bitmap::new_constant(true, result_block.num_rows()),
-                                )),
-                            }
+            for (index, (is_probe_nullable, is_build_nullable)) in
+                self.desc.probe_to_build.iter().cloned()
+            {
+                let entry = match (is_probe_nullable, is_build_nullable) {
+                    (true, true) | (false, false) => result_block.get_by_offset(index).clone(),
+                    (true, false) => result_block.get_by_offset(index).clone().remove_nullable(),
+                    (false, true) => {
+                        let entry = result_block.get_by_offset(index);
+                        let col = entry.to_column();
+
+                        match col.is_null() || col.is_nullable() {
+                            true => entry.clone(),
+                            false => BlockEntry::from(NullableColumn::new_column(
+                                col,
+                                Bitmap::new_constant(true, result_block.num_rows()),
+                            )),
                         }
-                    };
+                    }
+                };
 
-                    result_block.add_entry(entry);
-                }
+                result_block.add_entry(entry);
             }
 
             return Ok(Some(result_block));

src/query/service/src/pipelines/processors/transforms/new_hash_join/memory/outer_left_join.rs

@@ -112,7 +112,7 @@ impl Join for OuterLeftHashJoin {
                 .collect::<Vec<_>>();
 
             let build_block = null_build_block(&types, data.num_rows());
-            let probe_block = Some(data.project(&self.desc.probe_projections));
+            let probe_block = Some(data.project(&self.desc.probe_projection));
             let result_block = final_result_block(&self.desc, probe_block, build_block, num_rows);
             return Ok(Box::new(OneBlockJoinStream(Some(result_block))));
         }
@@ -128,7 +128,7 @@ impl Join for OuterLeftHashJoin {
         };
 
         self.desc.remove_keys_nullable(&mut keys);
-        let probe_block = data.project(&self.desc.probe_projections);
+        let probe_block = data.project(&self.desc.probe_projection);
 
         let probe_stream = with_join_hash_method!(|T| match self.basic_state.hash_table.deref() {
             HashJoinHashTable::T(table) => {