Ensure that field collapsing works with field aliases.

Previously, the following situation would throw an error:
* A search contains a `collapse` on a particular field.
* The search spans multiple indices, and in one index the field is mapped as a
  concrete field, but in another it is a field alias.

The error occurs when we attempt to merge `CollapseTopFieldDocs` across shards.
When merging, we validate that the name of the collapse field is the same across
shards. But the name has already been resolved to the concrete field name, so it
will be different on shards where the field was mapped as an alias vs. shards
where it was a concrete field.

This PR updates the collapse field name in `CollapseTopFieldDocs` to the
original requested field, so that it will always be consistent across shards.

Note that in #32648, we already made a fix around collapsing on field aliases.
However, we didn't test this specific scenario where the field was mapped as an
alias in only one of the indices being searched.

rest-api-spec/src/main/resources/rest-api-spec/test/search/110_field_collapsing.yml

@@ -347,30 +347,48 @@ setup:
 
 ---
 "field collapsing on a field alias":
-
+  - skip:
+      version: " - 7.99.99"
+      reason: the bug fix is not yet backported to 7.x
   - do:
-      indices.put_mapping:
-        index: test
+      indices.create:
+        index:  alias-test
         body:
-          properties:
-            group_alias: { type: alias, path: numeric_group }
+          mappings:
+            properties:
+              numeric_group: { type: alias, path: other_numeric_group }
+              other_numeric_group: { type: integer }
+  - do:
+      index:
+        index:  alias-test
+        id:     1
+        body:   { other_numeric_group: 1, sort: 6 }
+  - do:
+      index:
+        index:  alias-test
+        id:     2
+        body:   { other_numeric_group: 25, sort: 10 }
+  - do:
+      indices.refresh:
+        index: alias-test
+
   - do:
       search:
         rest_total_hits_as_int: true
-        index: test
+        index: [alias-test, test]
         body:
-          collapse: { field: group_alias, inner_hits: { name: sub_hits } }
+          collapse: { field: numeric_group, inner_hits: { name: sub_hits } }
           sort: [{ sort: desc }]
 
-  - match: { hits.total: 6 }
+  - match: { hits.total: 8 }
   - length: { hits.hits: 3 }
 
-  - match: { hits.hits.0.fields.group_alias: [3] }
+  - match: { hits.hits.0.fields.numeric_group: [3] }
   - match: { hits.hits.0.inner_hits.sub_hits.hits.total: 1}
-  - match: { hits.hits.1.fields.group_alias: [1] }
-  - match: { hits.hits.1.inner_hits.sub_hits.hits.total: 3}
-  - match: { hits.hits.2.fields.group_alias: [25] }
-  - match: { hits.hits.2.inner_hits.sub_hits.hits.total: 2}
+  - match: { hits.hits.1.fields.numeric_group: [1] }
+  - match: { hits.hits.1.inner_hits.sub_hits.hits.total: 4}
+  - match: { hits.hits.2.fields.numeric_group: [25] }
+  - match: { hits.hits.2.inner_hits.sub_hits.hits.total: 3}
 
 ---
 "field collapsing, inner_hits and seq_no":

server/src/main/java/org/apache/lucene/search/grouping/CollapsingDocValuesSource.java

@@ -30,6 +30,7 @@
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.index.fielddata.AbstractNumericDocValues;
 import org.elasticsearch.index.fielddata.AbstractSortedDocValues;
+import org.elasticsearch.index.mapper.MappedFieldType;
 
 import java.io.IOException;
 import java.util.Collection;
@@ -58,8 +59,8 @@ static class Numeric extends CollapsingDocValuesSource<Long> {
         private long value;
         private boolean hasValue;
 
-        Numeric(String field) {
-            super(field);
+        Numeric(MappedFieldType fieldType) {
+            super(fieldType.name());
         }
 
         @Override
@@ -148,8 +149,8 @@ static class Keyword extends CollapsingDocValuesSource<BytesRef> {
         private SortedDocValues values;
         private int ord;
 
-        Keyword(String field) {
-            super(field);
+        Keyword(MappedFieldType fieldType) {
+            super(fieldType.name());
         }
 
         @Override

server/src/main/java/org/apache/lucene/search/grouping/CollapsingTopDocsCollector.java

@@ -25,6 +25,7 @@
 import org.apache.lucene.search.Sort;
 import org.apache.lucene.search.SortField;
 import org.apache.lucene.search.TotalHits;
+import org.elasticsearch.index.mapper.MappedFieldType;
 
 import java.io.IOException;
 import java.util.Collection;
@@ -119,17 +120,19 @@ public void collect(int doc) throws IOException {
      * the collect will fail with an {@link IllegalStateException} if a document contains more than one value for the
      * field.
      *
-     * @param collapseField The sort field used to group
-     *                      documents.
-     * @param sort          The {@link Sort} used to sort the collapsed hits.
-     *                      The collapsing keeps only the top sorted document per collapsed key.
-     *                      This must be non-null, ie, if you want to groupSort by relevance
-     *                      use Sort.RELEVANCE.
-     * @param topN          How many top groups to keep.
+     * @param collapseField     The sort field used to group documents.
+     * @param collapseFieldType The {@link MappedFieldType} for this sort field.
+     * @param sort              The {@link Sort} used to sort the collapsed hits.
+     *                          The collapsing keeps only the top sorted document per collapsed key.
+     *                          This must be non-null, ie, if you want to groupSort by relevance
+     *                          use Sort.RELEVANCE.
+     * @param topN              How many top groups to keep.
      */
-    public static CollapsingTopDocsCollector<?> createNumeric(String collapseField, Sort sort,
+    public static CollapsingTopDocsCollector<?> createNumeric(String collapseField,
+                                                              MappedFieldType collapseFieldType,
+                                                              Sort sort,
                                                               int topN)  {
-        return new CollapsingTopDocsCollector<>(new CollapsingDocValuesSource.Numeric(collapseField),
+        return new CollapsingTopDocsCollector<>(new CollapsingDocValuesSource.Numeric(collapseFieldType),
                 collapseField, sort, topN);
     }
 
@@ -139,16 +142,18 @@ public static CollapsingTopDocsCollector<?> createNumeric(String collapseField,
      * the collect will fail with an {@link IllegalStateException} if a document contains more than one value for the
      * field.
      *
-     * @param collapseField The sort field used to group
-     *                      documents.
-     * @param sort          The {@link Sort} used to sort the collapsed hits. The collapsing keeps only the top sorted
-     *                      document per collapsed key.
-     *                      This must be non-null, ie, if you want to groupSort by relevance use Sort.RELEVANCE.
-     * @param topN          How many top groups to keep.
+     * @param collapseField     The sort field used to group documents.
+     * @param collapseFieldType The {@link MappedFieldType} for this sort field.
+     * @param sort              The {@link Sort} used to sort the collapsed hits. The collapsing keeps only the top sorted
+     *                          document per collapsed key.
+     *                          This must be non-null, ie, if you want to groupSort by relevance use Sort.RELEVANCE.
+     * @param topN              How many top groups to keep.
      */
-    public static CollapsingTopDocsCollector<?> createKeyword(String collapseField, Sort sort,
+    public static CollapsingTopDocsCollector<?> createKeyword(String collapseField,
+                                                              MappedFieldType collapseFieldType,
+                                                              Sort sort,
                                                               int topN)  {
-        return new CollapsingTopDocsCollector<>(new CollapsingDocValuesSource.Keyword(collapseField),
+        return new CollapsingTopDocsCollector<>(new CollapsingDocValuesSource.Keyword(collapseFieldType),
                 collapseField, sort, topN);
     }
 }

server/src/main/java/org/elasticsearch/search/collapse/CollapseContext.java

@@ -62,11 +62,11 @@ public List<InnerHitBuilder> getInnerHit() {
 
     public CollapsingTopDocsCollector<?> createTopDocs(Sort sort, int topN) {
         if (fieldType instanceof KeywordFieldMapper.KeywordFieldType) {
-            return CollapsingTopDocsCollector.createKeyword(fieldType.name(), sort, topN);
+            return CollapsingTopDocsCollector.createKeyword(fieldName, fieldType, sort, topN);
         } else if (fieldType instanceof NumberFieldMapper.NumberFieldType) {
-            return CollapsingTopDocsCollector.createNumeric(fieldType.name(), sort, topN);
+            return CollapsingTopDocsCollector.createNumeric(fieldName, fieldType, sort, topN);
         } else {
-            throw new IllegalStateException("unknown type for collapse field " + fieldType.name() +
+            throw new IllegalStateException("unknown type for collapse field " + fieldName +
                 ", only keywords and numbers are accepted");
         }
     }

server/src/test/java/org/apache/lucene/grouping/CollapsingTopDocsCollectorTests.java

@@ -48,6 +48,8 @@
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.NumericUtils;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.MockFieldMapper;
 import org.elasticsearch.test.ESTestCase;
 
 import java.io.IOException;
@@ -108,6 +110,7 @@ private <T extends Comparable<T>> void assertSearchCollapse(CollapsingDocValuesP
             w.addDocument(doc);
             totalHits++;
         }
+
         List<T> valueList = new ArrayList<>(values);
         Collections.sort(valueList);
         final IndexReader reader = w.getReader();
@@ -117,15 +120,18 @@ private <T extends Comparable<T>> void assertSearchCollapse(CollapsingDocValuesP
         final SortField sort2 = new SortField("sort2", SortField.Type.LONG);
         Sort sort = new Sort(sort1, sort2, collapseField);
 
+        MappedFieldType fieldType = new MockFieldMapper.FakeFieldType();
+        fieldType.setName(collapseField.getField());
+
         int expectedNumGroups = values.size();
 
         final CollapsingTopDocsCollector<?> collapsingCollector;
         if (numeric) {
             collapsingCollector =
-                CollapsingTopDocsCollector.createNumeric(collapseField.getField(), sort, expectedNumGroups);
+                CollapsingTopDocsCollector.createNumeric(collapseField.getField(), fieldType, sort, expectedNumGroups);
         } else {
             collapsingCollector =
-                CollapsingTopDocsCollector.createKeyword(collapseField.getField(), sort, expectedNumGroups);
+                CollapsingTopDocsCollector.createKeyword(collapseField.getField(), fieldType, sort, expectedNumGroups);
         }
 
         TopFieldCollector topFieldCollector =
@@ -195,9 +201,9 @@ private <T extends Comparable<T>> void assertSearchCollapse(CollapsingDocValuesP
             final SegmentSearcher subSearcher = subSearchers[shardIDX];
             final CollapsingTopDocsCollector<?> c;
             if (numeric) {
-                c = CollapsingTopDocsCollector.createNumeric(collapseField.getField(), sort, expectedNumGroups);
+                c = CollapsingTopDocsCollector.createNumeric(collapseField.getField(), fieldType, sort, expectedNumGroups);
             } else {
-                c = CollapsingTopDocsCollector.createKeyword(collapseField.getField(), sort, expectedNumGroups);
+                c = CollapsingTopDocsCollector.createKeyword(collapseField.getField(), fieldType, sort, expectedNumGroups);
             }
             subSearcher.search(weight, c);
             shardHits[shardIDX] = c.getTopDocs();
@@ -374,11 +380,16 @@ public void testEmptyNumericSegment() throws Exception {
         w.commit();
         final IndexReader reader = w.getReader();
         final IndexSearcher searcher = newSearcher(reader);
+
+        MappedFieldType fieldType = new MockFieldMapper.FakeFieldType();
+        fieldType.setName("group");
+
         SortField sortField = new SortField("group", SortField.Type.LONG);
         sortField.setMissingValue(Long.MAX_VALUE);
         Sort sort = new Sort(sortField);
+
         final CollapsingTopDocsCollector<?> collapsingCollector =
-                CollapsingTopDocsCollector.createNumeric("group", sort, 10);
+                CollapsingTopDocsCollector.createNumeric("group", fieldType, sort, 10);
         searcher.search(new MatchAllDocsQuery(), collapsingCollector);
         CollapseTopFieldDocs collapseTopFieldDocs = collapsingCollector.getTopDocs();
         assertEquals(4, collapseTopFieldDocs.scoreDocs.length);
@@ -412,9 +423,14 @@ public void testEmptySortedSegment() throws Exception {
         w.commit();
         final IndexReader reader = w.getReader();
         final IndexSearcher searcher = newSearcher(reader);
+
+        MappedFieldType fieldType = new MockFieldMapper.FakeFieldType();
+        fieldType.setName("group");
+
         Sort sort = new Sort(new SortField("group", SortField.Type.STRING_VAL));
+
         final CollapsingTopDocsCollector<?> collapsingCollector =
-            CollapsingTopDocsCollector.createKeyword("group", sort, 10);
+            CollapsingTopDocsCollector.createKeyword("group", fieldType, sort, 10);
         searcher.search(new MatchAllDocsQuery(), collapsingCollector);
         CollapseTopFieldDocs collapseTopFieldDocs = collapsingCollector.getTopDocs();
         assertEquals(4, collapseTopFieldDocs.scoreDocs.length);