[FLINK-12017][table-runtime-blink] Introduce Rank and Deduplicate operators for blink streaming runtime #8109

beyond1920 · 2019-04-03T11:26:17Z

What is the purpose of the change

Introduce Rank and Deduplicate operators for blink streaming runtime

Brief change log

StreamExecExchange, StreamExecRank, StreamExecDeduplicate implements StreamExecNode
Introduce RankFunctions and DeduplicateFunctions.
Fix some minor bug

Verifying this change

 IT Case

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

Dependencies (does it add or upgrade a dependency): (yes, for test)
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? (yes)
If yes, how is the feature documented? (JavaDocs)

flinkbot · 2019-04-03T11:27:45Z

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

KurtYoung

The indent config for your IDE seems not configured well, there are lots of indent issues.
I still need some more time to go through all rank function related classes.

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

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

KurtYoung · 2019-04-08T03:57:40Z

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

+
+    val rowTypeInfo = inputTransform.getOutputType.asInstanceOf[BaseRowTypeInfo]
+
+    val generateRetraction = true


I think keep first row will not generate retraction?

whether generate Retraction could be inferred based on Retraction rules, which is done in FLINK-12098. I could add a TODO message here.

Update here.

...nk/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecExchange.scala

...-blink/src/main/java/org/apache/flink/table/runtime/deduplicate/DeduplicateFunctionBase.java

KurtYoung · 2019-04-08T08:51:55Z

BTW, i think we need some dedicated tests for all rank functions.

...e/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/rank/SortedMap.java

...time-blink/src/main/java/org/apache/flink/table/runtime/rank/AbstractUpdateRankFunction.java

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

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

...ble/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryRow.java

.../flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryWriter.java

...k-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/TypeGetterSetters.java

...time-blink/src/main/java/org/apache/flink/table/runtime/deduplicate/DeduplicateFunction.java

JingsongLi · 2019-04-10T03:30:13Z

...le-runtime-blink/src/main/java/org/apache/flink/table/runtime/rank/AbstractRankFunction.java

+	protected long hitCount = 0L;
+	protected long requestCount = 0L;
+
+	AbstractRankFunction(long minRetentionTime, long maxRetentionTime, BaseRowTypeInfo inputRowType,


use inputArity outputArity instead of inputRowType and outputRowType

inputRowType is still useful in the other case, outputRowType could be removed.

JingsongLi · 2019-04-10T03:32:49Z

...able-runtime-blink/src/main/java/org/apache/flink/table/runtime/rank/AppendRankFunction.java

+	private transient Map<BaseRow, TopNBuffer> kvSortedMap;
+
+	public AppendRankFunction(
+			long minRetentionTime, long maxRetentionTime, BaseRowTypeInfo inputRowType, BaseRowTypeInfo outputRowType,


I think we can use BaseRowSerializer instead of inputRowType

inputRowType need to used when create StateDescriptor.
Such as :
ListTypeInfo<BaseRow> valueTypeInfo = new ListTypeInfo<>(inputRowType); MapStateDescriptor<BaseRow, List<BaseRow>> mapStateDescriptor = new MapStateDescriptor( "data-state-with-append", sortKeyType, valueTypeInfo);

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

KurtYoung · 2019-04-12T10:19:12Z

...nk/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecExchange.scala

+  with StreamPhysicalRel
+  with StreamExecNode[BaseRow] {
+
+  private val DEFAULT_MAX_PARALLELISM = 1 << 7


The variable's name is confusing, it's not max parallelism of operators, but max number of key-groups. Maybe it's better to use StreamGraphGenerator.DEFAULT_LOWER_BOUND_MAX_PARALLELISM or KeyGroupRangeAssignment.DEFAULT_LOWER_BOUND_MAX_PARALLELISM

+1 . I'm in favor of the later one which is also used in KeyedStream.java#134 and has a clear javadoc description.

...time-blink/src/main/java/org/apache/flink/table/runtime/deduplicate/DeduplicateFunction.java

KurtYoung · 2019-04-12T10:28:10Z

...time-blink/src/main/java/org/apache/flink/table/runtime/deduplicate/DeduplicateFunction.java

+		super.open(configure);
+		String stateName = keepLastRow ? "DeduplicateFunctionCleanupTime" : "DeduplicateFunctionCleanupTime";
+		initCleanupTimeState(stateName);
+		ValueStateDescriptor rowStateDesc = new ValueStateDescriptor("rowState", rowTypeInfo);


in case of firstRow, only pk is needed for state, we don't have to store the whole row.

Yes. I think we should only store the PK in state here.

If we only store PK, the ideal state schema should be ValueState<Boolean>, but this can't share the same state with LastRow mode. Maybe we need to separate the implementation for firstRow and lastRow.

Exactly, only keepLastRow and generate retract is true, we need to store complete row, else store pk is ok. Thanks.

...link/src/main/java/org/apache/flink/table/runtime/deduplicate/DeduplicateFunctionHelper.java

wuchong

Thanks @beyond1920 for the great work!

I think this PR is close to ready. Except we need to polish DeduplicateFunction a bit more, I only some other minor comments.

Cheers,
Jark

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

...link-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/utils/TableUtil.scala

...le-planner-blink/src/test/scala/org/apache/flink/table/runtime/utils/StreamingTestBase.scala

...r-blink/src/test/scala/org/apache/flink/table/runtime/utils/StreamingWithStateTestBase.scala

...time-blink/src/main/java/org/apache/flink/table/runtime/deduplicate/DeduplicateFunction.java

...le-runtime-blink/src/main/java/org/apache/flink/table/runtime/rank/AbstractRankFunction.java

...able-runtime-blink/src/main/java/org/apache/flink/table/runtime/rank/AppendRankFunction.java

...able-runtime-blink/src/main/java/org/apache/flink/table/runtime/rank/UpdateRankFunction.java

...le-runtime-blink/src/test/java/org/apache/flink/table/runtime/window/WindowOperatorTest.java

beyond1920

@wuchong thanks for your suggestions. I split keepFirstRow and keepLastRow into different functions. And Update the API of BaseRowKeySelector.

wuchong · 2019-04-15T04:11:20Z

@flinkbot approve all.

+1 to merge.

Wait until Travis turns green.

…plicate to StreamTransformation

2. Introduce SortedMapSerializerSnapshot to do snapshot for SortedMapTypeInfo.

2.Other minor update.

… DeduplicateKeepLastRowFunction 2. other minor update.

wuchong · 2019-04-16T01:55:21Z

Merging...

…rators for blink streaming runtime This closes apache#8109

rmetzger added the review=description? label Apr 3, 2019

beyond1920 force-pushed the flink-12017 branch from d6bb38e to 3c720b9 Compare April 3, 2019 11:36

rmetzger added the component=TableSQL/Planner label Apr 3, 2019

beyond1920 force-pushed the flink-12017 branch 10 times, most recently from 60252e4 to ea882fc Compare April 8, 2019 01:37

KurtYoung reviewed Apr 8, 2019

View reviewed changes

...e/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/rank/SortedMap.java Outdated Show resolved Hide resolved

KurtYoung reviewed Apr 8, 2019

View reviewed changes

...time-blink/src/main/java/org/apache/flink/table/runtime/rank/AbstractUpdateRankFunction.java Outdated Show resolved Hide resolved

beyond1920 force-pushed the flink-12017 branch 3 times, most recently from 615ae4b to 8e25cfd Compare April 10, 2019 03:03

JingsongLi reviewed Apr 10, 2019

View reviewed changes

beyond1920 force-pushed the flink-12017 branch 4 times, most recently from 4f9f70c to f88533d Compare April 12, 2019 09:33

KurtYoung reviewed Apr 12, 2019

View reviewed changes

beyond1920 force-pushed the flink-12017 branch 2 times, most recently from 2c6eb76 to b049244 Compare April 13, 2019 00:49

wuchong reviewed Apr 13, 2019

View reviewed changes

beyond1920 commented Apr 15, 2019

View reviewed changes

beyond1920 force-pushed the flink-12017 branch 5 times, most recently from bce8cf6 to 8cd47d6 Compare April 15, 2019 03:50

beyond1920 force-pushed the flink-12017 branch 3 times, most recently from 1d79a64 to 288fa78 Compare April 15, 2019 07:27

beyond1920 added 4 commits April 15, 2019 17:39

[FLINK-12017][table-planner-blink] Support translation from Rank/Dedu…

1b8e599

…plicate to StreamTransformation

Update code style.

ac1b502

1. add ut for Deduplicate functions and Rank functions.

2c794f9

2. Introduce SortedMapSerializerSnapshot to do snapshot for SortedMapTypeInfo.

1.Update Deduplicate Function state.

25bdd55

2.Other minor update.

beyond1920 force-pushed the flink-12017 branch from 288fa78 to 8832402 Compare April 15, 2019 09:40

1. split DeduplicateFunction into DeduplicateKeepFirstRowFunction and…

7de1517

… DeduplicateKeepLastRowFunction 2. other minor update.

beyond1920 force-pushed the flink-12017 branch from 8832402 to 7de1517 Compare April 15, 2019 10:36

beyond1920 changed the title ~~[FLINK-12017][table-planner-blink] Support translation from Rank/Deduplicate to StreamTransformation~~ [FLINK-12017][table-runtime-blink] Introduce Rank and Deduplicate operators for blink streaming runtime Apr 16, 2019

asfgit closed this in 8241954 Apr 16, 2019

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

[FLINK-12017][table-planner-blink] Introduce Rank and Deduplicate ope…

d8e6722

…rators for blink streaming runtime This closes apache#8109

tianchen92 pushed a commit to tianchen92/flink that referenced this pull request May 13, 2019

[FLINK-12017][table-planner-blink] Introduce Rank and Deduplicate ope…

79d000e

…rators for blink streaming runtime This closes apache#8109

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-12017][table-runtime-blink] Introduce Rank and Deduplicate operators for blink streaming runtime #8109

[FLINK-12017][table-runtime-blink] Introduce Rank and Deduplicate operators for blink streaming runtime #8109

beyond1920 commented Apr 3, 2019 •

edited

flinkbot commented Apr 3, 2019

KurtYoung left a comment

KurtYoung Apr 8, 2019

beyond1920 Apr 10, 2019

beyond1920 Apr 12, 2019

KurtYoung commented Apr 8, 2019

JingsongLi Apr 10, 2019

beyond1920 Apr 10, 2019

JingsongLi Apr 10, 2019

beyond1920 Apr 10, 2019

KurtYoung Apr 12, 2019

beyond1920 Apr 12, 2019

wuchong Apr 12, 2019

KurtYoung Apr 12, 2019

wuchong Apr 12, 2019 •

edited

beyond1920 Apr 12, 2019

wuchong left a comment

beyond1920 left a comment •

edited

wuchong commented Apr 15, 2019

wuchong commented Apr 16, 2019


		val rowTypeInfo = inputTransform.getOutputType.asInstanceOf[BaseRowTypeInfo]

		val generateRetraction = true

[FLINK-12017][table-runtime-blink] Introduce Rank and Deduplicate operators for blink streaming runtime #8109

[FLINK-12017][table-runtime-blink] Introduce Rank and Deduplicate operators for blink streaming runtime #8109

Conversation

beyond1920 commented Apr 3, 2019 • edited

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 Apr 3, 2019

Review Progress

KurtYoung 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

KurtYoung commented Apr 8, 2019

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

wuchong Apr 12, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wuchong left a comment

Choose a reason for hiding this comment

beyond1920 left a comment • edited

Choose a reason for hiding this comment

wuchong commented Apr 15, 2019

wuchong commented Apr 16, 2019

beyond1920 commented Apr 3, 2019 •

edited

wuchong Apr 12, 2019 •

edited

beyond1920 left a comment •

edited