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: [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]
