diff --git a/CHANGELOG.md b/CHANGELOG.md index 81d29509ac65c..0add5026cb75c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), ## [Unreleased 3.x] ### Added +- Add intra segment support for terms bucket aggregations ([#20829](https://github.com/opensearch-project/OpenSearch/pull/20829)) - Add bitmap64 query support ([#20606](https://github.com/opensearch-project/OpenSearch/pull/20606)) - Add ProfilingWrapper interface for plugin access to delegates in profiling decorators ([#20607](https://github.com/opensearch-project/OpenSearch/pull/20607)) - Support expected cluster name with validation in CCS Sniff mode ([#20532](https://github.com/opensearch-project/OpenSearch/pull/20532)) diff --git a/server/src/internalClusterTest/java/org/opensearch/search/aggregations/bucket/terms/StringTermsIT.java b/server/src/internalClusterTest/java/org/opensearch/search/aggregations/bucket/terms/StringTermsIT.java index f5d018b2ef491..96f8153b0f510 100644 --- a/server/src/internalClusterTest/java/org/opensearch/search/aggregations/bucket/terms/StringTermsIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/search/aggregations/bucket/terms/StringTermsIT.java @@ -31,7 +31,10 @@ package org.opensearch.search.aggregations.bucket.terms; +import com.carrotsearch.randomizedtesting.annotations.ParametersFactory; + import org.opensearch.OpenSearchException; +import org.opensearch.action.index.IndexRequestBuilder; import org.opensearch.action.search.SearchPhaseExecutionException; import org.opensearch.action.search.SearchResponse; import org.opensearch.common.settings.Settings; @@ -57,13 +60,19 @@ import org.opensearch.search.builder.SearchSourceBuilder; import org.opensearch.test.OpenSearchIntegTestCase; +import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.HashSet; import java.util.Iterator; +import java.util.List; import java.util.Set; import static org.opensearch.index.query.QueryBuilders.termQuery; +import static org.opensearch.search.SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING; +import static org.opensearch.search.SearchService.CONCURRENT_SEGMENT_SEARCH_PARTITION_MIN_SEGMENT_SIZE; +import static org.opensearch.search.SearchService.CONCURRENT_SEGMENT_SEARCH_PARTITION_STRATEGY; import static org.opensearch.search.aggregations.AggregationBuilders.avg; import static org.opensearch.search.aggregations.AggregationBuilders.extendedStats; import static org.opensearch.search.aggregations.AggregationBuilders.filter; @@ -85,6 +94,20 @@ public StringTermsIT(Settings staticSettings) { super(staticSettings); } + @ParametersFactory + public static Collection parameters() { + return Arrays.asList( + new Object[] { Settings.builder().put(CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey(), false).build() }, + new Object[] { Settings.builder().put(CONCURRENT_SEGMENT_SEARCH_PARTITION_STRATEGY.getKey(), "segment").build() }, + new Object[] { Settings.builder().put(CONCURRENT_SEGMENT_SEARCH_PARTITION_STRATEGY.getKey(), "force").build() }, + new Object[] { + Settings.builder() + .put(CONCURRENT_SEGMENT_SEARCH_PARTITION_STRATEGY.getKey(), "balanced") + .put(CONCURRENT_SEGMENT_SEARCH_PARTITION_MIN_SEGMENT_SIZE.getKey(), 1000) + .build() } + ); + } + // the main purpose of this test is to make sure we're not allocating 2GB of memory per shard public void testSizeIsZero() { final int minDocCount = randomInt(1); @@ -1067,6 +1090,40 @@ public void testDeferredSubAggs() { } } + public void testConcurrentStringAggregation() throws Exception { + createIndex("test_string_terms", Settings.builder().put("index.number_of_shards", 2).put("index.number_of_replicas", 1).build()); + try { + List builders = new ArrayList<>(5000); + for (int i = 0; i < 5; i++) { + builders.add(client().prepareIndex("test_string_terms").setSource("value", "val" + (i + 1))); + } + indexBulkWithSegments(builders, 2); + indexRandomForConcurrentSearch("test_string_terms"); + SearchResponse response = client().prepareSearch("test_string_terms") + .addAggregation( + terms("values").executionHint(randomExecutionHint()) + .field("value") + .collectMode(randomFrom(SubAggCollectionMode.values())) + ) + .get(); + + assertSearchResponse(response); + Terms values = response.getAggregations().get("values"); + assertThat(values, notNullValue()); + assertThat(values.getName(), equalTo("values")); + assertThat(values.getBuckets().size(), equalTo(5)); + + for (int i = 0; i < 5; i++) { + Terms.Bucket bucket = values.getBucketByKey("val" + (i + 1)); + assertThat(bucket, notNullValue()); + assertThat(key(bucket), equalTo("val" + (i + 1))); + assertThat(bucket.getDocCount(), equalTo(1L)); + } + } finally { + internalCluster().wipeIndices("test_string_terms"); + } + } + /** * Make sure that a request using a deterministic script or not using a script get cached. * Ensure requests using nondeterministic scripts do not get cached. 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 1f1e3199b72e3..88516c546a567 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 @@ -797,4 +797,9 @@ private static boolean isMatchAllQuery(Query query) { protected boolean supportsConcurrentSegmentSearch() { return true; } + + @Override + protected boolean supportsIntraSegmentSearch() { + return true; + } } 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 4294cd314ed6b..1bb50aa5f83ab 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 @@ -96,6 +96,7 @@ import org.opensearch.search.aggregations.AggregationBuilders; import org.opensearch.search.aggregations.AggregationExecutionException; import org.opensearch.search.aggregations.Aggregator; +import org.opensearch.search.aggregations.AggregatorFactories; import org.opensearch.search.aggregations.AggregatorTestCase; import org.opensearch.search.aggregations.BucketOrder; import org.opensearch.search.aggregations.InternalAggregation; @@ -1807,6 +1808,24 @@ public void testStringTermAggregatorForResultSelectionStrategy() throws IOExcept } } + public void testStringTermAggregatorWithIntrasegmentSearch() throws IOException { + MappedFieldType fieldtype = new KeywordFieldMapper.KeywordFieldType("value"); + try (Directory directory = newDirectory(); RandomIndexWriter indexWriter = new RandomIndexWriter(random(), directory)) { + indexWriter.addDocument(singleton(new StringField("value", "1", Field.Store.NO))); + try (IndexReader reader = indexWriter.getReader()) { + IndexSearcher searcher = newIndexSearcher(reader); + AggregatorFactories factories = AggregatorFactories.builder() + .addAggregator(new TermsAggregationBuilder("test").field("value")) + .build( + createSearchContext(searcher, createIndexSettings(), new MatchAllDocsQuery(), null, fieldtype) + .getQueryShardContext(), + null + ); + assertTrue(factories.allFactoriesSupportIntraSegmentSearch()); + } + } + } + private GlobalOrdinalsStringTermsAggregator createAndTestAggregator( IndexSearcher indexSearcher, MappedFieldType stringFieldType,