[Feature] Add Distributed Posting Router for SPANN

AnnService/CMakeLists.txt

@@ -10,6 +10,12 @@ include_directories(${Zstd}/lib)
 file(GLOB_RECURSE HDR_FILES ${AnnService}/inc/Core/*.h  ${AnnService}/inc/Helper/*.h)
 file(GLOB_RECURSE SRC_FILES ${AnnService}/src/Core/*.cpp ${AnnService}/src/Helper/*.cpp)
 
+# Include Socket sources in core lib for PostingRouter
+file(GLOB SOCKET_HDR_FILES ${AnnService}/inc/Socket/*.h)
+file(GLOB SOCKET_SRC_FILES ${AnnService}/src/Socket/*.cpp)
+list(APPEND HDR_FILES ${SOCKET_HDR_FILES})
+list(APPEND SRC_FILES ${SOCKET_SRC_FILES})
+
 set(SPDK_LIBRARIES "")
 if (SPDK)
     set(Spdk ${PROJECT_SOURCE_DIR}/ThirdParty/spdk/build)
@@ -73,7 +79,7 @@ endif()
 add_library (SPTAGLib SHARED ${SRC_FILES} ${HDR_FILES} ${TiKV_PROTO_SOURCES})
 target_link_libraries (SPTAGLib DistanceUtils ${RocksDB_LIBRARIES} ${uring_LIBRARIES} libzstd_shared ${NUMA_LIBRARY} ${TBB_LIBRARIES} ${SPDK_LIBRARIES} ${TiKV_LIBRARIES})
 add_library (SPTAGLibStatic STATIC ${SRC_FILES} ${HDR_FILES} ${TiKV_PROTO_SOURCES})
-target_link_libraries (SPTAGLibStatic DistanceUtils ${RocksDB_LIBRARIES} ${uring_LIBRARIES} libzstd_static ${NUMA_LIBRARY_STATIC} ${TBB_LIBRARIES} ${SPDK_LIBRARIES} ${TiKV_LIBRARIES})
+target_link_libraries (SPTAGLibStatic DistanceUtils ${RocksDB_LIBRARIES} ${uring_LIBRARIES} libzstd_static ${NUMA_LIBRARY_STATIC} ${TBB_LIBRARIES} ${SPDK_LIBRARIES} ${TiKV_LIBRARIES} ${Boost_LIBRARIES})
 
 if (MSVC)
     # SPANNIndex.cpp can exceed COFF section limits in Debug without /bigobj.

AnnService/inc/Core/Common/FineGrainedLock.h

@@ -60,6 +60,19 @@ namespace SPTAG
             {
                 return idx;
             }
+
+            // Bucket index for the internal mutex-sharded unordered_map of
+            // per-posting locks. Exposed for callers that need an array sized
+            // to BucketCount and indexed by the same granularity as the lock
+            // pool (e.g. ExtraDynamicSearcher::m_remoteBucketLocked).
+            static inline unsigned BucketIndex(SizeType idx)
+            {
+                unsigned key = static_cast<unsigned>(idx);
+                return ((unsigned)(key * 99991) + _rotl(key, 2) + 101) & BucketMask;
+            }
+
+            static const int BucketMask = 32767;
+            static const int BucketCount = BucketMask + 1;
         private:
             struct Bucket {
                 std::mutex mutex;
@@ -76,14 +89,6 @@ namespace SPTAG
                 return *iter->second;
             }
 
-            static inline unsigned BucketIndex(SizeType idx)
-            {
-                unsigned key = static_cast<unsigned>(idx);
-                return ((unsigned)(key * 99991) + _rotl(key, 2) + 101) & BucketMask;
-            }
-
-            static const int BucketMask = 32767;
-            static const int BucketCount = BucketMask + 1;
             mutable std::unique_ptr<Bucket[]> m_buckets;
         };
     }

AnnService/inc/Core/Common/IVersionMap.h

@@ -30,6 +30,30 @@ namespace SPTAG
 
             virtual void DeleteAll() = 0;
 
+            /// One-time per-layer setup performed at the end of BuildIndex.
+            ///   size      total VID count for this layer (== m_opt->m_vectorSize)
+            ///   blockSize/capacity  hints for array-backed legacy maps; ignored
+            ///                       by hashmap / TiKV implementations
+            ///   globalIDs (optional) set of GLOBAL VIDs that are alive on
+            ///                       this layer. Layers whose "default
+            ///                       version" semantics treat unknown VIDs as
+            ///                       DELETED (e.g. TiKV layer >0, hashmap
+            ///                       LocalVersionMap) MUST persist an
+            ///                       explicit alive byte for each globalID;
+            ///                       otherwise MergePostings'
+            ///                       Deleted()/version-mismatch filter
+            ///                       eats every base entry on the first
+            ///                       async merge and corrupts the head index.
+            /// Default impl: just bump the internal count via SetR.
+            virtual void Initialize(SizeType size, SizeType blockSize, SizeType capacity,
+                                    COMMON::Dataset<SizeType>* globalIDs = nullptr)
+            {
+                (void)blockSize;
+                (void)capacity;
+                (void)globalIDs;
+                SetR(size);
+            }
+
             virtual SizeType Count() = 0;
             virtual SizeType GetDeleteCount() = 0;
             virtual std::uint64_t BufferSize() = 0;
@@ -45,6 +69,18 @@ namespace SPTAG
             virtual bool TryGetDefaultVersionForNewVector(uint8_t& version) const { return false; }
             virtual void SetR(SizeType num) {}
             virtual void SetVersion(const SizeType& key, const uint8_t& version) = 0;
+
+            /// Batch SetVersion: apply (vids[i] -> versions[i]) for all i.
+            /// Default impl is a per-VID loop. TiKV-backed maps override this
+            /// to group writes by chunk so N records in the same chunk only
+            /// trigger 1 ReadChunk + 1 WriteChunk RPC pair
+            virtual void SetVersionBatch(const std::vector<SizeType>& vids, const std::vector<uint8_t>& versions)
+            {
+                size_t n = std::min(vids.size(), versions.size());
+                for (size_t i = 0; i < n; i++) {
+                    SetVersion(vids[i], versions[i]);
+                }
+            }
             /// Increment the version of a VID.
             /// @param expectedOld If not 0xff, the caller asserts the current version should be this value.
             ///   If TiKV already holds (expectedOld+1)&0x7f, treat as success (another node did the same increment).

AnnService/inc/Core/Common/LocalVersionMap.h

@@ -27,6 +27,29 @@ namespace SPTAG
                 m_label.clear(); 
             }
 
+            void Initialize(SizeType size, SizeType blockSize, SizeType capacity,
+                            COMMON::Dataset<SizeType>* globalIDs = nullptr) override
+            {
+                (void)size;
+                (void)blockSize;
+                (void)capacity;
+                if (globalIDs == nullptr || globalIDs->R() <= 0) return;
+
+                // Hashmap LocalVersionMap treats missing keys as deleted
+                // (Deleted() returns true, GetVersion() returns 0xfe).
+                // Layer-1 build calls Initialize with the alive-head global
+                // IDs; we must explicitly mark them alive (0x00) so that
+                // MergePostings' Deleted()/version-mismatch filter does not
+                // strip every base head entry on the first async merge.
+                std::unique_lock<std::shared_timed_mutex> lock(m_updateMutex);
+                for (SizeType i = 0; i < globalIDs->R(); i++) {
+                    SizeType globalID = *(globalIDs->At(i));
+                    if (globalID >= 0) {
+                        m_label[globalID] = 0x00;
+                    }
+                }
+            }
+
             SizeType Count() override { 
                 std::shared_lock<std::shared_timed_mutex> lock(m_updateMutex);
                 return (SizeType)(m_label.size()); 

AnnService/inc/Core/Common/TiKVVersionMap.h

@@ -6,6 +6,7 @@
 
 #include "IVersionMap.h"
 #include "inc/Helper/KeyValueIO.h"
+#include <algorithm>
 #include <atomic>
 #include <string>
 #include <vector>
@@ -212,7 +213,7 @@ namespace SPTAG
 
             std::shared_ptr<Helper::KeyValueIO> GetDB() const { return m_db; }
 
-            void Initialize(SizeType size, SizeType blockSize, SizeType capacity, COMMON::Dataset<SizeType>* globalIDs = nullptr)
+            void Initialize(SizeType size, SizeType blockSize, SizeType capacity, COMMON::Dataset<SizeType>* globalIDs = nullptr) override
             {
                 (void)blockSize;
                 (void)capacity;
@@ -233,10 +234,44 @@ namespace SPTAG
                     m_deleted = size;
                     SaveMetadata();
 
+                    // Batch the alive-marker writes via MultiPut so they
+                    // can be grouped per TiKV region and issued in parallel.
+                    // Serial PutByte was the build-time hotspot (~1-2ms
+                    // per write × ~200K alive heads at 1M-vector scale).
+                    std::vector<SizeType> aliveSorted;
+                    aliveSorted.reserve(aliveIDs.size());
+                    for (SizeType id : aliveIDs) aliveSorted.push_back(id);
+                    std::sort(aliveSorted.begin(), aliveSorted.end());
+
                     SizeType written = 0;
-                    for (SizeType globalID : aliveIDs) {
-                        if (PutByte(VersionKey(globalID), 0x00) == ErrorCode::Success) {
-                            written++;
+                    constexpr size_t kBatchSize = 4096;
+                    std::vector<std::string> keys;
+                    std::vector<std::string> values;
+                    keys.reserve(kBatchSize);
+                    values.reserve(kBatchSize);
+                    const std::string aliveByte(1, static_cast<char>(0x00));
+                    for (size_t i = 0; i < aliveSorted.size(); i++) {
+                        keys.push_back(VersionKey(aliveSorted[i]));
+                        values.push_back(aliveByte);
+                        if (keys.size() >= kBatchSize || i + 1 == aliveSorted.size()) {
+                            auto ret = m_db->MultiPut(keys, values, MaxTimeout, nullptr);
+                            if (ret == ErrorCode::Success) {
+                                written += static_cast<SizeType>(keys.size());
+                            } else if (ret == ErrorCode::Undefined) {
+                                // Backend lacks MultiPut: fall back to serial PutByte.
+                                for (const auto& k : keys) {
+                                    if (PutByte(k, 0x00) == ErrorCode::Success) written++;
+                                }
+                            } else {
+                                SPTAGLIB_LOG(Helper::LogLevel::LL_Warning,
+                                    "TiKVVersionMap::Initialize: MultiPut batch failed layer=%d ret=%d size=%zu; falling back to serial PutByte for this batch.\n",
+                                    m_layer, static_cast<int>(ret), keys.size());
+                                for (const auto& k : keys) {
+                                    if (PutByte(k, 0x00) == ErrorCode::Success) written++;
+                                }
+                            }
+                            keys.clear();
+                            values.clear();
                         }
                     }
                     m_deleted = size - written;
@@ -335,6 +370,56 @@ namespace SPTAG
                 UpdateDeleteCount(oldVal, storedVersion);
             }
 
+            // Per-VID batch write: mirrors SetVersion() for each (vid, ver) pair.
+            // Uses TiKVIO MultiPut so the writes are grouped per TiKV region
+            // and issued in parallel. m_deleted accounting is approximate
+            // here (we do not read the old byte to compute the exact delta);
+            // GetDeleteCount() returns 0 for the TiKV-backed version map so
+            // this approximation is acceptable. Callers that need precise
+            // accounting can call SetVersion() per-VID instead.
+            void SetVersionBatch(const std::vector<SizeType>& vids, const std::vector<uint8_t>& versions) override
+            {
+                size_t n = std::min(vids.size(), versions.size());
+                if (n == 0) return;
+
+                SizeType count = m_count.load();
+                std::vector<std::string> keys;
+                std::vector<std::string> values;
+                keys.reserve(n);
+                values.reserve(n);
+                for (size_t i = 0; i < n; ++i) {
+                    if (vids[i] < 0 || vids[i] >= count) {
+                        SPTAGLIB_LOG(Helper::LogLevel::LL_Error,
+                            "TiKVVersionMap::SetVersionBatch: invalid key %d (max %d)\n",
+                            vids[i], count);
+                        continue;
+                    }
+                    keys.push_back(VersionKey(vids[i]));
+                    values.push_back(std::string(1, static_cast<char>(versions[i])));
+                }
+                if (keys.empty()) return;
+
+                auto ret = m_db->MultiPut(keys, values, MaxTimeout, nullptr);
+                if (ret == ErrorCode::Undefined) {
+                    // Backend lacks MultiPut: fall back to serial SetVersion
+                    // which preserves m_deleted accounting.
+                    for (size_t i = 0; i < n; ++i) {
+                        if (vids[i] >= 0 && vids[i] < count) {
+                            SetVersion(vids[i], versions[i]);
+                        }
+                    }
+                } else if (ret != ErrorCode::Success) {
+                    SPTAGLIB_LOG(Helper::LogLevel::LL_Warning,
+                        "TiKVVersionMap::SetVersionBatch: MultiPut failed layer=%d ret=%d keys=%zu; falling back to per-VID SetVersion.\n",
+                        m_layer, static_cast<int>(ret), keys.size());
+                    for (size_t i = 0; i < n; ++i) {
+                        if (vids[i] >= 0 && vids[i] < count) {
+                            SetVersion(vids[i], versions[i]);
+                        }
+                    }
+                }
+            }
+
             bool IncVersion(const SizeType& key, uint8_t* newVersion, uint8_t expectedOld = 0xff) override
             {
                 if (key < 0 || key >= m_count.load()) {