diff --git a/server/src/internalClusterTest/java/org/opensearch/aliases/FilterAwareAliasIT.java b/server/src/internalClusterTest/java/org/opensearch/aliases/FilterAwareAliasIT.java index f47a5882a7d78..3bce00dcce8fc 100644 --- a/server/src/internalClusterTest/java/org/opensearch/aliases/FilterAwareAliasIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/aliases/FilterAwareAliasIT.java @@ -9,6 +9,7 @@ package org.opensearch.aliases; import org.opensearch.action.admin.indices.alias.IndicesAliasesRequest.AliasActions; +import org.opensearch.action.bulk.BulkRequestBuilder; import org.opensearch.action.search.SearchResponse; import org.opensearch.action.search.SearchType; import org.opensearch.action.support.WriteRequest.RefreshPolicy; @@ -45,15 +46,27 @@ public class FilterAwareAliasIT extends OpenSearchIntegTestCase { /** Build a corpus where RESTRICTED_TERM appears only in filtered-out docs, plus a single visible * sample doc that contains both the restricted term and the absent term. */ private void buildCorpus() throws Exception { + buildCorpus(200); + } + + /** + * As {@link #buildCorpus()}, with the number of restricted documents under the caller's control. The visible + * subset stays at 21 documents, so {@code restrictedDocs} sets how selective the view is. + */ + private void buildCorpus(int restrictedDocs) throws Exception { assertAcked( prepareCreate(INDEX).setMapping("dept", "type=keyword", "content", "type=text") .setSettings(Settings.builder().put("index.number_of_shards", 1).put("index.number_of_replicas", 0)) ); // Many restricted docs carrying the restricted term -> raises its corpus-wide df. - for (int i = 0; i < 200; i++) { - client().prepareIndex(INDEX).setSource("dept", RESTRICTED_DEPT, "content", "filler " + RESTRICTED_TERM + " noise" + i).get(); + BulkRequestBuilder bulk = client().prepareBulk(); + for (int i = 0; i < restrictedDocs; i++) { + bulk.add( + client().prepareIndex(INDEX).setSource("dept", RESTRICTED_DEPT, "content", "filler " + RESTRICTED_TERM + " noise" + i) + ); } + assertFalse(bulk.get().hasFailures()); // A handful of visible docs (without the restricted term of their own). for (int i = 0; i < 20; i++) { client().prepareIndex(INDEX).setSource("dept", VISIBLE_DEPT, "content", "cardio note " + i).get(); @@ -184,7 +197,20 @@ public void testPreFilterPrefixScoresOverVisibleSubsetOnly() throws Exception { * duration of the assertions and clear it afterwards. */ public void testFilteredStatisticsMatchPhysicallyFilteredIndex() throws Exception { - buildCorpus(); + assertFilteredStatisticsMatchPhysicallyFilteredIndex(200); // 21 visible of 221 documents + } + + /** + * The same equivalence, on a highly selective view: 21 visible of 2,021 documents, about 1%. Cost scales with + * the size of the view rather than the size of the index, so a selective view exercises a very different ratio + * of visible to total documents, and the statistics must come out identical either way. + */ + public void testFilteredStatisticsMatchPhysicallyFilteredIndexForSelectiveView() throws Exception { + assertFilteredStatisticsMatchPhysicallyFilteredIndex(2000); + } + + private void assertFilteredStatisticsMatchPhysicallyFilteredIndex(int restrictedDocs) throws Exception { + buildCorpus(restrictedDocs); addAlias("fs_pre", "pre_filter"); // Build a physical "visible-only" index: reindex just the cardiology docs. This is the @@ -226,6 +252,59 @@ public void testFilteredStatisticsMatchPhysicallyFilteredIndex() throws Exceptio } } + /** + * Small-view guardrail: filtered statistics cost O(visible postings) to build, so above a configured + * visible-doc budget the request falls back to the default constant_score behavior instead of paying that + * cost. Asserts both sides of the switch on the same corpus and alias: with a generous budget the sample + * scores with real BM25 (not 1.0), and with a budget below the visible-subset size every hit scores exactly + * 1.0 -- still leak-free, just unranked. + */ + public void testFilteredStatisticsGuardrailFallsBackToConstantScore() throws Exception { + buildCorpus(); + addAlias("fs_guard", "pre_filter"); + + final String gate = "opensearch.filter_aware_alias.filtered_stats"; + final String budget = "opensearch.filter_aware_alias.filtered_stats.max_visible_docs"; + final String previousGate = System.getProperty(gate); + final String previousBudget = System.getProperty(budget); + try { + System.setProperty(gate, "true"); + + // Budget well above the ~21 visible docs: filtered_stats applies, so the restricted-only term is + // scored with real (visible-subset) BM25 rather than flattened. + System.setProperty(budget, "1000000"); + float rankedScore = sampleScore("fs_guard", RESTRICTED_TERM); + assertTrue("sample found with filtered_stats in budget", rankedScore > 0f); + assertNotEquals( + "with filtered_stats applied the score should be real BM25, not the constant_score 1.0", + 1.0f, + rankedScore, + 0.0001f + ); + + // Budget below the visible-subset size: the guardrail trips and we fall back to constant_score, + // which scores every hit exactly 1.0. + System.setProperty(budget, "1"); + float guardedScore = sampleScore("fs_guard", RESTRICTED_TERM); + assertEquals("guardrail should fall back to constant_score (flat 1.0)", 1.0f, guardedScore, 0.0001f); + + // The fallback is still leak-free: a term confined to filtered-out docs is indistinguishable from + // a term that exists nowhere. + assertEquals("fallback stays leak-free", sampleScore("fs_guard", ABSENT_TERM), guardedScore, 0.0001f); + } finally { + restoreProperty(gate, previousGate); + restoreProperty(budget, previousBudget); + } + } + + private static void restoreProperty(String key, String previous) { + if (previous == null) { + System.clearProperty(key); + } else { + System.setProperty(key, previous); + } + } + /** Build a multi-shard corpus for the dfs test. Same shape as {@link #buildCorpus()} but spread * across several shards so dfs_query_then_fetch must aggregate per-shard statistics. */ private void buildMultiShardCorpus(String index, int shards) throws Exception { 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 b59e67749782b..10471e14d967c 100644 --- a/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java +++ b/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java @@ -50,6 +50,7 @@ import org.opensearch.index.MergeSchedulerConfig; import org.opensearch.index.SearchSlowLog; import org.opensearch.index.TieredMergePolicyProvider; +import org.opensearch.index.cache.filteredstats.FilteredStatsWarmer; import org.opensearch.index.compositeindex.datacube.startree.StarTreeIndexSettings; import org.opensearch.index.engine.EngineConfig; import org.opensearch.index.fielddata.IndexFieldDataService; @@ -209,6 +210,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings { MapperService.INDEX_MAPPING_FIELD_NAME_LENGTH_LIMIT_SETTING, MapperService.INDEX_MAPPING_DYNAMIC_PROPERTIES_LUCENE_FIELD_LIMIT_SETTING, IndicesBitsetFilterCache.INDEX_LOAD_RANDOM_ACCESS_FILTERS_EAGERLY_SETTING, + FilteredStatsWarmer.INDEX_WARM_FILTERED_STATS_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/IndexService.java b/server/src/main/java/org/opensearch/index/IndexService.java index aaca83985e348..ee2de12a9e76d 100644 --- a/server/src/main/java/org/opensearch/index/IndexService.java +++ b/server/src/main/java/org/opensearch/index/IndexService.java @@ -60,6 +60,7 @@ import org.opensearch.common.util.concurrent.AbstractRunnable; import org.opensearch.common.util.io.IOUtils; import org.opensearch.core.Assertions; +import org.opensearch.core.common.breaker.CircuitBreaker; import org.opensearch.core.common.io.stream.NamedWriteableRegistry; import org.opensearch.core.index.shard.ShardId; import org.opensearch.core.indices.breaker.CircuitBreakerService; @@ -72,6 +73,8 @@ import org.opensearch.index.analysis.IndexAnalyzers; import org.opensearch.index.cache.IndexCache; import org.opensearch.index.cache.bitset.BitsetFilterCache; +import org.opensearch.index.cache.filteredstats.FilteredStatsCache; +import org.opensearch.index.cache.filteredstats.FilteredStatsWarmer; import org.opensearch.index.cache.query.QueryCache; import org.opensearch.index.compositeindex.CompositeIndexSettings; import org.opensearch.index.engine.Engine; @@ -326,12 +329,17 @@ public IndexService( return shard == null ? IndicesFieldDataCache.Key.NO_SHARD_IDENTITY : System.identityHashCode(shard); }); this.bitsetFilterCache = new BitsetFilterCache(indexSettings, indicesBitsetFilterCache, new BitsetCacheListener(this)); + FilteredStatsCache filteredStatsCache = new FilteredStatsCache( + indexSettings, + circuitBreakerService.getBreaker(CircuitBreaker.FIELDDATA) + ); this.warmer = new IndexWarmer( threadPool, indexFieldData, - indicesBitsetFilterCache != null ? indicesBitsetFilterCache.createListener(threadPool) : null + indicesBitsetFilterCache != null ? indicesBitsetFilterCache.createListener(threadPool) : null, + new FilteredStatsWarmer(threadPool, filteredStatsCache) ); - this.indexCache = new IndexCache(indexSettings, queryCache, bitsetFilterCache); + this.indexCache = new IndexCache(indexSettings, queryCache, bitsetFilterCache, filteredStatsCache); } else { assert indexAnalyzers == null; this.mapperService = null; diff --git a/server/src/main/java/org/opensearch/index/cache/IndexCache.java b/server/src/main/java/org/opensearch/index/cache/IndexCache.java index 1067863fe9675..c84ade6dc9522 100644 --- a/server/src/main/java/org/opensearch/index/cache/IndexCache.java +++ b/server/src/main/java/org/opensearch/index/cache/IndexCache.java @@ -37,6 +37,7 @@ import org.opensearch.index.AbstractIndexComponent; import org.opensearch.index.IndexSettings; import org.opensearch.index.cache.bitset.BitsetFilterCache; +import org.opensearch.index.cache.filteredstats.FilteredStatsCache; import org.opensearch.index.cache.query.QueryCache; import java.io.Closeable; @@ -52,11 +53,22 @@ public class IndexCache extends AbstractIndexComponent implements Closeable { private final QueryCache queryCache; private final BitsetFilterCache bitsetFilterCache; + private final FilteredStatsCache filteredStatsCache; public IndexCache(IndexSettings indexSettings, QueryCache queryCache, BitsetFilterCache bitsetFilterCache) { + this(indexSettings, queryCache, bitsetFilterCache, null); + } + + public IndexCache( + IndexSettings indexSettings, + QueryCache queryCache, + BitsetFilterCache bitsetFilterCache, + FilteredStatsCache filteredStatsCache + ) { super(indexSettings); this.queryCache = queryCache; this.bitsetFilterCache = bitsetFilterCache; + this.filteredStatsCache = filteredStatsCache; } public QueryCache query() { @@ -70,14 +82,25 @@ public BitsetFilterCache bitsetFilterCache() { return bitsetFilterCache; } + /** + * Return the {@link FilteredStatsCache} for this index, or {@code null} when filtered-stats caching is unavailable + * (e.g. a cacheless index context). Callers must tolerate {@code null} and fall back to inline computation. + */ + public FilteredStatsCache filteredStatsCache() { + return filteredStatsCache; + } + @Override public void close() throws IOException { - IOUtils.close(queryCache, bitsetFilterCache); + IOUtils.close(queryCache, bitsetFilterCache, filteredStatsCache); } public void clear(String reason) { queryCache.clear(reason); bitsetFilterCache.clear(reason); + if (filteredStatsCache != null) { + filteredStatsCache.clear(reason); + } } } diff --git a/server/src/main/java/org/opensearch/index/cache/filteredstats/FilteredStatsCache.java b/server/src/main/java/org/opensearch/index/cache/filteredstats/FilteredStatsCache.java new file mode 100644 index 0000000000000..f15d5ff55f7dc --- /dev/null +++ b/server/src/main/java/org/opensearch/index/cache/filteredstats/FilteredStatsCache.java @@ -0,0 +1,366 @@ +/* + * 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.index.cache.filteredstats; + +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.NumericDocValues; +import org.apache.lucene.index.PostingsEnum; +import org.apache.lucene.index.Term; +import org.apache.lucene.index.Terms; +import org.apache.lucene.index.TermsEnum; +import org.apache.lucene.search.DocIdSetIterator; +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.ScorerSupplier; +import org.apache.lucene.search.Weight; +import org.apache.lucene.util.BitSet; +import org.apache.lucene.util.BitSetIterator; +import org.apache.lucene.util.Bits; +import org.apache.lucene.util.FixedBitSet; +import org.apache.lucene.util.SmallFloat; +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.core.common.breaker.CircuitBreaker; +import org.opensearch.index.AbstractIndexComponent; +import org.opensearch.index.IndexSettings; + +import java.io.Closeable; +import java.io.IOException; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Per-index cache of the visible-subset statistics used by the filter-aware-alias {@code filtered_stats} path. + *

+ * When a search runs {@code filtered_stats}, BM25 collection/term statistics are recomputed over only the documents + * that match the alias filter. That recomputation is O(total corpus) per query: it builds a per-segment "visible" + * bitset from the filter and then walks postings intersected with that bitset. Because Lucene segments are immutable + * and the alias filter is stable for a given view, the result is identical across every query hitting the same + * (segment, filter) pair -- so it is cached here and only computed on the first query per segment. + *

+ * Entries are keyed by the segment core {@link IndexReader.CacheKey} and evicted via a + * {@link IndexReader.CacheHelper#addClosedListener close listener}: when a segment is replaced on refresh or merge its + * core closes, its entry drops, and the next query recomputes against the new segment. This gives correct freshness + * with no explicit invalidation. + *

+ * Cached bytes are reported to the fielddata circuit breaker with {@link CircuitBreaker#addWithoutBreaking} + * (accounting only -- this cache never rejects), and returned on eviction. + * + * @opensearch.experimental + */ +@ExperimentalApi +public final class FilteredStatsCache extends AbstractIndexComponent implements Closeable { + + /** + * A computation that produces a value to cache on the first miss for a (segment, filter[, field|term]) key. + * Separate from {@code java.util.function.Supplier} so it can throw {@link IOException} from the postings walk. + */ + @ExperimentalApi + @FunctionalInterface + public interface Computation { + T compute() throws IOException; + } + + /** Sentinel {@code long[]} stored for a (segment, filter, term) that has no visible occurrences, so the "absent" answer is cached too. */ + private static final long[] TERM_ABSENT = new long[0]; + + /** + * Sentinel stored for a (segment, filter) with no visible docs. {@link ConcurrentHashMap} forbids {@code null} + * values, so we cache this identity and translate it back to {@code null} on read (the "no visible docs" contract). + */ + private static final BitSet NO_VISIBLE_DOCS = new FixedBitSet(1); + + /** Per-segment cached statistics for the visible subset, sub-keyed by the alias filter query. */ + private static final class SegmentEntry { + // filter -> visible-doc bitset for this segment (may be null when no visible docs match in this segment). + private final Map visibleBitSets = new ConcurrentHashMap<>(); + // filter -> field -> [docCount, sumTotalTermFreq, sumDocFreq] over the visible subset. + private final Map> fieldContrib = new ConcurrentHashMap<>(); + // filter -> term -> [docFreq, totalTermFreq] over the visible subset (TERM_ABSENT when none). + private final Map> termContrib = new ConcurrentHashMap<>(); + // approximate bytes this entry has reported to the breaker, returned when the segment closes. + private final AtomicLong bytes = new AtomicLong(); + } + + private final ConcurrentHashMap entries = new ConcurrentHashMap<>(); + private final CircuitBreaker accountingBreaker; + + public FilteredStatsCache(IndexSettings indexSettings, CircuitBreaker accountingBreaker) { + super(indexSettings); + this.accountingBreaker = accountingBreaker; + } + + /** + * Returns the visible-doc bitset for {@code ctx} under {@code filter}, computing and caching it on first access. + * A cached {@code null} (no visible docs in this segment) is preserved and returned as {@code null}. + */ + public BitSet getOrComputeVisibleBitSet(LeafReaderContext ctx, Query filter, Computation compute) throws IOException { + track(filter, null); + final SegmentEntry entry = entryFor(ctx); + if (entry == null) { + return compute.compute(); + } + final BitSet cached = entry.visibleBitSets.get(filter); + if (cached != null) { + return cached == NO_VISIBLE_DOCS ? null : cached; + } + final BitSet bitSet = compute.compute(); + entry.visibleBitSets.put(filter, bitSet == null ? NO_VISIBLE_DOCS : bitSet); + if (bitSet != null) { + account(entry, bitSet.ramBytesUsed()); + } + return bitSet; + } + + /** + * Returns the visible-subset field contribution {@code [docCount, sumTotalTermFreq, sumDocFreq]} for + * {@code (ctx, filter, field)}, computing and caching it on first access. + */ + public long[] getOrComputeFieldContribution(LeafReaderContext ctx, Query filter, String field, Computation compute) + throws IOException { + track(filter, field); + final SegmentEntry entry = entryFor(ctx); + if (entry == null) { + return compute.compute(); + } + final Map byField = entry.fieldContrib.computeIfAbsent(filter, f -> new ConcurrentHashMap<>()); + long[] contribution = byField.get(field); + if (contribution == null) { + contribution = compute.compute(); + byField.put(field, contribution); + account(entry, (long) contribution.length * Long.BYTES); + } + return contribution; + } + + /** + * Returns the visible-subset term contribution {@code [docFreq, totalTermFreq]} for {@code (ctx, filter, term)}, + * or {@code null} when the term has no visible occurrences. Computed and cached on first access; the "absent" + * answer is cached as well so repeated probes of a filtered-out term stay cheap. + */ + public long[] getOrComputeTermContribution(LeafReaderContext ctx, Query filter, Term term, Computation compute) + throws IOException { + final SegmentEntry entry = entryFor(ctx); + if (entry == null) { + return compute.compute(); + } + final Map byTerm = entry.termContrib.computeIfAbsent(filter, f -> new ConcurrentHashMap<>()); + long[] contribution = byTerm.get(term); + if (contribution == null) { + contribution = compute.compute(); + final long[] toStore = contribution == null ? TERM_ABSENT : contribution; + byTerm.put(term, toStore); + account(entry, (long) toStore.length * Long.BYTES); + } + return contribution == TERM_ABSENT ? null : contribution; + } + + /** + * Returns the cache entry for a segment, creating it (and registering the eviction listener) on first use. + * Returns {@code null} when the segment exposes no core cache helper, so the caller falls back to inline compute. + */ + private SegmentEntry entryFor(LeafReaderContext ctx) { + final IndexReader.CacheHelper cacheHelper = ctx.reader().getCoreCacheHelper(); + if (cacheHelper == null) { + return null; + } + final IndexReader.CacheKey key = cacheHelper.getKey(); + return entries.computeIfAbsent(key, k -> { + cacheHelper.addClosedListener(this::onSegmentClose); + return new SegmentEntry(); + }); + } + + private void onSegmentClose(IndexReader.CacheKey key) { + final SegmentEntry removed = entries.remove(key); + if (removed != null && accountingBreaker != null) { + final long bytes = removed.bytes.getAndSet(0); + if (bytes != 0) { + accountingBreaker.addWithoutBreaking(-bytes); + } + } + } + + private void account(SegmentEntry entry, long bytes) { + if (bytes <= 0) { + return; + } + entry.bytes.addAndGet(bytes); + if (accountingBreaker != null) { + accountingBreaker.addWithoutBreaking(bytes); + } + } + + /** + * Filters (and, per filter, the fields) this cache has actually been asked about, so a warmer can pre-populate + * exactly what is in use on a newly visible segment rather than guessing. Bounded so an unusual workload that + * cycles through many distinct filters cannot grow this without limit. + */ + private static final int MAX_TRACKED_FILTERS = 32; + private static final int MAX_TRACKED_FIELDS_PER_FILTER = 32; + private final Map> inUse = new ConcurrentHashMap<>(); + + private void track(Query filter, String field) { + Set fields = inUse.get(filter); + if (fields == null) { + if (inUse.size() >= MAX_TRACKED_FILTERS) { + return; + } + fields = inUse.computeIfAbsent(filter, f -> ConcurrentHashMap.newKeySet()); + } + if (field != null && fields.size() < MAX_TRACKED_FIELDS_PER_FILTER) { + fields.add(field); + } + } + + /** + * Pre-populates the visible bitset and per-field statistics for {@code ctx} for every (filter, field) pair this + * cache has seen in use, so the first query to touch a newly refreshed or merged segment does not pay the + * O(visible postings) build itself. Best-effort: any failure is logged and skipped, because warming is an + * optimization and the query path recomputes on a miss anyway. + */ + public void warm(IndexSearcher searcher, LeafReaderContext ctx) { + for (Map.Entry> entry : inUse.entrySet()) { + final Query filter = entry.getKey(); + try { + final Weight weight = searcher.createWeight(searcher.rewrite(filter), ScoreMode.COMPLETE_NO_SCORES, 1f); + final BitSet visible = getOrComputeVisibleBitSet(ctx, filter, () -> computeVisibleBitSet(weight, ctx)); + if (visible == null) { + continue; + } + for (String field : entry.getValue()) { + getOrComputeFieldContribution(ctx, filter, field, () -> computeFieldContribution(ctx, visible, field)); + } + } catch (IOException | RuntimeException e) { + logger.debug(() -> "failed to warm filtered statistics for filter [" + filter + "]", e); + } + } + } + + /** + * Builds the visible-doc bitset for a single segment from an (already rewritten) filter weight. Returns + * {@code null} when no live document in the segment matches, which callers treat as "no visible docs here". + * Only live documents are included so counts match what a physically filtered index would report. + */ + public static BitSet computeVisibleBitSet(Weight weight, LeafReaderContext ctx) throws IOException { + final ScorerSupplier scorerSupplier = weight.scorerSupplier(ctx); + if (scorerSupplier == null) { + return null; + } + final Scorer scorer = scorerSupplier.get(Long.MAX_VALUE); + if (scorer == null) { + return null; + } + final FixedBitSet bitSet = new FixedBitSet(ctx.reader().maxDoc()); + final DocIdSetIterator iterator = scorer.iterator(); + final Bits liveDocs = ctx.reader().getLiveDocs(); + for (int doc = iterator.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = iterator.nextDoc()) { + if (liveDocs == null || liveDocs.get(doc)) { + bitSet.set(doc); + } + } + return bitSet.cardinality() > 0 ? bitSet : null; + } + + /** + * Computes a single segment's {@code [docCount, sumTotalTermFreq, sumDocFreq]} contribution for a field over the + * visible subset. This is the O(field postings) walk that {@link #getOrComputeFieldContribution} memoizes. + */ + public static long[] computeFieldContribution(LeafReaderContext ctx, BitSet visible, String field) throws IOException { + final Terms terms = ctx.reader().terms(field); + if (terms == null) { + return new long[] { 0, 0, 0 }; + } + final NumericDocValues norms = ctx.reader().getNormValues(field); + if (norms == null) { + // Field indexed without norms: there is no per-document length to read, so fall back to the exact + // (but O(all postings in the field)) walk below. + return computeFieldContributionByTermWalk(ctx, visible, terms); + } + + // Fast path. BM25 reads exactly two values off CollectionStatistics -- docCount, and + // sumTotalTermFreq only to derive avgFieldLength = sumTotalTermFreq / docCount. Neither needs the + // term dictionary: norms carry one length value per document, so a single pass over the visible + // documents yields both in O(visible docs) instead of O(all postings in the field). + long docCount = 0; + long sumTotalTermFreq = 0; + final BitSetIterator visibleDocs = new BitSetIterator(visible, visible.approximateCardinality()); + for (int doc = visibleDocs.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = visibleDocs.nextDoc()) { + if (norms.advanceExact(doc)) { + docCount++; + // Norms encode field length the way the similarity wrote them; BM25 uses SmallFloat.intToByte4, + // and its own per-document length normalization decodes the same lossy value. A similarity that + // encodes something else would make avgFieldLength meaningless here -- see the class javadoc. + sumTotalTermFreq += SmallFloat.byte4ToInt((byte) norms.longValue()); + } + } + if (docCount == 0) { + return new long[] { 0, 0, 0 }; + } + + // sumDocFreq is never read by BM25, and computing it exactly is what forces a full term-dictionary + // walk. Estimate it by scaling the shard-wide value (free field metadata) by the visible fraction, + // then clamp to Lucene's CollectionStatistics invariants. + final long shardDocCount = terms.getDocCount(); + long sumDocFreq = shardDocCount <= 0 ? docCount : (long) ((double) terms.getSumDocFreq() * docCount / shardDocCount); + sumDocFreq = Math.min(Math.max(sumDocFreq, docCount), Math.max(sumTotalTermFreq, docCount)); + sumTotalTermFreq = Math.max(sumTotalTermFreq, sumDocFreq); + return new long[] { docCount, sumTotalTermFreq, sumDocFreq }; + } + + /** + * Exact contribution via a full walk of the field's term dictionary. Correct for any similarity, but costs + * O(all postings in the field) -- retained only for fields indexed without norms. + */ + private static long[] computeFieldContributionByTermWalk(LeafReaderContext ctx, BitSet visible, Terms terms) throws IOException { + final TermsEnum termsEnum = terms.iterator(); + final FixedBitSet docsWithField = new FixedBitSet(ctx.reader().maxDoc()); + long sumTotalTermFreq = 0; + long sumDocFreq = 0; + PostingsEnum postings = null; + while (termsEnum.next() != null) { + postings = termsEnum.postings(postings, PostingsEnum.FREQS); + long termDocFreq = 0; + for (int doc = postings.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = postings.nextDoc()) { + if (visible.get(doc)) { + termDocFreq++; + sumTotalTermFreq += postings.freq(); + docsWithField.set(doc); + } + } + sumDocFreq += termDocFreq; + } + return new long[] { docsWithField.cardinality(), sumTotalTermFreq, sumDocFreq }; + } + + /** Total bytes this cache is currently accounting for -- for tests and stats. */ + public long ramBytesUsed() { + long total = 0; + for (SegmentEntry entry : entries.values()) { + total += entry.bytes.get(); + } + return total; + } + + public void clear(String reason) { + logger.debug("clearing filtered-stats cache because [{}]", reason); + for (IndexReader.CacheKey key : entries.keySet()) { + onSegmentClose(key); + } + } + + @Override + public void close() { + clear("close"); + } +} diff --git a/server/src/main/java/org/opensearch/index/cache/filteredstats/FilteredStatsWarmer.java b/server/src/main/java/org/opensearch/index/cache/filteredstats/FilteredStatsWarmer.java new file mode 100644 index 0000000000000..3f0c01156647a --- /dev/null +++ b/server/src/main/java/org/opensearch/index/cache/filteredstats/FilteredStatsWarmer.java @@ -0,0 +1,78 @@ +/* + * 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.index.cache.filteredstats; + +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.search.IndexSearcher; +import org.opensearch.common.lucene.index.OpenSearchDirectoryReader; +import org.opensearch.common.settings.Setting; +import org.opensearch.index.IndexWarmer; +import org.opensearch.index.shard.IndexShard; +import org.opensearch.threadpool.ThreadPool; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; + +/** + * Warms {@link FilteredStatsCache} for segments that have just become visible. + *

+ * The cache is per-segment, so a refresh or merge that produces a new segment leaves that segment's visible-subset + * statistics uncomputed: the first query to touch it pays the O(visible postings) build. That is bounded (existing + * segments stay warm) but shows up as an occasional tail-latency spike on a write-heavy index, and it repeats on every + * node that has to build its own cache. This warmer moves that work off the query path and onto the warmer thread pool. + *

+ * It only warms (filter, field) pairs the cache has actually seen in use, so it does no speculative work on an index + * where nothing queries through a filtered alias. Warming is best-effort: on failure the query path simply recomputes. + * + * @opensearch.internal + */ +public final class FilteredStatsWarmer implements IndexWarmer.Listener { + + /** + * Whether to pre-compute visible-subset statistics for newly visible segments. Off by default: it costs warmer-pool + * work on every refresh, which is only worth paying for indices actually served through {@code pre_filter} aliases + * with {@code filtered_stats} scoring. + */ + public static final Setting INDEX_WARM_FILTERED_STATS_SETTING = Setting.boolSetting( + "index.filter_aware_alias.warm_filtered_stats", + false, + Setting.Property.IndexScope + ); + + private final Executor executor; + private final FilteredStatsCache cache; + + public FilteredStatsWarmer(ThreadPool threadPool, FilteredStatsCache cache) { + this.executor = threadPool.executor(ThreadPool.Names.WARMER); + this.cache = cache; + } + + @Override + public IndexWarmer.TerminationHandle warmReader(IndexShard indexShard, OpenSearchDirectoryReader reader) { + if (cache == null || indexShard.indexSettings().getValue(INDEX_WARM_FILTERED_STATS_SETTING) == false) { + return IndexWarmer.TerminationHandle.NO_WAIT; + } + // A plain searcher over this reader: we only need it to rewrite the filter and create weights, and we must not + // pollute (or read) the query cache while doing so. + final IndexSearcher searcher = new IndexSearcher(reader); + searcher.setQueryCache(null); + + final CountDownLatch latch = new CountDownLatch(reader.leaves().size()); + for (final LeafReaderContext ctx : reader.leaves()) { + executor.execute(() -> { + try { + cache.warm(searcher, ctx); + } finally { + latch.countDown(); + } + }); + } + return latch::await; + } +} diff --git a/server/src/main/java/org/opensearch/search/DefaultSearchContext.java b/server/src/main/java/org/opensearch/search/DefaultSearchContext.java index 158135e8729fd..edda83ad7d920 100644 --- a/server/src/main/java/org/opensearch/search/DefaultSearchContext.java +++ b/server/src/main/java/org/opensearch/search/DefaultSearchContext.java @@ -34,6 +34,7 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import org.apache.lucene.index.LeafReaderContext; import org.apache.lucene.search.BooleanClause.Occur; import org.apache.lucene.search.BooleanQuery; import org.apache.lucene.search.BoostQuery; @@ -43,6 +44,9 @@ import org.apache.lucene.search.FieldDoc; import org.apache.lucene.search.MatchNoDocsQuery; import org.apache.lucene.search.Query; +import org.apache.lucene.search.ScoreMode; +import org.apache.lucene.search.ScorerSupplier; +import org.apache.lucene.search.Weight; import org.opensearch.Version; import org.opensearch.action.search.SearchShardTask; import org.opensearch.action.search.SearchType; @@ -61,6 +65,7 @@ import org.opensearch.index.IndexService; import org.opensearch.index.IndexSettings; import org.opensearch.index.cache.bitset.BitsetFilterCache; +import org.opensearch.index.cache.filteredstats.FilteredStatsCache; import org.opensearch.index.compositeindex.CompositeIndexSettings; import org.opensearch.index.compositeindex.datacube.startree.StarTreeIndexSettings; import org.opensearch.index.engine.Engine; @@ -182,6 +187,35 @@ static boolean filteredStatisticsEnabled() { return Boolean.parseBoolean(System.getProperty(FILTERED_STATISTICS_PROPERTY, "false")); } + /** + * Guardrail for {@code filtered_stats}: the maximum number of visible (alias-filter-matching) documents for which + * filtered statistics will be computed. Filtered statistics cost O(visible postings) to build the first time a + * segment is queried, so on a very large view that first query can be slow; above this budget we fall back to the + * {@code constant_score} behavior, which is leak-free and flat but unranked. {@code -1} disables the guardrail. + */ + static final String FILTERED_STATISTICS_MAX_VISIBLE_DOCS_PROPERTY = "opensearch.filter_aware_alias.filtered_stats.max_visible_docs"; + + static final long FILTERED_STATISTICS_MAX_VISIBLE_DOCS_DEFAULT = 1_000_000L; + + static long filteredStatisticsMaxVisibleDocs() { + try { + return Long.parseLong( + System.getProperty( + FILTERED_STATISTICS_MAX_VISIBLE_DOCS_PROPERTY, + Long.toString(FILTERED_STATISTICS_MAX_VISIBLE_DOCS_DEFAULT) + ) + ); + } catch (NumberFormatException e) { + return FILTERED_STATISTICS_MAX_VISIBLE_DOCS_DEFAULT; + } + } + + /** + * Memoized result of the {@code filtered_stats} small-view guardrail; see {@link #withinFilteredStatisticsBudget()}. + * {@code null} until first evaluated. + */ + private Boolean filteredStatisticsWithinBudget; + private final ReaderContext readerContext; private final Engine.Searcher engineSearcher; private final ShardSearchRequest request; @@ -745,6 +779,11 @@ public BitsetFilterCache bitsetFilterCache() { return indexService.cache().bitsetFilterCache(); } + @Override + public FilteredStatsCache filteredStatsCache() { + return indexService.cache().filteredStatsCache(); + } + @Override public TimeValue timeout() { return timeout; @@ -877,7 +916,67 @@ public boolean useFilteredStatistics() { return false; } AliasFilter requestAliasFilter = request.getAliasFilter(); - return requestAliasFilter != null && requestAliasFilter.getEnforcement() == AliasFilter.Enforcement.PRE_FILTER; + if (requestAliasFilter == null || requestAliasFilter.getEnforcement() != AliasFilter.Enforcement.PRE_FILTER) { + return false; + } + return withinFilteredStatisticsBudget(); + } + + /** + * Small-view guardrail for {@code filtered_stats}. Estimates how many documents the alias filter matches and + * returns {@code false} when that exceeds {@link #filteredStatisticsMaxVisibleDocs()}, so the caller falls back to + * the {@code constant_score} behavior (still leak-free and flat, but unranked) instead of paying an unbounded + * first-query cost on a very large view. + *

+ * The estimate uses {@link ScorerSupplier#cost()}, which is an upper bound derived from postings metadata -- it + * does not iterate the filter, so the guardrail itself is cheap and, crucially, runs before the expensive + * visible-doc/statistics build it is protecting against. + *

+ * The result is memoized for the life of this context because it is consulted twice per request -- once when + * choosing the query shape in {@link #buildFilteredQuery(Query)} and again when the searcher computes statistics. + * Those two decisions must agree: a filtered-stats query shape scored with whole-shard statistics would reintroduce + * the very whole-corpus IDF the pre-filter is meant to exclude. + */ + private boolean withinFilteredStatisticsBudget() { + if (filteredStatisticsWithinBudget != null) { + return filteredStatisticsWithinBudget; + } + final long maxVisibleDocs = filteredStatisticsMaxVisibleDocs(); + if (maxVisibleDocs < 0) { + filteredStatisticsWithinBudget = true; + return true; + } + boolean withinBudget; + try { + final Weight weight = searcher.createWeight(searcher.rewrite(aliasFilter), ScoreMode.COMPLETE_NO_SCORES, 1f); + long estimate = 0; + for (LeafReaderContext ctx : searcher.getIndexReader().leaves()) { + final ScorerSupplier scorerSupplier = weight.scorerSupplier(ctx); + if (scorerSupplier != null) { + estimate += scorerSupplier.cost(); + if (estimate > maxVisibleDocs) { + break; + } + } + } + withinBudget = estimate <= maxVisibleDocs; + if (withinBudget == false) { + logger.debug( + "filtered_stats disabled for this request: alias filter matches ~{} docs, above the {} budget " + + "({}); falling back to constant_score", + estimate, + maxVisibleDocs, + FILTERED_STATISTICS_MAX_VISIBLE_DOCS_PROPERTY + ); + } + } catch (IOException e) { + // Fail closed onto constant_score: it is leak-free and cheaper, so an estimation failure degrades + // ranking rather than correctness or latency. + logger.debug("could not estimate alias-filter selectivity; falling back to constant_score", e); + withinBudget = false; + } + filteredStatisticsWithinBudget = withinBudget; + return withinBudget; } @Override diff --git a/server/src/main/java/org/opensearch/search/internal/ContextIndexSearcher.java b/server/src/main/java/org/opensearch/search/internal/ContextIndexSearcher.java index d182b913c294a..58e0a9f402075 100644 --- a/server/src/main/java/org/opensearch/search/internal/ContextIndexSearcher.java +++ b/server/src/main/java/org/opensearch/search/internal/ContextIndexSearcher.java @@ -68,13 +68,13 @@ import org.apache.lucene.util.BitSet; import org.apache.lucene.util.BitSetIterator; import org.apache.lucene.util.Bits; -import org.apache.lucene.util.FixedBitSet; import org.apache.lucene.util.SparseFixedBitSet; import org.opensearch.common.annotation.PublicApi; import org.opensearch.common.lease.Releasable; import org.opensearch.common.lucene.Lucene; import org.opensearch.common.lucene.search.TopDocsAndMaxScore; import org.opensearch.core.tasks.TaskCancelledException; +import org.opensearch.index.cache.filteredstats.FilteredStatsCache; import org.opensearch.lucene.util.CombinedBitSet; import org.opensearch.search.DocValueFormat; import org.opensearch.search.SearchHits; @@ -135,6 +135,12 @@ public class ContextIndexSearcher extends IndexSearcher implements Releasable { */ private BitSet[] visibleDocsPerSegment; private boolean visibleDocsInitialized = false; + /** + * The rewritten alias-filter query used to key {@link FilteredStatsCache} entries. Set when the visible bitsets + * are built so the collection/term-statistics lookups reuse the exact same key. Rewriting once per request also + * keeps the key stable across the three statistics methods. + */ + private Query visibleDocsFilterKey; public ContextIndexSearcher( IndexReader reader, @@ -623,9 +629,12 @@ private boolean useFilteredStatistics() { } /** - * Lazily builds and caches the per-segment bitsets of documents matching {@link SearchContext#aliasFilter()}. + * Lazily builds the per-segment bitsets of documents matching {@link SearchContext#aliasFilter()}. * These represent the "visible" subset over which filtered BM25 statistics are computed. Only live documents are - * included so the counts match what a physically-filtered index would report. + * included so the counts match what a physically-filtered index would report. Per-segment bitsets are served from + * {@link FilteredStatsCache} when available (keyed by segment + filter, evicted on segment close), so the + * O(corpus) filter scan is paid once per (segment, filter) rather than once per query; the result array is also + * memoized on this searcher for reuse across the terms of a single request. */ private synchronized BitSet[] getVisibleDocsPerSegment() throws IOException { if (visibleDocsInitialized) { @@ -633,28 +642,22 @@ private synchronized BitSet[] getVisibleDocsPerSegment() throws IOException { } final List leaves = getIndexReader().leaves(); final BitSet[] bitSets = new BitSet[leaves.size()]; - final Query aliasFilter = searchContext.aliasFilter(); + // Rewrite once and reuse as the FilteredStatsCache key for the bitset and the statistics lookups, so the key + // is identical across all three methods within a request. + final Query aliasFilter = rewrite(searchContext.aliasFilter()); + this.visibleDocsFilterKey = aliasFilter; // COMPLETE_NO_SCORES: we only need the matching doc ids, not scores. - final Weight weight = createWeight(rewrite(aliasFilter), ScoreMode.COMPLETE_NO_SCORES, 1f); + final Weight weight = createWeight(aliasFilter, ScoreMode.COMPLETE_NO_SCORES, 1f); + final FilteredStatsCache cache = filteredStatsCache(); for (LeafReaderContext ctx : leaves) { - final ScorerSupplier scorerSupplier = weight.scorerSupplier(ctx); - if (scorerSupplier == null) { - continue; - } - final Scorer scorer = scorerSupplier.get(Long.MAX_VALUE); - if (scorer == null) { - continue; - } - final FixedBitSet bitSet = new FixedBitSet(ctx.reader().maxDoc()); - final DocIdSetIterator iterator = scorer.iterator(); - final Bits liveDocs = ctx.reader().getLiveDocs(); - for (int doc = iterator.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = iterator.nextDoc()) { - if (liveDocs == null || liveDocs.get(doc)) { - bitSet.set(doc); - } - } - if (bitSet.cardinality() > 0) { - bitSets[ctx.ord] = bitSet; + if (cache != null) { + bitSets[ctx.ord] = cache.getOrComputeVisibleBitSet( + ctx, + aliasFilter, + () -> FilteredStatsCache.computeVisibleBitSet(weight, ctx) + ); + } else { + bitSets[ctx.ord] = FilteredStatsCache.computeVisibleBitSet(weight, ctx); } } this.visibleDocsPerSegment = bitSets; @@ -662,6 +665,10 @@ private synchronized BitSet[] getVisibleDocsPerSegment() throws IOException { return bitSets; } + private FilteredStatsCache filteredStatsCache() { + return searchContext == null ? null : searchContext.filteredStatsCache(); + } + /** * Computes {@link CollectionStatistics} for a field over only the visible (alias-filter) documents. Mirrors what * Lucene's {@link IndexSearcher#collectionStatistics(String)} does per segment, but intersects the field's postings @@ -671,35 +678,30 @@ private synchronized BitSet[] getVisibleDocsPerSegment() throws IOException { private CollectionStatistics filteredCollectionStatistics(String field) throws IOException { final BitSet[] visibleDocs = getVisibleDocsPerSegment(); final List leaves = getIndexReader().leaves(); - long docCount = 0; // number of visible docs that have at least one term in this field + final FilteredStatsCache cache = filteredStatsCache(); + long docCount = 0; // number of visible docs that have at least one term in this field long sumTotalTermFreq = 0; // total number of (visible) tokens in this field - long sumDocFreq = 0; // sum over terms of the number of visible docs containing the term + long sumDocFreq = 0; // sum over terms of the number of visible docs containing the term for (LeafReaderContext ctx : leaves) { final BitSet visible = visibleDocs[ctx.ord]; if (visible == null) { continue; } - final Terms terms = ctx.reader().terms(field); - if (terms == null) { - continue; - } - final TermsEnum termsEnum = terms.iterator(); - final FixedBitSet docsWithField = new FixedBitSet(ctx.reader().maxDoc()); - PostingsEnum postings = null; - while (termsEnum.next() != null) { - postings = termsEnum.postings(postings, PostingsEnum.FREQS); - int termDocFreq = 0; - for (int doc = postings.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = postings.nextDoc()) { - if (visible.get(doc)) { - termDocFreq++; - sumTotalTermFreq += postings.freq(); - docsWithField.set(doc); - } - } - sumDocFreq += termDocFreq; + final long[] contribution; + if (cache != null) { + contribution = cache.getOrComputeFieldContribution( + ctx, + visibleDocsFilterKey, + field, + () -> FilteredStatsCache.computeFieldContribution(ctx, visible, field) + ); + } else { + contribution = FilteredStatsCache.computeFieldContribution(ctx, visible, field); } - docCount += docsWithField.cardinality(); + docCount += contribution[0]; + sumTotalTermFreq += contribution[1]; + sumDocFreq += contribution[2]; } if (docCount == 0) { @@ -723,6 +725,7 @@ private CollectionStatistics filteredCollectionStatistics(String field) throws I private TermStatistics filteredTermStatistics(Term term) throws IOException { final BitSet[] visibleDocs = getVisibleDocsPerSegment(); final List leaves = getIndexReader().leaves(); + final FilteredStatsCache cache = filteredStatsCache(); long docFreq = 0; // number of visible docs containing the term long totalTermFreq = 0; // sum of term frequencies over visible docs @@ -731,20 +734,20 @@ private TermStatistics filteredTermStatistics(Term term) throws IOException { if (visible == null) { continue; } - final Terms terms = ctx.reader().terms(term.field()); - if (terms == null) { - continue; - } - final TermsEnum termsEnum = terms.iterator(); - if (termsEnum.seekExact(term.bytes()) == false) { - continue; + final long[] contribution; + if (cache != null) { + contribution = cache.getOrComputeTermContribution( + ctx, + visibleDocsFilterKey, + term, + () -> computeTermContribution(ctx, visible, term) + ); + } else { + contribution = computeTermContribution(ctx, visible, term); } - final PostingsEnum postings = termsEnum.postings(null, PostingsEnum.FREQS); - for (int doc = postings.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = postings.nextDoc()) { - if (visible.get(doc)) { - docFreq++; - totalTermFreq += postings.freq(); - } + if (contribution != null) { + docFreq += contribution[0]; + totalTermFreq += contribution[1]; } } @@ -758,6 +761,36 @@ private TermStatistics filteredTermStatistics(Term term) throws IOException { return new TermStatistics(term.bytes(), docFreq, totalTermFreq); } + /** + * Computes a single segment's {@code [docFreq, totalTermFreq]} contribution for a term over the visible subset, + * or {@code null} when the term occurs in no visible doc in this segment. Memoized per (segment, filter, term) by + * {@link FilteredStatsCache}. + */ + private static long[] computeTermContribution(LeafReaderContext ctx, BitSet visible, Term term) throws IOException { + final Terms terms = ctx.reader().terms(term.field()); + if (terms == null) { + return null; + } + final TermsEnum termsEnum = terms.iterator(); + if (termsEnum.seekExact(term.bytes()) == false) { + return null; + } + final PostingsEnum postings = termsEnum.postings(null, PostingsEnum.FREQS); + // Leapfrog the term's postings against the visible bitset instead of scanning the postings linearly and + // testing each hit: the conjunction advances whichever side is behind, so skip lists do the work and the + // cost is O(min(visible, docFreq)) rather than O(docFreq). The more selective the view, the cheaper it gets. + final DocIdSetIterator visibleAndTerm = ConjunctionUtils.intersectIterators( + Arrays.asList(new BitSetIterator(visible, visible.approximateCardinality()), postings) + ); + long docFreq = 0; + long totalTermFreq = 0; + for (int doc = visibleAndTerm.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = visibleAndTerm.nextDoc()) { + docFreq++; + totalTermFreq += postings.freq(); + } + return docFreq == 0 ? null : new long[] { docFreq, totalTermFreq }; + } + /** * Compute the leaf slices that will be used by concurrent segment search to spread work across threads * @param leaves all the segments diff --git a/server/src/main/java/org/opensearch/search/internal/SearchContext.java b/server/src/main/java/org/opensearch/search/internal/SearchContext.java index 602d966b70349..27dfd728ca2a5 100644 --- a/server/src/main/java/org/opensearch/search/internal/SearchContext.java +++ b/server/src/main/java/org/opensearch/search/internal/SearchContext.java @@ -47,6 +47,7 @@ import org.opensearch.common.unit.TimeValue; import org.opensearch.common.util.BigArrays; import org.opensearch.index.cache.bitset.BitsetFilterCache; +import org.opensearch.index.cache.filteredstats.FilteredStatsCache; import org.opensearch.index.mapper.MappedFieldType; import org.opensearch.index.mapper.MapperService; import org.opensearch.index.mapper.ObjectMapper; @@ -292,6 +293,15 @@ public final void assignRescoreDocIds(RescoreDocIds rescoreDocIds) { public abstract BitsetFilterCache bitsetFilterCache(); + /** + * The {@link FilteredStatsCache} for this context's index, or {@code null} when it is unavailable (e.g. a + * cacheless context). {@code DefaultSearchContext} overrides this; callers must tolerate {@code null} and fall + * back to inline computation. + */ + public FilteredStatsCache filteredStatsCache() { + return null; + } + public abstract TimeValue timeout(); public abstract void timeout(TimeValue timeout); diff --git a/server/src/test/java/org/opensearch/index/cache/filteredstats/FilteredStatsCacheTests.java b/server/src/test/java/org/opensearch/index/cache/filteredstats/FilteredStatsCacheTests.java new file mode 100644 index 0000000000000..fb63a36664edd --- /dev/null +++ b/server/src/test/java/org/opensearch/index/cache/filteredstats/FilteredStatsCacheTests.java @@ -0,0 +1,238 @@ +/* + * 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.index.cache.filteredstats; + +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.Term; +import org.apache.lucene.search.IndexSearcher; +import org.apache.lucene.search.Query; +import org.apache.lucene.search.TermQuery; +import org.apache.lucene.store.Directory; +import org.apache.lucene.util.BitSet; +import org.apache.lucene.util.FixedBitSet; +import org.opensearch.common.settings.Settings; +import org.opensearch.core.common.breaker.CircuitBreaker; +import org.opensearch.core.common.breaker.NoopCircuitBreaker; +import org.opensearch.index.IndexSettings; +import org.opensearch.test.IndexSettingsModule; +import org.opensearch.test.OpenSearchTestCase; + +import java.io.IOException; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.hamcrest.Matchers.greaterThan; + +public class FilteredStatsCacheTests extends OpenSearchTestCase { + + private FilteredStatsCache newCache() { + final IndexSettings indexSettings = IndexSettingsModule.newIndexSettings("idx", Settings.EMPTY); + return new FilteredStatsCache(indexSettings, new NoopCircuitBreaker(CircuitBreaker.FIELDDATA)); + } + + private static Directory writeSingleSegment() throws IOException { + final Directory dir = newDirectory(); + try (IndexWriter writer = new IndexWriter(dir, new IndexWriterConfig())) { + for (int i = 0; i < 3; i++) { + final Document doc = new Document(); + doc.add(new StringField("dept", i == 0 ? "cardiology" : "oncology", Field.Store.NO)); + writer.addDocument(doc); + } + writer.forceMerge(1); + writer.commit(); + } + return dir; + } + + public void testBitSetComputedOnceThenServedFromCache() throws IOException { + final Directory dir = writeSingleSegment(); + try (DirectoryReader reader = DirectoryReader.open(dir)) { + final LeafReaderContext leaf = reader.leaves().get(0); + final FilteredStatsCache cache = newCache(); + final Query filter = new TermQuery(new Term("dept", "cardiology")); + final AtomicInteger calls = new AtomicInteger(); + + final BitSet first = new FixedBitSet(leaf.reader().maxDoc()); + first.set(0); + final BitSet miss = cache.getOrComputeVisibleBitSet(leaf, filter, () -> { + calls.incrementAndGet(); + return first; + }); + // second call must not recompute -- the compute would fail the test if invoked. + final BitSet hit = cache.getOrComputeVisibleBitSet(leaf, filter, () -> { + calls.incrementAndGet(); + return new FixedBitSet(leaf.reader().maxDoc()); + }); + + assertSame(first, miss); + assertSame(first, hit); + assertEquals("bitset should be computed exactly once", 1, calls.get()); + assertThat("caching a bitset should account bytes", cache.ramBytesUsed(), greaterThan(0L)); + } + dir.close(); + } + + public void testNoVisibleDocsCachedAsNullWithoutRecompute() throws IOException { + final Directory dir = writeSingleSegment(); + try (DirectoryReader reader = DirectoryReader.open(dir)) { + final LeafReaderContext leaf = reader.leaves().get(0); + final FilteredStatsCache cache = newCache(); + final Query filter = new TermQuery(new Term("dept", "nonexistent")); + final AtomicInteger calls = new AtomicInteger(); + + final BitSet miss = cache.getOrComputeVisibleBitSet(leaf, filter, () -> { + calls.incrementAndGet(); + return null; // no visible docs in this segment + }); + final BitSet hit = cache.getOrComputeVisibleBitSet(leaf, filter, () -> { + calls.incrementAndGet(); + return new FixedBitSet(leaf.reader().maxDoc()); + }); + + assertNull("no-visible-docs must be returned as null", miss); + assertNull("the null answer must be served from cache", hit); + assertEquals("null bitset should be computed exactly once", 1, calls.get()); + } + dir.close(); + } + + public void testFieldAndTermContributionsAreCached() throws IOException { + final Directory dir = writeSingleSegment(); + try (DirectoryReader reader = DirectoryReader.open(dir)) { + final LeafReaderContext leaf = reader.leaves().get(0); + final FilteredStatsCache cache = newCache(); + final Query filter = new TermQuery(new Term("dept", "cardiology")); + + final long[] field = cache.getOrComputeFieldContribution(leaf, filter, "content", () -> new long[] { 5, 10, 7 }); + assertArrayEquals(new long[] { 5, 10, 7 }, field); + // cached: recompute lambda must not run. + final long[] fieldHit = cache.getOrComputeFieldContribution(leaf, filter, "content", () -> { + throw new AssertionError("field contribution should be served from cache"); + }); + assertArrayEquals(new long[] { 5, 10, 7 }, fieldHit); + + final Term term = new Term("content", "alpha"); + final long[] termMiss = cache.getOrComputeTermContribution(leaf, filter, term, () -> null); // term absent in visible subset + assertNull(termMiss); + // the "absent" answer is cached too. + final long[] termHit = cache.getOrComputeTermContribution(leaf, filter, term, () -> { + throw new AssertionError("absent term answer should be served from cache"); + }); + assertNull(termHit); + } + dir.close(); + } + + /** + * Warming a newly visible segment should pre-populate the (filter, field) pairs already in use, so the first query + * to touch that segment is a cache hit rather than paying the visible-postings build itself. + */ + public void testWarmPrepopulatesInUseFilterAndField() throws IOException { + final Query filter = new TermQuery(new Term("dept", "cardiology")); + final FilteredStatsCache cache = newCache(); + + // A first segment: querying it records (filter, field) as in-use. + final Directory first = writeSingleSegment(); + try (DirectoryReader reader = DirectoryReader.open(first)) { + final LeafReaderContext leaf = reader.leaves().get(0); + final FixedBitSet visible = new FixedBitSet(leaf.reader().maxDoc()); + visible.set(0); + cache.getOrComputeVisibleBitSet(leaf, filter, () -> visible); + cache.getOrComputeFieldContribution(leaf, filter, "dept", () -> new long[] { 1, 1, 1 }); + } + first.close(); + + // A second, independent segment stands in for one that just became visible after a refresh/merge. + final Directory second = writeSingleSegment(); + try (DirectoryReader reader = DirectoryReader.open(second)) { + final LeafReaderContext leaf = reader.leaves().get(0); + final IndexSearcher searcher = new IndexSearcher(reader); + searcher.setQueryCache(null); + + cache.warm(searcher, leaf); + + // Both the bitset and the field contribution must now be served from cache: these computations would + // fail the test if warming had not already populated them. + assertNotNull("warming should have built the visible bitset", cache.getOrComputeVisibleBitSet(leaf, filter, () -> { + throw new AssertionError("bitset should be warm"); + })); + assertNotNull( + "warming should have built the field contribution", + cache.getOrComputeFieldContribution(leaf, filter, "dept", () -> { + throw new AssertionError("field should be warm"); + }) + ); + } + second.close(); + } + + /** + * Cache memory scales with the number of distinct views (filters), not with how selective they are: a visible-doc + * bitset is {@code maxDoc} bits per (segment, filter) whatever its cardinality. This pins that down so the + * capacity-planning claim stays honest -- if the representation ever becomes sparse or shared, this test should be + * updated deliberately rather than silently drifting. + */ + public void testMemoryScalesWithNumberOfFiltersNotSelectivity() throws IOException { + final Directory dir = writeSingleSegment(); + try (DirectoryReader reader = DirectoryReader.open(dir)) { + final LeafReaderContext leaf = reader.leaves().get(0); + final int maxDoc = leaf.reader().maxDoc(); + final FilteredStatsCache cache = newCache(); + + // One very selective filter (a single visible doc) and one unselective filter (all docs visible). + final FixedBitSet sparse = new FixedBitSet(maxDoc); + sparse.set(0); + final FixedBitSet dense = new FixedBitSet(maxDoc); + dense.set(0, maxDoc); + + cache.getOrComputeVisibleBitSet(leaf, new TermQuery(new Term("v", "selective")), () -> sparse); + final long afterSelective = cache.ramBytesUsed(); + cache.getOrComputeVisibleBitSet(leaf, new TermQuery(new Term("v", "unselective")), () -> dense); + final long afterBoth = cache.ramBytesUsed(); + + final long selectiveCost = afterSelective; + final long unselectiveCost = afterBoth - afterSelective; + assertEquals( + "a 1-doc view and an all-docs view must cost the same: cost tracks maxDoc, not cardinality", + selectiveCost, + unselectiveCost + ); + + // Each additional distinct view adds another bitset of the same size -- memory is linear in view count. + for (int i = 0; i < 8; i++) { + final FixedBitSet bits = new FixedBitSet(maxDoc); + bits.set(i % maxDoc); + cache.getOrComputeVisibleBitSet(leaf, new TermQuery(new Term("v", "view" + i)), () -> bits); + } + assertEquals("memory should be linear in the number of distinct views", selectiveCost * 10, cache.ramBytesUsed()); + } + dir.close(); + } + + public void testEntriesEvictedWhenSegmentCloses() throws IOException { + final Directory dir = writeSingleSegment(); + final FilteredStatsCache cache = newCache(); + final Query filter = new TermQuery(new Term("dept", "cardiology")); + try (DirectoryReader reader = DirectoryReader.open(dir)) { + final LeafReaderContext leaf = reader.leaves().get(0); + final BitSet bitSet = new FixedBitSet(leaf.reader().maxDoc()); + bitSet.set(0); + cache.getOrComputeVisibleBitSet(leaf, filter, () -> bitSet); + assertThat(cache.ramBytesUsed(), greaterThan(0L)); + } + // Closing the reader closes the segment core, which fires the eviction listener. + assertEquals("closing the segment must evict its cached entry and return its accounted bytes", 0L, cache.ramBytesUsed()); + dir.close(); + } +}