Pulkitg64 commented on code in PR #16627:
URL: https://github.com/apache/lucene/pull/16627#discussion_r4175640111
##########
lucene/core/src/java/org/apache/lucene/util/hnsw/HnswConcurrentMergeBuilder.java:
##########
@@ -193,6 +201,7 @@ private void repairDisconnectedNodes() throws IOException {
// addConnections descends through the already-repaired upper levels.
taskExecutor.invokeAll(tasks);
}
+ return repairedNodes;
Review Comment:
Thanks!
##########
lucene/core/src/java/org/apache/lucene/util/hnsw/HnswConcurrentMergeBuilder.java:
##########
@@ -43,7 +43,10 @@ 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 disconnected nodes a repair worker claims atomically at a time.
+ // Number of disconnected nodes a repair worker claims atomically at a time.
Each repair is a
Review Comment:
I liked the shorter version of the comments. But I don't have any strong
opinion.
##########
lucene/core/src/test/org/apache/lucene/util/hnsw/TestHnswFloatVectorGraph.java:
##########
@@ -188,6 +219,303 @@ public void testAbortCheckInterruptsGraphInitAndRepair()
throws IOException {
}
}
+ /**
+ * 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,
+ 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,
+ 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", 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", Integer.toString(i), Field.Store.NO));
+ w.addDocument(doc);
+ }
+ w.flush(); // segment 2, deletion-free
+ // Delete 3 of every 10 base docs (30%): 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 = 30;
+ assertTrue(deletePct <
IncrementalHnswGraphMerger.DELETE_PCT_THRESHOLD);
Review Comment:
What do you think about selecting random delete % under DELETE_PCT_THRESHOLD
instead of using harcoded 30%?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
