Fix memory leak in Iceberg remove orphan files procedure

Author: grantatspothero

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java

@@ -206,10 +206,12 @@
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.OptionalLong;
+import java.util.Queue;
 import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
@@ -2251,7 +2253,9 @@ private void removeOrphanFiles(Table table, ConnectorSession session, SchemaTabl
         ImmutableSet.Builder<String> validMetadataFileNames = ImmutableSet.builder();
         ImmutableSet.Builder<String> validDataFileNames = ImmutableSet.builder();
 
-        for (Snapshot snapshot : table.snapshots()) {
+        Queue<Snapshot> allSnapshots = getAllSnapshots(table);
+        while (!allSnapshots.isEmpty()) {
+            Snapshot snapshot = allSnapshots.poll();
             if (snapshot.manifestListLocation() != null) {
                 validMetadataFileNames.add(fileName(snapshot.manifestListLocation()));
             }
@@ -4179,4 +4183,18 @@ private static TableStatistics mergeColumnStatistics(TableStatistics currentStat
         newStats.getColumnStatistics().forEach(statisticsBuilder::setColumnStatistics);
         return statisticsBuilder.build();
     }
+
+    /**
+     * Table.snapshots() returns Iterable but the actual implementation returned is a List
+     * snapshot.allManifests() caches manifests, so by keeping a List around GC is prevented from collecting any in memory manifests which leaks memory
+     * Workaround: use Queue, Queue#poll will unlink old Snapshots allowing for GC during traversal
+     */
+    private static Queue<Snapshot> getAllSnapshots(Table icebergTable)
+    {
+        Queue<Snapshot> allSnapshots = new LinkedList<>();
+        for (Snapshot snapshot : icebergTable.snapshots()) {
+            allSnapshots.add(snapshot);
+        }
+        return allSnapshots;
+    }
 }