Support joda style date patterns in 7.x (#52555)

If an index was created in version 6 and contain a date field with a joda-style pattern it should still be allowed to search and insert document into it.
Those created in 6 but date pattern starts with 8, should be considered as java style.

qa/rolling-upgrade/build.gradle

@@ -59,6 +59,7 @@ for (Version bwcVersion : bwcVersions.wireCompatible) {
     doFirst {
       project.delete("${buildDir}/cluster/shared/repo/${baseName}")
     }
+    systemProperty 'tests.upgrade_from_version', bwcVersion.toString()
     systemProperty 'tests.rest.suite', 'old_cluster'
     nonInputProperties.systemProperty('tests.rest.cluster', "${-> testClusters."${baseName}".allHttpSocketURI.join(",")}")
     nonInputProperties.systemProperty('tests.clustername', "${-> testClusters."${baseName}".getName()}")
@@ -71,7 +72,7 @@ for (Version bwcVersion : bwcVersions.wireCompatible) {
       testClusters."${baseName}".nextNodeToNextVersion()
     }
     systemProperty 'tests.rest.suite', 'mixed_cluster'
-    systemProperty 'tests.upgrade_from_version', project.version.replace("-SNAPSHOT", "")
+    systemProperty 'tests.upgrade_from_version', bwcVersion.toString()
     systemProperty 'tests.first_round', 'true'
     nonInputProperties.systemProperty('tests.rest.cluster', "${-> testClusters."${baseName}".allHttpSocketURI.join(",")}")
     nonInputProperties.systemProperty('tests.clustername', "${-> testClusters."${baseName}".getName()}")
@@ -84,7 +85,7 @@ for (Version bwcVersion : bwcVersions.wireCompatible) {
       testClusters."${baseName}".nextNodeToNextVersion()
     }
     systemProperty 'tests.rest.suite', 'mixed_cluster'
-    systemProperty 'tests.upgrade_from_version', project.version.replace("-SNAPSHOT", "")
+    systemProperty 'tests.upgrade_from_version', bwcVersion.toString()
     systemProperty 'tests.first_round', 'false'
     nonInputProperties.systemProperty('tests.rest.cluster', "${-> testClusters."${baseName}".allHttpSocketURI.join(",")}")
     nonInputProperties.systemProperty('tests.clustername', "${-> testClusters."${baseName}".getName()}")
@@ -97,6 +98,7 @@ for (Version bwcVersion : bwcVersions.wireCompatible) {
     }
     useCluster testClusters."${baseName}"
     systemProperty 'tests.rest.suite', 'upgraded_cluster'
+    systemProperty 'tests.upgrade_from_version', bwcVersion.toString()
 
     nonInputProperties.systemProperty('tests.rest.cluster', "${-> testClusters."${baseName}".allHttpSocketURI.join(",")}")
     nonInputProperties.systemProperty('tests.clustername', "${-> testClusters."${baseName}".getName()}")

qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/JodaCompatibilityIT.java

@@ -0,0 +1,269 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.upgrades;
+
+import org.apache.http.HttpStatus;
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.Version;
+import org.elasticsearch.client.Node;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.WarningsHandler;
+import org.elasticsearch.common.Booleans;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.search.DocValueFormat;
+import org.junit.BeforeClass;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Consumer;
+
+import static org.elasticsearch.rest.action.search.RestSearchAction.TOTAL_HITS_AS_INT_PARAM;
+
+/**
+ * This is test is meant to verify that when upgrading from 6.x version to 7.7 or newer it is able to parse date fields with joda pattern.
+ *
+ * The test is indexing documents and searches with use of joda or java pattern.
+ * In order to make sure that serialization logic is used a search call is executed 3 times (using all nodes).
+ * It cannot be guaranteed that serialization logic will always be used as it might happen that
+ * all shards are allocated on the same node and client is connecting to it.
+ * Because of this warnings assertions have to be ignored.
+ *
+ * A special flag used when serializing {@link DocValueFormat.DateTime#writeTo DocValueFormat.DateTime::writeTo}
+ * is used to indicate that an index was created in 6.x and has a joda pattern. The same flag is read when
+ * {@link DocValueFormat.DateTime#DateTime(StreamInput)} deserializing.
+ * When upgrading from 7.0-7.6 to 7.7 there is no way to tell if a pattern was created in 6.x as this flag cannot be added.
+ * Hence a skip assume section in init()
+ *
+ * @see org.elasticsearch.search.DocValueFormat.DateTime
+ */
+public class JodaCompatibilityIT extends AbstractRollingTestCase {
+
+    @BeforeClass
+    public static void init(){
+        assumeTrue("upgrading from 7.0-7.6 will fail parsing joda formats",
+            UPGRADE_FROM_VERSION.before(Version.V_7_0_0));
+    }
+
+    public void testJodaBackedDocValueAndDateFields() throws Exception {
+        switch (CLUSTER_TYPE) {
+            case OLD:
+                Request createTestIndex = indexWithDateField("joda_time", "YYYY-MM-dd'T'HH:mm:ssZZ");
+                createTestIndex.setOptions(ignoreWarnings());
+
+                Response resp = client().performRequest(createTestIndex);
+                assertEquals(HttpStatus.SC_OK, resp.getStatusLine().getStatusCode());
+
+                postNewDoc("joda_time", 1);
+
+                break;
+            case MIXED:
+                int minute = Booleans.parseBoolean(System.getProperty("tests.first_round")) ? 2 : 3;
+                postNewDoc("joda_time", minute);
+
+                Request search = dateRangeSearch("joda_time");
+                search.setOptions(ignoreWarnings());
+
+                performOnAllNodes(search, r -> assertEquals(HttpStatus.SC_OK, r.getStatusLine().getStatusCode()));
+                break;
+            case UPGRADED:
+                postNewDoc("joda_time", 4);
+
+                search = searchWithAgg("joda_time");
+                search.setOptions(ignoreWarnings());
+                //making sure all nodes were used for search
+                performOnAllNodes(search, r -> assertResponseHasAllDocuments(r));
+                break;
+        }
+    }
+
+    public void testJavaBackedDocValueAndDateFields() throws Exception {
+        switch (CLUSTER_TYPE) {
+            case OLD:
+                Request createTestIndex = indexWithDateField("java_time", "8yyyy-MM-dd'T'HH:mm:ssXXX");
+                Response resp = client().performRequest(createTestIndex);
+                assertEquals(HttpStatus.SC_OK, resp.getStatusLine().getStatusCode());
+
+                postNewDoc("java_time", 1);
+
+                break;
+            case MIXED:
+                int minute = Booleans.parseBoolean(System.getProperty("tests.first_round")) ? 2 : 3;
+                postNewDoc("java_time", minute);
+
+                Request search = dateRangeSearch("java_time");
+                Response searchResp = client().performRequest(search);
+                assertEquals(HttpStatus.SC_OK, searchResp.getStatusLine().getStatusCode());
+                break;
+            case UPGRADED:
+                postNewDoc("java_time", 4);
+
+                search = searchWithAgg("java_time");
+                //making sure all nodes were used for search
+                performOnAllNodes(search, r -> assertResponseHasAllDocuments(r));
+
+                break;
+        }
+    }
+
+    private RequestOptions ignoreWarnings() {
+        RequestOptions.Builder options = RequestOptions.DEFAULT.toBuilder();
+        options.setWarningsHandler(WarningsHandler.PERMISSIVE);
+        return options.build();
+    }
+
+    private void performOnAllNodes(Request search, Consumer<Response> consumer) throws IOException {
+        List<Node> nodes = client().getNodes();
+        for (Node node : nodes) {
+            client().setNodes(Collections.singletonList(node));
+            Response response = client().performRequest(search);
+            consumer.accept(response);
+            assertEquals(HttpStatus.SC_OK, response.getStatusLine().getStatusCode());
+        }
+        client().setNodes(nodes);
+    }
+
+    private void assertResponseHasAllDocuments(Response searchResp) {
+        assertEquals(HttpStatus.SC_OK, searchResp.getStatusLine().getStatusCode());
+        try {
+            assertEquals(removeWhiteSpace("{" +
+                    "  \"_shards\": {" +
+                    "    \"total\": 3," +
+                    "    \"successful\": 3" +
+                    "  },"+
+                    "  \"hits\": {" +
+                    "    \"total\": 4," +
+                    "    \"hits\": [" +
+                    "      {" +
+                    "        \"_source\": {" +
+                    "          \"datetime\": \"2020-01-01T00:00:01+01:00\"" +
+                    "        }" +
+                    "      }," +
+                    "      {" +
+                    "        \"_source\": {" +
+                    "          \"datetime\": \"2020-01-01T00:00:02+01:00\"" +
+                    "        }" +
+                    "      }," +
+                    "      {" +
+                    "        \"_source\": {" +
+                    "          \"datetime\": \"2020-01-01T00:00:03+01:00\"" +
+                    "        }" +
+                    "      }," +
+                    "      {" +
+                    "        \"_source\": {" +
+                    "          \"datetime\": \"2020-01-01T00:00:04+01:00\"" +
+                    "        }" +
+                    "      }" +
+                    "    ]" +
+                    "  }" +
+                    "}"),
+                EntityUtils.toString(searchResp.getEntity(), StandardCharsets.UTF_8));
+        } catch (IOException e) {
+            throw new AssertionError("Exception during response parising", e);
+        }
+    }
+
+    private String removeWhiteSpace(String input) {
+        return input.replaceAll("[\\n\\r\\t\\ ]", "");
+    }
+
+    private Request dateRangeSearch(String endpoint) {
+        Request search = new Request("GET", endpoint+"/_search");
+        search.addParameter(TOTAL_HITS_AS_INT_PARAM, "true");
+        search.addParameter("filter_path", "hits.total,hits.hits._source.datetime,_shards.total,_shards.successful");
+        search.setJsonEntity("" +
+                "{\n" +
+                "  \"track_total_hits\": true,\n" +
+                "  \"sort\": \"datetime\",\n" +
+                "  \"query\": {\n" +
+                "    \"range\": {\n" +
+                "      \"datetime\": {\n" +
+                "        \"gte\": \"2020-01-01T00:00:00+01:00\",\n" +
+                "        \"lte\": \"2020-01-02T00:00:00+01:00\"\n" +
+                "      }\n" +
+                "    }\n" +
+                "  }\n" +
+                "}\n"
+        );
+        return search;
+    }
+
+    private Request searchWithAgg(String endpoint) throws IOException {
+        Request search = new Request("GET", endpoint+"/_search");
+        search.addParameter(TOTAL_HITS_AS_INT_PARAM, "true");
+        search.addParameter("filter_path", "hits.total,hits.hits._source.datetime,_shards.total,_shards.successful");
+
+        search.setJsonEntity("{\n" +
+            "  \"track_total_hits\": true,\n" +
+            "  \"sort\": \"datetime\",\n" +
+            "  \"query\": {\n" +
+            "    \"range\": {\n" +
+            "      \"datetime\": {\n" +
+            "        \"gte\": \"2020-01-01T00:00:00+01:00\",\n" +
+            "        \"lte\": \"2020-01-02T00:00:00+01:00\"\n" +
+            "      }\n" +
+            "    }\n" +
+            "  },\n" +
+            "  \"aggs\" : {\n" +
+            "    \"docs_per_year\" : {\n" +
+            "      \"date_histogram\" : {\n" +
+            "        \"field\" : \"date\",\n" +
+            "        \"calendar_interval\" : \"year\"\n" +
+            "      }\n" +
+            "    }\n" +
+            "  }\n" +
+            "}\n"
+        );
+        return search;
+    }
+    private Request indexWithDateField(String indexName, String format) {
+        Request createTestIndex = new Request("PUT", indexName);
+        createTestIndex.addParameter("include_type_name", "false");
+        createTestIndex.setJsonEntity("{\n" +
+            "  \"settings\": {\n" +
+            "    \"index.number_of_shards\": 3\n" +
+            "  },\n" +
+            "  \"mappings\": {\n" +
+            "      \"properties\": {\n" +
+            "        \"datetime\": {\n" +
+            "          \"type\": \"date\",\n" +
+            "          \"format\": \"" + format + "\"\n" +
+            "        }\n" +
+            "      }\n" +
+            "  }\n" +
+            "}"
+        );
+        return createTestIndex;
+    }
+
+    private void postNewDoc(String endpoint, int minute) throws IOException {
+        Request putDoc = new Request("POST", endpoint+"/_doc");
+        putDoc.addParameter("refresh", "true");
+        putDoc.addParameter("wait_for_active_shards", "all");
+        putDoc.setJsonEntity("{\n" +
+            "  \"datetime\": \"2020-01-01T00:00:0" + minute + "+01:00\"\n" +
+            "}"
+        );
+        Response resp = client().performRequest(putDoc);
+        assertEquals(HttpStatus.SC_CREATED, resp.getStatusLine().getStatusCode());
+    }
+}

qa/rolling-upgrade/src/test/resources/rest-api-spec/test/mixed_cluster/20_date_range.yml

@@ -0,0 +1,39 @@
+---
+"Insert more docs to joda index":
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - '{"index": {"_index": "joda_for_range"}}'
+          - '{"time_frame": {"gte": "2019-01-01T00:00+01:00", "lte" : "2019-03-01T00:00+01:00"}}'
+
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: joda_for_range
+        body:
+          query:
+            range:
+              time_frame:
+                gte: "2019-02-01T00:00+01:00"
+                lte: "2019-02-01T00:00+01:00"
+
+---
+"Insert more docs to java index":
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - '{"index": {"_index": "java_for_range"}}'
+          - '{"time_frame": {"gte": "2019-01-01T00:00+01:00", "lte" : "2019-03-01T00:00+01:00"}}'
+
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: java_for_range
+        body:
+          query:
+            range:
+              time_frame:
+                gte: "2019-02-01T00:00+01:00"
+                lte: "2019-02-01T00:00+01:00"

qa/rolling-upgrade/src/test/resources/rest-api-spec/test/old_cluster/20_date_range.yml

@@ -0,0 +1,118 @@
+---
+"Create index with joda style index that is incompatible with java.time. (6.0)":
+  - skip:
+      features: "warnings"
+      version: "6.8.1 -"
+      reason: change of warning message
+  - do:
+      warnings:
+        - "Use of 'Y' (year-of-era) will change to 'y' in the next major version of Elasticsearch. Prefix your date format with '8' to use the new specifier."
+      indices.create:
+        index: joda_for_range
+        body:
+          settings:
+            index:
+              number_of_replicas: 2
+          mappings:
+              "properties":
+                "time_frame":
+                  "type": "date_range"
+                  "format": "YYYY-MM-dd'T'HH:mmZZ"
+
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - '{"index": {"_index": "joda_for_range"}}'
+          - '{"time_frame": {"gte": "2019-01-01T00:00+01:00", "lte" : "2019-03-01T00:00+01:00"}}'
+
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: joda_for_range
+        body:
+          query:
+            range:
+              time_frame:
+                gte: "2019-02-01T00:00+01:00"
+                lte: "2019-02-01T00:00+01:00"
+  - match: { hits.total: 1 }
+
+---
+"Create index with joda style index that is incompatible with java.time (>6.1)":
+  - skip:
+      features: "warnings"
+      version: " - 6.8.0, 7.0.0 -"
+      reason: change of warning message, we skip 7 becase this format will be considered java
+  - do:
+      warnings:
+        - "'Y' year-of-era should be replaced with 'y'. Use 'Y' for week-based-year.; 'Z' time zone offset/id fails when parsing 'Z' for Zulu timezone. Consider using 'X'. Prefix your date format with '8' to use the new specifier."
+      indices.create:
+        index: joda_for_range
+        body:
+          settings:
+            index:
+              number_of_replicas: 2
+          mappings:
+            "properties":
+              "time_frame":
+                "type": "date_range"
+                "format": "YYYY-MM-dd'T'HH:mmZZ"
+
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - '{"index": {"_index": "joda_for_range"}}'
+          - '{"time_frame": {"gte": "2019-01-01T00:00+01:00", "lte" : "2019-03-01T00:00+01:00"}}'
+
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: joda_for_range
+        body:
+          query:
+            range:
+              time_frame:
+                gte: "2019-02-01T00:00+01:00"
+                lte: "2019-02-01T00:00+01:00"
+  - match: { hits.total: 1 }
+
+---
+"Create index with java style index in 6":
+  - skip:
+      version: " - 6.7.99, 7.0.0 -"
+      reason: java.time patterns are allowed since 6.8
+  - do:
+      indices.create:
+        index: java_for_range
+        body:
+          settings:
+            index:
+              number_of_replicas: 2
+          mappings:
+            "properties":
+              "time_frame":
+                "type": "date_range"
+                "format": "8yyyy-MM-dd'T'HH:mmXXX"
+
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - '{"index": {"_index": "java_for_range"}}'
+          - '{"time_frame": {"gte": "2019-01-01T00:00+01:00", "lte" : "2019-03-01T00:00+01:00"}}'
+
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: java_for_range
+        body:
+          query:
+            range:
+              time_frame:
+                gte: "2019-02-01T00:00+01:00"
+                lte: "2019-02-01T00:00+01:00"
+  - match: { hits.total: 1 }
+
+

qa/rolling-upgrade/src/test/resources/rest-api-spec/test/upgraded_cluster/20_date_range.yml

@@ -0,0 +1,41 @@
+---
+"Verify that we can find results with joda style pattern":
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - '{"index": {"_index": "joda_for_range"}}'
+          - '{"time_frame": {"gte": "2019-01-01T00:00+01:00", "lte" : "2019-03-01T00:00+01:00"}}'
+
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: joda_for_range
+        body:
+          query:
+            range:
+              time_frame:
+                gte: "2019-02-01T00:00+01:00"
+                lte: "2019-02-01T00:00+01:00"
+
+
+---
+"Verify that we can find results with java style pattern":
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - '{"index": {"_index": "java_for_range"}}'
+          - '{"time_frame": {"gte": "2019-01-01T00:00+01:00", "lte" : "2019-03-01T00:00+01:00"}}'
+
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: java_for_range
+        body:
+          query:
+            range:
+              time_frame:
+                gte: "2019-02-01T00:00+01:00"
+                lte: "2019-02-01T00:00+01:00"
+

server/src/main/java/org/elasticsearch/common/joda/Joda.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.common.joda;
 
 import org.apache.logging.log4j.LogManager;
+import org.elasticsearch.Version;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.logging.DeprecationLogger;
 import org.elasticsearch.common.time.DateFormatter;
@@ -336,6 +337,18 @@ public class Joda {
         }
     };
 
+    /**
+     * Checks if a pattern is Joda-style.
+     * Joda style patterns are not always compatible with java.time patterns.
+     * @param version - creation version of the index where pattern was used
+     * @param pattern - the pattern to check
+     * @return - true if pattern is joda style, otherwise false
+     */
+    public static boolean isJodaPattern(Version version, String pattern) {
+        return version.before(Version.V_7_0_0)
+            && pattern.startsWith("8") == false;
+    }
+
     public static class EpochTimeParser implements DateTimeParser {
 
         private static final Pattern scientificNotation = Pattern.compile("[Ee]");

server/src/main/java/org/elasticsearch/common/joda/JodaDeprecationPatterns.java

@@ -30,7 +30,7 @@ import java.util.Set;
 import java.util.stream.Collectors;
 
 public class JodaDeprecationPatterns {
-    public static final String USE_NEW_FORMAT_SPECIFIERS = "Use new java.time date format specifiiers.";
+    public static final String USE_NEW_FORMAT_SPECIFIERS = "Use new java.time date format specifiers.";
     private static Map<String, String> JODA_PATTERNS_DEPRECATIONS = new LinkedHashMap<>();
 
     static {

server/src/main/java/org/elasticsearch/index/mapper/DateFieldMapper.java

@@ -38,6 +38,7 @@ import org.elasticsearch.common.Explicit;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.geo.ShapeRelation;
+import org.elasticsearch.common.joda.Joda;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.common.time.DateFormatters;
@@ -237,7 +238,13 @@ public final class DateFieldMapper extends FieldMapper {
 
             boolean hasPatternChanged = Strings.hasLength(pattern) && Objects.equals(pattern, dateTimeFormatter.pattern()) == false;
             if (hasPatternChanged || Objects.equals(builder.locale, dateTimeFormatter.locale()) == false) {
-                fieldType().setDateTimeFormatter(DateFormatter.forPattern(pattern).withLocale(locale));
+                DateFormatter formatter;
+                if (Joda.isJodaPattern(context.indexCreatedVersion(), pattern)) {
+                    formatter = Joda.forPattern(pattern).withLocale(locale);
+                } else {
+                    formatter = DateFormatter.forPattern(pattern).withLocale(locale);
+                }
+                fieldType().setDateTimeFormatter(formatter);
             }
 
             fieldType().setResolution(resolution);

server/src/main/java/org/elasticsearch/index/mapper/RangeFieldMapper.java

@@ -32,6 +32,7 @@ import org.elasticsearch.common.Explicit;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.geo.ShapeRelation;
+import org.elasticsearch.common.joda.Joda;
 import org.elasticsearch.common.network.InetAddresses;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
@@ -138,7 +139,13 @@ public class RangeFieldMapper extends FieldMapper {
                     Objects.equals(builder.pattern, formatter.pattern()) == false;
 
                 if (hasPatternChanged || Objects.equals(builder.locale, formatter.locale()) == false) {
-                    fieldType().setDateTimeFormatter(DateFormatter.forPattern(pattern).withLocale(locale));
+                    DateFormatter dateTimeFormatter;
+                    if (Joda.isJodaPattern(context.indexCreatedVersion(), pattern)) {
+                        dateTimeFormatter = Joda.forPattern(pattern).withLocale(locale);
+                    } else {
+                        dateTimeFormatter = DateFormatter.forPattern(pattern).withLocale(locale);
+                    }
+                    fieldType().setDateTimeFormatter(dateTimeFormatter);
                 }
             } else if (pattern != null) {
                 throw new IllegalArgumentException("field [" + name() + "] of type [" + fieldType().rangeType

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

@@ -25,6 +25,8 @@ import org.elasticsearch.Version;
 import org.elasticsearch.common.io.stream.NamedWriteable;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.joda.Joda;
+import org.elasticsearch.common.joda.JodaDateFormatter;
 import org.elasticsearch.common.network.InetAddresses;
 import org.elasticsearch.common.network.NetworkAddress;
 import org.elasticsearch.common.time.DateFormatter;
@@ -198,8 +200,8 @@ public interface DocValueFormat extends NamedWriteable {
         }
 
         public DateTime(StreamInput in) throws IOException {
-            this.formatter = DateFormatter.forPattern(in.readString());
+            String datePattern = in.readString();
-            this.parser = formatter.toDateMathParser();
+
             String zoneId = in.readString();
             if (in.getVersion().before(Version.V_7_0_0)) {
                 this.timeZone = DateUtils.of(zoneId);
@@ -208,6 +210,25 @@ public interface DocValueFormat extends NamedWriteable {
                 this.timeZone = ZoneId.of(zoneId);
                 this.resolution = DateFieldMapper.Resolution.ofOrdinal(in.readVInt());
             }
+            final boolean isJoda;
+            if (in.getVersion().onOrAfter(Version.V_7_7_0)) {
+                //if stream is from 7.7 Node it will have a flag indicating if format is joda
+                isJoda = in.readBoolean();
+            } else {
+                /*
+                 When received a stream from 6.0-6.latest Node it can be java if starts with 8 otherwise joda.
+
+                 If a stream is from [7.0 - 7.7) the boolean indicating that this is joda is not present.
+                 This means that if an index was created in 6.x using joda pattern and then cluster was upgraded to
+                 7.x but earlier then 7.0, there is no information that can tell that the index is using joda style pattern.
+                 It will be assumed that clusters upgrading from [7.0 - 7.7) are using java style patterns.
+                 */
+                isJoda = Joda.isJodaPattern(in.getVersion(), datePattern);
+            }
+            this.formatter = isJoda ? Joda.forPattern(datePattern) : DateFormatter.forPattern(datePattern);
+
+            this.parser = formatter.toDateMathParser();
+
         }
 
         @Override
@@ -224,6 +245,10 @@ public interface DocValueFormat extends NamedWriteable {
                 out.writeString(timeZone.getId());
                 out.writeVInt(resolution.ordinal());
             }
+            if (out.getVersion().onOrAfter(Version.V_7_7_0)) {
+                //in order not to loose information if the formatter is a joda we send a flag
+                out.writeBoolean(formatter instanceof JodaDateFormatter);//todo pg consider refactor to isJoda method..
+            }
         }
 
         @Override