EIC code displayed by LXR master/​include/​arrow/​acero/​bloom_filter.h	Source navigation Diff markup Identifier search General search
	Version: master test Revert   Change

File indexing completed on 2025-08-28 08:26:52

 // Licensed to the Apache Software Foundation (ASF) under one
 // or more contributor license agreements.  See the NOTICE file
 // distributed with this work for additional information
 // regarding copyright ownership.  The ASF licenses this file
 // to you under the Apache License, Version 2.0 (the
 // "License"); you may not use this file except in compliance
 // with the License.  You may obtain a copy of the License at
 //
 //   http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing,
 // software distributed under the License is distributed on an
 // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.
 
 #pragma once
 
 #include <atomic>
 #include <cstdint>
 #include <memory>
 
 #include "arrow/acero/partition_util.h"
 #include "arrow/acero/util.h"
 #include "arrow/memory_pool.h"
 #include "arrow/result.h"
 #include "arrow/status.h"
 #include "arrow/util/simd.h"
 
 namespace arrow {
 namespace acero {
 
 // A set of pre-generated bit masks from a 64-bit word.
 //
 // It is used to map selected bits of hash to a bit mask that will be used in
 // a Bloom filter.
 //
 // These bit masks need to look random and need to have a similar fractions of
 // bits set in order for a Bloom filter to have a low false positives rate.
 //
 struct ARROW_ACERO_EXPORT BloomFilterMasks {
   // Generate all masks as a single bit vector. Each bit offset in this bit
   // vector corresponds to a single mask.
   // In each consecutive kBitsPerMask bits, there must be between
   // kMinBitsSet and kMaxBitsSet bits set.
   //
   BloomFilterMasks();
 
   inline uint64_t mask(int bit_offset) {
 #if ARROW_LITTLE_ENDIAN
     return (arrow::util::SafeLoadAs<uint64_t>(masks_ + bit_offset / 8) >>
             (bit_offset % 8)) &
            kFullMask;
 #else
     return (BYTESWAP(arrow::util::SafeLoadAs<uint64_t>(masks_ + bit_offset / 8)) >>
             (bit_offset % 8)) &
            kFullMask;
 #endif
   }
 
   // Masks are 57 bits long because then they can be accessed at an
   // arbitrary bit offset using a single unaligned 64-bit load instruction.
   //
   static constexpr int kBitsPerMask = 57;
   static constexpr uint64_t kFullMask = (1ULL << kBitsPerMask) - 1;
 
   // Minimum and maximum number of bits set in each mask.
   // This constraint is enforced when generating the bit masks.
   // Values should be close to each other and chosen as to minimize a Bloom
   // filter false positives rate.
   //
   static constexpr int kMinBitsSet = 4;
   static constexpr int kMaxBitsSet = 5;
 
   // Number of generated masks.
   // Having more masks to choose will improve false positives rate of Bloom
   // filter but will also use more memory, which may lead to more CPU cache
   // misses.
   // The chosen value results in using only a few cache-lines for mask lookups,
   // while providing a good variety of available bit masks.
   //
   static constexpr int kLogNumMasks = 10;
   static constexpr int kNumMasks = 1 << kLogNumMasks;
 
   // Data of masks. Masks are stored in a single bit vector. Nth mask is
   // kBitsPerMask bits starting at bit offset N.
   //
   static constexpr int kTotalBytes = (kNumMasks + 64) / 8;
   uint8_t masks_[kTotalBytes];
 };
 
 // A variant of a blocked Bloom filter implementation.
 // A Bloom filter is a data structure that provides approximate membership test
 // functionality based only on the hash of the key. Membership test may return
 // false positives but not false negatives. Approximation of the result allows
 // in general case (for arbitrary data types of keys) to save on both memory and
 // lookup cost compared to the accurate membership test.
 // The accurate test may sometimes still be cheaper for a specific data types
 // and inputs, e.g. integers from a small range.
 //
 // This blocked Bloom filter is optimized for use in hash joins, to achieve a
 // good balance between the size of the filter, the cost of its building and
 // querying and the rate of false positives.
 //
 class ARROW_ACERO_EXPORT BlockedBloomFilter {
   friend class BloomFilterBuilder_SingleThreaded;
   friend class BloomFilterBuilder_Parallel;
 
  public:
   BlockedBloomFilter() : log_num_blocks_(0), num_blocks_(0), blocks_(NULLPTR) {}
 
   inline bool Find(uint64_t hash) const {
     uint64_t m = mask(hash);
     uint64_t b = blocks_[block_id(hash)];
     return (b & m) == m;
   }
 
   // Uses SIMD if available for smaller Bloom filters.
   // Uses memory prefetching for larger Bloom filters.
   //
   void Find(int64_t hardware_flags, int64_t num_rows, const uint32_t* hashes,
             uint8_t* result_bit_vector, bool enable_prefetch = true) const;
   void Find(int64_t hardware_flags, int64_t num_rows, const uint64_t* hashes,
             uint8_t* result_bit_vector, bool enable_prefetch = true) const;
 
   int log_num_blocks() const { return log_num_blocks_; }
 
   int NumHashBitsUsed() const;
 
   bool IsSameAs(const BlockedBloomFilter* other) const;
 
   int64_t NumBitsSet() const;
 
   // Folding of a block Bloom filter after the initial version
   // has been built.
   //
   // One of the parameters for creation of Bloom filter is the number
   // of bits allocated for it. The more bits allocated, the lower the
   // probability of false positives. A good heuristic is to aim for
   // half of the bits set in the constructed Bloom filter. This should
   // result in a good trade off between size (and following cost of
   // memory accesses) and false positives rate.
   //
   // There might have been many duplicate keys in the input provided
   // to Bloom filter builder. In that case the resulting bit vector
   // would be more sparse then originally intended. It is possible to
   // easily correct that and cut in half the size of Bloom filter
   // after it has already been constructed. The process to do that is
   // approximately equal to OR-ing bits from upper and lower half (the
   // way we address these bits when inserting or querying a hash makes
   // such folding in half possible).
   //
   // We will keep folding as long as the fraction of bits set is less
   // than 1/4. The resulting bit vector density should be in the [1/4,
   // 1/2) range.
   //
   void Fold();
 
  private:
   Status CreateEmpty(int64_t num_rows_to_insert, MemoryPool* pool);
 
   inline void Insert(uint64_t hash) {
     uint64_t m = mask(hash);
     uint64_t& b = blocks_[block_id(hash)];
     b |= m;
   }
 
   void Insert(int64_t hardware_flags, int64_t num_rows, const uint32_t* hashes);
   void Insert(int64_t hardware_flags, int64_t num_rows, const uint64_t* hashes);
 
   inline uint64_t mask(uint64_t hash) const {
     // The lowest bits of hash are used to pick mask index.
     //
     int mask_id = static_cast<int>(hash & (BloomFilterMasks::kNumMasks - 1));
     uint64_t result = masks_.mask(mask_id);
 
     // The next set of hash bits is used to pick the amount of bit
     // rotation of the mask.
     //
     int rotation = (hash >> BloomFilterMasks::kLogNumMasks) & 63;
     result = ROTL64(result, rotation);
 
     return result;
   }
 
   inline int64_t block_id(uint64_t hash) const {
     // The next set of hash bits following the bits used to select a
     // mask is used to pick block id (index of 64-bit word in a bit
     // vector).
     //
     return (hash >> (BloomFilterMasks::kLogNumMasks + 6)) & (num_blocks_ - 1);
   }
 
   template <typename T>
   inline void InsertImp(int64_t num_rows, const T* hashes);
 
   template <typename T>
   inline void FindImp(int64_t num_rows, const T* hashes, uint8_t* result_bit_vector,
                       bool enable_prefetch) const;
 
   void SingleFold(int num_folds);
 
 #if defined(ARROW_HAVE_RUNTIME_AVX2)
   inline __m256i mask_avx2(__m256i hash) const;
   inline __m256i block_id_avx2(__m256i hash) const;
   int64_t Insert_avx2(int64_t num_rows, const uint32_t* hashes);
   int64_t Insert_avx2(int64_t num_rows, const uint64_t* hashes);
   template <typename T>
   int64_t InsertImp_avx2(int64_t num_rows, const T* hashes);
   int64_t Find_avx2(int64_t num_rows, const uint32_t* hashes,
                     uint8_t* result_bit_vector) const;
   int64_t Find_avx2(int64_t num_rows, const uint64_t* hashes,
                     uint8_t* result_bit_vector) const;
   template <typename T>
   int64_t FindImp_avx2(int64_t num_rows, const T* hashes,
                        uint8_t* result_bit_vector) const;
 #endif
 
   bool UsePrefetch() const {
     return num_blocks_ * sizeof(uint64_t) > kPrefetchLimitBytes;
   }
 
   static constexpr int64_t kPrefetchLimitBytes = 256 * 1024;
 
   static BloomFilterMasks masks_;
 
   // Total number of bits used by block Bloom filter must be a power
   // of 2.
   //
   int log_num_blocks_;
   int64_t num_blocks_;
 
   // Buffer allocated to store an array of power of 2 64-bit blocks.
   //
   std::shared_ptr<Buffer> buf_;
   // Pointer to mutable data owned by Buffer
   //
   uint64_t* blocks_;
 };
 
 // We have two separate implementations of building a Bloom filter, multi-threaded and
 // single-threaded.
 //
 // Single threaded version is useful in two ways:
 // a) It allows to verify parallel implementation in tests (the single threaded one is
 // simpler and can be used as the source of truth).
 // b) It is preferred for small and medium size Bloom filters, because it skips extra
 // synchronization related steps from parallel variant (partitioning and taking locks).
 //
 enum class BloomFilterBuildStrategy {
   SINGLE_THREADED = 0,
   PARALLEL = 1,
 };
 
 class ARROW_ACERO_EXPORT BloomFilterBuilder {
  public:
   virtual ~BloomFilterBuilder() = default;
   virtual Status Begin(size_t num_threads, int64_t hardware_flags, MemoryPool* pool,
                        int64_t num_rows, int64_t num_batches,
                        BlockedBloomFilter* build_target) = 0;
   virtual int64_t num_tasks() const { return 0; }
   virtual Status PushNextBatch(size_t thread_index, int64_t num_rows,
                                const uint32_t* hashes) = 0;
   virtual Status PushNextBatch(size_t thread_index, int64_t num_rows,
                                const uint64_t* hashes) = 0;
   virtual void CleanUp() {}
   static std::unique_ptr<BloomFilterBuilder> Make(BloomFilterBuildStrategy strategy);
 };
 
 class ARROW_ACERO_EXPORT BloomFilterBuilder_SingleThreaded : public BloomFilterBuilder {
  public:
   Status Begin(size_t num_threads, int64_t hardware_flags, MemoryPool* pool,
                int64_t num_rows, int64_t num_batches,
                BlockedBloomFilter* build_target) override;
 
   Status PushNextBatch(size_t /*thread_index*/, int64_t num_rows,
                        const uint32_t* hashes) override;
 
   Status PushNextBatch(size_t /*thread_index*/, int64_t num_rows,
                        const uint64_t* hashes) override;
 
  private:
   template <typename T>
   void PushNextBatchImp(int64_t num_rows, const T* hashes);
 
   int64_t hardware_flags_;
   BlockedBloomFilter* build_target_;
 };
 
 class ARROW_ACERO_EXPORT BloomFilterBuilder_Parallel : public BloomFilterBuilder {
  public:
   Status Begin(size_t num_threads, int64_t hardware_flags, MemoryPool* pool,
                int64_t num_rows, int64_t num_batches,
                BlockedBloomFilter* build_target) override;
 
   Status PushNextBatch(size_t thread_id, int64_t num_rows,
                        const uint32_t* hashes) override;
 
   Status PushNextBatch(size_t thread_id, int64_t num_rows,
                        const uint64_t* hashes) override;
 
   void CleanUp() override;
 
  private:
   template <typename T>
   void PushNextBatchImp(size_t thread_id, int64_t num_rows, const T* hashes);
 
   int64_t hardware_flags_;
   BlockedBloomFilter* build_target_;
   int log_num_prtns_;
   struct ThreadLocalState {
     std::vector<uint32_t> partitioned_hashes_32;
     std::vector<uint64_t> partitioned_hashes_64;
     std::vector<uint16_t> partition_ranges;
     std::vector<int> unprocessed_partition_ids;
   };
   std::vector<ThreadLocalState> thread_local_states_;
   PartitionLocks prtn_locks_;
 };
 
 }  // namespace acero
 }  // namespace arrow

This page was automatically generated by the 2.3.7 LXR engine. The LXR team