[MOD-11765] streaming merge bench

src/VecSim/algorithms/hnsw/hnsw_disk_serializer.h

@@ -56,7 +56,7 @@ HNSWDiskIndex<DataType, DistType>::HNSWDiskIndex(
     : VecSimIndexAbstract<DataType, DistType>(abstractInitParams, components),
       idToMetaData(this->allocator), labelToIdMap(this->allocator), db(db), cf(cf), dbPath(""),
       indexDataGuard(), visitedNodesHandlerPool(INITIAL_CAPACITY, this->allocator),
-      delta_list(), new_elements_meta_data(this->allocator), batchThreshold(0), // Will be restored from file
+      new_elements_meta_data(this->allocator), batchThreshold(0), // Will be restored from file
       pendingVectorIds(this->allocator), pendingMetadata(this->allocator),
       pendingVectorCount(0), pendingDeleteIds(this->allocator),
       stagedInsertUpdates(this->allocator),
@@ -214,14 +214,14 @@ template <typename DataType, typename DistType>
 void HNSWDiskIndex<DataType, DistType>::restoreVectors(std::ifstream &input, EncodingVersion version) {
 // #ifdef HNSW_DISK_SERIALIZE_VECTORS_TO_FILE
     // NEW METHOD: Load vectors from metadata file
-    this->log(VecSimCommonStrings::LOG_VERBOSE_STRING,
-             "Loading vectors from metadata file (HNSW_DISK_SERIALIZE_VECTORS_TO_FILE enabled)");
-    restoreVectorsFromFile(input, version);
+    // this->log(VecSimCommonStrings::LOG_VERBOSE_STRING,
+    //          "Loading vectors from metadata file (HNSW_DISK_SERIALIZE_VECTORS_TO_FILE enabled)");
+    // restoreVectorsFromFile(input, version);
 // #else
-//     // CURRENT METHOD: Load vectors from RocksDB (default for backward compatibility)
-//     this->log(VecSimCommonStrings::LOG_VERBOSE_STRING,
-//              "Loading vectors from RocksDB checkpoint (default method)");
-//     restoreVectorsFromRocksDB(version);
+    // CURRENT METHOD: Load vectors from RocksDB (default for backward compatibility)
+    this->log(VecSimCommonStrings::LOG_VERBOSE_STRING,
+             "Loading vectors from RocksDB checkpoint (default method)");
+    restoreVectorsFromRocksDB(version);
 // #endif
 }
 
@@ -287,11 +287,7 @@ void HNSWDiskIndex<DataType, DistType>::saveIndexIMP(std::ofstream &output) {
     if (pendingDeleteIds.size() != 0) {
         throw std::runtime_error("Serialization error: pendingDeleteIds not empty after flush");
     }
-    // Note: delta_list and new_elements_meta_data are currently unused legacy variables
-    // but we verify them for future-proofing
-    if (!delta_list.empty()) {
-        throw std::runtime_error("Serialization error: delta_list not empty after flush");
-    }
+
     if (!new_elements_meta_data.empty()) {
         throw std::runtime_error("Serialization error: new_elements_meta_data not empty after flush");
     }

src/VecSim/index_factories/components/components_factory.h

@@ -59,8 +59,10 @@ CreateQuantizedIndexComponents(std::shared_ptr<VecSimAllocator> allocator, VecSi
     // Use INT8 distance function for quantized vectors
     spaces::dist_func_t<DistType> distFunc =
         spaces::GetDistFunc<int8_t, DistType>(distance_metric, dim, &alignment);
+    spaces::dist_func_t<DistType> rawDistFunc =
+        spaces::GetDistFunc<DataType, DistType>(distance_metric, dim, &alignment);
 
-    auto indexCalculator = new (allocator) DistanceCalculatorCommon<DistType>(allocator, distFunc);
+    auto indexCalculator = new (allocator) DistanceCalculatorQuantized<DistType>(allocator, distFunc, rawDistFunc);
 
     // Create preprocessor container with space for 2 preprocessors (normalization + quantization)
     auto preprocessors =

src/VecSim/index_factories/hnsw_disk_factory.cpp

@@ -191,12 +191,12 @@ class ManagedRocksDB {
                                    std::string(e.what()));
         }
 
-        // Open RocksDB from the temp checkpoint copy
-        // All writes (WAL, SST, MANIFEST, etc.) will go to the temp location
-        rocksdb::Options options;
-        options.create_if_missing = false;  // Checkpoint copy should exist
-        options.error_if_exists = false;
-        options.statistics = rocksdb::CreateDBStatistics();
+    // Open RocksDB from the temp checkpoint copy
+    // All writes (WAL, SST, MANIFEST, etc.) will go to the temp location
+    rocksdb::Options options;
+    options.create_if_missing = false;  // Checkpoint copy should exist
+    options.error_if_exists = false;
+    options.statistics = rocksdb::CreateDBStatistics();
 
         rocksdb::DB *db_ptr = nullptr;
         rocksdb::Status status = rocksdb::DB::Open(options, temp_checkpoint, &db_ptr);
@@ -223,10 +223,10 @@ class ManagedRocksDB {
         return instance;
     }
 
-    // Destructor: closes DB and optionally cleans up temp directory
+// Destructor: closes DB and optionally cleans up temp directory
     ~ManagedRocksDB() {
-        // Close DB first (unique_ptr handles this automatically)
-        db.reset();
+    // Close DB first (unique_ptr handles this automatically)
+    db.reset();
 
         // Delete temp directory only if it's actually temporary
         if (cleanup_temp_dir && !temp_dir.empty() && std::filesystem::exists(temp_dir)) {
@@ -257,8 +257,7 @@ static std::unique_ptr<ManagedRocksDB> managed_rocksdb;
 static AbstractIndexInitParams NewAbstractInitParams(const VecSimParams *params) {
     const HNSWParams *hnswParams = &params->algoParams.hnswParams;
 
-    size_t dataSize =
-        VecSimParams_GetDataSize(hnswParams->type, hnswParams->dim, hnswParams->metric);
+    size_t dataSize = hnswParams->dim * sizeof(int8_t); // Quantized storage
     AbstractIndexInitParams abstractInitParams = {.allocator =
                                                       VecSimAllocator::newVecsimAllocator(),
                                                   .dim = hnswParams->dim,

src/VecSim/spaces/computer/calculator.h

@@ -23,6 +23,7 @@ class IndexCalculatorInterface : public VecsimBaseObject {
     virtual ~IndexCalculatorInterface() = default;
 
     virtual DistType calcDistance(const void *v1, const void *v2, size_t dim) const = 0;
+    virtual DistType calcDistanceRaw(const void *v1, const void *v2, size_t dim) const = 0;
 };
 
 /**
@@ -39,7 +40,7 @@ class DistanceCalculatorInterface : public IndexCalculatorInterface<DistType> {
     DistanceCalculatorInterface(std::shared_ptr<VecSimAllocator> allocator, DistFuncType dist_func)
         : IndexCalculatorInterface<DistType>(allocator), dist_func(dist_func) {}
     virtual DistType calcDistance(const void *v1, const void *v2, size_t dim) const = 0;
-
+    virtual DistType calcDistanceRaw(const void *v1, const void *v2, size_t dim) const = 0;
 protected:
     DistFuncType dist_func;
 };
@@ -56,4 +57,28 @@ class DistanceCalculatorCommon
     DistType calcDistance(const void *v1, const void *v2, size_t dim) const override {
         return this->dist_func(v1, v2, dim);
     }
+    DistType calcDistanceRaw(const void *v1, const void *v2, size_t dim) const override {
+        return this->dist_func(v1, v2, dim);
+    }
+};
+
+template <typename DistType>
+class DistanceCalculatorQuantized
+    : public DistanceCalculatorInterface<DistType, spaces::dist_func_t<DistType>> {
+protected:
+    spaces::dist_func_t<DistType> raw_dist_func;
+
+public:
+    DistanceCalculatorQuantized(std::shared_ptr<VecSimAllocator> allocator,
+                                spaces::dist_func_t<DistType> quant_dist_func, spaces::dist_func_t<DistType> raw_dist_func)
+        : DistanceCalculatorInterface<DistType, spaces::dist_func_t<DistType>>(allocator,
+                                                                               quant_dist_func),
+          raw_dist_func(raw_dist_func) {}
+
+    DistType calcDistance(const void *v1, const void *v2, size_t dim) const override {
+        return this->dist_func(v1, v2, dim);
+    }
+    DistType calcDistanceRaw(const void *v1, const void *v2, size_t dim) const {
+        return this->raw_dist_func(v1, v2, dim) * 16129; // multiply by 127^2
+    }
 };

src/VecSim/spaces/computer/preprocessors.h

@@ -13,6 +13,7 @@
 #include <memory>
 #include <cassert>
 #include <cmath>
+#include <iostream>
 
 #include "VecSim/memory/vecsim_base.h"
 #include "VecSim/spaces/spaces.h"
@@ -216,5 +217,10 @@ class ScalarQuantizationPreprocessor : public PreprocessorInterface {
                 output_vec[i] = static_cast<int8_t>(std::round(scaled));
             }
         }
+
+        // std::cout << "quantized_0: " << static_cast<int>(output_vec[0]) << std::endl;
+        // std::cout << "original_0: " << input_vec[0] << std::endl;
+        // std::cout << "quantized_n: " << static_cast<int>(output_vec[dim - 1]) << std::endl;
+        // std::cout << "original_n: " << input_vec[dim - 1] << std::endl;
     }
 };

src/VecSim/utils/updatable_heap.h

@@ -41,9 +41,12 @@ class updatable_max_heap : public abstract_priority_queue<Priority, Value> {
     inline void emplace(Priority p, Value v) override;
     inline bool empty() const override;
     inline void pop() override;
+    inline bool exists(Value v) const;
     inline const std::pair<Priority, Value> top() const override;
     inline size_t size() const override;
-
+    // Random order iteration
+    const auto begin() const { return this->priorityToValue.begin(); }
+    const auto end() const { return this->priorityToValue.end(); }
 private:
     inline auto top_ptr() const;
 };
@@ -110,4 +113,8 @@ void updatable_max_heap<Priority, Value>::emplace(Priority p, Value v) {
     }
 }
 
+template <typename Priority, typename Value>
+bool updatable_max_heap<Priority, Value>::exists(Value v) const {
+    return valueToNode.find(v) != valueToNode.end();
+}
 } // namespace vecsim_stl

src/VecSim/vec_sim_common.h

@@ -328,6 +328,7 @@ typedef struct {
  */
 typedef struct {
     size_t memory;
+    size_t vectors_memory;
     size_t db_memory;
     size_t db_disk;
     size_t numberOfMarkedDeleted; // The number of vectors that are marked as deleted (HNSW/tiered

src/VecSim/vec_sim_index.h

@@ -136,6 +136,10 @@ struct VecSimIndexAbstract : public VecSimIndexInterface {
     DistType calcDistance(const void *vector_data1, const void *vector_data2) const {
         return indexCalculator->calcDistance(vector_data1, vector_data2, this->dim);
     }
+
+    DistType calcDistanceRaw(const void *vector_data1, const void *vector_data2) const {
+        return indexCalculator->calcDistanceRaw(vector_data1, vector_data2, this->dim);
+    }
 
     /**
      * @brief Preprocess a blob for both storage and query.
@@ -181,6 +185,7 @@ struct VecSimIndexAbstract : public VecSimIndexInterface {
     virtual inline VecSimIndexStatsInfo statisticInfo() const override {
         return VecSimIndexStatsInfo{
             .memory = this->getAllocationSize(),
+            .vectors_memory = 0,
             .db_memory = 0,
             .db_disk = 0,
             .numberOfMarkedDeleted = 0,

src/VecSim/vec_sim_tiered_index.h

@@ -308,6 +308,7 @@ template <typename DataType, typename DistType>
 VecSimIndexStatsInfo VecSimTieredIndex<DataType, DistType>::statisticInfo() const {
     auto stats = VecSimIndexStatsInfo{
         .memory = this->getAllocationSize(),
+        .vectors_memory = 0,
         .db_memory = 0,
         .db_disk = 0,
         .numberOfMarkedDeleted = 0, // Default value if cast fails

tests/benchmark/bm_common.h

@@ -94,6 +94,7 @@ void BM_VecSimCommon<index_type_t>::Memory(benchmark::State &st, IndexTypeIndex
         // Do nothing...
     }
     st.counters["memory"] = (double)VecSimIndex_StatsInfo(index).memory;
+    st.counters["vectors_memory"] = (double)VecSimIndex_StatsInfo(index).vectors_memory;
 }
 
 // TopK search BM
@@ -108,10 +109,11 @@ void BM_VecSimCommon<index_type_t>::TopK_HNSW_DISK(benchmark::State &st) {
     auto hnsw_index = GET_INDEX(INDEX_HNSW_DISK);
 
     // Get DB statistics if available
-    auto db_stats = dynamic_cast<HNSWDiskIndex<data_t, dist_t> *>(hnsw_index)->getDBStatistics();
-    size_t byte_reads = 0;
+    auto hnsw_disk_index = dynamic_cast<HNSWDiskIndex<data_t, dist_t> *>(hnsw_index);
+    auto db_stats = hnsw_disk_index->getDBStatistics();
+    size_t cache_misses = 0;
     if (db_stats) {
-        byte_reads = db_stats->getTickerCount(rocksdb::Tickers::BYTES_COMPRESSED_TO);
+        cache_misses = db_stats->getTickerCount(rocksdb::Tickers::BLOCK_CACHE_MISS);
     }
 
     for (auto _ : st) {
@@ -130,9 +132,13 @@ void BM_VecSimCommon<index_type_t>::TopK_HNSW_DISK(benchmark::State &st) {
     st.counters["Recall"] = (float)correct / (float)(k * iter);
 
     if (db_stats) {
-        byte_reads = db_stats->getTickerCount(rocksdb::Tickers::BYTES_COMPRESSED_TO) - byte_reads;
-        st.counters["byte_reads"] = static_cast<double>(byte_reads) / iter;
+        cache_misses = db_stats->getTickerCount(rocksdb::Tickers::BLOCK_CACHE_MISS) - cache_misses;
+        st.counters["cache_misses_per_query"] = static_cast<double>(cache_misses) / iter;
     }
+
+    // Output RocksDB memory usage
+    uint64_t db_memory = hnsw_disk_index->getDBMemorySize();
+    st.counters["rocksdb_memory"] = static_cast<double>(db_memory);
 }
 
 // Benchmark TopK performance with marked deleted vectors
@@ -759,11 +765,14 @@ void BM_VecSimCommon<index_type_t>::TopK_Tiered(benchmark::State &st, unsigned s
 #define REGISTER_TopK_HNSW_DISK(BM_CLASS, BM_FUNC)                                                 \
     BENCHMARK_REGISTER_F(BM_CLASS, BM_FUNC)                                                        \
         ->Args({10, 10})                                                                           \
-        ->Args({200, 10})                                                                          \
+        ->Args({100, 10})                                                                          \
+        ->Args({100, 50})                                                                         \
+        ->Args({200, 50})                                                                         \
         ->Args({100, 100})                                                                         \
         ->Args({200, 100})                                                                         \
+        ->Args({500, 100})                                                                         \
         ->ArgNames({"ef_runtime", "k"})                                                            \
-        ->Iterations(10)                                                                           \
+        ->Iterations(1000)                                                                           \
         ->Unit(benchmark::kMillisecond)
 
 // {ef_runtime, k, num_marked_deleted}
@@ -777,7 +786,7 @@ void BM_VecSimCommon<index_type_t>::TopK_Tiered(benchmark::State &st, unsigned s
         ->Args({200, 50, 10000})                                                                   \
         ->Args({200, 50, 25000})                                                                   \
         ->ArgNames({"ef_runtime", "k", "num_marked_deleted"})                                      \
-        ->Iterations(10)                                                                           \
+        ->Iterations(100)                                                                           \
         ->Unit(benchmark::kMillisecond)
 
 // {ef_runtime, k, num_deleted}

tests/benchmark/bm_initialization/bm_hnsw_disk_initialize_fp32.h

@@ -29,11 +29,6 @@ BENCHMARK_TEMPLATE_DEFINE_F(BM_VecSimCommon, BM_FUNC_NAME(Disk, HNSWDisk), fp32_
 (benchmark::State &st) { Disk(st, INDEX_HNSW_DISK); }
 BENCHMARK_REGISTER_F(BM_VecSimCommon, BM_FUNC_NAME(Disk, HNSWDisk))->Iterations(1);
 
-// AddLabel benchmarks
-BENCHMARK_TEMPLATE_DEFINE_F(BM_VecSimBasics, BM_ADD_LABEL, fp32_index_t)
-(benchmark::State &st) { AddLabel(st); }
-REGISTER_AddLabel(BM_ADD_LABEL, INDEX_HNSW_DISK);
-
 // TopK benchmark
 BENCHMARK_TEMPLATE_DEFINE_F(BM_VecSimCommon, BM_FUNC_NAME(TopK, HNSWDisk), fp32_index_t)
 (benchmark::State &st) { TopK_HNSW_DISK(st); }
@@ -48,6 +43,10 @@ REGISTER_TopK_HNSW_DISK_MarkDeleted(BM_VecSimCommon, BM_FUNC_NAME(TopK_MarkDelet
 BENCHMARK_TEMPLATE_DEFINE_F(BM_VecSimCommon, BM_FUNC_NAME(TopK_DeleteLabel, HNSWDisk), fp32_index_t)
 (benchmark::State &st) { TopK_HNSW_DISK_DeleteLabel(st); }
 REGISTER_TopK_HNSW_DISK_DeleteLabel(BM_VecSimCommon, BM_FUNC_NAME(TopK_DeleteLabel, HNSWDisk));
+// AddLabel benchmarks
+BENCHMARK_TEMPLATE_DEFINE_F(BM_VecSimBasics, BM_FLUSH_BATCH_DISK, fp32_index_t)
+(benchmark::State &st) { FlushBatchDisk(st); }
+REGISTER_FlushBatchDisk(BM_FLUSH_BATCH_DISK);
 
 // TopK benchmark after deleting vectors (with graph repair), protecting GT vectors for stable recall
 BENCHMARK_TEMPLATE_DEFINE_F(BM_VecSimCommon, BM_FUNC_NAME(TopK_DeleteLabel_ProtectGT, HNSWDisk), fp32_index_t)

tests/benchmark/bm_vecsim_basics.h

@@ -12,7 +12,7 @@
 #include "bm_common.h"
 #include <chrono>
 #include "types_ranges.h"
-
+#include "bm_vecsim_general.h"
 using namespace std::chrono;
 
 template <typename index_type_t>
@@ -42,6 +42,8 @@ class BM_VecSimBasics : public BM_VecSimCommon<index_type_t> {
     static void Range_BF(benchmark::State &st);
     static void Range_HNSW(benchmark::State &st);
 
+    static void FlushBatchDisk(benchmark::State &st);
+
 private:
     // Vectors of vector to store deleted labels' data.
     using LabelData = std::vector<std::vector<data_t>>;
@@ -317,6 +319,30 @@ void BM_VecSimBasics<index_type_t>::Range_HNSW(benchmark::State &st) {
     st.counters["Recall"] = (float)total_res / total_res_bf;
 }
 
+template <typename index_type_t>
+void BM_VecSimBasics<index_type_t>::FlushBatchDisk(benchmark::State &st) {
+    // Create a LOCAL ManagedRocksDB instance for this benchmark
+    // This ensures we don't interfere with the global managed_rocksdb used by other benchmarks
+    std::string folder_path = BM_VecSimGeneral::AttachRootPath(BM_VecSimGeneral::hnsw_index_file);
+    INDICES[INDEX_HNSW_DISK] = IndexPtr(HNSWDiskFactory::NewIndex(folder_path));
+    auto hnsw_index = GET_INDEX(INDEX_HNSW_DISK);
+    auto *hnsw_disk_index = dynamic_cast<HNSWDiskIndex<data_t, dist_t> *>(hnsw_index);
+
+    size_t flush_threshold = st.range(0);
+    hnsw_disk_index->setBatchThreshold(flush_threshold);
+    for (size_t i = 0; i < flush_threshold-1; i++) {
+        // add vectors to fill the batch
+        VecSimIndex_AddVector(hnsw_disk_index, QUERIES[i%N_QUERIES].data(), i);
+    }
+    for (auto _ : st) {
+        // add one vector to trigger flush
+        VecSimIndex_AddVector(hnsw_disk_index, QUERIES[(flush_threshold-1)%N_QUERIES].data(), flush_threshold-1);
+    }
+
+    // Clean up the index
+    VecSimIndex_Free(hnsw_disk_index);
+}
+
 #define UNIT_AND_ITERATIONS Unit(benchmark::kMillisecond)->Iterations(BM_VecSimGeneral::block_size)
 
 // These macros are used to make sure the expansion of other macros happens when needed
@@ -358,6 +384,15 @@ void BM_VecSimBasics<index_type_t>::Range_HNSW(benchmark::State &st) {
         ->UNIT_AND_ITERATIONS->Arg(VecSimAlgo)                                                     \
         ->ArgName(#VecSimAlgo)
 
+#define REGISTER_FlushBatchDisk(BM_FUNC)                                                           \
+    BENCHMARK_REGISTER_F(BM_VecSimBasics, BM_FUNC)                                                 \
+        ->Unit(benchmark::kMillisecond) \
+        ->Iterations(1) \
+        ->Args({100}) \
+        ->Args({1000}) \
+        ->Args({10000}) \
+        ->ArgName({"batch size"})                                                                   \
+
 // DeleteLabel define and register macros
 #define DEFINE_DELETE_LABEL(BM_FUNC, INDEX_TYPE, INDEX_NAME, DATA_TYPE, DIST_TYPE, VecSimAlgo)     \
     BENCHMARK_TEMPLATE_DEFINE_F(BM_VecSimBasics, BM_FUNC, INDEX_TYPE)(benchmark::State & st) {     \