flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From twal...@apache.org
Subject flink git commit: [FLINK-4359] [table] Add INTERVAL type
Date Thu, 11 Aug 2016 12:40:45 GMT
Repository: flink
Updated Branches:
  refs/heads/master f0992aa13 -> f170e04ab


[FLINK-4359] [table] Add INTERVAL type

This closes #2348.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f170e04a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f170e04a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f170e04a

Branch: refs/heads/master
Commit: f170e04ab93fba75c517b0f9184587d7b306113e
Parents: f0992aa
Author: twalthr <twalthr@apache.org>
Authored: Mon Aug 8 10:15:49 2016 +0200
Committer: twalthr <twalthr@apache.org>
Committed: Thu Aug 11 14:39:24 2016 +0200

----------------------------------------------------------------------
 docs/apis/table.md                              |  42 +-
 .../flink/api/scala/table/expressionDsl.scala   |  59 ++-
 .../flink/api/table/FlinkTypeFactory.scala      |  39 +-
 .../org/apache/flink/api/table/Types.scala      |   3 +
 .../flink/api/table/codegen/CodeGenUtils.scala  |  46 ++-
 .../flink/api/table/codegen/CodeGenerator.scala |  50 ++-
 .../table/codegen/calls/ScalarOperators.scala   | 113 +++++-
 .../table/expressions/ExpressionParser.scala    |  62 ++-
 .../api/table/expressions/ExpressionUtils.scala |  40 ++
 .../api/table/expressions/arithmetic.scala      |  48 ++-
 .../flink/api/table/expressions/cast.scala      |   3 +-
 .../flink/api/table/expressions/literals.scala  |  20 +
 .../api/table/typeutils/IntervalTypeInfo.scala  | 109 ++++++
 .../api/table/typeutils/TypeCheckUtils.scala    |  10 +-
 .../api/table/typeutils/TypeCoercion.scala      |   8 +
 .../api/java/batch/table/ExpressionsITCase.java |   5 +-
 .../scala/expression/TemporalTypesTest.scala    | 389 +++++++++++++++++++
 .../api/scala/expression/TimeTypesTest.scala    | 231 -----------
 18 files changed, 947 insertions(+), 330 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/docs/apis/table.md
----------------------------------------------------------------------
diff --git a/docs/apis/table.md b/docs/apis/table.md
index 57252d9..ea8e343 100644
--- a/docs/apis/table.md
+++ b/docs/apis/table.md
@@ -957,11 +957,13 @@ unary = [ "!" | "-" ] , composite ;
 
 composite = suffixed | atom ;
 
-suffixed = cast | as | aggregation | nullCheck | if | functionCall ;
+suffixed = interval | cast | as | aggregation | nullCheck | if | functionCall ;
+
+interval = composite , "." , ("year" | "month" | "day" | "hour" | "minute" | "second" | "milli") ;
 
 cast = composite , ".cast(" , dataType , ")" ;
 
-dataType = "BYTE" | "SHORT" | "INT" | "LONG" | "FLOAT" | "DOUBLE" | "BOOLEAN" | "STRING" | "DECIMAL" | "DATE" | "TIME" | "TIMESTAMP";
+dataType = "BYTE" | "SHORT" | "INT" | "LONG" | "FLOAT" | "DOUBLE" | "BOOLEAN" | "STRING" | "DECIMAL" | "DATE" | "TIME" | "TIMESTAMP" | "INTERVAL_MONTHS" | "INTERVAL_MILLIS" ;
 
 as = composite , ".as(" , fieldReference , ")" ;
 
@@ -986,6 +988,8 @@ If working with exact numeric values or large decimals is required, the Table AP
 
 In order to work with temporal values the Table API supports Java SQL's Date, Time, and Timestamp types. In the Scala Table API literals can be defined by using `java.sql.Date.valueOf("2016-06-27")`, `java.sql.Time.valueOf("10:10:42")`, or `java.sql.Timestamp.valueOf("2016-06-27 10:10:42.123")`. The Java and Scala Table API also support calling `"2016-06-27".toDate()`, `"10:10:42".toTime()`, and `"2016-06-27 10:10:42.123".toTimestamp()` for converting Strings into temporal types. *Note:* Since Java's temporal SQL types are time zone dependent, please make sure that the Flink Client and all TaskManagers use the same time zone.
 
+Temporal intervals can be represented as number of months (`Types.INTERVAL_MONTHS`) or number of milliseconds (`Types.INTERVAL_MILLIS`). Intervals of same type can be added or subtracted (e.g. `2.hour + 10.minutes`). Intervals of milliseconds can be added to time points (e.g. `"2016-08-10".toDate + 5.day`).
+
 {% top %}
 
 
@@ -1037,8 +1041,8 @@ The current version supports selection (filter), projection, inner equi-joins, g
 
 Among others, the following SQL features are not supported, yet:
 
-- Time interval data type (`INTERVAL`)
-- Timestamps are limited to milliseconds precision
+- Timestamps and intervals are limited to milliseconds precision
+- Interval arithmetic is currenly limited
 - Distinct aggregates (e.g., `COUNT(DISTINCT name)`)
 - Non-equi joins and Cartesian products
 - Grouping sets
@@ -1171,20 +1175,22 @@ Data Types
 
 The Table API is built on top of Flink's DataSet and DataStream API. Internally, it also uses Flink's `TypeInformation` to distinguish between types. The Table API does not support all Flink types so far. All supported simple types are listed in `org.apache.flink.api.table.Types`. The following table summarizes the relation between Table API types, SQL types, and the resulting Java class.
 
-| Table API              | SQL             | Java type              |
-| :--------------------- | :-------------- | :--------------------- |
-| `Types.STRING`         | `VARCHAR`       | `java.lang.String`     |
-| `Types.BOOLEAN`        | `BOOLEAN`       | `java.lang.Boolean`    |
-| `Types.BYTE`           | `TINYINT`       | `java.lang.Byte`       |
-| `Types.SHORT`          | `SMALLINT`      | `java.lang.Short`      |
-| `Types.INT`            | `INTEGER, INT`  | `java.lang.Integer`    |
-| `Types.LONG`           | `BIGINT`        | `java.lang.Long`       |
-| `Types.FLOAT`          | `REAL, FLOAT`   | `java.lang.Float`      |
-| `Types.DOUBLE`         | `DOUBLE`        | `java.lang.Double`     |
-| `Types.DECIMAL`        | `DECIMAL`       | `java.math.BigDecimal` |
-| `Types.DATE`           | `DATE`          | `java.sql.Date`        |
-| `Types.TIME`           | `TIME`          | `java.sql.Time`        |
-| `Types.TIMESTAMP`      | `TIMESTAMP`     | `java.sql.Timestamp`   |
+| Table API              | SQL                         | Java type              |
+| :--------------------- | :-------------------------- | :--------------------- |
+| `Types.STRING`         | `VARCHAR`                   | `java.lang.String`     |
+| `Types.BOOLEAN`        | `BOOLEAN`                   | `java.lang.Boolean`    |
+| `Types.BYTE`           | `TINYINT`                   | `java.lang.Byte`       |
+| `Types.SHORT`          | `SMALLINT`                  | `java.lang.Short`      |
+| `Types.INT`            | `INTEGER, INT`              | `java.lang.Integer`    |
+| `Types.LONG`           | `BIGINT`                    | `java.lang.Long`       |
+| `Types.FLOAT`          | `REAL, FLOAT`               | `java.lang.Float`      |
+| `Types.DOUBLE`         | `DOUBLE`                    | `java.lang.Double`     |
+| `Types.DECIMAL`        | `DECIMAL`                   | `java.math.BigDecimal` |
+| `Types.DATE`           | `DATE`                      | `java.sql.Date`        |
+| `Types.TIME`           | `TIME`                      | `java.sql.Time`        |
+| `Types.TIMESTAMP`      | `TIMESTAMP(3)`              | `java.sql.Timestamp`   |
+| `Types.INTERVAL_MONTHS`| `INTERVAL YEAR TO MONTH`    | `java.lang.Integer`    |
+| `Types.INTERVAL_MILLIS`| `INTERVAL DAY TO SECOND(3)` | `java.lang.Long`       |
 
 Advanced types such as generic types, composite types (e.g. POJOs or Tuples), and arrays can be fields of a row but can not be accessed yet. They are treated like a black box within Table API and SQL.
 

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
index cb91066..9ca9c8a 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
@@ -17,13 +17,15 @@
  */
 package org.apache.flink.api.scala.table
 
-import java.sql.{Timestamp, Time, Date}
-
-import scala.language.implicitConversions
+import java.sql.{Date, Time, Timestamp}
 
+import org.apache.calcite.avatica.util.DateTimeUtils._
 import org.apache.flink.api.common.typeinfo.{SqlTimeTypeInfo, TypeInformation}
+import org.apache.flink.api.table.expressions.ExpressionUtils.{toMilliInterval, toMonthInterval}
 import org.apache.flink.api.table.expressions._
 
+import scala.language.implicitConversions
+
 /**
  * These are all the operations that can be used to construct an [[Expression]] AST for expression
  * operations.
@@ -227,6 +229,57 @@ trait ImplicitExpressionOperations {
     * Parses a timestamp String in the form "yy-mm-dd hh:mm:ss.fff" to a SQL Timestamp.
     */
   def toTimestamp = Cast(expr, SqlTimeTypeInfo.TIMESTAMP)
+
+  // interval types
+
+  /**
+    * Creates an interval of the given number of years.
+    *
+    * @return interval of months
+    */
+  def year = toMonthInterval(expr, 12)
+
+  /**
+    * Creates an interval of the given number of months.
+    *
+    * @return interval of months
+    */
+  def month = toMonthInterval(expr, 1)
+
+  /**
+    * Creates an interval of the given number of days.
+    *
+    * @return interval of milliseconds
+    */
+  def day = toMilliInterval(expr, MILLIS_PER_DAY)
+
+    /**
+    * Creates an interval of the given number of hours.
+    *
+    * @return interval of milliseconds
+    */
+  def hour = toMilliInterval(expr, MILLIS_PER_HOUR)
+
+    /**
+    * Creates an interval of the given number of minutes.
+    *
+    * @return interval of milliseconds
+    */
+  def minute = toMilliInterval(expr, MILLIS_PER_MINUTE)
+
+    /**
+    * Creates an interval of the given number of seconds.
+    *
+    * @return interval of milliseconds
+    */
+  def second = toMilliInterval(expr, MILLIS_PER_SECOND)
+
+    /**
+    * Creates an interval of the given number of milliseconds.
+    *
+    * @return interval of milliseconds
+    */
+  def milli = toMilliInterval(expr, 1)
 }
 
 /**

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkTypeFactory.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkTypeFactory.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkTypeFactory.scala
index 6a31487..5a116db 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkTypeFactory.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkTypeFactory.scala
@@ -18,15 +18,19 @@
 
 package org.apache.flink.api.table
 
+import org.apache.calcite.avatica.util.TimeUnit
 import org.apache.calcite.jdbc.JavaTypeFactoryImpl
 import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeSystem}
+import org.apache.calcite.sql.SqlIntervalQualifier
 import org.apache.calcite.sql.`type`.SqlTypeName
 import org.apache.calcite.sql.`type`.SqlTypeName._
+import org.apache.calcite.sql.parser.SqlParserPos
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
 import org.apache.flink.api.common.typeinfo.{SqlTimeTypeInfo, TypeInformation}
 import org.apache.flink.api.java.typeutils.ValueTypeInfo._
 import org.apache.flink.api.table.FlinkTypeFactory.typeInfoToSqlTypeName
 import org.apache.flink.api.table.plan.schema.GenericRelDataType
+import org.apache.flink.api.table.typeutils.IntervalTypeInfo
 import org.apache.flink.api.table.typeutils.TypeCheckUtils.isSimple
 
 import scala.collection.mutable
@@ -42,7 +46,20 @@ class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImp
   def createTypeFromTypeInfo(typeInfo: TypeInformation[_]): RelDataType = {
     // simple type can be converted to SQL types and vice versa
     if (isSimple(typeInfo)) {
-      createSqlType(typeInfoToSqlTypeName(typeInfo))
+      val sqlType = typeInfoToSqlTypeName(typeInfo)
+      sqlType match {
+
+        case INTERVAL_YEAR_MONTH =>
+          createSqlIntervalType(
+            new SqlIntervalQualifier(TimeUnit.YEAR, TimeUnit.MONTH, SqlParserPos.ZERO))
+
+        case INTERVAL_DAY_TIME =>
+          createSqlIntervalType(
+            new SqlIntervalQualifier(TimeUnit.DAY, TimeUnit.SECOND, SqlParserPos.ZERO))
+
+        case _ =>
+          createSqlType(sqlType)
+      }
     }
     // advanced types require specific RelDataType
     // for storing the original TypeInformation
@@ -58,7 +75,7 @@ class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImp
       new GenericRelDataType(typeInfo, getTypeSystem.asInstanceOf[FlinkTypeSystem])
 
     case ti@_ =>
-      throw new TableException(s"Unsupported type information: $ti")
+      throw TableException(s"Unsupported type information: $ti")
   }
 }
 
@@ -75,16 +92,18 @@ object FlinkTypeFactory {
       case STRING_TYPE_INFO => VARCHAR
       case BIG_DEC_TYPE_INFO => DECIMAL
 
-      // date/time types
+      // temporal types
       case SqlTimeTypeInfo.DATE => DATE
       case SqlTimeTypeInfo.TIME => TIME
       case SqlTimeTypeInfo.TIMESTAMP => TIMESTAMP
+      case IntervalTypeInfo.INTERVAL_MONTHS => INTERVAL_YEAR_MONTH
+      case IntervalTypeInfo.INTERVAL_MILLIS => INTERVAL_DAY_TIME
 
       case CHAR_TYPE_INFO | CHAR_VALUE_TYPE_INFO =>
-        throw new TableException("Character type is not supported.")
+        throw TableException("Character type is not supported.")
 
       case _@t =>
-        throw new TableException(s"Type is not supported: $t")
+        throw TableException(s"Type is not supported: $t")
   }
 
   def toTypeInfo(relDataType: RelDataType): TypeInformation[_] = relDataType.getSqlTypeName match {
@@ -98,15 +117,15 @@ object FlinkTypeFactory {
     case VARCHAR | CHAR => STRING_TYPE_INFO
     case DECIMAL => BIG_DEC_TYPE_INFO
 
-    // date/time types
+    // temporal types
     case DATE => SqlTimeTypeInfo.DATE
     case TIME => SqlTimeTypeInfo.TIME
     case TIMESTAMP => SqlTimeTypeInfo.TIMESTAMP
-    case INTERVAL_DAY_TIME | INTERVAL_YEAR_MONTH =>
-      throw new TableException("Intervals are not supported yet.")
+    case INTERVAL_YEAR_MONTH => IntervalTypeInfo.INTERVAL_MONTHS
+    case INTERVAL_DAY_TIME => IntervalTypeInfo.INTERVAL_MILLIS
 
     case NULL =>
-      throw new TableException("Type NULL is not supported. " +
+      throw TableException("Type NULL is not supported. " +
         "Null values must have a supported type.")
 
     // symbol for special flags e.g. TRIM's BOTH, LEADING, TRAILING
@@ -119,6 +138,6 @@ object FlinkTypeFactory {
       genericRelDataType.typeInfo
 
     case _@t =>
-      throw new TableException(s"Type is not supported: $t")
+      throw TableException(s"Type is not supported: $t")
   }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/Types.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/Types.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/Types.scala
index d63683e..64d4612 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/Types.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/Types.scala
@@ -18,6 +18,7 @@
 package org.apache.flink.api.table
 
 import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, SqlTimeTypeInfo}
+import org.apache.flink.api.table.typeutils.IntervalTypeInfo
 
 /**
   * This class enumerates all supported types of the Table API.
@@ -38,5 +39,7 @@ object Types {
   val DATE = SqlTimeTypeInfo.DATE
   val TIME = SqlTimeTypeInfo.TIME
   val TIMESTAMP = SqlTimeTypeInfo.TIMESTAMP
+  val INTERVAL_MONTHS = IntervalTypeInfo.INTERVAL_MONTHS
+  val INTERVAL_MILLIS = IntervalTypeInfo.INTERVAL_MILLIS
 
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenUtils.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenUtils.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenUtils.scala
index c0f4fc8..18bf49f 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenUtils.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenUtils.scala
@@ -28,7 +28,7 @@ import org.apache.flink.api.common.typeinfo.{FractionalTypeInfo, SqlTimeTypeInfo
 import org.apache.flink.api.common.typeutils.CompositeType
 import org.apache.flink.api.java.typeutils.{PojoTypeInfo, TupleTypeInfo, TypeExtractor}
 import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
-import org.apache.flink.api.table.typeutils.{RowTypeInfo, TypeCheckUtils}
+import org.apache.flink.api.table.typeutils.{IntervalTypeInfo, RowTypeInfo, TypeCheckUtils}
 
 object CodeGenUtils {
 
@@ -65,11 +65,15 @@ object CodeGenUtils {
     case BOOLEAN_PRIMITIVE_ARRAY_TYPE_INFO => "boolean[]"
     case CHAR_PRIMITIVE_ARRAY_TYPE_INFO => "char[]"
 
-    // internal primitive representation of Date/Time/Timestamp
+    // internal primitive representation of time points
     case SqlTimeTypeInfo.DATE => "int"
     case SqlTimeTypeInfo.TIME => "int"
     case SqlTimeTypeInfo.TIMESTAMP => "long"
 
+    // internal primitive representation of time intervals
+    case IntervalTypeInfo.INTERVAL_MONTHS => "int"
+    case IntervalTypeInfo.INTERVAL_MILLIS => "long"
+
     case _ =>
       tpe.getTypeClass.getCanonicalName
   }
@@ -100,7 +104,10 @@ object CodeGenUtils {
     case BOOLEAN_TYPE_INFO => "false"
     case STRING_TYPE_INFO => "\"\""
     case CHAR_TYPE_INFO => "'\\0'"
-    case SqlTimeTypeInfo.DATE | SqlTimeTypeInfo.TIME | SqlTimeTypeInfo.TIMESTAMP => "-1"
+    case SqlTimeTypeInfo.DATE | SqlTimeTypeInfo.TIME => "-1"
+    case SqlTimeTypeInfo.TIMESTAMP => "-1L"
+    case IntervalTypeInfo.INTERVAL_MONTHS => "-1"
+    case IntervalTypeInfo.INTERVAL_MILLIS => "-1L"
 
     case _ => "null"
   }
@@ -113,7 +120,10 @@ object CodeGenUtils {
   def qualifyMethod(method: Method): String =
     method.getDeclaringClass.getCanonicalName + "." + method.getName
 
-  def internalToTemporalCode(resultType: TypeInformation[_], resultTerm: String) =
+  def qualifyEnum(enum: Enum[_]): String =
+    enum.getClass.getCanonicalName + "." + enum.name()
+
+  def internalToTimePointCode(resultType: TypeInformation[_], resultTerm: String) =
     resultType match {
       case SqlTimeTypeInfo.DATE =>
         s"${qualifyMethod(BuiltInMethod.INTERNAL_TO_DATE.method)}($resultTerm)"
@@ -123,7 +133,7 @@ object CodeGenUtils {
         s"${qualifyMethod(BuiltInMethod.INTERNAL_TO_TIMESTAMP.method)}($resultTerm)"
     }
 
-  def temporalToInternalCode(resultType: TypeInformation[_], resultTerm: String) =
+  def timePointToInternalCode(resultType: TypeInformation[_], resultTerm: String) =
     resultType match {
       case SqlTimeTypeInfo.DATE =>
         s"${qualifyMethod(BuiltInMethod.DATE_TO_INT.method)}($resultTerm)"
@@ -146,15 +156,25 @@ object CodeGenUtils {
       throw new CodeGenException(s"Comparable type expected, but was '${genExpr.resultType}'.")
     }
 
-  def requireString(genExpr: GeneratedExpression) = genExpr.resultType match {
-    case STRING_TYPE_INFO => // ok
-    case _ => throw new CodeGenException("String expression type expected.")
-  }
+  def requireString(genExpr: GeneratedExpression) =
+    if (!TypeCheckUtils.isString(genExpr.resultType)) {
+      throw new CodeGenException("String expression type expected.")
+    }
 
-  def requireBoolean(genExpr: GeneratedExpression) = genExpr.resultType match {
-    case BOOLEAN_TYPE_INFO => // ok
-    case _ => throw new CodeGenException("Boolean expression type expected.")
-  }
+  def requireBoolean(genExpr: GeneratedExpression) =
+    if (!TypeCheckUtils.isBoolean(genExpr.resultType)) {
+      throw new CodeGenException("Boolean expression type expected.")
+    }
+
+  def requireTemporal(genExpr: GeneratedExpression) =
+    if (!TypeCheckUtils.isTemporal(genExpr.resultType)) {
+      throw new CodeGenException("Temporal expression type expected.")
+    }
+
+  def requireTimeInterval(genExpr: GeneratedExpression) =
+    if (!TypeCheckUtils.isTimeInterval(genExpr.resultType)) {
+      throw new CodeGenException("Interval expression type expected.")
+    }
 
   // ----------------------------------------------------------------------------------------------
 

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
index c1508d8..92f9761 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
@@ -29,13 +29,13 @@ import org.apache.flink.api.common.typeinfo.{AtomicType, SqlTimeTypeInfo, TypeIn
 import org.apache.flink.api.common.typeutils.CompositeType
 import org.apache.flink.api.java.typeutils.{PojoTypeInfo, TupleTypeInfo}
 import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
-import org.apache.flink.api.table.{FlinkTypeFactory, TableConfig}
 import org.apache.flink.api.table.codegen.CodeGenUtils._
 import org.apache.flink.api.table.codegen.Indenter.toISC
 import org.apache.flink.api.table.codegen.calls.ScalarFunctions
 import org.apache.flink.api.table.codegen.calls.ScalarOperators._
 import org.apache.flink.api.table.typeutils.RowTypeInfo
-import org.apache.flink.api.table.typeutils.TypeCheckUtils.{isNumeric, isString, isTemporal}
+import org.apache.flink.api.table.typeutils.TypeCheckUtils._
+import org.apache.flink.api.table.{FlinkTypeFactory, TableConfig}
 
 import scala.collection.JavaConversions._
 import scala.collection.mutable
@@ -618,6 +618,20 @@ class CodeGenerator(
         generateNonNullLiteral(resultType, value.toString)
       case TIMESTAMP =>
         generateNonNullLiteral(resultType, value.toString + "L")
+      case INTERVAL_YEAR_MONTH =>
+        val decimal = BigDecimal(value.asInstanceOf[JBigDecimal])
+        if (decimal.isValidInt) {
+          generateNonNullLiteral(resultType, decimal.intValue().toString)
+        } else {
+          throw new CodeGenException("Decimal can not be converted to interval of months.")
+        }
+      case INTERVAL_DAY_TIME =>
+        val decimal = BigDecimal(value.asInstanceOf[JBigDecimal])
+        if (decimal.isValidLong) {
+          generateNonNullLiteral(resultType, decimal.longValue().toString + "L")
+        } else {
+          throw new CodeGenException("Decimal can not be converted to interval of milliseconds.")
+        }
 
       case t@_ =>
         throw new CodeGenException(s"Type not supported: $t")
@@ -645,11 +659,12 @@ class CodeGenerator(
         requireNumeric(right)
         generateArithmeticOperator("+", nullCheck, resultType, left, right)
 
-      case PLUS if isString(resultType) =>
+      case PLUS | DATETIME_PLUS if isTemporal(resultType) =>
         val left = operands.head
         val right = operands(1)
-        requireString(left)
-        generateStringConcatOperator(nullCheck, left, right)
+        requireTemporal(left)
+        requireTemporal(right)
+        generateTemporalPlusMinus(plus = true, nullCheck, left, right)
 
       case MINUS if isNumeric(resultType) =>
         val left = operands.head
@@ -658,6 +673,13 @@ class CodeGenerator(
         requireNumeric(right)
         generateArithmeticOperator("-", nullCheck, resultType, left, right)
 
+      case MINUS if isTemporal(resultType) =>
+        val left = operands.head
+        val right = operands(1)
+        requireTemporal(left)
+        requireTemporal(right)
+        generateTemporalPlusMinus(plus = false, nullCheck, left, right)
+
       case MULTIPLY if isNumeric(resultType) =>
         val left = operands.head
         val right = operands(1)
@@ -684,11 +706,21 @@ class CodeGenerator(
         requireNumeric(operand)
         generateUnaryArithmeticOperator("-", nullCheck, resultType, operand)
 
+      case UNARY_MINUS if isTimeInterval(resultType) =>
+        val operand = operands.head
+        requireTimeInterval(operand)
+        generateUnaryIntervalPlusMinus(plus = false, nullCheck, operand)
+
       case UNARY_PLUS if isNumeric(resultType) =>
         val operand = operands.head
         requireNumeric(operand)
         generateUnaryArithmeticOperator("+", nullCheck, resultType, operand)
 
+      case UNARY_PLUS if isTimeInterval(resultType) =>
+        val operand = operands.head
+        requireTimeInterval(operand)
+        generateUnaryIntervalPlusMinus(plus = true, nullCheck, operand)
+
       // comparison
       case EQUALS =>
         val left = operands.head
@@ -760,7 +792,7 @@ class CodeGenerator(
         generateIfElse(nullCheck, operands, resultType)
 
       // casting
-      case CAST =>
+      case CAST | REINTERPRET =>
         val operand = operands.head
         generateCast(nullCheck, operand, resultType)
 
@@ -946,8 +978,8 @@ class CodeGenerator(
     val defaultValue = primitiveDefaultValue(literalType)
 
     // explicit unboxing
-    val unboxedLiteralCode = if (isTemporal(literalType)) {
-      temporalToInternalCode(literalType, literalCode)
+    val unboxedLiteralCode = if (isTimePoint(literalType)) {
+      timePointToInternalCode(literalType, literalCode)
     } else {
       literalCode
     }
@@ -1023,7 +1055,7 @@ class CodeGenerator(
       case SqlTimeTypeInfo.DATE | SqlTimeTypeInfo.TIME | SqlTimeTypeInfo.TIMESTAMP =>
         val resultTerm = newName("result")
         val resultTypeTerm = boxedTypeTermForTypeInfo(expr.resultType)
-        val convMethod = internalToTemporalCode(expr.resultType, expr.resultTerm)
+        val convMethod = internalToTimePointCode(expr.resultType, expr.resultTerm)
 
         val resultCode = if (nullCheck) {
           s"""

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarOperators.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarOperators.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarOperators.scala
index 28c66b6..81187a5 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarOperators.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarOperators.scala
@@ -17,12 +17,14 @@
  */
 package org.apache.flink.api.table.codegen.calls
 
-import org.apache.calcite.avatica.util.DateTimeUtils
+import org.apache.calcite.avatica.util.DateTimeUtils.MILLIS_PER_DAY
+import org.apache.calcite.avatica.util.{DateTimeUtils, TimeUnitRange}
 import org.apache.calcite.util.BuiltInMethod
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
 import org.apache.flink.api.common.typeinfo.{NumericTypeInfo, SqlTimeTypeInfo, TypeInformation}
 import org.apache.flink.api.table.codegen.CodeGenUtils._
 import org.apache.flink.api.table.codegen.{CodeGenException, GeneratedExpression}
+import org.apache.flink.api.table.typeutils.IntervalTypeInfo
 import org.apache.flink.api.table.typeutils.TypeCheckUtils._
 
 object ScalarOperators {
@@ -70,10 +72,8 @@ object ScalarOperators {
           s"$operandTerm.negate()"
         } else if (isDecimal(operand.resultType) && operator == "+") {
           s"$operandTerm"
-        } else if (isNumeric(operand.resultType)) {
+        } else {
           s"$operator($operandTerm)"
-        }  else {
-          throw new CodeGenException("Unsupported unary operator.")
         }
     }
   }
@@ -396,7 +396,23 @@ object ScalarOperators {
     // Date/Time/Timestamp -> String
     case (dtt: SqlTimeTypeInfo[_], STRING_TYPE_INFO) =>
       generateUnaryOperatorIfNotNull(nullCheck, targetType, operand) {
-        (operandTerm) => s"""${internalToTemporalCode(dtt, operandTerm)}.toString()"""
+        (operandTerm) => s"""${internalToTimePointCode(dtt, operandTerm)}.toString()"""
+      }
+
+    // Interval Months -> String
+    case (IntervalTypeInfo.INTERVAL_MONTHS, STRING_TYPE_INFO) =>
+      val method = qualifyMethod(BuiltInMethod.INTERVAL_YEAR_MONTH_TO_STRING.method)
+      val timeUnitRange = qualifyEnum(TimeUnitRange.YEAR_TO_MONTH)
+      generateUnaryOperatorIfNotNull(nullCheck, targetType, operand) {
+        (operandTerm) => s"$method($operandTerm, $timeUnitRange)"
+      }
+
+    // Interval Millis -> String
+    case (IntervalTypeInfo.INTERVAL_MILLIS, STRING_TYPE_INFO) =>
+      val method = qualifyMethod(BuiltInMethod.INTERVAL_DAY_TIME_TO_STRING.method)
+      val timeUnitRange = qualifyEnum(TimeUnitRange.DAY_TO_SECOND)
+      generateUnaryOperatorIfNotNull(nullCheck, targetType, operand) {
+        (operandTerm) => s"$method($operandTerm, $timeUnitRange, 3)" // milli second precision
       }
 
     // * (not Date/Time/Timestamp) -> String
@@ -508,25 +524,28 @@ object ScalarOperators {
             s"${classOf[DateTimeUtils].getCanonicalName}.MILLIS_PER_DAY)"
       }
 
-    // Date -> Integer, Time -> Integer
-    case (SqlTimeTypeInfo.DATE, INT_TYPE_INFO) | (SqlTimeTypeInfo.TIME, INT_TYPE_INFO) =>
-      internalExprCasting(operand, INT_TYPE_INFO)
-
+    // internal temporal casting
+    // Date -> Integer
+    // Time -> Integer
     // Timestamp -> Long
-    case (SqlTimeTypeInfo.TIMESTAMP, LONG_TYPE_INFO) =>
-      internalExprCasting(operand, LONG_TYPE_INFO)
-
     // Integer -> Date
-    case (INT_TYPE_INFO, SqlTimeTypeInfo.DATE) =>
-      internalExprCasting(operand, SqlTimeTypeInfo.DATE)
-
     // Integer -> Time
-    case (INT_TYPE_INFO, SqlTimeTypeInfo.TIME) =>
-      internalExprCasting(operand, SqlTimeTypeInfo.TIME)
-
     // Long -> Timestamp
-    case (LONG_TYPE_INFO, SqlTimeTypeInfo.TIMESTAMP) =>
-      internalExprCasting(operand, SqlTimeTypeInfo.TIMESTAMP)
+    // Integer -> Interval Months
+    // Long -> Interval Millis
+    // Interval Months -> Integer
+    // Interval Millis -> Long
+    case (SqlTimeTypeInfo.DATE, INT_TYPE_INFO) |
+         (SqlTimeTypeInfo.TIME, INT_TYPE_INFO) |
+         (SqlTimeTypeInfo.TIMESTAMP, LONG_TYPE_INFO) |
+         (INT_TYPE_INFO, SqlTimeTypeInfo.DATE) |
+         (INT_TYPE_INFO, SqlTimeTypeInfo.TIME) |
+         (LONG_TYPE_INFO, SqlTimeTypeInfo.TIMESTAMP) |
+         (INT_TYPE_INFO, IntervalTypeInfo.INTERVAL_MONTHS) |
+         (LONG_TYPE_INFO, IntervalTypeInfo.INTERVAL_MILLIS) |
+         (IntervalTypeInfo.INTERVAL_MONTHS, INT_TYPE_INFO) |
+         (IntervalTypeInfo.INTERVAL_MONTHS, LONG_TYPE_INFO) =>
+      internalExprCasting(operand, targetType)
 
     case (from, to) =>
       throw new CodeGenException(s"Unsupported cast from '$from' to '$to'.")
@@ -591,6 +610,60 @@ object ScalarOperators {
     }
   }
 
+  def generateTemporalPlusMinus(
+      plus: Boolean,
+      nullCheck: Boolean,
+      left: GeneratedExpression,
+      right: GeneratedExpression)
+    : GeneratedExpression = {
+
+    val operator = if (plus) "+" else "-"
+
+    (left.resultType, right.resultType) match {
+      case (l: IntervalTypeInfo[_], r: IntervalTypeInfo[_]) if l == r =>
+        generateArithmeticOperator(operator, nullCheck, l, left, right)
+
+      case (SqlTimeTypeInfo.DATE, IntervalTypeInfo.INTERVAL_MILLIS) |
+           (IntervalTypeInfo.INTERVAL_MILLIS, SqlTimeTypeInfo.DATE) =>
+        generateOperatorIfNotNull(nullCheck, SqlTimeTypeInfo.DATE, left, right) {
+          if (isTimePoint(left.resultType)) {
+            (leftTerm, rightTerm) => s"$leftTerm + ((int) ($rightTerm / ${MILLIS_PER_DAY}L))"
+          } else {
+            (leftTerm, rightTerm) => s"((int) ($leftTerm / ${MILLIS_PER_DAY}L)) + $rightTerm"
+          }
+        }
+
+      case (SqlTimeTypeInfo.TIME, IntervalTypeInfo.INTERVAL_MILLIS) |
+           (IntervalTypeInfo.INTERVAL_MILLIS, SqlTimeTypeInfo.TIME) =>
+        generateOperatorIfNotNull(nullCheck, SqlTimeTypeInfo.TIME, left, right) {
+          if (isTimePoint(left.resultType)) {
+            (leftTerm, rightTerm) => s"$leftTerm + ((int) ($rightTerm))"
+          } else {
+            (leftTerm, rightTerm) => s"((int) ($leftTerm)) + $rightTerm"
+          }
+        }
+
+      case (SqlTimeTypeInfo.TIMESTAMP, IntervalTypeInfo.INTERVAL_MILLIS) =>
+        generateOperatorIfNotNull(nullCheck, SqlTimeTypeInfo.TIMESTAMP, left, right) {
+          (leftTerm, rightTerm) => s"$leftTerm + $rightTerm"
+        }
+
+      // TODO more operations when CALCITE-308 is fixed
+
+      case _ =>
+        throw new CodeGenException("Unsupported temporal arithmetic.")
+    }
+  }
+
+  def generateUnaryIntervalPlusMinus(
+      plus: Boolean,
+      nullCheck: Boolean,
+      operand: GeneratedExpression)
+    : GeneratedExpression = {
+    val operator = if (plus) "+" else "-"
+    generateUnaryArithmeticOperator(operator, nullCheck, operand.resultType, operand)
+  }
+
   // ----------------------------------------------------------------------------------------------
 
   private def generateUnaryOperatorIfNotNull(

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionParser.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionParser.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionParser.scala
index 6aa4f89..41222c8 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionParser.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionParser.scala
@@ -17,8 +17,11 @@
  */
 package org.apache.flink.api.table.expressions
 
-import org.apache.flink.api.common.typeinfo.{SqlTimeTypeInfo, TypeInformation, BasicTypeInfo}
+import org.apache.calcite.avatica.util.DateTimeUtils.{MILLIS_PER_DAY, MILLIS_PER_HOUR, MILLIS_PER_MINUTE, MILLIS_PER_SECOND}
+import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, SqlTimeTypeInfo, TypeInformation}
 import org.apache.flink.api.table.ExpressionParserException
+import org.apache.flink.api.table.expressions.ExpressionUtils.{toMilliInterval, toMonthInterval}
+import org.apache.flink.api.table.typeutils.IntervalTypeInfo
 
 import scala.util.parsing.combinator.{JavaTokenParsers, PackratParsers}
 
@@ -57,6 +60,13 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
   lazy val TO_DATE: Keyword = Keyword("toDate")
   lazy val TO_TIME: Keyword = Keyword("toTime")
   lazy val TO_TIMESTAMP: Keyword = Keyword("toTimestamp")
+  lazy val YEAR: Keyword = Keyword("year")
+  lazy val MONTH: Keyword = Keyword("month")
+  lazy val DAY: Keyword = Keyword("day")
+  lazy val HOUR: Keyword = Keyword("hour")
+  lazy val MINUTE: Keyword = Keyword("minute")
+  lazy val SECOND: Keyword = Keyword("second")
+  lazy val MILLI: Keyword = Keyword("milli")
 
   def functionIdent: ExpressionParser.Parser[String] =
     not(AS) ~ not(COUNT) ~ not(AVG) ~ not(MIN) ~ not(MAX) ~
@@ -68,6 +78,12 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
   lazy val dataType: PackratParser[TypeInformation[_]] =
     "BYTE" ^^ { ti => BasicTypeInfo.BYTE_TYPE_INFO } |
       "SHORT" ^^ { ti => BasicTypeInfo.SHORT_TYPE_INFO } |
+      "INTERVAL_MONTHS" ^^ {
+        ti => IntervalTypeInfo.INTERVAL_MONTHS.asInstanceOf[TypeInformation[_]]
+      } |
+      "INTERVAL_MILLIS" ^^ {
+        ti => IntervalTypeInfo.INTERVAL_MILLIS.asInstanceOf[TypeInformation[_]]
+      } |
       "INT" ^^ { ti => BasicTypeInfo.INT_TYPE_INFO } |
       "LONG" ^^ { ti => BasicTypeInfo.LONG_TYPE_INFO } |
       "FLOAT" ^^ { ti => BasicTypeInfo.FLOAT_TYPE_INFO } |
@@ -81,10 +97,14 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
 
   // Literals
 
+  // same as floatingPointNumber but we do not allow trailing dot "12.d" or "2."
+  lazy val floatingPointNumberFlink: Parser[String] =
+    """-?(\d+(\.\d+)?|\d*\.\d+)([eE][+-]?\d+)?[fFdD]?""".r
+
   lazy val numberLiteral: PackratParser[Expression] =
     (wholeNumber <~ ("l" | "L")) ^^ { n => Literal(n.toLong) } |
       (decimalNumber <~ ("p" | "P")) ^^ { n => Literal(BigDecimal(n)) } |
-      (floatingPointNumber | decimalNumber) ^^ {
+      (floatingPointNumberFlink | decimalNumber) ^^ {
         n =>
           if (n.matches("""-?\d+""")) {
             Literal(n.toInt)
@@ -109,7 +129,7 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
   }
 
   lazy val nullLiteral: PackratParser[Expression] = NULL ~ "(" ~> dataType <~ ")" ^^ {
-    case dt => Null(dt)
+    dt => Null(dt)
   }
 
   lazy val literalExpr: PackratParser[Expression] =
@@ -169,7 +189,7 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
   }
 
   lazy val suffixTrimWithoutArgs = composite <~ ".trim" ~ opt("()") ^^ {
-    case e =>
+    e =>
       Trim(TrimConstants.TRIM_BOTH, TrimConstants.TRIM_DEFAULT_CHAR, e)
   }
 
@@ -198,10 +218,29 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
   lazy val suffixToTime: PackratParser[Expression] =
     composite <~ "." ~ TO_TIME ~ opt("()") ^^ { e => Cast(e, SqlTimeTypeInfo.TIME) }
 
+  lazy val suffixTimeInterval : PackratParser[Expression] =
+    composite ~ "." ~ (YEAR | MONTH | DAY | HOUR | MINUTE | SECOND | MILLI) ^^ {
+
+    case expr ~ _ ~ YEAR.key => toMonthInterval(expr, 12)
+
+    case expr ~ _ ~ MONTH.key => toMonthInterval(expr, 1)
+
+    case expr ~ _ ~ DAY.key => toMilliInterval(expr, MILLIS_PER_DAY)
+
+    case expr ~ _ ~ HOUR.key => toMilliInterval(expr, MILLIS_PER_HOUR)
+
+    case expr ~ _ ~ MINUTE.key => toMilliInterval(expr, MILLIS_PER_MINUTE)
+
+    case expr ~ _ ~ SECOND.key => toMilliInterval(expr, MILLIS_PER_SECOND)
+
+    case expr ~ _ ~ MILLI.key => toMilliInterval(expr, 1)
+  }
+
   lazy val suffixed: PackratParser[Expression] =
-    suffixIsNull | suffixIsNotNull | suffixSum | suffixMin | suffixMax | suffixCount | suffixAvg |
-      suffixCast | suffixAs | suffixTrim | suffixTrimWithoutArgs | suffixIf | suffixFunctionCall |
-        suffixAsc | suffixDesc | suffixToDate | suffixToTimestamp | suffixToTime
+    suffixTimeInterval | suffixIsNull | suffixIsNotNull | suffixSum | suffixMin | suffixMax |
+      suffixCount | suffixAvg | suffixCast | suffixAs | suffixTrim | suffixTrimWithoutArgs |
+      suffixIf | suffixAsc | suffixDesc | suffixToDate | suffixToTimestamp | suffixToTime |
+      suffixFunctionCall // function call must always be at the end
 
   // prefix operators
 
@@ -263,7 +302,8 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
 
   lazy val prefixed: PackratParser[Expression] =
     prefixIsNull | prefixIsNotNull | prefixSum | prefixMin | prefixMax | prefixCount | prefixAvg |
-      prefixCast | prefixAs | prefixTrim | prefixTrimWithoutArgs | prefixIf | prefixFunctionCall
+      prefixCast | prefixAs | prefixTrim | prefixTrimWithoutArgs | prefixIf |
+      prefixFunctionCall // function call must always be at the end
 
   // suffix/prefix composite
 
@@ -339,10 +379,10 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
     parseAll(expressionList, expression) match {
       case Success(lst, _) => lst
 
-      case Failure(msg, _) => throw new ExpressionParserException(
+      case Failure(msg, _) => throw ExpressionParserException(
         "Could not parse expression: " + msg)
 
-      case Error(msg, _) => throw new ExpressionParserException(
+      case Error(msg, _) => throw ExpressionParserException(
         "Could not parse expression: " + msg)
     }
   }
@@ -352,7 +392,7 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
       case Success(lst, _) => lst
 
       case fail =>
-        throw new ExpressionParserException("Could not parse expression: " + fail.toString)
+        throw ExpressionParserException("Could not parse expression: " + fail.toString)
     }
   }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionUtils.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionUtils.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionUtils.scala
new file mode 100644
index 0000000..89671de
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionUtils.scala
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 org.apache.flink.api.table.expressions
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo
+import org.apache.flink.api.table.typeutils.IntervalTypeInfo
+
+object ExpressionUtils {
+
+  private[flink] def toMonthInterval(expr: Expression, multiplier: Int): Expression = expr match {
+    case Literal(value: Int, BasicTypeInfo.INT_TYPE_INFO) =>
+      Literal(value * multiplier, IntervalTypeInfo.INTERVAL_MONTHS)
+    case _ =>
+      Cast(Mul(expr, Literal(multiplier)), IntervalTypeInfo.INTERVAL_MONTHS)
+  }
+
+  private[flink] def toMilliInterval(expr: Expression, multiplier: Long): Expression = expr match {
+    case Literal(value: Int, BasicTypeInfo.INT_TYPE_INFO) =>
+      Literal(value * multiplier, IntervalTypeInfo.INTERVAL_MILLIS)
+    case _ =>
+      Cast(Mul(expr, Literal(multiplier)), IntervalTypeInfo.INTERVAL_MILLIS)
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/arithmetic.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/arithmetic.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/arithmetic.scala
index e4ed08e..b301f22 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/arithmetic.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/arithmetic.scala
@@ -22,8 +22,8 @@ import org.apache.calcite.sql.SqlOperator
 import org.apache.calcite.sql.fun.SqlStdOperatorTable
 import org.apache.calcite.tools.RelBuilder
 import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
-import org.apache.flink.api.table.typeutils.TypeCheckUtils.{isNumeric, isString}
-import org.apache.flink.api.table.typeutils.{TypeCheckUtils, TypeCoercion}
+import org.apache.flink.api.table.typeutils.TypeCheckUtils._
+import org.apache.flink.api.table.typeutils.TypeCoercion
 import org.apache.flink.api.table.validate._
 
 import scala.collection.JavaConversions._
@@ -61,10 +61,14 @@ case class Plus(left: Expression, right: Expression) extends BinaryArithmetic {
   override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     if(isString(left.resultType)) {
       val castedRight = Cast(right, BasicTypeInfo.STRING_TYPE_INFO)
-      relBuilder.call(SqlStdOperatorTable.PLUS, left.toRexNode, castedRight.toRexNode)
+      relBuilder.call(SqlStdOperatorTable.CONCAT, left.toRexNode, castedRight.toRexNode)
     } else if(isString(right.resultType)) {
       val castedLeft = Cast(left, BasicTypeInfo.STRING_TYPE_INFO)
-      relBuilder.call(SqlStdOperatorTable.PLUS, castedLeft.toRexNode, right.toRexNode)
+      relBuilder.call(SqlStdOperatorTable.CONCAT, castedLeft.toRexNode, right.toRexNode)
+    } else if (isTimeInterval(left.resultType) && left.resultType == right.resultType) {
+      relBuilder.call(SqlStdOperatorTable.PLUS, left.toRexNode, right.toRexNode)
+    } else if (isTemporal(left.resultType) && isTemporal(right.resultType)) {
+      relBuilder.call(SqlStdOperatorTable.DATETIME_PLUS, left.toRexNode, right.toRexNode)
     } else {
       val castedLeft = Cast(left, resultType)
       val castedRight = Cast(right, resultType)
@@ -72,15 +76,22 @@ case class Plus(left: Expression, right: Expression) extends BinaryArithmetic {
     }
   }
 
-  // TODO: tighten this rule once we implemented type coercion rules during validation
   override private[flink] def validateInput(): ExprValidationResult = {
     if (isString(left.resultType) || isString(right.resultType)) {
       ValidationSuccess
-    } else if (!isNumeric(left.resultType) || !isNumeric(right.resultType)) {
-      ValidationFailure(s"$this requires Numeric or String input," +
-        s" get $left : ${left.resultType} and $right : ${right.resultType}")
-    } else {
+    } else if (isTimeInterval(left.resultType) && left.resultType == right.resultType) {
+      ValidationSuccess
+    } else if (isTimePoint(left.resultType) && isTimeInterval(right.resultType)) {
+      ValidationSuccess
+    } else if (isTimeInterval(left.resultType) && isTimePoint(right.resultType)) {
+      ValidationSuccess
+    } else if (isNumeric(left.resultType) && isNumeric(right.resultType)) {
       ValidationSuccess
+    } else {
+      ValidationFailure(
+        s"$this requires Numeric, String, Intervals of same type, " +
+        s"or Interval and a time point input, " +
+        s"get $left : ${left.resultType} and $right : ${right.resultType}")
     }
   }
 }
@@ -94,14 +105,29 @@ case class UnaryMinus(child: Expression) extends UnaryExpression {
 
   override private[flink] def resultType = child.resultType
 
-  override private[flink] def validateInput(): ExprValidationResult =
-    TypeCheckUtils.assertNumericExpr(child.resultType, "unary minus")
+  override private[flink] def validateInput(): ExprValidationResult = {
+    if (isNumeric(child.resultType)) {
+      ValidationSuccess
+    } else if (isTimeInterval(child.resultType)) {
+      ValidationSuccess
+    } else {
+      ValidationFailure(s"$this requires Numeric, or Interval input, get ${child.resultType}")
+    }
+  }
 }
 
 case class Minus(left: Expression, right: Expression) extends BinaryArithmetic {
   override def toString = s"($left - $right)"
 
   private[flink] val sqlOperator = SqlStdOperatorTable.MINUS
+
+  override private[flink] def validateInput(): ExprValidationResult = {
+    if (isTimeInterval(left.resultType) && left.resultType == right.resultType) {
+      ValidationSuccess
+    } else {
+      super.validateInput()
+    }
+  }
 }
 
 case class Div(left: Expression, right: Expression) extends BinaryArithmetic {

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/cast.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/cast.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/cast.scala
index 525d010..f65dd5b 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/cast.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/cast.scala
@@ -32,7 +32,8 @@ case class Cast(child: Expression, resultType: TypeInformation[_]) extends Unary
     val typeFactory = relBuilder.getTypeFactory.asInstanceOf[FlinkTypeFactory]
     relBuilder
       .getRexBuilder
-      .makeCast(
+      // we use abstract cast here because RelBuilder.cast() has to many side effects
+      .makeAbstractCast(
         typeFactory.createTypeFromTypeInfo(resultType),
         child.toRexNode)
   }

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/literals.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/literals.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/literals.scala
index cd3de60..677160a 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/literals.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/literals.scala
@@ -20,11 +20,15 @@ package org.apache.flink.api.table.expressions
 import java.sql.{Date, Time, Timestamp}
 import java.util.{Calendar, TimeZone}
 
+import org.apache.calcite.avatica.util.TimeUnit
 import org.apache.calcite.rex.RexNode
+import org.apache.calcite.sql.SqlIntervalQualifier
 import org.apache.calcite.sql.`type`.SqlTypeName
+import org.apache.calcite.sql.parser.SqlParserPos
 import org.apache.calcite.tools.RelBuilder
 import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, SqlTimeTypeInfo, TypeInformation}
 import org.apache.flink.api.table.FlinkTypeFactory
+import org.apache.flink.api.table.typeutils.IntervalTypeInfo
 
 object Literal {
   private[flink] def apply(l: Any): Literal = l match {
@@ -63,6 +67,22 @@ case class Literal(value: Any, resultType: TypeInformation[_]) extends LeafExpre
       case SqlTimeTypeInfo.TIMESTAMP =>
         relBuilder.getRexBuilder.makeTimestampLiteral(dateToCalendar, 3)
 
+      case IntervalTypeInfo.INTERVAL_MONTHS =>
+        val interval = java.math.BigDecimal.valueOf(value.asInstanceOf[Int])
+        val intervalQualifier = new SqlIntervalQualifier(
+          TimeUnit.YEAR,
+          TimeUnit.MONTH,
+          SqlParserPos.ZERO)
+        relBuilder.getRexBuilder.makeIntervalLiteral(interval, intervalQualifier)
+
+      case IntervalTypeInfo.INTERVAL_MILLIS =>
+        val interval = java.math.BigDecimal.valueOf(value.asInstanceOf[Long])
+        val intervalQualifier = new SqlIntervalQualifier(
+          TimeUnit.DAY,
+          TimeUnit.SECOND,
+          SqlParserPos.ZERO)
+        relBuilder.getRexBuilder.makeIntervalLiteral(interval, intervalQualifier)
+
       case _ => relBuilder.literal(value)
     }
   }

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/IntervalTypeInfo.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/IntervalTypeInfo.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/IntervalTypeInfo.scala
new file mode 100644
index 0000000..85524fb
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/IntervalTypeInfo.scala
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 org.apache.flink.api.table.typeutils
+
+import java.util.Objects
+
+import org.apache.flink.api.common.ExecutionConfig
+import org.apache.flink.api.common.typeinfo.{AtomicType, TypeInformation}
+import org.apache.flink.api.common.typeutils.base.{IntComparator, IntSerializer, LongComparator, LongSerializer}
+import org.apache.flink.api.common.typeutils.{TypeComparator, TypeSerializer}
+import org.apache.flink.api.table.typeutils.IntervalTypeInfo.instantiateComparator
+import org.apache.flink.util.Preconditions._
+
+/**
+  * TypeInformation for SQL INTERVAL types.
+  */
+@SerialVersionUID(-1816179424364825258L)
+class IntervalTypeInfo[T](
+    val clazz: Class[T],
+    val serializer: TypeSerializer[T],
+    val comparatorClass: Class[_ <: TypeComparator[T]])
+  extends TypeInformation[T]
+  with AtomicType[T] {
+
+  checkNotNull(clazz)
+  checkNotNull(serializer)
+  checkNotNull(comparatorClass)
+
+  override def isBasicType: Boolean = false
+
+  override def isTupleType: Boolean = false
+
+  override def getArity: Int = 1
+
+  override def getTotalFields: Int = 1
+
+  override def getTypeClass: Class[T] = clazz
+
+  override def isKeyType: Boolean = true
+
+  override def createSerializer(config: ExecutionConfig): TypeSerializer[T] = serializer
+
+  override def createComparator(
+      sortOrderAscending: Boolean,
+      executionConfig: ExecutionConfig)
+    : TypeComparator[T] = instantiateComparator(comparatorClass, sortOrderAscending)
+
+  // ----------------------------------------------------------------------------------------------
+
+  override def hashCode: Int = Objects.hash(clazz, serializer, comparatorClass)
+
+  def canEqual(obj: Any): Boolean = obj.isInstanceOf[IntervalTypeInfo[_]]
+
+  override def equals(obj: Any): Boolean = {
+    obj match {
+      case other: IntervalTypeInfo[_] =>
+        other.canEqual(this) &&
+          (this.clazz eq other.clazz) &&
+          serializer == other.serializer &&
+          (this.comparatorClass eq other.comparatorClass)
+      case _ =>
+        false
+    }
+  }
+
+  override def toString: String = s"IntervalTypeInfo(${clazz.getSimpleName})"
+}
+
+object IntervalTypeInfo {
+
+  val INTERVAL_MONTHS =
+    new IntervalTypeInfo(classOf[java.lang.Integer], IntSerializer.INSTANCE, classOf[IntComparator])
+
+  val INTERVAL_MILLIS =
+    new IntervalTypeInfo(classOf[java.lang.Long], LongSerializer.INSTANCE, classOf[LongComparator])
+
+  // ----------------------------------------------------------------------------------------------
+
+  private def instantiateComparator[X](
+      comparatorClass: Class[_ <: TypeComparator[X]],
+      ascendingOrder: java.lang.Boolean)
+    : TypeComparator[X] = {
+    try {
+      val constructor = comparatorClass.getConstructor(classOf[java.lang.Boolean])
+      constructor.newInstance(ascendingOrder)
+    } catch {
+      case e: Exception =>
+        throw new RuntimeException(
+          s"Could not initialize comparator ${comparatorClass.getName}", e)
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCheckUtils.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCheckUtils.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCheckUtils.scala
index c19deec..4ff7888 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCheckUtils.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCheckUtils.scala
@@ -30,6 +30,7 @@ object TypeCheckUtils {
   def isAdvanced(dataType: TypeInformation[_]): Boolean = dataType match {
     case _: BasicTypeInfo[_] => false
     case _: SqlTimeTypeInfo[_] => false
+    case _: IntervalTypeInfo[_] => false
     case _ => true
   }
 
@@ -45,7 +46,14 @@ object TypeCheckUtils {
     case _ => false
   }
 
-  def isTemporal(dataType: TypeInformation[_]): Boolean = dataType.isInstanceOf[SqlTimeTypeInfo[_]]
+  def isTemporal(dataType: TypeInformation[_]): Boolean =
+    isTimePoint(dataType) || isTimeInterval(dataType)
+
+  def isTimePoint(dataType: TypeInformation[_]): Boolean =
+    dataType.isInstanceOf[SqlTimeTypeInfo[_]]
+
+  def isTimeInterval(dataType: TypeInformation[_]): Boolean =
+    dataType.isInstanceOf[IntervalTypeInfo[_]]
 
   def isString(dataType: TypeInformation[_]): Boolean = dataType == STRING_TYPE_INFO
 

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCoercion.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCoercion.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCoercion.scala
index dad13fb..bb3d060 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCoercion.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCoercion.scala
@@ -45,6 +45,9 @@ object TypeCoercion {
       case (_, BIG_DEC_TYPE_INFO) => Some(BIG_DEC_TYPE_INFO)
       case (BIG_DEC_TYPE_INFO, _) => Some(BIG_DEC_TYPE_INFO)
 
+      case (stti: SqlTimeTypeInfo[_], _: IntervalTypeInfo[_]) => Some(stti)
+      case (_: IntervalTypeInfo[_], stti: SqlTimeTypeInfo[_]) => Some(stti)
+
       case tuple if tuple.productIterator.forall(numericWideningPrecedence.contains) =>
         val higherIndex = numericWideningPrecedence.lastIndexWhere(t => t == tp1 || t == tp2)
         Some(numericWideningPrecedence(higherIndex))
@@ -100,6 +103,8 @@ object TypeCoercion {
     case (INT_TYPE_INFO, SqlTimeTypeInfo.DATE) => true
     case (INT_TYPE_INFO, SqlTimeTypeInfo.TIME) => true
     case (LONG_TYPE_INFO, SqlTimeTypeInfo.TIMESTAMP) => true
+    case (INT_TYPE_INFO, IntervalTypeInfo.INTERVAL_MONTHS) => true
+    case (LONG_TYPE_INFO, IntervalTypeInfo.INTERVAL_MILLIS) => true
 
     case (SqlTimeTypeInfo.DATE, SqlTimeTypeInfo.TIME) => false
     case (SqlTimeTypeInfo.TIME, SqlTimeTypeInfo.DATE) => false
@@ -108,6 +113,9 @@ object TypeCoercion {
     case (SqlTimeTypeInfo.TIME, INT_TYPE_INFO) => true
     case (SqlTimeTypeInfo.TIMESTAMP, LONG_TYPE_INFO) => true
 
+    case (IntervalTypeInfo.INTERVAL_MONTHS, INT_TYPE_INFO) => true
+    case (IntervalTypeInfo.INTERVAL_MILLIS, LONG_TYPE_INFO) => true
+
     case _ => false
   }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/ExpressionsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/ExpressionsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/ExpressionsITCase.java
index 7d225f3..b060540 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/ExpressionsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/ExpressionsITCase.java
@@ -192,11 +192,12 @@ public class ExpressionsITCase extends TableProgramsTestBase {
 					"a.cast(STRING) + a.cast(STRING)," +
 					"CAST(ISNULL(b), INT)," +
 					"ISNULL(CAST(b, INT).abs()) === false," +
-					"((((true) === true) || false).cast(STRING) + 'X ').trim");
+					"((((true) === true) || false).cast(STRING) + 'X ').trim," +
+					"12.isNull");
 
 		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		List<Row> results = ds.collect();
-		String expected = "false,10,55,0,true,trueX";
+		String expected = "false,10,55,0,true,trueX,false";
 		compareResultAsText(results, expected);
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/expression/TemporalTypesTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/expression/TemporalTypesTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/expression/TemporalTypesTest.scala
new file mode 100644
index 0000000..63d6346
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/expression/TemporalTypesTest.scala
@@ -0,0 +1,389 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 org.apache.flink.api.scala.expression
+
+import java.sql.{Date, Time, Timestamp}
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.scala.table._
+import org.apache.flink.api.table.expressions.utils.ExpressionTestBase
+import org.apache.flink.api.table.typeutils.RowTypeInfo
+import org.apache.flink.api.table.{Row, Types}
+import org.junit.Test
+
+class TemporalTypesTest extends ExpressionTestBase {
+
+  @Test
+  def testTimePointLiterals(): Unit = {
+    testAllApis(
+      "1990-10-14".toDate,
+      "'1990-10-14'.toDate",
+      "DATE '1990-10-14'",
+      "1990-10-14")
+
+    testTableApi(
+      Date.valueOf("2040-09-11"),
+      "'2040-09-11'.toDate",
+      "2040-09-11")
+
+    testAllApis(
+      "1500-04-30".cast(Types.DATE),
+      "'1500-04-30'.cast(DATE)",
+      "CAST('1500-04-30' AS DATE)",
+      "1500-04-30")
+
+    testAllApis(
+      "15:45:59".toTime,
+      "'15:45:59'.toTime",
+      "TIME '15:45:59'",
+      "15:45:59")
+
+    testTableApi(
+      Time.valueOf("00:00:00"),
+      "'00:00:00'.toTime",
+      "00:00:00")
+
+    testAllApis(
+      "1:30:00".cast(Types.TIME),
+      "'1:30:00'.cast(TIME)",
+      "CAST('1:30:00' AS TIME)",
+      "01:30:00")
+
+    testAllApis(
+      "1990-10-14 23:00:00.123".toTimestamp,
+      "'1990-10-14 23:00:00.123'.toTimestamp",
+      "TIMESTAMP '1990-10-14 23:00:00.123'",
+      "1990-10-14 23:00:00.123")
+
+    testTableApi(
+      Timestamp.valueOf("2040-09-11 00:00:00.000"),
+      "'2040-09-11 00:00:00.000'.toTimestamp",
+      "2040-09-11 00:00:00.0")
+
+    testAllApis(
+      "1500-04-30 12:00:00".cast(Types.TIMESTAMP),
+      "'1500-04-30 12:00:00'.cast(TIMESTAMP)",
+      "CAST('1500-04-30 12:00:00' AS TIMESTAMP)",
+      "1500-04-30 12:00:00.0")
+  }
+
+  @Test
+  def testTimeIntervalLiterals(): Unit = {
+    testAllApis(
+      1.year,
+      "1.year",
+      "INTERVAL '1' YEAR",
+      "+1-00")
+
+    testAllApis(
+      1.month,
+      "1.month",
+      "INTERVAL '1' MONTH",
+      "+0-01")
+
+    testAllApis(
+      12.day,
+      "12.day",
+      "INTERVAL '12' DAY",
+      "+12 00:00:00.000")
+
+    testAllApis(
+      1.hour,
+      "1.hour",
+      "INTERVAL '1' HOUR",
+      "+0 01:00:00.000")
+
+    testAllApis(
+      3.minute,
+      "3.minute",
+      "INTERVAL '3' MINUTE",
+      "+0 00:03:00.000")
+
+    testAllApis(
+      3.second,
+      "3.second",
+      "INTERVAL '3' SECOND",
+      "+0 00:00:03.000")
+
+    testAllApis(
+      3.milli,
+      "3.milli",
+      "INTERVAL '0.003' SECOND",
+      "+0 00:00:00.003")
+  }
+
+  @Test
+  def testTimePointInput(): Unit = {
+    testAllApis(
+      'f0,
+      "f0",
+      "f0",
+      "1990-10-14")
+
+    testAllApis(
+      'f1,
+      "f1",
+      "f1",
+      "10:20:45")
+
+    testAllApis(
+      'f2,
+      "f2",
+      "f2",
+      "1990-10-14 10:20:45.123")
+  }
+
+  @Test
+  def testTimeIntervalInput(): Unit = {
+    testAllApis(
+      'f9,
+      "f9",
+      "f9",
+      "+2-00")
+
+    testAllApis(
+      'f10,
+      "f10",
+      "f10",
+      "+0 00:00:12.000")
+  }
+
+  @Test
+  def testTimePointCasting(): Unit = {
+    testAllApis(
+      'f0.cast(Types.TIMESTAMP),
+      "f0.cast(TIMESTAMP)",
+      "CAST(f0 AS TIMESTAMP)",
+      "1990-10-14 00:00:00.0")
+
+    testAllApis(
+      'f1.cast(Types.TIMESTAMP),
+      "f1.cast(TIMESTAMP)",
+      "CAST(f1 AS TIMESTAMP)",
+      "1970-01-01 10:20:45.0")
+
+    testAllApis(
+      'f2.cast(Types.DATE),
+      "f2.cast(DATE)",
+      "CAST(f2 AS DATE)",
+      "1990-10-14")
+
+    testAllApis(
+      'f2.cast(Types.TIME),
+      "f2.cast(TIME)",
+      "CAST(f2 AS TIME)",
+      "10:20:45")
+
+    testAllApis(
+      'f2.cast(Types.TIME),
+      "f2.cast(TIME)",
+      "CAST(f2 AS TIME)",
+      "10:20:45")
+
+    testTableApi(
+      'f7.cast(Types.DATE),
+      "f7.cast(DATE)",
+      "2002-11-09")
+
+    testTableApi(
+      'f7.cast(Types.DATE).cast(Types.INT),
+      "f7.cast(DATE).cast(INT)",
+      "12000")
+
+    testTableApi(
+      'f7.cast(Types.TIME),
+      "f7.cast(TIME)",
+      "00:00:12")
+
+    testTableApi(
+      'f7.cast(Types.TIME).cast(Types.INT),
+      "f7.cast(TIME).cast(INT)",
+      "12000")
+
+    testTableApi(
+      'f8.cast(Types.TIMESTAMP),
+      "f8.cast(TIMESTAMP)",
+      "2016-06-27 07:23:33.0")
+
+    testTableApi(
+      'f8.cast(Types.TIMESTAMP).cast(Types.LONG),
+      "f8.cast(TIMESTAMP).cast(LONG)",
+      "1467012213000")
+  }
+
+  @Test
+  def testTimeIntervalCasting(): Unit = {
+    testTableApi(
+      'f7.cast(Types.INTERVAL_MONTHS),
+      "f7.cast(INTERVAL_MONTHS)",
+      "+1000-00")
+
+    testTableApi(
+      'f8.cast(Types.INTERVAL_MILLIS),
+      "f8.cast(INTERVAL_MILLIS)",
+      "+16979 07:23:33.000")
+  }
+
+  @Test
+  def testTimePointComparison(): Unit = {
+    testAllApis(
+      'f0 < 'f3,
+      "f0 < f3",
+      "f0 < f3",
+      "false")
+
+    testAllApis(
+      'f0 < 'f4,
+      "f0 < f4",
+      "f0 < f4",
+      "true")
+
+    testAllApis(
+      'f1 < 'f5,
+      "f1 < f5",
+      "f1 < f5",
+      "false")
+
+    testAllApis(
+      'f0.cast(Types.TIMESTAMP) !== 'f2,
+      "f0.cast(TIMESTAMP) !== f2",
+      "CAST(f0 AS TIMESTAMP) <> f2",
+      "true")
+
+    testAllApis(
+      'f0.cast(Types.TIMESTAMP) === 'f6,
+      "f0.cast(TIMESTAMP) === f6",
+      "CAST(f0 AS TIMESTAMP) = f6",
+      "true")
+  }
+
+  @Test
+  def testTimeIntervalArithmetic(): Unit = {
+    testAllApis(
+      12.month < 24.month,
+      "12.month < 24.month",
+      "INTERVAL '12' MONTH < INTERVAL '24' MONTH",
+      "true")
+
+    testAllApis(
+      8.milli > 10.milli,
+      "8.milli > 10.milli",
+      "INTERVAL '0.008' SECOND > INTERVAL '0.010' SECOND",
+      "false")
+
+    testAllApis(
+      8.year === 8.year,
+      "8.year === 8.year",
+      "INTERVAL '8' YEAR = INTERVAL '8' YEAR",
+      "true")
+
+    testAllApis(
+      8.year + 10.month,
+      "8.year + 10.month",
+      "INTERVAL '8' YEAR + INTERVAL '10' MONTH",
+      "+8-10")
+
+    testAllApis(
+      8.hour + 10.minute + 12.second + 5.milli,
+      "8.hour + 10.minute + 12.second + 5.milli",
+      "INTERVAL '8' HOUR + INTERVAL '10' MINUTE + INTERVAL '12.005' SECOND",
+      "+0 08:10:12.005")
+
+    testAllApis(
+      1.minute - 10.second,
+      "1.minute - 10.second",
+      "INTERVAL '1' MINUTE - INTERVAL '10' SECOND",
+      "+0 00:00:50.000")
+
+    testAllApis(
+      2.year - 12.month,
+      "2.year - 12.month",
+      "INTERVAL '2' YEAR - INTERVAL '12' MONTH",
+      "+1-00")
+
+    testAllApis(
+      -'f9.cast(Types.INTERVAL_MONTHS),
+      "-f9.cast(INTERVAL_MONTHS)",
+      "-CAST(f9 AS INTERVAL YEAR)",
+      "-2-00")
+
+    testAllApis(
+      'f0 + 2.day,
+      "f0 + 2.day",
+      "f0 + INTERVAL '2' DAY",
+      "1990-10-16")
+
+    testAllApis(
+      30.day + 'f0,
+      "30.day + f0",
+      "INTERVAL '30' DAY + f0",
+      "1990-11-13")
+
+    testAllApis(
+      'f1 + 12.hour,
+      "f1 + 12.hour",
+      "f1 + INTERVAL '12' HOUR",
+      "22:20:45")
+
+    testAllApis(
+      24.hour + 'f1,
+      "24.hour + f1",
+      "INTERVAL '24' HOUR + f1",
+      "10:20:45")
+
+    testAllApis(
+      'f2 + 10.day + 4.milli,
+      "f2 + 10.day + 4.milli",
+      "f2 + INTERVAL '10 00:00:00.004' DAY TO SECOND",
+      "1990-10-24 10:20:45.127")
+  }
+
+  // ----------------------------------------------------------------------------------------------
+
+  def testData = {
+    val testData = new Row(11)
+    testData.setField(0, Date.valueOf("1990-10-14"))
+    testData.setField(1, Time.valueOf("10:20:45"))
+    testData.setField(2, Timestamp.valueOf("1990-10-14 10:20:45.123"))
+    testData.setField(3, Date.valueOf("1990-10-13"))
+    testData.setField(4, Date.valueOf("1990-10-15"))
+    testData.setField(5, Time.valueOf("00:00:00"))
+    testData.setField(6, Timestamp.valueOf("1990-10-14 00:00:00.0"))
+    testData.setField(7, 12000)
+    testData.setField(8, 1467012213000L)
+    testData.setField(9, 24)
+    testData.setField(10, 12000L)
+    testData
+  }
+
+  def typeInfo = {
+    new RowTypeInfo(Seq(
+      Types.DATE,
+      Types.TIME,
+      Types.TIMESTAMP,
+      Types.DATE,
+      Types.DATE,
+      Types.TIME,
+      Types.TIMESTAMP,
+      Types.INT,
+      Types.LONG,
+      Types.INTERVAL_MONTHS,
+      Types.INTERVAL_MILLIS)).asInstanceOf[TypeInformation[Any]]
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f170e04a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/expression/TimeTypesTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/expression/TimeTypesTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/expression/TimeTypesTest.scala
deleted file mode 100644
index 532a3bd..0000000
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/expression/TimeTypesTest.scala
+++ /dev/null
@@ -1,231 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License 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 org.apache.flink.api.scala.expression
-
-import java.sql.{Date, Time, Timestamp}
-
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.scala.table._
-import org.apache.flink.api.table.expressions.utils.ExpressionTestBase
-import org.apache.flink.api.table.typeutils.RowTypeInfo
-import org.apache.flink.api.table.{Row, Types}
-import org.junit.Test
-
-class TimeTypesTest extends ExpressionTestBase {
-
-  @Test
-  def testTimeLiterals(): Unit = {
-    testAllApis(
-      "1990-10-14".toDate,
-      "'1990-10-14'.toDate",
-      "DATE '1990-10-14'",
-      "1990-10-14")
-
-    testTableApi(
-      Date.valueOf("2040-09-11"),
-      "'2040-09-11'.toDate",
-      "2040-09-11")
-
-    testAllApis(
-      "1500-04-30".cast(Types.DATE),
-      "'1500-04-30'.cast(DATE)",
-      "CAST('1500-04-30' AS DATE)",
-      "1500-04-30")
-
-    testAllApis(
-      "15:45:59".toTime,
-      "'15:45:59'.toTime",
-      "TIME '15:45:59'",
-      "15:45:59")
-
-    testTableApi(
-      Time.valueOf("00:00:00"),
-      "'00:00:00'.toTime",
-      "00:00:00")
-
-    testAllApis(
-      "1:30:00".cast(Types.TIME),
-      "'1:30:00'.cast(TIME)",
-      "CAST('1:30:00' AS TIME)",
-      "01:30:00")
-
-    testAllApis(
-      "1990-10-14 23:00:00.123".toTimestamp,
-      "'1990-10-14 23:00:00.123'.toTimestamp",
-      "TIMESTAMP '1990-10-14 23:00:00.123'",
-      "1990-10-14 23:00:00.123")
-
-    testTableApi(
-      Timestamp.valueOf("2040-09-11 00:00:00.000"),
-      "'2040-09-11 00:00:00.000'.toTimestamp",
-      "2040-09-11 00:00:00.0")
-
-    testAllApis(
-      "1500-04-30 12:00:00".cast(Types.TIMESTAMP),
-      "'1500-04-30 12:00:00'.cast(TIMESTAMP)",
-      "CAST('1500-04-30 12:00:00' AS TIMESTAMP)",
-      "1500-04-30 12:00:00.0")
-  }
-
-  @Test
-  def testTimeInput(): Unit = {
-    testAllApis(
-      'f0,
-      "f0",
-      "f0",
-      "1990-10-14")
-
-    testAllApis(
-      'f1,
-      "f1",
-      "f1",
-      "10:20:45")
-
-    testAllApis(
-      'f2,
-      "f2",
-      "f2",
-      "1990-10-14 10:20:45.123")
-  }
-
-  @Test
-  def testTimeCasting(): Unit = {
-    testAllApis(
-      'f0.cast(Types.TIMESTAMP),
-      "f0.cast(TIMESTAMP)",
-      "CAST(f0 AS TIMESTAMP)",
-      "1990-10-14 00:00:00.0")
-
-    testAllApis(
-      'f1.cast(Types.TIMESTAMP),
-      "f1.cast(TIMESTAMP)",
-      "CAST(f1 AS TIMESTAMP)",
-      "1970-01-01 10:20:45.0")
-
-    testAllApis(
-      'f2.cast(Types.DATE),
-      "f2.cast(DATE)",
-      "CAST(f2 AS DATE)",
-      "1990-10-14")
-
-    testAllApis(
-      'f2.cast(Types.TIME),
-      "f2.cast(TIME)",
-      "CAST(f2 AS TIME)",
-      "10:20:45")
-
-    testAllApis(
-      'f2.cast(Types.TIME),
-      "f2.cast(TIME)",
-      "CAST(f2 AS TIME)",
-      "10:20:45")
-
-    testTableApi(
-      'f7.cast(Types.DATE),
-      "f7.cast(DATE)",
-      "2002-11-09")
-
-    testTableApi(
-      'f7.cast(Types.DATE).cast(Types.INT),
-      "f7.cast(DATE).cast(INT)",
-      "12000")
-
-    testTableApi(
-      'f7.cast(Types.TIME),
-      "f7.cast(TIME)",
-      "00:00:12")
-
-    testTableApi(
-      'f7.cast(Types.TIME).cast(Types.INT),
-      "f7.cast(TIME).cast(INT)",
-      "12000")
-
-    testTableApi(
-      'f8.cast(Types.TIMESTAMP),
-      "f8.cast(TIMESTAMP)",
-      "2016-06-27 07:23:33.0")
-
-    testTableApi(
-      'f8.cast(Types.TIMESTAMP).cast(Types.LONG),
-      "f8.cast(TIMESTAMP).cast(LONG)",
-      "1467012213000")
-  }
-
-  @Test
-  def testTimeComparison(): Unit = {
-    testAllApis(
-      'f0 < 'f3,
-      "f0 < f3",
-      "f0 < f3",
-      "false")
-
-    testAllApis(
-      'f0 < 'f4,
-      "f0 < f4",
-      "f0 < f4",
-      "true")
-
-    testAllApis(
-      'f1 < 'f5,
-      "f1 < f5",
-      "f1 < f5",
-      "false")
-
-    testAllApis(
-      'f0.cast(Types.TIMESTAMP) !== 'f2,
-      "f0.cast(TIMESTAMP) !== f2",
-      "CAST(f0 AS TIMESTAMP) <> f2",
-      "true")
-
-    testAllApis(
-      'f0.cast(Types.TIMESTAMP) === 'f6,
-      "f0.cast(TIMESTAMP) === f6",
-      "CAST(f0 AS TIMESTAMP) = f6",
-      "true")
-  }
-
-  // ----------------------------------------------------------------------------------------------
-
-  def testData = {
-    val testData = new Row(9)
-    testData.setField(0, Date.valueOf("1990-10-14"))
-    testData.setField(1, Time.valueOf("10:20:45"))
-    testData.setField(2, Timestamp.valueOf("1990-10-14 10:20:45.123"))
-    testData.setField(3, Date.valueOf("1990-10-13"))
-    testData.setField(4, Date.valueOf("1990-10-15"))
-    testData.setField(5, Time.valueOf("00:00:00"))
-    testData.setField(6, Timestamp.valueOf("1990-10-14 00:00:00.0"))
-    testData.setField(7, 12000)
-    testData.setField(8, 1467012213000L)
-    testData
-  }
-
-  def typeInfo = {
-    new RowTypeInfo(Seq(
-      Types.DATE,
-      Types.TIME,
-      Types.TIMESTAMP,
-      Types.DATE,
-      Types.DATE,
-      Types.TIME,
-      Types.TIMESTAMP,
-      Types.INT,
-      Types.LONG)).asInstanceOf[TypeInformation[Any]]
-  }
-}


Mime
View raw message