The `nested` aggregator should also resolve and use the parentFilter of the closest `reverse_nested` aggregator.

Closes #6994
Closes #7048

src/main/java/org/elasticsearch/search/aggregations/bucket/nested/NestedAggregator.java

@@ -71,16 +71,13 @@ public class NestedAggregator extends SingleBucketAggregator implements ReaderCo
     @Override
     public void setNextReader(AtomicReaderContext reader) {
         if (parentFilter == null) {
-            NestedAggregator closestNestedAggregator = findClosestNestedAggregator(parentAggregator);
-            final Filter parentFilterNotCached;
-            if (closestNestedAggregator == null) {
-                parentFilterNotCached = NonNestedDocsFilter.INSTANCE;
-            } else {
             // The aggs are instantiated in reverse, first the most inner nested aggs and lastly the top level aggs
             // So at the time a nested 'nested' aggs is parsed its closest parent nested aggs hasn't been constructed.
             // So the trick to set at the last moment just before needed and we can use its child filter as the
             // parent filter.
-                parentFilterNotCached = closestNestedAggregator.childFilter;
+            Filter parentFilterNotCached = findClosestNestedPath(parentAggregator);
+            if (parentFilterNotCached == null) {
+                parentFilterNotCached = NonNestedDocsFilter.INSTANCE;
             }
             parentFilter = SearchContext.current().filterCache().cache(parentFilterNotCached);
             // if the filter cache is disabled, we still need to produce bit sets
@@ -103,10 +100,8 @@ public class NestedAggregator extends SingleBucketAggregator implements ReaderCo
 
     @Override
     public void collect(int parentDoc, long bucketOrd) throws IOException {
-
         // here we translate the parent doc to a list of its nested docs, and then call super.collect for evey one of them
         // so they'll be collected
-
         if (parentDoc == 0 || parentDocs == null) {
             return;
         }
@@ -135,10 +130,12 @@ public class NestedAggregator extends SingleBucketAggregator implements ReaderCo
         return nestedPath;
     }
 
-    static NestedAggregator findClosestNestedAggregator(Aggregator parent) {
+    private static Filter findClosestNestedPath(Aggregator parent) {
         for (; parent != null; parent = parent.parent()) {
             if (parent instanceof NestedAggregator) {
-                return (NestedAggregator) parent;
+                return ((NestedAggregator) parent).childFilter;
+            } else if (parent instanceof ReverseNestedAggregator) {
+                return ((ReverseNestedAggregator) parent).getParentFilter();
             }
         }
         return null;

src/main/java/org/elasticsearch/search/aggregations/bucket/nested/ReverseNestedAggregator.java

@@ -38,8 +38,6 @@ import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
 
-import static org.elasticsearch.search.aggregations.bucket.nested.NestedAggregator.findClosestNestedAggregator;
-
 /**
  *
  */
@@ -128,6 +126,14 @@ public class ReverseNestedAggregator extends SingleBucketAggregator implements R
         collectBucket(parentDoc, bucketOrd);
     }
 
+    private static NestedAggregator findClosestNestedAggregator(Aggregator parent) {
+        for (; parent != null; parent = parent.parent()) {
+            if (parent instanceof NestedAggregator) {
+                return (NestedAggregator) parent;
+            }
+        }
+        return null;
+    }
 
     @Override
     public InternalAggregation buildAggregation(long owningBucketOrdinal) {
@@ -139,6 +145,10 @@ public class ReverseNestedAggregator extends SingleBucketAggregator implements R
         return new InternalReverseNested(name, 0, buildEmptySubAggregations());
     }
 
+    Filter getParentFilter() {
+        return parentFilter;
+    }
+
     @Override
     protected void doClose() {
         Releasables.close(bucketOrdToLastCollectedParentDocRecycler);

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

@@ -129,6 +129,29 @@ public class ReverseNestedTests extends ElasticsearchIntegrationTest {
         verifyResults(response);
     }
 
+    @Test
+    public void simple_nested1ToRootToNested2() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type2")
+                .addAggregation(nested("nested1").path("nested1")
+                                .subAggregation(
+                                        reverseNested("nested1_to_root")
+                                                .subAggregation(nested("root_to_nested2").path("nested1.nested2"))
+                                        )
+                                )
+                .get();
+
+        assertSearchResponse(response);
+        Nested nested = response.getAggregations().get("nested1");
+        assertThat(nested.getName(), equalTo("nested1"));
+        assertThat(nested.getDocCount(), equalTo(9l));
+        ReverseNested reverseNested = nested.getAggregations().get("nested1_to_root");
+        assertThat(reverseNested.getName(), equalTo("nested1_to_root"));
+        assertThat(reverseNested.getDocCount(), equalTo(9l));
+        nested = reverseNested.getAggregations().get("root_to_nested2");
+        assertThat(nested.getName(), equalTo("root_to_nested2"));
+        assertThat(nested.getDocCount(), equalTo(25l));
+    }
+
     @Test
     public void simple_reverseNestedToNested1() throws Exception {
         SearchResponse response = client().prepareSearch("idx")