Allow aggs to disable offloading sequential collection

server/src/main/java/org/elasticsearch/search/aggregations/AggregationBuilder.java

@@ -220,15 +220,39 @@ public boolean isInSortOrderExecutionRequired() {
     }
 
     /**
-     * Return false if this aggregation or any of the child aggregations does not support concurrent search
+     * Return false if this aggregation or any of the child aggregations does not support concurrent search.
+     * As a result, such aggregation will always be executed sequentially despite concurrency is enabled for the query phase.
+     * Note: aggregations that don't support concurrency, may or may not support offloading their collection to the search worker threads,
+     * depending on what {@link #supportsOffloadingSequentialCollection()} returns.
      */
     public boolean supportsConcurrentExecution() {
+        if (isInSortOrderExecutionRequired()) {
+            return false;
+        }
         for (AggregationBuilder builder : factoriesBuilder.getAggregatorFactories()) {
             if (builder.supportsConcurrentExecution() == false) {
                 return false;
             }
         }
-        return isInSortOrderExecutionRequired() == false;
+        return supportsOffloadingSequentialCollection();
+    }
+
+    /**
+     * Returns false if this aggregation or any of its child aggregations does not support offloading its sequential collection
+     * to a separate thread. As a result, such aggregation will always be executed sequentially, and fully in the search thread,
+     * without offloading its collection to the search worker threads.
+     * Note: aggregations that don't support offloading sequential collection, don't support concurrency by definition.
+     */
+    public boolean supportsOffloadingSequentialCollection() {
+        if (isInSortOrderExecutionRequired()) {
+            return false;
+        }
+        for (AggregationBuilder builder : factoriesBuilder.getAggregatorFactories()) {
+            if (builder.supportsOffloadingSequentialCollection() == false) {
+                return false;
+            }
+        }
+        return true;
     }
 
     /**

server/src/main/java/org/elasticsearch/search/aggregations/bucket/composite/CompositeAggregationBuilder.java

@@ -97,7 +97,7 @@ public boolean supportsSampling() {
     }
 
     @Override
-    public boolean supportsConcurrentExecution() {
+    public boolean supportsOffloadingSequentialCollection() {
         return false;
     }
 

server/src/main/java/org/elasticsearch/search/aggregations/bucket/nested/NestedAggregationBuilder.java

@@ -81,7 +81,7 @@ public BucketCardinality bucketCardinality() {
     }
 
     @Override
-    public boolean supportsConcurrentExecution() {
+    public boolean supportsOffloadingSequentialCollection() {
         return false;
     }
 

server/src/main/java/org/elasticsearch/search/aggregations/metrics/CardinalityAggregationBuilder.java

@@ -113,7 +113,7 @@ public boolean supportsSampling() {
     }
 
     @Override
-    public boolean supportsConcurrentExecution() {
+    public boolean supportsOffloadingSequentialCollection() {
         return false;
     }
 

server/src/test/java/org/elasticsearch/search/aggregations/metrics/CardinalityAggregatorTests.java

@@ -89,8 +89,6 @@ public class CardinalityAggregatorTests extends AggregatorTestCase {
     /** Script to extract a collection of numeric values from the 'numbers' field **/
     public static final String NUMERIC_VALUES_SCRIPT = "doc['numbers']";
 
-    public static final int HASHER_DEFAULT_SEED = 17;
-
     @Override
     protected ScriptService getMockScriptService() {
         final Map<String, Function<Map<String, Object>, Object>> scripts = new HashMap<>();

test/framework/src/main/java/org/elasticsearch/search/aggregations/BaseAggregationTestCase.java

@@ -61,10 +61,25 @@ public void testFromXContent() throws IOException {
     public void testSupportsConcurrentExecution() {
         AB builder = createTestAggregatorBuilder();
         boolean supportsConcurrency = builder.supportsConcurrentExecution();
+        if (supportsConcurrency) {
+            assertTrue(builder.supportsOffloadingSequentialCollection());
+        }
         AggregationBuilder bucketBuilder = new HistogramAggregationBuilder("test");
-        assertThat(bucketBuilder.supportsConcurrentExecution(), equalTo(true));
+        assertTrue(bucketBuilder.supportsConcurrentExecution());
         bucketBuilder.subAggregation(builder);
         assertThat(bucketBuilder.supportsConcurrentExecution(), equalTo(supportsConcurrency));
+        if (bucketBuilder.supportsConcurrentExecution()) {
+            assertTrue(bucketBuilder.supportsOffloadingSequentialCollection());
+        }
+    }
+
+    public void testSupportsOffloadingSequentialCollection() {
+        AB builder = createTestAggregatorBuilder();
+        boolean supportsOffloadingSequentialCollection = builder.supportsOffloadingSequentialCollection();
+        AggregationBuilder bucketBuilder = new HistogramAggregationBuilder("test");
+        assertTrue(bucketBuilder.supportsOffloadingSequentialCollection());
+        bucketBuilder.subAggregation(builder);
+        assertThat(bucketBuilder.supportsOffloadingSequentialCollection(), equalTo(supportsOffloadingSequentialCollection));
     }
 
     /**