pvary commented on code in PR #18717:
URL: https://github.com/apache/iceberg/pull/18717#discussion_r4524313549
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java:
##########
@@ -968,6 +1384,138 @@ void bootstrapsAgainAfterRestoreFromFailedBootstrap()
throws Exception {
}
}
+ @Test
+ void rebuildsIndexWhenDeltaFails() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.open();
+
+ sendTrigger(harness);
+
+ // Two external appends. The manifest list of the older one is lost, so
every delta fails,
+ // while a rebuild reads only the head.
+ table.newAppend().appendFile(writeDataFile(table, createRecord(2,
"b"))).commit();
+ table.refresh();
+ List<File> olderManifestList =
+ Lists.newArrayList(
+ new
File(table.currentSnapshot().manifestListLocation().replace("file:", "")));
+ table.newAppend().appendFile(writeDataFile(table, createRecord(3,
"c"))).commit();
+ table.refresh();
+ hide(olderManifestList);
+
+ sendTrigger(harness);
+
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+ int afterSecond = harness.extractOutputValues().size();
+
+ // The next trigger rebuilds the index instead of repeating the delta.
+ sendTrigger(harness);
+
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+ List<ReadCommand> all = harness.extractOutputValues();
+ assertThat(all.subList(afterSecond, all.size()))
+ .filteredOn(c -> c.task() instanceof FileScanTask)
+ .extracting(TestEqualityConvertPlanner::filePath)
+ .containsExactlyInAnyOrderElementsOf(liveDataFiles(table));
+ // The rebuild also evicts the keys the failed delta left, at the
generation after the first
+ // build's.
+
assertThat(harness.getSideOutput(EqualityConvertPlanner.CLEAR_BROADCAST_STREAM))
+ .extracting(record -> record.getValue().indexGeneration())
+ .containsExactly(1L, 2L);
+ restore(olderManifestList);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void rebuildsIndexWhenDeltaAddsRemovedFileAgain(boolean sameCommit) throws
Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+ DataFile reused =
table.currentSnapshot().addedDataFiles(table.io()).iterator().next();
+ // Untouched rows keep the delta smaller than a rebuild.
+ insert(table, 10, "x");
+ insert(table, 11, "y");
+ insert(table, 12, "z");
+
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.open();
+
+ sendTrigger(harness);
+
+ // Removes the indexed file and adds it again, in two commits or in one.
+ if (sameCommit) {
+ table.newOverwrite().deleteFile(reused).addFile(reused).commit();
+ } else {
+ table.newDelete().deleteFile(reused).commit();
+ table.newAppend().appendFile(reused).commit();
+ }
Review Comment:
This is against the spec. We can keep and comment, or remove the test.
> Writers should not produce multiple manifest entries for the same content
file in a snapshot (for example, both ADDED and DELETED entries for the same
file)
--
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]
