You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by sz...@apache.org on 2026/09/16 22:04:35 UTC
(iceberg) branch main updated: Core, Spark 4.2: Validate snapshot when removing dangling deletes (#18061)
This is an automated email from the ASF dual-hosted git repository.
szehon-ho 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 bdd1ade1e5 Core, Spark 4.2: Validate snapshot when removing dangling deletes (#18061)
bdd1ade1e5 is described below
commit bdd1ade1e5bf8d5f840e76ff35ff4b6775071c2b
Author: Manu Zhang <Ow...@gmail.com>
AuthorDate: Thu Sep 17 06:04:28 2026 +0800
Core, Spark 4.2: Validate snapshot when removing dangling deletes (#18061)
---
.../actions/RemoveDanglingDeleteFilesAction.java | 20 +++-
.../TestRemoveDanglingDeleteFilesAction.java | 123 ++++++++++++++++++++-
.../actions/TestRemoveDanglingDeleteAction.java | 5 +-
3 files changed, 143 insertions(+), 5 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java b/core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java
index 515f83f051..2532b80793 100644
--- a/core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java
+++ b/core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java
@@ -21,6 +21,7 @@ package org.apache.iceberg.actions;
import java.io.IOException;
import java.io.Serializable;
import java.util.Collection;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -37,6 +38,7 @@ import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableScan;
import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -126,7 +128,23 @@ public class RemoveDanglingDeleteFilesAction
return EMPTY_RESULT;
}
- RewriteFiles rewriteFiles = table.newRewrite().validateFromSnapshot(snapshot.snapshotId());
+ long snapshotId = snapshot.snapshotId();
+ RewriteFiles rewriteFiles =
+ table
+ .newRewrite()
+ .validateFromSnapshot(snapshotId)
+ // validate on every commit attempt, including retries after concurrent updates
+ .validateWith(
+ snapshots -> {
+ Iterator<Snapshot> iterator = snapshots.iterator();
+ Snapshot currentSnapshot = iterator.hasNext() ? iterator.next() : null;
+ ValidationException.check(
+ currentSnapshot != null && currentSnapshot.snapshotId() == snapshotId,
+ "Cannot remove dangling deletes: current snapshot changed from %s to %s",
+ snapshotId,
+ currentSnapshot != null ? currentSnapshot.snapshotId() : null);
+ return true;
+ });
for (DeleteFile deleteFile : danglingDeletes) {
LOG.debug("Removing dangling delete file {}", deleteFile.location());
rewriteFiles.deleteFile(deleteFile);
diff --git a/core/src/test/java/org/apache/iceberg/actions/TestRemoveDanglingDeleteFilesAction.java b/core/src/test/java/org/apache/iceberg/actions/TestRemoveDanglingDeleteFilesAction.java
index c6beee692b..f80bde1c48 100644
--- a/core/src/test/java/org/apache/iceberg/actions/TestRemoveDanglingDeleteFilesAction.java
+++ b/core/src/test/java/org/apache/iceberg/actions/TestRemoveDanglingDeleteFilesAction.java
@@ -20,7 +20,14 @@ package org.apache.iceberg.actions;
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 static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import java.io.File;
import java.io.IOException;
@@ -29,12 +36,14 @@ import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.BaseTable;
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.FileScanTask;
+import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.Metrics;
import org.apache.iceberg.Parameter;
import org.apache.iceberg.ParameterizedTestExtension;
@@ -42,9 +51,12 @@ import org.apache.iceberg.Parameters;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.TestHelpers;
+import org.apache.iceberg.exceptions.CommitFailedException;
import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
@@ -55,6 +67,7 @@ import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.Pair;
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;
@@ -257,11 +270,15 @@ public class TestRemoveDanglingDeleteFilesAction {
}
private void setupPartitionedTable() {
+ setupPartitionedTable(formatVersion);
+ }
+
+ private void setupPartitionedTable(int tableFormatVersion) {
this.table =
TABLES.create(
SCHEMA,
SPEC,
- ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion)),
+ ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(tableFormatVersion)),
tableLocation);
}
@@ -297,7 +314,11 @@ public class TestRemoveDanglingDeleteFilesAction {
}
protected RemoveDanglingDeleteFiles removeDanglingDeleteFiles() {
- return new RemoveDanglingDeleteFilesAction(table);
+ return removeDanglingDeleteFiles(table);
+ }
+
+ protected RemoveDanglingDeleteFiles removeDanglingDeleteFiles(Table actionTable) {
+ return new RemoveDanglingDeleteFilesAction(actionTable);
}
@TestTemplate
@@ -499,6 +520,104 @@ public class TestRemoveDanglingDeleteFilesAction {
assertThat(actualAfter).containsExactlyInAnyOrderElementsOf(expectedAfter);
}
+ @TestTemplate
+ public void testEmptyTable() {
+ setupPartitionedTable();
+
+ RemoveDanglingDeleteFiles.Result result = removeDanglingDeleteFiles().execute();
+
+ assertThat(result.removedDeleteFiles()).isEmpty();
+ assertThat(table.currentSnapshot()).isNull();
+ }
+
+ @Test
+ public void testPlanningUsesStartingSnapshot() {
+ setupPartitionedTable(2);
+ DeleteFile deletes = FILE_A_POS_DELETES;
+ long originalSnapshotId = prepareDanglingDeletes(deletes);
+ long rewrittenSnapshotId = table.currentSnapshot().snapshotId();
+ table.manageSnapshots().rollbackTo(originalSnapshotId).commit();
+ Table actionTable = spy(table);
+ doAnswer(
+ invocation -> {
+ Object snapshot = invocation.callRealMethod();
+ table.manageSnapshots().setCurrentSnapshot(rewrittenSnapshotId).commit();
+ return snapshot;
+ })
+ .when(actionTable)
+ .snapshot(anyString());
+
+ RemoveDanglingDeleteFiles.Result result = removeDanglingDeleteFiles(actionTable).execute();
+
+ assertThat(result.removedDeleteFiles()).isEmpty();
+ assertThat(table.currentSnapshot().snapshotId()).isEqualTo(rewrittenSnapshotId);
+ }
+
+ @Test
+ public void testRollbackBeforeCommit() {
+ setupPartitionedTable(2);
+ DeleteFile deletes = FILE_A_POS_DELETES;
+ long rollbackSnapshotId = prepareDanglingDeletes(deletes);
+ long planningSnapshotId = table.currentSnapshot().snapshotId();
+ Table actionTable = spy(table);
+ doAnswer(
+ invocation -> {
+ table.manageSnapshots().rollbackTo(rollbackSnapshotId).commit();
+ return invocation.callRealMethod();
+ })
+ .when(actionTable)
+ .newRewrite();
+
+ assertThatThrownBy(() -> removeDanglingDeleteFiles(actionTable).execute())
+ .isInstanceOf(ValidationException.class)
+ .hasMessage(
+ "Cannot remove dangling deletes: current snapshot changed from %s to %s",
+ planningSnapshotId, rollbackSnapshotId);
+
+ assertThat(table.currentSnapshot().snapshotId()).isEqualTo(rollbackSnapshotId);
+ assertThat(liveEntries())
+ .extracting(Pair::second)
+ .contains(FILE_A.location(), deletes.location());
+ }
+
+ @Test
+ public void testRollbackDuringCommitRetry() {
+ setupPartitionedTable(2);
+ DeleteFile deletes = FILE_A_POS_DELETES;
+ long rollbackSnapshotId = prepareDanglingDeletes(deletes);
+ long planningSnapshotId = table.currentSnapshot().snapshotId();
+ TableOperations ops = spy(((HasTableOperations) table).operations());
+ Table actionTable = new BaseTable(ops, table.name());
+ doAnswer(
+ invocation -> {
+ table.manageSnapshots().rollbackTo(rollbackSnapshotId).commit();
+ throw new CommitFailedException("Injected concurrent rollback");
+ })
+ .doCallRealMethod()
+ .when(ops)
+ .commit(any(), any());
+
+ assertThatThrownBy(() -> removeDanglingDeleteFiles(actionTable).execute())
+ .isInstanceOf(ValidationException.class)
+ .hasMessage(
+ "Cannot remove dangling deletes: current snapshot changed from %s to %s",
+ planningSnapshotId, rollbackSnapshotId);
+
+ verify(ops, times(1)).commit(any(), any());
+ assertThat(table.currentSnapshot().snapshotId()).isEqualTo(rollbackSnapshotId);
+ assertThat(liveEntries())
+ .extracting(Pair::second)
+ .contains(FILE_A.location(), deletes.location());
+ }
+
+ private long prepareDanglingDeletes(DeleteFile deletes) {
+ table.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
+ table.newRowDelta().addDeletes(deletes).commit();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ table.newRewrite().validateFromSnapshot(snapshotId).deleteFile(FILE_A).commit();
+ return snapshotId;
+ }
+
private List<Pair<Long, String>> liveEntries() {
Set<Pair<Long, String>> results = Sets.newHashSet();
try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
diff --git a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java
index a1595dc505..9976311751 100644
--- a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java
+++ b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.actions;
import java.net.InetAddress;
+import org.apache.iceberg.Table;
import org.apache.iceberg.actions.RemoveDanglingDeleteFiles;
import org.apache.iceberg.actions.TestRemoveDanglingDeleteFilesAction;
import org.apache.iceberg.spark.TestBase;
@@ -52,7 +53,7 @@ public class TestRemoveDanglingDeleteAction extends TestRemoveDanglingDeleteFile
}
@Override
- protected RemoveDanglingDeleteFiles removeDanglingDeleteFiles() {
- return new RemoveDanglingDeletesSparkAction(spark, table);
+ protected RemoveDanglingDeleteFiles removeDanglingDeleteFiles(Table actionTable) {
+ return new RemoveDanglingDeletesSparkAction(spark, actionTable);
}
}