[FLINK-15961][table-planner][table-planner-blink] Introduce Python Physical Correlate RelNodes which are containers for Python TableFunction #11051

HuangXingBo · 2020-02-10T12:51:46Z

What is the purpose of the change

This pull request introduce Python Physical Correlate RelNodes which are containers for Python TableFunction

Brief change log

Add Physical Python Correlate Rule BatchExecPythonCorrelateRule ,StreamExecPythonCorrelateRule and DataStreamPythonCorrelateRule
Add Physical Python Correlate Node DataStreamPythonCorrelate, StreamExecPythonCorrelate and BatchExecPythonCorrelate

Verifying this change

This change added tests and can be verified as follows:

Added ut case testCorrelatePythonTableFunction

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: (no)

Documentation

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

flinkbot · 2020-02-10T12:54:15Z

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 68b0714 (Mon Feb 10 12:54:15 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.

Details

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-10T13:31:36Z

CI report:

68b0714 Travis: SUCCESS Azure: SUCCESS
9c17e49 Travis: FAILURE Azure: PENDING
67f8b75 Travis: FAILURE Azure: FAILURE
f0fbdc6 Travis: SUCCESS Azure: FAILURE

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 comments below.

hequn8128 · 2020-02-13T12:12:03Z

...er-blink/src/test/scala/org/apache/flink/table/planner/utils/UserDefinedTableFunctions.scala

+    for (i <- 0 until y) {
+      val row = new Row(2)
+      row.setField(0, x)
+      row.setField(1, i * i)
+      collect(row)
+    }


How about remove these lines. The code will never be called.

hequn8128 · 2020-02-13T12:12:51Z

...nk-table-planner/src/test/scala/org/apache/flink/table/utils/UserDefinedTableFunctions.scala

+    for (i <- 0 until y) {
+      val row = new Row(2)
+      row.setField(0, x)
+      row.setField(1, i * i)
+      collect(row)


How about remove these lines?

hequn8128 · 2020-02-13T12:29:43Z

...er/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamCorrelateRule.scala

    right match {
-      // right node is a table function
-      case scan: FlinkLogicalTableFunctionScan => true
+      // right node is a java table function


I don't think the right node must be a java table function. The original comment is ok.

hequn8128 · 2020-02-13T12:35:09Z

...er/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamCorrelateRule.scala

      // a filter is pushed above the table function
-      case calc: FlinkLogicalCalc if CorrelateUtil.getTableFunctionScan(calc).isDefined => true
+      case calc: FlinkLogicalCalc =>
+        PythonUtil.isNonPythonCall(CorrelateUtil.getTableFunctionScan(calc).get.getCall)


The method getTableFunctionScan may returns None.

hequn8128 · 2020-02-13T12:43:20Z

...er/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamCorrelateBase.scala

+    scan: FlinkLogicalTableFunctionScan,
+    condition: Option[RexNode],
+    schema: RowSchema,
+    joinSchema: RowSchema,


Remove this member. It has never been used.

hequn8128 · 2020-02-14T04:04:17Z

...in/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecCorrelate.scala


 /**
-  * Flink RelNode which matches along with join a user defined table function.
+  * Flink RelNode which matches along with join a java user defined table function.


java => Java/Scala

hequn8128 · 2020-02-14T04:09:32Z

...a/org/apache/flink/table/planner/plan/rules/physical/batch/BatchExecPythonCorrelateRule.java

+			// a filter is pushed above the table function
+			FlinkLogicalCalc calc = (FlinkLogicalCalc) right;
+			RelSubset input = (RelSubset) calc.getInput();
+			FlinkLogicalTableFunctionScan scan = (FlinkLogicalTableFunctionScan) input.getOriginal();


We can't cast the original node to FlinkLogicalTableFunctionScan.

hequn8128 · 2020-02-14T04:12:29Z

...a/org/apache/flink/table/planner/plan/rules/physical/batch/BatchExecPythonCorrelateRule.java

+		private final RelNode right;
+
+		BatchExecPythonCorrelateFactory(RelNode rel) {
+			this.correlateRel = rel;


Remove this member?

hequn8128 · 2020-02-14T04:16:54Z

...org/apache/flink/table/planner/plan/rules/physical/stream/StreamExecPythonCorrelateRule.java

+	 * The factory is responsible to creating {@link StreamExecPythonCorrelate}.
+	 */
+	private static class StreamExecPythonCorrelateFactory {
+		private final RelNode correlateRel;


Remove this member.

hequn8128 · 2020-02-14T04:17:23Z

...org/apache/flink/table/planner/plan/rules/physical/stream/StreamExecPythonCorrelateRule.java

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


responsible to => responsible for

… Correlate Rule And RelNode

… Correlate Rule And RelNode-fix-1

HuangXingBo · 2020-02-16T11:40:47Z

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

hequn8128

@HuangXingBo Thanks a lot for the update. The test failed due to the problems in BatchExecCorrelateBase and StreamExecCorrelateBase. See the detailed comments below.

As for the last commit that adding implementations for the Python Correlate RelNode, maybe it's better to add the commit later. Because in this PR, there are no ways to write tests to cover the implementation.

What do you think?

hequn8128 · 2020-02-17T03:04:50Z

...cala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecCorrelateBase.scala

+
+  override def explainTerms(pw: RelWriter): RelWriter = {
+    val rexCall = scan.getCall.asInstanceOf[RexCall]
+    val sqlFunction = rexCall.getOperator.asInstanceOf[TableSqlFunction]


The test failed due to this line. We can remove this line as it has never been used.
Remember to remove the useless import after removing this line.

hequn8128 · 2020-02-17T03:05:28Z

.../scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecCorrelateBase.scala

+
+  override def explainTerms(pw: RelWriter): RelWriter = {
+    val rexCall = scan.getCall.asInstanceOf[RexCall]
+    val sqlFunction = rexCall.getOperator.asInstanceOf[TableSqlFunction]


HuangXingBo · 2020-02-17T04:11:58Z

@HuangXingBo Thanks a lot for the update. The test failed due to the problems in BatchExecCorrelateBase and StreamExecCorrelateBase. See the detailed comments below.

As for the last commit that adding implementations for the Python Correlate RelNode, maybe it's better to add the commit later. Because in this PR, there are no ways to write tests to cover the implementation.

What do you think?

Thanks a lot for @hequn8128 , I will move the last commit to FLINK-15972.

… Correlate Rule And RelNode-fix-2

hequn8128 · 2020-02-18T06:35:52Z

@flinkbot run travis

… Correlate rules and RelNodes This closes apache#11051.

hequn8128 · 2020-02-18T09:56:28Z

@HuangXingBo Hi, the jira id in the commit log is wrong. Would be great if you can check on this next time.
Merging...

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

HuangXingBo force-pushed the FLINK-15961 branch from 68b0714 to 9c17e49 Compare February 13, 2020 05:52

hequn8128 requested changes Feb 14, 2020

View reviewed changes

HuangXingBo added 2 commits February 14, 2020 12:32

[FLINK-15691][table-planner][table-planner-blink] Add Physical Python…

15b130b

… Correlate Rule And RelNode

[FLINK-15691][table-planner][table-planner-blink] Add Physical Python…

a5685ea

… Correlate Rule And RelNode-fix-1

HuangXingBo force-pushed the FLINK-15961 branch from 9c17e49 to 67f8b75 Compare February 16, 2020 11:36

hequn8128 requested changes Feb 17, 2020

View reviewed changes

HuangXingBo force-pushed the FLINK-15961 branch from 67f8b75 to 30afe69 Compare February 17, 2020 04:23

[FLINK-15691][table-planner][table-planner-blink] Add Physical Python…

f0fbdc6

… Correlate Rule And RelNode-fix-2

HuangXingBo force-pushed the FLINK-15961 branch from 30afe69 to f0fbdc6 Compare February 17, 2020 04:44

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

[FLINK-15691][table-planner][table-planner-blink] Add physical Python…

53b7d35

… Correlate rules and RelNodes This closes apache#11051.

hequn8128 closed this in 91fcd83 Feb 18, 2020

hequn8128 mentioned this pull request Mar 5, 2020

[FLINK-16337][python][table-planner][table-planner-blink] Add support of vectorized Python UDF in blink planner and old planner #11252

Closed

[FLINK-15961][table-planner][table-planner-blink] Introduce Python Physical Correlate RelNodes which are containers for Python TableFunction #11051

[FLINK-15961][table-planner][table-planner-blink] Introduce Python Physical Correlate RelNodes which are containers for Python TableFunction #11051

Uh oh!

Conversation

HuangXingBo commented Feb 10, 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

Uh oh!

flinkbot commented Feb 10, 2020

Automated Checks

Review Progress

Uh oh!

flinkbot commented Feb 10, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

hequn8128 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HuangXingBo commented Feb 16, 2020

Uh oh!

hequn8128 left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hequn8128 Feb 17, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HuangXingBo commented Feb 17, 2020

Uh oh!

hequn8128 commented Feb 18, 2020

Uh oh!

hequn8128 commented Feb 18, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

flinkbot commented Feb 10, 2020 •

edited

Loading

hequn8128 left a comment •

edited

Loading

hequn8128 Feb 17, 2020 •

edited

Loading