Repository: apache/iceberg (language: python)

Issue:
Improve Position Deletes in V3
### Proposed Change

This proposal aims to enhance the handling of position deletes in Iceberg. It builds on lessons learned from deploying the current approach at scale and addresses all unresolved questions from past community discussions and proposals. This effort primarily targets the V3 spec but some parts can be completed independently.

Shortcomings of position deletes we observe today:
- Choosing between fewer delete files on disk (partition granularity) or targeted deletes (file granularity).
- Dependence on external maintenance for consistent write and read performance.
- Writing and reading overhead as in-memory and on-disk representations differ.


### Proposal document

https://docs.google.com/document/d/18Bqhr-vnzFfQk1S4AgRISkA_5_m5m32Nnc2Cw0zn2XM

### Specifications

- [X] Table
- [ ] View
- [ ] REST
- [X] Puffin
- [ ] Encryption
- [ ] Other


Discussion / hints:

Fix patch:
diff --git a/core/src/main/java/org/apache/iceberg/deletes/BaseDVFileWriter.java b/core/src/main/java/org/apache/iceberg/deletes/BaseDVFileWriter.java
new file mode 100644
index 000000000000..ab47d8ad78d6
--- /dev/null
+++ b/core/src/main/java/org/apache/iceberg/deletes/BaseDVFileWriter.java
@@ -0,0 +1,194 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.deletes;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileMetadata;
+import org.apache.iceberg.IcebergBuild;
+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.io.DeleteWriteResult;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.BlobMetadata;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+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.util.CharSequenceSet;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.StructLikeUtil;
+
+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 OutputFileFactory fileFactory;
+  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 = fileFactory;
+    this.loadPreviousDeletes = loadPreviousDeletes;
+  }
+
+  @Override
+  public void delete(String path, long pos, PartitionSpec spec, StructLike partition) {
+    Deletes deletes =
+        deletesByPath.computeIfAbsent(path, key -> new Deletes(path, spec, partition));
+    PositionDeleteIndex positions = deletes.positions();
+    positions.delete(pos);
+  }
+
+  @Override
+  public DeleteWriteResult result() {
+    Preconditions.checkState(result != null, "Cannot get result from unclosed writer");
+    return result;
+  }
+
+  @Override
+  public void close() throws IOException {
+    if (result == null) {
+      List<DeleteFile> dvs = Lists.newArrayList();
+      CharSequenceSet referencedDataFiles = CharSequenceSet.empty();
+      List<DeleteFile> rewrittenDeleteFiles = Lists.newArrayList();
+
+      PuffinWriter writer = newWriter();
+
+      try (PuffinWriter closeableWriter = writer) {
+        for (Deletes deletes : deletesByPath.values()) {
+          String path = deletes.path();
+          PositionDeleteIndex positions = deletes.positions();
+          PositionDeleteIndex previousPositions = loadPreviousDeletes.apply(path);
+          if (previousPositions != null) {
+            positions.merge(previousPositions);
+            for (DeleteFile previousDeleteFile : previousPositions.deleteFiles()) {
+              // only DVs and file-scoped deletes can be discarded from the table state
+              if (ContentFileUtil.isFileScoped(previousDeleteFile)) {
+                rewrittenDeleteFiles.add(previousDeleteFile);
+              }
+            }
+          }
+          write(closeableWriter, deletes);
+          referencedDataFiles.add(path);
+        }
+      }
+
+      // DVs share the Puffin path and file size but have different offsets
+      String puffinPath = writer.location();
+      long puffinFileSize = writer.fileSize();
+
+      for (String path : deletesByPath.keySet()) {
+        DeleteFile dv = createDV(puffinPath, puffinFileSize, path);
+        dvs.add(dv);
+      }
+
+      this.result = new DeleteWriteResult(dvs, referencedDataFiles, rewrittenDeleteFiles);
+    }
+  }
+
+  private DeleteFile createDV(String path, long size, String referencedDataFile) {
+    Deletes deletes = deletesByPath.get(referencedDataFile);
+    BlobMetadata blobMetadata = blobsByPath.get(referencedDataFile);
+    return FileMetadata.deleteFileBuilder(deletes.spec())
+        .ofPositionDeletes()
+        .withFormat(FileFormat.PUFFIN)
+        .withPath(path)
+        .withPartition(deletes.partition())
+        .withFileSizeInBytes(size)
+        .withReferencedDataFile(referencedDataFile)
+        .withContentOffset(blobMetadata.offset())
+        .withContentSizeInBytes(blobMetadata.length())
+        .withRecordCount(deletes.positions().cardinality())
+        .build();
+  }
+
+  private void write(PuffinWriter writer, Deletes deletes) {
+    String path = deletes.path();
+    PositionDeleteIndex positions = deletes.positions();
+    BlobMetadata blobMetadata = writer.write(toBlob(positions, path));
+    blobsByPath.put(path, blobMetadata);
+  }
+
+  private PuffinWriter newWriter() {
+    EncryptedOutputFile outputFile = fileFactory.newOutputFile();
+    String ident = "Iceberg " + IcebergBuild.fullVersion();
+    return Puffin.write(outputFile).createdBy(ident).build();
+  }
+
+  private Blob toBlob(PositionDeleteIndex positions, String path) {
+    return new Blob(
+        StandardBlobTypes.DV_V1,
+        ImmutableList.of(MetadataColumns.ROW_POSITION.fieldId()),
+        -1 /* snapshot ID is inherited */,
+        -1 /* sequence number is inherited */,
+        positions.serialize(),
+        null /* uncompressed */,
+        ImmutableMap.of(
+            REFERENCED_DATA_FILE_KEY,
+            path,
+            CARDINALITY_KEY,
+            String.valueOf(positions.cardinality())));
+  }
+
+  private static class Deletes {
+    private final String path;
+    private final PositionDeleteIndex positions;
+    private final PartitionSpec spec;
+    private final StructLike partition;
+
+    private Deletes(String path, PartitionSpec spec, StructLike partition) {
+      this.path = path;
+      this.positions = new BitmapPositionDeleteIndex();
+      this.spec = spec;
+      this.partition = StructLikeUtil.copy(partition);
+    }
+
+    public String path() {
+      return path;
+    }
+
+    public PositionDeleteIndex positions() {
+      return positions;
+    }
+
+    public PartitionSpec spec() {
+      return spec;
+    }
+
+    public StructLike partition() {
+      return partition;
+    }
+  }
+}
diff --git a/core/src/main/java/org/apache/iceberg/deletes/DVFileWriter.java b/core/src/main/java/org/apache/iceberg/deletes/DVFileWriter.java
new file mode 100644
index 000000000000..2561f7be3d34
--- /dev/null
+++ b/core/src/main/java/org/apache/iceberg/deletes/DVFileWriter.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.deletes;
+
+import java.io.Closeable;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.io.DeleteWriteResult;
+
+/** A deletion vector file writer. */
+public interface DVFileWriter extends Closeable {
+  /**
+   * Marks a position in a given data file as deleted.
+   *
+   * @param path the data file path
+   * @param pos the data file position
+   * @param spec the data file partition spec
+   * @param partition the data file partition
+   */
+  void delete(String path, long pos, PartitionSpec spec, StructLike partition);
+
+  /**
+   * Returns a result that contains information about written {@link DeleteFile}s. The result is
+   * valid only after the writer is closed.
+   *
+   * @return the writer result
+   */
+  DeleteWriteResult result();
+}
diff --git a/core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java b/core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java
index 471dc3e56035..968db0ab538b 100644
--- a/core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java
+++ b/core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java
@@ -40,6 +40,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.util.CharSequenceSet;
 import org.apache.iceberg.util.StructLikeMap;
+import org.apache.iceberg.util.StructLikeUtil;
 import org.apache.iceberg.util.StructProjection;
 import org.apache.iceberg.util.Tasks;
 import org.apache.iceberg.util.ThreadPools;
@@ -149,7 +150,7 @@ public void write(T row) throws IOException {
       PathOffset pathOffset = PathOffset.of(dataWriter.currentPath(), dataWriter.currentRows());
 
       // Create a copied key from this row.
-      StructLike copiedKey = StructCopy.copy(structProjection.wrap(asStructLike(row)));
+      StructLike copiedKey = StructLikeUtil.copy(structProjection.wrap(asStructLike(row)));
 
       // Adding a pos-delete to replace the old path-offset.
       PathOffset previous = insertedRowMap.put(copiedKey, pathOffset);
diff --git a/core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java b/core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
index 1dc4871f0a12..06093ff2a943 100644
--- a/core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
+++ b/core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
@@ -29,6 +29,7 @@
 import org.apache.iceberg.types.Comparators;
 import org.apache.iceberg.types.Types.StructType;
 import org.apache.iceberg.util.StructLikeSet;
+import org.apache.iceberg.util.StructLikeUtil;
 
 /**
  * A writer capable of writing to multiple specs and partitions that requires the incoming records
@@ -81,7 +82,7 @@ public void write(T row, PartitionSpec spec, StructLike partition) {
       this.partitionComparator = Comparators.forType(partitionType);
       this.completedPartitions = StructLikeSet.create(partitionType);
       // copy the partition key as the key object may be reused
-      this.currentPartition = StructCopy.copy(partition);
+      this.currentPartition = StructLikeUtil.copy(partition);
       this.currentWriter = newWriter(currentSpec, currentPartition);
 
     } else if (partition != currentPartition
@@ -96,7 +97,7 @@ public void write(T row, PartitionSpec spec, StructLike partition) {
       }
 
       // copy the partition key as the key object may be reused
-      this.currentPartition = StructCopy.copy(partition);
+      this.currentPartition = StructLikeUtil.copy(partition);
       this.currentWriter = newWriter(currentSpec, currentPartition);
     }
 
diff --git a/core/src/main/java/org/apache/iceberg/io/FanoutWriter.java b/core/src/main/java/org/apache/iceberg/io/FanoutWriter.java
index 95d5cc1afcc9..340f4b3558d9 100644
--- a/core/src/main/java/org/apache/iceberg/io/FanoutWriter.java
+++ b/core/src/main/java/org/apache/iceberg/io/FanoutWriter.java
@@ -25,6 +25,7 @@
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.util.StructLikeMap;
+import org.apache.iceberg.util.StructLikeUtil;
 
 /**
  * A writer capable of writing to multiple specs and partitions that keeps files for each seen
@@ -59,7 +60,7 @@ private FileWriter<T, R> writer(PartitionSpec spec, StructLike partition) {
 
     if (writer == null) {
       // copy the partition key as the key object may be reused
-      StructLike copiedPartition = StructCopy.copy(partition);
+      StructLike copiedPartition = StructLikeUtil.copy(partition);
       writer = newWriter(spec, copiedPartition);
       specWriters.put(copiedPartition, writer);
     }
diff --git a/core/src/main/java/org/apache/iceberg/io/StructCopy.java b/core/src/main/java/org/apache/iceberg/io/StructCopy.java
index 229dff371762..f8adbf62a1d6 100644
--- a/core/src/main/java/org/apache/iceberg/io/StructCopy.java
+++ b/core/src/main/java/org/apache/iceberg/io/StructCopy.java
@@ -20,7 +20,13 @@
 
 import org.apache.iceberg.StructLike;
 
-/** Copy the StructLike's values into a new one. It does not handle list or map values now. */
+/**
+ * Copy the StructLike's values into a new one. It does not handle list or map values now.
+ *
+ * @deprecated since 1.8.0, will be removed in 1.9.0; use {@link
+ *     org.apache.iceberg.util.StructLikeUtil#copy(StructLike)} instead.
+ */
+@Deprecated
 class StructCopy implements StructLike {
   static StructLike copy(StructLike struct) {
     return struct != null ? new StructCopy(struct) : null;
diff --git a/core/src/main/java/org/apache/iceberg/puffin/Puffin.java b/core/src/main/java/org/apache/iceberg/puffin/Puffin.java
index 251486d01e76..d20a5596c4d4 100644
--- a/core/src/main/java/org/apache/iceberg/puffin/Puffin.java
+++ b/core/src/main/java/org/apache/iceberg/puffin/Puffin.java
@@ -19,6 +19,7 @@
 package org.apache.iceberg.puffin;
 
 import java.util.Map;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.io.OutputFile;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -31,6 +32,10 @@ public static WriteBuilder write(OutputFile outputFile) {
     return new WriteBuilder(outputFile);
   }
 
+  public static WriteBuilder write(EncryptedOutputFile outputFile) {
+    return new WriteBuilder(outputFile.encryptingOutputFile());
+  }
+
   /** A builder for {@link PuffinWriter}. */
   public static class WriteBuilder {
     private final OutputFile outputFile;
diff --git a/core/src/main/java/org/apache/iceberg/puffin/PuffinWriter.java b/core/src/main/java/org/apache/iceberg/puffin/PuffinWriter.java
index 5728b7474885..cd44dab03b90 100644
--- a/core/src/main/java/org/apache/iceberg/puffin/PuffinWriter.java
+++ b/core/src/main/java/org/apache/iceberg/puffin/PuffinWriter.java
@@ -44,6 +44,7 @@ public class PuffinWriter implements FileAppender<Blob> {
   // Must not be modified
   private static final byte[] MAGIC = PuffinFormat.getMagic();
 
+  private final OutputFile outputFile;
   private final PositionOutputStream outputStream;
   private final Map<String, String> properties;
   private final PuffinCompressionCodec footerCompression;
@@ -63,6 +64,7 @@ public class PuffinWriter implements FileAppender<Blob> {
     Preconditions.checkNotNull(outputFile, "outputFile is null");
     Preconditions.checkNotNull(properties, "properties is null");
     Preconditions.checkNotNull(defaultBlobCompression, "defaultBlobCompression is null");
+    this.outputFile = outputFile;
     this.outputStream = outputFile.create();
     this.properties = ImmutableMap.copyOf(properties);
     this.footerCompression =
@@ -70,8 +72,16 @@ public class PuffinWriter implements FileAppender<Blob> {
     this.defaultBlobCompression = defaultBlobCompression;
   }
 
+  public String location() {
+    return outputFile.location();
+  }
+
   @Override
   public void add(Blob blob) {
+    write(blob);
+  }
+
+  public BlobMetadata write(Blob blob) {
     Preconditions.checkNotNull(blob, "blob is null");
     checkNotFinished();
     try {
@@ -82,7 +92,7 @@ public void add(Blob blob) {
       ByteBuffer rawData = PuffinFormat.compress(codec, blob.blobData());
       int length = rawData.remaining();
       IOUtil.writeFully(outputStream, rawData);
-      writtenBlobsMetadata.add(
+      BlobMetadata blobMetadata =
           new BlobMetadata(
               blob.type(),
               blob.inputFields(),
@@ -91,7 +101,9 @@ public void add(Blob blob) {
               fileOffset,
               length,
               codec.codecName(),
-              blob.properties()));
+              blob.properties());
+      writtenBlobsMetadata.add(blobMetadata);
+      return blobMetadata;
     } catch (IOException e) {
       throw new UncheckedIOException(e);
     }
diff --git a/core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java b/core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java
index 2be5df5f88b9..ce78375c4b1a 100644
--- a/core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java
+++ b/core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java
@@ -26,4 +26,7 @@ private StandardBlobTypes() {}
    * href="https://datasketches.apache.org/">Apache DataSketches</a> library
    */
   public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+  /** A serialized deletion vector according to the Iceberg spec */
+  public static final String DV_V1 = "deletion-vector-v1";
 }
diff --git a/core/src/main/java/org/apache/iceberg/util/StructLikeUtil.java b/core/src/main/java/org/apache/iceberg/util/StructLikeUtil.java
new file mode 100644
index 000000000000..5285793a4aad
--- /dev/null
+++ b/core/src/main/java/org/apache/iceberg/util/StructLikeUtil.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.util;
+
+import org.apache.iceberg.StructLike;
+
+public class StructLikeUtil {
+
+  private StructLikeUtil() {}
+
+  public static StructLike copy(StructLike struct) {
+    return StructCopy.copy(struct);
+  }
+
+  private static class StructCopy implements StructLike {
+    private static StructLike copy(StructLike struct) {
+      return struct != null ? new StructCopy(struct) : null;
+    }
+
+    private final Object[] values;
+
+    private StructCopy(StructLike toCopy) {
+      this.values = new Object[toCopy.size()];
+
+      for (int i = 0; i < values.length; i += 1) {
+        Object value = toCopy.get(i, Object.class);
+
+        if (value instanceof StructLike) {
+          values[i] = copy((StructLike) value);
+        } else {
+          values[i] = value;
+        }
+      }
+    }
+
+    @Override
+    public int size() {
+      return values.length;
+    }
+
+    @Override
+    public <T> T get(int pos, Class<T> javaClass) {
+      return javaClass.cast(values[pos]);
+    }
+
+    @Override
+    public <T> void set(int pos, T value) {
+      throw new UnsupportedOperationException("Struct copy cannot be modified");
+    }
+  }
+}
diff --git a/spark/v3.5/spark/src/jmh/java/org/apache/iceberg/spark/source/DVWriterBenchmark.java b/spark/v3.5/spark/src/jmh/java/org/apache/iceberg/spark/source/DVWriterBenchmark.java
new file mode 100644
index 000000000000..ac74fb5a109c
--- /dev/null
+++ b/spark/v3.5/spark/src/jmh/java/org/apache/iceberg/spark/source/DVWriterBenchmark.java
@@ -0,0 +1,251 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.source;
+
+import com.google.errorprone.annotations.FormatMethod;
+import com.google.errorprone.annotations.FormatString;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileGenerationUtil;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.deletes.DeleteGranularity;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.io.DeleteWriteResult;
+import org.apache.iceberg.io.FanoutPositionOnlyDeleteWriter;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.iceberg.spark.SparkSessionCatalog;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
+import org.apache.spark.sql.catalyst.parser.ParseException;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Timeout;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+/**
+ * A benchmark that compares the performance of DV and position delete writers.
+ *
+ * <p>To run this benchmark for spark-3.5: <code>
+ *   ./gradlew -DsparkVersions=3.5 :iceberg-spark:iceberg-spark-3.5_2.12:jmh
+ *       -PjmhIncludeRegex=DVWriterBenchmark
+ *       -PjmhOutputPath=benchmark/iceberg-dv-writer-benchmark-result.txt
+ * </code>
+ */
+@Fork(1)
+@State(Scope.Benchmark)
+@Warmup(iterations = 3)
+@Measurement(iterations = 10)
+@Timeout(time = 20, timeUnit = TimeUnit.MINUTES)
+@BenchmarkMode(Mode.SingleShotTime)
+public class DVWriterBenchmark {
+
+  private static final String TABLE_NAME = "test_table";
+  private static final int DATA_FILE_RECORD_COUNT = 2_000_000;
+  private static final long TARGET_FILE_SIZE = Long.MAX_VALUE;
+
+  @Param({"5", "10"})
+  private int referencedDataFileCount;
+
+  @Param({"0.01", "0.03", "0.05", "0.10", "0.2"})
+  private double deletedRowsRatio;
+
+  private final Configuration hadoopConf = new Configuration();
+  private final Random random = ThreadLocalRandom.current();
+  private SparkSession spark;
+  private Table table;
+  private Iterable<InternalRow> positionDeletes;
+
+  @Setup
+  public void setupBenchmark() throws NoSuchTableException, ParseException {
+    setupSpark();
+    initTable();
+    generatePositionDeletes();
+  }
+
+  @TearDown
+  public void tearDownBenchmark() {
+    dropTable();
+    tearDownSpark();
+  }
+
+  @Benchmark
+  @Threads(1)
+  public void dv(Blackhole blackhole) throws IOException {
+    OutputFileFactory fileFactory = newFileFactory(FileFormat.PUFFIN);
+    DVFileWriter writer = new BaseDVFileWriter(fileFactory, path -> null);
+
+    try (DVFileWriter closableWriter = writer) {
+      for (InternalRow row : positionDeletes) {
+        String path = row.getString(0);
+        long pos = row.getLong(1);
+        closableWriter.delete(path, pos, table.spec(), null);
+      }
+    }
+
+    DeleteWriteResult result = writer.result();
+    blackhole.consume(result);
+  }
+
+  @Benchmark
+  @Threads(1)
+  public void fileScopedParquetDeletes(Blackhole blackhole) throws IOException {
+    FanoutPositionOnlyDeleteWriter<InternalRow> writer = newWriter(DeleteGranularity.FILE);
+    write(writer, positionDeletes);
+    DeleteWriteResult result = writer.result();
+    blackhole.consume(result);
+  }
+
+  @Benchmark
+  @Threads(1)
+  public void partitionScopedParquetDeletes(Blackhole blackhole) throws IOException {
+    FanoutPositionOnlyDeleteWriter<InternalRow> writer = newWriter(DeleteGranularity.PARTITION);
+    write(writer, positionDeletes);
+    DeleteWriteResult result = writer.result();
+    blackhole.consume(result);
+  }
+
+  private FanoutPositionOnlyDeleteWriter<InternalRow> newWriter(DeleteGranularity granularity) {
+    return new FanoutPositionOnlyDeleteWriter<>(
+        newWriterFactory(),
+        newFileFactory(FileFormat.PARQUET),
+        table.io(),
+        TARGET_FILE_SIZE,
+        granularity);
+  }
+
+  private DeleteWriteResult write(
+      FanoutPositionOnlyDeleteWriter<InternalRow> writer, Iterable<InternalRow> rows)
+      throws IOException {
+
+    try (FanoutPositionOnlyDeleteWriter<InternalRow> closableWriter = writer) {
+      PositionDelete<InternalRow> positionDelete = PositionDelete.create();
+
+      for (InternalRow row : rows) {
+        String path = row.getString(0);
+        long pos = row.getLong(1);
+        positionDelete.set(path, pos, null /* no row */);
+        closableWriter.write(positionDelete, table.spec(), null);
+      }
+    }
+
+    return writer.result();
+  }
+
+  private SparkFileWriterFactory newWriterFactory() {
+    return SparkFileWriterFactory.builderFor(table).dataFileFormat(FileFormat.PARQUET).build();
+  }
+
+  private OutputFileFactory newFileFactory(FileFormat format) {
+    return OutputFileFactory.builderFor(table, 1, 1).format(format).build();
+  }
+
+  private void generatePositionDeletes() {
+    int numDeletesPerFile = (int) (DATA_FILE_RECORD_COUNT * deletedRowsRatio);
+    int numDeletes = referencedDataFileCount * numDeletesPerFile;
+    List<InternalRow> deletes = Lists.newArrayListWithExpectedSize(numDeletes);
+
+    for (int pathIndex = 0; pathIndex < referencedDataFileCount; pathIndex++) {
+      UTF8String dataFilePath = UTF8String.fromString(generateDataFilePath());
+      Set<Long> positions = generatePositions(numDeletesPerFile);
+      for (long pos : positions) {
+        deletes.add(new GenericInternalRow(new Object[] {dataFilePath, pos}));
+      }
+    }
+
+    Collections.shuffle(deletes);
+
+    this.positionDeletes = deletes;
+  }
+
+  public Set<Long> generatePositions(int numPositions) {
+    Set<Long> positions = Sets.newHashSet();
+
+    while (positions.size() < numPositions) {
+      long pos = random.nextInt(DATA_FILE_RECORD_COUNT);
+      positions.add(pos);
+    }
+
+    return positions;
+  }
+
+  private String generateDataFilePath() {
+    String fileName = FileGenerationUtil.generateFileName();
+    return table.locationProvider().newDataLocation(table.spec(), null, fileName);
+  }
+
+  private void setupSpark() {
+    this.spark =
+        SparkSession.builder()
+            .config("spark.ui.enabled", false)
+            .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
+            .config("spark.sql.catalog.spark_catalog", SparkSessionCatalog.class.getName())
+            .config("spark.sql.catalog.spark_catalog.type", "hadoop")
+            .config("spark.sql.catalog.spark_catalog.warehouse", newWarehouseDir())
+            .master("local[*]")
+            .getOrCreate();
+  }
+
+  private void tearDownSpark() {
+    spark.stop();
+  }
+
+  private void initTable() throws NoSuchTableException, ParseException {
+    sql("CREATE TABLE %s (c1 INT, c2 INT, c3 STRING) USING iceberg", TABLE_NAME);
+    this.table = Spark3Util.loadIcebergTable(spark, TABLE_NAME);
+  }
+
+  private void dropTable() {
+    sql("DROP TABLE IF EXISTS %s PURGE", TABLE_NAME);
+  }
+
+  private String newWarehouseDir() {
+    return hadoopConf.get("hadoop.tmp.dir") + UUID.randomUUID();
+  }
+
+  @FormatMethod
+  private void sql(@FormatString String query, Object... args) {
+    spark.sql(String.format(query, args));
+  }
+}


Test patch:
diff --git a/data/src/test/java/org/apache/iceberg/io/TestDVWriters.java b/data/src/test/java/org/apache/iceberg/io/TestDVWriters.java
new file mode 100644
index 000000000000..ce742b1c4685
--- /dev/null
+++ b/data/src/test/java/org/apache/iceberg/io/TestDVWriters.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.io;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.BaseDeleteLoader;
+import org.apache.iceberg.data.DeleteLoader;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.util.StructLikeSet;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public abstract class TestDVWriters<T> extends WriterTestBase<T> {
+
+  @Parameters(name = "formatVersion = {0}")
+  protected static List<Object> parameters() {
+    return Arrays.asList(new Object[] {3});
+  }
+
+  private OutputFileFactory fileFactory = null;
+
+  protected abstract StructLikeSet toSet(Iterable<T> records);
+
+  protected FileFormat dataFormat() {
+    return FileFormat.PARQUET;
+  }
+
+  @Override
+  @BeforeEach
+  public void setupTable() throws Exception {
+    this.table = create(SCHEMA, PartitionSpec.unpartitioned());
+    this.fileFactory = OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build();
+  }
+
+  @TestTemplate
+  public void testBasicDVs() throws IOException {
+    FileWriterFactory<T> writerFactory = newWriterFactory(table.schema());
+
+    // add the first data file
+    List<T> rows1 = ImmutableList.of(toRow(1, "aaa"), toRow(2, "aaa"), toRow(11, "aaa"));
+    DataFile dataFile1 = writeData(writerFactory, fileFactory, rows1, table.spec(), null);
+    table.newFastAppend().appendFile(dataFile1).commit();
+
+    // add the second data file
+    List<T> rows2 = ImmutableList.of(toRow(3, "aaa"), toRow(4, "aaa"), toRow(12, "aaa"));
+    DataFile dataFile2 = writeData(writerFactory, fileFactory, rows2, table.spec(), null);
+    table.newFastAppend().appendFile(dataFile2).commit();
+
+    // init the DV writer
+    DVFileWriter dvWriter =
+        new BaseDVFileWriter(fileFactory, new PreviousDeleteLoader(table, ImmutableMap.of()));
+
+    // write deletes for both data files (the order of records is mixed)
+    dvWriter.delete(dataFile1.location(), 1L, table.spec(), null);
+    dvWriter.delete(dataFile2.location(), 0L, table.spec(), null);
+    dvWriter.delete(dataFile1.location(), 0L, table.spec(), null);
+    dvWriter.delete(dataFile2.location(), 1L, table.spec(), null);
+    dvWriter.close();
+
+    // verify the writer result
+    DeleteWriteResult result = dvWriter.result();
+    assertThat(result.deleteFiles()).hasSize(2);
+    assertThat(result.referencedDataFiles())
+        .hasSize(2)
+        .contains(dataFile1.location())
+        .contains(dataFile2.location());
+    assertThat(result.referencesDataFiles()).isTrue();
+  }
+
+  private static class PreviousDeleteLoader implements Function<String, PositionDeleteIndex> {
+    private final Map<String, DeleteFile> deleteFiles;
+    private final DeleteLoader deleteLoader;
+
+    PreviousDeleteLoader(Table table, Map<String, DeleteFile> deleteFiles) {
+      this.deleteFiles = deleteFiles;
+      this.deleteLoader = new BaseDeleteLoader(deleteFile -> table.io().newInputFile(deleteFile));
+    }
+
+    @Override
+    public PositionDeleteIndex apply(String path) {
+      DeleteFile deleteFile = deleteFiles.get(path);
+      if (deleteFile == null) {
+        return null;
+      }
+      return deleteLoader.loadPositionDeletes(ImmutableList.of(deleteFile), path);
+    }
+  }
+}
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDVWriters.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDVWriters.java
new file mode 100644
index 000000000000..dfc693d3094d
--- /dev/null
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDVWriters.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.source;
+
+import java.util.List;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.io.FileWriterFactory;
+import org.apache.iceberg.io.TestDVWriters;
+import org.apache.iceberg.spark.SparkSchemaUtil;
+import org.apache.iceberg.util.ArrayUtil;
+import org.apache.iceberg.util.StructLikeSet;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+
+public class TestSparkDVWriters extends TestDVWriters<InternalRow> {
+
+  @Override
+  protected FileWriterFactory<InternalRow> newWriterFactory(
+      Schema dataSchema,
+      List<Integer> equalityFieldIds,
+      Schema equalityDeleteRowSchema,
+      Schema positionDeleteRowSchema) {
+    return SparkFileWriterFactory.builderFor(table)
+        .dataSchema(table.schema())
+        .dataFileFormat(dataFormat())
+        .deleteFileFormat(dataFormat())
+        .equalityFieldIds(ArrayUtil.toIntArray(equalityFieldIds))
+        .equalityDeleteRowSchema(equalityDeleteRowSchema)
+        .positionDeleteRowSchema(positionDeleteRowSchema)
+        .build();
+  }
+
+  @Override
+  protected InternalRow toRow(Integer id, String data) {
+    InternalRow row = new GenericInternalRow(2);
+    row.update(0, id);
+    row.update(1, UTF8String.fromString(data));
+    return row;
+  }
+
+  @Override
+  protected StructLikeSet toSet(Iterable<InternalRow> rows) {
+    StructLikeSet set = StructLikeSet.create(table.schema().asStruct());
+    StructType sparkType = SparkSchemaUtil.convert(table.schema());
+    for (InternalRow row : rows) {
+      InternalRowWrapper wrapper = new InternalRowWrapper(sparkType, table.schema().asStruct());
+      set.add(wrapper.wrap(row));
+    }
+    return set;
+  }
+}