feat: Add timestamp arithmetic functionality

docs/developer-guide/ksqldb-reference/scalar-functions.md

@@ -1151,6 +1151,28 @@ FROM_UNIXTIME(milliseconds)
 
 Converts a BIGINT millisecond timestamp value into a TIMESTAMP value.
 
+### TIMESTAMPADD
+
+Since: 0.16
+
+```sql
+TIMESTAMPADD(COL0, 5 MINUTES)
+```
+
+Adds an interval to a timestamp. Intervals are defined by an integral value and a supported
+[time unit](../../reference/sql/time.md#Time units).
+
+### TIMESTAMPSUB
+
+Since: 0.16
+
+```sql
+TIMESTAMPSUB(COL0, 5 MINUTES)
+```
+
+Subtracts an interval from a timestamp. Intervals are defined by an integral value and a supported
+[time unit](../../reference/sql/time.md#Time units).
+
 ## URLs
 
 !!! note

docs/reference/sql/time.md

@@ -1,7 +1,7 @@
 ### Time units
 
 The following list shows valid time units for the `SIZE`, `ADVANCE BY`,
-`SESSION`, and `WITHIN` clauses.
+`SESSION`, and `WITHIN` clauses, or to pass as intervals in functions.
 
 -   `DAY`, `DAYS`
 -   `HOUR`, `HOURS`

ksqldb-common/src/main/java/io/confluent/ksql/function/types/IntervalType.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2021 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.function.types;
+
+public final class IntervalType extends ObjectType {
+  public static final IntervalType INSTANCE = new IntervalType();
+
+  private IntervalType() {
+  }
+
+  @Override
+  public int hashCode() {
+    return 8;
+  }
+
+  @Override
+  public boolean equals(final Object obj) {
+    return obj instanceof  IntervalType;
+  }
+
+  @Override
+  public String toString() {
+    return "INTERVAL";
+  }
+}

ksqldb-common/src/main/java/io/confluent/ksql/function/types/ParamTypes.java

@@ -25,6 +25,7 @@
 import io.confluent.ksql.schema.ksql.types.SqlStruct;
 import io.confluent.ksql.schema.ksql.types.SqlStruct.Field;
 import io.confluent.ksql.schema.ksql.types.SqlType;
+import io.confluent.ksql.schema.ksql.types.SqlTypes;
 import java.util.Map.Entry;
 import java.util.Optional;
 
@@ -40,6 +41,7 @@ private ParamTypes() {
   public static final LongType LONG = LongType.INSTANCE;
   public static final ParamType DECIMAL = DecimalType.INSTANCE;
   public static final TimestampType TIMESTAMP = TimestampType.INSTANCE;
+  public static final IntervalType INTERVAL = IntervalType.INSTANCE;
 
   public static boolean areCompatible(final SqlArgument actual, final ParamType declared) {
     return areCompatible(actual, declared, false);
@@ -102,6 +104,10 @@ && areCompatible(
       return isStructCompatible(argumentSqlType, declared);
     }
 
+    if (argumentSqlType == SqlTypes.INTERVAL && declared instanceof IntervalType) {
+      return true;
+    }
+
     return isPrimitiveMatch(argumentSqlType, declared, allowCast);
   }
 

ksqldb-common/src/main/java/io/confluent/ksql/parser/DurationParser.java

@@ -27,6 +27,11 @@ public final class DurationParser {
   private DurationParser() {
   }
 
+  public static Duration buildDuration(final long size, final String timeUnitName) {
+    final TimeUnit timeUnit = parseTimeUnit(timeUnitName.toUpperCase());
+    return Duration.ofNanos(timeUnit.toNanos(size));
+  }
+
   public static Duration parse(final String text) {
     try {
       final String[] parts = text.split("\\s");
@@ -35,9 +40,7 @@ public static Duration parse(final String text) {
       }
 
       final long size = parseNumeric(parts[0]);
-      final TimeUnit timeUnit = parseTimeUnit(parts[1].toUpperCase());
-
-      return Duration.ofNanos(timeUnit.toNanos(size));
+      return buildDuration(size, parts[1]);
     } catch (final Exception e) {
       throw new IllegalArgumentException("Invalid duration: '" + text + "'. " + e.getMessage(), e);
     }

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

@@ -19,6 +19,7 @@
 import com.google.common.collect.ImmutableBiMap;
 import com.google.common.collect.ImmutableMap;
 import io.confluent.ksql.function.types.ArrayType;
+import io.confluent.ksql.function.types.IntervalType;
 import io.confluent.ksql.function.types.MapType;
 import io.confluent.ksql.function.types.ParamType;
 import io.confluent.ksql.function.types.ParamTypes;
@@ -381,6 +382,7 @@ private static class FunctionToSql implements FunctionToSqlConverter {
             .put(ParamTypes.LONG, SqlTypes.BIGINT)
             .put(ParamTypes.DOUBLE, SqlTypes.DOUBLE)
             .put(ParamTypes.TIMESTAMP, SqlTypes.TIMESTAMP)
+            .put(ParamTypes.INTERVAL, SqlTypes.INTERVAL)
             .build();
 
     @Override
@@ -444,6 +446,10 @@ public SqlBaseType toBaseType(final ParamType paramType) {
         return SqlBaseType.STRUCT;
       }
 
+      if (paramType instanceof IntervalType) {
+        return null;
+      }
+
       throw new KsqlException("Cannot convert param type to sql type: " + paramType);
     }
   }

ksqldb-common/src/test/java/io/confluent/ksql/function/types/ParamTypesTest.java

@@ -82,6 +82,9 @@ public void shouldPassCompatibleSchemas() {
         false),
         is(true));
 
+    assertThat(ParamTypes.areCompatible(SqlArgument.of(SqlTypes.INTERVAL), ParamTypes.INTERVAL),
+        is(true));
+
     assertThat(
         ParamTypes.areCompatible(
             SqlArgument.of(SqlTypes.array(SqlTypes.INTEGER)),
@@ -127,6 +130,7 @@ public void shouldNotPassInCompatibleSchemasWithImplicitCasting() {
     assertThat(ParamTypes.areCompatible(SqlArgument.of(SqlTypes.DOUBLE), ParamTypes.LONG, true), is(false));
 
     assertThat(ParamTypes.areCompatible(SqlArgument.of(SqlTypes.DOUBLE), ParamTypes.DECIMAL, true), is(false));
+    assertThat(ParamTypes.areCompatible(SqlArgument.of(SqlTypes.BIGINT), ParamTypes.INTERVAL, true), is(false));
   }
 }
 

ksqldb-common/src/test/java/io/confluent/ksql/parser/DurationParserTest.java

@@ -135,4 +135,9 @@ public void shouldSupportHours() {
   public void shouldSupportDays() {
     assertThat(DurationParser.parse("98 Day"), is(Duration.ofDays(98)));
   }
+
+  @Test
+  public void shouldBuildDuration() {
+    assertThat(DurationParser.buildDuration(20, "DAYS"), is(Duration.ofDays(20)));
+  }
 }

ksqldb-common/src/test/java/io/confluent/ksql/schema/ksql/SchemaConvertersTest.java

@@ -115,6 +115,7 @@ public class SchemaConvertersTest {
       .put(SqlTypes.DOUBLE, ParamTypes.DOUBLE)
       .put(SqlTypes.STRING, ParamTypes.STRING)
       .put(SqlTypes.TIMESTAMP, ParamTypes.TIMESTAMP)
+      .put(SqlTypes.INTERVAL, ParamTypes.INTERVAL)
       .put(SqlArray.of(SqlTypes.INTEGER), ArrayType.of(ParamTypes.INTEGER))
       .put(SqlDecimal.of(2, 1), ParamTypes.DECIMAL)
       .put(

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

@@ -37,6 +37,7 @@
 import io.confluent.ksql.execution.expression.tree.InListExpression;
 import io.confluent.ksql.execution.expression.tree.InPredicate;
 import io.confluent.ksql.execution.expression.tree.IntegerLiteral;
+import io.confluent.ksql.execution.expression.tree.IntervalExpression;
 import io.confluent.ksql.execution.expression.tree.IsNotNullPredicate;
 import io.confluent.ksql.execution.expression.tree.IsNullPredicate;
 import io.confluent.ksql.execution.expression.tree.LambdaFunctionCall;
@@ -519,6 +520,18 @@ public Expression visitDecimalLiteral(final DecimalLiteral node, final C context
       return plugin.apply(node, new Context<>(context, this)).orElse(node);
     }
 
+    @Override
+    public Expression visitIntervalExpression(final IntervalExpression node, final C context) {
+      final Optional<Expression> result
+          = plugin.apply(node, new Context<>(context, this));
+      if (result.isPresent()) {
+        return result.get();
+      }
+
+      final Expression expr = rewriter.apply(node.getExpression(), context);
+      return new IntervalExpression(node.getLocation(), expr, node.getTimeUnit());
+    }
+
     @Override
     public Expression visitTimeLiteral(final TimeLiteral node, final C context) {
       return plugin.apply(node, new Context<>(context, this)).orElse(node);

ksqldb-engine/src/main/java/io/confluent/ksql/function/UdafTypes.java

@@ -31,6 +31,7 @@
 import java.lang.reflect.TypeVariable;
 import java.math.BigDecimal;
 import java.sql.Timestamp;
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -56,6 +57,7 @@ class UdafTypes {
       .add(List.class)
       .add(Map.class)
       .add(Timestamp.class)
+      .add(Duration.class)
       .add(Function.class)
       .add(BiFunction.class)
       .add(TriFunction.class)

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

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2021 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.function.udf.datetime;
+
+import io.confluent.ksql.function.FunctionCategory;
+import io.confluent.ksql.function.udf.Udf;
+import io.confluent.ksql.function.udf.UdfDescription;
+import io.confluent.ksql.function.udf.UdfParameter;
+import io.confluent.ksql.util.KsqlConstants;
+import java.sql.Timestamp;
+import java.time.Duration;
+
+@UdfDescription(
+    name = "timestampadd",
+    category = FunctionCategory.DATE_TIME,
+    author = KsqlConstants.CONFLUENT_AUTHOR,
+    description = "Adds a duration to a TIMESTAMP value."
+)
+public class TimestampAdd {
+
+  @Udf(description = "Adds a duration to a timestamp")
+  public Timestamp timestampAdd(
+      @UdfParameter(description = "A TIMESTAMP value.") final Timestamp timestamp,
+      @UdfParameter(description = "The duration to add. This is denoted by [number] [time unit],"
+          + "for example 3 MONTHS or 10 SECONDS") final Duration duration) {
+    return new Timestamp(timestamp.getTime() + duration.toMillis());
+  }
+}

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

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2021 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.function.udf.datetime;
+
+import io.confluent.ksql.function.FunctionCategory;
+import io.confluent.ksql.function.udf.Udf;
+import io.confluent.ksql.function.udf.UdfDescription;
+import io.confluent.ksql.function.udf.UdfParameter;
+import io.confluent.ksql.util.KsqlConstants;
+import java.sql.Timestamp;
+import java.time.Duration;
+
+@UdfDescription(
+    name = "timestampsub",
+    category = FunctionCategory.DATE_TIME,
+    author = KsqlConstants.CONFLUENT_AUTHOR,
+    description = "Subtracts a duration from a TIMESTAMP value."
+)
+public class TimestampSub {
+
+  @Udf(description = "Subtracts a duration from a timestamp")
+  public Timestamp timestampSub(
+      @UdfParameter(description = "A TIMESTAMP value.") final Timestamp timestamp,
+      @UdfParameter(description = "The duration to subtract. This is denoted by"
+          + "[number] [time unit], for example 3 MONTHS or 10 SECONDS.") final Duration duration) {
+    return new Timestamp(timestamp.getTime() - duration.toMillis());
+  }
+}

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

@@ -46,6 +46,7 @@
 import io.confluent.ksql.execution.expression.tree.InListExpression;
 import io.confluent.ksql.execution.expression.tree.InPredicate;
 import io.confluent.ksql.execution.expression.tree.IntegerLiteral;
+import io.confluent.ksql.execution.expression.tree.IntervalExpression;
 import io.confluent.ksql.execution.expression.tree.IsNotNullPredicate;
 import io.confluent.ksql.execution.expression.tree.IsNullPredicate;
 import io.confluent.ksql.execution.expression.tree.LikePredicate;
@@ -78,6 +79,7 @@
 import java.sql.Timestamp;
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
 import org.junit.Before;
 import org.junit.Test;
@@ -746,6 +748,19 @@ public void shouldRewriteTypeUsingPlugin() {
     shouldRewriteUsingPlugin(type);
   }
 
+  @Test
+  public void shouldRewriteIntervalExpression() {
+    // Given:
+    final IntervalExpression expression = new IntervalExpression(LITERALS.get(0), TimeUnit.DAYS);
+    when(processor.apply(expression.getExpression(), context)).thenReturn(expr1);
+
+    // When:
+    final Expression rewritten = expressionRewriter.rewrite(expression, context);
+
+    // Then:
+    assertThat(rewritten, equalTo(new IntervalExpression(expression.getLocation(), expr1, expression.getTimeUnit())));
+  }
+
   @SuppressWarnings("unchecked")
   private <T extends Expression> T parseExpression(final String asText) {
     final String ksql = String.format("SELECT %s FROM test1;", asText);

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

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2021 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.function.udf.datetime;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import java.sql.Timestamp;
+import java.time.Duration;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TimestampAddTest {
+  private TimestampAdd udf;
+
+  @Before
+  public void setUp() {
+    udf = new TimestampAdd();
+  }
+
+  @Test
+  public void addToTimestamp() {
+    // When:
+    final Timestamp result = udf.timestampAdd(new Timestamp(100), Duration.ofMillis(50));
+
+    // Then:
+    final Timestamp expectedResult = new Timestamp(150);
+    assertThat(result, is(expectedResult));
+  }
+}