[FLINK-19609][table-planner-blink] Support streaming window join in planner #15195

beyond1920 · 2021-03-14T08:35:11Z

What is the purpose of the change

This pull request aims to support streaming WindowJoin in planner.

Brief change log

Introduce StreamPhysicalWindowJoin
Update FlinkChangelogModeInterenceProgram and FlinkRelMdWindowProperties to take StreamPhysicalWindowJoin into consideration.
Introduce StreamPhysicalWindowJoinRule to convert FlinkLogicalJoin to StreamPhysicalWindowJoin if join condition contains window starts equality of input tables and window ends equality of input tables.

Verifying this change

UT in WindowJoinTest.

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 · 2021-03-14T08:40:20Z

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 8f844ec (Sun Mar 14 08:40:20 UTC 2021)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{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 · 2021-03-14T08:58:25Z

CI report:

47ef3ba 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

wuchong

Thanks for the great work @beyond1920 . I left some comments.

wuchong · 2021-03-14T14:04:52Z

...src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowProperties.scala

+    val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
+    val leftWindowProperties = fmq.getRelWindowProperties(rel.getLeft)
+    val rightWindowProperties = fmq.getRelWindowProperties(rel.getRight)
+    assert(leftWindowProperties.getWindowSpec == rightWindowProperties.getWindowSpec)


Do we need to use assert here? I think the query is still valid, we just can't infer the window properties. I think we can return null in this case.

The case will not happen because the node will not be translated to WindowJoin if WindowSpec is different.

wuchong · 2021-03-14T14:19:27Z

...rg/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

@@ -266,6 +266,12 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
        }
        createNewNode(join, children, providedTrait, requiredTrait, requester)

+      case windowJoin: StreamPhysicalWindowJoin =>
+        // window join support all changes in input
+        val children = visitChildren(rel, ModifyKindSetTrait.ALL_CHANGES)


I would suggest to use INSERT_ONLY here (and we don't add any tests for this). We can

wuchong · 2021-03-14T14:26:35Z

...ala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalWindowJoin.scala

+    val leftWindowing: WindowingStrategy,
+    val rightWindowing: WindowingStrategy)


It seems that we do not allow different window strategy for left and right input. I think we can simplify the design to only have a single window strategy. We can evolve it in the future.

Left windowStrategy and Right windowStrategy only have same timeAttribute and WindowSpec. But window start index and window end index maybe different.

I see. You are right. Would be better to check the two window strategies in the physical node by overriding isValid.

wuchong · 2021-03-14T14:28:49Z

...-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowJoinUtil.scala

+  def containsWindowStartEqualityOrEndEquality(join: FlinkLogicalJoin): Boolean = {
+    val (windowStartEqualityLeftKeys, windowEndEqualityLeftKeys, _, _) =
+      excludeWindowStartEqualityAndEndEqualityFromJoinInfoPairs(join)
+    windowStartEqualityLeftKeys.nonEmpty || windowEndEqualityLeftKeys.nonEmpty


I think it would be better to use &&, because we don't allow start OR end for now in excludeWindowStartEqualityAndEndEqualityFromJoinInfoPairs.

wuchong · 2021-03-14T14:33:17Z

...-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowJoinUtil.scala

+          val rightFieldType = joinFieldsType.get(rightIndex).getType
+          val rightInputRef = new RexInputRef(rightIndex, rightFieldType)
+          val remainEqual = rexBuilder.makeCall(
+            SqlStdOperatorTable.EQUALS,


Why the remainCondition should contain the join key equality?

remainCondition contains two part:

remain equal condition exclude window start equality and window end equality

non-equal condition

Why not put the join key equality (exclude window start & end) into join key paires? I mean construct a new JoinInfo instead of a RexNode.

@wuchong Thanks, the solution you mentioned works well of course. However I prefer let StreamPhysicalWindowJoin extends CommonPhysicalJoin, so a complete condition which contains equal condition and non-equal condition is required, JoinInfo would be inferred in the Join node. What do you think?

Thanks for the explanation. Sounds good to me.

wuchong · 2021-03-14T14:46:17Z

...nk/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/WindowJoinTest.xml

+    <Resource name="optimized rel plan">
+      <![CDATA[
+Calc(select=[a, b, c, a0, b0, c0])
+- WindowJoin(leftWindow=[CUMULATE(win_start=[window_start], win_end=[window_end], max_size=[1 h], step=[10 min])], rightWindow=[CUMULATE(win_start=[window_start], win_end=[window_end], max_size=[1 h], step=[10 min])], joinType=[InnerJoin], where=[=(a, a0)], select=[a, b, c, window_start, window_end, a0, b0, c0, window_start0, window_end0])


Note: when implementing window join operator, we can't support proctime window join for now, becuase the window assigner is in an separate operator, that means we don't know when to trigger the processing-time window (we don't have something like watermark for proctime). We need to merge window assigner into window join to support proctime window join, but this can be a future work.

Good point! Do we need to throw UnsupportedException in planner, Or just when translate to ExecNode?

throwing a readable exception sounds good to me.

Throwing in planner or when translating to exec node are both fine. But this can be done in the next PR.

Ok. If I understand well, the problem also exists in WindowJoin, WindowAggregate, WindowJoin when input window strategy is WindowAttachedWindowingStrategy, and TimeAttribute is proc-time at the same time. Right?

wuchong · 2021-03-14T14:58:45Z

...nk/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/WindowJoinTest.xml

+    <Resource name="optimized rel plan">
+      <![CDATA[
+Calc(select=[window_start, window_end, a, cnt AS l_cnt, uv AS l_uv, cnt0 AS r_cnt, uv0 AS r_uv, rownum])
+- Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER], rankRange=[rankStart=1, rankEnd=3], partitionBy=[window_start, window_end], orderBy=[cnt DESC], select=[a, window_start, window_end, window_time, cnt, uv, a0, window_start0, window_end0, window_time0, cnt0, uv0, rownum])


Does it mean window properties are not propagated? Because here is Rank instead of WindowRank.

Good catch, I missed windowProperties inference for FlinkLogicalJoin in FlinkRelMdWindowProperties.

wuchong · 2021-03-14T15:08:02Z

...link/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/WindowJoinTest.scala

+  // ----------------------------------------------------------------------------------------
+
+  /** Window type in left and right child should be same **/
+  @Test(expected = classOf[TableException])


Use thrown.expect(...) and thrown.expectMessage(..) to assert exception and error message.

beyond1920 · 2021-03-15T04:12:22Z

@wuchong Thanks for review. I've address all your comments.

wuchong · 2021-03-15T10:06:34Z

...-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowJoinUtil.scala

+   * @return True if join condition contains window starts equality of input tables and window
+   *         ends equality of input tables. Else false.
+   */
+  def containsWindowStartEqualityOrEndEquality(join: FlinkLogicalJoin): Boolean = {


Suggested change

def containsWindowStartEqualityOrEndEquality(join: FlinkLogicalJoin): Boolean = {

def containsWindowStartEqualityAndEndEquality(join: FlinkLogicalJoin): Boolean = {

wuchong · 2021-03-15T10:13:48Z

...-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowJoinUtil.scala

+             |Currently, the windowing TVFs must be the same of left and right inputs.
+             |In the future, we could support different window TVFs, for example, tumbling windows
+             | join sliding windows with the same window size.
+             |""".stripMargin)


We should make it clear what's the current operator, and what's the difference.

Currently, window join doesn't support different window table function of left and right inputs. The left window table function is %s, the right window table function is %s.

wuchong · 2021-03-15T10:15:37Z

...-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowJoinUtil.scala

+             |Currently, time attribute type of left and right inputs should be both row-time or
+             |both proc-time. In the future, we could support different time attribute type.
+             |""".stripMargin)


Currently, window join doesn't support different time attribute type of left and right inputs. The left time attribute type is %s, the right time attribute type is %s.

wuchong · 2021-03-15T10:18:47Z

...-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowJoinUtil.scala

+           |Currently, window starts equality and window ends equality are both required for
+           |window join. In the future, we could support join clause which only includes window
+           |starts equality or window ends equality for TUMBLE or HOP window.
+           |""".stripMargin)


Currently, window join requires JOIN ON condition must contain both window starts equality of input tables and window ends equality of input tables. But the current JOIN ON condition is %s.

beyond1920 · 2021-03-15T14:02:09Z

@wuchong Thanks, I've updated based on your comment.

wuchong · 2021-03-15T14:45:20Z

...link/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/WindowJoinTest.scala

-         |starts equality or window ends equality for TUMBLE or HOP window.
-         |""".stripMargin)
+      "Currently, window join requires JOIN ON condition must contain both window starts " +
+        "equality of input tables and window ends equality of input tables.")


Please also verify the full exception message.

wuchong · 2021-03-15T14:45:22Z

...link/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/WindowJoinTest.scala

-         |starts equality or window ends equality for TUMBLE or HOP window.
-         |""".stripMargin)
+      "Currently, window join requires JOIN ON condition must contain both window starts " +
+        "equality of input tables and window ends equality of input tables.")


Please also verify the full exception message.

wuchong · 2021-03-15T14:50:15Z

...-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowJoinUtil.scala

-             |""".stripMargin)
+          "Currently, window join doesn't support different window table function of left and " +
+            "right inputs.\n" +
+            s"The left window table function is ${leftWindowProperties}.\n" +


We just need to print the windowSpec instead of the total leftWindowProperties. Besides, we need to implement toString method for WindowSpecs.

wuchong · 2021-03-15T14:58:00Z

...-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowJoinUtil.scala

-           |""".stripMargin)
+        "Currently, window join requires JOIN ON condition must contain both window starts " +
+          "equality of input tables and window ends equality of input tables.\n" +
+          s"But the current JOIN ON condition is ${join.getCondition}.")


Currently, it prints:

But the current JOIN ON condition is AND(=($2, $7), =($0, $5)).

I think it unreadable and not helpful for users. Would be better to display field names.

beyond1920 · 2021-03-17T04:19:22Z

@wuchong Thanks a lot, I've updated based on your comment.

…lanner.

wuchong · 2021-03-17T05:42:52Z

...-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowJoinUtil.scala

    if (windowStartEqualityLeftKeys.nonEmpty && windowEndEqualityLeftKeys.nonEmpty) {
      if (
        leftWindowProperties.getTimeAttributeType != rightWindowProperties.getTimeAttributeType) {
+
+        def timeAttributeTypeStr(isRowTime: Boolean): String = {
+          if (isRowTime) "ROWTIME" else "PROCTIME"


We can just print the logical type, because we will support TIMESTAMP_LTZ as time attribute soon, so the logical type maybe different.

wuchong · 2021-03-17T05:43:36Z

...-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowJoinUtil.scala

+            windowSpec: WindowSpec): String = {
+          val windowing = s"win_start=[${inputFieldNames(windowStartIdx)}]" +
+            s", win_end=[${inputFieldNames(windowEndIdx)}]"
+          windowSpec.toSummaryString(windowing)


WindowSpec doesn't contain window_start and window_end column information, so I think we don't need to print them.

wuchong · 2021-03-17T05:43:55Z

...-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowJoinUtil.scala

@@ -182,26 +183,65 @@ object WindowJoinUtil {
    }

    // Validate join
+    def getLeftFieldNames() = join.getLeft.getRowType.getFieldNames.toList
+
+    def getRightFieldNames() = join.getRight.getRowType.getFieldNames.toList


A local variable is enough?

wuchong · 2021-03-17T05:45:37Z

I appended a commit to improve the exception message. And also rebased the branch. If there is no objections, I will merge it once the build is passed.

wuchong

LGTM.

beyond1920 · 2021-03-17T10:15:36Z

@wuchong Thanks a lot. The update is good.

rmetzger added the review=description? label Mar 14, 2021

rmetzger added the component=TableSQL/Planner label Mar 14, 2021

wuchong reviewed Mar 14, 2021

View reviewed changes

beyond1920 force-pushed the window-join branch 3 times, most recently from 162ca63 to 0a79f7d Compare March 15, 2021 08:35

wuchong reviewed Mar 15, 2021

View reviewed changes

beyond1920 and others added 6 commits March 17, 2021 13:35

[FLINK-19609][table-planner-blink] Support streaming window join in p…

2aca81d

…lanner.

Address Jark's comment

bce3cb9

Resolve conflict and updated based on latest master

872dd27

Update exception comment

22bc617

Enrich exception message.

0bc1a2f

Jark: improve exception messages

47ef3ba

wuchong force-pushed the window-join branch from 6938603 to 47ef3ba Compare March 17, 2021 05:41

wuchong reviewed Mar 17, 2021

View reviewed changes

wuchong approved these changes Mar 17, 2021

View reviewed changes

wuchong merged commit 20cb469 into apache:master Mar 17, 2021

		val leftWindowing: WindowingStrategy,
		val rightWindowing: WindowingStrategy)

	def containsWindowStartEqualityOrEndEquality(join: FlinkLogicalJoin): Boolean = {
	def containsWindowStartEqualityAndEndEquality(join: FlinkLogicalJoin): Boolean = {

[FLINK-19609][table-planner-blink] Support streaming window join in planner #15195

[FLINK-19609][table-planner-blink] Support streaming window join in planner #15195

Conversation

beyond1920 commented Mar 14, 2021

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 14, 2021

Automated Checks

Review Progress

flinkbot commented Mar 14, 2021 • edited Loading

CI report:

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

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

beyond1920 Mar 15, 2021 • edited Loading

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

beyond1920 commented Mar 15, 2021

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

beyond1920 commented Mar 15, 2021

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

beyond1920 commented Mar 17, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wuchong commented Mar 17, 2021

wuchong left a comment

Choose a reason for hiding this comment

beyond1920 commented Mar 17, 2021

flinkbot commented Mar 14, 2021 •

edited

Loading

beyond1920 Mar 15, 2021 •

edited

Loading