chenjunjiedada commented on code in PR #4670:
URL: https://github.com/apache/iceberg/pull/4670#discussion_r863599148
##########
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:
Also, I would suggest removing table level configuration as I mentioned
before since in real cases we should always have offline service to optimize
tables which means two writers with different writing models.
--
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]
