refactor(query): improve hash join

src/common/hashtable/src/hashjoin_hashtable.rs

@@ -109,7 +109,9 @@ where
 
     fn contains(&self, key_ref: &Self::Key) -> bool {
         let index = key_ref.hash() as usize & self.hash_mask;
-        let mut raw_entry_ptr = self.pointers[index];
+        // # Safety
+        // `index` = hash & mask, it is less than the capacity of hash table.
+        let mut raw_entry_ptr = unsafe { *self.pointers.get_unchecked(index) };
         loop {
             if raw_entry_ptr == 0 {
                 break;
@@ -132,22 +134,18 @@ where
     ) -> (usize, u64) {
         let index = key_ref.hash() as usize & self.hash_mask;
         let origin = occupied;
-        let mut raw_entry_ptr = self.pointers[index];
+        // # Safety
+        // `index` = hash & mask, it is less than the capacity of hash table.
+        let mut raw_entry_ptr = unsafe { *self.pointers.get_unchecked(index) };
         loop {
             if raw_entry_ptr == 0 || occupied >= capacity {
                 break;
             }
-            let raw_entry = unsafe { &*(raw_entry_ptr as *mut RawEntry<K>) };
+            let raw_entry = unsafe { &*(raw_entry_ptr as *const RawEntry<K>) };
             if key_ref == &raw_entry.key {
                 // # Safety
                 // occupied is less than the capacity of vec_ptr.
-                unsafe {
-                    std::ptr::copy_nonoverlapping(
-                        &raw_entry.row_ptr as *const RowPtr,
-                        vec_ptr.add(occupied),
-                        1,
-                    )
-                };
+                unsafe { std::ptr::write(vec_ptr.add(occupied), raw_entry.row_ptr) };
                 occupied += 1;
             }
             raw_entry_ptr = raw_entry.next;
@@ -172,17 +170,11 @@ where
             if incomplete_ptr == 0 || occupied >= capacity {
                 break;
             }
-            let raw_entry = unsafe { &*(incomplete_ptr as *mut RawEntry<K>) };
+            let raw_entry = unsafe { &*(incomplete_ptr as *const RawEntry<K>) };
             if key_ref == &raw_entry.key {
                 // # Safety
                 // occupied is less than the capacity of vec_ptr.
-                unsafe {
-                    std::ptr::copy_nonoverlapping(
-                        &raw_entry.row_ptr as *const RowPtr,
-                        vec_ptr.add(occupied),
-                        1,
-                    )
-                };
+                unsafe { std::ptr::write(vec_ptr.add(occupied), raw_entry.row_ptr) };
                 occupied += 1;
             }
             incomplete_ptr = raw_entry.next;

src/common/hashtable/src/hashjoin_string_hashtable.rs

@@ -87,7 +87,9 @@ where A: Allocator + Clone + 'static
 
     fn contains(&self, key_ref: &Self::Key) -> bool {
         let index = key_ref.fast_hash() as usize & self.hash_mask;
-        let mut raw_entry_ptr = self.pointers[index];
+        // # Safety
+        // `index` = hash & mask, it is less than the capacity of hash table.
+        let mut raw_entry_ptr = unsafe { *self.pointers.get_unchecked(index) };
         loop {
             if raw_entry_ptr == 0 {
                 break;
@@ -125,12 +127,14 @@ where A: Allocator + Clone + 'static
     ) -> (usize, u64) {
         let index = key_ref.fast_hash() as usize & self.hash_mask;
         let origin = occupied;
-        let mut raw_entry_ptr = self.pointers[index];
+        // # Safety
+        // `index` = hash & mask, it is less than the capacity of hash table.
+        let mut raw_entry_ptr = unsafe { *self.pointers.get_unchecked(index) };
         loop {
             if raw_entry_ptr == 0 || occupied >= capacity {
                 break;
             }
-            let raw_entry = unsafe { &*(raw_entry_ptr as *mut StringRawEntry) };
+            let raw_entry = unsafe { &*(raw_entry_ptr as *const StringRawEntry) };
             // Compare `early` and the length of the string, the size of `early` is 4.
             let min_len = std::cmp::min(STRING_EARLY_SIZE, key_ref.len());
             if raw_entry.length as usize == key_ref.len()
@@ -145,13 +149,7 @@ where A: Allocator + Clone + 'static
                 if key == key_ref {
                     // # Safety
                     // occupied is less than the capacity of vec_ptr.
-                    unsafe {
-                        std::ptr::copy_nonoverlapping(
-                            &raw_entry.row_ptr as *const RowPtr,
-                            vec_ptr.add(occupied),
-                            1,
-                        )
-                    };
+                    unsafe { std::ptr::write(vec_ptr.add(occupied), raw_entry.row_ptr) };
                     occupied += 1;
                 }
             }
@@ -177,7 +175,7 @@ where A: Allocator + Clone + 'static
             if incomplete_ptr == 0 || occupied >= capacity {
                 break;
             }
-            let raw_entry = unsafe { &*(incomplete_ptr as *mut StringRawEntry) };
+            let raw_entry = unsafe { &*(incomplete_ptr as *const StringRawEntry) };
             // Compare `early` and the length of the string, the size of `early` is 4.
             let min_len = std::cmp::min(STRING_EARLY_SIZE, key_ref.len());
             if raw_entry.length as usize == key_ref.len()
@@ -192,13 +190,7 @@ where A: Allocator + Clone + 'static
                 if key == key_ref {
                     // # Safety
                     // occupied is less than the capacity of vec_ptr.
-                    unsafe {
-                        std::ptr::copy_nonoverlapping(
-                            &raw_entry.row_ptr as *const RowPtr,
-                            vec_ptr.add(occupied),
-                            1,
-                        )
-                    };
+                    unsafe { std::ptr::write(vec_ptr.add(occupied), raw_entry.row_ptr) };
                     occupied += 1;
                 }
             }