snuyanzin commented on code in PR #29697:
URL: https://github.com/apache/flink/pull/29697#discussion_r4436596079
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/JoinITCase.scala:
##########
@@ -697,6 +697,74 @@ class JoinITCase(miniBatch: MiniBatchMode, state:
StateBackendMode, enableAsyncS
assertThat(sink.getRetractResults.sorted).isEqualTo(expected.sorted)
}
+ /**
+ * user_regions is unique on its join columns (user_id, region), but orders
is joined on user_id
+ * only, so an order matches the user's row of every region and its key is
not unique in the
+ * output. Deleting one region must keep the order's row for the other
region.
+ */
+ @TestTemplate
+ def testMultiJoinKeyNotUniqueWhenJoinedOnFewerColumns(): Unit = {
+ tEnv.getConfig.getConfiguration
+
.setString(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED.key(),
"true")
+ val regions = TestValuesTableFactory.registerData(
+ Seq(
+ changelogRow("+I", "u1", "EU", "Ann"),
+ changelogRow("+I", "u1", "US", "Ann"),
+ changelogRow("-D", "u1", "EU", "Ann")))
+ val orders = TestValuesTableFactory.registerData(Seq(changelogRow("+I",
"o1", "u1", "book")))
+ val addresses = TestValuesTableFactory.registerData(
+ Seq(changelogRow("+I", "u1", "EU", "Berlin"), changelogRow("+I", "u1",
"US", "NYC")))
+ // the regions source sleeps after each row, so the delete comes after
both regions joined
+ tEnv.executeSql(s"""
+ |CREATE TABLE user_regions (
+ | user_id STRING, region STRING, name STRING,
+ | PRIMARY KEY (user_id, region) NOT ENFORCED
+ |) WITH (
+ | 'connector' = 'values',
+ | 'data-id' = '$regions',
+ | 'changelog-mode' = 'I,D',
+ | 'disable-lookup' = 'true',
+ | 'enable-watermark-push-down' = 'true',
+ | 'source.sleep-after-elements' = '1',
+ | 'source.sleep-time' = '500ms'
+ |)
+ |""".stripMargin)
+ tEnv.executeSql(
+ s"""
+ |CREATE TABLE orders (
+ | order_id STRING, user_id STRING, product STRING,
+ | PRIMARY KEY (order_id, user_id) NOT ENFORCED
+ |) WITH ('connector' = 'values', 'data-id' = '$orders',
'changelog-mode' = 'I,D')
Review Comment:
any reason to have different change log than in MultiJoinTest?
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
