hequn8128 commented on a change in pull request #13066: URL: https://github.com/apache/flink/pull/13066#discussion_r466100622
########## File path: flink-python/src/main/java/org/apache/flink/datastream/runtime/functions/python/DataStreamPythonFunction.java ########## @@ -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.datastream.runtime.functions.python; + +import org.apache.flink.table.functions.python.PythonEnv; +import org.apache.flink.table.functions.python.PythonFunction; + +/** + * DataStreamPythonFunction maintain the serialized python function and its function type, which will be used in Review comment: DataStreamPythonFunction => {@link DataStreamPythonFunction}. Same for other places. ########## File path: flink-python/src/main/java/org/apache/flink/datastream/runtime/runners/python/beam/BeamDataStreamPythonStatelessFunctionRunner.java ########## @@ -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.datastream.runtime.runners.python.beam; + +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.datastream.runtime.typeutils.python.PythonTypeUtils; +import org.apache.flink.fnexecution.v1.FlinkFnApi; +import org.apache.flink.python.env.PythonEnvironmentManager; +import org.apache.flink.python.metric.FlinkMetricContainer; +import org.apache.flink.streaming.api.runners.python.beam.BeamPythonStatelessFunctionRunner; + +import org.apache.beam.model.pipeline.v1.RunnerApi; + +import javax.annotation.Nullable; + +import java.util.Map; + +/** + * DataStreamPythonFunctionRunner is responsible for starting a beam python harness to execute user defined python + * function. + */ +public class BeamDataStreamPythonStatelessFunctionRunner extends BeamPythonStatelessFunctionRunner { Review comment: private static final long serialVersionUID = 1L; ########## File path: flink-python/src/main/java/org/apache/flink/datastream/runtime/functions/python/DataStreamPythonFunction.java ########## @@ -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.datastream.runtime.functions.python; + +import org.apache.flink.table.functions.python.PythonEnv; +import org.apache.flink.table.functions.python.PythonFunction; + +/** + * DataStreamPythonFunction maintain the serialized python function and its function type, which will be used in Review comment: maintains ########## File path: flink-python/src/main/java/org/apache/flink/datastream/runtime/operators/python/DataStreamPythonStatelessFunctionOperator.java ########## @@ -0,0 +1,154 @@ +package org.apache.flink.datastream.runtime.operators.python; + +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.memory.ByteArrayInputStreamWithPos; +import org.apache.flink.core.memory.ByteArrayOutputStreamWithPos; +import org.apache.flink.core.memory.DataInputViewStreamWrapper; +import org.apache.flink.core.memory.DataOutputViewStreamWrapper; +import org.apache.flink.datastream.runtime.functions.python.DataStreamPythonFunctionInfo; +import org.apache.flink.datastream.runtime.runners.python.beam.BeamDataStreamPythonStatelessFunctionRunner; +import org.apache.flink.datastream.runtime.typeutils.python.PythonTypeUtils; +import org.apache.flink.fnexecution.v1.FlinkFnApi; +import org.apache.flink.python.PythonFunctionRunner; +import org.apache.flink.streaming.api.operators.python.AbstractPythonFunctionOperator; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.table.functions.python.PythonEnv; +import org.apache.flink.table.runtime.util.StreamRecordCollector; + +import com.google.protobuf.ByteString; + +import java.util.Map; +import java.util.concurrent.LinkedBlockingQueue; + +/** + * DataStreamPythonFunctionOperator is responsible for launching beam runner which will start a python harness to Review comment: DataStreamPythonFunctionOperator => {@link DataStreamPythonFunctionOperator} ########## File path: flink-python/src/main/java/org/apache/flink/datastream/runtime/operators/python/DataStreamPythonStatelessFunctionOperator.java ########## @@ -0,0 +1,154 @@ +package org.apache.flink.datastream.runtime.operators.python; + +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.memory.ByteArrayInputStreamWithPos; +import org.apache.flink.core.memory.ByteArrayOutputStreamWithPos; +import org.apache.flink.core.memory.DataInputViewStreamWrapper; +import org.apache.flink.core.memory.DataOutputViewStreamWrapper; +import org.apache.flink.datastream.runtime.functions.python.DataStreamPythonFunctionInfo; +import org.apache.flink.datastream.runtime.runners.python.beam.BeamDataStreamPythonStatelessFunctionRunner; +import org.apache.flink.datastream.runtime.typeutils.python.PythonTypeUtils; +import org.apache.flink.fnexecution.v1.FlinkFnApi; +import org.apache.flink.python.PythonFunctionRunner; +import org.apache.flink.streaming.api.operators.python.AbstractPythonFunctionOperator; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.table.functions.python.PythonEnv; +import org.apache.flink.table.runtime.util.StreamRecordCollector; + +import com.google.protobuf.ByteString; + +import java.util.Map; +import java.util.concurrent.LinkedBlockingQueue; + +/** + * DataStreamPythonFunctionOperator is responsible for launching beam runner which will start a python harness to + * execute user defined python function. + */ +public class DataStreamPythonStatelessFunctionOperator<IN, OUT> extends AbstractPythonFunctionOperator<IN, OUT> { + + private static final long serialVersionUID = 1L; + + private static final String DATA_STREAM_STATELESS_PYTHON_FUNCTION_URN = "flink:transform:datastream_stateless_function:v1"; + private static final String DATA_STREAM_MAP_FUNCTION_CODER_URN = "flink:coder:datastream:map_function:v1"; + private static final String DATA_STREAM_FLAT_MAP_FUNCTION_CODER_URN = "flink:coder:datastream:flatmap_function:v1"; + + + protected final DataStreamPythonFunctionInfo pythonFunctionInfo; + + private final TypeInformation<IN> inputTypeInfo; + + private final TypeInformation<OUT> outputTypeInfo; + + private final Map<String, String> jobOptions; + + private transient TypeSerializer<IN> inputTypeSerializer; + + private transient TypeSerializer<OUT> outputTypeSerializer; + + protected transient LinkedBlockingQueue<byte[]> userDefinedFunctionResultQueue; Review comment: Please remove these unnecessary variables. ########## File path: flink-python/src/main/java/org/apache/flink/python/util/PythonConfigUtil.java ########## @@ -0,0 +1,66 @@ +/* + * 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.python.util; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; + +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; + +/** + * A Util class to get the StreamExecutionEnvironment configuration and merged configuration with environment settings. Review comment: StreamExecutionEnvironment => {@link StreamExecutionEnvironment} ########## File path: flink-python/src/main/java/org/apache/flink/datastream/runtime/typeutils/python/PythonTypeUtils.java ########## @@ -0,0 +1,310 @@ +/* + * 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.datastream.runtime.typeutils.python; + +import org.apache.flink.api.common.typeinfo.BasicTypeInfo; +import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.base.BigIntSerializer; +import org.apache.flink.api.common.typeutils.base.BooleanSerializer; +import org.apache.flink.api.common.typeutils.base.CharSerializer; +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.InstantSerializer; +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.VoidSerializer; +import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer; +import org.apache.flink.api.java.tuple.Tuple; +import org.apache.flink.api.java.typeutils.RowTypeInfo; +import org.apache.flink.api.java.typeutils.TupleTypeInfo; +import org.apache.flink.api.java.typeutils.runtime.RowSerializer; +import org.apache.flink.api.java.typeutils.runtime.TupleSerializer; +import org.apache.flink.datastream.typeinfo.python.PickledByteArrayTypeInfo; +import org.apache.flink.fnexecution.v1.FlinkFnApi; +import org.apache.flink.table.runtime.typeutils.serializers.python.BigDecSerializer; +import org.apache.flink.table.runtime.typeutils.serializers.python.DateSerializer; +import org.apache.flink.table.runtime.typeutils.serializers.python.StringSerializer; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; + +/** + * A util class for converting the given TypeInformation to other objects. + */ +public class PythonTypeUtils { Review comment: Add tests for this class. ########## File path: flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamPythonStatelessFunctionRunner.java ########## @@ -170,34 +150,11 @@ public ExecutableStage createExecutableStage() throws Exception { .build(); } - /** - * Gets the proto representation of the input coder. - */ - private RunnerApi.Coder getInputCoderProto() { - return getRowCoderProto(inputType); - } - - /** - * Gets the proto representation of the output coder. - */ - private RunnerApi.Coder getOutputCoderProto() { - return getRowCoderProto(outputType); - } + protected abstract byte[] getUserDefinedFunctionsProtoBytes(); Review comment: Keep one blank between protected and abstract. ########## File path: flink-python/src/main/java/org/apache/flink/datastream/runtime/functions/python/DataStreamPythonFunction.java ########## @@ -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.datastream.runtime.functions.python; + +import org.apache.flink.table.functions.python.PythonEnv; +import org.apache.flink.table.functions.python.PythonFunction; + +/** + * DataStreamPythonFunction maintain the serialized python function and its function type, which will be used in + * DataStreamPythonFunctionRunner. + */ +public class DataStreamPythonFunction implements PythonFunction { Review comment: private static final long serialVersionUID = 1L; ########## File path: flink-python/src/main/java/org/apache/flink/datastream/runtime/operators/python/DataStreamPythonStatelessFunctionOperator.java ########## @@ -0,0 +1,154 @@ +package org.apache.flink.datastream.runtime.operators.python; Review comment: Add license text for this class. Please check the test failures. ########## File path: flink-python/src/main/java/org/apache/flink/datastream/runtime/functions/python/DataStreamPythonFunctionInfo.java ########## @@ -0,0 +1,46 @@ +/* + * 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.datastream.runtime.functions.python; + +import org.apache.flink.table.functions.python.PythonFunction; + +import java.io.Serializable; + +/** + * DataStreamPythonFunctionInfo holds a PythonFunction and its function type. + * */ +public class DataStreamPythonFunctionInfo implements Serializable { + private static final long serialVersionUID = 1L; + + private final PythonFunction pythonFunction; + private final int functionType; + + public DataStreamPythonFunctionInfo(PythonFunction pythonFunction, + int functionType) { Review comment: We don't need to separate lines here. ########## File path: flink-python/src/main/java/org/apache/flink/datastream/runtime/functions/python/DataStreamPythonFunctionInfo.java ########## @@ -0,0 +1,46 @@ +/* + * 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.datastream.runtime.functions.python; + +import org.apache.flink.table.functions.python.PythonFunction; + +import java.io.Serializable; + +/** + * DataStreamPythonFunctionInfo holds a PythonFunction and its function type. Review comment: DataStreamPythonFunctionInfo => {@link DataStreamPythonFunctionInfo} ########## File path: flink-python/src/main/java/org/apache/flink/datastream/runtime/typeutils/python/PythonTypeUtils.java ########## @@ -0,0 +1,310 @@ +/* + * 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.datastream.runtime.typeutils.python; + +import org.apache.flink.api.common.typeinfo.BasicTypeInfo; +import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.base.BigIntSerializer; +import org.apache.flink.api.common.typeutils.base.BooleanSerializer; +import org.apache.flink.api.common.typeutils.base.CharSerializer; +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.InstantSerializer; +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.VoidSerializer; +import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer; +import org.apache.flink.api.java.tuple.Tuple; +import org.apache.flink.api.java.typeutils.RowTypeInfo; +import org.apache.flink.api.java.typeutils.TupleTypeInfo; +import org.apache.flink.api.java.typeutils.runtime.RowSerializer; +import org.apache.flink.api.java.typeutils.runtime.TupleSerializer; +import org.apache.flink.datastream.typeinfo.python.PickledByteArrayTypeInfo; +import org.apache.flink.fnexecution.v1.FlinkFnApi; +import org.apache.flink.table.runtime.typeutils.serializers.python.BigDecSerializer; +import org.apache.flink.table.runtime.typeutils.serializers.python.DateSerializer; +import org.apache.flink.table.runtime.typeutils.serializers.python.StringSerializer; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; + +/** + * A util class for converting the given TypeInformation to other objects. + */ +public class PythonTypeUtils { + + /** + * Get coder proto according to the given type information. + */ + public static class TypeInfoToProtoConverter { + + public static FlinkFnApi.TypeInfo.FieldType getFieldType(TypeInformation typeInformation) { + + if (typeInformation instanceof BasicTypeInfo) { + return buildBasicTypeProto((BasicTypeInfo) typeInformation); + } + + if (typeInformation instanceof PrimitiveArrayTypeInfo) { + return buildPrimitiveArrayTypeProto((PrimitiveArrayTypeInfo) typeInformation); + } + + if (typeInformation instanceof RowTypeInfo) { + return buildRowTypeProto((RowTypeInfo) typeInformation); + } + + if (typeInformation instanceof PickledByteArrayTypeInfo) { + return buildPickledBytesTypeProto((PickledByteArrayTypeInfo) typeInformation); + } + + if (typeInformation instanceof TupleTypeInfo) { + return buildTupleTypeProto((TupleTypeInfo) typeInformation); + } + + throw new UnsupportedOperationException( + String.format("The type information: %s is not supported in PyFlink currently.", + typeInformation.toString())); + } + + public static FlinkFnApi.TypeInfo toTypeInfoProto(FlinkFnApi.TypeInfo.FieldType fieldType) { + return FlinkFnApi.TypeInfo.newBuilder().addField(FlinkFnApi.TypeInfo.Field.newBuilder().setType(fieldType).build()).build(); + } + + private static FlinkFnApi.TypeInfo.FieldType buildBasicTypeProto(BasicTypeInfo basicTypeInfo) { + + FlinkFnApi.TypeInfo.TypeName typeName = null; + + if (basicTypeInfo.equals(BasicTypeInfo.BOOLEAN_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.BOOLEAN; + } + + if (basicTypeInfo.equals(BasicTypeInfo.BYTE_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.BYTE; + } + + if (basicTypeInfo.equals(BasicTypeInfo.STRING_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.STRING; + } + + if (basicTypeInfo.equals(BasicTypeInfo.SHORT_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.SHORT; + } + + if (basicTypeInfo.equals(BasicTypeInfo.INT_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.INT; + } + + if (basicTypeInfo.equals(BasicTypeInfo.LONG_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.LONG; + } + + if (basicTypeInfo.equals(BasicTypeInfo.FLOAT_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.FLOAT; + } + + if (basicTypeInfo.equals(BasicTypeInfo.DOUBLE_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.DOUBLE; + } + + if (basicTypeInfo.equals(BasicTypeInfo.CHAR_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.CHAR; + } + + if (basicTypeInfo.equals(BasicTypeInfo.DATE_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.LOCAL_DATE; + } + + if (basicTypeInfo.equals(BasicTypeInfo.VOID_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.VOID; + } + + if (basicTypeInfo.equals(BasicTypeInfo.BIG_INT_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.BIG_INT; + } + + if (basicTypeInfo.equals(BasicTypeInfo.BIG_DEC_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.BIG_DEC; + } + + if (basicTypeInfo.equals(BasicTypeInfo.INSTANT_TYPE_INFO)) { Review comment: Don't have INSTANCE type in datastream ########## File path: flink-python/src/main/java/org/apache/flink/datastream/runtime/runners/python/beam/BeamDataStreamPythonStatelessFunctionRunner.java ########## @@ -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.datastream.runtime.runners.python.beam; + +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.datastream.runtime.typeutils.python.PythonTypeUtils; +import org.apache.flink.fnexecution.v1.FlinkFnApi; +import org.apache.flink.python.env.PythonEnvironmentManager; +import org.apache.flink.python.metric.FlinkMetricContainer; +import org.apache.flink.streaming.api.runners.python.beam.BeamPythonStatelessFunctionRunner; + +import org.apache.beam.model.pipeline.v1.RunnerApi; + +import javax.annotation.Nullable; + +import java.util.Map; + +/** + * DataStreamPythonFunctionRunner is responsible for starting a beam python harness to execute user defined python Review comment: DataStreamPythonFunctionRunner => {@link DataStreamPythonFunctionRunner} ########## File path: flink-python/src/main/java/org/apache/flink/datastream/runtime/typeutils/python/PythonTypeUtils.java ########## @@ -0,0 +1,310 @@ +/* + * 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.datastream.runtime.typeutils.python; + +import org.apache.flink.api.common.typeinfo.BasicTypeInfo; +import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.base.BigIntSerializer; +import org.apache.flink.api.common.typeutils.base.BooleanSerializer; +import org.apache.flink.api.common.typeutils.base.CharSerializer; +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.InstantSerializer; +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.VoidSerializer; +import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer; +import org.apache.flink.api.java.tuple.Tuple; +import org.apache.flink.api.java.typeutils.RowTypeInfo; +import org.apache.flink.api.java.typeutils.TupleTypeInfo; +import org.apache.flink.api.java.typeutils.runtime.RowSerializer; +import org.apache.flink.api.java.typeutils.runtime.TupleSerializer; +import org.apache.flink.datastream.typeinfo.python.PickledByteArrayTypeInfo; +import org.apache.flink.fnexecution.v1.FlinkFnApi; +import org.apache.flink.table.runtime.typeutils.serializers.python.BigDecSerializer; +import org.apache.flink.table.runtime.typeutils.serializers.python.DateSerializer; +import org.apache.flink.table.runtime.typeutils.serializers.python.StringSerializer; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; + +/** + * A util class for converting the given TypeInformation to other objects. + */ +public class PythonTypeUtils { + + /** + * Get coder proto according to the given type information. + */ + public static class TypeInfoToProtoConverter { + + public static FlinkFnApi.TypeInfo.FieldType getFieldType(TypeInformation typeInformation) { + + if (typeInformation instanceof BasicTypeInfo) { + return buildBasicTypeProto((BasicTypeInfo) typeInformation); + } + + if (typeInformation instanceof PrimitiveArrayTypeInfo) { + return buildPrimitiveArrayTypeProto((PrimitiveArrayTypeInfo) typeInformation); + } + + if (typeInformation instanceof RowTypeInfo) { + return buildRowTypeProto((RowTypeInfo) typeInformation); + } + + if (typeInformation instanceof PickledByteArrayTypeInfo) { + return buildPickledBytesTypeProto((PickledByteArrayTypeInfo) typeInformation); + } + + if (typeInformation instanceof TupleTypeInfo) { + return buildTupleTypeProto((TupleTypeInfo) typeInformation); + } + + throw new UnsupportedOperationException( + String.format("The type information: %s is not supported in PyFlink currently.", + typeInformation.toString())); + } + + public static FlinkFnApi.TypeInfo toTypeInfoProto(FlinkFnApi.TypeInfo.FieldType fieldType) { + return FlinkFnApi.TypeInfo.newBuilder().addField(FlinkFnApi.TypeInfo.Field.newBuilder().setType(fieldType).build()).build(); + } + + private static FlinkFnApi.TypeInfo.FieldType buildBasicTypeProto(BasicTypeInfo basicTypeInfo) { + + FlinkFnApi.TypeInfo.TypeName typeName = null; + + if (basicTypeInfo.equals(BasicTypeInfo.BOOLEAN_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.BOOLEAN; + } + + if (basicTypeInfo.equals(BasicTypeInfo.BYTE_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.BYTE; + } + + if (basicTypeInfo.equals(BasicTypeInfo.STRING_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.STRING; + } + + if (basicTypeInfo.equals(BasicTypeInfo.SHORT_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.SHORT; + } + + if (basicTypeInfo.equals(BasicTypeInfo.INT_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.INT; + } + + if (basicTypeInfo.equals(BasicTypeInfo.LONG_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.LONG; + } + + if (basicTypeInfo.equals(BasicTypeInfo.FLOAT_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.FLOAT; + } + + if (basicTypeInfo.equals(BasicTypeInfo.DOUBLE_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.DOUBLE; + } + + if (basicTypeInfo.equals(BasicTypeInfo.CHAR_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.CHAR; + } + + if (basicTypeInfo.equals(BasicTypeInfo.DATE_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.LOCAL_DATE; + } + + if (basicTypeInfo.equals(BasicTypeInfo.VOID_TYPE_INFO)) { Review comment: We don't have VOID type. ########## File path: flink-python/src/main/java/org/apache/flink/table/runtime/runners/python/beam/BeamTablePythonStatelessFunctionRunner.java ########## @@ -0,0 +1,94 @@ +/* + * 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.runners.python.beam; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.fnexecution.v1.FlinkFnApi; +import org.apache.flink.python.env.PythonEnvironmentManager; +import org.apache.flink.python.metric.FlinkMetricContainer; +import org.apache.flink.streaming.api.runners.python.beam.BeamPythonFunctionRunner; +import org.apache.flink.streaming.api.runners.python.beam.BeamPythonStatelessFunctionRunner; +import org.apache.flink.table.runtime.typeutils.PythonTypeUtils; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.util.Preconditions; + +import org.apache.beam.model.pipeline.v1.RunnerApi; + +import java.util.Map; + +/** + * A {@link BeamPythonFunctionRunner} used to execute Python stateless functions. + */ +@Internal +public class BeamTablePythonStatelessFunctionRunner extends BeamPythonStatelessFunctionRunner { + + Review comment: private static final long serialVersionUID = 1L; ########## File path: flink-python/src/main/java/org/apache/flink/python/util/PythonConfigUtil.java ########## @@ -0,0 +1,66 @@ +/* + * 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.python.util; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; + +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; + +/** + * A Util class to get the StreamExecutionEnvironment configuration and merged configuration with environment settings. + */ +public class PythonConfigUtil { + + /** + * A static method to get the StreamExecutionEnvironment configuration merged with python dependency management + * configurations. + */ + public static Configuration getMergedConfig(StreamExecutionEnvironment env) throws InvocationTargetException, + IllegalAccessException, NoSuchMethodException { + Configuration envConfiguration = getEnvironmentConfig(env); + Configuration config = PythonDependencyUtils.configurePythonDependencies(env.getCachedFiles(), envConfiguration); + return config; + } + + /** + * Get the private method StreamExecutionEnvironment.getConfiguration() by reflection recursively. Then access the Review comment: {@link StreamExecutionEnvironment#getConfiguration()} ########## File path: flink-python/src/main/java/org/apache/flink/datastream/runtime/typeutils/python/PythonTypeUtils.java ########## @@ -0,0 +1,310 @@ +/* + * 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.datastream.runtime.typeutils.python; + +import org.apache.flink.api.common.typeinfo.BasicTypeInfo; +import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.base.BigIntSerializer; +import org.apache.flink.api.common.typeutils.base.BooleanSerializer; +import org.apache.flink.api.common.typeutils.base.CharSerializer; +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.InstantSerializer; +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.VoidSerializer; +import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer; +import org.apache.flink.api.java.tuple.Tuple; +import org.apache.flink.api.java.typeutils.RowTypeInfo; +import org.apache.flink.api.java.typeutils.TupleTypeInfo; +import org.apache.flink.api.java.typeutils.runtime.RowSerializer; +import org.apache.flink.api.java.typeutils.runtime.TupleSerializer; +import org.apache.flink.datastream.typeinfo.python.PickledByteArrayTypeInfo; +import org.apache.flink.fnexecution.v1.FlinkFnApi; +import org.apache.flink.table.runtime.typeutils.serializers.python.BigDecSerializer; +import org.apache.flink.table.runtime.typeutils.serializers.python.DateSerializer; +import org.apache.flink.table.runtime.typeutils.serializers.python.StringSerializer; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; + +/** + * A util class for converting the given TypeInformation to other objects. + */ +public class PythonTypeUtils { + + /** + * Get coder proto according to the given type information. + */ + public static class TypeInfoToProtoConverter { + + public static FlinkFnApi.TypeInfo.FieldType getFieldType(TypeInformation typeInformation) { + + if (typeInformation instanceof BasicTypeInfo) { + return buildBasicTypeProto((BasicTypeInfo) typeInformation); + } + + if (typeInformation instanceof PrimitiveArrayTypeInfo) { + return buildPrimitiveArrayTypeProto((PrimitiveArrayTypeInfo) typeInformation); + } + + if (typeInformation instanceof RowTypeInfo) { + return buildRowTypeProto((RowTypeInfo) typeInformation); + } + + if (typeInformation instanceof PickledByteArrayTypeInfo) { + return buildPickledBytesTypeProto((PickledByteArrayTypeInfo) typeInformation); + } + + if (typeInformation instanceof TupleTypeInfo) { + return buildTupleTypeProto((TupleTypeInfo) typeInformation); + } + + throw new UnsupportedOperationException( + String.format("The type information: %s is not supported in PyFlink currently.", + typeInformation.toString())); + } + + public static FlinkFnApi.TypeInfo toTypeInfoProto(FlinkFnApi.TypeInfo.FieldType fieldType) { + return FlinkFnApi.TypeInfo.newBuilder().addField(FlinkFnApi.TypeInfo.Field.newBuilder().setType(fieldType).build()).build(); + } + + private static FlinkFnApi.TypeInfo.FieldType buildBasicTypeProto(BasicTypeInfo basicTypeInfo) { + + FlinkFnApi.TypeInfo.TypeName typeName = null; + + if (basicTypeInfo.equals(BasicTypeInfo.BOOLEAN_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.BOOLEAN; + } + + if (basicTypeInfo.equals(BasicTypeInfo.BYTE_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.BYTE; + } + + if (basicTypeInfo.equals(BasicTypeInfo.STRING_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.STRING; + } + + if (basicTypeInfo.equals(BasicTypeInfo.SHORT_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.SHORT; + } + + if (basicTypeInfo.equals(BasicTypeInfo.INT_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.INT; + } + + if (basicTypeInfo.equals(BasicTypeInfo.LONG_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.LONG; + } + + if (basicTypeInfo.equals(BasicTypeInfo.FLOAT_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.FLOAT; + } + + if (basicTypeInfo.equals(BasicTypeInfo.DOUBLE_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.DOUBLE; + } + + if (basicTypeInfo.equals(BasicTypeInfo.CHAR_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.CHAR; + } + + if (basicTypeInfo.equals(BasicTypeInfo.DATE_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.LOCAL_DATE; + } + + if (basicTypeInfo.equals(BasicTypeInfo.VOID_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.VOID; + } + + if (basicTypeInfo.equals(BasicTypeInfo.BIG_INT_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.BIG_INT; + } + + if (basicTypeInfo.equals(BasicTypeInfo.BIG_DEC_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.BIG_DEC; + } + + if (basicTypeInfo.equals(BasicTypeInfo.INSTANT_TYPE_INFO)) { + typeName = FlinkFnApi.TypeInfo.TypeName.INSTANT; + } + + if (typeName == null) { + throw new UnsupportedOperationException( + String.format("The BasicTypeInfo: %s is not supported in PyFlink currently.", + basicTypeInfo.toString())); + } + + return FlinkFnApi.TypeInfo.FieldType.newBuilder() + .setTypeName(typeName).build(); + } + + private static FlinkFnApi.TypeInfo.FieldType buildPrimitiveArrayTypeProto( + PrimitiveArrayTypeInfo primitiveArrayTypeInfo) { + FlinkFnApi.TypeInfo.FieldType elementFieldType = null; + if (primitiveArrayTypeInfo.equals(PrimitiveArrayTypeInfo.BOOLEAN_PRIMITIVE_ARRAY_TYPE_INFO)) { + elementFieldType = buildBasicTypeProto(BasicTypeInfo.BOOLEAN_TYPE_INFO); + } + + if (primitiveArrayTypeInfo.equals(PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO)) { + elementFieldType = buildBasicTypeProto(BasicTypeInfo.BYTE_TYPE_INFO); + } + + if (primitiveArrayTypeInfo.equals(PrimitiveArrayTypeInfo.SHORT_PRIMITIVE_ARRAY_TYPE_INFO)) { + elementFieldType = buildBasicTypeProto(BasicTypeInfo.SHORT_TYPE_INFO); + } + + if (primitiveArrayTypeInfo.equals(PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO)) { + elementFieldType = buildBasicTypeProto(BasicTypeInfo.INT_TYPE_INFO); + } + + if (primitiveArrayTypeInfo.equals(PrimitiveArrayTypeInfo.LONG_PRIMITIVE_ARRAY_TYPE_INFO)) { + elementFieldType = buildBasicTypeProto(BasicTypeInfo.LONG_TYPE_INFO); + } + + if (primitiveArrayTypeInfo.equals(PrimitiveArrayTypeInfo.FLOAT_PRIMITIVE_ARRAY_TYPE_INFO)) { + elementFieldType = buildBasicTypeProto(BasicTypeInfo.FLOAT_TYPE_INFO); + } + + if (primitiveArrayTypeInfo.equals(PrimitiveArrayTypeInfo.DOUBLE_PRIMITIVE_ARRAY_TYPE_INFO)) { + elementFieldType = buildBasicTypeProto(BasicTypeInfo.DOUBLE_TYPE_INFO); + } + + if (primitiveArrayTypeInfo.equals(PrimitiveArrayTypeInfo.CHAR_PRIMITIVE_ARRAY_TYPE_INFO)) { + elementFieldType = buildBasicTypeProto(BasicTypeInfo.CHAR_TYPE_INFO); + } + + if (elementFieldType == null) { + throw new UnsupportedOperationException( + String.format("The element type of PrimitiveArrayTypeInfo: %s is not supported in PyFlink currently." + , primitiveArrayTypeInfo.toString())); + } + + FlinkFnApi.TypeInfo.FieldType.Builder builder = FlinkFnApi.TypeInfo.FieldType.newBuilder() + .setTypeName(FlinkFnApi.TypeInfo.TypeName.ARRAY); + builder.setCollectionElementType(elementFieldType); + return builder.build(); + } + + private static FlinkFnApi.TypeInfo.FieldType buildRowTypeProto(RowTypeInfo rowTypeInfo) { + FlinkFnApi.TypeInfo.FieldType.Builder builder = + FlinkFnApi.TypeInfo.FieldType.newBuilder() + .setTypeName(FlinkFnApi.TypeInfo.TypeName.ROW); + + FlinkFnApi.TypeInfo.Builder rowTypeInfoBuilder = FlinkFnApi.TypeInfo.newBuilder(); + + int arity = rowTypeInfo.getArity(); + for (int index = 0; index < arity; index++) { + rowTypeInfoBuilder.addField( + FlinkFnApi.TypeInfo.Field.newBuilder() + .setName(rowTypeInfo.getFieldNames()[index]) + .setType(TypeInfoToProtoConverter.getFieldType(rowTypeInfo.getTypeAt(index))) + .build()); + } + builder.setRowTypeInfo(rowTypeInfoBuilder.build()); + return builder.build(); + } + + private static FlinkFnApi.TypeInfo.FieldType buildPickledBytesTypeProto(PickledByteArrayTypeInfo pickledByteArrayTypeInfo) { + return FlinkFnApi.TypeInfo.FieldType.newBuilder() + .setTypeName(FlinkFnApi.TypeInfo.TypeName.PICKLED_BYTES).build(); + } + + private static FlinkFnApi.TypeInfo.FieldType buildTupleTypeProto(TupleTypeInfo tupleTypeInfo) { + FlinkFnApi.TypeInfo.FieldType.Builder builder = + FlinkFnApi.TypeInfo.FieldType.newBuilder() + .setTypeName(FlinkFnApi.TypeInfo.TypeName.TUPLE); + + FlinkFnApi.TypeInfo.Builder tupleTypeInfoBuilder = FlinkFnApi.TypeInfo.newBuilder(); + + int arity = tupleTypeInfo.getArity(); + for (int index = 0; index < arity; index++) { + tupleTypeInfoBuilder.addField( + FlinkFnApi.TypeInfo.Field.newBuilder() + .setName(tupleTypeInfo.getFieldNames()[index]) + .setType(TypeInfoToProtoConverter.getFieldType(tupleTypeInfo.getTypeAt(index))) + .build()); + } + builder.setTupleTypeInfo(tupleTypeInfoBuilder.build()); + return builder.build(); + } + } + + /** + * Get serializers according to the given typeInformation. + */ + public static class TypeInfoToSerializerConverter { + private static final Map<Class, TypeSerializer> typeInfoToSerialzerMap = new HashMap<>(); + + static { + typeInfoToSerialzerMap.put(BasicTypeInfo.BOOLEAN_TYPE_INFO.getTypeClass(), BooleanSerializer.INSTANCE); + typeInfoToSerialzerMap.put(BasicTypeInfo.INT_TYPE_INFO.getTypeClass(), IntSerializer.INSTANCE); + typeInfoToSerialzerMap.put(BasicTypeInfo.STRING_TYPE_INFO.getTypeClass(), StringSerializer.INSTANCE); + typeInfoToSerialzerMap.put(BasicTypeInfo.SHORT_TYPE_INFO.getTypeClass(), ShortSerializer.INSTANCE); + typeInfoToSerialzerMap.put(BasicTypeInfo.LONG_TYPE_INFO.getTypeClass(), LongSerializer.INSTANCE); + typeInfoToSerialzerMap.put(BasicTypeInfo.FLOAT_TYPE_INFO.getTypeClass(), FloatSerializer.INSTANCE); + typeInfoToSerialzerMap.put(BasicTypeInfo.DOUBLE_TYPE_INFO.getTypeClass(), DoubleSerializer.INSTANCE); + typeInfoToSerialzerMap.put(BasicTypeInfo.CHAR_TYPE_INFO.getTypeClass(), CharSerializer.INSTANCE); + typeInfoToSerialzerMap.put(BasicTypeInfo.DATE_TYPE_INFO.getTypeClass(), DateSerializer.INSTANCE); Review comment: Remove void and instance. ########## File path: flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamPythonStatelessFunctionRunner.java ########## @@ -53,8 +48,7 @@ /** * A {@link BeamPythonFunctionRunner} used to execute Python stateless functions. */ -@Internal Review comment: Recovery the change. ########## File path: flink-python/src/main/java/org/apache/flink/python/util/PythonConfigUtil.java ########## @@ -0,0 +1,66 @@ +/* + * 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.python.util; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; + +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; + +/** + * A Util class to get the StreamExecutionEnvironment configuration and merged configuration with environment settings. + */ +public class PythonConfigUtil { Review comment: Add tests for this class. ########## File path: flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughPythonTableFunctionRunner.java ########## @@ -32,10 +32,10 @@ import java.util.Map; /** - * A BeamPythonStatelessFunctionRunner that emit each input element in inner join and emit null in + * A BeamTablePythonStatelessFunctionRunner that emit each input element in inner join and emit null in Review comment: PassThroughPythonTableFunctionRunner ########## File path: flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughPythonScalarFunctionRunner.java ########## @@ -32,9 +32,9 @@ import java.util.Map; /** - * A BeamPythonStatelessFunctionRunner runner that just return the input elements as the execution results. + * A BeamTablePythonStatelessFunctionRunner runner that just return the input elements as the execution results. Review comment: PassThroughPythonScalarFunctionRunner ########## File path: flink-python/src/main/java/org/apache/flink/table/runtime/runners/python/beam/BeamTablePythonStatelessFunctionRunner.java ########## @@ -0,0 +1,94 @@ +/* + * 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.runners.python.beam; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.fnexecution.v1.FlinkFnApi; +import org.apache.flink.python.env.PythonEnvironmentManager; +import org.apache.flink.python.metric.FlinkMetricContainer; +import org.apache.flink.streaming.api.runners.python.beam.BeamPythonFunctionRunner; +import org.apache.flink.streaming.api.runners.python.beam.BeamPythonStatelessFunctionRunner; +import org.apache.flink.table.runtime.typeutils.PythonTypeUtils; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.util.Preconditions; + +import org.apache.beam.model.pipeline.v1.RunnerApi; + +import java.util.Map; + +/** + * A {@link BeamPythonFunctionRunner} used to execute Python stateless functions. Review comment: BeamTablePythonStatelessFunctionRunner ---------------------------------------------------------------- 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: us...@infra.apache.org
