[GitHub] [iceberg] yittg commented on a change in pull request #4189: Flink 1.14: Fix the flaky testHashDistributeMode by controlling rows in one checkpoint precisely

[GitBox <gi...@apache.org>]

yittg commented on a change in pull request #4189:
URL: https://github.com/apache/iceberg/pull/4189#discussion_r811748307



##########
File path: flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
##########
@@ -253,15 +258,24 @@ public void testHashDistributeMode() throws Exception {
         "write.format.default", format.name(),
         TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.HASH.modeName()
     );
+
+    // Initialize a BoundedSource table to precisely control that those 9 rows will be emitted in only one checkpoint.
+    List<Row> dataSet = ImmutableList.of(
+        Row.of(1, "aaa"), Row.of(1, "bbb"), Row.of(1, "ccc"),
+        Row.of(2, "aaa"), Row.of(2, "bbb"), Row.of(2, "ccc"),
+        Row.of(3, "aaa"), Row.of(3, "bbb"), Row.of(3, "ccc"));
+    String dataId = BoundedTableFactory.registerDataSet(ImmutableList.of(dataSet));

Review comment:
       shall we produce more than one checkpoint? and add enough records in each part instead of enumerating them?

##########
File path: flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
##########
@@ -253,15 +258,24 @@ public void testHashDistributeMode() throws Exception {
         "write.format.default", format.name(),
         TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.HASH.modeName()
     );
+
+    // Initialize a BoundedSource table to precisely control that those 9 rows will be emitted in only one checkpoint.
+    List<Row> dataSet = ImmutableList.of(
+        Row.of(1, "aaa"), Row.of(1, "bbb"), Row.of(1, "ccc"),
+        Row.of(2, "aaa"), Row.of(2, "bbb"), Row.of(2, "ccc"),
+        Row.of(3, "aaa"), Row.of(3, "bbb"), Row.of(3, "ccc"));
+    String dataId = BoundedTableFactory.registerDataSet(ImmutableList.of(dataSet));
+    sql("CREATE TABLE %s(id INT NOT NULL, data STRING NOT NULL)" +
+        " WITH ('connector'='BoundedSource', 'data-id'='%s')", SOURCE_TABLE, dataId);
+    Assert.assertEquals("Should have the expected rows", Sets.newHashSet(dataSet),
+        Sets.newHashSet(sql("SELECT * FROM %s", SOURCE_TABLE)));
+
     sql("CREATE TABLE %s(id INT, data VARCHAR) PARTITIONED BY (data) WITH %s",
         tableName, toWithClause(tableProps));
 
     try {
       // Insert data set.
-      sql("INSERT INTO %s VALUES " +
-          "(1, 'aaa'), (1, 'bbb'), (1, 'ccc'), " +
-          "(2, 'aaa'), (2, 'bbb'), (2, 'ccc'), " +
-          "(3, 'aaa'), (3, 'bbb'), (3, 'ccc')", tableName);
+      sql("INSERT INTO %s SELECT * FROM %s", tableName, SOURCE_TABLE);
 
       Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));
       SimpleDataUtil.assertTableRecords(table, ImmutableList.of(

Review comment:
       check records based on `dataSet`?




-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org