chenjunjiedada commented on code in PR #4670:
URL: https://github.com/apache/iceberg/pull/4670#discussion_r863595348
##########
flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/TestChangeLogTable.java:
##########
@@ -235,17 +236,59 @@ public void testPureInsertOnIdKey() throws Exception {
)
);
- testSqlChangeLog(TABLE_NAME, ImmutableList.of("data"),
elementsPerCheckpoint, expectedRecords);
+ testSqlChangeLog(TABLE_NAME, ImmutableList.of("data"),
elementsPerCheckpoint, expectedRecords, false);
+ }
+
+ @Test
+ public void testUpsertOnIdKey() throws Exception {
+ List<List<Row>> elementsPerCheckpoint = ImmutableList.of(
+ ImmutableList.of(
+ insertRow(1, "aaa")
+ ),
+ ImmutableList.of(
+ updateBeforeRow(1, "aaa"),
+ updateAfterRow(1, "bbb")
+ ),
+ ImmutableList.of(
+ updateBeforeRow(1, "bbb"),
+ updateAfterRow(1, "ccc")
+ ),
+ ImmutableList.of(
+ updateBeforeRow(1, "ccc"),
+ updateAfterRow(1, "ddd"),
+ updateBeforeRow(1, "ddd"),
+ updateAfterRow(1, "eee")
+ )
+ );
+
+ List<List<Row>> expectedRecords = ImmutableList.of(
+ ImmutableList.of(insertRow(1, "aaa")),
+ ImmutableList.of(insertRow(1, "bbb")),
+ ImmutableList.of(insertRow(1, "ccc")),
+ ImmutableList.of(insertRow(1, "eee"))
+ );
+
+ if (!partitioned) {
+ testSqlChangeLog(TABLE_NAME, ImmutableList.of("id"),
elementsPerCheckpoint, expectedRecords, true);
+ } else {
+ AssertHelpers.assertThrows("Should be error because equality field
columns don't include all partition keys",
+ IllegalStateException.class, "should be included in equality fields",
+ () -> {
+ testSqlChangeLog(TABLE_NAME, ImmutableList.of("id"),
elementsPerCheckpoint, expectedRecords, true);
+ return null;
+ });
+ }
}
private static Record record(int id, String data) {
return SimpleDataUtil.createRecord(id, data);
}
- private Table createTable(String tableName, List<String> key, boolean
isPartitioned) {
+ private Table createTable(String tableName, List<String> key, boolean
isPartitioned, boolean upsertEnabled) {
String partitionByCause = isPartitioned ? "PARTITIONED BY (data)" : "";
- sql("CREATE TABLE %s(id INT, data VARCHAR, PRIMARY KEY(%s) NOT ENFORCED)
%s",
- tableName, Joiner.on(',').join(key), partitionByCause);
+ sql("CREATE TABLE %s(id INT, data VARCHAR, PRIMARY KEY(%s) NOT ENFORCED)
%s " +
+ "WITH ('write-upsert-enabled'='%s')",
+ tableName, Joiner.on(',').join(key), partitionByCause, upsertEnabled ?
"true" : "false");
Review Comment:
The `write-upsert-enabled` is the option of the Flink logic table and the
`write.upsert.enabled` is the property of the iceberg table. Right now in this
unit test, it creates the table via Flink catalog and thus respects the job
level option.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
