Allow format sort values of date fields (#70357)

If a search after request targets multiple indices and some of its sort 
field has type `date` in one index but `date_nanos` in other indices,
then Elasticsearch won't interpret the search_after parameter correctly
in every target index. The sort value of a date field by default is a
long of milliseconds since the epoch while a date_nanos field is a long
of nanoseconds.

This commit introduces the `format` parameter in the sort field so a 
sort value of a date or date_nanos will be formatted using a date format
in a search response.

The below example illustrates how to use this new parameter.

```js
{
    "query": {
        "match_all": {}
    },
    "sort": [
        {
            "timestamp": { 
                "order": "asc",
                "format": "strict_date_optional_time_nanos"
           }
        }
    ]
}
```

```js
{
    "query": {
        "match_all": {}
    },
    "sort": [
        {
            "timestamp": { 
                "order": "asc",
                "format": "strict_date_optional_time_nanos"
            }
        }
    ],
    "search_after": [
        "2015-01-01T12:10:30.123456789Z" // in `strict_date_optional_time_nanos` format
    ]
}
```

Closes #69192

docs/reference/search/search-your-data/paginate-search-results.asciidoc

@@ -81,6 +81,12 @@ NOTE: Search after requests have optimizations that make them faster when the so
 order is `_shard_doc` and total hits are not tracked. If you want to iterate over all documents regardless of the
 order, this is the most efficient option.
 
+IMPORTANT: If the `sort` field is a <<date,`date`>> in some target data streams or indices
+but a <<date_nanos,`date_nanos`>> field in other targets, use the `numeric_type` parameter
+to convert the values to a single resolution and the `format` parameter to specify a
+<<mapping-date-format, date format>> for the `sort` field. Otherwise, {es} won't interpret
+the search after parameter correctly in each request.
+
 [source,console]
 ----
 GET /_search
@@ -96,7 +102,7 @@ GET /_search
 	    "keep_alive": "1m"
   },
   "sort": [ <2>
-    {"@timestamp": "asc"}
+    {"@timestamp": {"order": "asc", "format": "strict_date_optional_time_nanos", "numeric_type" : "date_nanos" }}
   ]
 }
 ----
@@ -107,7 +113,7 @@ GET /_search
 
 The search response includes an array of `sort` values for each hit. If you used
 a PIT, a tiebreaker is included as the last `sort` values for each hit.
-This tiebreaker called `_shard_doc` is added automically on every search requests that use a PIT.
+This tiebreaker called `_shard_doc` is added automatically on every search requests that use a PIT.
 The `_shard_doc` value is the combination of the shard index within the PIT and the Lucene's internal doc ID,
 it is unique per document and constant within a PIT.
 You can also add the tiebreaker explicitly in the search request to customize the order:
@@ -127,7 +133,7 @@ GET /_search
 	    "keep_alive": "1m"
   },
   "sort": [ <2>
-    {"@timestamp": "asc"},
+    {"@timestamp": {"order": "asc", "format": "strict_date_optional_time_nanos"}},
     {"_shard_doc": "desc"}
   ]
 }
@@ -156,7 +162,7 @@ GET /_search
         "_score" : null,
         "_source" : ...,
         "sort" : [                                <2>
-          4098435132000,
+          "2021-05-20T05:30:04.832Z",
           4294967298                              <3>
         ]
       }
@@ -190,10 +196,10 @@ GET /_search
 	    "keep_alive": "1m"
   },
   "sort": [
-    {"@timestamp": "asc"}
+    {"@timestamp": {"order": "asc", "format": "strict_date_optional_time_nanos"}}
   ],
   "search_after": [                                <2>
-    4098435132000,
+    "2021-05-20T05:30:04.832Z",
     4294967298
   ],
   "track_total_hits": false                        <3>

docs/reference/search/search-your-data/sort-search-results.asciidoc

@@ -31,7 +31,7 @@ PUT /my-index-000001
 GET /my-index-000001/_search
 {
   "sort" : [
-    { "post_date" : {"order" : "asc"}},
+    { "post_date" : {"order" : "asc", "format": "strict_date_optional_time_nanos"}},
     "user",
     { "name" : "desc" },
     { "age" : "desc" },
@@ -51,8 +51,25 @@ should sort by `_doc`. This especially helps when <<scroll-search-results,scroll
 [discrete]
 === Sort Values
 
-The sort values for each document returned are also returned as part of
-the response.
+The search response includes `sort` values for each document. Use the `format`
+parameter to specify a <<built-in-date-formats,date format>> for the `sort`
+values of <<date,`date`>> and <<date_nanos,`date_nanos`>> fields. The following
+search returns `sort` values for the `post_date` field in the
+`strict_date_optional_time_nanos` format.
+
+[source,console]
+--------------------------------------------------
+GET /my-index-000001/_search
+{
+  "sort" : [
+    { "post_date" : {"format": "strict_date_optional_time_nanos"}}
+  ],
+  "query" : {
+    "term" : { "user" : "kimchy" }
+  }
+}
+--------------------------------------------------
+// TEST[continued]
 
 [discrete]
 === Sort Order

rest-api-spec/src/main/resources/rest-api-spec/test/search/90_search_after.yml

@@ -242,3 +242,134 @@
           size: 1
           sort: ["_shard_doc"]
           search_after: [ 0L ]
+
+---
+"Format sort values":
+  - skip:
+      version: " - 7.99.99"
+      reason: Format sort output is introduced in 8.0
+
+  - do:
+      indices.create:
+        index: test
+        body:
+          mappings:
+            properties:
+              timestamp:
+                type: date
+                format: yyyy-MM-dd HH:mm:ss.SSS
+  - do:
+      indices.create:
+        index: test_nanos
+        body:
+          mappings:
+            properties:
+              timestamp:
+                type: date_nanos
+                format: dd/MM/yyyy HH:mm:ss.SSS
+  - do:
+      bulk:
+        refresh: true
+        index: test
+        body: |
+          {"index":{}}
+          {"timestamp":"2021-10-13 00:30:04.828"}
+          {"index":{}}
+          {"timestamp":"2021-06-11 04:30:04.828"}
+          {"index":{}}
+          {"timestamp":"2021-02-11 08:30:04.828"}
+  - do:
+      bulk:
+        refresh: true
+        index: test_nanos
+        body: |
+          {"index":{}}
+          {"timestamp":"21/08/2021 03:30:04.732"}
+          {"index":{}}
+          {"timestamp":"20/05/2021 05:30:04.832"}
+          {"index":{}}
+          {"timestamp":"15/04/2021 06:30:04.821"}
+
+  - do:
+      search:
+        index: test
+        body:
+          size: 1
+          sort: [{timestamp: {"order" : "asc", "format": "strict_date_optional_time_nanos"}}]
+  - match: {hits.total.value: 3 }
+  - length: {hits.hits: 1 }
+  - match: {hits.hits.0._source.timestamp: "2021-02-11 08:30:04.828" }
+  - match: {hits.hits.0.sort: ["2021-02-11T08:30:04.828Z"] }
+
+  - do:
+      search:
+        index: test
+        body:
+          size: 1
+          sort: [{timestamp: {"order" : "asc", "format": "strict_date_optional_time_nanos"}}]
+          search_after: ["2021-02-11T08:30:04.828Z"]
+  - match: {hits.total.value: 3 }
+  - length: {hits.hits: 1 }
+  - match: {hits.hits.0._source.timestamp: "2021-06-11 04:30:04.828" }
+  - match: {hits.hits.0.sort: ["2021-06-11T04:30:04.828Z"] }
+
+  # mismatch format
+  - do:
+      catch: /failed to parse date field/
+      search:
+        index: test
+        body:
+          size: 1
+          sort: [{ timestamp: {"order" : "asc", "format": "yyyy-MM-dd HH:mm:ss.SSS"}}]
+          search_after: [ "2021-02-11T08:30:04.828Z" ]
+  - do:
+      catch: /failed to parse date field/
+      search:
+        index: test
+        body:
+          size: 1
+          sort: [ { timestamp: { "order": "asc", "format": "epoch_millis" } } ]
+          search_after: [ "2021-02-11T08:30:04.828Z" ]
+  - do:
+      search:
+        index: test
+        body:
+          size: 1
+          sort: [{timestamp: {"order" : "asc", "format": "yyyy-MM-dd | HH:mm:ss.SSS"}}]
+          search_after: ["2021-02-11 | 08:30:04.828"]
+  - match: {hits.total.value: 3 }
+  - length: {hits.hits: 1 }
+  - match: {hits.hits.0._source.timestamp: "2021-06-11 04:30:04.828" }
+  - match: {hits.hits.0.sort: ["2021-06-11 | 04:30:04.828"] }
+
+  # Mixed two types with numeric
+  - do:
+      search:
+        index: tes*
+        body:
+          size: 2
+          sort: [ { timestamp: { "order": "asc", "format": "strict_date_optional_time_nanos", "numeric_type": "date_nanos" } } ]
+  - match: { hits.total.value: 6 }
+  - length: { hits.hits: 2 }
+  - match: { hits.hits.0._index: test }
+  - match: { hits.hits.0._source.timestamp: "2021-02-11 08:30:04.828" }
+  - match: { hits.hits.0.sort: [ "2021-02-11T08:30:04.828Z" ] }
+  - match: { hits.hits.1._index: test_nanos }
+  - match: { hits.hits.1._source.timestamp: "15/04/2021 06:30:04.821" }
+  - match: { hits.hits.1.sort: [ "2021-04-15T06:30:04.821Z" ] }
+
+  - do:
+      search:
+        index: test*
+        body:
+          size: 2
+          sort: [ { timestamp: { "order": "asc", "format": "strict_date_optional_time_nanos", "numeric_type": "date" } } ]
+          search_after: [ "2021-04-15T06:30:04.821Z" ]
+  - match: { hits.total.value: 6 }
+  - length: { hits.hits: 2 }
+  - match: { hits.hits.0._index: test_nanos }
+  - match: { hits.hits.0._source.timestamp: "20/05/2021 05:30:04.832" }
+  - match: { hits.hits.0.sort: [ "2021-05-20T05:30:04.832Z" ] }
+  - match: { hits.hits.1._index: test }
+  - match: { hits.hits.1._source.timestamp: "2021-06-11 04:30:04.828" }
+  - match: { hits.hits.1.sort: [ "2021-06-11T04:30:04.828Z" ] }

server/src/internalClusterTest/java/org/elasticsearch/search/searchafter/SearchAfterIT.java

@@ -9,14 +9,20 @@
 package org.elasticsearch.search.searchafter;
 
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
+import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.search.SearchPhaseExecutionException;
 import org.elasticsearch.action.search.SearchRequestBuilder;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.ShardSearchFailure;
+import org.elasticsearch.action.support.WriteRequest;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.sort.SortBuilders;
 import org.elasticsearch.search.sort.SortOrder;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.hamcrest.Matchers;
@@ -30,6 +36,9 @@ import java.util.Arrays;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertFailures;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
+import static org.hamcrest.Matchers.arrayContaining;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 
@@ -182,6 +191,80 @@ public class SearchAfterIT extends ESIntegTestCase {
         assertSearchFromWithSortValues(INDEX_NAME, documents, reqSize);
     }
 
+    public void testWithCustomFormatSortValueOfDateField() throws Exception {
+        final XContentBuilder mappings = jsonBuilder();
+        mappings.startObject().startObject("properties");
+        {
+            mappings.startObject("start_date");
+            mappings.field("type", "date");
+            mappings.field("format", "yyyy-MM-dd");
+            mappings.endObject();
+        }
+        {
+            mappings.startObject("end_date");
+            mappings.field("type", "date");
+            mappings.field("format", "yyyy-MM-dd");
+            mappings.endObject();
+        }
+        mappings.endObject().endObject();
+        assertAcked(client().admin().indices().prepareCreate("test")
+            .setSettings(Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, between(1, 3)))
+            .setMapping(mappings));
+
+
+        client().prepareBulk().setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
+            .add(new IndexRequest("test").id("1").source("start_date", "2019-03-24", "end_date", "2020-01-21"))
+            .add(new IndexRequest("test").id("2").source("start_date", "2018-04-23", "end_date", "2021-02-22"))
+            .add(new IndexRequest("test").id("3").source("start_date", "2015-01-22", "end_date", "2022-07-23"))
+            .add(new IndexRequest("test").id("4").source("start_date", "2016-02-21", "end_date", "2024-03-24"))
+            .add(new IndexRequest("test").id("5").source("start_date", "2017-01-20", "end_date", "2025-05-28"))
+            .get();
+
+        SearchResponse resp = client().prepareSearch("test")
+            .addSort(SortBuilders.fieldSort("start_date").setFormat("dd/MM/yyyy"))
+            .addSort(SortBuilders.fieldSort("end_date").setFormat("yyyy-MM-dd"))
+            .setSize(2)
+            .get();
+        assertNoFailures(resp);
+        assertThat(resp.getHits().getHits()[0].getSortValues(), arrayContaining("22/01/2015", "2022-07-23"));
+        assertThat(resp.getHits().getHits()[1].getSortValues(), arrayContaining("21/02/2016", "2024-03-24"));
+
+        resp = client().prepareSearch("test")
+            .addSort(SortBuilders.fieldSort("start_date").setFormat("dd/MM/yyyy"))
+            .addSort(SortBuilders.fieldSort("end_date").setFormat("yyyy-MM-dd"))
+            .searchAfter(new String[]{"21/02/2016", "2024-03-24"})
+            .setSize(2)
+            .get();
+        assertNoFailures(resp);
+        assertThat(resp.getHits().getHits()[0].getSortValues(), arrayContaining("20/01/2017", "2025-05-28"));
+        assertThat(resp.getHits().getHits()[1].getSortValues(), arrayContaining("23/04/2018", "2021-02-22"));
+
+        resp = client().prepareSearch("test")
+            .addSort(SortBuilders.fieldSort("start_date").setFormat("dd/MM/yyyy"))
+            .addSort(SortBuilders.fieldSort("end_date")) // it's okay because end_date has the format "yyyy-MM-dd"
+            .searchAfter(new String[]{"21/02/2016", "2024-03-24"})
+            .setSize(2)
+            .get();
+        assertNoFailures(resp);
+        assertThat(resp.getHits().getHits()[0].getSortValues(), arrayContaining("20/01/2017", 1748390400000L));
+        assertThat(resp.getHits().getHits()[1].getSortValues(), arrayContaining("23/04/2018", 1613952000000L));
+
+        SearchRequestBuilder searchRequest = client().prepareSearch("test")
+            .addSort(SortBuilders.fieldSort("start_date").setFormat("dd/MM/yyyy"))
+            .addSort(SortBuilders.fieldSort("end_date").setFormat("epoch_millis"))
+            .searchAfter(new Object[]{"21/02/2016", 1748390400000L})
+            .setSize(2);
+        assertNoFailures(searchRequest.get());
+
+        searchRequest = client().prepareSearch("test")
+            .addSort(SortBuilders.fieldSort("start_date").setFormat("dd/MM/yyyy"))
+            .addSort(SortBuilders.fieldSort("end_date").setFormat("epoch_millis")) // wrong format
+            .searchAfter(new Object[]{"21/02/2016", "23/04/2018"})
+            .setSize(2);
+        assertFailures(searchRequest, RestStatus.BAD_REQUEST,
+            containsString("failed to parse date field [23/04/2018] with format [epoch_millis]"));
+    }
+
     private static class ListComparator implements Comparator<List> {
         @Override
         public int compare(List o1, List o2) {

server/src/main/java/org/elasticsearch/search/DocValueFormat.java

@@ -80,6 +80,18 @@ public interface DocValueFormat extends NamedWriteable {
         throw new UnsupportedOperationException();
     }
 
+    /**
+     * Formats a value of a sort field in a search response. This is used by {@link SearchSortValues}
+     * to avoid sending the internal representation of a value of a sort field in a search response.
+     * The default implementation formats {@link BytesRef} but leave other types as-is.
+     */
+    default Object formatSortValue(Object value) {
+        if (value instanceof BytesRef) {
+            return format((BytesRef) value);
+        }
+        return value;
+    }
+
     DocValueFormat RAW = new DocValueFormat() {
 
         @Override
@@ -166,12 +178,21 @@ public interface DocValueFormat extends NamedWriteable {
     static DocValueFormat withNanosecondResolution(final DocValueFormat format) {
         if (format instanceof DateTime) {
             DateTime dateTime = (DateTime) format;
-            return new DateTime(dateTime.formatter, dateTime.timeZone, DateFieldMapper.Resolution.NANOSECONDS);
+            return new DateTime(dateTime.formatter, dateTime.timeZone, DateFieldMapper.Resolution.NANOSECONDS,
+                dateTime.formatSortValues);
         } else {
             throw new IllegalArgumentException("trying to convert a known date time formatter to a nanosecond one, wrong field used?");
         }
     }
 
+    static DocValueFormat enableFormatSortValues(DocValueFormat format) {
+        if (format instanceof DateTime) {
+            DateTime dateTime = (DateTime) format;
+            return new DateTime(dateTime.formatter, dateTime.timeZone, dateTime.resolution, true);
+        }
+        throw new IllegalArgumentException("require a date_time formatter; got [" + format.getWriteableName() + "]");
+    }
+
     final class DateTime implements DocValueFormat {
 
         public static final String NAME = "date_time";
@@ -180,12 +201,18 @@ public interface DocValueFormat extends NamedWriteable {
         final ZoneId timeZone;
         private final DateMathParser parser;
         final DateFieldMapper.Resolution resolution;
+        final boolean formatSortValues;
 
         public DateTime(DateFormatter formatter, ZoneId timeZone, DateFieldMapper.Resolution resolution) {
+            this(formatter, timeZone, resolution, false);
+        }
+
+        private DateTime(DateFormatter formatter, ZoneId timeZone, DateFieldMapper.Resolution resolution, boolean formatSortValues) {
             this.formatter = formatter;
             this.timeZone = Objects.requireNonNull(timeZone);
             this.parser = formatter.toDateMathParser();
             this.resolution = resolution;
+            this.formatSortValues = formatSortValues;
         }
 
         public DateTime(StreamInput in) throws IOException {
@@ -201,6 +228,11 @@ public interface DocValueFormat extends NamedWriteable {
                 */
                 in.readBoolean();
             }
+            if (in.getVersion().onOrAfter(Version.V_8_0_0)) {
+                this.formatSortValues = in.readBoolean();
+            } else {
+                this.formatSortValues = false;
+            }
         }
 
         @Override
@@ -220,6 +252,9 @@ public interface DocValueFormat extends NamedWriteable {
                 */
                 out.writeBoolean(false);
             }
+            if (out.getVersion().onOrAfter(Version.V_8_0_0)) {
+                out.writeBoolean(formatSortValues);
+            }
         }
 
         public DateMathParser getDateMathParser() {
@@ -236,6 +271,16 @@ public interface DocValueFormat extends NamedWriteable {
             return format((long) value);
         }
 
+        @Override
+        public Object formatSortValue(Object value) {
+            if (formatSortValues) {
+                if (value instanceof Long) {
+                    return format((Long) value);
+                }
+            }
+            return value;
+        }
+
         @Override
         public long parseLong(String value, boolean roundUp, LongSupplier now) {
             return resolution.convert(parser.parse(value, now, roundUp, timeZone));
@@ -516,6 +561,14 @@ public interface DocValueFormat extends NamedWriteable {
             }
         }
 
+        @Override
+        public Object formatSortValue(Object value) {
+            if (value instanceof Long) {
+                return format((Long) value);
+            }
+            return value;
+        }
+
         /**
          * Double docValues of the unsigned_long field type are already in the formatted representation,
          * so we don't need to do anything here

server/src/main/java/org/elasticsearch/search/SearchSortValues.java

@@ -8,7 +8,6 @@
 
 package org.elasticsearch.search;
 
-import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
@@ -43,16 +42,11 @@ public class SearchSortValues implements ToXContentFragment, Writeable {
             throw new IllegalArgumentException("formattedSortValues and sortValueFormats must hold the same number of items");
         }
         this.rawSortValues = rawSortValues;
-        this.formattedSortValues = Arrays.copyOf(rawSortValues, rawSortValues.length);
+        this.formattedSortValues = new Object[rawSortValues.length];
         for (int i = 0; i < rawSortValues.length; ++i) {
-            Object sortValue = rawSortValues[i];
-            if (sortValue instanceof BytesRef) {
-                this.formattedSortValues[i] = sortValueFormats[i].format((BytesRef) sortValue);
-            } else if ((sortValue instanceof Long) && (sortValueFormats[i] == DocValueFormat.UNSIGNED_LONG_SHIFTED)) {
-                this.formattedSortValues[i] = sortValueFormats[i].format((Long) sortValue);
-            } else {
-                this.formattedSortValues[i] = sortValue;
-            }
+            final Object v = sortValueFormats[i].formatSortValue(rawSortValues[i]);
+            assert v == null || v instanceof String || v instanceof Number || v instanceof Boolean: v + " was not formatted";
+            formattedSortValues[i] = v;
         }
     }
 

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

@@ -66,6 +66,7 @@ public class FieldSortBuilder extends SortBuilder<FieldSortBuilder> {
     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");
+    public static final ParseField FORMAT = new ParseField("format");
 
     /**
      * special field name to sort by index order
@@ -94,6 +95,8 @@ public class FieldSortBuilder extends SortBuilder<FieldSortBuilder> {
 
     private NestedSortBuilder nestedSort;
 
+    private String format;
+
     /** Copy constructor. */
     public FieldSortBuilder(FieldSortBuilder template) {
         this(template.fieldName);
@@ -141,6 +144,9 @@ public class FieldSortBuilder extends SortBuilder<FieldSortBuilder> {
         if (in.getVersion().onOrAfter(Version.V_7_2_0)) {
             numericType = in.readOptionalString();
         }
+        if (in.getVersion().onOrAfter(Version.V_8_0_0)) {
+            format = in.readOptionalString();
+        }
     }
 
     @Override
@@ -158,6 +164,13 @@ public class FieldSortBuilder extends SortBuilder<FieldSortBuilder> {
         if (out.getVersion().onOrAfter(Version.V_7_2_0)) {
             out.writeOptionalString(numericType);
         }
+        if (out.getVersion().onOrAfter(Version.V_8_0_0)) {
+            out.writeOptionalString(format);
+        } else {
+            if (format != null) {
+                throw new IllegalArgumentException("Custom format for output of sort fields requires all nodes on 8.0 or later");
+            }
+        }
     }
 
     /** Returns the document field this sort should be based on. */
@@ -273,6 +286,22 @@ public class FieldSortBuilder extends SortBuilder<FieldSortBuilder> {
         return this;
     }
 
+    /**
+     * Returns the external format that is specified via {@link #setFormat(String)}
+     */
+    public String getFormat() {
+        return format;
+    }
+
+    /**
+     * Specifies a format specification that will be used to format the output value of this sort field.
+     * Currently, only "date" and "data_nanos" date types support this external format (i.e., date format).
+     */
+    public FieldSortBuilder setFormat(String format) {
+        this.format = format;
+        return this;
+    }
+
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
@@ -293,6 +322,9 @@ public class FieldSortBuilder extends SortBuilder<FieldSortBuilder> {
         if (numericType != null) {
             builder.field(NUMERIC_TYPE.getPreferredName(), numericType);
         }
+        if (format != null) {
+            builder.field(FORMAT.getPreferredName(), format);
+        }
         builder.endObject();
         builder.endObject();
         return builder;
@@ -353,11 +385,14 @@ public class FieldSortBuilder extends SortBuilder<FieldSortBuilder> {
                 isNanosecond = ((IndexNumericFieldData) fieldData).getNumericType() == NumericType.DATE_NANOSECONDS;
             }
         }
-        DocValueFormat format = fieldType.docValueFormat(null, null);
+        DocValueFormat formatter = fieldType.docValueFormat(format, null);
+        if (format != null) {
+            formatter = DocValueFormat.enableFormatSortValues(formatter);
+        }
         if (isNanosecond) {
-            format = DocValueFormat.withNanosecondResolution(format);
+            formatter = DocValueFormat.withNanosecondResolution(formatter);
         }
-        return new SortFieldAndFormat(field, format);
+        return new SortFieldAndFormat(field, formatter);
     }
 
     public boolean canRewriteToMatchNone() {
@@ -622,13 +657,14 @@ public class FieldSortBuilder extends SortBuilder<FieldSortBuilder> {
         return (Objects.equals(this.fieldName, builder.fieldName) && 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.numericType, builder.numericType);
+                && Objects.equals(this.numericType, builder.numericType)
+                && Objects.equals(this.format, builder.format);
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(this.fieldName, this.nestedSort, this.missing, this.order, this.sortMode,
-            this.unmappedType, this.numericType);
+            this.unmappedType, this.numericType, this.format);
     }
 
     @Override
@@ -658,6 +694,7 @@ public class FieldSortBuilder extends SortBuilder<FieldSortBuilder> {
         PARSER.declareString((b, v) -> b.sortMode(SortMode.fromString(v)), SORT_MODE);
         PARSER.declareObject(FieldSortBuilder::setNestedSort, (p, c) -> NestedSortBuilder.fromXContent(p), NESTED_FIELD);
         PARSER.declareString(FieldSortBuilder::setNumericType, NUMERIC_TYPE);
+        PARSER.declareString(FieldSortBuilder::setFormat, FORMAT);
     }
 
     @Override

server/src/test/java/org/elasticsearch/search/DocValueFormatTests.java

@@ -25,6 +25,7 @@ import java.util.ArrayList;
 import java.util.List;
 
 import static org.elasticsearch.search.aggregations.bucket.geogrid.GeoTileUtils.longEncode;
+import static org.hamcrest.Matchers.equalTo;
 
 public class DocValueFormatTests extends ESTestCase {
 
@@ -64,6 +65,17 @@ public class DocValueFormatTests extends ESTestCase {
         assertEquals(ZoneOffset.ofHours(1), ((DocValueFormat.DateTime) vf).timeZone);
         assertEquals(Resolution.MILLISECONDS, ((DocValueFormat.DateTime) vf).resolution);
 
+        dateFormat = (DocValueFormat.DateTime) DocValueFormat.enableFormatSortValues(dateFormat);
+        out = new BytesStreamOutput();
+        out.writeNamedWriteable(dateFormat);
+        in = new NamedWriteableAwareStreamInput(out.bytes().streamInput(), registry);
+        vf = in.readNamedWriteable(DocValueFormat.class);
+        assertEquals(DocValueFormat.DateTime.class, vf.getClass());
+        assertEquals("epoch_second", ((DocValueFormat.DateTime) vf).formatter.pattern());
+        assertEquals(ZoneOffset.ofHours(1), ((DocValueFormat.DateTime) vf).timeZone);
+        assertEquals(Resolution.MILLISECONDS, ((DocValueFormat.DateTime) vf).resolution);
+        assertTrue(dateFormat.formatSortValues);
+
         DocValueFormat.DateTime nanosDateFormat = new DocValueFormat.DateTime(formatter, ZoneOffset.ofHours(1),Resolution.NANOSECONDS);
         out = new BytesStreamOutput();
         out.writeNamedWriteable(nanosDateFormat);
@@ -204,4 +216,12 @@ public class DocValueFormatTests extends ESTestCase {
         assertEquals(0.859d, parser.parseDouble("0.859", true, null), 0.0d);
         assertEquals(0.8598023539251286d, parser.parseDouble("0.8598023539251286", true, null), 0.0d);
     }
+
+    public void testFormatSortFieldOutput() {
+        DateFormatter formatter = DateFormatter.forPattern("yyyy-MM-dd HH:mm:ss");
+        DocValueFormat.DateTime dateFormat = new DocValueFormat.DateTime(formatter, ZoneOffset.ofHours(1), Resolution.MILLISECONDS);
+        assertThat(dateFormat.formatSortValue(1415580798601L), equalTo(1415580798601L));
+        dateFormat = (DocValueFormat.DateTime) DocValueFormat.enableFormatSortValues(dateFormat);
+        assertThat(dateFormat.formatSortValue(1415580798601L), equalTo("2014-11-10 01:53:18"));
+    }
 }

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

@@ -109,13 +109,16 @@ public class FieldSortBuilderTests extends AbstractSortTestCase<FieldSortBuilder
         if (randomBoolean()) {
             builder.setNumericType(randomFrom(random(), "long", "double"));
         }
+        if (fieldName.equals("custom_date") && randomBoolean()) {
+            builder.setFormat(randomFrom("yyyy-MM-dd", "yyyy/MM/dd"));
+        }
         return builder;
     }
 
     @Override
     protected FieldSortBuilder mutate(FieldSortBuilder original) throws IOException {
         FieldSortBuilder mutated = new FieldSortBuilder(original);
-        int parameter = randomIntBetween(0, 5);
+        int parameter = randomIntBetween(0, 6);
         switch (parameter) {
         case 0:
             mutated.setNestedSort(
@@ -139,6 +142,9 @@ public class FieldSortBuilderTests extends AbstractSortTestCase<FieldSortBuilder
             mutated.setNumericType(randomValueOtherThan(original.getNumericType(),
                 () -> randomFrom("long", "double")));
             break;
+        case 6:
+            mutated.setFormat(randomValueOtherThan(original.getFormat(), () -> randomFrom("yyyy-MM-dd", "yyyy/MM/dd")));
+            break;
         default:
             throw new IllegalStateException("Unsupported mutation.");
         }
@@ -330,6 +336,29 @@ public class FieldSortBuilderTests extends AbstractSortTestCase<FieldSortBuilder
         assertThat(sortAndFormat.format, equalTo(DocValueFormat.RAW));
     }
 
+    public void testFormatDateTime() throws Exception {
+        SearchExecutionContext searchExecutionContext = createMockSearchExecutionContext();
+
+        SortFieldAndFormat sortAndFormat = SortBuilders.fieldSort("custom-date").build(searchExecutionContext);
+        assertThat(sortAndFormat.format.formatSortValue(1615580798601L), equalTo(1615580798601L));
+
+        sortAndFormat = SortBuilders.fieldSort("custom-date").setFormat("yyyy-MM-dd").build(searchExecutionContext);
+        assertThat(sortAndFormat.format.formatSortValue(1615580798601L), equalTo("2021-03-12"));
+
+        sortAndFormat = SortBuilders.fieldSort("custom-date").setFormat("epoch_millis").build(searchExecutionContext);
+        assertThat(sortAndFormat.format.formatSortValue(1615580798601L), equalTo("1615580798601"));
+
+        sortAndFormat = SortBuilders.fieldSort("custom-date").setFormat("yyyy/MM/dd HH:mm:ss").build(searchExecutionContext);
+        assertThat(sortAndFormat.format.formatSortValue(1615580798601L), equalTo("2021/03/12 20:26:38"));
+    }
+
+    public void testInvalidFormat() {
+        SearchExecutionContext searchExecutionContext = createMockSearchExecutionContext();
+        IllegalArgumentException error = expectThrows(IllegalArgumentException.class,
+            () -> SortBuilders.fieldSort("custom-keyword").setFormat("yyyy/MM/dd HH:mm:ss").build(searchExecutionContext));
+        assertThat(error.getMessage(), equalTo("Field [custom-keyword] of type [keyword] does not support custom formats"));
+    }
+
     @Override
     protected MappedFieldType provideMappedFieldType(String name) {
         if (name.equals(MAPPED_STRING_FIELDNAME)) {