Report date data as a standardized format

build.gradle

@@ -223,6 +223,7 @@ dependencies {
     compile group: "org.elasticsearch.plugin", name: 'reindex-client', version: "${es_version}"
     compile group: 'com.google.guava', name: 'guava', version:'15.0'
     compile group: 'org.json', name: 'json', version:'20180813'
+    compile group: 'org.apache.commons', name: 'commons-lang3', version: '3.9'
 
     // ANTLR gradle plugin and runtime dependency
     antlr "org.antlr:antlr4:4.7.1"

src/main/java/com/amazon/opendistroforelasticsearch/sql/antlr/semantic/visitor/ESMappingLoader.java

@@ -105,9 +105,25 @@ private void defineIndexType(String indexName) {
 
     private void loadAllFieldsWithType(String indexName) {
         FieldMappings mappings = getFieldMappings(indexName);
+        saveDateFormats(mappings);
         mappings.flat(this::defineFieldName);
     }
 
+    private void saveDateFormats(FieldMappings mappings) {
+        for (Map.Entry<String, Map<String, Object>> data : mappings.data().entrySet()) {
+            String fieldName = data.getKey();
+            Object type = data.getValue().get("type");
+            if ("date".equals(type)) {
+                Object fieldDateFormat = data.getValue().get("format");
+                if (fieldDateFormat == null) {
+                    // Default format when not specified is "date_optional_time"
+                    fieldDateFormat = "date_optional_time";
+                }
+                clusterState.pushDateFieldFormat(fieldName, fieldDateFormat.toString());
+            }
+        }
+    }
+
     /*
      *  3.1 Define with alias if given: ex."SELECT * FROM accounts a".
      *      'accounts' -> INDEX

src/main/java/com/amazon/opendistroforelasticsearch/sql/esdomain/LocalClusterState.java

@@ -31,6 +31,7 @@
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -91,6 +92,7 @@ public class LocalClusterState {
      */
     private final Map<String, Object> latestSettings = new ConcurrentHashMap<>();
 
+    private Map<String, String> dateFieldFormatMap = new HashMap<>();
 
     public static synchronized LocalClusterState state() {
         if (INSTANCE == null) {
@@ -242,4 +244,11 @@ private <T> List<T> sortToList(T[] array) {
         return Arrays.asList(array);
     }
 
+    public void pushDateFieldFormat(String fieldName, String dateFormat) {
+        dateFieldFormatMap.put(fieldName, dateFormat);
+    }
+
+    public Map<String, String> getDateFieldFormatMap() {
+        return dateFieldFormatMap;
+    }
 }

src/main/java/com/amazon/opendistroforelasticsearch/sql/executor/format/DateFieldFormatter.java

@@ -0,0 +1,96 @@
+/*
+ *   Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License").
+ *   You may not use this file except in compliance with the License.
+ *   A copy of the License is located at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   or in the "license" file accompanying this file. This file 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 com.amazon.opendistroforelasticsearch.sql.executor.format;
+
+import org.apache.commons.lang3.time.DateUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.text.ParseException;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+
+public class DateFieldFormatter {
+    private static final Logger LOG = LogManager.getLogger(DateFieldFormatter.class);
+    private static final String FORMAT_JDBC = "yyyy-MM-dd HH:mm:ss.SSS";
+
+    private static final String FORMAT_DOT_DATE_AND_TIME = "yyyy-MM-dd'T'HH:mm:ss.SSSZ";
+    private static final String FORMAT_DOT_KIBANA_SAMPLE_DATA_LOGS_EXCEPTION = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'";
+    private static final String FORMAT_DOT_KIBANA_SAMPLE_DATA_FLIGHTS_EXCEPTION = "yyyy-MM-dd'T'HH:mm:ss";
+    private static final String FORMAT_DOT_KIBANA_SAMPLE_DATA_ECOMMERCE_EXCEPTION = "yyyy-MM-dd'T'HH:mm:ssXXX";
+    private static final String FORMAT_DOT_DATE = "yyyy-MM-dd";
+
+    private final Map<String, String> dateFieldFormatMap;
+    private List<Schema.Column> columns;
+
+    public DateFieldFormatter(Map<String, String> dateFieldFormatMap, List<Schema.Column> columns) {
+        this.dateFieldFormatMap = dateFieldFormatMap;
+        this.columns = columns;
+    }
+
+    public void applyJDBCDateFormat(Map<String, Object> rowSource) {
+        for (Schema.Column column : columns) {
+            String columnType = column.getType();
+            String columnName = column.getName();
+
+            if (columnType.equals(Schema.Type.DATE.nameLowerCase())) {
+                String columnFormat = dateFieldFormatMap.get(columnName);
+                DateFormat format = DateFormat.valueOf(columnFormat.toUpperCase());
+
+                Object columnOriginalDate = rowSource.get(columnName);
+                if (columnOriginalDate == null) {
+                    // Don't try to parse null date values
+                    continue;
+                }
+
+                Date date = parseDateString(format, columnOriginalDate.toString());
+                if (date != null) {
+                    rowSource.put(columnName, DateFormat.getFormattedDate(date, FORMAT_JDBC));
+                } else {
+                    LOG.warn("Could not parse date value; returning original value");
+                }
+            }
+        }
+    }
+
+    private Date parseDateString(DateFormat format, String columnOriginalDate) {
+        try {
+            switch (format) {
+                case DATE_OPTIONAL_TIME:
+                    return DateUtils.parseDate(
+                        columnOriginalDate,
+                        FORMAT_DOT_KIBANA_SAMPLE_DATA_LOGS_EXCEPTION,
+                        FORMAT_DOT_KIBANA_SAMPLE_DATA_FLIGHTS_EXCEPTION,
+                        FORMAT_DOT_KIBANA_SAMPLE_DATA_ECOMMERCE_EXCEPTION,
+                        FORMAT_DOT_DATE_AND_TIME,
+                        FORMAT_DOT_DATE);
+                case EPOCH_MILLIS:
+                    return new Date(Long.parseLong(columnOriginalDate));
+                case EPOCH_SECOND:
+                    return new Date(Long.parseLong(columnOriginalDate) * 1000);
+                default:
+                    return DateUtils.parseDate(columnOriginalDate, format.getFormatString());
+            }
+        } catch (ParseException e) {
+            LOG.error(
+                String.format("Error parsing date string %s as %s", columnOriginalDate, format.nameLowerCase()),
+                e);
+        }
+        return null;
+    }
+
+}

src/main/java/com/amazon/opendistroforelasticsearch/sql/executor/format/DateFormat.java

@@ -0,0 +1,142 @@
+/*
+ *   Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License").
+ *   You may not use this file except in compliance with the License.
+ *   A copy of the License is located at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   or in the "license" file accompanying this file. This file 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 com.amazon.opendistroforelasticsearch.sql.executor.format;
+
+// import java.text.SimpleDateFormat;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+// import java.util.TimeZone;
+
+public enum DateFormat {
+  // Special cases that are parsed separately
+  DATE_OPTIONAL_TIME(""),
+  EPOCH_MILLIS(""),
+  EPOCH_SECOND(""),
+
+  BASIC_DATE(Date.BASIC_DATE),
+  BASIC_DATE_TIME(Date.BASIC_DATE + Time.T + Time.BASIC_TIME + Time.MILLIS + Time.TZ),
+  BASIC_DATE_TIME_NO_MILLIS(Date.BASIC_DATE + Time.T + Time.BASIC_TIME + Time.TZ),
+
+  BASIC_ORDINAL_DATE(Date.BASIC_ORDINAL_DATE),
+  BASIC_ORDINAL_DATE_TIME(Date.BASIC_ORDINAL_DATE + Time.T + Time.BASIC_TIME + Time.MILLIS + Time.TZ),
+  BASIC_ORDINAL_DATE_TIME_NO_MILLIS(Date.BASIC_ORDINAL_DATE+ Time.T + Time.BASIC_TIME + Time.TZ),
+
+  BASIC_TIME(Time.BASIC_TIME + Time.MILLIS + Time.TZ),
+  BASIC_TIME_NO_MILLIS(Time.BASIC_TIME + Time.TZ),
+
+  BASIC_T_TIME(Time.T + Time.BASIC_TIME + Time.MILLIS + Time.TZ),
+  BASIC_T_TIME_NO_MILLIS(Time.T + Time.BASIC_TIME + Time.TZ),
+
+  BASIC_WEEK_DATE(Date.BASIC_WEEK_DATE),
+  BASIC_WEEK_DATE_TIME(Date.BASIC_WEEK_DATE + Time.T + Time.BASIC_TIME + Time.MILLIS + Time.TZ),
+  BASIC_WEEK_DATE_TIME_NO_MILLIS(Date.BASIC_WEEK_DATE + Time.T + Time.BASIC_TIME + Time.TZ),
+
+  DATE(Date.DATE),
+  DATE_HOUR(Date.DATE + Time.T + Time.HOUR),
+  DATE_HOUR_MINUTE(Date.DATE + Time.T + Time.HOUR_MINUTE),
+  DATE_HOUR_MINUTE_SECOND(Date.DATE + Time.T + Time.TIME),
+  DATE_HOUR_MINUTE_SECOND_FRACTION(Date.DATE + Time.T + Time.TIME + Time.MILLIS),
+  DATE_HOUR_MINUTE_SECOND_MILLIS(Date.DATE + Time.T + Time.TIME + Time.MILLIS),
+  DATE_TIME(Date.DATE + Time.T + Time.TIME + Time.MILLIS + Time.TZZ),
+  DATE_TIME_NO_MILLIS(Date.DATE + Time.T + Time.TIME + Time.TZZ),
+
+  HOUR(Time.HOUR),
+  HOUR_MINUTE(Time.HOUR_MINUTE),
+  HOUR_MINUTE_SECOND(Time.TIME),
+  HOUR_MINUTE_SECOND_FRACTION(Time.TIME + Time.MILLIS),
+  HOUR_MINUTE_SECOND_MILLIS(Time.TIME + Time.MILLIS),
+
+  ORDINAL_DATE(Date.ORDINAL_DATE),
+  ORDINAL_DATE_TIME(Date.ORDINAL_DATE + Time.T + Time.TIME + Time.MILLIS + Time.TZZ),
+  ORDINAL_DATE_TIME_NO_MILLIS(Date.ORDINAL_DATE + Time.T + Time.TIME + Time.TZZ),
+
+  TIME(Time.TIME + Time.MILLIS + Time.TZZ),
+  TIME_NO_MILLIS(Time.TIME + Time.TZZ),
+
+  T_TIME(Time.T + Time.TIME + Time.MILLIS + Time.TZZ),
+  T_TIME_NO_MILLIS(Time.T + Time.TIME + Time.TZZ),
+
+  WEEK_DATE(Date.WEEK_DATE),
+  WEEK_DATE_TIME(Date.WEEK_DATE + Time.T + Time.TIME + Time.MILLIS + Time.TZZ),
+  WEEK_DATE_TIME_NO_MILLIS(Date.WEEK_DATE + Time.T + Time.TIME + Time.TZZ),
+
+  // Note: input mapping is "weekyear", but output value is "week_year"
+  WEEK_YEAR(Date.WEEKYEAR),
+  WEEKYEAR_WEEK(Date.WEEKYEAR_WEEK),
+  WEEKYEAR_WEEK_DAY(Date.WEEK_DATE),
+
+  YEAR(Date.YEAR),
+  YEAR_MONTH(Date.YEAR_MONTH),
+  YEAR_MONTH_DAY(Date.DATE);
+
+  private static class Date {
+    static String BASIC_DATE = "yyyyMMdd";
+    static String BASIC_ORDINAL_DATE = "yyyyDDD";
+    static String BASIC_WEEK_DATE = "YYYY'W'wwu";
+
+    static String DATE = "yyyy-MM-dd";
+    static String ORDINAL_DATE = "yyyy-DDD";
+
+    static String YEAR = "yyyy";
+    static String YEAR_MONTH = "yyyy-MM";
+
+    static String WEEK_DATE = "YYYY-'W'ww-u";
+    static String WEEKYEAR = "YYYY";
+    static String WEEKYEAR_WEEK = "YYYY-'W'ww";
+  }
+
+  private static class Time {
+    static String T = "'T'";
+    static String BASIC_TIME = "HHmmss";
+    static String TIME = "HH:mm:ss";
+
+    static String HOUR = "HH";
+    static String HOUR_MINUTE = "HH:mm";
+
+    static String MILLIS = ".SSS";
+    static String TZ = "Z";
+    static String TZZ = "XX";
+  }
+
+  private String formatString;
+
+  DateFormat(String formatString) {
+    this.formatString = formatString;
+  }
+
+  public String getFormatString() {
+    return formatString;
+  }
+
+  public String nameLowerCase() {
+    return name().toLowerCase();
+  }
+
+  public static String getFormattedDate(java.util.Date date, String dateFormat) {
+    Instant instant = date.toInstant();
+    ZonedDateTime zdt = ZonedDateTime.ofInstant(instant, ZoneId.of("Etc/UTC"));
+    // return zdt.format(DateTimeFormatter.ofPattern(dateFormat));
+    return zdt.format(DateTimeFormatter.ofPattern(dateFormat));
+    // return DateTimeFormatter.ofLocalizedDateTime().format(date.toInstant());
+    // SimpleDateFormat formatter = new SimpleDateFormat(dateFormat);
+    // formatter.applyLocalizedPattern(dateFormat);
+    // formatter.setTimeZone(TimeZone.getTimeZone("UTC"));
+    // return formatter.format(date);
+    // formatter.
+  }
+}

src/main/java/com/amazon/opendistroforelasticsearch/sql/executor/format/Protocol.java

@@ -30,7 +30,9 @@
 import org.json.JSONArray;
 import org.json.JSONObject;
 
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
@@ -42,6 +44,7 @@ public class Protocol {
     static final int ERROR_STATUS = 500;
 
     private final String formatType;
+    private Map<String, String> dateFieldFormatMap = new HashMap<>();
     private int status;
     private long size;
     private long total;
@@ -56,6 +59,7 @@ public Protocol(Client client, QueryAction queryAction, Object queryResult, Stri
         }
         this.formatType = formatType;
         QueryStatement query = queryAction.getQueryStatement();
+        this.dateFieldFormatMap = queryAction.getDateFieldFormatMap();
         this.status = OK_STATUS;
         this.resultSet = loadResultSet(client, query, queryResult);
         this.size = resultSet.getDataRows().getSize();
@@ -75,7 +79,7 @@ private ResultSet loadResultSet(Client client, QueryStatement queryStatement, Ob
         if (queryStatement instanceof Delete) {
             return new DeleteResultSet(client, (Delete) queryStatement, queryResult);
         } else if (queryStatement instanceof Query) {
-            return new SelectResultSet(client, (Query) queryStatement, queryResult);
+            return new SelectResultSet(client, (Query) queryStatement, queryResult, dateFieldFormatMap);
         } else if (queryStatement instanceof IndexStatement) {
             IndexStatement statement = (IndexStatement) queryStatement;
             StatementType statementType = statement.getStatementType();

src/main/java/com/amazon/opendistroforelasticsearch/sql/executor/format/SelectResultSet.java

@@ -71,7 +71,9 @@ public class SelectResultSet extends ResultSet {
     private long totalHits;
     private List<DataRows.Row> rows;
 
-    public SelectResultSet(Client client, Query query, Object queryResult) {
+    private DateFieldFormatter dateFieldFormatter;
+
+    public SelectResultSet(Client client, Query query, Object queryResult, Map<String, String> dateFieldFormatMap) {
         this.client = client;
         this.query = query;
         this.queryResult = queryResult;
@@ -86,6 +88,7 @@ public SelectResultSet(Client client, Query query, Object queryResult) {
         }
         this.schema = new Schema(indexName, typeName, columns);
         this.head = schema.getHeaders();
+        this.dateFieldFormatter = new DateFieldFormatter(dateFieldFormatMap, columns);
 
         extractData();
         this.dataRows = new DataRows(size, totalHits, rows);
@@ -515,6 +518,8 @@ private List<DataRows.Row> populateRows(SearchHits searchHits) {
         Set<String> newKeys = new HashSet<>(head);
         for (SearchHit hit : searchHits) {
             Map<String, Object> rowSource = hit.getSourceAsMap();
+            dateFieldFormatter.applyJDBCDateFormat(rowSource);
+
             List<DataRows.Row> result;
 
             if (!isJoinQuery()) {

src/main/java/com/amazon/opendistroforelasticsearch/sql/plugin/RestSqlAction.java

@@ -72,6 +72,7 @@
 public class RestSqlAction extends BaseRestHandler {
 
     private static final Logger LOG = LogManager.getLogger(RestSqlAction.class);
+    private static Map<String, String> dateFieldFormatMap = new HashMap<>();
 
     private final boolean allowExplicitIndex;
 
@@ -150,6 +151,7 @@ private static QueryAction explainRequest(final NodeClient client, final SqlRequ
         final QueryAction queryAction = new SearchDao(client)
                 .explain(new QueryActionRequest(sqlRequest.getSql(), typeProvider, format));
         queryAction.setSqlRequest(sqlRequest);
+        queryAction.setDateFieldFormatMap(dateFieldFormatMap);
         return queryAction;
     }
 
@@ -219,6 +221,9 @@ private static ColumnTypeProvider performAnalysis(String sql) {
 
         OpenDistroSqlAnalyzer analyzer = new OpenDistroSqlAnalyzer(config);
         Optional<Type> outputColumnType = analyzer.analyze(sql, clusterState);
+
+        dateFieldFormatMap = clusterState.getDateFieldFormatMap();
+
         if (outputColumnType.isPresent()) {
             return new ColumnTypeProvider(outputColumnType.get());
         } else {