hequn8128 commented on a change in pull request #9977: [FLINK-14497][python] Support primitive data types in Python user-defined functions URL: https://github.com/apache/flink/pull/9977#discussion_r338925758
########## File path: flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java ########## @@ -0,0 +1,299 @@ +/* + * 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.runtime.typeutils; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.base.BooleanSerializer; +import org.apache.flink.api.common.typeutils.base.ByteSerializer; +import org.apache.flink.api.common.typeutils.base.DoubleSerializer; +import org.apache.flink.api.common.typeutils.base.FloatSerializer; +import org.apache.flink.api.common.typeutils.base.IntSerializer; +import org.apache.flink.api.common.typeutils.base.LongSerializer; +import org.apache.flink.api.common.typeutils.base.ShortSerializer; +import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer; +import org.apache.flink.api.java.typeutils.runtime.RowSerializer; +import org.apache.flink.fnexecution.v1.FlinkFnApi; +import org.apache.flink.table.runtime.typeutils.serializers.BaseRowSerializer; +import org.apache.flink.table.runtime.typeutils.serializers.DateSerializer; +import org.apache.flink.table.runtime.typeutils.serializers.StringSerializer; +import org.apache.flink.table.types.logical.BigIntType; +import org.apache.flink.table.types.logical.BinaryType; +import org.apache.flink.table.types.logical.BooleanType; +import org.apache.flink.table.types.logical.CharType; +import org.apache.flink.table.types.logical.DateType; +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.RowType; +import org.apache.flink.table.types.logical.SmallIntType; +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; + +/** + * Utilities for serializing Flink data. + */ +@Internal +public final class PythonTypeUtils { + + private static final String EMPTY_STRING = ""; + + public static TypeSerializer toFlinkTypeSerializer(LogicalType logicalType) { + return logicalType.accept(new LogicalTypeToTypeSerializerConverter()); + } + + public static TypeSerializer toBlinkTypeSerializer(LogicalType logicalType) { + return logicalType.accept(new LogicalTypeToBlinkTypeSerializerConverter()); + } + + public static FlinkFnApi.Schema.FieldType toProtoType(LogicalType logicalType) { + return logicalType.accept(new LogicalTypeToProtoTypeConverter()); + } + + private static class LogicalTypeToTypeSerializerConverter extends LogicalTypeDefaultVisitor<TypeSerializer> { + @Override + public TypeSerializer visit(BooleanType booleanType) { + return BooleanSerializer.INSTANCE; + } + + @Override + public TypeSerializer visit(TinyIntType tinyIntType) { + return ByteSerializer.INSTANCE; + } + + @Override + public TypeSerializer visit(SmallIntType smallIntType) { + return ShortSerializer.INSTANCE; + } + + @Override + public TypeSerializer visit(IntType intType) { + return IntSerializer.INSTANCE; + } + + @Override + public TypeSerializer visit(BigIntType bigIntType) { + return LongSerializer.INSTANCE; + } + + @Override + public TypeSerializer visit(FloatType floatType) { + return FloatSerializer.INSTANCE; + } + + @Override + public TypeSerializer visit(DoubleType doubleType) { + return DoubleSerializer.INSTANCE; + } + + @Override + public TypeSerializer visit(BinaryType binaryType) { + return BytePrimitiveArraySerializer.INSTANCE; + } + + @Override + public TypeSerializer visit(VarBinaryType varBinaryType) { + return BytePrimitiveArraySerializer.INSTANCE; + } + + @Override + public TypeSerializer visit(VarCharType varCharType) { + return StringSerializer.INSTANCE; + } + + @Override + public TypeSerializer visit(CharType charType) { + return StringSerializer.INSTANCE; + } + + @Override + public TypeSerializer visit(DateType dateType) { + return DateSerializer.INSTANCE; + } + + @Override + public TypeSerializer visit(RowType rowType) { + final TypeSerializer[] fieldTypeSerializers = rowType.getFields() + .stream() + .map(f -> f.getType().accept(this)) + .toArray(TypeSerializer[]::new); + return new RowSerializer(fieldTypeSerializers); + } + + @Override + protected TypeSerializer defaultMethod(LogicalType logicalType) { + throw new UnsupportedOperationException(String.format( Review comment: Add a test in `PythonTypeUtilsTest` to validate the exception, especially validate the error message. ---------------------------------------------------------------- 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] With regards, Apache Git Services
