EDBT22-LSH-JOIN

.gitignore

@@ -94,6 +94,8 @@ coverage.xml
 *.cover
 *.log
 htmlcov/
+Testing/Temporary/
+Testing/Temporary/CTestCostData.txt
 
 # Documentation
 docs/_build/

Testing/Temporary/CTestCostData.txt

@@ -0,0 +1 @@
+---

config/perf_join_datasource_modes.toml

@@ -66,3 +66,25 @@ file_path = "test/data/generated_test_data.json"
 #type = "random"
 
 log.level = "debug"
+
+# LSH performance quick pass
+[[performance_test]]
+name = "perf_join_lsh_random"
+mode = "generate_save_load"
+methods = ["lsh"]
+sizes = [1000]
+records_count = 10
+vector_dim = 64
+parallelism = [1,2,4,8,16,32]
+window_time_ms = [10000]
+window_trigger_ms = 50
+time_interval = 10
+similarity_threshold = 0.8
+seed = 42
+
+[performance_test.data_source]
+type = "random"
+
+[performance_test.storage]
+format = "json"
+file_path = "test/data/generated_test_data_lsh.json"

include/execution/partitioner.h

@@ -8,6 +8,7 @@
 #include <functional>
 #include <memory>
 #include "common/data_types.h"
+#include "execution/vector_space_partitioner.h"
 
 namespace sageFlow {
 class IPartitioner {
@@ -75,4 +76,21 @@ class BroadcastPartitioner : public IPartitioner {
   bool isBroadcast() const { return true; }
 };
 
+// 基于向量空间分区器的适配器，用于将 VectorSpacePartitioner 输出对接到运行时 IPartitioner 接口
+class LSHPartitionerAdapter : public IPartitioner {
+public:
+  explicit LSHPartitionerAdapter(std::shared_ptr<VectorSpacePartitioner> vsp)
+      : vsp_(std::move(vsp)) {}
+
+  size_t partition(const Response& data, size_t num_channels) override {
+    if (!vsp_ || !data.record_) {
+      return 0;
+    }
+    return vsp_->partition(*data.record_, num_channels);
+  }
+
+private:
+  std::shared_ptr<VectorSpacePartitioner> vsp_;
+};
+
 } // namespace sageFlow

include/operator/join_operator_methods/base_method.h

@@ -14,6 +14,7 @@ enum class JoinMethodType {
   IVF,
   HNSW,
   HDR_TREE,
+  LSH,
   CLUSTERED_JOIN,
   S3J
 };

include/operator/join_operator_methods/lsh_method.h

@@ -0,0 +1,106 @@
+#pragma once
+
+#include <random>
+#include <vector>
+#include <memory>
+#include <cstdint>
+#include <unordered_map>
+#include <mutex>
+
+#include "operator/join_operator_methods/base_method.h"
+#include "state/window_state.h"
+#include "compute_engine/simd_distance.h"
+#include "execution/runtime_context.h"
+
+namespace sageFlow {
+
+/**
+ * @brief 基于超平面的简易 LSH Join 方法（Eager 模式）
+ *
+ * 设计目标：
+ * - 不依赖外部索引，直接使用 WindowState 快照构建桶并筛选候选。
+ * - 先实现最小可用版本，后续可替换为更高性能的数据结构。
+ */
+class LSHMethod final : public BaseMethod {
+ public:
+  struct Config {
+    double similarity_threshold = 0.8;   ///< 相似度阈值
+    int num_tables = 4;                  ///< LSH 表数量
+    int num_hashes = 8;                  ///< 每表超平面数
+    int dimension = 128;                 ///< 向量维度
+    uint32_t seed = 42;                  ///< 随机种子
+    int64_t window_size_ms = 10000;      ///< 窗口大小（用于过期过滤）
+    int max_probes_per_table = 128;      ///< 每个表的最大多探测桶数（含主桶），提高覆盖率
+    int max_hamming_radius = 4;          ///< 多探测的最大汉明距离，取平衡的默认值
+    int sketch_bits = 8;                 ///< 轻量级预过滤 sketch 位数
+    int max_sketch_hamming = 6;          ///< 允许的 sketch 汉明距离阈值（<0 表示禁用预过滤）
+  };
+
+  explicit LSHMethod(const Config& config);
+
+  /**
+   * @brief 初始化方法，记录窗口指针与任务上下文
+   */
+  void open(const RuntimeContext& context,
+            WindowState* left_state,
+            WindowState* right_state);
+
+  /**
+   * @brief 设置窗口大小（由 JoinOperator 注入，以便进行过期过滤）
+   */
+  void setWindowSize(int64_t window_size_ms) { window_size_ms_ = window_size_ms; }
+
+  /**
+   * @brief 插入新记录到桶结构（由 JoinOperator 在窗口更新时调用）
+   */
+  void onRecordAdded(const VectorRecord& record, int slot);
+
+  std::vector<std::unique_ptr<VectorRecord>> ExecuteEager(
+      const VectorRecord& query_record,
+      int query_slot) override;
+
+ private:
+  using Hyperplane = std::vector<float>;
+    struct Entry {
+      std::shared_ptr<const VectorRecord> record;
+      uint64_t hash = 0;      ///< table 专属哈希（用于桶命中）
+      uint64_t sketch = 0;    ///< 轻量 sketch（用于快速过滤）
+      uint16_t left_sig = 0;  ///< 偶位签名（结构分片左）
+      uint16_t right_sig = 0; ///< 奇位签名（结构分片右）
+    };
+
+  Config config_;
+  WindowState* left_state_ = nullptr;
+  WindowState* right_state_ = nullptr;
+  size_t subtask_index_ = 0;
+  std::vector<std::vector<Hyperplane>> tables_;  // tables_[table][hash]
+    std::vector<Hyperplane> sketch_planes_;        // 用于生成轻量级 sketch
+
+  // 左右签名掩码（兼容保留）
+  uint32_t left_mask_ = 0;
+  uint32_t right_mask_ = 0;
+  int left_bits_ = 0;
+  int right_bits_ = 0;
+
+  // 桶结构：tables -> hash -> candidates（左右分开，避免同侧自匹配）
+    using BucketMap = std::unordered_map<uint64_t, std::vector<Entry>>;
+    std::vector<BucketMap> left_buckets_;
+    std::vector<BucketMap> right_buckets_;
+    std::vector<std::unique_ptr<std::mutex>> left_bucket_mutexes_;
+    std::vector<std::unique_ptr<std::mutex>> right_bucket_mutexes_;
+  int64_t window_size_ms_ = 10000;
+    bool use_sketch_filter_ = true;        ///< 是否启用 sketch 预过滤
+
+  std::vector<uint64_t> buildProbeKeys(uint64_t base_key,
+                                       int max_radius,
+                                       size_t max_probes) const;
+
+  void initHyperplanes();
+    void initSketchPlanes();
+  uint64_t hashVector(const VectorRecord& record, const std::vector<Hyperplane>& planes) const;
+    uint64_t sketchVector(const VectorRecord& record) const;
+  void splitHash(uint64_t full_hash, uint16_t& left, uint16_t& right) const;
+  static std::vector<float> toFloatVector(const VectorRecord& record);
+};
+
+}  // namespace sageFlow

include/operator/utils/join_strategy_config.h

@@ -14,6 +14,7 @@ enum class JoinAlgorithm {
     IVF,             ///< IVF-based approximate join
     HNSW,            ///< HNSW-based approximate join
     HDR_TREE,        ///< HDR-Tree baseline
+    LSH,             ///< Hyperplane LSH 近似 Join（移植 Danny baseline）
     CLUSTERED_JOIN,  ///< VectraFlow ClusteredJoin
     S3J,             ///< DEBS'23 S3J baseline
     VSJOIN           ///< Our method
@@ -81,6 +82,11 @@ struct JoinStrategyConfig {
     int hnsw_m = 16;                   ///< 每层最大邻居数
     int hnsw_ef_construction = 200;    ///< 构建时候选集大小
     int hnsw_ef_search = 50;           ///< 搜索时候选集大小
+
+    // ==================== LSH 参数 ====================
+    int lsh_num_tables = 4;            ///< LSH 表数量（重复次数）
+    int lsh_num_hashes = 8;            ///< 每张表的超平面数量
+    uint32_t lsh_seed = 42;            ///< 随机种子，确保可复现
 
     // ==================== VSJoin 参数 ====================
     int vsjoin_num_hash_functions = 8;    ///< LSH 哈希函数数量

include/operator/utils/join_strategy_factory.h

@@ -213,6 +213,11 @@ class JoinStrategyFactory {
         const JoinStrategyConfig& config,
         std::shared_ptr<ConcurrencyManager> cm,
         int left_idx, int right_idx);
+
+    static std::unique_ptr<BaseMethod> createLshMethod(
+        const JoinStrategyConfig& config,
+        std::shared_ptr<ConcurrencyManager> cm,
+        int left_idx, int right_idx);
 
     static std::unique_ptr<BaseMethod> createClusteredJoinMethod(
         const JoinStrategyConfig& config,

src/operator/CMakeLists.txt

@@ -22,6 +22,7 @@ add_lib(
         join_operator_methods/ivf.cpp
         join_operator_methods/hnsw.cpp
         join_operator_methods/hdr_tree_method.cpp
+        join_operator_methods/lsh_method.cpp
         join_operator_methods/clustered_join_method.cpp
         join_operator_methods/s3j_method.cpp
         join_operator_methods/vsjoin_method.cpp

src/operator/join_operator.cpp

@@ -7,6 +7,7 @@
 #include "operator/join_operator_methods/bruteforce_baseline.h"
 #include "operator/join_operator_methods/ivf_method.h"
 #include "operator/join_operator_methods/hdr_tree_method.h"
+#include "operator/join_operator_methods/lsh_method.h"
 #include "operator/join_metrics.h"
 #include "operator/utils/join_strategy_factory.h"
 #include "operator/utils/join_config_validator.h"
@@ -230,6 +231,23 @@ JoinOperator::JoinOperator(std::unique_ptr<Function> &join_func,
               -1, -1, join_similarity_threshold_, concurrency_manager_);
             SAGEFLOW_LOG_WARN("JOIN", "Failed to create HNSW index pair, falling back to BruteForce");
         }
+    } else if (algo == "lsh") {
+        // Hyperplane LSH: window-state only, no shared index
+        index_kind_ = InternalIndexKind::NONE;
+        use_shared_state_ = false;  // 与 inferDefaults 保持一致，使用分区窗口
+        use_index_ = false;
+
+        LSHMethod::Config lsh_cfg;
+        lsh_cfg.similarity_threshold = join_similarity_threshold_;
+        lsh_cfg.num_tables = 4;
+        lsh_cfg.num_hashes = 8;
+        lsh_cfg.dimension = join_func_->getDim();
+        lsh_cfg.seed = 42;
+        lsh_cfg.window_size_ms = join_func_->getWindowSize();
+
+        join_method_ = std::make_unique<LSHMethod>(lsh_cfg);
+        SAGEFLOW_LOG_INFO("JOIN", "LSH mode enabled (window-state buckets, partitioned state)");
+
     } else {
         index_kind_ = InternalIndexKind::NONE;
         use_index_ = false;
@@ -384,6 +402,11 @@ void JoinOperator::open(const RuntimeContext& context) {
           SAGEFLOW_LOG_INFO("JOIN", "IVFMethod initialized with ConcurrencyManager index, left_idx={} right_idx={}",
                            left_index_id_, right_index_id_);
       }
+      // LSH path: initialize hyperplane tables with window state pointers
+      else if (auto* lsh = dynamic_cast<LSHMethod*>(join_method_.get())) {
+          lsh->open(context, left_state_.get(), right_state_.get());
+          SAGEFLOW_LOG_INFO("JOIN", "LSHMethod initialized with WindowState");
+      }
   }
 
   SAGEFLOW_LOG_INFO("JOIN", "JoinOperator opened: subtask={}/{}, shared_state={}", 
@@ -414,6 +437,11 @@ auto JoinOperator::updateSideThreadSafe(
     } else {
       JoinMetrics::instance().total_records_right.fetch_add(1, std::memory_order_relaxed);
     }
+        if (auto* lsh = dynamic_cast<LSHMethod*>(join_method_.get())) {
+                if (data_ptr) {
+                        lsh->onRecordAdded(*data_ptr, slot);
+                }
+        }
     // 窗口插入阶段（仅插入）
     {
         MetricsTimer t_window_ins(JoinMetrics::instance().window_insert_ns);
@@ -716,6 +744,12 @@ auto JoinOperator::updateSideWithState(
     } else {
         JoinMetrics::instance().total_records_right.fetch_add(1, std::memory_order_relaxed);
     }
+
+    if (auto* lsh = dynamic_cast<LSHMethod*>(join_method_.get())) {
+        if (data_ptr) {
+            lsh->onRecordAdded(*data_ptr, slot);
+        }
+    }
 
     // 添加记录到窗口状态
     {
@@ -1087,6 +1121,11 @@ void JoinOperator::initializeWithStrategyConfig(const RuntimeContext& context) {
             use_index_ = true;
             SAGEFLOW_LOG_INFO("JOIN", "HNSWJoinMethod initialized via strategy config");
         }
+        // 尝试作为 LSHMethod 初始化（纯窗口 + LSH 桶过滤）
+        else if (auto* lsh = dynamic_cast<LSHMethod*>(join_method_.get())) {
+            lsh->open(context, left_state_.get(), right_state_.get());
+            SAGEFLOW_LOG_INFO("JOIN", "LSHMethod initialized via strategy config");
+        }
         // VSJoin 将通过 VSJoinMethod 处理，不再需要特殊初始化
         // 参考: include/operator/join_operator_methods/vsjoin_method.h
     }