Core: update manifest delete file size after rewrite table action

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<DataFile> 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<DeleteFile> rewriteDeleteManifest(
       ManifestFile manifestFile,
       Set<Long> snapshotIds,
@@ -389,6 +395,52 @@ public static RewriteResult<DeleteFile> 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.
+   *
+   * <p>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<DeleteFile> rewriteDeleteManifest(
+      ManifestFile manifestFile,
+      Set<Long> snapshotIds,
+      OutputFile outputFile,
+      FileIO io,
+      int format,
+      Map<Integer, PartitionSpec> specsById,
+      String sourcePrefix,
+      String targetPrefix,
+      String stagingLocation,
+      Map<String, Long> rewrittenDeleteFileSizes)
+      throws IOException {
     PartitionSpec spec = specsById.get(manifestFile.partitionSpecId());
     try (ManifestWriter<DeleteFile> writer =
             ManifestFiles.writeDeleteManifest(format, spec, outputFile, manifestFile.snapshotId());
@@ -405,7 +457,8 @@ public static RewriteResult<DeleteFile> rewriteDeleteManifest(
                       sourcePrefix,
                       targetPrefix,
                       stagingLocation,
-                      writer))
+                      writer,
+                      rewrittenDeleteFileSizes))
           .reduce(new RewriteResult<>(), RewriteResult::append);
     }
   }
@@ -445,14 +498,20 @@ private static RewriteResult<DeleteFile> writeDeleteFileEntry(
       String sourcePrefix,
       String targetPrefix,
       String stagingLocation,
-      ManifestWriter<DeleteFile> writer) {
+      ManifestWriter<DeleteFile> writer,
+      Map<String, Long> rewrittenDeleteFileSizes) {
 
     DeleteFile file = entry.file();
     RewriteResult<DeleteFile> 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<Record> 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.
+   *
+   * <p>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();
     }
   }
 

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<DeleteFile> reader =
+        ManifestFiles.readDeleteManifest(rewritten, table.io(), table.specs())) {
+      for (ManifestEntry<DeleteFile> 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<DeleteFile> writer =
+        ManifestFiles.writeDeleteManifest(formatVersion, SPEC, manifestFile, 1000L);
+    try {
+      writer.add(live);
+      writer.delete(deleted, 1, null);
+    } finally {
+      writer.close();
+    }
+
+    return writer.toManifestFile();
+  }
 }