refactor(query): refactor join and support grace hash join

src/common/hashtable/src/hashjoin_hashtable.rs

@@ -129,7 +129,7 @@ impl<K: Keyable, A: Allocator + Clone + Default> HashJoinHashTable<K, A> {
         hashtable
     }
 
-    pub fn insert(&mut self, key: K, entry_ptr: *mut RawEntry<K>) {
+    pub fn insert(&self, key: K, entry_ptr: *mut RawEntry<K>) {
         let hash = key.hash();
         let index = (hash >> self.hash_shift) as usize;
         let new_header = new_header(entry_ptr as u64, hash);

src/common/hashtable/src/hashjoin_string_hashtable.rs

@@ -63,7 +63,7 @@ impl<A: Allocator + Clone + Default> HashJoinStringHashTable<A> {
         hashtable
     }
 
-    pub fn insert(&mut self, key: &[u8], entry_ptr: *mut StringRawEntry) {
+    pub fn insert(&self, key: &[u8], entry_ptr: *mut StringRawEntry) {
         let hash = hash_join_fast_string_hash(key);
         let index = (hash >> self.hash_shift) as usize;
         let new_header = new_header(entry_ptr as u64, hash);

src/query/expression/src/block.rs

@@ -644,8 +644,8 @@ impl DataBlock {
     }
 
     #[inline]
-    pub fn remove_column(&mut self, index: usize) {
-        self.entries.remove(index);
+    pub fn remove_column(&mut self, index: usize) -> BlockEntry {
+        self.entries.remove(index)
     }
 
     #[inline]

src/query/expression/src/kernels/group_by_hash/method_fixed_keys.rs

@@ -78,7 +78,9 @@ where T: Clone + Default
     }
 }
 
-trait FixedKey: FastHash + 'static + Sized + Clone + Default + Eq + Debug + Sync + Send {
+pub trait FixedKey:
+    FastHash + 'static + Sized + Clone + Default + Eq + Debug + Sync + Send
+{
     fn downcast(keys_state: &KeysState) -> Option<&Buffer<Self>>;
 
     fn downcast_owned(keys_state: KeysState) -> Option<Buffer<Self>>;

src/query/expression/src/projected_block.rs

@@ -48,6 +48,16 @@ impl<'a> ProjectedBlock<'a> {
         }
     }
 
+    pub fn num_rows(&self) -> usize {
+        if self.entries.is_empty() {
+            return 0;
+        }
+
+        let num_rows = self.entries[0].len();
+        debug_assert!(self.entries.iter().all(|c| c.len() == num_rows));
+        num_rows
+    }
+
     pub fn slice<I>(&self, index: I) -> ProjectedBlock<'_>
     where I: SliceIndex<[usize], Output = [usize]> + SliceIndex<[BlockEntry], Output = [BlockEntry]>
     {

src/query/service/src/lib.rs

@@ -41,7 +41,9 @@
 #![allow(clippy::diverging_sub_expression)]
 #![allow(clippy::arc_with_non_send_sync)]
 #![feature(debug_closure_helpers)]
-
+#![feature(associated_type_defaults)]
+#![feature(mapped_lock_guards)]
+#![feature(unsafe_cell_access)]
 extern crate core;
 
 pub mod auth;

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

@@ -67,7 +67,9 @@ pub struct FixedKeyHashJoinHashTable<T: HashtableKeyable> {
     pub(crate) hash_method: HashMethodFixedKeys<T>,
 }
 
+#[derive(Default)]
 pub enum HashJoinHashTable {
+    #[default]
     Null,
     Serializer(SerializerHashJoinHashTable),
     SingleBinary(SingleBinaryHashJoinHashTable),

src/query/service/src/pipelines/processors/transforms/mod.rs

@@ -17,6 +17,7 @@ pub mod aggregator;
 mod broadcast;
 mod hash_join;
 mod materialized_cte;
+pub mod new_hash_join;
 pub(crate) mod range_join;
 mod runtime_pool;
 pub mod sort;