[FLINK-5481] [core] Add utility method to easily generate RowTypeInfos

This closes #3127

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

+/*
+ * 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.java.typeutils;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
+
+/**
+ * This class enumerates all supported types of
+ * the BasicTypeInfo, SqlTimeTypeInfo and RowTypeInfo for creation simplifying
+ */
+public class Types {
+
+	public static final BasicTypeInfo<String> STRING = BasicTypeInfo.STRING_TYPE_INFO;
+	public static final BasicTypeInfo<Boolean> BOOLEAN = BasicTypeInfo.BOOLEAN_TYPE_INFO;
+	public static final BasicTypeInfo<Byte> BYTE = BasicTypeInfo.BYTE_TYPE_INFO;
+	public static final BasicTypeInfo<Short> SHORT = BasicTypeInfo.SHORT_TYPE_INFO;
+	public static final BasicTypeInfo<Integer> INT = BasicTypeInfo.INT_TYPE_INFO;
+	public static final BasicTypeInfo<Long> LONG = BasicTypeInfo.LONG_TYPE_INFO;
+	public static final BasicTypeInfo<Float> FLOAT = BasicTypeInfo.FLOAT_TYPE_INFO;
+	public static final BasicTypeInfo<Double> DOUBLE = BasicTypeInfo.DOUBLE_TYPE_INFO;
+	public static final BasicTypeInfo<BigDecimal> DECIMAL = BasicTypeInfo.BIG_DEC_TYPE_INFO;
+
+	public static final SqlTimeTypeInfo<Date> SQL_DATE = SqlTimeTypeInfo.DATE;
+	public static final SqlTimeTypeInfo<Time> SQL_TIME = SqlTimeTypeInfo.TIME;
+	public static final SqlTimeTypeInfo<Timestamp> SQL_TIMESTAMP = SqlTimeTypeInfo.TIMESTAMP;
+
+	/**
+	 * Generates RowTypeInfo with default names (f1, f2 ..).
+	 * same as new RowTypeInfo(types)
+	 *
+	 * @param types of Row fields. e.g. ROW(Types.STRING, Types.INT)
+	 */
+	public static RowTypeInfo ROW(TypeInformation<?>... types) {
+		return new RowTypeInfo(types);
+	}
+
+	/**
+	 * Generates RowTypeInfo.
+	 * same as new RowTypeInfo(types, names)
+	 *
+	 * e.g. ROW(new String[]{"name", "number"}, Types.STRING, Types.INT)
+	 *
+	 * @param fieldNames array of field names
+	 * @param types array of field types
+	 */
+	public static RowTypeInfo ROW(String[] fieldNames, TypeInformation<?>... types) {
+		return new RowTypeInfo(types, fieldNames);
+	}
+}

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

@@ -17,29 +17,51 @@
  */
 package org.apache.flink.table.api
 
-import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, SqlTimeTypeInfo}
+import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.table.typeutils.TimeIntervalTypeInfo
+import org.apache.flink.api.java.typeutils.{Types => JTypes}
 
 /**
   * This class enumerates all supported types of the Table API.
   */
-object Types {
+object Types extends JTypes {
 
-  val STRING = BasicTypeInfo.STRING_TYPE_INFO
-  val BOOLEAN = BasicTypeInfo.BOOLEAN_TYPE_INFO
+  val STRING = JTypes.STRING
+  val BOOLEAN = JTypes.BOOLEAN
 
-  val BYTE = BasicTypeInfo.BYTE_TYPE_INFO
-  val SHORT = BasicTypeInfo.SHORT_TYPE_INFO
-  val INT = BasicTypeInfo.INT_TYPE_INFO
-  val LONG = BasicTypeInfo.LONG_TYPE_INFO
-  val FLOAT = BasicTypeInfo.FLOAT_TYPE_INFO
-  val DOUBLE = BasicTypeInfo.DOUBLE_TYPE_INFO
-  val DECIMAL = BasicTypeInfo.BIG_DEC_TYPE_INFO
+  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
 
-  val DATE = SqlTimeTypeInfo.DATE
-  val TIME = SqlTimeTypeInfo.TIME
-  val TIMESTAMP = SqlTimeTypeInfo.TIMESTAMP
+  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
 
+  /**
+    * Generates RowTypeInfo with default names (f1, f2 ..).
+    * same as ``new RowTypeInfo(types)``
+    *
+    * @param types of Row fields. e.g. ROW(Types.STRING, Types.INT)
+    */
+  def ROW[T: Manifest](types: TypeInformation[_]*) = {
+    JTypes.ROW(types: _*)
+  }
+
+  /**
+    * Generates RowTypeInfo.
+    * same as ``new RowTypeInfo(types, names)``
+    *
+    * @param fields of Row. e.g. ROW(("name", Types.STRING), ("number", Types.INT))
+    */
+  def ROW_NAMED(fields: (String, TypeInformation[_])*) = {
+    val names = fields.toList.map(_._1).toArray
+    val types = fields.toList.map(_._2)
+    JTypes.ROW(names, types: _*)
+  }
 }

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/batch/table/stringexpr/CalcStringExpressionTest.scala

@@ -357,8 +357,8 @@ class CalcStringExpressionTest {
       .toTable(tEnv, 'a, 'b, 'c, 'd, 'e)
 
     val t1 = t.select('a, 'b, 'c, 'd, 'e, BigDecimal("11.2"), BigDecimal("11.2").bigDecimal,
-        "1984-07-12".cast(Types.DATE), "14:34:24".cast(Types.TIME),
-        "1984-07-12 14:34:24".cast(Types.TIMESTAMP))
+        "1984-07-12".cast(Types.SQL_DATE), "14:34:24".cast(Types.SQL_TIME),
+        "1984-07-12 14:34:24".cast(Types.SQL_TIMESTAMP))
     val t2 = t.select("a, b, c, d, e, 11.2, 11.2," +
       "'1984-07-12'.toDate, '14:34:24'.toTime," +
       "'1984-07-12 14:34:24'.toTimestamp")

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ArrayTypeTest.scala

@@ -347,7 +347,7 @@ class ArrayTypeTest extends ExpressionTestBase {
       Types.INT,
       Types.INT,
       PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO,
-      ObjectArrayTypeInfo.getInfoFor(Types.DATE),
+      ObjectArrayTypeInfo.getInfoFor(Types.SQL_DATE),
       ObjectArrayTypeInfo.getInfoFor(ObjectArrayTypeInfo.getInfoFor(Types.INT)),
       ObjectArrayTypeInfo.getInfoFor(PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO),
       ObjectArrayTypeInfo.getInfoFor(Types.INT),

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala

@@ -1150,9 +1150,9 @@ class ScalarFunctionsTest extends ExpressionTestBase {
       Types.DOUBLE,
       Types.INT,
       Types.DECIMAL,
-      Types.DATE,
-      Types.TIME,
-      Types.TIMESTAMP,
+      Types.SQL_DATE,
+      Types.SQL_TIME,
+      Types.SQL_TIMESTAMP,
       Types.INTERVAL_MILLIS,
       Types.INTERVAL_MONTHS,
       Types.BOOLEAN,

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/TemporalTypesTest.scala

@@ -44,7 +44,7 @@ class TemporalTypesTest extends ExpressionTestBase {
       "2040-09-11")
 
     testAllApis(
-      "1500-04-30".cast(Types.DATE),
+      "1500-04-30".cast(Types.SQL_DATE),
       "'1500-04-30'.cast(DATE)",
       "CAST('1500-04-30' AS DATE)",
       "1500-04-30")
@@ -61,7 +61,7 @@ class TemporalTypesTest extends ExpressionTestBase {
       "00:00:00")
 
     testAllApis(
-      "1:30:00".cast(Types.TIME),
+      "1:30:00".cast(Types.SQL_TIME),
       "'1:30:00'.cast(TIME)",
       "CAST('1:30:00' AS TIME)",
       "01:30:00")
@@ -78,7 +78,7 @@ class TemporalTypesTest extends ExpressionTestBase {
       "2040-09-11 00:00:00.0")
 
     testAllApis(
-      "1500-04-30 12:00:00".cast(Types.TIMESTAMP),
+      "1500-04-30 12:00:00".cast(Types.SQL_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")
@@ -168,62 +168,62 @@ class TemporalTypesTest extends ExpressionTestBase {
   @Test
   def testTimePointCasting(): Unit = {
     testAllApis(
-      'f0.cast(Types.TIMESTAMP),
+      'f0.cast(Types.SQL_TIMESTAMP),
       "f0.cast(TIMESTAMP)",
       "CAST(f0 AS TIMESTAMP)",
       "1990-10-14 00:00:00.0")
 
     testAllApis(
-      'f1.cast(Types.TIMESTAMP),
+      'f1.cast(Types.SQL_TIMESTAMP),
       "f1.cast(TIMESTAMP)",
       "CAST(f1 AS TIMESTAMP)",
       "1970-01-01 10:20:45.0")
 
     testAllApis(
-      'f2.cast(Types.DATE),
+      'f2.cast(Types.SQL_DATE),
       "f2.cast(DATE)",
       "CAST(f2 AS DATE)",
       "1990-10-14")
 
     testAllApis(
-      'f2.cast(Types.TIME),
+      'f2.cast(Types.SQL_TIME),
       "f2.cast(TIME)",
       "CAST(f2 AS TIME)",
       "10:20:45")
 
     testAllApis(
-      'f2.cast(Types.TIME),
+      'f2.cast(Types.SQL_TIME),
       "f2.cast(TIME)",
       "CAST(f2 AS TIME)",
       "10:20:45")
 
     testTableApi(
-      'f7.cast(Types.DATE),
+      'f7.cast(Types.SQL_DATE),
       "f7.cast(DATE)",
       "2002-11-09")
 
     testTableApi(
-      'f7.cast(Types.DATE).cast(Types.INT),
+      'f7.cast(Types.SQL_DATE).cast(Types.INT),
       "f7.cast(DATE).cast(INT)",
       "12000")
 
     testTableApi(
-      'f7.cast(Types.TIME),
+      'f7.cast(Types.SQL_TIME),
       "f7.cast(TIME)",
       "00:00:12")
 
     testTableApi(
-      'f7.cast(Types.TIME).cast(Types.INT),
+      'f7.cast(Types.SQL_TIME).cast(Types.INT),
       "f7.cast(TIME).cast(INT)",
       "12000")
 
     testTableApi(
-      'f8.cast(Types.TIMESTAMP),
+      'f8.cast(Types.SQL_TIMESTAMP),
       "f8.cast(TIMESTAMP)",
       "2016-06-27 07:23:33.0")
 
     testTableApi(
-      'f8.cast(Types.TIMESTAMP).cast(Types.LONG),
+      'f8.cast(Types.SQL_TIMESTAMP).cast(Types.LONG),
       "f8.cast(TIMESTAMP).cast(LONG)",
       "1467012213000")
   }
@@ -262,13 +262,13 @@ class TemporalTypesTest extends ExpressionTestBase {
       "false")
 
     testAllApis(
-      'f0.cast(Types.TIMESTAMP) !== 'f2,
+      'f0.cast(Types.SQL_TIMESTAMP) !== 'f2,
       "f0.cast(TIMESTAMP) !== f2",
       "CAST(f0 AS TIMESTAMP) <> f2",
       "true")
 
     testAllApis(
-      'f0.cast(Types.TIMESTAMP) === 'f6,
+      'f0.cast(Types.SQL_TIMESTAMP) === 'f6,
       "f0.cast(TIMESTAMP) === f6",
       "CAST(f0 AS TIMESTAMP) = f6",
       "true")
@@ -558,13 +558,13 @@ class TemporalTypesTest extends ExpressionTestBase {
 
   def typeInfo = {
     new RowTypeInfo(
-      Types.DATE,
-      Types.TIME,
-      Types.TIMESTAMP,
-      Types.DATE,
-      Types.DATE,
-      Types.TIME,
-      Types.TIMESTAMP,
+      Types.SQL_DATE,
+      Types.SQL_TIME,
+      Types.SQL_TIMESTAMP,
+      Types.SQL_DATE,
+      Types.SQL_DATE,
+      Types.SQL_TIME,
+      Types.SQL_TIMESTAMP,
       Types.INT,
       Types.LONG,
       Types.INTERVAL_MONTHS,

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/UserDefinedScalarFunctionTest.scala

@@ -284,7 +284,7 @@ class UserDefinedScalarFunctionTest extends ExpressionTestBase {
       "7591 and 43810000 and 655906210000")
 
     testAllApis(
-      JavaFunc1(Null(Types.TIME), 15, Null(Types.TIMESTAMP)),
+      JavaFunc1(Null(Types.SQL_TIME), 15, Null(Types.SQL_TIMESTAMP)),
       "JavaFunc1(Null(TIME), 15, Null(TIMESTAMP))",
       "JavaFunc1(NULL, 15, NULL)",
       "null and 15 and null")
@@ -337,9 +337,9 @@ class UserDefinedScalarFunctionTest extends ExpressionTestBase {
       Types.STRING,
       Types.BOOLEAN,
       TypeInformation.of(classOf[SimplePojo]),
-      Types.DATE,
-      Types.TIME,
-      Types.TIMESTAMP,
+      Types.SQL_DATE,
+      Types.SQL_TIME,
+      Types.SQL_TIMESTAMP,
       Types.INTERVAL_MONTHS,
       Types.INTERVAL_MILLIS,
       TypeInformation.of(classOf[Seq[String]])

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/utils/UserDefinedScalarFunctions.scala

@@ -106,7 +106,7 @@ object Func10 extends ScalarFunction {
   }
 
   override def getResultType(signature: Array[Class[_]]): TypeInformation[_] = {
-    Types.TIMESTAMP
+    Types.SQL_TIMESTAMP
   }
 }