You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2021/08/17 17:09:45 UTC
[GitHub] [iceberg] szehon-ho opened a new pull request #2989: Flink: Fix flaky test TestFlinkTableSink
szehon-ho opened a new pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989
* New files may be added if pass checkpoint interval (set to 400)
--
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
[GitHub] [iceberg] szehon-ho commented on pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#issuecomment-900498323
@stevenzwu @openinx can you guys help check my understanding? Thanks
--
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
[GitHub] [iceberg] openinx commented on pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#issuecomment-912924920
LGTM, thanks for the contribution!
--
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
[GitHub] [iceberg] openinx commented on a change in pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#discussion_r701748870
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -259,4 +265,37 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
return dataFiles;
}
+
+ public static Map<Long, List<DataFile>> snapshotToDataFiles(
+ Table table)
+ throws IOException {
Review comment:
Nit: The method format looks quite strange for me, could we format it looks like the following?
```
public static Map<Long, List<DataFile>> snapshotToDataFiles(Table table) throws IOException {
```
--
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
[GitHub] [iceberg] openinx commented on a change in pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#discussion_r701750871
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -259,4 +265,37 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
return dataFiles;
}
+
+ public static Map<Long, List<DataFile>> snapshotToDataFiles(
+ Table table)
+ throws IOException {
+ table.refresh();
+ Map<Long, List<DataFile>> result = Maps.newHashMap();
+ List<ManifestFile> manifestFiles = table.currentSnapshot().dataManifests();
+ for (ManifestFile manifestFile : manifestFiles) {
+ try (ManifestReader<DataFile> reader = ManifestFiles.read(manifestFile, table.io())) {
+ List<DataFile> dataFiles = IteratorUtils.toList(reader.iterator());
Review comment:
I see a warning in IDE here, using the `Lists.newArrayList(reader)` could fix the warning.
--
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
[GitHub] [iceberg] stevenzwu commented on a change in pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#discussion_r697796028
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -259,4 +265,37 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
return dataFiles;
}
+
+ public static Map<Long, List<DataFile>> snapshotToDataFiles(
+ Table table)
+ throws IOException {
+ table.refresh();
+ Map<Long, List<DataFile>> res = Maps.newHashMap();
Review comment:
nit: I believe the style is to use more complete word
res -> result
mf -> manifestFile
--
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
[GitHub] [iceberg] szehon-ho commented on pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#issuecomment-911921554
@stevenzwu thanks again for taking a look, I updated based on the feedback
--
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
[GitHub] [iceberg] stevenzwu commented on a change in pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#discussion_r697798211
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
##########
@@ -170,114 +171,121 @@ public void testOverwriteTable() throws Exception {
public void testReplacePartitions() throws Exception {
Assume.assumeFalse("Flink unbounded streaming does not support overwrite operation", isStreamingJob);
String tableName = "test_partition";
-
sql("CREATE TABLE %s(id INT, data VARCHAR) PARTITIONED BY (data) WITH ('write.format.default'='%s')",
tableName, format.name());
- Table partitionedTable = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));
-
- sql("INSERT INTO %s SELECT 1, 'a'", tableName);
- sql("INSERT INTO %s SELECT 2, 'b'", tableName);
- sql("INSERT INTO %s SELECT 3, 'c'", tableName);
-
- SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
- SimpleDataUtil.createRecord(1, "a"),
- SimpleDataUtil.createRecord(2, "b"),
- SimpleDataUtil.createRecord(3, "c")
- ));
-
- sql("INSERT OVERWRITE %s SELECT 4, 'b'", tableName);
- sql("INSERT OVERWRITE %s SELECT 5, 'a'", tableName);
-
- SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
- SimpleDataUtil.createRecord(5, "a"),
- SimpleDataUtil.createRecord(4, "b"),
- SimpleDataUtil.createRecord(3, "c")
- ));
-
- sql("INSERT OVERWRITE %s PARTITION (data='a') SELECT 6", tableName);
-
- SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
- SimpleDataUtil.createRecord(6, "a"),
- SimpleDataUtil.createRecord(4, "b"),
- SimpleDataUtil.createRecord(3, "c")
- ));
-
- sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, tableName);
+ try {
+ Table partitionedTable = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));
+
+ sql("INSERT INTO %s SELECT 1, 'a'", tableName);
+ sql("INSERT INTO %s SELECT 2, 'b'", tableName);
+ sql("INSERT INTO %s SELECT 3, 'c'", tableName);
+
+ SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
+ SimpleDataUtil.createRecord(1, "a"),
+ SimpleDataUtil.createRecord(2, "b"),
+ SimpleDataUtil.createRecord(3, "c")
+ ));
+
+ sql("INSERT OVERWRITE %s SELECT 4, 'b'", tableName);
+ sql("INSERT OVERWRITE %s SELECT 5, 'a'", tableName);
+
+ SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
+ SimpleDataUtil.createRecord(5, "a"),
+ SimpleDataUtil.createRecord(4, "b"),
+ SimpleDataUtil.createRecord(3, "c")
+ ));
+
+ sql("INSERT OVERWRITE %s PARTITION (data='a') SELECT 6", tableName);
+
+ SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
+ SimpleDataUtil.createRecord(6, "a"),
+ SimpleDataUtil.createRecord(4, "b"),
+ SimpleDataUtil.createRecord(3, "c")
+ ));
+ } finally {
+ sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, tableName);
+ }
}
@Test
public void testInsertIntoPartition() throws Exception {
String tableName = "test_insert_into_partition";
-
sql("CREATE TABLE %s(id INT, data VARCHAR) PARTITIONED BY (data) WITH ('write.format.default'='%s')",
tableName, format.name());
- Table partitionedTable = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));
-
- // Full partition.
- sql("INSERT INTO %s PARTITION (data='a') SELECT 1", tableName);
- sql("INSERT INTO %s PARTITION (data='a') SELECT 2", tableName);
- sql("INSERT INTO %s PARTITION (data='b') SELECT 3", tableName);
-
- SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
- SimpleDataUtil.createRecord(1, "a"),
- SimpleDataUtil.createRecord(2, "a"),
- SimpleDataUtil.createRecord(3, "b")
- ));
-
- // Partial partition.
- sql("INSERT INTO %s SELECT 4, 'c'", tableName);
- sql("INSERT INTO %s SELECT 5, 'd'", tableName);
-
- SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
- SimpleDataUtil.createRecord(1, "a"),
- SimpleDataUtil.createRecord(2, "a"),
- SimpleDataUtil.createRecord(3, "b"),
- SimpleDataUtil.createRecord(4, "c"),
- SimpleDataUtil.createRecord(5, "d")
- ));
-
- sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, tableName);
+ try {
+ Table partitionedTable = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));
+
+ // Full partition.
+ sql("INSERT INTO %s PARTITION (data='a') SELECT 1", tableName);
+ sql("INSERT INTO %s PARTITION (data='a') SELECT 2", tableName);
+ sql("INSERT INTO %s PARTITION (data='b') SELECT 3", tableName);
+
+ SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
+ SimpleDataUtil.createRecord(1, "a"),
+ SimpleDataUtil.createRecord(2, "a"),
+ SimpleDataUtil.createRecord(3, "b")
+ ));
+
+ // Partial partition.
+ sql("INSERT INTO %s SELECT 4, 'c'", tableName);
+ sql("INSERT INTO %s SELECT 5, 'd'", tableName);
+
+ SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
+ SimpleDataUtil.createRecord(1, "a"),
+ SimpleDataUtil.createRecord(2, "a"),
+ SimpleDataUtil.createRecord(3, "b"),
+ SimpleDataUtil.createRecord(4, "c"),
+ SimpleDataUtil.createRecord(5, "d")
+ ));
+ } finally {
+ sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, tableName);
+ }
}
@Test
public void testHashDistributeMode() throws Exception {
String tableName = "test_hash_distribution_mode";
-
Map<String, String> tableProps = ImmutableMap.of(
"write.format.default", format.name(),
TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.HASH.modeName()
);
sql("CREATE TABLE %s(id INT, data VARCHAR) PARTITIONED BY (data) WITH %s",
tableName, toWithClause(tableProps));
- // 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);
-
- Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));
- SimpleDataUtil.assertTableRecords(table, ImmutableList.of(
- SimpleDataUtil.createRecord(1, "aaa"),
- SimpleDataUtil.createRecord(1, "bbb"),
- SimpleDataUtil.createRecord(1, "ccc"),
- SimpleDataUtil.createRecord(2, "aaa"),
- SimpleDataUtil.createRecord(2, "bbb"),
- SimpleDataUtil.createRecord(2, "ccc"),
- SimpleDataUtil.createRecord(3, "aaa"),
- SimpleDataUtil.createRecord(3, "bbb"),
- SimpleDataUtil.createRecord(3, "ccc")
- ));
-
- Assert.assertEquals("There should be only 1 data file in partition 'aaa'", 1,
- SimpleDataUtil.partitionDataFiles(table, ImmutableMap.of("data", "aaa")).size());
- Assert.assertEquals("There should be only 1 data file in partition 'bbb'", 1,
- SimpleDataUtil.partitionDataFiles(table, ImmutableMap.of("data", "bbb")).size());
- Assert.assertEquals("There should be only 1 data file in partition 'ccc'", 1,
- SimpleDataUtil.partitionDataFiles(table, ImmutableMap.of("data", "ccc")).size());
-
- sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, tableName);
+ 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);
+
+ Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));
+ SimpleDataUtil.assertTableRecords(table, ImmutableList.of(
+ SimpleDataUtil.createRecord(1, "aaa"),
+ SimpleDataUtil.createRecord(1, "bbb"),
+ SimpleDataUtil.createRecord(1, "ccc"),
+ SimpleDataUtil.createRecord(2, "aaa"),
+ SimpleDataUtil.createRecord(2, "bbb"),
+ SimpleDataUtil.createRecord(2, "ccc"),
+ SimpleDataUtil.createRecord(3, "aaa"),
+ SimpleDataUtil.createRecord(3, "bbb"),
+ SimpleDataUtil.createRecord(3, "ccc")
+ ));
+
+ // Sometimes we will have more than one checkpoint if we pass the auto checkpoint interval
Review comment:
maybe expand the comments a little more to provide more context to reader.
```
// As a result, we may produce multiple snapshots. So we need to assert that we only produces 1 file for every partition per snapshot
```
--
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
[GitHub] [iceberg] stevenzwu edited a comment on pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
stevenzwu edited a comment on pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#issuecomment-901366806
@szehon-ho I am not sure why this failed. maybe it is what you said. records were split across two checkpoint/commit cycles. However, this change will defeat the purpose of this unit test. with hash distribution, this unit test is to make sure all records for the same partition key is distributed to a single writer task and hence should generate one file per partition (per commit cycle).
Here is probably what we should do to fix the flakiness. we should assert that there is only one file per partition for each commit/snapshot.
--
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
[GitHub] [iceberg] szehon-ho commented on a change in pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on a change in pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#discussion_r701301837
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -259,4 +265,37 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
return dataFiles;
}
+
+ public static Map<Long, List<DataFile>> snapshotToDataFiles(
+ Table table)
+ throws IOException {
+ table.refresh();
+ Map<Long, List<DataFile>> res = Maps.newHashMap();
+ List<ManifestFile> manifestFiles = table.currentSnapshot().allManifests();
+ for (ManifestFile mf : manifestFiles) {
+ try (ManifestReader<DataFile> reader = ManifestFiles.read(mf, table.io())) {
+ List<DataFile> dataFiles = IteratorUtils.toList(reader.iterator());
+ if (res.containsKey(mf.snapshotId())) {
+ res.get(mf.snapshotId()).addAll(dataFiles);
+ } else {
+ res.put(mf.snapshotId(), dataFiles);
+ }
+ }
+ }
+ return res;
+ }
+
+ public static List<DataFile> matchingPartitions(
+ List<DataFile> dataFiles, PartitionSpec partitionSpec, Map<String, Object> partitionValues) {
+ Types.StructType spec = partitionSpec.partitionType();
+ Record partitionRecord = GenericRecord.create(spec).copy(partitionValues);
+ StructLikeWrapper expected = StructLikeWrapper
Review comment:
That's a good idea, but when I tried I found PartitionData is package protected so can't access it here unfortunately.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
##########
@@ -170,114 +171,121 @@ public void testOverwriteTable() throws Exception {
public void testReplacePartitions() throws Exception {
Assume.assumeFalse("Flink unbounded streaming does not support overwrite operation", isStreamingJob);
String tableName = "test_partition";
-
sql("CREATE TABLE %s(id INT, data VARCHAR) PARTITIONED BY (data) WITH ('write.format.default'='%s')",
tableName, format.name());
- Table partitionedTable = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));
-
- sql("INSERT INTO %s SELECT 1, 'a'", tableName);
- sql("INSERT INTO %s SELECT 2, 'b'", tableName);
- sql("INSERT INTO %s SELECT 3, 'c'", tableName);
-
- SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
- SimpleDataUtil.createRecord(1, "a"),
- SimpleDataUtil.createRecord(2, "b"),
- SimpleDataUtil.createRecord(3, "c")
- ));
-
- sql("INSERT OVERWRITE %s SELECT 4, 'b'", tableName);
- sql("INSERT OVERWRITE %s SELECT 5, 'a'", tableName);
-
- SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
- SimpleDataUtil.createRecord(5, "a"),
- SimpleDataUtil.createRecord(4, "b"),
- SimpleDataUtil.createRecord(3, "c")
- ));
-
- sql("INSERT OVERWRITE %s PARTITION (data='a') SELECT 6", tableName);
-
- SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
- SimpleDataUtil.createRecord(6, "a"),
- SimpleDataUtil.createRecord(4, "b"),
- SimpleDataUtil.createRecord(3, "c")
- ));
-
- sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, tableName);
+ try {
+ Table partitionedTable = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));
+
+ sql("INSERT INTO %s SELECT 1, 'a'", tableName);
+ sql("INSERT INTO %s SELECT 2, 'b'", tableName);
+ sql("INSERT INTO %s SELECT 3, 'c'", tableName);
+
+ SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
+ SimpleDataUtil.createRecord(1, "a"),
+ SimpleDataUtil.createRecord(2, "b"),
+ SimpleDataUtil.createRecord(3, "c")
+ ));
+
+ sql("INSERT OVERWRITE %s SELECT 4, 'b'", tableName);
+ sql("INSERT OVERWRITE %s SELECT 5, 'a'", tableName);
+
+ SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
+ SimpleDataUtil.createRecord(5, "a"),
+ SimpleDataUtil.createRecord(4, "b"),
+ SimpleDataUtil.createRecord(3, "c")
+ ));
+
+ sql("INSERT OVERWRITE %s PARTITION (data='a') SELECT 6", tableName);
+
+ SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
+ SimpleDataUtil.createRecord(6, "a"),
+ SimpleDataUtil.createRecord(4, "b"),
+ SimpleDataUtil.createRecord(3, "c")
+ ));
+ } finally {
+ sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, tableName);
+ }
}
@Test
public void testInsertIntoPartition() throws Exception {
String tableName = "test_insert_into_partition";
-
sql("CREATE TABLE %s(id INT, data VARCHAR) PARTITIONED BY (data) WITH ('write.format.default'='%s')",
tableName, format.name());
- Table partitionedTable = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));
-
- // Full partition.
- sql("INSERT INTO %s PARTITION (data='a') SELECT 1", tableName);
- sql("INSERT INTO %s PARTITION (data='a') SELECT 2", tableName);
- sql("INSERT INTO %s PARTITION (data='b') SELECT 3", tableName);
-
- SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
- SimpleDataUtil.createRecord(1, "a"),
- SimpleDataUtil.createRecord(2, "a"),
- SimpleDataUtil.createRecord(3, "b")
- ));
-
- // Partial partition.
- sql("INSERT INTO %s SELECT 4, 'c'", tableName);
- sql("INSERT INTO %s SELECT 5, 'd'", tableName);
-
- SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
- SimpleDataUtil.createRecord(1, "a"),
- SimpleDataUtil.createRecord(2, "a"),
- SimpleDataUtil.createRecord(3, "b"),
- SimpleDataUtil.createRecord(4, "c"),
- SimpleDataUtil.createRecord(5, "d")
- ));
-
- sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, tableName);
+ try {
+ Table partitionedTable = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));
+
+ // Full partition.
+ sql("INSERT INTO %s PARTITION (data='a') SELECT 1", tableName);
+ sql("INSERT INTO %s PARTITION (data='a') SELECT 2", tableName);
+ sql("INSERT INTO %s PARTITION (data='b') SELECT 3", tableName);
+
+ SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
+ SimpleDataUtil.createRecord(1, "a"),
+ SimpleDataUtil.createRecord(2, "a"),
+ SimpleDataUtil.createRecord(3, "b")
+ ));
+
+ // Partial partition.
+ sql("INSERT INTO %s SELECT 4, 'c'", tableName);
+ sql("INSERT INTO %s SELECT 5, 'd'", tableName);
+
+ SimpleDataUtil.assertTableRecords(partitionedTable, Lists.newArrayList(
+ SimpleDataUtil.createRecord(1, "a"),
+ SimpleDataUtil.createRecord(2, "a"),
+ SimpleDataUtil.createRecord(3, "b"),
+ SimpleDataUtil.createRecord(4, "c"),
+ SimpleDataUtil.createRecord(5, "d")
+ ));
+ } finally {
+ sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, tableName);
+ }
}
@Test
public void testHashDistributeMode() throws Exception {
String tableName = "test_hash_distribution_mode";
-
Map<String, String> tableProps = ImmutableMap.of(
"write.format.default", format.name(),
TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.HASH.modeName()
);
sql("CREATE TABLE %s(id INT, data VARCHAR) PARTITIONED BY (data) WITH %s",
tableName, toWithClause(tableProps));
- // 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);
-
- Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));
- SimpleDataUtil.assertTableRecords(table, ImmutableList.of(
- SimpleDataUtil.createRecord(1, "aaa"),
- SimpleDataUtil.createRecord(1, "bbb"),
- SimpleDataUtil.createRecord(1, "ccc"),
- SimpleDataUtil.createRecord(2, "aaa"),
- SimpleDataUtil.createRecord(2, "bbb"),
- SimpleDataUtil.createRecord(2, "ccc"),
- SimpleDataUtil.createRecord(3, "aaa"),
- SimpleDataUtil.createRecord(3, "bbb"),
- SimpleDataUtil.createRecord(3, "ccc")
- ));
-
- Assert.assertEquals("There should be only 1 data file in partition 'aaa'", 1,
- SimpleDataUtil.partitionDataFiles(table, ImmutableMap.of("data", "aaa")).size());
- Assert.assertEquals("There should be only 1 data file in partition 'bbb'", 1,
- SimpleDataUtil.partitionDataFiles(table, ImmutableMap.of("data", "bbb")).size());
- Assert.assertEquals("There should be only 1 data file in partition 'ccc'", 1,
- SimpleDataUtil.partitionDataFiles(table, ImmutableMap.of("data", "ccc")).size());
-
- sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, tableName);
+ 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);
+
+ Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));
+ SimpleDataUtil.assertTableRecords(table, ImmutableList.of(
+ SimpleDataUtil.createRecord(1, "aaa"),
+ SimpleDataUtil.createRecord(1, "bbb"),
+ SimpleDataUtil.createRecord(1, "ccc"),
+ SimpleDataUtil.createRecord(2, "aaa"),
+ SimpleDataUtil.createRecord(2, "bbb"),
+ SimpleDataUtil.createRecord(2, "ccc"),
+ SimpleDataUtil.createRecord(3, "aaa"),
+ SimpleDataUtil.createRecord(3, "bbb"),
+ SimpleDataUtil.createRecord(3, "ccc")
+ ));
+
+ // Sometimes we will have more than one checkpoint if we pass the auto checkpoint interval
Review comment:
Updated comment
--
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
[GitHub] [iceberg] stevenzwu commented on pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#issuecomment-907559375
@szehon-ho thx a lot for fixing the flaky tests. overall, it looks good to me. Left a few minor comments.
--
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
[GitHub] [iceberg] szehon-ho commented on a change in pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on a change in pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#discussion_r701301452
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -259,4 +265,37 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
return dataFiles;
}
+
+ public static Map<Long, List<DataFile>> snapshotToDataFiles(
+ Table table)
+ throws IOException {
+ table.refresh();
+ Map<Long, List<DataFile>> res = Maps.newHashMap();
Review comment:
Done
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -259,4 +265,37 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
return dataFiles;
}
+
+ public static Map<Long, List<DataFile>> snapshotToDataFiles(
+ Table table)
+ throws IOException {
+ table.refresh();
+ Map<Long, List<DataFile>> res = Maps.newHashMap();
+ List<ManifestFile> manifestFiles = table.currentSnapshot().allManifests();
Review comment:
Done
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -259,4 +265,37 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
return dataFiles;
}
+
+ public static Map<Long, List<DataFile>> snapshotToDataFiles(
+ Table table)
+ throws IOException {
+ table.refresh();
+ Map<Long, List<DataFile>> res = Maps.newHashMap();
+ List<ManifestFile> manifestFiles = table.currentSnapshot().allManifests();
+ for (ManifestFile mf : manifestFiles) {
+ try (ManifestReader<DataFile> reader = ManifestFiles.read(mf, table.io())) {
+ List<DataFile> dataFiles = IteratorUtils.toList(reader.iterator());
+ if (res.containsKey(mf.snapshotId())) {
+ res.get(mf.snapshotId()).addAll(dataFiles);
+ } else {
+ res.put(mf.snapshotId(), dataFiles);
+ }
+ }
+ }
+ return res;
+ }
+
+ public static List<DataFile> matchingPartitions(
+ List<DataFile> dataFiles, PartitionSpec partitionSpec, Map<String, Object> partitionValues) {
+ Types.StructType spec = partitionSpec.partitionType();
Review comment:
Done
--
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
[GitHub] [iceberg] szehon-ho commented on pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#issuecomment-901470021
Yea this makes sense, let me look into it.
--
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
[GitHub] [iceberg] szehon-ho commented on a change in pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on a change in pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#discussion_r702168639
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -259,4 +265,37 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
return dataFiles;
}
+
+ public static Map<Long, List<DataFile>> snapshotToDataFiles(
+ Table table)
+ throws IOException {
Review comment:
Done
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -259,4 +265,37 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
return dataFiles;
}
+
+ public static Map<Long, List<DataFile>> snapshotToDataFiles(
+ Table table)
+ throws IOException {
+ table.refresh();
+ Map<Long, List<DataFile>> result = Maps.newHashMap();
+ List<ManifestFile> manifestFiles = table.currentSnapshot().dataManifests();
+ for (ManifestFile manifestFile : manifestFiles) {
+ try (ManifestReader<DataFile> reader = ManifestFiles.read(manifestFile, table.io())) {
+ List<DataFile> dataFiles = IteratorUtils.toList(reader.iterator());
Review comment:
Good catch, done
--
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
[GitHub] [iceberg] szehon-ho commented on pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#issuecomment-912821574
@openinx done, thanks for taking a look!
--
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
[GitHub] [iceberg] szehon-ho commented on pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#issuecomment-907277450
@stevenzwu thanks for the review. I put back the one-file-per-partition, per commit/checkpoint. Also I added some test cleanup as well in the failure case.
--
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
[GitHub] [iceberg] szehon-ho commented on pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#issuecomment-900498323
@stevenzwu @openinx can you guys help check my understanding? Thanks
--
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
[GitHub] [iceberg] stevenzwu edited a comment on pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
stevenzwu edited a comment on pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#issuecomment-901366806
@szehon-ho I am not sure why this failed. maybe it is what you said. records were split across two checkpoint/commit cycles. However, this change will defeat the purpose of this unit test. with hash distribution, this unit test is to make sure all records for the same partition key is distributed to a single writer task and hence should generate one file (per commit cycle).
Here is probably what we should do to fix the flakiness. we should assert that there is only one file per partition for each commit/snapshot.
--
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
[GitHub] [iceberg] stevenzwu commented on a change in pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#discussion_r697797868
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -259,4 +265,37 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
return dataFiles;
}
+
+ public static Map<Long, List<DataFile>> snapshotToDataFiles(
+ Table table)
+ throws IOException {
+ table.refresh();
+ Map<Long, List<DataFile>> res = Maps.newHashMap();
+ List<ManifestFile> manifestFiles = table.currentSnapshot().allManifests();
+ for (ManifestFile mf : manifestFiles) {
+ try (ManifestReader<DataFile> reader = ManifestFiles.read(mf, table.io())) {
+ List<DataFile> dataFiles = IteratorUtils.toList(reader.iterator());
+ if (res.containsKey(mf.snapshotId())) {
+ res.get(mf.snapshotId()).addAll(dataFiles);
+ } else {
+ res.put(mf.snapshotId(), dataFiles);
+ }
+ }
+ }
+ return res;
+ }
+
+ public static List<DataFile> matchingPartitions(
+ List<DataFile> dataFiles, PartitionSpec partitionSpec, Map<String, Object> partitionValues) {
+ Types.StructType spec = partitionSpec.partitionType();
+ Record partitionRecord = GenericRecord.create(spec).copy(partitionValues);
+ StructLikeWrapper expected = StructLikeWrapper
Review comment:
`PartitionData` implements the `equals` method. we can construct `PartitionData` using this API from `DataFiles` class. not sure if it is better. but at least it is more specific.
```
public static PartitionData copy(PartitionSpec spec, StructLike partition) {
return copyPartitionData(spec, partition, null);
}
```
--
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
[GitHub] [iceberg] stevenzwu commented on pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#issuecomment-901366806
@szehon-ho I am not sure why this failed. maybe it is what you said. records were split across two checkpoint/commit cycles. However, this change will defeat the purpose of this unit test. with hash distribution, this unit test is to make sure all records for the same partition key is distributed to a single writer task.
Here is probably what we should do to fix the flakiness. we should assert that there is only one file per partition for each commit/snapshot.
--
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
[GitHub] [iceberg] stevenzwu commented on a change in pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#discussion_r697796518
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -259,4 +265,37 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
return dataFiles;
}
+
+ public static Map<Long, List<DataFile>> snapshotToDataFiles(
+ Table table)
+ throws IOException {
+ table.refresh();
+ Map<Long, List<DataFile>> res = Maps.newHashMap();
+ List<ManifestFile> manifestFiles = table.currentSnapshot().allManifests();
+ for (ManifestFile mf : manifestFiles) {
+ try (ManifestReader<DataFile> reader = ManifestFiles.read(mf, table.io())) {
+ List<DataFile> dataFiles = IteratorUtils.toList(reader.iterator());
+ if (res.containsKey(mf.snapshotId())) {
+ res.get(mf.snapshotId()).addAll(dataFiles);
+ } else {
+ res.put(mf.snapshotId(), dataFiles);
+ }
+ }
+ }
+ return res;
+ }
+
+ public static List<DataFile> matchingPartitions(
+ List<DataFile> dataFiles, PartitionSpec partitionSpec, Map<String, Object> partitionValues) {
+ Types.StructType spec = partitionSpec.partitionType();
Review comment:
nit: spec -> partitionType
--
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
[GitHub] [iceberg] openinx merged pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
openinx merged pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989
--
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
[GitHub] [iceberg] stevenzwu commented on a change in pull request #2989: Flink: Fix flaky test TestFlinkTableSink
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2989:
URL: https://github.com/apache/iceberg/pull/2989#discussion_r697796111
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -259,4 +265,37 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
return dataFiles;
}
+
+ public static Map<Long, List<DataFile>> snapshotToDataFiles(
+ Table table)
+ throws IOException {
+ table.refresh();
+ Map<Long, List<DataFile>> res = Maps.newHashMap();
+ List<ManifestFile> manifestFiles = table.currentSnapshot().allManifests();
Review comment:
since we are only looking for `DataFile`, we probably should call `dataManifests` (not `allManifests`) here
--
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