leonardBang commented on a change in pull request #14027:
URL: https://github.com/apache/flink/pull/14027#discussion_r521808720
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
##########
@@ -57,15 +57,19 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
// step2: satisfy UpdateKind trait
val rootModifyKindSet = getModifyKindSet(rootWithModifyKindSet)
// use the required UpdateKindTrait from parent blocks
- val requiredUpdateKindTraits = if (context.isUpdateBeforeRequired) {
- Seq(UpdateKindTrait.BEFORE_AND_AFTER)
- } else if (rootModifyKindSet.isInsertOnly) {
- Seq(UpdateKindTrait.NONE)
+ val requiredUpdateKindTraits = if
(rootModifyKindSet.contains(ModifyKind.UPDATE)) {
+ if (context.isUpdateBeforeRequired) {
+ Seq(UpdateKindTrait.BEFORE_AND_AFTER)
+ } else {
+ // update_before is not required, and input contains updates
+ // try ONLY_UPDATE_AFTER first, and then BEFORE_AND_AFTER
+ Seq(UpdateKindTrait.ONLY_UPDATE_AFTER,
UpdateKindTrait.BEFORE_AND_AFTER)
+ }
} else {
- // update_before is not required, and input contains updates
- // try ONLY_UPDATE_AFTER first, and then BEFORE_AND_AFTER
- Seq(UpdateKindTrait.ONLY_UPDATE_AFTER, UpdateKindTrait.BEFORE_AND_AFTER)
+ // there is no updates
+ Seq(UpdateKindTrait.NONE)
}
+
Review comment:
delete this line
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecJoin.scala
##########
@@ -57,6 +57,22 @@ class StreamExecJoin(
with StreamPhysicalRel
with StreamExecNode[RowData] {
+ /**
+ * This is mainly used in
`FlinkChangelogModeInferenceProgram.SatisfyUpdateKindTraitVisitor`.
+ * If the unique key of input contains join key, then it can support
ignoring UPDATE_BEFORE.
+ * Otherwise, it can't ignore UPDATE_BEFORE. For example, if the input
schema is [id, name, cnt]
+ * with the unique key (id). The join key is (id, name), then an insert and
update on the id:
Review comment:
```
* with the unique key (id). The join key is (id, name), then an insert
and update on the id:
```
I think current logic is unique key contains join key rather than unique key
is contained in join key, may this is an existed bug
`inputUniqueKeys.exists {
uniqueKey => joinKeys.forall(uniqueKey.toArray.contains(_))
}`
##########
File path:
flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.scala
##########
@@ -281,25 +281,40 @@ class TableScanTest extends TableTestBase {
@Test
def testJoinOnChangelogSource(): Unit = {
+ verifyJoinOnSource("I,UB,UA")
+ }
+
+ @Test
+ def testJoinOnNoUpdateSource(): Unit = {
+ verifyJoinOnSource("I,D")
+ }
+
+ @Test
+ def testJoinOnUpsertSource(): Unit = {
+ verifyJoinOnSource("UA,D")
+ }
+
+ private def verifyJoinOnSource(changelogMode: String): Unit = {
util.addTable(
"""
- |CREATE TABLE orders (
- | amount BIGINT,
- | currency STRING
- |) WITH (
- | 'connector' = 'values',
- | 'changelog-mode' = 'I'
- |)
- |""".stripMargin)
+ |CREATE TABLE orders (
+ | amount BIGINT,
+ | currency STRING
+ |) WITH (
+ | 'connector' = 'values',
+ | 'changelog-mode' = 'I'
+ |)
+ |""".stripMargin)
util.addTable(
- """
- |CREATE TABLE rates_history (
- | currency STRING,
- | rate BIGINT
- |) WITH (
- | 'connector' = 'values',
- | 'changelog-mode' = 'I,UB,UA'
- |)
+ s"""
+ |CREATE TABLE rates_history (
+ | currency STRING,
+ | rate BIGINT,
+ | PRIMARY KEY (currency) NOT ENFORCED
Review comment:
could you change the join key to `(currency, currency_no)` to check the
join key contains unique key case.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
