[FLINK-17013][python] Support Python UDTF in old planner under batch mode #11668

HuangXingBo · 2020-04-08T08:55:51Z

What is the purpose of the change

This pull request will support Python UDTF in old planner under batch mode

Brief change log

Add rules to convert RelNode to DataSetPythonCorrelate Node.
Add PythonTableFunctionFlatMap function to invoke Python TableFunctions for the legacy planner

Verifying this change

This change added tests and can be verified as follows:

Added integration tests PyFlinkBatchUserDefinedTableFunctionTests in test_udtf.py

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, Kubernetes/Yarn/Mesos, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

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

flinkbot · 2020-04-08T08:58:21Z

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 48fe02a (Wed Apr 08 08:58:21 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-04-08T09:11:04Z

CI report:

5f203f2 Travis: SUCCESS Azure: CANCELED

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. Some feedback about the code duplication problem. Other parts looks good overall. Besides, it would be great if you can rebase to the master. I will take a closer look then. Thank you

hequn8128 · 2020-04-16T05:49:18Z

flink-python/pyflink/table/tests/test_udtf.py

@@ -81,6 +81,42 @@ class PyFlinkBlinkStreamUserDefinedFunctionTests(UserDefinedTableFunctionTests,
    pass


+class PyFlinkBatchUserDefinedTableFunctionTests(PyFlinkBatchTableTestCase):
+
+    def test_table_function(self):


I'm thinking to avoid these code duplications. Most parts of the code are the same, we only have to extract the different parts into methods and override the methods in the child classes. For example, we can add a base method named get_output in the UserDefinedTableFunctionTests and override the method in PyFlinkBatchUserDefinedTableFunctionTests.

hequn8128 · 2020-04-16T06:31:03Z

...lanner/src/main/java/org/apache/flink/table/plan/rules/batch/DataSetPythonCorrelateRule.java

+ * The physical rule is responsible for convert {@link FlinkLogicalCorrelate} to
+ * {@link DataSetPythonCorrelate}.
+ */
+public class DataSetPythonCorrelateRule extends ConverterRule {


Maybe add some base classes for the two classes, i.e., DataSetPythonCorrelateRule and DataStreamPythonCorrelateRule to avoid the code duplications.

…mode

HuangXingBo · 2020-04-17T11:50:18Z

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 and in a good shape now.

hequn8128 · 2020-04-20T07:25:10Z

flink-python/pyflink/table/tests/test_udtf.py

+    def _register_table_sink(self, field_names: list, field_types: list):
+        table_sink = source_sink_utils.TestAppendSink(field_names, field_types)
+        self.t_env.register_table_sink("Results", table_sink)
+
+    def _get_output(self, t):
+        t.insert_into("Results")
+        self.t_env.execute("test")
+        return source_sink_utils.results()


How about making this two as the default implementation in the base class and only override them in the PyFlinkBatchUserDefinedTableFunctionTests?

hequn8128 · 2020-04-20T07:33:45Z

...planner/src/main/java/org/apache/flink/table/plan/rules/AbstractPythonCorrelateRuleBase.java

@@ -39,16 +40,14 @@
 import scala.Some;

 /**
- * The physical rule is responsible for convert {@link FlinkLogicalCorrelate} to
- * {@link DataStreamPythonCorrelate}.
+ * The abstract physical rule base is responsible for convert {@link FlinkLogicalCorrelate} to physical


responsible for converting

hequn8128 · 2020-04-20T07:34:06Z

...planner/src/main/java/org/apache/flink/table/plan/rules/AbstractPythonCorrelateRuleBase.java

 	/**
-	 * The factory is responsible to creating {@link DataStreamPythonCorrelate}.
+	 * The abstract factory is responsible to creating {@link DataSetPythonCorrelate} or {@link DataStreamPythonCorrelate}.


is responsible for creating

hequn8128 · 2020-04-20T07:34:31Z

...er/src/main/java/org/apache/flink/table/plan/rules/stream/DataStreamPythonCorrelateRule.java

+import scala.Option;
+
+/**
+ * The physical rule is responsible for convert {@link FlinkLogicalCorrelate} to


responsible for converting

hequn8128 · 2020-04-20T07:34:46Z

...er/src/main/java/org/apache/flink/table/plan/rules/stream/DataStreamPythonCorrelateRule.java

+	}
+
+	/**
+	 * The factory is responsible to creating {@link DataStreamPythonCorrelate}.


responsible for creating

hequn8128 · 2020-04-20T07:35:09Z

...lanner/src/main/java/org/apache/flink/table/plan/rules/batch/DataSetPythonCorrelateRule.java

+import scala.Option;
+
+/**
+ * The physical rule is responsible for convert {@link FlinkLogicalCorrelate} to


responsible for converting

hequn8128 · 2020-04-20T07:35:20Z

...lanner/src/main/java/org/apache/flink/table/plan/rules/batch/DataSetPythonCorrelateRule.java

+	}
+
+	/**
+	 * The factory is responsible to creating {@link DataSetPythonCorrelate}.


responsible for creating

hequn8128 · 2020-04-20T07:36:14Z

...er/src/main/java/org/apache/flink/table/plan/rules/stream/DataStreamPythonCorrelateRule.java

+	/**
+	 * The factory is responsible to creating {@link DataStreamPythonCorrelate}.
+	 */
+	private static class DataStreamPythonCorrelateFactory extends PythonCorrelateFactoryBase{


Add a blank here, i.e., PythonCorrelateFactoryBase {

hequn8128 · 2020-04-20T07:37:31Z

...rc/main/java/org/apache/flink/table/runtime/functions/python/PythonTableFunctionFlatMap.java

+
+	@Override
+	public void bufferInput(Row input) {
+		// always copy the input Row


Maybe more details that why we always copy the input Row. What do you think?

HuangXingBo · 2020-04-20T12:23:50Z

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

hequn8128 · 2020-04-21T05:00:01Z

@HuangXingBo Thanks a lot for the update. Merging...

rmetzger added the review=description? label Apr 8, 2020

hequn8128 self-assigned this Apr 8, 2020

rmetzger added the component=API/Python label Apr 8, 2020

hequn8128 requested changes Apr 16, 2020

View reviewed changes

HuangXingBo added 2 commits April 16, 2020 15:33

[FLINK-17013][python] Support Python UDTF in old planner under batch …

65362f1

…mode

fix

7d2e31f

HuangXingBo force-pushed the FLINK-17013 branch from 48fe02a to 7d2e31f Compare April 17, 2020 11:43

hequn8128 requested changes Apr 20, 2020

View reviewed changes

fix-2

5f203f2

hequn8128 merged commit bda9b77 into apache:master Apr 21, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-17013][python] Support Python UDTF in old planner under batch mode #11668

[FLINK-17013][python] Support Python UDTF in old planner under batch mode #11668

HuangXingBo commented Apr 8, 2020

flinkbot commented Apr 8, 2020

flinkbot commented Apr 8, 2020 •

edited

Loading

hequn8128 left a comment •

edited

Loading

hequn8128 Apr 16, 2020

hequn8128 Apr 16, 2020

HuangXingBo commented Apr 17, 2020

hequn8128 left a comment

hequn8128 Apr 20, 2020

hequn8128 Apr 20, 2020

hequn8128 Apr 20, 2020

hequn8128 Apr 20, 2020

hequn8128 Apr 20, 2020

hequn8128 Apr 20, 2020

hequn8128 Apr 20, 2020

hequn8128 Apr 20, 2020

hequn8128 Apr 20, 2020

HuangXingBo commented Apr 20, 2020

hequn8128 commented Apr 21, 2020

[FLINK-17013][python] Support Python UDTF in old planner under batch mode #11668

[FLINK-17013][python] Support Python UDTF in old planner under batch mode #11668

Conversation

HuangXingBo commented Apr 8, 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 Apr 8, 2020

Automated Checks

Review Progress

flinkbot commented Apr 8, 2020 • edited Loading

CI report:

hequn8128 left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HuangXingBo commented Apr 17, 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HuangXingBo commented Apr 20, 2020

hequn8128 commented Apr 21, 2020

flinkbot commented Apr 8, 2020 •

edited

Loading

hequn8128 left a comment •

edited

Loading