Skip to content

Commit 613ad56

Browse files
feat(vsjoin): implement VSJoin dual-index architecture (v0.1.3.1) (#101)
* feat(vsjoin): Refine VSJoin plan and add detailed task docs This commit introduces a comprehensive plan and task breakdown for the new VSJoin implementation. - **Refine VSJoin Plan:** - The main design document is updated to unify naming (removing "v2"). - Clarifies the replacement of v1 components with the new architecture (TwoTierWindowState + ConcurrencyManager). - Adds detailed sections on UID deduplication and the RCU-based load balancing mechanism (AssignmentTable). - **Add Detailed Task Documents:** - Creates a new directory. - Adds markdown files for each implementation step (Task 01 to 09), providing clear instructions for development. These files are force-added as the parent directory is in .gitignore. - **Cleanup Unused v1 Components:** - Deletes and as they are no longer needed in the new design. * feat(test): Add unit test for VSJoin rebuild functionality This commit adds a new unit test for the VSJoin rebuild process, enhancing the test coverage for the VSJoin implementation. The test file is located at UnitTest/test_vsjoin_rebuild.cpp and is configured to run with a specified timeout of 300 seconds. * feat(vsjoin): implement VSJoin dual-index architecture (tasks 01-04) - Task 01: VSJoinMethod basic implementation - Implement ExecuteEager() with dual-layer query logic (Global + Local) - Add UID deduplication using local unordered_set - Implement setGlobalIndexIds/setLocalIndexIds/setWindowStates interfaces - Task 02: JoinStrategyFactory integration - Add JoinAlgorithm::VSJOIN enum - Add vsjoin_* configuration parameters - Integrate VSJOIN case in factory (fallback to BruteForce for now) - Task 03: JoinOperator VSJoin special path - Add vsjoin_local_*_ids_ and vsjoin_global_*_id_ members - Implement VSJoin-specific updateSideWithState() (insert to local index only) - Support LSH partitioner for VSJoin - Task 04: Background rebuild mechanism - Implement globalIndexRebuildLoop() with periodic rebuild - Use std::call_once for thread-safe single startup - Implement local unordered_set deduplication (lock-free) - Add atomic index replacement via replace_index_by_id() Integration tests passed: - bruteforce: 7/7 tests, recall=1.000 - ivf: 7/7 tests, recall=0.999 - hdr_tree: 7/7 tests, recall=1.000 - clustered_join: 3/3 tests, recall=1.000 * fix(vsjoin): fix dangling pointer bug in globalIndexRebuildLoop The snapshot ownership was released prematurely in the rebuild loop, causing the pointers stored in unique_left_records and unique_right_records to become dangling. This led to IVF centroids being initialized with garbage data (dimension=0), causing 'Vectors must be of the same size' errors during query_for_join. Fix: Keep snapshot vectors alive until build_index_from_records completes by storing them in left_snapshots/right_snapshots containers. Also includes: - Improved test assertions for query record dimension - Code cleanup for VSJoin factory integration * chore: bump version to 0.1.3.1 * chore: add TODO issue links via todo-to-issue-action * fix(test): correct LSH window_state_type expectation to PARTITIONED The inferDefaults() for LSH algorithm returns PARTITIONED, not PARTITIONED_VECTOR. PARTITIONED_VECTOR is only used for VSJOIN algorithm. * feat(vsjoin): 完成 Task 06 - VSJoin 集成测试链路打通 sed -n '168,195p' /root/sageFlow/src/operator/utils/join_strategy_factory.cpp 1. 配置验证 ✅ - integration_test_cases.toml 包含 4 个启用的 VSJoin 测试用例 - 配置包含必要参数 (vsjoin_num_hash_functions, vsjoin_boundary_threshold 等) - num_partitions 参数设置合理 2. 链路打通验证 ✅ - JoinStrategyFactory::create() 正确创建 VSJoinMethod - TwoTierWindowState 正确初始化 - Global/Local Index 正确创建和管理 - 后台重建线程正常工作 3. 测试执行验证 ✅ - test_join_baseline_integration --gtest_filter='*vsjoin*' 执行成功 - run_integration_test.py --methods vsjoin 执行成功 - 测试报告正确生成 4. 召回率验证 ✅ - vsjoin_baseline: Recall=1.0 (预期>=0.70) - vsjoin_high_recall: Recall=1.0 (预期>=0.75) - vsjoin_parallelism_scaling: Recall=1.0 (并行度 1-16) - vsjoin_low_latency: Recall>=0.60 sed -n '168,195p' /root/sageFlow/src/operator/utils/join_strategy_factory.cpp - 修复 JoinConfigValidator 允许 LSH + TWO_TIER 组合 - 修复 JoinOperator 中 VSJoin 的 use_index_ 和 index_id 计算 - 更新 integration_test_cases.toml 添加 VSJoin 测试用例 - 更新 task06_integration_test.md 添加集成测试框架链路任务 * feat(vsjoin): implement Task 07-09 - AssignmentTable, LoadMonitor, Logical Partition Routing, and Load Balancing Tests Task 07: AssignmentTable (RCU) + LoadMonitor - Implement PartitionAssignment with RCU pattern for lock-free reads - Implement LoadMonitor for tracking partition load statistics - Support logical-to-physical partition mapping - Batch atomic updates for assignment table Task 08: Logical Partition Routing Integration - Add VSJoin routing methods in JoinOperator - Implement routeByLSHBucket() for query routing - Implement determineTargetPartitions() with multi-partition support - Integrate with CentroidPartitioner for initial assignment Task 09: Load Balancing Tests - Add comprehensive unit tests for LoadMonitor - Add comprehensive unit tests for PartitionAssignment - Add integration tests for VSJoin routing - Add load balancing scenario tests * test: isolate per-run outputs (reports/logs/charts) * feat(python): SAGE integration with Python bindings v0.1.3 (#100) * feat(python): add SAGE integration examples and documentation - Add test_sageflow_cpp_runtime.py: comprehensive C++ runtime verification tests - Add sage_sageflow_dual_stream_join.py: dual-stream Join pipeline demo for RAG - Add SAGEFLOW_SAGE_INTEGRATION_GUIDE.md: integration guide for SAGE + SageFlow The dual-stream Join demo shows: - Query Stream + Document Stream architecture - SageFlow C++ engine for vector similarity join - RAG context building from join results * feat(python): complete SAGE integration with Python bindings and examples Modified files: - sage_flow/__init__.py: update exports for SAGE integration - sage_flow/bindings.cpp: enhance Python bindings for dual-stream Join - test/CMakeLists.txt: add new test targets New files: - docs/LLM_INFERENCE_PIPELINE_GUIDE.md: LLM inference pipeline guide - docs/VSJOIN_DESIGN_REVIEW_REPORT.md: VSJoin design review - examples/python/llm_inference_service_demo.py: LLM service demo - examples/python/llm_pipeline_example.py: LLM pipeline example - examples/python/sage_integrated_pipeline_demo.py: SAGE integration demo - test/IntegrationTest/test_non_join_operators_pipeline.cpp: non-join ops test - test/UnitTest/python/: Python unit tests * chore: bump version to 0.1.3 for PyPI release This release includes: - SAGE integration Python bindings - Dual-stream Join pipeline support - Comprehensive C++ runtime verification tests - RAG pipeline examples and documentation * fix(test): correct LSH window_state_type expectation to PARTITIONED The inferDefaults() for LSH algorithm returns PARTITIONED, not PARTITIONED_VECTOR. PARTITIONED_VECTOR is only used for VSJOIN algorithm. * fix(test): remove index_id checks from LSH test LSH algorithm doesn't use external index, so left_index_id and right_index_id can be -1. Removed these checks to match feat/implement_vsjoin branch. * doc: add sage pipeline markdown file * fix: unitest failed * fix: update VSJoin test to expect TWO_TIER window state type * fix(pybind): fix alpha hardcode bug in NORMALIZED mode and add window_size_ms parameter - Fix BruteForceBaseline to use configured similarity_alpha instead of hardcoded 0.1 in NORMALIZED mode - Add 7-parameter join() overload with window_size_ms support in pybind interface - Add StreamingSource for dynamic streaming input (vs batch-oriented SimpleStreamSource) - Update JoinStrategyConfig to properly propagate similarity_alpha and window_size_ms * refactor(pybind): simplify join method bindings and fix test expectations - Remove unused include of join_strategy_config.h in bindings.cpp - Simplify join binding: use JoinFunction constructor without time_window param - Remove redundant SimilarityMode::NORMALIZED and alpha settings - Fix test expectation: LSH uses PARTITIONED instead of PARTITIONED_VECTOR - Add note about LSH not depending on external index * add refactor partitioner doc, stash changes --------- Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
1 parent 85986f9 commit 613ad56

74 files changed

Lines changed: 7241 additions & 2488 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎cmake/enableTest.cmake‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,10 +23,12 @@ macro(add_gtest TARGET_NAME SOURCE_FILE)
2323
add_test(NAME ${TARGET_NAME} COMMAND ${TARGET_NAME})
2424

2525
# 让 CLion 能枚举单测用例
26+
# 注意: 使用 POST_BUILD 模式避免并行构建时的 race condition
2627
if(COMMAND gtest_discover_tests)
2728
gtest_discover_tests(${TARGET_NAME}
2829
WORKING_DIRECTORY ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}
2930
DISCOVERY_TIMEOUT 30
31+
DISCOVERY_MODE PRE_TEST
3032
)
3133
endif()
3234
endmacro()

‎config/integration_test_cases.toml‎

Lines changed: 76 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ window_state_type = "shared"
4949
index_strategy = "shared"
5050
data_sizes = [500]
5151
window_size_ms = 1000
52-
parallelism = [1, 2, 4, 8, 16]
52+
parallelism = [1, 2, 4, 8, 16, 24, 32]
5353
expected_min_recall = 0.95
5454
expected_min_precision = 1.0
5555

@@ -341,7 +341,7 @@ clustered_training_samples = 50
341341
clustered_cold_start_enabled = true
342342
clustered_broadcast_dedup = true
343343
data_sizes = [2000]
344-
parallelism = [1, 2, 4, 8, 16]
344+
parallelism = [1, 2, 4, 8, 16, 24, 32]
345345
expected_min_recall = 0.90
346346
enabled = false
347347

@@ -551,63 +551,107 @@ expected_min_recall = 0.85
551551
enabled = false
552552

553553
# ==================== VSJoin 测试 ====================
554-
# Full VSJoin with LSH partitioning
554+
# VSJoin 双层索引(Global IVF + Local BruteForce)+ 后台重建
555+
#
556+
# 架构说明:
557+
# - Global Index: 2 个 IVF 索引(左/右流各一个),存储历史数据快照
558+
# - Local Index: 2*P 个 BruteForce 索引(每个分区左/右流各一个),存储实时数据
559+
# - 后台重建: 定期将 WindowState 中的数据重建到 Global Index
560+
# - LSH 分区: 使用局部敏感哈希进行向量分区
561+
562+
# VSJoin 基线测试 - 验证基本功能
555563
[[test_case]]
556-
name = "vsjoin_complete"
557-
description = "Full VSJoin with LSH partitioning and partitioned index"
564+
name = "vsjoin_baseline"
565+
description = "VSJoin baseline test with two-tier index"
558566
algorithm = "vsjoin"
559567
partition_strategy = "lsh"
560-
window_state_type = "partitioned_vector"
568+
window_state_type = "two_tier"
561569
index_strategy = "partitioned"
570+
num_partitions = 4
571+
# VSJoin 参数
562572
vsjoin_num_hash_functions = 8
563573
vsjoin_boundary_threshold = 0.1
564-
vsjoin_async_threads = 2
565-
vsjoin_allowed_lateness = 1000
566-
num_partitions = 8
574+
vsjoin_rebuild_interval_ms = 3000
575+
vsjoin_rebuild_threshold = 500
576+
# Global Index (IVF) 参数
567577
ivf_nlist = 50
568-
ivf_nprobes = 5
578+
ivf_nprobes = 10
579+
# 测试配置
569580
data_sizes = [500, 1000]
570-
parallelism = [2, 4]
581+
parallelism = [1, 2, 4]
571582
expected_min_recall = 0.70
572-
enabled = false
583+
enabled = true
573584

585+
# VSJoin 高召回测试 - 更多哈希函数提升分区质量
574586
[[test_case]]
575-
name = "vsjoin_high_hash"
576-
description = "VSJoin with more hash functions for better partitioning"
587+
name = "vsjoin_high_recall"
588+
description = "VSJoin with higher hash functions for better recall"
577589
algorithm = "vsjoin"
578590
partition_strategy = "lsh"
579-
window_state_type = "partitioned_vector"
591+
window_state_type = "two_tier"
580592
index_strategy = "partitioned"
593+
num_partitions = 4
594+
# VSJoin 参数 - 更多哈希函数
581595
vsjoin_num_hash_functions = 16
582-
vsjoin_boundary_threshold = 0.1
583-
vsjoin_async_threads = 2
584-
vsjoin_allowed_lateness = 1000
585-
num_partitions = 8
596+
vsjoin_boundary_threshold = 0.15
597+
vsjoin_rebuild_interval_ms = 2000
598+
vsjoin_rebuild_threshold = 300
599+
# Global Index (IVF) 参数 - 更大的 nprobes
586600
ivf_nlist = 50
587-
ivf_nprobes = 5
601+
ivf_nprobes = 15
602+
# 测试配置
588603
data_sizes = [500]
589-
parallelism = [2, 4]
604+
parallelism = [1, 2, 4]
590605
expected_min_recall = 0.75
591-
enabled = false
606+
enabled = true
592607

608+
# VSJoin 并行扩展测试 - 测试多并行度下的性能
609+
# 注意: VSJoin 的 LSH 分区在高并行度下召回率会下降(LSH 算法特性)
610+
# 当前配置下并行度 1-16 能保持 100% 召回率
593611
[[test_case]]
594-
name = "vsjoin_low_latency"
595-
description = "VSJoin optimized for low latency"
612+
name = "vsjoin_parallelism_scaling"
613+
description = "VSJoin parallelism scalability test"
596614
algorithm = "vsjoin"
597615
partition_strategy = "lsh"
598-
window_state_type = "partitioned_vector"
616+
window_state_type = "two_tier"
599617
index_strategy = "partitioned"
618+
num_partitions = 16
619+
# VSJoin 参数
600620
vsjoin_num_hash_functions = 8
601-
vsjoin_boundary_threshold = 0.05
602-
vsjoin_async_threads = 4
603-
vsjoin_allowed_lateness = 500
621+
vsjoin_boundary_threshold = 0.1
622+
vsjoin_rebuild_interval_ms = 3000
623+
vsjoin_rebuild_threshold = 500
624+
# Global Index (IVF) 参数
625+
ivf_nlist = 50
626+
ivf_nprobes = 10
627+
# 测试配置 - 并行度 1-16 稳定通过
628+
data_sizes = [2000]
629+
parallelism = [1, 2, 4, 8, 16, 24, 32]
630+
expected_min_recall = 0.90
631+
enabled = true
632+
633+
# VSJoin 低延迟测试 - 快速重建
634+
[[test_case]]
635+
name = "vsjoin_low_latency"
636+
description = "VSJoin optimized for low latency with fast rebuild"
637+
algorithm = "vsjoin"
638+
partition_strategy = "lsh"
639+
window_state_type = "two_tier"
640+
index_strategy = "partitioned"
604641
num_partitions = 4
642+
# VSJoin 参数 - 快速重建
643+
vsjoin_num_hash_functions = 8
644+
vsjoin_boundary_threshold = 0.1
645+
vsjoin_rebuild_interval_ms = 1000
646+
vsjoin_rebuild_threshold = 200
647+
# Global Index (IVF) 参数
605648
ivf_nlist = 30
606-
ivf_nprobes = 3
649+
ivf_nprobes = 8
650+
# 测试配置
607651
data_sizes = [500]
608-
parallelism = [2, 4]
609-
expected_min_recall = 0.65
610-
enabled = false
652+
parallelism = [1, 2, 4]
653+
expected_min_recall = 0.60
654+
enabled = true
611655
allow_approximate_match = true
612656

613657
# ==================== 混合测试(多并行度/多数据规模)====================

‎config/vsjoin_strategy.toml‎

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,111 @@
1+
# VSJoin 配置示例
2+
#
3+
# VSJoin 是一种基于双层索引的向量流 Join 算法,使用 LSH 分区和 Local/Global 索引。
4+
#
5+
# 关键特性:
6+
# - Local Index: 分区内的轻量级索引(推荐 BruteForce)
7+
# - Global Index: 跨分区的候选召回索引(推荐 IVF)
8+
# - 后台重建: 定期重建 Global Index 保持索引新鲜度
9+
# - 边界向量多播: 处理分区边界的向量以保证召回率
10+
11+
[default]
12+
# 基础配置
13+
algorithm = "vsjoin"
14+
is_eager = true
15+
similarity_threshold = 0.8
16+
dimension = 128
17+
18+
# 分区配置
19+
partition_strategy = "lsh"
20+
window_state_type = "partitioned"
21+
index_strategy = "partitioned"
22+
num_partitions = 4
23+
24+
# 窗口配置
25+
window_size_ms = 10000
26+
step_size_ms = 1000
27+
28+
# VSJoin 核心参数
29+
vsjoin_multicast_k = 2 # 边界向量多播到 k 个分区(推荐 2-3)
30+
vsjoin_rebuild_interval_ms = 5000 # Global Index 重建间隔(毫秒)
31+
vsjoin_rebuild_threshold = 1000 # 触发重建的记录数阈值
32+
33+
# VSJoin 索引类型
34+
vsjoin_local_index_type = "bruteforce" # Local Index 类型(推荐 bruteforce)
35+
vsjoin_global_index_type = "ivf" # Global Index 类型(推荐 ivf)
36+
37+
# LSH 分区器参数
38+
vsjoin_num_hash_functions = 8 # LSH 哈希函数数量
39+
vsjoin_boundary_threshold = 0.1 # 边界向量阈值
40+
41+
# IVF 参数(用于 Global Index)
42+
ivf_nlist = 100
43+
ivf_nprobes = 10
44+
45+
# HNSW 参数(备选 Global Index)
46+
hnsw_m = 16
47+
hnsw_ef_construction = 200
48+
hnsw_ef_search = 50
49+
50+
# ==================== 策略变体 ====================
51+
52+
[strategies.vsjoin_default]
53+
# 默认 VSJoin 配置
54+
algorithm = "vsjoin"
55+
partition_strategy = "lsh"
56+
window_state_type = "partitioned"
57+
index_strategy = "partitioned"
58+
vsjoin_multicast_k = 2
59+
vsjoin_rebuild_interval_ms = 5000
60+
vsjoin_rebuild_threshold = 1000
61+
vsjoin_local_index_type = "bruteforce"
62+
vsjoin_global_index_type = "ivf"
63+
vsjoin_num_hash_functions = 8
64+
vsjoin_boundary_threshold = 0.1
65+
66+
[strategies.vsjoin_high_recall]
67+
# 高召回率配置(更多多播)
68+
algorithm = "vsjoin"
69+
partition_strategy = "lsh"
70+
window_state_type = "partitioned"
71+
index_strategy = "partitioned"
72+
vsjoin_multicast_k = 3 # 更多多播提高召回
73+
vsjoin_rebuild_interval_ms = 3000 # 更频繁的重建
74+
vsjoin_rebuild_threshold = 500 # 更小的阈值
75+
vsjoin_local_index_type = "bruteforce"
76+
vsjoin_global_index_type = "ivf"
77+
vsjoin_num_hash_functions = 12 # 更多哈希函数
78+
vsjoin_boundary_threshold = 0.15 # 更宽的边界
79+
ivf_nprobes = 20 # 更多探测提高召回
80+
81+
[strategies.vsjoin_high_throughput]
82+
# 高吞吐量配置(减少多播开销)
83+
algorithm = "vsjoin"
84+
partition_strategy = "lsh"
85+
window_state_type = "partitioned"
86+
index_strategy = "partitioned"
87+
vsjoin_multicast_k = 1 # 减少多播
88+
vsjoin_rebuild_interval_ms = 10000 # 较少的重建
89+
vsjoin_rebuild_threshold = 2000 # 更大的阈值
90+
vsjoin_local_index_type = "bruteforce"
91+
vsjoin_global_index_type = "ivf"
92+
vsjoin_num_hash_functions = 6 # 较少哈希函数
93+
vsjoin_boundary_threshold = 0.05 # 更窄的边界
94+
ivf_nprobes = 5 # 较少探测提高速度
95+
96+
[strategies.vsjoin_hnsw_global]
97+
# 使用 HNSW 作为 Global Index
98+
algorithm = "vsjoin"
99+
partition_strategy = "lsh"
100+
window_state_type = "partitioned"
101+
index_strategy = "partitioned"
102+
vsjoin_multicast_k = 2
103+
vsjoin_rebuild_interval_ms = 5000
104+
vsjoin_rebuild_threshold = 1000
105+
vsjoin_local_index_type = "bruteforce"
106+
vsjoin_global_index_type = "hnsw" # 使用 HNSW
107+
vsjoin_num_hash_functions = 8
108+
vsjoin_boundary_threshold = 0.1
109+
hnsw_m = 16
110+
hnsw_ef_construction = 200
111+
hnsw_ef_search = 100 # HNSW 搜索参数

0 commit comments

Comments
 (0)