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/10/25 09:19:18 UTC
[GitHub] [iceberg] openinx opened a new pull request #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
openinx opened a new pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365
**( The following are copied from https://github.com/apache/iceberg/pull/3307#issuecomment-946683261 )**
I got a case that would break the flaky unit test from [here](https://github.com/apache/iceberg/blob/7ff9eb4a49c836dc8da0768153d77f19f405d649/flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java#L275):
In the unit test case, we are trying to write the following records into apache iceberg table by shuffling by partition field `data` (The parallelism is 2):
```
(1, 'aaa'), (1, 'bbb'), (1, 'ccc')
(2, 'aaa'), (2, 'bbb'), (2, 'ccc')
(3, 'aaa'), (3, 'bbb'), (3, 'ccc')
```
As we may produces multiple checkpoints when the streaming job is running, Then it's possible that we write the records in the following checkpoints:
* checkpoint#1
* (1, 'aaa')
* (1, 'bbb')
* (1, 'ccc')
* checkpoint#2
* (2, 'aaa'),
* (2, 'bbb'),
* (2, 'ccc')
* (3, 'aaa'),
* (3, 'bbb'),
* (3, 'ccc')
Then it will produces a seperate data file for each partition in the given checkpoint. Let's say:
* checkpoint#1
* produces `data-file-1` for partition `aaa`
* produces `data-file-2` for partition `bbb`
* produces `data-file-3` for partition `ccc`
* checkpoint#2
* produces `data-file-4` for partition `aaa`
* produces `data-file-5` for partition `bbb`
* produces `data-file-6` for partition `ccc`
In the [IcebergFilesCommitter](https://github.com/apache/iceberg/blob/0abaa7ce670f74493cdc5aad43ec56c25d4a54d3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java#L266) , we will use the default MergeAppend to merge the manifest. Then we will produces:
* checkpoint#1
* manifest-1: includes `data-file-1`, `data-file-2`, `data-file-3`
* checkpoint#2
* manifest-2: includes `data-file-1`, `data-file-2`, `data-file-3`, `data-file-4`, `data-file-5`, `data-file-6`
Then finally, in this [line](https://github.com/apache/iceberg/blob/7ff9eb4a49c836dc8da0768153d77f19f405d649/flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java#L275), we will encounter the `data-file1` twice in the `result` map. Finally, the assert would be failure.
I think we only need to find out the newly added data files for each given commit for fixing this unit test purpose.
--
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 #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on a change in pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#discussion_r735788226
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -267,17 +268,28 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
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 = Lists.newArrayList(reader);
- if (result.containsKey(manifestFile.snapshotId())) {
- result.get(manifestFile.snapshotId()).addAll(dataFiles);
- } else {
- result.put(manifestFile.snapshotId(), dataFiles);
- }
+ Snapshot current = table.currentSnapshot();
Review comment:
I think here ManifestEntry interface and methods are package-protected, hence they can't be used outside (As I mentioned over at the other thread, also looking to fix the original code).
And so was wondering, whether the interface can be exposed as public? I was curious the original reason why it was kept package protected, as "Entries" metadata table is already publically exposed via Spark to users, and it's already pubically documented on the spec.
--
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] zhongyujiang commented on pull request #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
zhongyujiang commented on pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#issuecomment-1020830310
https://github.com/apache/iceberg/blob/ca46fc93ae9a349e3920d0519ccf0f41e7210d44/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java#L261-L277
V1 table will merge results of multiple ckpts into one transaction.
--
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] rdblue commented on a change in pull request #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#discussion_r735755404
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -267,17 +268,28 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
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 = Lists.newArrayList(reader);
- if (result.containsKey(manifestFile.snapshotId())) {
- result.get(manifestFile.snapshotId()).addAll(dataFiles);
- } else {
- result.put(manifestFile.snapshotId(), dataFiles);
- }
+ Snapshot current = table.currentSnapshot();
Review comment:
I think it would be better to modify the old code rather than traversing snapshots and planning scans. All you need to do is to read the manifest entries and add the data files to the map using each entry's snapshot ID instead of the manifest's snapshot ID. You can use `ManifestGroup.entries` to get the entries.
--
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] zhongyujiang edited a comment on pull request #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
zhongyujiang edited a comment on pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#issuecomment-1020830310
--
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 #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#issuecomment-951534337
> @openinx, why would a manifest merge be triggered? It seems like this should not happen, or should happen consistently. I don't understand why this could cause a test to be flaky.
Reconsidered this question and checked[ the condition to trigger real manifest file merge](https://github.com/apache/iceberg/blob/90225d6c9413016d611e2ce5eff37db1bc1b4fc5/core/src/main/java/org/apache/iceberg/ManifestMergeManager.java#L134) in MergeAppend, in this testHashDistributeMode unit tests, we will produce 9 data files at most, so it's unlikely to trigger the manifests merge ( the default threshold is 100) .
```java
public static final String MANIFEST_MIN_MERGE_COUNT = "commit.manifest.min-count-to-merge";
public static final int MANIFEST_MIN_MERGE_COUNT_DEFAULT = 100;
```
I also wrote a small case to verify this:
```java
@RunWith(Parameterized.class)
public class TestSimpleDataUtil extends TableTestBase {
@Parameterized.Parameters(name = "formatVersion = {0}")
public static Object[] parameters() {
return new Object[] {1}; // We don't actually use the format version since everything is mock
}
public TestSimpleDataUtil(int formatVersion) {
super(formatVersion);
}
@Test
public void testDataFiles() throws IOException {
table.newAppend()
.appendFile(FILE_A)
.commit();
table.newAppend()
.appendFile(FILE_B)
.commit();
table.newAppend()
.appendFile(FILE_C)
.commit();
Map<Long, List<DataFile>> files = SimpleDataUtil.snapshotToDataFiles(table);
Assert.assertEquals(3, files.size());
for (Map.Entry<Long, List<DataFile>> entry : files.entrySet()) {
Assert.assertEquals(1, entry.getValue().size());
}
}
}
```
So I think @rdblue is right, we still don't get the real root cause why does it break the UT. Will need more careful work to catch 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] rdblue commented on pull request #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#issuecomment-951078308
@openinx, why would a manifest merge be triggered? It seems like this should not happen, or should happen consistently. I don't understand why this could cause a test to be flaky.
--
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 #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on a change in pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#discussion_r735788226
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -267,17 +268,28 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
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 = Lists.newArrayList(reader);
- if (result.containsKey(manifestFile.snapshotId())) {
- result.get(manifestFile.snapshotId()).addAll(dataFiles);
- } else {
- result.put(manifestFile.snapshotId(), dataFiles);
- }
+ Snapshot current = table.currentSnapshot();
Review comment:
I think here ManifestEntry interface and methods are package-protected, hence they can't be used outside (As I mentioned over at the other thread, also looking to fix the original code).
And so was wondering, whether the interface can be exposed as public? I was sure the original reason why it was kept package protected, as "Entries" metadata table is already publically exposed via Spark to users, and it's already pubically documented on the spec.
--
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] rdblue commented on pull request #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#issuecomment-951407789
@szehon-ho, I see that there could be multiple data files here, but I don't quite see how the current code would get it wrong. Still, being more careful and making sure the snapshot ID matches the actual data file is likely to fix it if that's the problem.
--
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] rdblue commented on a change in pull request #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#discussion_r736032154
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -267,17 +268,28 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
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 = Lists.newArrayList(reader);
- if (result.containsKey(manifestFile.snapshotId())) {
- result.get(manifestFile.snapshotId()).addAll(dataFiles);
- } else {
- result.put(manifestFile.snapshotId(), dataFiles);
- }
+ Snapshot current = table.currentSnapshot();
Review comment:
The reason why we don't expose the manifest entry is that it's confusing. We don't want users to read a manifest and assume that all of the entries represent files in the table because we track deleted files in the same metadata. So it is more that users would need to know more about the spec and we don't think that it is likely to be used correctly.
I'm still open to the idea of making this public. But if we don't need it then I'd opt not t.
--
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] rdblue commented on a change in pull request #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#discussion_r736032154
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -267,17 +268,28 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
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 = Lists.newArrayList(reader);
- if (result.containsKey(manifestFile.snapshotId())) {
- result.get(manifestFile.snapshotId()).addAll(dataFiles);
- } else {
- result.put(manifestFile.snapshotId(), dataFiles);
- }
+ Snapshot current = table.currentSnapshot();
Review comment:
The reason why we don't expose the manifest entry is that it's confusing. We don't want users to read a manifest and assume that all of the entries represent files in the table because we track deleted files in the same metadata. So it is more that users would need to know more about the spec and we don't think that it is likely to be used correctly.
I'm still open to the idea of making this public. But if we don't need it then I'd opt not to.
--
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] zhongyujiang commented on pull request #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
zhongyujiang commented on pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#issuecomment-1020781584
Hi @openinx, I am still seeing this error in CI sometimes, I think this is maybe caused by `notifyCheckpointComplete` pattern in IcebergFilesCommitter:
```
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
// It's possible that we have the following events:
// 1. snapshotState(ckpId);
// 2. snapshotState(ckpId+1);
// 3. notifyCheckpointComplete(ckpId+1);
// 4. notifyCheckpointComplete(ckpId);
// For step#4, we don't need to commit iceberg table again because in step#3 we've committed all the files,
// Besides, we need to maintain the max-committed-checkpoint-id to be increasing.
if (checkpointId > maxCommittedCheckpointId) {
commitUpToCheckpoint(dataFilesPerCheckpoint, flinkJobId, checkpointId);
this.maxCommittedCheckpointId = checkpointId;
}
}
```
My understanding is that not every ckpt of Flink corresponds to a commit in Iceberg, sometimes a snapshot may contains data files of several ckpts. And the checkpoint interval (400 ms) here is rather small, which makes this situation very likely. What do you think?
--
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] zhongyujiang edited a comment on pull request #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
zhongyujiang edited a comment on pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#issuecomment-1020830310
https://github.com/apache/iceberg/blob/ca46fc93ae9a349e3920d0519ccf0f41e7210d44/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java#L261-L277
Results of ckpts without delete file will be merged into one transaction.
--
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] zhongyujiang commented on pull request #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
zhongyujiang commented on pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#issuecomment-1020781584
--
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 #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on a change in pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#discussion_r735788226
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -267,17 +268,28 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
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 = Lists.newArrayList(reader);
- if (result.containsKey(manifestFile.snapshotId())) {
- result.get(manifestFile.snapshotId()).addAll(dataFiles);
- } else {
- result.put(manifestFile.snapshotId(), dataFiles);
- }
+ Snapshot current = table.currentSnapshot();
Review comment:
I think here ManifestEntry interface and methods are package-protected, hence they can't be used outside (As I mentioned over at the other thread, when I was also looking to fix the original code).
And so was wondering, whether the interface can be exposed as public? I was curious the original reason why it was kept package protected, as "Entries" metadata table is already publically exposed via Spark to users, and it's already publically documented on the spec.
--
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] zhongyujiang edited a comment on pull request #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
zhongyujiang edited a comment on pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#issuecomment-1020781584
Hi @openinx, I am still seeing this error in CI sometimes, I think this is maybe caused by the way `notifyCheckpointComplete` works in IcebergFilesCommitter:
```
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
// It's possible that we have the following events:
// 1. snapshotState(ckpId);
// 2. snapshotState(ckpId+1);
// 3. notifyCheckpointComplete(ckpId+1);
// 4. notifyCheckpointComplete(ckpId);
// For step#4, we don't need to commit iceberg table again because in step#3 we've committed all the files,
// Besides, we need to maintain the max-committed-checkpoint-id to be increasing.
if (checkpointId > maxCommittedCheckpointId) {
commitUpToCheckpoint(dataFilesPerCheckpoint, flinkJobId, checkpointId);
this.maxCommittedCheckpointId = checkpointId;
}
}
```
My understanding is that not every ckpt of Flink corresponds to a commit in Iceberg, sometimes a snapshot may contains data files of several ckpts. And the checkpoint interval (400 ms) here is rather small, which makes this situation very likely. What do you think?
--
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 #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#issuecomment-951113629
@rdblue for context, the possible issue is that FlinkSinks have auto-commit , and so sometimes there is a commit between the two writes of the same record value and thus two files are created. The test (checking that same records go to same file) was asserting only one file created.
Thanks @openinx for giving a try to fix 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] rdblue merged pull request #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
rdblue merged pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365
--
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] rdblue commented on pull request #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#issuecomment-952130831
@openinx, thanks for looking into it. Even if merge isn't the cause, having the wrong snapshot ID in the map could cause an issue so it's good to fix this and see if it helps. Thanks for working on this!
--
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 #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on a change in pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#discussion_r735788226
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -267,17 +268,28 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
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 = Lists.newArrayList(reader);
- if (result.containsKey(manifestFile.snapshotId())) {
- result.get(manifestFile.snapshotId()).addAll(dataFiles);
- } else {
- result.put(manifestFile.snapshotId(), dataFiles);
- }
+ Snapshot current = table.currentSnapshot();
Review comment:
I think here ManifestEntry interface and methods are package-protected, hence they can't be used outside (As I mentioned over at the other thread, also looking to fix the original code).
And so was wondering, whether the interface can be exposed as public? I was curious the original reason why it was kept package protected, as "Entries" metadata table is already publically exposed via Spark to users, and it's already publically documented on the spec.
--
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 #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#issuecomment-951534657
Thanks @szehon-ho and @stevenzwu for the explaining and check.
--
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 #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on a change in pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#discussion_r736041815
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -267,17 +268,28 @@ public static StructLikeSet actualRowSet(Table table, Long snapshotId, String...
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 = Lists.newArrayList(reader);
- if (result.containsKey(manifestFile.snapshotId())) {
- result.get(manifestFile.snapshotId()).addAll(dataFiles);
- } else {
- result.put(manifestFile.snapshotId(), dataFiles);
- }
+ Snapshot current = table.currentSnapshot();
Review comment:
I get what you mean. If you are curious, a real life use case for ManifestEntry we designed was a bit similar to this.
We were trying to build a Data Latency monitoring application that measures max data latency per partition. So we wanted to go through snapshots , then search reachable manifest entries for all ADDED data files matching a partition, and find the latest commit time from them all.
We end up trying to join snapshot + all_entries metadata tables, but due to perf issues and bugs with metadata tables aggregation, started to look at Table and ManifestReader API to explore a set of known snapshots/manifest files directly, as we kind of knew what time frame the partition was expected to land at latest. But without knowing the context of DataFile, this solution is hard to get working (DataFile in EXISTING mode resulting from metadata rewrite throws it off).
Not sure if there was an easier way to do 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] stevenzwu commented on pull request #3365: Flink: Fix the flaky test TestFlinkSink#testHashDistributeMode
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #3365:
URL: https://github.com/apache/iceberg/pull/3365#issuecomment-963755128
I am seeing the same issue in the CI for my PR: https://github.com/apache/iceberg/runs/4146901429?check_suite_focus=true
I have rebased the branch with the latest master. so it has this change already.
--
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