[SVS][PoC] 2-steps vectors update in SVSTiered

cmake/svs.cmake

@@ -64,7 +64,7 @@ if(USE_SVS)
     if (CMAKE_CXX_COMPILER_ID MATCHES "Clang")
         if(GLIBC_VERSION VERSION_GREATER_EQUAL "2.31" AND CMAKE_CXX_COMPILER_VERSION VERSION_GREATER_EQUAL "21.0")
             if(GLIBCXX_VERSION VERSION_GREATER_EQUAL "11")
-                set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/v0.3.1/svs-shared-library-reduced-clang21-gcc11.tar.gz" CACHE STRING "SVS URL")
+                set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/nightly/svs-shared-library-reduced-clang21-gcc11-pr330.tar.gz" CACHE STRING "SVS URL")
             else()
                 message(STATUS "libstdc++ >= GCC 11 is required for Clang SVS binaries - disabling SVS_SHARED_LIB")
                 set(SVS_SHARED_LIB OFF)
@@ -76,14 +76,12 @@ if(USE_SVS)
     else()
         if(GLIBC_VERSION VERSION_GREATER_EQUAL "2.28")
             if(CMAKE_CXX_COMPILER_VERSION VERSION_GREATER_EQUAL "14.0")
-                set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/v0.3.1/svs-shared-library-reduced-gcc14.tar.gz" CACHE STRING "SVS URL")
+                set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/nightly/svs-shared-library-reduced-gcc14-pr330.tar.gz" CACHE STRING "SVS URL")
             else()
-                set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/v0.3.1/svs-shared-library-reduced.tar.gz" CACHE STRING "SVS URL")
+                set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/nightly/svs-shared-library-reduced-pr330.tar.gz" CACHE STRING "SVS URL")
             endif()
-        elseif(GLIBC_VERSION VERSION_GREATER_EQUAL "2.26")
-            set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/v0.3.0/svs-shared-library-0.3.0-reduced-glibc2_26.tar.gz" CACHE STRING "SVS URL")
         else()
-            message(STATUS "GLIBC >= 2.26 is required for GCC SVS binaries - disabling SVS_SHARED_LIB")
+            message(STATUS "GLIBC >= 2.28 is required for GCC SVS binaries - disabling SVS_SHARED_LIB")
             set(SVS_SHARED_LIB OFF)
         endif()
     endif()

src/VecSim/algorithms/svs/svs.h

@@ -57,6 +57,10 @@ struct SVSIndexBase
     virtual std::unique_ptr<ImplHandler> createImpl(const void *vectors_data,
                                                     const labelType *labels, size_t n) = 0;
     virtual void setImpl(std::unique_ptr<ImplHandler> impl) = 0;
+
+    virtual std::unique_ptr<ImplHandler>
+    makeAddVectorsChanges(const void *vectors_data, const labelType *labels, size_t n) = 0;
+    virtual int applyAddVectorsChanges(std::unique_ptr<ImplHandler> changes) = 0;
 #ifdef BUILD_TESTS
     virtual svs::logging::logger_ptr getLogger() const = 0;
 #endif
@@ -252,6 +256,68 @@ class SVSIndex : public VecSimIndexAbstract<svs_details::vecsim_dt<DataType>, fl
         this->impl_ = std::move(svs_handler->impl);
     }
 
+    struct SVSChangesHandler : public ImplHandler {
+        using points_type = svs::data::SimpleDataView<DataType>;
+        using ids_type = std::span<const labelType>;
+        using changes_type = decltype(std::declval<impl_type>().add_points_compute_changes(
+            std::declval<points_type>(), std::declval<ids_type>()));
+
+        template <typename Impl>
+        SVSChangesHandler(MemoryUtils::unique_blob processed_blob, points_type points, ids_type ids,
+                          const Impl &impl, int result_num)
+            : processed_blob_{std::move(processed_blob)}, points_{std::move(points)},
+              ids_{std::move(ids)}, changes_{impl.add_points_compute_changes(points_, ids_)},
+              result_num_{result_num} {}
+        virtual ~SVSChangesHandler() = default;
+
+        MemoryUtils::unique_blob processed_blob_;
+        points_type points_;
+        ids_type ids_;
+        changes_type changes_;
+        int result_num_;
+    };
+
+    virtual std::unique_ptr<ImplHandler>
+    makeAddVectorsChanges(const void *vectors_data, const labelType *labels, size_t n) override {
+        assert(impl_ != nullptr);
+        if (n == 0) {
+            return nullptr;
+        }
+
+        int deleted_num = 0;
+        std::span<const labelType> ids(labels, n);
+        if constexpr (!isMulti) {
+            // SVS index does not support overriding vectors with the same label
+            // and we cannot delete vectors now, so these had to be deleted in advance.
+            // We expect that the caller (SVSTiered) has already deleted any existing labels from
+            // SVS index before calling this function inside addVector(). so just use assert() to
+            // check that no existing labels are present in the input.
+            assert(!std::any_of(ids.begin(), ids.end(),
+                                [this](labelType label) { return impl_->has_id(label); }));
+        }
+
+        auto processed_blob = this->preprocessForBatchStorage(vectors_data, n);
+        auto typed_vectors_data = static_cast<DataType *>(processed_blob.get());
+        // Wrap data into SVS SimpleDataView for SVS API
+        auto points = svs::data::SimpleDataView<DataType>{typed_vectors_data, n, this->dim};
+        return std::make_unique<SVSChangesHandler>(std::move(processed_blob), points, ids, *impl_,
+                                                   n - deleted_num);
+    }
+
+    virtual int applyAddVectorsChanges(std::unique_ptr<ImplHandler> changes) override {
+        assert(impl_ != nullptr);
+        if (!changes) {
+            return 0;
+        }
+
+        SVSChangesHandler *svs_changes_handler = dynamic_cast<SVSChangesHandler *>(changes.get());
+        if (!svs_changes_handler) {
+            throw std::logic_error("Failed to cast to SVSChangesHandler");
+        }
+        impl_->add_points_commit(svs_changes_handler->ids_, svs_changes_handler->changes_);
+        return svs_changes_handler->result_num_;
+    }
+
     // Assuming parallelism was updated to reflect the number of available threads before this
     // function was called.
     // This function assumes that the caller has already set parallelism to the appropriate value
@@ -587,29 +653,58 @@ class SVSIndex : public VecSimIndexAbstract<svs_details::vecsim_dt<DataType>, fl
         auto processed_query_ptr = this->preprocessQuery(queryBlob);
         const void *processed_query = processed_query_ptr.get();
 
-        auto query = svs::data::ConstSimpleDataView<DataType>{
-            static_cast<const DataType *>(processed_query), 1, this->dim};
-        auto result = svs::QueryResult<size_t>{query.size(), k};
         auto sp = svs_details::joinSearchParams(impl_->get_search_parameters(), queryParams,
                                                 is_two_level_lvq);
 
         auto timeoutCtx = queryParams ? queryParams->timeoutCtx : nullptr;
         auto cancel = [timeoutCtx]() { return VECSIM_TIMEOUT(timeoutCtx); };
 
-        impl_->search(result.view(), query, sp, cancel);
-        if (cancel()) {
-            rep->code = VecSim_QueryReply_TimedOut;
-            return rep;
-        }
+        if constexpr (isMulti) {
+            auto query = svs::data::ConstSimpleDataView<DataType>{
+                static_cast<const DataType *>(processed_query), 1, this->dim};
+            auto result = svs::QueryResult<size_t>{query.size(), k};
+            impl_->search(result.view(), query, sp, cancel);
+            if (cancel()) {
+                rep->code = VecSim_QueryReply_TimedOut;
+                return rep;
+            }
+
+            assert(result.n_queries() == 1);
 
-        assert(result.n_queries() == 1);
+            const auto n_neighbors = result.n_neighbors();
+            rep->results.reserve(n_neighbors);
 
-        const auto n_neighbors = result.n_neighbors();
-        rep->results.reserve(n_neighbors);
+            for (size_t i = 0; i < n_neighbors; i++) {
+                rep->results.push_back(
+                    VecSimQueryResult{result.index(0, i), toVecSimDistance(result.distance(0, i))});
+            }
+        } else {
+            // Use the single-query search path of MutableVamanaIndex to avoid the
+            // parallel batch-search overhead for a single query.
+            auto scratch = impl_->scratchspace(sp);
+            // Ensure the scratch buffer can hold at least `k` neighbors.
+            if (scratch.buffer.target_capacity() < k) {
+                scratch.buffer.change_maxsize(k);
+            }
 
-        for (size_t i = 0; i < n_neighbors; i++) {
-            rep->results.push_back(
-                VecSimQueryResult{result.index(0, i), toVecSimDistance(result.distance(0, i))});
+            auto query = std::span<const DataType>{static_cast<const DataType *>(processed_query),
+                                                   this->dim};
+            impl_->search(query, scratch, cancel);
+            if (cancel()) {
+                rep->code = VecSim_QueryReply_TimedOut;
+                return rep;
+            }
+
+            const auto &buffer = scratch.buffer;
+            const auto n_neighbors = std::min(k, buffer.size());
+            rep->results.reserve(n_neighbors);
+
+            for (size_t i = 0; i < n_neighbors; i++) {
+                const auto neighbor = buffer[i];
+                rep->results.push_back(
+                    VecSimQueryResult{impl_->translate_internal_id(neighbor.id()),
+                                      toVecSimDistance(neighbor.distance())});
+            }
         }
         // Workaround for VecSim merge_results() that expects results to be sorted
         // by score, then by id from both indices.

src/VecSim/algorithms/svs/svs_tiered.h

@@ -711,13 +711,14 @@ class TieredSVSIndex : public VecSimTieredIndex<DataType, float> {
                 std::lock_guard lock(this->mainIndexGuard);
                 svs_index->setImpl(std::move(impl));
             } else {
+                svs_index->setParallelism(std::min(availableThreads, labels_to_move.size()));
                 // Backend index is initialized - just add the vectors.
+                auto changes = svs_index->makeAddVectorsChanges(
+                    vectors_to_move.data(), labels_to_move.data(), labels_to_move.size());
+                // Upgrade to unique lock to add vectors
                 main_shared_lock.unlock();
                 std::lock_guard lock(this->mainIndexGuard);
-                // Upgrade to unique lock to add vectors
-                svs_index->setParallelism(std::min(availableThreads, labels_to_move.size()));
-                svs_index->addVectors(vectors_to_move.data(), labels_to_move.data(),
-                                      labels_to_move.size());
+                svs_index->applyAddVectorsChanges(std::move(changes));
             }
         }
         executeTracingCallback("UpdateJob::after_add_to_svs");

tests/benchmark/bm_initialization/bm_basics_svs_initialize_fp32.h

@@ -68,3 +68,19 @@ BENCHMARK_REGISTER_F(BM_VecSimSVS, BM_FUNC_NAME(BM_TopK))
     ->Args({200, 100})
     ->Args({500, 500})
     ->ArgNames({"window_size", "k"});
+
+// Parallel TopK searches running concurrently with a background update job.
+// Uses constant window_size=200 and k=100.
+// {update_threshold, n_parallel_searches, thread_count}
+BENCHMARK_TEMPLATE_DEFINE_F(BM_VecSimSVS, BM_FUNC_NAME(BM_TopKSearchDuringUpdate),
+                            DATA_TYPE_INDEX_T)
+(benchmark::State &st) { TopKSearchDuringUpdate(st); }
+BENCHMARK_REGISTER_F(BM_VecSimSVS, BM_FUNC_NAME(BM_TopKSearchDuringUpdate))
+    ->Unit(benchmark::kMillisecond)
+    ->Iterations(1)
+    ->ArgsProduct({{static_cast<long int>(BM_VecSimGeneral::block_size),
+                    static_cast<long int>(10 * BM_VecSimGeneral::block_size)},
+                   {10, 50},
+                   {2, 4}})
+    ->ArgNames({"update_threshold", "n_parallel_searches", "thread_count"})
+    ->UseRealTime();

tests/benchmark/bm_vecsim_svs.h

@@ -58,6 +58,11 @@ class BM_VecSimSVS : public BM_VecSimGeneral {
     // registration.
     void TopK_SVS(benchmark::State &st);
 
+    // Runs a number of parallel TopK searches while a background update job (moving vectors from
+    // the flat buffer into the SVS backend index) is in progress. Measures the time to complete
+    // all searches concurrently with the update.
+    void TopKSearchDuringUpdate(benchmark::State &st);
+
 private:
     static const char *svs_index_tar_file;
     static std::string base_path;
@@ -492,6 +497,68 @@ void BM_VecSimSVS<index_type_t>::TopK_SVS(benchmark::State &st) {
     VecSimIndex_Free(index);
 }
 
+template <typename index_type_t>
+void BM_VecSimSVS<index_type_t>::TopKSearchDuringUpdate(benchmark::State &st) {
+    // ensure mode is async
+    ASSERT_EQ(VecSimIndexInterface::asyncWriteMode, VecSim_WriteAsync);
+
+    const size_t window_size = 200;
+    const size_t k = 100;
+    size_t update_threshold = st.range(0);
+    size_t n_parallel_searches = st.range(1);
+    int unsigned num_threads = st.range(2);
+
+    if (num_threads > std::thread::hardware_concurrency()) {
+        GTEST_SKIP() << "Not enough threads available, skipping test...";
+    }
+
+    // Ensure we have enough vectors to fill the flat buffer and to run the searches.
+    ASSERT_GE(N_QUERIES, update_threshold);
+
+    auto mock_thread_pool = tieredIndexMock(num_threads);
+    ASSERT_EQ(mock_thread_pool.thread_pool_size, num_threads);
+    auto *tiered_index = CreateTieredSVSIndexFromFile(mock_thread_pool, update_threshold);
+
+    // Fill the flat buffer up to just below the update threshold, so a single AddVector will
+    // trigger the background update job.
+    for (size_t i = 0; i < update_threshold - 1; ++i) {
+        int ret = VecSimIndex_AddVector(tiered_index, test_vectors[i].data(), i + N_VECTORS);
+        ASSERT_EQ(ret, 1);
+    }
+
+    mock_thread_pool.init_threads();
+
+    for (auto _ : st) {
+        // Trigger the background update job by reaching the update threshold. It runs on the
+        // index's own thread pool.
+        int ret = VecSimIndex_AddVector(tiered_index, test_vectors[update_threshold - 1].data(),
+                                        update_threshold - 1 + N_VECTORS);
+        ASSERT_EQ(ret, 1);
+
+        // Run the TopK searches on independent threads, asynchronously to the index thread pool,
+        // so they execute concurrently with the ongoing background update.
+        std::vector<std::thread> search_threads;
+        search_threads.reserve(n_parallel_searches);
+        for (size_t i = 0; i < n_parallel_searches; ++i) {
+            search_threads.emplace_back([tiered_index, i, window_size, k]() {
+                SVSRuntimeParams svs_params = {.windowSize = window_size};
+                VecSimQueryParams query_params = {.svsRuntimeParams = svs_params};
+                auto results = VecSimIndex_TopKQuery(
+                    tiered_index, test_vectors[i % N_QUERIES].data(), k, &query_params, BY_SCORE);
+                VecSimQueryReply_Free(results);
+            });
+        }
+        for (auto &t : search_threads) {
+            t.join();
+        }
+
+        // Wait for the background update to complete.
+        mock_thread_pool.thread_pool_wait();
+    }
+
+    mock_thread_pool.thread_pool_join();
+}
+
 #define UNIT_AND_ITERATIONS Unit(benchmark::kMillisecond)->Iterations(2)
 
 #if HAVE_SVS_LVQ