aggs: Fix 2 bug in children agg

1) multiple nested children aggs result in a NPE
2) fixed a counting bug where the same readers where post collected twice

Closes #10158

src/main/java/org/elasticsearch/search/aggregations/Aggregator.java

@@ -331,8 +331,8 @@ public BucketAggregationMode bucketAggregationMode() {
      * Called after collection of all document is done.
      */
     public final void postCollection() throws IOException {
-        collectableSubAggregators.postCollection();
         doPostCollection();
+        collectableSubAggregators.postCollection();
     }
 
     /** Called upon release of the aggregator. */

src/main/java/org/elasticsearch/search/aggregations/bucket/children/ParentToChildrenAggregator.java

@@ -43,9 +43,9 @@
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 
 import java.io.IOException;
-import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.List;
+import java.util.LinkedHashSet;
+import java.util.Set;
 
 // The RecordingPerReaderBucketCollector assumes per segment recording which isn't the case for this
 // aggregation, for this reason that collector can't be used
@@ -66,7 +66,8 @@ public class ParentToChildrenAggregator extends SingleBucketAggregator implement
     private final LongObjectPagedHashMap<long[]> parentOrdToOtherBuckets;
     private boolean multipleBucketsPerParentOrd = false;
 
-    private List<AtomicReaderContext> replay = new ArrayList<>();
+    // This needs to be a Set to avoid duplicate reader context entries (#setNextReader(...) can get invoked multiple times with the same reader context)
+    private Set<AtomicReaderContext> replay = new LinkedHashSet<>();
     private SortedDocValues globalOrdinals;
     private Bits parentDocs;
 
@@ -143,7 +144,7 @@ public void setNextReader(AtomicReaderContext reader) {
 
     @Override
     protected void doPostCollection() throws IOException {
-        List<AtomicReaderContext> replay = this.replay;
+        Set<AtomicReaderContext> replay = this.replay;
         this.replay = null;
 
         for (AtomicReaderContext atomicReaderContext : replay) {
@@ -180,10 +181,6 @@ protected void doPostCollection() throws IOException {
                 }
             }
         }
-        // Need to invoke post collection on all aggs that the children agg is wrapping,
-        // otherwise any post work that is required, because we started to collect buckets
-        // in the method will not be performed.
-        collectableSubAggregators.postCollection();
     }
 
     @Override

src/test/java/org/elasticsearch/search/aggregations/bucket/ChildrenTests.java

@@ -21,6 +21,8 @@
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.update.UpdateResponse;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.common.settings.ImmutableSettings;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.aggregations.bucket.children.Children;
 import org.elasticsearch.search.aggregations.bucket.terms.Terms;
@@ -329,6 +331,53 @@ public void testPostCollection() throws Exception {
         assertThat(termsAgg.getBucketByKey("44").getDocCount(), equalTo(1l));
     }
 
+    @Test
+    public void testHierarchicalChildrenAggs() {
+        String indexName = "geo";
+        String grandParentType = "continent";
+        String parentType = "country";
+        String childType = "city";
+        assertAcked(
+                prepareCreate(indexName)
+                        .setSettings(ImmutableSettings.builder()
+                                .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
+                                .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0)
+                        )
+                        .addMapping(grandParentType)
+                        .addMapping(parentType, "_parent", "type=" + grandParentType)
+                        .addMapping(childType, "_parent", "type=" + parentType)
+        );
+
+        client().prepareIndex(indexName, grandParentType, "1").setSource("name", "europe").get();
+        client().prepareIndex(indexName, parentType, "2").setParent("1").setSource("name", "belgium").get();
+        client().prepareIndex(indexName, childType, "3").setParent("2").setRouting("1").setSource("name", "brussels").get();
+        refresh();
+
+        SearchResponse response = client().prepareSearch(indexName)
+                .setQuery(matchQuery("name", "europe"))
+                .addAggregation(
+                        children(parentType).childType(parentType).subAggregation(
+                                children(childType).childType(childType).subAggregation(
+                                        terms("name").field("name")
+                                )
+                        )
+                )
+                .get();
+        assertNoFailures(response);
+        assertHitCount(response, 1);
+
+        Children children = response.getAggregations().get(parentType);
+        assertThat(children.getName(), equalTo(parentType));
+        assertThat(children.getDocCount(), equalTo(1l));
+        children = children.getAggregations().get(childType);
+        assertThat(children.getName(), equalTo(childType));
+        assertThat(children.getDocCount(), equalTo(1l));
+        Terms terms = children.getAggregations().get("name");
+        assertThat(terms.getBuckets().size(), equalTo(1));
+        assertThat(terms.getBuckets().get(0).getKey().toString(), equalTo("brussels"));
+        assertThat(terms.getBuckets().get(0).getDocCount(), equalTo(1l));
+    }
+
     private static final class Control {
 
         final String category;