[MOD-17578] Isolate a deleted element from the graph once its repairs are done

src/VecSim/algorithms/hnsw/hnsw.h

@@ -205,7 +205,7 @@ class HNSWIndex : public VecSimIndexAbstract<DataType, DistType>,
                        idType id) const;
     void emplaceToHeap(vecsim_stl::abstract_priority_queue<DistType, labelType> &heap,
                        DistType dist, idType id) const;
-    void removeAndSwap(idType internalId);
+    void swapWithLast(idType removedId);
 
     size_t getVectorRelativeIndex(idType id) const { return id % this->blockSize; }
 
@@ -279,6 +279,26 @@ class HNSWIndex : public VecSimIndexAbstract<DataType, DistType>,
     void unmarkInProcess(idType internalId);
     HNSWAddVectorState storeNewElement(labelType label, const void *vector_data);
     void removeAndSwapMarkDeletedElement(idType internalId);
+    void removeFromGraph(idType internalId);
+    // Remove `id` from `level_data`'s links if it is still there. Unlike `removeLink`, tolerates
+    // its absence - an element that was already isolated (see `isolateDeletedElement`) holds no
+    // links at all, and a repair job running in parallel may have dropped the edge too.
+    static void removeLinkIfExists(ElementLevelData &level_data, idType id) {
+        for (size_t i = 0; i < level_data.getNumLinks(); i++) {
+            if (level_data.getLinkAtPos(i) == id) {
+                level_data.removeLink(id);
+                return;
+            }
+        }
+    }
+    // Take a marked-deleted element out of the graph entirely: drop every edge going out of it and
+    // every edge coming into it, at every level, keeping the bookkeeping of the other side
+    // consistent. To be called once all the repair jobs created for its deletion are done and
+    // before its swap job disposes of it, so that from that point no element refers to it and it
+    // refers to no element.
+    // Takes the per-element links locks (one at a time), so holding the main index guard for shared
+    // ownership is enough.
+    void isolateDeletedElement(idType internalId);
     void repairNodeConnections(idType node_id, size_t level);
     // For prefetching only.
     const ElementMetaData *getMetaDataAddress(idType internal_id) const {
@@ -951,7 +971,8 @@ void HNSWIndex<DataType, DistType>::repairConnectionsForDeletion(
     if (isMarkedDeleted(neighbour_id)) {
         // Just remove the deleted element from the neighbor's neighbors list. No need to repair as
         // this change is temporary, this neighbor is about to be removed from the graph as well.
-        neighbor_level.removeLink(element_internal_id);
+        // The link may already be gone if this neighbor was isolated upon completing its repairs.
+        removeLinkIfExists(neighbor_level, element_internal_id);
         return;
     }
 
@@ -1541,7 +1562,20 @@ void HNSWIndex<DataType, DistType>::mutuallyRemoveNeighborAtPos(ElementLevelData
     // mutually, so it should be sufficient to look at the removed node's incoming edges set
     // alone.
     if (!removed_node_level.removeIncomingUnidirectionalEdgeIfExists(node_id)) {
-        node_level.newIncomingUnidirectionalEdge(removed_node);
+        // No record of this edge on the other side. Normally that means the edge was bidirectional,
+        // but it also happens when the removed node was already isolated: `isolateDeletedElement`
+        // clears its incoming edges set together with its links. Check whether it actually points
+        // back before recording the remaining direction.
+        bool points_back = false;
+        for (size_t i = 0; i < removed_node_level.getNumLinks(); i++) {
+            if (removed_node_level.getLinkAtPos(i) == node_id) {
+                points_back = true;
+                break;
+            }
+        }
+        if (points_back) {
+            node_level.newIncomingUnidirectionalEdge(removed_node);
+        }
     }
 }
 
@@ -1644,20 +1678,73 @@ HNSWIndex<DataType, DistType>::~HNSWIndex() {
  */
 
 template <typename DataType, typename DistType>
-void HNSWIndex<DataType, DistType>::removeAndSwap(idType internalId) {
+void HNSWIndex<DataType, DistType>::isolateDeletedElement(idType internalId) {
+    assert(isMarkedDeleted(internalId) && "Only a marked-deleted element may be isolated");
+    auto element = getGraphDataByInternalId(internalId);
+    for (size_t level = 0; level <= element->toplevel; level++) {
+        // Take the element's edges at this level, and drop its own side of them right away - from
+        // here on it points to nothing, so a repair job that runs for it later finds no deleted
+        // neighbour and returns without touching it.
+        lockNodeLinks(internalId);
+        ElementLevelData &level_data = getElementLevelData(element, level);
+        auto neighbours = level_data.copyLinks();
+        std::vector<idType> incoming_edges(level_data.getIncomingEdges().begin(),
+                                           level_data.getIncomingEdges().end());
+        level_data.setNumLinks(0);
+        for (idType incoming_id : incoming_edges) {
+            level_data.removeIncomingUnidirectionalEdgeIfExists(incoming_id);
+        }
+        unlockNodeLinks(internalId);
+
+        // Drop the other side of the outgoing edges: the neighbour either recorded this element as
+        // an incoming unidirectional edge (the expected case, as no element points to it anymore),
+        // or still points back at it - an incoming edge whose repair job never ran, which is
+        // dropped here as well.
+        for (idType neighbour_id : neighbours) {
+            lockNodeLinks(neighbour_id);
+            ElementLevelData &neighbour = getElementLevelData(neighbour_id, level);
+            if (!neighbour.removeIncomingUnidirectionalEdgeIfExists(internalId)) {
+                // No record on the neighbour's side, so it still points back here. Every *live*
+                // element that pointed at this one had a repair job that removed its edge before
+                // this point, so this is an edge between two deleted elements - neither of them
+                // gets a repair job for the other, hence it is dropped here.
+                assert(isMarkedDeleted(neighbour_id) &&
+                       "a live element still points to a fully repaired deleted element");
+                removeLinkIfExists(neighbour, internalId);
+            }
+            unlockNodeLinks(neighbour_id);
+        }
+
+        // Same for any incoming edge that is still registered: remove it from its origin's links.
+        for (idType incoming_id : incoming_edges) {
+            // Same here: an incoming edge that survived all the repair jobs comes from another
+            // deleted element.
+            assert(isMarkedDeleted(incoming_id) &&
+                   "a live element still points to a fully repaired deleted element");
+            lockNodeLinks(incoming_id);
+            removeLinkIfExists(getElementLevelData(incoming_id, level), internalId);
+            unlockNodeLinks(incoming_id);
+        }
+    }
+}
+
+template <typename DataType, typename DistType>
+void HNSWIndex<DataType, DistType>::removeFromGraph(idType internalId) {
     // Sanity check - the id to remove cannot be the entry point, as it should have been replaced
     // upon marking it as deleted.
     assert(entrypointNode != internalId);
     auto element = getGraphDataByInternalId(internalId);
 
-    // Remove the deleted id form the relevant incoming edges sets in which it appears.
+    // Remove the deleted id form the relevant incoming edges sets in which it appears. For an
+    // asynchronously deleted element there is nothing to walk here: `isolateDeletedElement` already
+    // took all of its edges out when its last repair job completed.
     for (size_t level = 0; level <= element->toplevel; level++) {
         ElementLevelData &cur_level = getElementLevelData(element, level);
         for (size_t i = 0; i < cur_level.getNumLinks(); i++) {
             ElementLevelData &neighbour = getElementLevelData(cur_level.getLinkAtPos(i), level);
-            // Note that in case of in-place delete, we might have not accounted for this edge in
+            // Note that in case of in-place delete, we might have not accounted for this edge
             // in the unidirectional edges, since there is no point in keeping it there temporarily
-            // (we know we will get here and remove this deleted id permanently).
+            // . (We know we will get here and remove this deleted id permanently.)
             // However, upon asynchronous delete, this should always succeed since we do update
             // the incoming edges in the mutual update even for deleted elements.
             bool res = neighbour.removeIncomingUnidirectionalEdgeIfExists(internalId);
@@ -1673,16 +1760,19 @@ void HNSWIndex<DataType, DistType>::removeAndSwap(idType internalId) {
 
     // We can say now that the element has removed completely from index.
     --curElementCount;
+}
 
+template <typename DataType, typename DistType>
+void HNSWIndex<DataType, DistType>::swapWithLast(idType removedId) {
     // Get the last element's metadata and data.
-    // If we are deleting the last element, we already destroyed it's metadata.
+    // If we are deleting the last element, we already destroyed its metadata.
     auto *last_element_data = getDataByInternalId(curElementCount);
     DataBlock &last_gd_block = graphDataBlocks.back();
     auto last_element = (ElementGraphData *)last_gd_block.removeAndFetchLastElement();
 
     // Swap the last id with the deleted one, and invalidate the last id data.
-    if (curElementCount != internalId) {
-        SwapLastIdWithDeletedId(internalId, last_element, last_element_data);
+    if (curElementCount != removedId) {
+        SwapLastIdWithDeletedId(removedId, last_element, last_element_data);
     }
 
     // If we need to free a complete block and there is at least one block between the
@@ -1693,7 +1783,8 @@ void HNSWIndex<DataType, DistType>::removeAndSwap(idType internalId) {
 
 template <typename DataType, typename DistType>
 void HNSWIndex<DataType, DistType>::removeAndSwapMarkDeletedElement(idType internalId) {
-    removeAndSwap(internalId);
+    removeFromGraph(internalId);
+    swapWithLast(internalId);
     // element is permanently removed from the index, it is no longer counted as marked deleted.
     --numMarkedDeleted;
 }
@@ -1756,7 +1847,8 @@ void HNSWIndex<DataType, DistType>::removeVectorInPlace(const idType element_int
     }
     // Finally, remove the element from the index and make a swap with the last internal id to
     // avoid fragmentation and reclaim memory when needed.
-    removeAndSwap(element_internal_id);
+    removeFromGraph(element_internal_id);
+    swapWithLast(element_internal_id);
 }
 
 // Store the new element in the global data structures and keep the new state. In multithreaded

src/VecSim/algorithms/hnsw/hnsw_tiered.h

@@ -104,7 +104,8 @@ class TieredHNSWIndex : public VecSimTieredIndex<DataType, DistType> {
 
     // To be executed synchronously upon deleting a vector, doesn't require a wrapper. Main HNSW
     // lock is assumed to be held exclusive here.
-    void executeSwapJob(idType deleted_id, vecsim_stl::vector<idType> &idsToRemove);
+    void fixJobsAfterSwap(idType deleted_id, vecsim_stl::vector<idType> &idsToRemove);
+    void invalidateRepairJobs(idType deleted_id);
 
     // Execute the ready swap jobs, run no more than 'maxSwapsToRun' jobs (run all of them for -1).
     void executeReadySwapJobs(size_t maxSwapsToRun = -1);
@@ -128,6 +129,12 @@ class TieredHNSWIndex : public VecSimTieredIndex<DataType, DistType> {
     // while *HNSW shared lock is held* (shared locked).
     int deleteLabelFromHNSW(labelType label);
 
+    // Take a deleted element out of the graph once the last repair job created for its deletion is
+    // done - no edge in or out of it is left, so the swap job that disposes of it later only has to
+    // reclaim its slot. Called in the repair context, while the main index guard is held for shared
+    // ownership and `idToRepairJobsGuard` is not held.
+    void isolateRepairedElement(idType deleted_id);
+
     // Insert a single vector to HNSW. This can be called in both write modes - insert async and
     // in-place. For the async mode, we have to release the flat index guard that is held for shared
     // ownership (we do it right after we update the HNSW global data and receive the new state).
@@ -283,23 +290,11 @@ void TieredHNSWIndex<DataType, DistType>::executeRepairJobWrapper(AsyncJob *job)
 }
 
 template <typename DataType, typename DistType>
-void TieredHNSWIndex<DataType, DistType>::executeSwapJob(idType deleted_id,
-                                                         vecsim_stl::vector<idType> &idsToRemove) {
-    // Get the id that was last and was had been swapped with the job's deleted id.
+void TieredHNSWIndex<DataType, DistType>::fixJobsAfterSwap(
+    idType deleted_id, vecsim_stl::vector<idType> &idsToRemove) {
+    // Get the id that was last and had been swapped with the job's deleted id.
     idType prev_last_id = this->getHNSWIndex()->indexSize();
 
-    // Invalidate repair jobs for the disposed id (if exist), and update the associated swap jobs.
-    if (idToRepairJobs.find(deleted_id) != idToRepairJobs.end()) {
-        for (auto &job_it : idToRepairJobs.at(deleted_id)) {
-            job_it->node_id = this->setAndSaveInvalidJob(job_it);
-            for (auto &swap_job_it : job_it->associatedSwapJobs) {
-                if (swap_job_it->atomicDecreasePendingJobsNum() == 0) {
-                    readySwapJobs++;
-                }
-            }
-        }
-        idToRepairJobs.erase(deleted_id);
-    }
     // Swap the ids in the pending jobs for the current last id (if exist).
     if (idToRepairJobs.find(prev_last_id) != idToRepairJobs.end()) {
         for (auto &job_it : idToRepairJobs.at(prev_last_id)) {
@@ -324,11 +319,34 @@ void TieredHNSWIndex<DataType, DistType>::executeSwapJob(idType deleted_id,
     }
 }
 
+template <typename DataType, typename DistType>
+void TieredHNSWIndex<DataType, DistType>::invalidateRepairJobs(idType deleted_id) {
+    // Invalidate repair jobs for the disposed id (if exist), and update the associated swap jobs.
+    if (idToRepairJobs.find(deleted_id) == idToRepairJobs.end()) {
+        return;
+    }
+
+    for (auto &job_it : idToRepairJobs.at(deleted_id)) {
+        job_it->node_id = this->setAndSaveInvalidJob(job_it);
+        for (auto &swap_job_it : job_it->associatedSwapJobs) {
+            if (swap_job_it->atomicDecreasePendingJobsNum() == 0) {
+                readySwapJobs++;
+            }
+        }
+    }
+    idToRepairJobs.erase(deleted_id);
+}
+
 template <typename DataType, typename DistType>
 HNSWIndex<DataType, DistType> *TieredHNSWIndex<DataType, DistType>::getHNSWIndex() const {
     return dynamic_cast<HNSWIndex<DataType, DistType> *>(this->backendIndex);
 }
 
+template <typename DataType, typename DistType>
+void TieredHNSWIndex<DataType, DistType>::isolateRepairedElement(idType deleted_id) {
+    this->getHNSWIndex()->isolateDeletedElement(deleted_id);
+}
+
 template <typename DataType, typename DistType>
 void TieredHNSWIndex<DataType, DistType>::executeReadySwapJobs(size_t maxJobsToRun) {
 
@@ -342,10 +360,12 @@ void TieredHNSWIndex<DataType, DistType>::executeReadySwapJobs(size_t maxJobsToR
     idsToRemove.reserve(idToSwapJob.size());
     for (auto &it : idToSwapJob) {
         auto *swap_job = it.second;
+        // Swap job is ready for execution - execute and delete it.
         if (swap_job->pending_repair_jobs_counter.load() == 0) {
-            // Swap job is ready for execution - execute and delete it.
-            this->getHNSWIndex()->removeAndSwapMarkDeletedElement(swap_job->deleted_id);
-            this->executeSwapJob(swap_job->deleted_id, idsToRemove);
+            auto deleted_id = swap_job->deleted_id;
+            this->getHNSWIndex()->removeAndSwapMarkDeletedElement(deleted_id);
+            this->invalidateRepairJobs(deleted_id);
+            this->fixJobsAfterSwap(deleted_id, idsToRemove);
             delete swap_job;
         }
         if (maxJobsToRun > 0 && idsToRemove.size() >= maxJobsToRun) {
@@ -415,6 +435,13 @@ int TieredHNSWIndex<DataType, DistType>::deleteLabelFromHNSW(labelType label) {
         }
         this->idToRepairJobsGuard.unlock();
 
+        if (incomingEdges.size() == 0) {
+            // No repair job will ever run for this element, so this is already the point at which
+            // it can be taken out of the graph (outside the repair jobs guard, as isolating takes
+            // the per-element links locks).
+            this->isolateRepairedElement(id);
+        }
+
         this->submitJobs(repair_jobs);
         // Insert the swap job into the swap jobs lookup (for fast update in case that the
         // node id is changed due to swap job).
@@ -533,7 +560,8 @@ int TieredHNSWIndex<DataType, DistType>::deleteLabelFromHNSWInplace(labelType la
         // Get the id in every iteration, since the ids can be swapped in every iteration.
         idType id = hnsw_index->getElementIds(label).at(id_ind);
         hnsw_index->removeVectorInPlace(id);
-        this->executeSwapJob(id, idsToRemove);
+        this->invalidateRepairJobs(id);
+        this->fixJobsAfterSwap(id, idsToRemove);
     }
     hnsw_index->removeLabel(label);
     for (idType id : idsToRemove) {
@@ -639,14 +667,29 @@ void TieredHNSWIndex<DataType, DistType>::executeRepairJob(HNSWRepairJob *job) {
         *it = repair_jobs.back();
         repair_jobs.pop_back();
     }
+    this->idToRepairJobsGuard.unlock();
+
+    hnsw_index->repairNodeConnections(job->node_id, job->level);
+
+    // Account for this job only now that its repair has actually been performed. Decreasing the
+    // counter beforehand would let a swap job be seen as ready, and its element isolated, while
+    // an element still points to it from a repair that has not run yet.
+    vecsim_stl::vector<idType> fully_repaired_ids(this->allocator);
+    this->idToRepairJobsGuard.lock();
     for (auto &it : job->associatedSwapJobs) {
         if (it->atomicDecreasePendingJobsNum() == 0) {
             readySwapJobs++;
+            fully_repaired_ids.push_back(it->deleted_id);
         }
     }
     this->idToRepairJobsGuard.unlock();
 
-    hnsw_index->repairNodeConnections(job->node_id, job->level);
+    // These deleted elements have no pending repair job left, so nothing points to them anymore.
+    // Take them out of the graph entirely, leaving no edge in or out for the swap job to deal with.
+    // Done outside the repair jobs guard, as isolating takes the per-element links locks.
+    for (idType deleted_id : fully_repaired_ids) {
+        this->isolateRepairedElement(deleted_id);
+    }
 
     this->mainIndexGuard.unlock_shared();
 }

src/VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h

@@ -24,6 +24,7 @@ INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_deleteVectorAndRepairAsync_Test)
 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_invalidRepairJobOnSwap_Test)
 INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_deleteVectorsAndSwapSync_Test)
 INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_BatchIterator_Test)
 INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_BatchIteratorAdvanced_Test)