This is an automated email from the ASF dual-hosted git repository.
amogh-jahagirdar pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new c131e75958 Core: Fix file cleanup logic to retain Puffin files shared
by live deletion vectors (#18408)
c131e75958 is described below
commit c131e75958ff7bef67e22d241b02e5f8338b3955
Author: Yuya Ebihara <[email protected]>
AuthorDate: Wed Oct 14 11:59:42 2026 +0900
Core: Fix file cleanup logic to retain Puffin files shared by live deletion
vectors (#18408)
---
.../org/apache/iceberg/IncrementalFileCleanup.java | 65 ++++++++++++++++-
.../org/apache/iceberg/TestRemoveSnapshots.java | 83 ++++++++++++++++++++++
2 files changed, 146 insertions(+), 2 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java
b/core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java
index 911a94c13f..73a1e4584f 100644
--- a/core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java
+++ b/core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java
@@ -111,6 +111,7 @@ class IncrementalFileCleanup extends FileCleanupStrategy {
// find manifests to clean up that are still referenced by a valid
snapshot, but written by an
// expired snapshot
Set<String> validManifests = ConcurrentHashMap.newKeySet();
+ Set<ManifestFile> liveDeleteManifests = ConcurrentHashMap.newKeySet();
Set<ManifestFile> manifestsToScan = ConcurrentHashMap.newKeySet();
// Reads and deletes are done using
Tasks.foreach(...).suppressFailureWhenFinished to complete
@@ -131,6 +132,9 @@ class IncrementalFileCleanup extends FileCleanupStrategy {
try (CloseableIterable<ManifestFile> manifests =
readManifests(snapshot)) {
for (ManifestFile manifest : manifests) {
validManifests.add(manifest.path());
+ if (manifest.content() == ManifestContent.DELETES) {
+ liveDeleteManifests.add(manifest.copy());
+ }
long snapshotId = manifest.snapshotId();
// whether the manifest was created by a valid snapshot
(true) or an expired
@@ -261,7 +265,11 @@ class IncrementalFileCleanup extends FileCleanupStrategy {
if (ExpireSnapshots.CleanupLevel.ALL == cleanupLevel) {
Set<String> filesToDelete =
findFilesToDelete(
- manifestsToScan, manifestsToRevert, validIds,
beforeExpiration.specsById());
+ manifestsToScan,
+ manifestsToRevert,
+ liveDeleteManifests,
+ validIds,
+ beforeExpiration.specsById());
LOG.debug("Deleting {} data files", filesToDelete.size());
deleteFiles(filesToDelete, "data");
}
@@ -282,9 +290,11 @@ class IncrementalFileCleanup extends FileCleanupStrategy {
private Set<String> findFilesToDelete(
Set<ManifestFile> manifestsToScan,
Set<ManifestFile> manifestsToRevert,
+ Set<ManifestFile> liveDeleteManifests,
Set<Long> validIds,
Map<Integer, PartitionSpec> specsById) {
Set<String> filesToDelete = ConcurrentHashMap.newKeySet();
+ Set<String> dvFileRemovalCandidates = ConcurrentHashMap.newKeySet();
Tasks.foreach(manifestsToScan)
.retry(3)
.suppressFailureWhenFinished()
@@ -302,7 +312,12 @@ class IncrementalFileCleanup extends FileCleanupStrategy {
if (entry.status() == ManifestEntry.Status.DELETED
&& !validIds.contains(entry.snapshotId())) {
// use toString to ensure the path will not change (Utf8
is reused)
- filesToDelete.add(entry.file().location());
+ String location = entry.file().location();
+ if (entry.file().format() == FileFormat.PUFFIN) {
+ dvFileRemovalCandidates.add(location);
+ } else {
+ filesToDelete.add(location);
+ }
}
}
} catch (IOException e) {
@@ -310,6 +325,8 @@ class IncrementalFileCleanup extends FileCleanupStrategy {
}
});
+ filesToDelete.addAll(unreferencedDVs(dvFileRemovalCandidates,
liveDeleteManifests, specsById));
+
Tasks.foreach(manifestsToRevert)
.retry(3)
.suppressFailureWhenFinished()
@@ -335,4 +352,48 @@ class IncrementalFileCleanup extends FileCleanupStrategy {
return filesToDelete;
}
+
+ // A Puffin file can be shared by multiple deletion vectors, so keep it if
any live one remains
+ private Set<String> unreferencedDVs(
+ Set<String> dvFileRemovalCandidates,
+ Set<ManifestFile> liveDeleteManifests,
+ Map<Integer, PartitionSpec> specsById) {
+ if (dvFileRemovalCandidates.isEmpty()) {
+ return dvFileRemovalCandidates;
+ }
+
+ Set<String> unreferenced = ConcurrentHashMap.newKeySet();
+ unreferenced.addAll(dvFileRemovalCandidates);
+
+ try {
+ Tasks.foreach(liveDeleteManifests)
+ .retry(3)
+ .stopOnFailure()
+ .throwFailureWhenFinished()
+ .executeWith(planExecutorService)
+ .onFailure(
+ (item, exc) ->
+ LOG.warn("Failed to determine live DVs in manifest {}",
item.path(), exc))
+ .run(
+ manifest -> {
+ if (unreferenced.isEmpty()) {
+ return;
+ }
+
+ 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 determine live DVs, skipping DV cleanup", e);
+ return Set.of();
+ }
+
+ return unreferenced;
+ }
}
diff --git a/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java
b/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java
index 7a983bf585..e3dd980dfe 100644
--- a/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java
+++ b/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java
@@ -2144,6 +2144,89 @@ public class TestRemoveSnapshots extends TestBase {
.hasMessageContaining("Invalid cleanup level: null");
}
+ @TestTemplate
+ void expireDoesNotDeletePuffinFileSharedByLiveDeletionVectors() {
+ 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();
+
+ // 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);
+ }
+
+ @TestTemplate
+ void expireDeletesPuffinFileWhenAllSharedDeletionVectorsReplaced() {
+ assumeThat(formatVersion).as("Deletion vectors require V3 or
later").isGreaterThanOrEqualTo(3);
+
+ table.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
+
+ 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();
+
+ DeleteFile newDvA =
+ deletionVector("/path/to/new-dv-a-" + UUID.randomUUID() + ".puffin",
FILE_A, 4L);
+ DeleteFile newDvB =
+ deletionVector("/path/to/new-dv-b-" + UUID.randomUUID() + ".puffin",
FILE_B, 4L);
+ table
+ .newRowDelta()
+ .validateFromSnapshot(table.currentSnapshot().snapshotId())
+ .removeDeletes(dvA)
+ .removeDeletes(dvB)
+ .addDeletes(newDvA)
+ .addDeletes(newDvB)
+ .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).contains(sharedPuffinPath);
+ }
+
+ private DeleteFile deletionVector(String puffinPath, DataFile dataFile, long
contentOffset) {
+ return FileMetadata.deleteFileBuilder(table.spec())
+ .ofPositionDeletes()
+ .withPath(puffinPath)
+ .withFileSizeInBytes(100L)
+ .withPartition(dataFile.partition())
+ .withRecordCount(1L)
+ .withReferencedDataFile(dataFile.location())
+ .withContentOffset(contentOffset)
+ .withContentSizeInBytes(40L)
+ .build();
+ }
+
@TestTemplate
void readManifestsProjectsManifestContent() throws IOException {
assumeThat(formatVersion)
