[1.4.x] Core: Expired Snapshot files in a transaction should be deleted

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

@@ -44,6 +44,7 @@
 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;
@@ -505,9 +506,11 @@ 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 null;
+      return ImmutableSet.of();
     }
 
     Set<String> committedFiles = Sets.newHashSet();

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

@@ -212,6 +212,15 @@ List<File> listManifestFiles(File tableDirToList) {
                         && Files.getFileExtension(name).equalsIgnoreCase("avro")));
   }
 
+  List<File> listManifestLists(String tableDirToList) {
+    return Lists.newArrayList(
+        new File(tableDirToList, "metadata")
+            .listFiles(
+                (dir, name) ->
+                    name.startsWith("snap")
+                        && Files.getFileExtension(name).equalsIgnoreCase("avro")));
+  }
+
   public static long countAllMetadataFiles(File tableDir) {
     return Arrays.stream(new File(tableDir, "metadata").listFiles())
         .filter(f -> f.isFile())

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

@@ -440,7 +440,10 @@ public void testRetainNAvailableSnapshotsWithTransaction() {
       t3 = System.currentTimeMillis();
     }
 
-    // Retain last 2 snapshots
+    Assert.assertEquals(
+        "Should be 3 manifest lists", 3, listManifestLists(table.location()).size());
+
+    // Retain last 2 snapshots, which means 1 is deleted.
     Transaction tx = table.newTransaction();
     removeSnapshots(tx.table()).expireOlderThan(t3).retainLast(2).commit();
     tx.commitTransaction();
@@ -449,6 +452,8 @@ public void testRetainNAvailableSnapshotsWithTransaction() {
         "Should have two snapshots.", 2, Lists.newArrayList(table.snapshots()).size());
     Assert.assertEquals(
         "First snapshot should not present.", null, table.snapshot(firstSnapshotId));
+    Assert.assertEquals(
+        "Should be 2 manifest lists", 2, listManifestLists(table.location()).size());
   }
 
   @Test

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

@@ -309,6 +309,8 @@ public void testExpirationInTransaction() {
     V2Assert.assertEquals("Snapshot sequence number should be 1", 1, snap1.sequenceNumber());
     V2Assert.assertEquals(
         "Last sequence number should be 1", 1, readMetadata().lastSequenceNumber());
+    V2Assert.assertEquals(
+        "Should be 1 manifest list", 1, listManifestLists(table.location()).size());
 
     table.newAppend().appendFile(FILE_B).commit();
     Snapshot snap2 = table.currentSnapshot();
@@ -319,12 +321,18 @@ public void testExpirationInTransaction() {
     V2Assert.assertEquals("Snapshot sequence number should be 2", 2, snap2.sequenceNumber());
     V2Assert.assertEquals(
         "Last sequence number should be 2", 2, readMetadata().lastSequenceNumber());
+    V2Assert.assertEquals(
+        "Should be 2 manifest lists", 2, listManifestLists(table.location()).size());
 
     Transaction txn = table.newTransaction();
     txn.expireSnapshots().expireSnapshotId(commitId1).commit();
     txn.commitTransaction();
     V2Assert.assertEquals(
         "Last sequence number should be 2", 2, readMetadata().lastSequenceNumber());
+    V2Assert.assertEquals(
+        "Should be 1 manifest list as 1 was deleted",
+        1,
+        listManifestLists(table.location()).size());
   }
 
   @Test