diff --git a/core/src/main/java/org/apache/iceberg/BaseTransaction.java b/core/src/main/java/org/apache/iceberg/BaseTransaction.java index 9884ac297079..e180579cf543 100644 --- a/core/src/main/java/org/apache/iceberg/BaseTransaction.java +++ b/core/src/main/java/org/apache/iceberg/BaseTransaction.java @@ -45,7 +45,6 @@ import org.apache.iceberg.metrics.MetricsReporter; import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; -import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.util.PropertyUtil; @@ -393,24 +392,24 @@ private void commitSimpleTransaction() { // committed manifests to ensure that no committed manifest is deleted. // A manifest could be deleted in one successful operation commit, but reused in another // successful commit of that operation if the whole transaction is retried. - Set newSnapshots = Sets.newHashSet(); + // + // take the new snapshots from the metadata this transaction committed. re-reading + // through ops.current() can return stale metadata when the catalog caches table pointers, + // which would make the new snapshots appear missing and skip clean-up entirely. + Set newSnapshots = Sets.newHashSet(); for (Snapshot snapshot : current.snapshots()) { if (!startingSnapshots.contains(snapshot.snapshotId())) { - newSnapshots.add(snapshot.snapshotId()); + newSnapshots.add(snapshot); } } - Set committedFiles = committedFiles(ops, newSnapshots); - if (committedFiles != null) { - // delete all the files that were deleted in the most recent set of operation commits - Set uncommittedFiles = - deletedFiles.stream() - .filter(f -> !committedFiles.contains(f)) - .collect(Collectors.toSet()); - deleteUncommittedFiles(uncommittedFiles); - } else { - LOG.warn("Failed to load metadata for a committed snapshot, skipping clean-up"); - } + Set committedFiles = committedFiles(ops.io(), newSnapshots); + // delete all the files that were deleted in the most recent set of operation commits + Set uncommittedFiles = + deletedFiles.stream() + .filter(f -> !committedFiles.contains(f)) + .collect(Collectors.toSet()); + deleteUncommittedFiles(uncommittedFiles); } catch (RuntimeException e) { LOG.warn("Failed to load committed metadata, skipping clean-up", e); @@ -458,23 +457,13 @@ private void applyUpdates(TableOperations underlyingOps) { } } - // committedFiles returns null whenever the set of committed files - // cannot be determined from the provided snapshots - private static Set committedFiles(TableOperations ops, Set snapshotIds) { - if (snapshotIds.isEmpty()) { - return ImmutableSet.of(); - } - + // returns the manifest lists and manifests referenced by the given committed snapshots + private static Set committedFiles(FileIO io, Set snapshots) { Set committedFiles = Sets.newHashSet(); - for (long snapshotId : snapshotIds) { - Snapshot snap = ops.current().snapshot(snapshotId); - if (snap != null) { - committedFiles.add(snap.manifestListLocation()); - snap.allManifests(ops.io()).forEach(manifest -> committedFiles.add(manifest.path())); - } else { - return null; - } + for (Snapshot snap : snapshots) { + committedFiles.add(snap.manifestListLocation()); + snap.allManifests(io).forEach(manifest -> committedFiles.add(manifest.path())); } return committedFiles; diff --git a/core/src/test/java/org/apache/iceberg/TestTransaction.java b/core/src/test/java/org/apache/iceberg/TestTransaction.java index fe47ac62561d..7f371625d406 100644 --- a/core/src/test/java/org/apache/iceberg/TestTransaction.java +++ b/core/src/test/java/org/apache/iceberg/TestTransaction.java @@ -28,6 +28,9 @@ import java.util.List; import java.util.Set; import java.util.UUID; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; import org.apache.iceberg.ManifestEntry.Status; import org.apache.iceberg.exceptions.CommitFailedException; @@ -39,6 +42,7 @@ import org.apache.iceberg.types.Types; import org.junit.jupiter.api.TestTemplate; import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; @ExtendWith(ParameterizedTestExtension.class) @@ -323,6 +327,68 @@ public void testTransactionRetry() { .isEqualTo(appendManifests); } + @TestTemplate + public void testTransactionRetryCleansUpWhenCatalogReturnsStaleMetadata() throws IOException { + File location = java.nio.file.Files.createTempDirectory(temp, "junit").toFile(); + String tableName = "txnStaleMetadataCleanupTest"; + // fail the first two commits, each after a concurrent metadata update, so the transaction + // re-applies twice and writes a new manifest list each time. after the successful commit, + // serve stale metadata from current() to simulate a catalog that caches table pointers. + AtomicInteger injectedFailures = new AtomicInteger(0); + AtomicBoolean captureStaleMetadata = new AtomicBoolean(false); + AtomicReference staleMetadata = new AtomicReference<>(null); + TestTables.LocalFileIO spyFileIO = Mockito.spy(new TestTables.LocalFileIO()); + TestTables.TestTableOperations ops = + new TestTables.TestTableOperations(tableName, location, spyFileIO) { + @Override + public void commit(TableMetadata base, TableMetadata updatedMetadata) { + if (injectedFailures.getAndDecrement() > 0) { + TestTables.load(location, tableName) + .updateProperties() + .set("conflict-" + injectedFailures.get(), "true") + .commit(); + throw new CommitFailedException("Injected failure"); + } + super.commit(base, updatedMetadata); + if (captureStaleMetadata.get()) { + staleMetadata.compareAndSet(null, base); + } + } + + @Override + public TableMetadata current() { + if (staleMetadata.get() != null) { + return staleMetadata.get(); + } + return super.current(); + } + }; + TestTables.TestTable txnTable = + TestTables.create( + location, tableName, SCHEMA, SPEC, SortOrder.unsorted(), formatVersion, ops); + + txnTable.updateProperties().set(TableProperties.COMMIT_NUM_RETRIES, "3").commit(); + + Transaction txn = txnTable.newTransaction(); + txn.newFastAppend().appendFile(FILE_A).commit(); + + injectedFailures.set(2); + captureStaleMetadata.set(true); + txn.commitTransaction(); + + // clean-up must run off the transaction's committed metadata, not the stale catalog state + ArgumentCaptor deletedPaths = ArgumentCaptor.forClass(String.class); + Mockito.verify(spyFileIO, Mockito.atLeastOnce()).deleteFile(deletedPaths.capture()); + List deletedManifestLists = + deletedPaths.getAllValues().stream() + .filter(path -> path.contains("snap-")) + .collect(Collectors.toList()); + assertThat(deletedManifestLists).hasSize(2).doesNotHaveDuplicates(); + assertThat(deletedManifestLists) + .doesNotContain( + TestTables.load(location, tableName).currentSnapshot().manifestListLocation()); + } + @TestTemplate public void testTransactionRetryMergeAppend() { // use only one retry