[FLINK-11896] [table-planner-blink] Introduce stream physical nodes #7969

godfreyhe · 2019-03-13T05:59:27Z

What is the purpose of the change

Introduce stream physical RelNode

Brief change log

adds all stream physical nodes, excludes StreamExecCorrelate, StreamExecTemporalTableJoin, StreamExecGroupWindowAggregate, StreamExecMatch, StreamExecTemporalTableFunctionJoin

Verifying this change

This change is an initialization work without any test coverage.

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? (no)
If yes, how is the feature documented? (not documented)

flinkbot · 2019-03-13T06:00:18Z

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.

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

wuchong

Thanks @godfreyhe for the effort and the nice code comment.

I left some ideas below.

wuchong · 2019-03-14T10:07:22Z

...anner-blink/src/main/scala/org/apache/flink/table/plan/nodes/calcite/WatermarkAssigner.scala

-  extends SingleRel(cluster, traits, inputNode) {
+    inputRel: RelNode,
+    val rowtimeFieldIndex: Option[Int],
+    val watermarkOffset: Option[Long])


Rename to watermarkDelay ? In fact, it is the delay between max-timestamp and current watermark.
The offset terminology is usually means timezone offset in Flink.

wuchong · 2019-03-14T10:13:14Z

...ink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamPhysicalRel.scala

+  /**
+    * Whether the [[StreamPhysicalRel]] requires rowtime watermark in processing logic.
+    */
+  def requireWatermark: Boolean = false


How about to remove all default value for the interface methods to force the concrete node to implement them?

wuchong · 2019-03-14T10:17:56Z

...table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/common/CommonJoin.scala

+      .item("where",
+        RelExplainUtil.expressionToString(getCondition, inputRowType, getExpressionString))
+      .item("join", getRowType.getFieldNames.mkString(", "))
+      .item("joinType", RelExplainUtil.joinTypeToString(flinkJoinType))


How about to move joinType before where? IMO, the information of joinType is more important than where.

wuchong · 2019-03-14T10:18:17Z

...table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/common/CommonJoin.scala

+    pw.input("left", getLeft).input("right", getRight)
+      .item("where",
+        RelExplainUtil.expressionToString(getCondition, inputRowType, getExpressionString))
+      .item("join", getRowType.getFieldNames.mkString(", "))


How about rename join to select ?

wuchong · 2019-03-14T10:24:13Z

...rc/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecFirstLastRow.scala

+      .item("key", uniqueKeys.map(fieldNames.get).mkString(", "))
+      .item("select", fieldNames.mkString(", "))
+      .item("order", orderString)
+      .item("mode", if (isLastRowMode) "LastRow" else "FirstRow")


How about moving mode before key and remove select?
Do we really need to display field names in this node ?

field names is needless if the RelNode does not change output type

wuchong · 2019-03-14T10:40:28Z

...-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecRank.scala

+      .item("partitionBy", RelExplainUtil.fieldToString(partitionKey.toArray, inputRowType))
+      .item("orderBy", RelExplainUtil.collationToString(sortCollation, inputRowType))
+      .item("rankRange", rankRange.toString(inputRowType.getFieldNames))
+      .item("strategy", getStrategy())


I would suggest to move strategy before rankFunction as it is more important.

wuchong · 2019-03-14T12:15:54Z

...n/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecGroupAggregateBase.scala

+  * distinct agg. partial-aggregation produces a partial distinct agg result for each bucket group,
+  * and then final-aggregation produces final result based on partial result.
+  * Both partial-aggregation and final-aggregation need to shuffle data.
+  * partial-final aggregation is enabled only if all distinct aggregate calls are mergeable.


I think the local global can be enabled when all aggregate calls are mergeable. And the partial final can be enabled when all aggregates are splittable.

2. perfect comments 3. adjust explain terms order

godfreyhe · 2019-03-15T03:51:45Z

Thanks for your suggestion @wuchong. I have updated this PR based on comments

wuchong

Thanks @godfreyhe , I only left a minor comment.

Others looks good to me.

wuchong · 2019-03-15T06:36:17Z

...src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalWatermarkAssigner.scala

-    watermarkOffset: Long)
-  extends WatermarkAssigner(cluster, traits, input, rowtimeField, watermarkOffset)
+    rowtimeFieldIndex: Option[Int],
+    watermarkOffset: Option[Long])


Rename all the other watermarkOffset to watermarkDelay ?

KurtYoung · 2019-03-15T07:07:40Z

...table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/common/CommonJoin.scala

+/**
+  * Base physical class for flink [[Join]].
+  */
+trait CommonJoin extends Join with FlinkPhysicalRel {


rename to CommonPhysicalJoin?

BTW, i find out that CommonExchange is inherited directly from FlinkRelNode, but only have physical children. Could you also make CommonExchange to inherit from FlinkPhysicalRel?

wuchong · 2019-03-15T08:13:02Z

+1 from my side.

…nJoin` to `CommonPhysicalJoin`

This closes apache#7969

KurtYoung · 2019-03-15T11:47:50Z

LGTM, merging

…pache#7969)

rmetzger added the review=description? label Mar 13, 2019

godfreyhe force-pushed the FLINK-11896 branch from f8d0751 to 9a6d6aa Compare March 14, 2019 01:46

wuchong reviewed Mar 14, 2019

View reviewed changes

godfreyhe added 3 commits March 15, 2019 11:49

[FLINK-11896] [table-planner-blink] Introduce stream physical nodes

176114f

add license for CommonJoin

ace249d

1. remove default implementation in StreamPhysicalRel

e4f596d

2. perfect comments 3. adjust explain terms order

godfreyhe force-pushed the FLINK-11896 branch from d70f43a to e4f596d Compare March 15, 2019 03:50

wuchong reviewed Mar 15, 2019

View reviewed changes

rename watermarkOffset to watermarkDelay

2b6fd12

KurtYoung reviewed Mar 15, 2019

View reviewed changes

rename CommonExchange to CommonPhysicalExchange and rename `Commo…

8c78d6b

…nJoin` to `CommonPhysicalJoin`

wuchong pushed a commit to wuchong/flink that referenced this pull request Mar 15, 2019

[FLINK-11896] [table-planner-blink] Introduce stream physical nodes

abb70f3

This closes apache#7969

KurtYoung merged commit d814f04 into apache:master Mar 15, 2019

rmetzger added the component=API/TableSQL label Mar 18, 2019

HuangZhenQiu pushed a commit to HuangZhenQiu/flink that referenced this pull request Mar 20, 2019

[FLINK-11896] [table-planner-blink] Introduce stream physical nodes (a…

ddcb1b6

…pache#7969)

HuangZhenQiu pushed a commit to HuangZhenQiu/flink that referenced this pull request Apr 22, 2019

[FLINK-11896] [table-planner-blink] Introduce stream physical nodes (a…

2741423

…pache#7969)

rmetzger added component=TableSQL/Planner and removed component=API/TableSQL labels Apr 30, 2019

sunhaibotb pushed a commit to sunhaibotb/flink that referenced this pull request May 8, 2019

[FLINK-11896] [table-planner-blink] Introduce stream physical nodes (a…

5710bcb

…pache#7969)

godfreyhe deleted the FLINK-11896 branch June 1, 2019 09:54

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-11896] [table-planner-blink] Introduce stream physical nodes #7969

[FLINK-11896] [table-planner-blink] Introduce stream physical nodes #7969

godfreyhe commented Mar 13, 2019

flinkbot commented Mar 13, 2019

wuchong left a comment

wuchong Mar 14, 2019

wuchong Mar 14, 2019

wuchong Mar 14, 2019

wuchong Mar 14, 2019

wuchong Mar 14, 2019

godfreyhe Mar 15, 2019

wuchong Mar 14, 2019

wuchong Mar 14, 2019

godfreyhe commented Mar 15, 2019 •

edited

wuchong left a comment

wuchong Mar 15, 2019

godfreyhe Mar 15, 2019

KurtYoung Mar 15, 2019

KurtYoung Mar 15, 2019

godfreyhe Mar 15, 2019

wuchong commented Mar 15, 2019

KurtYoung commented Mar 15, 2019

[FLINK-11896] [table-planner-blink] Introduce stream physical nodes #7969

[FLINK-11896] [table-planner-blink] Introduce stream physical nodes #7969

Conversation

godfreyhe commented Mar 13, 2019

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 Mar 13, 2019

Review Progress

wuchong 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

godfreyhe commented Mar 15, 2019 • edited

wuchong 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

wuchong commented Mar 15, 2019

KurtYoung commented Mar 15, 2019

godfreyhe commented Mar 15, 2019 •

edited