diff --git a/CHANGELOG.md b/CHANGELOG.md index d82ac0fd18cca..8f78311258919 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -61,6 +61,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Remove child level directory on refresh for CompositeIndexWriter ([#20326](https://github.com/opensearch-project/OpenSearch/pull/20326)) - Fixes and refactoring in stream transport to make it more robust ([#20359](https://github.com/opensearch-project/OpenSearch/pull/20359)) - Harden detection of HTTP/3 support by ensuring Quic native libraries are available for the target platform ([#20680](https://github.com/opensearch-project/OpenSearch/pull/20680)) +- Fix the regression of terms agg optimization at high cardinality ([#20623](https://github.com/opensearch-project/OpenSearch/pull/20623)) ### Dependencies - Bump `com.google.auth:google-auth-library-oauth2-http` from 1.38.0 to 1.41.0 ([#20183](https://github.com/opensearch-project/OpenSearch/pull/20183)) diff --git a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java index 9b3fd09d041c6..b10396f083111 100644 --- a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java +++ b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java @@ -593,6 +593,7 @@ public void apply(Settings value, Settings current, Settings previous) { SearchService.SEARCH_MAX_QUERY_STRING_LENGTH, SearchService.SEARCH_MAX_QUERY_STRING_LENGTH_MONITOR_ONLY, SearchService.CARDINALITY_AGGREGATION_PRUNING_THRESHOLD, + SearchService.TERMS_AGGREGATION_MAX_PRECOMPUTE_CARDINALITY, CardinalityAggregator.CARDINALITY_AGGREGATION_HYBRID_COLLECTOR_ENABLED, CardinalityAggregator.CARDINALITY_AGGREGATION_HYBRID_COLLECTOR_MEMORY_THRESHOLD, SearchService.KEYWORD_INDEX_OR_DOC_VALUES_ENABLED, diff --git a/server/src/main/java/org/opensearch/search/DefaultSearchContext.java b/server/src/main/java/org/opensearch/search/DefaultSearchContext.java index cf55ef0518463..d95321e8b0761 100644 --- a/server/src/main/java/org/opensearch/search/DefaultSearchContext.java +++ b/server/src/main/java/org/opensearch/search/DefaultSearchContext.java @@ -230,6 +230,7 @@ final class DefaultSearchContext extends SearchContext { private final int maxAggRewriteFilters; private final int filterRewriteSegmentThreshold; private final int cardinalityAggregationPruningThreshold; + private final long termsAggregationMaxPrecomputeCardinality; private final CardinalityAggregationContext cardinalityAggregationContext; private final int bucketSelectionStrategyFactor; private final boolean keywordIndexOrDocValuesEnabled; @@ -298,6 +299,7 @@ final class DefaultSearchContext extends SearchContext { this.maxAggRewriteFilters = evaluateFilterRewriteSetting(); this.filterRewriteSegmentThreshold = evaluateAggRewriteFilterSegThreshold(); this.cardinalityAggregationPruningThreshold = evaluateCardinalityAggregationPruningThreshold(); + this.termsAggregationMaxPrecomputeCardinality = evaluateTermsAggregationMaxPrecomputeCardinality(); this.cardinalityAggregationContext = evaluateCardinalityAggregationContext(); this.bucketSelectionStrategyFactor = evaluateBucketSelectionStrategyFactor(); this.concurrentSearchDeciderFactories = concurrentSearchDeciderFactories; @@ -1260,6 +1262,11 @@ public int cardinalityAggregationPruningThreshold() { return cardinalityAggregationPruningThreshold; } + @Override + public long termsAggregationMaxPrecomputeCardinality() { + return termsAggregationMaxPrecomputeCardinality; + } + @Override public CardinalityAggregationContext cardinalityAggregationContext() { return cardinalityAggregationContext; @@ -1282,6 +1289,13 @@ private int evaluateCardinalityAggregationPruningThreshold() { return 0; } + private long evaluateTermsAggregationMaxPrecomputeCardinality() { + if (clusterService != null) { + return clusterService.getClusterSettings().get(SearchService.TERMS_AGGREGATION_MAX_PRECOMPUTE_CARDINALITY); + } + return 30_000L; + } + private CardinalityAggregationContext evaluateCardinalityAggregationContext() { if (clusterService != null) { boolean hybridCollectorEnabled = clusterService.getClusterSettings() diff --git a/server/src/main/java/org/opensearch/search/SearchService.java b/server/src/main/java/org/opensearch/search/SearchService.java index 7b694c24d7194..4a74170f71271 100644 --- a/server/src/main/java/org/opensearch/search/SearchService.java +++ b/server/src/main/java/org/opensearch/search/SearchService.java @@ -442,6 +442,14 @@ public class SearchService extends AbstractLifecycleComponent implements IndexEv Property.NodeScope ); + public static final Setting TERMS_AGGREGATION_MAX_PRECOMPUTE_CARDINALITY = Setting.longSetting( + "search.aggregations.terms.max_precompute_cardinality", + 30_000L, + 0L, + Property.Dynamic, + Property.NodeScope + ); + public static final int DEFAULT_BUCKET_SELECTION_STRATEGY_FACTOR = 5; public static final Setting BUCKET_SELECTION_STRATEGY_FACTOR_SETTING = Setting.intSetting( "search.aggregation.bucket_selection_strategy_factor", diff --git a/server/src/main/java/org/opensearch/search/aggregations/bucket/terms/GlobalOrdinalsStringTermsAggregator.java b/server/src/main/java/org/opensearch/search/aggregations/bucket/terms/GlobalOrdinalsStringTermsAggregator.java index 903155588880a..79cb9e39608fe 100644 --- a/server/src/main/java/org/opensearch/search/aggregations/bucket/terms/GlobalOrdinalsStringTermsAggregator.java +++ b/server/src/main/java/org/opensearch/search/aggregations/bucket/terms/GlobalOrdinalsStringTermsAggregator.java @@ -196,6 +196,11 @@ boolean tryCollectFromTermFrequencies(LeafReaderContext ctx, BiConsumer context.termsAggregationMaxPrecomputeCardinality()) { + return false; + } + NumericDocValues docCountValues = DocValues.getNumeric(ctx.reader(), DocCountFieldMapper.NAME); if (docCountValues.nextDoc() != NO_MORE_DOCS) { // This segment has at least one document with the _doc_count field. diff --git a/server/src/main/java/org/opensearch/search/aggregations/bucket/terms/TermsAggregatorFactory.java b/server/src/main/java/org/opensearch/search/aggregations/bucket/terms/TermsAggregatorFactory.java index 4759edff68168..1f1e3199b72e3 100644 --- a/server/src/main/java/org/opensearch/search/aggregations/bucket/terms/TermsAggregatorFactory.java +++ b/server/src/main/java/org/opensearch/search/aggregations/bucket/terms/TermsAggregatorFactory.java @@ -776,8 +776,9 @@ private boolean shouldDisableStreamingForOrdinals(SearchContext searchContext, W } // Check 2: Match-all query with the majority of docs in clean segments - // Traditional aggregator can use term frequency optimization for these segments - if (isMatchAllQuery(searchContext.query())) { + // and cardinality within the precompute threshold. + // Traditional aggregator can use term frequency optimization for these segments. + if (isMatchAllQuery(searchContext.query()) && maxCardinality <= searchContext.termsAggregationMaxPrecomputeCardinality()) { double cleanRatio = totalDocs > 0 ? (double) docsInCleanSegments / totalDocs : 0; return cleanRatio > 0.8; } 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 61c0b7bcb663b..45b0a7a11cb29 100644 --- a/server/src/main/java/org/opensearch/search/internal/SearchContext.java +++ b/server/src/main/java/org/opensearch/search/internal/SearchContext.java @@ -560,6 +560,11 @@ public int cardinalityAggregationPruningThreshold() { return 0; } + @ExperimentalApi + public long termsAggregationMaxPrecomputeCardinality() { + return 30_000L; + } + public CardinalityAggregationContext cardinalityAggregationContext() { return new CardinalityAggregationContext(false, Runtime.getRuntime().maxMemory() / 100); } diff --git a/server/src/test/java/org/opensearch/search/aggregations/FactoryStreamingCostEstimationTests.java b/server/src/test/java/org/opensearch/search/aggregations/FactoryStreamingCostEstimationTests.java index 0074a106d1cb1..d6dab9cbc315d 100644 --- a/server/src/test/java/org/opensearch/search/aggregations/FactoryStreamingCostEstimationTests.java +++ b/server/src/test/java/org/opensearch/search/aggregations/FactoryStreamingCostEstimationTests.java @@ -667,6 +667,37 @@ public void testTermsFactoryStreamableNonMatchAllQuery() throws IOException { } } + /** + * Test that match-all query with clean segments but cardinality above the precompute threshold IS streamable. + * When cardinality exceeds the threshold, Check 2 is skipped because tryCollectFromTermFrequencies + * would bail out on high cardinality anyway. + */ + public void testTermsFactoryStreamableMatchAllHighCardinality() throws IOException { + try (Directory directory = newDirectory()) { + try (IndexWriter writer = new IndexWriter(directory, new IndexWriterConfig())) { + // Create index with cardinality above the default threshold (30,000) + for (int i = 0; i < 50000; i++) { + Document doc = new Document(); + doc.add(new SortedSetDocValuesField("category", new BytesRef("cat_" + i))); + writer.addDocument(doc); + } + + try (IndexReader reader = DirectoryReader.open(writer)) { + IndexSearcher searcher = newIndexSearcher(reader); + MappedFieldType fieldType = new KeywordFieldMapper.KeywordFieldType("category"); + + TermsAggregationBuilder termsBuilder = new TermsAggregationBuilder("terms").field("category").size(10); + + // Match-all with clean segments but high cardinality - should be streamable + FactoryAndContext result = createAggregatorFactoryWithQuery(termsBuilder, searcher, new MatchAllDocsQuery(), fieldType); + StreamingCostMetrics metrics = ((StreamingCostEstimable) result.factory).estimateStreamingCost(result.searchContext); + + assertTrue("Match-all with high cardinality should be streamable", metrics.streamable()); + } + } + } + } + // ======================================== // Helper methods // ======================================== diff --git a/server/src/test/java/org/opensearch/search/aggregations/bucket/terms/TermsAggregatorTests.java b/server/src/test/java/org/opensearch/search/aggregations/bucket/terms/TermsAggregatorTests.java index 934050b9897bc..4294cd314ed6b 100644 --- a/server/src/test/java/org/opensearch/search/aggregations/bucket/terms/TermsAggregatorTests.java +++ b/server/src/test/java/org/opensearch/search/aggregations/bucket/terms/TermsAggregatorTests.java @@ -329,6 +329,96 @@ public void testSimpleAggregationLowCardinality() throws Exception { testSimple(ADD_SORTED_SET_FIELD_INDEXED, false, true, true, TermsAggregatorFactory.ExecutionMode.GLOBAL_ORDINALS, 4); } + /** + * When termsAggregationMaxPrecomputeCardinality is set to 0, tryCollectFromTermFrequencies should bail out + * even when all other conditions are met (indexed fields, no deletions, no _doc_count). + * This verifies the cardinality guard: with threshold=0, all documents must be visited via normal collection. + */ + public void testTermFrequencyCardinalityGuard() throws Exception { + try (Directory directory = newDirectory()) { + try ( + RandomIndexWriter indexWriter = new RandomIndexWriter( + random(), + directory, + newIndexWriterConfig().setMergePolicy(NoMergePolicy.INSTANCE) + ) + ) { + List documents = new ArrayList<>(); + Document document = new Document(); + ADD_SORTED_SET_FIELD_INDEXED.apply(document, "string", "a"); + ADD_SORTED_SET_FIELD_INDEXED.apply(document, "string", "b"); + documents.add(document); + + document = new Document(); + ADD_SORTED_SET_FIELD_INDEXED.apply(document, "string", ""); + ADD_SORTED_SET_FIELD_INDEXED.apply(document, "string", "c"); + ADD_SORTED_SET_FIELD_INDEXED.apply(document, "string", "a"); + documents.add(document); + + document = new Document(); + ADD_SORTED_SET_FIELD_INDEXED.apply(document, "string", "b"); + ADD_SORTED_SET_FIELD_INDEXED.apply(document, "string", "d"); + documents.add(document); + + document = new Document(); + ADD_SORTED_SET_FIELD_INDEXED.apply(document, "string", ""); + documents.add(document); + + indexWriter.addDocuments(documents); + + try (IndexReader indexReader = maybeWrapReaderEs(indexWriter.getReader())) { + IndexSearcher indexSearcher = newIndexSearcher(indexReader); + + TermsAggregationBuilder aggregationBuilder = new TermsAggregationBuilder("_name").userValueTypeHint(ValueType.STRING) + .executionHint(TermsAggregatorFactory.ExecutionMode.GLOBAL_ORDINALS.toString()) + .field("string") + .order(BucketOrder.key(true)); + MappedFieldType fieldType = new KeywordFieldMapper.KeywordFieldType("string"); + + TermsAggregatorFactory.COLLECT_SEGMENT_ORDS = false; + TermsAggregatorFactory.REMAP_GLOBAL_ORDS = false; + + // Set threshold to 0 so the cardinality guard bails out of tryCollectFromTermFrequencies + CountingAggregator aggregator = new CountingAggregator( + new AtomicInteger(), + createAggregatorWithCustomizableSearchContext( + new MatchAllDocsQuery(), + aggregationBuilder, + indexSearcher, + createIndexSettings(), + new MultiBucketConsumerService.MultiBucketConsumer( + DEFAULT_MAX_BUCKETS, + new NoneCircuitBreakerService().getBreaker(CircuitBreaker.REQUEST) + ), + searchContext -> when(searchContext.termsAggregationMaxPrecomputeCardinality()).thenReturn(0L), + fieldType + ) + ); + + aggregator.preCollection(); + indexSearcher.search(new MatchAllDocsQuery(), aggregator); + aggregator.postCollection(); + Terms result = reduce(aggregator); + assertEquals(5, result.getBuckets().size()); + assertEquals("", result.getBuckets().get(0).getKeyAsString()); + assertEquals(2L, result.getBuckets().get(0).getDocCount()); + assertEquals("a", result.getBuckets().get(1).getKeyAsString()); + assertEquals(2L, result.getBuckets().get(1).getDocCount()); + assertEquals("b", result.getBuckets().get(2).getKeyAsString()); + assertEquals(2L, result.getBuckets().get(2).getDocCount()); + assertEquals("c", result.getBuckets().get(3).getKeyAsString()); + assertEquals(1L, result.getBuckets().get(3).getDocCount()); + assertEquals("d", result.getBuckets().get(4).getKeyAsString()); + assertEquals(1L, result.getBuckets().get(4).getDocCount()); + + // With threshold=0, tryCollectFromTermFrequencies should bail out, + // so all 4 documents must be visited via normal collection + assertEquals(4, aggregator.getCollectCount().get()); + } + } + } + } + /** * This test case utilizes the MapStringTermsAggregator. */ 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 e6d24f2918c2b..f2eabf1b8c453 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 @@ -550,6 +550,7 @@ public boolean shouldCache(Query query) { fieldNameToType.putAll(getFieldAliases(fieldTypes)); when(searchContext.maxAggRewriteFilters()).thenReturn(10_000); + when(searchContext.termsAggregationMaxPrecomputeCardinality()).thenReturn(30_000L); when(searchContext.cardinalityAggregationContext()).thenReturn( new org.opensearch.search.aggregations.metrics.CardinalityAggregationContext(false, Runtime.getRuntime().maxMemory() / 100) );