Core: Prevent incremental file cleanup when expiring specified snapsh…

…ots (#10983)
apache · Sep 9, 2024 · 41d00ae · 41d00ae
1 parent 153b070
commit 41d00ae
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 2 deletions.
diff --git a/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java b/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java
@@ -84,6 +84,7 @@ public void accept(String file) {
   private ExecutorService deleteExecutorService = DEFAULT_DELETE_EXECUTOR_SERVICE;
   private ExecutorService planExecutorService = ThreadPools.getWorkerPool();
   private Boolean incrementalCleanup;
+  private boolean specifiedSnapshotId = false;
 
   RemoveSnapshots(TableOperations ops) {
     this.ops = ops;
@@ -116,6 +117,7 @@ public ExpireSnapshots cleanExpiredFiles(boolean clean) {
   public ExpireSnapshots expireSnapshotId(long expireSnapshotId) {
     LOG.info("Expiring snapshot with id: {}", expireSnapshotId);
     idsToRemove.add(expireSnapshotId);
+    specifiedSnapshotId = true;
     return this;
   }
 
@@ -321,6 +323,15 @@ ExpireSnapshots withIncrementalCleanup(boolean useIncrementalCleanup) {
   private void cleanExpiredSnapshots() {
     TableMetadata current = ops.refresh();
 
+    if (specifiedSnapshotId) {
+      if (incrementalCleanup != null && incrementalCleanup) {
+        throw new UnsupportedOperationException(
+            "Cannot clean files incrementally when snapshot IDs are specified");
+      }
+
+      incrementalCleanup = false;
+    }
+
     if (incrementalCleanup == null) {
       incrementalCleanup = current.refs().size() == 1;
     }

diff --git a/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java b/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java
@@ -370,7 +370,7 @@ public void testRetainLastWithExpireById() {
     }
 
     // Retain last 3 snapshots, but explicitly remove the first snapshot
-    removeSnapshots(table).expireSnapshotId(firstSnapshotId).retainLast(3).commit();
+    table.expireSnapshots().expireSnapshotId(firstSnapshotId).retainLast(3).commit();
 
     assertThat(table.snapshots()).hasSize(2);
     assertThat(table.snapshot(firstSnapshotId)).isNull();
@@ -956,7 +956,8 @@ public void testWithExpiringStagedThenCherrypick() {
     List<String> deletedFiles = Lists.newArrayList();
 
     // Expire `B` commit.
-    removeSnapshots(table)
+    table
+        .expireSnapshots()
         .deleteWith(deletedFiles::add)
         .expireSnapshotId(snapshotB.snapshotId())
         .commit();
@@ -1171,6 +1172,26 @@ public void testBranchExpiration() {
     assertThat(table.ops().current().ref(SnapshotRef.MAIN_BRANCH)).isNotNull();
   }
 
+  @TestTemplate
+  public void testIncrementalCleanupFailsWhenExpiringSnapshotId() {
+    table.newAppend().appendFile(FILE_A).commit();
+    table.newDelete().deleteFile(FILE_A).commit();
+    long snapshotId = table.currentSnapshot().snapshotId();
+    table.newAppend().appendFile(FILE_B).commit();
+    waitUntilAfter(table.currentSnapshot().timestampMillis());
+    RemoveSnapshots removeSnapshots = (RemoveSnapshots) table.expireSnapshots();
+
+    assertThatThrownBy(
+            () ->
+                removeSnapshots
+                    .withIncrementalCleanup(true)
+                    .expireSnapshotId(snapshotId)
+                    .cleanExpiredFiles(true)
+                    .commit())
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessage("Cannot clean files incrementally when snapshot IDs are specified");
+  }
+
   @TestTemplate
   public void testMultipleRefsAndCleanExpiredFilesFailsForIncrementalCleanup() {
     table.newAppend().appendFile(FILE_A).commit();