[FLINK-15913][python] Add Python TableFunction Runner and Operator in old planner #11020

HuangXingBo · 2020-02-05T10:05:03Z

What is the purpose of the change

This pull request add Python Table Function Runner and Operator for executing Python user-defined table function in legacy Planner

Brief change log

Add AbstractPythonTableFunctionOperator and PythonTableFunctionOperator
Add AbstractPythonStatelessFunctionRunner
Add AbstractPythonTableFunctionRunner and PythonTableFunctionRunner
TableSerializer and RowTableSerializer
Add Table Type in flink-fn-execution.proto

Verifying this change

This change added tests and can be verified as follows:

Add runner test in PythonTableFunctionRunnerTest
Add operator test in PythonTableFunctionOperatorTestBase

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable)

flinkbot · 2020-02-05T10:08:29Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 024d27c (Wed Feb 05 10:08:29 UTC 2020)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!
This pull request references an unassigned Jira ticket. According to the code contribution guide, tickets need to be assigned before starting with the implementation work.

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2020-02-05T10:41:22Z

CI report:

024d27c Travis: FAILURE Azure: SUCCESS
a2f0e14 Travis: FAILURE Azure: SUCCESS
e2ea604 Travis: FAILURE Azure: FAILURE
116926b Travis: FAILURE Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

hequn8128

@HuangXingBo
Thanks a lot for the PR and the great work! I left some comments below. The most important I think is we need to avoid some code duplications. Besides, some implementation may need to be simplified, such as it would be great if we can avoid adding the TABLE type in flink-fn-execution.proto. More details can be found in the comments below.

Furthermore, it would be great if you can rebase the code to the latest master and solve the conflicts. Thanks.

Best, Hequn

...c/main/java/org/apache/flink/table/runtime/typeutils/serializers/python/TableSerializer.java

flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java

hequn8128 · 2020-02-06T11:22:06Z

flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java

@@ -90,6 +91,39 @@ public static TypeSerializer toBlinkTypeSerializer(LogicalType logicalType) {
 		return logicalType.accept(new LogicalTypeToProtoTypeConverter());
 	}

+	public static TypeSerializer toFlinkTableTypeSerializer(LogicalType logicalType) {
+		RowType rowType = (RowType) logicalType;


It's not good to cast here. It's better to change the method from toFlinkTableTypeSerializer(LogicalType logicalType) to toFlinkTableTypeSerializer(RowType logicalType)

Same for toTableProtoType.

flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java

hequn8128 · 2020-02-06T11:25:02Z

flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java

+		return new RowTableSerializer(fieldTypeSerializers);
+	}
+
+	public static FlinkFnApi.Schema.FieldType toTableProtoType(LogicalType logicalType) {


Rename method name to toTableFunctionProtoType

hequn8128 · 2020-02-06T12:19:44Z

...c/main/java/org/apache/flink/table/runtime/operators/python/PythonTableFunctionOperator.java

+	/**
+	 * The collector is used to convert a {@link Row} to a {@link CRow}.
+	 */
+	private static class StreamRecordCRowWrappingCollector implements Collector<Row> {


This class is copied from PythonScalarFunctionOperator. I think we can do some code reuse.

hequn8128 · 2020-02-06T12:26:57Z

...ava/org/apache/flink/table/runtime/operators/python/AbstractPythonTableFunctionOperator.java

+ * @param <UDTFIN>  Type of the UDF input type.
+ * @param <UDTFOUT> Type of the UDF input type.
+ */
+public abstract class AbstractPythonTableFunctionOperator<IN, OUT, UDTFIN, UDTFOUT>


Please try to avoid code deduplication. For example, we can add a StatelessFunctionOperator and put the same code in this class. What do you think?

hequn8128 · 2020-02-06T12:28:08Z

...ava/org/apache/flink/table/runtime/operators/python/AbstractPythonTableFunctionOperator.java

+ * @param <UDTFIN>  Type of the UDF input type.
+ * @param <UDTFOUT> Type of the UDF input type.
+ */
+public abstract class AbstractPythonTableFunctionOperator<IN, OUT, UDTFIN, UDTFOUT>


Add @Internal

hequn8128 · 2020-02-06T12:28:51Z

...c/main/java/org/apache/flink/table/runtime/operators/python/PythonTableFunctionOperator.java

+/**
+ * The Python {@link TableFunction} operator for the legacy planner.
+ */
+public class PythonTableFunctionOperator extends AbstractPythonTableFunctionOperator<CRow, CRow, Row, Row> {


Add @Internal

hequn8128 · 2020-02-06T12:29:16Z

...in/java/org/apache/flink/table/runtime/runners/python/AbstractPythonTableFunctionRunner.java

+ * @param <IN>  Type of the input elements.
+ * @param <OUT> Type of the execution results.
+ */
+public abstract class AbstractPythonTableFunctionRunner<IN, OUT> extends AbstractPythonStatelessFunctionRunner<IN, OUT> {


Add @Internal

…n Legacy Planner

…n Legacy Planner-fix-1

HuangXingBo · 2020-02-07T08:13:10Z

Thanks a lot for @hequn8128 review, I have addressed the comments at the latest commit.

hequn8128

@HuangXingBo Thanks a lot for the update. The code looks much better now. With some minor comments below.

Best, Hequn

hequn8128 · 2020-02-07T12:34:03Z

flink-python/pyflink/proto/flink-fn-execution.proto

@@ -73,6 +73,7 @@ message Schema {
    ARRAY = 16;
    MAP = 17;
    MULTISET = 18;
+    TABLEFUNCTIONROW = 19;


TABLE_FUNCTION_ROW ?

hequn8128 · 2020-02-07T12:52:22Z

...ava/org/apache/flink/table/runtime/operators/python/AbstractPythonTableFunctionOperator.java

+	}
+
+	/**
+	 * The received udtf execution result is a finish message when it is a byte 0x00.


when it is a byte 0x00 => when it is a byte with value 0x00

hequn8128 · 2020-02-07T13:04:51Z

.../java/org/apache/flink/table/runtime/operators/python/AbstractStatelessFunctionOperator.java

+	protected transient LinkedBlockingQueue<IN> forwardedInputQueue;
+
+	/**
+	 * The queue holding the user-defined table function execution results. The execution results


user-defined table function => user-defined function

hequn8128 · 2020-02-07T13:09:26Z

.../java/org/apache/flink/table/runtime/operators/python/AbstractStatelessFunctionOperator.java

+	/**
+	 * The udf output logical type.
+	 */
+	protected transient RowType udfOutputType;


How about rename udfOutputType to functionOutputType or userDefinedFunctionOutputType? It seems strange to see the udfOutputType variable in AbstractPythonTableFunctionOperator.

Same for other variable names.

hequn8128 · 2020-02-07T13:12:39Z

...c/main/java/org/apache/flink/table/runtime/operators/python/PythonTableFunctionOperator.java

+	private transient TypeSerializer<CRow> forwardedInputSerializer;
+
+	/**
+	 * The TypeSerializer for udf execution results.


udf => udtf

hequn8128 · 2020-02-07T13:13:46Z

...c/main/java/org/apache/flink/table/runtime/operators/python/PythonTableFunctionOperator.java

+	@Override
+	public void bufferInput(CRow input) {
+		if (getExecutionConfig().isObjectReuseEnabled()) {
+			input =  forwardedInputSerializer.copy(input);


Remove the blank after =

…n Legacy Planner-fix-2

HuangXingBo · 2020-02-07T13:44:37Z

Thanks a lot for @hequn8128 review, I have addressed the comments at the latest commit.

hequn8128

LGTM. Merging this once travis passed.

…n Legacy Planner This closes apache#11020.

hequn8128 · 2020-02-10T02:27:59Z

It seems the test failure has nothing to do with this PR. I will merge this.

rmetzger added review=description? component=API/Python labels Feb 5, 2020

HuangXingBo force-pushed the FLINK-15913 branch from 024d27c to a2f0e14 Compare February 5, 2020 12:00

hequn8128 changed the title ~~[FLINK-15913][python] Add Python Table Function Runner And Operator In Legacy Planner~~ [FLINK-15913][python] Add Python TableFunction Runner and Operator in old planner Feb 6, 2020

hequn8128 requested changes Feb 6, 2020

View reviewed changes

HuangXingBo added 2 commits February 7, 2020 10:12

[FLINK-15913][python] Add Python Table Function Runner And Operator I…

b950cfa

…n Legacy Planner

[FLINK-15913][python] Add Python Table Function Runner And Operator I…

e2ea604

…n Legacy Planner-fix-1

HuangXingBo force-pushed the FLINK-15913 branch from a2f0e14 to e2ea604 Compare February 7, 2020 08:08

hequn8128 requested changes Feb 7, 2020

View reviewed changes

[FLINK-15913][python] Add Python Table Function Runner And Operator I…

116926b

…n Legacy Planner-fix-2

hequn8128 approved these changes Feb 7, 2020

View reviewed changes

hequn8128 pushed a commit to hequn8128/flink that referenced this pull request Feb 7, 2020

[FLINK-15913][python] Add Python Table Function Runner And Operator I…

fc82b01

…n Legacy Planner This closes apache#11020.

hequn8128 merged commit fee187e into apache:master Feb 10, 2020

HuangXingBo deleted the FLINK-15913 branch February 10, 2020 02:30

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-15913][python] Add Python TableFunction Runner and Operator in old planner #11020

[FLINK-15913][python] Add Python TableFunction Runner and Operator in old planner #11020

HuangXingBo commented Feb 5, 2020

flinkbot commented Feb 5, 2020

flinkbot commented Feb 5, 2020 •

edited

hequn8128 left a comment •

edited

hequn8128 Feb 6, 2020

hequn8128 Feb 6, 2020

hequn8128 Feb 6, 2020

hequn8128 Feb 6, 2020

hequn8128 Feb 6, 2020

hequn8128 Feb 6, 2020

hequn8128 Feb 6, 2020

HuangXingBo commented Feb 7, 2020

hequn8128 left a comment

hequn8128 Feb 7, 2020

hequn8128 Feb 7, 2020

hequn8128 Feb 7, 2020

hequn8128 Feb 7, 2020

hequn8128 Feb 7, 2020

hequn8128 Feb 7, 2020

HuangXingBo commented Feb 7, 2020

hequn8128 left a comment

hequn8128 commented Feb 10, 2020

[FLINK-15913][python] Add Python TableFunction Runner and Operator in old planner #11020

[FLINK-15913][python] Add Python TableFunction Runner and Operator in old planner #11020

Conversation

HuangXingBo commented Feb 5, 2020

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Feb 5, 2020

Automated Checks

Review Progress

flinkbot commented Feb 5, 2020 • edited

CI report:

hequn8128 left a comment • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HuangXingBo commented Feb 7, 2020

hequn8128 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HuangXingBo commented Feb 7, 2020

hequn8128 left a comment

Choose a reason for hiding this comment

hequn8128 commented Feb 10, 2020

flinkbot commented Feb 5, 2020 •

edited

hequn8128 left a comment •

edited