SPARK-47547 BloomFilter fpp degradation: pull up long and byte hashin…

…g to abstract base to make them extensible

common/sketch/src/main/java/org/apache/spark/util/sketch/BloomFilterBase.java

@@ -162,4 +162,23 @@ public long cardinality() {
   protected abstract BloomFilterBase checkCompatibilityForMerge(BloomFilter other)
     throws IncompatibleMergeException;
 
+  public record HiLoHash (int hi, int lo) {}
+
+  protected HiLoHash hashLongToIntPair(long item, int seed) {
+    // Here we first hash the input long element into 2 int hash values, h1 and h2, then produce n
+    // hash values by `h1 + i * h2` with 1 <= i <= numHashFunctions.
+    // Note that `CountMinSketch` use a different strategy, it hash the input long element with
+    // every i to produce n hash values.
+    // TODO: the strategy of `CountMinSketch` looks more advanced, should we follow it here?
+    int h1 = Murmur3_x86_32.hashLong(item, seed);
+    int h2 = Murmur3_x86_32.hashLong(item, h1);
+    return new HiLoHash(h1, h2);
+  }
+
+  protected HiLoHash hashBytesToIntPair(byte[] item, int seed) {
+    int h1 = Murmur3_x86_32.hashUnsafeBytes(item, Platform.BYTE_ARRAY_OFFSET, item.length, seed);
+    int h2 = Murmur3_x86_32.hashUnsafeBytes(item, Platform.BYTE_ARRAY_OFFSET, item.length, h1);
+    return new HiLoHash(h1, h2);
+  }
+
 }

common/sketch/src/main/java/org/apache/spark/util/sketch/BloomFilterImpl.java

@@ -29,8 +29,9 @@ private BloomFilterImpl() {}
 
   @Override
   public boolean putBinary(byte[] item) {
-    int h1 = Murmur3_x86_32.hashUnsafeBytes(item, Platform.BYTE_ARRAY_OFFSET, item.length, 0);
-    int h2 = Murmur3_x86_32.hashUnsafeBytes(item, Platform.BYTE_ARRAY_OFFSET, item.length, h1);
+    HiLoHash hiLoHash = hashBytesToIntPair(item, seed);
+    int h1 = hiLoHash.hi();
+    int h2 = hiLoHash.lo();
 
     long bitSize = bits.bitSize();
     boolean bitsChanged = false;
@@ -47,8 +48,9 @@ public boolean putBinary(byte[] item) {
 
   @Override
   public boolean mightContainBinary(byte[] item) {
-    int h1 = Murmur3_x86_32.hashUnsafeBytes(item, Platform.BYTE_ARRAY_OFFSET, item.length, 0);
-    int h2 = Murmur3_x86_32.hashUnsafeBytes(item, Platform.BYTE_ARRAY_OFFSET, item.length, h1);
+    HiLoHash hiLoHash = hashBytesToIntPair(item, seed);
+    int h1 = hiLoHash.hi();
+    int h2 = hiLoHash.lo();
 
     long bitSize = bits.bitSize();
     for (int i = 1; i <= numHashFunctions; i++) {
@@ -66,13 +68,9 @@ public boolean mightContainBinary(byte[] item) {
 
   @Override
   public boolean putLong(long item) {
-    // Here we first hash the input long element into 2 int hash values, h1 and h2, then produce n
-    // hash values by `h1 + i * h2` with 1 <= i <= numHashFunctions.
-    // Note that `CountMinSketch` use a different strategy, it hash the input long element with
-    // every i to produce n hash values.
-    // TODO: the strategy of `CountMinSketch` looks more advanced, should we follow it here?
-    int h1 = Murmur3_x86_32.hashLong(item, 0);
-    int h2 = Murmur3_x86_32.hashLong(item, h1);
+    HiLoHash hiLoHash = hashLongToIntPair(item, seed);
+    int h1 = hiLoHash.hi();
+    int h2 = hiLoHash.lo();
 
     long bitSize = bits.bitSize();
     boolean bitsChanged = false;
@@ -89,8 +87,9 @@ public boolean putLong(long item) {
 
   @Override
   public boolean mightContainLong(long item) {
-    int h1 = Murmur3_x86_32.hashLong(item, 0);
-    int h2 = Murmur3_x86_32.hashLong(item, h1);
+    HiLoHash hiLoHash = hashLongToIntPair(item, seed);
+    int h1 = hiLoHash.hi();
+    int h2 = hiLoHash.lo();
 
     long bitSize = bits.bitSize();
     for (int i = 1; i <= numHashFunctions; i++) {

common/sketch/src/main/java/org/apache/spark/util/sketch/BloomFilterImplV2.java

@@ -49,8 +49,9 @@ public boolean equals(Object other) {
 
   @Override
   public boolean putBinary(byte[] item) {
-    int h1 = Murmur3_x86_32.hashUnsafeBytes(item, Platform.BYTE_ARRAY_OFFSET, item.length, seed);
-    int h2 = Murmur3_x86_32.hashUnsafeBytes(item, Platform.BYTE_ARRAY_OFFSET, item.length, h1);
+    HiLoHash hiLoHash = hashBytesToIntPair(item, seed);
+    int h1 = hiLoHash.hi();
+    int h2 = hiLoHash.lo();
 
     long bitSize = bits.bitSize();
     boolean bitsChanged = false;
@@ -70,8 +71,9 @@ public boolean putBinary(byte[] item) {
 
   @Override
   public boolean mightContainBinary(byte[] item) {
-    int h1 = Murmur3_x86_32.hashUnsafeBytes(item, Platform.BYTE_ARRAY_OFFSET, item.length, seed);
-    int h2 = Murmur3_x86_32.hashUnsafeBytes(item, Platform.BYTE_ARRAY_OFFSET, item.length, h1);
+    HiLoHash hiLoHash = hashBytesToIntPair(item, seed);
+    int h1 = hiLoHash.hi();
+    int h2 = hiLoHash.lo();
 
     long bitSize = bits.bitSize();
 
@@ -92,13 +94,9 @@ public boolean mightContainBinary(byte[] item) {
 
   @Override
   public boolean putLong(long item) {
-    // Here we first hash the input long element into 2 int hash values, h1 and h2, then produce n
-    // hash values by `h1 + i * h2` with 1 <= i <= numHashFunctions.
-    // Note that `CountMinSketch` use a different strategy, it hash the input long element with
-    // every i to produce n hash values.
-    // TODO: the strategy of `CountMinSketch` looks more advanced, should we follow it here?
-    int h1 = Murmur3_x86_32.hashLong(item, seed);
-    int h2 = Murmur3_x86_32.hashLong(item, h1);
+    HiLoHash hiLoHash = hashLongToIntPair(item, seed);
+    int h1 = hiLoHash.hi();
+    int h2 = hiLoHash.lo();
 
     long bitSize = bits.bitSize();
     boolean bitsChanged = false;
@@ -118,8 +116,9 @@ public boolean putLong(long item) {
 
   @Override
   public boolean mightContainLong(long item) {
-    int h1 = Murmur3_x86_32.hashLong(item, seed);
-    int h2 = Murmur3_x86_32.hashLong(item, h1);
+    HiLoHash hiLoHash = hashLongToIntPair(item, seed);
+    int h1 = hiLoHash.hi();
+    int h2 = hiLoHash.lo();
 
     long bitSize = bits.bitSize();