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/SnapshotSummary.java b/core/src/main/java/org/apache/iceberg/SnapshotSummary.java
index ad832a5e78e2..6043424cd7fc 100644
--- a/core/src/main/java/org/apache/iceberg/SnapshotSummary.java
+++ b/core/src/main/java/org/apache/iceberg/SnapshotSummary.java
@@ -25,6 +25,7 @@
 import org.apache.iceberg.relocated.com.google.common.base.Strings;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.ContentFileUtil;
 import org.apache.iceberg.util.ScanTaskUtil;
 
 public class SnapshotSummary {
@@ -36,6 +37,8 @@ public class SnapshotSummary {
   public static final String REMOVED_EQ_DELETE_FILES_PROP = "removed-equality-delete-files";
   public static final String ADD_POS_DELETE_FILES_PROP = "added-position-delete-files";
   public static final String REMOVED_POS_DELETE_FILES_PROP = "removed-position-delete-files";
+  public static final String ADDED_DVS_PROP = "added-dvs";
+  public static final String REMOVED_DVS_PROP = "removed-dvs";
   public static final String REMOVED_DELETE_FILES_PROP = "removed-delete-files";
   public static final String TOTAL_DELETE_FILES_PROP = "total-delete-files";
   public static final String ADDED_RECORDS_PROP = "added-records";
@@ -222,6 +225,8 @@ private static class UpdateMetrics {
     private int removedEqDeleteFiles = 0;
     private int addedPosDeleteFiles = 0;
     private int removedPosDeleteFiles = 0;
+    private int addedDVs = 0;
+    private int removedDVs = 0;
     private int addedDeleteFiles = 0;
     private int removedDeleteFiles = 0;
     private long addedRecords = 0L;
@@ -243,6 +248,8 @@ void clear() {
       this.removedPosDeleteFiles = 0;
       this.addedDeleteFiles = 0;
       this.removedDeleteFiles = 0;
+      this.addedDVs = 0;
+      this.removedDVs = 0;
       this.addedRecords = 0L;
       this.deletedRecords = 0L;
       this.addedPosDeletes = 0L;
@@ -262,6 +269,8 @@ void addTo(ImmutableMap.Builder<String, String> builder) {
           removedPosDeleteFiles > 0, builder, REMOVED_POS_DELETE_FILES_PROP, removedPosDeleteFiles);
       setIf(addedDeleteFiles > 0, builder, ADDED_DELETE_FILES_PROP, addedDeleteFiles);
       setIf(removedDeleteFiles > 0, builder, REMOVED_DELETE_FILES_PROP, removedDeleteFiles);
+      setIf(addedDVs > 0, builder, ADDED_DVS_PROP, addedDVs);
+      setIf(removedDVs > 0, builder, REMOVED_DVS_PROP, removedDVs);
       setIf(addedRecords > 0, builder, ADDED_RECORDS_PROP, addedRecords);
       setIf(deletedRecords > 0, builder, DELETED_RECORDS_PROP, deletedRecords);
 
@@ -283,8 +292,13 @@ void addedFile(ContentFile<?> file) {
           this.addedRecords += file.recordCount();
           break;
         case POSITION_DELETES:
+          DeleteFile deleteFile = (DeleteFile) file;
+          if (ContentFileUtil.isDV(deleteFile)) {
+            this.addedDVs += 1;
+          } else {
+            this.addedPosDeleteFiles += 1;
+          }
           this.addedDeleteFiles += 1;
-          this.addedPosDeleteFiles += 1;
           this.addedPosDeletes += file.recordCount();
           break;
         case EQUALITY_DELETES:
@@ -306,8 +320,13 @@ void removedFile(ContentFile<?> file) {
           this.deletedRecords += file.recordCount();
           break;
         case POSITION_DELETES:
+          DeleteFile deleteFile = (DeleteFile) file;
+          if (ContentFileUtil.isDV(deleteFile)) {
+            this.removedDVs += 1;
+          } else {
+            this.removedPosDeleteFiles += 1;
+          }
           this.removedDeleteFiles += 1;
-          this.removedPosDeleteFiles += 1;
           this.removedPosDeletes += file.recordCount();
           break;
         case EQUALITY_DELETES:
@@ -344,6 +363,8 @@ void merge(UpdateMetrics other) {
       this.removedEqDeleteFiles += other.removedEqDeleteFiles;
       this.addedPosDeleteFiles += other.addedPosDeleteFiles;
       this.removedPosDeleteFiles += other.removedPosDeleteFiles;
+      this.addedDVs += other.addedDVs;
+      this.removedDVs += other.removedDVs;
       this.addedDeleteFiles += other.addedDeleteFiles;
       this.removedDeleteFiles += other.removedDeleteFiles;
       this.addedSize += other.addedSize;
diff --git a/core/src/main/java/org/apache/iceberg/metrics/CommitMetricsResult.java b/core/src/main/java/org/apache/iceberg/metrics/CommitMetricsResult.java
index ad66e8d32408..7a87172708f6 100644
--- a/core/src/main/java/org/apache/iceberg/metrics/CommitMetricsResult.java
+++ b/core/src/main/java/org/apache/iceberg/metrics/CommitMetricsResult.java
@@ -34,7 +34,9 @@ public interface CommitMetricsResult {
   String ADDED_DELETE_FILES = "added-delete-files";
   String ADDED_EQ_DELETE_FILES = "added-equality-delete-files";
   String ADDED_POS_DELETE_FILES = "added-positional-delete-files";
+  String ADDED_DVS = "added-dvs";
   String REMOVED_POS_DELETE_FILES = "removed-positional-delete-files";
+  String REMOVED_DVS = "removed-dvs";
   String REMOVED_EQ_DELETE_FILES = "removed-equality-delete-files";
   String REMOVED_DELETE_FILES = "removed-delete-files";
   String TOTAL_DELETE_FILES = "total-delete-files";
@@ -75,6 +77,12 @@ public interface CommitMetricsResult {
   @Nullable
   CounterResult addedPositionalDeleteFiles();
 
+  @Nullable
+  @Value.Default
+  default CounterResult addedDVs() {
+    return null;
+  }
+
   @Nullable
   CounterResult removedDeleteFiles();
 
@@ -84,6 +92,12 @@ public interface CommitMetricsResult {
   @Nullable
   CounterResult removedPositionalDeleteFiles();
 
+  @Nullable
+  @Value.Default
+  default CounterResult removedDVs() {
+    return null;
+  }
+
   @Nullable
   CounterResult totalDeleteFiles();
 
@@ -136,6 +150,7 @@ static CommitMetricsResult from(
         .addedDeleteFiles(counterFrom(snapshotSummary, SnapshotSummary.ADDED_DELETE_FILES_PROP))
         .addedPositionalDeleteFiles(
             counterFrom(snapshotSummary, SnapshotSummary.ADD_POS_DELETE_FILES_PROP))
+        .addedDVs(counterFrom(snapshotSummary, SnapshotSummary.ADDED_DVS_PROP))
         .addedEqualityDeleteFiles(
             counterFrom(snapshotSummary, SnapshotSummary.ADD_EQ_DELETE_FILES_PROP))
         .removedDeleteFiles(counterFrom(snapshotSummary, SnapshotSummary.REMOVED_DELETE_FILES_PROP))
@@ -143,6 +158,7 @@ static CommitMetricsResult from(
             counterFrom(snapshotSummary, SnapshotSummary.REMOVED_EQ_DELETE_FILES_PROP))
         .removedPositionalDeleteFiles(
             counterFrom(snapshotSummary, SnapshotSummary.REMOVED_POS_DELETE_FILES_PROP))
+        .removedDVs(counterFrom(snapshotSummary, SnapshotSummary.REMOVED_DVS_PROP))
         .totalDeleteFiles(counterFrom(snapshotSummary, SnapshotSummary.TOTAL_DELETE_FILES_PROP))
         .addedRecords(counterFrom(snapshotSummary, SnapshotSummary.ADDED_RECORDS_PROP))
         .removedRecords(counterFrom(snapshotSummary, SnapshotSummary.DELETED_RECORDS_PROP))
diff --git a/core/src/main/java/org/apache/iceberg/metrics/CommitMetricsResultParser.java b/core/src/main/java/org/apache/iceberg/metrics/CommitMetricsResultParser.java
index d4fd883c4375..2c45581ba5d6 100644
--- a/core/src/main/java/org/apache/iceberg/metrics/CommitMetricsResultParser.java
+++ b/core/src/main/java/org/apache/iceberg/metrics/CommitMetricsResultParser.java
@@ -81,6 +81,11 @@ static void toJson(CommitMetricsResult metrics, JsonGenerator gen) throws IOExce
       CounterResultParser.toJson(metrics.addedPositionalDeleteFiles(), gen);
     }
 
+    if (null != metrics.addedDVs()) {
+      gen.writeFieldName(CommitMetricsResult.ADDED_DVS);
+      CounterResultParser.toJson(metrics.addedDVs(), gen);
+    }
+
     if (null != metrics.removedDeleteFiles()) {
       gen.writeFieldName(CommitMetricsResult.REMOVED_DELETE_FILES);
       CounterResultParser.toJson(metrics.removedDeleteFiles(), gen);
@@ -91,6 +96,11 @@ static void toJson(CommitMetricsResult metrics, JsonGenerator gen) throws IOExce
       CounterResultParser.toJson(metrics.removedPositionalDeleteFiles(), gen);
     }
 
+    if (null != metrics.removedDVs()) {
+      gen.writeFieldName(CommitMetricsResult.REMOVED_DVS);
+      CounterResultParser.toJson(metrics.removedDVs(), gen);
+    }
+
     if (null != metrics.removedEqualityDeleteFiles()) {
       gen.writeFieldName(CommitMetricsResult.REMOVED_EQ_DELETE_FILES);
       CounterResultParser.toJson(metrics.removedEqualityDeleteFiles(), gen);
@@ -186,10 +196,12 @@ static CommitMetricsResult fromJson(JsonNode json) {
             CounterResultParser.fromJson(CommitMetricsResult.ADDED_EQ_DELETE_FILES, json))
         .addedPositionalDeleteFiles(
             CounterResultParser.fromJson(CommitMetricsResult.ADDED_POS_DELETE_FILES, json))
+        .addedDVs(CounterResultParser.fromJson(CommitMetricsResult.ADDED_DVS, json))
         .removedEqualityDeleteFiles(
             CounterResultParser.fromJson(CommitMetricsResult.REMOVED_EQ_DELETE_FILES, json))
         .removedPositionalDeleteFiles(
             CounterResultParser.fromJson(CommitMetricsResult.REMOVED_POS_DELETE_FILES, json))
+        .removedDVs(CounterResultParser.fromJson(CommitMetricsResult.REMOVED_DVS, json))
         .removedDeleteFiles(
             CounterResultParser.fromJson(CommitMetricsResult.REMOVED_DELETE_FILES, json))
         .totalDeleteFiles(
diff --git a/core/src/main/java/org/apache/iceberg/metrics/ScanMetrics.java b/core/src/main/java/org/apache/iceberg/metrics/ScanMetrics.java
index 421466f0fa85..0f7def37638e 100644
--- a/core/src/main/java/org/apache/iceberg/metrics/ScanMetrics.java
+++ b/core/src/main/java/org/apache/iceberg/metrics/ScanMetrics.java
@@ -40,6 +40,7 @@ public abstract class ScanMetrics {
   public static final String INDEXED_DELETE_FILES = "indexed-delete-files";
   public static final String EQUALITY_DELETE_FILES = "equality-delete-files";
   public static final String POSITIONAL_DELETE_FILES = "positional-delete-files";
+  public static final String DVS = "dvs";
 
   public static ScanMetrics noop() {
     return ScanMetrics.of(MetricsContext.nullMetrics());
@@ -127,6 +128,11 @@ public Counter positionalDeleteFiles() {
     return metricsContext().counter(POSITIONAL_DELETE_FILES);
   }
 
+  @Value.Derived
+  public Counter dvs() {
+    return metricsContext().counter(DVS);
+  }
+
   public static ScanMetrics of(MetricsContext metricsContext) {
     return ImmutableScanMetrics.builder().metricsContext(metricsContext).build();
   }
diff --git a/core/src/main/java/org/apache/iceberg/metrics/ScanMetricsResult.java b/core/src/main/java/org/apache/iceberg/metrics/ScanMetricsResult.java
index b930dd83adef..2137e52e0a89 100644
--- a/core/src/main/java/org/apache/iceberg/metrics/ScanMetricsResult.java
+++ b/core/src/main/java/org/apache/iceberg/metrics/ScanMetricsResult.java
@@ -73,6 +73,12 @@ public interface ScanMetricsResult {
   @Nullable
   CounterResult positionalDeleteFiles();
 
+  @Nullable
+  @Value.Default
+  default CounterResult dvs() {
+    return null;
+  }
+
   static ScanMetricsResult fromScanMetrics(ScanMetrics scanMetrics) {
     Preconditions.checkArgument(null != scanMetrics, "Invalid scan metrics: null");
     return ImmutableScanMetricsResult.builder()
@@ -93,6 +99,7 @@ static ScanMetricsResult fromScanMetrics(ScanMetrics scanMetrics) {
         .indexedDeleteFiles(CounterResult.fromCounter(scanMetrics.indexedDeleteFiles()))
         .equalityDeleteFiles(CounterResult.fromCounter(scanMetrics.equalityDeleteFiles()))
         .positionalDeleteFiles(CounterResult.fromCounter(scanMetrics.positionalDeleteFiles()))
+        .dvs(CounterResult.fromCounter(scanMetrics.dvs()))
         .build();
   }
 }
diff --git a/core/src/main/java/org/apache/iceberg/metrics/ScanMetricsResultParser.java b/core/src/main/java/org/apache/iceberg/metrics/ScanMetricsResultParser.java
index 5cff1ae8e0db..f85c26753211 100644
--- a/core/src/main/java/org/apache/iceberg/metrics/ScanMetricsResultParser.java
+++ b/core/src/main/java/org/apache/iceberg/metrics/ScanMetricsResultParser.java
@@ -121,6 +121,11 @@ static void toJson(ScanMetricsResult metrics, JsonGenerator gen) throws IOExcept
       CounterResultParser.toJson(metrics.positionalDeleteFiles(), gen);
     }
 
+    if (null != metrics.dvs()) {
+      gen.writeFieldName(ScanMetrics.DVS);
+      CounterResultParser.toJson(metrics.dvs(), gen);
+    }
+
     gen.writeEndObject();
   }
 
@@ -159,6 +164,7 @@ static ScanMetricsResult fromJson(JsonNode json) {
         .equalityDeleteFiles(CounterResultParser.fromJson(ScanMetrics.EQUALITY_DELETE_FILES, json))
         .positionalDeleteFiles(
             CounterResultParser.fromJson(ScanMetrics.POSITIONAL_DELETE_FILES, json))
+        .dvs(CounterResultParser.fromJson(ScanMetrics.DVS, json))
         .build();
   }
 }
diff --git a/core/src/main/java/org/apache/iceberg/metrics/ScanMetricsUtil.java b/core/src/main/java/org/apache/iceberg/metrics/ScanMetricsUtil.java
index 1ba891f58474..6e6aa25636bd 100644
--- a/core/src/main/java/org/apache/iceberg/metrics/ScanMetricsUtil.java
+++ b/core/src/main/java/org/apache/iceberg/metrics/ScanMetricsUtil.java
@@ -21,6 +21,7 @@
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.FileContent;
+import org.apache.iceberg.util.ContentFileUtil;
 import org.apache.iceberg.util.ScanTaskUtil;
 
 public class ScanMetricsUtil {
@@ -31,7 +32,11 @@ public static void indexedDeleteFile(ScanMetrics metrics, DeleteFile deleteFile)
     metrics.indexedDeleteFiles().increment();
 
     if (deleteFile.content() == FileContent.POSITION_DELETES) {
-      metrics.positionalDeleteFiles().increment();
+      if (ContentFileUtil.isDV(deleteFile)) {
+        metrics.dvs().increment();
+      } else {
+        metrics.positionalDeleteFiles().increment();
+      }
     } else if (deleteFile.content() == FileContent.EQUALITY_DELETES) {
       metrics.equalityDeleteFiles().increment();
     }


Test patch:
diff --git a/core/src/test/java/org/apache/iceberg/TestSnapshotSummary.java b/core/src/test/java/org/apache/iceberg/TestSnapshotSummary.java
index 529e0cc614f6..b0b9d003e35b 100644
--- a/core/src/test/java/org/apache/iceberg/TestSnapshotSummary.java
+++ b/core/src/test/java/org/apache/iceberg/TestSnapshotSummary.java
@@ -358,4 +358,76 @@ public void rewriteWithDeletesAndDuplicates() {
         .containsEntry(SnapshotSummary.TOTAL_FILE_SIZE_PROP, "20")
         .containsEntry(SnapshotSummary.TOTAL_RECORDS_PROP, "1");
   }
+
+  @TestTemplate
+  public void testFileSizeSummaryWithDVs() {
+    assumeThat(formatVersion).isGreaterThanOrEqualTo(3);
+
+    DeleteFile dv1 = newDV(FILE_A);
+    table.newRowDelta().addDeletes(dv1).commit();
+
+    DeleteFile dv2 = newDV(FILE_B);
+    table.newRowDelta().addDeletes(dv2).commit();
+
+    Map<String, String> summary1 = table.currentSnapshot().summary();
+    long addedPosDeletes1 = dv2.recordCount();
+    long addedFileSize1 = dv2.contentSizeInBytes();
+    long totalPosDeletes1 = dv1.recordCount() + dv2.recordCount();
+    long totalFileSize1 = dv1.contentSizeInBytes() + dv2.contentSizeInBytes();
+    assertThat(summary1)
+        .hasSize(12)
+        .doesNotContainKey(SnapshotSummary.ADD_POS_DELETE_FILES_PROP)
+        .doesNotContainKey(SnapshotSummary.REMOVED_POS_DELETE_FILES_PROP)
+        .containsEntry(SnapshotSummary.ADDED_DELETE_FILES_PROP, "1")
+        .doesNotContainKey(SnapshotSummary.REMOVED_DELETE_FILES_PROP)
+        .containsEntry(SnapshotSummary.ADDED_DVS_PROP, "1")
+        .doesNotContainKey(SnapshotSummary.REMOVED_DVS_PROP)
+        .containsEntry(SnapshotSummary.ADDED_POS_DELETES_PROP, String.valueOf(addedPosDeletes1))
+        .doesNotContainKey(SnapshotSummary.REMOVED_POS_DELETES_PROP)
+        .containsEntry(SnapshotSummary.ADDED_FILE_SIZE_PROP, String.valueOf(addedFileSize1))
+        .doesNotContainKey(SnapshotSummary.REMOVED_FILE_SIZE_PROP)
+        .containsEntry(SnapshotSummary.TOTAL_DELETE_FILES_PROP, "2")
+        .containsEntry(SnapshotSummary.TOTAL_POS_DELETES_PROP, String.valueOf(totalPosDeletes1))
+        .containsEntry(SnapshotSummary.TOTAL_FILE_SIZE_PROP, String.valueOf(totalFileSize1))
+        .containsEntry(SnapshotSummary.TOTAL_DATA_FILES_PROP, "0")
+        .containsEntry(SnapshotSummary.TOTAL_EQ_DELETES_PROP, "0")
+        .containsEntry(SnapshotSummary.TOTAL_RECORDS_PROP, "0")
+        .containsEntry(SnapshotSummary.CHANGED_PARTITION_COUNT_PROP, "1");
+
+    DeleteFile dv3 = newDV(FILE_A);
+    table
+        .newRowDelta()
+        .removeDeletes(dv1)
+        .removeDeletes(dv2)
+        .addDeletes(dv3)
+        .validateFromSnapshot(table.currentSnapshot().snapshotId())
+        .commit();
+
+    Map<String, String> summary2 = table.currentSnapshot().summary();
+    long addedPosDeletes2 = dv3.recordCount();
+    long removedPosDeletes2 = dv1.recordCount() + dv2.recordCount();
+    long addedFileSize2 = dv3.contentSizeInBytes();
+    long removedFileSize2 = dv1.contentSizeInBytes() + dv2.contentSizeInBytes();
+    long totalPosDeletes2 = dv3.recordCount();
+    long totalFileSize2 = dv3.contentSizeInBytes();
+    assertThat(summary2)
+        .hasSize(16)
+        .doesNotContainKey(SnapshotSummary.ADD_POS_DELETE_FILES_PROP)
+        .doesNotContainKey(SnapshotSummary.REMOVED_POS_DELETE_FILES_PROP)
+        .containsEntry(SnapshotSummary.ADDED_DELETE_FILES_PROP, "1")
+        .containsEntry(SnapshotSummary.REMOVED_DELETE_FILES_PROP, "2")
+        .containsEntry(SnapshotSummary.ADDED_DVS_PROP, "1")
+        .containsEntry(SnapshotSummary.REMOVED_DVS_PROP, "2")
+        .containsEntry(SnapshotSummary.ADDED_POS_DELETES_PROP, String.valueOf(addedPosDeletes2))
+        .containsEntry(SnapshotSummary.REMOVED_POS_DELETES_PROP, String.valueOf(removedPosDeletes2))
+        .containsEntry(SnapshotSummary.ADDED_FILE_SIZE_PROP, String.valueOf(addedFileSize2))
+        .containsEntry(SnapshotSummary.REMOVED_FILE_SIZE_PROP, String.valueOf(removedFileSize2))
+        .containsEntry(SnapshotSummary.TOTAL_DELETE_FILES_PROP, "1")
+        .containsEntry(SnapshotSummary.TOTAL_POS_DELETES_PROP, String.valueOf(totalPosDeletes2))
+        .containsEntry(SnapshotSummary.TOTAL_FILE_SIZE_PROP, String.valueOf(totalFileSize2))
+        .containsEntry(SnapshotSummary.TOTAL_DATA_FILES_PROP, "0")
+        .containsEntry(SnapshotSummary.TOTAL_EQ_DELETES_PROP, "0")
+        .containsEntry(SnapshotSummary.TOTAL_RECORDS_PROP, "0")
+        .containsEntry(SnapshotSummary.CHANGED_PARTITION_COUNT_PROP, "2");
+  }
 }
diff --git a/core/src/test/java/org/apache/iceberg/metrics/TestCommitMetricsResultParser.java b/core/src/test/java/org/apache/iceberg/metrics/TestCommitMetricsResultParser.java
index 5aa2660143a4..1b51066cf15c 100644
--- a/core/src/test/java/org/apache/iceberg/metrics/TestCommitMetricsResultParser.java
+++ b/core/src/test/java/org/apache/iceberg/metrics/TestCommitMetricsResultParser.java
@@ -74,6 +74,8 @@ public void roundTripSerde() {
             .put(SnapshotSummary.ADDED_DELETE_FILES_PROP, "4")
             .put(SnapshotSummary.ADD_EQ_DELETE_FILES_PROP, "5")
             .put(SnapshotSummary.ADD_POS_DELETE_FILES_PROP, "6")
+            .put(SnapshotSummary.ADDED_DVS_PROP, "1")
+            .put(SnapshotSummary.REMOVED_DVS_PROP, "4")
             .put(SnapshotSummary.REMOVED_POS_DELETE_FILES_PROP, "7")
             .put(SnapshotSummary.REMOVED_EQ_DELETE_FILES_PROP, "8")
             .put(SnapshotSummary.REMOVED_DELETE_FILES_PROP, "9")
@@ -101,6 +103,8 @@ public void roundTripSerde() {
     assertThat(result.addedDeleteFiles().value()).isEqualTo(4L);
     assertThat(result.addedEqualityDeleteFiles().value()).isEqualTo(5L);
     assertThat(result.addedPositionalDeleteFiles().value()).isEqualTo(6L);
+    assertThat(result.addedDVs().value()).isEqualTo(1L);
+    assertThat(result.removedDVs().value()).isEqualTo(4L);
     assertThat(result.removedPositionalDeleteFiles().value()).isEqualTo(7L);
     assertThat(result.removedEqualityDeleteFiles().value()).isEqualTo(8L);
     assertThat(result.removedDeleteFiles().value()).isEqualTo(9L);
@@ -153,6 +157,10 @@ public void roundTripSerde() {
             + "    \"unit\" : \"count\",\n"
             + "    \"value\" : 6\n"
             + "  },\n"
+            + "  \"added-dvs\" : {\n"
+            + "    \"unit\" : \"count\",\n"
+            + "    \"value\" : 1\n"
+            + "  },\n"
             + "  \"removed-delete-files\" : {\n"
             + "    \"unit\" : \"count\",\n"
             + "    \"value\" : 9\n"
@@ -161,6 +169,10 @@ public void roundTripSerde() {
             + "    \"unit\" : \"count\",\n"
             + "    \"value\" : 7\n"
             + "  },\n"
+            + "  \"removed-dvs\" : {\n"
+            + "    \"unit\" : \"count\",\n"
+            + "    \"value\" : 4\n"
+            + "  },\n"
             + "  \"removed-equality-delete-files\" : {\n"
             + "    \"unit\" : \"count\",\n"
             + "    \"value\" : 8\n"
diff --git a/core/src/test/java/org/apache/iceberg/metrics/TestScanMetricsResultParser.java b/core/src/test/java/org/apache/iceberg/metrics/TestScanMetricsResultParser.java
index 44d5803c4a3a..f5cb1e237307 100644
--- a/core/src/test/java/org/apache/iceberg/metrics/TestScanMetricsResultParser.java
+++ b/core/src/test/java/org/apache/iceberg/metrics/TestScanMetricsResultParser.java
@@ -178,6 +178,7 @@ public void extraFields() {
     scanMetrics.skippedDeleteManifests().increment(3L);
     scanMetrics.indexedDeleteFiles().increment(10L);
     scanMetrics.positionalDeleteFiles().increment(6L);
+    scanMetrics.dvs().increment();
     scanMetrics.equalityDeleteFiles().increment(4L);
 
     ScanMetricsResult scanMetricsResult = ScanMetricsResult.fromScanMetrics(scanMetrics);
@@ -199,6 +200,7 @@ public void extraFields() {
                     + "\"indexed-delete-files\":{\"unit\":\"count\",\"value\":10},"
                     + "\"equality-delete-files\":{\"unit\":\"count\",\"value\":4},"
                     + "\"positional-delete-files\":{\"unit\":\"count\",\"value\":6},"
+                    + "\"dvs\":{\"unit\":\"count\",\"value\":1},"
                     + "\"extra\": \"value\",\"extra2\":23}"))
         .isEqualTo(scanMetricsResult);
   }
@@ -242,6 +244,7 @@ public void roundTripSerde() {
     scanMetrics.skippedDeleteManifests().increment(3L);
     scanMetrics.indexedDeleteFiles().increment(10L);
     scanMetrics.positionalDeleteFiles().increment(6L);
+    scanMetrics.dvs().increment(3L);
     scanMetrics.equalityDeleteFiles().increment(4L);
 
     ScanMetricsResult scanMetricsResult = ScanMetricsResult.fromScanMetrics(scanMetrics);
@@ -312,6 +315,10 @@ public void roundTripSerde() {
             + "  \"positional-delete-files\" : {\n"
             + "    \"unit\" : \"count\",\n"
             + "    \"value\" : 6\n"
+            + "  },\n"
+            + "  \"dvs\" : {\n"
+            + "    \"unit\" : \"count\",\n"
+            + "    \"value\" : 3\n"
             + "  }\n"
             + "}";
 
diff --git a/core/src/test/java/org/apache/iceberg/metrics/TestScanReportParser.java b/core/src/test/java/org/apache/iceberg/metrics/TestScanReportParser.java
index 51e21ad9bf01..caac0704cd3f 100644
--- a/core/src/test/java/org/apache/iceberg/metrics/TestScanReportParser.java
+++ b/core/src/test/java/org/apache/iceberg/metrics/TestScanReportParser.java
@@ -84,6 +84,7 @@ public void extraFields() {
     scanMetrics.skippedDeleteManifests().increment(3L);
     scanMetrics.indexedDeleteFiles().increment(10L);
     scanMetrics.positionalDeleteFiles().increment(6L);
+    scanMetrics.dvs().increment();
     scanMetrics.equalityDeleteFiles().increment(4L);
 
     String tableName = "roundTripTableName";
@@ -118,6 +119,7 @@ public void extraFields() {
                     + "\"indexed-delete-files\":{\"unit\":\"count\",\"value\":10},"
                     + "\"equality-delete-files\":{\"unit\":\"count\",\"value\":4},"
                     + "\"positional-delete-files\":{\"unit\":\"count\",\"value\":6},"
+                    + "\"dvs\":{\"unit\":\"count\",\"value\":1},"
                     + "\"extra-metric\":\"extra-val\"},"
                     + "\"extra\":\"extraVal\"}"))
         .isEqualTo(scanReport);
@@ -279,6 +281,10 @@ public void roundTripSerde() {
             + "    \"positional-delete-files\" : {\n"
             + "      \"unit\" : \"count\",\n"
             + "      \"value\" : 6\n"
+            + "    },\n"
+            + "    \"dvs\" : {\n"
+            + "      \"unit\" : \"count\",\n"
+            + "      \"value\" : 0\n"
             + "    }\n"
             + "  }\n"
             + "}";