Add an option to force the numeric type of a field sort

This change adds an option to the `FieldSortBuilder` that allows to transform the type
of a numeric field into another. Possible values for this option are `long` that transforms
the source field into an integer and `double` that transforms the source field into a floating point.
This new option is useful for cross-index search when the sort field is mapped differently on some
indices. For instance if a field is mapped as a floating point in one index and as an integer in another
it is possible to align the type for both indices using the `numeric_type` option:

```
{
   "sort": {
    "field": "my_field",
    "numeric_type": "double" <1>
   }
}
```

<1> Ensure that values for this field are transformed to a floating point if needed.

Only `long` and `double` are supported at the moment but the goal is to also handle `date` and `date_nanos`
when elastic#32601 is merged.

docs/reference/search/request/sort.asciidoc

@@ -108,6 +108,73 @@ POST /_search
 --------------------------------------------------
 // CONSOLE
 
+==== Sorting numeric fields
+
+For numeric fields it is also possible to cast the values from one type
+to another using the `numeric_type` option.
+This option accepts the following values: [`"double", "long"`] and can be useful
+for cross-index search if the sort field is mapped differently on some
+indices.
+
+Consider for instance these two indices:
+
+[source,js]
+--------------------------------------------------
+PUT /index_double
+{
+    "mappings": {
+        "properties": {
+            "field": { "type": "double" }
+        }
+    }
+}
+--------------------------------------------------
+// CONSOLE
+
+[source,js]
+--------------------------------------------------
+PUT /index_long
+{
+    "mappings": {
+        "properties": {
+            "field": { "type": "long" }
+        }
+    }
+}
+--------------------------------------------------
+// CONSOLE
+// TEST[continued]
+
+Since `field` is mapped as a `double` in the first index and as a `long`
+in the second index, it is not possible to use this field to sort requests
+that query both indices bu default. However you can force the type to one
+or the other with the `numeric_type` option in order to force a specific
+type for all indices:
+
+[source,js]
+--------------------------------------------------
+POST /index_long,index_double/_search
+{
+   "sort" : [
+      {
+        "field" : {
+            "numeric_type" : "double"
+        }
+      }
+   ]
+}
+--------------------------------------------------
+// CONSOLE
+// TEST[continued]
+
+In the example above, values for the `index_long` index are casted to
+a double in order to be compatible with the values produced by the
+`index_double` index.
+It is also possible to transform a floating point field into a `long`
+but note that in this case floating points are replaced by the largest
+value that is less than or equal to the argument and is equal to a mathematical
+integer.
+
 [[nested-sorting]]
 ==== Sorting within nested objects.
 

server/src/main/java/org/elasticsearch/index/fielddata/plain/SortedNumericDVIndexFieldData.java

@@ -31,6 +31,7 @@
 import org.apache.lucene.search.SortedNumericSortField;
 import org.apache.lucene.util.Accountable;
 import org.apache.lucene.util.NumericUtils;
+import org.elasticsearch.common.inject.name.Named;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.fielddata.AtomicNumericFieldData;
 import org.elasticsearch.index.fielddata.FieldData;
@@ -46,6 +47,9 @@
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Predicate;
 
 /**
  * FieldData backed by {@link LeafReader#getSortedNumericDocValues(String)}
@@ -62,10 +66,15 @@ public SortedNumericDVIndexFieldData(Index index, String fieldNames, NumericType
         this.numericType = numericType;
     }
 
-    @Override
-    public SortField sortField(Object missingValue, MultiValueMode sortMode, Nested nested, boolean reverse) {
+    /**
+     * Returns the {@link SortField} to used for sorting.
+     * Values are casted to the provided <code>targetNumericType</code> type if it doesn't
+     * match the field's <code>numericType</code>.
+     */
+    public SortField sortField(NumericType targetNumericType, Object missingValue, MultiValueMode sortMode, Nested nested,
+                               boolean reverse) {
         final XFieldComparatorSource source;
-        switch (numericType) {
+        switch (targetNumericType) {
             case HALF_FLOAT:
             case FLOAT:
                 source = new FloatValuesComparatorSource(this, missingValue, sortMode, nested);
@@ -76,7 +85,7 @@ public SortField sortField(Object missingValue, MultiValueMode sortMode, Nested
                 break;
 
             default:
-                assert !numericType.isFloatingPoint();
+                assert !targetNumericType.isFloatingPoint();
                 source = new LongValuesComparatorSource(this, missingValue, sortMode, nested);
                 break;
         }
@@ -86,8 +95,9 @@ public SortField sortField(Object missingValue, MultiValueMode sortMode, Nested
          * returns a custom sort field otherwise.
          */
         if (nested != null
-                || (sortMode != MultiValueMode.MAX && sortMode != MultiValueMode.MIN)
-                || numericType == NumericType.HALF_FLOAT) {
+            || (sortMode != MultiValueMode.MAX && sortMode != MultiValueMode.MIN)
+            || numericType == NumericType.HALF_FLOAT
+            || targetNumericType != numericType) {
             return new SortField(fieldName, source, reverse);
         }
 
@@ -112,6 +122,11 @@ public SortField sortField(Object missingValue, MultiValueMode sortMode, Nested
         return sortField;
     }
 
+    @Override
+    public SortField sortField(Object missingValue, MultiValueMode sortMode, Nested nested, boolean reverse) {
+        return sortField(numericType, missingValue, sortMode, nested, reverse);
+    }
+
     @Override
     public NumericType getNumericType() {
         return numericType;

server/src/main/java/org/elasticsearch/search/sort/FieldSortBuilder.java

@@ -33,6 +33,8 @@
 import org.elasticsearch.index.fielddata.IndexFieldData;
 import org.elasticsearch.index.fielddata.IndexFieldData.XFieldComparatorSource.Nested;
 import org.elasticsearch.index.fielddata.IndexNumericFieldData;
+import org.elasticsearch.index.fielddata.IndexNumericFieldData.NumericType;
+import org.elasticsearch.index.fielddata.plain.SortedNumericDVIndexFieldData;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.QueryRewriteContext;
@@ -42,6 +44,7 @@
 import org.elasticsearch.search.MultiValueMode;
 
 import java.io.IOException;
+import java.util.Locale;
 import java.util.Objects;
 
 import static org.elasticsearch.search.sort.NestedSortBuilder.NESTED_FIELD;
@@ -56,6 +59,7 @@ public class FieldSortBuilder extends SortBuilder<FieldSortBuilder> {
     public static final ParseField MISSING = new ParseField("missing");
     public static final ParseField SORT_MODE = new ParseField("mode");
     public static final ParseField UNMAPPED_TYPE = new ParseField("unmapped_type");
+    public static final ParseField NUMERIC_TYPE = new ParseField("numeric_type");
 
     /**
      * special field name to sort by index order
@@ -72,6 +76,8 @@ public class FieldSortBuilder extends SortBuilder<FieldSortBuilder> {
 
     private String unmappedType;
 
+    private String numericType;
+
     private SortMode sortMode;
 
     private QueryBuilder nestedFilter;
@@ -94,6 +100,7 @@ public FieldSortBuilder(FieldSortBuilder template) {
         if (template.getNestedSort() != null) {
             this.setNestedSort(template.getNestedSort());
         }
+        this.numericType = template.numericType;
     }
 
     /**
@@ -123,6 +130,9 @@ public FieldSortBuilder(StreamInput in) throws IOException {
         if (in.getVersion().onOrAfter(Version.V_6_1_0)) {
             nestedSort = in.readOptionalWriteable(NestedSortBuilder::new);
         }
+        if (in.getVersion().onOrAfter(Version.V_7_0_0)) {
+            numericType = in.readOptionalString();
+        }
     }
 
     @Override
@@ -137,6 +147,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(Version.V_6_1_0)) {
             out.writeOptionalWriteable(nestedSort);
         }
+        if (out.getVersion().onOrAfter(Version.V_7_0_0)) {
+            out.writeOptionalString(numericType);
+        }
     }
 
     /** Returns the document field this sort should be based on. */
@@ -274,6 +287,36 @@ public FieldSortBuilder setNestedSort(final NestedSortBuilder nestedSort) {
         return this;
     }
 
+    /**
+     * Returns the numeric type that values should translated to or null
+     * if the original numeric type should be preserved.
+     */
+    public String getNumericType() {
+        return numericType;
+    }
+
+    /**
+     * Forces the numeric type to use for the field. The query will fail if this option
+     * is set on a field that is not mapped as a numeric in some indices.
+     * Specifying a numeric type tells Elasticsearch what type the sort values should
+     * have, which is important for cross-index search, if a field does not have
+     * the same type on all indices.
+     * Allowed values are <code>long</code> and <code>double</code>.
+     */
+    public FieldSortBuilder setNumericType(String numericType) {
+        String upperCase = numericType.toUpperCase(Locale.ENGLISH);
+        switch (upperCase) {
+            case "LONG":
+            case "DOUBLE":
+                break;
+
+            default:
+                throw new IllegalArgumentException("invalid value for [numeric_type], must be [LONG, DOUBLE], got " + numericType);
+        }
+        this.numericType = upperCase;
+        return this;
+    }
+
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
@@ -297,6 +340,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         if (nestedSort != null) {
             builder.field(NESTED_FIELD.getPreferredName(), nestedSort);
         }
+        if (numericType != null) {
+            builder.field(NUMERIC_TYPE.getPreferredName(), numericType);
+        }
         builder.endObject();
         builder.endObject();
         return builder;
@@ -351,7 +397,18 @@ public SortFieldAndFormat build(QueryShardContext context) throws IOException {
                     && (sortMode == SortMode.SUM || sortMode == SortMode.AVG || sortMode == SortMode.MEDIAN)) {
                 throw new QueryShardException(context, "we only support AVG, MEDIAN and SUM on number based fields");
             }
-            SortField field = fieldData.sortField(missing, localSortMode, nested, reverse);
+            final SortField field;
+            if (numericType != null) {
+                if (fieldData instanceof SortedNumericDVIndexFieldData == false) {
+                    throw new QueryShardException(context,
+                        "[numeric_type] option cannot be set on a non-numeric field, got " + fieldType.typeName());
+                }
+                SortedNumericDVIndexFieldData numericFieldData = (SortedNumericDVIndexFieldData) fieldData;
+                NumericType resolvedType = NumericType.valueOf(numericType);
+                field = numericFieldData.sortField(resolvedType, missing, localSortMode, nested, reverse);
+            } else {
+                field = fieldData.sortField(missing, localSortMode, nested, reverse);
+            }
             return new SortFieldAndFormat(field, fieldType.docValueFormat(null, null));
         }
     }
@@ -370,13 +427,14 @@ public boolean equals(Object other) {
         return (Objects.equals(this.fieldName, builder.fieldName) && Objects.equals(this.nestedFilter, builder.nestedFilter)
                 && Objects.equals(this.nestedPath, builder.nestedPath) && Objects.equals(this.missing, builder.missing)
                 && Objects.equals(this.order, builder.order) && Objects.equals(this.sortMode, builder.sortMode)
-                && Objects.equals(this.unmappedType, builder.unmappedType) && Objects.equals(this.nestedSort, builder.nestedSort));
+                && Objects.equals(this.unmappedType, builder.unmappedType) && Objects.equals(this.nestedSort, builder.nestedSort))
+                && Objects.equals(this.nestedFilter, builder.numericType);
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(this.fieldName, this.nestedFilter, this.nestedPath, this.nestedSort, this.missing, this.order, this.sortMode,
-            this.unmappedType);
+            this.unmappedType, this.numericType);
     }
 
     @Override
@@ -413,6 +471,7 @@ public static FieldSortBuilder fromXContent(XContentParser parser, String fieldN
             return SortBuilder.parseNestedFilter(p);
         }, NESTED_FILTER_FIELD);
         PARSER.declareObject(FieldSortBuilder::setNestedSort, (p, c) -> NestedSortBuilder.fromXContent(p), NESTED_FIELD);
+        PARSER.declareString((b, v) -> b.setNumericType(v), NUMERIC_TYPE);
     }
 
     @Override

server/src/test/java/org/elasticsearch/search/sort/FieldSortBuilderTests.java

@@ -102,13 +102,16 @@ public FieldSortBuilder randomFieldSortBuilder() {
                 }
             }
         }
+        if (randomBoolean()) {
+            builder.setNumericType(randomFrom(random(), "long", "double"));
+        }
         return builder;
     }
 
     @Override
     protected FieldSortBuilder mutate(FieldSortBuilder original) throws IOException {
         FieldSortBuilder mutated = new FieldSortBuilder(original);
-        int parameter = randomIntBetween(0, 4);
+        int parameter = randomIntBetween(0, 5);
         switch (parameter) {
         case 0:
             if (original.getNestedPath() == null && original.getNestedFilter() == null) {
@@ -136,6 +139,9 @@ protected FieldSortBuilder mutate(FieldSortBuilder original) throws IOException
         case 4:
             mutated.order(randomValueOtherThan(original.order(), () -> randomFrom(SortOrder.values())));
             break;
+        case 5:
+            mutated.setNumericType(randomValueOtherThan(original.getNumericType(), () -> randomFrom("long", "double")));
+            break;
         default:
             throw new IllegalStateException("Unsupported mutation.");
         }