Core: Use transaction metadata for post-commit file cleanup

core/src/main/java/org/apache/iceberg/BaseTransaction.java

@@ -45,7 +45,6 @@
 import org.apache.iceberg.metrics.MetricsReporter;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.util.PropertyUtil;
@@ -393,24 +392,24 @@ private void commitSimpleTransaction() {
       // committed manifests to ensure that no committed manifest is deleted.
       // A manifest could be deleted in one successful operation commit, but reused in another
       // successful commit of that operation if the whole transaction is retried.
-      Set<Long> newSnapshots = Sets.newHashSet();
+      //
+      // take the new snapshots from the metadata this transaction committed. re-reading
+      // through ops.current() can return stale metadata when the catalog caches table pointers,
+      // which would make the new snapshots appear missing and skip clean-up entirely.
+      Set<Snapshot> newSnapshots = Sets.newHashSet();
       for (Snapshot snapshot : current.snapshots()) {
         if (!startingSnapshots.contains(snapshot.snapshotId())) {
-          newSnapshots.add(snapshot.snapshotId());
+          newSnapshots.add(snapshot);
         }
       }
 
-      Set<String> committedFiles = committedFiles(ops, newSnapshots);
-      if (committedFiles != null) {
-        // delete all the files that were deleted in the most recent set of operation commits
-        Set<String> uncommittedFiles =
-            deletedFiles.stream()
-                .filter(f -> !committedFiles.contains(f))
-                .collect(Collectors.toSet());
-        deleteUncommittedFiles(uncommittedFiles);
-      } else {
-        LOG.warn("Failed to load metadata for a committed snapshot, skipping clean-up");
-      }
+      Set<String> committedFiles = committedFiles(ops.io(), newSnapshots);
+      // delete all the files that were deleted in the most recent set of operation commits
+      Set<String> uncommittedFiles =
+          deletedFiles.stream()
+              .filter(f -> !committedFiles.contains(f))
+              .collect(Collectors.toSet());
+      deleteUncommittedFiles(uncommittedFiles);
 
     } catch (RuntimeException e) {
       LOG.warn("Failed to load committed metadata, skipping clean-up", e);
@@ -458,23 +457,13 @@ private void applyUpdates(TableOperations underlyingOps) {
     }
   }
 
-  // committedFiles returns null whenever the set of committed files
-  // cannot be determined from the provided snapshots
-  private static Set<String> committedFiles(TableOperations ops, Set<Long> snapshotIds) {
-    if (snapshotIds.isEmpty()) {
-      return ImmutableSet.of();
-    }
-
+  // returns the manifest lists and manifests referenced by the given committed snapshots
+  private static Set<String> committedFiles(FileIO io, Set<Snapshot> snapshots) {
     Set<String> committedFiles = Sets.newHashSet();
 
-    for (long snapshotId : snapshotIds) {
-      Snapshot snap = ops.current().snapshot(snapshotId);
-      if (snap != null) {
-        committedFiles.add(snap.manifestListLocation());
-        snap.allManifests(ops.io()).forEach(manifest -> committedFiles.add(manifest.path()));
-      } else {
-        return null;
-      }
+    for (Snapshot snap : snapshots) {
+      committedFiles.add(snap.manifestListLocation());
+      snap.allManifests(io).forEach(manifest -> committedFiles.add(manifest.path()));
     }
 
     return committedFiles;

core/src/test/java/org/apache/iceberg/TestTransaction.java

@@ -28,6 +28,9 @@
 import java.util.List;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import org.apache.iceberg.ManifestEntry.Status;
 import org.apache.iceberg.exceptions.CommitFailedException;
@@ -39,6 +42,7 @@
 import org.apache.iceberg.types.Types;
 import org.junit.jupiter.api.TestTemplate;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
 
 @ExtendWith(ParameterizedTestExtension.class)
@@ -323,6 +327,68 @@ public void testTransactionRetry() {
         .isEqualTo(appendManifests);
   }
 
+  @TestTemplate
+  public void testTransactionRetryCleansUpWhenCatalogReturnsStaleMetadata() throws IOException {
+    File location = java.nio.file.Files.createTempDirectory(temp, "junit").toFile();
+    String tableName = "txnStaleMetadataCleanupTest";
+    // fail the first two commits, each after a concurrent metadata update, so the transaction
+    // re-applies twice and writes a new manifest list each time. after the successful commit,
+    // serve stale metadata from current() to simulate a catalog that caches table pointers.
+    AtomicInteger injectedFailures = new AtomicInteger(0);
+    AtomicBoolean captureStaleMetadata = new AtomicBoolean(false);
+    AtomicReference<TableMetadata> staleMetadata = new AtomicReference<>(null);
+    TestTables.LocalFileIO spyFileIO = Mockito.spy(new TestTables.LocalFileIO());
+    TestTables.TestTableOperations ops =
+        new TestTables.TestTableOperations(tableName, location, spyFileIO) {
+          @Override
+          public void commit(TableMetadata base, TableMetadata updatedMetadata) {
+            if (injectedFailures.getAndDecrement() > 0) {
+              TestTables.load(location, tableName)
+                  .updateProperties()
+                  .set("conflict-" + injectedFailures.get(), "true")
+                  .commit();
+              throw new CommitFailedException("Injected failure");
+            }
+            super.commit(base, updatedMetadata);
+            if (captureStaleMetadata.get()) {
+              staleMetadata.compareAndSet(null, base);
+            }
+          }
+
+          @Override
+          public TableMetadata current() {
+            if (staleMetadata.get() != null) {
+              return staleMetadata.get();
+            }
+            return super.current();
+          }
+        };
+    TestTables.TestTable txnTable =
+        TestTables.create(
+            location, tableName, SCHEMA, SPEC, SortOrder.unsorted(), formatVersion, ops);
+
+    txnTable.updateProperties().set(TableProperties.COMMIT_NUM_RETRIES, "3").commit();
+
+    Transaction txn = txnTable.newTransaction();
+    txn.newFastAppend().appendFile(FILE_A).commit();
+
+    injectedFailures.set(2);
+    captureStaleMetadata.set(true);
+    txn.commitTransaction();
+
+    // clean-up must run off the transaction's committed metadata, not the stale catalog state
+    ArgumentCaptor<String> deletedPaths = ArgumentCaptor.forClass(String.class);
+    Mockito.verify(spyFileIO, Mockito.atLeastOnce()).deleteFile(deletedPaths.capture());
+    List<String> deletedManifestLists =
+        deletedPaths.getAllValues().stream()
+            .filter(path -> path.contains("snap-"))
+            .collect(Collectors.toList());
+    assertThat(deletedManifestLists).hasSize(2).doesNotHaveDuplicates();
+    assertThat(deletedManifestLists)
+        .doesNotContain(
+            TestTables.load(location, tableName).currentSnapshot().manifestListLocation());
+  }
+
   @TestTemplate
   public void testTransactionRetryMergeAppend() {
     // use only one retry