Remove ConcurrentHnswMerger and related classes / params

Author: carlosdelest

server/src/main/java/org/elasticsearch/index/codec/vectors/ES814HnswScalarQuantizedVectorsFormat.java

@@ -61,7 +61,7 @@ public ES814HnswScalarQuantizedVectorsFormat(int maxConn, int beamWidth, Float c
 
     @Override
     public KnnVectorsWriter fieldsWriter(SegmentWriteState state) throws IOException {
-        return new ES910HnswVectorsWriter(state, maxConn, beamWidth, flatVectorsFormat.fieldsWriter(state), 1, null);
+        return new ES910HnswVectorsWriter(state, maxConn, beamWidth, flatVectorsFormat.fieldsWriter(state));
     }
 
     @Override

server/src/main/java/org/elasticsearch/index/codec/vectors/ES815HnswBitVectorsFormat.java

@@ -56,7 +56,7 @@ public ES815HnswBitVectorsFormat(int maxConn, int beamWidth) {
 
     @Override
     public KnnVectorsWriter fieldsWriter(SegmentWriteState state) throws IOException {
-        return new ES910HnswVectorsWriter(state, maxConn, beamWidth, flatVectorsFormat.fieldsWriter(state), 1, null);
+        return new ES910HnswVectorsWriter(state, maxConn, beamWidth, flatVectorsFormat.fieldsWriter(state));
     }
 
     @Override

server/src/main/java/org/elasticsearch/index/codec/vectors/es818/ES818HnswBinaryQuantizedVectorsFormat.java

@@ -119,7 +119,7 @@ public ES818HnswBinaryQuantizedVectorsFormat(int maxConn, int beamWidth, int num
 
     @Override
     public KnnVectorsWriter fieldsWriter(SegmentWriteState state) throws IOException {
-        return new ES910HnswVectorsWriter(state, maxConn, beamWidth, flatVectorsFormat.fieldsWriter(state), numMergeWorkers, mergeExec);
+        return new ES910HnswVectorsWriter(state, maxConn, beamWidth, flatVectorsFormat.fieldsWriter(state));
     }
 
     @Override

server/src/main/java/org/elasticsearch/index/codec/vectors/es910/ES910HnswVectorsFormat.java

@@ -28,19 +28,14 @@
 import org.apache.lucene.codecs.lucene99.Lucene99FlatVectorsFormat;
 import org.apache.lucene.codecs.lucene99.Lucene99HnswVectorsFormat;
 import org.apache.lucene.codecs.lucene99.Lucene99HnswVectorsReader;
-import org.apache.lucene.index.MergePolicy;
-import org.apache.lucene.index.MergeScheduler;
 import org.apache.lucene.index.SegmentReadState;
 import org.apache.lucene.index.SegmentWriteState;
-import org.apache.lucene.search.TaskExecutor;
 import org.apache.lucene.util.hnsw.HnswGraph;
 
 import java.io.IOException;
-import java.util.concurrent.ExecutorService;
 
 import static org.apache.lucene.codecs.lucene99.Lucene99HnswVectorsFormat.DEFAULT_BEAM_WIDTH;
 import static org.apache.lucene.codecs.lucene99.Lucene99HnswVectorsFormat.DEFAULT_MAX_CONN;
-import static org.apache.lucene.codecs.lucene99.Lucene99HnswVectorsFormat.DEFAULT_NUM_MERGE_WORKER;
 import static org.apache.lucene.codecs.lucene99.Lucene99HnswVectorsFormat.MAXIMUM_BEAM_WIDTH;
 import static org.apache.lucene.codecs.lucene99.Lucene99HnswVectorsFormat.MAXIMUM_MAX_CONN;
 
@@ -75,12 +70,9 @@ public class ES910HnswVectorsFormat extends KnnVectorsFormat {
         FlatVectorScorerUtil.getLucene99FlatVectorsScorer()
     );
 
-    private final int numMergeWorkers;
-    private final TaskExecutor mergeExec;
-
     /** Constructs a format using default graph construction parameters */
     public ES910HnswVectorsFormat() {
-        this(DEFAULT_MAX_CONN, DEFAULT_BEAM_WIDTH, DEFAULT_NUM_MERGE_WORKER, null);
+        this(DEFAULT_MAX_CONN, DEFAULT_BEAM_WIDTH);
     }
 
     /**
@@ -90,21 +82,6 @@ public ES910HnswVectorsFormat() {
      * @param beamWidth the size of the queue maintained during graph construction.
      */
     public ES910HnswVectorsFormat(int maxConn, int beamWidth) {
-        this(maxConn, beamWidth, DEFAULT_NUM_MERGE_WORKER, null);
-    }
-
-    /**
-     * Constructs a format using the given graph construction parameters and scalar quantization.
-     *
-     * @param maxConn the maximum number of connections to a node in the HNSW graph
-     * @param beamWidth the size of the queue maintained during graph construction.
-     * @param numMergeWorkers number of workers (threads) that will be used when doing merge. If
-     *     larger than 1, a non-null {@link ExecutorService} must be passed as mergeExec
-     * @param mergeExec the {@link ExecutorService} that will be used by ALL vector writers that are
-     *     generated by this format to do the merge. If null, the configured {@link
-     *     MergeScheduler#getIntraMergeExecutor(MergePolicy.OneMerge)} is used.
-     */
-    public ES910HnswVectorsFormat(int maxConn, int beamWidth, int numMergeWorkers, ExecutorService mergeExec) {
         super(ES910HnswVectorsFormat.NAME);
         if (maxConn <= 0 || maxConn > MAXIMUM_MAX_CONN) {
             throw new IllegalArgumentException(
@@ -118,20 +95,11 @@ public ES910HnswVectorsFormat(int maxConn, int beamWidth, int numMergeWorkers, E
         }
         this.maxConn = maxConn;
         this.beamWidth = beamWidth;
-        if (numMergeWorkers == 1 && mergeExec != null) {
-            throw new IllegalArgumentException("No executor service is needed as we'll use single thread to merge");
-        }
-        this.numMergeWorkers = numMergeWorkers;
-        if (mergeExec != null) {
-            this.mergeExec = new TaskExecutor(mergeExec);
-        } else {
-            this.mergeExec = null;
-        }
     }
 
     @Override
     public KnnVectorsWriter fieldsWriter(SegmentWriteState state) throws IOException {
-        return new ES910HnswVectorsWriter(state, maxConn, beamWidth, flatVectorsFormat.fieldsWriter(state), numMergeWorkers, mergeExec);
+        return new ES910HnswVectorsWriter(state, maxConn, beamWidth, flatVectorsFormat.fieldsWriter(state));
     }
 
     @Override

server/src/main/java/org/elasticsearch/index/codec/vectors/es910/ES910HnswVectorsWriter.java

@@ -36,7 +36,6 @@
 import org.apache.lucene.index.SegmentWriteState;
 import org.apache.lucene.index.Sorter;
 import org.apache.lucene.index.VectorSimilarityFunction;
-import org.apache.lucene.search.TaskExecutor;
 import org.apache.lucene.store.IndexOutput;
 import org.apache.lucene.util.IOUtils;
 import org.apache.lucene.util.InfoStream;
@@ -46,7 +45,6 @@
 import org.apache.lucene.util.hnsw.RandomVectorScorerSupplier;
 import org.apache.lucene.util.hnsw.UpdateableRandomVectorScorer;
 import org.apache.lucene.util.packed.DirectMonotonicWriter;
-import org.elasticsearch.index.codec.vectors.es910.hnsw.ConcurrentHnswMerger;
 import org.elasticsearch.index.codec.vectors.es910.hnsw.HnswGraph;
 import org.elasticsearch.index.codec.vectors.es910.hnsw.HnswGraphBuilder;
 import org.elasticsearch.index.codec.vectors.es910.hnsw.HnswGraphMerger;
@@ -76,25 +74,14 @@ public class ES910HnswVectorsWriter extends KnnVectorsWriter {
     private final int M;
     private final int beamWidth;
     private final FlatVectorsWriter flatVectorWriter;
-    private final int numMergeWorkers;
-    private final TaskExecutor mergeExec;
 
     private final List<ES910HnswVectorsWriter.FieldWriter<?>> fields = new ArrayList<>();
     private boolean finished;
 
-    public ES910HnswVectorsWriter(
-        SegmentWriteState state,
-        int M,
-        int beamWidth,
-        FlatVectorsWriter flatVectorWriter,
-        int numMergeWorkers,
-        TaskExecutor mergeExec
-    ) throws IOException {
+    public ES910HnswVectorsWriter(SegmentWriteState state, int M, int beamWidth, FlatVectorsWriter flatVectorWriter) throws IOException {
         this.M = M;
         this.flatVectorWriter = flatVectorWriter;
         this.beamWidth = beamWidth;
-        this.numMergeWorkers = numMergeWorkers;
-        this.mergeExec = mergeExec;
         segmentWriteState = state;
 
         String metaFileName = IndexFileNames.segmentFileName(
@@ -367,12 +354,7 @@ public void mergeOneField(FieldInfo fieldInfo, MergeState mergeState) throws IOE
             int[][] vectorIndexNodeOffsets = null;
             if (scorerSupplier.totalVectorCount() > 0) {
                 // build graph
-                HnswGraphMerger merger = createGraphMerger(
-                    fieldInfo,
-                    scorerSupplier,
-                    mergeState.intraMergeTaskExecutor == null ? null : new TaskExecutor(mergeState.intraMergeTaskExecutor),
-                    numMergeWorkers
-                );
+                HnswGraphMerger merger = createGraphMerger(fieldInfo, scorerSupplier);
                 for (int i = 0; i < mergeState.liveDocs.length; i++) {
                     if (hasVectorValues(mergeState.fieldInfos[i], fieldInfo.name)) {
                         merger.addReader(mergeState.knnVectorsReaders[i], mergeState.docMaps[i], mergeState.liveDocs[i]);
@@ -508,18 +490,7 @@ private void writeMeta(
         }
     }
 
-    private HnswGraphMerger createGraphMerger(
-        FieldInfo fieldInfo,
-        RandomVectorScorerSupplier scorerSupplier,
-        TaskExecutor parallelMergeTaskExecutor,
-        int numParallelMergeWorkers
-    ) {
-        if (mergeExec != null) {
-            return new ConcurrentHnswMerger(fieldInfo, scorerSupplier, M, beamWidth, mergeExec, numMergeWorkers);
-        }
-        if (parallelMergeTaskExecutor != null && numParallelMergeWorkers > 1) {
-            return new ConcurrentHnswMerger(fieldInfo, scorerSupplier, M, beamWidth, parallelMergeTaskExecutor, numParallelMergeWorkers);
-        }
+    private HnswGraphMerger createGraphMerger(FieldInfo fieldInfo, RandomVectorScorerSupplier scorerSupplier) {
         return new IncrementalHnswGraphMerger(fieldInfo, scorerSupplier, M, beamWidth);
     }