refactor(query): use InEvaluator to reuse the hashset

src/query/datavalues/Cargo.toml

@@ -14,6 +14,7 @@ test = false
 # Workspace dependencies
 common-arrow = { path = "../../common/arrow" }
 common-exception = { path = "../../common/exception" }
+common-hashtable = { path = "../../common/hashtable" }
 common-io = { path = "../../common/io" }
 primitive-types = "0.11.1"
 

src/query/datavalues/src/scalars/type_.rs

@@ -15,6 +15,8 @@
 use std::any::Any;
 
 use common_exception::Result;
+use common_hashtable::KeysRef;
+use ordered_float::OrderedFloat;
 
 use super::column::ScalarColumn;
 use crate::prelude::*;
@@ -33,6 +35,8 @@ where for<'a> Self::ColumnType: ScalarColumn<RefItem<'a> = Self::RefType<'a>>
     type Viewer<'a>: ScalarViewer<'a, ScalarItem = Self>
     where Self: 'a;
 
+    type KeyType;
+
     /// Get a reference of the current value.
     fn as_scalar_ref(&self) -> Self::RefType<'_>;
 
@@ -52,18 +56,22 @@ pub trait ScalarRef<'a>: std::fmt::Debug + Clone + Copy + Send + 'a {
     /// Convert the reference into an owned value.
     fn to_owned_scalar(&self) -> Self::ScalarType;
 
+    /// Convert the reference into a key type (which is used for hashmap/hashset)
+    fn to_key(&self) -> <<Self as ScalarRef<'a>>::ScalarType as Scalar>::KeyType;
+
     /// Whether to_owned_scalar has heap allocation which is unhandled by Bumplao
     fn has_alloc_beyond_bump() -> bool {
         false
     }
 }
 
 macro_rules! impl_primitive_scalar_type {
-    ($native:ident) => {
+    ($native:ident, $key_type: ident) => {
         impl Scalar for $native {
             type ColumnType = PrimitiveColumn<$native>;
             type RefType<'a> = $native;
             type Viewer<'a> = PrimitiveViewer<'a, $native>;
+            type KeyType = $key_type;
 
             #[inline]
             fn as_scalar_ref(&self) -> $native {
@@ -86,25 +94,35 @@ macro_rules! impl_primitive_scalar_type {
             fn to_owned_scalar(&self) -> $native {
                 *self
             }
+
+            #[inline]
+            fn to_key(&self) -> $key_type {
+                (*self).into()
+            }
         }
     };
 }
 
-impl_primitive_scalar_type!(u8);
-impl_primitive_scalar_type!(u16);
-impl_primitive_scalar_type!(u32);
-impl_primitive_scalar_type!(u64);
-impl_primitive_scalar_type!(i8);
-impl_primitive_scalar_type!(i16);
-impl_primitive_scalar_type!(i32);
-impl_primitive_scalar_type!(i64);
-impl_primitive_scalar_type!(f32);
-impl_primitive_scalar_type!(f64);
+impl_primitive_scalar_type!(u8, u8);
+impl_primitive_scalar_type!(u16, u16);
+impl_primitive_scalar_type!(u32, u32);
+impl_primitive_scalar_type!(u64, u64);
+impl_primitive_scalar_type!(i8, i8);
+impl_primitive_scalar_type!(i16, i16);
+impl_primitive_scalar_type!(i32, i32);
+impl_primitive_scalar_type!(i64, i64);
+
+type F32 = OrderedFloat<f32>;
+type F64 = OrderedFloat<f64>;
+
+impl_primitive_scalar_type!(f32, F32);
+impl_primitive_scalar_type!(f64, F64);
 
 impl Scalar for bool {
     type ColumnType = BooleanColumn;
     type RefType<'a> = bool;
     type Viewer<'a> = BooleanViewer;
+    type KeyType = bool;
 
     #[inline]
     fn as_scalar_ref(&self) -> bool {
@@ -126,12 +144,18 @@ impl<'a> ScalarRef<'a> for bool {
     fn to_owned_scalar(&self) -> bool {
         *self
     }
+
+    #[inline]
+    fn to_key(&self) -> bool {
+        *self
+    }
 }
 
 impl Scalar for Vec<u8> {
     type ColumnType = StringColumn;
     type RefType<'a> = &'a [u8];
     type Viewer<'a> = StringViewer<'a>;
+    type KeyType = KeysRef;
 
     #[inline]
     fn as_scalar_ref(&self) -> &[u8] {
@@ -153,6 +177,11 @@ impl<'a> ScalarRef<'a> for &'a [u8] {
         self.to_vec()
     }
 
+    #[inline]
+    fn to_key(&self) -> KeysRef {
+        KeysRef::create(self.as_ptr() as usize, self.len())
+    }
+
     fn has_alloc_beyond_bump() -> bool {
         true
     }
@@ -162,6 +191,7 @@ impl Scalar for VariantValue {
     type ColumnType = ObjectColumn<VariantValue>;
     type RefType<'a> = &'a VariantValue;
     type Viewer<'a> = ObjectViewer<'a, VariantValue>;
+    type KeyType = VariantValue;
 
     #[inline]
     fn as_scalar_ref(&self) -> &VariantValue {
@@ -184,6 +214,11 @@ impl<'a> ScalarRef<'a> for &'a VariantValue {
         (*self).clone()
     }
 
+    #[inline]
+    fn to_key(&self) -> VariantValue {
+        (*self).clone()
+    }
+
     fn has_alloc_beyond_bump() -> bool {
         true
     }
@@ -193,6 +228,7 @@ impl Scalar for ArrayValue {
     type ColumnType = ArrayColumn;
     type RefType<'a> = ArrayValueRef<'a>;
     type Viewer<'a> = ArrayViewer<'a>;
+    type KeyType = ArrayValue;
 
     #[inline]
     fn as_scalar_ref(&self) -> ArrayValueRef<'_> {
@@ -217,12 +253,18 @@ impl<'a> ScalarRef<'a> for ArrayValueRef<'a> {
             ArrayValueRef::ValueRef { val } => (*val).clone(),
         }
     }
+
+    #[inline]
+    fn to_key(&self) -> ArrayValue {
+        self.to_owned_scalar()
+    }
 }
 
 impl Scalar for StructValue {
     type ColumnType = StructColumn;
     type RefType<'a> = StructValueRef<'a>;
     type Viewer<'a> = StructViewer<'a>;
+    type KeyType = StructValue;
 
     #[inline]
     fn as_scalar_ref(&self) -> StructValueRef<'_> {
@@ -247,4 +289,9 @@ impl<'a> ScalarRef<'a> for StructValueRef<'a> {
             StructValueRef::ValueRef { val } => (*val).clone(),
         }
     }
+
+    #[inline]
+    fn to_key(&self) -> StructValue {
+        self.to_owned_scalar()
+    }
 }