Adding asynchronous fetching for DirectIO directory

qa/vector/build.gradle

@@ -55,6 +55,9 @@ tasks.register("checkVec", JavaExec) {
   if (buildParams.getRuntimeJavaVersion().map { it.majorVersion.toInteger() }.get() >= 21) {
     jvmArgs '--add-modules=jdk.incubator.vector', '--enable-native-access=ALL-UNNAMED'
   }
+  if (System.getenv("DO_ASYNC_PROFILING") != null) {
+    jvmArgs System.getenv("DO_ASYNC_PROFILING")
+  }
   if (System.getenv("DO_PROFILING") != null) {
     jvmArgs '-XX:StartFlightRecording=dumponexit=true,maxsize=250M,filename=knn.jfr,settings=profile.jfc'
   }

qa/vector/src/main/java/module-info.java

@@ -18,5 +18,6 @@
     requires org.elasticsearch.logging;
     requires java.management;
     requires jdk.management;
+    requires org.apache.lucene.misc;
     requires org.elasticsearch.gpu;
 }

qa/vector/src/main/java/org/elasticsearch/test/knn/KnnIndexer.java

@@ -34,16 +34,11 @@
 import org.apache.lucene.index.VectorSimilarityFunction;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.FSDirectory;
-import org.apache.lucene.store.IOContext;
-import org.apache.lucene.store.IndexInput;
 import org.apache.lucene.store.MMapDirectory;
-import org.apache.lucene.store.NIOFSDirectory;
 import org.apache.lucene.store.NativeFSLockFactory;
 import org.apache.lucene.util.PrintStreamInfoStream;
 import org.elasticsearch.common.io.Channels;
-import org.elasticsearch.core.IOUtils;
-import org.elasticsearch.index.store.LuceneFilesExtensions;
-import org.elasticsearch.index.store.Store;
+import org.elasticsearch.index.store.FsDirectoryFactory;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -233,7 +228,7 @@ public boolean isEnabled(String component) {
     static Directory getDirectory(Path indexPath) throws IOException {
         Directory dir = FSDirectory.open(indexPath);
         if (dir instanceof MMapDirectory mmapDir) {
-            return new HybridDirectory(mmapDir);
+            return new FsDirectoryFactory.HybridDirectory(NativeFSLockFactory.INSTANCE, mmapDir);
         }
         return dir;
     }
@@ -375,64 +370,4 @@ synchronized void next(byte[] dest) throws IOException {
             bytes.get(dest);
         }
     }
-
-    // Copy of Elastic's HybridDirectory which extends NIOFSDirectory and uses MMapDirectory for certain files.
-    static final class HybridDirectory extends NIOFSDirectory {
-        private final MMapDirectory delegate;
-
-        HybridDirectory(MMapDirectory delegate) throws IOException {
-            super(delegate.getDirectory(), NativeFSLockFactory.INSTANCE);
-            this.delegate = delegate;
-        }
-
-        @Override
-        public IndexInput openInput(String name, IOContext context) throws IOException {
-            if (useDelegate(name, context)) {
-                // we need to do these checks on the outer directory since the inner doesn't know about pending deletes
-                ensureOpen();
-                ensureCanRead(name);
-                // we switch the context here since mmap checks for the READONCE context by identity
-                context = context == Store.READONCE_CHECKSUM ? IOContext.READONCE : context;
-                // we only use the mmap to open inputs. Everything else is managed by the NIOFSDirectory otherwise
-                // we might run into trouble with files that are pendingDelete in one directory but still
-                // listed in listAll() from the other. We on the other hand don't want to list files from both dirs
-                // and intersect for perf reasons.
-                return delegate.openInput(name, context);
-            } else {
-                return super.openInput(name, context);
-            }
-        }
-
-        @Override
-        public void close() throws IOException {
-            IOUtils.close(super::close, delegate);
-        }
-
-        private static String getExtension(String name) {
-            // Unlike FileSwitchDirectory#getExtension, we treat `tmp` as a normal file extension, which can have its own rules for mmaping.
-            final int lastDotIndex = name.lastIndexOf('.');
-            if (lastDotIndex == -1) {
-                return "";
-            } else {
-                return name.substring(lastDotIndex + 1);
-            }
-        }
-
-        static boolean useDelegate(String name, IOContext ioContext) {
-            if (ioContext == Store.READONCE_CHECKSUM) {
-                // If we're just reading the footer for the checksum then mmap() isn't really necessary, and it's desperately inefficient
-                // if pre-loading is enabled on this file.
-                return false;
-            }
-
-            final LuceneFilesExtensions extension = LuceneFilesExtensions.fromExtension(getExtension(name));
-            if (extension == null || extension.shouldMmap() == false) {
-                // Other files are either less performance-sensitive (e.g. stored field index, norms metadata)
-                // or are large and have a random access pattern and mmap leads to page cache trashing
-                // (e.g. stored fields and term vectors).
-                return false;
-            }
-            return true;
-        }
-    }
 }