From 152c478e096d27bd92eba6b43add5365d5376ed4 Mon Sep 17 00:00:00 2001 From: sharonrosenfeld Date: Mon, 19 Oct 2026 10:22:16 +0300 Subject: [PATCH] Refactor addVector to support reuse flows Split addVector into addVectorInPlace and addVectorToFlat, and move the single-value branch of updateVectors into updateVectorSingleValue. A single-value update with zero vectors now removes the label instead of inserting from a null blob. Co-Authored-By: Claude Sonnet 5.5 --- src/VecSim/algorithms/hnsw/hnsw_tiered.h | 179 ++++++++++++----------- tests/unit/test_hnsw_tiered.cpp | 24 +++ 2 files changed, 119 insertions(+), 84 deletions(-) diff --git a/src/VecSim/algorithms/hnsw/hnsw_tiered.h b/src/VecSim/algorithms/hnsw/hnsw_tiered.h index 21ebaa342..301163767 100644 --- a/src/VecSim/algorithms/hnsw/hnsw_tiered.h +++ b/src/VecSim/algorithms/hnsw/hnsw_tiered.h @@ -300,6 +300,10 @@ class TieredHNSWIndex : public VecSimTieredIndex { // the label and adding the blobs one by one. bool updateMultiValueAsync(labelType label, const void *new_blobs, size_t n); + VecSimUpdateCode updateVectorSingleValue(labelType label, const void *new_blobs, size_t n); + int addVectorInPlace(const void *blob, labelType label); + HNSWInsertJob *addVectorToFlat(const void *blob, labelType label, int &ret); + #ifdef BUILD_TESTS #include "VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h" #endif @@ -389,6 +393,8 @@ class TieredHNSWIndex : public VecSimTieredIndex { #endif int addVector(const void *blob, labelType label) override; + // 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. int deleteVector(labelType label) override; VecSimRelabelCode relabelVector(labelType old_label, labelType new_label) override; @@ -1262,63 +1268,29 @@ template auto TieredHNSWIndex::shouldDefrag() { return readySwapJobs.load() >= this->pendingSwapJobsThreshold; } -// In the tiered index, we assume that the blobs are processed by the flat buffer -// before being transferred to the HNSW index. -// When inserting vectors directly into the HNSW index—such as in VecSim_WriteInPlace mode— or when -// the flat buffer is full, we must manually preprocess the blob according to the **frontend** index -// parameters. template -int TieredHNSWIndex::addVector(const void *blob, labelType label) { - if (sqAccumulationState) { - return addVectorDuringAccumulation(blob, label); - } +int TieredHNSWIndex::addVectorInPlace(const void *blob, labelType label) { int ret = 1; auto hnsw_index = this->getHNSWIndex(); - // writeMode is not protected since it is assumed to be called only from the "main thread" - // (that is the thread that is exclusively calling add/delete vector). - if (this->getWriteMode() == VecSim_WriteInPlace) { - // Use the frontend parameters to manually prepare the blob for its transfer to the HNSW - // index. - auto storage_blob = this->frontendIndex->preprocessForStorage(blob); - if (this->reuseIdOnUpdate) { - ret = updateVectorInPlace(label, hnsw_index, storage_blob); - } else { - // Reuse disabled: delete the label outright (from both tiers) and append a fresh - // element, as this index did before id reuse existed. - if (!this->backendIndex->isMultiValue()) { - ret -= this->deleteVector(label); - } - const auto main_index_lock = this->acquireMainIndexGuard(); - hnsw_index->addVector(storage_blob.get(), label); - } - // Track direct insertion to HNSW (bypassing flat buffer) - ++this->directHNSWInsertions; - // Removing the old id in place can complete a reserved id's repairs. - this->submitDeferredReuseInsertJobs(); - return ret; - } - if (this->frontendIndex->indexSize() >= this->flatBufferLimit) { - // Handle overwrite situation. + auto storage_blob = this->frontendIndex->preprocessForStorage(blob); + if (this->reuseIdOnUpdate) { + ret = updateVectorInPlace(label, hnsw_index, storage_blob); + } else { 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. - // Use the frontend parameters to manually prepare the blob for its transfer to the HNSW - // index. - auto storage_blob = this->frontendIndex->preprocessForStorage(blob); - this->insertVectorToHNSW(hnsw_index, label, storage_blob.get()); - // Track direct insertion to HNSW (flat buffer was full) - ++this->directHNSWInsertions; - 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). + const auto main_index_lock = this->acquireMainIndexGuard(); + hnsw_index->addVector(storage_blob.get(), label); } + // Track direct insertion to HNSW (bypassing flat buffer) + ++this->directHNSWInsertions; + // Removing the old id in place can complete a reserved id's repairs. + this->submitDeferredReuseInsertJobs(); + return ret; +} +template +HNSWInsertJob *TieredHNSWIndex::addVectorToFlat(const void *blob, + labelType label, int &ret) { this->flatIndexGuard.lock(); idType new_flat_id = this->frontendIndex->indexSize(); if (this->frontendIndex->isLabelExists(label) && !this->frontendIndex->isMultiValue()) { @@ -1338,7 +1310,7 @@ int TieredHNSWIndex::addVector(const void *blob, labelType l // If this label already exists, this will do overwrite. this->frontendIndex->addVector(blob, label); - auto *new_insert_job = new (this->allocator) + auto new_insert_job = new (this->allocator) HNSWInsertJob(this->allocator, label, new_flat_id, executeInsertJobWrapper, this); // Construct the job vector only for a new label; multi-value labels append to the existing one. auto [it, inserted] = @@ -1348,7 +1320,44 @@ int TieredHNSWIndex::addVector(const void *blob, labelType l it->second.push_back(new_insert_job); } this->flatIndexGuard.unlock(); - + return new_insert_job; +} +// In the tiered index, we assume that the blobs are processed by the flat buffer +// before being transferred to the HNSW index. +// When inserting vectors directly into the HNSW index—such as in VecSim_WriteInPlace mode— or when +// the flat buffer is full, we must manually preprocess the blob according to the **frontend** index +// parameters. +template +int TieredHNSWIndex::addVector(const void *blob, labelType label) { + if (sqAccumulationState) { + return addVectorDuringAccumulation(blob, label); + } + // writeMode is not protected since it is assumed to be called only from the "main thread" + // (that is the thread that is exclusively calling add/delete vector). + if (this->getWriteMode() == VecSim_WriteInPlace) { + return addVectorInPlace(blob, label); + } + int ret = 1; + if (this->frontendIndex->indexSize() >= this->flatBufferLimit) { + // Handle overwrite situation. + if (!this->backendIndex->isMultiValue()) { + 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. + auto storage_blob = this->frontendIndex->preprocessForStorage(blob); + auto hnsw_index = this->getHNSWIndex(); + this->insertVectorToHNSW(hnsw_index, label, storage_blob.get()); + // Track direct insertion to HNSW (flat buffer was full) + ++this->directHNSWInsertions; + 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). + } + auto new_insert_job = addVectorToFlat(blob, label, ret); // Here, a worker might ingest the previous vector that was stored under "label" // (in case of override in non-MULTI index) - so if it's there, we remove it (and create the // required repair jobs), *before* we submit the insert job. @@ -1363,9 +1372,6 @@ int TieredHNSWIndex::addVector(const void *blob, labelType l } // Apply ready swap jobs if number of deleted vectors reached the threshold (under exclusive // lock of the main index guard). - // If swapJobs size is equal or larger than a threshold, go over the swap jobs and execute a - // batch of jobs for which all of its pending repair jobs were executed (otherwise finish and - // return). if (shouldDefrag()) { this->defrag(); } @@ -1446,6 +1452,26 @@ int TieredHNSWIndex::deleteVector(labelType label) { return num_deleted_vectors; } +template +VecSimUpdateCode TieredHNSWIndex::updateVectorSingleValue(labelType label, + const void *new_blobs, + size_t n) { + if (n > 1) { + return VecSimUpdate_MultiNotSupported; + } + if (n == 0) { + // Nothing to store, so replacing the label's content means removing it. + this->deleteVector(label); + return VecSimUpdate_OK; + } + // One vector replacing a single-value label *is* an overwrite, and `addVector` serves that + // better than a deleted and an insert would: it writes over the buffered copy in place, keeping + // its id and the pending job that will ingest it, and only then marks the backend's copy + // deleted - so the label is never without a vector, where removing first would leave a window + // with none. + this->addVector(new_blobs, label); + return VecSimUpdate_OK; +} /** * Replace everything stored under a label with the given vectors. * @@ -1469,38 +1495,23 @@ template VecSimUpdateCode TieredHNSWIndex::updateVectors(labelType label, const void *new_blobs, size_t n) { - if (!this->backendIndex->isMultiValue() && n > 1) { - return VecSimUpdate_MultiNotSupported; - } - // One vector replacing a single-value label *is* an overwrite, and `addVector` serves that - // better than a deleted and an insert would: it writes over the buffered copy in place, keeping - // its id and the pending job that will ingest it, and only then marks the backend's copy - // deleted - so the label is never without a vector, where removing first would leave a window - // with none. It is also the case a caller hits most, a hash document holding one vector per - // field. - if (n == 1 && !this->backendIndex->isMultiValue()) { - this->addVector(new_blobs, label); - return VecSimUpdate_OK; - } + if (!this->backendIndex->isMultiValue()) + return updateVectorSingleValue(label, new_blobs, n); - // From here on the backend is always multi-value - single-value with n > 1 already returned - // above. + // -- Multi Value -- // During SQ accumulation, training writes must stay in FLAT regardless of write mode - the // in-place path writes straight into the (still untrained) HNSW backend and bypasses the - // running-sum bookkeeping that `addVector`/`deleteVector` maintain for it, so it is skipped - // here in favor of the generic delete-then-add path below, which routes through both of those - // and so stays accumulation-safe. - if (this->getWriteMode() == VecSim_WriteInPlace && this->reuseIdOnUpdate && - !sqAccumulationState) { - this->updateMultiValueInPlace(label, new_blobs, n); - // Removing ids in place can complete a reserved id's repairs. - this->submitDeferredReuseInsertJobs(); - return VecSimUpdate_OK; - } - - if (this->getWriteMode() == VecSim_WriteAsync && this->reuseIdOnUpdate && - !sqAccumulationState && n > 0 && this->updateMultiValueAsync(label, new_blobs, n)) { - return VecSimUpdate_OK; + // running-sum bookkeeping that `addVector`/`deleteVector` maintain for it + if (this->reuseIdOnUpdate && !sqAccumulationState) { + if (this->getWriteMode() == VecSim_WriteInPlace) { + this->updateMultiValueInPlace(label, new_blobs, n); + // Removing ids in place can complete a reserved id's repairs. + this->submitDeferredReuseInsertJobs(); + return VecSimUpdate_OK; + } + if (n > 0 && this->updateMultiValueAsync(label, new_blobs, n)) { + return VecSimUpdate_OK; + } } this->deleteVector(label); diff --git a/tests/unit/test_hnsw_tiered.cpp b/tests/unit/test_hnsw_tiered.cpp index 539f239d2..84e314150 100644 --- a/tests/unit/test_hnsw_tiered.cpp +++ b/tests/unit/test_hnsw_tiered.cpp @@ -5827,6 +5827,30 @@ TYPED_TEST(HNSWTieredIndexTestBasic, updateVectorsMultiInPlaceShrinkToZeroRemove ASSERT_EQ(tiered_index->getDistanceFrom_Unsafe(2, v2), 0); } +// On a single-value index, replacing a label with zero vectors must remove it, like it does on a +// multi-value one - not read the (null) blob as a vector to overwrite it with. Covers a label +// still in the flat buffer (async) and one already in HNSW (in-place). +TYPED_TEST(HNSWTieredIndexTestBasic, updateVectorsSingleValueZeroVectorsRemovesLabel) { + size_t dim = 4; + HNSWParams params = { + .type = TypeParam::get_index_type(), .dim = dim, .metric = VecSimMetric_L2, .multi = false}; + VecSimParams hnsw_params = CreateParams(params); + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = this->CreateTieredHNSWIndex(hnsw_params, mock_thread_pool); + + for (auto mode : {VecSim_WriteAsync, VecSim_WriteInPlace}) { + VecSim_SetWriteMode(mode); + labelType label = mode == VecSim_WriteAsync ? 1 : 11; + GenerateAndAddVector(tiered_index, dim, label, label); + size_t labels_before = tiered_index->indexLabelCount(); + ASSERT_EQ(labels_before, 1); + + ASSERT_EQ(tiered_index->updateVectors(label, nullptr, 0), VecSimUpdate_OK); + + ASSERT_EQ(tiered_index->indexLabelCount(), 0); + } +} + // A repair job filed against a multi-value label's id (because some other, unrelated element was // deleted and needs that id's edge fixed) can still be sitting in the queue when the label itself // is overwritten in place. `updateMultiValueInPlace` must invalidate that stale job before reusing