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/BaseContentScanTask.java b/core/src/main/java/org/apache/iceberg/BaseContentScanTask.java
index 8d38a48309c6..960c04cc0f37 100644
--- a/core/src/main/java/org/apache/iceberg/BaseContentScanTask.java
+++ b/core/src/main/java/org/apache/iceberg/BaseContentScanTask.java
@@ -23,6 +23,7 @@
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.util.ArrayUtil;
+import org.apache.iceberg.util.ScanTaskUtil;
 
 abstract class BaseContentScanTask<ThisT extends ContentScanTask<F>, F extends ContentFile<F>>
     implements ContentScanTask<F>, SplittableScanTask<ThisT> {
@@ -82,7 +83,7 @@ public long start() {
 
   @Override
   public long length() {
-    return file.fileSizeInBytes();
+    return ScanTaskUtil.contentSizeInBytes(file);
   }
 
   @Override
diff --git a/core/src/main/java/org/apache/iceberg/BaseFileScanTask.java b/core/src/main/java/org/apache/iceberg/BaseFileScanTask.java
index aa37f40be7c0..2cc406444552 100644
--- a/core/src/main/java/org/apache/iceberg/BaseFileScanTask.java
+++ b/core/src/main/java/org/apache/iceberg/BaseFileScanTask.java
@@ -176,6 +176,8 @@ public boolean canMerge(ScanTask other) {
     @Override
     public SplitScanTask merge(ScanTask other) {
       SplitScanTask that = (SplitScanTask) other;
+      // don't use deletesSizeBytes() here so that deletesSizeBytes is only calculated once after
+      // merging rather than for each task before merging
       return new SplitScanTask(offset, len + that.length(), fileScanTask, deletesSizeBytes);
     }
 


Test patch:
diff --git a/core/src/test/java/org/apache/iceberg/TestMetadataTableFilters.java b/core/src/test/java/org/apache/iceberg/TestMetadataTableFilters.java
index f8c34019875f..7c5a860db15f 100644
--- a/core/src/test/java/org/apache/iceberg/TestMetadataTableFilters.java
+++ b/core/src/test/java/org/apache/iceberg/TestMetadataTableFilters.java
@@ -52,18 +52,26 @@ protected static List<Object> parameters() {
     return Arrays.asList(
         new Object[] {1, MetadataTableType.DATA_FILES},
         new Object[] {2, MetadataTableType.DATA_FILES},
+        new Object[] {3, MetadataTableType.DATA_FILES},
         new Object[] {2, MetadataTableType.DELETE_FILES},
+        new Object[] {3, MetadataTableType.DELETE_FILES},
         new Object[] {1, MetadataTableType.FILES},
         new Object[] {2, MetadataTableType.FILES},
+        new Object[] {3, MetadataTableType.FILES},
         new Object[] {1, MetadataTableType.ALL_DATA_FILES},
         new Object[] {2, MetadataTableType.ALL_DATA_FILES},
+        new Object[] {3, MetadataTableType.ALL_DATA_FILES},
         new Object[] {2, MetadataTableType.ALL_DELETE_FILES},
+        new Object[] {3, MetadataTableType.ALL_DELETE_FILES},
         new Object[] {1, MetadataTableType.ALL_FILES},
         new Object[] {2, MetadataTableType.ALL_FILES},
+        new Object[] {3, MetadataTableType.ALL_FILES},
         new Object[] {1, MetadataTableType.ENTRIES},
         new Object[] {2, MetadataTableType.ENTRIES},
+        new Object[] {3, MetadataTableType.ENTRIES},
         new Object[] {1, MetadataTableType.ALL_ENTRIES},
-        new Object[] {2, MetadataTableType.ALL_ENTRIES});
+        new Object[] {2, MetadataTableType.ALL_ENTRIES},
+        new Object[] {3, MetadataTableType.ALL_ENTRIES});
   }
 
   @BeforeEach
@@ -76,9 +84,9 @@ public void setupTable() throws Exception {
     table.newFastAppend().appendFile(FILE_D).commit();
     table.newFastAppend().appendFile(FILE_B).commit();
 
-    if (formatVersion == 2) {
-      table.newRowDelta().addDeletes(FILE_A_DELETES).commit();
-      table.newRowDelta().addDeletes(FILE_B_DELETES).commit();
+    if (formatVersion >= 2) {
+      table.newRowDelta().addDeletes(fileADeletes()).commit();
+      table.newRowDelta().addDeletes(fileBDeletes()).commit();
       table.newRowDelta().addDeletes(FILE_C2_DELETES).commit();
       table.newRowDelta().addDeletes(FILE_D2_DELETES).commit();
     }
@@ -366,7 +374,7 @@ public void testPartitionSpecEvolutionRemovalV1() {
 
   @TestTemplate
   public void testPartitionSpecEvolutionRemovalV2() {
-    assumeThat(formatVersion).isEqualTo(2);
+    assumeThat(formatVersion).isGreaterThanOrEqualTo(2);
 
     // Change spec and add two data and delete files each
     table.updateSpec().removeField(Expressions.bucket("data", 16)).addField("id").commit();
@@ -388,27 +396,13 @@ public void testPartitionSpecEvolutionRemovalV2() {
             .withPartitionPath("id=11")
             .build();
 
-    DeleteFile delete10 =
-        FileMetadata.deleteFileBuilder(newSpec)
-            .ofPositionDeletes()
-            .withPath("/path/to/data-10-deletes.parquet")
-            .withFileSizeInBytes(10)
-            .withPartitionPath("id=10")
-            .withRecordCount(1)
-            .build();
-    DeleteFile delete11 =
-        FileMetadata.deleteFileBuilder(newSpec)
-            .ofPositionDeletes()
-            .withPath("/path/to/data-11-deletes.parquet")
-            .withFileSizeInBytes(10)
-            .withPartitionPath("id=11")
-            .withRecordCount(1)
-            .build();
+    DeleteFile delete10 = posDelete(table, data10);
+    DeleteFile delete11 = posDelete(table, data11);
 
     table.newFastAppend().appendFile(data10).commit();
     table.newFastAppend().appendFile(data11).commit();
 
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       table.newRowDelta().addDeletes(delete10).commit();
       table.newRowDelta().addDeletes(delete11).commit();
     }
@@ -447,6 +441,12 @@ public void testPartitionSpecEvolutionRemovalV2() {
     assertThat(tasks).hasSize(expectedScanTaskCount(3));
   }
 
+  private DeleteFile posDelete(Table table, DataFile dataFile) {
+    return formatVersion >= 3
+        ? FileGenerationUtil.generateDV(table, dataFile)
+        : FileGenerationUtil.generatePositionDeleteFile(table, dataFile);
+  }
+
   @TestTemplate
   public void testPartitionSpecEvolutionAdditiveV1() {
     assumeThat(formatVersion).isEqualTo(1);
@@ -514,8 +514,8 @@ public void testPartitionSpecEvolutionAdditiveV1() {
   }
 
   @TestTemplate
-  public void testPartitionSpecEvolutionAdditiveV2() {
-    assumeThat(formatVersion).isEqualTo(2);
+  public void testPartitionSpecEvolutionAdditiveV2AndAbove() {
+    assumeThat(formatVersion).isGreaterThanOrEqualTo(2);
 
     // Change spec and add two data and delete files each
     table.updateSpec().addField("id").commit();
@@ -537,27 +537,13 @@ public void testPartitionSpecEvolutionAdditiveV2() {
             .withPartitionPath("data_bucket=1/id=11")
             .build();
 
-    DeleteFile delete10 =
-        FileMetadata.deleteFileBuilder(newSpec)
-            .ofPositionDeletes()
-            .withPath("/path/to/data-10-deletes.parquet")
-            .withFileSizeInBytes(10)
-            .withPartitionPath("data_bucket=0/id=10")
-            .withRecordCount(1)
-            .build();
-    DeleteFile delete11 =
-        FileMetadata.deleteFileBuilder(newSpec)
-            .ofPositionDeletes()
-            .withPath("/path/to/data-11-deletes.parquet")
-            .withFileSizeInBytes(10)
-            .withPartitionPath("data_bucket=1/id=11")
-            .withRecordCount(1)
-            .build();
+    DeleteFile delete10 = posDelete(table, data10);
+    DeleteFile delete11 = posDelete(table, data11);
 
     table.newFastAppend().appendFile(data10).commit();
     table.newFastAppend().appendFile(data11).commit();
 
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       table.newRowDelta().addDeletes(delete10).commit();
       table.newRowDelta().addDeletes(delete11).commit();
     }
diff --git a/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java b/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java
index f811dac02043..a31e02144167 100644
--- a/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java
+++ b/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java
@@ -1733,7 +1733,7 @@ public void testFilesTableEstimateSize() throws Exception {
     assertEstimatedRowCount(new AllDataFilesTable(table), 4);
     assertEstimatedRowCount(new AllFilesTable(table), 4);
 
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       assertEstimatedRowCount(new DeleteFilesTable(table), 4);
       assertEstimatedRowCount(new AllDeleteFilesTable(table), 4);
     }
diff --git a/data/src/test/java/org/apache/iceberg/data/FileHelpers.java b/data/src/test/java/org/apache/iceberg/data/FileHelpers.java
index 62df1634aa3f..181ca18138a1 100644
--- a/data/src/test/java/org/apache/iceberg/data/FileHelpers.java
+++ b/data/src/test/java/org/apache/iceberg/data/FileHelpers.java
@@ -23,9 +23,12 @@
 import java.util.List;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
 import org.apache.iceberg.deletes.EqualityDeleteWriter;
 import org.apache.iceberg.deletes.PositionDelete;
 import org.apache.iceberg.deletes.PositionDeleteWriter;
@@ -35,6 +38,8 @@
 import org.apache.iceberg.io.DataWriter;
 import org.apache.iceberg.io.FileWriterFactory;
 import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.CharSequenceSet;
 import org.apache.iceberg.util.Pair;
@@ -47,21 +52,53 @@ public static Pair<DeleteFile, CharSequenceSet> writeDeleteFile(
     return writeDeleteFile(table, out, null, deletes);
   }
 
+  public static Pair<DeleteFile, CharSequenceSet> writeDeleteFile(
+      Table table, OutputFile out, List<Pair<CharSequence, Long>> deletes, int formatVersion)
+      throws IOException {
+    return writeDeleteFile(table, out, null, deletes, formatVersion);
+  }
+
   public static Pair<DeleteFile, CharSequenceSet> writeDeleteFile(
       Table table, OutputFile out, StructLike partition, List<Pair<CharSequence, Long>> deletes)
       throws IOException {
-    FileWriterFactory<Record> factory = GenericFileWriterFactory.builderFor(table).build();
+    return writeDeleteFile(table, out, partition, deletes, 2);
+  }
 
-    PositionDeleteWriter<Record> writer =
-        factory.newPositionDeleteWriter(encrypt(out), table.spec(), partition);
-    PositionDelete<Record> posDelete = PositionDelete.create();
-    try (Closeable toClose = writer) {
-      for (Pair<CharSequence, Long> delete : deletes) {
-        writer.write(posDelete.set(delete.first(), delete.second(), null));
+  public static Pair<DeleteFile, CharSequenceSet> writeDeleteFile(
+      Table table,
+      OutputFile out,
+      StructLike partition,
+      List<Pair<CharSequence, Long>> deletes,
+      int formatVersion)
+      throws IOException {
+    if (formatVersion >= 3) {
+      OutputFileFactory fileFactory =
+          OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build();
+      DVFileWriter writer = new BaseDVFileWriter(fileFactory, p -> null);
+      try (DVFileWriter closeableWriter = writer) {
+        for (Pair<CharSequence, Long> delete : deletes) {
+          closeableWriter.delete(
+              delete.first().toString(), delete.second(), table.spec(), partition);
+        }
+      }
+
+      return Pair.of(
+          Iterables.getOnlyElement(writer.result().deleteFiles()),
+          writer.result().referencedDataFiles());
+    } else {
+      FileWriterFactory<Record> factory = GenericFileWriterFactory.builderFor(table).build();
+
+      PositionDeleteWriter<Record> writer =
+          factory.newPositionDeleteWriter(encrypt(out), table.spec(), partition);
+      PositionDelete<Record> posDelete = PositionDelete.create();
+      try (Closeable toClose = writer) {
+        for (Pair<CharSequence, Long> delete : deletes) {
+          writer.write(posDelete.set(delete.first(), delete.second(), null));
+        }
       }
-    }
 
-    return Pair.of(writer.toDeleteFile(), writer.referencedDataFiles());
+      return Pair.of(writer.toDeleteFile(), writer.referencedDataFiles());
+    }
   }
 
   public static DeleteFile writeDeleteFile(
@@ -121,18 +158,43 @@ public static DataFile writeDataFile(
   public static DeleteFile writePosDeleteFile(
       Table table, OutputFile out, StructLike partition, List<PositionDelete<?>> deletes)
       throws IOException {
-    FileWriterFactory<Record> factory =
-        GenericFileWriterFactory.builderFor(table).positionDeleteRowSchema(table.schema()).build();
+    return writePosDeleteFile(table, out, partition, deletes, 2);
+  }
 
-    PositionDeleteWriter<?> writer =
-        factory.newPositionDeleteWriter(encrypt(out), table.spec(), partition);
-    try (Closeable toClose = writer) {
-      for (PositionDelete delete : deletes) {
-        writer.write(delete);
+  public static DeleteFile writePosDeleteFile(
+      Table table,
+      OutputFile out,
+      StructLike partition,
+      List<PositionDelete<?>> deletes,
+      int formatVersion)
+      throws IOException {
+    if (formatVersion >= 3) {
+      OutputFileFactory fileFactory =
+          OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build();
+      DVFileWriter writer = new BaseDVFileWriter(fileFactory, p -> null);
+      try (DVFileWriter closeableWriter = writer) {
+        for (PositionDelete<?> delete : deletes) {
+          closeableWriter.delete(delete.path().toString(), delete.pos(), table.spec(), partition);
+        }
       }
-    }
 
-    return writer.toDeleteFile();
+      return Iterables.getOnlyElement(writer.result().deleteFiles());
+    } else {
+      FileWriterFactory<Record> factory =
+          GenericFileWriterFactory.builderFor(table)
+              .positionDeleteRowSchema(table.schema())
+              .build();
+
+      PositionDeleteWriter<?> writer =
+          factory.newPositionDeleteWriter(encrypt(out), table.spec(), partition);
+      try (Closeable toClose = writer) {
+        for (PositionDelete delete : deletes) {
+          writer.write(delete);
+        }
+      }
+
+      return writer.toDeleteFile();
+    }
   }
 
   private static EncryptedOutputFile encrypt(OutputFile out) {
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/actions/TestRewriteDataFilesAction.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/actions/TestRewriteDataFilesAction.java
index 6b8399f666d4..085e7e48204c 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/actions/TestRewriteDataFilesAction.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/actions/TestRewriteDataFilesAction.java
@@ -25,6 +25,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Path;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.Set;
@@ -46,6 +47,7 @@
 import org.apache.iceberg.Parameters;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.actions.RewriteDataFilesActionResult;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
@@ -77,6 +79,9 @@ public class TestRewriteDataFilesAction extends CatalogTestBase {
   @Parameter(index = 2)
   private FileFormat format;
 
+  @Parameter(index = 3)
+  private int formatVersion;
+
   private Table icebergTableUnPartitioned;
   private Table icebergTablePartitioned;
   private Table icebergTableWithPk;
@@ -87,15 +92,17 @@ protected TableEnvironment getTableEnv() {
     return super.getTableEnv();
   }
 
-  @Parameters(name = "catalogName={0}, baseNamespace={1}, format={2}")
+  @Parameters(name = "catalogName={0}, baseNamespace={1}, format={2}, formatVersion={3}")
   public static List<Object[]> parameters() {
     List<Object[]> parameters = Lists.newArrayList();
     for (FileFormat format :
         new FileFormat[] {FileFormat.AVRO, FileFormat.ORC, FileFormat.PARQUET}) {
       for (Object[] catalogParams : CatalogTestBase.parameters()) {
-        String catalogName = (String) catalogParams[0];
-        Namespace baseNamespace = (Namespace) catalogParams[1];
-        parameters.add(new Object[] {catalogName, baseNamespace, format});
+        for (int version : Arrays.asList(2, 3)) {
+          String catalogName = (String) catalogParams[0];
+          Namespace baseNamespace = (Namespace) catalogParams[1];
+          parameters.add(new Object[] {catalogName, baseNamespace, format, version});
+        }
       }
     }
     return parameters;
@@ -111,21 +118,21 @@ public void before() {
     sql("USE CATALOG %s", catalogName);
     sql("USE %s", DATABASE);
     sql(
-        "CREATE TABLE %s (id int, data varchar) with ('write.format.default'='%s')",
-        TABLE_NAME_UNPARTITIONED, format.name());
+        "CREATE TABLE %s (id int, data varchar) with ('write.format.default'='%s', '%s'='%s')",
+        TABLE_NAME_UNPARTITIONED, format.name(), TableProperties.FORMAT_VERSION, formatVersion);
     icebergTableUnPartitioned =
         validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME_UNPARTITIONED));
 
     sql(
         "CREATE TABLE %s (id int, data varchar,spec varchar) "
-            + " PARTITIONED BY (data,spec) with ('write.format.default'='%s')",
-        TABLE_NAME_PARTITIONED, format.name());
+            + " PARTITIONED BY (data,spec) with ('write.format.default'='%s', '%s'='%s')",
+        TABLE_NAME_PARTITIONED, format.name(), TableProperties.FORMAT_VERSION, formatVersion);
     icebergTablePartitioned =
         validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME_PARTITIONED));
 
     sql(
-        "CREATE TABLE %s (id int, data varchar, PRIMARY KEY(`id`) NOT ENFORCED) with ('write.format.default'='%s', 'format-version'='2')",
-        TABLE_NAME_WITH_PK, format.name());
+        "CREATE TABLE %s (id int, data varchar, PRIMARY KEY(`id`) NOT ENFORCED) with ('write.format.default'='%s', '%s'='%s')",
+        TABLE_NAME_WITH_PK, format.name(), TableProperties.FORMAT_VERSION, formatVersion);
     icebergTableWithPk =
         validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME_WITH_PK));
   }
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java
index bfa09552396a..ad93b80baf81 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java
@@ -23,6 +23,8 @@
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.io.File;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
@@ -32,8 +34,12 @@
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
 import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileGenerationUtil;
 import org.apache.iceberg.FileMetadata;
 import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
@@ -47,14 +53,15 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 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.TestBase;
 import org.apache.iceberg.types.Types;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.io.TempDir;
 
+@ExtendWith(ParameterizedTestExtension.class)
 public class TestDeleteReachableFilesAction extends TestBase {
   private static final HadoopTables TABLES = new HadoopTables(new Configuration());
   private static final Schema SCHEMA =
@@ -112,13 +119,24 @@ public class TestDeleteReachableFilesAction extends TestBase {
           .build();
 
   @TempDir private File tableDir;
+  @Parameter private int formatVersion;
+
+  @Parameters(name = "formatVersion = {0}")
+  protected static List<Object> parameters() {
+    return Arrays.asList(2, 3);
+  }
 
   private Table table;
 
   @BeforeEach
   public void setupTableLocation() throws Exception {
     String tableLocation = tableDir.toURI().toString();
-    this.table = TABLES.create(SCHEMA, SPEC, Maps.newHashMap(), tableLocation);
+    this.table =
+        TABLES.create(
+            SCHEMA,
+            SPEC,
+            ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion)),
+            tableLocation);
     spark.conf().set("spark.sql.shuffle.partitions", SHUFFLE_PARTITIONS);
   }
 
@@ -155,7 +173,7 @@ private void checkRemoveFilesResults(
         .isEqualTo(expectedOtherFilesDeleted);
   }
 
-  @Test
+  @TestTemplate
   public void dataFilesCleanupWithParallelTasks() {
     table.newFastAppend().appendFile(FILE_A).commit();
 
@@ -206,7 +224,7 @@ public void dataFilesCleanupWithParallelTasks() {
     checkRemoveFilesResults(4L, 0, 0, 6L, 4L, 6, result);
   }
 
-  @Test
+  @TestTemplate
   public void testWithExpiringDanglingStageCommit() {
     table.location();
     // `A` commit
@@ -224,7 +242,7 @@ public void testWithExpiringDanglingStageCommit() {
     checkRemoveFilesResults(3L, 0, 0, 3L, 3L, 5, result);
   }
 
-  @Test
+  @TestTemplate
   public void testRemoveFileActionOnEmptyTable() {
     DeleteReachableFiles.Result result =
         sparkActions().deleteReachableFiles(metadataLocation(table)).io(table.io()).execute();
@@ -232,7 +250,7 @@ public void testRemoveFileActionOnEmptyTable() {
     checkRemoveFilesResults(0, 0, 0, 0, 0, 2, result);
   }
 
-  @Test
+  @TestTemplate
   public void testRemoveFilesActionWithReducedVersionsTable() {
     table.updateProperties().set(TableProperties.METADATA_PREVIOUS_VERSIONS_MAX, "2").commit();
     table.newAppend().appendFile(FILE_A).commit();
@@ -252,7 +270,7 @@ public void testRemoveFilesActionWithReducedVersionsTable() {
     checkRemoveFilesResults(4, 0, 0, 5, 5, 8, result);
   }
 
-  @Test
+  @TestTemplate
   public void testRemoveFilesAction() {
     table.newAppend().appendFile(FILE_A).commit();
 
@@ -263,20 +281,20 @@ public void testRemoveFilesAction() {
     checkRemoveFilesResults(2, 0, 0, 2, 2, 4, baseRemoveFilesSparkAction.execute());
   }
 
-  @Test
+  @TestTemplate
   public void testPositionDeleteFiles() {
     table.newAppend().appendFile(FILE_A).commit();
 
     table.newAppend().appendFile(FILE_B).commit();
 
-    table.newRowDelta().addDeletes(FILE_A_POS_DELETES).commit();
+    table.newRowDelta().addDeletes(fileADeletes()).commit();
 
     DeleteReachableFiles baseRemoveFilesSparkAction =
         sparkActions().deleteReachableFiles(metadataLocation(table)).io(table.io());
     checkRemoveFilesResults(2, 1, 0, 3, 3, 5, baseRemoveFilesSparkAction.execute());
   }
 
-  @Test
+  @TestTemplate
   public void testEqualityDeleteFiles() {
     table.newAppend().appendFile(FILE_A).commit();
 
@@ -289,7 +307,7 @@ public void testEqualityDeleteFiles() {
     checkRemoveFilesResults(2, 0, 1, 3, 3, 5, baseRemoveFilesSparkAction.execute());
   }
 
-  @Test
+  @TestTemplate
   public void testRemoveFilesActionWithDefaultIO() {
     table.newAppend().appendFile(FILE_A).commit();
 
@@ -302,7 +320,7 @@ public void testRemoveFilesActionWithDefaultIO() {
     checkRemoveFilesResults(2, 0, 0, 2, 2, 4, baseRemoveFilesSparkAction.execute());
   }
 
-  @Test
+  @TestTemplate
   public void testUseLocalIterator() {
     table.newFastAppend().appendFile(FILE_A).commit();
 
@@ -333,7 +351,7 @@ public void testUseLocalIterator() {
         });
   }
 
-  @Test
+  @TestTemplate
   public void testIgnoreMetadataFilesNotFound() {
     table.updateProperties().set(TableProperties.METADATA_PREVIOUS_VERSIONS_MAX, "1").commit();
 
@@ -354,7 +372,7 @@ public void testIgnoreMetadataFilesNotFound() {
     checkRemoveFilesResults(1, 0, 0, 1, 1, 4, res);
   }
 
-  @Test
+  @TestTemplate
   public void testEmptyIOThrowsException() {
     DeleteReachableFiles baseRemoveFilesSparkAction =
         sparkActions().deleteReachableFiles(metadataLocation(table)).io(null);
@@ -364,7 +382,7 @@ public void testEmptyIOThrowsException() {
         .hasMessage("File IO cannot be null");
   }
 
-  @Test
+  @TestTemplate
   public void testRemoveFilesActionWhenGarbageCollectionDisabled() {
     table.updateProperties().set(TableProperties.GC_ENABLED, "false").commit();
 
@@ -381,4 +399,8 @@ private String metadataLocation(Table tbl) {
   private ActionsProvider sparkActions() {
     return SparkActions.get();
   }
+
+  private DeleteFile fileADeletes() {
+    return formatVersion >= 3 ? FileGenerationUtil.generateDV(table, FILE_A) : FILE_A_POS_DELETES;
+  }
 }
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestExpireSnapshotsAction.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestExpireSnapshotsAction.java
index 5909dec51c05..661df99ef3e6 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestExpireSnapshotsAction.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestExpireSnapshotsAction.java
@@ -21,10 +21,12 @@
 import static org.apache.iceberg.types.Types.NestedField.optional;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assumptions.assumeThat;
 
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Path;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -36,8 +38,12 @@
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
 import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileGenerationUtil;
 import org.apache.iceberg.FileMetadata;
 import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.ReachableFileUtil;
 import org.apache.iceberg.Schema;
@@ -53,16 +59,17 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 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.TestBase;
 import org.apache.iceberg.spark.data.TestHelpers;
 import org.apache.iceberg.types.Types;
 import org.apache.spark.sql.Dataset;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.io.TempDir;
 
+@ExtendWith(ParameterizedTestExtension.class)
 public class TestExpireSnapshotsAction extends TestBase {
   private static final HadoopTables TABLES = new HadoopTables(new Configuration());
   private static final Schema SCHEMA =
@@ -120,6 +127,12 @@ public class TestExpireSnapshotsAction extends TestBase {
           .build();
 
   @TempDir private Path temp;
+  @Parameter private int formatVersion;
+
+  @Parameters(name = "formatVersion = {0}")
+  protected static List<Object> parameters() {
+    return Arrays.asList(2, 3);
+  }
 
   @TempDir private File tableDir;
   private String tableLocation;
@@ -128,7 +141,12 @@ public class TestExpireSnapshotsAction extends TestBase {
   @BeforeEach
   public void setupTableLocation() throws Exception {
     this.tableLocation = tableDir.toURI().toString();
-    this.table = TABLES.create(SCHEMA, SPEC, Maps.newHashMap(), tableLocation);
+    this.table =
+        TABLES.create(
+            SCHEMA,
+            SPEC,
+            ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion)),
+            tableLocation);
     spark.conf().set("spark.sql.shuffle.partitions", SHUFFLE_PARTITIONS);
   }
 
@@ -144,6 +162,10 @@ private Long rightAfterSnapshot(long snapshotId) {
     return end;
   }
 
+  private DeleteFile fileADeletes() {
+    return formatVersion >= 3 ? FileGenerationUtil.generateDV(table, FILE_A) : FILE_A_POS_DELETES;
+  }
+
   private void checkExpirationResults(
       long expectedDatafiles,
       long expectedPosDeleteFiles,
@@ -173,7 +195,7 @@ private void checkExpirationResults(
         .isEqualTo(expectedManifestListsDeleted);
   }
 
-  @Test
+  @TestTemplate
   public void testFilesCleaned() throws Exception {
     table.newFastAppend().appendFile(FILE_A).commit();
 
@@ -191,7 +213,7 @@ public void testFilesCleaned() throws Exception {
     checkExpirationResults(1L, 0L, 0L, 1L, 2L, results);
   }
 
-  @Test
+  @TestTemplate
   public void dataFilesCleanupWithParallelTasks() throws IOException {
 
     table.newFastAppend().appendFile(FILE_A).commit();
@@ -245,7 +267,7 @@ public void dataFilesCleanupWithParallelTasks() throws IOException {
     checkExpirationResults(2L, 0L, 0L, 3L, 3L, result);
   }
 
-  @Test
+  @TestTemplate
   public void testNoFilesDeletedWhenNoSnapshotsExpired() throws Exception {
     table.newFastAppend().appendFile(FILE_A).commit();
 
@@ -253,7 +275,7 @@ public void testNoFilesDeletedWhenNoSnapshotsExpired() throws Exception {
     checkExpirationResults(0L, 0L, 0L, 0L, 0L, results);
   }
 
-  @Test
+  @TestTemplate
   public void testCleanupRepeatedOverwrites() throws Exception {
     table.newFastAppend().appendFile(FILE_A).commit();
 
@@ -269,7 +291,7 @@ public void testCleanupRepeatedOverwrites() throws Exception {
     checkExpirationResults(1L, 0L, 0L, 39L, 20L, results);
   }
 
-  @Test
+  @TestTemplate
   public void testRetainLastWithExpireOlderThan() {
     table
         .newAppend()
@@ -300,7 +322,7 @@ public void testRetainLastWithExpireOlderThan() {
     assertThat(table.snapshot(firstSnapshotId)).as("First snapshot should not present.").isNull();
   }
 
-  @Test
+  @TestTemplate
   public void testExpireTwoSnapshotsById() throws Exception {
     table
         .newAppend()
@@ -335,7 +357,7 @@ public void testExpireTwoSnapshotsById() throws Exception {
     checkExpirationResults(0L, 0L, 0L, 0L, 2L, result);
   }
 
-  @Test
+  @TestTemplate
   public void testRetainLastWithExpireById() {
     table
         .newAppend()
@@ -366,7 +388,7 @@ public void testRetainLastWithExpireById() {
     checkExpirationResults(0L, 0L, 0L, 0L, 1L, result);
   }
 
-  @Test
+  @TestTemplate
   public void testRetainLastWithTooFewSnapshots() {
     table
         .newAppend()
@@ -393,7 +415,7 @@ public void testRetainLastWithTooFewSnapshots() {
     checkExpirationResults(0L, 0L, 0L, 0L, 0L, result);
   }
 
-  @Test
+  @TestTemplate
   public void testRetainLastKeepsExpiringSnapshot() {
     table
         .newAppend()
@@ -432,7 +454,7 @@ public void testRetainLastKeepsExpiringSnapshot() {
     checkExpirationResults(0L, 0L, 0L, 0L, 1L, result);
   }
 
-  @Test
+  @TestTemplate
   public void testExpireSnapshotsWithDisabledGarbageCollection() {
     table.updateProperties().set(TableProperties.GC_ENABLED, "false").commit();
 
@@ -444,7 +466,7 @@ public void testExpireSnapshotsWithDisabledGarbageCollection() {
             "Cannot expire snapshots: GC is disabled (deleting files may corrupt other tables)");
   }
 
-  @Test
+  @TestTemplate
   public void testExpireOlderThanMultipleCalls() {
     table
         .newAppend()
@@ -480,7 +502,7 @@ public void testExpireOlderThanMultipleCalls() {
     checkExpirationResults(0L, 0L, 0L, 0L, 2L, result);
   }
 
-  @Test
+  @TestTemplate
   public void testRetainLastMultipleCalls() {
     table
         .newAppend()
@@ -517,14 +539,14 @@ public void testRetainLastMultipleCalls() {
     checkExpirationResults(0L, 0L, 0L, 0L, 2L, result);
   }
 
-  @Test
+  @TestTemplate
   public void testRetainZeroSnapshots() {
     assertThatThrownBy(() -> SparkActions.get().expireSnapshots(table).retainLast(0).execute())
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessage("Number of snapshots to retain must be at least 1, cannot be: 0");
   }
 
-  @Test
+  @TestTemplate
   public void testScanExpiredManifestInValidSnapshotAppend() {
     table.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
 
@@ -547,7 +569,7 @@ public void testScanExpiredManifestInValidSnapshotAppend() {
     checkExpirationResults(1L, 0L, 0L, 1L, 2L, result);
   }
 
-  @Test
+  @TestTemplate
   public void testScanExpiredManifestInValidSnapshotFastAppend() {
     table
         .updateProperties()
@@ -580,7 +602,7 @@ public void testScanExpiredManifestInValidSnapshotFastAppend() {
    * Test on table below, and expiring the staged commit `B` using `expireOlderThan` API. Table: A -
    * C ` B (staged)
    */
-  @Test
+  @TestTemplate
   public void testWithExpiringDanglingStageCommit() {
     // `A` commit
     table.newAppend().appendFile(FILE_A).commit();
@@ -641,7 +663,7 @@ public void testWithExpiringDanglingStageCommit() {
    * Expire cherry-pick the commit as shown below, when `B` is in table's current state Table: A - B
    * - C <--current snapshot `- D (source=B)
    */
-  @Test
+  @TestTemplate
   public void testWithCherryPickTableSnapshot() {
     // `A` commit
     table.newAppend().appendFile(FILE_A).commit();
@@ -696,7 +718,7 @@ public void testWithCherryPickTableSnapshot() {
    * Test on table below, and expiring `B` which is not in current table state. 1) Expire `B` 2) All
    * commit Table: A - C - D (B) ` B (staged)
    */
-  @Test
+  @TestTemplate
   public void testWithExpiringStagedThenCherrypick() {
     // `A` commit
     table.newAppend().appendFile(FILE_A).commit();
@@ -760,7 +782,7 @@ public void testWithExpiringStagedThenCherrypick() {
     checkExpirationResults(0L, 0L, 0L, 0L, 2L, secondResult);
   }
 
-  @Test
+  @TestTemplate
   public void testExpireOlderThan() {
     table.newAppend().appendFile(FILE_A).commit();
 
@@ -796,7 +818,7 @@ public void testExpireOlderThan() {
     checkExpirationResults(0, 0, 0, 0, 1, result);
   }
 
-  @Test
+  @TestTemplate
   public void testExpireOlderThanWithDelete() {
     table.newAppend().appendFile(FILE_A).commit();
 
@@ -858,7 +880,7 @@ public void testExpireOlderThanWithDelete() {
     checkExpirationResults(1, 0, 0, 2, 2, result);
   }
 
-  @Test
+  @TestTemplate
   public void testExpireOlderThanWithDeleteInMergedManifests() {
     // merge every commit
     table.updateProperties().set(TableProperties.MANIFEST_MIN_MERGE_COUNT, "0").commit();
@@ -924,7 +946,7 @@ public void testExpireOlderThanWithDeleteInMergedManifests() {
     checkExpirationResults(1, 0, 0, 1, 2, result);
   }
 
-  @Test
+  @TestTemplate
   public void testExpireOlderThanWithRollback() {
     // merge every commit
     table.updateProperties().set(TableProperties.MANIFEST_MIN_MERGE_COUNT, "0").commit();
@@ -982,7 +1004,7 @@ public void testExpireOlderThanWithRollback() {
     checkExpirationResults(0, 0, 0, 1, 1, result);
   }
 
-  @Test
+  @TestTemplate
   public void testExpireOlderThanWithRollbackAndMergedManifests() {
     table.newAppend().appendFile(FILE_A).commit();
 
@@ -1037,20 +1059,18 @@ public void testExpireOlderThanWithRollbackAndMergedManifests() {
     checkExpirationResults(1, 0, 0, 1, 1, result);
   }
 
-  @Test
+  @TestTemplate
   public void testExpireOlderThanWithDeleteFile() {
-    table
-        .updateProperties()
-        .set(TableProperties.FORMAT_VERSION, "2")
-        .set(TableProperties.MANIFEST_MERGE_ENABLED, "false")
-        .commit();
+    assumeThat(formatVersion).isGreaterThanOrEqualTo(2);
+    table.updateProperties().set(TableProperties.MANIFEST_MERGE_ENABLED, "false").commit();
 
     // Add Data File
     table.newAppend().appendFile(FILE_A).commit();
     Snapshot firstSnapshot = table.currentSnapshot();
 
     // Add POS Delete
-    table.newRowDelta().addDeletes(FILE_A_POS_DELETES).commit();
+    DeleteFile fileADeletes = fileADeletes();
+    table.newRowDelta().addDeletes(fileADeletes).commit();
     Snapshot secondSnapshot = table.currentSnapshot();
 
     // Add EQ Delete
@@ -1081,7 +1101,7 @@ public void testExpireOlderThanWithDeleteFile() {
             thirdSnapshot.manifestListLocation(),
             fourthSnapshot.manifestListLocation(),
             FILE_A.path().toString(),
-            FILE_A_POS_DELETES.path().toString(),
+            fileADeletes.path().toString(),
             FILE_A_EQ_DELETES.path().toString());
 
     expectedDeletes.addAll(
@@ -1103,7 +1123,7 @@ public void testExpireOlderThanWithDeleteFile() {
     checkExpirationResults(1, 1, 1, 6, 4, result);
   }
 
-  @Test
+  @TestTemplate
   public void testExpireOnEmptyTable() {
     Set<String> deletedFiles = Sets.newHashSet();
 
@@ -1118,7 +1138,7 @@ public void testExpireOnEmptyTable() {
     checkExpirationResults(0, 0, 0, 0, 0, result);
   }
 
-  @Test
+  @TestTemplate
   public void testExpireAction() {
     table.newAppend().appendFile(FILE_A).commit();
 
@@ -1167,7 +1187,7 @@ public void testExpireAction() {
         .isEqualTo(pendingDeletes.count());
   }
 
-  @Test
+  @TestTemplate
   public void testUseLocalIterator() {
     table.newFastAppend().appendFile(FILE_A).commit();
 
@@ -1201,7 +1221,7 @@ public void testUseLocalIterator() {
         });
   }
 
-  @Test
+  @TestTemplate
   public void testExpireAfterExecute() {
     table
         .newAppend()
@@ -1236,12 +1256,12 @@ public void testExpireAfterExecute() {
     assertThat(untypedExpiredFiles).as("Expired results must match").hasSize(1);
   }
 
-  @Test
+  @TestTemplate
   public void testExpireFileDeletionMostExpired() {
     textExpireAllCheckFilesDeleted(5, 2);
   }
 
-  @Test
+  @TestTemplate
   public void testExpireFileDeletionMostRetained() {
     textExpireAllCheckFilesDeleted(2, 5);
   }
@@ -1303,7 +1323,7 @@ public void textExpireAllCheckFilesDeleted(int dataFilesExpired, int dataFilesRe
         .isEqualTo(expectedDeletes);
   }
 
-  @Test
+  @TestTemplate
   public void testExpireSomeCheckFilesDeleted() {
 
     table.newAppend().appendFile(FILE_A).commit();
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java
index 3b4dce73fee5..e58966cfea3f 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java
@@ -22,6 +22,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.io.File;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -30,7 +31,11 @@
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
 import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileGenerationUtil;
 import org.apache.iceberg.FileMetadata;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
@@ -44,10 +49,12 @@
 import org.apache.spark.sql.Encoders;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.io.TempDir;
 import scala.Tuple2;
 
+@ExtendWith(ParameterizedTestExtension.class)
 public class TestRemoveDanglingDeleteAction extends TestBase {
 
   private static final HadoopTables TABLES = new HadoopTables(new Configuration());
@@ -202,6 +209,12 @@ public class TestRemoveDanglingDeleteAction extends TestBase {
           .build();
 
   @TempDir private File tableDir;
+  @Parameter private int formatVersion;
+
+  @Parameters(name = "formatVersion = {0}")
+  protected static List<Object> parameters() {
+    return Arrays.asList(2, 3);
+  }
 
   private String tableLocation = null;
   private Table table;
@@ -219,7 +232,10 @@ public void after() {
   private void setupPartitionedTable() {
     this.table =
         TABLES.create(
-            SCHEMA, SPEC, ImmutableMap.of(TableProperties.FORMAT_VERSION, "2"), tableLocation);
+            SCHEMA,
+            SPEC,
+            ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion)),
+            tableLocation);
   }
 
   private void setupUnpartitionedTable() {
@@ -227,11 +243,33 @@ private void setupUnpartitionedTable() {
         TABLES.create(
             SCHEMA,
             PartitionSpec.unpartitioned(),
-            ImmutableMap.of(TableProperties.FORMAT_VERSION, "2"),
+            ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion)),
             tableLocation);
   }
 
-  @Test
+  private DeleteFile fileADeletes() {
+    return formatVersion >= 3 ? FileGenerationUtil.generateDV(table, FILE_A) : FILE_A_POS_DELETES;
+  }
+
+  private DeleteFile fileA2Deletes() {
+    return formatVersion >= 3 ? FileGenerationUtil.generateDV(table, FILE_A2) : FILE_A2_POS_DELETES;
+  }
+
+  private DeleteFile fileBDeletes() {
+    return formatVersion >= 3 ? FileGenerationUtil.generateDV(table, FILE_B) : FILE_B_POS_DELETES;
+  }
+
+  private DeleteFile fileB2Deletes() {
+    return formatVersion >= 3 ? FileGenerationUtil.generateDV(table, FILE_B2) : FILE_B2_POS_DELETES;
+  }
+
+  private DeleteFile fileUnpartitionedDeletes() {
+    return formatVersion >= 3
+        ? FileGenerationUtil.generateDV(table, FILE_UNPARTITIONED)
+        : FILE_UNPARTITIONED_POS_DELETE;
+  }
+
+  @TestTemplate
   public void testPartitionedDeletesWithLesserSeqNo() {
     setupPartitionedTable();
 
@@ -239,12 +277,16 @@ public void testPartitionedDeletesWithLesserSeqNo() {
     table.newAppend().appendFile(FILE_B).appendFile(FILE_C).appendFile(FILE_D).commit();
 
     // Add Delete Files
+    DeleteFile fileADeletes = fileADeletes();
+    DeleteFile fileA2Deletes = fileA2Deletes();
+    DeleteFile fileBDeletes = fileBDeletes();
+    DeleteFile fileB2Deletes = fileB2Deletes();
     table
         .newRowDelta()
-        .addDeletes(FILE_A_POS_DELETES)
-        .addDeletes(FILE_A2_POS_DELETES)
-        .addDeletes(FILE_B_POS_DELETES)
-        .addDeletes(FILE_B2_POS_DELETES)
+        .addDeletes(fileADeletes)
+        .addDeletes(fileA2Deletes)
+        .addDeletes(fileBDeletes)
+        .addDeletes(fileB2Deletes)
         .addDeletes(FILE_A_EQ_DELETES)
         .addDeletes(FILE_A2_EQ_DELETES)
         .addDeletes(FILE_B_EQ_DELETES)
@@ -275,18 +317,18 @@ public void testPartitionedDeletesWithLesserSeqNo() {
             Tuple2.apply(1L, FILE_C.path().toString()),
             Tuple2.apply(1L, FILE_D.path().toString()),
             Tuple2.apply(2L, FILE_A_EQ_DELETES.path().toString()),
-            Tuple2.apply(2L, FILE_A_POS_DELETES.path().toString()),
+            Tuple2.apply(2L, fileADeletes.path().toString()),
             Tuple2.apply(2L, FILE_A2_EQ_DELETES.path().toString()),
-            Tuple2.apply(2L, FILE_A2_POS_DELETES.path().toString()),
+            Tuple2.apply(2L, fileA2Deletes.path().toString()),
             Tuple2.apply(2L, FILE_B_EQ_DELETES.path().toString()),
-            Tuple2.apply(2L, FILE_B_POS_DELETES.path().toString()),
+            Tuple2.apply(2L, fileBDeletes.path().toString()),
             Tuple2.apply(2L, FILE_B2_EQ_DELETES.path().toString()),
-            Tuple2.apply(2L, FILE_B2_POS_DELETES.path().toString()),
+            Tuple2.apply(2L, fileB2Deletes.path().toString()),
             Tuple2.apply(3L, FILE_A2.path().toString()),
             Tuple2.apply(3L, FILE_B2.path().toString()),
             Tuple2.apply(3L, FILE_C2.path().toString()),
             Tuple2.apply(3L, FILE_D2.path().toString()));
-    assertThat(actual).isEqualTo(expected);
+    assertThat(actual).containsExactlyInAnyOrderElementsOf(expected);
 
     RemoveDanglingDeleteFiles.Result result =
         SparkActions.get().removeDanglingDeleteFiles(table).execute();
@@ -302,8 +344,8 @@ public void testPartitionedDeletesWithLesserSeqNo() {
         .as("Expected 4 delete files removed")
         .hasSize(4)
         .containsExactlyInAnyOrder(
-            FILE_A_POS_DELETES.path(),
-            FILE_A2_POS_DELETES.path(),
+            fileADeletes.path(),
+            fileA2Deletes.path(),
             FILE_A_EQ_DELETES.path(),
             FILE_A2_EQ_DELETES.path());
 
@@ -323,17 +365,17 @@ public void testPartitionedDeletesWithLesserSeqNo() {
             Tuple2.apply(1L, FILE_C.path().toString()),
             Tuple2.apply(1L, FILE_D.path().toString()),
             Tuple2.apply(2L, FILE_B_EQ_DELETES.path().toString()),
-            Tuple2.apply(2L, FILE_B_POS_DELETES.path().toString()),
+            Tuple2.apply(2L, fileBDeletes.path().toString()),
             Tuple2.apply(2L, FILE_B2_EQ_DELETES.path().toString()),
-            Tuple2.apply(2L, FILE_B2_POS_DELETES.path().toString()),
+            Tuple2.apply(2L, fileB2Deletes.path().toString()),
             Tuple2.apply(3L, FILE_A2.path().toString()),
             Tuple2.apply(3L, FILE_B2.path().toString()),
             Tuple2.apply(3L, FILE_C2.path().toString()),
             Tuple2.apply(3L, FILE_D2.path().toString()));
-    assertThat(actualAfter).isEqualTo(expectedAfter);
+    assertThat(actualAfter).containsExactlyInAnyOrderElementsOf(expectedAfter);
   }
 
-  @Test
+  @TestTemplate
   public void testPartitionedDeletesWithEqSeqNo() {
     setupPartitionedTable();
 
@@ -341,18 +383,22 @@ public void testPartitionedDeletesWithEqSeqNo() {
     table.newAppend().appendFile(FILE_A).appendFile(FILE_C).appendFile(FILE_D).commit();
 
     // Add Data Files with EQ and POS deletes
+    DeleteFile fileADeletes = fileADeletes();
+    DeleteFile fileA2Deletes = fileA2Deletes();
+    DeleteFile fileBDeletes = fileBDeletes();
+    DeleteFile fileB2Deletes = fileB2Deletes();
     table
         .newRowDelta()
         .addRows(FILE_A2)
         .addRows(FILE_B2)
         .addRows(FILE_C2)
         .addRows(FILE_D2)
-        .addDeletes(FILE_A_POS_DELETES)
-        .addDeletes(FILE_A2_POS_DELETES)
+        .addDeletes(fileADeletes)
+        .addDeletes(fileA2Deletes)
         .addDeletes(FILE_A_EQ_DELETES)
         .addDeletes(FILE_A2_EQ_DELETES)
-        .addDeletes(FILE_B_POS_DELETES)
-        .addDeletes(FILE_B2_POS_DELETES)
+        .addDeletes(fileBDeletes)
+        .addDeletes(fileB2Deletes)
         .addDeletes(FILE_B_EQ_DELETES)
         .addDeletes(FILE_B2_EQ_DELETES)
         .commit();
@@ -372,18 +418,18 @@ public void testPartitionedDeletesWithEqSeqNo() {
             Tuple2.apply(1L, FILE_C.path().toString()),
             Tuple2.apply(1L, FILE_D.path().toString()),
             Tuple2.apply(2L, FILE_A_EQ_DELETES.path().toString()),
-            Tuple2.apply(2L, FILE_A_POS_DELETES.path().toString()),
+            Tuple2.apply(2L, fileADeletes.path().toString()),
             Tuple2.apply(2L, FILE_A2.path().toString()),
             Tuple2.apply(2L, FILE_A2_EQ_DELETES.path().toString()),
-            Tuple2.apply(2L, FILE_A2_POS_DELETES.path().toString()),
+            Tuple2.apply(2L, fileA2Deletes.path().toString()),
             Tuple2.apply(2L, FILE_B_EQ_DELETES.path().toString()),
-            Tuple2.apply(2L, FILE_B_POS_DELETES.path().toString()),
+            Tuple2.apply(2L, fileBDeletes.path().toString()),
             Tuple2.apply(2L, FILE_B2.path().toString()),
             Tuple2.apply(2L, FILE_B2_EQ_DELETES.path().toString()),
-            Tuple2.apply(2L, FILE_B2_POS_DELETES.path().toString()),
+            Tuple2.apply(2L, fileB2Deletes.path().toString()),
             Tuple2.apply(2L, FILE_C2.path().toString()),
             Tuple2.apply(2L, FILE_D2.path().toString()));
-    assertThat(actual).isEqualTo(expected);
+    assertThat(actual).containsExactlyInAnyOrderElementsOf(expected);
 
     RemoveDanglingDeleteFiles.Result result =
         SparkActions.get().removeDanglingDeleteFiles(table).execute();
@@ -415,25 +461,25 @@ public void testPartitionedDeletesWithEqSeqNo() {
             Tuple2.apply(1L, FILE_C.path().toString()),
             Tuple2.apply(1L, FILE_D.path().toString()),
             Tuple2.apply(2L, FILE_A_EQ_DELETES.path().toString()),
-            Tuple2.apply(2L, FILE_A_POS_DELETES.path().toString()),
+            Tuple2.apply(2L, fileADeletes.path().toString()),
             Tuple2.apply(2L, FILE_A2.path().toString()),
             Tuple2.apply(2L, FILE_A2_EQ_DELETES.path().toString()),
-            Tuple2.apply(2L, FILE_A2_POS_DELETES.path().toString()),
-            Tuple2.apply(2L, FILE_B_POS_DELETES.path().toString()),
+            Tuple2.apply(2L, fileA2Deletes.path().toString()),
+            Tuple2.apply(2L, fileBDeletes.path().toString()),
             Tuple2.apply(2L, FILE_B2.path().toString()),
-            Tuple2.apply(2L, FILE_B2_POS_DELETES.path().toString()),
+            Tuple2.apply(2L, fileB2Deletes.path().toString()),
             Tuple2.apply(2L, FILE_C2.path().toString()),
             Tuple2.apply(2L, FILE_D2.path().toString()));
-    assertThat(actualAfter).isEqualTo(expectedAfter);
+    assertThat(actualAfter).containsExactlyInAnyOrderElementsOf(expectedAfter);
   }
 
-  @Test
+  @TestTemplate
   public void testUnpartitionedTable() {
     setupUnpartitionedTable();
 
     table
         .newRowDelta()
-        .addDeletes(FILE_UNPARTITIONED_POS_DELETE)
+        .addDeletes(fileUnpartitionedDeletes())
         .addDeletes(FILE_UNPARTITIONED_EQ_DELETE)
         .commit();
     table.newAppend().appendFile(FILE_UNPARTITIONED).commit();
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction.java
index 12defafff06d..d36898d4c464 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction.java
@@ -21,6 +21,7 @@
 import static org.apache.iceberg.types.Types.NestedField.optional;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assumptions.assumeThat;
 
 import java.io.File;
 import java.io.IOException;
@@ -35,6 +36,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
@@ -45,6 +47,9 @@
 import org.apache.iceberg.Files;
 import org.apache.iceberg.GenericBlobMetadata;
 import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Snapshot;
@@ -81,9 +86,11 @@
 import org.apache.spark.sql.types.DataTypes;
 import org.apache.spark.sql.types.StructType;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.io.TempDir;
 
+@ExtendWith(ParameterizedTestExtension.class)
 public abstract class TestRemoveOrphanFilesAction extends TestBase {
 
   private static final HadoopTables TABLES = new HadoopTables(new Configuration());
@@ -97,16 +104,23 @@ public abstract class TestRemoveOrphanFilesAction extends TestBase {
 
   @TempDir private File tableDir = null;
   protected String tableLocation = null;
+  protected Map<String, String> properties;
+  @Parameter private int formatVersion;
+
+  @Parameters(name = "formatVersion = {0}")
+  protected static List<Object> parameters() {
+    return Arrays.asList(2, 3);
+  }
 
   @BeforeEach
   public void setupTableLocation() throws Exception {
     this.tableLocation = tableDir.toURI().toString();
+    properties = ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion));
   }
 
-  @Test
-  public void testDryRun() throws IOException, InterruptedException {
-    Table table =
-        TABLES.create(SCHEMA, PartitionSpec.unpartitioned(), Maps.newHashMap(), tableLocation);
+  @TestTemplate
+  public void testDryRun() throws IOException {
+    Table table = TABLES.create(SCHEMA, PartitionSpec.unpartitioned(), properties, tableLocation);
 
     List<ThreeColumnRecord> records =
         Lists.newArrayList(new ThreeColumnRecord(1, "AAAAAAAAAA", "AAAA"));
@@ -184,9 +198,9 @@ public void testDryRun() throws IOException, InterruptedException {
     assertThat(actualRecords).isEqualTo(expectedRecords);
   }
 
-  @Test
-  public void testAllValidFilesAreKept() throws IOException, InterruptedException {
-    Table table = TABLES.create(SCHEMA, SPEC, Maps.newHashMap(), tableLocation);
+  @TestTemplate
+  public void testAllValidFilesAreKept() throws IOException {
+    Table table = TABLES.create(SCHEMA, SPEC, properties, tableLocation);
 
     List<ThreeColumnRecord> records1 =
         Lists.newArrayList(new ThreeColumnRecord(1, "AAAAAAAAAA", "AAAA"));
@@ -246,9 +260,9 @@ public void testAllValidFilesAreKept() throws IOException, InterruptedException
     }
   }
 
-  @Test
-  public void orphanedFileRemovedWithParallelTasks() throws InterruptedException, IOException {
-    Table table = TABLES.create(SCHEMA, SPEC, Maps.newHashMap(), tableLocation);
+  @TestTemplate
+  public void orphanedFileRemovedWithParallelTasks() {
+    Table table = TABLES.create(SCHEMA, SPEC, properties, tableLocation);
 
     List<ThreeColumnRecord> records1 =
         Lists.newArrayList(new ThreeColumnRecord(1, "AAAAAAAAAA", "AAAA"));
@@ -310,10 +324,12 @@ public void orphanedFileRemovedWithParallelTasks() throws InterruptedException,
     assertThat(deletedFiles).hasSize(4);
   }
 
-  @Test
-  public void testWapFilesAreKept() throws InterruptedException {
+  @TestTemplate
+  public void testWapFilesAreKept() {
+    assumeThat(formatVersion).as("currently fails with DVs").isEqualTo(2);
     Map<String, String> props = Maps.newHashMap();
     props.put(TableProperties.WRITE_AUDIT_PUBLISH_ENABLED, "true");
+    props.putAll(properties);
     Table table = TABLES.create(SCHEMA, SPEC, props, tableLocation);
 
     List<ThreeColumnRecord> records =
@@ -331,6 +347,8 @@ public void testWapFilesAreKept() throws InterruptedException {
     Dataset<Row> resultDF = spark.read().format("iceberg").load(tableLocation);
     List<ThreeColumnRecord> actualRecords =
         resultDF.as(Encoders.bean(ThreeColumnRecord.class)).collectAsList();
+
+    // TODO: currently fails because DVs delete stuff from WAP branch
     assertThat(actualRecords)
         .as("Should not return data from the staged snapshot")
         .isEqualTo(records);
@@ -345,11 +363,12 @@ public void testWapFilesAreKept() throws InterruptedException {
     assertThat(result.orphanFileLocations()).as("Should not delete any files").isEmpty();
   }
 
-  @Test
-  public void testMetadataFolderIsIntact() throws InterruptedException {
+  @TestTemplate
+  public void testMetadataFolderIsIntact() {
     // write data directly to the table location
     Map<String, String> props = Maps.newHashMap();
     props.put(TableProperties.WRITE_DATA_LOCATION, tableLocation);
+    props.putAll(properties);
     Table table = TABLES.create(SCHEMA, SPEC, props, tableLocation);
 
     List<ThreeColumnRecord> records =
@@ -375,9 +394,9 @@ public void testMetadataFolderIsIntact() throws InterruptedException {
     assertThat(actualRecords).as("Rows must match").isEqualTo(records);
   }
 
-  @Test
-  public void testOlderThanTimestamp() throws InterruptedException {
-    Table table = TABLES.create(SCHEMA, SPEC, Maps.newHashMap(), tableLocation);
+  @TestTemplate
+  public void testOlderThanTimestamp() {
+    Table table = TABLES.create(SCHEMA, SPEC, properties, tableLocation);
 
     List<ThreeColumnRecord> records =
         Lists.newArrayList(new ThreeColumnRecord(1, "AAAAAAAAAA", "AAAA"));
@@ -404,11 +423,12 @@ public void testOlderThanTimestamp() throws InterruptedException {
     assertThat(result.orphanFileLocations()).as("Should delete only 2 files").hasSize(2);
   }
 
-  @Test
-  public void testRemoveUnreachableMetadataVersionFiles() throws InterruptedException {
+  @TestTemplate
+  public void testRemoveUnreachableMetadataVersionFiles() {
     Map<String, String> props = Maps.newHashMap();
     props.put(TableProperties.WRITE_DATA_LOCATION, tableLocation);
     props.put(TableProperties.METADATA_PREVIOUS_VERSIONS_MAX, "1");
+    props.putAll(properties);
     Table table = TABLES.create(SCHEMA, SPEC, props, tableLocation);
 
     List<ThreeColumnRecord> records =
@@ -441,9 +461,9 @@ public void testRemoveUnreachableMetadataVersionFiles() throws InterruptedExcept
     assertThat(actualRecords).as("Rows must match").isEqualTo(expectedRecords);
   }
 
-  @Test
-  public void testManyTopLevelPartitions() throws InterruptedException {
-    Table table = TABLES.create(SCHEMA, SPEC, Maps.newHashMap(), tableLocation);
+  @TestTemplate
+  public void testManyTopLevelPartitions() {
+    Table table = TABLES.create(SCHEMA, SPEC, properties, tableLocation);
 
     List<ThreeColumnRecord> records = Lists.newArrayList();
     for (int i = 0; i < 100; i++) {
@@ -467,9 +487,9 @@ public void testManyTopLevelPartitions() throws InterruptedException {
     assertThat(resultDF.count()).as("Rows count must match").isEqualTo(records.size());
   }
 
-  @Test
-  public void testManyLeafPartitions() throws InterruptedException {
-    Table table = TABLES.create(SCHEMA, SPEC, Maps.newHashMap(), tableLocation);
+  @TestTemplate
+  public void testManyLeafPartitions() {
+    Table table = TABLES.create(SCHEMA, SPEC, properties, tableLocation);
 
     List<ThreeColumnRecord> records = Lists.newArrayList();
     for (int i = 0; i < 100; i++) {
@@ -493,15 +513,15 @@ public void testManyLeafPartitions() throws InterruptedException {
     assertThat(resultDF.count()).as("Row count must match").isEqualTo(records.size());
   }
 
-  @Test
-  public void testHiddenPartitionPaths() throws InterruptedException {
+  @TestTemplate
+  public void testHiddenPartitionPaths() {
     Schema schema =
         new Schema(
             optional(1, "c1", Types.IntegerType.get()),
             optional(2, "_c2", Types.StringType.get()),
             optional(3, "c3", Types.StringType.get()));
     PartitionSpec spec = PartitionSpec.builderFor(schema).truncate("_c2", 2).identity("c3").build();
-    Table table = TABLES.create(schema, spec, Maps.newHashMap(), tableLocation);
+    Table table = TABLES.create(schema, spec, properties, tableLocation);
 
     StructType structType =
         new StructType()
@@ -526,15 +546,15 @@ public void testHiddenPartitionPaths() throws InterruptedException {
     assertThat(result.orphanFileLocations()).as("Should delete 2 files").hasSize(2);
   }
 
-  @Test
-  public void testHiddenPartitionPathsWithPartitionEvolution() throws InterruptedException {
+  @TestTemplate
+  public void testHiddenPartitionPathsWithPartitionEvolution() {
     Schema schema =
         new Schema(
             optional(1, "_c1", Types.IntegerType.get()),
             optional(2, "_c2", Types.StringType.get()),
             optional(3, "c3", Types.StringType.get()));
     PartitionSpec spec = PartitionSpec.builderFor(schema).truncate("_c2", 2).build();
-    Table table = TABLES.create(schema, spec, Maps.newHashMap(), tableLocation);
+    Table table = TABLES.create(schema, spec, properties, tableLocation);
 
     StructType structType =
         new StructType()
@@ -562,16 +582,15 @@ public void testHiddenPartitionPathsWithPartitionEvolution() throws InterruptedE
     assertThat(result.orphanFileLocations()).as("Should delete 2 files").hasSize(2);
   }
 
-  @Test
-  public void testHiddenPathsStartingWithPartitionNamesAreIgnored()
-      throws InterruptedException, IOException {
+  @TestTemplate
+  public void testHiddenPathsStartingWithPartitionNamesAreIgnored() throws IOException {
     Schema schema =
         new Schema(
             optional(1, "c1", Types.IntegerType.get()),
             optional(2, "_c2", Types.StringType.get()),
             optional(3, "c3", Types.StringType.get()));
     PartitionSpec spec = PartitionSpec.builderFor(schema).truncate("_c2", 2).identity("c3").build();
-    Table table = TABLES.create(schema, spec, Maps.newHashMap(), tableLocation);
+    Table table = TABLES.create(schema, spec, properties, tableLocation);
 
     StructType structType =
         new StructType()
@@ -610,11 +629,11 @@ private List<String> snapshotFiles(long snapshotId) {
         .collectAsList();
   }
 
-  @Test
-  public void testRemoveOrphanFilesWithRelativeFilePath() throws IOException, InterruptedException {
+  @TestTemplate
+  public void testRemoveOrphanFilesWithRelativeFilePath() throws IOException {
     Table table =
         TABLES.create(
-            SCHEMA, PartitionSpec.unpartitioned(), Maps.newHashMap(), tableDir.getAbsolutePath());
+            SCHEMA, PartitionSpec.unpartitioned(), properties, tableDir.getAbsolutePath());
 
     List<ThreeColumnRecord> records =
         Lists.newArrayList(new ThreeColumnRecord(1, "AAAAAAAAAA", "AAAA"));
@@ -670,7 +689,7 @@ public void testRemoveOrphanFilesWithRelativeFilePath() throws IOException, Inte
         .isTrue();
   }
 
-  @Test
+  @TestTemplate
   public void testRemoveOrphanFilesWithHadoopCatalog() throws InterruptedException {
     HadoopCatalog catalog = new HadoopCatalog(new Configuration(), tableLocation);
     String namespaceName = "testDb";
@@ -705,15 +724,11 @@ public void testRemoveOrphanFilesWithHadoopCatalog() throws InterruptedException
     assertThat(actualRecords).as("Rows must match").isEqualTo(records);
   }
 
-  @Test
+  @TestTemplate
   public void testHiveCatalogTable() throws IOException {
-    Table table =
-        catalog.createTable(
-            TableIdentifier.of("default", "hivetestorphan"),
-            SCHEMA,
-            SPEC,
-            tableLocation,
-            Maps.newHashMap());
+    TableIdentifier identifier =
+        TableIdentifier.of("default", "hivetestorphan" + ThreadLocalRandom.current().nextInt(1000));
+    Table table = catalog.createTable(identifier, SCHEMA, SPEC, tableLocation, properties);
 
     List<ThreeColumnRecord> records =
         Lists.newArrayList(new ThreeColumnRecord(1, "AAAAAAAAAA", "AAAA"));
@@ -724,7 +739,7 @@ public void testHiveCatalogTable() throws IOException {
         .write()
         .format("iceberg")
         .mode("append")
-        .save("default.hivetestorphan");
+        .save(identifier.toString());
 
     String location = table.location().replaceFirst("file:", "");
     new File(location + "/data/trashfile").createNewFile();
@@ -739,10 +754,9 @@ public void testHiveCatalogTable() throws IOException {
         .anyMatch(file -> file.contains("file:" + location + "/data/trashfile"));
   }
 
-  @Test
+  @TestTemplate
   public void testGarbageCollectionDisabled() {
-    Table table =
-        TABLES.create(SCHEMA, PartitionSpec.unpartitioned(), Maps.newHashMap(), tableLocation);
+    Table table = TABLES.create(SCHEMA, PartitionSpec.unpartitioned(), properties, tableLocation);
 
     List<ThreeColumnRecord> records =
         Lists.newArrayList(new ThreeColumnRecord(1, "AAAAAAAAAA", "AAAA"));
@@ -759,10 +773,9 @@ public void testGarbageCollectionDisabled() {
             "Cannot delete orphan files: GC is disabled (deleting files may corrupt other tables)");
   }
 
-  @Test
-  public void testCompareToFileList() throws IOException, InterruptedException {
-    Table table =
-        TABLES.create(SCHEMA, PartitionSpec.unpartitioned(), Maps.newHashMap(), tableLocation);
+  @TestTemplate
+  public void testCompareToFileList() throws IOException {
+    Table table = TABLES.create(SCHEMA, PartitionSpec.unpartitioned(), properties, tableLocation);
 
     List<ThreeColumnRecord> records =
         Lists.newArrayList(new ThreeColumnRecord(1, "AAAAAAAAAA", "AAAA"));
@@ -890,14 +903,10 @@ protected long waitUntilAfter(long timestampMillis) {
     return current;
   }
 
-  @Test
+  @TestTemplate
   public void testRemoveOrphanFilesWithStatisticFiles() throws Exception {
-    Table table =
-        TABLES.create(
-            SCHEMA,
-            PartitionSpec.unpartitioned(),
-            ImmutableMap.of(TableProperties.FORMAT_VERSION, "2"),
-            tableLocation);
+    assumeThat(formatVersion).isGreaterThanOrEqualTo(2);
+    Table table = TABLES.create(SCHEMA, PartitionSpec.unpartitioned(), properties, tableLocation);
 
     List<ThreeColumnRecord> records =
         Lists.newArrayList(new ThreeColumnRecord(1, "AAAAAAAAAA", "AAAA"));
@@ -966,28 +975,28 @@ public void testRemoveOrphanFilesWithStatisticFiles() throws Exception {
     assertThat(statsLocation.exists()).as("stats file should be deleted").isFalse();
   }
 
-  @Test
+  @TestTemplate
   public void testPathsWithExtraSlashes() {
     List<String> validFiles = Lists.newArrayList("file:///dir1/dir2/file1");
     List<String> actualFiles = Lists.newArrayList("file:///dir1/////dir2///file1");
     executeTest(validFiles, actualFiles, Lists.newArrayList());
   }
 
-  @Test
+  @TestTemplate
   public void testPathsWithValidFileHavingNoAuthority() {
     List<String> validFiles = Lists.newArrayList("hdfs:///dir1/dir2/file1");
     List<String> actualFiles = Lists.newArrayList("hdfs://servicename/dir1/dir2/file1");
     executeTest(validFiles, actualFiles, Lists.newArrayList());
   }
 
-  @Test
+  @TestTemplate
   public void testPathsWithActualFileHavingNoAuthority() {
     List<String> validFiles = Lists.newArrayList("hdfs://servicename/dir1/dir2/file1");
     List<String> actualFiles = Lists.newArrayList("hdfs:///dir1/dir2/file1");
     executeTest(validFiles, actualFiles, Lists.newArrayList());
   }
 
-  @Test
+  @TestTemplate
   public void testPathsWithEqualSchemes() {
     List<String> validFiles = Lists.newArrayList("scheme1://bucket1/dir1/dir2/file1");
     List<String> actualFiles = Lists.newArrayList("scheme2://bucket1/dir1/dir2/file1");
@@ -1016,7 +1025,7 @@ public void testPathsWithEqualSchemes() {
         DeleteOrphanFiles.PrefixMismatchMode.ERROR);
   }
 
-  @Test
+  @TestTemplate
   public void testPathsWithEqualAuthorities() {
     List<String> validFiles = Lists.newArrayList("hdfs://servicename1/dir1/dir2/file1");
     List<String> actualFiles = Lists.newArrayList("hdfs://servicename2/dir1/dir2/file1");
@@ -1045,7 +1054,7 @@ public void testPathsWithEqualAuthorities() {
         DeleteOrphanFiles.PrefixMismatchMode.ERROR);
   }
 
-  @Test
+  @TestTemplate
   public void testRemoveOrphanFileActionWithDeleteMode() {
     List<String> validFiles = Lists.newArrayList("hdfs://servicename1/dir1/dir2/file1");
     List<String> actualFiles = Lists.newArrayList("hdfs://servicename2/dir1/dir2/file1");
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction3.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction3.java
index 2476d1bb7078..14784da4f74f 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction3.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction3.java
@@ -21,10 +21,9 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.io.File;
-import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.stream.StreamSupport;
 import org.apache.iceberg.actions.DeleteOrphanFiles;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.spark.SparkCatalog;
 import org.apache.iceberg.spark.SparkSchemaUtil;
 import org.apache.iceberg.spark.SparkSessionCatalog;
@@ -32,10 +31,10 @@
 import org.apache.spark.sql.connector.catalog.Identifier;
 import org.apache.spark.sql.connector.expressions.Transform;
 import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
 
 public class TestRemoveOrphanFilesAction3 extends TestRemoveOrphanFilesAction {
-  @Test
+  @TestTemplate
   public void testSparkCatalogTable() throws Exception {
     spark.conf().set("spark.sql.catalog.mycat", "org.apache.iceberg.spark.SparkCatalog");
     spark.conf().set("spark.sql.catalog.mycat.type", "hadoop");
@@ -43,16 +42,16 @@ public void testSparkCatalogTable() throws Exception {
     SparkCatalog cat = (SparkCatalog) spark.sessionState().catalogManager().catalog("mycat");
 
     String[] database = {"default"};
-    Identifier id = Identifier.of(database, "table");
-    Map<String, String> options = Maps.newHashMap();
+    Identifier id = Identifier.of(database, "table" + ThreadLocalRandom.current().nextInt(1000));
     Transform[] transforms = {};
-    cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms, options);
+    cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms, properties);
     SparkTable table = (SparkTable) cat.loadTable(id);
 
-    spark.sql("INSERT INTO mycat.default.table VALUES (1,1,1)");
+    sql("INSERT INTO mycat.default.%s VALUES (1,1,1)", id.name());
 
     String location = table.table().location().replaceFirst("file:", "");
-    new File(location + "/data/trashfile").createNewFile();
+    String trashFile = "/data/trashfile" + ThreadLocalRandom.current().nextInt(1000);
+    new File(location + trashFile).createNewFile();
 
     DeleteOrphanFiles.Result results =
         SparkActions.get()
@@ -61,10 +60,10 @@ public void testSparkCatalogTable() throws Exception {
             .execute();
     assertThat(StreamSupport.stream(results.orphanFileLocations().spliterator(), false))
         .as("trash file should be removed")
-        .anyMatch(file -> file.contains("file:" + location + "/data/trashfile"));
+        .anyMatch(file -> file.contains("file:" + location + trashFile));
   }
 
-  @Test
+  @TestTemplate
   public void testSparkCatalogNamedHadoopTable() throws Exception {
     spark.conf().set("spark.sql.catalog.hadoop", "org.apache.iceberg.spark.SparkCatalog");
     spark.conf().set("spark.sql.catalog.hadoop.type", "hadoop");
@@ -72,16 +71,16 @@ public void testSparkCatalogNamedHadoopTable() throws Exception {
     SparkCatalog cat = (SparkCatalog) spark.sessionState().catalogManager().catalog("hadoop");
 
     String[] database = {"default"};
-    Identifier id = Identifier.of(database, "table");
-    Map<String, String> options = Maps.newHashMap();
+    Identifier id = Identifier.of(database, "table" + ThreadLocalRandom.current().nextInt(1000));
     Transform[] transforms = {};
-    cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms, options);
+    cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms, properties);
     SparkTable table = (SparkTable) cat.loadTable(id);
 
-    spark.sql("INSERT INTO hadoop.default.table VALUES (1,1,1)");
+    sql("INSERT INTO hadoop.default.%s VALUES (1,1,1)", id.name());
 
     String location = table.table().location().replaceFirst("file:", "");
-    new File(location + "/data/trashfile").createNewFile();
+    String trashFile = "/data/trashfile" + ThreadLocalRandom.current().nextInt(1000);
+    new File(location + trashFile).createNewFile();
 
     DeleteOrphanFiles.Result results =
         SparkActions.get()
@@ -90,10 +89,10 @@ public void testSparkCatalogNamedHadoopTable() throws Exception {
             .execute();
     assertThat(StreamSupport.stream(results.orphanFileLocations().spliterator(), false))
         .as("trash file should be removed")
-        .anyMatch(file -> file.contains("file:" + location + "/data/trashfile"));
+        .anyMatch(file -> file.contains("file:" + location + trashFile));
   }
 
-  @Test
+  @TestTemplate
   public void testSparkCatalogNamedHiveTable() throws Exception {
     spark.conf().set("spark.sql.catalog.hive", "org.apache.iceberg.spark.SparkCatalog");
     spark.conf().set("spark.sql.catalog.hive.type", "hadoop");
@@ -101,16 +100,16 @@ public void testSparkCatalogNamedHiveTable() throws Exception {
     SparkCatalog cat = (SparkCatalog) spark.sessionState().catalogManager().catalog("hive");
 
     String[] database = {"default"};
-    Identifier id = Identifier.of(database, "table");
-    Map<String, String> options = Maps.newHashMap();
+    Identifier id = Identifier.of(database, "table" + ThreadLocalRandom.current().nextInt(1000));
     Transform[] transforms = {};
-    cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms, options);
+    cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms, properties);
     SparkTable table = (SparkTable) cat.loadTable(id);
 
-    spark.sql("INSERT INTO hive.default.table VALUES (1,1,1)");
+    sql("INSERT INTO hive.default.%s VALUES (1,1,1)", id.name());
 
     String location = table.table().location().replaceFirst("file:", "");
-    new File(location + "/data/trashfile").createNewFile();
+    String trashFile = "/data/trashfile" + ThreadLocalRandom.current().nextInt(1000);
+    new File(location + trashFile).createNewFile();
 
     DeleteOrphanFiles.Result results =
         SparkActions.get()
@@ -120,10 +119,10 @@ public void testSparkCatalogNamedHiveTable() throws Exception {
 
     assertThat(StreamSupport.stream(results.orphanFileLocations().spliterator(), false))
         .as("trash file should be removed")
-        .anyMatch(file -> file.contains("file:" + location + "/data/trashfile"));
+        .anyMatch(file -> file.contains("file:" + location + trashFile));
   }
 
-  @Test
+  @TestTemplate
   public void testSparkSessionCatalogHadoopTable() throws Exception {
     spark
         .conf()
@@ -134,16 +133,16 @@ public void testSparkSessionCatalogHadoopTable() throws Exception {
         (SparkSessionCatalog) spark.sessionState().catalogManager().v2SessionCatalog();
 
     String[] database = {"default"};
-    Identifier id = Identifier.of(database, "table");
-    Map<String, String> options = Maps.newHashMap();
+    Identifier id = Identifier.of(database, "table" + ThreadLocalRandom.current().nextInt(1000));
     Transform[] transforms = {};
-    cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms, options);
+    cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms, properties);
     SparkTable table = (SparkTable) cat.loadTable(id);
 
-    spark.sql("INSERT INTO default.table VALUES (1,1,1)");
+    sql("INSERT INTO default.%s VALUES (1,1,1)", id.name());
 
     String location = table.table().location().replaceFirst("file:", "");
-    new File(location + "/data/trashfile").createNewFile();
+    String trashFile = "/data/trashfile" + ThreadLocalRandom.current().nextInt(1000);
+    new File(location + trashFile).createNewFile();
 
     DeleteOrphanFiles.Result results =
         SparkActions.get()
@@ -152,10 +151,10 @@ public void testSparkSessionCatalogHadoopTable() throws Exception {
             .execute();
     assertThat(StreamSupport.stream(results.orphanFileLocations().spliterator(), false))
         .as("trash file should be removed")
-        .anyMatch(file -> file.contains("file:" + location + "/data/trashfile"));
+        .anyMatch(file -> file.contains("file:" + location + trashFile));
   }
 
-  @Test
+  @TestTemplate
   public void testSparkSessionCatalogHiveTable() throws Exception {
     spark
         .conf()
@@ -166,16 +165,16 @@ public void testSparkSessionCatalogHiveTable() throws Exception {
 
     String[] database = {"default"};
     Identifier id = Identifier.of(database, "sessioncattest");
-    Map<String, String> options = Maps.newHashMap();
     Transform[] transforms = {};
     cat.dropTable(id);
-    cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms, options);
+    cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms, properties);
     SparkTable table = (SparkTable) cat.loadTable(id);
 
     spark.sql("INSERT INTO default.sessioncattest VALUES (1,1,1)");
 
     String location = table.table().location().replaceFirst("file:", "");
-    new File(location + "/data/trashfile").createNewFile();
+    String trashFile = "/data/trashfile" + ThreadLocalRandom.current().nextInt(1000);
+    new File(location + trashFile).createNewFile();
 
     DeleteOrphanFiles.Result results =
         SparkActions.get()
@@ -184,7 +183,7 @@ public void testSparkSessionCatalogHiveTable() throws Exception {
             .execute();
     assertThat(StreamSupport.stream(results.orphanFileLocations().spliterator(), false))
         .as("trash file should be removed")
-        .anyMatch(file -> file.contains("file:" + location + "/data/trashfile"));
+        .anyMatch(file -> file.contains("file:" + location + trashFile));
   }
 
   @AfterEach
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
index b7ab47f865b5..38c4d32a90d2 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
@@ -27,6 +27,7 @@
 import static org.apache.spark.sql.functions.min;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assumptions.assumeThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.Mockito.doAnswer;
@@ -38,7 +39,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.UncheckedIOException;
-import java.nio.file.Path;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
@@ -56,6 +57,9 @@
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
 import org.apache.iceberg.PartitionData;
 import org.apache.iceberg.PartitionKey;
 import org.apache.iceberg.PartitionSpec;
@@ -77,6 +81,8 @@
 import org.apache.iceberg.data.GenericAppenderFactory;
 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.EqualityDeleteWriter;
 import org.apache.iceberg.deletes.PositionDelete;
 import org.apache.iceberg.deletes.PositionDeleteWriter;
@@ -96,7 +102,6 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 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.Streams;
 import org.apache.iceberg.spark.FileRewriteCoordinator;
 import org.apache.iceberg.spark.ScanTaskSetManager;
@@ -119,11 +124,13 @@
 import org.apache.spark.sql.internal.SQLConf;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.io.TempDir;
 import org.mockito.ArgumentMatcher;
 import org.mockito.Mockito;
 
+@ExtendWith(ParameterizedTestExtension.class)
 public class TestRewriteDataFilesAction extends TestBase {
 
   @TempDir private File tableDir;
@@ -138,7 +145,12 @@ public class TestRewriteDataFilesAction extends TestBase {
 
   private static final PartitionSpec SPEC = PartitionSpec.builderFor(SCHEMA).identity("c1").build();
 
-  @TempDir private Path temp;
+  @Parameter private int formatVersion;
+
+  @Parameters(name = "formatVersion = {0}")
+  protected static List<Object> parameters() {
+    return Arrays.asList(2, 3);
+  }
 
   private final FileRewriteCoordinator coordinator = FileRewriteCoordinator.get();
   private final ScanTaskSetManager manager = ScanTaskSetManager.get();
@@ -161,10 +173,11 @@ private RewriteDataFilesSparkAction basicRewrite(Table table) {
     return actions().rewriteDataFiles(table).option(SizeBasedFileRewriter.MIN_INPUT_FILES, "1");
   }
 
-  @Test
+  @TestTemplate
   public void testEmptyTable() {
     PartitionSpec spec = PartitionSpec.unpartitioned();
-    Map<String, String> options = Maps.newHashMap();
+    Map<String, String> options =
+        ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion));
     Table table = TABLES.create(SCHEMA, spec, options, tableLocation);
 
     assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
@@ -174,7 +187,7 @@ public void testEmptyTable() {
     assertThat(table.currentSnapshot()).as("Table must stay empty").isNull();
   }
 
-  @Test
+  @TestTemplate
   public void testBinPackUnpartitionedTable() {
     Table table = createTable(4);
     shouldHaveFiles(table, 4);
@@ -194,7 +207,7 @@ public void testBinPackUnpartitionedTable() {
     assertEquals("Rows must match", expectedRecords, actual);
   }
 
-  @Test
+  @TestTemplate
   public void testBinPackPartitionedTable() {
     Table table = createTablePartitioned(4, 2);
     shouldHaveFiles(table, 8);
@@ -214,7 +227,7 @@ public void testBinPackPartitionedTable() {
     assertEquals("Rows must match", expectedRecords, actualRecords);
   }
 
-  @Test
+  @TestTemplate
   public void testBinPackWithFilter() {
     Table table = createTablePartitioned(4, 2);
     shouldHaveFiles(table, 8);
@@ -239,7 +252,7 @@ public void testBinPackWithFilter() {
     assertEquals("Rows must match", expectedRecords, actualRecords);
   }
 
-  @Test
+  @TestTemplate
   public void testBinPackWithFilterOnBucketExpression() {
     Table table = createTablePartitioned(4, 2);
 
@@ -265,7 +278,7 @@ public void testBinPackWithFilterOnBucketExpression() {
     assertEquals("Rows must match", expectedRecords, actualRecords);
   }
 
-  @Test
+  @TestTemplate
   public void testBinPackAfterPartitionChange() {
     Table table = createTable();
 
@@ -300,10 +313,10 @@ public void testBinPackAfterPartitionChange() {
     shouldHaveFiles(table, 20);
   }
 
-  @Test
-  public void testBinPackWithDeletes() {
+  @TestTemplate
+  public void testBinPackWithDeletes() throws IOException {
+    assumeThat(formatVersion).isGreaterThanOrEqualTo(2);
     Table table = createTablePartitioned(4, 2);
-    table.updateProperties().set(TableProperties.FORMAT_VERSION, "2").commit();
     shouldHaveFiles(table, 8);
     table.refresh();
 
@@ -311,14 +324,28 @@ public void testBinPackWithDeletes() {
     int total = (int) dataFiles.stream().mapToLong(ContentFile::recordCount).sum();
 
     RowDelta rowDelta = table.newRowDelta();
-    // add 1 delete file for data files 0, 1, 2
-    for (int i = 0; i < 3; i++) {
-      writePosDeletesToFile(table, dataFiles.get(i), 1).forEach(rowDelta::addDeletes);
-    }
+    if (formatVersion >= 3) {
+      // delete 1 position for data files 0, 1, 2
+      for (int i = 0; i < 3; i++) {
+        writeDV(table, dataFiles.get(i).partition(), dataFiles.get(i).location(), 1)
+            .forEach(rowDelta::addDeletes);
+      }
 
-    // add 2 delete files for data files 3, 4
-    for (int i = 3; i < 5; i++) {
-      writePosDeletesToFile(table, dataFiles.get(i), 2).forEach(rowDelta::addDeletes);
+      // delete 2 positions for data files 3, 4
+      for (int i = 3; i < 5; i++) {
+        writeDV(table, dataFiles.get(i).partition(), dataFiles.get(i).location(), 2)
+            .forEach(rowDelta::addDeletes);
+      }
+    } else {
+      // add 1 delete file for data files 0, 1, 2
+      for (int i = 0; i < 3; i++) {
+        writePosDeletesToFile(table, dataFiles.get(i), 1).forEach(rowDelta::addDeletes);
+      }
+
+      // add 2 delete files for data files 3, 4
+      for (int i = 3; i < 5; i++) {
+        writePosDeletesToFile(table, dataFiles.get(i), 2).forEach(rowDelta::addDeletes);
+      }
     }
 
     rowDelta.commit();
@@ -326,32 +353,49 @@ public void testBinPackWithDeletes() {
     List<Object[]> expectedRecords = currentData();
     long dataSizeBefore = testDataSize(table);
 
-    Result result =
-        actions()
-            .rewriteDataFiles(table)
-            // do not include any file based on bin pack file size configs
-            .option(SizeBasedFileRewriter.MIN_FILE_SIZE_BYTES, "0")
-            .option(RewriteDataFiles.TARGET_FILE_SIZE_BYTES, Long.toString(Long.MAX_VALUE - 1))
-            .option(SizeBasedFileRewriter.MAX_FILE_SIZE_BYTES, Long.toString(Long.MAX_VALUE))
-            .option(SizeBasedDataRewriter.DELETE_FILE_THRESHOLD, "2")
-            .execute();
-    assertThat(result.rewrittenDataFilesCount())
-        .as("Action should rewrite 2 data files")
-        .isEqualTo(2);
-    assertThat(result.rewrittenBytesCount()).isGreaterThan(0L).isLessThan(dataSizeBefore);
+    if (formatVersion >= 3) {
+      Result result =
+          actions()
+              .rewriteDataFiles(table)
+              // do not include any file based on bin pack file size configs
+              .option(SizeBasedFileRewriter.MIN_FILE_SIZE_BYTES, "0")
+              .option(RewriteDataFiles.TARGET_FILE_SIZE_BYTES, Long.toString(Long.MAX_VALUE - 1))
+              .option(SizeBasedFileRewriter.MAX_FILE_SIZE_BYTES, Long.toString(Long.MAX_VALUE))
+              // set DELETE_FILE_THRESHOLD to 1 since DVs only produce one delete file per data file
+              .option(SizeBasedDataRewriter.DELETE_FILE_THRESHOLD, "1")
+              .execute();
+      assertThat(result.rewrittenDataFilesCount())
+          .as("Action should rewrite 5 data files")
+          .isEqualTo(5);
+      assertThat(result.rewrittenBytesCount()).isGreaterThan(0L).isLessThan(dataSizeBefore);
+    } else {
+      Result result =
+          actions()
+              .rewriteDataFiles(table)
+              // do not include any file based on bin pack file size configs
+              .option(SizeBasedFileRewriter.MIN_FILE_SIZE_BYTES, "0")
+              .option(RewriteDataFiles.TARGET_FILE_SIZE_BYTES, Long.toString(Long.MAX_VALUE - 1))
+              .option(SizeBasedFileRewriter.MAX_FILE_SIZE_BYTES, Long.toString(Long.MAX_VALUE))
+              .option(SizeBasedDataRewriter.DELETE_FILE_THRESHOLD, "2")
+              .execute();
+      assertThat(result.rewrittenDataFilesCount())
+          .as("Action should rewrite 2 data files")
+          .isEqualTo(2);
+      assertThat(result.rewrittenBytesCount()).isGreaterThan(0L).isLessThan(dataSizeBefore);
+    }
 
     List<Object[]> actualRecords = currentData();
     assertEquals("Rows must match", expectedRecords, actualRecords);
     assertThat(actualRecords).as("7 rows are removed").hasSize(total - 7);
   }
 
-  @Test
+  @TestTemplate
   public void testRemoveDangledEqualityDeletesPartitionEvolution() {
     Table table =
         TABLES.create(
             SCHEMA,
             SPEC,
-            Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"),
+            ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion)),
             tableLocation);
 
     // data seq = 1, write 4 files in 2 partitions
@@ -414,13 +458,13 @@ public void testRemoveDangledEqualityDeletesPartitionEvolution() {
     shouldHaveFiles(table, 5);
   }
 
-  @Test
-  public void testRemoveDangledPositionDeletesPartitionEvolution() {
+  @TestTemplate
+  public void testRemoveDangledPositionDeletesPartitionEvolution() throws IOException {
     Table table =
         TABLES.create(
             SCHEMA,
             SPEC,
-            Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"),
+            ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion)),
             tableLocation);
 
     // data seq = 1, write 4 files in 2 partitions
@@ -428,11 +472,15 @@ public void testRemoveDangledPositionDeletesPartitionEvolution() {
     List<DataFile> dataFilesBefore = TestHelpers.dataFiles(table, null);
     shouldHaveFiles(table, 4);
 
+    DeleteFile deleteFile;
     // data seq = 2, write 1 position deletes in c1=1
-    table
-        .newRowDelta()
-        .addDeletes(writePosDeletesToFile(table, dataFilesBefore.get(3), 1).get(0))
-        .commit();
+    DataFile dataFile = dataFilesBefore.get(3);
+    if (formatVersion >= 3) {
+      deleteFile = writeDV(table, dataFile.partition(), dataFile.location(), 1).get(0);
+    } else {
+      deleteFile = writePosDeletesToFile(table, dataFile, 1).get(0);
+    }
+    table.newRowDelta().addDeletes(deleteFile).commit();
 
     // partition evolution
     table.updateSpec().addField(Expressions.ref("c3")).commit();
@@ -464,11 +512,10 @@ public void testRemoveDangledPositionDeletesPartitionEvolution() {
     assertEquals("Rows must match", expectedRecords, currentData());
   }
 
-  @Test
-  public void testBinPackWithDeleteAllData() {
-    Map<String, String> options = Maps.newHashMap();
-    options.put(TableProperties.FORMAT_VERSION, "2");
-    Table table = createTablePartitioned(1, 1, 1, options);
+  @TestTemplate
+  public void testBinPackWithDeleteAllData() throws IOException {
+    assumeThat(formatVersion).isGreaterThanOrEqualTo(2);
+    Table table = createTablePartitioned(1, 1, 1);
     shouldHaveFiles(table, 1);
     table.refresh();
 
@@ -476,8 +523,14 @@ public void testBinPackWithDeleteAllData() {
     int total = (int) dataFiles.stream().mapToLong(ContentFile::recordCount).sum();
 
     RowDelta rowDelta = table.newRowDelta();
+    DataFile dataFile = dataFiles.get(0);
     // remove all data
-    writePosDeletesToFile(table, dataFiles.get(0), total).forEach(rowDelta::addDeletes);
+    if (formatVersion >= 3) {
+      writeDV(table, dataFile.partition(), dataFile.location(), total)
+          .forEach(rowDelta::addDeletes);
+    } else {
+      writePosDeletesToFile(table, dataFile, total).forEach(rowDelta::addDeletes);
+    }
 
     rowDelta.commit();
     table.refresh();
@@ -507,12 +560,12 @@ public void testBinPackWithDeleteAllData() {
         .isEqualTo(total);
   }
 
-  @Test
+  @TestTemplate
   public void testBinPackWithStartingSequenceNumber() {
+    assumeThat(formatVersion).isGreaterThanOrEqualTo(2);
     Table table = createTablePartitioned(4, 2);
     shouldHaveFiles(table, 8);
     List<Object[]> expectedRecords = currentData();
-    table.updateProperties().set(TableProperties.FORMAT_VERSION, "2").commit();
     table.refresh();
     long oldSequenceNumber = table.currentSnapshot().sequenceNumber();
     long dataSizeBefore = testDataSize(table);
@@ -544,7 +597,7 @@ public void testBinPackWithStartingSequenceNumber() {
     }
   }
 
-  @Test
+  @TestTemplate
   public void testBinPackWithStartingSequenceNumberV1Compatibility() {
     Map<String, String> properties = ImmutableMap.of(TableProperties.FORMAT_VERSION, "1");
     Table table = createTablePartitioned(4, 2, SCALE, properties);
@@ -580,11 +633,15 @@ public void testBinPackWithStartingSequenceNumberV1Compatibility() {
     }
   }
 
-  @Test
+  @TestTemplate
   public void testRewriteLargeTableHasResiduals() {
     PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).build();
-    Map<String, String> options = Maps.newHashMap();
-    options.put(TableProperties.PARQUET_ROW_GROUP_SIZE_BYTES, "100");
+    Map<String, String> options =
+        ImmutableMap.of(
+            TableProperties.FORMAT_VERSION,
+            String.valueOf(formatVersion),
+            TableProperties.PARQUET_ROW_GROUP_SIZE_BYTES,
+            "100");
     Table table = TABLES.create(SCHEMA, spec, options, tableLocation);
 
     // all records belong to the same partition
@@ -622,7 +679,7 @@ public void testRewriteLargeTableHasResiduals() {
     assertEquals("Rows must match", expectedRecords, actualRecords);
   }
 
-  @Test
+  @TestTemplate
   public void testBinPackSplitLargeFile() {
     Table table = createTable(1);
     shouldHaveFiles(table, 1);
@@ -647,7 +704,7 @@ public void testBinPackSplitLargeFile() {
     assertEquals("Rows must match", expectedRecords, actualRecords);
   }
 
-  @Test
+  @TestTemplate
   public void testBinPackCombineMixedFiles() {
     Table table = createTable(1); // 400000
     shouldHaveFiles(table, 1);
@@ -683,7 +740,7 @@ public void testBinPackCombineMixedFiles() {
     assertEquals("Rows must match", expectedRecords, actualRecords);
   }
 
-  @Test
+  @TestTemplate
   public void testBinPackCombineMediumFiles() {
     Table table = createTable(4);
     shouldHaveFiles(table, 4);
@@ -716,7 +773,7 @@ public void testBinPackCombineMediumFiles() {
     assertEquals("Rows must match", expectedRecords, actualRecords);
   }
 
-  @Test
+  @TestTemplate
   public void testPartialProgressEnabled() {
     Table table = createTable(20);
     int fileSize = averageFileSize(table);
@@ -747,7 +804,7 @@ public void testPartialProgressEnabled() {
     assertEquals("We shouldn't have changed the data", originalData, postRewriteData);
   }
 
-  @Test
+  @TestTemplate
   public void testMultipleGroups() {
     Table table = createTable(20);
     int fileSize = averageFileSize(table);
@@ -775,7 +832,7 @@ public void testMultipleGroups() {
     shouldHaveACleanCache(table);
   }
 
-  @Test
+  @TestTemplate
   public void testPartialProgressMaxCommits() {
     Table table = createTable(20);
     int fileSize = averageFileSize(table);
@@ -804,7 +861,7 @@ public void testPartialProgressMaxCommits() {
     shouldHaveACleanCache(table);
   }
 
-  @Test
+  @TestTemplate
   public void testSingleCommitWithRewriteFailure() {
     Table table = createTable(20);
     int fileSize = averageFileSize(table);
@@ -838,7 +895,7 @@ public void testSingleCommitWithRewriteFailure() {
     shouldHaveACleanCache(table);
   }
 
-  @Test
+  @TestTemplate
   public void testSingleCommitWithCommitFailure() {
     Table table = createTable(20);
     int fileSize = averageFileSize(table);
@@ -872,7 +929,7 @@ public void testSingleCommitWithCommitFailure() {
     shouldHaveACleanCache(table);
   }
 
-  @Test
+  @TestTemplate
   public void testCommitFailsWithUncleanableFailure() {
     Table table = createTable(20);
     int fileSize = averageFileSize(table);
@@ -906,7 +963,7 @@ public void testCommitFailsWithUncleanableFailure() {
     shouldHaveACleanCache(table);
   }
 
-  @Test
+  @TestTemplate
   public void testParallelSingleCommitWithRewriteFailure() {
     Table table = createTable(20);
     int fileSize = averageFileSize(table);
@@ -941,7 +998,7 @@ public void testParallelSingleCommitWithRewriteFailure() {
     shouldHaveACleanCache(table);
   }
 
-  @Test
+  @TestTemplate
   public void testPartialProgressWithRewriteFailure() {
     Table table = createTable(20);
     int fileSize = averageFileSize(table);
@@ -983,7 +1040,7 @@ public void testPartialProgressWithRewriteFailure() {
     shouldHaveACleanCache(table);
   }
 
-  @Test
+  @TestTemplate
   public void testParallelPartialProgressWithRewriteFailure() {
     Table table = createTable(20);
     int fileSize = averageFileSize(table);
@@ -1026,7 +1083,7 @@ public void testParallelPartialProgressWithRewriteFailure() {
     shouldHaveACleanCache(table);
   }
 
-  @Test
+  @TestTemplate
   public void testParallelPartialProgressWithCommitFailure() {
     Table table = createTable(20);
     int fileSize = averageFileSize(table);
@@ -1071,7 +1128,7 @@ public void testParallelPartialProgressWithCommitFailure() {
     shouldHaveACleanCache(table);
   }
 
-  @Test
+  @TestTemplate
   public void testParallelPartialProgressWithMaxFailedCommits() {
     Table table = createTable(20);
     int fileSize = averageFileSize(table);
@@ -1113,7 +1170,7 @@ public void testParallelPartialProgressWithMaxFailedCommits() {
     shouldHaveACleanCache(table);
   }
 
-  @Test
+  @TestTemplate
   public void testInvalidOptions() {
     Table table = createTable(20);
 
@@ -1157,7 +1214,7 @@ public void testInvalidOptions() {
         .hasMessageContaining("requires enabling Iceberg Spark session extensions");
   }
 
-  @Test
+  @TestTemplate
   public void testSortMultipleGroups() {
     Table table = createTable(20);
     shouldHaveFiles(table, 20);
@@ -1189,7 +1246,7 @@ public void testSortMultipleGroups() {
     shouldHaveACleanCache(table);
   }
 
-  @Test
+  @TestTemplate
   public void testSimpleSort() {
     Table table = createTable(20);
     shouldHaveFiles(table, 20);
@@ -1222,7 +1279,7 @@ public void testSimpleSort() {
     shouldHaveLastCommitSorted(table, "c2");
   }
 
-  @Test
+  @TestTemplate
   public void testSortAfterPartitionChange() {
     Table table = createTable(20);
     shouldHaveFiles(table, 20);
@@ -1258,7 +1315,7 @@ public void testSortAfterPartitionChange() {
     shouldHaveLastCommitSorted(table, "c2");
   }
 
-  @Test
+  @TestTemplate
   public void testSortCustomSortOrder() {
     Table table = createTable(20);
     shouldHaveLastCommitUnsorted(table, "c2");
@@ -1289,7 +1346,7 @@ public void testSortCustomSortOrder() {
     shouldHaveLastCommitSorted(table, "c2");
   }
 
-  @Test
+  @TestTemplate
   public void testSortCustomSortOrderRequiresRepartition() {
     int partitions = 4;
     Table table = createTable();
@@ -1329,7 +1386,7 @@ public void testSortCustomSortOrderRequiresRepartition() {
     shouldHaveLastCommitSorted(table, "c3");
   }
 
-  @Test
+  @TestTemplate
   public void testAutoSortShuffleOutput() {
     Table table = createTable(20);
     shouldHaveLastCommitUnsorted(table, "c2");
@@ -1369,7 +1426,7 @@ public void testAutoSortShuffleOutput() {
     shouldHaveLastCommitSorted(table, "c2");
   }
 
-  @Test
+  @TestTemplate
   public void testCommitStateUnknownException() {
     Table table = createTable(20);
     shouldHaveFiles(table, 20);
@@ -1401,7 +1458,7 @@ public void testCommitStateUnknownException() {
     shouldHaveSnapshots(table, 2); // Commit actually Succeeded
   }
 
-  @Test
+  @TestTemplate
   public void testZOrderSort() {
     int originalFiles = 20;
     Table table = createTable(originalFiles);
@@ -1461,7 +1518,7 @@ public void testZOrderSort() {
         .isGreaterThan(filesScannedC2C3);
   }
 
-  @Test
+  @TestTemplate
   public void testZOrderAllTypesSort() {
     Table table = createTypeTestTable();
     shouldHaveFiles(table, 10);
@@ -1505,7 +1562,7 @@ public void testZOrderAllTypesSort() {
     shouldHaveACleanCache(table);
   }
 
-  @Test
+  @TestTemplate
   public void testInvalidAPIUsage() {
     Table table = createTable(1);
 
@@ -1524,14 +1581,14 @@ public void testInvalidAPIUsage() {
         .hasMessage("Must use only one rewriter type (bin-pack, sort, zorder)");
   }
 
-  @Test
+  @TestTemplate
   public void testRewriteJobOrderBytesAsc() {
+    assumeThat(formatVersion).isGreaterThanOrEqualTo(2);
     Table table = createTablePartitioned(4, 2);
     writeRecords(1, SCALE, 1);
     writeRecords(2, SCALE, 2);
     writeRecords(3, SCALE, 3);
     writeRecords(4, SCALE, 4);
-    table.updateProperties().set(TableProperties.FORMAT_VERSION, "2").commit();
 
     RewriteDataFilesSparkAction basicRewrite = basicRewrite(table).binPack();
     List<Long> expected =
@@ -1556,14 +1613,14 @@ public void testRewriteJobOrderBytesAsc() {
     assertThat(actual).as("Size in bytes order should not be descending").isNotEqualTo(expected);
   }
 
-  @Test
+  @TestTemplate
   public void testRewriteJobOrderBytesDesc() {
+    assumeThat(formatVersion).isGreaterThanOrEqualTo(2);
     Table table = createTablePartitioned(4, 2);
     writeRecords(1, SCALE, 1);
     writeRecords(2, SCALE, 2);
     writeRecords(3, SCALE, 3);
     writeRecords(4, SCALE, 4);
-    table.updateProperties().set(TableProperties.FORMAT_VERSION, "2").commit();
 
     RewriteDataFilesSparkAction basicRewrite = basicRewrite(table).binPack();
     List<Long> expected =
@@ -1588,14 +1645,14 @@ public void testRewriteJobOrderBytesDesc() {
     assertThat(actual).as("Size in bytes order should not be ascending").isNotEqualTo(expected);
   }
 
-  @Test
+  @TestTemplate
   public void testRewriteJobOrderFilesAsc() {
+    assumeThat(formatVersion).isGreaterThanOrEqualTo(2);
     Table table = createTablePartitioned(4, 2);
     writeRecords(1, SCALE, 1);
     writeRecords(2, SCALE, 2);
     writeRecords(3, SCALE, 3);
     writeRecords(4, SCALE, 4);
-    table.updateProperties().set(TableProperties.FORMAT_VERSION, "2").commit();
 
     RewriteDataFilesSparkAction basicRewrite = basicRewrite(table).binPack();
     List<Long> expected =
@@ -1620,14 +1677,14 @@ public void testRewriteJobOrderFilesAsc() {
     assertThat(actual).as("Number of files order should not be descending").isNotEqualTo(expected);
   }
 
-  @Test
+  @TestTemplate
   public void testRewriteJobOrderFilesDesc() {
+    assumeThat(formatVersion).isGreaterThanOrEqualTo(2);
     Table table = createTablePartitioned(4, 2);
     writeRecords(1, SCALE, 1);
     writeRecords(2, SCALE, 2);
     writeRecords(3, SCALE, 3);
     writeRecords(4, SCALE, 4);
-    table.updateProperties().set(TableProperties.FORMAT_VERSION, "2").commit();
 
     RewriteDataFilesSparkAction basicRewrite = basicRewrite(table).binPack();
     List<Long> expected =
@@ -1652,7 +1709,7 @@ public void testRewriteJobOrderFilesDesc() {
     assertThat(actual).as("Number of files order should not be ascending").isNotEqualTo(expected);
   }
 
-  @Test
+  @TestTemplate
   public void testSnapshotProperty() {
     Table table = createTable(4);
     Result ignored = basicRewrite(table).snapshotProperty("key", "value").execute();
@@ -1669,7 +1726,7 @@ public void testSnapshotProperty() {
     assertThat(table.currentSnapshot().summary()).containsKeys(commitMetricsKeys);
   }
 
-  @Test
+  @TestTemplate
   public void testBinPackRewriterWithSpecificUnparitionedOutputSpec() {
     Table table = createTable(10);
     shouldHaveFiles(table, 10);
@@ -1691,7 +1748,7 @@ public void testBinPackRewriterWithSpecificUnparitionedOutputSpec() {
     shouldRewriteDataFilesWithPartitionSpec(table, outputSpecId);
   }
 
-  @Test
+  @TestTemplate
   public void testBinPackRewriterWithSpecificOutputSpec() {
     Table table = createTable(10);
     shouldHaveFiles(table, 10);
@@ -1714,7 +1771,7 @@ public void testBinPackRewriterWithSpecificOutputSpec() {
     shouldRewriteDataFilesWithPartitionSpec(table, outputSpecId);
   }
 
-  @Test
+  @TestTemplate
   public void testBinpackRewriteWithInvalidOutputSpecId() {
     Table table = createTable(10);
     shouldHaveFiles(table, 10);
@@ -1730,7 +1787,7 @@ public void testBinpackRewriteWithInvalidOutputSpecId() {
             "Cannot use output spec id 1234 because the table does not contain a reference to this spec-id.");
   }
 
-  @Test
+  @TestTemplate
   public void testSortRewriterWithSpecificOutputSpecId() {
     Table table = createTable(10);
     shouldHaveFiles(table, 10);
@@ -1753,7 +1810,7 @@ public void testSortRewriterWithSpecificOutputSpecId() {
     shouldRewriteDataFilesWithPartitionSpec(table, outputSpecId);
   }
 
-  @Test
+  @TestTemplate
   public void testZOrderRewriteWithSpecificOutputSpecId() {
     Table table = createTable(10);
     shouldHaveFiles(table, 10);
@@ -1956,7 +2013,8 @@ private <T> List<Pair<Pair<T, T>, Pair<T, T>>> checkForOverlappingFiles(
 
   protected Table createTable() {
     PartitionSpec spec = PartitionSpec.unpartitioned();
-    Map<String, String> options = Maps.newHashMap();
+    Map<String, String> options =
+        ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion));
     Table table = TABLES.create(SCHEMA, spec, options, tableLocation);
     table
         .updateProperties()
@@ -1989,7 +2047,19 @@ protected Table createTablePartitioned(
   }
 
   protected Table createTablePartitioned(int partitions, int files) {
-    return createTablePartitioned(partitions, files, SCALE, Maps.newHashMap());
+    return createTablePartitioned(
+        partitions,
+        files,
+        SCALE,
+        ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion)));
+  }
+
+  protected Table createTablePartitioned(int partitions, int files, int numRecords) {
+    return createTablePartitioned(
+        partitions,
+        files,
+        numRecords,
+        ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion)));
   }
 
   private Table createTypeTestTable() {
@@ -2005,7 +2075,8 @@ private Table createTypeTestTable() {
             optional(8, "booleanCol", Types.BooleanType.get()),
             optional(9, "binaryCol", Types.BinaryType.get()));
 
-    Map<String, String> options = Maps.newHashMap();
+    Map<String, String> options =
+        ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion));
     Table table = TABLES.create(schema, PartitionSpec.unpartitioned(), options, tableLocation);
 
     spark
@@ -2123,6 +2194,20 @@ private List<DeleteFile> writePosDeletes(
     return results;
   }
 
+  private List<DeleteFile> writeDV(
+      Table table, StructLike partition, String path, int numPositionsToDelete) throws IOException {
+    OutputFileFactory fileFactory =
+        OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build();
+    DVFileWriter writer = new BaseDVFileWriter(fileFactory, p -> null);
+    try (DVFileWriter closeableWriter = writer) {
+      for (int row = 0; row < numPositionsToDelete; row++) {
+        closeableWriter.delete(path, row, table.spec(), partition);
+      }
+    }
+
+    return writer.result().deleteFiles();
+  }
+
   private void writeEqDeleteRecord(
       Table table, String partCol, Object partVal, String delCol, Object delVal) {
     List<Integer> equalityFieldIds = Lists.newArrayList(table.schema().findField(delCol).fieldId());
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java
index 11d61e599eba..44971843547b 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java
@@ -37,6 +37,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
@@ -107,6 +108,7 @@ public static Object[] parameters() {
       new Object[] {"false", "true", true, 1},
       new Object[] {"true", "false", false, 2},
       new Object[] {"false", "false", false, 2},
+      new Object[] {"true", "false", false, 3},
       new Object[] {"false", "false", false, 3}
     };
   }
@@ -1108,13 +1110,23 @@ private DeleteFile newDV(Table table, DataFile dataFile) {
   }
 
   private DeleteFile newDeleteFile(Table table, String partitionPath) {
-    return FileMetadata.deleteFileBuilder(table.spec())
-        .ofPositionDeletes()
-        .withPath("/path/to/pos-deletes-" + UUID.randomUUID() + ".parquet")
-        .withFileSizeInBytes(5)
-        .withPartitionPath(partitionPath)
-        .withRecordCount(1)
-        .build();
+    return formatVersion >= 3
+        ? FileMetadata.deleteFileBuilder(table.spec())
+            .ofPositionDeletes()
+            .withPath("/path/to/pos-deletes-" + UUID.randomUUID() + ".puffin")
+            .withFileSizeInBytes(5)
+            .withPartitionPath(partitionPath)
+            .withRecordCount(1)
+            .withContentOffset(ThreadLocalRandom.current().nextInt())
+            .withContentSizeInBytes(ThreadLocalRandom.current().nextInt())
+            .build()
+        : FileMetadata.deleteFileBuilder(table.spec())
+            .ofPositionDeletes()
+            .withPath("/path/to/pos-deletes-" + UUID.randomUUID() + ".parquet")
+            .withFileSizeInBytes(5)
+            .withPartitionPath(partitionPath)
+            .withRecordCount(1)
+            .build();
   }
 
   private List<Pair<CharSequence, Long>> generatePosDeletes(String predicate) {
@@ -1145,7 +1157,7 @@ private Pair<DeleteFile, CharSequenceSet> writePosDeletes(
       Table table, StructLike partition, List<Pair<CharSequence, Long>> deletes)
       throws IOException {
     OutputFile outputFile = Files.localOutput(File.createTempFile("junit", null, temp.toFile()));
-    return FileHelpers.writeDeleteFile(table, outputFile, partition, deletes);
+    return FileHelpers.writeDeleteFile(table, outputFile, partition, deletes, formatVersion);
   }
 
   private DeleteFile writeEqDeletes(Table table, String key, Object... values) throws IOException {