amogh-jahagirdar commented on code in PR #18408:
URL: https://github.com/apache/iceberg/pull/18408#discussion_r4254056705
##########
core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java:
##########
@@ -335,4 +355,47 @@ private Set<String> findFilesToDelete(
return filesToDelete;
}
+
+ // A Puffin file can be shared by multiple deletion vectors, so keep it if
any live one remains
+ private Set<String> unreferencedDeletionVectors(
+ Set<String> deletionVectorsToDelete,
+ Collection<ManifestFile> liveDeleteManifests,
+ Map<Integer, PartitionSpec> specsById) {
+ if (deletionVectorsToDelete.isEmpty()) {
+ return deletionVectorsToDelete;
+ }
+
+ Set<String> unreferenced = ConcurrentHashMap.newKeySet();
+ unreferenced.addAll(deletionVectorsToDelete);
+
+ try {
+ Tasks.foreach(liveDeleteManifests)
+ .retry(3)
+ .stopOnFailure()
+ .throwFailureWhenFinished()
+ .executeWith(planExecutorService)
+ .onFailure(
+ (item, exc) ->
+ LOG.warn(
+ "Failed to determine live deletion vectors in manifest
{}. Retrying",
+ item.path(),
+ exc))
+ .run(
+ manifest -> {
+ try (CloseableIterable<String> paths =
+ ManifestFiles.readPaths(manifest, fileIO, specsById)) {
+ for (String path : paths) {
+ unreferenced.remove(path);
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to read manifest
file: %s", manifest);
+ }
+ });
+ } catch (Throwable e) {
+ LOG.warn("Failed to list all live deletion vectors, retaining deletion
vector files", e);
Review Comment:
"Failed to determine live DVs, skipping DV cleanup" would be a bit cleaner
of a user facing error message imo or something like that, main points are
around avoiding the term "list" as that may be confusing and relaying the main
consequence to the user which is DVs are still hanging around.
##########
core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java:
##########
@@ -2144,6 +2144,105 @@ public void testCleanupLevelNullValidation() {
.hasMessageContaining("Invalid cleanup level: null");
}
+ @TestTemplate
+ public void testExpireDoesNotDeletePuffinFileSharedByLiveDeletionVectors()
throws Exception {
+ assumeThat(formatVersion).as("Deletion vectors require V3 or
later").isGreaterThanOrEqualTo(3);
+
+ table.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
+
+ // one Puffin file with deletion vectors for FILE_A and FILE_B
+ String sharedPuffinPath = "/path/to/shared-dvs-" + UUID.randomUUID() +
".puffin";
+ DeleteFile dvA = deletionVector(sharedPuffinPath, FILE_A, 4L);
+ DeleteFile dvB = deletionVector(sharedPuffinPath, FILE_B, 44L);
+ table.newRowDelta().addDeletes(dvA).addDeletes(dvB).commit();
+
+ assertThat(liveDeleteFilePaths())
+ .as("Table should have two deletion vectors in the shared Puffin file")
+ .containsExactly(sharedPuffinPath, sharedPuffinPath);
+
+ // replace FILE_A's deletion vector; FILE_B's stays in the shared Puffin
file
+ DeleteFile newDvA =
+ deletionVector("/path/to/new-dv-" + UUID.randomUUID() + ".puffin",
FILE_A, 4L);
+ table
+ .newRowDelta()
+ .validateFromSnapshot(table.currentSnapshot().snapshotId())
+ .removeDeletes(dvA)
+ .addDeletes(newDvA)
+ .commit();
+
+ table.newAppend().appendFile(FILE_C).commit();
+
+ long tAfterCommits =
waitUntilAfter(table.currentSnapshot().timestampMillis());
+
+ Set<String> deletedFiles = Sets.newHashSet();
+
removeSnapshots(table).expireOlderThan(tAfterCommits).deleteWith(deletedFiles::add).commit();
+
+ assertThat(table.snapshots()).hasSize(1);
+ assertThat(deletedFiles)
+ .as("Puffin file with a live deletion vector must not be deleted")
+ .doesNotContain(sharedPuffinPath);
+ assertThat(liveDeleteFilePaths())
Review Comment:
Same as above. This is basically testing scan planning behavior which is
independent of this.
##########
core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java:
##########
@@ -2144,6 +2144,105 @@ public void testCleanupLevelNullValidation() {
.hasMessageContaining("Invalid cleanup level: null");
}
+ @TestTemplate
+ public void testExpireDoesNotDeletePuffinFileSharedByLiveDeletionVectors()
throws Exception {
+ assumeThat(formatVersion).as("Deletion vectors require V3 or
later").isGreaterThanOrEqualTo(3);
+
+ table.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
+
+ // one Puffin file with deletion vectors for FILE_A and FILE_B
+ String sharedPuffinPath = "/path/to/shared-dvs-" + UUID.randomUUID() +
".puffin";
+ DeleteFile dvA = deletionVector(sharedPuffinPath, FILE_A, 4L);
+ DeleteFile dvB = deletionVector(sharedPuffinPath, FILE_B, 44L);
+ table.newRowDelta().addDeletes(dvA).addDeletes(dvB).commit();
+
+ assertThat(liveDeleteFilePaths())
Review Comment:
Minor: I think I'd prefer to not have this assertion, it's basically just
asserting scan planning behavior which is independent from the behavior we're
trying to validate here. It's not wrong though, just trying to avoid
addtitional assertions that make the test hard to parse in terms of what
behavior is being validated at this abstraction (specifically that would just
be what is actually deleted in expiration not what files scan planning kicks
back)
Check out the other test testExpireWithDeleteFiles.
##########
core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java:
##########
@@ -335,4 +355,47 @@ private Set<String> findFilesToDelete(
return filesToDelete;
}
+
+ // A Puffin file can be shared by multiple deletion vectors, so keep it if
any live one remains
+ private Set<String> unreferencedDeletionVectors(
Review Comment:
Minor: I would just stick with the project naming of `DV` or `DVs`, it's an
acroynym that is fairly clear. So I'd call it `unreferencedDVs`
##########
core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java:
##########
@@ -2144,6 +2144,105 @@ public void testCleanupLevelNullValidation() {
.hasMessageContaining("Invalid cleanup level: null");
}
+ @TestTemplate
+ public void testExpireDoesNotDeletePuffinFileSharedByLiveDeletionVectors()
throws Exception {
Review Comment:
Sorry missed this earlier, but seems like for new changes, we're dropping
the test prefix in test method names. The below test already does it, so we
should do it here for consistency in this change.
##########
core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java:
##########
@@ -335,4 +355,47 @@ private Set<String> findFilesToDelete(
return filesToDelete;
}
+
+ // A Puffin file can be shared by multiple deletion vectors, so keep it if
any live one remains
+ private Set<String> unreferencedDeletionVectors(
Review Comment:
Same with params passed in, see my comment on naming above
##########
core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java:
##########
@@ -282,9 +292,11 @@ public void cleanFiles(
private Set<String> findFilesToDelete(
Set<ManifestFile> manifestsToScan,
Set<ManifestFile> manifestsToRevert,
+ Collection<ManifestFile> liveDeleteManifests,
Set<Long> validIds,
Map<Integer, PartitionSpec> specsById) {
Set<String> filesToDelete = ConcurrentHashMap.newKeySet();
+ Set<String> deletionVectorsToDelete = ConcurrentHashMap.newKeySet();
Review Comment:
Should this be renamed to dvFileRemovalCandidates? The current name
indicates as though they will be for sure removed, but that's not really the
case. They are candidates that may be removed if still referenced.
--
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]
