Fix nested aggregation top_hits with query inner_hits

docs/changelog/137351.yaml

@@ -0,0 +1,6 @@
+pr: 137351
+summary: Fix nested aggregation `top_hits` with query `inner_hits`
+area: Search
+type: bug
+issues:
+ - 136893

modules/aggregations/src/main/java/module-info.java

@@ -23,4 +23,5 @@
     opens org.elasticsearch.aggregations to org.elasticsearch.painless.spi; // whitelist resource access
 
     provides org.elasticsearch.painless.spi.PainlessExtension with org.elasticsearch.aggregations.AggregationsPainlessExtension;
+    provides org.elasticsearch.features.FeatureSpecification with org.elasticsearch.aggregations.AggregationsFeatures;
 }

modules/aggregations/src/main/java/org/elasticsearch/aggregations/AggregationsFeatures.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.aggregations;
+
+import org.elasticsearch.features.FeatureSpecification;
+import org.elasticsearch.features.NodeFeature;
+
+import java.util.Set;
+
+public final class AggregationsFeatures implements FeatureSpecification {
+
+    public static final NodeFeature NESTED_AGG_TOP_HITS_WITH_INNER_HITS = new NodeFeature("nested_agg_top_hits_with_inner_hits");
+
+    @Override
+    public Set<NodeFeature> getFeatures() {
+        return Set.of(NESTED_AGG_TOP_HITS_WITH_INNER_HITS);
+    }
+}

modules/aggregations/src/main/resources/META-INF/services/org.elasticsearch.features.FeatureSpecification

@@ -0,0 +1,10 @@
+#
+# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+# or more contributor license agreements. Licensed under the "Elastic License
+# 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+# Public License v 1"; you may not use this file except in compliance with, at
+# your election, the "Elastic License 2.0", the "GNU Affero General Public
+# License v3.0 only", or the "Server Side Public License, v 1".
+#
+
+org.elasticsearch.aggregations.AggregationsFeatures

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

@@ -157,3 +157,70 @@ setup:
   - match: { aggregations.groups.buckets.0.users.hits.hits.1._index: my-index }
   - gte: { aggregations.groups.buckets.0.users.hits.hits.1._seq_no: 0 }
   - gte: { aggregations.groups.buckets.0.users.hits.hits.1._primary_term: 1 }
+
+---
+"nested top_hits aggregation with query inner_hits":
+  - requires:
+      cluster_features: ["nested_agg_top_hits_with_inner_hits"]
+      reason: "Fix for nested aggregation top_hits with query inner_hits"
+  - do:
+      indices.create:
+        index: nested-inner-hits-test
+        body:
+          settings:
+            number_of_shards: 1
+          mappings:
+            properties:
+              nested_item:
+                type: nested
+                properties:
+                  title:
+                    type: text
+                  content:
+                    type: text
+                  number:
+                    type: long
+
+  - do:
+      index:
+        index: nested-inner-hits-test
+        id: "1"
+        refresh: true
+        body:
+          nested_item:
+            - title: "foo"
+            - title: "foobar"
+              content: "foobar foobar"
+            - title: "baz"
+              content: "foobar"
+              number: 2
+
+  - do:
+      search:
+        index: nested-inner-hits-test
+        rest_total_hits_as_int: true
+        body:
+          query:
+            nested:
+              path: nested_item
+              inner_hits: {}
+              query:
+                match_all: {}
+          aggs:
+            nested_item:
+              nested:
+                path: nested_item
+              aggs:
+                top_nested:
+                  top_hits: {}
+
+  - match: { hits.total: 1 }
+  - length: { hits.hits.0.inner_hits.nested_item.hits.hits: 3 }
+  - match: { aggregations.nested_item.top_nested.hits.total: 3 }
+  - length: { aggregations.nested_item.top_nested.hits.hits: 3 }
+  - match: { aggregations.nested_item.top_nested.hits.hits.0._nested.field: nested_item }
+  - match: { aggregations.nested_item.top_nested.hits.hits.0._nested.offset: 0 }
+  - match: { aggregations.nested_item.top_nested.hits.hits.1._nested.field: nested_item }
+  - match: { aggregations.nested_item.top_nested.hits.hits.1._nested.offset: 1 }
+  - match: { aggregations.nested_item.top_nested.hits.hits.2._nested.field: nested_item }
+  - match: { aggregations.nested_item.top_nested.hits.hits.2._nested.offset: 2 }

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

@@ -198,7 +198,7 @@ public InternalAggregation buildAggregation(long owningBucketOrdinal) throws IOE
         for (int i = 0; i < topDocs.scoreDocs.length; i++) {
             docIdsToLoad[i] = topDocs.scoreDocs[i].doc;
         }
-        FetchSearchResult fetchResult = runFetchPhase(subSearchContext, docIdsToLoad);
+        FetchSearchResult fetchResult = runFetchPhase(docIdsToLoad);
         if (fetchProfiles != null) {
             fetchProfiles.add(fetchResult.profileResult());
         }
@@ -222,13 +222,20 @@ public InternalAggregation buildAggregation(long owningBucketOrdinal) throws IOE
         );
     }
 
-    private static FetchSearchResult runFetchPhase(SubSearchContext subSearchContext, int[] docIdsToLoad) {
+    private FetchSearchResult runFetchPhase(int[] docIdsToLoad) {
         // Fork the search execution context for each slice, because the fetch phase does not support concurrent execution yet.
         SearchExecutionContext searchExecutionContext = new SearchExecutionContext(subSearchContext.getSearchExecutionContext());
-        // InnerHitSubContext is not thread-safe, so we fork it as well to support concurrent execution
-        InnerHitsContext innerHitsContext = new InnerHitsContext(
-            getForkedInnerHits(subSearchContext.innerHits().getInnerHits(), searchExecutionContext)
-        );
+
+        // When top_hits aggregation is inside a nested aggregation, do not inherit inner_hits from parent query.
+        // Inheriting inner_hits means get nested documents from already nested documents which is incorrect
+        InnerHitsContext innerHitsContext;
+        if (isNestedAggregationContext()) {
+            innerHitsContext = new InnerHitsContext();
+        } else {
+            innerHitsContext = new InnerHitsContext(
+                getForkedInnerHits(subSearchContext.innerHits().getInnerHits(), searchExecutionContext)
+            );
+        }
 
         SubSearchContext fetchSubSearchContext = new SubSearchContext(subSearchContext) {
             @Override
@@ -246,6 +253,21 @@ public InnerHitsContext innerHits() {
         return fetchSubSearchContext.fetchResult();
     }
 
+    /**
+     * Check if this top_hits aggregator is a child of a nested aggregator.
+     * If true, we should not inherit inner_hits from parent query context.
+     */
+    private boolean isNestedAggregationContext() {
+        Aggregator current = parent;
+        while (current != null) {
+            if (current instanceof org.elasticsearch.search.aggregations.bucket.nested.NestedAggregator) {
+                return true;
+            }
+            current = current.parent();
+        }
+        return false;
+    }
+
     private static Map<String, InnerHitSubContext> getForkedInnerHits(
         Map<String, InnerHitSubContext> originalInnerHits,
         SearchExecutionContext searchExecutionContext