refactor(query): use stable siphash for bucket

Author: KKould

Cargo.lock

@@ -358,7 +358,6 @@ memchr = { version = "2", default-features = false }
 micromarshal = "0.7.0"
 mlua = { version = "0.11", features = ["lua54", "vendored", "async", "serialize"] }
 mockall = "0.11.2"
-murmur3 = "0.5.2"
 mysql_async = { version = "0.34", default-features = false, features = ["native-tls-tls"] }
 naive-cityhash = "0.2.0"
 ndarray = "0.15.6"

Cargo.toml

@@ -41,7 +41,6 @@ jsonb = { workspace = true }
 libm = { workspace = true }
 match-template = { workspace = true }
 md-5 = { workspace = true }
-murmur3 = { workspace = true }
 naive-cityhash = { workspace = true }
 num-traits = { workspace = true }
 proptest = { workspace = true }

src/query/functions/Cargo.toml

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::hash_map::DefaultHasher;
 use std::hash::Hash;
 use std::hash::Hasher;
 
@@ -48,6 +47,7 @@ use md5::Digest;
 use md5::Md5 as Md5Hasher;
 use naive_cityhash::cityhash64_with_seed;
 use num_traits::AsPrimitive;
+use siphasher::sip::SipHasher13;
 use twox_hash::XxHash32;
 use twox_hash::XxHash64;
 
@@ -70,7 +70,7 @@ pub fn register(registry: &mut FunctionRegistry) {
             NumberClass::Decimal128 => {
                 struct Siphash64;
                 impl HashFunction for Siphash64 {
-                    type Hasher = DefaultHasher;
+                    type Hasher = SipHasher13;
                     fn name() -> &'static str {
                         "siphash64"
                     }
@@ -115,9 +115,7 @@ pub fn register(registry: &mut FunctionRegistry) {
             GenericType<0>,
             NumberType<u64>,
         >(|val, output, _| {
-            let mut hasher = DefaultHasher::default();
-            DFHash::hash(&val.to_owned(), &mut hasher);
-            output.push(hasher.finish());
+            output.push(siphash64(&val.to_owned()));
         }))
         .register();
 
@@ -231,7 +229,7 @@ fn register_bucket(registry: &mut FunctionRegistry) {
             |_, _, _| FunctionDomain::MayThrow,
             vectorize_with_builder_2_arg::<NumberType<u64>, StringType, NumberType<i32>>(
                 |buckets, value, output, ctx| {
-                    output.push(bucket(buckets, value.as_bytes(), output.len(), ctx));
+                    output.push(bucket(buckets, value, output.len(), ctx));
                 },
             ),
         );
@@ -241,7 +239,7 @@ fn register_bucket(registry: &mut FunctionRegistry) {
             |_, _, _| FunctionDomain::MayThrow,
             vectorize_with_builder_2_arg::<NumberType<u64>, DateType, NumberType<i32>>(
                 |buckets, value, output, ctx| {
-                    output.push(bucket(buckets, &value.to_le_bytes(), output.len(), ctx));
+                    output.push(bucket(buckets, &value, output.len(), ctx));
                 },
             ),
         );
@@ -256,12 +254,7 @@ fn register_bucket(registry: &mut FunctionRegistry) {
         |_, _, _| FunctionDomain::MayThrow,
         vectorize_with_builder_2_arg::<NumberType<u64>, TimestampType, NumberType<i32>>(
             |buckets, value, output, ctx| {
-                output.push(bucket(
-                    buckets,
-                    &value.to_le_bytes(),
-                    output.len(),
-                    ctx,
-                ));
+                output.push(bucket(buckets, &value, output.len(), ctx));
             },
         ),
     );
@@ -282,13 +275,7 @@ fn register_bucket(registry: &mut FunctionRegistry) {
                         NumberType<NUM_TYPE>,
                         NumberType<i32>,
                     >(|buckets, value, output, ctx| {
-                        let value: i64 = value.as_();
-                        output.push(bucket(
-                            buckets,
-                            &value.to_le_bytes(),
-                            output.len(),
-                            ctx,
-                        ));
+                        output.push(bucket(buckets, &value, output.len(), ctx));
                     }),
                 );
             }
@@ -297,9 +284,9 @@ fn register_bucket(registry: &mut FunctionRegistry) {
     }
 }
 
-fn bucket(
+fn bucket<T: DFHash + ?Sized>(
     buckets: u64,
-    bytes: &[u8],
+    value: &T,
     row: usize,
     ctx: &mut databend_common_expression::EvalContext,
 ) -> i32 {
@@ -312,9 +299,14 @@ fn bucket(
         return 0;
     }
 
-    let mut bytes = bytes;
-    let hash = murmur3::murmur3_32(&mut bytes, 0).unwrap() as i32;
-    (hash & i32::MAX) % buckets
+    (siphash64(value) % buckets as u64) as i32
+}
+
+fn siphash64<T: DFHash + ?Sized>(value: &T) -> u64 {
+    // Keep SQL hash results independent of Rust's unspecified DefaultHasher.
+    let mut hasher = SipHasher13::new_with_keys(0, 0);
+    DFHash::hash(value, &mut hasher);
+    hasher.finish()
 }
 
 fn register_simple_domain_type_hash<T: ArgType>(registry: &mut FunctionRegistry)
@@ -327,9 +319,7 @@ where for<'a> T::ScalarRef<'a>: DFHash {
         .calc_domain(|_, _| FunctionDomain::Full)
         .vectorized(vectorize_with_builder_1_arg::<T, NumberType<u64>>(
             |val, output, _| {
-                let mut hasher = DefaultHasher::default();
-                DFHash::hash(&val, &mut hasher);
-                output.push(hasher.finish());
+                output.push(siphash64(&val));
             },
         ))
         .register();

src/query/functions/src/scalars/hash.rs

@@ -854,37 +854,37 @@ evaluation (internal):
 ast            : bucket(16, 34)
 raw expr       : bucket(16, 34)
 checked expr   : bucket<UInt64, UInt8>(CAST<UInt8>(16_u8 AS UInt64), 34_u8)
-optimized expr : 3_i32
+optimized expr : 5_i32
 output type    : Int32
-output domain  : {3..=3}
-output         : 3
+output domain  : {5..=5}
+output         : 5
 
 
 ast            : bucket(16, 'iceberg')
 raw expr       : bucket(16, 'iceberg')
 checked expr   : bucket<UInt64, String>(CAST<UInt8>(16_u8 AS UInt64), "iceberg")
-optimized expr : 9_i32
+optimized expr : 6_i32
 output type    : Int32
-output domain  : {9..=9}
-output         : 9
+output domain  : {6..=6}
+output         : 6
 
 
 ast            : bucket(16, to_date(10000))
 raw expr       : bucket(16, to_date(10000))
 checked expr   : bucket<UInt64, Date>(CAST<UInt8>(16_u8 AS UInt64), CAST<Int64>(CAST<UInt16>(10000_u16 AS Int64) AS Date))
-optimized expr : 12_i32
+optimized expr : 0_i32
 output type    : Int32
-output domain  : {12..=12}
-output         : 12
+output domain  : {0..=0}
+output         : 0
 
 
 ast            : bucket(16, to_datetime(100000))
 raw expr       : bucket(16, to_datetime(100000))
 checked expr   : bucket<UInt64, Timestamp>(CAST<UInt8>(16_u8 AS UInt64), CAST<Int64>(CAST<UInt32>(100000_u32 AS Int64) AS Timestamp))
-optimized expr : 15_i32
+optimized expr : 11_i32
 output type    : Int32
-output domain  : {15..=15}
-output         : 15
+output domain  : {11..=11}
+output         : 11
 
 
 error: 
@@ -905,17 +905,17 @@ evaluation:
 +--------+---------+---------+
 | Type   | Int64   | Int32   |
 | Domain | {1..=4} | Unknown |
-| Row 0  | 1       | 4       |
-| Row 1  | 2       | 4       |
-| Row 2  | 3       | 3       |
-| Row 3  | 4       | 6       |
+| Row 0  | 1       | 1       |
+| Row 1  | 2       | 6       |
+| Row 2  | 3       | 2       |
+| Row 3  | 4       | 1       |
 +--------+---------+---------+
 evaluation (internal):
 +--------+-----------------------------+
 | Column | Data                        |
 +--------+-----------------------------+
 | a      | Column(Int64([1, 2, 3, 4])) |
-| Output | Int32([4, 4, 3, 6])         |
+| Output | Int32([1, 6, 2, 1])         |
 +--------+-----------------------------+