[FLINK-7452] [types] Add helper methods for all built-in Flink types to Types

This closes #4612.

docs/dev/table/sql.md

@@ -821,9 +821,9 @@ The SQL runtime is built on top of Flink's DataSet and DataStream APIs. Internal
 | `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.SQL_DATE`       | `DATE`                      | `java.sql.Date`        |
+| `Types.SQL_TIME`       | `TIME`                      | `java.sql.Time`        |
+| `Types.SQL_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`       |
 | `Types.PRIMITIVE_ARRAY`| `ARRAY`                     | e.g. `int[]`           |

docs/dev/table/tableApi.md

@@ -1559,9 +1559,9 @@ The Table API is built on top of Flink's DataSet and DataStream APIs. Internally
 | `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.SQL_DATE`       | `DATE`                      | `java.sql.Date`        |
+| `Types.SQL_TIME`       | `TIME`                      | `java.sql.Time`        |
+| `Types.SQL_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`       |
 | `Types.PRIMITIVE_ARRAY`| `ARRAY`                     | e.g. `int[]`           |

flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceTest.java

@@ -108,7 +108,7 @@ public class OrcTableSourceTest {
 		assertTrue(returnType instanceof RowTypeInfo);
 		RowTypeInfo rowType = (RowTypeInfo) returnType;
 
-		RowTypeInfo expected = Types.ROW_NAMED(getNestedFieldNames(), getNestedFieldTypes());
+		TypeInformation<Row> expected = Types.ROW_NAMED(getNestedFieldNames(), getNestedFieldTypes());
 		assertEquals(expected, rowType);
 	}
 

flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeInfoParser.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-                                           
+
 package org.apache.flink.api.java.typeutils;
 
 import java.lang.reflect.Field;
@@ -31,6 +31,10 @@ import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.types.Value;
 
+/**
+ * @deprecated Use {@link org.apache.flink.api.common.typeinfo.Types} instead.
+ */
+@Deprecated
 @Public
 public class TypeInfoParser {
 	private static final String TUPLE_PACKAGE = "org.apache.flink.api.java.tuple";

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/Types.scala

@@ -17,6 +17,8 @@
  */
 package org.apache.flink.table.api
 
+import _root_.java.{lang, math, sql, util}
+
 import org.apache.flink.api.common.typeinfo.{PrimitiveArrayTypeInfo, TypeInformation, Types => JTypes}
 import org.apache.flink.api.java.typeutils.{MapTypeInfo, MultisetTypeInfo, ObjectArrayTypeInfo}
 import org.apache.flink.table.typeutils.TimeIntervalTypeInfo
@@ -25,35 +27,95 @@ import org.apache.flink.types.Row
 import _root_.scala.annotation.varargs
 
 /**
-  * This class enumerates all supported types of the Table API.
+  * This class enumerates all supported types of the Table API & SQL.
   */
 object Types {
 
-  val STRING = JTypes.STRING
-  val BOOLEAN = JTypes.BOOLEAN
+  /**
+    * Returns type information for a Table API string or SQL VARCHAR type.
+    */
+  val STRING: TypeInformation[String] = JTypes.STRING
+
+  /**
+    * Returns type information for a Table API boolean or SQL BOOLEAN type.
+    */
+  val BOOLEAN: TypeInformation[lang.Boolean] = JTypes.BOOLEAN
+
+  /**
+    * Returns type information for a Table API byte or SQL TINYINT type.
+    */
+  val BYTE: TypeInformation[lang.Byte] = JTypes.BYTE
+
+  /**
+    * Returns type information for a Table API short or SQL SMALLINT type.
+    */
+  val SHORT: TypeInformation[lang.Short] = JTypes.SHORT
+
+  /**
+    * Returns type information for a Table API integer or SQL INT/INTEGER type.
+    */
+  val INT: TypeInformation[lang.Integer] = JTypes.INT
 
-  val BYTE = JTypes.BYTE
-  val SHORT = JTypes.SHORT
-  val INT = JTypes.INT
-  val LONG = JTypes.LONG
-  val FLOAT = JTypes.FLOAT
-  val DOUBLE = JTypes.DOUBLE
-  val DECIMAL = JTypes.DECIMAL
+  /**
+    * Returns type information for a Table API long or SQL BIGINT type.
+    */
+  val LONG: TypeInformation[lang.Long] = JTypes.LONG
 
-  val SQL_DATE = JTypes.SQL_DATE
-  val SQL_TIME = JTypes.SQL_TIME
-  val SQL_TIMESTAMP = JTypes.SQL_TIMESTAMP
-  val INTERVAL_MONTHS = TimeIntervalTypeInfo.INTERVAL_MONTHS
-  val INTERVAL_MILLIS = TimeIntervalTypeInfo.INTERVAL_MILLIS
+  /**
+    * Returns type information for a Table API float or SQL FLOAT/REAL type.
+    */
+  val FLOAT: TypeInformation[lang.Float] = JTypes.FLOAT
+
+  /**
+    * Returns type information for a Table API integer or SQL DOUBLE type.
+    */
+  val DOUBLE: TypeInformation[lang.Double] = JTypes.DOUBLE
 
   /**
-    * Generates row type information.
+    * Returns type information for a Table API big decimal or SQL DECIMAL type.
+    */
+  val DECIMAL: TypeInformation[math.BigDecimal] = JTypes.BIG_DEC
+
+  /**
+    * Returns type information for a Table API SQL date or SQL DATE type.
+    */
+  val SQL_DATE: TypeInformation[sql.Date] = JTypes.SQL_DATE
+
+  /**
+    * Returns type information for a Table API SQL time or SQL TIME type.
+    */
+  val SQL_TIME: TypeInformation[sql.Time] = JTypes.SQL_TIME
+
+  /**
+    * Returns type information for a Table API SQL timestamp or SQL TIMESTAMP type.
+    */
+  val SQL_TIMESTAMP: TypeInformation[sql.Timestamp] = JTypes.SQL_TIMESTAMP
+
+  /**
+    * Returns type information for a Table API interval of months.
+    */
+  val INTERVAL_MONTHS: TypeInformation[lang.Integer] = TimeIntervalTypeInfo.INTERVAL_MONTHS
+
+  /**
+    * Returns type information for a Table API interval milliseconds.
+    */
+  val INTERVAL_MILLIS: TypeInformation[lang.Long] = TimeIntervalTypeInfo.INTERVAL_MILLIS
+
+  /**
+    * Returns type information for [[org.apache.flink.types.Row]] with fields of the given types.
+    *
+    * A row is a variable-length, null-aware composite type for storing multiple values in a
+    * deterministic field order. Every field can be null regardless of the field's type.
+    * The type of row fields cannot be automatically inferred; therefore, it is required to provide
+    * type information whenever a row is used.
     *
-    * A row type consists of zero or more fields with a field name and a corresponding type.
+    * <p>The schema of rows can have up to <code>Integer.MAX_VALUE</code> fields, however, all
+    * row instances must strictly adhere to the schema defined by the type info.
     *
-    * The fields have the default names (f0, f1, f2 ..).
+    * This method generates type information with fields of the given types; the fields have
+    * the default names (f0, f1, f2 ..).
     *
-    * @param types types of row fields; e.g. Types.STRING, Types.INT
+    * @param types The types of the row fields, e.g., Types.STRING, Types.INT
     */
   @varargs
   def ROW(types: TypeInformation[_]*): TypeInformation[Row] = {
@@ -61,19 +123,29 @@ object Types {
   }
 
   /**
-    * Generates row type information.
+    * Returns type information for [[org.apache.flink.types.Row]] with fields of the given types
+    * and with given names.
+    *
+    * A row is a variable-length, null-aware composite type for storing multiple values in a
+    * deterministic field order. Every field can be null independent of the field's type.
+    * The type of row fields cannot be automatically inferred; therefore, it is required to provide
+    * type information whenever a row is used.
+    *
+    * <p>The schema of rows can have up to <code>Integer.MAX_VALUE</code> fields, however, all
+    * row instances must strictly adhere to the schema defined by the type info.
     *
-    * A row type consists of zero or more fields with a field name and a corresponding type.
+    * Example use: `Types.ROW(Array("name", "number"), Array(Types.STRING, Types.INT))`.
     *
-    * @param names names of row fields, e.g. "userid", "name"
-    * @param types types of row fields; e.g. Types.STRING, Types.INT
+    * @param fieldNames array of field names
+    * @param types      array of field types
     */
-  def ROW(names: Array[String], types: Array[TypeInformation[_]]): TypeInformation[Row] = {
-    JTypes.ROW_NAMED(names, types: _*)
+  def ROW(fieldNames: Array[String], types: Array[TypeInformation[_]]): TypeInformation[Row] = {
+    JTypes.ROW_NAMED(fieldNames, types: _*)
   }
 
   /**
-    * Generates type information for an array consisting of Java primitive elements.
+    * Generates type information for an array consisting of Java primitive elements. The elements
+    * do not support null values.
     *
     * @param elementType type of the array elements; e.g. Types.INT
     */
@@ -93,30 +165,35 @@ object Types {
   }
 
   /**
-    * Generates type information for an array consisting of Java object elements.
+    * Generates type information for an array consisting of Java object elements. Null values for
+    * elements are supported.
     *
     * @param elementType type of the array elements; e.g. Types.STRING or Types.INT
     */
-  def OBJECT_ARRAY(elementType: TypeInformation[_]): TypeInformation[_] = {
+  def OBJECT_ARRAY[E](elementType: TypeInformation[E]): TypeInformation[Array[E]] = {
     ObjectArrayTypeInfo.getInfoFor(elementType)
   }
 
   /**
-    * Generates type information for a Java HashMap.
+    * Generates type information for a Java HashMap. Null values in keys are not supported. An
+    * entry's value can be null.
     *
     * @param keyType type of the keys of the map e.g. Types.STRING
     * @param valueType type of the values of the map e.g. Types.STRING
     */
-  def MAP(keyType: TypeInformation[_], valueType: TypeInformation[_]): TypeInformation[_] = {
+  def MAP[K, V](
+      keyType: TypeInformation[K],
+      valueType: TypeInformation[V]): TypeInformation[util.Map[K, V]] = {
     new MapTypeInfo(keyType, valueType)
   }
 
   /**
-    * Generates type information for a Multiset.
+    * Generates type information for a Multiset. A Multiset is baked by a Java HashMap and maps an
+    * arbitrary key to an integer value. Null values in keys are not supported.
     *
     * @param elementType type of the elements of the multiset e.g. Types.STRING
     */
-  def MULTISET(elementType: TypeInformation[_]): TypeInformation[_] = {
+  def MULTISET[E](elementType: TypeInformation[E]): TypeInformation[util.Map[E, lang.Integer]] = {
     new MultisetTypeInfo(elementType)
   }
 }

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/stringexpr/CorrelateStringExpressionTest.scala

@@ -20,9 +20,8 @@ package org.apache.flink.table.api.stream.table.stringexpr
 
 import org.apache.flink.api.java.typeutils.RowTypeInfo
 import org.apache.flink.api.scala._
-import org.apache.flink.table.api.scala._
 import org.apache.flink.table.api._
-import org.apache.flink.table.runtime.utils._
+import org.apache.flink.table.api.scala._
 import org.apache.flink.table.utils._
 import org.apache.flink.types.Row
 import org.junit.Test

flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/Types.scala

+/*
+ * 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.typeutils
+
+import org.apache.flink.annotation.PublicEvolving
+import org.apache.flink.api.common.functions.InvalidTypesException
+import org.apache.flink.api.common.typeinfo.{TypeInformation, Types => JTypes}
+import org.apache.flink.types.Row
+
+import _root_.scala.collection.JavaConverters._
+import _root_.scala.util.{Either, Try}
+
+/**
+  * This class gives access to the type information of the most common Scala types for which Flink
+  * has built-in serializers and comparators.
+  *
+  * This class contains types of [[org.apache.flink.api.common.typeinfo.Types]] and adds
+  * types for Scala specific classes (such as [[Unit]] or case classes).
+  *
+  * In many cases, Flink tries to analyze generic signatures of functions to determine return
+  * types automatically. This class is intended for cases where type information has to be
+  * supplied manually or cases where automatic type inference results in an inefficient type.
+  *
+  * Scala macros allow to determine type information of classes and type parameters. You can
+  * use [[Types.of]] to let type information be determined automatically.
+  */
+@PublicEvolving
+object Types {
+
+  /**
+    * Generates type information based on the given class and/or its type parameters.
+    *
+    * The definition is similar to a [[org.apache.flink.api.common.typeinfo.TypeHint]] but does
+    * not require to implement anonymous classes.
+    *
+    * If the class could not be analyzed by the Scala type analyzer, the Java analyzer
+    * will be used.
+    *
+    * Example use:
+    *
+    * `Types.of[(Int, String, String)]` for Scala tuples
+    * `Types.of[Unit]` for Scala specific types
+    *
+    * @tparam T class to be analyzed
+    */
+  def of[T: TypeInformation]: TypeInformation[T] = {
+    val typeInfo: TypeInformation[T] = implicitly[TypeInformation[T]]
+    typeInfo
+  }
+
+  /**
+    * Returns type information for Scala [[Nothing]]. Does not support a null value.
+    */
+  val NOTHING: TypeInformation[Nothing] = new ScalaNothingTypeInfo
+
+  /**
+    * Returns type information for Scala [[Unit]]. Does not support a null value.
+    */
+  val UNIT: TypeInformation[Unit] = new UnitTypeInfo
+
+  /**
+    * Returns type information for [[String]] and [[java.lang.String]]. Supports a null value.
+    */
+  val STRING: TypeInformation[String] = JTypes.STRING
+
+  /**
+    * Returns type information for primitive [[Byte]] and [[java.lang.Byte]]. Does not
+    * support a null value.
+    */
+  val BYTE: TypeInformation[java.lang.Byte] = JTypes.BYTE
+
+  /**
+    * Returns type information for primitive [[Boolean]] and [[java.lang.Boolean]]. Does not
+    * support a null value.
+    */
+  val BOOLEAN: TypeInformation[java.lang.Boolean] = JTypes.BOOLEAN
+
+  /**
+    * Returns type information for primitive [[Short]] and [[java.lang.Short]]. Does not
+    * support a null value.
+    */
+  val SHORT: TypeInformation[java.lang.Short] = JTypes.SHORT
+
+  /**
+    * Returns type information for primitive [[Int]] and [[java.lang.Integer]]. Does not
+    * support a null value.
+    */
+  val INT: TypeInformation[java.lang.Integer] = JTypes.INT
+
+  /**
+    * Returns type information for primitive [[Long]] and [[java.lang.Long]]. Does not
+    * support a null value.
+    */
+  val LONG: TypeInformation[java.lang.Long] = JTypes.LONG
+
+  /**
+    * Returns type information for primitive [[Float]] and [[java.lang.Float]]. Does not
+    * support a null value.
+    */
+  val FLOAT: TypeInformation[java.lang.Float] = JTypes.FLOAT
+
+  /**
+    * Returns type information for primitive [[Double]] and [[java.lang.Double]]. Does not
+    * support a null value.
+    */
+  val DOUBLE: TypeInformation[java.lang.Double] = JTypes.DOUBLE
+
+  /**
+    * Returns type information for primitive [[Char]] and [[java.lang.Character]]. Does not
+    * support a null value.
+    */
+  val CHAR: TypeInformation[java.lang.Character] = JTypes.CHAR
+
+  /**
+    * Returns type information for Java [[java.math.BigDecimal]]. Supports a null value.
+    *
+    * Note that Scala [[BigDecimal]] is not supported yet.
+    */
+  val JAVA_BIG_DEC: TypeInformation[java.math.BigDecimal] = JTypes.BIG_DEC
+
+  /**
+    * Returns type information for Java [[java.math.BigInteger]]. Supports a null value.
+    *
+    * Note that Scala [[BigInt]] is not supported yet.
+    */
+  val JAVA_BIG_INT: TypeInformation[java.math.BigInteger] = JTypes.BIG_INT
+
+  /**
+    * Returns type information for [[java.sql.Date]]. Supports a null value.
+    */
+  val SQL_DATE: TypeInformation[java.sql.Date] = JTypes.SQL_DATE
+
+  /**
+    * Returns type information for [[java.sql.Time]]. Supports a null value.
+    */
+  val SQL_TIME: TypeInformation[java.sql.Time] = JTypes.SQL_TIME
+
+  /**
+    * Returns type information for [[java.sql.Timestamp]]. Supports a null value.
+    */
+  val SQL_TIMESTAMP: TypeInformation[java.sql.Timestamp] = JTypes.SQL_TIMESTAMP
+
+  /**
+    * Returns type information for [[org.apache.flink.types.Row]] with fields of the given types.
+    * A row itself must not be null.
+    *
+    * A row is a fixed-length, null-aware composite type for storing multiple values in a
+    * deterministic field order. Every field can be null regardless of the field's type.
+    * The type of row fields cannot be automatically inferred; therefore, it is required to provide
+    * type information whenever a row is used.
+    *
+    * <p>The schema of rows can have up to <code>Integer.MAX_VALUE</code> fields, however, all row
+    * instances must strictly adhere to the schema defined by the type info.
+    *
+    * This method generates type information with fields of the given types; the fields have
+    * the default names (f0, f1, f2 ..).
+    *
+    * @param types The types of the row fields, e.g., Types.STRING, Types.INT
+    */
+  def ROW(types: TypeInformation[_]*): TypeInformation[Row] = JTypes.ROW(types: _*)
+
+  /**
+    * Returns type information for [[org.apache.flink.types.Row]] with fields of the given types
+    * and with given names. A row must not be null.
+    *
+    * A row is a variable-length, null-aware composite type for storing multiple values in a
+    * deterministic field order. Every field can be null independent of the field's type.
+    * The type of row fields cannot be automatically inferred; therefore, it is required to provide
+    * type information whenever a row is used.
+    *
+    * <p>The schema of rows can have up to <code>Integer.MAX_VALUE</code> fields, however, all row
+    * instances must strictly adhere to the schema defined by the type info.
+    *
+    * Example use: `Types.ROW(Array("name", "number"), Array(Types.STRING, Types.INT))`.
+    *
+    * @param fieldNames array of field names
+    * @param types      array of field types
+    */
+  def ROW(fieldNames: Array[String], types: Array[TypeInformation[_]]): TypeInformation[Row] =
+    JTypes.ROW_NAMED(fieldNames, types: _*)
+
+  /**
+    * Returns type information for a POJO (Plain Old Java Object).
+    *
+    * A POJO class is public and standalone (no non-static inner class). It has a public
+    * no-argument constructor. All non-static, non-transient fields in the class (and all
+    * superclasses) are either public (and non-final) or have a public getter and a setter
+    * method that follows the Java beans naming conventions for getters and setters.
+    *
+    * A POJO is a fixed-length, null-aware composite type with non-deterministic field order.
+    * Every field can be null independent of the field's type.
+    *
+    * The generic types for all fields of the POJO can be defined in a hierarchy of subclasses.
+    *
+    * If Flink's type analyzer is unable to extract a valid POJO type information with
+    * type information for all fields, an
+    * [[org.apache.flink.api.common.functions.InvalidTypesException}]] is thrown.
+    * Alternatively, you can use [[Types.POJO(Class, Map)]] to specify all fields manually.
+    *
+    * @param pojoClass POJO class to be analyzed by Flink
+    */
+  def POJO[T](pojoClass: Class[T]): TypeInformation[T] = {
+    JTypes.POJO(pojoClass)
+  }
+
+  /**
+    * Returns type information for a POJO (Plain Old Java Object) and allows to specify all
+    * fields manually.
+    *
+    * A POJO class is public and standalone (no non-static inner class). It has a public no-argument
+    * constructor. All non-static, non-transient fields in the class (and all superclasses) are
+    * either public (and non-final) or have a public getter and a setter method that follows the
+    * Java beans naming conventions for getters and setters.
+    *
+    * A POJO is a fixed-length, null-aware composite type with non-deterministic field order.
+    * Every field can be null independent of the field's type.
+    *
+    * The generic types for all fields of the POJO can be defined in a hierarchy of subclasses.
+    *
+    * If Flink's type analyzer is unable to extract a POJO field, an
+    * [[org.apache.flink.api.common.functions.InvalidTypesException]] is thrown.
+    *
+    * '''Note:''' In most cases the type information of fields can be determined automatically,
+    * we recommend to use [[Types.POJO(Class)]].
+    *
+    * @param pojoClass POJO class
+    * @param fields    map of fields that map a name to type information. The map key is the name of
+    *                  the field and the value is its type.
+    */
+  def POJO[T](pojoClass: Class[T], fields: Map[String, TypeInformation[_]]): TypeInformation[T] = {
+    JTypes.POJO(pojoClass, fields.asJava)
+  }
+
+  /**
+    * Returns generic type information for any Scala/Java object. The serialization logic will
+    * use the general purpose serializer Kryo.
+    *
+    * Generic types are black-boxes for Flink, but allow any object and null values in fields.
+    *
+    * By default, serialization of this type is not very efficient. Please read the documentation
+    * about how to improve efficiency (namely by pre-registering classes).
+    *
+    * @param genericClass any Scala/Java class
+    */
+  def GENERIC[T](genericClass: Class[T]): TypeInformation[T] = JTypes.GENERIC(genericClass)
+
+  /**
+    * Returns type information for a Scala case class and Scala tuples.
+    *
+    * A Scala case class is a fixed-length composite type for storing multiple values in a
+    * deterministic field order. Fields of a case class are typed. Case classes and tuples are
+    * the most efficient composite type; therefore, they do not not support null-valued fields
+    * unless the type of the field supports nullability.
+    *
+    * Example use: `Types.CASE_CLASS[MyCaseClass]`
+    *
+    * @tparam T case class to be analyzed
+    */
+  def CASE_CLASS[T: TypeInformation]: TypeInformation[T] = {
+    val t = Types.of[T]
+    if (t.isInstanceOf[CaseClassTypeInfo[_]]) {
+      t
+    } else {
+      throw new InvalidTypesException("Case class type expected but was: " + t)
+    }
+  }
+
+  /**
+    * Returns type information for a Scala tuple.
+    *
+    * A Scala tuple is a fixed-length composite type for storing multiple values in a
+    * deterministic field order. Fields of a tuple are typed. Tuples are
+    * the most efficient composite type; therefore, they do not not support null-valued fields
+    * unless the type of the field supports nullability.
+    *
+    * Example use: `Types.TUPLE[(String, Int)]`
+    *
+    * @tparam T tuple to be analyzed
+    */
+  def TUPLE[T: TypeInformation]: TypeInformation[T] = {
+    CASE_CLASS[T]
+  }
+
+  /**
+    * Returns type information for Scala/Java arrays of primitive type (such as `Array[Byte]`).
+    * The array and its elements do not support null values.
+    *
+    * @param elementType element type of the array (e.g. Types.BOOLEAN, Types.INT, Types.DOUBLE)
+    */
+  def PRIMITIVE_ARRAY(elementType: TypeInformation[_]): TypeInformation[_] = {
+    JTypes.PRIMITIVE_ARRAY(elementType)
+  }
+
+  /**
+    * Returns type information for Scala/Java arrays of object types (such as `Array[String]`,
+    * `Array[java.lang.Integer]`). The array itself must not be null. Null values for elements
+    * are supported.
+    *
+    * @param elementType element type of the array
+    */
+  def OBJECT_ARRAY[E <: AnyRef](elementType: TypeInformation[E]): TypeInformation[Array[E]] = {
+    // necessary for the Scala compiler
+    JTypes.OBJECT_ARRAY(elementType).asInstanceOf[TypeInformation[Array[E]]]
+  }
+
+  /**
+    * Returns type information for Scala [[Either]] type. Null values are not supported.
+    *
+    * The either type can be used for a value of two possible types.
+    *
+    * Example use: `Types.EITHER(Types.INT, Types.NOTHING]`
+    *
+    * @param leftType type information of left side / [[Left]]
+    * @param rightType type information of right side / [[Right]]
+    */
+  def EITHER[A, B](
+      leftType: TypeInformation[A],
+      rightType: TypeInformation[B]): TypeInformation[Either[A, B]] = {
+    new EitherTypeInfo(classOf[Either[A, B]], leftType, rightType)
+  }
+
+  /**
+    * Returns type information for Scala [[Option]] type. Null values are not supported.
+    *
+    * The option type can be used for distinguishing between a value or no value.
+    *
+    * Example use: `Types.OPTION(Types.INT)`
+    *
+    * @param valueType type information of the option's value
+    */
+  def OPTION[A, T <: Option[A]](valueType: TypeInformation[A]): TypeInformation[T] = {
+    new OptionTypeInfo(valueType)
+  }
+
+  /**
+    * Returns type information for Scala [[Try]] type. Null values are not supported.
+    *
+    * The try type can be used for distinguishing between a value or throwable.
+    *
+    * Example use: `Types.TRY(Types.INT)`
+    *
+    * @param valueType type information of the try's value
+    */
+  def TRY[A, T <: Try[A]](valueType: TypeInformation[A]): TypeInformation[T] = {
+    new TryTypeInfo(valueType)
+  }
+
+  /**
+    * Returns type information for Scala enumerations. Null values are not supported.
+    *
+    * @param enum enumeration object
+    * @param valueClass value class
+    */
+  def ENUMERATION[E <: Enumeration](
+      enum: E,
+      valueClass: Class[E#Value]): TypeInformation[E#Value] = {
+    new EnumValueTypeInfo(enum, valueClass)
+  }
+
+  /**
+    * Returns type information for Scala collections that implement [[Traversable]]. Null values
+    * are not supported.
+    */
+  def TRAVERSABLE[T: TypeInformation]: TypeInformation[T] = {
+    val t = Types.of[T]
+    if (t.isInstanceOf[TraversableTypeInfo[_, _]]) {
+      t
+    } else {
+      throw new InvalidTypesException("Traversable type expected but was: " + t)
+    }
+  }
+}