feat: New UNIX_TIMESTAMP and UNIX_DATE datetime functions

docs/developer-guide/syntax-reference.rst

@@ -1241,6 +1241,18 @@ Scalar functions
 +------------------------+---------------------------------------------------------------------------+---------------------------------------------------+
 | CONCAT                 |  ``CONCAT(col1, '_hello')``                                               | Concatenate two strings.                          |
 +------------------------+---------------------------------------------------------------------------+---------------------------------------------------+
+| CURRENTDATE            |  ``CURRENTDATE()``                                                        | Gets an integer representing days since epoch.    |
++------------------------+---------------------------------------------------------------------------+---------------------------------------------------+
+| UNIX_TIMESTAMP         |  ``UNIX_TIMESTAMP([TIMEZONE])``                                           | Gets the Unix timestamp in milliseconds,          |
+|                        |                                                                           | represented as a BIGINT.                          |
+|                        |                                                                           | TIMEZONE is an optional parameter and it is a     |
+|                        |                                                                           | ``java.util.TimeZone`` ID format, for example:    |
+|                        |                                                                           | "UTC", "America/Los_Angeles", "PDT",              |
+|                        |                                                                           | "Europe/London".                                  |
+|                        |                                                                           | Notice that the returned timestamp might be       |
+|                        |                                                                           | different in case of the local time in KSQL       |
+|                        |                                                                           | servers are not synchronized between them.        |
++------------------------+---------------------------------------------------------------------------+---------------------------------------------------+
 | DATETOSTRING           |  ``DATETOSTRING(START_DATE, 'yyyy-MM-dd')``                               | Converts an integer representation of a date into |
 |                        |                                                                           | a string representing the date in                 |
 |                        |                                                                           | the given format. Single quotes in the            |
@@ -1350,8 +1362,9 @@ Scalar functions
 |                        |                                                                           | two successive single quotes, ``''``, for         |
 |                        |                                                                           | example: ``'yyyy-MM-dd''T''HH:mm:ssX'``.          |
 |                        |                                                                           | TIMEZONE is an optional parameter and it is a     |
-|                        |                                                                           | java.util.TimeZone ID format, for example: "UTC", |
-|                        |                                                                           | "America/Los_Angeles", "PDT", "Europe/London"     |
+|                        |                                                                           | ``java.util.TimeZone`` ID format, for example:    |
+|                        |                                                                           | "UTC", "America/Los_Angeles", "PDT",              |
+|                        |                                                                           | "Europe/London".                                  |
 +------------------------+---------------------------------------------------------------------------+---------------------------------------------------+
 | SUBSTRING              |  ``SUBSTRING(col1, 2, 5)``                                                | ``SUBSTRING(str, pos, [len]``.                    |
 |                        |                                                                           | Returns a substring of ``str`` that starts at     |
@@ -1383,8 +1396,9 @@ Scalar functions
 |                        |                                                                           | successive single quotes, ``''``, for example:    |
 |                        |                                                                           | ``'yyyy-MM-dd''T''HH:mm:ssX'``.                   |
 |                        |                                                                           | TIMEZONE is an optional parameter and it is a     |
-|                        |                                                                           | java.util.TimeZone ID format, for example: "UTC", |
-|                        |                                                                           | "America/Los_Angeles", "PDT", "Europe/London"     |
+|                        |                                                                           | ``java.util.TimeZone`` ID format, for example:    |
+|                        |                                                                           | "UTC", "America/Los_Angeles", "PDT",              |
+|                        |                                                                           | "Europe/London".                                  |
 +------------------------+---------------------------------------------------------------------------+---------------------------------------------------+
 | TRIM                   |  ``TRIM(col1)``                                                           | Trim the spaces from the beginning and end of     |
 |                        |                                                                           | a string.                                         |

ksql-engine/src/main/java/io/confluent/ksql/function/udf/datetime/CurrentDate.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Confluent Community 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.function.udf.datetime;
+
+import io.confluent.ksql.function.udf.Udf;
+import io.confluent.ksql.function.udf.UdfDescription;
+import java.time.LocalDate;
+
+@UdfDescription(name = "currentdate",
+    description = "Gets an integer representing days since epoch.")
+public class CurrentDate {
+
+  @Udf(description = "Gets an integer representing days since epoch.")
+  public int currentDate() {
+    return ((int) LocalDate.now().toEpochDay());
+  }
+
+}

ksql-engine/src/main/java/io/confluent/ksql/function/udf/datetime/UnixTimestamp.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Confluent Community 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.function.udf.datetime;
+
+import io.confluent.ksql.function.KsqlFunctionException;
+import io.confluent.ksql.function.udf.Udf;
+import io.confluent.ksql.function.udf.UdfDescription;
+import io.confluent.ksql.function.udf.UdfParameter;
+import java.sql.Timestamp;
+import java.time.ZoneId;
+
+@UdfDescription(name = "unix_timestamp",
+    description = "Gets the Unix timestamp in milliseconds, represented as a BIGINT.")
+public class UnixTimestamp {
+
+  @Udf(description = "Gets a BIGINT millisecond from the Unix timestamp.")
+  public long unixTimestamp() {
+    return System.currentTimeMillis();
+  }
+
+  @Udf(description = "Gets a BIGINT millisecond from the Unix timestamp"
+      + " in the given time zone.")
+  public long unixTimestamp(
+      @UdfParameter(value = "timeZone",
+          description =  "timeZone is a java.util.TimeZone ID format, for example: \"UTC\","
+              + " \"America/Los_Angeles\", \"PDT\", \"Europe/London\"") final String timeZone) {
+    try {
+      final Timestamp timestamp = new Timestamp(System.currentTimeMillis());
+      final ZoneId zoneId = ZoneId.of(timeZone);
+      timestamp.toLocalDateTime().atZone(zoneId);
+      return timestamp.getTime();
+    } catch (final RuntimeException e) {
+      throw new KsqlFunctionException("Failed to get the Unix timestamp': "
+          + "' at timezone '" + timeZone + "': " + e.getMessage(), e);
+    }
+  }
+
+}

ksql-engine/src/test/java/io/confluent/ksql/function/udf/datetime/CurrentDateTest.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Confluent Community 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.function.udf.datetime;
+
+import org.junit.Before;
+import org.junit.Test;
+import java.time.LocalDate;
+
+import static org.junit.Assert.assertEquals;
+
+public class CurrentDateTest {
+
+  private CurrentDate udf;
+
+  @Before
+  public void setUp() {
+    udf = new CurrentDate();
+  }
+
+  @Test
+  public void shouldGetTheCurrentDate() {
+    // Given:
+    final int now = ((int) LocalDate.now().toEpochDay());
+
+    // When:
+    final int result = udf.currentDate();
+
+    // Then:
+    assertEquals(now, result);
+  }
+
+}

ksql-engine/src/test/java/io/confluent/ksql/function/udf/datetime/UnixTimestampTest.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Confluent Community 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.function.udf.datetime;
+
+import io.confluent.ksql.function.KsqlFunctionException;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import java.sql.Timestamp;
+import java.time.ZoneId;
+
+import static org.junit.Assert.assertTrue;
+
+public class UnixTimestampTest {
+
+  private UnixTimestamp udf;
+
+  @Rule
+  public final ExpectedException expectedException = ExpectedException.none();
+
+  @Before
+  public void setUp() {
+    udf = new UnixTimestamp();
+  }
+
+  @Test
+  public void shouldGetTheUnixTimestamp() {
+    // Given:
+    final long before = System.currentTimeMillis();
+
+    // When:
+    final long result = udf.unixTimestamp();
+    final long after = System.currentTimeMillis();
+
+    // Then:
+    assertTrue(before <= result && result <= after);
+  }
+
+  @Test
+  public void shouldGetTheUnixTimestampWithTimeZone() {
+    // Given:
+    final String timeZone = "UTC";
+    final Timestamp before = new Timestamp(System.currentTimeMillis());
+    before.toLocalDateTime().atZone(ZoneId.of(timeZone));
+
+    // When:
+    final long result = udf.unixTimestamp(timeZone);
+    final Timestamp after = new Timestamp(System.currentTimeMillis());
+    after.toLocalDateTime().atZone(ZoneId.of(timeZone));
+
+    // Then:
+    assertTrue(before.getTime() <= result && result <= after.getTime());
+  }
+
+  @Test
+  public void shouldThrowIfInvalidTimeZone() {
+    // Given:
+    expectedException.expect(KsqlFunctionException.class);
+    expectedException.expectMessage("Unknown time-zone ID: PST");
+
+    // When:
+    udf.unixTimestamp("PST");
+  }
+
+}