Merge remote-tracking branch 'origin/meiravg_bufferlimit_bm' into meiravg_bufferlimit_bm

Author: meiravgri

src/VecSim/algorithms/hnsw/hnsw_tiered.h

@@ -112,7 +112,7 @@ class TieredHNSWIndex : public VecSimTieredIndex<DataType, DistType> {
     // ownership (we do it right after we update the HNSW global data and receive the new state).
     template <bool releaseFlatGuard>
     void insertVectorToHNSW(HNSWIndex<DataType, DistType> *hnsw_index, labelType label,
-                            const void *blob);
+                            DataType *blob);
 
 #ifdef BUILD_TESTS
 #include "VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h"
@@ -366,7 +366,7 @@ void TieredHNSWIndex<DataType, DistType>::updateInsertJobInternalId(idType prev_
 template <typename DataType, typename DistType>
 template <bool releaseFlatGuard>
 void TieredHNSWIndex<DataType, DistType>::insertVectorToHNSW(
-    HNSWIndex<DataType, DistType> *hnsw_index, labelType label, const void *blob) {
+    HNSWIndex<DataType, DistType> *hnsw_index, labelType label, DataType *blob) {
     // Acquire the index data lock, so we know what is the exact index size at this time. Acquire
     // the main r/w lock before to avoid deadlocks.
     AddVectorCtx state = {0};
@@ -430,6 +430,9 @@ void TieredHNSWIndex<DataType, DistType>::insertVectorToHNSW(
 /******************** Job's callbacks **********************************/
 template <typename DataType, typename DistType>
 void TieredHNSWIndex<DataType, DistType>::executeInsertJob(HNSWInsertJob *job) {
+    // Note: this method had not been tested with yet overwriting scenarios, where job may
+    // have been invalidate before it is executed (TODO in the future).
+    HNSWIndex<DataType, DistType> *hnsw_index = this->getHNSWIndex();
     // Note that accessing the job fields should occur with flat index guard held (here and later).
     this->flatIndexGuard.lock_shared();
     if (job->id == INVALID_JOB_ID) {
@@ -438,7 +441,6 @@ void TieredHNSWIndex<DataType, DistType>::executeInsertJob(HNSWInsertJob *job) {
         return;
     }
 
-    HNSWIndex<DataType, DistType> *hnsw_index = this->getHNSWIndex();
     // Copy the vector blob from the flat buffer, so we can release the flat lock while we are
     // indexing the vector into HNSW index.
     DataType blob_copy[this->frontendIndex->getDim()];
@@ -593,35 +595,18 @@ size_t TieredHNSWIndex<DataType, DistType>::indexLabelCount() const {
 template <typename DataType, typename DistType>
 int TieredHNSWIndex<DataType, DistType>::addVector(const void *blob, labelType label,
                                                    void *auxiliaryCtx) {
-    int ret = 1;
-    auto hnsw_index = this->getHNSWIndex();
-    if (this->getWriteMode() == VecSim_WriteInPlace) {
-        this->mainIndexGuard.lock();
-        // Internally, we may overwrite (delete the previous vector stored under this label), and
-        // may need to increase the capacity when we append the new vector afterwards.
-        ret = hnsw_index->addVector(blob, label);
-        this->mainIndexGuard.unlock();
+
+    if (this->getWriteMode() == VecSim_WriteInPlace ||
+        this->frontendIndex->indexSize() >= this->flatBufferLimit) {
+        auto hnsw_index = this->getHNSWIndex();
+        // Insert vector directly to HNSW (since flat buffer guard was not held, no need to release
+        // it internally).
+        this->insertVectorToHNSW<false>(hnsw_index, label, (DataType *)blob);
         this->UpdateIndexMemory(this->memoryCtx, this->getAllocationSize());
-        return ret;
-    }
-    if (this->frontendIndex->indexSize() >= this->flatBufferLimit) {
-        // Handle overwrite situation.
-        if (!this->backendIndex->isMultiValue()) {
-            // This will do nothing (and return 0) if this label doesn't exist. Otherwise, it may
-            // remove vector from the flat buffer and/or the HNSW index.
-            ret -= this->deleteVector(label);
-        }
-        if (this->frontendIndex->indexSize() >= this->flatBufferLimit) {
-            // We didn't remove a vector from flat buffer due to overwrite, insert the new vector
-            // directly to HNSW. Since flat buffer guard was not held, no need to release it
-            // internally.
-            this->insertVectorToHNSW<false>(hnsw_index, label, blob);
-            this->UpdateIndexMemory(this->memoryCtx, this->getAllocationSize());
-            return ret;
-        }
-        // Otherwise, we fall back to the "regular" insertion into the flat buffer
-        // (since it is not full anymore after removing the previous vector stored under the label).
+        return 1;
     }
+
+    /* Note: this currently doesn't support overriding (assuming that the label doesn't exist)! */
     this->flatIndexGuard.lock();
     idType new_flat_id = this->frontendIndex->indexSize();
     if (this->frontendIndex->isLabelExists(label) && !this->frontendIndex->isMultiValue()) {
@@ -713,7 +698,9 @@ int TieredHNSWIndex<DataType, DistType>::deleteVector(labelType label) {
     // Note that we may remove the same vector that has been removed from the flat index, if it was
     // being ingested at that time.
     if (this->getWriteMode() == VecSim_WriteAsync) {
+        this->mainIndexGuard.lock_shared();
         num_deleted_vectors += this->deleteLabelFromHNSW(label);
+        this->mainIndexGuard.unlock_shared();
         // Apply ready swap jobs if number of deleted vectors reached the threshold
         // (under exclusive lock of the main index guard).
         this->executeReadySwapJobs();

src/VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h

@@ -16,20 +16,10 @@ INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_alternateInsertDeleteAsync_Test)
 INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_swapJobBasic_Test)
 INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_swapJobBasic2_Test)
 INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_deleteVectorsAndSwapSync_Test)
-INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_BatchIterator_Test)
-INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_BatchIteratorAdvanced_Test)
-INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_BatchIteratorSize1_Test)
-INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_BatchIteratorReset_Test)
-INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_BatchIteratorWithOverlaps_Test)
-INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_parallelBatchIteratorSearch_Test)
-INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_testInfo_Test)
-INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_testInfoIterator_Test)
 INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_writeInPlaceMode_Test)
-INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_switchWriteModes_Test)
+INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_burstThreadsScenario_Test)
 INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_bufferLimit_Test)
 INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_bufferLimitAsync_Test)
-INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_RangeSearch_Test)
-INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_parallelRangeSearch_Test)
 
 INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTestBasic_insertJobAsync_Test)
 INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTestBasic_insertJobAsyncMulti_Test)

src/VecSim/vec_sim.cpp

@@ -22,7 +22,9 @@ extern "C" void VecSim_SetLogCallbackFunction(logCallbackFunction callback) {
     VecSimIndex::setLogCallbackFunction(callback);
 }
 
-extern "C" void VecSim_SetWriteMode(VecSimWriteMode mode) { VecSimIndex::setWriteMode(mode); }
+extern "C" void VecSim_SetInPlaceWriteMode(VecSimWriteMode mode) {
+    VecSimIndex::setWriteMode(mode);
+}
 
 static VecSimResolveCode _ResolveParams_EFRuntime(VecSimAlgo index_type, VecSimRawParam rparam,
                                                   VecSimQueryParams *qparams,

src/VecSim/vec_sim.h

@@ -209,9 +209,9 @@ void VecSim_SetLogCallbackFunction(logCallbackFunction callback);
  * @brief Allow 3rd party to set the write mode for tiered index - async insert/delete using
  * background jobs, or insert/delete inplace.
  *
- * @param mode VecSimWriteMode the mode in which we add/remove vectors (async or in-place).
+ * @param mode VecSimWriteMode to mode to write in.
  */
-void VecSim_SetWriteMode(VecSimWriteMode mode);
+void VecSim_SetInPlaceWriteMode(VecSimWriteMode mode);
 
 #ifdef __cplusplus
 }

src/VecSim/vec_sim_index.h

@@ -133,6 +133,7 @@ struct VecSimIndexAbstract : public VecSimIndexInterface {
             delete[] buf;
         }
     }
+
 #ifdef BUILD_TESTS
     // Set new log context to be sent to the log callback.
     // Returns the previous logctx.
@@ -142,92 +143,4 @@ struct VecSimIndexAbstract : public VecSimIndexInterface {
         return prev_logCtx;
     }
 #endif
-
-    void addCommonInfoToIterator(VecSimInfoIterator *infoIterator, const CommonInfo &info) const {
-        infoIterator->addInfoField(VecSim_InfoField{
-            .fieldName = VecSimCommonStrings::TYPE_STRING,
-            .fieldType = INFOFIELD_STRING,
-            .fieldValue = {FieldValue{.stringValue = VecSimType_ToString(info.type)}}});
-        infoIterator->addInfoField(
-            VecSim_InfoField{.fieldName = VecSimCommonStrings::DIMENSION_STRING,
-                             .fieldType = INFOFIELD_UINT64,
-                             .fieldValue = {FieldValue{.uintegerValue = info.dim}}});
-        infoIterator->addInfoField(VecSim_InfoField{
-            .fieldName = VecSimCommonStrings::METRIC_STRING,
-            .fieldType = INFOFIELD_STRING,
-            .fieldValue = {FieldValue{.stringValue = VecSimMetric_ToString(info.metric)}}});
-        infoIterator->addInfoField(
-            VecSim_InfoField{.fieldName = VecSimCommonStrings::IS_MULTI_STRING,
-                             .fieldType = INFOFIELD_UINT64,
-                             .fieldValue = {FieldValue{.uintegerValue = info.isMulti}}});
-        infoIterator->addInfoField(
-            VecSim_InfoField{.fieldName = VecSimCommonStrings::INDEX_SIZE_STRING,
-                             .fieldType = INFOFIELD_UINT64,
-                             .fieldValue = {FieldValue{.uintegerValue = info.indexSize}}});
-        infoIterator->addInfoField(
-            VecSim_InfoField{.fieldName = VecSimCommonStrings::INDEX_LABEL_COUNT_STRING,
-                             .fieldType = INFOFIELD_UINT64,
-                             .fieldValue = {FieldValue{.uintegerValue = info.indexLabelCount}}});
-        infoIterator->addInfoField(
-            VecSim_InfoField{.fieldName = VecSimCommonStrings::BLOCK_SIZE_STRING,
-                             .fieldType = INFOFIELD_UINT64,
-                             .fieldValue = {FieldValue{.uintegerValue = info.blockSize}}});
-        infoIterator->addInfoField(
-            VecSim_InfoField{.fieldName = VecSimCommonStrings::MEMORY_STRING,
-                             .fieldType = INFOFIELD_UINT64,
-                             .fieldValue = {FieldValue{.uintegerValue = info.memory}}});
-        infoIterator->addInfoField(VecSim_InfoField{
-            .fieldName = VecSimCommonStrings::SEARCH_MODE_STRING,
-            .fieldType = INFOFIELD_STRING,
-            .fieldValue = {FieldValue{.stringValue = VecSimSearchMode_ToString(info.last_mode)}}});
-    }
-    const void *processBlob(const void *original_blob, void *processed_blob) const {
-        // if the metric is cosine, we need to normalize
-        if (this->metric == VecSimMetric_Cosine) {
-            // copy original blob to the output blob
-            memcpy(processed_blob, original_blob, this->data_size);
-            // normalize the copy in place
-            normalize_func(processed_blob, this->dim);
-
-            return processed_blob;
-        }
-
-        // Else no process is needed, return the original blob
-        return original_blob;
-    }
-
-protected:
-    virtual int addVectorWrapper(const void *blob, labelType label, void *auxiliaryCtx) override {
-        char processed_blob[this->data_size];
-        const void *vector_to_add = processBlob(blob, processed_blob);
-
-        return this->addVector(vector_to_add, label, auxiliaryCtx);
-    }
-
-    virtual VecSimQueryResult_List topKQueryWrapper(const void *queryBlob, size_t k,
-                                                    VecSimQueryParams *queryParams) override {
-        char processed_blob[this->data_size];
-        const void *query_to_send = processBlob(queryBlob, processed_blob);
-
-        return this->topKQuery(query_to_send, k, queryParams);
-    }
-
-    virtual VecSimQueryResult_List rangeQueryWrapper(const void *queryBlob, double radius,
-                                                     VecSimQueryParams *queryParams,
-                                                     VecSimQueryResult_Order order) override {
-        char processed_blob[this->data_size];
-        const void *query_to_send = processBlob(queryBlob, processed_blob);
-
-        return this->rangeQuery(query_to_send, radius, queryParams, order);
-    }
-
-    virtual VecSimBatchIterator *
-    newBatchIteratorWrapper(const void *queryBlob, VecSimQueryParams *queryParams) const override {
-        char processed_blob[this->data_size];
-        const void *query_to_send = processBlob(queryBlob, processed_blob);
-
-        return this->newBatchIterator(query_to_send, queryParams);
-    }
-
-
 };

src/VecSim/vec_sim_interface.h

@@ -234,7 +234,7 @@ struct VecSimIndexInterface : public VecsimBaseObject {
      * @brief Allow 3rd party to set the write mode for tiered index - async insert/delete using
      * background jobs, or insert/delete inplace.
      *
-     * @param mode VecSimWriteMode the mode in which we add/remove vectors (async or in-place).
+     * @param mode VecSimWriteMode to mode to write in.
      */
     static VecSimWriteMode asyncWriteMode;
     inline static void setWriteMode(VecSimWriteMode mode) {

src/VecSim/vec_sim_tiered_index.h

@@ -66,66 +66,11 @@ class VecSimTieredIndex : public VecSimIndexInterface {
     VecSimQueryResult_List topKQuery(const void *queryBlob, size_t k,
                                      VecSimQueryParams *queryParams) override;
 
-    virtual inline int64_t getAllocationSize() const override {
-        return this->allocator->getAllocationSize() + this->backendIndex->getAllocationSize() +
-               this->frontendIndex->getAllocationSize();
-    }
-
-    virtual VecSimIndexInfo info() const override;
-    virtual VecSimInfoIterator *infoIterator() const override;
-
-    VecSimQueryResult_List rangeQuery(const void *queryBlob, double radius,
-                                      VecSimQueryParams *queryParams,
-                                      VecSimQueryResult_Order order) override;
-
-    bool preferAdHocSearch(size_t subsetSize, size_t k, bool initial_check) override {
-        // For now, decide according to the bigger index.
-        return this->backendIndex->indexSize() > this->frontendIndex->indexSize()
-                   ? this->backendIndex->preferAdHocSearch(subsetSize, k, initial_check)
-                   : this->frontendIndex->preferAdHocSearch(subsetSize, k, initial_check);
-    }
+    static VecSimWriteMode getWriteMode() { return asyncWriteMode; }
 
-    // Return the current state of the global write mode (async/in-place).
-    static VecSimWriteMode getWriteMode() { return VecSimIndexInterface::asyncWriteMode; }
 #ifdef BUILD_TESTS
     inline VecSimIndexAbstract<DistType> *getFlatbufferIndex() { return this->frontendIndex; }
 #endif
-private:
-    virtual int addVectorWrapper(const void *blob, labelType label, void *auxiliaryCtx) override {
-        // Will be used only if a processing stage is needed
-        char processed_blob[this->backendIndex->getDataSize()];
-        const void *vector_to_add = this->backendIndex->processBlob(blob, processed_blob);
-        return this->addVector(vector_to_add, label, auxiliaryCtx);
-    }
-
-    virtual VecSimQueryResult_List topKQueryWrapper(const void *queryBlob, size_t k,
-                                                    VecSimQueryParams *queryParams) override {
-        // Will be used only if a processing stage is needed
-        char processed_blob[this->backendIndex->getDataSize()];
-        const void *query_to_send = this->backendIndex->processBlob(queryBlob, processed_blob);
-        return this->topKQuery(query_to_send, k, queryParams);
-    }
-
-    virtual VecSimQueryResult_List rangeQueryWrapper(const void *queryBlob, double radius,
-                                                     VecSimQueryParams *queryParams,
-                                                     VecSimQueryResult_Order order) override {
-        // Will be used only if a processing stage is needed
-        char processed_blob[this->backendIndex->getDataSize()];
-        const void *query_to_send = this->backendIndex->processBlob(queryBlob, processed_blob);
-
-        return this->rangeQuery(query_to_send, radius, queryParams, order);
-    }
-
-    virtual VecSimBatchIterator *
-    newBatchIteratorWrapper(const void *queryBlob, VecSimQueryParams *queryParams) const override {
-        // Will be used only if a processing stage is needed
-        char processed_blob[this->backendIndex->getDataSize()];
-        const void *query_to_send = this->backendIndex->processBlob(queryBlob, processed_blob);
-
-        return this->newBatchIterator(query_to_send, queryParams);
-    }
-
-
 };
 
 template <typename DataType, typename DistType>