diff --git a/core/src/main/java/org/apache/iceberg/BaseTable.java b/core/src/main/java/org/apache/iceberg/BaseTable.java index 4d73c96fec28..5fcd18fee03c 100644 --- a/core/src/main/java/org/apache/iceberg/BaseTable.java +++ b/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 diff --git a/core/src/main/java/org/apache/iceberg/BaseTransaction.java b/core/src/main/java/org/apache/iceberg/BaseTransaction.java index 9884ac297079..040daa3ae7e0 100644 --- a/core/src/main/java/org/apache/iceberg/BaseTransaction.java +++ b/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 diff --git a/core/src/main/java/org/apache/iceberg/FileCleanupStrategy.java b/core/src/main/java/org/apache/iceberg/FileCleanupStrategy.java index 573aef057ff6..5c46ef0c8496 100644 --- a/core/src/main/java/org/apache/iceberg/FileCleanupStrategy.java +++ b/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 @@ protected FileCleanupStrategy( * @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,8 +110,10 @@ protected CloseableIterable readManifests(Snapshot snapshot) { } } - protected void deleteFiles(Set pathsToDelete, String fileType) { + protected void deleteFiles( + Set pathsToDelete, DeletedFileType fileType, DeleteSummary summary) { if (deleteFunc == null && fileIO instanceof SupportsBulkOperations) { + int failures = 0; try { ((SupportsBulkOperations) fileIO).deleteFiles(pathsToDelete); } catch (BulkDeletionFailureException e) { @@ -112,11 +121,13 @@ protected void deleteFiles(Set pathsToDelete, String fileType) { "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 deleteFuncToUse = deleteFunc == null ? defaultDeleteFunc : deleteFunc; @@ -127,8 +138,89 @@ protected void deleteFiles(Set pathsToDelete, String fileType) { .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 counts; + + DeleteSummary() { + Map 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 @@ protected Set expiredStatisticsFilesLocations( 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 statsFileLocations(TableMetadata tableMetadata) { Set statsFileLocations = Sets.newHashSet(); diff --git a/core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java b/core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java index 911a94c13f28..78c200cb2882 100644 --- a/core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java +++ b/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,10 +49,11 @@ 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 @@ -59,7 +61,7 @@ public void cleanFiles( // 4. Delete the manifest lists if (ExpireSnapshots.CleanupLevel.NONE == cleanupLevel) { LOG.info("Nothing to clean."); - return; + return summary; } Set 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 snapshots = afterExpiration.snapshots(); @@ -259,32 +261,44 @@ public void cleanFiles( }); if (ExpireSnapshots.CleanupLevel.ALL == cleanupLevel) { - Set filesToDelete = + Set filesToDelete = findFilesToDelete( manifestsToScan, manifestsToRevert, validIds, beforeExpiration.specsById()); - LOG.debug("Deleting {} data files", filesToDelete.size()); - deleteFiles(filesToDelete, "data"); + Map> groupedFilesToDelete = + filesToDelete.stream() + .collect( + Collectors.groupingBy( + file -> DeletedFileType.fromContent(file.getContent()), + Collectors.mapping(ExpiredContentFile::getPath, Collectors.toSet()))); + + for (Map.Entry> entry : groupedFilesToDelete.entrySet()) { + Set 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 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 findFilesToDelete( + private Set findFilesToDelete( Set manifestsToScan, Set manifestsToRevert, Set validIds, Map specsById) { - Set filesToDelete = ConcurrentHashMap.newKeySet(); + Set filesToDelete = ConcurrentHashMap.newKeySet(); Tasks.foreach(manifestsToScan) .retry(3) .suppressFailureWhenFinished() @@ -295,14 +309,15 @@ private Set 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> reader = + ManifestFiles.open(manifest, fileIO, specsById)) { + for (ManifestEntry> 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 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> reader = + ManifestFiles.open(manifest, fileIO, specsById)) { + for (ManifestEntry> 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) { diff --git a/core/src/main/java/org/apache/iceberg/ManifestFiles.java b/core/src/main/java/org/apache/iceberg/ManifestFiles.java index 8c16a0e44359..56ec05f35bac 100644 --- a/core/src/main/java/org/apache/iceberg/ManifestFiles.java +++ b/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 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 readColumns( + ManifestFile manifest, + FileIO io, + Map specsById, + Collection columns) { + return CloseableIterable.transform( + read(manifest, io, specsById).select(columns).liveEntries(), ManifestEntry::file); + } + /** * Returns a new {@link ManifestReader} for a {@link ManifestFile}. * diff --git a/core/src/main/java/org/apache/iceberg/ReachableFileCleanup.java b/core/src/main/java/org/apache/iceberg/ReachableFileCleanup.java index b5a565b81e7e..b9acc9185a59 100644 --- a/core/src/main/java/org/apache/iceberg/ReachableFileCleanup.java +++ b/core/src/main/java/org/apache/iceberg/ReachableFileCleanup.java @@ -19,6 +19,7 @@ package org.apache.iceberg; import java.io.IOException; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -28,6 +29,7 @@ import org.apache.iceberg.exceptions.RuntimeIOException; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.util.Tasks; import org.slf4j.Logger; @@ -40,6 +42,8 @@ class ReachableFileCleanup extends FileCleanupStrategy { private static final Logger LOG = LoggerFactory.getLogger(ReachableFileCleanup.class); + private static final List DELETE_CANDIDATE_COLUMNS = + ImmutableList.of("content", "file_path"); ReachableFileCleanup( FileIO fileIO, @@ -50,13 +54,14 @@ class ReachableFileCleanup extends FileCleanupStrategy { } @Override - public void cleanFiles( + public DeleteSummary cleanFiles( TableMetadata beforeExpiration, TableMetadata afterExpiration, ExpireSnapshots.CleanupLevel cleanupLevel) { + DeleteSummary summary = new DeleteSummary(); if (ExpireSnapshots.CleanupLevel.NONE == cleanupLevel) { LOG.info("Nothing to clean."); - return; + return summary; } Set manifestListsToDelete = Sets.newHashSet(); @@ -82,28 +87,40 @@ public void cleanFiles( if (!manifestsToDelete.isEmpty()) { if (ExpireSnapshots.CleanupLevel.ALL == cleanupLevel) { - Set dataFilesToDelete = + Set filesToDelete = findFilesToDelete(manifestsToDelete, currentManifests, beforeExpiration.specsById()); - LOG.debug("Deleting {} data files", dataFilesToDelete.size()); - deleteFiles(dataFilesToDelete, "data"); + Map> groupedFilesToDelete = + filesToDelete.stream() + .collect( + Collectors.groupingBy( + file -> DeletedFileType.fromContent(file.getContent()), + Collectors.mapping(ExpiredContentFile::getPath, Collectors.toSet()))); + + for (Map.Entry> entry : groupedFilesToDelete.entrySet()) { + Set filesToDeleteGroup = entry.getValue(); + DeletedFileType fileType = entry.getKey(); + LOG.debug("Deleting {} {} files", filesToDeleteGroup.size(), fileType.displayName()); + deleteFiles(filesToDeleteGroup, fileType, summary); + } } Set manifestPathsToDelete = manifestsToDelete.stream().map(ManifestFile::path).collect(Collectors.toSet()); LOG.debug("Deleting {} manifest files", manifestPathsToDelete.size()); - deleteFiles(manifestPathsToDelete, "manifest"); + deleteFiles(manifestPathsToDelete, 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 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 pruneReferencedManifests( @@ -167,11 +184,12 @@ private Set readManifests(Set snapshots) { return manifestFiles; } - private Set findFilesToDelete( + // Helper to determine data files to delete + private Set findFilesToDelete( Set manifestFilesToDelete, Set currentManifestFiles, Map specsById) { - Set filesToDelete = ConcurrentHashMap.newKeySet(); + Set filesToDelete = ConcurrentHashMap.newKeySet(); Tasks.foreach(manifestFilesToDelete) .retry(3) @@ -183,9 +201,13 @@ private Set findFilesToDelete( "Failed to determine live files in manifest {}. Retrying", item.path(), exc)) .run( manifest -> { - try (CloseableIterable paths = - ManifestFiles.readPaths(manifest, fileIO, specsById)) { - paths.forEach(filesToDelete::add); + try (CloseableIterable entries = + ManifestFiles.readColumns( + manifest, fileIO, specsById, DELETE_CANDIDATE_COLUMNS)) { + entries.forEach( + entry -> + filesToDelete.add( + new ExpiredContentFile(entry.content(), entry.location()))); } catch (IOException e) { throw new RuntimeIOException(e, "Failed to read manifest file: %s", manifest); } @@ -212,9 +234,13 @@ private Set findFilesToDelete( } // Remove all the live files from the candidate deletion set - try (CloseableIterable paths = - ManifestFiles.readPaths(manifest, fileIO, specsById)) { - paths.forEach(filesToDelete::remove); + try (CloseableIterable entries = + ManifestFiles.readColumns( + manifest, fileIO, specsById, DELETE_CANDIDATE_COLUMNS)) { + entries.forEach( + entry -> + filesToDelete.remove( + new ExpiredContentFile(entry.content(), entry.location()))); } catch (IOException e) { throw new RuntimeIOException(e, "Failed to read manifest file: %s", manifest); } diff --git a/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java b/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java index dbccdb86ed4e..9c6203da9f53 100644 --- a/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java +++ b/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java @@ -44,6 +44,8 @@ import java.util.stream.Collectors; import org.apache.iceberg.exceptions.CommitFailedException; import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.metrics.MetricsReporter; +import org.apache.iceberg.metrics.RemoveSnapshotsReport; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; @@ -80,6 +82,7 @@ class RemoveSnapshots implements ExpireSnapshots { private boolean cleanExpiredMetadata = false; private boolean cleanExpiredFiles = true; private CleanupLevel cleanupLevel = CleanupLevel.ALL; + private MetricsReporter metricsReporter; RemoveSnapshots(TableOperations ops) { this.ops = ops; @@ -193,6 +196,12 @@ public List apply() { return removed; } + /** Report metrics about the snapshot expiration to the given reporter. */ + RemoveSnapshots reportWith(MetricsReporter reporter) { + this.metricsReporter = reporter; + return this; + } + private TableMetadata internalApply() { this.base = ops.refresh(); // attempt to clean expired metadata even if there are no snapshots to expire @@ -407,7 +416,18 @@ private void cleanExpiredSnapshots() { : new ReachableFileCleanup( ops.io(), deleteExecutorService, planExecutorService(), deleteFunc); - cleanupStrategy.cleanFiles(base, current, cleanupLevel); + FileCleanupStrategy.DeleteSummary summary = + cleanupStrategy.cleanFiles(base, current, cleanupLevel); + if (metricsReporter != null) { + metricsReporter.report( + RemoveSnapshotsReport.of( + summary.dataFilesCount(), + summary.positionDeleteFilesCount(), + summary.equalityDeleteFilesCount(), + summary.manifestsCount(), + summary.manifestListsCount(), + summary.statisticsFilesCount())); + } } private void validateCleanupCanBeIncremental(TableMetadata current) { diff --git a/core/src/main/java/org/apache/iceberg/metrics/InMemoryMetricsReporter.java b/core/src/main/java/org/apache/iceberg/metrics/InMemoryMetricsReporter.java index 2fa9281003e7..5dd7492924d4 100644 --- a/core/src/main/java/org/apache/iceberg/metrics/InMemoryMetricsReporter.java +++ b/core/src/main/java/org/apache/iceberg/metrics/InMemoryMetricsReporter.java @@ -45,4 +45,11 @@ public CommitReport commitReport() { return null; } } + + public RemoveSnapshotsReport removeSnapshotsReport() { + Preconditions.checkArgument( + metricsReport == null || metricsReport instanceof RemoveSnapshotsReport, + "Metrics report is not a remove snapshots report"); + return (RemoveSnapshotsReport) metricsReport; + } } diff --git a/core/src/main/java/org/apache/iceberg/metrics/RemoveSnapshotsReport.java b/core/src/main/java/org/apache/iceberg/metrics/RemoveSnapshotsReport.java new file mode 100644 index 000000000000..4b00e1886953 --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/metrics/RemoveSnapshotsReport.java @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.metrics; + +import org.immutables.value.Value; + +@Value.Immutable +public abstract class RemoveSnapshotsReport implements MetricsReport { + public abstract long dataFilesCount(); + + public abstract long positionDeleteFilesCount(); + + public abstract long equalityDeleteFilesCount(); + + public abstract long manifestsCount(); + + public abstract long manifestListsCount(); + + public abstract long statisticsFilesCount(); + + public static RemoveSnapshotsReport of( + long dataFilesCount, + long positionDeleteFilesCount, + long equalityDeleteFilesCount, + long manifestsCount, + long manifestListsCount, + long statisticsFilesCount) { + return ImmutableRemoveSnapshotsReport.builder() + .dataFilesCount(dataFilesCount) + .positionDeleteFilesCount(positionDeleteFilesCount) + .equalityDeleteFilesCount(equalityDeleteFilesCount) + .manifestsCount(manifestsCount) + .manifestListsCount(manifestListsCount) + .statisticsFilesCount(statisticsFilesCount) + .build(); + } +} diff --git a/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java b/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java index eab95f3256ce..ec45f5fe9555 100644 --- a/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java +++ b/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java @@ -45,6 +45,8 @@ import org.apache.iceberg.io.BulkDeletionFailureException; import org.apache.iceberg.io.FileIO; import org.apache.iceberg.io.PositionOutputStream; +import org.apache.iceberg.metrics.InMemoryMetricsReporter; +import org.apache.iceberg.metrics.RemoveSnapshotsReport; import org.apache.iceberg.puffin.Blob; import org.apache.iceberg.puffin.Puffin; import org.apache.iceberg.puffin.PuffinWriter; @@ -1074,7 +1076,12 @@ public void testExpireWithDeleteFiles() { long fourthSnapshotTs = waitUntilAfter(fourthSnapshot.timestampMillis()); Set deletedFiles = Sets.newHashSet(); - removeSnapshots(table).expireOlderThan(fourthSnapshotTs).deleteWith(deletedFiles::add).commit(); + InMemoryMetricsReporter reporter = new InMemoryMetricsReporter(); + removeSnapshots(table) + .reportWith(reporter) + .expireOlderThan(fourthSnapshotTs) + .deleteWith(deletedFiles::add) + .commit(); assertThat(deletedFiles) .as("Should remove old delete files and delete file manifests") @@ -1091,6 +1098,14 @@ public void testExpireWithDeleteFiles() { .map(ManifestFile::path) .collect(Collectors.toList())) .build()); + + RemoveSnapshotsReport report = reporter.removeSnapshotsReport(); + assertThat(report.dataFilesCount()).isEqualTo(1); + assertThat(report.positionDeleteFilesCount()).isEqualTo(1); + assertThat(report.equalityDeleteFilesCount()).isEqualTo(0); + assertThat(report.manifestListsCount()).isEqualTo(3); + assertThat(report.manifestsCount()).isEqualTo(4); + assertThat(report.statisticsFilesCount()).isEqualTo(0); } @TestTemplate @@ -1212,7 +1227,8 @@ public void testExpireWithStatisticsFiles() throws IOException { assertThat(table.statisticsFiles()).hasSize(2); long tAfterCommits = waitUntilAfter(table.currentSnapshot().timestampMillis()); - removeSnapshots(table).expireOlderThan(tAfterCommits).commit(); + InMemoryMetricsReporter reporter = new InMemoryMetricsReporter(); + removeSnapshots(table).reportWith(reporter).expireOlderThan(tAfterCommits).commit(); // only the current snapshot and its stats file should be retained assertThat(table.snapshots()).hasSize(1); @@ -1224,6 +1240,14 @@ public void testExpireWithStatisticsFiles() throws IOException { assertThat(new File(URI.create(statsFileLocation1))).doesNotExist(); assertThat(new File(URI.create(statsFileLocation2))).exists(); + + RemoveSnapshotsReport report = reporter.removeSnapshotsReport(); + assertThat(report.dataFilesCount()).isEqualTo(0); + assertThat(report.positionDeleteFilesCount()).isEqualTo(0); + assertThat(report.equalityDeleteFilesCount()).isEqualTo(0); + assertThat(report.manifestListsCount()).isEqualTo(1); + assertThat(report.manifestsCount()).isEqualTo(0); + assertThat(report.statisticsFilesCount()).isEqualTo(1); } @TestTemplate