Merge pull request #7645 from sundy-li/in-improve

refactor(query): improve in function

Author: mergify[bot]

Cargo.lock

@@ -15,6 +15,7 @@ test = false
 common-base = { path = "../base" }
 
 # Crates.io dependencies
+ahash = "0.7.6"
 ordered-float = { git = "https://github.com/andylokandy/rust-ordered-float.git", branch = "as", features = ["serde"] }
 primitive-types = "0.11.1"
 

src/common/hashtable/Cargo.toml

@@ -30,4 +30,8 @@ impl<Key: HashTableKeyable, Grower: HashTableGrower, Allocator: AllocatorTrait +
             self.insert_key(value.get_key(), &mut inserted);
         }
     }
+
+    pub fn contains(&self, key: &Key) -> bool {
+        self.find_key(key).is_some()
+    }
 }

src/common/hashtable/src/hash_set.rs

@@ -126,7 +126,7 @@ impl HashTableKeyable for OrderedFloat<f32> {
 
     #[inline(always)]
     fn is_zero(&self) -> bool {
-        self.is_nan()
+        self.0 == 0.0
     }
 
     #[inline(always)]
@@ -144,7 +144,7 @@ impl HashTableKeyable for OrderedFloat<f64> {
 
     #[inline(always)]
     fn is_zero(&self) -> bool {
-        self.is_nan()
+        self.0 == 0.0
     }
 
     #[inline(always)]

src/common/hashtable/src/hash_table_key.rs

@@ -0,0 +1,79 @@
+// Copyright 2021 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::hash::Hasher;
+
+use ahash::AHasher;
+
+use super::HashTableKeyable;
+
+#[derive(Clone, Copy)]
+pub struct KeysRef {
+    pub length: usize,
+    pub address: usize,
+}
+
+impl KeysRef {
+    pub fn create(address: usize, length: usize) -> KeysRef {
+        KeysRef { length, address }
+    }
+}
+
+impl Eq for KeysRef {}
+
+impl PartialEq for KeysRef {
+    fn eq(&self, other: &Self) -> bool {
+        if self.length != other.length {
+            return false;
+        }
+
+        unsafe {
+            let self_value = std::slice::from_raw_parts(self.address as *const u8, self.length);
+            let other_value = std::slice::from_raw_parts(other.address as *const u8, other.length);
+            self_value == other_value
+        }
+    }
+}
+
+impl HashTableKeyable for KeysRef {
+    const BEFORE_EQ_HASH: bool = true;
+
+    fn is_zero(&self) -> bool {
+        self.length == 0
+    }
+
+    fn fast_hash(&self) -> u64 {
+        unsafe {
+            // TODO(Winter) We need more efficient hash algorithm
+            let value = std::slice::from_raw_parts(self.address as *const u8, self.length);
+
+            let mut hasher = AHasher::default();
+            hasher.write(value);
+            hasher.finish()
+        }
+    }
+
+    fn set_key(&mut self, new_value: &Self) {
+        self.length = new_value.length;
+        self.address = new_value.address;
+    }
+}
+
+impl std::hash::Hash for KeysRef {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        let self_value =
+            unsafe { std::slice::from_raw_parts(self.address as *const u8, self.length) };
+        self_value.hash(state);
+    }
+}

src/common/hashtable/src/keys_ref.rs

@@ -36,8 +36,11 @@ mod hash_table_entity;
 mod hash_table_grower;
 mod hash_table_iter;
 mod hash_table_key;
+mod keys_ref;
 mod two_level_hash_table;
 
+pub use keys_ref::KeysRef;
+
 #[cfg(not(target_os = "linux"))]
 type HashTableAllocator = common_base::mem_allocator::JEAllocator;
 #[cfg(target_os = "linux")]

src/common/hashtable/src/lib.rs

@@ -51,7 +51,7 @@ where T: Scalar
         }
     }
 
-    fn build_nonull(&mut self, length: usize) -> ColumnRef {
+    pub fn build_nonull(&mut self, length: usize) -> ColumnRef {
         let size = self.len();
         let col = self.builder.to_column();
         if length != size && size == 1 {

src/query/datavalues/src/columns/builder.rs

@@ -26,14 +26,14 @@ use common_expression::with_number_mapped_type;
 use common_expression::Column;
 use common_expression::ColumnBuilder;
 use common_expression::Scalar;
+use common_hashtable::KeysRef;
 use common_io::prelude::*;
 
 use super::aggregate_distinct_state::AggregateDistinctNumberState;
 use super::aggregate_distinct_state::AggregateDistinctState;
 use super::aggregate_distinct_state::AggregateDistinctStringState;
 use super::aggregate_distinct_state::DataGroupValue;
 use super::aggregate_distinct_state::DistinctStateFunc;
-use super::aggregate_distinct_state::KeysRef;
 use super::aggregate_function::AggregateFunction;
 use super::aggregate_function_factory::AggregateFunctionCreator;
 use super::aggregate_function_factory::AggregateFunctionDescription;

src/query/functions-v2/src/aggregates/aggregate_combinator_distinct.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 use std::collections::hash_map::RandomState;
 use std::collections::HashSet;
-use std::hash::Hash;
 use std::marker::Send;
 use std::marker::Sync;
 
@@ -34,6 +33,7 @@ use common_expression::Scalar;
 use common_hashtable::HashSetWithStackMemory;
 use common_hashtable::HashTableEntity;
 use common_hashtable::HashTableKeyable;
+use common_hashtable::KeysRef;
 use common_io::prelude::*;
 use serde::de::DeserializeOwned;
 use serde::Serialize;
@@ -327,39 +327,3 @@ where T: Number + Serialize + DeserializeOwned + HashTableKeyable
         Ok(vec![NumberType::<T>::upcast_column(values)])
     }
 }
-
-#[derive(Clone, Copy)]
-pub struct KeysRef {
-    pub length: usize,
-    pub address: usize,
-}
-
-impl KeysRef {
-    pub fn create(address: usize, length: usize) -> KeysRef {
-        KeysRef { length, address }
-    }
-}
-
-impl Eq for KeysRef {}
-
-impl PartialEq for KeysRef {
-    fn eq(&self, other: &Self) -> bool {
-        if self.length != other.length {
-            return false;
-        }
-
-        unsafe {
-            let self_value = std::slice::from_raw_parts(self.address as *const u8, self.length);
-            let other_value = std::slice::from_raw_parts(other.address as *const u8, other.length);
-            self_value == other_value
-        }
-    }
-}
-
-impl Hash for KeysRef {
-    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
-        let self_value =
-            unsafe { std::slice::from_raw_parts(self.address as *const u8, self.length) };
-        self_value.hash(state);
-    }
-}

src/query/functions-v2/src/aggregates/aggregate_distinct_state.rs

@@ -20,6 +20,7 @@ use std::sync::Arc;
 use common_arrow::arrow::bitmap::Bitmap;
 use common_datavalues::prelude::*;
 use common_exception::Result;
+use common_hashtable::KeysRef;
 use common_io::prelude::*;
 use ordered_float::OrderedFloat;
 
@@ -28,7 +29,6 @@ use super::aggregate_distinct_state::AggregateDistinctState;
 use super::aggregate_distinct_state::AggregateDistinctStringState;
 use super::aggregate_distinct_state::DataGroupValues;
 use super::aggregate_distinct_state::DistinctStateFunc;
-use super::aggregate_distinct_state::KeysRef;
 use super::aggregate_function::AggregateFunction;
 use super::aggregate_function_factory::AggregateFunctionCreator;
 use super::aggregate_function_factory::AggregateFunctionDescription;