diff --git a/core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java b/core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java index 69f82931833e..e72dbef6dbcc 100644 --- a/core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java +++ b/core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java @@ -47,6 +47,7 @@ import org.apache.iceberg.puffin.PuffinReader; import org.apache.iceberg.puffin.PuffinWriter; 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.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Sets; @@ -377,7 +378,12 @@ public static RewriteResult rewriteDataManifest( * @param stagingLocation staging location for rewritten files (referred delete file will be * rewritten here) * @return a copy plan of content files in the manifest that was rewritten + * @deprecated since 1.12.0, will be removed in 1.13.0; use the overload that accepts the map of + * rewritten position delete file sizes. This overload records the original {@code + * file_size_in_bytes}, which can be inconsistent with the rewritten file size on disk once + * embedded data file paths change length. */ + @Deprecated public static RewriteResult rewriteDeleteManifest( ManifestFile manifestFile, Set snapshotIds, @@ -389,6 +395,52 @@ public static RewriteResult rewriteDeleteManifest( String targetPrefix, String stagingLocation) throws IOException { + return rewriteDeleteManifest( + manifestFile, + snapshotIds, + outputFile, + io, + format, + specsById, + sourcePrefix, + targetPrefix, + stagingLocation, + ImmutableMap.of()); + } + + /** + * Rewrite a delete manifest, replacing path references. + * + *

This is a metadata-only operation: position delete file content is rewritten separately (see + * {@link #rewritePositionDelete}). The actual sizes of those rewritten files are supplied via + * {@code rewrittenDeleteFileSizes} and recorded in the manifest so that {@code + * file_size_in_bytes} stays consistent with the rewritten file on disk. + * + * @param manifestFile source delete manifest to rewrite + * @param snapshotIds snapshot ids for filtering returned delete manifest entries + * @param outputFile output file to rewrite manifest file to + * @param io file io + * @param format format of the manifest file + * @param specsById map of partition specs by id + * @param sourcePrefix source prefix that will be replaced + * @param targetPrefix target prefix that will replace it + * @param stagingLocation staging location for rewritten position delete files + * @param rewrittenDeleteFileSizes map from source position delete file path to the actual size of + * the rewritten file; entries absent from the map keep their original size + * @return a copy plan of content files in the manifest that was rewritten + */ + public static RewriteResult rewriteDeleteManifest( + ManifestFile manifestFile, + Set snapshotIds, + OutputFile outputFile, + FileIO io, + int format, + Map specsById, + String sourcePrefix, + String targetPrefix, + String stagingLocation, + Map rewrittenDeleteFileSizes) + throws IOException { PartitionSpec spec = specsById.get(manifestFile.partitionSpecId()); try (ManifestWriter writer = ManifestFiles.writeDeleteManifest(format, spec, outputFile, manifestFile.snapshotId()); @@ -405,7 +457,8 @@ public static RewriteResult rewriteDeleteManifest( sourcePrefix, targetPrefix, stagingLocation, - writer)) + writer, + rewrittenDeleteFileSizes)) .reduce(new RewriteResult<>(), RewriteResult::append); } } @@ -445,14 +498,20 @@ private static RewriteResult writeDeleteFileEntry( String sourcePrefix, String targetPrefix, String stagingLocation, - ManifestWriter writer) { + ManifestWriter writer, + Map rewrittenDeleteFileSizes) { DeleteFile file = entry.file(); RewriteResult result = new RewriteResult<>(); switch (file.content()) { case POSITION_DELETES: - DeleteFile posDeleteFile = newPositionDeleteEntry(file, spec, sourcePrefix, targetPrefix); + // Path rewrites change the file size; use the measured size, falling back to the original + // for entries that were not rewritten (e.g. deleted entries not copied to the target). + long fileSizeInBytes = + rewrittenDeleteFileSizes.getOrDefault(file.location(), file.fileSizeInBytes()); + DeleteFile posDeleteFile = + newPositionDeleteEntry(file, spec, sourcePrefix, targetPrefix, fileSizeInBytes); appendEntryWithFile(entry, writer, posDeleteFile); // keep the following entries in metadata but exclude them from copyPlan // 1) deleted position delete files @@ -523,7 +582,11 @@ private static DeleteFile newEqualityDeleteEntry( } private static DeleteFile newPositionDeleteEntry( - DeleteFile file, PartitionSpec spec, String sourcePrefix, String targetPrefix) { + DeleteFile file, + PartitionSpec spec, + String sourcePrefix, + String targetPrefix, + long fileSizeInBytes) { String path = file.location(); Preconditions.checkArgument( path.startsWith(sourcePrefix), @@ -535,6 +598,7 @@ private static DeleteFile newPositionDeleteEntry( FileMetadata.deleteFileBuilder(spec) .copy(file) .withPath(newPath(path, sourcePrefix, targetPrefix)) + .withFileSizeInBytes(fileSizeInBytes) .withMetrics(ContentFileUtil.replacePathBounds(file, sourcePrefix, targetPrefix)); // Update referencedDataFile for DV files @@ -607,7 +671,11 @@ PositionDeleteWriter writer( * @param sourcePrefix source prefix that will be replaced * @param targetPrefix target prefix to replace it * @param posDeleteReaderWriter class to read and write position delete files + * @deprecated since 1.12.0, will be removed in 1.13.0; use {@link #rewritePositionDelete} which + * returns the size of the rewritten file so callers can record an accurate {@code + * file_size_in_bytes}. */ + @Deprecated public static void rewritePositionDeleteFile( DeleteFile deleteFile, OutputFile outputFile, @@ -617,6 +685,37 @@ public static void rewritePositionDeleteFile( String targetPrefix, PositionDeleteReaderWriter posDeleteReaderWriter) throws IOException { + rewritePositionDelete( + deleteFile, outputFile, io, spec, sourcePrefix, targetPrefix, posDeleteReaderWriter); + } + + /** + * Rewrite a position delete file, replacing path references, and return the size of the rewritten + * file. + * + *

The size is measured from the writer after it is closed (rather than via a separate {@code + * getLength()}/HEAD call), so it is accurate even on file systems where the length of an + * in-progress write underreports. Callers record this size as {@code file_size_in_bytes} in the + * rewritten manifest. + * + * @param deleteFile source position delete file to rewrite + * @param outputFile output file to write the rewritten delete file to + * @param io file io + * @param spec spec of delete file + * @param sourcePrefix source prefix that will be replaced + * @param targetPrefix target prefix to replace it + * @param posDeleteReaderWriter class to read and write position delete files + * @return the size in bytes of the rewritten file + */ + public static long rewritePositionDelete( + DeleteFile deleteFile, + OutputFile outputFile, + FileIO io, + PartitionSpec spec, + String sourcePrefix, + String targetPrefix, + PositionDeleteReaderWriter posDeleteReaderWriter) + throws IOException { String path = deleteFile.location(); if (!path.startsWith(sourcePrefix)) { throw new UnsupportedOperationException( @@ -625,8 +724,7 @@ public static void rewritePositionDeleteFile( // DV files (Puffin format for v3+) need special handling to rewrite internal blob metadata if (ContentFileUtil.isDV(deleteFile)) { - rewriteDVFile(deleteFile, outputFile, io, sourcePrefix, targetPrefix); - return; + return rewriteDVFile(deleteFile, outputFile, io, sourcePrefix, targetPrefix); } // For non-DV position delete files (v2), rewrite using the reader/writer @@ -655,9 +753,14 @@ record = recordIt.next(); writer.write(newPositionDeleteRecord(record, sourcePrefix, targetPrefix)); } } + + writer.close(); + return writer.length(); } } } + + return 0; } /** @@ -668,8 +771,9 @@ record = recordIt.next(); * @param io file io * @param sourcePrefix source prefix that will be replaced * @param targetPrefix target prefix to replace it + * @return the size in bytes of the rewritten DV file */ - private static void rewriteDVFile( + private static long rewriteDVFile( DeleteFile deleteFile, OutputFile outputFile, FileIO io, @@ -708,6 +812,8 @@ private static void rewriteDVFile( try (PuffinWriter writer = Puffin.write(outputFile).createdBy(IcebergBuild.fullVersion()).build()) { rewrittenBlobs.forEach(writer::write); + writer.close(); + return writer.length(); } } diff --git a/core/src/test/java/org/apache/iceberg/TestRewriteTablePathUtil.java b/core/src/test/java/org/apache/iceberg/TestRewriteTablePathUtil.java index bedd8dd66d71..1b0f5f6b1c70 100644 --- a/core/src/test/java/org/apache/iceberg/TestRewriteTablePathUtil.java +++ b/core/src/test/java/org/apache/iceberg/TestRewriteTablePathUtil.java @@ -24,6 +24,9 @@ import java.io.IOException; import java.util.Set; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestTemplate; import org.junit.jupiter.api.extension.ExtendWith; @@ -277,8 +280,100 @@ public void testRewritingMultiplePositionDeleteEntriesWithinManifestFile() throw table.specs(), sourcePrefix, targetPrefix, - stagingDir); + stagingDir, + ImmutableMap.of()); assertThat(deleteFileRewriteResult.toRewrite()).hasSize(2); } + + // A position delete entry that is not rewritten (e.g. a DELETED entry not copied to the target) + // is absent from the measured-size map and must keep its original file_size_in_bytes. + @TestTemplate + public void testRewriteDeleteManifestFallsBackToOriginalSizeForDeletedEntries() + throws IOException { + assumeThat(formatVersion) + .as("Delete files only work for format version 2+") + .isGreaterThanOrEqualTo(2); + + String sourcePrefix = "/path/to/"; + String targetPrefix = "/path/new/"; + String stagingDir = "/staging/"; + + // FILE_A_DELETES is live, so its rewritten size is measured and supplied; FILE_B_DELETES is a + // DELETED entry, absent from the size map. + ManifestFile manifest = deleteManifestWithLiveAndDeletedEntry(FILE_A_DELETES, FILE_B_DELETES); + + long measuredSizeForA = 9999L; + OutputFile output = + Files.localOutput( + FileFormat.AVRO.addExtension( + temp.resolve("junit" + System.nanoTime()).toFile().toString())); + RewriteTablePathUtil.rewriteDeleteManifest( + manifest, + Set.of(1000L), + output, + table.io(), + formatVersion, + table.specs(), + sourcePrefix, + targetPrefix, + stagingDir, + ImmutableMap.of(FILE_A_DELETES.location(), measuredSizeForA)); + + InputFile rewrittenInput = output.toInputFile(); + ManifestFile rewritten = + new GenericManifestFile( + rewrittenInput.location(), + rewrittenInput.getLength(), + SPEC.specId(), + ManifestContent.DELETES, + 0L, + 0L, + 1000L, + null, + null, + null, + null, + null, + null, + null, + null, + null); + int seen = 0; + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(rewritten, table.io(), table.specs())) { + for (ManifestEntry entry : reader.entries()) { + seen++; + if (entry.status() == ManifestEntry.Status.DELETED) { + assertThat(entry.file().fileSizeInBytes()) + .as("DELETED entry should fall back to its original size") + .isEqualTo(FILE_B_DELETES.fileSizeInBytes()); + } else { + assertThat(entry.file().fileSizeInBytes()) + .as("Live entry should use the measured rewritten size") + .isEqualTo(measuredSizeForA); + } + } + } + + assertThat(seen).as("Both the live and deleted entries should be present").isEqualTo(2); + } + + private ManifestFile deleteManifestWithLiveAndDeletedEntry(DeleteFile live, DeleteFile deleted) + throws IOException { + OutputFile manifestFile = + Files.localOutput( + FileFormat.AVRO.addExtension( + temp.resolve("junit" + System.nanoTime()).toFile().toString())); + ManifestWriter writer = + ManifestFiles.writeDeleteManifest(formatVersion, SPEC, manifestFile, 1000L); + try { + writer.add(live); + writer.delete(deleted, 1, null); + } finally { + writer.close(); + } + + return writer.toManifestFile(); + } } diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java index aedb25e4a4a6..11935e815e76 100644 --- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java +++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java @@ -22,6 +22,7 @@ import java.io.IOException; import java.io.OutputStreamWriter; import java.nio.charset.StandardCharsets; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Set; @@ -31,9 +32,13 @@ import org.apache.iceberg.ContentFile; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; import org.apache.iceberg.FileFormat; import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.ManifestContent; import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.ManifestReader; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.PartitionStatisticsFile; import org.apache.iceberg.RewriteTablePathUtil; @@ -69,13 +74,14 @@ import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; 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; import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.spark.JobGroupInfo; import org.apache.iceberg.spark.source.SerializableTableWithSize; import org.apache.iceberg.util.DeleteFileSet; import org.apache.iceberg.util.Pair; import org.apache.iceberg.util.Tasks; -import org.apache.spark.api.java.function.ForeachFunction; +import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.MapFunction; import org.apache.spark.api.java.function.ReduceFunction; import org.apache.spark.broadcast.Broadcast; @@ -87,6 +93,7 @@ import org.apache.spark.sql.functions; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import scala.Tuple2; public class RewriteTablePathSparkAction extends BaseSparkAction implements RewriteTablePath { @@ -272,7 +279,8 @@ private String jobDesc() { *

    *
  • Rebuild version files to staging *
  • Rebuild manifest list files to staging - *
  • Rebuild manifest to staging + *
  • Rewrite referenced position delete files to staging + *
  • Rebuild manifests to staging *
  • Get all files needed to move *
*/ @@ -308,24 +316,29 @@ private Result rebuildMetadata() { RewriteResult rewriteManifestListResult = new RewriteResult<>(); manifestListResults.forEach(rewriteManifestListResult::append); - // rebuild manifest files - Set metaFiles = rewriteManifestListResult.toRewrite(); - RewriteContentFileResult rewriteManifestResult = - rewriteManifests(deltaSnapshots, endMetadata, metaFiles); + Set manifestFiles = rewriteManifestListResult.toRewrite(); // rebuild position delete files - Set deleteFiles = - rewriteManifestResult.toRewrite().stream() - .filter(e -> e instanceof DeleteFile) - .map(e -> (DeleteFile) e) - .collect(Collectors.toCollection(DeleteFileSet::create)); - rewritePositionDeletes(deleteFiles); + Set deleteManifests = + manifestFiles.stream() + .filter(manifest -> manifest.content() == ManifestContent.DELETES) + .collect(Collectors.toSet()); + Set deleteFilesToRewrite = positionDeletesToRewrite(deleteManifests); + Map rewrittenDeleteFileSizes = rewritePositionDeletes(deleteFilesToRewrite); + + // rebuild manifest files + RewriteContentFileResult rewriteManifestResult = + rewriteManifests( + deltaSnapshots, + endMetadata, + manifestFiles, + sparkContext().broadcast(rewrittenDeleteFileSizes)); ImmutableRewriteTablePath.Result.Builder builder = ImmutableRewriteTablePath.Result.builder() .stagingLocation(stagingDir) - .rewrittenDeleteFilePathsCount(deleteFiles.size()) - .rewrittenManifestFilePathsCount(metaFiles.size()) + .rewrittenDeleteFilePathsCount(deleteFilesToRewrite.size()) + .rewrittenManifestFilePathsCount(manifestFiles.size()) .latestVersion(RewriteTablePathUtil.fileName(endVersionName)); if (!createFileList) { @@ -561,7 +574,10 @@ public RewriteContentFileResult appendDeleteFile(RewriteResult r1) { /** Rewrite manifest files in a distributed manner and return rewritten data files path pairs. */ private RewriteContentFileResult rewriteManifests( - Set deltaSnapshots, TableMetadata tableMetadata, Set toRewrite) { + Set deltaSnapshots, + TableMetadata tableMetadata, + Set toRewrite, + Broadcast> rewrittenDeleteFileSizes) { if (toRewrite.isEmpty()) { return new RewriteContentFileResult(); } @@ -581,7 +597,8 @@ private RewriteContentFileResult rewriteManifests( stagingDir, tableMetadata.formatVersion(), sourcePrefix, - targetPrefix), + targetPrefix, + rewrittenDeleteFileSizes), Encoders.bean(RewriteContentFileResult.class)) // duplicates are expected here as the same data file can have different statuses // (e.g. added and deleted) @@ -594,7 +611,8 @@ private static MapFunction toManifests( String stagingLocation, int format, String sourcePrefix, - String targetPrefix) { + String targetPrefix, + Broadcast> rewrittenDeleteFileSizes) { return manifestFile -> { RewriteContentFileResult result = new RewriteContentFileResult(); @@ -619,7 +637,8 @@ private static MapFunction toManifests( stagingLocation, format, sourcePrefix, - targetPrefix)); + targetPrefix, + rewrittenDeleteFileSizes)); break; default: throw new UnsupportedOperationException( @@ -665,7 +684,8 @@ private static RewriteResult writeDeleteManifest( String stagingLocation, int format, String sourcePrefix, - String targetPrefix) { + String targetPrefix, + Broadcast> rewrittenDeleteFileSizes) { try { String stagingPath = RewriteTablePathUtil.stagingPath(manifestFile.path(), sourcePrefix, stagingLocation); @@ -682,27 +702,120 @@ private static RewriteResult writeDeleteManifest( specsById, sourcePrefix, targetPrefix, - stagingLocation); + stagingLocation, + rewrittenDeleteFileSizes.value()); } catch (IOException e) { throw new RuntimeIOException(e); } } - private void rewritePositionDeletes(Set toRewrite) { + /** + * Enumerate the distinct position delete files referenced by the given delete manifests. Deduped + * by identity (location, offset, size) so a file shared across manifests is counted once; the + * physical rewrite is further deduped by location in {@link #rewritePositionDeletes}. + */ + private Set positionDeletesToRewrite(Set deleteManifests) { + if (deleteManifests.isEmpty()) { + return Collections.emptySet(); + } + + Encoder manifestFileEncoder = Encoders.javaSerialization(ManifestFile.class); + Dataset manifestDS = + spark().createDataset(Lists.newArrayList(deleteManifests), manifestFileEncoder); + Encoder deleteFileEncoder = Encoders.javaSerialization(DeleteFile.class); + + List referencedDeleteFiles = + manifestDS + .repartition(deleteManifests.size()) + .flatMap(positionDeletesInManifest(tableBroadcast()), deleteFileEncoder) + .collectAsList(); + + return DeleteFileSet.of(referencedDeleteFiles); + } + + private static FlatMapFunction positionDeletesInManifest( + Broadcast tableArg) { + return manifestFile -> { + Table table = tableArg.getValue(); + List deleteFiles = Lists.newArrayList(); + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifestFile, table.io(), table.specs())) { + for (DeleteFile deleteFile : reader) { + if (deleteFile.content() == FileContent.POSITION_DELETES) { + deleteFiles.add(deleteFile.copy()); + } + } + } + return deleteFiles.iterator(); + }; + } + + /** + * Rewrite the given position delete files in parallel, returning a map from each source delete + * file path to the size of its rewritten file. Physical files are deduped by location, so a + * Puffin file holding multiple DVs is rewritten once and its size keyed once. + */ + private Map rewritePositionDeletes(Set toRewrite) { if (toRewrite.isEmpty()) { - return; + return Collections.emptyMap(); + } + + // Multiple DVs can share one Puffin file at different blob offsets; rewrite each physical file + // once. The measured size is keyed by location and applied to every referencing manifest entry. + Map byLocation = Maps.newHashMapWithExpectedSize(toRewrite.size()); + for (DeleteFile deleteFile : toRewrite) { + byLocation.putIfAbsent(deleteFile.location(), deleteFile); } + List physicalFiles = Lists.newArrayList(byLocation.values()); Encoder deleteFileEncoder = Encoders.javaSerialization(DeleteFile.class); - Dataset deleteFileDs = - spark().createDataset(Lists.newArrayList(toRewrite), deleteFileEncoder); + Dataset deleteFileDS = spark().createDataset(physicalFiles, deleteFileEncoder); PositionDeleteReaderWriter posDeleteReaderWriter = new SparkPositionDeleteReaderWriter(); - deleteFileDs - .repartition(toRewrite.size()) - .foreach( - rewritePositionDelete( - tableBroadcast(), sourcePrefix, targetPrefix, stagingDir, posDeleteReaderWriter)); + List> rewrittenSizes = + deleteFileDS + .repartition(physicalFiles.size()) + .map( + rewritePositionDelete( + tableBroadcast(), + sourcePrefix, + targetPrefix, + stagingDir, + posDeleteReaderWriter), + Encoders.tuple(Encoders.STRING(), Encoders.LONG())) + .collectAsList(); + + Map sizesBySourcePath = Maps.newHashMapWithExpectedSize(rewrittenSizes.size()); + for (Tuple2 entry : rewrittenSizes) { + sizesBySourcePath.put(entry._1(), entry._2()); + } + return sizesBySourcePath; + } + + private static MapFunction> rewritePositionDelete( + Broadcast
tableArg, + String sourcePrefixArg, + String targetPrefixArg, + String stagingLocationArg, + PositionDeleteReaderWriter posDeleteReaderWriter) { + return deleteFile -> { + FileIO io = tableArg.getValue().io(); + String newPath = + RewriteTablePathUtil.stagingPath( + deleteFile.location(), sourcePrefixArg, stagingLocationArg); + OutputFile outputFile = io.newOutputFile(newPath); + PartitionSpec spec = tableArg.getValue().specs().get(deleteFile.specId()); + long rewrittenLength = + RewriteTablePathUtil.rewritePositionDelete( + deleteFile, + outputFile, + io, + spec, + sourcePrefixArg, + targetPrefixArg, + posDeleteReaderWriter); + return new Tuple2<>(deleteFile.location(), rewrittenLength); + }; } private static class SparkPositionDeleteReaderWriter implements PositionDeleteReaderWriter { @@ -724,30 +837,6 @@ public PositionDeleteWriter writer( } } - private ForeachFunction rewritePositionDelete( - Broadcast
tableArg, - String sourcePrefixArg, - String targetPrefixArg, - String stagingLocationArg, - PositionDeleteReaderWriter posDeleteReaderWriter) { - return deleteFile -> { - FileIO io = tableArg.getValue().io(); - String newPath = - RewriteTablePathUtil.stagingPath( - deleteFile.location(), sourcePrefixArg, stagingLocationArg); - OutputFile outputFile = io.newOutputFile(newPath); - PartitionSpec spec = tableArg.getValue().specs().get(deleteFile.specId()); - RewriteTablePathUtil.rewritePositionDeleteFile( - deleteFile, - outputFile, - io, - spec, - sourcePrefixArg, - targetPrefixArg, - posDeleteReaderWriter); - }; - } - private static CloseableIterable positionDeletesReader( InputFile inputFile, FileFormat format, PartitionSpec spec) { return FormatModelRegistry.readBuilder(format, Record.class, inputFile) diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java index c5db04762f21..9660beae187e 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java @@ -30,6 +30,7 @@ import java.nio.file.Path; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.function.Predicate; @@ -41,15 +42,21 @@ import org.apache.iceberg.BaseTable; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileFormat; import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.ManifestReader; import org.apache.iceberg.Parameter; import org.apache.iceberg.ParameterizedTestExtension; import org.apache.iceberg.Parameters; import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.RowDelta; import org.apache.iceberg.Schema; import org.apache.iceberg.Snapshot; import org.apache.iceberg.SnapshotChanges; import org.apache.iceberg.StaticTableOperations; +import org.apache.iceberg.StructLike; import org.apache.iceberg.Table; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.TableProperties; @@ -62,14 +69,18 @@ import org.apache.iceberg.data.FileHelpers; import org.apache.iceberg.data.GenericRecord; import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.BaseDVFileWriter; +import org.apache.iceberg.deletes.DVFileWriter; import org.apache.iceberg.deletes.PositionDelete; import org.apache.iceberg.hadoop.HadoopTables; import org.apache.iceberg.io.FileIO; import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.io.OutputFileFactory; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; 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.spark.SparkCatalog; import org.apache.iceberg.spark.TestBase; import org.apache.iceberg.spark.source.ThreeColumnRecord; @@ -629,12 +640,7 @@ public void testPositionDeletesDeduplication() throws Exception { // in a new manifest, which will cause duplicate DeleteFile objects when processing tableWithPosDeletes.newRowDelta().addDeletes(positionDeletes).commit(); - // This should NOT throw AlreadyExistsException - the fix uses DeleteFileSet to deduplicate - // Without the fix (using Collectors.toSet()), this would fail because: - // 1. Both manifests contain entries for the same delete file - // 2. Processing returns two different DeleteFile objects for the same file - // 3. HashSet doesn't deduplicate them (DeleteFile doesn't override equals()) - // 4. rewritePositionDeletes tries to write the same file twice -> AlreadyExistsException + // This should NOT throw AlreadyExistsException RewriteTablePath.Result result = actions() .rewriteTablePath(tableWithPosDeletes) @@ -642,13 +648,295 @@ public void testPositionDeletesDeduplication() throws Exception { .rewriteLocationPrefix(tableWithPosDeletes.location(), targetTableLocation()) .execute(); - // Verify the rewrite completed successfully - should have rewritten exactly 1 delete file - // (the duplicate should be deduplicated by DeleteFileSet) assertThat(result.rewrittenDeleteFilePathsCount()) .as("Should have rewritten exactly 1 delete file after deduplication") .isEqualTo(1); } + // Regression test: when the same position delete file is referenced from manifests in different + // snapshots, it must be rewritten once and the resulting size stamped consistently into every + // manifest that references it. The delete file is enumerated and deduped by path before the + // rewrite, so its measured size is shared across all referencing delete manifests. + @TestTemplate + public void testSharedDeleteFileSizeAcrossManifests() throws Exception { + assumeThat(formatVersion) + .as("Format versions 3+ use DVs with different validation rules") + .isEqualTo(2); + + Table tableWithPosDeletes = + createTableWithSnapshots( + tableDir.toFile().toURI().toString().concat("tableWithSharedDelete"), + 1, + Map.of(TableProperties.DELETE_DEFAULT_FILE_FORMAT, "parquet")); + + DataFile dataFile = + tableWithPosDeletes + .currentSnapshot() + .addedDataFiles(tableWithPosDeletes.io()) + .iterator() + .next(); + + List> deletes = Lists.newArrayList(Pair.of(dataFile.location(), 0L)); + File deleteFile = + new File( + removePrefix(tableWithPosDeletes.location() + "/data/deeply/nested/deletes.parquet")); + DeleteFile positionDeletes = + FileHelpers.writeDeleteFile( + tableWithPosDeletes, + tableWithPosDeletes.io().newOutputFile(deleteFile.toURI().toString()), + deletes, + formatVersion) + .first(); + + tableWithPosDeletes.newRowDelta().addDeletes(positionDeletes).commit(); + tableWithPosDeletes.newRowDelta().addDeletes(positionDeletes).commit(); + + RewriteTablePath.Result result = + actions() + .rewriteTablePath(tableWithPosDeletes) + .stagingLocation(stagingLocation()) + .rewriteLocationPrefix(tableWithPosDeletes.location(), targetTableLocation()) + .execute(); + copyTableFiles(result); + + Table targetTable = TABLES.load(targetTableLocation()); + List deleteManifests = + targetTable.currentSnapshot().deleteManifests(targetTable.io()); + assertThat(deleteManifests) + .as("Expected the shared delete file to be referenced by multiple manifests") + .hasSizeGreaterThanOrEqualTo(2); + for (ManifestFile manifest : deleteManifests) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, targetTable.io(), targetTable.specs())) { + for (DeleteFile df : reader) { + long manifestSize = df.fileSizeInBytes(); + long actualSize = targetTable.io().newInputFile(df.location()).getLength(); + assertThat(manifestSize) + .as( + "file_size_in_bytes in rewritten manifest should match actual file size for %s", + df.location()) + .isEqualTo(actualSize); + } + } + } + } + + // Regression test: a single delete manifest can reference multiple distinct position delete + // files, and each entry must be stamped with its own rewritten size. The two delete files carry + // a different number of records so they rewrite to different sizes, which catches a per-path size + // map that collapses entries to a single size or falls back to the stale original size. + @TestTemplate + public void testMultipleDistinctDeleteFileSizesAfterRewrite() throws Exception { + assumeThat(formatVersion) + .as("Format versions 3+ use DVs with different validation rules") + .isEqualTo(2); + + Table tableWithPosDeletes = + createTableWithSnapshots( + tableDir.toFile().toURI().toString().concat("tableWithDistinctDeletes"), + 2, + Map.of(TableProperties.DELETE_DEFAULT_FILE_FORMAT, "parquet")); + + List dataFiles = Lists.newArrayList(); + tableWithPosDeletes + .snapshots() + .forEach( + snapshot -> snapshot.addedDataFiles(tableWithPosDeletes.io()).forEach(dataFiles::add)); + assertThat(dataFiles).as("Expected two data files to reference from deletes").hasSize(2); + + // One delete file holds a single record, the other holds two, so they rewrite to distinct + // on-disk sizes. + File smallDeleteFile = + new File( + removePrefix( + tableWithPosDeletes.location() + "/data/deeply/nested/deletes-small.parquet")); + DeleteFile smallDelete = + FileHelpers.writeDeleteFile( + tableWithPosDeletes, + tableWithPosDeletes.io().newOutputFile(smallDeleteFile.toURI().toString()), + Lists.newArrayList(Pair.of(dataFiles.get(0).location(), 0L)), + formatVersion) + .first(); + + File largeDeleteFile = + new File( + removePrefix( + tableWithPosDeletes.location() + "/data/deeply/nested/deletes-large.parquet")); + DeleteFile largeDelete = + FileHelpers.writeDeleteFile( + tableWithPosDeletes, + tableWithPosDeletes.io().newOutputFile(largeDeleteFile.toURI().toString()), + Lists.newArrayList( + Pair.of(dataFiles.get(0).location(), 0L), + Pair.of(dataFiles.get(1).location(), 0L)), + formatVersion) + .first(); + + tableWithPosDeletes.newRowDelta().addDeletes(smallDelete).addDeletes(largeDelete).commit(); + + RewriteTablePath.Result result = + actions() + .rewriteTablePath(tableWithPosDeletes) + .stagingLocation(stagingLocation()) + .rewriteLocationPrefix(tableWithPosDeletes.location(), targetTableLocation()) + .execute(); + copyTableFiles(result); + + Table targetTable = TABLES.load(targetTableLocation()); + List rewrittenSizes = Lists.newArrayList(); + for (ManifestFile manifest : targetTable.currentSnapshot().deleteManifests(targetTable.io())) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, targetTable.io(), targetTable.specs())) { + for (DeleteFile df : reader) { + long manifestSize = df.fileSizeInBytes(); + long actualSize = targetTable.io().newInputFile(df.location()).getLength(); + assertThat(manifestSize) + .as( + "file_size_in_bytes in rewritten manifest should match actual file size for %s", + df.location()) + .isEqualTo(actualSize); + rewrittenSizes.add(manifestSize); + } + } + } + + assertThat(rewrittenSizes) + .as( + "The two distinct delete files should rewrite to distinct, independently recorded sizes") + .hasSize(2) + .doesNotHaveDuplicates(); + } + + // Regression test: rewriting delete file paths changes the file size (since the + // embedded data file paths may differ in length), but file_size_in_bytes in the rewritten + // manifest was not updated. Readers that use file_size_in_bytes to elide a stat() call may + // fail. + @TestTemplate + public void testDeleteFileSizeInBytesAfterRewrite() throws Exception { + List> deletes = + Lists.newArrayList( + Pair.of( + SnapshotChanges.builderFor(table) + .build() + .addedDataFiles() + .iterator() + .next() + .location(), + 0L)); + + File file = new File(removePrefix(table.location() + "/data/deeply/nested/deletes.parquet")); + DeleteFile positionDeletes = + FileHelpers.writeDeleteFile( + table, table.io().newOutputFile(file.toURI().toString()), deletes, formatVersion) + .first(); + table.newRowDelta().addDeletes(positionDeletes).commit(); + + RewriteTablePath.Result result = + actions() + .rewriteTablePath(table) + .stagingLocation(stagingLocation()) + .rewriteLocationPrefix(table.location(), targetTableLocation()) + .execute(); + copyTableFiles(result); + + Table targetTable = TABLES.load(targetTableLocation()); + for (ManifestFile manifest : targetTable.currentSnapshot().deleteManifests(targetTable.io())) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, targetTable.io(), targetTable.specs())) { + for (DeleteFile df : reader) { + long manifestSize = df.fileSizeInBytes(); + long actualSize = targetTable.io().newInputFile(df.location()).getLength(); + assertThat(manifestSize) + .as( + "file_size_in_bytes in rewritten manifest should match actual file size for %s", + df.location()) + .isEqualTo(actualSize); + } + } + } + } + + // Regression test: a single Puffin file can hold multiple DVs (one blob per data file) + // referenced by distinct DeleteFile entries that share the same location. The rewrite must + // rewrite + // the physical Puffin file once (rather than colliding on the staging path) and stamp the + // rewritten size into every referencing manifest entry. + @TestTemplate + public void testSharedPuffinDeleteFileSizeAfterRewrite() throws Exception { + assumeThat(formatVersion) + .as("DVs are introduced in v3; v4 writes parquet manifests the test setup cannot read") + .isEqualTo(3); + + Table tableWithDVs = + createTableWithSnapshots( + tableDir.toFile().toURI().toString().concat("tableWithSharedPuffin"), 2); + + List dataFilePaths = Lists.newArrayList(); + tableWithDVs + .snapshots() + .forEach( + snapshot -> + snapshot + .addedDataFiles(tableWithDVs.io()) + .forEach(dataFile -> dataFilePaths.add(dataFile.location()))); + assertThat(dataFilePaths).as("Expected two data files to back two DVs").hasSize(2); + + List dvs = writeDVsForDataFiles(tableWithDVs, dataFilePaths); + assertThat(dvs) + .as("Both DVs should live in a single Puffin file") + .hasSize(2) + .allSatisfy(dv -> assertThat(dv.location()).isEqualTo(dvs.get(0).location())); + + RowDelta rowDelta = tableWithDVs.newRowDelta(); + dvs.forEach(rowDelta::addDeletes); + rowDelta.commit(); + + RewriteTablePath.Result result = + actions() + .rewriteTablePath(tableWithDVs) + .stagingLocation(stagingLocation()) + .rewriteLocationPrefix(tableWithDVs.location(), targetTableLocation()) + .execute(); + assertThat(result.rewrittenDeleteFilePathsCount()) + .as("Two DVs are two delete files with rewritten paths") + .isEqualTo(2); + copyTableFiles(result); + + Table targetTable = TABLES.load(targetTableLocation()); + Set rewrittenLocations = Sets.newHashSet(); + for (ManifestFile manifest : targetTable.currentSnapshot().deleteManifests(targetTable.io())) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, targetTable.io(), targetTable.specs())) { + for (DeleteFile df : reader) { + rewrittenLocations.add(df.location()); + long actualSize = targetTable.io().newInputFile(df.location()).getLength(); + assertThat(df.fileSizeInBytes()) + .as("file_size_in_bytes should match the rewritten Puffin size for %s", df.location()) + .isEqualTo(actualSize); + } + } + } + assertThat(rewrittenLocations) + .as("Both DVs should point at the single rewritten Puffin file") + .hasSize(1); + } + + // Writes one DV per data file path into a single Puffin file, returning the resulting DeleteFiles + // (which share a location but carry distinct blob offsets). + private List writeDVsForDataFiles(Table targetTable, List dataFilePaths) + throws IOException { + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(targetTable, 1, 1).format(FileFormat.PUFFIN).build(); + DVFileWriter writer = new BaseDVFileWriter(fileFactory, p -> null); + try (DVFileWriter closeableWriter = writer) { + for (String path : dataFilePaths) { + closeableWriter.delete(path, 0L, targetTable.spec(), (StructLike) null); + } + } + + return writer.result().deleteFiles(); + } + @TestTemplate public void testEqualityDeletes() throws Exception { Table sourceTable = createTableWithSnapshots(newTableLocation(), 1); diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java index aedb25e4a4a6..11935e815e76 100644 --- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java +++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java @@ -22,6 +22,7 @@ import java.io.IOException; import java.io.OutputStreamWriter; import java.nio.charset.StandardCharsets; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Set; @@ -31,9 +32,13 @@ import org.apache.iceberg.ContentFile; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; import org.apache.iceberg.FileFormat; import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.ManifestContent; import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.ManifestReader; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.PartitionStatisticsFile; import org.apache.iceberg.RewriteTablePathUtil; @@ -69,13 +74,14 @@ import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; 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; import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.spark.JobGroupInfo; import org.apache.iceberg.spark.source.SerializableTableWithSize; import org.apache.iceberg.util.DeleteFileSet; import org.apache.iceberg.util.Pair; import org.apache.iceberg.util.Tasks; -import org.apache.spark.api.java.function.ForeachFunction; +import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.MapFunction; import org.apache.spark.api.java.function.ReduceFunction; import org.apache.spark.broadcast.Broadcast; @@ -87,6 +93,7 @@ import org.apache.spark.sql.functions; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import scala.Tuple2; public class RewriteTablePathSparkAction extends BaseSparkAction implements RewriteTablePath { @@ -272,7 +279,8 @@ private String jobDesc() { *
    *
  • Rebuild version files to staging *
  • Rebuild manifest list files to staging - *
  • Rebuild manifest to staging + *
  • Rewrite referenced position delete files to staging + *
  • Rebuild manifests to staging *
  • Get all files needed to move *
*/ @@ -308,24 +316,29 @@ private Result rebuildMetadata() { RewriteResult rewriteManifestListResult = new RewriteResult<>(); manifestListResults.forEach(rewriteManifestListResult::append); - // rebuild manifest files - Set metaFiles = rewriteManifestListResult.toRewrite(); - RewriteContentFileResult rewriteManifestResult = - rewriteManifests(deltaSnapshots, endMetadata, metaFiles); + Set manifestFiles = rewriteManifestListResult.toRewrite(); // rebuild position delete files - Set deleteFiles = - rewriteManifestResult.toRewrite().stream() - .filter(e -> e instanceof DeleteFile) - .map(e -> (DeleteFile) e) - .collect(Collectors.toCollection(DeleteFileSet::create)); - rewritePositionDeletes(deleteFiles); + Set deleteManifests = + manifestFiles.stream() + .filter(manifest -> manifest.content() == ManifestContent.DELETES) + .collect(Collectors.toSet()); + Set deleteFilesToRewrite = positionDeletesToRewrite(deleteManifests); + Map rewrittenDeleteFileSizes = rewritePositionDeletes(deleteFilesToRewrite); + + // rebuild manifest files + RewriteContentFileResult rewriteManifestResult = + rewriteManifests( + deltaSnapshots, + endMetadata, + manifestFiles, + sparkContext().broadcast(rewrittenDeleteFileSizes)); ImmutableRewriteTablePath.Result.Builder builder = ImmutableRewriteTablePath.Result.builder() .stagingLocation(stagingDir) - .rewrittenDeleteFilePathsCount(deleteFiles.size()) - .rewrittenManifestFilePathsCount(metaFiles.size()) + .rewrittenDeleteFilePathsCount(deleteFilesToRewrite.size()) + .rewrittenManifestFilePathsCount(manifestFiles.size()) .latestVersion(RewriteTablePathUtil.fileName(endVersionName)); if (!createFileList) { @@ -561,7 +574,10 @@ public RewriteContentFileResult appendDeleteFile(RewriteResult r1) { /** Rewrite manifest files in a distributed manner and return rewritten data files path pairs. */ private RewriteContentFileResult rewriteManifests( - Set deltaSnapshots, TableMetadata tableMetadata, Set toRewrite) { + Set deltaSnapshots, + TableMetadata tableMetadata, + Set toRewrite, + Broadcast> rewrittenDeleteFileSizes) { if (toRewrite.isEmpty()) { return new RewriteContentFileResult(); } @@ -581,7 +597,8 @@ private RewriteContentFileResult rewriteManifests( stagingDir, tableMetadata.formatVersion(), sourcePrefix, - targetPrefix), + targetPrefix, + rewrittenDeleteFileSizes), Encoders.bean(RewriteContentFileResult.class)) // duplicates are expected here as the same data file can have different statuses // (e.g. added and deleted) @@ -594,7 +611,8 @@ private static MapFunction toManifests( String stagingLocation, int format, String sourcePrefix, - String targetPrefix) { + String targetPrefix, + Broadcast> rewrittenDeleteFileSizes) { return manifestFile -> { RewriteContentFileResult result = new RewriteContentFileResult(); @@ -619,7 +637,8 @@ private static MapFunction toManifests( stagingLocation, format, sourcePrefix, - targetPrefix)); + targetPrefix, + rewrittenDeleteFileSizes)); break; default: throw new UnsupportedOperationException( @@ -665,7 +684,8 @@ private static RewriteResult writeDeleteManifest( String stagingLocation, int format, String sourcePrefix, - String targetPrefix) { + String targetPrefix, + Broadcast> rewrittenDeleteFileSizes) { try { String stagingPath = RewriteTablePathUtil.stagingPath(manifestFile.path(), sourcePrefix, stagingLocation); @@ -682,27 +702,120 @@ private static RewriteResult writeDeleteManifest( specsById, sourcePrefix, targetPrefix, - stagingLocation); + stagingLocation, + rewrittenDeleteFileSizes.value()); } catch (IOException e) { throw new RuntimeIOException(e); } } - private void rewritePositionDeletes(Set toRewrite) { + /** + * Enumerate the distinct position delete files referenced by the given delete manifests. Deduped + * by identity (location, offset, size) so a file shared across manifests is counted once; the + * physical rewrite is further deduped by location in {@link #rewritePositionDeletes}. + */ + private Set positionDeletesToRewrite(Set deleteManifests) { + if (deleteManifests.isEmpty()) { + return Collections.emptySet(); + } + + Encoder manifestFileEncoder = Encoders.javaSerialization(ManifestFile.class); + Dataset manifestDS = + spark().createDataset(Lists.newArrayList(deleteManifests), manifestFileEncoder); + Encoder deleteFileEncoder = Encoders.javaSerialization(DeleteFile.class); + + List referencedDeleteFiles = + manifestDS + .repartition(deleteManifests.size()) + .flatMap(positionDeletesInManifest(tableBroadcast()), deleteFileEncoder) + .collectAsList(); + + return DeleteFileSet.of(referencedDeleteFiles); + } + + private static FlatMapFunction positionDeletesInManifest( + Broadcast
tableArg) { + return manifestFile -> { + Table table = tableArg.getValue(); + List deleteFiles = Lists.newArrayList(); + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifestFile, table.io(), table.specs())) { + for (DeleteFile deleteFile : reader) { + if (deleteFile.content() == FileContent.POSITION_DELETES) { + deleteFiles.add(deleteFile.copy()); + } + } + } + return deleteFiles.iterator(); + }; + } + + /** + * Rewrite the given position delete files in parallel, returning a map from each source delete + * file path to the size of its rewritten file. Physical files are deduped by location, so a + * Puffin file holding multiple DVs is rewritten once and its size keyed once. + */ + private Map rewritePositionDeletes(Set toRewrite) { if (toRewrite.isEmpty()) { - return; + return Collections.emptyMap(); + } + + // Multiple DVs can share one Puffin file at different blob offsets; rewrite each physical file + // once. The measured size is keyed by location and applied to every referencing manifest entry. + Map byLocation = Maps.newHashMapWithExpectedSize(toRewrite.size()); + for (DeleteFile deleteFile : toRewrite) { + byLocation.putIfAbsent(deleteFile.location(), deleteFile); } + List physicalFiles = Lists.newArrayList(byLocation.values()); Encoder deleteFileEncoder = Encoders.javaSerialization(DeleteFile.class); - Dataset deleteFileDs = - spark().createDataset(Lists.newArrayList(toRewrite), deleteFileEncoder); + Dataset deleteFileDS = spark().createDataset(physicalFiles, deleteFileEncoder); PositionDeleteReaderWriter posDeleteReaderWriter = new SparkPositionDeleteReaderWriter(); - deleteFileDs - .repartition(toRewrite.size()) - .foreach( - rewritePositionDelete( - tableBroadcast(), sourcePrefix, targetPrefix, stagingDir, posDeleteReaderWriter)); + List> rewrittenSizes = + deleteFileDS + .repartition(physicalFiles.size()) + .map( + rewritePositionDelete( + tableBroadcast(), + sourcePrefix, + targetPrefix, + stagingDir, + posDeleteReaderWriter), + Encoders.tuple(Encoders.STRING(), Encoders.LONG())) + .collectAsList(); + + Map sizesBySourcePath = Maps.newHashMapWithExpectedSize(rewrittenSizes.size()); + for (Tuple2 entry : rewrittenSizes) { + sizesBySourcePath.put(entry._1(), entry._2()); + } + return sizesBySourcePath; + } + + private static MapFunction> rewritePositionDelete( + Broadcast
tableArg, + String sourcePrefixArg, + String targetPrefixArg, + String stagingLocationArg, + PositionDeleteReaderWriter posDeleteReaderWriter) { + return deleteFile -> { + FileIO io = tableArg.getValue().io(); + String newPath = + RewriteTablePathUtil.stagingPath( + deleteFile.location(), sourcePrefixArg, stagingLocationArg); + OutputFile outputFile = io.newOutputFile(newPath); + PartitionSpec spec = tableArg.getValue().specs().get(deleteFile.specId()); + long rewrittenLength = + RewriteTablePathUtil.rewritePositionDelete( + deleteFile, + outputFile, + io, + spec, + sourcePrefixArg, + targetPrefixArg, + posDeleteReaderWriter); + return new Tuple2<>(deleteFile.location(), rewrittenLength); + }; } private static class SparkPositionDeleteReaderWriter implements PositionDeleteReaderWriter { @@ -724,30 +837,6 @@ public PositionDeleteWriter writer( } } - private ForeachFunction rewritePositionDelete( - Broadcast
tableArg, - String sourcePrefixArg, - String targetPrefixArg, - String stagingLocationArg, - PositionDeleteReaderWriter posDeleteReaderWriter) { - return deleteFile -> { - FileIO io = tableArg.getValue().io(); - String newPath = - RewriteTablePathUtil.stagingPath( - deleteFile.location(), sourcePrefixArg, stagingLocationArg); - OutputFile outputFile = io.newOutputFile(newPath); - PartitionSpec spec = tableArg.getValue().specs().get(deleteFile.specId()); - RewriteTablePathUtil.rewritePositionDeleteFile( - deleteFile, - outputFile, - io, - spec, - sourcePrefixArg, - targetPrefixArg, - posDeleteReaderWriter); - }; - } - private static CloseableIterable positionDeletesReader( InputFile inputFile, FileFormat format, PartitionSpec spec) { return FormatModelRegistry.readBuilder(format, Record.class, inputFile) diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java index c5db04762f21..9660beae187e 100644 --- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java +++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java @@ -30,6 +30,7 @@ import java.nio.file.Path; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.function.Predicate; @@ -41,15 +42,21 @@ import org.apache.iceberg.BaseTable; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileFormat; import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.ManifestReader; import org.apache.iceberg.Parameter; import org.apache.iceberg.ParameterizedTestExtension; import org.apache.iceberg.Parameters; import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.RowDelta; import org.apache.iceberg.Schema; import org.apache.iceberg.Snapshot; import org.apache.iceberg.SnapshotChanges; import org.apache.iceberg.StaticTableOperations; +import org.apache.iceberg.StructLike; import org.apache.iceberg.Table; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.TableProperties; @@ -62,14 +69,18 @@ import org.apache.iceberg.data.FileHelpers; import org.apache.iceberg.data.GenericRecord; import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.BaseDVFileWriter; +import org.apache.iceberg.deletes.DVFileWriter; import org.apache.iceberg.deletes.PositionDelete; import org.apache.iceberg.hadoop.HadoopTables; import org.apache.iceberg.io.FileIO; import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.io.OutputFileFactory; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; 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.spark.SparkCatalog; import org.apache.iceberg.spark.TestBase; import org.apache.iceberg.spark.source.ThreeColumnRecord; @@ -629,12 +640,7 @@ public void testPositionDeletesDeduplication() throws Exception { // in a new manifest, which will cause duplicate DeleteFile objects when processing tableWithPosDeletes.newRowDelta().addDeletes(positionDeletes).commit(); - // This should NOT throw AlreadyExistsException - the fix uses DeleteFileSet to deduplicate - // Without the fix (using Collectors.toSet()), this would fail because: - // 1. Both manifests contain entries for the same delete file - // 2. Processing returns two different DeleteFile objects for the same file - // 3. HashSet doesn't deduplicate them (DeleteFile doesn't override equals()) - // 4. rewritePositionDeletes tries to write the same file twice -> AlreadyExistsException + // This should NOT throw AlreadyExistsException RewriteTablePath.Result result = actions() .rewriteTablePath(tableWithPosDeletes) @@ -642,13 +648,295 @@ public void testPositionDeletesDeduplication() throws Exception { .rewriteLocationPrefix(tableWithPosDeletes.location(), targetTableLocation()) .execute(); - // Verify the rewrite completed successfully - should have rewritten exactly 1 delete file - // (the duplicate should be deduplicated by DeleteFileSet) assertThat(result.rewrittenDeleteFilePathsCount()) .as("Should have rewritten exactly 1 delete file after deduplication") .isEqualTo(1); } + // Regression test: when the same position delete file is referenced from manifests in different + // snapshots, it must be rewritten once and the resulting size stamped consistently into every + // manifest that references it. The delete file is enumerated and deduped by path before the + // rewrite, so its measured size is shared across all referencing delete manifests. + @TestTemplate + public void testSharedDeleteFileSizeAcrossManifests() throws Exception { + assumeThat(formatVersion) + .as("Format versions 3+ use DVs with different validation rules") + .isEqualTo(2); + + Table tableWithPosDeletes = + createTableWithSnapshots( + tableDir.toFile().toURI().toString().concat("tableWithSharedDelete"), + 1, + Map.of(TableProperties.DELETE_DEFAULT_FILE_FORMAT, "parquet")); + + DataFile dataFile = + tableWithPosDeletes + .currentSnapshot() + .addedDataFiles(tableWithPosDeletes.io()) + .iterator() + .next(); + + List> deletes = Lists.newArrayList(Pair.of(dataFile.location(), 0L)); + File deleteFile = + new File( + removePrefix(tableWithPosDeletes.location() + "/data/deeply/nested/deletes.parquet")); + DeleteFile positionDeletes = + FileHelpers.writeDeleteFile( + tableWithPosDeletes, + tableWithPosDeletes.io().newOutputFile(deleteFile.toURI().toString()), + deletes, + formatVersion) + .first(); + + tableWithPosDeletes.newRowDelta().addDeletes(positionDeletes).commit(); + tableWithPosDeletes.newRowDelta().addDeletes(positionDeletes).commit(); + + RewriteTablePath.Result result = + actions() + .rewriteTablePath(tableWithPosDeletes) + .stagingLocation(stagingLocation()) + .rewriteLocationPrefix(tableWithPosDeletes.location(), targetTableLocation()) + .execute(); + copyTableFiles(result); + + Table targetTable = TABLES.load(targetTableLocation()); + List deleteManifests = + targetTable.currentSnapshot().deleteManifests(targetTable.io()); + assertThat(deleteManifests) + .as("Expected the shared delete file to be referenced by multiple manifests") + .hasSizeGreaterThanOrEqualTo(2); + for (ManifestFile manifest : deleteManifests) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, targetTable.io(), targetTable.specs())) { + for (DeleteFile df : reader) { + long manifestSize = df.fileSizeInBytes(); + long actualSize = targetTable.io().newInputFile(df.location()).getLength(); + assertThat(manifestSize) + .as( + "file_size_in_bytes in rewritten manifest should match actual file size for %s", + df.location()) + .isEqualTo(actualSize); + } + } + } + } + + // Regression test: a single delete manifest can reference multiple distinct position delete + // files, and each entry must be stamped with its own rewritten size. The two delete files carry + // a different number of records so they rewrite to different sizes, which catches a per-path size + // map that collapses entries to a single size or falls back to the stale original size. + @TestTemplate + public void testMultipleDistinctDeleteFileSizesAfterRewrite() throws Exception { + assumeThat(formatVersion) + .as("Format versions 3+ use DVs with different validation rules") + .isEqualTo(2); + + Table tableWithPosDeletes = + createTableWithSnapshots( + tableDir.toFile().toURI().toString().concat("tableWithDistinctDeletes"), + 2, + Map.of(TableProperties.DELETE_DEFAULT_FILE_FORMAT, "parquet")); + + List dataFiles = Lists.newArrayList(); + tableWithPosDeletes + .snapshots() + .forEach( + snapshot -> snapshot.addedDataFiles(tableWithPosDeletes.io()).forEach(dataFiles::add)); + assertThat(dataFiles).as("Expected two data files to reference from deletes").hasSize(2); + + // One delete file holds a single record, the other holds two, so they rewrite to distinct + // on-disk sizes. + File smallDeleteFile = + new File( + removePrefix( + tableWithPosDeletes.location() + "/data/deeply/nested/deletes-small.parquet")); + DeleteFile smallDelete = + FileHelpers.writeDeleteFile( + tableWithPosDeletes, + tableWithPosDeletes.io().newOutputFile(smallDeleteFile.toURI().toString()), + Lists.newArrayList(Pair.of(dataFiles.get(0).location(), 0L)), + formatVersion) + .first(); + + File largeDeleteFile = + new File( + removePrefix( + tableWithPosDeletes.location() + "/data/deeply/nested/deletes-large.parquet")); + DeleteFile largeDelete = + FileHelpers.writeDeleteFile( + tableWithPosDeletes, + tableWithPosDeletes.io().newOutputFile(largeDeleteFile.toURI().toString()), + Lists.newArrayList( + Pair.of(dataFiles.get(0).location(), 0L), + Pair.of(dataFiles.get(1).location(), 0L)), + formatVersion) + .first(); + + tableWithPosDeletes.newRowDelta().addDeletes(smallDelete).addDeletes(largeDelete).commit(); + + RewriteTablePath.Result result = + actions() + .rewriteTablePath(tableWithPosDeletes) + .stagingLocation(stagingLocation()) + .rewriteLocationPrefix(tableWithPosDeletes.location(), targetTableLocation()) + .execute(); + copyTableFiles(result); + + Table targetTable = TABLES.load(targetTableLocation()); + List rewrittenSizes = Lists.newArrayList(); + for (ManifestFile manifest : targetTable.currentSnapshot().deleteManifests(targetTable.io())) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, targetTable.io(), targetTable.specs())) { + for (DeleteFile df : reader) { + long manifestSize = df.fileSizeInBytes(); + long actualSize = targetTable.io().newInputFile(df.location()).getLength(); + assertThat(manifestSize) + .as( + "file_size_in_bytes in rewritten manifest should match actual file size for %s", + df.location()) + .isEqualTo(actualSize); + rewrittenSizes.add(manifestSize); + } + } + } + + assertThat(rewrittenSizes) + .as( + "The two distinct delete files should rewrite to distinct, independently recorded sizes") + .hasSize(2) + .doesNotHaveDuplicates(); + } + + // Regression test: rewriting delete file paths changes the file size (since the + // embedded data file paths may differ in length), but file_size_in_bytes in the rewritten + // manifest was not updated. Readers that use file_size_in_bytes to elide a stat() call may + // fail. + @TestTemplate + public void testDeleteFileSizeInBytesAfterRewrite() throws Exception { + List> deletes = + Lists.newArrayList( + Pair.of( + SnapshotChanges.builderFor(table) + .build() + .addedDataFiles() + .iterator() + .next() + .location(), + 0L)); + + File file = new File(removePrefix(table.location() + "/data/deeply/nested/deletes.parquet")); + DeleteFile positionDeletes = + FileHelpers.writeDeleteFile( + table, table.io().newOutputFile(file.toURI().toString()), deletes, formatVersion) + .first(); + table.newRowDelta().addDeletes(positionDeletes).commit(); + + RewriteTablePath.Result result = + actions() + .rewriteTablePath(table) + .stagingLocation(stagingLocation()) + .rewriteLocationPrefix(table.location(), targetTableLocation()) + .execute(); + copyTableFiles(result); + + Table targetTable = TABLES.load(targetTableLocation()); + for (ManifestFile manifest : targetTable.currentSnapshot().deleteManifests(targetTable.io())) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, targetTable.io(), targetTable.specs())) { + for (DeleteFile df : reader) { + long manifestSize = df.fileSizeInBytes(); + long actualSize = targetTable.io().newInputFile(df.location()).getLength(); + assertThat(manifestSize) + .as( + "file_size_in_bytes in rewritten manifest should match actual file size for %s", + df.location()) + .isEqualTo(actualSize); + } + } + } + } + + // Regression test: a single Puffin file can hold multiple DVs (one blob per data file) + // referenced by distinct DeleteFile entries that share the same location. The rewrite must + // rewrite + // the physical Puffin file once (rather than colliding on the staging path) and stamp the + // rewritten size into every referencing manifest entry. + @TestTemplate + public void testSharedPuffinDeleteFileSizeAfterRewrite() throws Exception { + assumeThat(formatVersion) + .as("DVs are introduced in v3; v4 writes parquet manifests the test setup cannot read") + .isEqualTo(3); + + Table tableWithDVs = + createTableWithSnapshots( + tableDir.toFile().toURI().toString().concat("tableWithSharedPuffin"), 2); + + List dataFilePaths = Lists.newArrayList(); + tableWithDVs + .snapshots() + .forEach( + snapshot -> + snapshot + .addedDataFiles(tableWithDVs.io()) + .forEach(dataFile -> dataFilePaths.add(dataFile.location()))); + assertThat(dataFilePaths).as("Expected two data files to back two DVs").hasSize(2); + + List dvs = writeDVsForDataFiles(tableWithDVs, dataFilePaths); + assertThat(dvs) + .as("Both DVs should live in a single Puffin file") + .hasSize(2) + .allSatisfy(dv -> assertThat(dv.location()).isEqualTo(dvs.get(0).location())); + + RowDelta rowDelta = tableWithDVs.newRowDelta(); + dvs.forEach(rowDelta::addDeletes); + rowDelta.commit(); + + RewriteTablePath.Result result = + actions() + .rewriteTablePath(tableWithDVs) + .stagingLocation(stagingLocation()) + .rewriteLocationPrefix(tableWithDVs.location(), targetTableLocation()) + .execute(); + assertThat(result.rewrittenDeleteFilePathsCount()) + .as("Two DVs are two delete files with rewritten paths") + .isEqualTo(2); + copyTableFiles(result); + + Table targetTable = TABLES.load(targetTableLocation()); + Set rewrittenLocations = Sets.newHashSet(); + for (ManifestFile manifest : targetTable.currentSnapshot().deleteManifests(targetTable.io())) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, targetTable.io(), targetTable.specs())) { + for (DeleteFile df : reader) { + rewrittenLocations.add(df.location()); + long actualSize = targetTable.io().newInputFile(df.location()).getLength(); + assertThat(df.fileSizeInBytes()) + .as("file_size_in_bytes should match the rewritten Puffin size for %s", df.location()) + .isEqualTo(actualSize); + } + } + } + assertThat(rewrittenLocations) + .as("Both DVs should point at the single rewritten Puffin file") + .hasSize(1); + } + + // Writes one DV per data file path into a single Puffin file, returning the resulting DeleteFiles + // (which share a location but carry distinct blob offsets). + private List writeDVsForDataFiles(Table targetTable, List dataFilePaths) + throws IOException { + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(targetTable, 1, 1).format(FileFormat.PUFFIN).build(); + DVFileWriter writer = new BaseDVFileWriter(fileFactory, p -> null); + try (DVFileWriter closeableWriter = writer) { + for (String path : dataFilePaths) { + closeableWriter.delete(path, 0L, targetTable.spec(), (StructLike) null); + } + } + + return writer.result().deleteFiles(); + } + @TestTemplate public void testEqualityDeletes() throws Exception { Table sourceTable = createTableWithSnapshots(newTableLocation(), 1); diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java index aedb25e4a4a6..11935e815e76 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java @@ -22,6 +22,7 @@ import java.io.IOException; import java.io.OutputStreamWriter; import java.nio.charset.StandardCharsets; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Set; @@ -31,9 +32,13 @@ import org.apache.iceberg.ContentFile; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; import org.apache.iceberg.FileFormat; import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.ManifestContent; import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.ManifestReader; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.PartitionStatisticsFile; import org.apache.iceberg.RewriteTablePathUtil; @@ -69,13 +74,14 @@ import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; 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; import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.spark.JobGroupInfo; import org.apache.iceberg.spark.source.SerializableTableWithSize; import org.apache.iceberg.util.DeleteFileSet; import org.apache.iceberg.util.Pair; import org.apache.iceberg.util.Tasks; -import org.apache.spark.api.java.function.ForeachFunction; +import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.MapFunction; import org.apache.spark.api.java.function.ReduceFunction; import org.apache.spark.broadcast.Broadcast; @@ -87,6 +93,7 @@ import org.apache.spark.sql.functions; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import scala.Tuple2; public class RewriteTablePathSparkAction extends BaseSparkAction implements RewriteTablePath { @@ -272,7 +279,8 @@ private String jobDesc() { *
    *
  • Rebuild version files to staging *
  • Rebuild manifest list files to staging - *
  • Rebuild manifest to staging + *
  • Rewrite referenced position delete files to staging + *
  • Rebuild manifests to staging *
  • Get all files needed to move *
*/ @@ -308,24 +316,29 @@ private Result rebuildMetadata() { RewriteResult rewriteManifestListResult = new RewriteResult<>(); manifestListResults.forEach(rewriteManifestListResult::append); - // rebuild manifest files - Set metaFiles = rewriteManifestListResult.toRewrite(); - RewriteContentFileResult rewriteManifestResult = - rewriteManifests(deltaSnapshots, endMetadata, metaFiles); + Set manifestFiles = rewriteManifestListResult.toRewrite(); // rebuild position delete files - Set deleteFiles = - rewriteManifestResult.toRewrite().stream() - .filter(e -> e instanceof DeleteFile) - .map(e -> (DeleteFile) e) - .collect(Collectors.toCollection(DeleteFileSet::create)); - rewritePositionDeletes(deleteFiles); + Set deleteManifests = + manifestFiles.stream() + .filter(manifest -> manifest.content() == ManifestContent.DELETES) + .collect(Collectors.toSet()); + Set deleteFilesToRewrite = positionDeletesToRewrite(deleteManifests); + Map rewrittenDeleteFileSizes = rewritePositionDeletes(deleteFilesToRewrite); + + // rebuild manifest files + RewriteContentFileResult rewriteManifestResult = + rewriteManifests( + deltaSnapshots, + endMetadata, + manifestFiles, + sparkContext().broadcast(rewrittenDeleteFileSizes)); ImmutableRewriteTablePath.Result.Builder builder = ImmutableRewriteTablePath.Result.builder() .stagingLocation(stagingDir) - .rewrittenDeleteFilePathsCount(deleteFiles.size()) - .rewrittenManifestFilePathsCount(metaFiles.size()) + .rewrittenDeleteFilePathsCount(deleteFilesToRewrite.size()) + .rewrittenManifestFilePathsCount(manifestFiles.size()) .latestVersion(RewriteTablePathUtil.fileName(endVersionName)); if (!createFileList) { @@ -561,7 +574,10 @@ public RewriteContentFileResult appendDeleteFile(RewriteResult r1) { /** Rewrite manifest files in a distributed manner and return rewritten data files path pairs. */ private RewriteContentFileResult rewriteManifests( - Set deltaSnapshots, TableMetadata tableMetadata, Set toRewrite) { + Set deltaSnapshots, + TableMetadata tableMetadata, + Set toRewrite, + Broadcast> rewrittenDeleteFileSizes) { if (toRewrite.isEmpty()) { return new RewriteContentFileResult(); } @@ -581,7 +597,8 @@ private RewriteContentFileResult rewriteManifests( stagingDir, tableMetadata.formatVersion(), sourcePrefix, - targetPrefix), + targetPrefix, + rewrittenDeleteFileSizes), Encoders.bean(RewriteContentFileResult.class)) // duplicates are expected here as the same data file can have different statuses // (e.g. added and deleted) @@ -594,7 +611,8 @@ private static MapFunction toManifests( String stagingLocation, int format, String sourcePrefix, - String targetPrefix) { + String targetPrefix, + Broadcast> rewrittenDeleteFileSizes) { return manifestFile -> { RewriteContentFileResult result = new RewriteContentFileResult(); @@ -619,7 +637,8 @@ private static MapFunction toManifests( stagingLocation, format, sourcePrefix, - targetPrefix)); + targetPrefix, + rewrittenDeleteFileSizes)); break; default: throw new UnsupportedOperationException( @@ -665,7 +684,8 @@ private static RewriteResult writeDeleteManifest( String stagingLocation, int format, String sourcePrefix, - String targetPrefix) { + String targetPrefix, + Broadcast> rewrittenDeleteFileSizes) { try { String stagingPath = RewriteTablePathUtil.stagingPath(manifestFile.path(), sourcePrefix, stagingLocation); @@ -682,27 +702,120 @@ private static RewriteResult writeDeleteManifest( specsById, sourcePrefix, targetPrefix, - stagingLocation); + stagingLocation, + rewrittenDeleteFileSizes.value()); } catch (IOException e) { throw new RuntimeIOException(e); } } - private void rewritePositionDeletes(Set toRewrite) { + /** + * Enumerate the distinct position delete files referenced by the given delete manifests. Deduped + * by identity (location, offset, size) so a file shared across manifests is counted once; the + * physical rewrite is further deduped by location in {@link #rewritePositionDeletes}. + */ + private Set positionDeletesToRewrite(Set deleteManifests) { + if (deleteManifests.isEmpty()) { + return Collections.emptySet(); + } + + Encoder manifestFileEncoder = Encoders.javaSerialization(ManifestFile.class); + Dataset manifestDS = + spark().createDataset(Lists.newArrayList(deleteManifests), manifestFileEncoder); + Encoder deleteFileEncoder = Encoders.javaSerialization(DeleteFile.class); + + List referencedDeleteFiles = + manifestDS + .repartition(deleteManifests.size()) + .flatMap(positionDeletesInManifest(tableBroadcast()), deleteFileEncoder) + .collectAsList(); + + return DeleteFileSet.of(referencedDeleteFiles); + } + + private static FlatMapFunction positionDeletesInManifest( + Broadcast
tableArg) { + return manifestFile -> { + Table table = tableArg.getValue(); + List deleteFiles = Lists.newArrayList(); + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifestFile, table.io(), table.specs())) { + for (DeleteFile deleteFile : reader) { + if (deleteFile.content() == FileContent.POSITION_DELETES) { + deleteFiles.add(deleteFile.copy()); + } + } + } + return deleteFiles.iterator(); + }; + } + + /** + * Rewrite the given position delete files in parallel, returning a map from each source delete + * file path to the size of its rewritten file. Physical files are deduped by location, so a + * Puffin file holding multiple DVs is rewritten once and its size keyed once. + */ + private Map rewritePositionDeletes(Set toRewrite) { if (toRewrite.isEmpty()) { - return; + return Collections.emptyMap(); + } + + // Multiple DVs can share one Puffin file at different blob offsets; rewrite each physical file + // once. The measured size is keyed by location and applied to every referencing manifest entry. + Map byLocation = Maps.newHashMapWithExpectedSize(toRewrite.size()); + for (DeleteFile deleteFile : toRewrite) { + byLocation.putIfAbsent(deleteFile.location(), deleteFile); } + List physicalFiles = Lists.newArrayList(byLocation.values()); Encoder deleteFileEncoder = Encoders.javaSerialization(DeleteFile.class); - Dataset deleteFileDs = - spark().createDataset(Lists.newArrayList(toRewrite), deleteFileEncoder); + Dataset deleteFileDS = spark().createDataset(physicalFiles, deleteFileEncoder); PositionDeleteReaderWriter posDeleteReaderWriter = new SparkPositionDeleteReaderWriter(); - deleteFileDs - .repartition(toRewrite.size()) - .foreach( - rewritePositionDelete( - tableBroadcast(), sourcePrefix, targetPrefix, stagingDir, posDeleteReaderWriter)); + List> rewrittenSizes = + deleteFileDS + .repartition(physicalFiles.size()) + .map( + rewritePositionDelete( + tableBroadcast(), + sourcePrefix, + targetPrefix, + stagingDir, + posDeleteReaderWriter), + Encoders.tuple(Encoders.STRING(), Encoders.LONG())) + .collectAsList(); + + Map sizesBySourcePath = Maps.newHashMapWithExpectedSize(rewrittenSizes.size()); + for (Tuple2 entry : rewrittenSizes) { + sizesBySourcePath.put(entry._1(), entry._2()); + } + return sizesBySourcePath; + } + + private static MapFunction> rewritePositionDelete( + Broadcast
tableArg, + String sourcePrefixArg, + String targetPrefixArg, + String stagingLocationArg, + PositionDeleteReaderWriter posDeleteReaderWriter) { + return deleteFile -> { + FileIO io = tableArg.getValue().io(); + String newPath = + RewriteTablePathUtil.stagingPath( + deleteFile.location(), sourcePrefixArg, stagingLocationArg); + OutputFile outputFile = io.newOutputFile(newPath); + PartitionSpec spec = tableArg.getValue().specs().get(deleteFile.specId()); + long rewrittenLength = + RewriteTablePathUtil.rewritePositionDelete( + deleteFile, + outputFile, + io, + spec, + sourcePrefixArg, + targetPrefixArg, + posDeleteReaderWriter); + return new Tuple2<>(deleteFile.location(), rewrittenLength); + }; } private static class SparkPositionDeleteReaderWriter implements PositionDeleteReaderWriter { @@ -724,30 +837,6 @@ public PositionDeleteWriter writer( } } - private ForeachFunction rewritePositionDelete( - Broadcast
tableArg, - String sourcePrefixArg, - String targetPrefixArg, - String stagingLocationArg, - PositionDeleteReaderWriter posDeleteReaderWriter) { - return deleteFile -> { - FileIO io = tableArg.getValue().io(); - String newPath = - RewriteTablePathUtil.stagingPath( - deleteFile.location(), sourcePrefixArg, stagingLocationArg); - OutputFile outputFile = io.newOutputFile(newPath); - PartitionSpec spec = tableArg.getValue().specs().get(deleteFile.specId()); - RewriteTablePathUtil.rewritePositionDeleteFile( - deleteFile, - outputFile, - io, - spec, - sourcePrefixArg, - targetPrefixArg, - posDeleteReaderWriter); - }; - } - private static CloseableIterable positionDeletesReader( InputFile inputFile, FileFormat format, PartitionSpec spec) { return FormatModelRegistry.readBuilder(format, Record.class, inputFile) diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java index c5db04762f21..9660beae187e 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java @@ -30,6 +30,7 @@ import java.nio.file.Path; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.function.Predicate; @@ -41,15 +42,21 @@ import org.apache.iceberg.BaseTable; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileFormat; import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.ManifestReader; import org.apache.iceberg.Parameter; import org.apache.iceberg.ParameterizedTestExtension; import org.apache.iceberg.Parameters; import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.RowDelta; import org.apache.iceberg.Schema; import org.apache.iceberg.Snapshot; import org.apache.iceberg.SnapshotChanges; import org.apache.iceberg.StaticTableOperations; +import org.apache.iceberg.StructLike; import org.apache.iceberg.Table; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.TableProperties; @@ -62,14 +69,18 @@ import org.apache.iceberg.data.FileHelpers; import org.apache.iceberg.data.GenericRecord; import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.BaseDVFileWriter; +import org.apache.iceberg.deletes.DVFileWriter; import org.apache.iceberg.deletes.PositionDelete; import org.apache.iceberg.hadoop.HadoopTables; import org.apache.iceberg.io.FileIO; import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.io.OutputFileFactory; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; 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.spark.SparkCatalog; import org.apache.iceberg.spark.TestBase; import org.apache.iceberg.spark.source.ThreeColumnRecord; @@ -629,12 +640,7 @@ public void testPositionDeletesDeduplication() throws Exception { // in a new manifest, which will cause duplicate DeleteFile objects when processing tableWithPosDeletes.newRowDelta().addDeletes(positionDeletes).commit(); - // This should NOT throw AlreadyExistsException - the fix uses DeleteFileSet to deduplicate - // Without the fix (using Collectors.toSet()), this would fail because: - // 1. Both manifests contain entries for the same delete file - // 2. Processing returns two different DeleteFile objects for the same file - // 3. HashSet doesn't deduplicate them (DeleteFile doesn't override equals()) - // 4. rewritePositionDeletes tries to write the same file twice -> AlreadyExistsException + // This should NOT throw AlreadyExistsException RewriteTablePath.Result result = actions() .rewriteTablePath(tableWithPosDeletes) @@ -642,13 +648,295 @@ public void testPositionDeletesDeduplication() throws Exception { .rewriteLocationPrefix(tableWithPosDeletes.location(), targetTableLocation()) .execute(); - // Verify the rewrite completed successfully - should have rewritten exactly 1 delete file - // (the duplicate should be deduplicated by DeleteFileSet) assertThat(result.rewrittenDeleteFilePathsCount()) .as("Should have rewritten exactly 1 delete file after deduplication") .isEqualTo(1); } + // Regression test: when the same position delete file is referenced from manifests in different + // snapshots, it must be rewritten once and the resulting size stamped consistently into every + // manifest that references it. The delete file is enumerated and deduped by path before the + // rewrite, so its measured size is shared across all referencing delete manifests. + @TestTemplate + public void testSharedDeleteFileSizeAcrossManifests() throws Exception { + assumeThat(formatVersion) + .as("Format versions 3+ use DVs with different validation rules") + .isEqualTo(2); + + Table tableWithPosDeletes = + createTableWithSnapshots( + tableDir.toFile().toURI().toString().concat("tableWithSharedDelete"), + 1, + Map.of(TableProperties.DELETE_DEFAULT_FILE_FORMAT, "parquet")); + + DataFile dataFile = + tableWithPosDeletes + .currentSnapshot() + .addedDataFiles(tableWithPosDeletes.io()) + .iterator() + .next(); + + List> deletes = Lists.newArrayList(Pair.of(dataFile.location(), 0L)); + File deleteFile = + new File( + removePrefix(tableWithPosDeletes.location() + "/data/deeply/nested/deletes.parquet")); + DeleteFile positionDeletes = + FileHelpers.writeDeleteFile( + tableWithPosDeletes, + tableWithPosDeletes.io().newOutputFile(deleteFile.toURI().toString()), + deletes, + formatVersion) + .first(); + + tableWithPosDeletes.newRowDelta().addDeletes(positionDeletes).commit(); + tableWithPosDeletes.newRowDelta().addDeletes(positionDeletes).commit(); + + RewriteTablePath.Result result = + actions() + .rewriteTablePath(tableWithPosDeletes) + .stagingLocation(stagingLocation()) + .rewriteLocationPrefix(tableWithPosDeletes.location(), targetTableLocation()) + .execute(); + copyTableFiles(result); + + Table targetTable = TABLES.load(targetTableLocation()); + List deleteManifests = + targetTable.currentSnapshot().deleteManifests(targetTable.io()); + assertThat(deleteManifests) + .as("Expected the shared delete file to be referenced by multiple manifests") + .hasSizeGreaterThanOrEqualTo(2); + for (ManifestFile manifest : deleteManifests) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, targetTable.io(), targetTable.specs())) { + for (DeleteFile df : reader) { + long manifestSize = df.fileSizeInBytes(); + long actualSize = targetTable.io().newInputFile(df.location()).getLength(); + assertThat(manifestSize) + .as( + "file_size_in_bytes in rewritten manifest should match actual file size for %s", + df.location()) + .isEqualTo(actualSize); + } + } + } + } + + // Regression test: a single delete manifest can reference multiple distinct position delete + // files, and each entry must be stamped with its own rewritten size. The two delete files carry + // a different number of records so they rewrite to different sizes, which catches a per-path size + // map that collapses entries to a single size or falls back to the stale original size. + @TestTemplate + public void testMultipleDistinctDeleteFileSizesAfterRewrite() throws Exception { + assumeThat(formatVersion) + .as("Format versions 3+ use DVs with different validation rules") + .isEqualTo(2); + + Table tableWithPosDeletes = + createTableWithSnapshots( + tableDir.toFile().toURI().toString().concat("tableWithDistinctDeletes"), + 2, + Map.of(TableProperties.DELETE_DEFAULT_FILE_FORMAT, "parquet")); + + List dataFiles = Lists.newArrayList(); + tableWithPosDeletes + .snapshots() + .forEach( + snapshot -> snapshot.addedDataFiles(tableWithPosDeletes.io()).forEach(dataFiles::add)); + assertThat(dataFiles).as("Expected two data files to reference from deletes").hasSize(2); + + // One delete file holds a single record, the other holds two, so they rewrite to distinct + // on-disk sizes. + File smallDeleteFile = + new File( + removePrefix( + tableWithPosDeletes.location() + "/data/deeply/nested/deletes-small.parquet")); + DeleteFile smallDelete = + FileHelpers.writeDeleteFile( + tableWithPosDeletes, + tableWithPosDeletes.io().newOutputFile(smallDeleteFile.toURI().toString()), + Lists.newArrayList(Pair.of(dataFiles.get(0).location(), 0L)), + formatVersion) + .first(); + + File largeDeleteFile = + new File( + removePrefix( + tableWithPosDeletes.location() + "/data/deeply/nested/deletes-large.parquet")); + DeleteFile largeDelete = + FileHelpers.writeDeleteFile( + tableWithPosDeletes, + tableWithPosDeletes.io().newOutputFile(largeDeleteFile.toURI().toString()), + Lists.newArrayList( + Pair.of(dataFiles.get(0).location(), 0L), + Pair.of(dataFiles.get(1).location(), 0L)), + formatVersion) + .first(); + + tableWithPosDeletes.newRowDelta().addDeletes(smallDelete).addDeletes(largeDelete).commit(); + + RewriteTablePath.Result result = + actions() + .rewriteTablePath(tableWithPosDeletes) + .stagingLocation(stagingLocation()) + .rewriteLocationPrefix(tableWithPosDeletes.location(), targetTableLocation()) + .execute(); + copyTableFiles(result); + + Table targetTable = TABLES.load(targetTableLocation()); + List rewrittenSizes = Lists.newArrayList(); + for (ManifestFile manifest : targetTable.currentSnapshot().deleteManifests(targetTable.io())) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, targetTable.io(), targetTable.specs())) { + for (DeleteFile df : reader) { + long manifestSize = df.fileSizeInBytes(); + long actualSize = targetTable.io().newInputFile(df.location()).getLength(); + assertThat(manifestSize) + .as( + "file_size_in_bytes in rewritten manifest should match actual file size for %s", + df.location()) + .isEqualTo(actualSize); + rewrittenSizes.add(manifestSize); + } + } + } + + assertThat(rewrittenSizes) + .as( + "The two distinct delete files should rewrite to distinct, independently recorded sizes") + .hasSize(2) + .doesNotHaveDuplicates(); + } + + // Regression test: rewriting delete file paths changes the file size (since the + // embedded data file paths may differ in length), but file_size_in_bytes in the rewritten + // manifest was not updated. Readers that use file_size_in_bytes to elide a stat() call may + // fail. + @TestTemplate + public void testDeleteFileSizeInBytesAfterRewrite() throws Exception { + List> deletes = + Lists.newArrayList( + Pair.of( + SnapshotChanges.builderFor(table) + .build() + .addedDataFiles() + .iterator() + .next() + .location(), + 0L)); + + File file = new File(removePrefix(table.location() + "/data/deeply/nested/deletes.parquet")); + DeleteFile positionDeletes = + FileHelpers.writeDeleteFile( + table, table.io().newOutputFile(file.toURI().toString()), deletes, formatVersion) + .first(); + table.newRowDelta().addDeletes(positionDeletes).commit(); + + RewriteTablePath.Result result = + actions() + .rewriteTablePath(table) + .stagingLocation(stagingLocation()) + .rewriteLocationPrefix(table.location(), targetTableLocation()) + .execute(); + copyTableFiles(result); + + Table targetTable = TABLES.load(targetTableLocation()); + for (ManifestFile manifest : targetTable.currentSnapshot().deleteManifests(targetTable.io())) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, targetTable.io(), targetTable.specs())) { + for (DeleteFile df : reader) { + long manifestSize = df.fileSizeInBytes(); + long actualSize = targetTable.io().newInputFile(df.location()).getLength(); + assertThat(manifestSize) + .as( + "file_size_in_bytes in rewritten manifest should match actual file size for %s", + df.location()) + .isEqualTo(actualSize); + } + } + } + } + + // Regression test: a single Puffin file can hold multiple DVs (one blob per data file) + // referenced by distinct DeleteFile entries that share the same location. The rewrite must + // rewrite + // the physical Puffin file once (rather than colliding on the staging path) and stamp the + // rewritten size into every referencing manifest entry. + @TestTemplate + public void testSharedPuffinDeleteFileSizeAfterRewrite() throws Exception { + assumeThat(formatVersion) + .as("DVs are introduced in v3; v4 writes parquet manifests the test setup cannot read") + .isEqualTo(3); + + Table tableWithDVs = + createTableWithSnapshots( + tableDir.toFile().toURI().toString().concat("tableWithSharedPuffin"), 2); + + List dataFilePaths = Lists.newArrayList(); + tableWithDVs + .snapshots() + .forEach( + snapshot -> + snapshot + .addedDataFiles(tableWithDVs.io()) + .forEach(dataFile -> dataFilePaths.add(dataFile.location()))); + assertThat(dataFilePaths).as("Expected two data files to back two DVs").hasSize(2); + + List dvs = writeDVsForDataFiles(tableWithDVs, dataFilePaths); + assertThat(dvs) + .as("Both DVs should live in a single Puffin file") + .hasSize(2) + .allSatisfy(dv -> assertThat(dv.location()).isEqualTo(dvs.get(0).location())); + + RowDelta rowDelta = tableWithDVs.newRowDelta(); + dvs.forEach(rowDelta::addDeletes); + rowDelta.commit(); + + RewriteTablePath.Result result = + actions() + .rewriteTablePath(tableWithDVs) + .stagingLocation(stagingLocation()) + .rewriteLocationPrefix(tableWithDVs.location(), targetTableLocation()) + .execute(); + assertThat(result.rewrittenDeleteFilePathsCount()) + .as("Two DVs are two delete files with rewritten paths") + .isEqualTo(2); + copyTableFiles(result); + + Table targetTable = TABLES.load(targetTableLocation()); + Set rewrittenLocations = Sets.newHashSet(); + for (ManifestFile manifest : targetTable.currentSnapshot().deleteManifests(targetTable.io())) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, targetTable.io(), targetTable.specs())) { + for (DeleteFile df : reader) { + rewrittenLocations.add(df.location()); + long actualSize = targetTable.io().newInputFile(df.location()).getLength(); + assertThat(df.fileSizeInBytes()) + .as("file_size_in_bytes should match the rewritten Puffin size for %s", df.location()) + .isEqualTo(actualSize); + } + } + } + assertThat(rewrittenLocations) + .as("Both DVs should point at the single rewritten Puffin file") + .hasSize(1); + } + + // Writes one DV per data file path into a single Puffin file, returning the resulting DeleteFiles + // (which share a location but carry distinct blob offsets). + private List writeDVsForDataFiles(Table targetTable, List dataFilePaths) + throws IOException { + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(targetTable, 1, 1).format(FileFormat.PUFFIN).build(); + DVFileWriter writer = new BaseDVFileWriter(fileFactory, p -> null); + try (DVFileWriter closeableWriter = writer) { + for (String path : dataFilePaths) { + closeableWriter.delete(path, 0L, targetTable.spec(), (StructLike) null); + } + } + + return writer.result().deleteFiles(); + } + @TestTemplate public void testEqualityDeletes() throws Exception { Table sourceTable = createTableWithSnapshots(newTableLocation(), 1);