Replace custom sort field with SortedSetSortField and SortedNumericSortField when possible (#23827)

Currently for field sorting we always use a custom sort field and a custom comparator source.
Though for numeric fields this custom sort field could be replaced with a standard SortedNumericSortField unless
the field is nested especially since we removed the FieldData for numerics.
We can also use a SortedSetSortField for string sort based on doc_values when the field is not nested.

This change replaces IndexFieldData#comparatorSource with IndexFieldData#sortField that returns a Sorted{Set,Numeric}SortField when possible or a custom
 sort field when the field sort spec is not handled by the SortedSortFields.

core/src/main/java/org/elasticsearch/common/lucene/Lucene.java

@@ -57,6 +57,8 @@ import org.apache.lucene.search.TopFieldDocs;
 import org.apache.lucene.search.TwoPhaseIterator;
 import org.apache.lucene.search.Weight;
 import org.apache.lucene.search.grouping.CollapseTopFieldDocs;
+import org.apache.lucene.search.SortedNumericSortField;
+import org.apache.lucene.search.SortedSetSortField;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.IOContext;
 import org.apache.lucene.store.IndexInput;
@@ -552,7 +554,22 @@ public class Lucene {
             SortField newSortField = new SortField(sortField.getField(), SortField.Type.DOUBLE);
             newSortField.setMissingValue(sortField.getMissingValue());
             sortField = newSortField;
+        } else if (sortField.getClass() == SortedSetSortField.class) {
+            // for multi-valued sort field, we replace the SortedSetSortField with a simple SortField.
+            // It works because the sort field is only used to merge results from different shards.
+            SortField newSortField = new SortField(sortField.getField(), SortField.Type.STRING, sortField.getReverse());
+            newSortField.setMissingValue(sortField.getMissingValue());
+            sortField = newSortField;
+        } else if (sortField.getClass() == SortedNumericSortField.class) {
+            // for multi-valued sort field, we replace the SortedSetSortField with a simple SortField.
+            // It works because the sort field is only used to merge results from different shards.
+            SortField newSortField = new SortField(sortField.getField(),
+                ((SortedNumericSortField) sortField).getNumericType(),
+                sortField.getReverse());
+            newSortField.setMissingValue(sortField.getMissingValue());
+            sortField = newSortField;
         }
+
         if (sortField.getClass() != SortField.class) {
             throw new IllegalArgumentException("Cannot serialize SortField impl [" + sortField + "]");
         }

core/src/main/java/org/elasticsearch/index/fielddata/IndexFieldData.java

@@ -85,9 +85,9 @@ public interface IndexFieldData<FD extends AtomicFieldData> extends IndexCompone
     FD loadDirect(LeafReaderContext context) throws Exception;
 
     /**
-     * Comparator used for sorting.
+     * Returns the {@link SortField} to used for sorting.
      */
-    XFieldComparatorSource comparatorSource(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested);
+    SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested, boolean reverse);
 
     /**
      * Clears any resources associated with this field data.
@@ -136,17 +136,17 @@ public interface IndexFieldData<FD extends AtomicFieldData> extends IndexCompone
         }
 
         /** Whether missing values should be sorted first. */
-        protected final boolean sortMissingFirst(Object missingValue) {
+        public final boolean sortMissingFirst(Object missingValue) {
             return "_first".equals(missingValue);
         }
 
         /** Whether missing values should be sorted last, this is the default. */
-        protected final boolean sortMissingLast(Object missingValue) {
+        public final boolean sortMissingLast(Object missingValue) {
             return missingValue == null || "_last".equals(missingValue);
         }
 
         /** Return the missing object value according to the reduced type of the comparator. */
-        protected final Object missingObject(Object missingValue, boolean reversed) {
+        public final Object missingObject(Object missingValue, boolean reversed) {
             if (sortMissingFirst(missingValue) || sortMissingLast(missingValue)) {
                 final boolean min = sortMissingFirst(missingValue) ^ reversed;
                 switch (reducedType()) {

core/src/main/java/org/elasticsearch/index/fielddata/ordinals/GlobalOrdinalsIndexFieldData.java

@@ -20,6 +20,7 @@ package org.elasticsearch.index.fielddata.ordinals;
 
 import org.apache.lucene.index.DirectoryReader;
 import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.search.SortField;
 import org.apache.lucene.util.Accountable;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.index.AbstractIndexComponent;
@@ -68,7 +69,7 @@ public abstract class GlobalOrdinalsIndexFieldData extends AbstractIndexComponen
     }
 
     @Override
-    public XFieldComparatorSource comparatorSource(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested) {
+    public SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested, boolean reverse) {
         throw new UnsupportedOperationException("no global ordinals sorting yet");
     }
 

core/src/main/java/org/elasticsearch/index/fielddata/plain/AbstractGeoPointDVIndexFieldData.java

@@ -21,6 +21,7 @@ package org.elasticsearch.index.fielddata.plain;
 
 import org.apache.lucene.index.DocValues;
 import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.search.SortField;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexSettings;
@@ -43,7 +44,7 @@ public abstract class AbstractGeoPointDVIndexFieldData extends DocValuesIndexFie
     }
 
     @Override
-    public final XFieldComparatorSource comparatorSource(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested) {
+    public SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested, boolean reverse) {
         throw new IllegalArgumentException("can't sort on geo_point field without using specific sorting feature, like geo_distance");
     }
 

core/src/main/java/org/elasticsearch/index/fielddata/plain/AbstractIndexGeoPointFieldData.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.index.fielddata.plain;
 
+import org.apache.lucene.search.SortField;
 import org.apache.lucene.spatial.geopoint.document.GeoPointField;
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.BytesRefIterator;
@@ -28,6 +29,7 @@ import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.geo.GeoPoint;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.fielddata.AtomicGeoPointFieldData;
+import org.elasticsearch.index.fielddata.IndexFieldData;
 import org.elasticsearch.index.fielddata.IndexFieldData.XFieldComparatorSource.Nested;
 import org.elasticsearch.index.fielddata.IndexFieldDataCache;
 import org.elasticsearch.index.fielddata.IndexGeoPointFieldData;
@@ -104,7 +106,7 @@ abstract class AbstractIndexGeoPointFieldData extends AbstractIndexFieldData<Ato
     }
 
     @Override
-    public final XFieldComparatorSource comparatorSource(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested) {
+    public SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested, boolean reverse) {
         throw new IllegalArgumentException("can't sort on geo_point field without using specific sorting feature, like geo_distance");
     }
 

core/src/main/java/org/elasticsearch/index/fielddata/plain/AbstractIndexOrdinalsFieldData.java

@@ -55,11 +55,6 @@ public abstract class AbstractIndexOrdinalsFieldData extends AbstractIndexFieldD
         this.minSegmentSize = minSegmentSize;
     }
 
-    @Override
-    public XFieldComparatorSource comparatorSource(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested) {
-        return new BytesRefFieldComparatorSource(this, missingValue, sortMode, nested);
-    }
-
     @Override
     public IndexOrdinalsFieldData loadGlobal(DirectoryReader indexReader) {
         if (indexReader.leaves().size() <= 1) {

core/src/main/java/org/elasticsearch/index/fielddata/plain/AbstractLatLonPointDVIndexFieldData.java

@@ -24,6 +24,7 @@ import org.apache.lucene.index.DocValuesType;
 import org.apache.lucene.index.FieldInfo;
 import org.apache.lucene.index.LeafReader;
 import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.search.SortField;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.index.Index;
@@ -46,8 +47,7 @@ public abstract class AbstractLatLonPointDVIndexFieldData extends DocValuesIndex
     }
 
     @Override
-    public final XFieldComparatorSource comparatorSource(@Nullable Object missingValue, MultiValueMode sortMode,
+    public SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, XFieldComparatorSource.Nested nested, boolean reverse) {
-                                                         XFieldComparatorSource.Nested nested) {
         throw new IllegalArgumentException("can't sort on geo_point field without using specific sorting feature, like geo_distance");
     }
 

core/src/main/java/org/elasticsearch/index/fielddata/plain/BinaryDVIndexFieldData.java

@@ -20,9 +20,12 @@
 package org.elasticsearch.index.fielddata.plain;
 
 import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.search.SortField;
+import org.apache.lucene.search.SortedSetSortField;
+import org.apache.lucene.search.SortedSetSelector;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.fielddata.IndexFieldData;
-import org.elasticsearch.index.fielddata.IndexFieldData.XFieldComparatorSource.Nested;
 import org.elasticsearch.index.fielddata.fieldcomparator.BytesRefFieldComparatorSource;
 import org.elasticsearch.search.MultiValueMode;
 
@@ -43,7 +46,21 @@ public class BinaryDVIndexFieldData extends DocValuesIndexFieldData implements I
     }
 
     @Override
-    public org.elasticsearch.index.fielddata.IndexFieldData.XFieldComparatorSource comparatorSource(Object missingValue, MultiValueMode sortMode, Nested nested) {
+    public SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, XFieldComparatorSource.Nested nested, boolean reverse) {
-        return new BytesRefFieldComparatorSource(this, missingValue, sortMode, nested);
+        XFieldComparatorSource source = new BytesRefFieldComparatorSource(this, missingValue, sortMode, nested);
+        /**
+         * Check if we can use a simple {@link SortedSetSortField} compatible with index sorting and
+         * returns a custom sort field otherwise.
+         */
+        if (nested != null ||
+                (sortMode != MultiValueMode.MAX && sortMode != MultiValueMode.MIN) ||
+                (source.sortMissingFirst(missingValue) == false && source.sortMissingLast(missingValue) == false)) {
+            return new SortField(getFieldName(), source, reverse);
+        }
+        SortField sortField = new SortedSetSortField(fieldName, reverse,
+            sortMode == MultiValueMode.MAX ? SortedSetSelector.Type.MAX : SortedSetSelector.Type.MIN);
+        sortField.setMissingValue(source.sortMissingLast(missingValue) ^ reverse ?
+            SortedSetSortField.STRING_LAST : SortedSetSortField.STRING_FIRST);
+        return sortField;
     }
 }

core/src/main/java/org/elasticsearch/index/fielddata/plain/BytesBinaryDVIndexFieldData.java

@@ -21,6 +21,7 @@ package org.elasticsearch.index.fielddata.plain;
 
 import org.apache.lucene.index.DocValues;
 import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.search.SortField;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexSettings;
@@ -41,7 +42,7 @@ public class BytesBinaryDVIndexFieldData extends DocValuesIndexFieldData impleme
     }
 
     @Override
-    public final XFieldComparatorSource comparatorSource(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested) {
+    public SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested, boolean reverse) {
         throw new IllegalArgumentException("can't sort on binary field");
     }
 

core/src/main/java/org/elasticsearch/index/fielddata/plain/IndexIndexFieldData.java

@@ -24,17 +24,21 @@ import org.apache.lucene.index.DocValues;
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.index.RandomAccessOrds;
 import org.apache.lucene.index.SortedDocValues;
+import org.apache.lucene.search.SortField;
 import org.apache.lucene.util.Accountable;
 import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.fielddata.AtomicOrdinalsFieldData;
 import org.elasticsearch.index.fielddata.IndexFieldData;
 import org.elasticsearch.index.fielddata.IndexFieldDataCache;
 import org.elasticsearch.index.fielddata.IndexOrdinalsFieldData;
+import org.elasticsearch.index.fielddata.fieldcomparator.BytesRefFieldComparatorSource;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.TextFieldMapper;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
+import org.elasticsearch.search.MultiValueMode;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -124,6 +128,12 @@ public class IndexIndexFieldData extends AbstractIndexOrdinalsFieldData {
         return atomicFieldData;
     }
 
+    @Override
+    public SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, XFieldComparatorSource.Nested nested, boolean reverse) {
+        final XFieldComparatorSource source = new BytesRefFieldComparatorSource(this, missingValue, sortMode, nested);
+        return new SortField(getFieldName(), source, reverse);
+    }
+
     @Override
     public IndexOrdinalsFieldData loadGlobal(DirectoryReader indexReader) {
         return this;

core/src/main/java/org/elasticsearch/index/fielddata/plain/PagedBytesIndexFieldData.java

@@ -27,10 +27,12 @@ import org.apache.lucene.index.PostingsEnum;
 import org.apache.lucene.index.Terms;
 import org.apache.lucene.index.TermsEnum;
 import org.apache.lucene.search.DocIdSetIterator;
+import org.apache.lucene.search.SortField;
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.PagedBytes;
 import org.apache.lucene.util.packed.PackedInts;
 import org.apache.lucene.util.packed.PackedLongValues;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.breaker.CircuitBreaker;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.fielddata.AtomicOrdinalsFieldData;
@@ -38,11 +40,13 @@ import org.elasticsearch.index.fielddata.IndexFieldData;
 import org.elasticsearch.index.fielddata.IndexFieldDataCache;
 import org.elasticsearch.index.fielddata.IndexOrdinalsFieldData;
 import org.elasticsearch.index.fielddata.RamAccountingTermsEnum;
+import org.elasticsearch.index.fielddata.fieldcomparator.BytesRefFieldComparatorSource;
 import org.elasticsearch.index.fielddata.ordinals.Ordinals;
 import org.elasticsearch.index.fielddata.ordinals.OrdinalsBuilder;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
+import org.elasticsearch.search.MultiValueMode;
 
 import java.io.IOException;
 
@@ -74,6 +78,12 @@ public class PagedBytesIndexFieldData extends AbstractIndexOrdinalsFieldData {
         super(indexSettings, fieldName, cache, breakerService, minFrequency, maxFrequency, minSegmentSize);
     }
 
+    @Override
+    public SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, XFieldComparatorSource.Nested nested, boolean reverse) {
+        XFieldComparatorSource source = new BytesRefFieldComparatorSource(this, missingValue, sortMode, nested);
+        return new SortField(getFieldName(), source, reverse);
+    }
+
     @Override
     public AtomicOrdinalsFieldData loadDirect(LeafReaderContext context) throws Exception {
         LeafReader reader = context.reader();

core/src/main/java/org/elasticsearch/index/fielddata/plain/ParentChildIndexFieldData.java

@@ -27,6 +27,7 @@ import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.index.MultiDocValues;
 import org.apache.lucene.index.MultiDocValues.OrdinalMap;
 import org.apache.lucene.index.SortedDocValues;
+import org.apache.lucene.search.SortField;
 import org.apache.lucene.util.Accountable;
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.LongValues;
@@ -89,8 +90,9 @@ public class ParentChildIndexFieldData extends AbstractIndexFieldData<AtomicPare
     }
 
     @Override
-    public XFieldComparatorSource comparatorSource(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested) {
+    public SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested, boolean reverse) {
-        return new BytesRefFieldComparatorSource(this, missingValue, sortMode, nested);
+        final XFieldComparatorSource source = new BytesRefFieldComparatorSource(this, missingValue, sortMode, nested);
+        return new SortField(getFieldName(), source, reverse);
     }
 
     @Override
@@ -335,7 +337,7 @@ public class ParentChildIndexFieldData extends AbstractIndexFieldData<AtomicPare
         }
 
         @Override
-        public XFieldComparatorSource comparatorSource(Object missingValue, MultiValueMode sortMode, Nested nested) {
+        public SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested, boolean reverse) {
             throw new UnsupportedOperationException("No sorting on global ords");
         }
 

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

@@ -26,6 +26,9 @@ import org.apache.lucene.index.LeafReader;
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.index.NumericDocValues;
 import org.apache.lucene.index.SortedNumericDocValues;
+import org.apache.lucene.search.SortField;
+import org.apache.lucene.search.SortedNumericSelector;
+import org.apache.lucene.search.SortedNumericSortField;
 import org.apache.lucene.util.Accountable;
 import org.apache.lucene.util.NumericUtils;
 import org.elasticsearch.index.Index;
@@ -60,17 +63,53 @@ public class SortedNumericDVIndexFieldData extends DocValuesIndexFieldData imple
     }
 
     @Override
-    public org.elasticsearch.index.fielddata.IndexFieldData.XFieldComparatorSource comparatorSource(Object missingValue, MultiValueMode sortMode, Nested nested) {
+    public SortField sortField(Object missingValue, MultiValueMode sortMode, Nested nested, boolean reverse) {
+        final XFieldComparatorSource source;
         switch (numericType) {
             case HALF_FLOAT:
             case FLOAT:
-                return new FloatValuesComparatorSource(this, missingValue, sortMode, nested);
+                source = new FloatValuesComparatorSource(this, missingValue, sortMode, nested);
+                break;
+
             case DOUBLE:
-                return new DoubleValuesComparatorSource(this, missingValue, sortMode, nested);
+                source = new DoubleValuesComparatorSource(this, missingValue, sortMode, nested);
+                break;
+
             default:
                 assert !numericType.isFloatingPoint();
-                return new LongValuesComparatorSource(this, missingValue, sortMode, nested);
+                source = new LongValuesComparatorSource(this, missingValue, sortMode, nested);
+                break;
         }
+
+        /**
+         * Check if we can use a simple {@link SortedNumericSortField} compatible with index sorting and
+         * returns a custom sort field otherwise.
+         */
+        if (nested != null
+                || (sortMode != MultiValueMode.MAX && sortMode != MultiValueMode.MIN)
+                || numericType == NumericType.HALF_FLOAT) {
+            return new SortField(fieldName, source, reverse);
+        }
+
+        final SortField sortField;
+        final SortedNumericSelector.Type selectorType = sortMode == MultiValueMode.MAX ?
+            SortedNumericSelector.Type.MAX : SortedNumericSelector.Type.MIN;
+        switch (numericType) {
+            case FLOAT:
+                sortField = new SortedNumericSortField(fieldName, SortField.Type.FLOAT, reverse, selectorType);
+                break;
+
+            case DOUBLE:
+                sortField = new SortedNumericSortField(fieldName, SortField.Type.DOUBLE, reverse, selectorType);
+                break;
+
+            default:
+                assert !numericType.isFloatingPoint();
+                sortField = new SortedNumericSortField(fieldName, SortField.Type.LONG, reverse, selectorType);
+                break;
+        }
+        sortField.setMissingValue(source.missingObject(missingValue, reverse));
+        return sortField;
     }
 
     @Override

core/src/main/java/org/elasticsearch/index/fielddata/plain/SortedSetDVOrdinalsIndexFieldData.java

@@ -22,10 +22,13 @@ package org.elasticsearch.index.fielddata.plain;
 import org.apache.lucene.index.DirectoryReader;
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.index.RandomAccessOrds;
+import org.apache.lucene.search.SortField;
+import org.apache.lucene.search.SortedSetSortField;
+import org.apache.lucene.search.SortedSetSelector;
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.fielddata.AtomicOrdinalsFieldData;
-import org.elasticsearch.index.fielddata.IndexFieldData;
 import org.elasticsearch.index.fielddata.IndexFieldData.XFieldComparatorSource.Nested;
 import org.elasticsearch.index.fielddata.IndexFieldDataCache;
 import org.elasticsearch.index.fielddata.IndexOrdinalsFieldData;
@@ -55,8 +58,22 @@ public class SortedSetDVOrdinalsIndexFieldData extends DocValuesIndexFieldData i
     }
 
     @Override
-    public org.elasticsearch.index.fielddata.IndexFieldData.XFieldComparatorSource comparatorSource(Object missingValue, MultiValueMode sortMode, Nested nested) {
+    public SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested, boolean reverse) {
-        return new BytesRefFieldComparatorSource((IndexFieldData<?>) this, missingValue, sortMode, nested);
+        XFieldComparatorSource source = new BytesRefFieldComparatorSource(this, missingValue, sortMode, nested);
+        /**
+         * Check if we can use a simple {@link SortedSetSortField} compatible with index sorting and
+         * returns a custom sort field otherwise.
+         */
+        if (nested != null ||
+                (sortMode != MultiValueMode.MAX && sortMode != MultiValueMode.MIN) ||
+                (source.sortMissingLast(missingValue) == false && source.sortMissingFirst(missingValue) == false)) {
+            return new SortField(getFieldName(), source, reverse);
+        }
+        SortField sortField = new SortedSetSortField(fieldName, reverse,
+            sortMode == MultiValueMode.MAX ? SortedSetSelector.Type.MAX : SortedSetSelector.Type.MIN);
+        sortField.setMissingValue(source.sortMissingLast(missingValue) ^ reverse ?
+            SortedSetSortField.STRING_LAST : SortedSetSortField.STRING_FIRST);
+        return sortField;
     }
 
     @Override

core/src/main/java/org/elasticsearch/index/mapper/ScaledFloatFieldMapper.java

@@ -28,8 +28,10 @@ import org.apache.lucene.index.NumericDocValues;
 import org.apache.lucene.index.SortedNumericDocValues;
 import org.apache.lucene.search.BoostQuery;
 import org.apache.lucene.search.Query;
+import org.apache.lucene.search.SortField;
 import org.elasticsearch.action.fieldstats.FieldStats;
 import org.elasticsearch.common.Explicit;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
@@ -492,9 +494,9 @@ public class ScaledFloatFieldMapper extends FieldMapper {
         }
 
         @Override
-        public org.elasticsearch.index.fielddata.IndexFieldData.XFieldComparatorSource comparatorSource(Object missingValue,
+        public SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested, boolean reverse) {
-                MultiValueMode sortMode, Nested nested) {
+            final XFieldComparatorSource source = new DoubleValuesComparatorSource(this, missingValue, sortMode, nested);
-            return new DoubleValuesComparatorSource(this, missingValue, sortMode, nested);
+            return new SortField(getFieldName(), source, reverse);
         }
 
         @Override

core/src/main/java/org/elasticsearch/search/searchafter/SearchAfterBuilder.java

@@ -21,6 +21,8 @@ package org.elasticsearch.search.searchafter;
 
 import org.apache.lucene.search.FieldDoc;
 import org.apache.lucene.search.SortField;
+import org.apache.lucene.search.SortedNumericSortField;
+import org.apache.lucene.search.SortedSetSortField;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.ParsingException;
@@ -128,12 +130,23 @@ public class SearchAfterBuilder implements ToXContent, Writeable {
         return new FieldDoc(Integer.MAX_VALUE, 0, fieldValues);
     }
 
+    private static SortField.Type extractSortType(SortField sortField) {
+        if (sortField instanceof SortedSetSortField) {
+            return SortField.Type.STRING;
+        } else if (sortField instanceof SortedNumericSortField) {
+            return ((SortedNumericSortField) sortField).getNumericType();
+        } else {
+            return sortField.getType();
+        }
+    }
+
     private static Object convertValueFromSortField(Object value, SortField sortField, DocValueFormat format) {
         if (sortField.getComparatorSource() instanceof IndexFieldData.XFieldComparatorSource) {
             IndexFieldData.XFieldComparatorSource cmpSource = (IndexFieldData.XFieldComparatorSource) sortField.getComparatorSource();
             return convertValueFromSortType(sortField.getField(), cmpSource.reducedType(), value, format);
         }
-        return convertValueFromSortType(sortField.getField(), sortField.getType(), value, format);
+        SortField.Type sortType = extractSortType(sortField);
+        return convertValueFromSortType(sortField.getField(), sortType, value, format);
     }
 
     private static Object convertValueFromSortType(String fieldName, SortField.Type sortType, Object value, DocValueFormat format) {

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

@@ -279,9 +279,7 @@ public class FieldSortBuilder extends SortBuilder<FieldSortBuilder> {
                     && (sortMode == SortMode.SUM || sortMode == SortMode.AVG || sortMode == SortMode.MEDIAN)) {
                 throw new QueryShardException(context, "we only support AVG, MEDIAN and SUM on number based fields");
             }
-            IndexFieldData.XFieldComparatorSource fieldComparatorSource = fieldData
+            SortField field = fieldData.sortField(missing, localSortMode, nested, reverse);
-                    .comparatorSource(missing, localSortMode, nested);
-            SortField field = new SortField(fieldType.name(), fieldComparatorSource, reverse);
             return new SortFieldAndFormat(field, fieldType.docValueFormat(null, null));
         }
     }

core/src/test/java/org/elasticsearch/index/fielddata/AbstractFieldDataImplTestCase.java

@@ -109,9 +109,9 @@ public abstract class AbstractFieldDataImplTestCase extends AbstractFieldDataTes
 
             IndexSearcher searcher = new IndexSearcher(readerContext.reader());
             TopFieldDocs topDocs;
-
+            SortField sortField = indexFieldData.sortField(null, MultiValueMode.MIN, null, false);
             topDocs = searcher.search(new MatchAllDocsQuery(), 10,
-                    new Sort(new SortField("value", indexFieldData.comparatorSource(null, MultiValueMode.MIN, null))));
+                    new Sort(sortField));
             assertThat(topDocs.totalHits, equalTo(3));
             assertThat(topDocs.scoreDocs[0].doc, equalTo(1));
             assertThat(toString(((FieldDoc) topDocs.scoreDocs[0]).fields[0]), equalTo(one()));
@@ -120,8 +120,9 @@ public abstract class AbstractFieldDataImplTestCase extends AbstractFieldDataTes
             assertThat(topDocs.scoreDocs[2].doc, equalTo(2));
             assertThat(toString(((FieldDoc) topDocs.scoreDocs[2]).fields[0]), equalTo(three()));
 
+            sortField = indexFieldData.sortField(null, MultiValueMode.MAX, null, true);
             topDocs = searcher.search(new MatchAllDocsQuery(), 10,
-                    new Sort(new SortField("value", indexFieldData.comparatorSource(null, MultiValueMode.MAX, null), true)));
+                    new Sort(sortField));
             assertThat(topDocs.totalHits, equalTo(3));
             assertThat(topDocs.scoreDocs[0].doc, equalTo(2));
             assertThat(topDocs.scoreDocs[1].doc, equalTo(0));
@@ -182,14 +183,16 @@ public abstract class AbstractFieldDataImplTestCase extends AbstractFieldDataTes
             assertValues(bytesValues, 2, three());
 
             IndexSearcher searcher = new IndexSearcher(DirectoryReader.open(writer));
-            TopFieldDocs topDocs = searcher.search(new MatchAllDocsQuery(), 10, new Sort(new SortField("value", indexFieldData.comparatorSource(null, MultiValueMode.MIN, null))));
+            SortField sortField = indexFieldData.sortField(null, MultiValueMode.MIN, null, false);
+            TopFieldDocs topDocs = searcher.search(new MatchAllDocsQuery(), 10, new Sort(sortField));
             assertThat(topDocs.totalHits, equalTo(3));
             assertThat(topDocs.scoreDocs.length, equalTo(3));
             assertThat(topDocs.scoreDocs[0].doc, equalTo(1));
             assertThat(topDocs.scoreDocs[1].doc, equalTo(0));
-            assertThat(topDocs.scoreDocs[2].doc, equalTo(2));
+            assertThat(topDocs.scoreDocs[2].doc, equalTo(2))
-
+            ;
-            topDocs = searcher.search(new MatchAllDocsQuery(), 10, new Sort(new SortField("value", indexFieldData.comparatorSource(null, MultiValueMode.MAX, null), true)));
+            sortField = indexFieldData.sortField(null, MultiValueMode.MAX, null, true);
+            topDocs = searcher.search(new MatchAllDocsQuery(), 10, new Sort(sortField));
             assertThat(topDocs.totalHits, equalTo(3));
             assertThat(topDocs.scoreDocs.length, equalTo(3));
             assertThat(topDocs.scoreDocs[0].doc, equalTo(0));
@@ -245,8 +248,10 @@ public abstract class AbstractFieldDataImplTestCase extends AbstractFieldDataTes
         IndexFieldData indexFieldData = getForField("value");
 
         IndexSearcher searcher = new IndexSearcher(DirectoryReader.open(writer));
+        SortField sortField =
+            indexFieldData.sortField(null, MultiValueMode.MIN, null, false);
         TopFieldDocs topDocs = searcher.search(new MatchAllDocsQuery(), 10,
-                new Sort(new SortField("value", indexFieldData.comparatorSource(null, MultiValueMode.MIN, null))));
+                new Sort(sortField));
         assertThat(topDocs.totalHits, equalTo(8));
         assertThat(topDocs.scoreDocs.length, equalTo(8));
         assertThat(topDocs.scoreDocs[0].doc, equalTo(7));
@@ -266,8 +271,9 @@ public abstract class AbstractFieldDataImplTestCase extends AbstractFieldDataTes
         assertThat(topDocs.scoreDocs[7].doc, equalTo(5));
         assertThat((BytesRef) ((FieldDoc) topDocs.scoreDocs[7]).fields[0], equalTo(null));
 
+        sortField = indexFieldData.sortField(null, MultiValueMode.MAX, null, true);
         topDocs = searcher.search(new MatchAllDocsQuery(), 10,
-                new Sort(new SortField("value", indexFieldData.comparatorSource(null, MultiValueMode.MAX, null), true)));
+                new Sort(sortField));
         assertThat(topDocs.totalHits, equalTo(8));
         assertThat(topDocs.scoreDocs.length, equalTo(8));
         assertThat(topDocs.scoreDocs[0].doc, equalTo(6));

core/src/test/java/org/elasticsearch/index/fielddata/AbstractStringFieldDataTestCase.java

@@ -264,8 +264,8 @@ public abstract class AbstractStringFieldDataTestCase extends AbstractFieldDataI
         final IndexFieldData indexFieldData = getForField("value");
         final String missingValue = values[1];
         IndexSearcher searcher = new IndexSearcher(DirectoryReader.open(writer));
-        XFieldComparatorSource comparator = indexFieldData.comparatorSource(missingValue, MultiValueMode.MIN, null);
+        SortField sortField = indexFieldData.sortField(missingValue, MultiValueMode.MIN, null, reverse);
-        TopFieldDocs topDocs = searcher.search(new MatchAllDocsQuery(), randomBoolean() ? numDocs : randomIntBetween(10, numDocs), new Sort(new SortField("value", comparator, reverse)));
+        TopFieldDocs topDocs = searcher.search(new MatchAllDocsQuery(), randomBoolean() ? numDocs : randomIntBetween(10, numDocs), new Sort(sortField));
         assertEquals(numDocs, topDocs.totalHits);
         BytesRef previousValue = reverse ? UnicodeUtil.BIG_TERM : new BytesRef();
         for (int i = 0; i < topDocs.scoreDocs.length; ++i) {
@@ -318,8 +318,8 @@ public abstract class AbstractStringFieldDataTestCase extends AbstractFieldDataI
         }
         final IndexFieldData indexFieldData = getForField("value");
         IndexSearcher searcher = new IndexSearcher(DirectoryReader.open(writer));
-        XFieldComparatorSource comparator = indexFieldData.comparatorSource(first ? "_first" : "_last", MultiValueMode.MIN, null);
+        SortField sortField = indexFieldData.sortField(first ? "_first" : "_last", MultiValueMode.MIN, null, reverse);
-        TopFieldDocs topDocs = searcher.search(new MatchAllDocsQuery(), randomBoolean() ? numDocs : randomIntBetween(10, numDocs), new Sort(new SortField("value", comparator, reverse)));
+        TopFieldDocs topDocs = searcher.search(new MatchAllDocsQuery(), randomBoolean() ? numDocs : randomIntBetween(10, numDocs), new Sort(sortField));
         assertEquals(numDocs, topDocs.totalHits);
         BytesRef previousValue = first ? null : reverse ? UnicodeUtil.BIG_TERM : new BytesRef();
         for (int i = 0; i < topDocs.scoreDocs.length; ++i) {

core/src/test/java/org/elasticsearch/index/fielddata/NoOrdinalsStringFieldDataTests.java

@@ -20,6 +20,8 @@
 package org.elasticsearch.index.fielddata;
 
 import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.search.SortField;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.fielddata.IndexFieldData.XFieldComparatorSource.Nested;
 import org.elasticsearch.index.fielddata.fieldcomparator.BytesRefFieldComparatorSource;
@@ -53,8 +55,9 @@ public class NoOrdinalsStringFieldDataTests extends PagedBytesStringFieldDataTes
             }
 
             @Override
-            public XFieldComparatorSource comparatorSource(Object missingValue, MultiValueMode sortMode, Nested nested) {
+            public SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, Nested nested, boolean reverse) {
-                return new BytesRefFieldComparatorSource(this, missingValue, sortMode, nested);
+                XFieldComparatorSource source = new BytesRefFieldComparatorSource(this, missingValue, sortMode, nested);
+                return new SortField(getFieldName(), source, reverse);
             }
 
             @Override

core/src/test/java/org/elasticsearch/index/fielddata/ParentChildFieldDataTests.java

@@ -172,9 +172,8 @@ public class ParentChildFieldDataTests extends AbstractFieldDataTestCase {
     public void testSorting() throws Exception {
         IndexFieldData indexFieldData = getForField(parentType);
         IndexSearcher searcher = new IndexSearcher(DirectoryReader.open(writer));
-        IndexFieldData.XFieldComparatorSource comparator = indexFieldData.comparatorSource("_last", MultiValueMode.MIN, null);
+        SortField sortField = indexFieldData.sortField("_last", MultiValueMode.MIN, null, false);
-
+        TopFieldDocs topDocs = searcher.search(new MatchAllDocsQuery(), 10, new Sort(sortField));
-        TopFieldDocs topDocs = searcher.search(new MatchAllDocsQuery(), 10, new Sort(new SortField(ParentFieldMapper.joinField(parentType), comparator, false)));
         assertThat(topDocs.totalHits, equalTo(8));
         assertThat(topDocs.scoreDocs.length, equalTo(8));
         assertThat(topDocs.scoreDocs[0].doc, equalTo(0));
@@ -194,7 +193,8 @@ public class ParentChildFieldDataTests extends AbstractFieldDataTestCase {
         assertThat(topDocs.scoreDocs[7].doc, equalTo(7));
         assertThat(((BytesRef) ((FieldDoc) topDocs.scoreDocs[7]).fields[0]), equalTo(null));
 
-        topDocs = searcher.search(new MatchAllDocsQuery(), 10, new Sort(new SortField(ParentFieldMapper.joinField(parentType), comparator, true)));
+        sortField = indexFieldData.sortField("_last", MultiValueMode.MIN, null, true);
+        topDocs = searcher.search(new MatchAllDocsQuery(), 10, new Sort(sortField));
         assertThat(topDocs.totalHits, equalTo(8));
         assertThat(topDocs.scoreDocs.length, equalTo(8));
         assertThat(topDocs.scoreDocs[0].doc, equalTo(3));

core/src/test/java/org/elasticsearch/index/query/functionscore/FunctionScoreTests.java

@@ -37,6 +37,7 @@ import org.apache.lucene.search.Scorer;
 import org.apache.lucene.search.TermQuery;
 import org.apache.lucene.search.TopDocs;
 import org.apache.lucene.search.Weight;
+import org.apache.lucene.search.SortField;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.util.Accountable;
 import org.apache.lucene.util.BytesRef;
@@ -135,8 +136,7 @@ public class FunctionScoreTests extends ESTestCase {
         }
 
         @Override
-        public IndexFieldData.XFieldComparatorSource comparatorSource(@Nullable Object missingValue, MultiValueMode sortMode,
+        public SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, XFieldComparatorSource.Nested nested, boolean reverse) {
-                IndexFieldData.XFieldComparatorSource.Nested nested) {
             throw new UnsupportedOperationException(UNSUPPORTED);
         }
 
@@ -225,8 +225,7 @@ public class FunctionScoreTests extends ESTestCase {
         }
 
         @Override
-        public XFieldComparatorSource comparatorSource(@Nullable Object missingValue, MultiValueMode sortMode,
+        public SortField sortField(@Nullable Object missingValue, MultiValueMode sortMode, XFieldComparatorSource.Nested nested, boolean reverse) {
-                XFieldComparatorSource.Nested nested) {
             throw new UnsupportedOperationException(UNSUPPORTED);
         }
 

core/src/test/java/org/elasticsearch/index/search/nested/NestedSortingTests.java

@@ -113,9 +113,9 @@ public class NestedSortingTests extends AbstractFieldDataTestCase {
     private TopDocs getTopDocs(IndexSearcher searcher, IndexFieldData<?> indexFieldData, String missingValue, MultiValueMode sortMode, int n, boolean reverse) throws IOException {
         Query parentFilter = new TermQuery(new Term("__type", "parent"));
         Query childFilter = new TermQuery(new Term("__type", "child"));
-        XFieldComparatorSource nestedComparatorSource = indexFieldData.comparatorSource(missingValue, sortMode, createNested(searcher, parentFilter, childFilter));
+        SortField sortField = indexFieldData.sortField(missingValue, sortMode, createNested(searcher, parentFilter, childFilter), reverse);
         Query query = new ConstantScoreQuery(parentFilter);
-        Sort sort = new Sort(new SortField("f", nestedComparatorSource, reverse));
+        Sort sort = new Sort(sortField);
         return searcher.search(query, n, sort);
     }
 

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

@@ -39,7 +39,7 @@ public class FieldSortBuilderTests extends AbstractSortTestCase<FieldSortBuilder
     private List<Object> missingContent = Arrays.asList(
             "_last",
             "_first",
-            randomAsciiOfLength(10), randomUnicodeOfCodepointLengthBetween(5, 15),
+            Integer.toString(randomInt()),
             randomInt());