Disable concurrency when top_hits sorts on anything but _score

We already disable inter-segment concurrency in SearchSourceBuilder whenever
the top-level sort provided is not _score. We shoudl apply the same rules
in top_hits. We recenly stumbled upon non deterministic behaviour caused by
script sorting defined within top hits. That is to be expected given that
script sorting does not support search concurrency.

The sort script can be replaced with a runtime field, either defined in the
mapping or in the search request, which does support concurrency and guarantees
predictable behaviour.

Author: javanna

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

@@ -49,6 +49,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.function.ToLongFunction;
 
 public class TopHitsAggregationBuilder extends AbstractAggregationBuilder<TopHitsAggregationBuilder> {
     public static final String NAME = "top_hits";
@@ -822,4 +823,18 @@ public String getType() {
     public TransportVersion getMinimalSupportedVersion() {
         return TransportVersions.ZERO;
     }
+
+    @Override
+    public boolean supportsParallelCollection(ToLongFunction<String> fieldCardinalityResolver) {
+        if (sorts != null) {
+            // the implicit sorting is by _score, which supports parallel collection
+            for (SortBuilder<?> sortBuilder : sorts) {
+                if (sortBuilder.supportsParallelCollection() == false) {
+                    return false;
+                }
+            }
+        }
+
+        return super.supportsParallelCollection(fieldCardinalityResolver);
+    }
 }

server/src/test/java/org/elasticsearch/search/builder/SearchSourceBuilderTests.java

@@ -36,6 +36,7 @@
 import org.elasticsearch.search.SearchExtBuilder;
 import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
 import org.elasticsearch.search.aggregations.metrics.MaxAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.TopHitsAggregationBuilder;
 import org.elasticsearch.search.collapse.CollapseBuilder;
 import org.elasticsearch.search.collapse.CollapseBuilderTests;
 import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
@@ -993,6 +994,26 @@ public void testSupportsParallelCollection() {
             searchSourceBuilder.aggregation(new TermsAggregationBuilder("terms"));
             assertFalse(searchSourceBuilder.supportsParallelCollection(fieldCardinality));
         }
+        {
+            SearchSourceBuilder searchSourceBuilder = newSearchSourceBuilder.get();
+            searchSourceBuilder.aggregation(new TopHitsAggregationBuilder("terms"));
+            assertTrue(searchSourceBuilder.supportsParallelCollection(fieldCardinality));
+        }
+        {
+            SearchSourceBuilder searchSourceBuilder = newSearchSourceBuilder.get();
+            searchSourceBuilder.aggregation(new TopHitsAggregationBuilder("terms").sort("_score"));
+            assertTrue(searchSourceBuilder.supportsParallelCollection(fieldCardinality));
+        }
+        {
+            SearchSourceBuilder searchSourceBuilder = newSearchSourceBuilder.get();
+            searchSourceBuilder.aggregation(new TopHitsAggregationBuilder("terms").sort(SortBuilders.fieldSort("field")));
+            assertFalse(searchSourceBuilder.supportsParallelCollection(fieldCardinality));
+        }
+        {
+            SearchSourceBuilder searchSourceBuilder = newSearchSourceBuilder.get();
+            searchSourceBuilder.aggregation(new TopHitsAggregationBuilder("terms").sort(SortBuilders.scriptSort(new Script("id"), ScriptSortBuilder.ScriptSortType.NUMBER)));
+            assertFalse(searchSourceBuilder.supportsParallelCollection(fieldCardinality));
+        }
         {
             SearchSourceBuilder searchSourceBuilder = newSearchSourceBuilder.get();
             searchSourceBuilder.collapse(CollapseBuilderTests.randomCollapseBuilder());