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 52fc764127 Core: Skip manifest list re-reads after successful commits
(#18074)
52fc764127 is described below
commit 52fc764127402ade8170b1b53ddd95fde13cb75c
Author: Raunaq Morarka <[email protected]>
AuthorDate: Tue Nov 10 00:05:12 2026 +0530
Core: Skip manifest list re-reads after successful commits (#18074)
---
.../org/apache/iceberg/BaseRewriteManifests.java | 2 +-
.../java/org/apache/iceberg/BaseTransaction.java | 22 ++++-
.../main/java/org/apache/iceberg/FastAppend.java | 2 +-
.../org/apache/iceberg/ManifestFilterManager.java | 4 +-
.../org/apache/iceberg/ManifestMergeManager.java | 4 +-
.../apache/iceberg/MergingSnapshotProducer.java | 4 +-
.../java/org/apache/iceberg/SnapshotProducer.java | 50 +++++++---
.../org/apache/iceberg/TestSnapshotProducer.java | 108 +++++++++++++++++++++
.../java/org/apache/iceberg/TestTransaction.java | 76 ++++++++++++++-
9 files changed, 245 insertions(+), 27 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java
b/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java
index e98027ec4a..b0188b9854 100644
--- a/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java
+++ b/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java
@@ -343,7 +343,7 @@ public class BaseRewriteManifests extends
SnapshotProducer<RewriteManifests>
}
@Override
- protected void cleanUncommitted(Set<ManifestFile> committed) {
+ protected void cleanUncommitted(Set<String> committed) {
deleteUncommitted(newManifests, committed, false);
// clean up only rewrittenAddedManifests as they are always owned by the
table
// don't clean up addedManifests as they are added to the manifest list
and are not compacted
diff --git a/core/src/main/java/org/apache/iceberg/BaseTransaction.java
b/core/src/main/java/org/apache/iceberg/BaseTransaction.java
index af353f9429..da8f134578 100644
--- a/core/src/main/java/org/apache/iceberg/BaseTransaction.java
+++ b/core/src/main/java/org/apache/iceberg/BaseTransaction.java
@@ -400,7 +400,7 @@ public class BaseTransaction implements Transaction {
}
}
- Set<String> committedFiles = committedFiles(ops.io(), newSnapshots);
+ Set<String> committedFiles = committedFiles(newSnapshots);
// delete all the files that were deleted in the most recent set of
operation commits
Set<String> uncommittedFiles =
deletedFiles.stream()
@@ -455,17 +455,33 @@ public class BaseTransaction implements Transaction {
}
// returns the manifest lists and manifests referenced by the given
committed snapshots
- private static Set<String> committedFiles(FileIO io, Set<Snapshot>
snapshots) {
+ private Set<String> committedFiles(Set<Snapshot> snapshots) {
Set<String> committedFiles = Sets.newHashSet();
for (Snapshot snap : snapshots) {
committedFiles.add(snap.manifestListLocation());
- snap.allManifests(io).forEach(manifest ->
committedFiles.add(manifest.path()));
+ committedFiles.addAll(committedManifestPaths(snap));
}
return committedFiles;
}
+ /** Returns this snapshot's manifest paths. */
+ private Set<String> committedManifestPaths(Snapshot snapshot) {
+ for (PendingUpdate update : updates) {
+ if (update instanceof SnapshotProducer<?> producer) {
+ Set<String> paths =
producer.writtenManifestPaths(snapshot.manifestListLocation());
+ if (paths != null) {
+ return paths;
+ }
+ }
+ }
+
+ return snapshot.allManifests(ops.io()).stream()
+ .map(ManifestFile::path)
+ .collect(Collectors.toSet());
+ }
+
public class TransactionTableOperations implements TableOperations {
private TableOperations tempOps = ops.temp(current);
diff --git a/core/src/main/java/org/apache/iceberg/FastAppend.java
b/core/src/main/java/org/apache/iceberg/FastAppend.java
index a6caed574a..c6c81291e3 100644
--- a/core/src/main/java/org/apache/iceberg/FastAppend.java
+++ b/core/src/main/java/org/apache/iceberg/FastAppend.java
@@ -181,7 +181,7 @@ class FastAppend extends SnapshotProducer<AppendFiles>
implements AppendFiles {
}
@Override
- protected void cleanUncommitted(Set<ManifestFile> committed) {
+ protected void cleanUncommitted(Set<String> committed) {
if (newManifests != null) {
deleteUncommitted(newManifests, committed, true /* clear manifests */);
}
diff --git a/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java
b/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java
index 96fa1b1a68..254464966f 100644
--- a/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java
+++ b/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java
@@ -335,7 +335,7 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
*
* @param committed the set of manifest files that were committed
*/
- void cleanUncommitted(Set<ManifestFile> committed) {
+ void cleanUncommitted(Set<String> committed) {
// iterate over a copy of entries to avoid concurrent modification
List<Map.Entry<ManifestFile, ManifestFile>> filterEntries =
Lists.newArrayList(filteredManifests.entrySet());
@@ -344,7 +344,7 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
// remove any new filtered manifests that aren't in the committed list
ManifestFile manifest = entry.getKey();
ManifestFile filtered = entry.getValue();
- if (!committed.contains(filtered)) {
+ if (!committed.contains(filtered.path())) {
// only delete if the filtered copy was created (manifest was replaced)
if (!manifest.equals(filtered)) {
deleteFile(filtered.path());
diff --git a/core/src/main/java/org/apache/iceberg/ManifestMergeManager.java
b/core/src/main/java/org/apache/iceberg/ManifestMergeManager.java
index 0aec0ac69a..efc0068081 100644
--- a/core/src/main/java/org/apache/iceberg/ManifestMergeManager.java
+++ b/core/src/main/java/org/apache/iceberg/ManifestMergeManager.java
@@ -105,7 +105,7 @@ abstract class ManifestMergeManager<F extends
ContentFile<F>> {
return replacedManifestsCount.get();
}
- void cleanUncommitted(Set<ManifestFile> committed) {
+ void cleanUncommitted(Set<String> committed) {
// iterate over a copy of entries to avoid concurrent modification
List<Map.Entry<List<ManifestFile>, ManifestFile>> entries =
Lists.newArrayList(mergedManifests.entrySet());
@@ -113,7 +113,7 @@ abstract class ManifestMergeManager<F extends
ContentFile<F>> {
for (Map.Entry<List<ManifestFile>, ManifestFile> entry : entries) {
// delete any new merged manifests that aren't in the committed list
ManifestFile merged = entry.getValue();
- if (!committed.contains(merged)) {
+ if (!committed.contains(merged.path())) {
deleteFile(merged.path());
List<ManifestFile> bin = entry.getKey();
mergedManifests.remove(bin);
diff --git a/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
b/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
index b0ebe41586..3b2ce91d22 100644
--- a/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
+++ b/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
@@ -1120,7 +1120,7 @@ abstract class MergingSnapshotProducer<ThisT> extends
SnapshotProducer<ThisT> {
}
@Override
- protected void cleanUncommitted(Set<ManifestFile> committed) {
+ protected void cleanUncommitted(Set<String> committed) {
mergeManager.cleanUncommitted(committed);
filterManager.cleanUncommitted(committed);
deleteMergeManager.cleanUncommitted(committed);
@@ -1128,7 +1128,7 @@ abstract class MergingSnapshotProducer<ThisT> extends
SnapshotProducer<ThisT> {
cleanUncommittedAppends(committed);
}
- private void cleanUncommittedAppends(Set<ManifestFile> committed) {
+ private void cleanUncommittedAppends(Set<String> committed) {
deleteUncommitted(cachedNewDataManifests, committed, true /* clear
manifests */);
deleteUncommitted(cachedNewDeleteManifests, committed, true /* clear
manifests */);
// rewritten manifests are always owned by the table
diff --git a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java
b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java
index 06281188c3..93133d2fe5 100644
--- a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java
+++ b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java
@@ -47,6 +47,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
+import java.util.stream.Collectors;
import org.apache.iceberg.encryption.EncryptedOutputFile;
import
org.apache.iceberg.encryption.StandardEncryptionManager.FileEncryptionKeys;
import org.apache.iceberg.events.CreateSnapshotEvent;
@@ -67,8 +68,8 @@ import org.apache.iceberg.metrics.Timer.Timed;
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.ImmutableMap;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.math.IntMath;
import org.apache.iceberg.util.Exceptions;
import org.apache.iceberg.util.PropertyUtil;
@@ -88,7 +89,7 @@ import org.slf4j.LoggerFactory;
abstract class SnapshotProducer<ThisT> implements SnapshotUpdate<ThisT> {
private static final Logger LOG =
LoggerFactory.getLogger(SnapshotProducer.class);
static final int MIN_FILE_GROUP_SIZE = 10_000;
- static final Set<ManifestFile> EMPTY_SET = Sets.newHashSet();
+ static final Set<String> EMPTY_SET = ImmutableSet.of();
/** Default callback used to delete files. */
private final Consumer<String> defaultDelete =
@@ -108,7 +109,7 @@ abstract class SnapshotProducer<ThisT> implements
SnapshotUpdate<ThisT> {
private final String commitUUID = UUID.randomUUID().toString();
private final AtomicInteger manifestCount = new AtomicInteger(0);
private final AtomicInteger attempt = new AtomicInteger(0);
- private final List<String> manifestLists = Lists.newArrayList();
+ private final Map<String, Set<String>> manifestsByManifestList =
Maps.newHashMap();
private final long targetManifestSizeBytes;
private final FileFormat manifestFormat;
private final Map<String, String> manifestWriterProps;
@@ -265,7 +266,7 @@ abstract class SnapshotProducer<ThisT> implements
SnapshotUpdate<ThisT> {
*
* @param committed a set of manifest paths that were actually committed
*/
- protected abstract void cleanUncommitted(Set<ManifestFile> committed);
+ protected abstract void cleanUncommitted(Set<String> committed);
/**
* A string that describes the action that produced the new snapshot.
@@ -318,8 +319,10 @@ abstract class SnapshotProducer<ThisT> implements
SnapshotUpdate<ThisT> {
base.nextRowId());
try (writer) {
- // keep track of the manifest lists created
- manifestLists.add(manifestList.location());
+ // keep track of the manifest lists created and the manifests they
reference
+ manifestsByManifestList.put(
+ manifestList.location(),
+
manifests.stream().map(ManifestFile::path).collect(Collectors.toSet()));
ManifestFile[] manifestFiles = new ManifestFile[manifests.size()];
@@ -555,11 +558,11 @@ abstract class SnapshotProducer<ThisT> implements
SnapshotUpdate<ThisT> {
Snapshot saved = ops.refresh().snapshot(newSnapshotId.get());
if (saved != null) {
if (cleanupAfterCommit()) {
- cleanUncommitted(Sets.newHashSet(saved.allManifests(ops.io())));
+ cleanUncommitted(committedManifestPaths(saved));
}
// also clean up unused manifest lists created by multiple attempts
- for (String manifestList : manifestLists) {
+ for (String manifestList : manifestsByManifestList.keySet()) {
if (!saved.manifestListLocation().equals(manifestList)) {
deleteFile(manifestList);
}
@@ -610,13 +613,34 @@ abstract class SnapshotProducer<ThisT> implements
SnapshotUpdate<ThisT> {
}
protected void cleanAll() {
- for (String manifestList : manifestLists) {
+ for (String manifestList : manifestsByManifestList.keySet()) {
deleteFile(manifestList);
}
- manifestLists.clear();
+ manifestsByManifestList.clear();
cleanUncommitted(EMPTY_SET);
}
+ private Set<String> committedManifestPaths(Snapshot saved) {
+ Set<String> paths = writtenManifestPaths(saved.manifestListLocation());
+ if (paths != null) {
+ // this producer wrote the manifest list, so its manifest paths are
already in memory
+ return paths;
+ }
+
+ // this producer did not write the manifest list (e.g. a fast-forward
cherry-pick), so read it
+ return saved.allManifests(ops.io()).stream()
+ .map(ManifestFile::path)
+ .collect(Collectors.toSet());
+ }
+
+ /**
+ * Returns the paths of the manifests written to the given manifest list, or
null if not written
+ * by this producer.
+ */
+ Set<String> writtenManifestPaths(String manifestListLocation) {
+ return manifestsByManifestList.get(manifestListLocation);
+ }
+
protected void deleteFile(String path) {
deleteFunc.accept(path);
}
@@ -770,10 +794,10 @@ abstract class SnapshotProducer<ThisT> implements
SnapshotUpdate<ThisT> {
// Deletes uncommitted manifests; clears list if clearManifests and any
deleted.
protected void deleteUncommitted(
- Collection<ManifestFile> manifests, Set<ManifestFile> committed, boolean
clearManifests) {
+ Collection<ManifestFile> manifests, Set<String> committed, boolean
clearManifests) {
boolean anyDeleted = false;
for (ManifestFile manifest : manifests) {
- if (!committed.contains(manifest)) {
+ if (!committed.contains(manifest.path())) {
deleteFile(manifest.path());
anyDeleted = true;
}
diff --git a/core/src/test/java/org/apache/iceberg/TestSnapshotProducer.java
b/core/src/test/java/org/apache/iceberg/TestSnapshotProducer.java
index 8d128b9b6b..9e653b3380 100644
--- a/core/src/test/java/org/apache/iceberg/TestSnapshotProducer.java
+++ b/core/src/test/java/org/apache/iceberg/TestSnapshotProducer.java
@@ -31,16 +31,21 @@ import java.time.Clock;
import java.time.Instant;
import java.time.ZoneOffset;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import javax.annotation.Nonnull;
+import org.apache.iceberg.exceptions.CommitFailedException;
import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.relocated.com.google.common.collect.Streams;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mockito;
@ExtendWith(ParameterizedTestExtension.class)
public class TestSnapshotProducer extends TestBase {
@@ -403,4 +408,107 @@ public class TestSnapshotProducer extends TestBase {
.isEqualTo(mainNewTs)
.isLessThan(branchTs);
}
+
+ @TestTemplate
+ public void commitDoesNotReadCommittedManifestList() {
+ String tableName = "no-manifest-list-reread";
+ TestTables.LocalFileIO spyFileIO = Mockito.spy(new
TestTables.LocalFileIO());
+ TestTables.TestTable testTable =
+ TestTables.create(
+ tableDir,
+ tableName,
+ SCHEMA,
+ SPEC,
+ SortOrder.unsorted(),
+ formatVersion,
+ new TestTables.TestTableOperations(tableName, tableDir,
spyFileIO));
+
+ testTable.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
+
+ Snapshot committed = testTable.currentSnapshot();
+ assertThat(committed).isNotNull();
+ Mockito.verify(spyFileIO,
Mockito.never()).newInputFile(committed.manifestListLocation());
+ }
+
+ @TestTemplate
+ public void fastForwardCherryPickReadsCommittedManifestList() {
+ String tableName = "fast-forward-cherry-pick";
+ TestTables.LocalFileIO spyFileIO = Mockito.spy(new
TestTables.LocalFileIO());
+ TestTables.TestTable testTable =
+ TestTables.create(
+ tableDir,
+ tableName,
+ SCHEMA,
+ SPEC,
+ SortOrder.unsorted(),
+ formatVersion,
+ new TestTables.TestTableOperations(tableName, tableDir,
spyFileIO));
+
+ testTable.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
+ long parentId = testTable.currentSnapshot().snapshotId();
+ // cherry-pick fast-forwards a delete without loading its manifests
+ testTable.newDelete().deleteFile(FILE_A).stageOnly().commit();
+ Snapshot staged =
+ Streams.stream(testTable.snapshots())
+ .filter(snapshot -> snapshot.snapshotId() != parentId)
+ .findFirst()
+ .orElseThrow();
+
+ Mockito.clearInvocations(spyFileIO);
+ testTable.manageSnapshots().cherrypick(staged.snapshotId()).commit();
+
+
assertThat(testTable.currentSnapshot().snapshotId()).isEqualTo(staged.snapshotId());
+ Mockito.verify(spyFileIO,
Mockito.atLeastOnce()).newInputFile(staged.manifestListLocation());
+ assertThat(staged.allManifests(testTable.io()))
+ .allSatisfy(manifest -> assertThat(new
File(manifest.path())).exists());
+ }
+
+ @TestTemplate
+ public void cleanupUsesManifestsOfCommittedAttempt() {
+ String tableName = "committed-attempt-cleanup";
+ AtomicBoolean throwAfterCommit = new AtomicBoolean(false);
+ TestTables.TestTableOperations ops =
+ new TestTables.TestTableOperations(tableName, tableDir) {
+ @Override
+ public void commit(TableMetadata base, TableMetadata
updatedMetadata) {
+ super.commit(base, updatedMetadata);
+ if (throwAfterCommit.getAndSet(false)) {
+ throw new CommitFailedException("Injected failure after commit");
+ }
+ }
+ };
+ TestTables.TestTable retryTable =
+ TestTables.create(
+ tableDir, tableName, SCHEMA, SPEC, SortOrder.unsorted(),
formatVersion, ops);
+ retryTable
+ .updateProperties()
+ .set(TableProperties.MANIFEST_MIN_MERGE_COUNT, "1")
+ .set(TableProperties.COMMIT_MIN_RETRY_WAIT_MS, "0")
+ .commit();
+ retryTable.newAppend().appendFile(FILE_A).commit();
+
+ // the first attempt commits but reports failure, so the retry finds its
snapshot committed
+ throwAfterCommit.set(true);
+ retryTable.newAppend().appendFile(FILE_B).commit();
+
+ Snapshot committed = retryTable.currentSnapshot();
+ assertThat(committed.allManifests(retryTable.io()))
+ .allSatisfy(manifest -> assertThat(new
File(manifest.path())).exists());
+
+ Set<String> referencedManifests = Sets.newHashSet();
+ for (Snapshot snapshot : retryTable.snapshots()) {
+ for (ManifestFile manifest : snapshot.allManifests(retryTable.io())) {
+ referencedManifests.add(new File(manifest.path()).getName());
+ }
+ }
+ assertThat(listManifestFiles(tableDir))
+ .extracting(File::getName)
+ .containsExactlyInAnyOrderElementsOf(referencedManifests);
+ assertThat(listManifestLists(tableDir))
+ .extracting(File::getName)
+ .containsExactlyInAnyOrder(
+ Streams.stream(retryTable.snapshots())
+ .map(snapshot -> new
File(snapshot.manifestListLocation()).getName())
+ .toArray(String[]::new));
+ }
}
diff --git a/core/src/test/java/org/apache/iceberg/TestTransaction.java
b/core/src/test/java/org/apache/iceberg/TestTransaction.java
index 8fa2658e3a..855bafe179 100644
--- a/core/src/test/java/org/apache/iceberg/TestTransaction.java
+++ b/core/src/test/java/org/apache/iceberg/TestTransaction.java
@@ -683,9 +683,9 @@ public class TestTransaction extends TestBase {
}
@TestTemplate
- public void testTransactionCleanupReadsOnlyOwnManifestLists() throws
IOException {
+ public void transactionCleanupReadsNoManifestLists() throws IOException {
File location = java.nio.file.Files.createTempDirectory(temp,
"junit").toFile();
- String tableName = "txnCleanupOwnManifestListsTest";
+ String tableName = "txnCleanupNoManifestListsTest";
// record the manifest lists opened after the last successful commit
List<String> manifestListsRead = Lists.newArrayList();
AtomicInteger injectedFailures = new AtomicInteger(0);
@@ -742,7 +742,7 @@ public class TestTransaction extends TestBase {
Table committedTable = TestTables.load(location, tableName);
assertThat(committedTable.snapshots()).hasSize(5);
String committedManifestList =
committedTable.currentSnapshot().manifestListLocation();
- assertThat(manifestListsRead).containsExactly(committedManifestList);
+ assertThat(manifestListsRead).isEmpty();
assertThat(new File(committedManifestList)).exists();
assertThat(((BaseTransaction)
txn).deletedFiles()).contains(firstManifestList);
assertThat(new File(firstManifestList)).doesNotExist();
@@ -1119,6 +1119,76 @@ public class TestTransaction extends TestBase {
.hasSize(2);
}
+ @TestTemplate
+ public void commitTransactionRetryKeepsMergedManifest() throws IOException {
+ File location = java.nio.file.Files.createTempDirectory(temp,
"junit").toFile();
+ String tableName = "txnRetryMergedManifest";
+ TestTables.LocalFileIO spyFileIO = Mockito.spy(new
TestTables.LocalFileIO());
+ TestTables.TestTable txnTable =
+ TestTables.create(
+ location,
+ tableName,
+ SCHEMA,
+ SPEC,
+ SortOrder.unsorted(),
+ formatVersion,
+ new TestTables.TestTableOperations(tableName, location,
spyFileIO));
+ txnTable
+ .updateProperties()
+ .set(TableProperties.MANIFEST_MIN_MERGE_COUNT, "1")
+ .set("random-snapshot-ids", "true")
+ .commit();
+
+ Transaction txn = txnTable.newTransaction();
+ txn.newAppend().appendFile(FILE_A).commit();
+
+ // a concurrent commit forces the transaction to re-apply its update on
top of a new snapshot
+ TestTables.load(location,
tableName).newAppend().appendFile(FILE_B).commit();
+ Snapshot concurrent = TestTables.load(location,
tableName).currentSnapshot();
+
+ Mockito.clearInvocations(spyFileIO);
+ txn.commitTransaction();
+
+ Snapshot committed = txnTable.currentSnapshot();
+ Mockito.verify(spyFileIO,
Mockito.never()).newInputFile(committed.manifestListLocation());
+ assertThat(committed.parentId()).isEqualTo(concurrent.snapshotId());
+ assertThat(committed.allManifests(txnTable.io())).hasSize(1);
+ for (Snapshot snapshot : txnTable.snapshots()) {
+ assertThat(snapshot.allManifests(txnTable.io()))
+ .allSatisfy(manifest -> assertThat(new
File(manifest.path())).exists());
+ }
+ }
+
+ @TestTemplate
+ public void transactionCleanupReadsManifestListOfExternalSnapshot() throws
IOException {
+ File location = java.nio.file.Files.createTempDirectory(temp,
"junit").toFile();
+ String tableName = "txnCleanupExternalSnapshot";
+ TestTables.LocalFileIO spyFileIO = Mockito.spy(new
TestTables.LocalFileIO());
+ TestTables.TestTable txnTable =
+ TestTables.create(
+ location,
+ tableName,
+ SCHEMA,
+ SPEC,
+ SortOrder.unsorted(),
+ formatVersion,
+ new TestTables.TestTableOperations(tableName, location,
spyFileIO));
+
+ Transaction txn = txnTable.newTransaction();
+ // a snapshot committed through the transaction's operations has no update
that wrote it
+ Snapshot external = txnTable.newFastAppend().appendFile(FILE_B).apply();
+ TableOperations txnOps = ((HasTableOperations) txn.table()).operations();
+ txnOps.commit(
+ txnOps.current(),
TableMetadata.buildFrom(txnOps.current()).addSnapshot(external).build());
+ txn.newAppend().appendFile(FILE_A).commit();
+
+ Mockito.clearInvocations(spyFileIO);
+ txn.commitTransaction();
+
+ assertThat(txnTable.snapshot(external.snapshotId())).isNotNull();
+ Mockito.verify(spyFileIO,
Mockito.atLeastOnce()).newInputFile(external.manifestListLocation());
+ }
+
private static class AppendToBranchTransaction extends BaseTransaction {
AppendToBranchTransaction(
