Improve error when sorting on incompatible types (#88399)

Currently when sorting on incompatible types, we get
class_cast_exception error (code 500). This patch improves
the error to explain that the problem is because
of incompatible sort types for the field across
different shards and returns user error (code 400).

Closes #73146

docs/changelog/88399.yaml

@@ -0,0 +1,6 @@
+pr: 88399
+summary: Improve error when sorting on incompatible types
+area: Search
+type: enhancement
+issues:
+ - 73146

server/src/internalClusterTest/java/org/elasticsearch/search/sort/FieldSortIT.java

@@ -2122,4 +2122,43 @@ public void testLongSortOptimizationCorrectResults() {
         }
     }
 
+    public void testSortMixedFieldTypes() {
+        assertAcked(prepareCreate("index_long").setMapping("foo", "type=long").get());
+        assertAcked(prepareCreate("index_integer").setMapping("foo", "type=integer").get());
+        assertAcked(prepareCreate("index_double").setMapping("foo", "type=double").get());
+        assertAcked(prepareCreate("index_keyword").setMapping("foo", "type=keyword").get());
+
+        client().prepareIndex("index_long").setId("1").setSource("foo", "123").get();
+        client().prepareIndex("index_integer").setId("1").setSource("foo", "123").get();
+        client().prepareIndex("index_double").setId("1").setSource("foo", "123").get();
+        client().prepareIndex("index_keyword").setId("1").setSource("foo", "123").get();
+        refresh();
+
+        { // mixing long and integer types is ok, as we convert integer sort to long sort
+            SearchResponse searchResponse = client().prepareSearch("index_long", "index_integer")
+                .addSort(new FieldSortBuilder("foo"))
+                .setSize(10)
+                .get();
+            assertSearchResponse(searchResponse);
+        }
+
+        String errMsg = "Can't sort on field [foo]; the field has incompatible sort types";
+
+        { // mixing long and double types is not allowed
+            SearchPhaseExecutionException exc = expectThrows(
+                SearchPhaseExecutionException.class,
+                () -> client().prepareSearch("index_long", "index_double").addSort(new FieldSortBuilder("foo")).setSize(10).get()
+            );
+            assertThat(exc.getCause().toString(), containsString(errMsg));
+        }
+
+        { // mixing long and keyword types is not allowed
+            SearchPhaseExecutionException exc = expectThrows(
+                SearchPhaseExecutionException.class,
+                () -> client().prepareSearch("index_long", "index_keyword").addSort(new FieldSortBuilder("foo")).setSize(10).get()
+            );
+            assertThat(exc.getCause().toString(), containsString(errMsg));
+        }
+    }
+
 }

server/src/main/java/org/elasticsearch/action/search/SearchPhaseController.java

@@ -14,6 +14,8 @@
 import org.apache.lucene.search.ScoreDoc;
 import org.apache.lucene.search.Sort;
 import org.apache.lucene.search.SortField;
+import org.apache.lucene.search.SortedNumericSortField;
+import org.apache.lucene.search.SortedSetSortField;
 import org.apache.lucene.search.TermStatistics;
 import org.apache.lucene.search.TopDocs;
 import org.apache.lucene.search.TopFieldDocs;
@@ -197,6 +199,7 @@ static TopDocs mergeTopDocs(Collection<TopDocs> results, int topN, int from) {
             final TopFieldGroups[] shardTopDocs = results.toArray(new TopFieldGroups[numShards]);
             mergedTopDocs = TopFieldGroups.merge(sort, from, topN, shardTopDocs, false);
         } else if (topDocs instanceof TopFieldDocs firstTopDocs) {
+            checkSameSortTypes(results, firstTopDocs.fields);
             final Sort sort = new Sort(firstTopDocs.fields);
             final TopFieldDocs[] shardTopDocs = results.toArray(new TopFieldDocs[numShards]);
             mergedTopDocs = TopDocs.merge(sort, from, topN, shardTopDocs);
@@ -207,6 +210,56 @@ static TopDocs mergeTopDocs(Collection<TopDocs> results, int topN, int from) {
         return mergedTopDocs;
     }
 
+    private static void checkSameSortTypes(Collection<TopDocs> results, SortField[] firstSortFields) {
+        if (results.size() < 2) return;
+
+        SortField.Type[] firstTypes = new SortField.Type[firstSortFields.length];
+        boolean isFirstResult = true;
+        for (TopDocs topDocs : results) {
+            SortField[] curSortFields = ((TopFieldDocs) topDocs).fields;
+            if (isFirstResult) {
+                for (int i = 0; i < curSortFields.length; i++) {
+                    firstTypes[i] = getType(firstSortFields[i]);
+                    if (firstTypes[i] == SortField.Type.CUSTOM) {
+                        // for custom types that we can't resolve, we can't do the check
+                        return;
+                    }
+                }
+                isFirstResult = false;
+            } else {
+                for (int i = 0; i < curSortFields.length; i++) {
+                    SortField.Type curType = getType(curSortFields[i]);
+                    if (curType != firstTypes[i]) {
+                        if (curType == SortField.Type.CUSTOM) {
+                            // for custom types that we can't resolve, we can't do the check
+                            return;
+                        }
+                        throw new IllegalArgumentException(
+                            "Can't sort on field ["
+                                + curSortFields[i].getField()
+                                + "]; the field has incompatible sort types: ["
+                                + firstTypes[i]
+                                + "] and ["
+                                + curType
+                                + "] across shards!"
+                        );
+                    }
+                }
+            }
+        }
+    }
+
+    private static SortField.Type getType(SortField sortField) {
+        if (sortField instanceof SortedNumericSortField) {
+            return ((SortedNumericSortField) sortField).getNumericType();
+        }
+        if (sortField instanceof SortedSetSortField) {
+            return SortField.Type.STRING;
+        } else {
+            return sortField.getType();
+        }
+    }
+
     static void setShardIndex(TopDocs topDocs, int shardIndex) {
         assert topDocs.scoreDocs.length == 0 || topDocs.scoreDocs[0].shardIndex == -1 : "shardIndex is already set";
         for (ScoreDoc doc : topDocs.scoreDocs) {