[FLINK-12665] [table-planner-blink] Introduce MiniBatchIntervalInferRule and watermark assigner operators #8562

godfreyhe · 2019-05-29T07:37:31Z

What is the purpose of the change

Introduce MiniBatchIntervalInferRule and watermark assigner operators

Brief change log

Introduce MiniBatchIntervalInferRule
Introduce watermark assigner operators, including: WatermarkAssignerOperator, MiniBatchedWatermarkAssignerOperator, MiniBatchAssignerOperator

Verifying this change

This change added tests and can be verified as follows:

Added operators tests for watermark assigner operators
Added MiniBatchIntervalInferTest that validates the plan after MiniBatchIntervalInferRule is applied

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

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / no / don't know)
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 / docs / JavaDocs / not documented)

flinkbot · 2019-05-29T07:39:44Z

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

cshuo

Looks good overall. I've left some minor comments.

cshuo · 2019-07-02T07:19:38Z

...a/org/apache/flink/table/runtime/watermarkassigner/MiniBatchedWatermarkAssignerOperator.java

+
+	private final long watermarkDelay;
+
+	// timezone watermarkDelay.


use "offset" here may be more precise.

cshuo · 2019-07-02T07:43:04Z

...planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/AggregateITCase.scala

-        .toTable(tEnv, 'a, 'b, 'c, 'd, 'e, 'rowtime)
+    val t = failingDataSource(TestData.tupleData5.map {
+      case (a, b, c, d, e) => (b, a, c, d, e)
+    }).assignTimestampsAndWatermarks(


It's ok here as mini-batch window is supported now. But it's necessary to construct a StreamExecWatermarkAssigner node to fully verify the correctness of mini-batch watermark.

cshuo · 2019-07-03T06:32:24Z

...src/main/scala/org/apache/flink/table/plan/optimize/StreamCommonSubGraphBasedOptimizer.scala

          val tableName = scan.getTable.getQualifiedName.mkString(".")
          val inputBlocks = block.children.filter(b => tableName.equals(b.getOutputTableName))
          Preconditions.checkArgument(inputBlocks.size <= 1)
          if (inputBlocks.size == 1) {
+            val mergedInterval = FlinkRelOptUtil.mergeMiniBatchInterval(


traits of sinkBlock have already been initialized before the first round of optimization, so miniBatchInterval can be ignored if the block is sink block.

cshuo · 2019-07-03T06:40:12Z

...ain/scala/org/apache/flink/table/plan/rules/physical/stream/MiniBatchIntervalInferRule.scala

+    val updatedTrait = rel match {
+      case _: StreamExecGroupWindowAggregate =>
+        // TODO introduce mini-batch window aggregate later
+        MiniBatchIntervalTrait.NONE


It's not proper to use MiniBatchIntervalTrait.NONE here, since we can not distinguish window without miniBatch with other cases which yield MiniBatchIntervalTrait.NONE.
A specific value may be needed.

cshuo · 2019-07-03T11:41:37Z

...ink-table-planner-blink/src/test/resources/org/apache/flink/table/api/stream/ExplainTest.xml

+                     +- LogicalTableScan(table=[[_DataStreamTable_3]])
+
+== Optimized Logical Plan ==
+Calc(select=[id1, rowtime AS ts, text], updateAsRetraction=[true], accMode=[Acc], reuse_id=[1]): rowcount = , cumulative cost = {rows, cpu, io, network, memory}


It's necessary to add cases with multi-layer blocks to validate the correctness of trait propagating among blocks.

…ule and watermark assigner operators

…TableTestUtil

cshuo · 2019-07-04T06:21:49Z

Looks good. 👍

KurtYoung

+1, verifying this locally and merge

rmetzger added review=description? component=TableSQL/Planner component=TableSQL/Runtime labels May 29, 2019

godfreyhe force-pushed the FLINK-12665 branch 2 times, most recently from f70b276 to 6c15cdc Compare May 31, 2019 03:12

godfreyhe force-pushed the FLINK-12665 branch 3 times, most recently from bcb020a to 560eab3 Compare July 2, 2019 03:32

cshuo reviewed Jul 3, 2019

View reviewed changes

godfreyhe added 4 commits July 4, 2019 10:50

[FLINK-12665] [table-planner-blink] Introduce MiniBatchIntervalInferR…

0a18672

…ule and watermark assigner operators

fix checkstyle error

3c7547c

move registerTableWithWatermark from StreamTableEnvironment to Stream…

ea2943e

…TableTestUtil

address the comments

2737dc7

godfreyhe force-pushed the FLINK-12665 branch from 560eab3 to 2737dc7 Compare July 4, 2019 04:12

KurtYoung approved these changes Jul 4, 2019

View reviewed changes

fix failed tests

412b0fd

KurtYoung merged commit 4fbf323 into apache:master Jul 4, 2019

godfreyhe deleted the FLINK-12665 branch May 20, 2020 02:07

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-12665] [table-planner-blink] Introduce MiniBatchIntervalInferRule and watermark assigner operators #8562

[FLINK-12665] [table-planner-blink] Introduce MiniBatchIntervalInferRule and watermark assigner operators #8562

godfreyhe commented May 29, 2019

flinkbot commented May 29, 2019

cshuo left a comment

cshuo Jul 2, 2019

cshuo Jul 2, 2019

cshuo Jul 3, 2019

cshuo Jul 3, 2019

cshuo Jul 3, 2019

cshuo commented Jul 4, 2019

KurtYoung left a comment


		private final long watermarkDelay;

		// timezone watermarkDelay.

[FLINK-12665] [table-planner-blink] Introduce MiniBatchIntervalInferRule and watermark assigner operators #8562

[FLINK-12665] [table-planner-blink] Introduce MiniBatchIntervalInferRule and watermark assigner operators #8562

Conversation

godfreyhe commented May 29, 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 May 29, 2019

Review Progress

cshuo left a comment

Choose a reason for hiding this comment

cshuo Jul 2, 2019

Choose a reason for hiding this comment

cshuo Jul 2, 2019

Choose a reason for hiding this comment

cshuo Jul 3, 2019

Choose a reason for hiding this comment

cshuo Jul 3, 2019

Choose a reason for hiding this comment

cshuo Jul 3, 2019

Choose a reason for hiding this comment

cshuo commented Jul 4, 2019

KurtYoung left a comment

Choose a reason for hiding this comment