Feat/s3j adaptive join

config/integration_test_cases.toml

@@ -503,8 +503,9 @@ name = "s3j_adaptive"
 description = "DEBS'23 S3J with adaptive clustering enabled"
 algorithm = "s3j"
 partition_strategy = "centroid"
-window_state_type = "partitioned"
+window_state_type = "partitioned_vector"
 index_strategy = "partitioned"
+alpha = 5.0
 s3j_num_centroids = 16
 s3j_enable_adaptive = true
 s3j_adapt_interval_ms = 1000
@@ -514,32 +515,34 @@ ivf_nprobes = 5
 data_sizes = [500, 1000]
 parallelism = [2, 4]
 expected_min_recall = 0.80
-enabled = false
+enabled = true
 
 [[test_case]]
 name = "s3j_static"
 description = "S3J with static centroid configuration"
 algorithm = "s3j"
 partition_strategy = "centroid"
-window_state_type = "partitioned"
+window_state_type = "partitioned_vector"
 index_strategy = "partitioned"
+alpha = 5.0
 s3j_num_centroids = 16
 s3j_enable_adaptive = false
 ivf_nlist = 50
 ivf_nprobes = 5
 data_sizes = [500]
 parallelism = [2, 4]
 expected_min_recall = 0.75
-enabled = false
+enabled = true
 
 [[test_case]]
 name = "s3j_high_centroids"
 description = "S3J with more centroids for finer partitioning"
 algorithm = "s3j"
 partition_strategy = "centroid"
-window_state_type = "partitioned"
+window_state_type = "partitioned_vector"
 index_strategy = "partitioned"
-s3j_num_centroids = 32
+alpha = 5.0
+    s3j_num_centroids = 32
 s3j_enable_adaptive = true
 s3j_adapt_interval_ms = 500
 s3j_load_threshold = 0.2
@@ -548,7 +551,7 @@ ivf_nprobes = 8
 data_sizes = [1000]
 parallelism = [4, 8]
 expected_min_recall = 0.85
-enabled = false
+enabled = true
 
 # ==================== VSJoin 测试 ====================
 # Full VSJoin with LSH partitioning

config/perf_join_datasource_modes.toml

@@ -1,63 +1,176 @@
+[log]
+level = "info"
+
+# ==================== S3J 多并行度性能测试 ====================
+# 测试 S3J 在不同并行度下的性能表现
+
+# 小规模基准 (500 条) - parallelism=1 保持召回
 [[performance_test]]
-# CI Performance Test: Generate-Direct-Use (no file I/O for faster CI execution)
-# Using smaller dataset size optimized for CI environment
-name = "ci_perf_join_parallel_test"
+name = "s3j_small"
 mode = "generate_direct_use"
-methods = ["bruteforce", "ivf", "hdr_tree"]
-sizes = [1000]  # Smaller size for CI - fast execution while testing parallelism
-records_count = 1000
-vector_dim = 64
-parallelism = [1, 2, 4, 8, 16, 32]  # Test all parallelism levels from 1 to 32
-window_time_ms = [10000]
+methods = ["s3j"]
+sizes = [500]
+records_count = 500
+vector_dim = 128
+parallelism = [1]
+window_time_ms = [5000]
 window_trigger_ms = 50
 time_interval = 10
 similarity_threshold = 0.8
 seed = 42
 
-# HDR-Tree specific parameters
-[performance_test.hdr_tree_params]
-projected_dim = 8
-max_node_size = 100
-delta_buffer_size = 1000
-pca_sample_size = 3000  # Smaller for CI
-
-[performance_test.data_source]
-type = "random"
+[performance_test.clustered_join_params]
+index_type = "bruteforce"
+training_samples = 250
+multicast_enabled = 1
+overlap_ratio = 0.1
 
-log.level = "info"
+[performance_test.s3j_params]
+num_centroids = 4
+enable_adaptive = 0
+adapt_interval_ms = 1000
+load_threshold = 0.2
+multicast_k = 1
 
+# 中规模多并行度测试 (1000 条) - 低并行度保证召回
 [[performance_test]]
-# CI Performance Test for HDR_Tree
-name = "ci_perf_join_hdrtree_test"
+name = "s3j_medium_low_par"
 mode = "generate_direct_use"
-methods = ["hdrtree"]
+methods = ["s3j"]
 sizes = [1000]
 records_count = 1000
-vector_dim = 64
-parallelism = [1, 4]
-window_time_ms = [10000]
+vector_dim = 128
+parallelism = [1, 2, 4]
+window_time_ms = [15000]
 window_trigger_ms = 50
 time_interval = 10
 similarity_threshold = 0.8
 seed = 42
 
-[performance_test.data_source]
-type = "random"
+[performance_test.clustered_join_params]
+index_type = "bruteforce"
+training_samples = 500
+multicast_enabled = 1
+overlap_ratio = 0.1
+
+[performance_test.s3j_params]
+num_centroids = 8
+enable_adaptive = 0
+adapt_interval_ms = 1000
+load_threshold = 0.2
+multicast_k = 2
 
-# LSH performance quick pass
+# 中规模高并行度测试 - 分层测试策略
+# p8-10: 1000条 (验证功能)
+# p12-16: 500条 (验证高并行度场景)
 [[performance_test]]
-name = "perf_join_lsh_random"
+name = "s3j_medium_high_par"
 mode = "generate_direct_use"
-methods = ["lsh"]
+methods = ["s3j"]
 sizes = [1000]
 records_count = 1000
-vector_dim = 64
+vector_dim = 128
+parallelism = [8, 10]
+window_time_ms = [120000]
+window_trigger_ms = 200
+time_interval = 20
+similarity_threshold = 0.8
+seed = 42
+
+[performance_test.clustered_join_params]
+index_type = "bruteforce"
+training_samples = 500
+multicast_enabled = 1
+overlap_ratio = 0.2
+
+[performance_test.s3j_params]
+num_centroids = 16
+enable_adaptive = 0
+adapt_interval_ms = 2000
+load_threshold = 0.3
+multicast_k = 2
+
+# 高并行度测试 - 小规模数据验证扩展性
+[[performance_test]]
+name = "s3j_scalability_high_par"
+mode = "generate_direct_use"
+methods = ["s3j"]
+sizes = [500]
+records_count = 500
+vector_dim = 128
+parallelism = [12, 16]
+window_time_ms = [180000]
+window_trigger_ms = 300
+time_interval = 30
+similarity_threshold = 0.8
+seed = 42
+
+[performance_test.clustered_join_params]
+index_type = "bruteforce"
+training_samples = 250
+multicast_enabled = 1
+overlap_ratio = 0.2
+
+[performance_test.s3j_params]
+num_centroids = 16
+enable_adaptive = 0
+adapt_interval_ms = 2000
+load_threshold = 0.3
+multicast_k = 2
+
+# 大规模验证
+[[performance_test]]
+name = "s3j_large"
+mode = "generate_direct_use"
+methods = ["s3j"]
+sizes = [2000]
+records_count = 2000
+vector_dim = 128
 parallelism = [1, 2, 4]
-window_time_ms = [10000]
-window_trigger_ms = 50
+window_time_ms = [30000]
+window_trigger_ms = 100
+time_interval = 10
+similarity_threshold = 0.8
+seed = 42
+
+[performance_test.clustered_join_params]
+index_type = "bruteforce"
+training_samples = 1000
+multicast_enabled = 1
+overlap_ratio = 0.1
+
+[performance_test.s3j_params]
+num_centroids = 16
+enable_adaptive = 0
+adapt_interval_ms = 1000
+load_threshold = 0.2
+multicast_k = 2
+
+# 自适应负载均衡测试 - 验证 greedy balancing 被调用
+# 大规模数据 + 高并行度，充分测试贪心策略
+[[performance_test]]
+name = "s3j_adaptive_test"
+mode = "generate_direct_use"
+methods = ["s3j"]
+sizes = [1000]
+records_count = 1000
+vector_dim = 128
+parallelism = [4, 8]
+window_time_ms = [30000]
+window_trigger_ms = 100
 time_interval = 10
 similarity_threshold = 0.8
 seed = 42
 
-[performance_test.data_source]
-type = "random"
+[performance_test.clustered_join_params]
+index_type = "bruteforce"
+training_samples = 500
+multicast_enabled = 1
+overlap_ratio = 0.15
+
+[performance_test.s3j_params]
+num_centroids = 16
+enable_adaptive = 1
+adapt_interval_ms = 500
+load_threshold = 0.15
+multicast_k = 2

config/perf_join_s3j_skew.toml

@@ -0,0 +1,16 @@
+[[performance_test]]
+name = "s3j_skew_benchmark"
+mode = "generate_direct_use"
+methods = ["s3j"]
+sizes = [2000]
+parallelism = [1]
+window_time_ms = [5000]
+similarity_threshold = 0.8
+vector_dim = 128
+seed = 42
+
+[performance_test.data_source]
+type = "skewed"
+num_clusters = 50
+zipf_skew = 1.2
+cluster_spread = 0.05

include/concurrency/blank_controller.h

@@ -1,4 +1,6 @@
 #include <memory>
+#include <shared_mutex>
+#include <unordered_set>
 
 #include "concurrency/concurrency_controller.h"
 #include "index/index.h"
@@ -32,5 +34,7 @@ class BlankController final : public ConcurrencyController {
 
  private:
   std::shared_ptr<Index> index_;
+  std::unordered_set<uint64_t> local_uids_;
+  mutable std::shared_mutex local_uids_mutex_;
 };
 }  // namespace sageFlow

include/coordination/workset_directory.h

@@ -0,0 +1,82 @@
+#pragma once
+
+#include <cstdint>
+#include <optional>
+#include <unordered_map>
+#include <shared_mutex>
+#include <mutex>
+#include <vector>
+#include <atomic>
+
+namespace sageFlow {
+
+struct WorksetProfile {
+    uint64_t id;
+    int owner;
+    double load;
+};
+
+/**
+ * @brief Interface for managing Workset ownership and Load info.
+ */
+class WorksetDirectory {
+public:
+    virtual ~WorksetDirectory() = default;
+
+    virtual std::optional<int> getOwner(uint64_t workset_id) const = 0;
+    virtual void setOwner(uint64_t workset_id, int worker_id) = 0;
+
+    // Report load for a specific workset (accumulative or absolute? let's say absolute rate)
+    virtual void reportWorksetLoad(uint64_t workset_id, double load) = 0;
+
+    // Get global view for rebalancing
+    virtual std::vector<WorksetProfile> getAllWorksetProfiles() const = 0;
+};
+
+class LocalWorksetDirectory : public WorksetDirectory {
+public:
+    std::optional<int> getOwner(uint64_t workset_id) const override {
+        std::shared_lock<std::shared_mutex> lock(mutex_);
+        auto it = owners_.find(workset_id);
+        if (it != owners_.end()) {
+            return it->second;
+        }
+        return std::nullopt;
+    }
+
+    void setOwner(uint64_t workset_id, int worker_id) override {
+        std::unique_lock<std::shared_mutex> lock(mutex_);
+        owners_[workset_id] = worker_id;
+    }
+
+    void reportWorksetLoad(uint64_t workset_id, double load) override {
+        std::lock_guard<std::mutex> lock(load_mutex_);
+        loads_[workset_id] = load;
+    }
+
+    std::vector<WorksetProfile> getAllWorksetProfiles() const override {
+        std::shared_lock<std::shared_mutex> owner_lock(mutex_);
+        std::lock_guard<std::mutex> load_lock(load_mutex_);
+
+        std::vector<WorksetProfile> profiles;
+        profiles.reserve(owners_.size());
+
+        for (const auto& kv : owners_) {
+            double load = 0.0;
+            if (loads_.count(kv.first)) {
+                load = loads_.at(kv.first);
+            }
+            profiles.push_back({kv.first, kv.second, load});
+        }
+        return profiles;
+    }
+
+private:
+    mutable std::shared_mutex mutex_;
+    std::unordered_map<uint64_t, int> owners_;
+
+    mutable std::mutex load_mutex_;
+    std::unordered_map<uint64_t, double> loads_;
+};
+
+} // namespace sageFlow

include/execution/blocking_queue.h

@@ -44,6 +44,10 @@ class BlockingQueue final : public IQueue {
      * 并使后续的 push 调用立即返回，pop 调用在队列为空后返回 std::nullopt。
      */
     void stop() override;
+
+    bool isStopped() const override {
+        return stopped_.load(std::memory_order_acquire);
+    }
 
 private:
     std::queue<TaggedResponse> queue_;