[FLINK-13289][table-planner-blink] Blink-planner should setKeyFields to upsert table sink #9195

JingsongLi · 2019-07-22T08:08:03Z

What is the purpose of the change

Support upsert table sink in blink-planner

Verifying this change

TableSinkITCase

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

flinkbot · 2019-07-22T08:09:36Z

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 1eeb005 (Tue Aug 06 15:51:04 UTC 2019)

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 · 2019-07-22T08:17:57Z

CI report:

5a275f7 : FAILURE Build
e959fd3 : FAILURE Build
0941235 : FAILURE Build
4c5fe1a : FAILURE Build
1eeb005 : SUCCESS Build

tsreaper

Thanks for the PR @JingsongLi . Looks good to me overall, just left a minor comment.

tsreaper · 2019-07-24T08:49:46Z

...able-planner-blink/src/main/scala/org/apache/flink/table/plan/util/UpdatingPlanChecker.scala

+    val uniqueKeys = fmq.getUniqueKeys(relNode)
+    if (uniqueKeys != null) {
+      uniqueKeys.filter(_.nonEmpty).toList.map { uniqueKey =>
+        val keys = new util.HashSet[String]()


Why a HashSet is needed? It seems that the indices returned by the ImmutableBitSet is strictly ascending. Why not just uniqueKey.asList().toArray?

I'll remove HashSet.

tsreaper

Looks good to me.

wuchong

Thanks for the effort @JingsongLi . I left some comments.

wuchong · 2019-07-24T12:58:03Z

...rc/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecSink.scala

+            // Now we pick shortest one to sink
+            // TODO UpsertStreamTableSink setKeyFields interface should be Array[Array[String]]
+            val tableKeys: Option[Array[String]] =
+              UpdatingPlanChecker.getUniqueKeys(getInput, planner).sortBy(_.length).headOption


There is a potential NPE at this line. Because UpdatingPlanChecker.getUniqueKeys may return null.
What about returning Option[Array[String]] like flink planner? We can move the sortBy and headOption into it.

I think we can return Option[Array[Array[String]]], let sortBy and headOption still remain in caller, keep the TODO with UpsertStreamTableSink.setKeyFields

wuchong · 2019-07-24T13:06:09Z

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

+    val uniqueKeys = fmq.getUniqueKeys(relNode)
+    if (uniqueKeys != null) {
+      uniqueKeys.filter(_.nonEmpty).toList
+          .map(_.asList().map(rowType.getFieldNames.get(_)).toArray).toArray


Can be simplified to uniqueKeys.filter(_.nonEmpty).map(_.toArray.map(fieldNames.get)).toArray

wuchong · 2019-07-24T13:09:39Z

...anner-blink/src/test/scala/org/apache/flink/table/runtime/stream/table/TableSinkITCase.scala

+
+}
+
+private[flink] class TestUpsertSink(


I would like to avoid introducing another set of testing sink (incl. upsert sink, retract sink, append sink). We also have TestingRetractTableSink, TestingUpsertTableSink, TestingAppendTableSink. Can we leverage them in this class? Maybe we need some changes for TestingUpsertTableSink.setKeyFields/setIsAppendOnly.

Yeah, we can make TestingUpsertTableSink support expectedKeys and expectedIsAppendOnly

…to upsert table sink

JingsongLi · 2019-07-25T03:43:26Z

@wuchong Please take a look again.

wuchong

Thanks @JingsongLi for the updating. I left two thoughts. However, overall looks good to me now.

wuchong · 2019-07-25T14:49:17Z

...ink/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/TableSinkITCase.scala

+      .assignAscendingTimestamps(_._1.toLong)
+      .toTable(tEnv, 'id, 'num, 'text, 'rowtime.rowtime)
+
+    val sink = new TestingUpsertTableSink(Array(0, 1, 2), TimeZone.getDefault)


Key array can be "0, 1" to align with "wend", "num"

wuchong · 2019-07-25T14:51:05Z

...rc/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecSink.scala

+            // TODO UpsertStreamTableSink setKeyFields interface should be Array[Array[String]]
+            val tableKeys = {
+              UpdatingPlanChecker.getUniqueKeyFields(getInput, planner) match {
+                case Some(keys) => keys.sortBy(_.length).headOption


Should we return the longest instead of the shortest one? So that it can be the same with flink planner when window start and window end are both selected.

wuchong

Hi @JingsongLi , I looked into the FlinkRelMdUniqueKeys. Currently, we didn't permutate all the possible unique keys and it may touch too many things. I think it's not a blocking problem, we can fix it later.

So I will merge this pull request.

…to upsert table sink This closes #9195

…to upsert table sink This closes apache#9195

rmetzger added the review=description? label Jul 22, 2019

rmetzger added component=Connectors/JDBC component=TableSQL/Planner labels Jul 22, 2019

JingsongLi force-pushed the upsert branch from 5a275f7 to e959fd3 Compare July 23, 2019 06:50

tsreaper reviewed Jul 24, 2019

View reviewed changes

JingsongLi force-pushed the upsert branch from e959fd3 to 138a2ba Compare July 24, 2019 10:33

tsreaper approved these changes Jul 24, 2019

View reviewed changes

wuchong requested changes Jul 24, 2019

View reviewed changes

JingsongLi added 5 commits July 25, 2019 11:42

[FLINK-13289][table-planner-blink] Blink-planner should setKeyFields …

6a02aba

…to upsert table sink

checkstyle

bb0f3fe

Fix comments

0f32020

Fix rebase compile

4d18a28

Fix wuchong's comments

4c5fe1a

JingsongLi force-pushed the upsert branch from 0941235 to 4c5fe1a Compare July 25, 2019 03:43

Fix cases

1eeb005

wuchong reviewed Jul 25, 2019

View reviewed changes

wuchong approved these changes Jul 26, 2019

View reviewed changes

asfgit closed this in 35237ae Jul 26, 2019

asfgit pushed a commit that referenced this pull request Jul 26, 2019

[FLINK-13289][table-planner-blink] Blink planner should setKeyFields …

cdcd15f

…to upsert table sink This closes #9195

JingsongLi deleted the upsert branch July 28, 2019 11:51

wzhero1 pushed a commit to wzhero1/flink that referenced this pull request Aug 2, 2019

[FLINK-13289][table-planner-blink] Blink planner should setKeyFields …

a6ec778

…to upsert table sink This closes apache#9195

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-13289][table-planner-blink] Blink-planner should setKeyFields to upsert table sink #9195

[FLINK-13289][table-planner-blink] Blink-planner should setKeyFields to upsert table sink #9195

JingsongLi commented Jul 22, 2019

flinkbot commented Jul 22, 2019 •

edited

flinkbot commented Jul 22, 2019 •

edited

tsreaper left a comment

tsreaper Jul 24, 2019

JingsongLi Jul 24, 2019

tsreaper left a comment

wuchong left a comment

wuchong Jul 24, 2019

JingsongLi Jul 25, 2019

wuchong Jul 24, 2019

wuchong Jul 24, 2019

JingsongLi Jul 25, 2019

JingsongLi commented Jul 25, 2019

wuchong left a comment

wuchong Jul 25, 2019

wuchong Jul 25, 2019

wuchong left a comment

[FLINK-13289][table-planner-blink] Blink-planner should setKeyFields to upsert table sink #9195

[FLINK-13289][table-planner-blink] Blink-planner should setKeyFields to upsert table sink #9195

Conversation

JingsongLi commented Jul 22, 2019

What is the purpose of the change

Verifying this change

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

Documentation

flinkbot commented Jul 22, 2019 • edited

Automated Checks

Review Progress

flinkbot commented Jul 22, 2019 • edited

CI report:

tsreaper left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tsreaper left a comment

Choose a reason for hiding this comment

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

JingsongLi commented Jul 25, 2019

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

wuchong left a comment

Choose a reason for hiding this comment

flinkbot commented Jul 22, 2019 •

edited

flinkbot commented Jul 22, 2019 •

edited