Skip to content

Invalidate exact bitset filter cache keys on reader close #22498

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
jainankitk merged 4 commits into from
Aug 18, 2026
Merged

Invalidate exact bitset filter cache keys on reader close #22498

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a2036c8
Invalidate exact bitset filter cache keys on reader close
jwils Jul 23, 2026
9452a6c
Merge branch 'main' into fix-bitset-stale-retention
jwils Jul 27, 2026
2e255f5
Merge branch 'main' into fix-bitset-stale-retention
jwils Aug 13, 2026
413decd
Merge branch 'main' into fix-bitset-stale-retention
jwils Aug 15, 2026
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
106 changes: 31 additions & 75 deletions server/src/main/java/org/opensearch/indices/IndicesBitsetFilterCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,6 @@

package org.opensearch.indices;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.lucene.index.FilterLeafReader;
import org.apache.lucene.index.IndexReader;
Expand All @@ -30,7 +28,6 @@
import org.opensearch.common.cache.CacheBuilder;
import org.opensearch.common.cache.RemovalListener;
import org.opensearch.common.cache.RemovalNotification;
import org.opensearch.common.lease.Releasable;
import org.opensearch.common.lucene.index.OpenSearchDirectoryReader;
import org.opensearch.common.lucene.search.Queries;
import org.opensearch.common.settings.Setting;
Expand All @@ -55,16 +52,16 @@
import java.util.HashSet;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.ToLongBiFunction;

/**
* Node-level cache for {@link BitSet} based filters. Manages a single flat cache shared across
* all indices on the node, with a configurable size limit and async stale entry cleanup.
* Stale entries from closed readers are purged periodically by a background cleanup task.
* all indices on the node, with a configurable size limit. Entries for a reader are invalidated
* by exact key as soon as the reader closes.
*
* @opensearch.api
*/
Expand All @@ -75,8 +72,6 @@ public class IndicesBitsetFilterCache
RemovalListener<IndicesBitsetFilterCache.BitsetCacheKey, IndicesBitsetFilterCache.Value>,
Closeable {

private static final Logger logger = LogManager.getLogger(IndicesBitsetFilterCache.class);

public static final Setting<Boolean> INDEX_LOAD_RANDOM_ACCESS_FILTERS_EAGERLY_SETTING = Setting.boolSetting(
"index.load_fixed_bitset_filters_eagerly",
true,
Expand All @@ -89,28 +84,24 @@ public class IndicesBitsetFilterCache
Property.NodeScope
);

public static final Setting<TimeValue> INDICES_BITSET_FILTER_CACHE_CLEAN_INTERVAL_SETTING = Setting.positiveTimeSetting(
"indices.cache.bitset.cleanup_interval",
TimeValue.timeValueSeconds(60),
Property.NodeScope
);

private final Cache<BitsetCacheKey, Value> cache;
private final Set<IndexReader.CacheKey> staleCacheKeys = ConcurrentCollections.newConcurrentSet();
private final Set<IndexReader.CacheKey> registeredKeys = ConcurrentCollections.newConcurrentSet();
private final BitsetCacheCleaner cacheCleaner;
/**
* Every key cached for a reader, so {@link #onClose(IndexReader.CacheKey)} can invalidate the
* exact keys when the reader closes. Keys are only removed on reader close: trimming on
* eviction would race with a concurrent reload of the same key re-registering it, and a lost
* registration would leave the entry with no cleanup path. A registered key whose entry was
* evicted costs two references, is bounded by the reader's lifetime, and invalidating it on
* close is a no-op.
*/
private final ConcurrentMap<IndexReader.CacheKey, Set<BitsetCacheKey>> keysByReader = ConcurrentCollections.newConcurrentMap();

public IndicesBitsetFilterCache(Settings settings, ThreadPool threadPool) {
public IndicesBitsetFilterCache(Settings settings) {
long sizeInBytes = INDICES_BITSET_FILTER_CACHE_SIZE_SETTING.get(settings).getBytes();
CacheBuilder<BitsetCacheKey, Value> cacheBuilder = CacheBuilder.<BitsetCacheKey, Value>builder().removalListener(this);
if (sizeInBytes > 0) {
cacheBuilder.setMaximumWeight(sizeInBytes).weigher(new BitsetWeigher());
}
this.cache = cacheBuilder.build();

TimeValue cleanInterval = INDICES_BITSET_FILTER_CACHE_CLEAN_INTERVAL_SETTING.get(settings);
this.cacheCleaner = new BitsetCacheCleaner(this, threadPool, cleanInterval);
threadPool.schedule(cacheCleaner, cleanInterval, ThreadPool.Names.SAME);
}

public BitSetProducer getBitSetProducer(Query query, BitsetFilterCache.Listener listener) {
Expand Down Expand Up @@ -143,11 +134,14 @@ BitSet getAndLoadIfNotPresent(final Query query, final LeafReaderContext context
final IndexReader.CacheKey coreCacheReader = cacheHelper.getKey();
final ShardId shardId = ShardUtils.extractShardId(context.reader());

if (registeredKeys.add(coreCacheReader)) {
cacheHelper.addClosedListener(this);
}

final BitsetCacheKey cacheKey = new BitsetCacheKey(coreCacheReader, query);
// Register the closed listener at most once per reader (the mapping function runs
// atomically) and record the key so onClose can invalidate it exactly. The caller holds
// the reader open, so onClose cannot fire before the key is recorded.
keysByReader.computeIfAbsent(coreCacheReader, k -> {
cacheHelper.addClosedListener(this);
return ConcurrentCollections.newConcurrentSet();
}).add(cacheKey);
return cache.computeIfAbsent(cacheKey, key -> {
final BitSet bitSet = bitsetFromQuery(query, context);
Value value = new Value(bitSet, shardId, listener);
Expand All @@ -158,19 +152,26 @@ BitSet getAndLoadIfNotPresent(final Query query, final LeafReaderContext context

@Override
public void onClose(IndexReader.CacheKey ownerCoreCacheKey) {
staleCacheKeys.add(ownerCoreCacheKey);
// Invalidate the exact keys synchronously rather than deferring to a periodic cache
// sweep: a reader closes only once, and exact-key invalidation is O(#queries cached for
// the reader), so there is no need to batch it. A key whose entry was already evicted is
// a no-op to invalidate.
Set<BitsetCacheKey> keys = keysByReader.remove(ownerCoreCacheKey);
if (keys != null) {
for (BitsetCacheKey key : keys) {
cache.invalidate(key);
}
}
Comment thread
jainankitk marked this conversation as resolved.
Resolved
}

@Override
public void close() {
cacheCleaner.close();
clear();
}

public void clear() {
cache.invalidateAll();
staleCacheKeys.clear();
registeredKeys.clear();
keysByReader.clear();
}

@Override
Expand All @@ -182,22 +183,6 @@ public void onRemoval(RemovalNotification<BitsetCacheKey, Value> notification) {
value.listener.onRemoval(value.shardId, value.bitset);
}

public void purgeStaleEntries() {
if (staleCacheKeys.isEmpty()) {
return;
}
Set<IndexReader.CacheKey> staleSnapshot = new HashSet<>(staleCacheKeys);

for (BitsetCacheKey key : cache.keys()) {
if (staleSnapshot.contains(key.readerCacheKey)) {
cache.invalidate(key);
}
}

staleCacheKeys.removeAll(staleSnapshot);
registeredKeys.removeAll(staleSnapshot);
}

public Cache<BitsetCacheKey, Value> getCache() {
return cache;
}
Expand Down Expand Up @@ -373,33 +358,4 @@ public void onRemoval(ShardId shardId, Accountable accountable) {
}
}

private static final class BitsetCacheCleaner implements Runnable, Releasable {
private final IndicesBitsetFilterCache cache;
private final ThreadPool threadPool;
private final TimeValue interval;
private final AtomicBoolean closed = new AtomicBoolean(false);

BitsetCacheCleaner(IndicesBitsetFilterCache cache, ThreadPool threadPool, TimeValue interval) {
this.cache = cache;
this.threadPool = threadPool;
this.interval = interval;
}

@Override
public void run() {
try {
cache.purgeStaleEntries();
} catch (Exception e) {
logger.warn("Exception during periodic bitset filter cache cleanup:", e);
}
if (closed.get() == false) {
threadPool.scheduleUnlessShuttingDown(interval, ThreadPool.Names.SAME, this);
}
}

@Override
public void close() {
closed.compareAndSet(false, true);
}
}
}
2 changes: 1 addition & 1 deletion server/src/main/java/org/opensearch/indices/IndicesService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -588,7 +588,7 @@ public void onRemoval(ShardId shardId, String fieldName, boolean wasEvicted, lon
}, clusterService, threadPool);
this.cleanInterval = INDICES_CACHE_CLEAN_INTERVAL_SETTING.get(settings);
this.cacheCleaner = new CacheCleaner(indicesFieldDataCache, logger, threadPool, this.cleanInterval);
this.indicesBitsetFilterCache = new IndicesBitsetFilterCache(settings, threadPool);
this.indicesBitsetFilterCache = new IndicesBitsetFilterCache(settings);
this.metaStateService = metaStateService;
this.engineFactoryProviders = engineFactoryProviders;

Expand Down
19 changes: 7 additions & 12 deletions server/src/test/java/org/opensearch/index/cache/bitset/BitSetFilterCacheTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,7 +121,7 @@ public void testInvalidateEntries() throws Exception {
DirectoryReader reader = DirectoryReader.open(writer);
reader = OpenSearchDirectoryReader.wrap(reader, new ShardId("test", "_na_", 0));

IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool);
IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY);
BitsetFilterCache cache = new BitsetFilterCache(INDEX_SETTINGS, indicesCache, new BitsetFilterCache.Listener() {
@Override
public void onCache(ShardId shardId, Accountable accountable) {}
Expand All @@ -146,15 +146,11 @@ public void onRemoval(ShardId shardId, Accountable accountable) {}

// now cached
assertThat(matchCount(filter, reader), equalTo(3));
// Old 3 segments were closed (stale entries purged on next access via cleaner), new merged segment cached = 1
// Trigger purge explicitly since the scheduled cleaner may not have run yet.
indicesCache.purgeStaleEntries();
// Old 3 segments were closed (their entries invalidated on reader close), new merged segment cached = 1
assertThat(indicesCache.getCache().count(), equalTo(1));

reader.close();
writer.close();
// Trigger purge for the last closed reader.
indicesCache.purgeStaleEntries();
assertThat(indicesCache.getCache().count(), equalTo(0));

indicesCache.close();
Expand All @@ -176,7 +172,7 @@ public void testListener() throws IOException {
final AtomicInteger onCacheCalls = new AtomicInteger();
final AtomicInteger onRemoveCalls = new AtomicInteger();

IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool);
IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY);
BitsetFilterCache cache = new BitsetFilterCache(INDEX_SETTINGS, indicesCache, new BitsetFilterCache.Listener() {
@Override
public void onCache(ShardId shardId, Accountable accountable) {
Expand Down Expand Up @@ -210,7 +206,6 @@ public void onRemoval(ShardId shardId, Accountable accountable) {
assertEquals(1, onCacheCalls.get());
assertEquals(0, onRemoveCalls.get());
IOUtils.close(reader, writer);
indicesCache.purgeStaleEntries();
assertEquals(1, onRemoveCalls.get());
assertEquals(0, stats.get());

Expand All @@ -219,7 +214,7 @@ public void onRemoval(ShardId shardId, Accountable accountable) {

public void testSetNullListener() {
try {
new BitsetFilterCache(INDEX_SETTINGS, new IndicesBitsetFilterCache(Settings.EMPTY, threadPool), null);
new BitsetFilterCache(INDEX_SETTINGS, new IndicesBitsetFilterCache(Settings.EMPTY), null);
fail("listener can't be null");
} catch (IllegalArgumentException ex) {
assertEquals("listener must not be null", ex.getMessage());
Expand All @@ -246,7 +241,7 @@ public void onRemoval(ShardId shardId, Accountable accountable) {}
}

public void testCreateListenerWithIndicesCache() throws IOException {
IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool);
IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY);
BitsetFilterCache cache = new BitsetFilterCache(INDEX_SETTINGS, indicesCache, new BitsetFilterCache.Listener() {
@Override
public void onCache(ShardId shardId, Accountable accountable) {}
Expand Down Expand Up @@ -284,7 +279,7 @@ public void testBitsetFromQuery() throws IOException {
}

public void testNoOpDelegationMethods() throws IOException {
IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool);
IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY);
BitsetFilterCache cache = new BitsetFilterCache(INDEX_SETTINGS, indicesCache, new BitsetFilterCache.Listener() {
@Override
public void onCache(ShardId shardId, Accountable accountable) {}
Expand All @@ -303,7 +298,7 @@ public void onRemoval(ShardId shardId, Accountable accountable) {}
}

public void testRejectOtherIndex() throws IOException {
IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool);
IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY);
BitsetFilterCache cache = new BitsetFilterCache(INDEX_SETTINGS, indicesCache, new BitsetFilterCache.Listener() {
@Override
public void onCache(ShardId shardId, Accountable accountable) {}
Expand Down
Loading
Loading