| /* |
| * 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 java.util |
| |
| 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`.{BasicSqlType, SqlTypeName, SqlTypeUtil} |
| import org.apache.calcite.sql.parser.SqlParserPos |
| import org.apache.flink.api.common.typeinfo.BasicTypeInfo.{INT_TYPE_INFO, _} |
| import org.apache.flink.api.common.typeinfo._ |
| import org.apache.flink.api.common.typeutils.CompositeType |
| import org.apache.flink.api.java.typeutils.ValueTypeInfo._ |
| import org.apache.flink.api.java.typeutils.{MapTypeInfo, MultisetTypeInfo, ObjectArrayTypeInfo, RowTypeInfo, PojoField => _} |
| import org.apache.flink.table.api.scala._ |
| import org.apache.flink.table.api.types._ |
| import org.apache.flink.table.api.{TableException, TableSchema} |
| import org.apache.flink.table.calcite.FlinkTypeFactory.typeInfoToSqlTypeName |
| import org.apache.flink.table.dataformat.{BaseRow, Decimal} |
| import org.apache.flink.table.plan.schema._ |
| import org.apache.flink.table.typeutils._ |
| import org.apache.flink.types.Row |
| |
| import scala.collection.JavaConverters._ |
| import scala.collection.mutable |
| |
| /** |
| * Flink specific type factory that represents the interface between Flink's [[TypeInformation]] |
| * and Calcite's [[RelDataType]]. |
| */ |
| class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImpl(typeSystem) { |
| |
| // NOTE: for future data types it might be necessary to |
| // override more methods of RelDataTypeFactoryImpl |
| |
| private val seenTypes = mutable.HashMap[(TypeInformation[_], Boolean), RelDataType]() |
| |
| def isAdvanced(dataType: TypeInformation[_]): Boolean = dataType match { |
| case PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO => false |
| case _: TimeIndicatorTypeInfo => false |
| case _: BasicTypeInfo[_] => false |
| case _: SqlTimeTypeInfo[_] => false |
| case _: TimeIntervalTypeInfo[_] => false |
| case _ => true |
| } |
| |
| def createTypeFromInternalType( |
| t: InternalType, |
| isNullable: Boolean) |
| : RelDataType = { |
| createTypeFromTypeInfo(t, isNullable) |
| } |
| |
| def createTypeFromTypeInfo( |
| typeInfo: TypeInformation[_], |
| 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 DECIMAL => |
| val dec = typeInfo.asInstanceOf[BigDecimalTypeInfo] |
| createSqlType(DECIMAL, dec.precision, dec.scale) |
| |
| 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 TIMESTAMP if typeInfo.isInstanceOf[TimeIndicatorTypeInfo] => |
| if (typeInfo.asInstanceOf[TimeIndicatorTypeInfo].isEventTime) { |
| createRowtimeIndicatorType() |
| } else { |
| createProctimeIndicatorType() |
| } |
| |
| case _ => |
| createSqlType(sqlType) |
| } |
| } else { |
| // advanced types require specific RelDataType |
| // for storing the original TypeInformation |
| seenTypes.getOrElseUpdate((typeInfo, isNullable), createAdvancedType(typeInfo, isNullable)) |
| } |
| |
| createTypeWithNullability(relType, isNullable) |
| } |
| |
| /** |
| * Creates a indicator type for processing-time, but with similar properties as SQL timestamp. |
| */ |
| def createProctimeIndicatorType(): RelDataType = { |
| val originalType = createTypeFromTypeInfo(SqlTimeTypeInfo.TIMESTAMP, isNullable = false) |
| canonize( |
| new TimeIndicatorRelDataType( |
| getTypeSystem, |
| originalType.asInstanceOf[BasicSqlType], |
| isEventTime = false) |
| ) |
| } |
| |
| /** |
| * Creates a indicator type for event-time, but with similar properties as SQL timestamp. |
| */ |
| def createRowtimeIndicatorType(): RelDataType = { |
| val originalType = createTypeFromTypeInfo(SqlTimeTypeInfo.TIMESTAMP, isNullable = false) |
| canonize( |
| new TimeIndicatorRelDataType( |
| getTypeSystem, |
| originalType.asInstanceOf[BasicSqlType], |
| isEventTime = true) |
| ) |
| } |
| |
| /** |
| * Creates types that create custom [[RelDataType]]s that wrap Flink's [[TypeInformation]]. |
| */ |
| private def createAdvancedType( |
| typeInfo: TypeInformation[_], |
| isNullable: Boolean): RelDataType = { |
| |
| val relType = typeInfo match { |
| |
| case ct: CompositeType[_] => |
| new CompositeRelDataType(ct, isNullable, this) |
| |
| case pa: PrimitiveArrayTypeInfo[_] => |
| new ArrayRelDataType( |
| pa, |
| createTypeFromTypeInfo(pa.getComponentType, isNullable = false), |
| isNullable) |
| |
| case ba: BasicArrayTypeInfo[_, _] => |
| new ArrayRelDataType( |
| ba, |
| createTypeFromTypeInfo(ba.getComponentInfo, isNullable = true), |
| isNullable) |
| |
| case oa: ObjectArrayTypeInfo[_, _] => |
| new ArrayRelDataType( |
| oa, |
| createTypeFromTypeInfo(oa.getComponentInfo, isNullable = true), |
| isNullable) |
| |
| case mp: MapTypeInfo[_, _] => |
| new MapRelDataType( |
| mp, |
| createTypeFromTypeInfo(mp.getKeyTypeInfo, isNullable = true), |
| createTypeFromTypeInfo(mp.getValueTypeInfo, isNullable = true), |
| isNullable) |
| |
| case mts: MultisetTypeInfo[_] => |
| new MultisetRelDataType( |
| mts, |
| createTypeFromTypeInfo(mts.getElementTypeInfo, isNullable = true), |
| isNullable |
| ) |
| |
| case ti: TypeInformation[_] => |
| new GenericRelDataType( |
| ti, |
| isNullable, |
| getTypeSystem.asInstanceOf[FlinkTypeSystem]) |
| } |
| |
| canonize(relType) |
| } |
| |
| /** |
| * 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]] |
| * @return a struct type with the input fieldNames, input fieldTypes, and system fields |
| */ |
| def buildLogicalRowType( |
| fieldNames: Seq[String], |
| fieldTypes: Seq[TypeInformation[_]]) |
| : RelDataType = { |
| buildLogicalRowType( |
| fieldNames, |
| fieldTypes, |
| fieldTypes.map(!FlinkTypeFactory.isTimeIndicatorType(_))) |
| } |
| |
| /** |
| * Creates a struct type with the input fieldNames, input fieldTypes and input fieldNullables |
| * using FlinkTypeFactory |
| * |
| * @param fieldNames field names |
| * @param fieldTypes field types, every element is Flink's [[TypeInformation]] |
| * @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[_]], |
| fieldNullables: Seq[Boolean]) |
| : RelDataType = { |
| val logicalRowTypeBuilder = builder |
| val fields = fieldNames.zip(fieldTypes).zip(fieldNullables) |
| fields foreach { |
| case ((fieldName, fieldType), fieldNullable) => |
| if (FlinkTypeFactory.isTimeIndicatorType(fieldType) && fieldNullable) { |
| throw new TableException( |
| s"$fieldName can not be nullable because it is TimeIndicatorType!") |
| } |
| logicalRowTypeBuilder.add(fieldName, createTypeFromTypeInfo(fieldType, fieldNullable)) |
| } |
| logicalRowTypeBuilder.build |
| } |
| |
| /** |
| * Created a struct type with the input table schema using FlinkTypeFactory |
| * @param tableSchema the table schema |
| * @return a struct type with the input fieldNames, input fieldTypes, and system fields |
| */ |
| def buildLogicalRowType(tableSchema: TableSchema, isStreaming: Boolean): RelDataType = { |
| buildRelDataType( |
| tableSchema.getFieldNames.toSeq, |
| tableSchema.getFieldTypes map { |
| case DataTypes.PROCTIME_INDICATOR if !isStreaming => DataTypes.TIMESTAMP |
| case DataTypes.ROWTIME_INDICATOR if !isStreaming => DataTypes.TIMESTAMP |
| case tpe: InternalType => tpe |
| }, |
| tableSchema.getFieldNullables) |
| } |
| |
| def buildRelDataType( |
| fieldNames: Seq[String], |
| fieldTypes: Seq[InternalType]) |
| : RelDataType = { |
| buildRelDataType( |
| fieldNames, |
| fieldTypes, |
| fieldTypes.map(!FlinkTypeFactory.isTimeIndicatorType(_))) |
| } |
| |
| def buildRelDataType( |
| fieldNames: Seq[String], |
| fieldTypes: Seq[InternalType], |
| fieldNullables: Seq[Boolean]) |
| : RelDataType = { |
| val b = builder |
| val fields = fieldNames.zip(fieldTypes).zip(fieldNullables) |
| fields foreach { |
| case ((fieldName, fieldType), fieldNullable) => |
| if (FlinkTypeFactory.isTimeIndicatorType(fieldType) && fieldNullable) { |
| throw new TableException( |
| s"$fieldName can not be nullable because it is TimeIndicatorType!") |
| } |
| b.add(fieldName, createTypeFromInternalType(fieldType, fieldNullable)) |
| } |
| b.build |
| } |
| |
| // ---------------------------------------------------------------------------------------------- |
| |
| |
| override def getJavaClass(`type`: RelDataType): java.lang.reflect.Type = { |
| if (`type`.getSqlTypeName == FLOAT) { |
| if (`type`.isNullable) { |
| classOf[java.lang.Float] |
| } else { |
| java.lang.Float.TYPE |
| } |
| } else { |
| super.getJavaClass(`type`) |
| } |
| } |
| |
| override def createSqlType(typeName: SqlTypeName, precision: Int): RelDataType = { |
| // it might happen that inferred VARCHAR types overflow as we set them to Int.MaxValue |
| // Calcite will limit the length of the VARCHAR type to 65536. |
| if (typeName == VARCHAR && precision < 0) { |
| createSqlType(typeName, getTypeSystem.getDefaultPrecision(typeName)) |
| } else { |
| super.createSqlType(typeName, precision) |
| } |
| } |
| |
| override def createSqlType(typeName: SqlTypeName): RelDataType = { |
| if (typeName == DECIMAL) { |
| // if we got here, the precision and scale are not specified, here we |
| // keep precision/scale in sync with our type system's default value, |
| // see DecimalType.USER_DEFAULT. |
| createSqlType(typeName, DecimalType.USER_DEFAULT.precision(), |
| DecimalType.USER_DEFAULT.scale()) |
| } else { |
| super.createSqlType(typeName) |
| } |
| } |
| |
| override def createArrayType(elementType: RelDataType, maxCardinality: Long): RelDataType = { |
| val arrayType = FlinkTypeFactory.toInternalType(elementType) match { |
| case DataTypes.BOOLEAN => BasicArrayTypeInfo.BOOLEAN_ARRAY_TYPE_INFO |
| case DataTypes.SHORT => BasicArrayTypeInfo.SHORT_ARRAY_TYPE_INFO |
| case DataTypes.INT => BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO |
| case DataTypes.LONG => BasicArrayTypeInfo.LONG_ARRAY_TYPE_INFO |
| case DataTypes.FLOAT => BasicArrayTypeInfo.FLOAT_ARRAY_TYPE_INFO |
| case DataTypes.DOUBLE => BasicArrayTypeInfo.DOUBLE_ARRAY_TYPE_INFO |
| case DataTypes.CHAR => BasicArrayTypeInfo.CHAR_ARRAY_TYPE_INFO |
| case DataTypes.STRING => BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO |
| |
| // object |
| case _ => ObjectArrayTypeInfo.getInfoFor(FlinkTypeFactory.toTypeInfo(elementType)) |
| } |
| val relType = new ArrayRelDataType( |
| arrayType, |
| elementType, |
| isNullable = false) |
| canonize(relType) |
| } |
| |
| override def createMapType(keyType: RelDataType, valueType: RelDataType): RelDataType = { |
| val relType = new MapRelDataType( |
| new MapTypeInfo( |
| FlinkTypeFactory.toTypeInfo(keyType), |
| FlinkTypeFactory.toTypeInfo(valueType)), |
| keyType, |
| valueType, |
| isNullable = false) |
| canonize(relType) |
| } |
| |
| override def createMultisetType(elementType: RelDataType, maxCardinality: Long): RelDataType = { |
| val relType = new MultisetRelDataType( |
| MultisetTypeInfo.getInfoFor(FlinkTypeFactory.toTypeInfo(elementType)), |
| elementType, |
| isNullable = false) |
| canonize(relType) |
| } |
| |
| override def createTypeWithNullability( |
| relDataType: RelDataType, |
| isNullable: Boolean): RelDataType = { |
| |
| // nullability change not necessary |
| if (relDataType.isNullable == isNullable) { |
| return canonize(relDataType) |
| } |
| |
| // change nullability |
| val newType = relDataType match { |
| |
| case composite: CompositeRelDataType => |
| new CompositeRelDataType(composite.compositeType, isNullable, this) |
| |
| case array: ArrayRelDataType => |
| new ArrayRelDataType(array.typeInfo, array.getComponentType, isNullable) |
| |
| case map: MapRelDataType => |
| new MapRelDataType(map.typeInfo, map.keyType, map.valueType, isNullable) |
| |
| case multiSet: MultisetRelDataType => |
| new MultisetRelDataType(multiSet.typeInfo, multiSet.getComponentType, isNullable) |
| |
| case generic: GenericRelDataType => |
| new GenericRelDataType(generic.typeInfo, isNullable, typeSystem) |
| |
| case timeIndicator: TimeIndicatorRelDataType => |
| timeIndicator |
| |
| case _ => |
| super.createTypeWithNullability(relDataType, isNullable) |
| } |
| |
| canonize(newType) |
| } |
| |
| override def leastRestrictive(types: util.List[RelDataType]): RelDataType = { |
| val type0 = types.get(0) |
| if (type0.getSqlTypeName != null) { |
| val resultType = resolveAllIdenticalTypes(types) |
| if (resultType.isDefined) { |
| // result type for identical types |
| return resultType.get |
| } |
| } |
| // fall back to super |
| super.leastRestrictive(types) |
| } |
| |
| private def resolveAllIdenticalTypes(types: util.List[RelDataType]): Option[RelDataType] = { |
| val allTypes = types.asScala |
| |
| val head = allTypes.head |
| // check if all types are the same |
| if (allTypes.forall(_ == head)) { |
| // types are the same, check nullability |
| val nullable = allTypes |
| .exists(sqlType => sqlType.isNullable || sqlType.getSqlTypeName == SqlTypeName.NULL) |
| // return type with nullability |
| Some(createTypeWithNullability(head, nullable)) |
| } else { |
| // types are not all the same |
| if (allTypes.exists(_.getSqlTypeName == SqlTypeName.ANY)) { |
| // one of the type was ANY. |
| // we cannot generate a common type if it differs from other types. |
| throw new TableException("Generic ANY types must have a common type information.") |
| } else { |
| // cannot resolve a common type for different input types |
| None |
| } |
| } |
| } |
| |
| // Calcite's default impl for division is apparently borrowed from T-SQL, |
| // but the details are a little different, e.g. when Decimal(34,0)/Decimal(10,0) |
| // To avoid confusion, follow the exact T-SQL behavior. |
| // Note that for (+-*), Calcite is also different from T-SQL; |
| // however, Calcite conforms to SQL2003 while T-SQL does not. |
| // therefore we keep Calcite's behavior on (+-*) |
| override def createDecimalQuotient(type1: RelDataType, type2: RelDataType): RelDataType = { |
| if (SqlTypeUtil.isExactNumeric(type1) && SqlTypeUtil.isExactNumeric(type2) && |
| (SqlTypeUtil.isDecimal(type1) || SqlTypeUtil.isDecimal(type2))) { |
| val result = Decimal.inferDivisionType( |
| type1.getPrecision, type1.getScale, |
| type2.getPrecision, type2.getScale) |
| createSqlType(SqlTypeName.DECIMAL, result.precision, result.scale) |
| } |
| else { |
| null |
| } |
| } |
| |
| } |
| |
| 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 |
| case _: BigDecimalTypeInfo => DECIMAL |
| |
| // 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") |
| } |
| |
| /** |
| * Converts a Calcite logical record into a Flink type information. |
| */ |
| @deprecated("Use the RowSchema class instead because it handles both logical and physical rows.") |
| def toInternalRowTypeInfo(logicalRowType: RelDataType): TypeInformation[Row] = { |
| // convert to type information |
| val logicalFieldTypes = logicalRowType.getFieldList.asScala map { relDataType => |
| FlinkTypeFactory.toTypeInfo(relDataType.getType) |
| } |
| // field names |
| val logicalFieldNames = logicalRowType.getFieldNames.asScala |
| new RowTypeInfo(logicalFieldTypes.toArray, logicalFieldNames.toArray) |
| } |
| |
| /** |
| * Converts a Calcite logical record into a Flink BaseRow information. |
| */ |
| def toInternalBaseRowTypeInfo(logicalRowType: RelDataType): BaseRowTypeInfo = { |
| // convert to type information |
| val logicalFieldTypes = logicalRowType.getFieldList.asScala map { relDataType => |
| FlinkTypeFactory.toTypeInfo(relDataType.getType) |
| } |
| // field names |
| val logicalFieldNames = logicalRowType.getFieldNames.asScala |
| new BaseRowTypeInfo(logicalFieldTypes.toArray, logicalFieldNames.toArray) |
| } |
| |
| def toInternalRowType(logicalRowType: RelDataType): RowType = { |
| // convert to type information |
| val logicalFieldTypes = logicalRowType.getFieldList.asScala map { relDataType => |
| FlinkTypeFactory.toInternalType(relDataType.getType) |
| } |
| // field names |
| val logicalFieldNames = logicalRowType.getFieldNames.asScala |
| new RowType(logicalFieldTypes.toArray[DataType], logicalFieldNames.toArray) |
| } |
| |
| def toInternalFieldTypes(logicalRowType: RelDataType): Seq[InternalType] = { |
| logicalRowType.getFieldList.asScala map { relDataType => |
| FlinkTypeFactory.toInternalType(relDataType.getType) |
| } |
| } |
| |
| def newBaseRowTypeInfo( |
| types: Array[TypeInformation[_]]): BaseRowTypeInfo = { |
| new BaseRowTypeInfo(types: _*) |
| } |
| |
| def isProctimeIndicatorType(relDataType: RelDataType): Boolean = relDataType match { |
| case ti: TimeIndicatorRelDataType if !ti.isEventTime => true |
| case _ => false |
| } |
| |
| def isProctimeIndicatorType(typeInfo: TypeInformation[_]): Boolean = typeInfo match { |
| case ti: TimeIndicatorTypeInfo if !ti.isEventTime => true |
| case _ => false |
| } |
| |
| def isProctimeIndicatorType(dataType: DataType): Boolean = { |
| isProctimeIndicatorType(TypeConverters.createExternalTypeInfoFromDataType(dataType)) |
| } |
| |
| def isRowtimeIndicatorType(relDataType: RelDataType): Boolean = relDataType match { |
| case ti: TimeIndicatorRelDataType if ti.isEventTime => true |
| case _ => false |
| } |
| |
| def isRowtimeIndicatorType(typeInfo: TypeInformation[_]): Boolean = typeInfo match { |
| case ti: TimeIndicatorTypeInfo if ti.isEventTime => true |
| case _ => false |
| } |
| |
| def isRowtimeIndicatorType(dataType: DataType): Boolean = { |
| isRowtimeIndicatorType(dataType) |
| } |
| |
| def isTimeIndicatorType(relDataType: RelDataType): Boolean = relDataType match { |
| case _: TimeIndicatorRelDataType => true |
| case _ => false |
| } |
| |
| def isTimeIndicatorType(typeInfo: TypeInformation[_]): Boolean = typeInfo match { |
| case _: TimeIndicatorTypeInfo => true |
| case _ => false |
| } |
| |
| def isTimeIndicatorType(t: InternalType): Boolean = t match { |
| case DataTypes.ROWTIME_INDICATOR | DataTypes.PROCTIME_INDICATOR => true |
| case _ => false |
| } |
| |
| def toTypeInfo(relDataType: RelDataType): TypeInformation[_] = relDataType.getSqlTypeName match { |
| case BOOLEAN => BOOLEAN_TYPE_INFO |
| case TINYINT => BYTE_TYPE_INFO |
| case SMALLINT => SHORT_TYPE_INFO |
| case INTEGER => INT_TYPE_INFO |
| case BIGINT => LONG_TYPE_INFO |
| case FLOAT => FLOAT_TYPE_INFO |
| case DOUBLE => DOUBLE_TYPE_INFO |
| case VARCHAR | CHAR => STRING_TYPE_INFO |
| case DECIMAL => BigDecimalTypeInfo.of(relDataType.getPrecision, relDataType.getScale) |
| |
| // time indicators |
| case TIMESTAMP if relDataType.isInstanceOf[TimeIndicatorRelDataType] => |
| val indicator = relDataType.asInstanceOf[TimeIndicatorRelDataType] |
| if (indicator.isEventTime) { |
| TimeIndicatorTypeInfo.ROWTIME_INDICATOR |
| } else { |
| TimeIndicatorTypeInfo.PROCTIME_INDICATOR |
| } |
| |
| // temporal types |
| case DATE => SqlTimeTypeInfo.DATE |
| case TIME => SqlTimeTypeInfo.TIME |
| case TIMESTAMP => SqlTimeTypeInfo.TIMESTAMP |
| case typeName if YEAR_INTERVAL_TYPES.contains(typeName) => TimeIntervalTypeInfo.INTERVAL_MONTHS |
| case typeName if DAY_INTERVAL_TYPES.contains(typeName) => TimeIntervalTypeInfo.INTERVAL_MILLIS |
| |
| 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 => INT_TYPE_INFO |
| |
| // extract encapsulated TypeInformation |
| case ANY if relDataType.isInstanceOf[GenericRelDataType] => |
| val genericRelDataType = relDataType.asInstanceOf[GenericRelDataType] |
| genericRelDataType.typeInfo |
| |
| case ROW if relDataType.isInstanceOf[CompositeRelDataType] => |
| val compositeRelDataType = relDataType.asInstanceOf[CompositeRelDataType] |
| compositeRelDataType.compositeType |
| |
| case ROW if relDataType.isInstanceOf[RelRecordType] => |
| val relRecordType = relDataType.asInstanceOf[RelRecordType] |
| new BaseRowSchema(relRecordType).typeInfo() |
| |
| // CURSOR for UDTF case, whose type info will never be used, just a placeholder |
| case CURSOR => new NothingTypeInfo |
| |
| case ARRAY if relDataType.isInstanceOf[ArrayRelDataType] => |
| val arrayRelDataType = relDataType.asInstanceOf[ArrayRelDataType] |
| arrayRelDataType.typeInfo |
| |
| case MAP if relDataType.isInstanceOf[MapRelDataType] => |
| val mapRelDataType = relDataType.asInstanceOf[MapRelDataType] |
| mapRelDataType.typeInfo |
| |
| case VARBINARY | BINARY => PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO |
| |
| case MULTISET if relDataType.isInstanceOf[MultisetRelDataType] => |
| val multisetRelDataType = relDataType.asInstanceOf[MultisetRelDataType] |
| multisetRelDataType.typeInfo |
| |
| case _@t => |
| throw new TableException(s"Type is not supported: $t") |
| } |
| |
| def toInternalType(relDataType: RelDataType): InternalType = |
| TypeConverters.createInternalTypeFromTypeInfo(FlinkTypeFactory.toTypeInfo(relDataType)) |
| |
| def toDataType(relDataType: RelDataType): DataType = FlinkTypeFactory.toTypeInfo(relDataType) |
| } |
