Core: Report metrics about deleted files in ExpireSnapshots

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

@@ -238,7 +238,7 @@ public UpdatePartitionStatistics updatePartitionStatistics() {
 
   @Override
   public ExpireSnapshots expireSnapshots() {
-    return new RemoveSnapshots(ops);
+    return new RemoveSnapshots(ops).reportWith(reporter);
   }
 
   @Override

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

@@ -227,7 +227,7 @@ public UpdatePartitionStatistics updatePartitionStatistics() {
 
   @Override
   public ExpireSnapshots expireSnapshots() {
-    return appendUpdate(new RemoveSnapshots(transactionOps));
+    return appendUpdate(new RemoveSnapshots(transactionOps).reportWith(reporter));
   }
 
   @Override

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

@@ -18,15 +18,22 @@
  */
 package org.apache.iceberg;
 
+import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.io.BulkDeletionFailureException;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.SupportsBulkOperations;
+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.ImmutableSet;
+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.TypeUtil;
 import org.apache.iceberg.util.Tasks;
@@ -74,7 +81,7 @@
    * @param afterExpiration table metadata after snapshot expiration
    * @param cleanupLevel controls which types of files are eligible for deletion
    */
-  public abstract void cleanFiles(
+  public abstract DeleteSummary cleanFiles(
       TableMetadata beforeExpiration,
       TableMetadata afterExpiration,
       ExpireSnapshots.CleanupLevel cleanupLevel);
@@ -103,20 +110,24 @@
     }
   }
 
-  protected void deleteFiles(Set<String> pathsToDelete, String fileType) {
+  protected void deleteFiles(
+      Set<String> pathsToDelete, DeletedFileType fileType, DeleteSummary summary) {
     if (deleteFunc == null && fileIO instanceof SupportsBulkOperations) {
+      int failures = 0;
       try {
         ((SupportsBulkOperations) fileIO).deleteFiles(pathsToDelete);
       } catch (BulkDeletionFailureException e) {
         LOG.warn(
             "Bulk deletion failed for {} of {} {} file(s)",
             e.numberFailedObjects(),
             pathsToDelete.size(),
-            fileType,
+            fileType.displayName(),
             e);
+        failures = e.numberFailedObjects();
       } catch (RuntimeException e) {
         LOG.warn("Bulk deletion failed", e);
       }
+      summary.deletedFiles(fileType, pathsToDelete.size() - failures);
     } else {
       Consumer<String> deleteFuncToUse = deleteFunc == null ? defaultDeleteFunc : deleteFunc;
 
@@ -127,8 +138,89 @@
           .stopOnFailure()
           .suppressFailureWhenFinished()
           .onFailure(
-              (file, thrown) -> LOG.warn("Delete failed for {} file: {}", fileType, file, thrown))
-          .run(deleteFuncToUse::accept);
+              (file, thrown) ->
+                  LOG.warn("Delete failed for {} file: {}", fileType.displayName(), file, thrown))
+          .run(
+              file -> {
+                deleteFuncToUse.accept(file);
+                summary.deletedFile(fileType);
+              });
+    }
+  }
+
+  enum DeletedFileType {
+    DATA("data"),
+    POSITION_DELETES("position delete"),
+    EQUALITY_DELETES("equality delete"),
+    MANIFEST("manifest"),
+    MANIFEST_LIST("manifest list"),
+    STATISTICS_FILES("statistics files");
+
+    private final String displayName;
+
+    DeletedFileType(String displayName) {
+      this.displayName = displayName;
+    }
+
+    public String displayName() {
+      return displayName;
+    }
+
+    public static DeletedFileType fromContent(FileContent content) {
+      switch (content) {
+        case DATA:
+          return DATA;
+        case POSITION_DELETES:
+          return POSITION_DELETES;
+        case EQUALITY_DELETES:
+          return EQUALITY_DELETES;
+        default:
+          throw new ValidationException("Illegal file content: %s", content);
+      }
+    }
+  }
+
+  static class DeleteSummary {
+    private final Map<DeletedFileType, AtomicLong> counts;
+
+    DeleteSummary() {
+      Map<DeletedFileType, AtomicLong> map = Maps.newEnumMap(DeletedFileType.class);
+      for (DeletedFileType type : DeletedFileType.values()) {
+        map.put(type, new AtomicLong(0L));
+      }
+      this.counts = map;
+    }
+
+    public void deletedFiles(DeletedFileType type, int numFiles) {
+      counts.get(type).addAndGet(numFiles);
+    }
+
+    public void deletedFile(DeletedFileType type) {
+      deletedFiles(type, 1);
+    }
+
+    public long dataFilesCount() {
+      return counts.get(DeletedFileType.DATA).get();
+    }
+
+    public long positionDeleteFilesCount() {
+      return counts.get(DeletedFileType.POSITION_DELETES).get();
+    }
+
+    public long equalityDeleteFilesCount() {
+      return counts.get(DeletedFileType.EQUALITY_DELETES).get();
+    }
+
+    public long manifestsCount() {
+      return counts.get(DeletedFileType.MANIFEST).get();
+    }
+
+    public long manifestListsCount() {
+      return counts.get(DeletedFileType.MANIFEST_LIST).get();
+    }
+
+    public long statisticsFilesCount() {
+      return counts.get(DeletedFileType.STATISTICS_FILES).get();
     }
   }
 
@@ -145,6 +237,46 @@
     return Sets.difference(statsFileLocationsBeforeExpiration, statsFileLocationsAfterExpiration);
   }
 
+  protected static class ExpiredContentFile {
+    private final FileContent content;
+    private final String path;
+
+    public ExpiredContentFile(FileContent content, String path) {
+      this.content = Preconditions.checkNotNull(content, "content is null");
+      this.path = Preconditions.checkNotNull(path, "path is null");
+    }
+
+    public FileContent getContent() {
+      return content;
+    }
+
+    public String getPath() {
+      return path;
+    }
+
+    @Override
+    public boolean equals(Object other) {
+      if (this == other) {
+        return true;
+      } else if (other == null || getClass() != other.getClass()) {
+        return false;
+      }
+
+      ExpiredContentFile that = (ExpiredContentFile) other;
+      return Objects.equals(content, that.content) && Objects.equals(path, that.path);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(content, path);
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this).add("content", content).add("path", path).toString();
+    }
+  }
+
   private Set<String> statsFileLocations(TableMetadata tableMetadata) {
     Set<String> statsFileLocations = Sets.newHashSet();
 

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

@@ -25,6 +25,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.function.Consumer;
+import java.util.stream.Collectors;
 import org.apache.iceberg.exceptions.RuntimeIOException;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.FileIO;
@@ -48,18 +49,19 @@ class IncrementalFileCleanup extends FileCleanupStrategy {
 
   @Override
   @SuppressWarnings({"checkstyle:CyclomaticComplexity", "MethodLength"})
-  public void cleanFiles(
+  public DeleteSummary cleanFiles(
       TableMetadata beforeExpiration,
       TableMetadata afterExpiration,
       ExpireSnapshots.CleanupLevel cleanupLevel) {
+    DeleteSummary summary = new DeleteSummary();
     // clean up required underlying files based on the expired snapshots
     // 1. Get a list of the snapshots that were removed
     // 2. Delete any data files that were deleted by those snapshots and are not in the table
     // 3. Delete any manifests that are no longer used by current snapshots
     // 4. Delete the manifest lists
     if (ExpireSnapshots.CleanupLevel.NONE == cleanupLevel) {
       LOG.info("Nothing to clean.");
-      return;
+      return summary;
     }
 
     Set<Long> validIds = Sets.newHashSet();
@@ -79,12 +81,12 @@ public void cleanFiles(
 
     if (expiredIds.isEmpty()) {
       // if no snapshots were expired, skip cleanup
-      return;
+      return summary;
     }
 
     Snapshot latest = beforeExpiration.currentSnapshot();
     if (latest == null) {
-      return;
+      return summary;
     }
 
     List<Snapshot> snapshots = afterExpiration.snapshots();
@@ -259,32 +261,44 @@ public void cleanFiles(
             });
 
     if (ExpireSnapshots.CleanupLevel.ALL == cleanupLevel) {
-      Set<String> filesToDelete =
+      Set<ExpiredContentFile> filesToDelete =
           findFilesToDelete(
               manifestsToScan, manifestsToRevert, validIds, beforeExpiration.specsById());
-      LOG.debug("Deleting {} data files", filesToDelete.size());
-      deleteFiles(filesToDelete, "data");
+      Map<DeletedFileType, Set<String>> groupedFilesToDelete =
+          filesToDelete.stream()
+              .collect(
+                  Collectors.groupingBy(
+                      file -> DeletedFileType.fromContent(file.getContent()),
+                      Collectors.mapping(ExpiredContentFile::getPath, Collectors.toSet())));
+
+      for (Map.Entry<DeletedFileType, Set<String>> entry : groupedFilesToDelete.entrySet()) {
+        Set<String> filesToDeleteGroup = entry.getValue();
+        DeletedFileType fileType = entry.getKey();
+        LOG.debug("Deleting {} {} files", filesToDeleteGroup.size(), fileType.displayName());
+        deleteFiles(filesToDeleteGroup, fileType, summary);
+      }
     }
 
     LOG.debug("Deleting {} manifest files", manifestsToDelete.size());
-    deleteFiles(manifestsToDelete, "manifest");
+    deleteFiles(manifestsToDelete, DeletedFileType.MANIFEST, summary);
     LOG.debug("Deleting {} manifest-list files", manifestListsToDelete.size());
-    deleteFiles(manifestListsToDelete, "manifest list");
+    deleteFiles(manifestListsToDelete, DeletedFileType.MANIFEST_LIST, summary);
 
     if (hasAnyStatisticsFiles(beforeExpiration)) {
       Set<String> expiredStatisticsFilesLocations =
           expiredStatisticsFilesLocations(beforeExpiration, afterExpiration);
       LOG.debug("Deleting {} statistics files", expiredStatisticsFilesLocations.size());
-      deleteFiles(expiredStatisticsFilesLocations, "statistics files");
+      deleteFiles(expiredStatisticsFilesLocations, DeletedFileType.STATISTICS_FILES, summary);
     }
+    return summary;
   }
 
-  private Set<String> findFilesToDelete(
+  private Set<ExpiredContentFile> findFilesToDelete(
       Set<ManifestFile> manifestsToScan,
       Set<ManifestFile> manifestsToRevert,
       Set<Long> validIds,
       Map<Integer, PartitionSpec> specsById) {
-    Set<String> filesToDelete = ConcurrentHashMap.newKeySet();
+    Set<ExpiredContentFile> filesToDelete = ConcurrentHashMap.newKeySet();
     Tasks.foreach(manifestsToScan)
         .retry(3)
         .suppressFailureWhenFinished()
@@ -295,14 +309,15 @@ private Set<String> findFilesToDelete(
         .run(
             manifest -> {
               // the manifest has deletes, scan it to find files to delete
-              try (ManifestReader<?> reader = ManifestFiles.open(manifest, fileIO, specsById)) {
-                for (ManifestEntry<?> entry : reader.entries()) {
+              try (ManifestReader<? extends ContentFile<?>> reader =
+                  ManifestFiles.open(manifest, fileIO, specsById)) {
+                for (ManifestEntry<? extends ContentFile<?>> entry : reader.entries()) {
                   // if the snapshot ID of the DELETE entry is no longer valid, the data can be
                   // deleted
                   if (entry.status() == ManifestEntry.Status.DELETED
                       && !validIds.contains(entry.snapshotId())) {
-                    // use toString to ensure the path will not change (Utf8 is reused)
-                    filesToDelete.add(entry.file().location());
+                    ContentFile<?> file = entry.file();
+                    filesToDelete.add(new ExpiredContentFile(file.content(), file.location()));
                   }
                 }
               } catch (IOException e) {
@@ -320,12 +335,13 @@ private Set<String> findFilesToDelete(
         .run(
             manifest -> {
               // the manifest has deletes, scan it to find files to delete
-              try (ManifestReader<?> reader = ManifestFiles.open(manifest, fileIO, specsById)) {
-                for (ManifestEntry<?> entry : reader.entries()) {
+              try (ManifestReader<? extends ContentFile<?>> reader =
+                  ManifestFiles.open(manifest, fileIO, specsById)) {
+                for (ManifestEntry<? extends ContentFile<?>> entry : reader.entries()) {
                   // delete any ADDED file from manifests that were reverted
                   if (entry.status() == ManifestEntry.Status.ADDED) {
-                    // use toString to ensure the path will not change (Utf8 is reused)
-                    filesToDelete.add(entry.file().location());
+                    ContentFile<?> file = entry.file();
+                    filesToDelete.add(new ExpiredContentFile(file.content(), file.location()));
                   }
                 }
               } catch (IOException e) {

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

@@ -103,7 +103,7 @@ public static CacheMetricsReport contentCacheStats(FileIO io) {
   }
 
   /**
-   * Returns a {@link CloseableIterable} of file paths in the {@link ManifestFile}.
+   * Returns a {@link CloseableIterable} of {@link DataFile}s in the {@link ManifestFile}.
    *
    * @param manifest a ManifestFile
    * @param io a FileIO
@@ -117,6 +117,25 @@ public static CloseableIterable<String> readPaths(
         entry -> entry.file().location());
   }
 
+  /**
+   * Returns a {@link CloseableIterable} of {@link DataFile}s in the {@link ManifestFile}, reading
+   * only the given columns.
+   *
+   * @param manifest a ManifestFile
+   * @param io a FileIO
+   * @param specsById a Map from spec ID to partition spec
+   * @param columns columns to read
+   * @return a manifest reader
+   */
+  public static CloseableIterable<DataFile> readColumns(
+      ManifestFile manifest,
+      FileIO io,
+      Map<Integer, PartitionSpec> specsById,
+      Collection<String> columns) {
+    return CloseableIterable.transform(
+        read(manifest, io, specsById).select(columns).liveEntries(), ManifestEntry::file);
+  }
+
   /**
    * Returns a new {@link ManifestReader} for a {@link ManifestFile}.
    *