From 8d5a817109ce70330dcd0f0fed7b869d377e7d81 Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Fri, 11 Sep 2026 18:08:24 +0900 Subject: [PATCH] Core, Spark: Validate snapshot when removing dangling deletes Validate that the planning snapshot is still the current snapshot of the target branch on every commit attempt so concurrent rollbacks cannot remove valid deletes. Generated-by: Codex Co-Authored-By: Claude Opus 5 (1M context) --- .../RemoveDanglingDeleteFilesAction.java | 20 ++- .../TestRemoveDanglingDeleteFilesAction.java | 123 +++++++++++++++++- .../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 515f83f051ed..2532b8079306 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 @@ 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.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 Result execute() { 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 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 c6beee692bde..f80bde1c4839 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 @@ 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.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.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.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 void after() { } 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 @@ private DeleteFile fileUnpartitionedDeletes() { } 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 void testPartitionedDeletesWithDanglingDvs() { 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> liveEntries() { Set> results = Sets.newHashSet(); try (CloseableIterable 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 a1595dc50567..997631175195 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 static void stopSpark() { } @Override - protected RemoveDanglingDeleteFiles removeDanglingDeleteFiles() { - return new RemoveDanglingDeletesSparkAction(spark, table); + protected RemoveDanglingDeleteFiles removeDanglingDeleteFiles(Table actionTable) { + return new RemoveDanglingDeletesSparkAction(spark, actionTable); } }