(iceberg) branch main updated: Core: Skip concurrent DV validation for added data files (#18626)

[pvary.apache.org via commits]

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

pvary 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 ae869756da Core: Skip concurrent DV validation for added data files 
(#18626)
ae869756da is described below

commit ae869756da0f30a4bf67723351c8de2a2a8b924b
Author: Sergei Nikolaev <kinol...@gmail.com>
AuthorDate: Tue Nov 3 12:52:37 2026 +0100

    Core: Skip concurrent DV validation for added data files (#18626)
---
 .../apache/iceberg/MergingSnapshotProducer.java    | 11 ++++++++++
 .../test/java/org/apache/iceberg/TestRowDelta.java | 25 +++++++++++++++++++++-
 2 files changed, 35 insertions(+), 1 deletion(-)

diff --git a/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java 
b/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
index 70617cbc42..b0ebe41586 100644
--- a/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
+++ b/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
@@ -876,6 +876,17 @@ abstract class MergingSnapshotProducer<ThisT> extends 
SnapshotProducer<ThisT> {
       return;
     }
 
+    // DVs for data files added in this operation can't conflict with 
concurrent commits
+    Set<String> unmatchedReferencedDataFiles = 
Sets.newHashSet(dvsByReferencedFile.keySet());
+    for (DataFileSet dataFiles : newDataFilesBySpec.values()) {
+      for (DataFile dataFile : dataFiles) {
+        if (unmatchedReferencedDataFiles.remove(dataFile.location())
+            && unmatchedReferencedDataFiles.isEmpty()) {
+          return;
+        }
+      }
+    }
+
     Pair<List<ManifestFile>, Set<Long>> history =
         validationHistory(
             base,
diff --git a/core/src/test/java/org/apache/iceberg/TestRowDelta.java 
b/core/src/test/java/org/apache/iceberg/TestRowDelta.java
index bacd4b9546..b60cd6696d 100644
--- a/core/src/test/java/org/apache/iceberg/TestRowDelta.java
+++ b/core/src/test/java/org/apache/iceberg/TestRowDelta.java
@@ -1967,7 +1967,13 @@ public class TestRowDelta extends TestBase {
     RowDelta rowDelta1 = table.newRowDelta().addDeletes(deleteFile1);
 
     DeleteFile deleteFile2 = newDeletes(dataFile);
-    RowDelta rowDelta2 = table.newRowDelta().addDeletes(deleteFile2);
+    DataFile dataFile2 = newDataFile("data_bucket=0");
+    RowDelta rowDelta2 =
+        table
+            .newRowDelta()
+            .addDeletes(deleteFile2)
+            .addRows(dataFile2)
+            .addDeletes(newDeletes(dataFile2));
 
     commit(table, rowDelta1, branch);
 
@@ -1976,6 +1982,23 @@ public class TestRowDelta extends TestBase {
         .hasMessageContaining("Found concurrently added DV for %s", 
dataFile.location());
   }
 
+  @TestTemplate
+  public void testDvsForAddedDataFilesSkipValidation() {
+    assumeThat(formatVersion).isGreaterThanOrEqualTo(3);
+
+    commit(table, table.newRowDelta().addRows(newDataFile("data_bucket=0")), 
branch);
+    long expiredSnapshotId = latestSnapshot(table, branch).snapshotId();
+    commit(table, table.newRowDelta().addRows(newDataFile("data_bucket=0")), 
branch);
+    // validating without a starting snapshot fails once the oldest ancestor 
is expired
+    table.expireSnapshots().expireSnapshotId(expiredSnapshotId).commit();
+
+    DataFile dataFile = newDataFile("data_bucket=0");
+    commit(table, 
table.newRowDelta().addRows(dataFile).addDeletes(newDeletes(dataFile)), branch);
+
+    assertThat(latestSnapshot(table, branch).summary())
+        .containsEntry(SnapshotSummary.ADDED_DVS_PROP, "1");
+  }
+
   @TestTemplate
   public void testDuplicateDVsAreMerged() throws IOException {
     assumeThat(formatVersion).isGreaterThanOrEqualTo(3);