From ff2efecf5296457a52e5f674aebc384155518f3e Mon Sep 17 00:00:00 2001 From: Grant Nicholas Date: Mon, 21 Sep 2026 08:42:24 -0500 Subject: [PATCH 1/4] Core: Fix canContainDroppedFiles when minSequenceNumber set Orphaned DVs were sometimes never removed in ManifestFilterManager. canContainDroppedFiles only returned true for a delete manifest whose minSequenceNumber was below the dropDeleteFilesOlderThan threshold when data files were also being removed in the same commit. A commit that only calls dropDeleteFilesOlderThan (e.g. a plain append) never admitted delete manifests for that check. --- .../apache/iceberg/ManifestFilterManager.java | 6 + .../iceberg/TestManifestFilterManager.java | 111 ++++++++++++++++++ 2 files changed, 117 insertions(+) create mode 100644 core/src/test/java/org/apache/iceberg/TestManifestFilterManager.java diff --git a/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java b/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java index 96fa1b1a68b3..55806b3165d2 100644 --- a/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java +++ b/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java @@ -436,6 +436,12 @@ private boolean canContainDroppedPartitions(ManifestFile manifest) { } private boolean canContainDroppedFiles(ManifestFile manifest) { + if (manifest.content() == ManifestContent.DELETES + && minSequenceNumber > 0 + && manifest.minSequenceNumber() < minSequenceNumber) { + return true; + } + if (!deletePaths.isEmpty()) { return true; } else if (!deleteFiles.isEmpty()) { diff --git a/core/src/test/java/org/apache/iceberg/TestManifestFilterManager.java b/core/src/test/java/org/apache/iceberg/TestManifestFilterManager.java new file mode 100644 index 000000000000..27746b449641 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/TestManifestFilterManager.java @@ -0,0 +1,111 @@ +/* + * 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; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assumptions.assumeThat; + +import java.io.IOException; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.util.DeleteFileSet; +import org.apache.iceberg.util.ThreadPools; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; + +@ExtendWith(ParameterizedTestExtension.class) +public class TestManifestFilterManager extends TestBase { + + @TestTemplate + public void obsoleteDeleteFilesAreFoundWithoutRemovedDataFiles() throws IOException { + assumeThat(formatVersion).as("delete files require v2+").isGreaterThanOrEqualTo(2); + + // a manifest holding an obsolete delete file must be read even when no data files are removed + // in the same commit + ManifestEntry entry = + manifestEntry(ManifestEntry.Status.EXISTING, 1L, 5L, 5L, FILE_B_DELETES); + ManifestFile manifestB = writeManifest(1L, entry); + + CountingFilterManager filterManager = new CountingFilterManager(); + filterManager.dropDeleteFilesOlderThan(6L); + + filterManager.filterManifests(SCHEMA, ImmutableList.of(manifestB)); + + assertThat(filterManager.opened) + .as("A manifest that can hold a delete file below the sequence number must still be read") + .containsExactly(manifestB.path()); + } + + @TestTemplate + public void manifestsAtOrAboveTheSequenceNumberAreNotOpened() throws IOException { + assumeThat(formatVersion).as("delete files require v2+").isGreaterThanOrEqualTo(2); + + // a manifest whose minimum data sequence number is already at or above the threshold cannot + // contain an obsolete delete file, so it should not be opened + ManifestEntry entry = + manifestEntry(ManifestEntry.Status.EXISTING, 1L, 5L, 5L, FILE_B_DELETES); + ManifestFile manifestB = writeManifest(1L, entry); + + CountingFilterManager filterManager = new CountingFilterManager(); + filterManager.dropDeleteFilesOlderThan(5L); + + List filtered = + filterManager.filterManifests(SCHEMA, ImmutableList.of(manifestB)); + + assertThat(filterManager.opened) + .as("A manifest that cannot hold an obsolete delete file should not be read") + .isEmpty(); + assertThat(filtered).containsExactly(manifestB); + } + + /** A delete-manifest filter manager that records every manifest it opens. */ + private class CountingFilterManager extends ManifestFilterManager { + final Set opened = ConcurrentHashMap.newKeySet(); + + CountingFilterManager() { + super(table.specs(), ThreadPools::getWorkerPool); + } + + @Override + protected void deleteFile(String location) {} + + @Override + protected ManifestWriter newManifestWriter(PartitionSpec spec) { + OutputFile outputFile = + Files.localOutput( + manifestFormat() + .addExtension(temp.resolve("filtered" + System.nanoTime()).toFile().toString())); + return ManifestFiles.writeDeleteManifest(formatVersion, spec, outputFile, 2L); + } + + @Override + protected ManifestReader newManifestReader(ManifestFile manifest) { + opened.add(manifest.path()); + return ManifestFiles.readDeleteManifest(manifest, FILE_IO, table.specs()); + } + + @Override + protected Set newFileSet() { + return DeleteFileSet.create(); + } + } +} From 78b4b8f156f3bff6f1ad304ea5af8547efae74d6 Mon Sep 17 00:00:00 2001 From: Grant Nicholas Date: Mon, 21 Sep 2026 08:43:08 -0500 Subject: [PATCH 2/4] Core: Fix canContainDroppedFiles to check each condition independently The deletePaths/deleteFiles/removedDataFilePaths checks were chained as else-if branches, so once one of the earlier conditions applied, later ones were never evaluated. This meant a manifest could be skipped (and a dangling DV missed) whenever both deleteFiles and removedDataFilePaths were non-empty --- .../apache/iceberg/ManifestFilterManager.java | 14 +++++++---- .../iceberg/TestManifestFilterManager.java | 23 +++++++++++++++++++ 2 files changed, 32 insertions(+), 5 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java b/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java index 55806b3165d2..b348480aad23 100644 --- a/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java +++ b/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java @@ -85,8 +85,7 @@ public String partition() { private boolean failMissingDeletePaths = false; private boolean caseSensitive = true; private boolean allDeletesReferenceManifests = true; - // this is only being used for the DeleteManifestFilterManager to detect orphaned DVs for removed - // data file paths + // only used for the DeleteManifestFilterManager to detect orphaned DVs for removed data files private Set removedDataFilePaths = Sets.newHashSet(); // cache filtered manifests to avoid extra work when commits fail. @@ -444,9 +443,14 @@ private boolean canContainDroppedFiles(ManifestFile manifest) { if (!deletePaths.isEmpty()) { return true; - } else if (!deleteFiles.isEmpty()) { - return ManifestFileUtil.canContainAny(manifest, deleteFilePartitions, specsById); - } else if (!removedDataFilePaths.isEmpty()) { + } + + if (!deleteFiles.isEmpty() + && ManifestFileUtil.canContainAny(manifest, deleteFilePartitions, specsById)) { + return true; + } + + if (!removedDataFilePaths.isEmpty()) { return true; } diff --git a/core/src/test/java/org/apache/iceberg/TestManifestFilterManager.java b/core/src/test/java/org/apache/iceberg/TestManifestFilterManager.java index 27746b449641..98b9b7affcd4 100644 --- a/core/src/test/java/org/apache/iceberg/TestManifestFilterManager.java +++ b/core/src/test/java/org/apache/iceberg/TestManifestFilterManager.java @@ -27,6 +27,7 @@ import java.util.concurrent.ConcurrentHashMap; import org.apache.iceberg.io.OutputFile; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; import org.apache.iceberg.util.DeleteFileSet; import org.apache.iceberg.util.ThreadPools; import org.junit.jupiter.api.TestTemplate; @@ -77,6 +78,28 @@ public void manifestsAtOrAboveTheSequenceNumberAreNotOpened() throws IOException assertThat(filtered).containsExactly(manifestB); } + @TestTemplate + public void danglingDVsAreFoundWhenDeleteFilesAreAlsoRemoved() throws IOException { + assumeThat(formatVersion).as("DVs are only written in v3 and later").isGreaterThanOrEqualTo(3); + + ManifestFile manifestA = writeDeleteManifest(formatVersion, 1L, newDV(FILE_A)); + + // FILE_B_DELETES has no manifest location, so canTrustManifestReferences is false + // This tests only tests the non-trusted manifest path + assertThat(FILE_B_DELETES.manifestLocation()).isNull(); + + CountingFilterManager filterManager = new CountingFilterManager(); + filterManager.delete(FILE_B_DELETES); + filterManager.removeDanglingDeletesFor(ImmutableSet.of(FILE_A)); + filterManager.filterManifests(SCHEMA, ImmutableList.of(manifestA)); + + assertThat(filterManager.opened) + .as( + "A dangling DV must be found even when an unrelated delete file in a different " + + "partition is also explicitly removed in the same commit") + .containsExactly(manifestA.path()); + } + /** A delete-manifest filter manager that records every manifest it opens. */ private class CountingFilterManager extends ManifestFilterManager { final Set opened = ConcurrentHashMap.newKeySet(); From 838b40129e8f205b8a1d28395a53dbc918f730f1 Mon Sep 17 00:00:00 2001 From: Grant Nicholas Date: Wed, 16 Sep 2026 14:42:29 -0500 Subject: [PATCH 3/4] Core: Measure partitioned tables and object store latency in RewriteDataFilesBenchmark The benchmark builds an unpartitioned table on the local filesystem, so it cannot show the cost of reading or writing manifests. A local open takes well under a millisecond while an object store charges tens of milliseconds per round trip, and manifest work against remote storage is dominated by those round trips rather than by decoding. --- .../iceberg/RewriteDataFilesBenchmark.java | 297 +++++++++++++++++- 1 file changed, 281 insertions(+), 16 deletions(-) diff --git a/core/src/jmh/java/org/apache/iceberg/RewriteDataFilesBenchmark.java b/core/src/jmh/java/org/apache/iceberg/RewriteDataFilesBenchmark.java index 34b5b58599db..cdb9397e08db 100644 --- a/core/src/jmh/java/org/apache/iceberg/RewriteDataFilesBenchmark.java +++ b/core/src/jmh/java/org/apache/iceberg/RewriteDataFilesBenchmark.java @@ -25,8 +25,14 @@ import java.util.Map; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; +import org.apache.iceberg.encryption.EncryptionManager; import org.apache.iceberg.hadoop.HadoopTables; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.InputFile; import org.apache.iceberg.io.LocationProvider; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.io.PositionOutputStream; +import org.apache.iceberg.io.SeekableInputStream; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; @@ -73,23 +79,46 @@ public class RewriteDataFilesBenchmark { required(5, "timestamp_col", Types.TimestampType.withoutZone()), required(6, "timestamp_tz_col", Types.TimestampType.withZone()), required(7, "str_col", Types.StringType.get())); - private static final PartitionSpec SPEC = PartitionSpec.unpartitioned(); private static final HadoopTables TABLES = new HadoopTables(); private Table table; private DataFileSet dataFilesToRemove; private DataFileSet dataFilesToAdd; - @Param({"50000", "100000", "500000", "1000000", "2000000"}) + @Param({"50000", "100000", "500000"}) private int numFiles; - @Param({"5", "25", "50", "100"}) + @Param({"1", "5", "50", "100"}) private int percentDataFilesRewritten; + /** Number of partitions to spread the data files over, or 0 for an unpartitioned table. */ + @Param({"0", "1000"}) + private int numPartitions; + + /** + * Latency charged to every metadata file read and write, or 0 for no latency. Helps simulate + * object storage like latency even though using local disk. + */ + @Param({"0", "50"}) + private int metadataFileLatencyMs; + + /** + * Target manifest size, which controls how many manifests the files are spread over. Helps + * simulate larger tables with many manifests without generating extremely large table metadata. + */ + @Param({"65536", "8388608"}) + private long manifestTargetSizeBytes; + @Setup public void setupBenchmark() throws IOException { initTable(); initFiles(); + + this.table = + new BaseTable( + new LatencyInjectingTableOperations( + ((HasTableOperations) table).operations(), metadataFileLatencyMs), + TABLE_IDENT); } @TearDown @@ -116,7 +145,27 @@ private void initTable() { this.table = TABLES.create( - SCHEMA, SPEC, ImmutableMap.of(TableProperties.FORMAT_VERSION, "3"), TABLE_IDENT); + SCHEMA, + spec(), + ImmutableMap.of( + TableProperties.FORMAT_VERSION, + "3", + TableProperties.MANIFEST_TARGET_SIZE_BYTES, + String.valueOf(manifestTargetSizeBytes)), + TABLE_IDENT); + } + + private PartitionSpec spec() { + return numPartitions > 0 + ? PartitionSpec.builderFor(SCHEMA).identity("int_col").build() + : PartitionSpec.unpartitioned(); + } + + private int partitionValue(int ordinal) { + // Assign files to partitions in contiguous blocks to pack them by partition value. + // This allows partition pruning optimization, as manifests are clustered by partition value. + int filesPerPartition = (numFiles + numPartitions - 1) / numPartitions; + return ordinal / filesPerPartition; } private void dropTable() { @@ -129,13 +178,14 @@ private void initFiles() throws IOException { Map filesToReplace = Maps.newHashMapWithExpectedSize(numDataFilesToRewrite); RowDelta rowDelta = table.newRowDelta(); for (int ordinal = 0; ordinal < numFiles; ordinal++) { - DataFile dataFile = generateDataFile(); + DataFile dataFile = generateDataFile(ordinal); rowDelta.addRows(dataFile); DeleteFile deleteFile = FileGenerationUtil.generateDV(table, dataFile); rowDelta.addDeletes(deleteFile); if (numDataFilesToRewrite > 0) { filesToReplace.put(dataFile.location(), dataFile); - DataFile pendingDataFile = generateDataFile(dataFile.recordCount()); + // the replacement lands in the same partition as the file it replaces + DataFile pendingDataFile = generateDataFile(ordinal, dataFile.recordCount()); rowDelta.addRows(pendingDataFile); pendingDataFiles.add(pendingDataFile); numDataFilesToRewrite--; @@ -163,15 +213,224 @@ private void initFiles() throws IOException { this.dataFilesToAdd = DataFileSet.of(pendingDataFiles); } - private DataFile generateDataFile() { - return generateDataFile(-1L); + /** Delegating operations whose FileIO charges a fixed latency for each open. */ + private static class LatencyInjectingTableOperations implements TableOperations { + private final TableOperations delegate; + private final int latencyMs; + + LatencyInjectingTableOperations(TableOperations delegate, int latencyMs) { + this.delegate = delegate; + this.latencyMs = latencyMs; + } + + @Override + public TableMetadata current() { + return delegate.current(); + } + + @Override + public TableMetadata refresh() { + return delegate.refresh(); + } + + @Override + public void commit(TableMetadata base, TableMetadata metadata) { + delegate.commit(base, metadata); + } + + @Override + public FileIO io() { + return new LatencyInjectingFileIO(delegate.io(), latencyMs); + } + + @Override + public EncryptionManager encryption() { + return delegate.encryption(); + } + + @Override + public String metadataFileLocation(String fileName) { + return delegate.metadataFileLocation(fileName); + } + + @Override + public LocationProvider locationProvider() { + return delegate.locationProvider(); + } + + @Override + public TableOperations temp(TableMetadata uncommittedMetadata) { + return new LatencyInjectingTableOperations(delegate.temp(uncommittedMetadata), latencyMs); + } + + @Override + public long newSnapshotId() { + return delegate.newSnapshotId(); + } + + @Override + public boolean requireStrictCleanup() { + return delegate.requireStrictCleanup(); + } + } + + private static class LatencyInjectingFileIO implements FileIO { + private final FileIO delegate; + private final int latencyMs; + + LatencyInjectingFileIO(FileIO delegate, int latencyMs) { + this.delegate = delegate; + this.latencyMs = latencyMs; + } + + @Override + public InputFile newInputFile(String path) { + return new LatencyInjectingInputFile(delegate.newInputFile(path), latencyMs); + } + + @Override + public InputFile newInputFile(String path, long length) { + return new LatencyInjectingInputFile(delegate.newInputFile(path, length), latencyMs); + } + + @Override + public OutputFile newOutputFile(String path) { + return new LatencyInjectingOutputFile(delegate.newOutputFile(path), latencyMs); + } + + @Override + public void deleteFile(String path) { + delegate.deleteFile(path); + } } - private DataFile generateDataFile(long recordCount) { + /** + * An object store pays for a write when the object is completed rather than when the handle is + * created, so the latency is charged on close. + */ + private static class LatencyInjectingOutputFile implements OutputFile { + private final OutputFile delegate; + private final int latencyMs; + + LatencyInjectingOutputFile(OutputFile delegate, int latencyMs) { + this.delegate = delegate; + this.latencyMs = latencyMs; + } + + @Override + public PositionOutputStream create() { + return new LatencyInjectingPositionOutputStream(delegate.create(), latencyMs); + } + + @Override + public PositionOutputStream createOrOverwrite() { + return new LatencyInjectingPositionOutputStream(delegate.createOrOverwrite(), latencyMs); + } + + @Override + public String location() { + return delegate.location(); + } + + @Override + public InputFile toInputFile() { + return new LatencyInjectingInputFile(delegate.toInputFile(), latencyMs); + } + } + + private static class LatencyInjectingPositionOutputStream extends PositionOutputStream { + private final PositionOutputStream delegate; + private final int latencyMs; + + LatencyInjectingPositionOutputStream(PositionOutputStream delegate, int latencyMs) { + this.delegate = delegate; + this.latencyMs = latencyMs; + } + + @Override + public long getPos() throws IOException { + return delegate.getPos(); + } + + @Override + public void write(int b) throws IOException { + delegate.write(b); + } + + @Override + public void write(byte[] buffer, int offset, int length) throws IOException { + delegate.write(buffer, offset, length); + } + + @Override + public void flush() throws IOException { + delegate.flush(); + } + + @Override + public void close() throws IOException { + sleep(latencyMs); + delegate.close(); + } + } + + private static class LatencyInjectingInputFile implements InputFile { + private final InputFile delegate; + private final int latencyMs; + + LatencyInjectingInputFile(InputFile delegate, int latencyMs) { + this.delegate = delegate; + this.latencyMs = latencyMs; + } + + @Override + public long getLength() { + return delegate.getLength(); + } + + @Override + public SeekableInputStream newStream() { + sleep(latencyMs); + return delegate.newStream(); + } + + @Override + public String location() { + return delegate.location(); + } + + @Override + public boolean exists() { + return delegate.exists(); + } + } + + private static void sleep(int latencyMs) { + if (latencyMs <= 0) { + return; + } + + try { + Thread.sleep(latencyMs); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException("Interrupted while simulating latency", e); + } + } + + private DataFile generateDataFile(int ordinal) { + return generateDataFile(ordinal, -1L); + } + + private DataFile generateDataFile(int ordinal, long recordCount) { Schema schema = table.schema(); PartitionSpec spec = table.spec(); LocationProvider locations = table.locationProvider(); - String path = locations.newDataLocation(spec, null, FileGenerationUtil.generateFileName()); + String fileName = FileGenerationUtil.generateFileName(); + String path = + numPartitions > 0 + ? locations.newDataLocation(fileName) + : locations.newDataLocation(spec, null, fileName); long fileSize = ThreadLocalRandom.current().nextLong(50_000L); MetricsConfig metricsConfig = MetricsConfig.forTable(table); Metrics metrics = @@ -187,11 +446,17 @@ private DataFile generateDataFile(long recordCount) { metrics.nanValueCounts()); } - return DataFiles.builder(spec) - .withPath(path) - .withFileSizeInBytes(fileSize) - .withFormat(FileFormat.PARQUET) - .withMetrics(metrics) - .build(); + DataFiles.Builder builder = + DataFiles.builder(spec) + .withPath(path) + .withFileSizeInBytes(fileSize) + .withFormat(FileFormat.PARQUET) + .withMetrics(metrics); + + if (numPartitions > 0) { + builder.withPartitionPath("int_col=" + partitionValue(ordinal)); + } + + return builder.build(); } } From b247cb4efb20b203f4c019bffab1a7c598ca39fc Mon Sep 17 00:00:00 2001 From: Grant Nicholas Date: Wed, 16 Sep 2026 14:42:29 -0500 Subject: [PATCH 4/4] Core: Prune delete manifests that cannot hold a dangling DV Removing data files makes every DV that references them dangling, so ManifestFilterManager scans delete manifests to find them. Previously canContainDroppedFiles admitted every delete manifest unconditionally, so a commit that replaces data files opened every delete manifest. Now uses partition pruning to determine which delete manifests to open. This mirrors partition pruning done with data manifests. --- .../apache/iceberg/ManifestFilterManager.java | 21 ++++-- .../iceberg/TestManifestFilterManager.java | 74 +++++++++++++++++++ 2 files changed, 89 insertions(+), 6 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java b/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java index b348480aad23..ac539ba0b82c 100644 --- a/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java +++ b/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java @@ -87,6 +87,8 @@ public String partition() { private boolean allDeletesReferenceManifests = true; // only used for the DeleteManifestFilterManager to detect orphaned DVs for removed data files private Set removedDataFilePaths = Sets.newHashSet(); + // partitions of the removed data files, used to skip delete manifests that cannot hold a DV + private PartitionSet removedDataFilePartitions; // cache filtered manifests to avoid extra work when commits fail. private final Map filteredManifests = Maps.newConcurrentMap(); @@ -102,6 +104,7 @@ protected ManifestFilterManager( this.specsById = specsById; this.deleteFilePartitions = PartitionSet.create(specsById); this.dropPartitions = PartitionSet.create(specsById); + this.removedDataFilePartitions = PartitionSet.create(specsById); this.workerPoolSupplier = executorSupplier; } @@ -169,6 +172,13 @@ void caseSensitive(boolean newCaseSensitive) { protected void removeDanglingDeletesFor(Set dataFiles) { this.removedDataFilePaths = dataFiles.stream().map(ContentFile::location).collect(Collectors.toSet()); + + PartitionSet partitions = PartitionSet.create(specsById); + for (DataFile dataFile : dataFiles) { + partitions.add(dataFile.specId(), dataFile.partition()); + } + + this.removedDataFilePartitions = partitions; } /** Add a specific path to be deleted in the new snapshot. */ @@ -445,12 +455,11 @@ private boolean canContainDroppedFiles(ManifestFile manifest) { return true; } - if (!deleteFiles.isEmpty() - && ManifestFileUtil.canContainAny(manifest, deleteFilePartitions, specsById)) { - return true; - } - - if (!removedDataFilePaths.isEmpty()) { + if ((!deleteFiles.isEmpty() || !removedDataFilePaths.isEmpty()) + && ManifestFileUtil.canContainAny( + manifest, + Iterables.concat(deleteFilePartitions, removedDataFilePartitions), + specsById)) { return true; } diff --git a/core/src/test/java/org/apache/iceberg/TestManifestFilterManager.java b/core/src/test/java/org/apache/iceberg/TestManifestFilterManager.java index 98b9b7affcd4..9d70e5e8b06f 100644 --- a/core/src/test/java/org/apache/iceberg/TestManifestFilterManager.java +++ b/core/src/test/java/org/apache/iceberg/TestManifestFilterManager.java @@ -36,6 +36,80 @@ @ExtendWith(ParameterizedTestExtension.class) public class TestManifestFilterManager extends TestBase { + @TestTemplate + public void removedDataFileSkipsDeleteManifestsInOtherPartitions() throws IOException { + assumeThat(formatVersion).as("DVs are only written in v3 and later").isGreaterThanOrEqualTo(3); + + // FILE_A is in data_bucket=0 and FILE_B is in data_bucket=1 + ManifestFile manifestA = writeDeleteManifest(formatVersion, 1L, newDV(FILE_A)); + ManifestFile manifestB = writeDeleteManifest(formatVersion, 1L, newDV(FILE_B)); + + CountingFilterManager filterManager = new CountingFilterManager(); + filterManager.removeDanglingDeletesFor(ImmutableSet.of(FILE_A)); + + List filtered = + filterManager.filterManifests(SCHEMA, ImmutableList.of(manifestA, manifestB)); + + assertThat(filterManager.opened) + .as("Only the delete manifest covering the removed data file's partition should be read") + .containsExactly(manifestA.path()); + + assertThat(filtered.get(0).path()) + .as("The manifest holding the dangling DV should be rewritten") + .isNotEqualTo(manifestA.path()); + assertThat(filtered.get(1)) + .as("The manifest in an unrelated partition should pass through untouched") + .isEqualTo(manifestB); + } + + @TestTemplate + public void obsoleteDeleteFilesAreStillFoundInOtherPartitions() throws IOException { + assumeThat(formatVersion).as("DVs are only written in v3 and later").isGreaterThanOrEqualTo(3); + + ManifestEntry entry = + manifestEntry(ManifestEntry.Status.EXISTING, 1L, 5L, 5L, newDV(FILE_B)); + ManifestFile manifestB = writeManifest(1L, entry); + assertThat(manifestB.minSequenceNumber()).isEqualTo(5L); + + // cannot partition prune the delete manifest due to the minSequenceNumber + CountingFilterManager obsolete = new CountingFilterManager(); + obsolete.removeDanglingDeletesFor(ImmutableSet.of(FILE_A)); + obsolete.dropDeleteFilesOlderThan(6L); + obsolete.filterManifests(SCHEMA, ImmutableList.of(manifestB)); + + assertThat(obsolete.opened) + .as("A manifest that can hold a delete file below the sequence number must still be read") + .containsExactly(manifestB.path()); + + // can partition prune the delete manifest due to the minSequenceNumber + CountingFilterManager notObsolete = new CountingFilterManager(); + notObsolete.removeDanglingDeletesFor(ImmutableSet.of(FILE_A)); + notObsolete.dropDeleteFilesOlderThan(5L); + notObsolete.filterManifests(SCHEMA, ImmutableList.of(manifestB)); + + assertThat(notObsolete.opened) + .as("A manifest whose delete files are not obsolete must still be pruned by partition") + .isEmpty(); + } + + @TestTemplate + public void removedDataFileWithoutDanglingDVsReadsNothing() throws IOException { + assumeThat(formatVersion).as("DVs are only written in v3 and later").isGreaterThanOrEqualTo(3); + + ManifestFile manifestB = writeDeleteManifest(formatVersion, 1L, newDV(FILE_B)); + + CountingFilterManager filterManager = new CountingFilterManager(); + filterManager.removeDanglingDeletesFor(ImmutableSet.of(FILE_A)); + + List filtered = + filterManager.filterManifests(SCHEMA, ImmutableList.of(manifestB)); + + assertThat(filterManager.opened) + .as("No delete manifest can hold a DV for the removed data file, so none should be read") + .isEmpty(); + assertThat(filtered).containsExactly(manifestB); + } + @TestTemplate public void obsoleteDeleteFilesAreFoundWithoutRemovedDataFiles() throws IOException { assumeThat(formatVersion).as("delete files require v2+").isGreaterThanOrEqualTo(2);