Deduplicate delete files by path in $files table

Deduplicate files by path in FilesTablePageSource since the same
delete file can appear multiple times. Add a HashSet to track seen
file paths and skip duplicates.

Add testFilesTableDeleteFileDeduplication to BaseIcebergSystemTables
that verifies the $files table shows each delete file exactly once.

Follow-up to #28911 as requested by findinpath.

Author: kaveti

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/system/files/FilesTablePageSource.java

@@ -49,11 +49,13 @@
 import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
 import java.util.Comparator;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.OptionalLong;
+import java.util.Set;
 import java.util.function.BiConsumer;
 import java.util.function.Supplier;
 
@@ -110,6 +112,7 @@ public final class FilesTablePageSource
     private final Iterator<? extends ContentFile<?>> contentIterator;
     private final Map<String, Integer> columnNameToIndex;
     private final PageBuilder pageBuilder;
+    private final Set<String> seenFilePaths;
     private final long completedBytes;
     private long completedPositions;
     private long readTimeNanos;
@@ -146,6 +149,7 @@ public FilesTablePageSource(
         this.columnNameToIndex = mapWithIndex(requiredColumns.stream(),
                 (columnName, position) -> immutableEntry(columnName, Long.valueOf(position).intValue()))
                 .collect(toImmutableMap(Map.Entry::getKey, Map.Entry::getValue));
+        this.seenFilePaths = new HashSet<>();
         this.completedBytes = split.manifestFile().length();
         this.completedPositions = 0L;
         this.readTimeNanos = 0L;
@@ -184,9 +188,13 @@ public SourcePage getNextSourcePage()
         }
 
         while (contentIterator.hasNext() && !pageBuilder.isFull()) {
+            ContentFile<?> contentFile = contentIterator.next();
+            // Deduplicate files by path since the same file can appear multiple times
+            if (!seenFilePaths.add(contentFile.location())) {
+                continue;
+            }
             pageBuilder.declarePosition();
             long start = System.nanoTime();
-            ContentFile<?> contentFile = contentIterator.next();
 
             // content
             writeValueOrNull(pageBuilder, CONTENT_COLUMN_NAME, () -> contentFile.content().id(), INTEGER::writeInt);

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergSystemTables.java

@@ -431,6 +431,48 @@ public void testFilesTable()
         }
     }
 
+    @Test
+    public void testFilesTableDeleteFileDeduplication()
+            throws Exception
+    {
+        try (TestTable testTable = newTrinoTable("test_files_delete_dedup_",
+                "WITH (partitioning = ARRAY['regionkey']) AS SELECT * FROM tpch.tiny.nation")) {
+            String tableName = testTable.getName();
+            Table icebergTable = loadTable(tableName);
+
+            // Verify initial state: only data files, no delete files
+            assertQuery(
+                    "SELECT count(*) FROM \"" + tableName + "$files\" WHERE content = 0",
+                    "VALUES 5"); // one data file per regionkey partition
+            assertQuery(
+                    "SELECT count(*) FROM \"" + tableName + "$files\" WHERE content != 0",
+                    "VALUES 0");
+
+            // Write a position delete via MOR path
+            assertUpdate("DELETE FROM " + tableName + " WHERE nationkey = 7", 1);
+
+            // Write an equality delete file for regionkey=2
+            writeEqualityDeleteForTable(icebergTable, fileSystemFactory,
+                    Optional.of(icebergTable.spec()),
+                    Optional.of(new PartitionData(new Long[] {2L})),
+                    ImmutableMap.of("regionkey", 2L),
+                    Optional.empty());
+
+            // Verify: each file path should appear exactly once (no duplicates)
+            assertQuery(
+                    "SELECT count(*) FROM \"" + tableName + "$files\" WHERE content = 1",
+                    "VALUES 1"); // exactly 1 position delete file
+            assertQuery(
+                    "SELECT count(*) FROM \"" + tableName + "$files\" WHERE content = 2",
+                    "VALUES 1"); // exactly 1 equality delete file
+
+            // Verify no duplicate file paths exist
+            assertQuery(
+                    "SELECT count(file_path) - count(DISTINCT file_path) FROM \"" + tableName + "$files\"",
+                    "VALUES 0");
+        }
+    }
+
     @Test
     public void testFilesPartitionTable()
     {