(iceberg) branch main updated: Core: Validate v2 deletes against concurrent format upgrade (#16146)

[am...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

amogh-jahagirdar pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/main by this push:
     new 099ef477c8 Core: Validate v2 deletes against concurrent format upgrade (#16146)
099ef477c8 is described below

commit 099ef477c8024b3bcb4d12abca18944c25b0eb45
Author: Hongyue/Steve Zhang <ho...@gmail.com>
AuthorDate: Tue Apr 28 20:11:36 2026 -0700

    Core: Validate v2 deletes against concurrent format upgrade (#16146)
    
    * Core: validate buffered v2 deletes against concurrent format upgrade
    
    * rename to validateDeleteFilesForVersion
---
 .../org/apache/iceberg/MergingSnapshotProducer.java | 18 +++++++++++++++---
 .../test/java/org/apache/iceberg/TestRowDelta.java  | 21 +++++++++++++++++++++
 2 files changed, 36 insertions(+), 3 deletions(-)

diff --git a/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java b/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
index eed4e56dc0..e072382543 100644
--- a/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
+++ b/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
@@ -289,7 +289,11 @@ abstract class MergingSnapshotProducer<ThisT> extends SnapshotProducer<ThisT> {
 
   protected void validateNewDeleteFile(DeleteFile file) {
     Preconditions.checkNotNull(file, "Invalid delete file: null");
-    switch (formatVersion()) {
+    validateDeleteFileForVersion(file, formatVersion());
+  }
+
+  private static void validateDeleteFileForVersion(DeleteFile file, int formatVersion) {
+    switch (formatVersion) {
       case 1:
         throw new IllegalArgumentException("Deletes are supported in V2 and above");
       case 2:
@@ -303,11 +307,11 @@ abstract class MergingSnapshotProducer<ThisT> extends SnapshotProducer<ThisT> {
         Preconditions.checkArgument(
             file.content() == FileContent.EQUALITY_DELETES || ContentFileUtil.isDV(file),
             "Must use DVs for position deletes in V%s: %s",
-            formatVersion(),
+            formatVersion,
             file.location());
         break;
       default:
-        throw new IllegalArgumentException("Unsupported format version: " + formatVersion());
+        throw new IllegalArgumentException("Unsupported format version: " + formatVersion);
     }
   }
 
@@ -959,8 +963,16 @@ abstract class MergingSnapshotProducer<ThisT> extends SnapshotProducer<ThisT> {
     return summaryBuilder.build();
   }
 
+  // guard buffered deletes against concurrent format upgrade
+  private void validateDeleteFilesForVersion(int currentFormatVersion) {
+    for (DeleteFile file : v2Deletes) {
+      validateDeleteFileForVersion(file, currentFormatVersion);
+    }
+  }
+
   @Override
   public List<ManifestFile> apply(TableMetadata base, Snapshot snapshot) {
+    validateDeleteFilesForVersion(base.formatVersion());
     // filter any existing manifests
     List<ManifestFile> filtered =
         filterManager.filterManifests(
diff --git a/core/src/test/java/org/apache/iceberg/TestRowDelta.java b/core/src/test/java/org/apache/iceberg/TestRowDelta.java
index c442541289..aaccf41224 100644
--- a/core/src/test/java/org/apache/iceberg/TestRowDelta.java
+++ b/core/src/test/java/org/apache/iceberg/TestRowDelta.java
@@ -2419,6 +2419,27 @@ public class TestRowDelta extends TestBase {
     assertThat(taskDV.contentSizeInBytes()).isEqualTo(dv.contentSizeInBytes());
   }
 
+  @TestTemplate
+  public void testV2StagedPositionDeleteCannotCommitToV3() {
+    assumeThat(formatVersion).isEqualTo(2);
+
+    Snapshot initial = commit(table, table.newAppend().appendFile(FILE_A), branch);
+
+    // Stage RowDelta at v2: position delete for FILE_A + add new data FILE_B.
+    RowDelta rowDelta = table.newRowDelta().addDeletes(FILE_A_DELETES).addRows(FILE_B);
+
+    // upgrade the table
+    table.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit();
+
+    assertThatThrownBy(rowDelta::commit)
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Must use DVs for position deletes in V3");
+
+    table.refresh();
+    assertThat(table.operations().current().formatVersion()).isEqualTo(3);
+    assertThat(table.snapshot(branch)).isEqualTo(initial);
+  }
+
   @TestTemplate
   public void testInabilityToAddPositionDeleteFilesInTablesWithDVs() {
     assumeThat(formatVersion).isGreaterThanOrEqualTo(3);