[FLINK-12345] [table-planner-blink] Add support for generating optimized logical plan for stream window aggregate #8288

godfreyhe · 2019-04-27T12:23:46Z

What is the purpose of the change

Add support for generating optimized logical plan for stream window aggregate

Brief change log

refactor batch window aggregate
add RelTimeIndicatorConverter to materialize rowtime/proctime type to timestamp type
add WindowAggregateReduceFunctionsRule to reduce complex aggregate function to simpler forms
add StreamExecGroupWindowAggregateRule to convert logical window aggregate to stream physical window aggregate

Verifying this change

This change added tests and can be verified as follows:

Added WindowAggregateTest that validates physical plan for window aggregate queries
Added RelTimeIndicatorConverterTest that validates plan after materialized

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-04-27T12:23:58Z

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.

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

KurtYoung · 2019-05-07T02:53:26Z

...-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/TableConfig.scala

+    * 0 means no delay (fire on every element).
+    * > 0 means the fire interval
+    */
+  private var earlyFireInterval = DEFAULT_FIRE_INTERVAL


cc @wuchong here, do we want to support early fire in 1.9.0?

I think we can support it if we expose early&late interval through options in TableConfig.

I think it's nice to have it.

KurtYoung · 2019-05-07T02:54:33Z

...-planner-blink/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala

+    visitSetOp(minus)
+
+  override def visit(sort: LogicalSort): RelNode = {
+


delete blank line

wuchong · 2019-05-07T03:27:19Z

...scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecGroupWindowAggregate.scala

+/**
+  * Streaming group window aggregate physical node which will be translate to window operator.
+  *
+  * If requirements satisfied, it will be translated into minibatch window operator, otherwise,


The minibatch-window is still in beta in Blink. I think we can support it after 1.9 release.

What do you think about removing minibatch-window relative comments and codes?

It makes sense to me

wuchong · 2019-05-07T03:39:58Z

...le-planner-blink/src/main/java/org/apache/flink/table/functions/sql/ProctimeSqlFunction.java

-			null,
-			OperandTypes.NILADIC,
-			SqlFunctionCategory.TIMEDATE);
+			ReturnTypes.cascade(ReturnTypes.explicit(SqlTypeName.TIMESTAMP), SqlTypeTransforms.TO_NULLABLE),


PROCTIME() is a built-in function used to generate proctime attribute field, whose return type must be proctime time indicator.

I think what you want is ProctimeMaterializeSqlFunction which materialize a proctime field into Timestamp field.

So I think we don't need to change this file. We should use FlinkSqlOperatorTable.PROCTIME_MATERIALIZE in RelTimeIndicatorConverter.

I didn't notice ProctimeMaterializeSqlFunction's existence before, i will revert the changes in ProctimeSqlFunction

…zed logical plan for stream window aggregate

…dowAggregateRule

…rter and revert changes in ProctimeSqlFunction

wuchong · 2019-05-07T07:11:25Z

Thanks @godfreyhe , it looks good to me now.

rmetzger added review=description? component=TableSQL/Planner labels Apr 27, 2019

godfreyhe force-pushed the FLINK-12345 branch 2 times, most recently from 00d941d to 64561df Compare April 30, 2019 02:12

godfreyhe force-pushed the FLINK-12345 branch from 64561df to b25d12e Compare May 6, 2019 10:27

KurtYoung reviewed May 7, 2019

View reviewed changes

wuchong reviewed May 7, 2019

View reviewed changes

godfreyhe added 6 commits May 7, 2019 15:06

[FLINK-12345] [table-planner-blink] Add support for generating optimi…

212a66d

…zed logical plan for stream window aggregate

fix failed test and refactor logical window aggregate

4d43338

fix failed test after rebase master

e054c11

fix checkstyle error and add some comments

650f31d

fix inputIndex of FieldReferenceExpression in Batch(Stream)LogicalWin…

62208cd

…dowAggregateRule

use PROCTIME_MATERIALIZE instead of PROCTIME in RelTimeIndicatorConve…

4a7dca5

…rter and revert changes in ProctimeSqlFunction

godfreyhe force-pushed the FLINK-12345 branch from b25d12e to 4a7dca5 Compare May 7, 2019 07:06

KurtYoung merged commit c6ac93a into apache:master May 7, 2019

godfreyhe deleted the FLINK-12345 branch June 1, 2019 09:53

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-12345] [table-planner-blink] Add support for generating optimized logical plan for stream window aggregate #8288

[FLINK-12345] [table-planner-blink] Add support for generating optimized logical plan for stream window aggregate #8288

Uh oh!

godfreyhe commented Apr 27, 2019

Uh oh!

flinkbot commented Apr 27, 2019

Uh oh!

KurtYoung May 7, 2019

Uh oh!

wuchong May 7, 2019

Uh oh!

KurtYoung May 7, 2019

Uh oh!

godfreyhe May 7, 2019

Uh oh!

wuchong May 7, 2019

Uh oh!

godfreyhe May 7, 2019

Uh oh!

wuchong May 7, 2019

Uh oh!

godfreyhe May 7, 2019

Uh oh!

wuchong commented May 7, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

		visitSetOp(minus)

		override def visit(sort: LogicalSort): RelNode = {

[FLINK-12345] [table-planner-blink] Add support for generating optimized logical plan for stream window aggregate #8288

[FLINK-12345] [table-planner-blink] Add support for generating optimized logical plan for stream window aggregate #8288

Uh oh!

Conversation

godfreyhe commented Apr 27, 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

Uh oh!

flinkbot commented Apr 27, 2019

Review Progress

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!

wuchong commented May 7, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants