Fix reading encrypted Deletion Vectors

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

@@ -29,10 +29,12 @@
 import org.apache.iceberg.deletes.BaseDVFileWriter;
 import org.apache.iceberg.deletes.DVFileWriter;
 import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptingFileIO;
+import org.apache.iceberg.encryption.EncryptionUtil;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.IOUtil;
 import org.apache.iceberg.io.InputFile;
-import org.apache.iceberg.io.OutputFile;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -188,8 +190,16 @@ private static List<DeleteFile> writeDVs(
       FileIO fileIO,
       String dvOutputLocation,
       Map<String, Pair<PartitionSpec, StructLike>> partitions) {
-    OutputFile dvOutputFile = fileIO.newOutputFile(dvOutputLocation);
-    try (DVFileWriter dvFileWriter = new BaseDVFileWriter(() -> dvOutputFile, path -> null)) {
+    EncryptedOutputFile dvEncryptedOutputFile;
+    if (fileIO instanceof EncryptingFileIO) {
+      dvEncryptedOutputFile = ((EncryptingFileIO) fileIO).newEncryptingOutputFile(dvOutputLocation);
+    } else {
+      dvEncryptedOutputFile =
+          EncryptionUtil.plainAsEncryptedOutput(fileIO.newOutputFile(dvOutputLocation));
+    }
+
+    try (DVFileWriter dvFileWriter =
+        new BaseDVFileWriter(() -> dvEncryptedOutputFile, path -> null)) {
       for (Map.Entry<String, PositionDeleteIndex> entry : mergedIndexByFile.entrySet()) {
         String referencedLocation = entry.getKey();
         PositionDeleteIndex mergedPositions = entry.getValue();

core/src/main/java/org/apache/iceberg/deletes/BaseDVFileWriter.java

@@ -19,6 +19,7 @@
 package org.apache.iceberg.deletes;
 
 import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
@@ -30,8 +31,10 @@
 import org.apache.iceberg.MetadataColumns;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.encryption.EncryptionUtil;
 import org.apache.iceberg.io.DeleteWriteResult;
-import org.apache.iceberg.io.OutputFile;
 import org.apache.iceberg.io.OutputFileFactory;
 import org.apache.iceberg.puffin.Blob;
 import org.apache.iceberg.puffin.BlobMetadata;
@@ -52,19 +55,19 @@ public class BaseDVFileWriter implements DVFileWriter {
   private static final String REFERENCED_DATA_FILE_KEY = "referenced-data-file";
   private static final String CARDINALITY_KEY = "cardinality";
 
-  private final Supplier<OutputFile> dvOutputFile;
+  private final Supplier<EncryptedOutputFile> dvOutputFile;
   private final Function<String, PositionDeleteIndex> loadPreviousDeletes;
   private final Map<String, Deletes> deletesByPath = Maps.newHashMap();
   private final Map<String, BlobMetadata> blobsByPath = Maps.newHashMap();
   private DeleteWriteResult result = null;
 
   public BaseDVFileWriter(
       OutputFileFactory fileFactory, Function<String, PositionDeleteIndex> loadPreviousDeletes) {
-    this(() -> fileFactory.newOutputFile().encryptingOutputFile(), loadPreviousDeletes);
+    this(fileFactory::newOutputFile, loadPreviousDeletes);
   }
 
   public BaseDVFileWriter(
-      Supplier<OutputFile> dvOutputFile,
+      Supplier<EncryptedOutputFile> dvOutputFile,
       Function<String, PositionDeleteIndex> loadPreviousDeletes) {
     this.dvOutputFile = dvOutputFile;
     this.loadPreviousDeletes = loadPreviousDeletes;
@@ -108,7 +111,11 @@ public void close() throws IOException {
         return;
       }
 
-      PuffinWriter writer = newWriter();
+      EncryptedOutputFile encryptedOut = dvOutputFile.get();
+      PuffinWriter writer =
+          Puffin.write(encryptedOut.encryptingOutputFile())
+              .createdBy(IcebergBuild.fullVersion())
+              .build();
 
       try (PuffinWriter closeableWriter = writer) {
         for (Deletes deletes : deletesByPath.values()) {
@@ -133,23 +140,31 @@ public void close() throws IOException {
       String puffinPath = writer.location();
       long puffinFileSize = writer.fileSize();
 
+      ByteBuffer keyMetadataBuffer = null;
+      EncryptionKeyMetadata keyMetadata = encryptedOut.keyMetadata();
+      if (keyMetadata != null && keyMetadata.buffer() != null) {
+        keyMetadataBuffer = EncryptionUtil.setFileLength(keyMetadata.buffer(), puffinFileSize);
+      }
+
       for (String path : deletesByPath.keySet()) {
-        DeleteFile dv = createDV(puffinPath, puffinFileSize, path);
+        DeleteFile dv = createDV(puffinPath, puffinFileSize, path, keyMetadataBuffer);
         dvs.add(dv);
       }
 
       this.result = new DeleteWriteResult(dvs, referencedDataFiles, rewrittenDeleteFiles);
     }
   }
 
-  private DeleteFile createDV(String path, long size, String referencedDataFile) {
+  private DeleteFile createDV(
+      String path, long size, String referencedDataFile, ByteBuffer keyMetadataBuffer) {
     Deletes deletes = deletesByPath.get(referencedDataFile);
     BlobMetadata blobMetadata = blobsByPath.get(referencedDataFile);
     return FileMetadata.deleteFileBuilder(deletes.spec())
         .ofPositionDeletes()
         .withFormat(FileFormat.PUFFIN)
         .withPath(path)
         .withPartition(deletes.partition())
+        .withEncryptionKeyMetadata(keyMetadataBuffer)
         .withFileSizeInBytes(size)
         .withReferencedDataFile(referencedDataFile)
         .withContentOffset(blobMetadata.offset())
@@ -165,11 +180,6 @@ private void write(PuffinWriter writer, Deletes deletes) {
     blobsByPath.put(path, blobMetadata);
   }
 
-  private PuffinWriter newWriter() {
-    OutputFile outputFile = dvOutputFile.get();
-    return Puffin.write(outputFile).createdBy(IcebergBuild.fullVersion()).build();
-  }
-
   private Blob toBlob(PositionDeleteIndex positions, String path) {
     return new Blob(
         StandardBlobTypes.DV_V1,

spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestTableEncryption.java

@@ -212,16 +212,27 @@ public void testInsertAndDelete() {
         expected,
         sql("SELECT * FROM %s ORDER BY id", tableName));
 
-    sql("DELETE FROM %s WHERE id < 4", tableName);
+    sql("DELETE FROM %s WHERE id < 3", tableName);
 
-    expected = ImmutableList.of(row(4L, "d", 4.0F), row(5L, "e", 5.0F), row(6L, "f", Float.NaN));
+    expected =
+        ImmutableList.of(
+            row(3L, "c", Float.NaN),
+            row(4L, "d", 4.0F),
+            row(5L, "e", 5.0F),
+            row(6L, "f", Float.NaN));
 
     assertEquals(
         "Should return all expected rows",
         expected,
         sql("SELECT * FROM %s ORDER BY id", tableName));
   }
 
+  @TestTemplate
+  public void testInsertAndDeleteMOR() {
+    sql("ALTER TABLE %s SET TBLPROPERTIES ('write.delete.mode'='merge-on-read')", tableName);
+    testInsertAndDelete();
+  }
+
   @TestTemplate
   public void testMetadataTamperproofing() throws IOException {
     ChecksumFileSystem fs = ((ChecksumFileSystem) FileSystem.newInstance(new Configuration()));

spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestTableEncryption.java

@@ -212,16 +212,27 @@ public void testInsertAndDelete() {
         expected,
         sql("SELECT * FROM %s ORDER BY id", tableName));
 
-    sql("DELETE FROM %s WHERE id < 4", tableName);
+    sql("DELETE FROM %s WHERE id < 3", tableName);
 
-    expected = ImmutableList.of(row(4L, "d", 4.0F), row(5L, "e", 5.0F), row(6L, "f", Float.NaN));
+    expected =
+        ImmutableList.of(
+            row(3L, "c", Float.NaN),
+            row(4L, "d", 4.0F),
+            row(5L, "e", 5.0F),
+            row(6L, "f", Float.NaN));
 
     assertEquals(
         "Should return all expected rows",
         expected,
         sql("SELECT * FROM %s ORDER BY id", tableName));
   }
 
+  @TestTemplate
+  public void testInsertAndDeleteMOR() {
+    sql("ALTER TABLE %s SET TBLPROPERTIES ('write.delete.mode'='merge-on-read')", tableName);
+    testInsertAndDelete();
+  }
+
   @TestTemplate
   public void testMetadataTamperproofing() throws IOException {
     ChecksumFileSystem fs = ((ChecksumFileSystem) FileSystem.newInstance(new Configuration()));