You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ho...@apache.org on 2026/10/21 22:55:29 UTC
(lucene) branch branch_10_6 updated: Parallelize the repair process for disconnected nodes during HNSW merge-re-use (#16627)
This is an automated email from the ASF dual-hosted git repository.
HoustonPutman pushed a commit to branch branch_10_6
in repository https://gitbox.apache.org/repos/asf/lucene.git
The following commit(s) were added to refs/heads/branch_10_6 by this push:
new a58062065eb Parallelize the repair process for disconnected nodes during HNSW merge-re-use (#16627)
a58062065eb is described below
commit a58062065ebda6bf7e05e1b453be54565738b181
Author: Ethan Baker <1e...@gmail.com>
AuthorDate: Thu Oct 8 14:22:06 2026 -0400
Parallelize the repair process for disconnected nodes during HNSW merge-re-use (#16627)
(cherry picked from commit d8468fd93757116c11760bbdaad90c212ffa62f1)
---
lucene/CHANGES.txt | 4 +
.../lucene/util/hnsw/ConcurrentHnswMerger.java | 20 +-
.../util/hnsw/HnswConcurrentMergeBuilder.java | 90 +++++-
.../apache/lucene/util/hnsw/HnswGraphBuilder.java | 131 +++++++-
.../util/hnsw/InitializedHnswGraphBuilder.java | 201 +++++-------
.../lucene/util/hnsw/TestHnswFloatVectorGraph.java | 338 ++++++++++++++++++++-
6 files changed, 657 insertions(+), 127 deletions(-)
diff --git a/lucene/CHANGES.txt b/lucene/CHANGES.txt
index 95aee0e3360..c7ebe87af70 100644
--- a/lucene/CHANGES.txt
+++ b/lucene/CHANGES.txt
@@ -237,6 +237,10 @@ Optimizations
* GITHUB#16797: Stored fields data is written with DataAccessHint.SEQUENTIAL and NoReuseHint,
matching how it is read. (Jim Ferenczi)
+* GITHUB#16618: Repair disconnected nodes during HNSW merge-reuse across the merge worker
+ pool instead of serially on the merge thread, so reusing a graph with deletes is no longer
+ slower than a full rebuild. (Ethan Baker)
+
* GITHUB#16796: Merges read inner files of compound files through a mapping of their own, so the
access they advise does not replace the advice of searches. (Jim Ferenczi)
diff --git a/lucene/core/src/java/org/apache/lucene/util/hnsw/ConcurrentHnswMerger.java b/lucene/core/src/java/org/apache/lucene/util/hnsw/ConcurrentHnswMerger.java
index f784513d6cb..a57c5f34757 100644
--- a/lucene/core/src/java/org/apache/lucene/util/hnsw/ConcurrentHnswMerger.java
+++ b/lucene/core/src/java/org/apache/lucene/util/hnsw/ConcurrentHnswMerger.java
@@ -74,6 +74,7 @@ public class ConcurrentHnswMerger extends IncrementalHnswGraphMerger {
throws IOException {
OnHeapHnswGraph graph;
BitSet initializedNodes = null;
+ InitializedHnswGraphBuilder.PrunedGraph prunedGraph = null;
if (largestGraphReader == null) {
graph = new OnHeapHnswGraph(M, maxOrd);
@@ -95,18 +96,27 @@ public class ConcurrentHnswMerger extends IncrementalHnswGraphMerger {
initGraphSize,
mergedVectorValues,
initializedNodes);
- graph =
- InitializedHnswGraphBuilder.initGraph(
+ InitializedHnswGraphBuilder.PrunedGraph result =
+ InitializedHnswGraphBuilder.pruneGraph(
+ scorerSupplier,
+ beamWidth,
initializerGraph,
oldToNewOrdinalMap,
maxOrd,
- beamWidth,
- scorerSupplier,
abortCheck);
+ graph = result.graph();
+ prunedGraph = result.hasDeletes() ? result : null;
}
}
return new HnswConcurrentMergeBuilder(
- taskExecutor, numWorker, scorerSupplier, M, beamWidth, graph, initializedNodes);
+ taskExecutor,
+ numWorker,
+ scorerSupplier,
+ M,
+ beamWidth,
+ graph,
+ initializedNodes,
+ prunedGraph);
}
/**
diff --git a/lucene/core/src/java/org/apache/lucene/util/hnsw/HnswConcurrentMergeBuilder.java b/lucene/core/src/java/org/apache/lucene/util/hnsw/HnswConcurrentMergeBuilder.java
index 6a63ba596a6..c4312b45027 100644
--- a/lucene/core/src/java/org/apache/lucene/util/hnsw/HnswConcurrentMergeBuilder.java
+++ b/lucene/core/src/java/org/apache/lucene/util/hnsw/HnswConcurrentMergeBuilder.java
@@ -33,6 +33,7 @@ import org.apache.lucene.util.BitSet;
import org.apache.lucene.util.FixedBitSet;
import org.apache.lucene.util.IORunnable;
import org.apache.lucene.util.InfoStream;
+import org.apache.lucene.util.IntsRef;
/**
* A graph builder that manages multiple workers, it only supports adding the whole graph all at
@@ -40,12 +41,16 @@ import org.apache.lucene.util.InfoStream;
*/
public class HnswConcurrentMergeBuilder implements HnswBuilder {
- private static final int DEFAULT_BATCH_SIZE =
- 2048; // number of vectors the worker handles sequentially at one batch
+ // Number of vectors the worker handles sequentially at one batch.
+ private static final int DEFAULT_BATCH_SIZE = 2048;
+
+ // Number of disconnected nodes a repair worker claims at a time.
+ private static final int REPAIR_BATCH_SIZE = 64;
private final TaskExecutor taskExecutor;
private final ConcurrentMergeWorker[] workers;
private final HnswLock hnswLock;
+ private final InitializedHnswGraphBuilder.PrunedGraph prunedGraph;
private InfoStream infoStream = InfoStream.getDefault();
private boolean frozen;
@@ -58,6 +63,25 @@ public class HnswConcurrentMergeBuilder implements HnswBuilder {
OnHeapHnswGraph hnsw,
BitSet initializedNodes)
throws IOException {
+ this(taskExecutor, numWorker, scorerSupplier, M, beamWidth, hnsw, initializedNodes, null);
+ }
+
+ /**
+ * Creates a builder that, when the pruned source graph had deletes, repairs its disconnected
+ * nodes across the worker pool and rebalances it before inserting the remaining vectors.
+ *
+ * @param prunedGraph the deferred repair state, or null when the reused graph had no deletes
+ */
+ HnswConcurrentMergeBuilder(
+ TaskExecutor taskExecutor,
+ int numWorker,
+ RandomVectorScorerSupplier scorerSupplier,
+ int M,
+ int beamWidth,
+ OnHeapHnswGraph hnsw,
+ BitSet initializedNodes,
+ InitializedHnswGraphBuilder.PrunedGraph prunedGraph)
+ throws IOException {
this.taskExecutor = taskExecutor;
AtomicInteger workProgress = new AtomicInteger(0);
workers = new ConcurrentMergeWorker[numWorker];
@@ -74,6 +98,7 @@ public class HnswConcurrentMergeBuilder implements HnswBuilder {
initializedNodes,
workProgress);
}
+ this.prunedGraph = prunedGraph;
}
@Override
@@ -92,6 +117,24 @@ public class HnswConcurrentMergeBuilder implements HnswBuilder {
worker.setMergeStartTimeNs(mergeStartTimeNs);
worker.setCumulativeWorkTimeNs(cumulativeWorkTimeNs);
}
+ if (prunedGraph != null) {
+ long repairStartNs = System.nanoTime();
+ int repairedNodes = repairDisconnectedNodes();
+ long rebalanceStartNs = System.nanoTime();
+ prunedGraph.builder().rebalanceGraph();
+ long rebalanceEndNs = System.nanoTime();
+ if (infoStream.isEnabled(HNSW_COMPONENT)) {
+ infoStream.message(
+ HNSW_COMPONENT,
+ String.format(
+ Locale.ROOT,
+ "repaired reused graph: %d nodes in %.2f ms with %d workers, %.2f ms rebalance",
+ repairedNodes,
+ (rebalanceStartNs - repairStartNs) / 1_000_000.0,
+ workers.length,
+ (rebalanceEndNs - rebalanceStartNs) / 1_000_000.0));
+ }
+ }
List<Callable<Void>> futures = new ArrayList<>();
for (int i = 0; i < workers.length; i++) {
int finalI = i;
@@ -119,6 +162,49 @@ public class HnswConcurrentMergeBuilder implements HnswBuilder {
return getCompletedGraph();
}
+ /**
+ * Repairs the pruned graph's disconnected nodes across the worker pool, one level at a time from
+ * the top down.
+ *
+ * @return the number of nodes repaired, counting a node once per level it was repaired on
+ */
+ private int repairDisconnectedNodes() throws IOException {
+ int repairedNodes = 0;
+ for (int level = prunedGraph.numLevels() - 1; level >= 0; level--) {
+ IntsRef disconnectedNodes = prunedGraph.disconnectedNodesByLevel()[level];
+ if (disconnectedNodes == null) {
+ continue;
+ }
+ int total = disconnectedNodes.length;
+ repairedNodes += total;
+ // Use at most one task per worker and no more tasks than repair batches; each task
+ // dynamically claims batches below.
+ int taskCount = Math.min(workers.length, Math.ceilDiv(total, REPAIR_BATCH_SIZE));
+ AtomicInteger repairProgress = new AtomicInteger(0);
+ int repairLevel = level;
+ List<Callable<Void>> tasks = new ArrayList<>(taskCount);
+ for (int t = 0; t < taskCount; t++) {
+ ConcurrentMergeWorker worker = workers[t];
+ tasks.add(
+ () -> {
+ int from;
+ while ((from = repairProgress.getAndAdd(REPAIR_BATCH_SIZE)) < total) {
+ int length = Math.min(REPAIR_BATCH_SIZE, total - from);
+ worker.fixDisconnectedNodes(
+ new IntsRef(disconnectedNodes.ints, disconnectedNodes.offset + from, length),
+ repairLevel,
+ worker.scorer);
+ }
+ return null;
+ });
+ }
+ // Finish every repair task at this level before repairing the next lower level, because
+ // addConnections descends through the already-repaired upper levels.
+ taskExecutor.invokeAll(tasks);
+ }
+ return repairedNodes;
+ }
+
@Override
public void addGraphNode(int node) throws IOException {
throw new UnsupportedOperationException("This builder is for merge only");
diff --git a/lucene/core/src/java/org/apache/lucene/util/hnsw/HnswGraphBuilder.java b/lucene/core/src/java/org/apache/lucene/util/hnsw/HnswGraphBuilder.java
index 400c7531edb..da4423a3bfc 100644
--- a/lucene/core/src/java/org/apache/lucene/util/hnsw/HnswGraphBuilder.java
+++ b/lucene/core/src/java/org/apache/lucene/util/hnsw/HnswGraphBuilder.java
@@ -35,6 +35,7 @@ import org.apache.lucene.search.knn.KnnSearchStrategy;
import org.apache.lucene.util.FixedBitSet;
import org.apache.lucene.util.IORunnable;
import org.apache.lucene.util.InfoStream;
+import org.apache.lucene.util.IntsRef;
import org.apache.lucene.util.hnsw.HnswUtil.Component;
/**
@@ -440,8 +441,21 @@ public class HnswGraphBuilder implements HnswBuilder {
*/
NeighborArray neighbors = hnsw.getNeighbors(level, node);
int maxConnOnLevel = level == 0 ? M * 2 : M;
- boolean[] mask =
- selectAndLinkDiverse(node, neighbors, candidates, maxConnOnLevel, scorer, isLinkRepair);
+ boolean[] mask;
+ if (isLinkRepair && hnswLock != null) {
+ // A repaired node is already discoverable, so lock its own array against a concurrent
+ // reciprocal write.
+ Lock selfLock = hnswLock.write(level, node);
+ try {
+ mask =
+ selectAndLinkDiverse(node, neighbors, candidates, maxConnOnLevel, scorer, isLinkRepair);
+ } finally {
+ selfLock.unlock();
+ }
+ } else {
+ mask =
+ selectAndLinkDiverse(node, neighbors, candidates, maxConnOnLevel, scorer, isLinkRepair);
+ }
// Link the selected nodes to the new node, and the new node to the selected nodes (again
// applying diversity heuristic)
@@ -699,6 +713,119 @@ public class HnswGraphBuilder implements HnswBuilder {
}
}
+ /**
+ * Fixes disconnected nodes at a specific level by performing graph searches from their existing
+ * neighbors to find additional connections.
+ *
+ * <p>For each disconnected node:
+ *
+ * <ol>
+ * <li>Use existing neighbors as entry points for graph search
+ * <li>Search the level to find candidate neighbors
+ * <li>Add diverse neighbors using the HNSW heuristic selection algorithm
+ * </ol>
+ *
+ * <p>A node with no neighbors is instead connected from scratch via {@link #addConnections}.
+ *
+ * <p>When {@link #hnswLock} is set (concurrent repair) each node's existing neighbors are
+ * snapshot under its read lock, since another worker may be adding a reciprocal link into the
+ * same array.
+ *
+ * @param disconnectedNodes node ordinals that need additional neighbors
+ * @param level the level at which to repair connections
+ * @param scorer vector similarity scorer for distance calculations
+ * @throws IOException if an I/O error occurs during search operations
+ */
+ void fixDisconnectedNodes(
+ IntsRef disconnectedNodes, int level, UpdateableRandomVectorScorer scorer)
+ throws IOException {
+ int beamWidth = beamCandidates.k();
+ GraphBuilderKnnCollector candidates = new GraphBuilderKnnCollector(beamWidth);
+ NeighborArray scratchArray = new NeighborArray(beamWidth, false);
+
+ for (int i = disconnectedNodes.offset;
+ i < disconnectedNodes.offset + disconnectedNodes.length;
+ i++) {
+ int node = disconnectedNodes.ints[i];
+ maybeAbort();
+ scorer.setScoringOrdinal(node);
+
+ int[] entryPoints;
+ if (hnswLock != null) {
+ Lock readLock = hnswLock.read(level, node);
+ try {
+ NeighborArray existingNeighbors = hnsw.getNeighbors(level, node);
+ int size = existingNeighbors.size();
+ entryPoints = new int[size];
+ System.arraycopy(existingNeighbors.nodes(), 0, entryPoints, 0, size);
+ } finally {
+ readLock.unlock();
+ }
+ } else {
+ NeighborArray existingNeighbors = hnsw.getNeighbors(level, node);
+ int size = existingNeighbors.size();
+ entryPoints = new int[size];
+ System.arraycopy(existingNeighbors.nodes(), 0, entryPoints, 0, size);
+ }
+
+ // Use all existing neighbors as entry points for search
+ if (entryPoints.length > 0) {
+ // Search from entry points to find candidate neighbors
+ graphSearcher.searchLevel(candidates, scorer, level, entryPoints, hnsw, null);
+ popToScratch(candidates, scratchArray);
+ // Add diverse neighbors using HNSW heuristic (prunes similar neighbors)
+ addDiverseNeighbors(level, node, scratchArray, scorer, true);
+ } else {
+ // Node has no neighbors, add connections from scratch
+ addConnections(node, level, scorer);
+ }
+
+ // Clear for next iteration
+ scratchArray.clear();
+ candidates.clear();
+ }
+ }
+
+ /**
+ * Adds connections for an existing node at a specific level in the graph hierarchy.
+ *
+ * <p>The process involves:
+ *
+ * <ol>
+ * <li>Navigate down from the top level to find the closest node at the target level
+ * <li>Perform a full search at the target level to find neighbors
+ * <li>Add diverse neighbors using the HNSW heuristic selection
+ * </ol>
+ *
+ * @param node the node ordinal to add connections for
+ * @param targetLevel the level to add connections at
+ * @param scorer vector similarity scorer for distance calculations
+ * @throws IOException if an I/O error occurs during search or neighbor addition
+ */
+ void addConnections(int node, int targetLevel, UpdateableRandomVectorScorer scorer)
+ throws IOException {
+
+ int beamWidth = beamCandidates.k();
+ GraphBuilderKnnCollector candidates = new GraphBuilderKnnCollector(beamWidth);
+ int[] eps = {hnsw.entryNode()};
+
+ // Navigate down from top to target level, greedily moving toward the new node
+ for (int level = hnsw.numLevels() - 1; level > targetLevel; level--) {
+ graphSearcher.searchLevel(candidates, scorer, level, eps, hnsw, null);
+ eps[0] = candidates.popNode();
+ candidates.clear();
+ }
+
+ // Perform full search at target level to find neighbors
+ graphSearcher.searchLevel(candidates, scorer, targetLevel, eps, hnsw, null);
+
+ NeighborArray scratchArray = new NeighborArray(beamWidth, false);
+ popToScratch(candidates, scratchArray);
+
+ // Add diverse neighbors and establish bidirectional connections
+ addDiverseNeighbors(targetLevel, node, scratchArray, scorer, true);
+ }
+
/**
* A restricted, specialized knnCollector that can be used when building a graph.
*
diff --git a/lucene/core/src/java/org/apache/lucene/util/hnsw/InitializedHnswGraphBuilder.java b/lucene/core/src/java/org/apache/lucene/util/hnsw/InitializedHnswGraphBuilder.java
index 8e6a8c03a7b..fe6452334ad 100644
--- a/lucene/core/src/java/org/apache/lucene/util/hnsw/InitializedHnswGraphBuilder.java
+++ b/lucene/core/src/java/org/apache/lucene/util/hnsw/InitializedHnswGraphBuilder.java
@@ -20,17 +20,15 @@ package org.apache.lucene.util.hnsw;
import static org.apache.lucene.search.DocIdSetIterator.NO_MORE_DOCS;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
import java.util.SplittableRandom;
import org.apache.lucene.internal.hppc.IntArrayList;
import org.apache.lucene.internal.hppc.IntCursor;
import org.apache.lucene.util.ArrayUtil;
import org.apache.lucene.util.BitSet;
import org.apache.lucene.util.IORunnable;
+import org.apache.lucene.util.IntsRef;
+import org.apache.lucene.util.IntsRefBuilder;
/**
* This creates a graph builder that is initialized with the provided HnswGraph. This is useful for
@@ -92,6 +90,9 @@ public final class InitializedHnswGraphBuilder extends HnswGraphBuilder {
// Tracks if the graph has deletes
private boolean hasDeletes = false;
+ /** Seed for the random promotions in {@link #rebalanceGraph}. */
+ private final long seed;
+
/**
* Creates an initialized HNSW graph builder from an existing graph.
*
@@ -228,6 +229,60 @@ public final class InitializedHnswGraphBuilder extends HnswGraphBuilder {
throws IOException {
super(scorerSupplier, beamWidth, seed, initializedGraph);
this.initializedNodes = initializedNodes;
+ this.seed = seed;
+ }
+
+ /**
+ * Prunes deleted nodes and edges from the initializer graph without repairing or rebalancing it,
+ * returning the pruned graph plus the state a caller needs to run those phases itself.
+ *
+ * @param scorerSupplier provides vector similarity scoring for graph operations
+ * @param beamWidth the search beam width for graph construction
+ * @param initializerGraph the source graph to prune
+ * @param newOrdMap maps old ordinals to new ordinals; -1 indicates deleted documents
+ * @param totalNumberOfVectors the total number of vectors in the merged graph
+ * @param abortCheck optional check invoked during pruning; may be null
+ * @return the pruned graph and its deferred repair/rebalance state
+ * @throws IOException if an I/O error occurs while pruning
+ */
+ static PrunedGraph pruneGraph(
+ RandomVectorScorerSupplier scorerSupplier,
+ int beamWidth,
+ HnswGraph initializerGraph,
+ int[] newOrdMap,
+ int totalNumberOfVectors,
+ IORunnable abortCheck)
+ throws IOException {
+ InitializedHnswGraphBuilder builder =
+ new InitializedHnswGraphBuilder(
+ scorerSupplier,
+ beamWidth,
+ randSeed,
+ new OnHeapHnswGraph(initializerGraph.maxConn(), totalNumberOfVectors),
+ null);
+ if (abortCheck != null) {
+ builder.setAbortCheck(abortCheck);
+ }
+ IntsRef[] disconnectedNodesByLevel =
+ builder.copyAndPruneGraphStructure(initializerGraph, newOrdMap);
+ return new PrunedGraph(builder, disconnectedNodesByLevel, builder.hasDeletes);
+ }
+
+ /**
+ * A pruned graph and the state needed to repair and rebalance it, as produced by {@link
+ * #pruneGraph}. Repair and rebalance are deferred so a caller can parallelize the repair phase;
+ * the builder is retained because the serial rebalance runs on it.
+ */
+ record PrunedGraph(
+ InitializedHnswGraphBuilder builder, IntsRef[] disconnectedNodesByLevel, boolean hasDeletes) {
+
+ OnHeapHnswGraph graph() {
+ return builder.getGraph();
+ }
+
+ int numLevels() {
+ return disconnectedNodesByLevel.length;
+ }
}
/**
@@ -247,8 +302,7 @@ public final class InitializedHnswGraphBuilder extends HnswGraphBuilder {
private void initializeFromGraph(HnswGraph initializerGraph, int[] newOrdMap) throws IOException {
hasDeletes = false;
// Phase 1: Copy structure and identify nodes that lost too many neighbors
- Map<Integer, List<Integer>> disconnectedNodesByLevel =
- copyGraphStructure(initializerGraph, newOrdMap);
+ IntsRef[] disconnectedNodesByLevel = copyAndPruneGraphStructure(initializerGraph, newOrdMap);
// Repair graph if it has deletes
if (hasDeletes) {
@@ -261,8 +315,8 @@ public final class InitializedHnswGraphBuilder extends HnswGraphBuilder {
}
/**
- * Copies the graph structure from the initializer graph, applying ordinal remapping and
- * identifying nodes that have lost neighbors.
+ * Copies the surviving graph structure while pruning deleted nodes and edges and applying ordinal
+ * remapping.
*
* <p>Deleted neighbors show up as -1 in newOrdMap and are dropped. A node that loses any this way
* is flagged for repair when it crosses one of the two thresholds ({@link
@@ -277,18 +331,19 @@ public final class InitializedHnswGraphBuilder extends HnswGraphBuilder {
*
* @param initializerGraph the source graph to copy from
* @param newOrdMap maps old ordinals to new ordinals; -1 indicates deleted documents
- * @return map of level to list of disconnected node ordinals at that level
+ * @return disconnected node ordinals by level; a null entry means no nodes need repair at that
+ * level
* @throws IOException if an I/O error occurs during graph traversal
*/
- private Map<Integer, List<Integer>> copyGraphStructure(
- HnswGraph initializerGraph, int[] newOrdMap) throws IOException {
+ private IntsRef[] copyAndPruneGraphStructure(HnswGraph initializerGraph, int[] newOrdMap)
+ throws IOException {
int numLevels = initializerGraph.numLevels();
levelToNodes = new IntArrayList[numLevels];
- Map<Integer, List<Integer>> disconnectedNodesByLevel = new HashMap<>(numLevels);
+ IntsRef[] disconnectedNodesByLevel = new IntsRef[numLevels];
for (int level = numLevels - 1; level >= 0; level--) {
levelToNodes[level] = new IntArrayList();
- List<Integer> disconnectedNodes = new ArrayList<>();
+ IntsRefBuilder disconnectedNodes = null;
HnswGraph.NodesIterator it = initializerGraph.getNodesOnLevel(level);
while (it.hasNext()) {
@@ -328,10 +383,15 @@ public final class InitializedHnswGraphBuilder extends HnswGraphBuilder {
if (newNeighbors.size() < oldNeighbourCount * DISCONNECTED_NODE_FACTOR
|| (newNeighbors.size() < oldNeighbourCount
&& newNeighbors.size() < maxConnOnLevel * CUMULATIVE_DEGREE_FLOOR_FACTOR)) {
- disconnectedNodes.add(newOrd);
+ if (disconnectedNodes == null) {
+ disconnectedNodes = new IntsRefBuilder();
+ }
+ disconnectedNodes.append(newOrd);
}
}
- disconnectedNodesByLevel.put(level, disconnectedNodes);
+ if (disconnectedNodes != null) {
+ disconnectedNodesByLevel[level] = disconnectedNodes.get();
+ }
}
return disconnectedNodesByLevel;
}
@@ -340,71 +400,18 @@ public final class InitializedHnswGraphBuilder extends HnswGraphBuilder {
* Repairs disconnected nodes at all levels by finding additional neighbors to restore
* connectivity.
*
- * @param disconnectedNodesByLevel map of level to disconnected nodes at that level
+ * @param disconnectedNodesByLevel disconnected node ordinals by level; a null entry means no
+ * nodes need repair at that level
* @param numLevels total number of levels in the graph hierarchy
* @throws IOException if an I/O error occurs during repair operations
*/
- private void repairDisconnectedNodes(
- Map<Integer, List<Integer>> disconnectedNodesByLevel, int numLevels) throws IOException {
- for (int level = numLevels - 1; level >= 0; level--) {
- fixDisconnectedNodes(disconnectedNodesByLevel.get(level), level, scorer);
- }
- }
-
- /**
- * Fixes disconnected nodes at a specific level by performing graph searches from their existing
- * neighbors to find additional connections.
- *
- * <p>For each disconnected node:
- *
- * <ol>
- * <li>Use existing neighbors as entry points for graph search
- * <li>Search the level to find candidate neighbors
- * <li>Add diverse neighbors using the HNSW heuristic selection algorithm
- * </ol>
- *
- * <p>If a node has no neighbors at all, it cannot be repaired at this level and will rely on the
- * rebalancing phase.
- *
- * @param disconnectedNodes list of node ordinals that need additional neighbors
- * @param level the level at which to repair connections
- * @param scorer vector similarity scorer for distance calculations
- * @throws IOException if an I/O error occurs during search operations
- */
- private void fixDisconnectedNodes(
- List<Integer> disconnectedNodes, int level, UpdateableRandomVectorScorer scorer)
+ private void repairDisconnectedNodes(IntsRef[] disconnectedNodesByLevel, int numLevels)
throws IOException {
- if (disconnectedNodes.isEmpty()) return;
-
- int beamWidth = beamCandidates.k();
- GraphBuilderKnnCollector candidates = new GraphBuilderKnnCollector(beamWidth);
- NeighborArray scratchArray = new NeighborArray(beamWidth, false);
-
- for (int node : disconnectedNodes) {
- maybeAbort();
- scorer.setScoringOrdinal(node);
- NeighborArray existingNeighbors = hnsw.getNeighbors(level, node);
-
- // Only repair if node has at least one neighbor to use as entry point
- if (existingNeighbors.size() > 0) {
- // Use all existing neighbors as entry points for search
- int[] entryPoints = new int[existingNeighbors.size()];
- System.arraycopy(existingNeighbors.nodes(), 0, entryPoints, 0, existingNeighbors.size());
-
- // Search from entry points to find candidate neighbors
- graphSearcher.searchLevel(candidates, scorer, level, entryPoints, hnsw, null);
- popToScratch(candidates, scratchArray);
-
- // Add diverse neighbors using HNSW heuristic (prunes similar neighbors)
- addDiverseNeighbors(level, node, scratchArray, scorer, true);
- } else {
- // Node has no nighbors, add connections from scratch
- addConnections(node, level, scorer);
+ for (int level = numLevels - 1; level >= 0; level--) {
+ IntsRef disconnectedNodes = disconnectedNodesByLevel[level];
+ if (disconnectedNodes != null) {
+ fixDisconnectedNodes(disconnectedNodes, level, scorer);
}
-
- // Clear for next iteration
- scratchArray.clear();
- candidates.clear();
}
}
@@ -424,8 +431,8 @@ public final class InitializedHnswGraphBuilder extends HnswGraphBuilder {
*
* @throws IOException if an I/O error occurs during node promotion
*/
- private void rebalanceGraph() throws IOException {
- SplittableRandom random = new SplittableRandom();
+ void rebalanceGraph() throws IOException {
+ SplittableRandom random = new SplittableRandom(seed);
int size = hnsw.size();
double invMaxConn = 1.0 / M;
@@ -476,46 +483,6 @@ public final class InitializedHnswGraphBuilder extends HnswGraphBuilder {
}
}
- /**
- * Adds connections for an existing node at a specific level in the graph hierarchy.
- *
- * <p>The process involves:
- *
- * <ol>
- * <li>Navigate down from the top level to find the closest node at the target level
- * <li>Perform a full search at the target level to find neighbors
- * <li>Add diverse neighbors using the HNSW heuristic selection
- * </ol>
- *
- * @param node the node ordinal to add connections for
- * @param targetLevel the level to add connections at
- * @param scorer vector similarity scorer for distance calculations
- * @throws IOException if an I/O error occurs during search or neighbor addition
- */
- private void addConnections(int node, int targetLevel, UpdateableRandomVectorScorer scorer)
- throws IOException {
-
- int beamWidth = beamCandidates.k();
- GraphBuilderKnnCollector candidates = new GraphBuilderKnnCollector(beamWidth);
- int[] eps = {hnsw.entryNode()};
-
- // Navigate down from top to target level, greedily moving toward the new node
- for (int level = hnsw.numLevels() - 1; level > targetLevel; level--) {
- graphSearcher.searchLevel(candidates, scorer, level, eps, hnsw, null);
- eps[0] = candidates.popNode();
- candidates.clear();
- }
-
- // Perform full search at target level to find neighbors
- graphSearcher.searchLevel(candidates, scorer, targetLevel, eps, hnsw, null);
-
- NeighborArray scratchArray = new NeighborArray(beamWidth, false);
- popToScratch(candidates, scratchArray);
-
- // Add diverse neighbors and establish bidirectional connections
- addDiverseNeighbors(targetLevel, node, scratchArray, scorer, true);
- }
-
@Override
public void addGraphNode(int node) throws IOException {
if (initializedNodes.get(node)) {
diff --git a/lucene/core/src/test/org/apache/lucene/util/hnsw/TestHnswFloatVectorGraph.java b/lucene/core/src/test/org/apache/lucene/util/hnsw/TestHnswFloatVectorGraph.java
index 329da3a0275..0ba34acdbaf 100644
--- a/lucene/core/src/test/org/apache/lucene/util/hnsw/TestHnswFloatVectorGraph.java
+++ b/lucene/core/src/test/org/apache/lucene/util/hnsw/TestHnswFloatVectorGraph.java
@@ -17,31 +17,65 @@
package org.apache.lucene.util.hnsw;
+import static org.apache.lucene.search.DocIdSetIterator.NO_MORE_DOCS;
+
import com.carrotsearch.randomizedtesting.RandomizedTest;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.io.PrintStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.lucene.codecs.Codec;
+import org.apache.lucene.codecs.hnsw.HnswGraphProvider;
+import org.apache.lucene.codecs.lucene99.Lucene99HnswVectorsFormat;
+import org.apache.lucene.codecs.perfield.PerFieldKnnVectorsFormat;
+import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.document.KnnFloatVectorField;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.index.CodecReader;
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.FilterLeafReader;
import org.apache.lucene.index.FloatVectorValues;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
import org.apache.lucene.index.KnnVectorValues;
import org.apache.lucene.index.LeafReader;
+import org.apache.lucene.index.LeafReaderContext;
import org.apache.lucene.index.MergePolicy;
+import org.apache.lucene.index.NoMergePolicy;
+import org.apache.lucene.index.Term;
import org.apache.lucene.index.VectorEncoding;
import org.apache.lucene.index.VectorSimilarityFunction;
+import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.KnnCollector;
import org.apache.lucene.search.KnnFloatVectorQuery;
import org.apache.lucene.search.Query;
import org.apache.lucene.search.ScoreDoc;
+import org.apache.lucene.search.TaskExecutor;
import org.apache.lucene.search.TopDocs;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.tests.util.TestUtil;
import org.apache.lucene.util.ArrayUtil;
import org.apache.lucene.util.FixedBitSet;
import org.apache.lucene.util.IORunnable;
+import org.apache.lucene.util.IntsRef;
+import org.apache.lucene.util.NamedThreadFactory;
+import org.apache.lucene.util.PrintStreamInfoStream;
import org.junit.Before;
/** Tests HNSW KNN graphs */
public class TestHnswFloatVectorGraph extends HnswGraphTestCase<float[]> {
+ private static final String ID_FIELD = "id";
+
@Before
public void setup() {
similarityFunction = RandomizedTest.randomFrom(VectorSimilarityFunction.values());
@@ -148,7 +182,8 @@ public class TestHnswFloatVectorGraph extends HnswGraphTestCase<float[]> {
newOrdMap[i] = (i == entry || (i & 1) == 0) ? i : -1;
}
- // copyGraphStructure polls the abort check exactly once per source node across every level.
+ // copyAndPruneGraphStructure polls the abort check exactly once per source node across every
+ // level.
int copyLoopChecks = 0;
for (int level = 0; level < graph.numLevels(); level++) {
copyLoopChecks += graph.getNodesOnLevel(level).size();
@@ -188,6 +223,307 @@ public class TestHnswFloatVectorGraph extends HnswGraphTestCase<float[]> {
}
}
+ /**
+ * With a single worker, {@link HnswConcurrentMergeBuilder}'s locked repair branch (non-null
+ * {@code hnswLock}) must produce the same graph as the serial {@link
+ * InitializedHnswGraphBuilder#initGraph} path (null {@code hnswLock}). Copy, repair and the
+ * seeded rebalance are all deterministic, and one worker repairs in the same order as the serial
+ * loop, so every level must match.
+ */
+ public void testSingleWorkerRepairMatchesSerial() throws IOException {
+ long savedRandSeed = HnswGraphBuilder.randSeed;
+ ExecutorService exec = Executors.newFixedThreadPool(1, new NamedThreadFactory("hnswRepair1"));
+ try {
+ HnswGraphBuilder.randSeed = 42;
+ similarityFunction = VectorSimilarityFunction.EUCLIDEAN;
+ int M = 16;
+ int beamWidth = 100;
+ int size = 512;
+ int dim = 16;
+ MockVectorValues vectors =
+ MockVectorValues.fromValues(createRandomFloatVectors(size, dim, new Random(42)));
+ RandomVectorScorerSupplier supplier = buildScorerSupplier(vectors);
+ OnHeapHnswGraph base = HnswGraphBuilder.create(supplier, M, beamWidth, 42).build(size);
+
+ // Delete ~33%, compacting survivors into a dense ordinal space so the merged graph has no
+ // holes. pruneGraph is called directly here, so the merger's reuse threshold does not apply.
+ int[] newOrdMap = new int[size];
+ int liveCount = 0;
+ for (int old = 0; old < size; old++) {
+ newOrdMap[old] = (old % 3 == 1) ? -1 : liveCount++;
+ }
+
+ OnHeapHnswGraph serial =
+ InitializedHnswGraphBuilder.initGraph(base, newOrdMap, liveCount, beamWidth, supplier);
+
+ InitializedHnswGraphBuilder.PrunedGraph prunedGraph =
+ InitializedHnswGraphBuilder.pruneGraph(
+ supplier, beamWidth, base, newOrdMap, liveCount, null);
+ assertTrue(
+ "deletes must trigger repair for this test to be meaningful", prunedGraph.hasDeletes());
+ IntsRef flaggedAtLevel0 = prunedGraph.disconnectedNodesByLevel()[0];
+ assertNotNull("level 0 must have flagged nodes", flaggedAtLevel0);
+ assertTrue(
+ "level 0 must have flagged nodes, else the equality is vacuous",
+ flaggedAtLevel0.length > 0);
+ FixedBitSet initialized = new FixedBitSet(liveCount);
+ initialized.set(0, liveCount); // all live nodes pre-populated; no extra inserts in build()
+ HnswConcurrentMergeBuilder builder =
+ new HnswConcurrentMergeBuilder(
+ new TaskExecutor(exec),
+ 1,
+ supplier,
+ M,
+ beamWidth,
+ prunedGraph.graph(),
+ initialized,
+ prunedGraph);
+ builder.build(liveCount);
+ OnHeapHnswGraph concurrent = builder.getCompletedGraph();
+
+ assertEquals(serial.size(), concurrent.size());
+ assertEquals(serial.numLevels(), concurrent.numLevels());
+ assertEquals(serial.entryNode(), concurrent.entryNode());
+ for (int level = 0; level < serial.numLevels(); level++) {
+ for (int node = 0; node < liveCount; node++) {
+ assertEquals(
+ "node existence differs at level " + level + " for node " + node,
+ serial.nodeExistAtLevel(level, node),
+ concurrent.nodeExistAtLevel(level, node));
+ if (serial.nodeExistAtLevel(level, node)) {
+ assertEquals(
+ "neighbors differ at level " + level + " for node " + node,
+ sortedNeighbors(serial, level, node),
+ sortedNeighbors(concurrent, level, node));
+ }
+ }
+ }
+ } finally {
+ exec.shutdownNow();
+ HnswGraphBuilder.randSeed = savedRandSeed;
+ }
+ }
+
+ /**
+ * Multi-worker repair must be thread-safe and correct: no exception or deadlock, the merged graph
+ * has the right size and a single rooted level-0 component, and every flagged node is repaired
+ * (gains at least one neighbor and never drops below its pre-repair degree). The graph data is
+ * pinned so the flagged set is deterministic; the real thread pool supplies the interleaving that
+ * {@code -Ptests.dups} stresses. Assertions are floors/invariants only, never full-graph
+ * equality.
+ */
+ public void testConcurrentRepairIsThreadSafeAndRepairsFlaggedNodes() throws IOException {
+ long savedRandSeed = HnswGraphBuilder.randSeed;
+ ExecutorService exec = Executors.newFixedThreadPool(2, new NamedThreadFactory("hnswRepair2"));
+ try {
+ HnswGraphBuilder.randSeed = 42;
+ similarityFunction = VectorSimilarityFunction.EUCLIDEAN;
+ int M = 16;
+ int beamWidth = 100;
+ int size = 512;
+ int dim = 16;
+ MockVectorValues vectors =
+ MockVectorValues.fromValues(createRandomFloatVectors(size, dim, new Random(42)));
+ RandomVectorScorerSupplier supplier = buildScorerSupplier(vectors);
+ OnHeapHnswGraph base = HnswGraphBuilder.create(supplier, M, beamWidth, 42).build(size);
+
+ int[] newOrdMap = new int[size];
+ int liveCount = 0;
+ for (int old = 0; old < size; old++) {
+ newOrdMap[old] = (old % 3 == 1) ? -1 : liveCount++; // ~33% deleted
+ }
+
+ InitializedHnswGraphBuilder.PrunedGraph prunedGraph =
+ InitializedHnswGraphBuilder.pruneGraph(
+ supplier, beamWidth, base, newOrdMap, liveCount, null);
+ assertTrue("deletes must trigger repair", prunedGraph.hasDeletes());
+
+ // Snapshot the flagged nodes and their pre-repair degrees, and confirm repair partitions
+ // across >= 2 levels so the top-down per-level barrier is actually exercised.
+ IntsRef[] flaggedByLevel = prunedGraph.disconnectedNodesByLevel();
+ List<int[]> flaggedBefore = new ArrayList<>(); // {level, node, degreeBeforeRepair}
+ int levelsWithFlags = 0;
+ for (int level = 0; level < flaggedByLevel.length; level++) {
+ IntsRef nodesAtLevel = flaggedByLevel[level];
+ if (nodesAtLevel == null) {
+ continue;
+ }
+ levelsWithFlags++;
+ for (int i = nodesAtLevel.offset; i < nodesAtLevel.offset + nodesAtLevel.length; i++) {
+ int node = nodesAtLevel.ints[i];
+ flaggedBefore.add(
+ new int[] {level, node, prunedGraph.graph().getNeighbors(level, node).size()});
+ }
+ }
+ assertTrue(
+ "expected flagged nodes on >= 2 levels, got " + levelsWithFlags, levelsWithFlags >= 2);
+ assertFalse("expected some flagged nodes", flaggedBefore.isEmpty());
+ int maxFlaggedOnAnyLevel = 0;
+ for (IntsRef nodesAtLevel : flaggedByLevel) {
+ if (nodesAtLevel != null) {
+ maxFlaggedOnAnyLevel = Math.max(maxFlaggedOnAnyLevel, nodesAtLevel.length);
+ }
+ }
+ assertTrue(
+ "need > 64 flagged nodes on some level to submit multiple same-level repair tasks, got "
+ + maxFlaggedOnAnyLevel,
+ maxFlaggedOnAnyLevel > 64);
+
+ FixedBitSet initialized = new FixedBitSet(liveCount);
+ initialized.set(0, liveCount);
+ HnswConcurrentMergeBuilder builder =
+ new HnswConcurrentMergeBuilder(
+ new TaskExecutor(exec),
+ 2,
+ supplier,
+ M,
+ beamWidth,
+ prunedGraph.graph(),
+ initialized,
+ prunedGraph);
+ builder.build(liveCount);
+ OnHeapHnswGraph merged = builder.getCompletedGraph();
+
+ assertEquals("merged graph should contain every live node", liveCount, merged.size());
+ // No isolated nodes: level 0 is a single rooted component after concurrent repair.
+ assertEquals(
+ "graph should be a single rooted component after repair",
+ 1,
+ HnswUtil.componentSizes(merged).size());
+
+ // Every flagged node gained >= 1 neighbor and never lost neighbors during repair.
+ for (int[] fb : flaggedBefore) {
+ int level = fb[0];
+ int node = fb[1];
+ int before = fb[2];
+ int after = merged.getNeighbors(level, node).size();
+ assertTrue(
+ "flagged node "
+ + node
+ + " lost neighbors on level "
+ + level
+ + ": "
+ + before
+ + " -> "
+ + after,
+ after >= before);
+ assertTrue(
+ "flagged node " + node + " has no neighbors after repair on level " + level, after > 0);
+ }
+ } finally {
+ exec.shutdownNow();
+ HnswGraphBuilder.randSeed = savedRandSeed;
+ }
+ }
+
+ /**
+ * End-to-end: a real force-merge with 2 merge workers over an index whose largest segment carries
+ * deletes (below {@code IncrementalHnswGraphMerger.DELETE_PCT_THRESHOLD}, so its graph is reused
+ * as the merge base) drives {@link ConcurrentHnswMerger} through {@code pruneGraph} and the
+ * concurrent repair path. The merged graph must be sized correctly, rooted, and searchable.
+ */
+ public void testConcurrentMergeReuseWithDeletesEndToEnd() throws IOException {
+ similarityFunction = RandomizedTest.randomFrom(VectorSimilarityFunction.values());
+ int M = 16;
+ int beamWidth = 100;
+ int dim = 16;
+ String field = "vec";
+ int baseSize = atLeast(1200);
+ int addSize = 300;
+ int total = baseSize + addSize;
+ MockVectorValues vectors =
+ MockVectorValues.fromValues(createRandomFloatVectors(total, dim, random()));
+ ExecutorService mergeExec =
+ Executors.newFixedThreadPool(2, new NamedThreadFactory("hnsw-merge-worker"));
+ try (Directory dir = newDirectory()) {
+ ByteArrayOutputStream infoBytes = new ByteArrayOutputStream();
+ Codec codec =
+ TestUtil.alwaysKnnVectorsFormat(
+ // 2 merge workers + always build a graph (tinySegmentsThreshold = 0)
+ new Lucene99HnswVectorsFormat(M, beamWidth, 2, mergeExec, 0));
+ // ingest with merging off, so the force-merge below sees exactly these two segments
+ try (IndexWriter w =
+ new IndexWriter(
+ dir,
+ new IndexWriterConfig().setCodec(codec).setMergePolicy(NoMergePolicy.INSTANCE))) {
+ for (int i = 0; i < baseSize; i++) {
+ Document doc = new Document();
+ doc.add(knnVectorField(field, vectors.vectorValue(i), similarityFunction));
+ doc.add(new StringField(ID_FIELD, Integer.toString(i), Field.Store.NO));
+ w.addDocument(doc);
+ }
+ w.flush(); // segment 1 = the base graph
+ for (int i = baseSize; i < total; i++) {
+ Document doc = new Document();
+ doc.add(knnVectorField(field, vectors.vectorValue(i), similarityFunction));
+ doc.add(new StringField(ID_FIELD, Integer.toString(i), Field.Store.NO));
+ w.addDocument(doc);
+ }
+ w.flush(); // segment 2, deletion-free
+ // Delete under the merger's reuse threshold, so the base segment is still chosen as the
+ // reuse base, but carrying deletes so repair runs.
+ int deletePct =
+ TestUtil.nextInt(random(), 1, IncrementalHnswGraphMerger.DELETE_PCT_THRESHOLD - 1);
+ int toDelete = baseSize * deletePct / 100;
+ for (int d = 0; d < toDelete; d++) {
+ w.deleteDocuments(new Term(ID_FIELD, Integer.toString(d)));
+ }
+ w.commit();
+ }
+ try (IndexWriter w =
+ new IndexWriter(
+ dir,
+ new IndexWriterConfig()
+ .setCodec(codec)
+ .setInfoStream(
+ new PrintStreamInfoStream(
+ new PrintStream(infoBytes, true, StandardCharsets.UTF_8))))) {
+ w.forceMerge(1);
+ }
+ assertTrue(
+ "expected the concurrent reuse+repair path to run",
+ infoBytes.toString(StandardCharsets.UTF_8).contains("repaired reused graph"));
+ try (IndexReader reader = DirectoryReader.open(dir)) {
+ int liveCount = reader.numDocs();
+ assertTrue("expected some docs to be deleted", liveCount < total);
+ for (LeafReaderContext ctx : reader.leaves()) {
+ HnswGraph graph =
+ ((HnswGraphProvider)
+ ((PerFieldKnnVectorsFormat.FieldsReader)
+ ((CodecReader) FilterLeafReader.unwrap(ctx.reader()))
+ .getVectorReader())
+ .getFieldReader(field))
+ .getGraph(field);
+ assertEquals(liveCount, graph.size());
+ for (int node = 0; node < graph.size(); node++) {
+ graph.seek(0, node);
+ assertNotEquals(
+ "node " + node + " has no level-0 neighbors after repair",
+ NO_MORE_DOCS,
+ graph.nextNeighbor());
+ }
+ }
+
+ IndexSearcher searcher = newSearcher(reader);
+ TopDocs td = searcher.search(knnQuery(field, randomVector(dim), 10), 10);
+ assertEquals("KNN search should return k results", 10, td.scoreDocs.length);
+ }
+ } finally {
+ mergeExec.shutdownNow();
+ }
+ }
+
+ private static List<Integer> sortedNeighbors(OnHeapHnswGraph graph, int level, int node) {
+ NeighborArray arr = graph.getNeighbors(level, node);
+ List<Integer> neighbors = new ArrayList<>(arr.size());
+ int[] nodes = arr.nodes();
+ for (int i = 0; i < arr.size(); i++) {
+ neighbors.add(nodes[i]);
+ }
+ Collections.sort(neighbors);
+ return neighbors;
+ }
+
@Override
MockVectorValues vectorValues(LeafReader reader, String fieldName) throws IOException {
FloatVectorValues vectorValues = reader.getFloatVectorValues(fieldName);