Speed up InternalEngine#resolveDocVersion(...) method

server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -106,6 +106,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
@@ -1021,21 +1022,13 @@ private VersionValue resolveDocVersion(final Operation op, boolean loadSeqNo) th
         if (versionValue == null) {
             assert incrementIndexVersionLookup(); // used for asserting in tests
             final VersionsAndSeqNoResolver.DocIdAndVersion docIdAndVersion;
-            try (Searcher searcher = acquireSearcher("load_version", SearcherScope.INTERNAL)) {
+            try (var directoryReaderSupplier = acquireDirectoryReaderSupplier(SearcherScope.INTERNAL)) {
+                var indexReader = directoryReaderSupplier.getDirectoryReader();
                 if (engineConfig.getIndexSettings().getMode() == IndexMode.TIME_SERIES) {
                     assert engineConfig.getLeafSorter() == DataStream.TIMESERIES_LEAF_READERS_SORTER;
-                    docIdAndVersion = VersionsAndSeqNoResolver.timeSeriesLoadDocIdAndVersion(
-                        searcher.getIndexReader(),
-                        op.uid(),
-                        op.id(),
-                        loadSeqNo
-                    );
+                    docIdAndVersion = VersionsAndSeqNoResolver.timeSeriesLoadDocIdAndVersion(indexReader, op.uid(), op.id(), loadSeqNo);
                 } else {
-                    docIdAndVersion = VersionsAndSeqNoResolver.timeSeriesLoadDocIdAndVersion(
-                        searcher.getIndexReader(),
-                        op.uid(),
-                        loadSeqNo
-                    );
+                    docIdAndVersion = VersionsAndSeqNoResolver.timeSeriesLoadDocIdAndVersion(indexReader, op.uid(), loadSeqNo);
                 }
             }
             if (docIdAndVersion != null) {
@@ -3470,4 +3463,58 @@ public LiveVersionMap getLiveVersionMap() {
     protected long getPreCommitSegmentGeneration() {
         return preCommitSegmentGeneration.get();
     }
+
+    DirectoryReaderSupplier acquireDirectoryReaderSupplier(SearcherScope scope) throws EngineException {
+        assert scope == SearcherScope.INTERNAL : "acquireDirectoryReaderSupplier(...) isn't prepared for external usage";
+        assert store.hasReferences();
+        try {
+            ReferenceManager<ElasticsearchDirectoryReader> referenceManager = getReferenceManager(scope);
+            ElasticsearchDirectoryReader acquire = referenceManager.acquire();
+            return new DirectoryReaderSupplier(acquire) {
+
+                @Override
+                void doClose() {
+                    try {
+                        referenceManager.release(acquire);
+                    } catch (IOException e) {
+                        throw new UncheckedIOException("failed to close", e);
+                    } catch (AlreadyClosedException e) {
+                        // This means there's a bug somewhere: don't suppress it
+                        throw new AssertionError(e);
+                    }
+                }
+            };
+        } catch (AlreadyClosedException ex) {
+            throw ex;
+        } catch (Exception ex) {
+            maybeFailEngine("acquire_directory_reader", ex);
+            ensureOpen(ex); // throw EngineCloseException here if we are already closed
+            logger.error("failed to acquire directory reader", ex);
+            throw new EngineException(shardId, "failed to acquire directory reader", ex);
+        }
+    }
+
+    abstract static class DirectoryReaderSupplier implements Releasable {
+        private final DirectoryReader directoryReader;
+        private final AtomicBoolean released = new AtomicBoolean(false);
+
+        DirectoryReaderSupplier(DirectoryReader directoryReader) {
+            this.directoryReader = directoryReader;
+        }
+
+        public DirectoryReader getDirectoryReader() {
+            return directoryReader;
+        }
+
+        @Override
+        public final void close() {
+            if (released.compareAndSet(false, true)) {
+                doClose();
+            } else {
+                assert false : "DirectoryReaderSupplier was released twice";
+            }
+        }
+
+        abstract void doClose();
+    }
 }