Reject invalid reverse_nested aggs

docs/changelog/137047.yaml

@@ -0,0 +1,5 @@
+pr: 137047
+summary: Reject invalid `reverse_nested` aggs
+area: Aggregations
+type: bug
+issues: []

modules/aggregations/src/yamlRestTest/resources/rest-api-spec/test/aggregations/nested.yml

@@ -194,3 +194,33 @@ setup:
   - match: { aggregations.courses.highpass_filter.unnest.department.buckets.0.doc_count: 1 }
   - match: { aggregations.courses.highpass_filter.unnest.department.buckets.1.key: math }
   - match: { aggregations.courses.highpass_filter.unnest.department.buckets.1.doc_count: 1 }
+---
+"Illegal reverse nested aggregation to a child nested object":
+  - requires:
+        capabilities:
+          - method: POST
+            path: /_search
+            capabilities: [ reject_invalid_reverse_nesting ]
+        test_runner_features: [ capabilities ]
+        reason: "search does not yet reject invalid reverse nesting paths"
+  - do:
+      catch: /Reverse nested path \[courses.sessions\] is not a parent of the current nested scope \[courses\]/
+      search:
+        index: test
+        body:
+          {
+            "aggs": {
+              "parent_nested": {
+                "nested": {
+                  "path": "courses"
+                },
+                "aggs": {
+                  "invalid_reverse_nested": {
+                    "reverse_nested": {
+                      "path": "courses.sessions"
+                    }
+                  }
+                }
+              }
+            }
+          }

server/src/main/java/org/elasticsearch/rest/action/search/SearchCapabilities.java

@@ -60,6 +60,7 @@ private SearchCapabilities() {}
     private static final String BUCKET_SCRIPT_PARENT_MULTI_BUCKET_ERROR = "bucket_script_parent_multi_bucket_error";
     private static final String EXCLUDE_SOURCE_VECTORS_SETTING = "exclude_source_vectors_setting";
     private static final String CLUSTER_STATS_EXTENDED_USAGE = "extended-search-usage-stats";
+    private static final String REJECT_INVALID_REVERSE_NESTING = "reject_invalid_reverse_nesting";
 
     public static final Set<String> CAPABILITIES;
     static {
@@ -90,6 +91,7 @@ private SearchCapabilities() {}
         capabilities.add(BUCKET_SCRIPT_PARENT_MULTI_BUCKET_ERROR);
         capabilities.add(EXCLUDE_SOURCE_VECTORS_SETTING);
         capabilities.add(CLUSTER_STATS_EXTENDED_USAGE);
+        capabilities.add(REJECT_INVALID_REVERSE_NESTING);
         CAPABILITIES = Set.copyOf(capabilities);
     }
 }

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

@@ -84,6 +84,18 @@ protected AggregatorFactory doBuild(AggregationContext context, AggregatorFactor
             throw new IllegalArgumentException("Reverse nested aggregation [" + name + "] can only be used inside a [nested] aggregation");
         }
 
+        if (path != null) {
+            NestedObjectMapper currentParent = context.nestedScope().getObjectMapper();
+            if (currentParent != null) {
+                String parentPath = currentParent.fullPath();
+                if (parentPath.equals(path) == false && parentPath.startsWith(path + ".") == false) {
+                    throw new IllegalArgumentException(
+                        "Reverse nested path [" + path + "] is not a parent of the current nested scope [" + parentPath + "]"
+                    );
+                }
+            }
+        }
+
         NestedObjectMapper nestedMapper = null;
         if (path != null) {
             nestedMapper = context.nestedLookup().getNestedMappers().get(path);

server/src/test/java/org/elasticsearch/search/aggregations/bucket/nested/ReverseNestedAggregatorTests.java

@@ -247,4 +247,29 @@ public void testNestedUnderTerms() throws IOException {
     protected List<ObjectMapper> objectMappers() {
         return NestedAggregatorTests.MOCK_OBJECT_MAPPERS;
     }
+
+    public void testErrorOnInvalidReverseNestedWithPath() throws IOException {
+        AggregationBuilder aggregationBuilder = nested("n1", NESTED_OBJECT).subAggregation(
+            nested("n2", NESTED_OBJECT + ".child").subAggregation(
+                reverseNested("r1").path(NESTED_OBJECT).subAggregation(reverseNested("r2").path(NESTED_OBJECT + ".child"))
+            )
+        );
+
+        try (Directory directory = newDirectory()) {
+            try (RandomIndexWriter iw = newRandomIndexWriterWithLogDocMergePolicy(directory)) {
+                // No docs needed
+            }
+            try (DirectoryReader indexReader = wrapInMockESDirectoryReader(DirectoryReader.open(directory))) {
+                IllegalArgumentException e = assertThrows(
+                    IllegalArgumentException.class,
+                    () -> searchAndReduce(indexReader, new AggTestConfig(aggregationBuilder))
+                );
+                assertThat(
+                    e.getMessage(),
+                    equalTo("Reverse nested path [nested_object.child] is not a parent of the current nested scope [nested_object]")
+                );
+
+            }
+        }
+    }
 }