refactor: improve the partition compute (#1151)

## Rationale
The previous implementation of computing partition key is coarse,
leading to poor performance.

## Detailed Changes
Refactor the implementation of computing partition key, and the most of
memory allocation is avoided.

## Test Plan
Existing unit tests.

common_types/src/datum.rs

@@ -498,7 +498,7 @@ impl Datum {
             Datum::Timestamp(v) => v.as_i64() as u64,
             Datum::Double(v) => *v as u64,
             Datum::Float(v) => *v as u64,
-            Datum::Varbinary(v) => hash64(v),
+            Datum::Varbinary(v) => hash64(&v[..]),
             Datum::String(v) => hash64(v.as_bytes()),
             Datum::UInt64(v) => *v,
             Datum::UInt32(v) => *v as u64,
@@ -937,7 +937,6 @@ impl Datum {
         }
     }
 
-    #[cfg(test)]
     pub fn as_view(&self) -> DatumView {
         match self {
             Datum::Null => DatumView::Null,
@@ -1056,7 +1055,7 @@ impl Serialize for Datum {
 /// A view to a datum.
 ///
 /// Holds copy of integer like datum and reference of string like datum.
-#[derive(Debug, PartialEq, PartialOrd)]
+#[derive(Clone, Debug, PartialEq, PartialOrd)]
 pub enum DatumView<'a> {
     Null,
     Timestamp(Timestamp),

components/hash_ext/src/lib.rs

@@ -16,7 +16,7 @@
 /// - Memory: aHash
 /// - Disk: SeaHash
 /// https://github.com/CeresDB/hash-benchmark-rs
-use std::hash::BuildHasher;
+use std::{hash::BuildHasher, io::Read};
 
 pub use ahash;
 use byteorder::{ByteOrder, LittleEndian};
@@ -34,9 +34,9 @@ impl BuildHasher for SeaHasherBuilder {
     }
 }
 
-pub fn hash64(mut bytes: &[u8]) -> u64 {
+pub fn hash64<R: Read>(mut source: R) -> u64 {
     let mut out = [0; 16];
-    murmur3_x64_128(&mut bytes, 0, &mut out);
+    murmur3_x64_128(&mut source, 0, &mut out);
     // in most cases we run on little endian target
     LittleEndian::read_u64(&out[0..8])
 }
@@ -53,13 +53,13 @@ mod test {
 
     #[test]
     fn test_murmur_hash() {
-        assert_eq!(hash64(&[]), 0);
+        assert_eq!(hash64(&(vec![])[..]), 0);
 
         for (key, code) in [
             (b"cse_engine_hash_mod_test_bytes1", 6401327391689448380),
             (b"cse_engine_hash_mod_test_bytes2", 10824100215277000151),
         ] {
-            assert_eq!(code, hash64(key));
+            assert_eq!(code, hash64(key.as_slice()));
         }
     }
 

interpreters/src/insert.rs

@@ -204,7 +204,7 @@ impl<'a> TsidBuilder<'a> {
     }
 
     fn finish(self) -> u64 {
-        hash64(self.hash_bytes)
+        hash64(&self.hash_bytes[..])
     }
 }
 

query_frontend/src/promql/udf.rs

@@ -152,7 +152,7 @@ impl UUIDBuilder {
     }
 
     fn finish(self) -> u64 {
-        hash64(&self.buf)
+        hash64(&self.buf[..])
     }
 }
 

table_engine/Cargo.toml

@@ -48,3 +48,6 @@ smallvec = { workspace = true }
 snafu = { workspace = true }
 tokio = { workspace = true }
 trace_metric = { workspace = true }
+
+[dev-dependencies]
+common_types = { workspace = true, features = ["test"] }

table_engine/src/partition/rule/df_adapter/mod.rs

@@ -74,7 +74,6 @@ impl DfPartitionRuleAdapter {
 
 #[cfg(test)]
 mod tests {
-    use bytes_ext::BytesMut;
     use common_types::{
         column_schema,
         datum::{Datum, DatumKind},
@@ -121,9 +120,8 @@ mod tests {
             Datum::String(StringBytes::from("test")),
             Datum::UInt64(42),
         ];
-        let partition_key_refs = partition_keys.iter().collect::<Vec<_>>();
-        let mut buf = BytesMut::new();
-        let expected = compute_partition(&partition_key_refs, partition_num, &mut buf);
+        let partition_key_refs = partition_keys.iter().map(Datum::as_view);
+        let expected = compute_partition(partition_key_refs, partition_num);
 
         assert_eq!(partitions[0], expected);
 
@@ -239,12 +237,11 @@ mod tests {
 
         // Expected
         let partition_keys_1 = test_datums[0].clone();
-        let partition_key_refs_1 = partition_keys_1.iter().collect::<Vec<_>>();
+        let partition_key_refs_1 = partition_keys_1.iter().map(Datum::as_view);
         let partition_keys_2 = test_datums[1].clone();
-        let partition_key_refs_2 = partition_keys_2.iter().collect::<Vec<_>>();
-        let mut buf = BytesMut::new();
-        let expected_1 = compute_partition(&partition_key_refs_1, partition_num, &mut buf);
-        let expected_2 = compute_partition(&partition_key_refs_2, partition_num, &mut buf);
+        let partition_key_refs_2 = partition_keys_2.iter().map(Datum::as_view);
+        let expected_1 = compute_partition(partition_key_refs_1, partition_num);
+        let expected_2 = compute_partition(partition_key_refs_2, partition_num);
         let expecteds = vec![expected_1, expected_2];
 
         assert_eq!(partitions, expecteds);