fix

Author: forsaken628

src/common/exception/src/exception_into.rs

@@ -16,7 +16,6 @@ use std::error::Error;
 use std::fmt::Debug;
 use std::fmt::Display;
 use std::fmt::Formatter;
-use std::sync::PoisonError;
 
 use geozero::error::GeozeroError;
 
@@ -437,9 +436,3 @@ impl From<redis::RedisError> for ErrorCode {
         ErrorCode::DictionarySourceError(format!("Dictionary Redis Error, cause: {}", error))
     }
 }
-
-impl<T> From<PoisonError<T>> for ErrorCode {
-    fn from(error: PoisonError<T>) -> Self {
-        ErrorCode::Internal(format!("{error}"))
-    }
-}

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

@@ -141,26 +141,26 @@ impl BasicHashJoin {
             return;
         }
 
-        if matches!(
-            self.state.hash_table.deref(),
-            HashJoinHashTable::NestedLoop(_)
-        ) {
-            return;
-        }
-
         let locked = self.state.mutex.lock();
         let _locked = locked.unwrap_or_else(PoisonError::into_inner);
 
-        if self.state.chunks.is_empty() || !self.state.columns.is_empty() {
+        debug_assert!(!matches!(
+            self.state.hash_table.deref(),
+            HashJoinHashTable::NestedLoop(_)
+        ));
+
+        if !self.state.columns.is_empty() {
             return;
         }
-
         if let Some(block) = self.state.chunks.first() {
-            for offset in 0..self.desc.build_projection.len() {
-                let column_type = self.state.column_types.as_mut();
-                column_type.push(block.get_by_offset(offset).data_type());
-            }
-        }
+            let column_type = self.state.column_types.as_mut();
+            column_type.extend(
+                (0..self.desc.build_projection.len())
+                    .map(|offset| block.get_by_offset(offset).data_type()),
+            );
+        } else {
+            return;
+        };
 
         let mut columns = Vec::with_capacity(self.desc.build_projection.len());
         for offset in 0..self.desc.build_projection.len() {

src/query/service/src/pipelines/processors/transforms/new_hash_join/transform_hash_join.rs

@@ -120,7 +120,7 @@ impl Processor for TransformHashJoin {
         }
     }
 
-    fn process<'a>(&'a mut self) -> Result<()> {
+    fn process(&mut self) -> Result<()> {
         match &mut self.stage {
             Stage::Finished => Ok(()),
             Stage::Build(state) => {
@@ -147,7 +147,7 @@ impl Processor for TransformHashJoin {
                     let stream = self.join.probe_block(probe_data)?;
                     // This is safe because both join and stream are properties of the struct.
                     state.stream = Some(unsafe {
-                        std::mem::transmute::<Box<dyn JoinStream + 'a>, Box<dyn JoinStream>>(stream)
+                        std::mem::transmute::<Box<dyn JoinStream + '_>, Box<dyn JoinStream>>(stream)
                     });
                 }
 
@@ -166,7 +166,7 @@ impl Processor for TransformHashJoin {
                         state.initialize = true;
                         // This is safe because both join and stream are properties of the struct.
                         state.stream = Some(unsafe {
-                            std::mem::transmute::<Box<dyn JoinStream + 'a>, Box<dyn JoinStream>>(
+                            std::mem::transmute::<Box<dyn JoinStream + '_>, Box<dyn JoinStream>>(
                                 final_stream,
                             )
                         });