mxm commented on code in PR #18717:
URL: https://github.com/apache/iceberg/pull/18717#discussion_r4524366736
##########
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:
I'm assuming you are referring to the block above the the `else`. Removed
that edge case from the test.
--
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]
