feat(static): support for partial datetimes on WindowStart bounds (#…

…3435)

With the change you no longer need to supply the full datetime when adding bounds on `WindowStart`.  You can drop precision on the right as much as you like, all the way until you have only a year. For example,

```sql
SELECT * FROM AGGREGATE WHERE '2020-02-23T23:45' <= WindowStart AND WindowStart < '2020-02-23T24' AND ROWKEY='10';
```

Numeric timezones are now also supported

```sql
SELECT * FROM AGGREGATE WHERE ROWKEY='10' AND WindowStart='2020-02-23T22:45:12.000-0100';
```

ksql-common/src/main/java/io/confluent/ksql/util/timestamp/PartialStringToTimestampParser.java

@@ -0,0 +1,111 @@
+/*
+ * Copyright 2019 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.util.timestamp;
+
+import static io.confluent.ksql.util.KsqlConstants.TIME_PATTERN;
+
+import io.confluent.ksql.util.KsqlConstants;
+import io.confluent.ksql.util.KsqlException;
+import java.time.ZoneId;
+
+/**
+ * A parser that can handle partially complete date-times.
+ *
+ * <p>A hack around the fact we do not as yet have a DATETIME type.
+ */
+public class PartialStringToTimestampParser {
+
+  private static final String HELP_MESSAGE = System.lineSeparator()
+      + "Required format is: \"" + KsqlConstants.DATE_TIME_PATTERN + "\", "
+      + "with an optional numeric timezone. "
+      + "Partials are also supported, for example \"2020-05-26\"";
+
+  private static final StringToTimestampParser PARSER =
+      new StringToTimestampParser(KsqlConstants.DATE_TIME_PATTERN);
+
+  @SuppressWarnings("MethodMayBeStatic") // Non-static to support DI.
+  public long parse(final String text) {
+
+    final String date;
+    final String time;
+    final String timezone;
+
+    if (text.contains("T")) {
+      date = text.substring(0, text.indexOf('T'));
+      final String withTimezone = completeTime(
+          text.substring(text.indexOf('T') + 1)
+      );
+      timezone = getTimezone(withTimezone);
+      time = completeTime(withTimezone.substring(0, withTimezone.length() - timezone.length()));
+    } else {
+      date = completeDate(text);
+      time = completeTime("");
+      timezone = "";
+    }
+
+    try {
+      if (timezone.length() > 0) {
+        return PARSER.parse(date + "T" + time, ZoneId.of(timezone));
+      } else {
+        return PARSER.parse(date + "T" + time);
+      }
+    } catch (final RuntimeException e) {
+      throw new KsqlException("Failed to parse timestamp '" + text
+          + "': " + e.getMessage()
+          + HELP_MESSAGE,
+          e
+      );
+    }
+  }
+
+  private static String getTimezone(final String time) {
+    if (time.contains("+")) {
+      return time.substring(time.indexOf('+'));
+    }
+
+    if (time.contains("-")) {
+      return time.substring(time.indexOf('-'));
+    }
+
+    return "";
+  }
+
+  private static String completeDate(final String date) {
+    final String[] parts = date.split("-");
+    if (parts.length == 1) {
+      return date + "-01-01";
+    }
+
+    if (parts.length == 2) {
+      return date + "-01";
+    }
+
+    // It is either a complete date or an incorrectly formatted one.
+    // In the latter case, we can pass the incorrectly formed string
+    // to the timestamp parser which will deal with the error handling.
+    return date;
+  }
+
+  private static String completeTime(final String time) {
+    if (time.length() >= TIME_PATTERN.length()) {
+      return time;
+    }
+
+    return time + TIME_PATTERN
+        .substring(time.length())
+        .replaceAll("[a-zA-Z]", "0");
+  }
+}

ksql-common/src/test/java/io/confluent/ksql/util/timestamp/PartialStringToTimestampParserTest.java

@@ -0,0 +1,129 @@
+/*
+ * Copyright 2019 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.util.timestamp;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+import io.confluent.ksql.util.KsqlConstants;
+import io.confluent.ksql.util.KsqlException;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+public class PartialStringToTimestampParserTest {
+
+  private static final StringToTimestampParser FULL_PARSER =
+      new StringToTimestampParser(KsqlConstants.DATE_TIME_PATTERN);
+
+  @Rule
+  public final ExpectedException expectedException = ExpectedException.none();
+
+  private PartialStringToTimestampParser parser;
+
+  @Before
+  public void init() {
+    parser = new PartialStringToTimestampParser();
+  }
+
+  @Test
+  public void shouldParseYear() {
+    // When:
+    assertThat(parser.parse("2017"), is(fullParse("2017-01-01T00:00:00.000")));
+  }
+
+  @Test
+  public void shouldParseYearMonth() {
+    // When:
+    assertThat(parser.parse("2020-02"), is(fullParse("2020-02-01T00:00:00.000")));
+  }
+
+  @Test
+  public void shouldParseFullDate() {
+    // When:
+    assertThat(parser.parse("2020-01-02"), is(fullParse("2020-01-02T00:00:00.000")));
+    assertThat(parser.parse("2020-01-02T"), is(fullParse("2020-01-02T00:00:00.000")));
+  }
+
+  @Test
+  public void shouldParseDateWithHour() {
+    // When:
+    assertThat(parser.parse("2020-12-02T13"), is(fullParse("2020-12-02T13:00:00.000")));
+  }
+
+  @Test
+  public void shouldParseDateWithHourMinute() {
+    // When:
+    assertThat(parser.parse("2020-12-02T13:59"), is(fullParse("2020-12-02T13:59:00.000")));
+  }
+
+  @Test
+  public void shouldParseDateWithHourMinuteSecond() {
+    // When:
+    assertThat(parser.parse("2020-12-02T13:59:58"), is(fullParse("2020-12-02T13:59:58.000")));
+  }
+
+  @Test
+  public void shouldParseFullDateTime() {
+    // When:
+    assertThat(parser.parse("2020-12-02T13:59:58.123"), is(fullParse("2020-12-02T13:59:58.123")));
+  }
+
+  @Test
+  public void shouldParseDateTimeWithPositiveTimezones() {
+    assertThat(parser.parse("2017-11-13T23:59:58.999+0100"), is(1510613998999L));
+  }
+
+  @Test
+  public void shouldParseDateTimeWithNegativeTimezones() {
+    assertThat(parser.parse("2017-11-13T23:59:58.999-0100"), is(1510621198999L));
+  }
+
+  @Test
+  public void shouldThrowOnIncorrectlyFormattedDateTime() {
+    // Expect:
+    expectedException.expect(KsqlException.class);
+    expectedException.expectMessage("Failed to parse timestamp '2017-1-1'");
+
+    // When:
+    parser.parse("2017-1-1");
+  }
+
+  @Test
+  public void shouldThrowOnTimezoneParseError() {
+    // Expect:
+    expectedException.expect(KsqlException.class);
+    expectedException.expectMessage("Failed to parse timestamp '2017-01-01T00:00:00.000+foo'");
+
+    // When:
+    parser.parse("2017-01-01T00:00:00.000+foo");
+  }
+
+  @Test
+  public void shouldIncludeRequiredFormatInErrorMessage() {
+    // Expect:
+    expectedException.expectMessage("Required format is: \"yyyy-MM-dd'T'HH:mm:ss.SSS\", "
+        + "with an optional numeric timezone. Partials are also supported, for example \"2020-05-26\"");
+
+    // When:
+    parser.parse("2017-01-01T00:00:00.000+foo");
+  }
+
+  private static long fullParse(final String text) {
+    return FULL_PARSER.parse(text);
+  }
+}