split_block.hpp

Go to the documentation of this file.

// SPDX-License-Identifier: AGPL-3.0-or-later
// Copyright 2026 Johnson Ogundeji
#pragma once
 
 
#include "signet/bloom/xxhash.hpp"
 
#include <algorithm>
#include <cassert>
#include <cmath>
#include <cstdint>
#include <cstring>
#include <stdexcept>
#include <string>
#include <vector>
 
namespace signet::forge {
 
class SplitBlockBloomFilter {
public:
    static constexpr size_t kBytesPerBlock = 32;
 
    static constexpr size_t kWordsPerBlock = 8;
 
    static constexpr size_t kMinBytes = kBytesPerBlock;
 
    static constexpr size_t kMaxBytes = 128ULL * 1024 * 1024;
 
    // -----------------------------------------------------------------------
    // Construction
    // -----------------------------------------------------------------------
 
    explicit SplitBlockBloomFilter(size_t num_distinct_values,
                                    double fpr = 0.01)
    {
        if (fpr <= 0.0 || fpr >= 1.0) {
            throw std::invalid_argument(
                "SplitBlockBloomFilter: fpr must be in (0, 1)");
        }
        if (num_distinct_values == 0) {
            num_distinct_values = 1; // at least one block
        }
 
        // Optimal byte count: -8 * ndv / ln(1 - fpr^(1/8))
        double exponent  = 1.0 / 8.0;
        double fpr_root  = std::pow(fpr, exponent);
        double denom     = std::log(1.0 - fpr_root);
        double raw_bytes = -8.0 * static_cast<double>(num_distinct_values) / denom;
 
        // Round up to multiple of 32, clamp to [kMinBytes, kMaxBytes]
        auto num_bytes = static_cast<size_t>(std::ceil(raw_bytes));
        num_bytes = round_up_to_block(num_bytes);
        num_bytes = (std::max)(num_bytes, kMinBytes);
        num_bytes = (std::min)(num_bytes, kMaxBytes);
 
        data_.resize(num_bytes, 0);
    }
 
    static SplitBlockBloomFilter with_size(size_t num_bytes) {
        if (num_bytes == 0 || (num_bytes % kBytesPerBlock) != 0) {
            throw std::invalid_argument(
                "SplitBlockBloomFilter::with_size: "
                "num_bytes must be a positive multiple of 32");
        }
        num_bytes = (std::min)(num_bytes, kMaxBytes);
        SplitBlockBloomFilter f;
        f.data_.resize(num_bytes, 0);
        return f;
    }
 
    // -----------------------------------------------------------------------
    // Core operations
    // -----------------------------------------------------------------------
 
    void insert(uint64_t hash) {
        const size_t nblocks = num_blocks();
        // Block index: use upper 32 bits to select the block
        // Invariant: nblocks > 0 (enforced by constructor min 1 block)
        const auto block_idx = static_cast<size_t>(
            (static_cast<uint64_t>(hash >> 32) * nblocks) >> 32);
        assert(block_idx < nblocks);
 
        const auto key = static_cast<uint32_t>(hash);
 
        for (size_t i = 0; i < kWordsPerBlock; ++i) {
            uint32_t bit_pos = (key * kSalt[i]) >> 27;  // top 5 bits
            uint32_t word = block_read(block_idx, i);
            word |= (UINT32_C(1) << bit_pos);
            block_write(block_idx, i, word);
        }
    }
 
    [[nodiscard]] bool might_contain(uint64_t hash) const {
        const size_t nblocks = num_blocks();
        const auto block_idx = static_cast<size_t>(
            (static_cast<uint64_t>(hash >> 32) * nblocks) >> 32);
        assert(block_idx < nblocks);
 
        const auto key = static_cast<uint32_t>(hash);
 
        for (size_t i = 0; i < kWordsPerBlock; ++i) {
            uint32_t bit_pos = (key * kSalt[i]) >> 27;
            uint32_t word = block_read(block_idx, i);
            if ((word & (UINT32_C(1) << bit_pos)) == 0) {
                return false;
            }
        }
        return true;
    }
 
    // -----------------------------------------------------------------------
    // Convenience: insert/check typed values using xxHash64
    // -----------------------------------------------------------------------
 
    void insert_value(const std::string& val) {
        insert(xxhash::hash64(val));
    }
    void insert_value(int32_t val) {
        insert(xxhash::hash64_value(val));
    }
    void insert_value(int64_t val) {
        insert(xxhash::hash64_value(val));
    }
    void insert_value(float val) {
        insert(xxhash::hash64_value(val));
    }
    void insert_value(double val) {
        insert(xxhash::hash64_value(val));
    }
 
    [[nodiscard]] bool might_contain_value(const std::string& val) const {
        return might_contain(xxhash::hash64(val));
    }
    [[nodiscard]] bool might_contain_value(int32_t val) const {
        return might_contain(xxhash::hash64_value(val));
    }
    [[nodiscard]] bool might_contain_value(int64_t val) const {
        return might_contain(xxhash::hash64_value(val));
    }
    [[nodiscard]] bool might_contain_value(float val) const {
        return might_contain(xxhash::hash64_value(val));
    }
    [[nodiscard]] bool might_contain_value(double val) const {
        return might_contain(xxhash::hash64_value(val));
    }
 
    // -----------------------------------------------------------------------
    // Serialization / deserialization
    // -----------------------------------------------------------------------
 
    [[nodiscard]] const std::vector<uint8_t>& data() const { return data_; }
 
    [[nodiscard]] size_t size_bytes() const { return data_.size(); }
 
    [[nodiscard]] size_t num_blocks() const {
        return data_.size() / kBytesPerBlock;
    }
 
    static SplitBlockBloomFilter from_data(const uint8_t* src, size_t size) {
        if (size == 0 || (size % kBytesPerBlock) != 0) {
            throw std::invalid_argument(
                "SplitBlockBloomFilter::from_data: "
                "size must be a positive multiple of 32");
        }
        // CWE-400: Uncontrolled Resource Consumption — kMaxBytes (128 MiB) cap
        if (size > kMaxBytes) {
            throw std::invalid_argument(
                "SplitBlockBloomFilter::from_data: "
                "data exceeds maximum filter size");
        }
        SplitBlockBloomFilter f;
        f.data_.assign(src, src + size);
        return f;
    }
 
    // -----------------------------------------------------------------------
    // Utilities
    // -----------------------------------------------------------------------
 
    void merge(const SplitBlockBloomFilter& other) {
        if (data_.size() != other.data_.size()) {
            throw std::invalid_argument(
                "SplitBlockBloomFilter::merge: filter sizes must match");
        }
        for (size_t i = 0; i < data_.size(); ++i) {
            data_[i] |= other.data_[i];
        }
    }
 
    void reset() {
        std::fill(data_.begin(), data_.end(), uint8_t{0});
    }
 
    [[nodiscard]] double estimated_fpr(size_t num_insertions) const {
        if (num_insertions == 0) return 0.0;
 
        const double k = 8.0;
        const double m = static_cast<double>(num_blocks()) * 256.0; // total bits
        const double n = static_cast<double>(num_insertions);
 
        // Probability that a single bit is set after n insertions
        double p = 1.0 - std::exp(-k * n / m);
        // FPR = probability all k bits match
        return std::pow(p, k);
    }
 
private:
    SplitBlockBloomFilter() = default;
 
    static constexpr uint64_t kBloomSeed = 0;
    static_assert(kBloomSeed == 0, "Parquet spec requires bloom filter seed = 0");
 
    static constexpr uint32_t kSalt[kWordsPerBlock] = {
        0x47b6137bU,
        0x44974d91U,
        0x8824ad5bU,
        0xa2b7289dU,
        0x705495c7U,
        0x2df1424bU,
        0x9efc4947U,
        0x5c6bfb31U
    };
 
    // -----------------------------------------------------------------------
    // Internal helpers
    // -----------------------------------------------------------------------
 
    static constexpr size_t round_up_to_block(size_t bytes) {
        return ((bytes + kBytesPerBlock - 1) / kBytesPerBlock) * kBytesPerBlock;
    }
 
    uint32_t block_read(size_t block_idx, size_t word_idx) const {
        uint32_t val;
        std::memcpy(&val, data_.data() + block_idx * kBytesPerBlock + word_idx * sizeof(uint32_t), sizeof(uint32_t));
        return val;
    }
 
    void block_write(size_t block_idx, size_t word_idx, uint32_t val) {
        std::memcpy(data_.data() + block_idx * kBytesPerBlock + word_idx * sizeof(uint32_t), &val, sizeof(uint32_t));
    }
 
    [[deprecated("use block_read/block_write")]]
    uint32_t* block_ptr(size_t idx) {
        return reinterpret_cast<uint32_t*>(data_.data() + idx * kBytesPerBlock);
    }
 
    [[deprecated("use block_read/block_write")]]
    const uint32_t* block_ptr(size_t idx) const {
        return reinterpret_cast<const uint32_t*>(
            data_.data() + idx * kBytesPerBlock);
    }
 
    std::vector<uint8_t> data_;
};
 
} // namespace signet::forge