fix: bloom filter core implementation

- Use uint64_t with bswap64 for portable serialization
- Add input validation for zero expected_elements and invalid FPR
- Fix truncated payload handling in BloomFilter constructor
- Make h2 in get_bit_position key-dependent via rehashing
- Add num_bits_ == 0 guards in insert() and might_contain()

Author: poyrazK

include/network/rpc_message.hpp

@@ -449,8 +449,8 @@ struct BloomFilterArgs {
     std::string probe_table;
     std::string probe_key_col;  // Join key column on probe side for filtering
     std::vector<uint8_t> filter_data;
-    size_t expected_elements;
-    size_t num_hashes;
+    size_t expected_elements = 0;
+    size_t num_hashes = 0;
 
     [[nodiscard]] std::vector<uint8_t> serialize() const {
         std::vector<uint8_t> out;
@@ -466,11 +466,13 @@ struct BloomFilterArgs {
         std::memcpy(out.data() + off, &filter_len, Serializer::VAL_SIZE_32);
         out.insert(out.end(), filter_data.begin(), filter_data.end());
 
-        // Serialize metadata
+        // Serialize metadata using fixed-width temporaries
+        uint64_t tmp_expected = static_cast<uint64_t>(expected_elements);
+        uint8_t tmp_hashes = static_cast<uint8_t>(num_hashes);
         const size_t off2 = out.size();
-        out.resize(off2 + Serializer::VAL_SIZE_32);
-        std::memcpy(out.data() + off2, &expected_elements, Serializer::VAL_SIZE_32);
-        out.push_back(static_cast<uint8_t>(num_hashes));
+        out.resize(off2 + 9);  // 8 bytes for expected_elements + 1 for num_hashes
+        std::memcpy(out.data() + off2, &tmp_expected, 8);
+        out[off2 + 8] = tmp_hashes;
         return out;
     }
 
@@ -493,12 +495,13 @@ struct BloomFilterArgs {
             offset += filter_len;
         }
 
-        if (offset + Serializer::VAL_SIZE_32 <= in.size()) {
-            std::memcpy(&args.expected_elements, in.data() + offset, Serializer::VAL_SIZE_32);
-            offset += Serializer::VAL_SIZE_32;
-        }
-        if (offset < in.size()) {
-            args.num_hashes = in[offset];
+        // Deserialize metadata using fixed-width temporaries
+        if (offset + 9 <= in.size()) {
+            uint64_t tmp_expected = 0;
+            std::memcpy(&tmp_expected, in.data() + offset, 8);
+            args.expected_elements = static_cast<size_t>(tmp_expected);
+            offset += 8;
+            args.num_hashes = static_cast<size_t>(in[offset]);
         }
         return args;
     }

src/common/bloom_filter.cpp

@@ -6,15 +6,36 @@
 #include "common/bloom_filter.hpp"
 
 #include <cmath>
+#include <cstdint>
+
+#if defined(__APPLE__)
+#include <libkern/OSByteOrder.h>
+#define bswap64(x) OSSwapInt64(x)
+#else
+#include <endian.h>
+#define bswap64(x) __builtin_bswap64(x)
+#endif
 
 namespace cloudsql::common {
 
 BloomFilter::BloomFilter(size_t expected_elements, double false_positive_rate)
     : expected_elements_(expected_elements) {
+    // Handle zero expected_elements as empty filter
+    if (expected_elements == 0) {
+        num_bits_ = 0;
+        num_hashes_ = 0;
+        return;
+    }
+
+    // Clamp false_positive_rate to safe range [0.001, 0.99]
+    double p = false_positive_rate;
+    if (p <= 0.0 || p >= 1.0) {
+        p = 0.01;  // Safe default
+    }
+
     // m = -n * ln(p) / (ln(2)^2)
     // k = m/n * ln(2)
     double n = static_cast<double>(expected_elements);
-    double p = false_positive_rate;
 
     double m = -n * std::log(p) / (std::log(2) * std::log(2));
     double k = (m / n) * std::log(2);
@@ -37,25 +58,45 @@ BloomFilter::BloomFilter(size_t expected_elements, double false_positive_rate)
 }
 
 BloomFilter::BloomFilter(const uint8_t* data, size_t size) {
-    if (size < sizeof(size_t) * 3) {
+    // Minimum size: 3 x uint64_t header + at least 1 byte of bits
+    if (size < sizeof(uint64_t) * 3 + 1) {
         return;  // Invalid data
     }
 
     size_t offset = 0;
-    std::memcpy(&num_bits_, data + offset, sizeof(size_t));
-    offset += sizeof(size_t);
-
-    std::memcpy(&num_hashes_, data + offset, sizeof(size_t));
-    offset += sizeof(size_t);
-
-    std::memcpy(&expected_elements_, data + offset, sizeof(size_t));
-    offset += sizeof(size_t);
 
+    // Read with fixed-width uint64_t and proper byte-order conversion
+    uint64_t tmp_num_bits = 0;
+    std::memcpy(&tmp_num_bits, data + offset, sizeof(uint64_t));
+    tmp_num_bits = bswap64(tmp_num_bits);
+    num_bits_ = static_cast<size_t>(tmp_num_bits);
+    offset += sizeof(uint64_t);
+
+    uint64_t tmp_num_hashes = 0;
+    std::memcpy(&tmp_num_hashes, data + offset, sizeof(uint64_t));
+    tmp_num_hashes = bswap64(tmp_num_hashes);
+    num_hashes_ = static_cast<size_t>(tmp_num_hashes);
+    offset += sizeof(uint64_t);
+
+    uint64_t tmp_expected = 0;
+    std::memcpy(&tmp_expected, data + offset, sizeof(uint64_t));
+    tmp_expected = bswap64(tmp_expected);
+    expected_elements_ = static_cast<size_t>(tmp_expected);
+    offset += sizeof(uint64_t);
+
+    // Validate bit array size
     size_t bit_bytes = (num_bits_ + 7) / 8;
-    if (size >= offset + bit_bytes) {
-        bits_.resize(bit_bytes);
-        std::memcpy(bits_.data(), data + offset, bit_bytes);
+    if (size < offset + bit_bytes) {
+        // Truncated payload - reset to safe empty state
+        num_bits_ = 0;
+        num_hashes_ = 0;
+        expected_elements_ = 0;
+        bits_.clear();
+        return;
     }
+
+    bits_.resize(bit_bytes);
+    std::memcpy(bits_.data(), data + offset, bit_bytes);
 }
 
 size_t BloomFilter::murmur3_hash(const Value& key) const {
@@ -84,14 +125,21 @@ size_t BloomFilter::murmur3_hash(const uint8_t* data, size_t len, size_t seed) c
 
 size_t BloomFilter::get_bit_position(size_t hash, size_t i) const {
     // Double hashing technique: h(i) = h1 + i * h2
-    // Use two different hash seeds
+    // Make h2 key-dependent by rehashing the input hash with a different seed
     size_t h1 = hash;
-    size_t h2 = murmur3_hash(reinterpret_cast<const uint8_t*>("salt"), 4, 0xcafebabe);
+    size_t h2 = murmur3_hash(reinterpret_cast<const uint8_t*>(&hash), sizeof(hash), 0xcafebabe);
+
+    // Ensure h2 is non-zero to avoid degenerate probing
+    if (h2 == 0) {
+        h2 = 1;
+    }
 
     return (h1 + i * h2) % num_bits_;
 }
 
 void BloomFilter::insert(const Value& key) {
+    if (num_bits_ == 0) return;  // Empty filter
+
     size_t base_hash = murmur3_hash(key);
 
     for (size_t i = 0; i < num_hashes_; ++i) {
@@ -103,6 +151,8 @@ void BloomFilter::insert(const Value& key) {
 }
 
 bool BloomFilter::might_contain(const Value& key) const {
+    if (num_bits_ == 0) return false;  // Empty filter
+
     size_t base_hash = murmur3_hash(key);
 
     for (size_t i = 0; i < num_hashes_; ++i) {
@@ -121,17 +171,21 @@ bool BloomFilter::might_contain(const Value& key) const {
 std::vector<uint8_t> BloomFilter::serialize() const {
     std::vector<uint8_t> out;
 
-    // Store metadata
-    out.resize(sizeof(size_t) * 3);
+    // Store metadata using fixed-width uint64_t with byte-order conversion
+    out.resize(sizeof(uint64_t) * 3);
     size_t offset = 0;
-    std::memcpy(out.data() + offset, &num_bits_, sizeof(size_t));
-    offset += sizeof(size_t);
 
-    std::memcpy(out.data() + offset, &num_hashes_, sizeof(size_t));
-    offset += sizeof(size_t);
+    uint64_t tmp_num_bits = bswap64(static_cast<uint64_t>(num_bits_));
+    std::memcpy(out.data() + offset, &tmp_num_bits, sizeof(uint64_t));
+    offset += sizeof(uint64_t);
+
+    uint64_t tmp_num_hashes = bswap64(static_cast<uint64_t>(num_hashes_));
+    std::memcpy(out.data() + offset, &tmp_num_hashes, sizeof(uint64_t));
+    offset += sizeof(uint64_t);
 
-    std::memcpy(out.data() + offset, &expected_elements_, sizeof(size_t));
-    offset += sizeof(size_t);
+    uint64_t tmp_expected = bswap64(static_cast<uint64_t>(expected_elements_));
+    std::memcpy(out.data() + offset, &tmp_expected, sizeof(uint64_t));
+    offset += sizeof(uint64_t);
 
     // Store bits
     size_t bit_bytes = (num_bits_ + 7) / 8;