Skip to content

Revert NonClosingReaderWrapper cache (#20921) #21830

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bugmakerrrrrr merged 1 commit into from
May 26, 2026
+6 −114
Merged

Revert NonClosingReaderWrapper cache (#20921) #21830

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 6 additions & 66 deletions server/src/main/java/org/opensearch/index/shard/IndexShard.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -333,8 +332,6 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl

private final IndexingOperationListener indexingOperationListeners;
private final Runnable globalCheckpointSyncer;
private final ConcurrentHashMap<DirectoryReader, NonClosingReaderWrapper> nonClosingReaderWrapperCache = new ConcurrentHashMap<>();
private final Function<DirectoryReader, DirectoryReader> nonClosingReaderWrapperSupplier;

Runnable getGlobalCheckpointSyncer() {
return globalCheckpointSyncer;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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;
Expand All @@ -2469,29 +2436,15 @@ private Engine.Searcher wrapSearcher(Engine.Searcher searcher) {
public static Engine.Searcher wrapSearcher(
Engine.Searcher engineSearcher,
CheckedFunction<DirectoryReader, DirectoryReader, IOException> readerWrapper
) throws IOException {
return wrapSearcher(engineSearcher, readerWrapper, null);
}

public static Engine.Searcher wrapSearcher(
Engine.Searcher engineSearcher,
CheckedFunction<DirectoryReader, DirectoryReader, IOException> readerWrapper,
Function<DirectoryReader, DirectoryReader> 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()) {
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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.
Comment thread
kkewwei marked this conversation as resolved.
// 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(),
Expand Down Expand Up @@ -6369,16 +6319,6 @@ public static <T> T applyOnEngine(Indexer indexer, Function<Engine, T> applier)
}
}

// Visible for testing
Function<DirectoryReader, DirectoryReader> nonClosingReaderWrapperSupplier() {
return nonClosingReaderWrapperSupplier;
}

// Visible for testing
ConcurrentHashMap<DirectoryReader, NonClosingReaderWrapper> nonClosingReaderWrapperCache() {
return nonClosingReaderWrapperCache;
}

// Visible for testing
Object getEngineMutex() {
return engineMutex;
Expand Down
48 changes: 0 additions & 48 deletions server/src/test/java/org/opensearch/index/shard/IndexShardTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<DirectoryReader, DirectoryReader, IOException> 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,
Expand Down
Loading