Clone IndexInput when creating MemorySegmentPostingsVisitor (#129690)

Author: iverase

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

@@ -122,7 +122,7 @@ NeighborQueue scorePostingLists(FieldInfo fieldInfo, KnnCollector knnCollector,
     PostingVisitor getPostingVisitor(FieldInfo fieldInfo, IndexInput indexInput, float[] target, IntPredicate needsScoring)
         throws IOException {
         FieldEntry entry = fields.get(fieldInfo.number);
-        return new MemorySegmentPostingsVisitor(target, indexInput, entry, fieldInfo, needsScoring);
+        return new MemorySegmentPostingsVisitor(target, indexInput.clone(), entry, fieldInfo, needsScoring);
     }
 
     // TODO can we do this in off-heap blocks?

server/src/test/java/org/elasticsearch/index/codec/vectors/IVFVectorsFormatTests.java

@@ -34,6 +34,7 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.Locale;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static java.lang.String.format;
 import static org.elasticsearch.index.codec.vectors.IVFVectorsFormat.MAX_VECTORS_PER_CLUSTER;
@@ -128,4 +129,49 @@ public void testSimpleOffHeapSize() throws IOException {
             }
         }
     }
+
+    // this is a modified version of lucene's TestSearchWithThreads test case
+    public void testWithThreads() throws Exception {
+        final int numThreads = random().nextInt(2, 5);
+        final int numSearches = atLeast(100);
+        final int numDocs = atLeast(1000);
+        final int dimensions = random().nextInt(12, 500);
+        try (Directory dir = newDirectory(); IndexWriter w = new IndexWriter(dir, newIndexWriterConfig())) {
+            for (int docCount = 0; docCount < numDocs; docCount++) {
+                final Document doc = new Document();
+                doc.add(new KnnFloatVectorField("f", randomVector(dimensions), VectorSimilarityFunction.EUCLIDEAN));
+                w.addDocument(doc);
+            }
+            w.forceMerge(1);
+            try (IndexReader reader = DirectoryReader.open(w)) {
+                final AtomicBoolean failed = new AtomicBoolean();
+                Thread[] threads = new Thread[numThreads];
+                for (int threadID = 0; threadID < numThreads; threadID++) {
+                    threads[threadID] = new Thread(() -> {
+                        try {
+                            long totSearch = 0;
+                            for (; totSearch < numSearches && failed.get() == false; totSearch++) {
+                                float[] vector = randomVector(dimensions);
+                                LeafReader leafReader = getOnlyLeafReader(reader);
+                                leafReader.searchNearestVectors("f", vector, 10, leafReader.getLiveDocs(), Integer.MAX_VALUE);
+                            }
+                            assertTrue(totSearch > 0);
+                        } catch (Exception exc) {
+                            failed.set(true);
+                            throw new RuntimeException(exc);
+                        }
+                    });
+                    threads[threadID].setDaemon(true);
+                }
+
+                for (Thread t : threads) {
+                    t.start();
+                }
+
+                for (Thread t : threads) {
+                    t.join();
+                }
+            }
+        }
+    }
 }