Refactor index engines to manage readers instead of searchers (#43860)

This commit changes the way we manage refreshes in the index engines.
Instead of relying on a SearcherManager, this change uses a ReaderManager that
creates ElasticsearchDirectoryReader when needed. Searchers are now created on-demand
(when acquireSearcher is called) from the current ElasticsearchDirectoryReader.
It also slightly changes the Engine.Searcher to extend IndexSearcher in order
to simplify the usage in the consumer.

Author: jimczi

modules/percolator/src/test/java/org/elasticsearch/percolator/PercolatorQuerySearchTests.java

@@ -18,7 +18,6 @@
  */
 package org.elasticsearch.percolator;
 
-import org.apache.lucene.search.IndexSearcher;
 import org.apache.lucene.search.Query;
 import org.apache.lucene.search.join.ScoreMode;
 import org.elasticsearch.action.search.SearchResponse;
@@ -256,19 +255,18 @@ public void testRangeQueriesWithNow() throws Exception {
             .get();
         client().admin().indices().prepareRefresh().get();
 
-        try (Engine.Searcher engineSearcher = indexService.getShard(0).acquireSearcher("test")) {
-            IndexSearcher indexSearcher = engineSearcher.searcher();
+        try (Engine.Searcher searcher = indexService.getShard(0).acquireSearcher("test")) {
             long[] currentTime = new long[] {System.currentTimeMillis()};
             QueryShardContext queryShardContext =
-                indexService.newQueryShardContext(0, engineSearcher.reader(), () -> currentTime[0], null);
+                indexService.newQueryShardContext(0, searcher.getIndexReader(), () -> currentTime[0], null);
 
             BytesReference source = BytesReference.bytes(jsonBuilder().startObject()
                 .field("field1", "value")
                 .field("field2", currentTime[0])
                 .endObject());
             QueryBuilder queryBuilder = new PercolateQueryBuilder("query", source, XContentType.JSON);
             Query query = queryBuilder.toQuery(queryShardContext);
-            assertThat(indexSearcher.count(query), equalTo(3));
+            assertThat(searcher.count(query), equalTo(3));
 
             currentTime[0] = currentTime[0] + 10800000; // + 3 hours
             source = BytesReference.bytes(jsonBuilder().startObject()
@@ -277,7 +275,7 @@ public void testRangeQueriesWithNow() throws Exception {
                 .endObject());
             queryBuilder = new PercolateQueryBuilder("query", source, XContentType.JSON);
             query = queryBuilder.toQuery(queryShardContext);
-            assertThat(indexSearcher.count(query), equalTo(3));
+            assertThat(searcher.count(query), equalTo(3));
         }
     }
 

server/src/main/java/org/elasticsearch/index/IndexService.java

@@ -398,10 +398,10 @@ public synchronized IndexShard createShard(
 
             logger.debug("creating shard_id {}", shardId);
             // if we are on a shared FS we only own the shard (ie. we can safely delete it) if we are the primary.
-            final Engine.Warmer engineWarmer = (searcher) -> {
+            final Engine.Warmer engineWarmer = (reader) -> {
                 IndexShard shard =  getShardOrNull(shardId.getId());
                 if (shard != null) {
-                    warmer.warm(searcher, shard, IndexService.this.indexSettings);
+                    warmer.warm(reader, shard, IndexService.this.indexSettings);
                 }
             };
             Directory directory = directoryFactory.newDirectory(this.indexSettings, path);

server/src/main/java/org/elasticsearch/index/IndexWarmer.java

@@ -22,9 +22,8 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
-import org.apache.lucene.index.DirectoryReader;
+import org.elasticsearch.common.lucene.index.ElasticsearchDirectoryReader;
 import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.fielddata.IndexFieldData;
 import org.elasticsearch.index.fielddata.IndexFieldDataService;
 import org.elasticsearch.index.mapper.MappedFieldType;
@@ -58,22 +57,22 @@ public final class IndexWarmer {
         this.listeners = Collections.unmodifiableList(list);
     }
 
-    void warm(Engine.Searcher searcher, IndexShard shard, IndexSettings settings) {
+    void warm(ElasticsearchDirectoryReader reader, IndexShard shard, IndexSettings settings) {
         if (shard.state() == IndexShardState.CLOSED) {
             return;
         }
         if (settings.isWarmerEnabled() == false) {
             return;
         }
         if (logger.isTraceEnabled()) {
-            logger.trace("{} top warming [{}]", shard.shardId(), searcher.reader());
+            logger.trace("{} top warming [{}]", shard.shardId(), reader);
         }
         shard.warmerService().onPreWarm();
         long time = System.nanoTime();
         final List<TerminationHandle> terminationHandles = new ArrayList<>();
         // get a handle on pending tasks
         for (final Listener listener : listeners) {
-            terminationHandles.add(listener.warmReader(shard, searcher));
+            terminationHandles.add(listener.warmReader(shard, reader));
         }
         // wait for termination
         for (TerminationHandle terminationHandle : terminationHandles) {
@@ -103,7 +102,7 @@ public interface TerminationHandle {
     public interface Listener {
         /** Queue tasks to warm-up the given segments and return handles that allow to wait for termination of the
          *  execution of those tasks. */
-        TerminationHandle warmReader(IndexShard indexShard, Engine.Searcher searcher);
+        TerminationHandle warmReader(IndexShard indexShard, ElasticsearchDirectoryReader reader);
     }
 
     private static class FieldDataWarmer implements IndexWarmer.Listener {
@@ -117,7 +116,7 @@ private static class FieldDataWarmer implements IndexWarmer.Listener {
         }
 
         @Override
-        public TerminationHandle warmReader(final IndexShard indexShard, final Engine.Searcher searcher) {
+        public TerminationHandle warmReader(final IndexShard indexShard, final ElasticsearchDirectoryReader reader) {
             final MapperService mapperService = indexShard.mapperService();
             final Map<String, MappedFieldType> warmUpGlobalOrdinals = new HashMap<>();
             for (MappedFieldType fieldType : mapperService.fieldTypes()) {
@@ -133,7 +132,6 @@ public TerminationHandle warmReader(final IndexShard indexShard, final Engine.Se
                     try {
                         final long start = System.nanoTime();
                         IndexFieldData.Global ifd = indexFieldDataService.getForField(fieldType);
-                        DirectoryReader reader = searcher.getDirectoryReader();
                         IndexFieldData<?> global = ifd.loadGlobal(reader);
                         if (reader.leaves().isEmpty() == false) {
                             global.load(reader.leaves().get(0));

server/src/main/java/org/elasticsearch/index/cache/bitset/BitsetFilterCache.java

@@ -38,6 +38,7 @@
 import org.elasticsearch.common.cache.CacheBuilder;
 import org.elasticsearch.common.cache.RemovalListener;
 import org.elasticsearch.common.cache.RemovalNotification;
+import org.elasticsearch.common.lucene.index.ElasticsearchDirectoryReader;
 import org.elasticsearch.common.lucene.search.Queries;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
@@ -46,7 +47,6 @@
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.IndexWarmer;
 import org.elasticsearch.index.IndexWarmer.TerminationHandle;
-import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.mapper.DocumentMapper;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.ObjectMapper;
@@ -222,7 +222,7 @@ final class BitSetProducerWarmer implements IndexWarmer.Listener {
         }
 
         @Override
-        public IndexWarmer.TerminationHandle warmReader(final IndexShard indexShard, final Engine.Searcher searcher) {
+        public IndexWarmer.TerminationHandle warmReader(final IndexShard indexShard, final ElasticsearchDirectoryReader reader) {
             if (indexSettings.getIndex().equals(indexShard.indexSettings().getIndex()) == false) {
                 // this is from a different index
                 return TerminationHandle.NO_WAIT;
@@ -254,8 +254,8 @@ public IndexWarmer.TerminationHandle warmReader(final IndexShard indexShard, fin
                 warmUp.add(Queries.newNonNestedFilter());
             }
 
-            final CountDownLatch latch = new CountDownLatch(searcher.reader().leaves().size() * warmUp.size());
-            for (final LeafReaderContext ctx : searcher.reader().leaves()) {
+            final CountDownLatch latch = new CountDownLatch(reader.leaves().size() * warmUp.size());
+            for (final LeafReaderContext ctx : reader.leaves()) {
                 for (final Query filterToWarm : warmUp) {
                     executor.execute(() -> {
                         try {

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

@@ -0,0 +1,82 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.index.engine;
+
+import java.io.IOException;
+import java.util.function.BiConsumer;
+
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.search.ReferenceManager;
+
+import org.apache.lucene.search.SearcherManager;
+import org.elasticsearch.common.SuppressForbidden;
+import org.elasticsearch.common.lucene.index.ElasticsearchDirectoryReader;
+
+/**
+ * Utility class to safely share {@link ElasticsearchDirectoryReader} instances across
+ * multiple threads, while periodically reopening. This class ensures each
+ * reader is closed only once all threads have finished using it.
+ *
+ * @see SearcherManager
+ *
+ */
+@SuppressForbidden(reason = "reference counting is required here")
+class ElasticsearchReaderManager extends ReferenceManager<ElasticsearchDirectoryReader> {
+    private final BiConsumer<ElasticsearchDirectoryReader, ElasticsearchDirectoryReader> refreshListener;
+
+    /**
+     * Creates and returns a new ElasticsearchReaderManager from the given
+     * already-opened {@link ElasticsearchDirectoryReader}, stealing
+     * the incoming reference.
+     *
+     * @param reader            the directoryReader to use for future reopens
+     * @param refreshListener   A consumer that is called every time a new reader is opened
+     */
+    ElasticsearchReaderManager(ElasticsearchDirectoryReader reader,
+                               BiConsumer<ElasticsearchDirectoryReader, ElasticsearchDirectoryReader> refreshListener) {
+        this.current = reader;
+        this.refreshListener = refreshListener;
+        refreshListener.accept(current, null);
+    }
+
+    @Override
+    protected void decRef(ElasticsearchDirectoryReader reference) throws IOException {
+        reference.decRef();
+    }
+
+    @Override
+    protected ElasticsearchDirectoryReader refreshIfNeeded(ElasticsearchDirectoryReader referenceToRefresh) throws IOException {
+        final ElasticsearchDirectoryReader reader = (ElasticsearchDirectoryReader) DirectoryReader.openIfChanged(referenceToRefresh);
+        if (reader != null) {
+            refreshListener.accept(reader, referenceToRefresh);
+        }
+        return reader;
+    }
+
+    @Override
+    protected boolean tryIncRef(ElasticsearchDirectoryReader reference) {
+        return reference.tryIncRef();
+    }
+
+    @Override
+    protected int getRefCount(ElasticsearchDirectoryReader reference) {
+        return reference.getRefCount();
+    }
+}