thebalu commented on a change in pull request #78: URL: https://github.com/apache/bahir-flink/pull/78#discussion_r412262673
########## File path: flink-connector-kudu/src/main/java/org/apache/flink/connectors/kudu/table/utils/KuduTypeUtils.java ########## @@ -0,0 +1,152 @@ +/* + * 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.connectors.kudu.table.utils; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.types.AtomicDataType; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.BigIntType; +import org.apache.flink.table.types.logical.BooleanType; +import org.apache.flink.table.types.logical.DecimalType; +import org.apache.flink.table.types.logical.DoubleType; +import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.SmallIntType; +import org.apache.flink.table.types.logical.TimestampType; +import org.apache.flink.table.types.logical.TinyIntType; +import org.apache.flink.table.types.logical.VarBinaryType; +import org.apache.flink.table.types.logical.VarCharType; +import org.apache.flink.table.types.logical.utils.LogicalTypeDefaultVisitor; + +import org.apache.kudu.ColumnTypeAttributes; +import org.apache.kudu.Type; + +import java.sql.Timestamp; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +public class KuduTypeUtils { + + public static DataType toFlinkType(Type type, ColumnTypeAttributes typeAttributes) { + switch (type) { + case STRING: + return DataTypes.STRING(); + case FLOAT: + return DataTypes.FLOAT(); + case INT8: + return DataTypes.TINYINT(); + case INT16: + return DataTypes.SMALLINT(); + case INT32: + return DataTypes.INT(); + case INT64: + return DataTypes.BIGINT(); + case DOUBLE: + return DataTypes.DOUBLE(); + case DECIMAL: + return DataTypes.DECIMAL(typeAttributes.getPrecision(), typeAttributes.getScale()); + case BOOL: + return DataTypes.BOOLEAN(); + case BINARY: + return DataTypes.BYTES(); + case UNIXTIME_MICROS: + return new AtomicDataType(new TimestampType(3), Timestamp.class); + + default: + throw new IllegalArgumentException("Illegal var type: " + type); + } + } + + public static Type toKuduType(DataType dataType) { + checkNotNull(dataType, "type cannot be null"); + LogicalType logicalType = dataType.getLogicalType(); + return logicalType.accept(new KuduTypeLogicalTypeVisitor(dataType)); + } + + private static class KuduTypeLogicalTypeVisitor extends LogicalTypeDefaultVisitor<Type> { + + private final DataType dataType; + + KuduTypeLogicalTypeVisitor(DataType dataType) { + this.dataType = dataType; + } + + @Override + public Type visit(BooleanType booleanType) { + return Type.BOOL; + } + + @Override + public Type visit(TinyIntType tinyIntType) { + return Type.INT8; + } + + @Override + public Type visit(SmallIntType smallIntType) { + return Type.INT16; + } + + @Override + public Type visit(IntType intType) { + return Type.INT32; + } + + @Override + public Type visit(BigIntType bigIntType) { + return Type.INT64; + } + + @Override + public Type visit(FloatType floatType) { + return Type.FLOAT; + } + + @Override + public Type visit(DoubleType doubleType) { + return Type.DOUBLE; + } + + @Override + public Type visit(DecimalType decimalType) { + return Type.DECIMAL; + } + + @Override + public Type visit(TimestampType timestampType) { + return Type.UNIXTIME_MICROS; + } + + @Override + public Type visit(VarCharType varCharType) { Review comment: VarChar seems to be not yet supported by the Blink planner ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected]
