You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2019/03/05 13:17:22 UTC
[flink] branch master updated: [FLINK-11803][table-planner-blink]
Improve FlinkTypeFactory for Blink (#7898)
This is an automated email from the ASF dual-hosted git repository.
kurt pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new fb256d4 [FLINK-11803][table-planner-blink] Improve FlinkTypeFactory for Blink (#7898)
fb256d4 is described below
commit fb256d48e52c490b6a899c17c1225d30f7a6c92f
Author: Jingsong Lee <lz...@aliyun.com>
AuthorDate: Tue Mar 5 21:17:15 2019 +0800
[FLINK-11803][table-planner-blink] Improve FlinkTypeFactory for Blink (#7898)
---
.../flink/table/calcite/FlinkTypeFactory.scala | 167 ++++++++++++---------
.../flink/table/calcite/FlinkTypeSystem.scala | 51 +++++++
.../flink/table/plan/schema/ArrayRelDataType.scala | 55 +++++++
.../flink/table/plan/schema/InlineTable.scala | 18 +--
.../flink/table/plan/schema/MapRelDataType.scala | 50 ++++++
.../flink/table/plan/schema/RowRelDataType.scala | 89 +++++++++++
.../apache/flink/table/plan/schema/RowSchema.scala | 72 +++++++++
.../flink/table/calcite/FlinkTypeFactoryTest.scala | 75 +++++++++
8 files changed, 497 insertions(+), 80 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
index ba52152..0ba2252 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
@@ -18,28 +18,22 @@
package org.apache.flink.table.calcite
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
-import org.apache.flink.api.common.typeinfo._
-import org.apache.flink.api.java.typeutils.ValueTypeInfo._
-import org.apache.flink.table.`type`.DecimalType
+import org.apache.flink.table.`type`.{ArrayType, DecimalType, InternalType, InternalTypes, MapType, RowType}
import org.apache.flink.table.api.TableException
-import org.apache.flink.table.calcite.FlinkTypeFactory.typeInfoToSqlTypeName
-import org.apache.flink.table.typeutils._
+import org.apache.flink.table.plan.schema.{ArrayRelDataType, MapRelDataType, RowRelDataType, RowSchema}
-import org.apache.calcite.avatica.util.TimeUnit
import org.apache.calcite.jdbc.JavaTypeFactoryImpl
import org.apache.calcite.rel.`type`._
-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 java.util
import scala.collection.JavaConverters._
+import scala.collection.mutable
/**
- * Flink specific type factory that represents the interface between Flink's [[TypeInformation]]
+ * Flink specific type factory that represents the interface between Flink's [[InternalType]]
* and Calcite's [[RelDataType]].
*/
class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImpl(typeSystem) {
@@ -47,39 +41,51 @@ class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImp
// NOTE: for future data types it might be necessary to
// override more methods of RelDataTypeFactoryImpl
- def isAdvanced(dataType: TypeInformation[_]): Boolean = dataType match {
- case PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO => false
- case _: BasicTypeInfo[_] => false
- case _: SqlTimeTypeInfo[_] => false
- case _: TimeIntervalTypeInfo[_] => false
- case _ => true
- }
+ private val seenTypes = mutable.HashMap[(InternalType, Boolean), RelDataType]()
- def createTypeFromTypeInfo(
- typeInfo: TypeInformation[_],
+ def createTypeFromInternalType(
+ tp: InternalType,
isNullable: Boolean): RelDataType = {
- // we cannot use seenTypes for simple types,
- // because time indicators and timestamps would be the same
-
- val relType = if (!isAdvanced(typeInfo)) {
- // simple types can be converted to SQL types and vice versa
- val sqlType = typeInfoToSqlTypeName(typeInfo)
- sqlType match {
-
- case INTERVAL_YEAR_MONTH =>
- createSqlIntervalType(
- new SqlIntervalQualifier(TimeUnit.YEAR, TimeUnit.MONTH, SqlParserPos.ZERO))
-
- case INTERVAL_DAY_SECOND =>
- createSqlIntervalType(
- new SqlIntervalQualifier(TimeUnit.DAY, TimeUnit.SECOND, SqlParserPos.ZERO))
-
- case _ =>
- createSqlType(sqlType)
- }
- } else {
- throw new TableException("Unsupported now")
+ val relType = seenTypes.get((tp, isNullable)) match {
+ case Some(retType: RelDataType) => retType
+ case None =>
+ val refType = tp match {
+ case InternalTypes.BOOLEAN => createSqlType(BOOLEAN)
+ case InternalTypes.BYTE => createSqlType(TINYINT)
+ case InternalTypes.SHORT => createSqlType(SMALLINT)
+ case InternalTypes.INT => createSqlType(INTEGER)
+ case InternalTypes.LONG => createSqlType(BIGINT)
+ case InternalTypes.FLOAT => createSqlType(FLOAT)
+ case InternalTypes.DOUBLE => createSqlType(DOUBLE)
+ case InternalTypes.STRING => createSqlType(VARCHAR)
+
+ // temporal types
+ case InternalTypes.DATE => createSqlType(DATE)
+ case InternalTypes.TIME => createSqlType(TIME)
+ case InternalTypes.TIMESTAMP => createSqlType(TIMESTAMP)
+
+ case InternalTypes.BINARY => createSqlType(VARBINARY)
+
+ case InternalTypes.CHAR =>
+ throw new TableException("Character type is not supported.")
+
+ case rowType: RowType => new RowRelDataType(rowType, isNullable, this)
+
+ case arrayType: ArrayType => new ArrayRelDataType(arrayType,
+ createTypeFromInternalType(arrayType.getElementType, isNullable = true), isNullable)
+
+ case mapType: MapType => new MapRelDataType(
+ mapType,
+ createTypeFromInternalType(mapType.getKeyType, isNullable = true),
+ createTypeFromInternalType(mapType.getValueType, isNullable = true),
+ isNullable)
+
+ case _@t =>
+ throw new TableException(s"Type is not supported: $t")
+ }
+ seenTypes.put((tp, isNullable), refType)
+ refType
}
createTypeWithNullability(relType, isNullable)
@@ -89,12 +95,12 @@ class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImp
* Creates a struct type with the input fieldNames and input fieldTypes using FlinkTypeFactory
*
* @param fieldNames field names
- * @param fieldTypes field types, every element is Flink's [[TypeInformation]]
+ * @param fieldTypes field types, every element is Flink's [[InternalType]]
* @return a struct type with the input fieldNames, input fieldTypes, and system fields
*/
def buildLogicalRowType(
fieldNames: Seq[String],
- fieldTypes: Seq[TypeInformation[_]]): RelDataType = {
+ fieldTypes: Seq[InternalType]): RelDataType = {
buildLogicalRowType(
fieldNames,
fieldTypes,
@@ -106,19 +112,19 @@ class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImp
* using FlinkTypeFactory
*
* @param fieldNames field names
- * @param fieldTypes field types, every element is Flink's [[TypeInformation]]
+ * @param fieldTypes field types, every element is Flink's [[InternalType]]
* @param fieldNullables field nullable properties
* @return a struct type with the input fieldNames, input fieldTypes, and system fields
*/
def buildLogicalRowType(
fieldNames: Seq[String],
- fieldTypes: Seq[TypeInformation[_]],
+ fieldTypes: Seq[InternalType],
fieldNullables: Seq[Boolean]): RelDataType = {
val logicalRowTypeBuilder = builder
val fields = fieldNames.zip(fieldTypes).zip(fieldNullables)
fields foreach {
case ((fieldName, fieldType), fieldNullable) =>
- logicalRowTypeBuilder.add(fieldName, createTypeFromTypeInfo(fieldType, fieldNullable))
+ logicalRowTypeBuilder.add(fieldName, createTypeFromInternalType(fieldType, fieldNullable))
}
logicalRowTypeBuilder.build
}
@@ -215,30 +221,49 @@ class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImp
object FlinkTypeFactory {
- private[flink] def typeInfoToSqlTypeName(typeInfo: TypeInformation[_]): SqlTypeName =
- typeInfo match {
- case BOOLEAN_TYPE_INFO => BOOLEAN
- case BYTE_TYPE_INFO => TINYINT
- case SHORT_TYPE_INFO => SMALLINT
- case INT_TYPE_INFO => INTEGER
- case LONG_TYPE_INFO => BIGINT
- case FLOAT_TYPE_INFO => FLOAT
- case DOUBLE_TYPE_INFO => DOUBLE
- case STRING_TYPE_INFO => VARCHAR
-
- // temporal types
- case SqlTimeTypeInfo.DATE => DATE
- case SqlTimeTypeInfo.TIME => TIME
- case SqlTimeTypeInfo.TIMESTAMP => TIMESTAMP
- case TimeIntervalTypeInfo.INTERVAL_MONTHS => INTERVAL_YEAR_MONTH
- case TimeIntervalTypeInfo.INTERVAL_MILLIS => INTERVAL_DAY_SECOND
-
- case PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO => VARBINARY
-
- case CHAR_TYPE_INFO | CHAR_VALUE_TYPE_INFO =>
- throw new TableException("Character type is not supported.")
-
- case _@t =>
- throw new TableException(s"Type is not supported: $t")
- }
+ def toInternalType(relDataType: RelDataType): InternalType = relDataType.getSqlTypeName match {
+ case BOOLEAN => InternalTypes.BOOLEAN
+ case TINYINT => InternalTypes.BYTE
+ case SMALLINT => InternalTypes.SHORT
+ case INTEGER => InternalTypes.INT
+ case BIGINT => InternalTypes.LONG
+ case FLOAT => InternalTypes.FLOAT
+ case DOUBLE => InternalTypes.DOUBLE
+ case VARCHAR | CHAR => InternalTypes.STRING
+ case DECIMAL => throw new RuntimeException("Not support yet.")
+
+ // temporal types
+ case DATE => InternalTypes.DATE
+ case TIME => InternalTypes.TIME
+ case TIMESTAMP => InternalTypes.TIMESTAMP
+
+ case VARBINARY => InternalTypes.BINARY
+
+ case NULL =>
+ throw new TableException(
+ "Type NULL is not supported. Null values must have a supported type.")
+
+ // symbol for special flags e.g. TRIM's BOTH, LEADING, TRAILING
+ // are represented as integer
+ case SYMBOL => InternalTypes.INT
+
+ case ROW if relDataType.isInstanceOf[RowRelDataType] =>
+ val compositeRelDataType = relDataType.asInstanceOf[RowRelDataType]
+ compositeRelDataType.rowType
+
+ case ROW if relDataType.isInstanceOf[RelRecordType] =>
+ val relRecordType = relDataType.asInstanceOf[RelRecordType]
+ new RowSchema(relRecordType).internalType
+
+ case ARRAY if relDataType.isInstanceOf[ArrayRelDataType] =>
+ val arrayRelDataType = relDataType.asInstanceOf[ArrayRelDataType]
+ arrayRelDataType.arrayType
+
+ case MAP if relDataType.isInstanceOf[MapRelDataType] =>
+ val mapRelDataType = relDataType.asInstanceOf[MapRelDataType]
+ mapRelDataType.mapType
+
+ case _@t =>
+ throw new TableException(s"Type is not supported: $t")
+ }
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeSystem.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeSystem.scala
new file mode 100644
index 0000000..99d8cab
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeSystem.scala
@@ -0,0 +1,51 @@
+/*
+ * 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.table.calcite
+
+import org.apache.calcite.rel.`type`.RelDataTypeSystemImpl
+import org.apache.calcite.sql.`type`.SqlTypeName
+
+/**
+ * Custom type system for Flink.
+ */
+class FlinkTypeSystem extends RelDataTypeSystemImpl {
+
+ // we cannot use Int.MaxValue because of an overflow in Calcite's type inference logic
+ // half should be enough for all use cases
+ override def getMaxNumericScale: Int = Int.MaxValue / 2
+
+ // we cannot use Int.MaxValue because of an overflow in Calcite's type inference logic
+ // half should be enough for all use cases
+ override def getMaxNumericPrecision: Int = Int.MaxValue / 2
+
+ override def getDefaultPrecision(typeName: SqlTypeName): Int = typeName match {
+
+ // Calcite will limit the length of the VARCHAR field to 65536
+ case SqlTypeName.VARCHAR =>
+ Int.MaxValue
+
+ // we currently support only timestamps with milliseconds precision
+ case SqlTypeName.TIMESTAMP =>
+ 3
+
+ case _ =>
+ super.getDefaultPrecision(typeName)
+ }
+
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/ArrayRelDataType.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/ArrayRelDataType.scala
new file mode 100644
index 0000000..7b58ce5
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/ArrayRelDataType.scala
@@ -0,0 +1,55 @@
+/*
+ * 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.table.plan.schema
+
+import org.apache.flink.table.`type`.ArrayType
+
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.sql.`type`.ArraySqlType
+
+/**
+ * Flink distinguishes between primitive arrays (int[], double[], ...) and
+ * object arrays (Integer[], MyPojo[], ...). This custom type supports both cases.
+ */
+class ArrayRelDataType(
+ val arrayType: ArrayType,
+ elementType: RelDataType,
+ isNullable: Boolean)
+ extends ArraySqlType(
+ elementType,
+ isNullable) {
+
+ override def toString = s"ARRAY($arrayType)"
+
+ def canEqual(other: Any): Boolean = other.isInstanceOf[ArrayRelDataType]
+
+ override def equals(other: Any): Boolean = other match {
+ case that: ArrayRelDataType =>
+ super.equals(that) &&
+ (that canEqual this) &&
+ arrayType == that.arrayType &&
+ isNullable == that.isNullable
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ arrayType.hashCode()
+ }
+
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/InlineTable.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/InlineTable.scala
index fce6bcf..84d3977 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/InlineTable.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/InlineTable.scala
@@ -19,7 +19,7 @@
package org.apache.flink.table.plan.schema
import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.table.`type`.{InternalType, RowType, TypeConverters}
import org.apache.flink.table.api.TableException
import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.plan.stats.FlinkStatistic
@@ -55,21 +55,21 @@ abstract class InlineTable[T](
s"List of all fields: ${fieldNames.mkString("[", ", ", "]")}.")
}
- val fieldTypes: Array[TypeInformation[_]] =
- typeInfo match {
+ val fieldTypes: Array[InternalType] =
+ TypeConverters.createInternalTypeFromTypeInfo(typeInfo) match {
- case ct: CompositeType[_] =>
+ case rt: RowType =>
// it is ok to leave out fields
- if (fieldIndexes.count(_ >= 0) > ct.getArity) {
+ if (fieldIndexes.count(_ >= 0) > rt.getArity) {
throw new TableException(
- s"Arity of type (" + ct.getFieldNames.deep + ") " +
+ s"Arity of type (" + rt.getFieldNames.deep + ") " +
"must not be greater than number of field names " + fieldNames.deep + ".")
}
- fieldIndexes.map(i => ct.getTypeAt(i).asInstanceOf[TypeInformation[_]])
+ fieldIndexes.map(i => rt.getTypeAt(i))
- case t: TypeInformation[_] =>
+ case t: InternalType =>
val cnt = fieldIndexes.length
- val types = fieldIndexes.map(_.asInstanceOf[TypeInformation[_]])
+ val types = Array(t)
// ensure that the atomic type is matched at most once.
if (cnt > 1) {
throw new TableException(
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/MapRelDataType.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/MapRelDataType.scala
new file mode 100644
index 0000000..4a21904
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/MapRelDataType.scala
@@ -0,0 +1,50 @@
+/*
+ * 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.table.plan.schema
+
+import org.apache.flink.table.`type`.MapType
+
+import com.google.common.base.Objects
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.sql.`type`.MapSqlType
+
+class MapRelDataType(
+ val mapType: MapType,
+ val keyType: RelDataType,
+ val valueType: RelDataType,
+ isNullable: Boolean) extends MapSqlType(keyType, valueType, isNullable) {
+
+ override def toString: String = s"MAP($mapType)"
+
+ def canEqual(other: Any): Boolean = other.isInstanceOf[MapRelDataType]
+
+ override def equals(other: Any): Boolean = other match {
+ case that: MapRelDataType =>
+ super.equals(that) &&
+ (that canEqual this) &&
+ keyType == that.keyType &&
+ valueType == that.valueType &&
+ isNullable == that.isNullable
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ Objects.hashCode(keyType, valueType)
+ }
+
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/RowRelDataType.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/RowRelDataType.scala
new file mode 100644
index 0000000..ebb465c
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/RowRelDataType.scala
@@ -0,0 +1,89 @@
+/*
+ * 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.table.plan.schema
+
+import java.util
+import org.apache.calcite.rel.`type`.{RelDataTypeField, RelDataTypeFieldImpl, RelRecordType, StructKind}
+import org.apache.flink.table.`type`.RowType
+import org.apache.flink.table.calcite.FlinkTypeFactory
+import org.apache.flink.table.plan.schema.RowRelDataType.createFieldList
+
+import scala.collection.JavaConverters._
+
+/**
+ * Row type for encapsulating Flink's [[RowType]].
+ *
+ * @param rowType rowType to encapsulate
+ * @param nullable flag if type can be nullable
+ * @param typeFactory Flink's type factory
+ */
+class RowRelDataType(
+ val rowType: RowType,
+ val nullable: Boolean,
+ typeFactory: FlinkTypeFactory)
+ extends RelRecordType(
+ StructKind.PEEK_FIELDS_NO_EXPAND,
+ createFieldList(rowType, typeFactory)) {
+
+ override def toString = s"ROW($rowType)"
+
+ def canEqual(other: Any): Boolean = other.isInstanceOf[RowRelDataType]
+
+ override def equals(other: Any): Boolean = other match {
+ case that: RowRelDataType =>
+ super.equals(that) &&
+ (that canEqual this) &&
+ rowType == that.rowType &&
+ nullable == that.nullable
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ rowType.hashCode()
+ }
+
+ override def isNullable: Boolean = nullable
+
+}
+
+object RowRelDataType {
+
+ /**
+ * Converts the fields of a row type to list of [[RelDataTypeField]].
+ */
+ private def createFieldList(
+ rowType: RowType,
+ typeFactory: FlinkTypeFactory)
+ : util.List[RelDataTypeField] = {
+
+ rowType
+ .getFieldNames
+ .zipWithIndex
+ .map { case (name, index) =>
+ new RelDataTypeFieldImpl(
+ name,
+ index,
+ // TODO the row type should provide the information if subtypes are nullable
+ typeFactory.createTypeFromInternalType(rowType.getTypeAt(index), isNullable = true)
+ ).asInstanceOf[RelDataTypeField]
+ }
+ .toList
+ .asJava
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/RowSchema.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/RowSchema.scala
new file mode 100644
index 0000000..0d40074
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/RowSchema.scala
@@ -0,0 +1,72 @@
+/*
+ * 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.table.plan.schema
+
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.table.`type`.{InternalType, RowType}
+import org.apache.flink.table.calcite.FlinkTypeFactory
+
+import scala.collection.JavaConversions._
+
+/**
+ * Schema that describes both a logical and physical row.
+ */
+class RowSchema(private val logicalRowType: RelDataType) {
+
+ private lazy val internalFieldTypes: Seq[InternalType] =
+ logicalRowType.getFieldList map { f => FlinkTypeFactory.toInternalType(f.getType) }
+
+ private lazy val internalRowType: RowType = new RowType(
+ internalFieldTypes.toArray, fieldNames.toArray)
+
+ /**
+ * Returns the arity of the schema.
+ */
+ def arity: Int = logicalRowType.getFieldCount
+
+ /**
+ * Returns the [[RelDataType]] of the schema
+ */
+ def relDataType: RelDataType = logicalRowType
+
+ /**
+ * Returns the [[RowType]] of the schema
+ */
+ def internalType: RowType = internalRowType
+
+ /**
+ * Returns the [[RowType]] of fields of the schema
+ */
+ def fieldInternalTypes: Seq[InternalType] = internalFieldTypes
+
+ /**
+ * Returns the fields names
+ */
+ def fieldNames: Seq[String] = logicalRowType.getFieldNames
+
+ /**
+ * Returns a projected [[TypeInformation]] of the schema.
+ */
+ def projectedInternalType(fields: Array[Int]): RowType = {
+ val projectedTypes = fields.map(fieldInternalTypes(_))
+ val projectedNames = fields.map(fieldNames(_))
+ new RowType(projectedTypes, projectedNames)
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/calcite/FlinkTypeFactoryTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/calcite/FlinkTypeFactoryTest.scala
new file mode 100644
index 0000000..b714b3d
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/calcite/FlinkTypeFactoryTest.scala
@@ -0,0 +1,75 @@
+/*
+ * 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.table.calcite
+
+import org.apache.flink.table.`type`.{InternalType, InternalTypes}
+
+import org.junit.{Assert, Test}
+
+class FlinkTypeFactoryTest {
+
+ @Test
+ def testInternalToRelType(): Unit = {
+ val typeFactory = new FlinkTypeFactory(new FlinkTypeSystem)
+
+ def test(t: InternalType): Unit = {
+ Assert.assertEquals(
+ t,
+ FlinkTypeFactory.toInternalType(
+ typeFactory.createTypeFromInternalType(t, isNullable = true))
+ )
+
+ Assert.assertEquals(
+ t,
+ FlinkTypeFactory.toInternalType(
+ typeFactory.createTypeFromInternalType(t, isNullable = false))
+ )
+
+ // twice for cache.
+ Assert.assertEquals(
+ t,
+ FlinkTypeFactory.toInternalType(
+ typeFactory.createTypeFromInternalType(t, isNullable = true))
+ )
+
+ Assert.assertEquals(
+ t,
+ FlinkTypeFactory.toInternalType(
+ typeFactory.createTypeFromInternalType(t, isNullable = false))
+ )
+ }
+
+ test(InternalTypes.BOOLEAN)
+ test(InternalTypes.BYTE)
+ test(InternalTypes.STRING)
+ test(InternalTypes.DOUBLE)
+ test(InternalTypes.FLOAT)
+ test(InternalTypes.INT)
+ test(InternalTypes.LONG)
+ test(InternalTypes.SHORT)
+ test(InternalTypes.BINARY)
+ test(InternalTypes.DATE)
+ test(InternalTypes.TIME)
+ test(InternalTypes.TIMESTAMP)
+
+ test(InternalTypes.createArrayType(InternalTypes.DOUBLE))
+ test(InternalTypes.createMapType(InternalTypes.DOUBLE, InternalTypes.STRING))
+ test(InternalTypes.createRowType(InternalTypes.DOUBLE, InternalTypes.STRING))
+ }
+}