fix: per-context bloom filter metadata to avoid race conditions

- Store expected_elements and num_hashes in per-context maps instead of globals
- Update set_local_bloom_bits to write to per-context maps
- Update getters to take context_id parameter
- clear_bloom_filter now also erases local_bloom_bits and metadata maps
- Update BloomFilterBits handler to use per-context getters

Author: poyrazK

include/common/cluster_manager.hpp

@@ -286,8 +286,8 @@ class ClusterManager {
                               size_t expected_elements, size_t num_hashes) {
         const std::scoped_lock<std::mutex> lock(mutex_);
         local_bloom_bits_[context_id] = std::move(bits);
-        local_expected_elements_ = expected_elements;
-        local_num_hashes_ = num_hashes;
+        local_expected_elements_map_[context_id] = expected_elements;
+        local_num_hashes_map_[context_id] = num_hashes;
     }
 
     /**
@@ -305,17 +305,25 @@ class ClusterManager {
     /**
      * @brief Get expected_elements for local bloom filter
      */
-    [[nodiscard]] size_t get_local_expected_elements() const {
+    [[nodiscard]] size_t get_local_expected_elements(const std::string& context_id) const {
         const std::scoped_lock<std::mutex> lock(mutex_);
-        return local_expected_elements_;
+        auto it = local_expected_elements_map_.find(context_id);
+        if (it != local_expected_elements_map_.end()) {
+            return it->second;
+        }
+        return 0;
     }
 
     /**
      * @brief Get num_hashes for local bloom filter
      */
-    [[nodiscard]] size_t get_local_num_hashes() const {
+    [[nodiscard]] size_t get_local_num_hashes(const std::string& context_id) const {
         const std::scoped_lock<std::mutex> lock(mutex_);
-        return local_num_hashes_;
+        auto it = local_num_hashes_map_.find(context_id);
+        if (it != local_num_hashes_map_.end()) {
+            return it->second;
+        }
+        return 0;
     }
 
     /**
@@ -324,6 +332,9 @@ class ClusterManager {
     void clear_bloom_filter(const std::string& context_id) {
         const std::scoped_lock<std::mutex> lock(mutex_);
         bloom_filters_.erase(context_id);
+        local_bloom_bits_.erase(context_id);
+        local_expected_elements_map_.erase(context_id);
+        local_num_hashes_map_.erase(context_id);
     }
 
    private:
@@ -352,8 +363,8 @@ class ClusterManager {
     std::unordered_map<std::string, BloomFilterEntry> bloom_filters_;
     /* context_id -> local bloom filter bits (for aggregation during distributed build) */
     std::unordered_map<std::string, std::vector<uint8_t>> local_bloom_bits_;
-    size_t local_expected_elements_ = 0;
-    size_t local_num_hashes_ = 0;
+    std::unordered_map<std::string, size_t> local_expected_elements_map_;
+    std::unordered_map<std::string, size_t> local_num_hashes_map_;
     mutable std::mutex mutex_;
 };
 

src/main.cpp

@@ -529,8 +529,8 @@ int main(int argc, char* argv[]) {
                         reply_args.filter_data =
                             cluster_manager->get_local_bloom_bits(args.context_id);
                         reply_args.expected_elements =
-                            cluster_manager->get_local_expected_elements();
-                        reply_args.num_hashes = cluster_manager->get_local_num_hashes();
+                            cluster_manager->get_local_expected_elements(args.context_id);
+                        reply_args.num_hashes = cluster_manager->get_local_num_hashes(args.context_id);
 
                         auto resp_p = reply_args.serialize();
                         cloudsql::network::RpcHeader resp_h;