[FLINK-19282][table sql/planner]Supports watermark push down with Wat… #13449

fsk119 · 2020-09-22T04:41:20Z

…ermarkStrategy

convert class FunctionContext to interface which allows us convert WatermarkStrategy.Context to FunctionContext;
add bindConstructorTerm in ExprCodeGenerator;
add switch that WatermarkGeneratorCodeGenerator will generate WatermarkGenerator with parameterized constructor and use the term to init udf ;
add rule that transpose between project and watermark assigner. After optimization, watermark assigner will replace reference with rowtime expr;

What is the purpose of the change

Allow planner to push watermark strategy into DynamicTableSource.

Brief change log

convert class FunctionContext to interface which allows us convert WatermarkStrategy.Context to FunctionContext;
add bindConstructorTerm in ExprCodeGenerator;
add switch that WatermarkGeneratorCodeGenerator will generate WatermarkGenerator with parameterized constructor and use the term to open udf ;
add rule that transpose between project and watermark assigner. After optimization, watermark assigner will replace reference with rowtime expr;

Verifying this change

This change added tests and can be verified as follows:

Added test that WatermarkGeneartor can be init with passed context && classload in constructor in WatermarkGeneratorCodeGenTest
Added test for single rule, all rule and ITcase

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, Kubernetes/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 · 2020-09-22T04:43:42Z

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 b042779 (Tue Sep 22 04:43:42 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-09-22T05:02:25Z

CI report:

f80fd5b UNKNOWN
6c252f9 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

fsk119 · 2020-09-25T04:16:58Z

@flinkbot run azure

twalthr · 2020-10-06T11:56:10Z

Thanks for the PR @fsk119. I would like to take a look at this PR soon. Some code changes might conflict with FLIP-107. We need to make sure that both efforts are aligned.

fsk119 · 2020-10-06T15:08:42Z

Thanks for your reply. Looking forward for your suggestions!

godfreyhe

Thanks for the contribution @fsk119 , I left some comments

godfreyhe · 2020-09-30T03:37:39Z

...-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala

+  tableConfig: TableConfig,
+  contextTerm: String = "parameters")


godfreyhe · 2020-09-30T03:39:02Z

.../src/main/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenerator.scala

@@ -18,14 +18,16 @@

 package org.apache.flink.table.planner.codegen

+import org.apache.calcite.rex.RexNode


nit: reorder imports

godfreyhe · 2020-09-30T03:41:01Z

.../src/main/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenerator.scala

@@ -35,7 +37,8 @@ object WatermarkGeneratorCodeGenerator {
  def generateWatermarkGenerator(
      config: TableConfig,
      inputType: RowType,
-      watermarkExpr: RexNode): GeneratedWatermarkGenerator = {
+      watermarkExpr: RexNode,
+      contextTerm: String = null): GeneratedWatermarkGenerator = {


please use None instead of null in Scala

godfreyhe · 2020-09-30T03:44:59Z

.../src/main/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenerator.scala

@@ -44,11 +47,33 @@ object WatermarkGeneratorCodeGenerator {
          " but is " + watermarkOutputType)
    }
    val funcName = newName("WatermarkGenerator")
-    val ctx = CodeGeneratorContext(config)
+    val ctx = if (contextTerm != null) {
+      new ConstantCodeGeneratorContext(config, contextTerm)


why we use ConstantCodeGeneratorContext here ? ConstantCodeGeneratorContext is used for constant reducer, we should create a new special CodeGeneratorContext for watermark generator

godfreyhe · 2020-09-30T04:28:21Z

...nk/src/test/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenTest.scala

 import java.util.Collections
 import java.util.function.{Function => JFunction, Supplier => JSupplier}

+import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier


reorder imports

godfreyhe · 2020-10-07T08:10:10Z

...-blink/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableSourceBase.java

+	/*
+	* Used by apply method to deal with.
+	* */


godfreyhe · 2020-10-07T08:22:05Z

.../apache/flink/table/planner/plan/rules/logical/PushWatermarkIntoTableSourceScanRuleTest.java

+		util.buildStreamProgram(FlinkStreamProgram.DEFAULT_REWRITE());
+		CalciteConfig calciteConfig = TableConfigUtils.getCalciteConfig(util.tableEnv().getConfig());
+		calciteConfig.getStreamProgram().get().addLast(
+				"PushWatermarkIntoTableSourceScanRule",
+				FlinkHepRuleSetProgramBuilder.<StreamOptimizeContext>newBuilder()
+						.setHepRulesExecutionType(HEP_RULES_EXECUTION_TYPE.RULE_SEQUENCE())
+						.setHepMatchOrder(HepMatchOrder.BOTTOM_UP)
+						.add(RuleSets.ofList(
+								WatermarkAssignerProjectTransposeRule.INSTANCE,
+								PushWatermarkIntoTableSourceScanRule.INSTANCE))
+						.build()
+		);


it's better to build a specific program that only contains the rules needed for the current test, this could avoid interference with other rules.

godfreyhe · 2020-10-07T08:25:13Z

.../apache/flink/table/planner/plan/rules/logical/PushWatermarkIntoTableSourceScanRuleTest.java

+				" 'bounded' = 'false'\n" +
+				")";
+		util.tableEnv().executeSql(ddl);
+		util.verifyPlan("select * from MyTable");


add some tests about

projection/filter in select,

watermark expression contains two field references

Please refer to testWatermarkOnComputedColumnWithQuery
and testWatermarkOnComputedColumnWithMultipleInputs.

godfreyhe · 2020-10-07T08:28:50Z

.../apache/flink/table/planner/plan/rules/logical/PushWatermarkIntoTableSourceScanRuleTest.java

+import java.time.ZoneOffset;
+
+/**
+ * Test rule PushWatermarkIntoTableSourceScanRule.


Test for [[PushWatermarkIntoTableSourceScanRule]] and [[WatermarkAssignerProjectTransposeRule]]

godfreyhe · 2020-10-07T08:31:24Z

...src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SourceWatermarkITCase.scala

+  @Before
+  override def before(): Unit = {
+    super.before()
+    env.setParallelism(1)


why parallelism is 1 ?

godfreyhe

Thanks for the update, I left some comments

...link/table/planner/plan/rules/logical/PushWatermarkIntoTableSourceScanAcrossProjectRule.java

.../apache/flink/table/planner/plan/rules/logical/PushWatermarkIntoTableSourceScanBaseRule.java

...lanner-blink/src/main/java/org/apache/flink/table/planner/operations/MergeTableLikeUtil.java

...e-planner-blink/src/main/java/org/apache/flink/table/planner/sources/DynamicSourceUtils.java

godfreyhe · 2020-11-05T13:46:53Z

...he/flink/table/planner/plan/rules/logical/PushWatermarkIntoFlinkTableSourceScanRuleTest.java

+
+	@Before
+	public void setup() {
+		util.buildStreamProgram(FlinkStreamProgram.LOGICAL_REWRITE());


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

godfreyhe · 2020-11-05T13:54:51Z

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

+    </Resource>
+    <Resource name="planAfter">
+      <![CDATA[
+WatermarkAssigner(rowtime=[c], watermark=[-(c, 5000:INTERVAL SECOND)])


why the watermark node can't be pushed down

Oh! I add the wrong rule into the rule set. It works now.

godfreyhe · 2020-11-05T13:57:37Z

...he/flink/table/planner/plan/rules/logical/PushWatermarkIntoFlinkTableSourceScanRuleTest.java

+						"  b bigint,\n" +
+						"  c timestamp(3) not null,\n" +
+						"  d as c + interval '5' second,\n" +
+						"  watermark for d as d - interval '5' second\n" +


add a test about watermark for d as d - interval '5' second + other timestamp field

The watermark expression only allows one input reference. The expression mentioned is illegal

...test/resources/org/apache/flink/table/planner/plan/stream/sql/MiniBatchIntervalInferTest.xml

...org/apache/flink/table/planner/plan/rules/logical/ProjectWatermarkAssignerTransposeRule.java

...e/flink/table/planner/plan/rules/logical/PushWatermarkIntoTableSourceScanAcrossCalcRule.java

.../apache/flink/table/planner/plan/rules/logical/PushWatermarkIntoTableSourceScanRuleTest.java

godfreyhe

Thanks for the update @fsk119 , LGTM

fsk119 · 2020-11-06T13:26:17Z

Thanks for your review! Very thanks!

rmetzger added the review=description? label Sep 22, 2020

rmetzger added the component=TableSQL/Planner label Sep 22, 2020

fsk119 force-pushed the FLINK-19282 branch 2 times, most recently from 3925844 to 28d2598 Compare September 25, 2020 02:54

godfreyhe requested changes Oct 7, 2020

View reviewed changes

fsk119 force-pushed the FLINK-19282 branch from fc8be6b to 614d193 Compare November 5, 2020 10:01

godfreyhe requested changes Nov 5, 2020

View reviewed changes

godfreyhe reviewed Nov 6, 2020

View reviewed changes

godfreyhe approved these changes Nov 6, 2020

View reviewed changes

[FLINK-19282][planner] Support watermark push down in planner

6c252f9

fsk119 force-pushed the FLINK-19282 branch from f80fd5b to 6c252f9 Compare November 6, 2020 13:25

godfreyhe merged commit 2c1e24b into apache:master Nov 7, 2020

fsk119 mentioned this pull request Apr 2, 2021

[FLINK-21675][table-planner-blink] Allow Predicate Pushdown with Watermark Assigner Between Filter and Scan #15307

Closed

fsk119 deleted the FLINK-19282 branch November 22, 2021 07:42

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-19282][table sql/planner]Supports watermark push down with Wat… #13449

[FLINK-19282][table sql/planner]Supports watermark push down with Wat… #13449

fsk119 commented Sep 22, 2020

flinkbot commented Sep 22, 2020

flinkbot commented Sep 22, 2020 •

edited

fsk119 commented Sep 25, 2020

twalthr commented Oct 6, 2020

fsk119 commented Oct 6, 2020

godfreyhe left a comment

godfreyhe Sep 30, 2020

godfreyhe Sep 30, 2020

godfreyhe Sep 30, 2020

godfreyhe Sep 30, 2020

godfreyhe Sep 30, 2020

godfreyhe Oct 7, 2020

godfreyhe Oct 7, 2020

godfreyhe Oct 7, 2020

fsk119 Oct 9, 2020

godfreyhe Oct 7, 2020

godfreyhe Oct 7, 2020

godfreyhe left a comment

godfreyhe Nov 5, 2020

godfreyhe Nov 5, 2020

fsk119 Nov 5, 2020

godfreyhe Nov 5, 2020

fsk119 Nov 5, 2020

godfreyhe left a comment

fsk119 commented Nov 6, 2020

		tableConfig: TableConfig,
		contextTerm: String = "parameters")

		@@ -18,14 +18,16 @@

		package org.apache.flink.table.planner.codegen

		import org.apache.calcite.rex.RexNode

[FLINK-19282][table sql/planner]Supports watermark push down with Wat… #13449

[FLINK-19282][table sql/planner]Supports watermark push down with Wat… #13449

Conversation

fsk119 commented Sep 22, 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 Sep 22, 2020

Automated Checks

Review Progress

flinkbot commented Sep 22, 2020 • edited

CI report:

fsk119 commented Sep 25, 2020

twalthr commented Oct 6, 2020

fsk119 commented Oct 6, 2020

godfreyhe 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

godfreyhe 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

godfreyhe left a comment

Choose a reason for hiding this comment

fsk119 commented Nov 6, 2020

flinkbot commented Sep 22, 2020 •

edited