SQL: Move internals from Joda to java.time (#35649)

Remove/Limit usage of Joda through-out the processors and functions
Use ZonedDateTime wherever possible instead of long/tzId

Fix #35633

x-pack/plugin/sql/sql-action/src/main/java/org/elasticsearch/xpack/sql/action/CliFormatter.java

@@ -9,8 +9,10 @@ import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.xpack.sql.proto.ColumnInfo;
+import org.elasticsearch.xpack.sql.proto.DateUtils;
 
 import java.io.IOException;
+import java.time.ZonedDateTime;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
@@ -24,7 +26,7 @@ public class CliFormatter implements Writeable {
      * The minimum width for any column in the formatted results.
      */
     private static final int MIN_COLUMN_WIDTH = 15;
-
+    
     private int[] width;
 
     /**
@@ -45,7 +47,7 @@ public class CliFormatter implements Writeable {
             for (int i = 0; i < width.length; i++) {
                 // TODO are we sure toString is correct here? What about dates that come back as longs.
                 // Tracked by https://github.com/elastic/x-pack-elasticsearch/issues/3081
-                width[i] = Math.max(width[i], Objects.toString(row.get(i)).length());
+                width[i] = Math.max(width[i], toString(row.get(i)).length());
             }
         }
     }
@@ -116,10 +118,10 @@ public class CliFormatter implements Writeable {
                 if (i > 0) {
                     sb.append('|');
                 }
-
                 // TODO are we sure toString is correct here? What about dates that come back as longs.
                 // Tracked by https://github.com/elastic/x-pack-elasticsearch/issues/3081
-                String string = Objects.toString(row.get(i));
+                String string = toString(row.get(i));
+
                 if (string.length() <= width[i]) {
                     // Pad
                     sb.append(string);
@@ -138,6 +140,14 @@ public class CliFormatter implements Writeable {
         return sb.toString();
     }
 
+    private static String toString(Object object) {
+        if (object instanceof ZonedDateTime) {
+            return DateUtils.toString((ZonedDateTime) object);
+        } else {
+            return Objects.toString(object);
+        }
+    }
+
     /**
      * Pick a good estimate of the buffer size needed to contain the rows.
      */
@@ -154,8 +164,12 @@ public class CliFormatter implements Writeable {
 
     @Override
     public boolean equals(Object o) {
-        if (this == o) return true;
+        if (this == o) {
-        if (o == null || getClass() != o.getClass()) return false;
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
         CliFormatter that = (CliFormatter) o;
         return Arrays.equals(width, that.width);
     }

x-pack/plugin/sql/sql-action/src/main/java/org/elasticsearch/xpack/sql/action/SqlQueryResponse.java

@@ -13,11 +13,12 @@ import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.xpack.sql.proto.ColumnInfo;
+import org.elasticsearch.xpack.sql.proto.DateUtils;
 import org.elasticsearch.xpack.sql.proto.Mode;
-import org.joda.time.ReadableDateTime;
 
 import java.io.IOException;
 import java.sql.JDBCType;
+import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
@@ -167,9 +168,17 @@ public class SqlQueryResponse extends ActionResponse implements ToXContentObject
      * Serializes the provided value in SQL-compatible way based on the client mode
      */
     public static XContentBuilder value(XContentBuilder builder, Mode mode, Object value) throws IOException {
-        if (Mode.isDriver(mode) && value instanceof ReadableDateTime) {
+        if (value instanceof ZonedDateTime) {
-            // JDBC cannot parse dates in string format
+            ZonedDateTime zdt = (ZonedDateTime) value;
-            builder.value(((ReadableDateTime) value).getMillis());
+            if (Mode.isDriver(mode)) {
+                // JDBC cannot parse dates in string format and ODBC can have issues with it
+                // so instead, use the millis since epoch (in UTC)
+                builder.value(zdt.toInstant().toEpochMilli());
+            }
+            // otherwise use the ISO format
+            else {
+                builder.value(DateUtils.toString(zdt));
+            }
         } else {
             builder.value(value);
         }

x-pack/plugin/sql/sql-proto/src/main/java/org/elasticsearch/xpack/sql/proto/DateUtils.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.sql.proto;
+
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
+import java.util.Locale;
+
+import static java.time.format.DateTimeFormatter.ISO_LOCAL_DATE;
+import static java.time.temporal.ChronoField.HOUR_OF_DAY;
+import static java.time.temporal.ChronoField.MILLI_OF_SECOND;
+import static java.time.temporal.ChronoField.MINUTE_OF_HOUR;
+import static java.time.temporal.ChronoField.SECOND_OF_MINUTE;
+
+public class DateUtils {
+
+    private static final DateTimeFormatter ISO_WITH_MILLIS = new DateTimeFormatterBuilder()
+            .parseCaseInsensitive()
+            .append(ISO_LOCAL_DATE)
+            .appendLiteral('T')
+            .appendValue(HOUR_OF_DAY, 2)
+            .appendLiteral(':')
+            .appendValue(MINUTE_OF_HOUR, 2)
+            .appendLiteral(':')
+            .appendValue(SECOND_OF_MINUTE, 2)
+            .appendFraction(MILLI_OF_SECOND, 3, 3, true)
+            .appendOffsetId()
+            .toFormatter(Locale.ROOT);
+
+    private DateUtils() {}
+
+
+    public static String toString(ZonedDateTime dateTime) {
+        return dateTime.format(ISO_WITH_MILLIS);
+    }
+}

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/execution/search/extractor/CompositeKeyExtractor.java

@@ -5,16 +5,15 @@
  */
 package org.elasticsearch.xpack.sql.execution.search.extractor;
 
-import org.elasticsearch.Version;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.search.aggregations.bucket.MultiBucketsAggregation.Bucket;
 import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
 import org.elasticsearch.xpack.sql.querydsl.container.GroupByRef.Property;
-import org.joda.time.DateTime;
+import org.elasticsearch.xpack.sql.util.DateUtils;
-import org.joda.time.DateTimeZone;
 
 import java.io.IOException;
+import java.time.ZoneId;
 import java.util.Map;
 import java.util.Objects;
 import java.util.TimeZone;
@@ -29,6 +28,7 @@ public class CompositeKeyExtractor implements BucketExtractor {
     private final String key;
     private final Property property;
     private final TimeZone timeZone;
+    private final ZoneId zoneId;
 
     /**
      * Constructs a new <code>CompositeKeyExtractor</code> instance.
@@ -38,40 +38,29 @@ public class CompositeKeyExtractor implements BucketExtractor {
         this.key = key;
         this.property = property;
         this.timeZone = timeZone;
+        this.zoneId = timeZone != null ? timeZone.toZoneId() : null;
     }
 
     CompositeKeyExtractor(StreamInput in) throws IOException {
         key = in.readString();
         property = in.readEnum(Property.class);
-        if (in.getVersion().onOrAfter(Version.V_6_3_0)) {
+        if (in.readBoolean()) {
-            if (in.readBoolean()) {
+            timeZone = TimeZone.getTimeZone(in.readString());
-                timeZone = TimeZone.getTimeZone(in.readString());
-            } else {
-                timeZone = null;
-            }
         } else {
-            DateTimeZone dtz = in.readOptionalTimeZone();
+            timeZone = null;
-            if (dtz == null) {
-                timeZone = null;
-            } else {
-                timeZone = dtz.toTimeZone();
-            }
         }
+        this.zoneId = timeZone != null ? timeZone.toZoneId() : null;
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeString(key);
         out.writeEnum(property);
-        if (out.getVersion().onOrAfter(Version.V_6_3_0)) {
+        if (timeZone == null) {
-            if (timeZone == null) {
+            out.writeBoolean(false);
-                out.writeBoolean(false);
-            } else {
-                out.writeBoolean(true);
-                out.writeString(timeZone.getID());
-            }
         } else {
-            out.writeOptionalTimeZone(timeZone == null ? null : DateTimeZone.forTimeZone(timeZone));
+            out.writeBoolean(true);
+            out.writeString(timeZone.getID());
         }
     }
 
@@ -110,7 +99,7 @@ public class CompositeKeyExtractor implements BucketExtractor {
             if (object == null) {
                 return object;
             } else if (object instanceof Long) {
-                object = new DateTime(((Long) object).longValue(), DateTimeZone.forTimeZone(timeZone));
+                object = DateUtils.of(((Long) object).longValue(), zoneId);
             } else {
                 throw new SqlIllegalArgumentException("Invalid date key returned: {}", object);
             }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/execution/search/extractor/FieldHitExtractor.java

@@ -13,9 +13,8 @@ import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
 import org.elasticsearch.xpack.sql.type.DataType;
+import org.elasticsearch.xpack.sql.util.DateUtils;
 import org.joda.time.DateTime;
-import org.joda.time.DateTimeZone;
-import org.joda.time.ReadableDateTime;
 
 import java.io.IOException;
 import java.util.List;
@@ -136,11 +135,16 @@ public class FieldHitExtractor implements HitExtractor {
         if (values instanceof Map) {
             throw new SqlIllegalArgumentException("Objects (returned by [{}]) are not supported", fieldName);
         }
-        if (values instanceof String && dataType == DataType.DATE) {
+        if (dataType == DataType.DATE) {
-            return new DateTime(Long.parseLong(values.toString()), DateTimeZone.UTC);
+            if (values instanceof String) {
+                return DateUtils.of(Long.parseLong(values.toString()));
+            }
+            // returned by nested types...
+            if (values instanceof DateTime) {
+                return DateUtils.of((DateTime) values);
+            }
         }
-        if (values instanceof Long || values instanceof Double || values instanceof String || values instanceof Boolean
+        if (values instanceof Long || values instanceof Double || values instanceof String || values instanceof Boolean) {
-                || values instanceof ReadableDateTime) {
             return values;
         }
         throw new SqlIllegalArgumentException("Type {} (returned by [{}]) is not supported", values.getClass().getSimpleName(), fieldName);

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/BaseDateTimeFunction.java

@@ -12,19 +12,22 @@ import org.elasticsearch.xpack.sql.expression.Expressions.ParamOrdinal;
 import org.elasticsearch.xpack.sql.expression.function.scalar.UnaryScalarFunction;
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.tree.NodeInfo;
-import org.joda.time.DateTime;
 
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
 import java.util.Objects;
 import java.util.TimeZone;
 
 abstract class BaseDateTimeFunction extends UnaryScalarFunction {
     
     private final TimeZone timeZone;
+    private final ZoneId zoneId;
     private final String name;
 
     BaseDateTimeFunction(Location location, Expression field, TimeZone timeZone) {
         super(location, field);
         this.timeZone = timeZone;
+        this.zoneId = timeZone != null ? timeZone.toZoneId() : null;
 
         StringBuilder sb = new StringBuilder(super.name());
         // add timezone as last argument
@@ -61,15 +64,15 @@ abstract class BaseDateTimeFunction extends UnaryScalarFunction {
 
     @Override
     public Object fold() {
-        DateTime folded = (DateTime) field().fold();
+        ZonedDateTime folded = (ZonedDateTime) field().fold();
         if (folded == null) {
             return null;
         }
 
-        return doFold(folded.getMillis(), timeZone().getID());
+        return doFold(folded.withZoneSameInstant(zoneId));
     }
 
-    protected abstract Object doFold(long millis, String tzId);
+    protected abstract Object doFold(ZonedDateTime dateTime);
     
 
     @Override

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/BaseDateTimeProcessor.java

@@ -10,21 +10,25 @@ import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
 import org.elasticsearch.xpack.sql.expression.gen.processor.Processor;
-import org.joda.time.ReadableInstant;
 
 import java.io.IOException;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
 import java.util.TimeZone;
 
 public abstract class BaseDateTimeProcessor implements Processor {
 
     private final TimeZone timeZone;
+    private final ZoneId zoneId;
     
     BaseDateTimeProcessor(TimeZone timeZone) {
         this.timeZone = timeZone;
+        this.zoneId = timeZone.toZoneId();
     }
     
     BaseDateTimeProcessor(StreamInput in) throws IOException {
         timeZone = TimeZone.getTimeZone(in.readString());
+        zoneId = timeZone.toZoneId();
     }
 
     @Override
@@ -37,23 +41,17 @@ public abstract class BaseDateTimeProcessor implements Processor {
     }
 
     @Override
-    public Object process(Object l) {
+    public Object process(Object input) {
-        if (l == null) {
+        if (input == null) {
             return null;
         }
-        long millis;
+
-        if (l instanceof String) {
+        if (!(input instanceof ZonedDateTime)) {
-            // 6.4+
+            throw new SqlIllegalArgumentException("A date is required; received {}", input);
-            millis = Long.parseLong(l.toString());
-        } else if (l instanceof ReadableInstant) {
-            // 6.3-
-            millis = ((ReadableInstant) l).getMillis();
-        } else {
-            throw new SqlIllegalArgumentException("A string or a date is required; received {}", l);
         }
-        
+
-        return doProcess(millis);
+        return doProcess(((ZonedDateTime) input).withZoneSameInstant(zoneId));
     }
 
-    abstract Object doProcess(long millis);
+    abstract Object doProcess(ZonedDateTime dateTime);
-}
+}

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DateTimeFunction.java

@@ -14,7 +14,6 @@ import org.elasticsearch.xpack.sql.expression.gen.script.ScriptTemplate;
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.type.DataType;
 
-import java.time.Instant;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
 import java.time.temporal.ChronoField;
@@ -24,23 +23,25 @@ import static org.elasticsearch.xpack.sql.expression.gen.script.ParamsBuilder.pa
 
 public abstract class DateTimeFunction extends BaseDateTimeFunction {
 
-    DateTimeFunction(Location location, Expression field, TimeZone timeZone) {
+    private final DateTimeExtractor extractor;
+
+    DateTimeFunction(Location location, Expression field, TimeZone timeZone, DateTimeExtractor extractor) {
         super(location, field, timeZone);
+        this.extractor = extractor;
     }
 
     @Override
-    protected Object doFold(long millis, String tzId) {
+    protected Object doFold(ZonedDateTime dateTime) {
-        return dateTimeChrono(millis, tzId, chronoField().name());
+        return dateTimeChrono(dateTime, extractor.chronoField());
     }
 
-    public static Integer dateTimeChrono(long millis, String tzId, String chronoName) {
+    public static Integer dateTimeChrono(ZonedDateTime dateTime, String tzId, String chronoName) {
-        ZonedDateTime time = ZonedDateTime.ofInstant(Instant.ofEpochMilli(millis), ZoneId.of(tzId));
+        ZonedDateTime zdt = dateTime.withZoneSameInstant(ZoneId.of(tzId));
-        return Integer.valueOf(time.get(ChronoField.valueOf(chronoName)));
+        return dateTimeChrono(zdt, ChronoField.valueOf(chronoName));
     }
 
-    public static Integer dateTimeChrono(ZonedDateTime millis, String tzId, String chronoName) {
+    private static Integer dateTimeChrono(ZonedDateTime dateTime, ChronoField field) {
-        ZonedDateTime time = millis.withZoneSameInstant(ZoneId.of(tzId));
+        return Integer.valueOf(dateTime.get(field));
-        return Integer.valueOf(time.get(ChronoField.valueOf(chronoName)));
     }
 
     @Override
@@ -51,21 +52,14 @@ public abstract class DateTimeFunction extends BaseDateTimeFunction {
         template = formatTemplate("{sql}.dateTimeChrono(doc[{}].value, {}, {})");
         params.variable(field.name())
               .variable(timeZone().getID())
-              .variable(chronoField().name());
+              .variable(extractor.chronoField().name());
         
         return new ScriptTemplate(template, params.build(), dataType());
     }
 
-    /**
-     * Used for generating the painless script version of this function when the time zone is not UTC
-     */
-    protected abstract ChronoField chronoField();
-
-    protected abstract DateTimeExtractor extractor();
-
     @Override
     protected Processor makeProcessor() {
-        return new DateTimeProcessor(extractor(), timeZone());
+        return new DateTimeProcessor(extractor, timeZone());
     }
 
     @Override

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DateTimeHistogramFunction.java

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 
 import org.elasticsearch.xpack.sql.expression.Expression;
+import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeProcessor.DateTimeExtractor;
 import org.elasticsearch.xpack.sql.tree.Location;
 
 import java.util.TimeZone;
@@ -16,8 +17,8 @@ import java.util.TimeZone;
  */
 public abstract class DateTimeHistogramFunction extends DateTimeFunction {
 
-    DateTimeHistogramFunction(Location location, Expression field, TimeZone timeZone) {
+    DateTimeHistogramFunction(Location location, Expression field, TimeZone timeZone, DateTimeExtractor extractor) {
-        super(location, field, timeZone);
+        super(location, field, timeZone, extractor);
     }
 
     /**

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DateTimeProcessor.java

@@ -7,38 +7,40 @@ package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.joda.time.DateTime;
-import org.joda.time.DateTimeFieldType;
-import org.joda.time.DateTimeZone;
-import org.joda.time.ReadableDateTime;
 
 import java.io.IOException;
+import java.time.ZonedDateTime;
+import java.time.temporal.ChronoField;
 import java.util.Objects;
 import java.util.TimeZone;
 
 public class DateTimeProcessor extends BaseDateTimeProcessor {
     
     public enum DateTimeExtractor {
-        DAY_OF_MONTH(DateTimeFieldType.dayOfMonth()),
+        DAY_OF_MONTH(ChronoField.DAY_OF_MONTH),
-        DAY_OF_WEEK(DateTimeFieldType.dayOfWeek()),
+        DAY_OF_WEEK(ChronoField.DAY_OF_WEEK),
-        DAY_OF_YEAR(DateTimeFieldType.dayOfYear()),
+        DAY_OF_YEAR(ChronoField.DAY_OF_YEAR),
-        HOUR_OF_DAY(DateTimeFieldType.hourOfDay()),
+        HOUR_OF_DAY(ChronoField.HOUR_OF_DAY),
-        MINUTE_OF_DAY(DateTimeFieldType.minuteOfDay()),
+        MINUTE_OF_DAY(ChronoField.MINUTE_OF_DAY),
-        MINUTE_OF_HOUR(DateTimeFieldType.minuteOfHour()),
+        MINUTE_OF_HOUR(ChronoField.MINUTE_OF_HOUR),
-        MONTH_OF_YEAR(DateTimeFieldType.monthOfYear()),
+        MONTH_OF_YEAR(ChronoField.MONTH_OF_YEAR),
-        SECOND_OF_MINUTE(DateTimeFieldType.secondOfMinute()),
+        SECOND_OF_MINUTE(ChronoField.SECOND_OF_MINUTE),
-        WEEK_OF_YEAR(DateTimeFieldType.weekOfWeekyear()),
+        WEEK_OF_YEAR(ChronoField.ALIGNED_WEEK_OF_YEAR),
-        YEAR(DateTimeFieldType.year());
+        YEAR(ChronoField.YEAR);
 
-        private final DateTimeFieldType field;
+        private final ChronoField field;
 
-        DateTimeExtractor(DateTimeFieldType field) {
+        DateTimeExtractor(ChronoField field) {
             this.field = field;
         }
 
-        public int extract(ReadableDateTime dt) {
+        public int extract(ZonedDateTime dt) {
             return dt.get(field);
         }
+
+        public ChronoField chronoField() {
+            return field;
+        }
     }
     
     public static final String NAME = "dt";
@@ -70,10 +72,8 @@ public class DateTimeProcessor extends BaseDateTimeProcessor {
     }
 
     @Override
-    public Object doProcess(long millis) {
+    public Object doProcess(ZonedDateTime dateTime) {
-        ReadableDateTime dt = new DateTime(millis, DateTimeZone.forTimeZone(timeZone()));
+        return extractor.extract(dateTime);
-
-        return extractor.extract(dt);
     }
 
     @Override
@@ -95,4 +95,4 @@ public class DateTimeProcessor extends BaseDateTimeProcessor {
     public String toString() {
         return extractor.toString();
     }
-}
+}

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DayOfMonth.java

@@ -10,7 +10,6 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeP
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.tree.NodeInfo.NodeCtor2;
 
-import java.time.temporal.ChronoField;
 import java.util.TimeZone;
 
 /**
@@ -18,7 +17,7 @@ import java.util.TimeZone;
  */
 public class DayOfMonth extends DateTimeFunction {
     public DayOfMonth(Location location, Expression field, TimeZone timeZone) {
-        super(location, field, timeZone);
+        super(location, field, timeZone, DateTimeExtractor.DAY_OF_MONTH);
     }
 
     @Override
@@ -35,14 +34,4 @@ public class DayOfMonth extends DateTimeFunction {
     public String dateTimeFormat() {
         return "d";
     }
-
-    @Override
-    protected ChronoField chronoField() {
-        return ChronoField.DAY_OF_MONTH;
-    }
-
-    @Override
-    protected DateTimeExtractor extractor() {
-        return DateTimeExtractor.DAY_OF_MONTH;
-    }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DayOfWeek.java

@@ -10,7 +10,6 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeP
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.tree.NodeInfo.NodeCtor2;
 
-import java.time.temporal.ChronoField;
 import java.util.TimeZone;
 
 /**
@@ -18,7 +17,7 @@ import java.util.TimeZone;
  */
 public class DayOfWeek extends DateTimeFunction {
     public DayOfWeek(Location location, Expression field, TimeZone timeZone) {
-        super(location, field, timeZone);
+        super(location, field, timeZone, DateTimeExtractor.DAY_OF_WEEK);
     }
 
     @Override
@@ -35,14 +34,4 @@ public class DayOfWeek extends DateTimeFunction {
     public String dateTimeFormat() {
         return "e";
     }
-
-    @Override
-    protected ChronoField chronoField() {
-        return ChronoField.DAY_OF_WEEK;
-    }
-
-    @Override
-    protected DateTimeExtractor extractor() {
-        return DateTimeExtractor.DAY_OF_WEEK;
-    }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DayOfYear.java

@@ -11,7 +11,6 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeP
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.tree.NodeInfo.NodeCtor2;
 
-import java.time.temporal.ChronoField;
 import java.util.TimeZone;
 
 /**
@@ -19,7 +18,7 @@ import java.util.TimeZone;
  */
 public class DayOfYear extends DateTimeFunction {
     public DayOfYear(Location location, Expression field, TimeZone timeZone) {
-        super(location, field, timeZone);
+        super(location, field, timeZone, DateTimeExtractor.DAY_OF_YEAR);
     }
 
     @Override
@@ -36,14 +35,4 @@ public class DayOfYear extends DateTimeFunction {
     public String dateTimeFormat() {
         return "D";
     }
-
-    @Override
-    protected ChronoField chronoField() {
-        return ChronoField.DAY_OF_YEAR;
-    }
-
-    @Override
-    protected DateTimeExtractor extractor() {
-        return DateTimeExtractor.DAY_OF_YEAR;
-    }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/HourOfDay.java

@@ -10,7 +10,6 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeP
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.tree.NodeInfo.NodeCtor2;
 
-import java.time.temporal.ChronoField;
 import java.util.TimeZone;
 
 /**
@@ -18,7 +17,7 @@ import java.util.TimeZone;
  */
 public class HourOfDay extends DateTimeFunction {
     public HourOfDay(Location location, Expression field, TimeZone timeZone) {
-        super(location, field, timeZone);
+        super(location, field, timeZone, DateTimeExtractor.HOUR_OF_DAY);
     }
 
     @Override
@@ -35,14 +34,4 @@ public class HourOfDay extends DateTimeFunction {
     public String dateTimeFormat() {
         return "hour";
     }
-
-    @Override
-    protected ChronoField chronoField() {
-        return ChronoField.HOUR_OF_DAY;
-    }
-
-    @Override
-    protected DateTimeExtractor extractor() {
-        return DateTimeExtractor.HOUR_OF_DAY;
-    }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/MinuteOfDay.java

@@ -10,7 +10,6 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeP
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.tree.NodeInfo.NodeCtor2;
 
-import java.time.temporal.ChronoField;
 import java.util.TimeZone;
 
 /**
@@ -19,7 +18,7 @@ import java.util.TimeZone;
 public class MinuteOfDay extends DateTimeFunction {
 
     public MinuteOfDay(Location location, Expression field, TimeZone timeZone) {
-        super(location, field, timeZone);
+        super(location, field, timeZone, DateTimeExtractor.MINUTE_OF_DAY);
     }
 
     @Override
@@ -36,14 +35,4 @@ public class MinuteOfDay extends DateTimeFunction {
     public String dateTimeFormat() {
         throw new UnsupportedOperationException("is there a format for it?");
     }
-
-    @Override
-    protected ChronoField chronoField() {
-        return ChronoField.MINUTE_OF_DAY;
-    }
-
-    @Override
-    protected DateTimeExtractor extractor() {
-        return DateTimeExtractor.MINUTE_OF_DAY;
-    }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/MinuteOfHour.java

@@ -10,7 +10,6 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeP
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.tree.NodeInfo.NodeCtor2;
 
-import java.time.temporal.ChronoField;
 import java.util.TimeZone;
 
 /**
@@ -18,7 +17,7 @@ import java.util.TimeZone;
  */
 public class MinuteOfHour extends DateTimeFunction {
     public MinuteOfHour(Location location, Expression field, TimeZone timeZone) {
-        super(location, field, timeZone);
+        super(location, field, timeZone, DateTimeExtractor.MINUTE_OF_HOUR);
     }
 
     @Override
@@ -35,14 +34,4 @@ public class MinuteOfHour extends DateTimeFunction {
     public String dateTimeFormat() {
         return "m";
     }
-
-    @Override
-    protected ChronoField chronoField() {
-        return ChronoField.MINUTE_OF_HOUR;
-    }
-
-    @Override
-    protected DateTimeExtractor extractor() {
-        return DateTimeExtractor.MINUTE_OF_HOUR;
-    }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/MonthOfYear.java

@@ -10,7 +10,6 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeP
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.tree.NodeInfo.NodeCtor2;
 
-import java.time.temporal.ChronoField;
 import java.util.TimeZone;
 
 /**
@@ -18,7 +17,7 @@ import java.util.TimeZone;
  */
 public class MonthOfYear extends DateTimeFunction {
     public MonthOfYear(Location location, Expression field, TimeZone timeZone) {
-        super(location, field, timeZone);
+        super(location, field, timeZone, DateTimeExtractor.MONTH_OF_YEAR);
     }
 
     @Override
@@ -35,14 +34,4 @@ public class MonthOfYear extends DateTimeFunction {
     public String dateTimeFormat() {
         return "M";
     }
-
-    @Override
-    protected ChronoField chronoField() {
-        return ChronoField.MONTH_OF_YEAR;
-    }
-
-    @Override
-    protected DateTimeExtractor extractor() {
-        return DateTimeExtractor.MONTH_OF_YEAR;
-    }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/NamedDateTimeFunction.java

@@ -14,6 +14,7 @@ import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.type.DataType;
 import org.elasticsearch.xpack.sql.util.StringUtils;
 
+import java.time.ZonedDateTime;
 import java.util.Locale;
 import java.util.TimeZone;
 
@@ -33,8 +34,8 @@ abstract class NamedDateTimeFunction extends BaseDateTimeFunction {
     }
 
     @Override
-    protected Object doFold(long millis, String tzId) {
+    protected Object doFold(ZonedDateTime dateTime) {
-        return nameExtractor.extract(millis, tzId);
+        return nameExtractor.extract(dateTime);
     }
 
     @Override

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/NamedDateTimeProcessor.java

@@ -9,7 +9,6 @@ import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 
 import java.io.IOException;
-import java.time.Instant;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
@@ -31,8 +30,8 @@ public class NamedDateTimeProcessor extends BaseDateTimeProcessor {
             this.apply = apply;
         }
 
-        public final String extract(Long millis, String tzId) {
+        public final String extract(ZonedDateTime dateTime) {
-            return extract(ZonedDateTime.ofInstant(Instant.ofEpochMilli(millis), ZoneId.of(tzId)), tzId);
+            return apply.apply(dateTime);
         }
 
         public final String extract(ZonedDateTime millis, String tzId) {
@@ -73,8 +72,8 @@ public class NamedDateTimeProcessor extends BaseDateTimeProcessor {
     }
 
     @Override
-    public Object doProcess(long millis) {
+    public Object doProcess(ZonedDateTime dateTime) {
-        return extractor.extract(millis, timeZone().getID());
+        return extractor.extract(dateTime);
     }
 
     @Override

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/Quarter.java

@@ -14,6 +14,7 @@ import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.tree.NodeInfo.NodeCtor2;
 import org.elasticsearch.xpack.sql.type.DataType;
 
+import java.time.ZonedDateTime;
 import java.util.TimeZone;
 
 import static org.elasticsearch.xpack.sql.expression.function.scalar.datetime.QuarterProcessor.quarter;
@@ -26,8 +27,8 @@ public class Quarter extends BaseDateTimeFunction {
     }
 
     @Override
-    protected Object doFold(long millis, String tzId) {
+    protected Object doFold(ZonedDateTime dateTime) {
-        return quarter(millis, tzId);
+        return quarter(dateTime);
     }
 
     @Override

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/QuarterProcessor.java

@@ -9,7 +9,6 @@ package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 import org.elasticsearch.common.io.stream.StreamInput;
 
 import java.io.IOException;
-import java.time.Instant;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
@@ -36,17 +35,16 @@ public class QuarterProcessor extends BaseDateTimeProcessor {
     }
 
     @Override
-    public Object doProcess(long millis) {
+    public Object doProcess(ZonedDateTime zdt) {
-        return quarter(millis, timeZone().getID());
+        return quarter(zdt);
     }
     
-    public static Integer quarter(long millis, String tzId) {
+    public static Integer quarter(ZonedDateTime dateTime, String tzId) {
-        return quarter(ZonedDateTime.ofInstant(Instant.ofEpochMilli(millis), ZoneId.of(tzId)), tzId);
+        return quarter(dateTime.withZoneSameInstant(ZoneId.of(tzId)));
     }
 
-    public static Integer quarter(ZonedDateTime zdt, String tzId) {
+    static Integer quarter(ZonedDateTime zdt) {
-        ZonedDateTime time = zdt.withZoneSameInstant(ZoneId.of(tzId));
+        return Integer.valueOf(zdt.format(QUARTER_FORMAT));
-        return Integer.valueOf(time.format(QUARTER_FORMAT));
     }
 
     @Override

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/SecondOfMinute.java

@@ -10,7 +10,6 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeP
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.tree.NodeInfo.NodeCtor2;
 
-import java.time.temporal.ChronoField;
 import java.util.TimeZone;
 
 /**
@@ -18,7 +17,7 @@ import java.util.TimeZone;
  */
 public class SecondOfMinute extends DateTimeFunction {
     public SecondOfMinute(Location location, Expression field, TimeZone timeZone) {
-        super(location, field, timeZone);
+        super(location, field, timeZone, DateTimeExtractor.SECOND_OF_MINUTE);
     }
 
     @Override
@@ -35,14 +34,4 @@ public class SecondOfMinute extends DateTimeFunction {
     public String dateTimeFormat() {
         return "s";
     }
-
-    @Override
-    protected ChronoField chronoField() {
-        return ChronoField.SECOND_OF_MINUTE;
-    }
-
-    @Override
-    protected DateTimeExtractor extractor() {
-        return DateTimeExtractor.SECOND_OF_MINUTE;
-    }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/WeekOfYear.java

@@ -10,7 +10,6 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeP
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.tree.NodeInfo.NodeCtor2;
 
-import java.time.temporal.ChronoField;
 import java.util.TimeZone;
 
 /**
@@ -18,7 +17,7 @@ import java.util.TimeZone;
  */
 public class WeekOfYear extends DateTimeFunction {
     public WeekOfYear(Location location, Expression field, TimeZone timeZone) {
-        super(location, field, timeZone);
+        super(location, field, timeZone, DateTimeExtractor.WEEK_OF_YEAR);
     }
 
     @Override
@@ -35,14 +34,4 @@ public class WeekOfYear extends DateTimeFunction {
     public String dateTimeFormat() {
         return "w";
     }
-
-    @Override
-    protected ChronoField chronoField() {
-        return ChronoField.ALIGNED_WEEK_OF_YEAR;
-    }
-
-    @Override
-    protected DateTimeExtractor extractor() {
-        return DateTimeExtractor.WEEK_OF_YEAR;
-    }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/Year.java

@@ -10,7 +10,6 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeP
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.tree.NodeInfo.NodeCtor2;
 
-import java.time.temporal.ChronoField;
 import java.util.TimeZone;
 
 /**
@@ -18,7 +17,7 @@ import java.util.TimeZone;
  */
 public class Year extends DateTimeHistogramFunction {
     public Year(Location location, Expression field, TimeZone timeZone) {
-        super(location, field, timeZone);
+        super(location, field, timeZone, DateTimeExtractor.YEAR);
     }
 
     @Override
@@ -41,16 +40,6 @@ public class Year extends DateTimeHistogramFunction {
         return field();
     }
 
-    @Override
-    protected ChronoField chronoField() {
-        return ChronoField.YEAR;
-    }
-
-    @Override
-    protected DateTimeExtractor extractor() {
-        return DateTimeExtractor.YEAR;
-    }
-
     @Override
     public String interval() {
         return "year";

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/predicate/operator/arithmetic/Arithmetics.java

@@ -5,6 +5,10 @@
  */
 package org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic;
 
+import java.time.Duration;
+import java.time.Period;
+import java.time.ZonedDateTime;
+
 /**
  * Arithmetic operation using the type widening rules of the JLS 5.6.2 namely
  * widen to double or float or long or int in this order.
@@ -29,6 +33,38 @@ public abstract class Arithmetics {
         return Integer.valueOf(Math.addExact(l.intValue(), r.intValue()));
     }
 
+    static Period add(Period l, Period r) {
+        if (l == null || r == null) {
+            return null;
+        }
+
+        return l.plus(r);
+    }
+
+    static Duration add(Duration l, Duration r) {
+        if (l == null || r == null) {
+            return null;
+        }
+
+        return l.plus(r);
+    }
+
+    static ZonedDateTime add(ZonedDateTime l, Period r) {
+        if (l == null || r == null) {
+            return null;
+        }
+
+        return l.plus(r);
+    }
+
+    static ZonedDateTime add(ZonedDateTime l, Duration r) {
+        if (l == null || r == null) {
+            return null;
+        }
+
+        return l.plus(r);
+    }
+
     static Number sub(Number l, Number r) {
         if (l == null || r == null) {
             return null;
@@ -47,6 +83,38 @@ public abstract class Arithmetics {
         return Integer.valueOf(Math.subtractExact(l.intValue(), r.intValue()));
     }
 
+    static Period sub(Period l, Period r) {
+        if (l == null || r == null) {
+            return null;
+        }
+
+        return l.minus(r);
+    }
+
+    static Duration sub(Duration l, Duration r) {
+        if (l == null || r == null) {
+            return null;
+        }
+
+        return l.minus(r);
+    }
+
+    static ZonedDateTime sub(ZonedDateTime l, Period r) {
+        if (l == null || r == null) {
+            return null;
+        }
+
+        return l.minus(r);
+    }
+
+    static ZonedDateTime sub(ZonedDateTime l, Duration r) {
+        if (l == null || r == null) {
+            return null;
+        }
+
+        return l.minus(r);
+    }
+
     static Number mul(Number l, Number r) {
         if (l == null || r == null) {
             return null;

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/parser/ExpressionBuilder.java

@@ -24,7 +24,6 @@ import org.elasticsearch.xpack.sql.expression.UnresolvedStar;
 import org.elasticsearch.xpack.sql.expression.function.Function;
 import org.elasticsearch.xpack.sql.expression.function.UnresolvedFunction;
 import org.elasticsearch.xpack.sql.expression.function.scalar.Cast;
-import org.elasticsearch.xpack.sql.expression.predicate.nulls.IsNull;
 import org.elasticsearch.xpack.sql.expression.predicate.Range;
 import org.elasticsearch.xpack.sql.expression.predicate.fulltext.MatchQueryPredicate;
 import org.elasticsearch.xpack.sql.expression.predicate.fulltext.MultiMatchQueryPredicate;
@@ -33,6 +32,7 @@ import org.elasticsearch.xpack.sql.expression.predicate.logical.And;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.Not;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.Or;
 import org.elasticsearch.xpack.sql.expression.predicate.nulls.IsNotNull;
+import org.elasticsearch.xpack.sql.expression.predicate.nulls.IsNull;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.Add;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.Div;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.Mod;
@@ -94,6 +94,7 @@ import org.elasticsearch.xpack.sql.proto.SqlTypedParamValue;
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.type.DataType;
 import org.elasticsearch.xpack.sql.type.DataTypes;
+import org.elasticsearch.xpack.sql.util.DateUtils;
 import org.joda.time.DateTime;
 import org.joda.time.format.DateTimeFormatter;
 import org.joda.time.format.DateTimeFormatterBuilder;
@@ -631,7 +632,7 @@ abstract class ExpressionBuilder extends IdentifierBuilder {
         } catch(IllegalArgumentException ex) {
             throw new ParsingException(loc, "Invalid date received; {}", ex.getMessage());
         }
-        return new Literal(loc, dt, DataType.DATE);
+        return new Literal(loc, DateUtils.of(dt), DataType.DATE);
     }
 
     @Override
@@ -667,7 +668,7 @@ abstract class ExpressionBuilder extends IdentifierBuilder {
         } catch (IllegalArgumentException ex) {
             throw new ParsingException(loc, "Invalid timestamp received; {}", ex.getMessage());
         }
-        return new Literal(loc, dt, DataType.DATE);
+        return new Literal(loc, DateUtils.of(dt), DataType.DATE);
     }
 
     @Override

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/plugin/TextFormat.java

@@ -7,13 +7,15 @@ package org.elasticsearch.xpack.sql.plugin;
 
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.rest.RestRequest;
-import org.elasticsearch.xpack.sql.action.SqlQueryResponse;
 import org.elasticsearch.xpack.sql.action.CliFormatter;
+import org.elasticsearch.xpack.sql.action.SqlQueryResponse;
 import org.elasticsearch.xpack.sql.proto.ColumnInfo;
 import org.elasticsearch.xpack.sql.session.Cursor;
 import org.elasticsearch.xpack.sql.session.Cursors;
+import org.elasticsearch.xpack.sql.util.DateUtils;
 import org.elasticsearch.xpack.sql.util.StringUtils;
 
+import java.time.ZonedDateTime;
 import java.util.List;
 import java.util.Locale;
 import java.util.Objects;
@@ -225,7 +227,7 @@ enum TextFormat {
         }
 
         for (List<Object> row : response.rows()) {
-            row(sb, row, f -> Objects.toString(f, StringUtils.EMPTY));
+            row(sb, row, f -> f instanceof ZonedDateTime ? DateUtils.toString((ZonedDateTime) f) : Objects.toString(f, StringUtils.EMPTY));
         }
 
         return sb.toString();

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/type/DataTypeConversion.java

@@ -8,12 +8,9 @@ package org.elasticsearch.xpack.sql.type;
 import org.elasticsearch.common.Booleans;
 import org.elasticsearch.common.network.InetAddresses;
 import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
-import org.joda.time.DateTime;
+import org.elasticsearch.xpack.sql.util.DateUtils;
-import org.joda.time.DateTimeZone;
-import org.joda.time.ReadableInstant;
-import org.joda.time.format.DateTimeFormatter;
-import org.joda.time.format.ISODateTimeFormat;
 
+import java.time.ZonedDateTime;
 import java.util.Locale;
 import java.util.function.DoubleFunction;
 import java.util.function.Function;
@@ -32,8 +29,6 @@ import static org.elasticsearch.xpack.sql.type.DataType.NULL;
  */
 public abstract class DataTypeConversion {
 
-    private static final DateTimeFormatter UTC_DATE_FORMATTER = ISODateTimeFormat.dateOptionalTimeParser().withZoneUTC();
-
     /**
      * Returns the type compatible with both left and right types
      * <p>
@@ -374,7 +369,7 @@ public abstract class DataTypeConversion {
         IDENTITY(Function.identity()),
         NULL(value -> null),
         
-        DATE_TO_STRING(Object::toString),
+        DATE_TO_STRING(o -> DateUtils.toString((ZonedDateTime) o)),
         OTHER_TO_STRING(String::valueOf),
 
         RATIONAL_TO_LONG(fromDouble(DataTypeConversion::safeToLong)),
@@ -416,7 +411,7 @@ public abstract class DataTypeConversion {
         RATIONAL_TO_DATE(toDate(RATIONAL_TO_LONG)),
         INTEGER_TO_DATE(toDate(INTEGER_TO_LONG)),
         BOOL_TO_DATE(toDate(BOOL_TO_INT)),
-        STRING_TO_DATE(fromString(UTC_DATE_FORMATTER::parseDateTime, "Date")),
+        STRING_TO_DATE(fromString(DateUtils::of, "Date")),
 
         NUMERIC_TO_BOOLEAN(fromLong(value -> value != 0)),
         STRING_TO_BOOLEAN(fromString(DataTypeConversion::convertToBoolean, "Boolean")),
@@ -462,11 +457,11 @@ public abstract class DataTypeConversion {
         }
         
         private static Function<Object, Object> fromDate(Function<Long, Object> converter) {
-            return l -> ((ReadableInstant) l).getMillis();
+            return l -> ((ZonedDateTime) l).toEpochSecond();
         }
 
         private static Function<Object, Object> toDate(Conversion conversion) {
-            return l -> new DateTime(((Number) conversion.convert(l)).longValue(), DateTimeZone.UTC);
+            return l -> DateUtils.of(((Number) conversion.convert(l)).longValue());
         }
 
         public Object convert(Object l) {

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/type/DataTypes.java

@@ -6,7 +6,9 @@
 package org.elasticsearch.xpack.sql.type;
 
 import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
-import org.joda.time.DateTime;
+
+import java.time.OffsetDateTime;
+import java.time.ZonedDateTime;
 
 public final class DataTypes {
 
@@ -45,7 +47,7 @@ public final class DataTypes {
         if (value instanceof Short) {
             return DataType.SHORT;
         }
-        if (value instanceof DateTime) {
+        if (value instanceof ZonedDateTime || value instanceof OffsetDateTime) {
             return DataType.DATE;
         }
         if (value instanceof String || value instanceof Character) {

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/util/DateUtils.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.sql.util;
+
+import org.joda.time.DateTime;
+import org.joda.time.format.DateTimeFormatter;
+import org.joda.time.format.ISODateTimeFormat;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+
+public class DateUtils {
+
+    // TODO: do we have a java.time based parser we can use instead?
+    private static final DateTimeFormatter UTC_DATE_FORMATTER = ISODateTimeFormat.dateOptionalTimeParser().withZoneUTC();
+
+    public static ZoneId UTC = ZoneId.of("UTC");
+
+    private DateUtils() {}
+
+
+    /**
+     * Creates a date from the millis since epoch (thus the time-zone is UTC).
+     */
+    public static ZonedDateTime of(long millis) {
+        return ZonedDateTime.ofInstant(Instant.ofEpochMilli(millis), UTC);
+    }
+
+    /**
+     * Creates a date from the millis since epoch then translates the date into the given timezone.
+     */
+    public static ZonedDateTime of(long millis, ZoneId id) {
+        return ZonedDateTime.ofInstant(Instant.ofEpochMilli(millis), id);
+    }
+
+    /**
+     * Parses the given string into a DateTime using UTC as a default timezone.
+     */
+    public static ZonedDateTime of(String dateFormat) {
+        return of(UTC_DATE_FORMATTER.parseDateTime(dateFormat));
+    }
+
+    public static ZonedDateTime of(DateTime dateTime) {
+        LocalDateTime ldt = LocalDateTime.of(
+                dateTime.getYear(),
+                dateTime.getMonthOfYear(),
+                dateTime.getDayOfMonth(),
+                dateTime.getHourOfDay(),
+                dateTime.getMinuteOfHour(),
+                dateTime.getSecondOfMinute(),
+                dateTime.getMillisOfSecond() * 1_000_000);
+        
+        return ZonedDateTime.ofStrict(ldt,
+                ZoneOffset.ofTotalSeconds(dateTime.getZone().getOffset(dateTime) / 1000),
+                org.elasticsearch.common.time.DateUtils.dateTimeZoneToZoneId(dateTime.getZone()));
+    }
+    
+    public static String toString(ZonedDateTime dateTime) {
+        return org.elasticsearch.xpack.sql.proto.DateUtils.toString(dateTime);
+    }
+}

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/execution/search/extractor/CompositeKeyExtractorTests.java

@@ -11,8 +11,7 @@ import org.elasticsearch.search.aggregations.bucket.MultiBucketsAggregation.Buck
 import org.elasticsearch.test.AbstractWireSerializingTestCase;
 import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
 import org.elasticsearch.xpack.sql.querydsl.container.GroupByRef.Property;
-import org.joda.time.DateTime;
+import org.elasticsearch.xpack.sql.util.DateUtils;
-import org.joda.time.DateTimeZone;
 
 import java.io.IOException;
 import java.util.TimeZone;
@@ -63,7 +62,7 @@ public class CompositeKeyExtractorTests extends AbstractWireSerializingTestCase<
 
         long millis = System.currentTimeMillis();
         Bucket bucket = new TestBucket(singletonMap(extractor.key(), millis), randomLong(), new Aggregations(emptyList()));
-        assertEquals(new DateTime(millis, DateTimeZone.forTimeZone(extractor.timeZone())), extractor.extract(bucket));
+        assertEquals(DateUtils.of(millis, extractor.timeZone().toZoneId()), extractor.extract(bucket));
     }
 
     public void testExtractIncorrectDateKey() {

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/execution/search/extractor/FieldHitExtractorTests.java

@@ -15,8 +15,7 @@ import org.elasticsearch.test.AbstractWireSerializingTestCase;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.sql.SqlException;
 import org.elasticsearch.xpack.sql.type.DataType;
-import org.joda.time.DateTime;
+import org.elasticsearch.xpack.sql.util.DateUtils;
-import org.joda.time.DateTimeZone;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -144,7 +143,7 @@ public class FieldHitExtractorTests extends AbstractWireSerializingTestCase<Fiel
         DocumentField field = new DocumentField("my_date_field", documentFieldValues);
         hit.fields(singletonMap("my_date_field", field));
         FieldHitExtractor extractor = new FieldHitExtractor("my_date_field", DataType.DATE, true);
-        assertEquals(new DateTime(millis, DateTimeZone.UTC), extractor.extract(hit));
+        assertEquals(DateUtils.of(millis), extractor.extract(hit));
     }
 
     public void testGetSource() throws IOException {

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DateTimeProcessorTests.java

@@ -8,12 +8,12 @@ package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 import org.elasticsearch.common.io.stream.Writeable.Reader;
 import org.elasticsearch.test.AbstractWireSerializingTestCase;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeProcessor.DateTimeExtractor;
-import org.joda.time.DateTime;
-import org.joda.time.DateTimeZone;
 
 import java.io.IOException;
 import java.util.TimeZone;
 
+import static org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeTestUtils.dateTime;
+
 public class DateTimeProcessorTests extends AbstractWireSerializingTestCase<DateTimeProcessor> {
     private static final TimeZone UTC = TimeZone.getTimeZone("UTC");
 
@@ -39,12 +39,12 @@ public class DateTimeProcessorTests extends AbstractWireSerializingTestCase<Date
 
     public void testApply() {
         DateTimeProcessor proc = new DateTimeProcessor(DateTimeExtractor.YEAR, UTC);
-        assertEquals(1970, proc.process(new DateTime(0L, DateTimeZone.UTC)));
+        assertEquals(1970, proc.process(dateTime(0L)));
-        assertEquals(2017, proc.process(new DateTime(2017, 01, 02, 10, 10, DateTimeZone.UTC)));
+        assertEquals(2017, proc.process(dateTime(2017, 01, 02, 10, 10)));
 
         proc = new DateTimeProcessor(DateTimeExtractor.DAY_OF_MONTH, UTC);
-        assertEquals(1, proc.process(new DateTime(0L, DateTimeZone.UTC)));
+        assertEquals(1, proc.process(dateTime(0L)));
-        assertEquals(2, proc.process(new DateTime(2017, 01, 02, 10, 10, DateTimeZone.UTC)));
+        assertEquals(2, proc.process(dateTime(2017, 01, 02, 10, 10)));
-        assertEquals(31, proc.process(new DateTime(2017, 01, 31, 10, 10, DateTimeZone.UTC)));
+        assertEquals(31, proc.process(dateTime(2017, 01, 31, 10, 10)));
     }
 }

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DateTimeTestUtils.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
+
+import org.elasticsearch.xpack.sql.util.DateUtils;
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+
+import java.time.ZonedDateTime;
+
+import static org.junit.Assert.assertEquals;
+
+public class DateTimeTestUtils {
+
+    private DateTimeTestUtils() {}
+
+    public static ZonedDateTime dateTime(int year, int month, int day, int hour, int minute) {
+        DateTime dateTime = new DateTime(year, month, day, hour, minute, DateTimeZone.UTC);
+        ZonedDateTime zdt = ZonedDateTime.of(year, month, day, hour, minute, 0, 0, DateUtils.UTC);
+        assertEquals(dateTime.getMillis() / 1000, zdt.toEpochSecond());
+        return zdt;
+    }
+
+    public static ZonedDateTime dateTime(long millisSinceEpoch) {
+        return DateUtils.of(millisSinceEpoch);
+    }
+}

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DayOfYearTests.java

@@ -8,11 +8,11 @@ package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.sql.expression.Literal;
 import org.elasticsearch.xpack.sql.type.DataType;
-import org.joda.time.DateTime;
-import org.joda.time.DateTimeZone;
 
 import java.util.TimeZone;
 
+import static org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeTestUtils.dateTime;
+
 public class DayOfYearTests extends ESTestCase {
     private static final TimeZone UTC = TimeZone.getTimeZone("UTC");
 
@@ -22,10 +22,6 @@ public class DayOfYearTests extends ESTestCase {
         assertEquals(365, extract(dateTime(0), TimeZone.getTimeZone("GMT-01:00")));
     }
 
-    private DateTime dateTime(long millisSinceEpoch) {
-        return new DateTime(millisSinceEpoch, DateTimeZone.forTimeZone(UTC));
-    }
-
     private Object extract(Object value, TimeZone timeZone) {
         return build(value, timeZone).asPipe().asProcessor().process(value);
     }

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/NamedDateTimeProcessorTests.java

@@ -10,13 +10,13 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.Writeable.Reader;
 import org.elasticsearch.test.AbstractWireSerializingTestCase;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.NamedDateTimeProcessor.NameExtractor;
-import org.joda.time.DateTime;
-import org.joda.time.DateTimeZone;
 import org.junit.Assume;
 
 import java.io.IOException;
 import java.util.TimeZone;
 
+import static org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeTestUtils.dateTime;
+
 public class NamedDateTimeProcessorTests extends AbstractWireSerializingTestCase<NamedDateTimeProcessor> {
     
     private static final TimeZone UTC = TimeZone.getTimeZone("UTC");
@@ -44,55 +44,55 @@ public class NamedDateTimeProcessorTests extends AbstractWireSerializingTestCase
     public void testValidDayNamesInUTC() {
         assumeJava9PlusAndCompatLocaleProviderSetting();
         NamedDateTimeProcessor proc = new NamedDateTimeProcessor(NameExtractor.DAY_NAME, UTC);
-        assertEquals("Thursday", proc.process("0"));
+        assertEquals("Thursday", proc.process(dateTime(0L)));
-        assertEquals("Saturday", proc.process("-64164233612338"));
+        assertEquals("Saturday", proc.process(dateTime(-64164233612338L)));
-        assertEquals("Monday", proc.process("64164233612338"));
+        assertEquals("Monday", proc.process(dateTime(64164233612338L)));
         
-        assertEquals("Thursday", proc.process(new DateTime(0L, DateTimeZone.UTC)));
+        assertEquals("Thursday", proc.process(dateTime(0L)));
-        assertEquals("Thursday", proc.process(new DateTime(-5400, 12, 25, 2, 0, DateTimeZone.UTC)));
+        assertEquals("Thursday", proc.process(dateTime(-5400, 12, 25, 2, 0)));
-        assertEquals("Friday", proc.process(new DateTime(30, 2, 1, 12, 13, DateTimeZone.UTC)));
+        assertEquals("Friday", proc.process(dateTime(30, 2, 1, 12, 13)));
-        assertEquals("Tuesday", proc.process(new DateTime(10902, 8, 22, 11, 11, DateTimeZone.UTC)));
+        assertEquals("Tuesday", proc.process(dateTime(10902, 8, 22, 11, 11)));
     }
 
     public void testValidDayNamesWithNonUTCTimeZone() {
         assumeJava9PlusAndCompatLocaleProviderSetting();
         NamedDateTimeProcessor proc = new NamedDateTimeProcessor(NameExtractor.DAY_NAME, TimeZone.getTimeZone("GMT-10:00"));
-        assertEquals("Wednesday", proc.process("0"));
+        assertEquals("Wednesday", proc.process(dateTime(0)));
-        assertEquals("Friday", proc.process("-64164233612338"));
+        assertEquals("Friday", proc.process(dateTime(-64164233612338L)));
-        assertEquals("Monday", proc.process("64164233612338"));
+        assertEquals("Monday", proc.process(dateTime(64164233612338L)));
 
-        assertEquals("Wednesday", proc.process(new DateTime(0L, DateTimeZone.UTC)));
+        assertEquals("Wednesday", proc.process(dateTime(0L)));
-        assertEquals("Wednesday", proc.process(new DateTime(-5400, 12, 25, 2, 0, DateTimeZone.UTC)));
+        assertEquals("Wednesday", proc.process(dateTime(-5400, 12, 25, 2, 0)));
-        assertEquals("Friday", proc.process(new DateTime(30, 2, 1, 12, 13, DateTimeZone.UTC)));
+        assertEquals("Friday", proc.process(dateTime(30, 2, 1, 12, 13)));
-        assertEquals("Tuesday", proc.process(new DateTime(10902, 8, 22, 11, 11, DateTimeZone.UTC)));
+        assertEquals("Tuesday", proc.process(dateTime(10902, 8, 22, 11, 11)));
-        assertEquals("Monday", proc.process(new DateTime(10902, 8, 22, 9, 59, DateTimeZone.UTC)));
+        assertEquals("Monday", proc.process(dateTime(10902, 8, 22, 9, 59)));
     }
 
     public void testValidMonthNamesInUTC() {
         assumeJava9PlusAndCompatLocaleProviderSetting();
         NamedDateTimeProcessor proc  = new NamedDateTimeProcessor(NameExtractor.MONTH_NAME, UTC);
-        assertEquals("January", proc.process("0"));
+        assertEquals("January", proc.process(dateTime(0)));
-        assertEquals("September", proc.process("-64164233612338"));
+        assertEquals("September", proc.process(dateTime(-64165813612338L)));
-        assertEquals("April", proc.process("64164233612338"));
+        assertEquals("April", proc.process(dateTime(64164233612338L)));
 
-        assertEquals("January", proc.process(new DateTime(0L, DateTimeZone.UTC)));
+        assertEquals("January", proc.process(dateTime(0L)));
-        assertEquals("December", proc.process(new DateTime(-5400, 12, 25, 10, 10, DateTimeZone.UTC)));
+        assertEquals("December", proc.process(dateTime(-5400, 12, 25, 10, 10)));
-        assertEquals("February", proc.process(new DateTime(30, 2, 1, 12, 13, DateTimeZone.UTC)));
+        assertEquals("February", proc.process(dateTime(30, 2, 1, 12, 13)));
-        assertEquals("August", proc.process(new DateTime(10902, 8, 22, 11, 11, DateTimeZone.UTC)));
+        assertEquals("August", proc.process(dateTime(10902, 8, 22, 11, 11)));
     }
 
     public void testValidMonthNamesWithNonUTCTimeZone() {
         assumeJava9PlusAndCompatLocaleProviderSetting();
         NamedDateTimeProcessor proc = new NamedDateTimeProcessor(NameExtractor.MONTH_NAME, TimeZone.getTimeZone("GMT-3:00"));
-        assertEquals("December", proc.process("0"));
+        assertEquals("December", proc.process(dateTime(0)));
-        assertEquals("August", proc.process("-64165813612338")); // GMT: Tuesday, September 1, -0064 2:53:07.662 AM
+        assertEquals("August", proc.process(dateTime(-64165813612338L))); // GMT: Tuesday, September 1, -0064 2:53:07.662 AM
-        assertEquals("April", proc.process("64164233612338")); // GMT: Monday, April 14, 4003 2:13:32.338 PM
+        assertEquals("April", proc.process(dateTime(64164233612338L))); // GMT: Monday, April 14, 4003 2:13:32.338 PM
 
-        assertEquals("December", proc.process(new DateTime(0L, DateTimeZone.UTC)));
+        assertEquals("December", proc.process(dateTime(0L)));
-        assertEquals("November", proc.process(new DateTime(-5400, 12, 1, 1, 1, DateTimeZone.UTC)));
+        assertEquals("November", proc.process(dateTime(-5400, 12, 1, 1, 1)));
-        assertEquals("February", proc.process(new DateTime(30, 2, 1, 12, 13, DateTimeZone.UTC)));
+        assertEquals("February", proc.process(dateTime(30, 2, 1, 12, 13)));
-        assertEquals("July", proc.process(new DateTime(10902, 8, 1, 2, 59, DateTimeZone.UTC)));
+        assertEquals("July", proc.process(dateTime(10902, 8, 1, 2, 59)));
-        assertEquals("August", proc.process(new DateTime(10902, 8, 1, 3, 00, DateTimeZone.UTC)));
+        assertEquals("August", proc.process(dateTime(10902, 8, 1, 3, 00)));
     }
     
     /*
@@ -109,8 +109,8 @@ public class NamedDateTimeProcessorTests extends AbstractWireSerializingTestCase
         }
         String beforeJava9CompatibleLocale = System.getProperty("java.locale.providers");
         // and COMPAT setting needs to be first on the list
-        boolean isBeforeJava9Compatible = beforeJava9CompatibleLocale != null 
+        boolean isBeforeJava9Compatible = beforeJava9CompatibleLocale != null
                 && Strings.tokenizeToStringArray(beforeJava9CompatibleLocale, ",")[0].equals("COMPAT");
         Assume.assumeTrue(isBeforeJava9Compatible);
     }
-}
+}

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/QuarterProcessorTests.java

@@ -7,11 +7,11 @@
 package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 
 import org.elasticsearch.test.ESTestCase;
-import org.joda.time.DateTime;
-import org.joda.time.DateTimeZone;
 
 import java.util.TimeZone;
 
+import static org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeTestUtils.dateTime;
+
 public class QuarterProcessorTests extends ESTestCase {
 
     private static final TimeZone UTC = TimeZone.getTimeZone("UTC");
@@ -19,28 +19,28 @@ public class QuarterProcessorTests extends ESTestCase {
     public void testQuarterWithUTCTimezone() {
         QuarterProcessor proc = new QuarterProcessor(UTC);
         
-        assertEquals(1, proc.process(new DateTime(0L, DateTimeZone.UTC)));
+        assertEquals(1, proc.process(dateTime(0L)));
-        assertEquals(4, proc.process(new DateTime(-5400, 12, 25, 10, 10, DateTimeZone.UTC)));
+        assertEquals(4, proc.process(dateTime(-5400, 12, 25, 10, 10)));
-        assertEquals(1, proc.process(new DateTime(30, 2, 1, 12, 13, DateTimeZone.UTC)));
+        assertEquals(1, proc.process(dateTime(30, 2, 1, 12, 13)));
-        assertEquals(3, proc.process(new DateTime(10902, 8, 22, 11, 11, DateTimeZone.UTC)));
+        assertEquals(3, proc.process(dateTime(10902, 8, 22, 11, 11)));
         
-        assertEquals(1, proc.process("0"));
+        assertEquals(1, proc.process(dateTime(0L)));
-        assertEquals(3, proc.process("-64164233612338"));
+        assertEquals(3, proc.process(dateTime(-64164233612338L)));
-        assertEquals(2, proc.process("64164233612338"));
+        assertEquals(2, proc.process(dateTime(64164233612338L)));
     }
     
     public void testValidDayNamesWithNonUTCTimeZone() {
         QuarterProcessor proc = new QuarterProcessor(TimeZone.getTimeZone("GMT-10:00"));
-        assertEquals(4, proc.process(new DateTime(0L, DateTimeZone.UTC)));
+        assertEquals(4, proc.process(dateTime(0L)));
-        assertEquals(4, proc.process(new DateTime(-5400, 1, 1, 5, 0, DateTimeZone.UTC)));
+        assertEquals(4, proc.process(dateTime(-5400, 1, 1, 5, 0)));
-        assertEquals(1, proc.process(new DateTime(30, 4, 1, 9, 59, DateTimeZone.UTC)));
+        assertEquals(1, proc.process(dateTime(30, 4, 1, 9, 59)));
         
         proc = new QuarterProcessor(TimeZone.getTimeZone("GMT+10:00"));
-        assertEquals(4, proc.process(new DateTime(10902, 9, 30, 14, 1, DateTimeZone.UTC)));
+        assertEquals(4, proc.process(dateTime(10902, 9, 30, 14, 1)));
-        assertEquals(3, proc.process(new DateTime(10902, 9, 30, 13, 59, DateTimeZone.UTC)));
+        assertEquals(3, proc.process(dateTime(10902, 9, 30, 13, 59)));
         
-        assertEquals(1, proc.process("0"));
+        assertEquals(1, proc.process(dateTime(0L)));
-        assertEquals(3, proc.process("-64164233612338"));
+        assertEquals(3, proc.process(dateTime(-64164233612338L)));
-        assertEquals(2, proc.process("64164233612338"));
+        assertEquals(2, proc.process(dateTime(64164233612338L)));
     }
 }

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/planner/QueryTranslatorTests.java

@@ -28,7 +28,7 @@ import org.elasticsearch.xpack.sql.querydsl.query.TermQuery;
 import org.elasticsearch.xpack.sql.querydsl.query.TermsQuery;
 import org.elasticsearch.xpack.sql.type.EsField;
 import org.elasticsearch.xpack.sql.type.TypesTests;
-import org.joda.time.DateTime;
+import org.elasticsearch.xpack.sql.util.DateUtils;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 
@@ -150,7 +150,7 @@ public class QueryTranslatorTests extends ESTestCase {
         assertTrue(query instanceof RangeQuery);
         RangeQuery rq = (RangeQuery) query;
         assertEquals("date", rq.field());
-        assertEquals(DateTime.parse("1969-05-13T12:34:56Z"), rq.lower());
+        assertEquals(DateUtils.of("1969-05-13T12:34:56Z"), rq.lower());
     }
     
     public void testLikeConstructsNotSupported() {

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/type/DataTypeConversionTests.java

@@ -9,9 +9,11 @@ import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
 import org.elasticsearch.xpack.sql.expression.Literal;
 import org.elasticsearch.xpack.sql.type.DataTypeConversion.Conversion;
-import org.joda.time.DateTime;
+import org.elasticsearch.xpack.sql.util.DateUtils;
-import org.joda.time.DateTimeZone;
 
+import java.time.ZonedDateTime;
+
+import static org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeTestUtils.dateTime;
 import static org.elasticsearch.xpack.sql.tree.Location.EMPTY;
 
 public class DataTypeConversionTests extends ESTestCase {
@@ -22,7 +24,7 @@ public class DataTypeConversionTests extends ESTestCase {
 
         conversion = DataTypeConversion.conversionFor(DataType.DATE, DataType.KEYWORD);
         assertNull(conversion.convert(null));
-        assertEquals("1970-01-01T00:00:00.000Z", conversion.convert(new DateTime(0, DateTimeZone.UTC)));
+        assertEquals("1970-01-01T00:00:00.000Z", conversion.convert(dateTime(0)));
     }
 
     /**
@@ -64,33 +66,33 @@ public class DataTypeConversionTests extends ESTestCase {
         {
             Conversion conversion = DataTypeConversion.conversionFor(DataType.DOUBLE, to);
             assertNull(conversion.convert(null));
-            assertEquals(new DateTime(10L, DateTimeZone.UTC), conversion.convert(10.0));
+            assertEquals(dateTime(10L), conversion.convert(10.0));
-            assertEquals(new DateTime(10L, DateTimeZone.UTC), conversion.convert(10.1));
+            assertEquals(dateTime(10L), conversion.convert(10.1));
-            assertEquals(new DateTime(11L, DateTimeZone.UTC), conversion.convert(10.6));
+            assertEquals(dateTime(11L), conversion.convert(10.6));
             Exception e = expectThrows(SqlIllegalArgumentException.class, () -> conversion.convert(Double.MAX_VALUE));
             assertEquals("[" + Double.MAX_VALUE + "] out of [Long] range", e.getMessage());
         }
         {
             Conversion conversion = DataTypeConversion.conversionFor(DataType.INTEGER, to);
             assertNull(conversion.convert(null));
-            assertEquals(new DateTime(10L, DateTimeZone.UTC), conversion.convert(10));
+            assertEquals(dateTime(10L), conversion.convert(10));
-            assertEquals(new DateTime(-134L, DateTimeZone.UTC), conversion.convert(-134));
+            assertEquals(dateTime(-134L), conversion.convert(-134));
         }
         {
             Conversion conversion = DataTypeConversion.conversionFor(DataType.BOOLEAN, to);
             assertNull(conversion.convert(null));
-            assertEquals(new DateTime(1, DateTimeZone.UTC), conversion.convert(true));
+            assertEquals(dateTime(1), conversion.convert(true));
-            assertEquals(new DateTime(0, DateTimeZone.UTC), conversion.convert(false));
+            assertEquals(dateTime(0), conversion.convert(false));
         }
         Conversion conversion = DataTypeConversion.conversionFor(DataType.KEYWORD, to);
         assertNull(conversion.convert(null));
 
-        assertEquals(new DateTime(1000L, DateTimeZone.UTC), conversion.convert("1970-01-01T00:00:01Z"));
+        assertEquals(dateTime(1000L), conversion.convert("1970-01-01T00:00:01Z"));
-        assertEquals(new DateTime(1483228800000L, DateTimeZone.UTC), conversion.convert("2017-01-01T00:00:00Z"));
+        assertEquals(dateTime(1483228800000L), conversion.convert("2017-01-01T00:00:00Z"));
-        assertEquals(new DateTime(18000000L, DateTimeZone.UTC), conversion.convert("1970-01-01T00:00:00-05:00"));
+        assertEquals(dateTime(18000000L), conversion.convert("1970-01-01T00:00:00-05:00"));
         
         // double check back and forth conversion
-        DateTime dt = DateTime.now(DateTimeZone.UTC);
+        ZonedDateTime dt = ZonedDateTime.now(DateUtils.UTC);
         Conversion forward = DataTypeConversion.conversionFor(DataType.DATE, DataType.KEYWORD);
         Conversion back = DataTypeConversion.conversionFor(DataType.KEYWORD, DataType.DATE);
         assertEquals(dt, back.convert(forward.convert(dt)));