feat: define how DATE data is formatted

ksqldb-engine-common/src/main/java/io/confluent/ksql/schema/ksql/DefaultSqlValueCoercer.java

@@ -299,7 +299,7 @@ private static final class Rules {
             .put(key(DOUBLE, DOUBLE), Coercer.PASS_THROUGH)
             // STRING:
             .put(key(STRING, STRING), Coercer.PASS_THROUGH)
-            .put(key(STRING, TIMESTAMP), parser((v, t) -> SqlTimestamps.parseTimestamp(v)))
+            .put(key(STRING, TIMESTAMP), parser((v, t) -> SqlTimeTypes.parseTimestamp(v)))
             // ARRAY:
             .put(key(ARRAY, ARRAY), coercer(
                 DefaultSqlValueCoercer::canCoerceToArray,
@@ -343,7 +343,7 @@ private static final class Rules {
             .put(key(STRING, DOUBLE), parser((v, t) -> SqlDoubles.parseDouble(v)))
             // TIMESTAMP:
             .put(key(TIMESTAMP, STRING), coercer((c, v, t)
-                -> Result.of(SqlTimestamps.formatTimestamp((Timestamp) v))))
+                -> Result.of(SqlTimeTypes.formatTimestamp((Timestamp) v))))
             .build();
 
     private static Coercer parser(final BiFunction<String, SqlType, Object> parserFunction) {

ksqldb-engine-common/src/main/java/io/confluent/ksql/schema/ksql/SqlTimestamps.java → ksqldb-engine-common/src/main/java/io/confluent/ksql/schema/ksql/SqlTimeTypes.java

@@ -17,20 +17,23 @@
 
 import io.confluent.ksql.util.KsqlConstants;
 import io.confluent.ksql.util.timestamp.PartialStringToTimestampParser;
+import java.sql.Date;
 import java.sql.Time;
 import java.sql.Timestamp;
+import java.time.LocalDate;
 import java.time.LocalTime;
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Helpers for working with SQL time types.
  */
-public final class SqlTimestamps {
+public final class SqlTimeTypes {
 
   private static PartialStringToTimestampParser PARSER = new PartialStringToTimestampParser();
 
-  private SqlTimestamps() {
+  private SqlTimeTypes() {
   }
 
   /**
@@ -55,4 +58,8 @@ public static String formatTimestamp(final Timestamp timestamp) {
   public static String formatTime(final Time time) {
     return LocalTime.ofSecondOfDay(time.getTime() / 1000).toString();
   }
+
+  public static String formatDate(final Date date) {
+    return LocalDate.ofEpochDay(TimeUnit.MILLISECONDS.toDays(date.getTime())).toString();
+  }
 }

ksqldb-engine-common/src/test/java/io/confluent/ksql/schema/ksql/SqlTimestampsTest.java → ksqldb-engine-common/src/test/java/io/confluent/ksql/schema/ksql/SqlTimeTypesTest.java

@@ -20,31 +20,38 @@
 import static org.junit.Assert.assertThrows;
 
 import io.confluent.ksql.util.KsqlException;
+import java.sql.Date;
 import java.sql.Time;
 import java.sql.Timestamp;
 import org.junit.Test;
 
-public class SqlTimestampsTest {
+public class SqlTimeTypesTest {
   @Test
   public void shouldParseTimestamp() {
-    assertThat(SqlTimestamps.parseTimestamp("2019-03-17T10:00:00"), is(new Timestamp(1552816800000L)));
-    assertThat(SqlTimestamps.parseTimestamp("2019-03-17T03:00-0700"), is(new Timestamp(1552816800000L)));
+    assertThat(SqlTimeTypes.parseTimestamp("2019-03-17T10:00:00"), is(new Timestamp(1552816800000L)));
+    assertThat(SqlTimeTypes.parseTimestamp("2019-03-17T03:00-0700"), is(new Timestamp(1552816800000L)));
   }
 
   @Test
   public void shouldNotParseTimestamp() {
-    assertThrows(KsqlException.class, () -> SqlTimestamps.parseTimestamp("abc"));
-    assertThrows(KsqlException.class, () -> SqlTimestamps.parseTimestamp("2019-03-17 03:00"));
+    assertThrows(KsqlException.class, () -> SqlTimeTypes.parseTimestamp("abc"));
+    assertThrows(KsqlException.class, () -> SqlTimeTypes.parseTimestamp("2019-03-17 03:00"));
   }
 
   @Test
   public void shouldFormatTimestamp() {
-    assertThat(SqlTimestamps.formatTimestamp(new Timestamp(1552816800000L)), is("2019-03-17T10:00:00.000"));
+    assertThat(SqlTimeTypes.formatTimestamp(new Timestamp(1552816800000L)), is("2019-03-17T10:00:00.000"));
   }
 
   @Test
   public void shouldFormatTime() {
-    assertThat(SqlTimestamps.formatTime(new Time(1000)), is("00:00:01"));
-    assertThat(SqlTimestamps.formatTime(new Time(1005)), is("00:00:01"));
+    assertThat(SqlTimeTypes.formatTime(new Time(1000)), is("00:00:01"));
+    assertThat(SqlTimeTypes.formatTime(new Time(1005)), is("00:00:01"));
+  }
+
+  @Test
+  public void shouldFormatDate() {
+    assertThat(SqlTimeTypes.formatDate(new Date(864000000)), is("1970-01-11"));
+    assertThat(SqlTimeTypes.formatDate(new Date(864000000 - 10)), is("1970-01-10"));
   }
 }

ksqldb-engine/src/main/java/io/confluent/ksql/engine/rewrite/ExpressionTreeRewriter.java

@@ -28,6 +28,7 @@
 import io.confluent.ksql.execution.expression.tree.CreateMapExpression;
 import io.confluent.ksql.execution.expression.tree.CreateStructExpression;
 import io.confluent.ksql.execution.expression.tree.CreateStructExpression.Field;
+import io.confluent.ksql.execution.expression.tree.DateLiteral;
 import io.confluent.ksql.execution.expression.tree.DecimalLiteral;
 import io.confluent.ksql.execution.expression.tree.DereferenceExpression;
 import io.confluent.ksql.execution.expression.tree.DoubleLiteral;
@@ -530,6 +531,11 @@ public Expression visitTimeLiteral(final TimeLiteral node, final C context) {
       return plugin.apply(node, new Context<>(context, this)).orElse(node);
     }
 
+    @Override
+    public Expression visitDateLiteral(final DateLiteral node, final C context) {
+      return plugin.apply(node, new Context<>(context, this)).orElse(node);
+    }
+
     @Override
     public Expression visitTimestampLiteral(
         final TimestampLiteral node,

ksqldb-engine/src/test/java/io/confluent/ksql/engine/rewrite/ExpressionTreeRewriterTest.java

@@ -38,6 +38,7 @@
 import io.confluent.ksql.execution.expression.tree.CreateMapExpression;
 import io.confluent.ksql.execution.expression.tree.CreateStructExpression;
 import io.confluent.ksql.execution.expression.tree.CreateStructExpression.Field;
+import io.confluent.ksql.execution.expression.tree.DateLiteral;
 import io.confluent.ksql.execution.expression.tree.DecimalLiteral;
 import io.confluent.ksql.execution.expression.tree.DereferenceExpression;
 import io.confluent.ksql.execution.expression.tree.DoubleLiteral;
@@ -76,6 +77,7 @@
 import io.confluent.ksql.util.KsqlParserTestUtil;
 import io.confluent.ksql.util.MetaStoreFixture;
 import java.math.BigDecimal;
+import java.sql.Date;
 import java.sql.Time;
 import java.sql.Timestamp;
 import java.util.List;
@@ -101,6 +103,7 @@ public class ExpressionTreeRewriterTest {
       new NullLiteral(),
       new DecimalLiteral(BigDecimal.ONE),
       new TimeLiteral(new Time(1000L)),
+      new DateLiteral(new Date(864000000L)),
       new TimestampLiteral(new Timestamp(0))
   );
 

ksqldb-execution/src/main/java/io/confluent/ksql/execution/codegen/SqlToJavaVisitor.java

@@ -46,6 +46,7 @@
 import io.confluent.ksql.execution.expression.tree.CreateMapExpression;
 import io.confluent.ksql.execution.expression.tree.CreateStructExpression;
 import io.confluent.ksql.execution.expression.tree.CreateStructExpression.Field;
+import io.confluent.ksql.execution.expression.tree.DateLiteral;
 import io.confluent.ksql.execution.expression.tree.DecimalLiteral;
 import io.confluent.ksql.execution.expression.tree.DereferenceExpression;
 import io.confluent.ksql.execution.expression.tree.DoubleLiteral;
@@ -95,7 +96,7 @@
 import io.confluent.ksql.schema.ksql.SchemaConverters;
 import io.confluent.ksql.schema.ksql.SqlBooleans;
 import io.confluent.ksql.schema.ksql.SqlDoubles;
-import io.confluent.ksql.schema.ksql.SqlTimestamps;
+import io.confluent.ksql.schema.ksql.SqlTimeTypes;
 import io.confluent.ksql.schema.ksql.types.SqlArray;
 import io.confluent.ksql.schema.ksql.types.SqlBaseType;
 import io.confluent.ksql.schema.ksql.types.SqlDecimal;
@@ -163,7 +164,7 @@ public class SqlToJavaVisitor {
       InListEvaluator.class.getCanonicalName(),
       SqlDoubles.class.getCanonicalName(),
       SqlBooleans.class.getCanonicalName(),
-      SqlTimestamps.class.getCanonicalName()
+      SqlTimeTypes.class.getCanonicalName()
   );
 
   private static final Map<Operator, String> DECIMAL_OPERATOR_NAME = ImmutableMap
@@ -345,6 +346,14 @@ public Pair<String, SqlType> visitTimeLiteral(
       return new Pair<>(node.toString(), SqlTypes.TIME);
     }
 
+    @Override
+    public Pair<String, SqlType> visitDateLiteral(
+        final DateLiteral node,
+        final Context context
+    ) {
+      return new Pair<>(node.toString(), SqlTypes.DATE);
+    }
+
     @Override
     public Pair<String, SqlType> visitSimpleCaseExpression(
         final SimpleCaseExpression simpleCaseExpression,
@@ -725,7 +734,7 @@ private String toTimestamp(final SqlType schema, final int index) {
         case TIMESTAMP:
           return "%" + index + "$s";
         case STRING:
-          return "SqlTimestamps.parseTimestamp(%" + index + "$s)";
+          return "SqlTimeTypes.parseTimestamp(%" + index + "$s)";
         default:
           throw new KsqlException("Unexpected comparison to TIMESTAMP: " + schema.baseType());
       }

ksqldb-execution/src/main/java/io/confluent/ksql/execution/codegen/helpers/CastEvaluator.java

@@ -89,7 +89,7 @@ public final class CastEvaluator {
       .put(key(STRING, BIGINT), nonNullSafeCode("Long.parseLong(%s.trim())"))
       .put(key(STRING, DECIMAL), CastEvaluator::castToDecimal)
       .put(key(STRING, DOUBLE), nonNullSafeCode("SqlDoubles.parseDouble(%s.trim())"))
-      .put(key(STRING, TIMESTAMP), nonNullSafeCode("SqlTimestamps.parseTimestamp(%s.trim())"))
+      .put(key(STRING, TIMESTAMP), nonNullSafeCode("SqlTimeTypes.parseTimestamp(%s.trim())"))
       // ARRAY:
       .put(key(ARRAY, ARRAY), CastEvaluator::castArrayToArray)
       .put(key(ARRAY, STRING), CastEvaluator::castToString)
@@ -100,7 +100,7 @@ public final class CastEvaluator {
       .put(key(STRUCT, STRUCT), CastEvaluator::castStructToStruct)
       .put(key(STRUCT, STRING), CastEvaluator::castToString)
       // TIMESTAMP:
-      .put(key(TIMESTAMP, STRING), nonNullSafeCode("SqlTimestamps.formatTimestamp(%s)"))
+      .put(key(TIMESTAMP, STRING), nonNullSafeCode("SqlTimeTypes.formatTimestamp(%s)"))
       .build();
 
   private CastEvaluator() {

ksqldb-execution/src/main/java/io/confluent/ksql/execution/expression/formatter/ExpressionFormatter.java

@@ -29,6 +29,7 @@
 import io.confluent.ksql.execution.expression.tree.CreateArrayExpression;
 import io.confluent.ksql.execution.expression.tree.CreateMapExpression;
 import io.confluent.ksql.execution.expression.tree.CreateStructExpression;
+import io.confluent.ksql.execution.expression.tree.DateLiteral;
 import io.confluent.ksql.execution.expression.tree.DecimalLiteral;
 import io.confluent.ksql.execution.expression.tree.DereferenceExpression;
 import io.confluent.ksql.execution.expression.tree.DoubleLiteral;
@@ -59,7 +60,7 @@
 import io.confluent.ksql.execution.expression.tree.UnqualifiedColumnReferenceExp;
 import io.confluent.ksql.execution.expression.tree.WhenClause;
 import io.confluent.ksql.name.Name;
-import io.confluent.ksql.schema.ksql.SqlTimestamps;
+import io.confluent.ksql.schema.ksql.SqlTimeTypes;
 import io.confluent.ksql.schema.utils.FormatOptions;
 import io.confluent.ksql.util.KsqlConstants;
 import java.util.List;
@@ -191,12 +192,17 @@ public String visitDecimalLiteral(final DecimalLiteral node, final Context conte
 
     @Override
     public String visitTimeLiteral(final TimeLiteral node, final Context context) {
-      return SqlTimestamps.formatTime(node.getValue());
+      return SqlTimeTypes.formatTime(node.getValue());
+    }
+
+    @Override
+    public String visitDateLiteral(final DateLiteral node, final Context context) {
+      return SqlTimeTypes.formatDate(node.getValue());
     }
 
     @Override
     public String visitTimestampLiteral(final TimestampLiteral node, final Context context) {
-      return SqlTimestamps.formatTimestamp(node.getValue());
+      return SqlTimeTypes.formatTimestamp(node.getValue());
     }
 
     @Override

ksqldb-execution/src/main/java/io/confluent/ksql/execution/expression/tree/DateLiteral.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.execution.expression.tree;
+
+import static java.util.Objects.requireNonNull;
+
+import com.google.errorprone.annotations.Immutable;
+import io.confluent.ksql.parser.NodeLocation;
+import java.sql.Date;
+import java.util.Objects;
+import java.util.Optional;
+
+@Immutable
+public class DateLiteral extends Literal {
+
+  private final long value;
+
+  public DateLiteral(final Date value) {
+    this(Optional.empty(), value);
+  }
+
+  public DateLiteral(final Optional<NodeLocation> location, final Date value) {
+    super(location);
+    this.value = requireNonNull(value, "value").getTime();
+  }
+
+  @Override
+  public Date getValue() {
+    return new Date(value);
+  }
+
+  @Override
+  public <R, C> R accept(final ExpressionVisitor<R, C> visitor, final C context) {
+    return visitor.visitDateLiteral(this, context);
+  }
+
+  @Override
+  public boolean equals(final Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    final DateLiteral that = (DateLiteral) o;
+    return Objects.equals(value, that.value);
+  }
+
+  @Override
+  public int hashCode() {
+    return Long.hashCode(value);
+  }
+}

ksqldb-execution/src/main/java/io/confluent/ksql/execution/expression/tree/ExpressionVisitor.java

@@ -81,6 +81,8 @@ default R process(final Expression node, final C context) {
 
   R visitTimeLiteral(TimeLiteral exp, C context);
 
+  R visitDateLiteral(DateLiteral exp, C context);
+
   R visitTimestampLiteral(TimestampLiteral exp, C context);
 
   R visitType(Type exp, C context);

ksqldb-execution/src/main/java/io/confluent/ksql/execution/expression/tree/TraversalExpressionVisitor.java

@@ -188,6 +188,11 @@ public Void visitTimeLiteral(final TimeLiteral node, final C context) {
     return null;
   }
 
+  @Override
+  public Void visitDateLiteral(final DateLiteral node, final C context) {
+    return null;
+  }
+
   @Override
   public Void visitTimestampLiteral(final TimestampLiteral node, final C context) {
     return null;

ksqldb-execution/src/main/java/io/confluent/ksql/execution/expression/tree/VisitParentExpressionVisitor.java

@@ -76,6 +76,11 @@ public R visitTimeLiteral(final TimeLiteral node, final C context) {
     return visitLiteral(node, context);
   }
 
+  @Override
+  public R visitDateLiteral(final DateLiteral node, final C context) {
+    return visitLiteral(node, context);
+  }
+
   @Override
   public R visitTimestampLiteral(final TimestampLiteral node, final C context) {
     return visitLiteral(node, context);

ksqldb-execution/src/main/java/io/confluent/ksql/execution/interpreter/CastInterpreter.java

@@ -22,7 +22,7 @@
 import io.confluent.ksql.execution.interpreter.terms.Term;
 import io.confluent.ksql.schema.ksql.SqlBooleans;
 import io.confluent.ksql.schema.ksql.SqlDoubles;
-import io.confluent.ksql.schema.ksql.SqlTimestamps;
+import io.confluent.ksql.schema.ksql.SqlTimeTypes;
 import io.confluent.ksql.schema.ksql.types.SqlArray;
 import io.confluent.ksql.schema.ksql.types.SqlBaseType;
 import io.confluent.ksql.schema.ksql.types.SqlDecimal;
@@ -180,7 +180,7 @@ public static CastFunction castToStringFunction(
     if (from.baseType() == SqlBaseType.DECIMAL) {
       return object -> ((BigDecimal) object).toPlainString();
     } else if (from.baseType() == SqlBaseType.TIMESTAMP) {
-      return object -> SqlTimestamps.formatTimestamp(((Timestamp) object));
+      return object -> SqlTimeTypes.formatTimestamp(((Timestamp) object));
     }
     return object -> config.getBoolean(KsqlConfig.KSQL_STRING_CASE_CONFIG_TOGGLE)
         ? Objects.toString(object, null)
@@ -202,7 +202,7 @@ public static ComparableCastFunction<Date> castToTimestampFunction(
       final SqlType from
   ) {
     if (from.baseType() == SqlBaseType.STRING) {
-      return object -> SqlTimestamps.parseTimestamp(((String) object).trim());
+      return object -> SqlTimeTypes.parseTimestamp(((String) object).trim());
     } else if (from.baseType() == SqlBaseType.TIMESTAMP) {
       return object -> (Timestamp) object;
     }

ksqldb-execution/src/main/java/io/confluent/ksql/execution/interpreter/TermCompiler.java

@@ -35,6 +35,7 @@
 import io.confluent.ksql.execution.expression.tree.CreateMapExpression;
 import io.confluent.ksql.execution.expression.tree.CreateStructExpression;
 import io.confluent.ksql.execution.expression.tree.CreateStructExpression.Field;
+import io.confluent.ksql.execution.expression.tree.DateLiteral;
 import io.confluent.ksql.execution.expression.tree.DecimalLiteral;
 import io.confluent.ksql.execution.expression.tree.DereferenceExpression;
 import io.confluent.ksql.execution.expression.tree.DoubleLiteral;
@@ -224,6 +225,14 @@ public Term visitTimeLiteral(
     return visitUnsupported(timeLiteral);
   }
 
+  @Override
+  public Term visitDateLiteral(
+      final DateLiteral dateLiteral,
+      final Context context
+  ) {
+    return visitUnsupported(dateLiteral);
+  }
+
   @Override
   public Term visitSimpleCaseExpression(
       final SimpleCaseExpression simpleCaseExpression, final Context context