diff --git a/server/src/main/java/org/opensearch/index/shard/IndexShard.java b/server/src/main/java/org/opensearch/index/shard/IndexShard.java index 7444460d694cc..90bb8f3cb050b 100644 --- a/server/src/main/java/org/opensearch/index/shard/IndexShard.java +++ b/server/src/main/java/org/opensearch/index/shard/IndexShard.java @@ -252,7 +252,6 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.CompletionService; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorCompletionService; @@ -333,8 +332,6 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl private final IndexingOperationListener indexingOperationListeners; private final Runnable globalCheckpointSyncer; - private final ConcurrentHashMap nonClosingReaderWrapperCache = new ConcurrentHashMap<>(); - private final Function nonClosingReaderWrapperSupplier; Runnable getGlobalCheckpointSyncer() { return globalCheckpointSyncer; @@ -559,34 +556,6 @@ public boolean shouldCache(Query query) { } else { readerWrapper = indexReaderWrapper; } - - nonClosingReaderWrapperSupplier = directoryReader -> { - int[] fromCache = new int[] { 0 }; - try { - // To prevent instantiating a new NonClosingReaderWrapper per query/get/update request, - // the wrapper can be shared across all uses of the same NonClosingReaderWrapper. - return nonClosingReaderWrapperCache.computeIfAbsent(directoryReader, key -> { - try { - NonClosingReaderWrapper closingReaderWrapper = new NonClosingReaderWrapper(key); - fromCache[0] = 1; - return closingReaderWrapper; - } catch (IOException e) { - fromCache[0] = 2; - throw new OpenSearchException("failed to wrap searcher", e); - } - }); - } finally { - if (fromCache[0] == 1) { - OpenSearchDirectoryReader.addReaderCloseListener( - directoryReader, - cacheKey -> nonClosingReaderWrapperCache.remove(directoryReader) - ); - } else if (fromCache[0] == 2) { - nonClosingReaderWrapperCache.remove(directoryReader); - } - } - }; - refreshListeners = buildRefreshListeners(); lastSearcherAccess.set(threadPool.relativeTimeInMillis()); persistMetadata(path, indexSettings, shardRouting, null, logger); @@ -2448,9 +2417,7 @@ private Engine.Searcher wrapSearcher(Engine.Searcher searcher) { : "DirectoryReader must be an instance or OpenSearchDirectoryReader"; boolean success = false; try { - final Engine.Searcher newSearcher = readerWrapper == null - ? searcher - : wrapSearcher(searcher, readerWrapper, nonClosingReaderWrapperSupplier); + final Engine.Searcher newSearcher = readerWrapper == null ? searcher : wrapSearcher(searcher, readerWrapper); assert newSearcher != null; success = true; return newSearcher; @@ -2469,29 +2436,15 @@ private Engine.Searcher wrapSearcher(Engine.Searcher searcher) { public static Engine.Searcher wrapSearcher( Engine.Searcher engineSearcher, CheckedFunction readerWrapper - ) throws IOException { - return wrapSearcher(engineSearcher, readerWrapper, null); - } - - public static Engine.Searcher wrapSearcher( - Engine.Searcher engineSearcher, - CheckedFunction readerWrapper, - Function nonClosingReaderWrapperSupplier ) throws IOException { assert readerWrapper != null; - DirectoryReader directoryReader = engineSearcher.getDirectoryReader(); - final OpenSearchDirectoryReader openSearchDirectoryReader = OpenSearchDirectoryReader.getOpenSearchDirectoryReader(directoryReader); + final OpenSearchDirectoryReader openSearchDirectoryReader = OpenSearchDirectoryReader.getOpenSearchDirectoryReader( + engineSearcher.getDirectoryReader() + ); if (openSearchDirectoryReader == null) { throw new IllegalStateException("Can't wrap non opensearch directory reader"); } - - DirectoryReader nonClosingReaderWrapper; - if (nonClosingReaderWrapperSupplier == null) { - nonClosingReaderWrapper = new NonClosingReaderWrapper(directoryReader); - } else { - nonClosingReaderWrapper = nonClosingReaderWrapperSupplier.apply(directoryReader); - assert nonClosingReaderWrapper instanceof NonClosingReaderWrapper; - } + NonClosingReaderWrapper nonClosingReaderWrapper = new NonClosingReaderWrapper(engineSearcher.getDirectoryReader()); DirectoryReader reader = readerWrapper.apply(nonClosingReaderWrapper); if (reader != nonClosingReaderWrapper) { if (reader.getReaderCacheHelper() != openSearchDirectoryReader.getReaderCacheHelper()) { @@ -2588,7 +2541,6 @@ public void close(String reason, boolean flushEngine, boolean deleted) throws IO changeState(IndexShardState.CLOSED, reason); } } finally { - nonClosingReaderWrapperCache.clear(); final Indexer engine = this.currentEngineReference.getAndSet(null); try { if (engine != null && flushEngine) { @@ -4595,9 +4547,7 @@ public void beforeRefresh() {} public void afterRefresh(boolean didRefresh) { if (!didRefresh) return; // Use the engine directly (not IndexShard.acquireSearcher) so that we do NOT - // go through IndexShard.wrapSearcher / nonClosingReaderWrapperSupplier. - // Going through the shard-level wrapper would create entries in the - // nonClosingReaderWrapperCache that callers do not expect. + // go through IndexShard.wrapSearcher. try ( Engine.Searcher searcher = applyOnEngine( getIndexer(), @@ -6369,16 +6319,6 @@ public static T applyOnEngine(Indexer indexer, Function applier) } } - // Visible for testing - Function nonClosingReaderWrapperSupplier() { - return nonClosingReaderWrapperSupplier; - } - - // Visible for testing - ConcurrentHashMap nonClosingReaderWrapperCache() { - return nonClosingReaderWrapperCache; - } - // Visible for testing Object getEngineMutex() { return engineMutex; diff --git a/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java b/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java index 6b2d88d448226..3b4cedcd180a0 100644 --- a/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java +++ b/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java @@ -5514,54 +5514,6 @@ public void testPeriodicFlushTaskExecutesFlush() throws Exception { closeShards(primary); } - public void testCacheWrapperReader() throws IOException { - Settings settings = Settings.builder() - .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT) - .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0) - .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1) - .put(IndexSettings.INDEX_PERIODIC_FLUSH_INTERVAL_SETTING.getKey(), "1s") - .build(); - - IndexMetadata metadata = IndexMetadata.builder("test") - .putMapping("{ \"properties\": { \"foo\": { \"type\": \"text\"}}}") - .settings(settings) - .primaryTerm(0, 1) - .build(); - - CheckedFunction wrapper = reader -> reader; - - IndexShard primary = newShard(new ShardId(metadata.getIndex(), 0), true, "n1", metadata, wrapper); - recoverShardFromStore(primary); - indexDoc(primary, "_doc", "0", "{\"foo\" : \"bar\"}"); - primary.flush(new FlushRequest()); - - try ( - Engine.SearcherSupplier searcherSupplier = primary.acquireSearcherSupplier(); - Engine.Searcher searcher = searcherSupplier.acquireSearcher("foo") - ) { - DirectoryReader directoryReader = searcher.getDirectoryReader(); - Engine.Searcher wrap = IndexShard.wrapSearcher(searcher, wrapper, primary.nonClosingReaderWrapperSupplier()); - wrap.close(); - assertEquals(1, primary.nonClosingReaderWrapperCache().size()); - DirectoryReader nonClosingReaderWrapper = primary.nonClosingReaderWrapperCache().get(directoryReader); - assertNotNull(nonClosingReaderWrapper); - - // use the cache - wrap = IndexShard.wrapSearcher(searcher, wrapper, primary.nonClosingReaderWrapperSupplier()); - wrap.close(); - assertEquals(1, primary.nonClosingReaderWrapperCache().size()); - DirectoryReader newNonClosingReaderWrapper = primary.nonClosingReaderWrapperCache().get(directoryReader); - assertEquals(nonClosingReaderWrapper, newNonClosingReaderWrapper); - - // not use the cache - wrap = IndexShard.wrapSearcher(searcher, wrapper, null); - assertNotEquals(wrap, newNonClosingReaderWrapper); - wrap.close(); - } - closeShards(primary); - assertTrue(primary.nonClosingReaderWrapperCache().isEmpty()); - } - /** * Verifies that {@code isRemoteSegmentStoreInSync} uses {@code getCatalogSnapshot()} (the unified * catalog API) rather than the legacy {@code getSegmentInfosSnapshot()}. After indexing and refreshing,