Allow nested fields in the composite aggregation (#37178)

This changes adds the support to handle `nested` fields in the `composite`
aggregation. A `nested` aggregation can be used as parent of a `composite`
aggregation in order to target `nested` fields in the `sources`.

Closes #28611

docs/reference/aggregations/bucket/composite-aggregation.asciidoc

@@ -31,7 +31,24 @@ PUT /sales
       },
       "shop": {
           "type": "keyword"
-      }
+      },
+      "nested": {
+          "type": "nested",
+          "properties": {
+            "product": {
+                "type": "keyword"
+            },
+            "timestamp": {
+                "type": "date"
+            },
+            "price": {
+                "type": "long"
+            },
+            "shop": {
+                "type": "keyword"
+            }
+          }
+       }
     }
   }
 }
@@ -287,7 +304,6 @@ GET /_search
 --------------------------------------------------
 // CONSOLE
 
-
 This will create composite buckets from the values created by two values source, a `date_histogram` and a `terms`.
 Each bucket is composed of two values, one for each value source defined in the aggregation.
 Any type of combinations is allowed and the order in the array is preserved

rest-api-spec/src/main/resources/rest-api-spec/test/search.aggregation/230_composite.yml

@@ -13,13 +13,18 @@ setup:
                       type: keyword
                     long:
                       type: long
+                    nested:
+                      type: nested
+                      properties:
+                        nested_long:
+                          type: long
 
   - do:
       index:
         index: test
         type:  doc
         id:    1
-        body:  { "keyword": "foo", "long": [10, 20] }
+        body:  { "keyword": "foo", "long": [10, 20], "nested": [{"nested_long": 10}, {"nested_long": 20}] }
 
   - do:
       index:
@@ -33,14 +38,14 @@ setup:
         index: test
         type:  doc
         id:    3
-        body:  { "keyword": "bar", "long": [100, 0] }
+        body:  { "keyword": "bar", "long": [100, 0], "nested": [{"nested_long": 10}, {"nested_long": 0}] }
 
   - do:
       index:
         index: test
         type:  doc
         id:    4
-        body:  { "keyword": "bar", "long": [1000, 0] }
+        body:  { "keyword": "bar", "long": [1000, 0], "nested": [{"nested_long": 1000}, {"nested_long": 20}] }
 
   - do:
       index:
@@ -66,7 +71,6 @@ setup:
       version: " - 6.0.99"
       reason:  this uses a new API that has been added in 6.1
 
-
   - do:
       search:
         rest_total_hits_as_int: true
@@ -357,3 +361,68 @@ setup:
                       }
                     }
                   ]
+
+---
+"Composite aggregation with nested parent":
+  - skip:
+      version: " - 6.99.99"
+      reason:  the ability to set a nested parent aggregation was added in 7.0.
+
+  - do:
+        search:
+          rest_total_hits_as_int: true
+          index: test
+          body:
+            aggregations:
+              1:
+                nested:
+                  path: nested
+                aggs:
+                  2:
+                    composite:
+                      sources: [
+                        "nested": {
+                          "terms": {
+                            "field": "nested.nested_long"
+                        }
+                      }
+                    ]
+
+  - match: {hits.total: 6}
+  - length: { aggregations.1.2.buckets: 4 }
+  - match: { aggregations.1.2.buckets.0.key.nested: 0 }
+  - match: { aggregations.1.2.buckets.0.doc_count:  1 }
+  - match: { aggregations.1.2.buckets.1.key.nested: 10 }
+  - match: { aggregations.1.2.buckets.1.doc_count:  2 }
+  - match: { aggregations.1.2.buckets.2.key.nested: 20 }
+  - match: { aggregations.1.2.buckets.2.doc_count:  2 }
+  - match: { aggregations.1.2.buckets.3.key.nested: 1000 }
+  - match: { aggregations.1.2.buckets.3.doc_count:  1 }
+
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: test
+        body:
+          aggregations:
+            1:
+              nested:
+                path: nested
+              aggs:
+                2:
+                  composite:
+                    after: { "nested": 10 }
+                    sources: [
+                      "nested": {
+                        "terms": {
+                          "field": "nested.nested_long"
+                        }
+                      }
+                    ]
+
+  - match: {hits.total: 6}
+  - length: { aggregations.1.2.buckets: 2 }
+  - match: { aggregations.1.2.buckets.0.key.nested: 20 }
+  - match: { aggregations.1.2.buckets.0.doc_count:  2 }
+  - match: { aggregations.1.2.buckets.1.key.nested: 1000 }
+  - match: { aggregations.1.2.buckets.1.doc_count:  1 }

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

@@ -29,6 +29,7 @@ import org.elasticsearch.search.aggregations.AbstractAggregationBuilder;
 import org.elasticsearch.search.aggregations.AggregationBuilder;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
+import org.elasticsearch.search.aggregations.bucket.nested.NestedAggregatorFactory;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -151,11 +152,28 @@ public class CompositeAggregationBuilder extends AbstractAggregationBuilder<Comp
         return size;
     }
 
+    /**
+     * Returns null if the provided factory and his parents are compatible with
+     * this aggregator or the instance of the parent's factory that is incompatible with
+     * the composite aggregation.
+     */
+    private AggregatorFactory<?> checkParentIsNullOrNested(AggregatorFactory<?> factory) {
+        if (factory == null) {
+            return null;
+        } else if (factory instanceof NestedAggregatorFactory) {
+            return checkParentIsNullOrNested(factory.getParent());
+        } else {
+            return factory;
+        }
+    }
+
     @Override
     protected AggregatorFactory<?> doBuild(SearchContext context, AggregatorFactory<?> parent,
                                            AggregatorFactories.Builder subfactoriesBuilder) throws IOException {
-        if (parent != null) {
-            throw new IllegalArgumentException("[composite] aggregation cannot be used with a parent aggregation");
+        AggregatorFactory<?> invalid = checkParentIsNullOrNested(parent);
+        if (invalid != null) {
+            throw new IllegalArgumentException("[composite] aggregation cannot be used with a parent aggregation of" +
+                " type: [" + invalid.getClass().getSimpleName() + "]");
         }
         CompositeValuesSourceConfig[] configs = new CompositeValuesSourceConfig[sources.size()];
         for (int i = 0; i < configs.length; i++) {

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

@@ -32,7 +32,7 @@ import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 
-class NestedAggregatorFactory extends AggregatorFactory<NestedAggregatorFactory> {
+public class NestedAggregatorFactory extends AggregatorFactory<NestedAggregatorFactory> {
 
     private final ObjectMapper parentObjectMapper;
     private final ObjectMapper childObjectMapper;