[FLINK-20370][table] part1: Fix wrong results when sink primary key is not the same with query result's changelog upsert key #17699

lincoln-lil · 2021-11-05T14:53:38Z

What is the purpose of the change

The root cause of this case is FlinkChangelogModeInferenceProgram didn't consider the case when sink's primary key differs from input's upsert keys when infer required ChangeLogMode from sink node.
So we should add releated logic to FlinkChangelogModeInferenceProgram and also update upsertMaterialization process in StreamPhysicalSink

Brief change log

update the logic of FlinkChangelogModeInferenceProgram and StreamPhysicalSink

Verifying this change

Streaming sql's RankTest, AggregateTest and JoinTest

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, Kubernetes/Yarn, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)

flinkbot · 2021-11-05T14:57:56Z

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 0c17938 (Fri Nov 05 14:57:55 UTC 2021)

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 · 2021-11-05T14:58:11Z

CI report:

cf97485 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

JingsongLi · 2021-11-08T05:44:06Z

Thanks @lincoln-lil
Note SinkUpsertMaterializer requires update before.
We should not separate the inference of change log mode from the sink upsert materializer, which will lead to bugs. We should put both the inference and sink upsert mode in one place.
I think we should introduce something like RankProcessStrategy of rank in FlinkChangelogModeInferenceProgram.

lincoln-lil · 2021-11-08T07:03:01Z

@JingsongLi agree with you putting these logic together. I've updated the pr.

lincoln-lil · 2021-11-10T15:55:54Z

@flinkbot run azure

zentol · 2021-11-10T16:51:12Z

@flinkbot run azure

JingsongLi · 2021-11-22T06:55:41Z

docs/layouts/shortcodes/generated/execution_config_configuration.html

@@ -58,6 +58,12 @@
            <td><p>Enum</p></td>
            <td>The NOT NULL column constraint on a table enforces that null values can't be inserted into the table. Flink supports 'error' (default) and 'drop' enforcement behavior. By default, Flink will check values and throw runtime exception when null values writing into NOT NULL columns. Users can change the behavior to 'drop' to silently drop such records without throwing exception.<br /><br />Possible values:<ul><li>"ERROR"</li><li>"DROP"</li></ul></td>
        </tr>
+        <tr>
+            <td><h5>table.exec.sink.pk-shuffle</h5><br> <span class="label label-primary">Streaming</span></td>


Can we create separate PR for this?

Okay, I thought the target issue is the same one. I'll create a separate pr.

JingsongLi · 2021-11-22T06:58:27Z

...rg/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

+        // if sink's pk(s) are not exactly match input changeLogUpsertKeys then it will fallback
+        // to beforeAndAfter mode for the correctness
+        var shouldFallback: Boolean = false
+        val sinkDefinedPks = toScala(sink.catalogTable.getResolvedSchema


Maybe you just want getPrimaryKeyIndexes?

Good point! I simply moved the code from StreamPhysicalSink to here, getPrimaryKeyIndexes is more simpler.

JingsongLi · 2021-11-22T07:00:07Z

...rg/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

+          // Notice: even sink pk(s) contains input upsert key we cannot optimize to UA only,
+          // this differs from batch job's unique key inference
+          if (changeLogUpsertKeys == null || changeLogUpsertKeys.size() == 0
+              || !changeLogUpsertKeys.exists {0 == _.compareTo(sinkPks)}) {


compareTo -> equals?

JingsongLi · 2021-11-22T07:00:19Z

...rg/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

+          // fallback to beforeAndAfter.
+          // Notice: even sink pk(s) contains input upsert key we cannot optimize to UA only,
+          // this differs from batch job's unique key inference
+          if (changeLogUpsertKeys == null || changeLogUpsertKeys.size() == 0


remove changeLogUpsertKeys.size() == 0?

This should be reserved because the metadata query may return a empty changeLogUpsertKeys set.

I think changeLogUpsertKeys.exists {_.equals(sinkPks)} should cover this?

You're right, I overlooked it.

JingsongLi · 2021-11-22T07:01:04Z

...rg/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

+      val sinkRequiredTraits = if (sinkTrait.equals(ONLY_UPDATE_AFTER)) {
+        // if sink's pk(s) are not exactly match input changeLogUpsertKeys then it will fallback
+        // to beforeAndAfter mode for the correctness
+        var shouldFallback: Boolean = false


requireBeforeAndAfter?

yes, it's more clearly.

JingsongLi · 2021-11-22T07:01:38Z

...rg/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

+      val inputChangelogMode = ChangelogPlanUtils.getChangelogMode(
+        sink.getInput.asInstanceOf[StreamPhysicalRel]).get
+      val catalogTable = sink.catalogTable
+      val primaryKeys = toScala(catalogTable.getResolvedSchema


JingsongLi · 2021-11-22T07:04:25Z

...rg/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

+      sinkRequiredTraits
+    }
+
+    private def analyzeUpsertMaterializeStrategy(sink: StreamPhysicalSink): Boolean = {


Can we merge this method to inferSinkRequiredTraits? For example, inferSinkRequiredTraits returns (Seq[UpdateKindTrait], Boolean)

Initially I put the two methods together, but seems a little bit complex, and the two methods do the different things indeed, so I change to the current version.

lincoln-lil

@JingsongLi Thans for your comments!
I'll separate the pr into two.

lincoln-lil · 2021-11-22T07:18:04Z

docs/layouts/shortcodes/generated/execution_config_configuration.html

@@ -58,6 +58,12 @@
            <td><p>Enum</p></td>
            <td>The NOT NULL column constraint on a table enforces that null values can't be inserted into the table. Flink supports 'error' (default) and 'drop' enforcement behavior. By default, Flink will check values and throw runtime exception when null values writing into NOT NULL columns. Users can change the behavior to 'drop' to silently drop such records without throwing exception.<br /><br />Possible values:<ul><li>"ERROR"</li><li>"DROP"</li></ul></td>
        </tr>
+        <tr>
+            <td><h5>table.exec.sink.pk-shuffle</h5><br> <span class="label label-primary">Streaming</span></td>


Okay, I thought the target issue is the same one. I'll create a separate pr.

lincoln-lil · 2021-11-22T07:24:28Z

...rg/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

+      val sinkRequiredTraits = if (sinkTrait.equals(ONLY_UPDATE_AFTER)) {
+        // if sink's pk(s) are not exactly match input changeLogUpsertKeys then it will fallback
+        // to beforeAndAfter mode for the correctness
+        var shouldFallback: Boolean = false


yes, it's more clearly.

lincoln-lil · 2021-11-22T07:25:42Z

...rg/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

+        // if sink's pk(s) are not exactly match input changeLogUpsertKeys then it will fallback
+        // to beforeAndAfter mode for the correctness
+        var shouldFallback: Boolean = false
+        val sinkDefinedPks = toScala(sink.catalogTable.getResolvedSchema


Good point! I simply moved the code from StreamPhysicalSink to here, getPrimaryKeyIndexes is more simpler.

lincoln-lil · 2021-11-22T07:28:05Z

...rg/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

+          // fallback to beforeAndAfter.
+          // Notice: even sink pk(s) contains input upsert key we cannot optimize to UA only,
+          // this differs from batch job's unique key inference
+          if (changeLogUpsertKeys == null || changeLogUpsertKeys.size() == 0


This should be reserved because the metadata query may return a empty changeLogUpsertKeys set.

lincoln-lil · 2021-11-22T07:34:31Z

...rg/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

+      sinkRequiredTraits
+    }
+
+    private def analyzeUpsertMaterializeStrategy(sink: StreamPhysicalSink): Boolean = {


Initially I put the two methods together, but seems a little bit complex, and the two methods do the different things indeed, so I change to the current version.

…s not the same with query result's changelog upsert key

JingsongLi · 2021-11-25T02:41:34Z

...rg/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

@@ -762,6 +757,81 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
        Some(sink.copy(sinkTrait, children.head).asInstanceOf[StreamPhysicalRel])
      }
    }
+


Can you add document here to explain inferSinkRequiredTraits and analyzeUpsertMaterializeStrategy?

.... should ...

.... should ...

...

JingsongLi · 2021-11-25T02:43:50Z

I think this is almost good. Left two comments.

lincoln-lil · 2021-11-26T01:47:53Z

@JingsongLi Thanks for your reviewing! I've updated the pr according to your comments.

JingsongLi

Looks good to me!

…s not the same with query result's changelog upsert key This closes apache#17699

rmetzger added the component=TableSQL/Planner label Nov 5, 2021

lincoln-lil force-pushed the FLINK-20370 branch from 6405f23 to 4115ff9 Compare November 6, 2021 16:27

lincoln-lil force-pushed the FLINK-20370 branch from 64160d5 to 1550e7e Compare November 10, 2021 11:41

lincoln-lil force-pushed the FLINK-20370 branch from 1550e7e to a3f2c4f Compare November 11, 2021 03:39

JingsongLi reviewed Nov 22, 2021

View reviewed changes

lincoln-lil commented Nov 22, 2021

View reviewed changes

lincoln-lil added 2 commits November 22, 2021 15:58

[FLINK-20370][table] part1: Fix wrong results when sink primary key i…

b81c5c6

…s not the same with query result's changelog upsert key

address review comments.

974cd1e

lincoln-lil force-pushed the FLINK-20370 branch from 7f9e229 to 974cd1e Compare November 22, 2021 08:05

lincoln-lil changed the title ~~[FLINK-20370][table] Fix wrong results when sink primary key is not the same with query result's changelog upsert key~~ [FLINK-20370][table] part1: Fix wrong results when sink primary key is not the same with query result's changelog upsert key Nov 22, 2021

lincoln-lil added 2 commits November 22, 2021 16:15

remove unnecessary newline

23a439a

fix pk compute change

8a63281

JingsongLi reviewed Nov 25, 2021

View reviewed changes

address comments: add comments for the inference methods.

cf97485

JingsongLi approved these changes Nov 29, 2021

View reviewed changes

JingsongLi merged commit f8f6935 into apache:master Nov 29, 2021

lincoln-lil added a commit to lincoln-lil/flink that referenced this pull request Dec 8, 2021

[FLINK-20370][table] part1: Fix wrong results when sink primary key i…

03c879c

…s not the same with query result's changelog upsert key This closes apache#17699

niklassemmler pushed a commit to niklassemmler/flink that referenced this pull request Feb 3, 2022

[FLINK-20370][table] part1: Fix wrong results when sink primary key i…

7169439

…s not the same with query result's changelog upsert key This closes apache#17699

yittg mentioned this pull request Apr 15, 2022

Add Support For Flink 1.15 apache/iceberg#4553

Merged

jnh5y pushed a commit to jnh5y/flink that referenced this pull request Dec 18, 2023

[FLINK-20370][table] part1: Fix wrong results when sink primary key i…

cebcb61

…s not the same with query result's changelog upsert key This closes apache#17699

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-20370][table] part1: Fix wrong results when sink primary key is not the same with query result's changelog upsert key #17699

[FLINK-20370][table] part1: Fix wrong results when sink primary key is not the same with query result's changelog upsert key #17699

lincoln-lil commented Nov 5, 2021

flinkbot commented Nov 5, 2021

flinkbot commented Nov 5, 2021 •

edited

Loading

JingsongLi commented Nov 8, 2021

lincoln-lil commented Nov 8, 2021

lincoln-lil commented Nov 10, 2021

zentol commented Nov 10, 2021

JingsongLi Nov 22, 2021

lincoln-lil Nov 22, 2021

JingsongLi Nov 22, 2021

lincoln-lil Nov 22, 2021

JingsongLi Nov 22, 2021

JingsongLi Nov 22, 2021

lincoln-lil Nov 22, 2021

JingsongLi Nov 25, 2021

lincoln-lil Nov 26, 2021

JingsongLi Nov 22, 2021

lincoln-lil Nov 22, 2021

JingsongLi Nov 22, 2021

JingsongLi Nov 22, 2021

lincoln-lil Nov 22, 2021

lincoln-lil left a comment

lincoln-lil Nov 22, 2021

lincoln-lil Nov 22, 2021

lincoln-lil Nov 22, 2021

lincoln-lil Nov 22, 2021

lincoln-lil Nov 22, 2021

JingsongLi Nov 25, 2021

JingsongLi commented Nov 25, 2021

lincoln-lil commented Nov 26, 2021

JingsongLi left a comment

[FLINK-20370][table] part1: Fix wrong results when sink primary key is not the same with query result's changelog upsert key #17699

[FLINK-20370][table] part1: Fix wrong results when sink primary key is not the same with query result's changelog upsert key #17699

Conversation

lincoln-lil commented Nov 5, 2021

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 Nov 5, 2021

Automated Checks

Review Progress

flinkbot commented Nov 5, 2021 • edited Loading

CI report:

JingsongLi commented Nov 8, 2021

lincoln-lil commented Nov 8, 2021

lincoln-lil commented Nov 10, 2021

zentol commented Nov 10, 2021

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lincoln-lil 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

JingsongLi commented Nov 25, 2021

lincoln-lil commented Nov 26, 2021

JingsongLi left a comment

Choose a reason for hiding this comment

flinkbot commented Nov 5, 2021 •

edited

Loading