Disable concurrent search path for composite aggregations. For more d…

…etails see: opensearch-project#12331

Signed-off-by: Sorabh Hamirwasia <[email protected]>

release-notes/opensearch.release-notes-2.12.0.md

@@ -142,6 +142,7 @@
 - [Query Insights] Implement Top N Queries feature to collect and gather information about high latency queries in a window ([#11904](https://github.com/opensearch-project/OpenSearch/pull/11904))
 - Add override support for sampling based on action ([#9621](https://github.com/opensearch-project/OpenSearch/issues/9621))
 - Added custom sampler support based on transport action in request ([#9621](https://github.com/opensearch-project/OpenSearch/issues/9621))
+- Disable concurrent search for composite aggregation([#12375](https://github.com/opensearch-project/OpenSearch/pull/12375))
 
 ### Removed
 - Remove deprecated classes for Rounding ([#10956](https://github.com/opensearch-project/OpenSearch/issues/10956))

server/src/internalClusterTest/java/org/opensearch/search/aggregations/bucket/CompositeAggIT.java

@@ -0,0 +1,99 @@
+/*
+ * 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.search.aggregations.bucket;
+
+import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+
+import org.opensearch.action.search.SearchResponse;
+import org.opensearch.cluster.health.ClusterHealthStatus;
+import org.opensearch.cluster.metadata.IndexMetadata;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.search.aggregations.bucket.composite.CompositeAggregationBuilder;
+import org.opensearch.search.aggregations.bucket.composite.CompositeValuesSourceBuilder;
+import org.opensearch.search.aggregations.bucket.composite.TermsValuesSourceBuilder;
+import org.opensearch.search.aggregations.metrics.MaxAggregationBuilder;
+import org.opensearch.test.OpenSearchIntegTestCase;
+import org.opensearch.test.ParameterizedStaticSettingsOpenSearchIntegTestCase;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+
+import static org.opensearch.search.SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING;
+import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
+import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertSearchResponse;
+
+@OpenSearchIntegTestCase.SuiteScopeTestCase
+public class CompositeAggIT extends ParameterizedStaticSettingsOpenSearchIntegTestCase {
+
+    public CompositeAggIT(Settings staticSettings) {
+        super(staticSettings);
+    }
+
+    @ParametersFactory
+    public static Collection<Object[]> parameters() {
+        return Arrays.asList(
+            new Object[] { Settings.builder().put(CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey(), false).build() },
+            new Object[] { Settings.builder().put(CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey(), true).build() }
+        );
+    }
+
+    @Override
+    public void setupSuiteScopeCluster() throws Exception {
+        assertAcked(
+            prepareCreate(
+                "idx",
+                Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1).put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+            ).setMapping("type", "type=keyword", "num", "type=integer", "score", "type=integer")
+        );
+        waitForRelocation(ClusterHealthStatus.GREEN);
+
+        client().prepareIndex("idx").setId("1").setSource("type", "type1", "num", "1", "score", "5").get();
+        client().prepareIndex("idx").setId("1").setSource("type", "type2", "num", "11", "score", "50").get();
+        refresh("idx");
+        client().prepareIndex("idx").setId("1").setSource("type", "type1", "num", "1", "score", "2").get();
+        client().prepareIndex("idx").setId("1").setSource("type", "type2", "num", "12", "score", "20").get();
+        refresh("idx");
+        client().prepareIndex("idx").setId("1").setSource("type", "type1", "num", "3", "score", "10").get();
+        client().prepareIndex("idx").setId("1").setSource("type", "type2", "num", "13", "score", "15").get();
+        refresh("idx");
+        client().prepareIndex("idx").setId("1").setSource("type", "type1", "num", "3", "score", "1").get();
+        client().prepareIndex("idx").setId("1").setSource("type", "type2", "num", "13", "score", "100").get();
+        refresh("idx");
+
+        waitForRelocation(ClusterHealthStatus.GREEN);
+        refresh();
+    }
+
+    public void testCompositeAggWithNoSubAgg() {
+        SearchResponse rsp = client().prepareSearch("idx")
+            .addAggregation(new CompositeAggregationBuilder("my_composite", getTestValueSources()))
+            .get();
+        assertSearchResponse(rsp);
+    }
+
+    public void testCompositeAggWithSubAgg() {
+        SearchResponse rsp = client().prepareSearch("idx")
+            .addAggregation(
+                new CompositeAggregationBuilder("my_composite", getTestValueSources()).subAggregation(
+                    new MaxAggregationBuilder("max").field("score")
+                )
+            )
+            .get();
+        assertSearchResponse(rsp);
+    }
+
+    private List<CompositeValuesSourceBuilder<?>> getTestValueSources() {
+        final List<CompositeValuesSourceBuilder<?>> sources = new ArrayList<>();
+        sources.add(new TermsValuesSourceBuilder("keyword_vs").field("type"));
+        sources.add(new TermsValuesSourceBuilder("num_vs").field("num"));
+        return sources;
+    }
+}

server/src/main/java/org/opensearch/search/aggregations/bucket/composite/CompositeAggregationFactory.java

@@ -80,6 +80,7 @@ protected Aggregator createInternal(
 
     @Override
     protected boolean supportsConcurrentSegmentSearch() {
-        return true;
+        // See https://github.com/opensearch-project/OpenSearch/issues/12331 for details
+        return false;
     }
 }