diff --git a/modules/percolator/src/test/java/org/opensearch/percolator/PercolatorQuerySearchTests.java b/modules/percolator/src/test/java/org/opensearch/percolator/PercolatorQuerySearchTests.java index 5f4925a4ae577..74f346d677f14 100644 --- a/modules/percolator/src/test/java/org/opensearch/percolator/PercolatorQuerySearchTests.java +++ b/modules/percolator/src/test/java/org/opensearch/percolator/PercolatorQuerySearchTests.java @@ -41,13 +41,13 @@ import org.opensearch.core.xcontent.MediaTypeRegistry; import org.opensearch.core.xcontent.XContentBuilder; import org.opensearch.index.IndexService; -import org.opensearch.index.cache.bitset.BitsetFilterCache; import org.opensearch.index.engine.Engine; import org.opensearch.index.fielddata.ScriptDocValues; import org.opensearch.index.query.Operator; import org.opensearch.index.query.QueryBuilder; import org.opensearch.index.query.QueryBuilders; import org.opensearch.index.query.QueryShardContext; +import org.opensearch.indices.IndicesBitsetFilterCache; import org.opensearch.plugins.Plugin; import org.opensearch.script.MockScriptPlugin; import org.opensearch.script.Script; @@ -149,7 +149,9 @@ public void testPercolateQueryWithNestedDocuments_doNotLeakBitsetCacheEntries() .indices() .prepareCreate("test") // to avoid normal document from being cached by BitsetFilterCache - .setSettings(Settings.builder().put(BitsetFilterCache.INDEX_LOAD_RANDOM_ACCESS_FILTERS_EAGERLY_SETTING.getKey(), false)) + .setSettings( + Settings.builder().put(IndicesBitsetFilterCache.INDEX_LOAD_RANDOM_ACCESS_FILTERS_EAGERLY_SETTING.getKey(), false) + ) .setMapping(mapping) ); client().prepareIndex("test") diff --git a/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java b/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java index a84b6fa411c5c..8ab506438b3a8 100644 --- a/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java +++ b/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java @@ -50,7 +50,6 @@ import org.opensearch.index.MergeSchedulerConfig; import org.opensearch.index.SearchSlowLog; import org.opensearch.index.TieredMergePolicyProvider; -import org.opensearch.index.cache.bitset.BitsetFilterCache; import org.opensearch.index.compositeindex.datacube.startree.StarTreeIndexSettings; import org.opensearch.index.engine.EngineConfig; import org.opensearch.index.fielddata.IndexFieldDataService; @@ -59,6 +58,7 @@ import org.opensearch.index.similarity.SimilarityService; import org.opensearch.index.store.FsDirectoryFactory; import org.opensearch.index.store.Store; +import org.opensearch.indices.IndicesBitsetFilterCache; import org.opensearch.indices.IndicesRequestCache; import org.opensearch.search.streaming.FlushModeResolver; @@ -206,7 +206,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings { MapperService.INDEX_MAPPING_TOTAL_FIELDS_LIMIT_SETTING, MapperService.INDEX_MAPPING_DEPTH_LIMIT_SETTING, MapperService.INDEX_MAPPING_FIELD_NAME_LENGTH_LIMIT_SETTING, - BitsetFilterCache.INDEX_LOAD_RANDOM_ACCESS_FILTERS_EAGERLY_SETTING, + IndicesBitsetFilterCache.INDEX_LOAD_RANDOM_ACCESS_FILTERS_EAGERLY_SETTING, IndexModule.INDEX_STORE_TYPE_SETTING, IndexModule.INDEX_COMPOSITE_STORE_TYPE_SETTING, IndexModule.INDEX_STORE_FACTORY_SETTING, diff --git a/server/src/main/java/org/opensearch/index/IndexModule.java b/server/src/main/java/org/opensearch/index/IndexModule.java index 47ce26abc948c..56bd6e22884a7 100644 --- a/server/src/main/java/org/opensearch/index/IndexModule.java +++ b/server/src/main/java/org/opensearch/index/IndexModule.java @@ -94,6 +94,7 @@ import org.opensearch.index.store.remote.filecache.FileCache; import org.opensearch.index.translog.TranslogFactory; import org.opensearch.indices.ClusterMergeSchedulerConfig; +import org.opensearch.indices.IndicesBitsetFilterCache; import org.opensearch.indices.IndicesQueryCache; import org.opensearch.indices.RemoteStoreSettings; import org.opensearch.indices.fielddata.cache.IndicesFieldDataCache; @@ -824,6 +825,79 @@ public IndexService newIndexService( indicesQueryCache, mapperRegistry, indicesFieldDataCache, + null, + namedWriteableRegistry, + idFieldDataEnabled, + valuesSourceRegistry, + remoteDirectoryFactory, + translogFactorySupplier, + clusterDefaultRefreshIntervalSupplier, + fixedRefreshIntervalSchedulingEnabled, + shardLevelRefreshEnabled, + recoverySettings, + remoteStoreSettings, + replicator, + segmentReplicationStatsProvider, + clusterDefaultMaxMergeAtOnceSupplier, + clusterMergeSchedulerConfig, + (DataFormatRegistry) null + ); + } + + /** + * @deprecated Use the overload that accepts {@code indicesBitsetFilterCache} and {@code dataFormatRegistry} parameters. + */ + @Deprecated(forRemoval = true) + public IndexService newIndexService( + IndexService.IndexCreationContext indexCreationContext, + NodeEnvironment environment, + NamedXContentRegistry xContentRegistry, + IndexService.ShardStoreDeleter shardStoreDeleter, + CircuitBreakerService circuitBreakerService, + BigArrays bigArrays, + ThreadPool threadPool, + ScriptService scriptService, + ClusterService clusterService, + Client client, + IndicesQueryCache indicesQueryCache, + MapperRegistry mapperRegistry, + IndicesFieldDataCache indicesFieldDataCache, + NamedWriteableRegistry namedWriteableRegistry, + BooleanSupplier idFieldDataEnabled, + ValuesSourceRegistry valuesSourceRegistry, + IndexStorePlugin.DirectoryFactory remoteDirectoryFactory, + BiFunction translogFactorySupplier, + Supplier clusterDefaultRefreshIntervalSupplier, + Supplier fixedRefreshIntervalSchedulingEnabled, + Supplier shardLevelRefreshEnabled, + RecoverySettings recoverySettings, + RemoteStoreSettings remoteStoreSettings, + Consumer replicator, + Function segmentReplicationStatsProvider, + Supplier clusterDefaultMaxMergeAtOnceSupplier, + ClusterMergeSchedulerConfig clusterMergeSchedulerConfig, + CheckedTriFunction< + ShardPath, + MapperService, + IndexSettings, + DataFormatAwareEngineFactory, + IOException> dataFormatAwareEngineFactorySupplier + ) throws IOException { + return newIndexService( + indexCreationContext, + environment, + xContentRegistry, + shardStoreDeleter, + circuitBreakerService, + bigArrays, + threadPool, + scriptService, + clusterService, + client, + indicesQueryCache, + mapperRegistry, + indicesFieldDataCache, + null, namedWriteableRegistry, idFieldDataEnabled, valuesSourceRegistry, @@ -860,6 +934,7 @@ public IndexService newIndexService( IndicesQueryCache indicesQueryCache, MapperRegistry mapperRegistry, IndicesFieldDataCache indicesFieldDataCache, + IndicesBitsetFilterCache indicesBitsetFilterCache, NamedWriteableRegistry namedWriteableRegistry, BooleanSupplier idFieldDataEnabled, ValuesSourceRegistry valuesSourceRegistry, @@ -896,6 +971,7 @@ public IndexService newIndexService( indicesQueryCache, mapperRegistry, indicesFieldDataCache, + indicesBitsetFilterCache, namedWriteableRegistry, idFieldDataEnabled, valuesSourceRegistry, @@ -928,6 +1004,7 @@ public IndexService newIndexService( IndicesQueryCache indicesQueryCache, MapperRegistry mapperRegistry, IndicesFieldDataCache indicesFieldDataCache, + IndicesBitsetFilterCache indicesBitsetFilterCache, NamedWriteableRegistry namedWriteableRegistry, BooleanSupplier idFieldDataEnabled, ValuesSourceRegistry valuesSourceRegistry, @@ -1000,6 +1077,7 @@ public IndexService newIndexService( readerWrapperFactory, mapperRegistry, indicesFieldDataCache, + indicesBitsetFilterCache, searchOperationListeners, indexOperationListeners, namedWriteableRegistry, diff --git a/server/src/main/java/org/opensearch/index/IndexService.java b/server/src/main/java/org/opensearch/index/IndexService.java index 5e579227cbcdd..65bcfdcc565c5 100644 --- a/server/src/main/java/org/opensearch/index/IndexService.java +++ b/server/src/main/java/org/opensearch/index/IndexService.java @@ -106,6 +106,7 @@ import org.opensearch.index.translog.Translog; import org.opensearch.index.translog.TranslogFactory; import org.opensearch.indices.ClusterMergeSchedulerConfig; +import org.opensearch.indices.IndicesBitsetFilterCache; import org.opensearch.indices.RemoteStoreSettings; import org.opensearch.indices.cluster.IndicesClusterStateService; import org.opensearch.indices.fielddata.cache.IndicesFieldDataCache; @@ -243,6 +244,7 @@ public IndexService( Function> wrapperFactory, MapperRegistry mapperRegistry, IndicesFieldDataCache indicesFieldDataCache, + IndicesBitsetFilterCache indicesBitsetFilterCache, List searchOperationListeners, List indexingOperationListeners, NamedWriteableRegistry namedWriteableRegistry, @@ -310,8 +312,12 @@ public IndexService( this.indexSortSupplier = () -> null; } indexFieldData.setListener(new FieldDataCacheListener(this)); - this.bitsetFilterCache = new BitsetFilterCache(indexSettings, new BitsetCacheListener(this)); - this.warmer = new IndexWarmer(threadPool, indexFieldData, bitsetFilterCache.createListener(threadPool)); + this.bitsetFilterCache = new BitsetFilterCache(indexSettings, indicesBitsetFilterCache, new BitsetCacheListener(this)); + this.warmer = new IndexWarmer( + threadPool, + indexFieldData, + indicesBitsetFilterCache != null ? indicesBitsetFilterCache.createListener(threadPool) : null + ); this.indexCache = new IndexCache(indexSettings, queryCache, bitsetFilterCache); } else { assert indexAnalyzers == null; @@ -448,6 +454,7 @@ public IndexService( wrapperFactory, mapperRegistry, indicesFieldDataCache, + null, searchOperationListeners, indexingOperationListeners, namedWriteableRegistry, diff --git a/server/src/main/java/org/opensearch/index/cache/bitset/BitsetFilterCache.java b/server/src/main/java/org/opensearch/index/cache/bitset/BitsetFilterCache.java index 96a867c662137..c397e2d554617 100644 --- a/server/src/main/java/org/opensearch/index/cache/bitset/BitsetFilterCache.java +++ b/server/src/main/java/org/opensearch/index/cache/bitset/BitsetFilterCache.java @@ -32,58 +32,34 @@ package org.opensearch.index.cache.bitset; -import org.apache.logging.log4j.message.ParameterizedMessage; -import org.apache.lucene.index.FilterLeafReader; import org.apache.lucene.index.IndexReader; -import org.apache.lucene.index.IndexReaderContext; import org.apache.lucene.index.LeafReaderContext; -import org.apache.lucene.index.ReaderUtil; -import org.apache.lucene.search.IndexSearcher; import org.apache.lucene.search.Query; -import org.apache.lucene.search.ScoreMode; -import org.apache.lucene.search.Scorer; -import org.apache.lucene.search.Weight; import org.apache.lucene.search.join.BitSetProducer; import org.apache.lucene.util.Accountable; import org.apache.lucene.util.BitDocIdSet; import org.apache.lucene.util.BitSet; -import org.opensearch.ExceptionsHelper; import org.opensearch.common.annotation.PublicApi; import org.opensearch.common.cache.Cache; -import org.opensearch.common.cache.CacheBuilder; import org.opensearch.common.cache.RemovalListener; import org.opensearch.common.cache.RemovalNotification; -import org.opensearch.common.lucene.index.OpenSearchDirectoryReader; -import org.opensearch.common.lucene.search.Queries; import org.opensearch.common.settings.Setting; -import org.opensearch.common.settings.Setting.Property; -import org.opensearch.common.unit.TimeValue; import org.opensearch.core.index.shard.ShardId; import org.opensearch.index.AbstractIndexComponent; import org.opensearch.index.IndexSettings; import org.opensearch.index.IndexWarmer; -import org.opensearch.index.IndexWarmer.TerminationHandle; -import org.opensearch.index.mapper.DocumentMapper; -import org.opensearch.index.mapper.MapperService; -import org.opensearch.index.mapper.ObjectMapper; -import org.opensearch.index.shard.IndexShard; -import org.opensearch.index.shard.ShardUtils; +import org.opensearch.indices.IndicesBitsetFilterCache; import org.opensearch.threadpool.ThreadPool; import java.io.Closeable; import java.io.IOException; -import java.util.HashSet; -import java.util.Objects; -import java.util.Set; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Executor; /** - * This is a cache for {@link BitDocIdSet} based filters and is unbounded by size or time. + * Per-index view into the node-level {@link IndicesBitsetFilterCache}. *

- * Use this cache with care, only components that require that a filter is to be materialized as a {@link BitDocIdSet} - * and require that it should always be around should use this cache, otherwise the + * This is a cache for {@link BitDocIdSet} based filters. Use this cache with care, only components + * that require a filter to be materialized as a {@link BitDocIdSet} and require that it should always + * be around should use this cache, otherwise the * {@link org.opensearch.index.cache.query.QueryCache} should be used instead. * * @opensearch.api @@ -95,103 +71,74 @@ public final class BitsetFilterCache extends AbstractIndexComponent RemovalListener>, Closeable { - public static final Setting INDEX_LOAD_RANDOM_ACCESS_FILTERS_EAGERLY_SETTING = Setting.boolSetting( - "index.load_fixed_bitset_filters_eagerly", - true, - Property.IndexScope - ); + /** + * @deprecated Use {@link IndicesBitsetFilterCache#INDEX_LOAD_RANDOM_ACCESS_FILTERS_EAGERLY_SETTING} instead. + */ + @Deprecated + public static final Setting INDEX_LOAD_RANDOM_ACCESS_FILTERS_EAGERLY_SETTING = + IndicesBitsetFilterCache.INDEX_LOAD_RANDOM_ACCESS_FILTERS_EAGERLY_SETTING; - private final boolean loadRandomAccessFiltersEagerly; - private final Cache> loadedFilters; + private final IndicesBitsetFilterCache indicesCache; private final Listener listener; + /** + * @deprecated Use {@link #BitsetFilterCache(IndexSettings, IndicesBitsetFilterCache, Listener)} instead. + */ + @Deprecated public BitsetFilterCache(IndexSettings indexSettings, Listener listener) { + this(indexSettings, null, listener); + } + + public BitsetFilterCache(IndexSettings indexSettings, IndicesBitsetFilterCache indicesCache, Listener listener) { super(indexSettings); if (listener == null) { throw new IllegalArgumentException("listener must not be null"); } - this.loadRandomAccessFiltersEagerly = this.indexSettings.getValue(INDEX_LOAD_RANDOM_ACCESS_FILTERS_EAGERLY_SETTING); - this.loadedFilters = CacheBuilder.>builder().removalListener(this).build(); + this.indicesCache = indicesCache; this.listener = listener; } public static BitSet bitsetFromQuery(Query query, LeafReaderContext context) throws IOException { - final IndexReaderContext topLevelContext = ReaderUtil.getTopLevelContext(context); - final IndexSearcher searcher = new IndexSearcher(topLevelContext); - searcher.setQueryCache(null); - final Weight weight = searcher.createWeight(searcher.rewrite(query), ScoreMode.COMPLETE_NO_SCORES, 1f); - Scorer s = weight.scorer(context); - if (s == null) { - return null; - } else { - return BitSet.of(s.iterator(), context.reader().maxDoc()); - } + return IndicesBitsetFilterCache.bitsetFromQuery(query, context); } + /** + * @deprecated The warmer is now created by {@link IndicesBitsetFilterCache#createListener(ThreadPool)}. + */ + @Deprecated public IndexWarmer.Listener createListener(ThreadPool threadPool) { - return new BitSetProducerWarmer(threadPool); + if (indicesCache != null) { + return indicesCache.createListener(threadPool); + } + return null; } public BitSetProducer getBitSetProducer(Query query) { - return new QueryWrapperBitSetProducer(query); + if (indicesCache != null) { + return indicesCache.getBitSetProducer(query, listener); + } + throw new IllegalStateException("IndicesBitsetFilterCache is not available"); } @Override public void onClose(IndexReader.CacheKey ownerCoreCacheKey) { - loadedFilters.invalidate(ownerCoreCacheKey); + // Delegated to node-level cache } @Override public void close() { - clear("close"); + // Per-index close is a no-op; entries are cleaned up by the node-level + // periodic stale-key purge after the index's readers close. } public void clear(String reason) { logger.debug("clearing all bitsets because [{}]", reason); - loadedFilters.invalidateAll(); - } - - private BitSet getAndLoadIfNotPresent(final Query query, final LeafReaderContext context) throws ExecutionException { - final IndexReader.CacheHelper cacheHelper = FilterLeafReader.unwrap(context.reader()).getCoreCacheHelper(); - if (cacheHelper == null) { - throw new IllegalArgumentException("Reader " + context.reader() + " does not support caching"); - } - final IndexReader.CacheKey coreCacheReader = cacheHelper.getKey(); - final ShardId shardId = ShardUtils.extractShardId(context.reader()); - if (indexSettings.getIndex().equals(shardId.getIndex()) == false) { - // insanity - throw new IllegalStateException( - "Trying to load bit set for index " + shardId.getIndex() + " with cache of index " + indexSettings.getIndex() - ); - } - Cache filterToFbs = loadedFilters.computeIfAbsent(coreCacheReader, key -> { - cacheHelper.addClosedListener(BitsetFilterCache.this); - return CacheBuilder.builder().build(); - }); - - return filterToFbs.computeIfAbsent(query, key -> { - final BitSet bitSet = bitsetFromQuery(query, context); - Value value = new Value(bitSet, shardId); - listener.onCache(shardId, value.bitset); - return value; - }).bitset; + // Per-index clear is a no-op; entries are evicted by the node-level cache. } @Override public void onRemoval(RemovalNotification> notification) { - if (notification.getKey() == null) { - return; - } - - Cache valueCache = notification.getValue(); - if (valueCache == null) { - return; - } - - for (Value value : valueCache.values()) { - listener.onRemoval(value.shardId, value.bitset); - // if null then this means the shard has already been removed and the stats are 0 anyway for the shard this key belongs to - } + // Delegated to node-level cache } /** @@ -211,122 +158,12 @@ public Value(BitSet bitset, ShardId shardId) { } } - final class QueryWrapperBitSetProducer implements BitSetProducer { - - final Query query; - - QueryWrapperBitSetProducer(Query query) { - this.query = Objects.requireNonNull(query); - } - - // TODO: convertToElastic might need to be renamed - @Override - public BitSet getBitSet(LeafReaderContext context) throws IOException { - try { - return getAndLoadIfNotPresent(query, context); - } catch (ExecutionException e) { - throw ExceptionsHelper.convertToOpenSearchException(e); - } - } - - @Override - public String toString() { - return "random_access(" + query + ")"; - } - - @Override - public boolean equals(Object o) { - if (!(o instanceof QueryWrapperBitSetProducer other)) return false; - return this.query.equals(other.query); - } - - @Override - public int hashCode() { - return 31 * getClass().hashCode() + query.hashCode(); - } - } - - final class BitSetProducerWarmer implements IndexWarmer.Listener { - - private final Executor executor; - - BitSetProducerWarmer(ThreadPool threadPool) { - this.executor = threadPool.executor(ThreadPool.Names.WARMER); - } - - @Override - public IndexWarmer.TerminationHandle warmReader(final IndexShard indexShard, final OpenSearchDirectoryReader reader) { - if (indexSettings.getIndex().equals(indexShard.indexSettings().getIndex()) == false) { - // this is from a different index - return TerminationHandle.NO_WAIT; - } - - if (!loadRandomAccessFiltersEagerly) { - return TerminationHandle.NO_WAIT; - } - - boolean hasNested = false; - final Set warmUp = new HashSet<>(); - final MapperService mapperService = indexShard.mapperService(); - DocumentMapper docMapper = mapperService.documentMapper(); - if (docMapper != null) { - if (docMapper.hasNestedObjects()) { - hasNested = true; - for (ObjectMapper objectMapper : docMapper.objectMappers().values()) { - if (objectMapper.nested().isNested()) { - ObjectMapper parentObjectMapper = objectMapper.getParentObjectMapper(mapperService); - if (parentObjectMapper != null && parentObjectMapper.nested().isNested()) { - warmUp.add(parentObjectMapper.nestedTypeFilter()); - } - } - } - } - } - - if (hasNested) { - warmUp.add(Queries.newNonNestedFilter()); - } - - final CountDownLatch latch = new CountDownLatch(reader.leaves().size() * warmUp.size()); - for (final LeafReaderContext ctx : reader.leaves()) { - for (final Query filterToWarm : warmUp) { - executor.execute(() -> { - try { - final long start = System.nanoTime(); - getAndLoadIfNotPresent(filterToWarm, ctx); - if (indexShard.warmerService().logger().isTraceEnabled()) { - indexShard.warmerService() - .logger() - .trace( - "warmed bitset for [{}], took [{}]", - filterToWarm, - TimeValue.timeValueNanos(System.nanoTime() - start) - ); - } - } catch (Exception e) { - indexShard.warmerService() - .logger() - .warn(() -> new ParameterizedMessage("failed to load " + "bitset for [{}]", filterToWarm), e); - } finally { - latch.countDown(); - } - }); - } - } - return () -> latch.await(); - } - - } - - Cache> getLoadedFilters() { - return loadedFilters; - } - /** - * A listener interface that is executed for each onCache / onRemoval event + * A listener interface that is executed for each onCache / onRemoval event * * @opensearch.internal */ + @PublicApi(since = "1.0.0") public interface Listener { /** * Called for each cached bitset on the cache event. diff --git a/server/src/main/java/org/opensearch/indices/IndicesBitsetFilterCache.java b/server/src/main/java/org/opensearch/indices/IndicesBitsetFilterCache.java new file mode 100644 index 0000000000000..0d6830ab75060 --- /dev/null +++ b/server/src/main/java/org/opensearch/indices/IndicesBitsetFilterCache.java @@ -0,0 +1,405 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +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; +import org.apache.lucene.index.IndexReaderContext; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.ReaderUtil; +import org.apache.lucene.search.IndexSearcher; +import org.apache.lucene.search.Query; +import org.apache.lucene.search.ScoreMode; +import org.apache.lucene.search.Scorer; +import org.apache.lucene.search.Weight; +import org.apache.lucene.search.join.BitSetProducer; +import org.apache.lucene.util.Accountable; +import org.apache.lucene.util.BitSet; +import org.opensearch.ExceptionsHelper; +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.common.cache.Cache; +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; +import org.opensearch.common.settings.Setting.Property; +import org.opensearch.common.settings.Settings; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.common.util.concurrent.ConcurrentCollections; +import org.opensearch.core.common.unit.ByteSizeValue; +import org.opensearch.core.index.shard.ShardId; +import org.opensearch.index.IndexWarmer; +import org.opensearch.index.IndexWarmer.TerminationHandle; +import org.opensearch.index.cache.bitset.BitsetFilterCache; +import org.opensearch.index.mapper.DocumentMapper; +import org.opensearch.index.mapper.MapperService; +import org.opensearch.index.mapper.ObjectMapper; +import org.opensearch.index.shard.IndexShard; +import org.opensearch.index.shard.ShardUtils; +import org.opensearch.threadpool.ThreadPool; + +import java.io.Closeable; +import java.io.IOException; +import java.util.HashSet; +import java.util.Objects; +import java.util.Set; +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. + * + * @opensearch.api + */ +@ExperimentalApi +public class IndicesBitsetFilterCache + implements + IndexReader.ClosedListener, + RemovalListener, + Closeable { + + private static final Logger logger = LogManager.getLogger(IndicesBitsetFilterCache.class); + + public static final Setting INDEX_LOAD_RANDOM_ACCESS_FILTERS_EAGERLY_SETTING = Setting.boolSetting( + "index.load_fixed_bitset_filters_eagerly", + true, + Property.IndexScope + ); + + public static final Setting INDICES_BITSET_FILTER_CACHE_SIZE_SETTING = Setting.memorySizeSetting( + "indices.cache.bitset.size", + "5%", + Property.NodeScope + ); + + public static final Setting INDICES_BITSET_FILTER_CACHE_CLEAN_INTERVAL_SETTING = Setting.positiveTimeSetting( + "indices.cache.bitset.cleanup_interval", + TimeValue.timeValueSeconds(60), + Property.NodeScope + ); + + private final Cache cache; + private final Set staleCacheKeys = ConcurrentCollections.newConcurrentSet(); + private final Set registeredKeys = ConcurrentCollections.newConcurrentSet(); + private final BitsetCacheCleaner cacheCleaner; + + public IndicesBitsetFilterCache(Settings settings, ThreadPool threadPool) { + long sizeInBytes = INDICES_BITSET_FILTER_CACHE_SIZE_SETTING.get(settings).getBytes(); + CacheBuilder cacheBuilder = CacheBuilder.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) { + return new QueryWrapperBitSetProducer(query, listener); + } + + public IndexWarmer.Listener createListener(ThreadPool threadPool) { + return new BitSetProducerWarmer(threadPool); + } + + public static BitSet bitsetFromQuery(Query query, LeafReaderContext context) throws IOException { + final IndexReaderContext topLevelContext = ReaderUtil.getTopLevelContext(context); + final IndexSearcher searcher = new IndexSearcher(topLevelContext); + searcher.setQueryCache(null); + final Weight weight = searcher.createWeight(searcher.rewrite(query), ScoreMode.COMPLETE_NO_SCORES, 1f); + Scorer s = weight.scorer(context); + if (s == null) { + return null; + } else { + return BitSet.of(s.iterator(), context.reader().maxDoc()); + } + } + + BitSet getAndLoadIfNotPresent(final Query query, final LeafReaderContext context, final BitsetFilterCache.Listener listener) + throws ExecutionException { + final IndexReader.CacheHelper cacheHelper = FilterLeafReader.unwrap(context.reader()).getCoreCacheHelper(); + if (cacheHelper == null) { + throw new IllegalArgumentException("Reader " + context.reader() + " does not support caching"); + } + 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); + return cache.computeIfAbsent(cacheKey, key -> { + final BitSet bitSet = bitsetFromQuery(query, context); + Value value = new Value(bitSet, shardId, listener); + listener.onCache(shardId, value.bitset); + return value; + }).bitset; + } + + @Override + public void onClose(IndexReader.CacheKey ownerCoreCacheKey) { + staleCacheKeys.add(ownerCoreCacheKey); + } + + @Override + public void close() { + cacheCleaner.close(); + clear(); + } + + public void clear() { + cache.invalidateAll(); + staleCacheKeys.clear(); + registeredKeys.clear(); + } + + @Override + public void onRemoval(RemovalNotification notification) { + Value value = notification.getValue(); + if (value == null || value.listener == null) { + return; + } + value.listener.onRemoval(value.shardId, value.bitset); + } + + public void purgeStaleEntries() { + if (staleCacheKeys.isEmpty()) { + return; + } + Set 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 getCache() { + return cache; + } + + /** + * Composite key combining a reader segment key with a query. + * + * @opensearch.internal + */ + @ExperimentalApi + public static final class BitsetCacheKey { + final IndexReader.CacheKey readerCacheKey; + final Query query; + + public BitsetCacheKey(IndexReader.CacheKey readerCacheKey, Query query) { + this.readerCacheKey = Objects.requireNonNull(readerCacheKey); + this.query = Objects.requireNonNull(query); + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (!(o instanceof BitsetCacheKey other)) return false; + return readerCacheKey == other.readerCacheKey && query.equals(other.query); + } + + @Override + public int hashCode() { + return 31 * System.identityHashCode(readerCacheKey) + query.hashCode(); + } + } + + /** + * Cached value holding the bitset, shard identity, and the per-index listener for stats. + * + * @opensearch.internal + */ + @ExperimentalApi + public static final class Value { + final BitSet bitset; + final ShardId shardId; + final BitsetFilterCache.Listener listener; + + Value(BitSet bitset, ShardId shardId, BitsetFilterCache.Listener listener) { + this.bitset = bitset; + this.shardId = shardId; + this.listener = listener; + } + } + + static class BitsetWeigher implements ToLongBiFunction { + @Override + public long applyAsLong(BitsetCacheKey key, Value value) { + long weight = (value.bitset != null) ? value.bitset.ramBytesUsed() : 0; + return weight == 0 ? 1 : weight; + } + } + + final class QueryWrapperBitSetProducer implements BitSetProducer { + final Query query; + final BitsetFilterCache.Listener listener; + + QueryWrapperBitSetProducer(Query query, BitsetFilterCache.Listener listener) { + this.query = Objects.requireNonNull(query); + this.listener = Objects.requireNonNull(listener); + } + + @Override + public BitSet getBitSet(LeafReaderContext context) throws IOException { + try { + return getAndLoadIfNotPresent(query, context, listener); + } catch (ExecutionException e) { + throw ExceptionsHelper.convertToOpenSearchException(e); + } + } + + @Override + public String toString() { + return "random_access(" + query + ")"; + } + + @Override + public boolean equals(Object o) { + if (!(o instanceof QueryWrapperBitSetProducer other)) return false; + return this.query.equals(other.query); + } + + @Override + public int hashCode() { + return 31 * getClass().hashCode() + query.hashCode(); + } + } + + final class BitSetProducerWarmer implements IndexWarmer.Listener { + private final Executor executor; + + BitSetProducerWarmer(ThreadPool threadPool) { + this.executor = threadPool.executor(ThreadPool.Names.WARMER); + } + + @Override + public IndexWarmer.TerminationHandle warmReader(final IndexShard indexShard, final OpenSearchDirectoryReader reader) { + if (!indexShard.indexSettings().getValue(INDEX_LOAD_RANDOM_ACCESS_FILTERS_EAGERLY_SETTING)) { + return TerminationHandle.NO_WAIT; + } + + boolean hasNested = false; + final Set warmUp = new HashSet<>(); + final MapperService mapperService = indexShard.mapperService(); + DocumentMapper docMapper = mapperService.documentMapper(); + if (docMapper != null) { + if (docMapper.hasNestedObjects()) { + hasNested = true; + for (ObjectMapper objectMapper : docMapper.objectMappers().values()) { + if (objectMapper.nested().isNested()) { + ObjectMapper parentObjectMapper = objectMapper.getParentObjectMapper(mapperService); + if (parentObjectMapper != null && parentObjectMapper.nested().isNested()) { + warmUp.add(parentObjectMapper.nestedTypeFilter()); + } + } + } + } + } + + if (hasNested) { + warmUp.add(Queries.newNonNestedFilter()); + } + + // Build a listener that routes stats to the correct shard. + final BitsetFilterCache.Listener listener = new BitsetFilterCache.Listener() { + @Override + public void onCache(ShardId shardId, Accountable accountable) { + if (shardId != null && accountable != null) { + indexShard.shardBitsetFilterCache().onCached(accountable.ramBytesUsed()); + } + } + + @Override + public void onRemoval(ShardId shardId, Accountable accountable) { + if (shardId != null && accountable != null) { + indexShard.shardBitsetFilterCache().onRemoval(accountable.ramBytesUsed()); + } + } + }; + + final CountDownLatch latch = new CountDownLatch(reader.leaves().size() * warmUp.size()); + for (final LeafReaderContext ctx : reader.leaves()) { + for (final Query filterToWarm : warmUp) { + executor.execute(() -> { + try { + final long start = System.nanoTime(); + getAndLoadIfNotPresent(filterToWarm, ctx, listener); + if (indexShard.warmerService().logger().isTraceEnabled()) { + indexShard.warmerService() + .logger() + .trace( + "warmed bitset for [{}], took [{}]", + filterToWarm, + TimeValue.timeValueNanos(System.nanoTime() - start) + ); + } + } catch (Exception e) { + indexShard.warmerService() + .logger() + .warn(() -> new ParameterizedMessage("failed to load bitset for [{}]", filterToWarm), e); + } finally { + latch.countDown(); + } + }); + } + } + return () -> latch.await(); + } + } + + 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); + } + } +} diff --git a/server/src/main/java/org/opensearch/indices/IndicesService.java b/server/src/main/java/org/opensearch/indices/IndicesService.java index 20516116f072a..9bfb4d2e295d5 100644 --- a/server/src/main/java/org/opensearch/indices/IndicesService.java +++ b/server/src/main/java/org/opensearch/indices/IndicesService.java @@ -378,6 +378,7 @@ public class IndicesService extends AbstractLifecycleComponent private final IndexNameExpressionResolver indexNameExpressionResolver; private final IndexScopedSettings indexScopedSettings; private final IndicesFieldDataCache indicesFieldDataCache; + private final IndicesBitsetFilterCache indicesBitsetFilterCache; private final CacheCleaner cacheCleaner; private final ThreadPool threadPool; private final CircuitBreakerService circuitBreakerService; @@ -522,6 +523,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.metaStateService = metaStateService; this.engineFactoryProviders = engineFactoryProviders; @@ -1010,6 +1012,7 @@ public void onStoreClosed(ShardId shardId) { indexMetadata, indicesQueryCache, indicesFieldDataCache, + indicesBitsetFilterCache, finalListeners, indexingMemoryController ); @@ -1065,6 +1068,7 @@ public void onStoreCreated(ShardId shardId) { indexMetadata, indicesQueryCache, indicesFieldDataCache, + indicesBitsetFilterCache, finalListeners, indexingMemoryController ); @@ -1081,6 +1085,7 @@ private synchronized IndexService createIndexService( IndexMetadata indexMetadata, IndicesQueryCache indicesQueryCache, IndicesFieldDataCache indicesFieldDataCache, + IndicesBitsetFilterCache indicesBitsetFilterCache, List builtInListeners, IndexingOperationListener... indexingOperationListeners ) throws IOException { @@ -1137,6 +1142,7 @@ private synchronized IndexService createIndexService( indicesQueryCache, mapperRegistry, indicesFieldDataCache, + indicesBitsetFilterCache, namedWriteableRegistry, this::isIdFieldDataEnabled, valuesSourceRegistry, @@ -1267,6 +1273,7 @@ public synchronized void verifyIndexMetadata(IndexMetadata metadata, IndexMetada metadata, indicesQueryCache, indicesFieldDataCache, + indicesBitsetFilterCache, emptyList() ); closeables.add(() -> service.close("metadata verification", false)); diff --git a/server/src/test/java/org/opensearch/index/IndexModuleTests.java b/server/src/test/java/org/opensearch/index/IndexModuleTests.java index 5ecabffe35c70..8b19e2fa475c7 100644 --- a/server/src/test/java/org/opensearch/index/IndexModuleTests.java +++ b/server/src/test/java/org/opensearch/index/IndexModuleTests.java @@ -270,6 +270,7 @@ private IndexService newIndexService(IndexModule module) throws IOException { indicesQueryCache, mapperRegistry, new IndicesFieldDataCache(settings, listener, clusterService, threadPool), + null, writableRegistry(), () -> false, null, diff --git a/server/src/test/java/org/opensearch/index/cache/bitset/BitSetFilterCacheTests.java b/server/src/test/java/org/opensearch/index/cache/bitset/BitSetFilterCacheTests.java index f3cac6abd6ced..9ab8f3d1c4702 100644 --- a/server/src/test/java/org/opensearch/index/cache/bitset/BitSetFilterCacheTests.java +++ b/server/src/test/java/org/opensearch/index/cache/bitset/BitSetFilterCacheTests.java @@ -55,18 +55,37 @@ import org.opensearch.common.util.io.IOUtils; import org.opensearch.core.index.shard.ShardId; import org.opensearch.index.IndexSettings; +import org.opensearch.indices.IndicesBitsetFilterCache; import org.opensearch.test.IndexSettingsModule; import org.opensearch.test.OpenSearchTestCase; +import org.opensearch.threadpool.TestThreadPool; +import org.opensearch.threadpool.ThreadPool; import java.io.IOException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.notNullValue; +import static org.hamcrest.Matchers.nullValue; public class BitSetFilterCacheTests extends OpenSearchTestCase { private static final IndexSettings INDEX_SETTINGS = IndexSettingsModule.newIndexSettings("test", Settings.EMPTY); + private ThreadPool threadPool; + + @Override + public void setUp() throws Exception { + super.setUp(); + threadPool = new TestThreadPool("bitset_filter_cache_test"); + } + + @Override + public void tearDown() throws Exception { + ThreadPool.terminate(threadPool, 10, TimeUnit.SECONDS); + super.tearDown(); + } private static int matchCount(BitSetProducer producer, IndexReader reader) throws IOException { int count = 0; @@ -102,24 +121,21 @@ public void testInvalidateEntries() throws Exception { DirectoryReader reader = DirectoryReader.open(writer); reader = OpenSearchDirectoryReader.wrap(reader, new ShardId("test", "_na_", 0)); - BitsetFilterCache cache = new BitsetFilterCache(INDEX_SETTINGS, new BitsetFilterCache.Listener() { + IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool); + BitsetFilterCache cache = new BitsetFilterCache(INDEX_SETTINGS, indicesCache, new BitsetFilterCache.Listener() { @Override - public void onCache(ShardId shardId, Accountable accountable) { - - } + public void onCache(ShardId shardId, Accountable accountable) {} @Override - public void onRemoval(ShardId shardId, Accountable accountable) { - - } + public void onRemoval(ShardId shardId, Accountable accountable) {} }); BitSetProducer filter = cache.getBitSetProducer(new TermQuery(new Term("field", "value"))); assertThat(matchCount(filter, reader), equalTo(3)); // now cached assertThat(matchCount(filter, reader), equalTo(3)); - // There are 3 segments - assertThat(cache.getLoadedFilters().weight(), equalTo(3L)); + // There are 3 segments, each with 1 query = 3 entries in the flat cache + assertThat(indicesCache.getCache().count(), equalTo(3)); writer.forceMerge(1); reader.close(); @@ -130,13 +146,18 @@ public void onRemoval(ShardId shardId, Accountable accountable) { // now cached assertThat(matchCount(filter, reader), equalTo(3)); - // Only one segment now, so the size must be 1 - assertThat(cache.getLoadedFilters().weight(), equalTo(1L)); + // 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(); + assertThat(indicesCache.getCache().count(), equalTo(1)); reader.close(); writer.close(); - // There is no reference from readers and writer to any segment in the test index, so the size in the fbs cache must be 0 - assertThat(cache.getLoadedFilters().weight(), equalTo(0L)); + // Trigger purge for the last closed reader. + indicesCache.purgeStaleEntries(); + assertThat(indicesCache.getCache().count(), equalTo(0)); + + indicesCache.close(); } public void testListener() throws IOException { @@ -155,7 +176,8 @@ public void testListener() throws IOException { final AtomicInteger onCacheCalls = new AtomicInteger(); final AtomicInteger onRemoveCalls = new AtomicInteger(); - final BitsetFilterCache cache = new BitsetFilterCache(INDEX_SETTINGS, new BitsetFilterCache.Listener() { + IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool); + BitsetFilterCache cache = new BitsetFilterCache(INDEX_SETTINGS, indicesCache, new BitsetFilterCache.Listener() { @Override public void onCache(ShardId shardId, Accountable accountable) { onCacheCalls.incrementAndGet(); @@ -188,31 +210,106 @@ 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()); + + indicesCache.close(); } public void testSetNullListener() { try { - new BitsetFilterCache(INDEX_SETTINGS, null); + new BitsetFilterCache(INDEX_SETTINGS, new IndicesBitsetFilterCache(Settings.EMPTY, threadPool), null); fail("listener can't be null"); } catch (IllegalArgumentException ex) { assertEquals("listener must not be null", ex.getMessage()); - // all is well } } - public void testRejectOtherIndex() throws IOException { + public void testDeprecatedConstructorAndCreateListener() throws IOException { + // The deprecated 2-arg constructor sets indicesCache to null. BitsetFilterCache cache = new BitsetFilterCache(INDEX_SETTINGS, new BitsetFilterCache.Listener() { @Override - public void onCache(ShardId shardId, Accountable accountable) { + public void onCache(ShardId shardId, Accountable accountable) {} - } + @Override + public void onRemoval(ShardId shardId, Accountable accountable) {} + }); + // createListener returns null when indicesCache is null + assertThat(cache.createListener(threadPool), nullValue()); + + // getBitSetProducer throws when indicesCache is null + expectThrows(IllegalStateException.class, () -> cache.getBitSetProducer(new MatchAllDocsQuery())); + + cache.close(); + } + + public void testCreateListenerWithIndicesCache() throws IOException { + IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool); + BitsetFilterCache cache = new BitsetFilterCache(INDEX_SETTINGS, indicesCache, new BitsetFilterCache.Listener() { @Override - public void onRemoval(ShardId shardId, Accountable accountable) { + public void onCache(ShardId shardId, Accountable accountable) {} - } + @Override + public void onRemoval(ShardId shardId, Accountable accountable) {} + }); + + // createListener returns non-null when indicesCache is present + assertThat(cache.createListener(threadPool), notNullValue()); + + cache.close(); + indicesCache.close(); + } + + public void testBitsetFromQuery() throws IOException { + Directory dir = newDirectory(); + IndexWriter writer = new IndexWriter(dir, newIndexWriterConfig()); + Document doc = new Document(); + doc.add(new StringField("field", "value", Field.Store.NO)); + writer.addDocument(doc); + writer.commit(); + DirectoryReader reader = DirectoryReader.open(writer); + + // Matching query returns non-null bitset + BitSet bitSet = BitsetFilterCache.bitsetFromQuery(new TermQuery(new Term("field", "value")), reader.leaves().get(0)); + assertNotNull(bitSet); + assertEquals(1, bitSet.cardinality()); + + // Non-matching query returns null bitset + BitSet emptyBitSet = BitsetFilterCache.bitsetFromQuery(new TermQuery(new Term("field", "missing")), reader.leaves().get(0)); + assertNull(emptyBitSet); + + IOUtils.close(reader, writer, dir); + } + + public void testNoOpDelegationMethods() throws IOException { + IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool); + BitsetFilterCache cache = new BitsetFilterCache(INDEX_SETTINGS, indicesCache, new BitsetFilterCache.Listener() { + @Override + public void onCache(ShardId shardId, Accountable accountable) {} + + @Override + public void onRemoval(ShardId shardId, Accountable accountable) {} + }); + + // These are all no-ops delegated to the node-level cache; just verify they don't throw. + cache.onClose(null); + cache.clear("test"); + cache.onRemoval(null); + cache.close(); + + indicesCache.close(); + } + + public void testRejectOtherIndex() throws IOException { + IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool); + BitsetFilterCache cache = new BitsetFilterCache(INDEX_SETTINGS, indicesCache, new BitsetFilterCache.Listener() { + @Override + public void onCache(ShardId shardId, Accountable accountable) {} + + @Override + public void onRemoval(ShardId shardId, Accountable accountable) {} }); Directory dir = newDirectory(); @@ -224,14 +321,17 @@ public void onRemoval(ShardId shardId, Accountable accountable) { BitSetProducer producer = cache.getBitSetProducer(new MatchAllDocsQuery()); + // The node-level cache doesn't validate index identity — it just caches. + // This test verifies the producer works for any index. try { - producer.getBitSet(reader.leaves().get(0)); - fail(); - } catch (IllegalStateException expected) { - assertEquals("Trying to load bit set for index [test2] with cache of index [test]", expected.getMessage()); + for (LeafReaderContext ctx : reader.leaves()) { + producer.getBitSet(ctx); + } } finally { IOUtils.close(reader, dir); } + + indicesCache.close(); } } diff --git a/server/src/test/java/org/opensearch/indices/IndicesBitsetFilterCacheTests.java b/server/src/test/java/org/opensearch/indices/IndicesBitsetFilterCacheTests.java new file mode 100644 index 0000000000000..b6e9cf224dde4 --- /dev/null +++ b/server/src/test/java/org/opensearch/indices/IndicesBitsetFilterCacheTests.java @@ -0,0 +1,370 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.indices; + +import org.apache.lucene.analysis.standard.StandardAnalyzer; +import org.apache.lucene.document.Document; +import org.apache.lucene.document.Field; +import org.apache.lucene.document.StringField; +import org.apache.lucene.index.DirectoryReader; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexWriterConfig; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.LogByteSizeMergePolicy; +import org.apache.lucene.index.Term; +import org.apache.lucene.search.TermQuery; +import org.apache.lucene.search.join.BitSetProducer; +import org.apache.lucene.store.ByteBuffersDirectory; +import org.apache.lucene.util.Accountable; +import org.opensearch.common.lucene.index.OpenSearchDirectoryReader; +import org.opensearch.common.settings.Settings; +import org.opensearch.core.index.shard.ShardId; +import org.opensearch.index.cache.bitset.BitsetFilterCache; +import org.opensearch.test.OpenSearchTestCase; +import org.opensearch.threadpool.TestThreadPool; +import org.opensearch.threadpool.ThreadPool; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.greaterThan; +import static org.hamcrest.Matchers.lessThanOrEqualTo; + +public class IndicesBitsetFilterCacheTests extends OpenSearchTestCase { + + private ThreadPool threadPool; + + @Override + public void setUp() throws Exception { + super.setUp(); + threadPool = new TestThreadPool("indices_bitset_filter_cache_test"); + } + + @Override + public void tearDown() throws Exception { + ThreadPool.terminate(threadPool, 10, TimeUnit.SECONDS); + super.tearDown(); + } + + private static final BitsetFilterCache.Listener NO_OP_LISTENER = new BitsetFilterCache.Listener() { + @Override + public void onCache(ShardId shardId, Accountable accountable) {} + + @Override + public void onRemoval(ShardId shardId, Accountable accountable) {} + }; + + /** + * Verifies that when the cache size setting is configured, the cache evicts entries + * once the total weight exceeds the configured limit. + */ + public void testCacheSizeLimitIsHonored() throws Exception { + // First, figure out how large a single bitset entry is by caching one. + long singleEntryBytes; + try (IndicesBitsetFilterCache probeCache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool)) { + IndexWriter writer = new IndexWriter( + new ByteBuffersDirectory(), + new IndexWriterConfig(new StandardAnalyzer()).setMergePolicy(new LogByteSizeMergePolicy()) + ); + Document doc = new Document(); + doc.add(new StringField("field", "value", Field.Store.NO)); + writer.addDocument(doc); + writer.commit(); + + DirectoryReader reader = DirectoryReader.open(writer); + reader = OpenSearchDirectoryReader.wrap(reader, new ShardId("probe", "_na_", 0)); + + BitSetProducer producer = probeCache.getBitSetProducer(new TermQuery(new Term("field", "value")), NO_OP_LISTENER); + producer.getBitSet(reader.leaves().get(0)); + + assertThat(probeCache.getCache().count(), equalTo(1)); + singleEntryBytes = probeCache.getCache().weight(); + assertThat(singleEntryBytes, greaterThan(0L)); + + reader.close(); + writer.close(); + } + + // Now create a cache with a size limit that fits exactly 2 entries. + long cacheSizeBytes = singleEntryBytes * 2; + Settings settings = Settings.builder().put("indices.cache.bitset.size", cacheSizeBytes + "b").build(); + + try (IndicesBitsetFilterCache cache = new IndicesBitsetFilterCache(settings, threadPool)) { + // Create an index with 3 separate segments (3 commits), each with one doc matching a different query. + IndexWriter writer = new IndexWriter( + new ByteBuffersDirectory(), + new IndexWriterConfig(new StandardAnalyzer()).setMergePolicy(new LogByteSizeMergePolicy()) + ); + + for (int i = 0; i < 3; i++) { + Document doc = new Document(); + doc.add(new StringField("field", "val" + i, Field.Store.NO)); + writer.addDocument(doc); + writer.commit(); + } + + DirectoryReader reader = DirectoryReader.open(writer); + reader = OpenSearchDirectoryReader.wrap(reader, new ShardId("test", "_na_", 0)); + + // 3 segments, cache 3 different queries — one per segment. + // Each segment has exactly 1 leaf. + assertThat(reader.leaves().size(), equalTo(3)); + + for (int i = 0; i < 3; i++) { + LeafReaderContext leaf = reader.leaves().get(i); + BitSetProducer producer = cache.getBitSetProducer(new TermQuery(new Term("field", "val" + i)), NO_OP_LISTENER); + producer.getBitSet(leaf); + } + + // We inserted 3 entries but the cache can only hold 2. + // The LRU eviction should have kicked in. + assertThat(cache.getCache().count(), lessThanOrEqualTo(2)); + assertThat(cache.getCache().weight(), lessThanOrEqualTo(cacheSizeBytes)); + + reader.close(); + writer.close(); + } + } + + /** + * Verifies that the onRemoval listener is called when entries are evicted due to size limit. + */ + public void testEvictionTriggersOnRemovalListener() throws Exception { + // Probe for single entry size. + long singleEntryBytes; + try (IndicesBitsetFilterCache probeCache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool)) { + IndexWriter writer = new IndexWriter( + new ByteBuffersDirectory(), + new IndexWriterConfig(new StandardAnalyzer()).setMergePolicy(new LogByteSizeMergePolicy()) + ); + Document doc = new Document(); + doc.add(new StringField("field", "value", Field.Store.NO)); + writer.addDocument(doc); + writer.commit(); + + DirectoryReader reader = DirectoryReader.open(writer); + reader = OpenSearchDirectoryReader.wrap(reader, new ShardId("probe", "_na_", 0)); + + probeCache.getBitSetProducer(new TermQuery(new Term("field", "value")), NO_OP_LISTENER).getBitSet(reader.leaves().get(0)); + singleEntryBytes = probeCache.getCache().weight(); + + reader.close(); + writer.close(); + } + + // Cache fits only 1 entry. + long cacheSizeBytes = singleEntryBytes; + Settings settings = Settings.builder().put("indices.cache.bitset.size", cacheSizeBytes + "b").build(); + + final AtomicLong removedBytes = new AtomicLong(); + BitsetFilterCache.Listener trackingListener = new BitsetFilterCache.Listener() { + @Override + public void onCache(ShardId shardId, Accountable accountable) {} + + @Override + public void onRemoval(ShardId shardId, Accountable accountable) { + if (accountable != null) { + removedBytes.addAndGet(accountable.ramBytesUsed()); + } + } + }; + + try (IndicesBitsetFilterCache cache = new IndicesBitsetFilterCache(settings, threadPool)) { + IndexWriter writer = new IndexWriter( + new ByteBuffersDirectory(), + new IndexWriterConfig(new StandardAnalyzer()).setMergePolicy(new LogByteSizeMergePolicy()) + ); + + for (int i = 0; i < 2; i++) { + Document doc = new Document(); + doc.add(new StringField("field", "val" + i, Field.Store.NO)); + writer.addDocument(doc); + writer.commit(); + } + + DirectoryReader reader = DirectoryReader.open(writer); + reader = OpenSearchDirectoryReader.wrap(reader, new ShardId("test", "_na_", 0)); + assertThat(reader.leaves().size(), equalTo(2)); + + // Cache first entry. + cache.getBitSetProducer(new TermQuery(new Term("field", "val0")), trackingListener).getBitSet(reader.leaves().get(0)); + assertThat(cache.getCache().count(), equalTo(1)); + assertThat(removedBytes.get(), equalTo(0L)); + + // Cache second entry — should evict the first since limit is 1 entry. + cache.getBitSetProducer(new TermQuery(new Term("field", "val1")), trackingListener).getBitSet(reader.leaves().get(1)); + assertThat(cache.getCache().count(), equalTo(1)); + assertThat(removedBytes.get(), greaterThan(0L)); + + reader.close(); + writer.close(); + } + } + + /** + * Verifies that stale entries from closed readers are purged. + */ + public void testStaleEntriesPurgedAfterReaderClose() throws Exception { + try (IndicesBitsetFilterCache cache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool)) { + IndexWriter writer = new IndexWriter( + new ByteBuffersDirectory(), + new IndexWriterConfig(new StandardAnalyzer()).setMergePolicy(new LogByteSizeMergePolicy()) + ); + + Document doc = new Document(); + doc.add(new StringField("field", "value", Field.Store.NO)); + writer.addDocument(doc); + writer.commit(); + + DirectoryReader reader = DirectoryReader.open(writer); + reader = OpenSearchDirectoryReader.wrap(reader, new ShardId("test", "_na_", 0)); + + cache.getBitSetProducer(new TermQuery(new Term("field", "value")), NO_OP_LISTENER).getBitSet(reader.leaves().get(0)); + assertThat(cache.getCache().count(), equalTo(1)); + + // Close writer first so no references remain, then close reader. + writer.close(); + reader.close(); + + // Purge stale entries. + cache.purgeStaleEntries(); + assertThat(cache.getCache().count(), equalTo(0)); + } + } + + /** + * Verifies that when one index's readers are closed (simulating index close), + * only that index's entries are purged while other indices' entries remain. + */ + public void testIndexCloseOnlyPurgesItsOwnEntries() throws Exception { + try (IndicesBitsetFilterCache cache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool)) { + // Create two separate "indices" with their own writers. + IndexWriter writer1 = new IndexWriter( + new ByteBuffersDirectory(), + new IndexWriterConfig(new StandardAnalyzer()).setMergePolicy(new LogByteSizeMergePolicy()) + ); + Document doc1 = new Document(); + doc1.add(new StringField("field", "val1", Field.Store.NO)); + writer1.addDocument(doc1); + writer1.commit(); + + IndexWriter writer2 = new IndexWriter( + new ByteBuffersDirectory(), + new IndexWriterConfig(new StandardAnalyzer()).setMergePolicy(new LogByteSizeMergePolicy()) + ); + Document doc2 = new Document(); + doc2.add(new StringField("field", "val2", Field.Store.NO)); + writer2.addDocument(doc2); + writer2.commit(); + + DirectoryReader reader1 = OpenSearchDirectoryReader.wrap(DirectoryReader.open(writer1), new ShardId("index1", "_na_", 0)); + DirectoryReader reader2 = OpenSearchDirectoryReader.wrap(DirectoryReader.open(writer2), new ShardId("index2", "_na_", 0)); + + // Cache one entry from each index. + cache.getBitSetProducer(new TermQuery(new Term("field", "val1")), NO_OP_LISTENER).getBitSet(reader1.leaves().get(0)); + cache.getBitSetProducer(new TermQuery(new Term("field", "val2")), NO_OP_LISTENER).getBitSet(reader2.leaves().get(0)); + assertThat(cache.getCache().count(), equalTo(2)); + + // Simulate index1 close: close its reader. + reader1.close(); + writer1.close(); + cache.purgeStaleEntries(); + + // Only index1's entry should be purged; index2's entry remains. + assertThat(cache.getCache().count(), equalTo(1)); + + reader2.close(); + writer2.close(); + cache.purgeStaleEntries(); + assertThat(cache.getCache().count(), equalTo(0)); + } + } + + /** + * Verifies that entries from multiple indices share the same cache and the size limit applies globally. + */ + public void testMultipleIndicesShareCacheWithGlobalSizeLimit() throws Exception { + // Probe for single entry size. + long singleEntryBytes; + try (IndicesBitsetFilterCache probeCache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool)) { + IndexWriter writer = new IndexWriter( + new ByteBuffersDirectory(), + new IndexWriterConfig(new StandardAnalyzer()).setMergePolicy(new LogByteSizeMergePolicy()) + ); + Document doc = new Document(); + doc.add(new StringField("field", "value", Field.Store.NO)); + writer.addDocument(doc); + writer.commit(); + + DirectoryReader reader = DirectoryReader.open(writer); + reader = OpenSearchDirectoryReader.wrap(reader, new ShardId("probe", "_na_", 0)); + + probeCache.getBitSetProducer(new TermQuery(new Term("field", "value")), NO_OP_LISTENER).getBitSet(reader.leaves().get(0)); + singleEntryBytes = probeCache.getCache().weight(); + + reader.close(); + writer.close(); + } + + // Cache fits 2 entries total across all indices. + long cacheSizeBytes = singleEntryBytes * 2; + Settings settings = Settings.builder().put("indices.cache.bitset.size", cacheSizeBytes + "b").build(); + + try (IndicesBitsetFilterCache cache = new IndicesBitsetFilterCache(settings, threadPool)) { + // Create two separate "indices" (different ShardIds, different writers). + IndexWriter writer1 = new IndexWriter( + new ByteBuffersDirectory(), + new IndexWriterConfig(new StandardAnalyzer()).setMergePolicy(new LogByteSizeMergePolicy()) + ); + Document doc1 = new Document(); + doc1.add(new StringField("field", "val1", Field.Store.NO)); + writer1.addDocument(doc1); + writer1.commit(); + + IndexWriter writer2 = new IndexWriter( + new ByteBuffersDirectory(), + new IndexWriterConfig(new StandardAnalyzer()).setMergePolicy(new LogByteSizeMergePolicy()) + ); + Document doc2 = new Document(); + doc2.add(new StringField("field", "val2", Field.Store.NO)); + writer2.addDocument(doc2); + writer2.commit(); + + IndexWriter writer3 = new IndexWriter( + new ByteBuffersDirectory(), + new IndexWriterConfig(new StandardAnalyzer()).setMergePolicy(new LogByteSizeMergePolicy()) + ); + Document doc3 = new Document(); + doc3.add(new StringField("field", "val3", Field.Store.NO)); + writer3.addDocument(doc3); + writer3.commit(); + + DirectoryReader reader1 = OpenSearchDirectoryReader.wrap(DirectoryReader.open(writer1), new ShardId("index1", "_na_", 0)); + DirectoryReader reader2 = OpenSearchDirectoryReader.wrap(DirectoryReader.open(writer2), new ShardId("index2", "_na_", 0)); + DirectoryReader reader3 = OpenSearchDirectoryReader.wrap(DirectoryReader.open(writer3), new ShardId("index3", "_na_", 0)); + + // Cache one entry from each "index". + cache.getBitSetProducer(new TermQuery(new Term("field", "val1")), NO_OP_LISTENER).getBitSet(reader1.leaves().get(0)); + cache.getBitSetProducer(new TermQuery(new Term("field", "val2")), NO_OP_LISTENER).getBitSet(reader2.leaves().get(0)); + cache.getBitSetProducer(new TermQuery(new Term("field", "val3")), NO_OP_LISTENER).getBitSet(reader3.leaves().get(0)); + + // 3 entries inserted but only 2 fit — global eviction should have kicked in. + assertThat(cache.getCache().count(), lessThanOrEqualTo(2)); + assertThat(cache.getCache().weight(), lessThanOrEqualTo(cacheSizeBytes)); + + reader1.close(); + reader2.close(); + reader3.close(); + writer1.close(); + writer2.close(); + writer3.close(); + } + } +} diff --git a/server/src/test/java/org/opensearch/search/aggregations/bucket/nested/NestedAggregatorTests.java b/server/src/test/java/org/opensearch/search/aggregations/bucket/nested/NestedAggregatorTests.java index c7fbca538c6ee..63fc3bbe58b26 100644 --- a/server/src/test/java/org/opensearch/search/aggregations/bucket/nested/NestedAggregatorTests.java +++ b/server/src/test/java/org/opensearch/search/aggregations/bucket/nested/NestedAggregatorTests.java @@ -49,6 +49,7 @@ import org.apache.lucene.search.DocIdSetIterator; import org.apache.lucene.search.MatchAllDocsQuery; import org.apache.lucene.search.TermQuery; +import org.apache.lucene.search.join.BitSetProducer; import org.apache.lucene.search.join.ScoreMode; import org.apache.lucene.store.Directory; import org.apache.lucene.tests.index.RandomIndexWriter; @@ -76,6 +77,7 @@ import org.opensearch.index.query.QueryShardContext; import org.opensearch.index.query.TermsQueryBuilder; import org.opensearch.index.query.support.NestedScope; +import org.opensearch.indices.IndicesBitsetFilterCache; import org.opensearch.script.MockScriptEngine; import org.opensearch.script.Script; import org.opensearch.script.ScriptEngine; @@ -104,6 +106,7 @@ import org.opensearch.search.aggregations.pipeline.InternalSimpleValue; import org.opensearch.search.aggregations.support.AggregationInspectionHelper; import org.opensearch.search.aggregations.support.ValueType; +import org.opensearch.threadpool.TestThreadPool; import java.io.IOException; import java.util.ArrayList; @@ -1132,8 +1135,17 @@ protected QueryShardContext createQueryShardContext(String fieldName, IndexSetti QueryShardContext queryShardContext = mock(QueryShardContext.class); when(queryShardContext.nestedScope()).thenReturn(new NestedScope(indexSettings)); - BitsetFilterCache bitsetFilterCache = new BitsetFilterCache(indexSettings, Mockito.mock(BitsetFilterCache.Listener.class)); - when(queryShardContext.bitsetFilter(any())).thenReturn(bitsetFilterCache.getBitSetProducer(Queries.newNonNestedFilter())); + if (aggTestThreadPool == null) { + aggTestThreadPool = new TestThreadPool("nested_agg_test"); + aggTestIndicesBitsetFilterCache = new IndicesBitsetFilterCache(Settings.EMPTY, aggTestThreadPool); + } + BitsetFilterCache bitsetFilterCache = new BitsetFilterCache( + indexSettings, + aggTestIndicesBitsetFilterCache, + Mockito.mock(BitsetFilterCache.Listener.class) + ); + BitSetProducer nonNestedFilter = bitsetFilterCache.getBitSetProducer(Queries.newNonNestedFilter()); + when(queryShardContext.bitsetFilter(any())).thenReturn(nonNestedFilter); when(queryShardContext.fieldMapper(anyString())).thenReturn(fieldType); when(queryShardContext.getSearchQuoteAnalyzer(any())).thenCallRealMethod(); when(queryShardContext.getSearchAnalyzer(any())).thenCallRealMethod(); diff --git a/server/src/test/java/org/opensearch/search/internal/ContextIndexSearcherTests.java b/server/src/test/java/org/opensearch/search/internal/ContextIndexSearcherTests.java index e806fe764c0a2..6ea54e619c277 100644 --- a/server/src/test/java/org/opensearch/search/internal/ContextIndexSearcherTests.java +++ b/server/src/test/java/org/opensearch/search/internal/ContextIndexSearcherTests.java @@ -81,6 +81,7 @@ import org.opensearch.index.cache.bitset.BitsetFilterCache; import org.opensearch.index.shard.IndexShard; import org.opensearch.index.shard.SearchOperationListener; +import org.opensearch.indices.IndicesBitsetFilterCache; import org.opensearch.lucene.util.CombinedBitSet; import org.opensearch.search.aggregations.InternalAggregation; import org.opensearch.search.aggregations.LeafBucketCollector; @@ -90,6 +91,8 @@ import org.opensearch.search.query.QuerySearchResult; import org.opensearch.test.IndexSettingsModule; import org.opensearch.test.OpenSearchTestCase; +import org.opensearch.threadpool.TestThreadPool; +import org.opensearch.threadpool.ThreadPool; import java.io.IOException; import java.io.UncheckedIOException; @@ -256,7 +259,9 @@ public void onRemoval(ShardId shardId, Accountable accountable) { } }; DirectoryReader reader = OpenSearchDirectoryReader.wrap(DirectoryReader.open(w), new ShardId(settings.getIndex(), 0)); - BitsetFilterCache cache = new BitsetFilterCache(settings, listener); + ThreadPool tp = new TestThreadPool("test"); + IndicesBitsetFilterCache indicesCache = new IndicesBitsetFilterCache(Settings.EMPTY, tp); + BitsetFilterCache cache = new BitsetFilterCache(settings, indicesCache, listener); Query roleQuery = new TermQuery(new Term("allowed", "yes")); BitSet bitSet = cache.getBitSetProducer(roleQuery).getBitSet(reader.leaves().get(0)); if (sparse) { @@ -315,7 +320,8 @@ public void onRemoval(ShardId shardId, Accountable accountable) { assertEquals(1, topDocs.scoreDocs.length); assertEquals(3f, topDocs.scoreDocs[0].score, 0); - IOUtils.close(reader, w, dir); + IOUtils.close(reader, w, dir, indicesCache); + ThreadPool.terminate(tp, 10, java.util.concurrent.TimeUnit.SECONDS); } public void testSlicesWithMaxTargetSliceSupplier() throws Exception { diff --git a/server/src/test/java/org/opensearch/search/sort/AbstractSortTestCase.java b/server/src/test/java/org/opensearch/search/sort/AbstractSortTestCase.java index eb50793e939c8..c7df3b038fe41 100644 --- a/server/src/test/java/org/opensearch/search/sort/AbstractSortTestCase.java +++ b/server/src/test/java/org/opensearch/search/sort/AbstractSortTestCase.java @@ -64,6 +64,7 @@ import org.opensearch.index.query.QueryShardContext; import org.opensearch.index.query.Rewriteable; import org.opensearch.index.query.TermQueryBuilder; +import org.opensearch.indices.IndicesBitsetFilterCache; import org.opensearch.script.MockScriptEngine; import org.opensearch.script.ScriptEngine; import org.opensearch.script.ScriptModule; @@ -208,7 +209,11 @@ protected final QueryShardContext createMockShardContext(IndexSearcher searcher) index, Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT).build() ); - BitsetFilterCache bitsetFilterCache = new BitsetFilterCache(idxSettings, Mockito.mock(BitsetFilterCache.Listener.class)); + BitsetFilterCache bitsetFilterCache = new BitsetFilterCache( + idxSettings, + Mockito.mock(IndicesBitsetFilterCache.class, Mockito.RETURNS_DEEP_STUBS), + Mockito.mock(BitsetFilterCache.Listener.class) + ); TriFunction, IndexFieldData> indexFieldDataLookup = ( fieldType, fieldIndexName, diff --git a/test/framework/src/main/java/org/opensearch/search/aggregations/AggregatorTestCase.java b/test/framework/src/main/java/org/opensearch/search/aggregations/AggregatorTestCase.java index f2eabf1b8c453..14affb7366ac0 100644 --- a/test/framework/src/main/java/org/opensearch/search/aggregations/AggregatorTestCase.java +++ b/test/framework/src/main/java/org/opensearch/search/aggregations/AggregatorTestCase.java @@ -90,7 +90,6 @@ import org.opensearch.index.analysis.IndexAnalyzers; import org.opensearch.index.analysis.NamedAnalyzer; import org.opensearch.index.cache.bitset.BitsetFilterCache; -import org.opensearch.index.cache.bitset.BitsetFilterCache.Listener; import org.opensearch.index.cache.query.DisabledQueryCache; import org.opensearch.index.codec.composite.CompositeIndexFieldInfo; import org.opensearch.index.compositeindex.datacube.Dimension; @@ -128,6 +127,7 @@ import org.opensearch.index.query.QueryShardContext; import org.opensearch.index.shard.IndexShard; import org.opensearch.index.shard.SearchOperationListener; +import org.opensearch.indices.IndicesBitsetFilterCache; import org.opensearch.indices.IndicesModule; import org.opensearch.indices.mapper.MapperRegistry; import org.opensearch.plugins.SearchPlugin; @@ -152,6 +152,8 @@ import org.opensearch.search.streaming.FlushMode; import org.opensearch.test.InternalAggregationTestCase; import org.opensearch.test.OpenSearchTestCase; +import org.opensearch.threadpool.TestThreadPool; +import org.opensearch.threadpool.ThreadPool; import org.junit.After; import org.junit.Before; @@ -196,6 +198,8 @@ public abstract class AggregatorTestCase extends OpenSearchTestCase { private List releasables = new ArrayList<>(); private static final String TYPE_NAME = "type"; protected ValuesSourceRegistry valuesSourceRegistry; + protected ThreadPool aggTestThreadPool; + protected IndicesBitsetFilterCache aggTestIndicesBitsetFilterCache; // A list of field types that should not be tested, or are not currently supported private static List TYPE_TEST_DENYLIST; @@ -502,7 +506,13 @@ public boolean shouldCache(Query query) { when(searchContext.numberOfShards()).thenReturn(1); when(searchContext.searcher()).thenReturn(contextIndexSearcher); when(searchContext.fetchPhase()).thenReturn(new FetchPhase(Arrays.asList(new FetchSourcePhase(), new FetchDocValuesPhase()))); - when(searchContext.bitsetFilterCache()).thenReturn(new BitsetFilterCache(indexSettings, mock(Listener.class))); + if (aggTestThreadPool == null) { + aggTestThreadPool = new TestThreadPool("agg_test"); + aggTestIndicesBitsetFilterCache = new IndicesBitsetFilterCache(Settings.EMPTY, aggTestThreadPool); + } + when(searchContext.bitsetFilterCache()).thenReturn( + new BitsetFilterCache(indexSettings, aggTestIndicesBitsetFilterCache, mock(BitsetFilterCache.Listener.class)) + ); IndexShard indexShard = mock(IndexShard.class); when(indexShard.shardId()).thenReturn(new ShardId("test", "test", 0)); when(indexShard.indexSettings()).thenReturn(indexSettings); @@ -1378,6 +1388,14 @@ public IndexAnalyzers getIndexAnalyzers() { private void cleanupReleasables() { Releasables.close(releasables); releasables.clear(); + if (aggTestIndicesBitsetFilterCache != null) { + aggTestIndicesBitsetFilterCache.close(); + aggTestIndicesBitsetFilterCache = null; + } + if (aggTestThreadPool != null) { + ThreadPool.terminate(aggTestThreadPool, 10, java.util.concurrent.TimeUnit.SECONDS); + aggTestThreadPool = null; + } } /** diff --git a/test/framework/src/main/java/org/opensearch/test/AbstractBuilderTestCase.java b/test/framework/src/main/java/org/opensearch/test/AbstractBuilderTestCase.java index befef4ff57088..c7dff0343d34c 100644 --- a/test/framework/src/main/java/org/opensearch/test/AbstractBuilderTestCase.java +++ b/test/framework/src/main/java/org/opensearch/test/AbstractBuilderTestCase.java @@ -75,6 +75,7 @@ import org.opensearch.index.query.QueryBuilderVisitor; import org.opensearch.index.query.QueryShardContext; import org.opensearch.index.similarity.SimilarityService; +import org.opensearch.indices.IndicesBitsetFilterCache; import org.opensearch.indices.IndicesModule; import org.opensearch.indices.analysis.AnalysisModule; import org.opensearch.indices.fielddata.cache.IndicesFieldDataCache; @@ -380,6 +381,7 @@ private static class ServiceHolder implements Closeable { private final SimilarityService similarityService; private final MapperService mapperService; private final BitsetFilterCache bitsetFilterCache; + private final IndicesBitsetFilterCache indicesBitsetFilterCache; private final ScriptService scriptService; private final Client client; private final long nowInMillis; @@ -450,7 +452,8 @@ private static class ServiceHolder implements Closeable { mapperService, threadPool ); - bitsetFilterCache = new BitsetFilterCache(idxSettings, new BitsetFilterCache.Listener() { + indicesBitsetFilterCache = new IndicesBitsetFilterCache(Settings.EMPTY, threadPool); + bitsetFilterCache = new BitsetFilterCache(idxSettings, indicesBitsetFilterCache, new BitsetFilterCache.Listener() { @Override public void onCache(ShardId shardId, Accountable accountable) { @@ -528,7 +531,9 @@ public static Predicate indexNameMatcher() { } @Override - public void close() throws IOException {} + public void close() throws IOException { + indicesBitsetFilterCache.close(); + } QueryShardContext createShardContext(IndexSearcher searcher) { return new QueryShardContext(