Core: Add Data/Delete file comparators

api/src/main/java/org/apache/iceberg/types/Comparators.java

@@ -22,6 +22,8 @@
 import java.util.Comparator;
 import java.util.List;
 import java.util.function.IntFunction;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.util.UnicodeUtil;
@@ -179,6 +181,10 @@ public static Comparator<CharSequence> filePath() {
     return FilePathComparator.INSTANCE;
   }
 
+  public static Comparator<ContentFile<?>> contentFile() {
+    return ContentFileComparator.INSTANCE;
+  }
+
   private static class NullsFirst<T> implements Comparator<T> {
     private static final NullsFirst<?> INSTANCE = new NullsFirst<>();
 
@@ -394,4 +400,31 @@ public int compare(CharSequence s1, CharSequence s2) {
       return 0;
     }
   }
+
+  private static class DeleteFileComparator implements Comparator<DeleteFile> {
+    private static final DeleteFileComparator INSTANCE = new DeleteFileComparator();
+
+    private DeleteFileComparator() {}
+
+    @Override
+    public int compare(DeleteFile s1, DeleteFile s2) {
+      // this needs to be updated once support for DVs is added
+      return CharSeqComparator.INSTANCE.compare(s1.path(), s2.path());
+    }
+  }
+
+  private static class ContentFileComparator implements Comparator<ContentFile<?>> {
+    private static final ContentFileComparator INSTANCE = new ContentFileComparator();
+
+    private ContentFileComparator() {}
+
+    @Override
+    public int compare(ContentFile<?> s1, ContentFile<?> s2) {
+      if (s1 instanceof DeleteFile && s2 instanceof DeleteFile) {
+        return DeleteFileComparator.INSTANCE.compare((DeleteFile) s1, (DeleteFile) s2);
+      }
+
+      return CharSeqComparator.INSTANCE.compare(s1.path(), s2.path());
+    }
+  }
 }

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

@@ -27,10 +27,11 @@
 import org.apache.iceberg.expressions.StrictMetricsEvaluator;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
 
 public class BaseOverwriteFiles extends MergingSnapshotProducer<OverwriteFiles>
     implements OverwriteFiles {
-  private final Set<DataFile> deletedDataFiles = Sets.newHashSet();
+  private final Set<DataFile> deletedDataFiles = Sets.newTreeSet(Comparators.contentFile());
   private boolean validateAddedFilesMatchOverwriteFilter = false;
   private Long startingSnapshotId = null;
   private Expression conflictDetectionFilter = null;

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

@@ -22,9 +22,10 @@
 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.Sets;
+import org.apache.iceberg.types.Comparators;
 
 class BaseRewriteFiles extends MergingSnapshotProducer<RewriteFiles> implements RewriteFiles {
-  private final Set<DataFile> replacedDataFiles = Sets.newHashSet();
+  private final Set<DataFile> replacedDataFiles = Sets.newTreeSet(Comparators.contentFile());
   private Long startingSnapshotId = null;
 
   BaseRewriteFiles(String tableName, TableOperations ops) {

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

@@ -30,7 +30,8 @@
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.apache.iceberg.util.CharSequenceSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
 
 /**
  * {@link AppendFiles Append} implementation that adds a new manifest file for the write.
@@ -44,7 +45,7 @@ class FastAppend extends SnapshotProducer<AppendFiles> implements AppendFiles {
   private final PartitionSpec spec;
   private final SnapshotSummary.Builder summaryBuilder = SnapshotSummary.builder();
   private final List<DataFile> newFiles = Lists.newArrayList();
-  private final CharSequenceSet newFilePaths = CharSequenceSet.empty();
+  private final Set<DataFile> newFileSet = Sets.newTreeSet(Comparators.contentFile());
   private final List<ManifestFile> appendManifests = Lists.newArrayList();
   private final List<ManifestFile> rewrittenAppendManifests = Lists.newArrayList();
   private List<ManifestFile> newManifests = null;
@@ -86,7 +87,7 @@ protected Map<String, String> summary() {
   @Override
   public FastAppend appendFile(DataFile file) {
     Preconditions.checkNotNull(file, "Invalid data file: null");
-    if (newFilePaths.add(file.path())) {
+    if (newFileSet.add(file)) {
       this.hasNewFiles = true;
       newFiles.add(file);
       summaryBuilder.addedFile(spec, file);

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

@@ -25,6 +25,7 @@
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 import org.apache.iceberg.exceptions.RuntimeIOException;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.expressions.Expression;
@@ -36,10 +37,10 @@
 import org.apache.iceberg.relocated.com.google.common.base.Joiner;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
-import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
 import org.apache.iceberg.util.CharSequenceSet;
 import org.apache.iceberg.util.CharSequenceWrapper;
 import org.apache.iceberg.util.ManifestFileUtil;
@@ -71,6 +72,7 @@ public String partition() {
   private final PartitionSet deleteFilePartitions;
   private final PartitionSet dropPartitions;
   private final CharSequenceSet deletePaths = CharSequenceSet.empty();
+  private final Set<F> filesToDelete = Sets.newTreeSet(Comparators.contentFile());
   private Expression deleteExpression = Expressions.alwaysFalse();
   private long minSequenceNumber = 0;
   private boolean hasPathOnlyDeletes = false;
@@ -153,7 +155,7 @@ void caseSensitive(boolean newCaseSensitive) {
   void delete(F file) {
     Preconditions.checkNotNull(file, "Cannot delete file: null");
     invalidateFilteredCache();
-    deletePaths.add(file.path());
+    filesToDelete.add(file);
     deleteFilePartitions.add(file.specId(), file.partition());
   }
 
@@ -167,6 +169,7 @@ void delete(CharSequence path) {
 
   boolean containsDeletes() {
     return !deletePaths.isEmpty()
+        || !filesToDelete.isEmpty()
         || deleteExpression != Expressions.alwaysFalse()
         || !dropPartitions.isEmpty();
   }
@@ -233,23 +236,28 @@ SnapshotSummary.Builder buildSummary(Iterable<ManifestFile> manifests) {
   @SuppressWarnings("CollectionUndefinedEquality")
   private void validateRequiredDeletes(ManifestFile... manifests) {
     if (failMissingDeletePaths) {
-      CharSequenceSet deletedFiles = deletedFiles(manifests);
+      Set<F> deletedFiles = deletedFiles(manifests);
+
       ValidationException.check(
-          deletedFiles.containsAll(deletePaths),
+          deletedFiles.containsAll(filesToDelete),
           "Missing required files to delete: %s",
-          COMMA.join(Iterables.filter(deletePaths, path -> !deletedFiles.contains(path))));
+          COMMA.join(
+              filesToDelete.stream()
+                  .filter(f -> !deletedFiles.contains(f))
+                  .map(ContentFile::path)
+                  .collect(Collectors.toList())));
     }
   }
 
-  private CharSequenceSet deletedFiles(ManifestFile[] manifests) {
-    CharSequenceSet deletedFiles = CharSequenceSet.empty();
+  private Set<F> deletedFiles(ManifestFile[] manifests) {
+    Set<F> deletedFiles = Sets.newTreeSet(Comparators.contentFile());
 
     if (manifests != null) {
       for (ManifestFile manifest : manifests) {
         Iterable<F> manifestDeletes = filteredManifestToDeletedFiles.get(manifest);
         if (manifestDeletes != null) {
           for (F file : manifestDeletes) {
-            deletedFiles.add(file.path());
+            deletedFiles.add(file);
           }
         }
       }
@@ -347,7 +355,7 @@ private boolean canContainDeletedFiles(ManifestFile manifest) {
     boolean canContainDroppedFiles;
     if (hasPathOnlyDeletes) {
       canContainDroppedFiles = true;
-    } else if (!deletePaths.isEmpty()) {
+    } else if (!filesToDelete.isEmpty()) {
       // because there were no path-only deletes, the set of deleted file partitions is valid
       canContainDroppedFiles =
           ManifestFileUtil.canContainAny(manifest, deleteFilePartitions, specsById);
@@ -372,8 +380,14 @@ private boolean manifestHasDeletedFiles(
 
     for (ManifestEntry<F> entry : reader.liveEntries()) {
       F file = entry.file();
+
+      // add path-based delete to set of files to be deleted
+      if (deletePaths.contains(CharSequenceWrapper.wrap(file.path()))) {
+        filesToDelete.add(file);
+      }
+
       boolean markedForDelete =
-          deletePaths.contains(file.path())
+          filesToDelete.contains(file)
               || dropPartitions.contains(file.specId(), file.partition())
               || (isDelete
                   && entry.isLive()
@@ -409,8 +423,7 @@ private ManifestFile filterManifestWithDeletedFiles(
     boolean isDelete = reader.isDeleteManifestReader();
     // when this point is reached, there is at least one file that will be deleted in the
     // manifest. produce a copy of the manifest with all deleted files removed.
-    List<F> deletedFiles = Lists.newArrayList();
-    Set<CharSequenceWrapper> deletedPaths = Sets.newHashSet();
+    Set<F> deleted = Sets.newTreeSet(Comparators.contentFile());
 
     try {
       ManifestWriter<F> writer = newManifestWriter(reader.spec());
@@ -421,7 +434,7 @@ private ManifestFile filterManifestWithDeletedFiles(
                 entry -> {
                   F file = entry.file();
                   boolean markedForDelete =
-                      deletePaths.contains(file.path())
+                      filesToDelete.contains(file)
                           || dropPartitions.contains(file.specId(), file.partition())
                           || (isDelete
                               && entry.isLive()
@@ -441,18 +454,16 @@ private ManifestFile filterManifestWithDeletedFiles(
                       if (allRowsMatch) {
                         writer.delete(entry);
 
-                        CharSequenceWrapper wrapper = CharSequenceWrapper.wrap(entry.file().path());
-                        if (deletedPaths.contains(wrapper)) {
+                        if (deleted.contains(entry.file())) {
                           LOG.warn(
                               "Deleting a duplicate path from manifest {}: {}",
                               manifest.path(),
-                              wrapper.get());
+                              entry.file().path());
                           duplicateDeleteCount += 1;
                         } else {
                           // only add the file to deletes if it is a new delete
                           // this keeps the snapshot summary accurate for non-duplicate data
-                          deletedFiles.add(entry.file().copyWithoutStats());
-                          deletedPaths.add(wrapper);
+                          deleted.add(entry.file().copyWithoutStats());
                         }
                       } else {
                         writer.existing(entry);
@@ -472,7 +483,7 @@ private ManifestFile filterManifestWithDeletedFiles(
 
       // update caches
       filteredManifests.put(manifest, filtered);
-      filteredManifestToDeletedFiles.put(filtered, deletedFiles);
+      filteredManifestToDeletedFiles.put(filtered, deleted);
 
       return filtered;
 

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

@@ -49,6 +49,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
 import org.apache.iceberg.util.CharSequenceSet;
 import org.apache.iceberg.util.Pair;
 import org.apache.iceberg.util.PartitionSet;
@@ -81,8 +82,8 @@ abstract class MergingSnapshotProducer<ThisT> extends SnapshotProducer<ThisT> {
 
   // update data
   private final Map<PartitionSpec, List<DataFile>> newDataFilesBySpec = Maps.newHashMap();
-  private final CharSequenceSet newDataFilePaths = CharSequenceSet.empty();
-  private final CharSequenceSet newDeleteFilePaths = CharSequenceSet.empty();
+  private final Set<DataFile> newDataFiles = Sets.newTreeSet(Comparators.contentFile());
+  private final Set<DeleteFile> newDeleteFiles = Sets.newTreeSet(Comparators.contentFile());
   private Long newDataFilesDataSequenceNumber;
   private final Map<Integer, List<DeleteFileHolder>> newDeleteFilesBySpec = Maps.newHashMap();
   private final List<ManifestFile> appendManifests = Lists.newArrayList();
@@ -234,7 +235,7 @@ protected boolean addsDeleteFiles() {
   /** Add a data file to the new snapshot. */
   protected void add(DataFile file) {
     Preconditions.checkNotNull(file, "Invalid data file: null");
-    if (newDataFilePaths.add(file.path())) {
+    if (newDataFiles.add(file)) {
       PartitionSpec fileSpec = ops.current().spec(file.specId());
       Preconditions.checkArgument(
           fileSpec != null,
@@ -244,9 +245,9 @@ protected void add(DataFile file) {
 
       addedFilesSummary.addedFile(fileSpec, file);
       hasNewDataFiles = true;
-      List<DataFile> newDataFiles =
+      List<DataFile> dataFiles =
           newDataFilesBySpec.computeIfAbsent(fileSpec, ignored -> Lists.newArrayList());
-      newDataFiles.add(file);
+      dataFiles.add(file);
     }
   }
 
@@ -268,7 +269,7 @@ private void add(DeleteFileHolder fileHolder) {
     List<DeleteFileHolder> deleteFiles =
         newDeleteFilesBySpec.computeIfAbsent(specId, s -> Lists.newArrayList());
 
-    if (newDeleteFilePaths.add(fileHolder.deleteFile().path())) {
+    if (newDeleteFiles.add(fileHolder.deleteFile())) {
       deleteFiles.add(fileHolder);
       addedFilesSummary.addedFile(fileSpec, fileHolder.deleteFile());
       hasNewDeleteFiles = true;
@@ -970,9 +971,9 @@ private List<ManifestFile> newDataFilesAsManifests() {
 
     if (cachedNewDataManifests.isEmpty()) {
       newDataFilesBySpec.forEach(
-          (dataSpec, newDataFiles) -> {
+          (dataSpec, dataFiles) -> {
             List<ManifestFile> newDataManifests =
-                writeDataManifests(newDataFiles, newDataFilesDataSequenceNumber, dataSpec);
+                writeDataManifests(dataFiles, newDataFilesDataSequenceNumber, dataSpec);
             cachedNewDataManifests.addAll(newDataManifests);
           });
       this.hasNewDataFiles = false;

core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java

@@ -28,6 +28,7 @@
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
 import org.apache.iceberg.util.Tasks;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -73,8 +74,8 @@ public RewriteDataFilesCommitManager(
    * @param fileGroups fileSets to commit
    */
   public void commitFileGroups(Set<RewriteFileGroup> fileGroups) {
-    Set<DataFile> rewrittenDataFiles = Sets.newHashSet();
-    Set<DataFile> addedDataFiles = Sets.newHashSet();
+    Set<DataFile> rewrittenDataFiles = Sets.newTreeSet(Comparators.contentFile());
+    Set<DataFile> addedDataFiles = Sets.newTreeSet(Comparators.contentFile());
     for (RewriteFileGroup group : fileGroups) {
       rewrittenDataFiles.addAll(group.rewrittenFiles());
       addedDataFiles.addAll(group.addedFiles());

core/src/main/java/org/apache/iceberg/actions/RewriteFileGroup.java

@@ -18,7 +18,6 @@
  */
 package org.apache.iceberg.actions;
 
-import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Set;
@@ -29,6 +28,8 @@
 import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
 
 /**
  * Container class representing a set of files to be rewritten by a RewriteAction and the new files
@@ -38,7 +39,7 @@ public class RewriteFileGroup {
   private final FileGroupInfo info;
   private final List<FileScanTask> fileScanTasks;
 
-  private Set<DataFile> addedFiles = Collections.emptySet();
+  private Set<DataFile> addedFiles = Sets.newTreeSet(Comparators.contentFile());
 
   public RewriteFileGroup(FileGroupInfo info, List<FileScanTask> fileScanTasks) {
     this.info = info;
@@ -58,7 +59,9 @@ public void setOutputFiles(Set<DataFile> files) {
   }
 
   public Set<DataFile> rewrittenFiles() {
-    return fileScans().stream().map(FileScanTask::file).collect(Collectors.toSet());
+    return fileScans().stream()
+        .map(FileScanTask::file)
+        .collect(Collectors.toCollection(() -> Sets.newTreeSet(Comparators.contentFile())));
   }
 
   public Set<DataFile> addedFiles() {