Merge upstream/users/qiazh/pre-merge-tikv-bugfix into users/zhangt/merge-distributed-to-tikv

Resolve conflicts preserving features from both sides:
- HEAD: distributed worker/dispatcher/router, AsyncAppend, MergeAsyncJob(3-arg),
  m_asyncStatus, headVec layout with metaDataSize prefix, NumSearchDuringInsertThreads
- qiazh: VersionReadPolicy, TiKV PutBaseChunkAndCount/AsyncGetPostingCounts batch
  RPCs, AddIndexAsync* (multi-chunk + single-key), Append RMW data-loss fix
  (distinguish Key_NotFound from errors), GetCachedPostingCount -1 sentinel,
  extensive DIAG stats and histograms (reassignJobLatency, queueDepth,
  splitNewHeadCount, MC histograms), batch barrier wait timing.

Build verification: Release/SPTAGTest builds cleanly with -DTIKV=ON -DTBB=ON.
Added absl_synchronization to SPTAGTest link deps (qiazh's main.cpp uses
absl::SetMutexDeadlockDetectionMode).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: TerrenceZhangX

AnnService/inc/Core/Common/FineGrainedLock.h

@@ -8,6 +8,7 @@
 #include <vector>
 #include <mutex>
 #include <memory>
+#include <unordered_map>
 
 namespace SPTAG
 {
@@ -43,27 +44,47 @@ namespace SPTAG
         class FineGrainedRWLock {
         public:
             FineGrainedRWLock() {
-                m_locks.reset(new std::shared_timed_mutex[PoolSize + 1]);
+                m_buckets.reset(new Bucket[BucketCount]);
             }
             ~FineGrainedRWLock() {}
 
             std::shared_timed_mutex& operator[](SizeType idx) {
-                unsigned index = hash_func((unsigned)idx);
-                return m_locks[index];
+                return GetLock(idx);
             }
 
             const std::shared_timed_mutex& operator[](SizeType idx) const {
-                unsigned index = hash_func((unsigned)idx);
-                return m_locks[index];
+                return GetLock(idx);
             }
 
             static inline unsigned hash_func(unsigned idx)
             {
-                return ((unsigned)(idx * 99991) + _rotl(idx, 2) + 101) & PoolSize;
+                return idx;
             }
         private:
-            static const int PoolSize = 32767;
-            std::unique_ptr<std::shared_timed_mutex[]> m_locks;
+            struct Bucket {
+                std::mutex mutex;
+                std::unordered_map<SizeType, std::unique_ptr<std::shared_timed_mutex>> locks;
+            };
+
+            std::shared_timed_mutex& GetLock(SizeType idx) const {
+                Bucket& bucket = m_buckets[BucketIndex(idx)];
+                std::lock_guard<std::mutex> guard(bucket.mutex);
+                auto iter = bucket.locks.find(idx);
+                if (iter == bucket.locks.end()) {
+                    iter = bucket.locks.emplace(idx, std::unique_ptr<std::shared_timed_mutex>(new std::shared_timed_mutex())).first;
+                }
+                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

@@ -15,6 +15,12 @@ namespace SPTAG
 {
     namespace COMMON
     {
+        enum class VersionReadPolicy
+        {
+            UseCache,
+            BypassCacheNoFill
+        };
+
         /// Abstract interface for version map, allowing both local (in-memory)
         /// and distributed (TiKV-backed) implementations.
         class IVersionMap
@@ -31,9 +37,11 @@ namespace SPTAG
             virtual std::uint64_t BufferSize() const = 0;
 
             virtual bool Deleted(const SizeType& key) const = 0;
+            virtual bool Deleted(const SizeType& key, VersionReadPolicy policy) const { return Deleted(key); }
             virtual bool Delete(const SizeType& key) = 0;
 
             virtual uint8_t GetVersion(const SizeType& key) = 0;
+            virtual uint8_t GetVersion(const SizeType& key, VersionReadPolicy policy) { return GetVersion(key); }
             virtual void SetVersion(const SizeType& key, const uint8_t& version) = 0;
             /// Increment the version of a VID.
             /// @param expectedOld If not 0xff, the caller asserts the current version should be this value.
@@ -55,13 +63,18 @@ namespace SPTAG
             /// Returns a vector of versions (0xfe = deleted) in the same order as vids.
             /// Default implementation does per-VID lookup.
             virtual void BatchGetVersions(const std::vector<SizeType>& vids, std::vector<uint8_t>& versions)
+            {
+                BatchGetVersions(vids, versions, VersionReadPolicy::UseCache);
+            }
+
+            virtual void BatchGetVersions(const std::vector<SizeType>& vids, std::vector<uint8_t>& versions, VersionReadPolicy policy)
             {
                 versions.resize(vids.size());
                 for (size_t i = 0; i < vids.size(); i++) {
                     if (vids[i] < 0 || vids[i] >= Count()) {
                         versions[i] = 0xfe;
                     } else {
-                        versions[i] = GetVersion(vids[i]);
+                        versions[i] = GetVersion(vids[i], policy);
                     }
                 }
             }

AnnService/inc/Core/Common/TiKVVersionMap.h

@@ -48,25 +48,22 @@ namespace SPTAG
             struct CachedChunk {
                 SizeType chunkId;
                 std::string data;
-                std::chrono::steady_clock::time_point fetchTime;
             };
             using LruList = std::list<CachedChunk>;
             mutable std::shared_mutex m_cacheMutex;
             mutable LruList m_lruList;
             mutable std::unordered_map<SizeType, LruList::iterator> m_cacheMap;
-            int m_cacheTTLMs{0};        // TTL in ms, 0 = cache disabled
-            int m_cacheMaxChunks{10000}; // max cached chunks (~40MB at 4096 chunk size)
+            int m_cacheMaxChunks{10000}; // max cached chunks; <= 0 disables caching
 
             // Insert or update a chunk in the LRU cache. Caller must hold exclusive m_cacheMutex.
-            void CachePut(SizeType chunkId, const std::string& data, std::chrono::steady_clock::time_point now) const
+            void CachePut(SizeType chunkId, const std::string& data) const
             {
                 if (m_cacheMaxChunks <= 0) return; // cache disabled
 
                 auto it = m_cacheMap.find(chunkId);
                 if (it != m_cacheMap.end()) {
                     // Update existing: move to front
                     it->second->data = data;
-                    it->second->fetchTime = now;
                     m_lruList.splice(m_lruList.begin(), m_lruList, it->second);
                 } else {
                     // Evict LRU entries if at capacity
@@ -75,7 +72,7 @@ namespace SPTAG
                         m_cacheMap.erase(back.chunkId);
                         m_lruList.pop_back();
                     }
-                    m_lruList.push_front({chunkId, data, now});
+                    m_lruList.push_front({chunkId, data});
                     m_cacheMap[chunkId] = m_lruList.begin();
                 }
             }
@@ -112,7 +109,7 @@ namespace SPTAG
                 auto ret = m_db->Put(ChunkKey(chunkId), data, MaxTimeout, nullptr);
                 if (ret == ErrorCode::Success) {
                     std::unique_lock<std::shared_mutex> lock(m_cacheMutex);
-                    CachePut(chunkId, data, std::chrono::steady_clock::now());
+                    CachePut(chunkId, data);
                 }
                 return ret;
             }
@@ -136,16 +133,17 @@ namespace SPTAG
                 std::string data = ReadChunk(chunkId);
                 if (!data.empty()) {
                     std::unique_lock<std::shared_mutex> lock(m_cacheMutex);
-                    CachePut(chunkId, data, std::chrono::steady_clock::now());
+                    CachePut(chunkId, data);
                 }
                 return data;
             }
 
-            // Read a single byte for a VID using cache. Returns 0xfe on error/miss.
-            uint8_t ReadVersionByte(SizeType vid) const
+            // Read a single byte for a VID. Returns 0xfe on error/miss.
+            uint8_t ReadVersionByte(SizeType vid, VersionReadPolicy policy = VersionReadPolicy::UseCache) const
             {
                 SizeType cid = ChunkId(vid);
-                std::string chunk = ReadChunkCached(cid);
+                std::string chunk = (policy == VersionReadPolicy::BypassCacheNoFill) ?
+                    ReadChunk(cid) : ReadChunkCached(cid);
                 if (chunk.empty() || (int)chunk.size() <= ChunkOffset(vid)) {
                     return 0xfe;
                 }
@@ -190,7 +188,6 @@ namespace SPTAG
             void SetDB(std::shared_ptr<Helper::KeyValueIO> db) { m_db = db; }
             void SetLayer(int layer) { m_layer = layer; }
             void SetChunkSize(int chunkSize) { m_chunkSize = chunkSize; }
-            void SetCacheTTL(int ttlMs) { m_cacheTTLMs = ttlMs; }
             void SetCacheMaxChunks(int maxChunks) { m_cacheMaxChunks = maxChunks; }
 
             std::shared_ptr<Helper::KeyValueIO> GetDB() const { return m_db; }
@@ -293,12 +290,17 @@ namespace SPTAG
             std::uint64_t BufferSize() const override { return static_cast<std::uint64_t>(m_count.load()) + sizeof(SizeType); }
 
             bool Deleted(const SizeType& key) const override
+            {
+                return Deleted(key, VersionReadPolicy::UseCache);
+            }
+
+            bool Deleted(const SizeType& key, VersionReadPolicy policy) const override
             {
                 if (key < 0 || key >= m_count.load()) {
                     SPTAGLIB_LOG(Helper::LogLevel::LL_Error, "TiKVVersionMap::Deleted: invalid key %d (max %d)\n", key, m_count.load());
                     return true;
                 }
-                return ReadVersionByte(key) == 0xfe;
+                return ReadVersionByte(key, policy) == 0xfe;
             }
 
             bool Delete(const SizeType& key) override
@@ -323,12 +325,17 @@ namespace SPTAG
             }
 
             uint8_t GetVersion(const SizeType& key) override
+            {
+                return GetVersion(key, VersionReadPolicy::UseCache);
+            }
+
+            uint8_t GetVersion(const SizeType& key, VersionReadPolicy policy) override
             {
                 if (key < 0 || key >= m_count.load()) {
                     SPTAGLIB_LOG(Helper::LogLevel::LL_Error, "TiKVVersionMap::GetVersion: invalid key %d (max %d)\n", key, m_count.load());
                     return 0xfe;
                 }
-                return ReadVersionByte(key);
+                return ReadVersionByte(key, policy);
             }
 
             void SetVersion(const SizeType& key, const uint8_t& version) override
@@ -451,13 +458,18 @@ namespace SPTAG
             /// Batch version lookup with local cache support.
             /// Checks cache first, only fetches misses from TiKV via BatchGet.
             void BatchGetVersions(const std::vector<SizeType>& vids, std::vector<uint8_t>& versions) override
+            {
+                BatchGetVersions(vids, versions, VersionReadPolicy::UseCache);
+            }
+
+            void BatchGetVersions(const std::vector<SizeType>& vids, std::vector<uint8_t>& versions, VersionReadPolicy policy) override
             {
                 versions.resize(vids.size());
                 if (vids.empty()) return;
 
                 SizeType count = m_count.load();
-                auto now = std::chrono::steady_clock::now();
-                bool cacheEnabled = (m_cacheTTLMs > 0);
+                bool bypassCache = (policy == VersionReadPolicy::BypassCacheNoFill);
+                bool cacheEnabled = (m_cacheMaxChunks > 0 && !bypassCache);
 
                 // Group VIDs by chunk
                 std::unordered_map<SizeType, std::vector<size_t>> chunkToIndices;
@@ -479,11 +491,8 @@ namespace SPTAG
                     for (auto& [cid, indices] : chunkToIndices) {
                         auto it = m_cacheMap.find(cid);
                         if (it != m_cacheMap.end()) {
-                            auto ageMs = std::chrono::duration_cast<std::chrono::milliseconds>(now - it->second->fetchTime).count();
-                            if (ageMs < m_cacheTTLMs) {
-                                resolvedChunks[cid] = it->second->data; // copy
-                                continue;
-                            }
+                            resolvedChunks[cid] = it->second->data; // copy
+                            continue;
                         }
                         missChunkIds.push_back(cid);
                     }
@@ -522,11 +531,11 @@ namespace SPTAG
                                      m_layer, missingChunks, static_cast<int>(missChunkIds.size()), missChunkIds[0]);
                     }
 
-                    // Update LRU cache with fetched chunks (exclusive lock)
+                    // Update LRU cache with fetched chunks (exclusive lock). Bypass reads never fill cache.
                     if (cacheEnabled && !fetchedChunks.empty()) {
                         std::unique_lock<std::shared_mutex> lock(m_cacheMutex);
                         for (auto& [cid, data] : fetchedChunks) {
-                            CachePut(cid, data, now);
+                            CachePut(cid, data);
                         }
                     }
                 }