This is an automated email from the ASF dual-hosted git repository.
findepi 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 cd3ae3c805 Core: Sweep only the transaction's own snapshots after
commit (#18095)
cd3ae3c805 is described below
commit cd3ae3c8051e586e5b695182db44e682b54dcca3
Author: Raunaq Morarka <[email protected]>
AuthorDate: Mon Oct 12 23:44:08 2026 +0530
Core: Sweep only the transaction's own snapshots after commit (#18095)
Snapshots committed by other writers while the transaction was open
cannot reference files it wrote, so reading their manifest lists during
post-commit clean-up is wasted work that grows with the number of
concurrent commits.
---
.../java/org/apache/iceberg/BaseTransaction.java | 5 +-
.../java/org/apache/iceberg/TestTransaction.java | 111 +++++++++++++++++++++
2 files changed, 114 insertions(+), 2 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/BaseTransaction.java
b/core/src/main/java/org/apache/iceberg/BaseTransaction.java
index 46eda9e0c9..af353f9429 100644
--- a/core/src/main/java/org/apache/iceberg/BaseTransaction.java
+++ b/core/src/main/java/org/apache/iceberg/BaseTransaction.java
@@ -353,8 +353,6 @@ public class BaseTransaction implements Transaction {
return;
}
- Set<Long> startingSnapshots =
-
base.snapshots().stream().map(Snapshot::snapshotId).collect(Collectors.toSet());
try {
Tasks.foreach(ops)
.retry(base.propertyAsInt(COMMIT_NUM_RETRIES,
COMMIT_NUM_RETRIES_DEFAULT))
@@ -392,6 +390,9 @@ public class BaseTransaction implements Transaction {
// 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.
+ // only snapshots added on top of the committed base can reference this
transaction's files
+ Set<Long> startingSnapshots =
+
base.snapshots().stream().map(Snapshot::snapshotId).collect(Collectors.toSet());
Set<Snapshot> newSnapshots = Sets.newHashSet();
for (Snapshot snapshot : current.snapshots()) {
if (!startingSnapshots.contains(snapshot.snapshotId())) {
diff --git a/core/src/test/java/org/apache/iceberg/TestTransaction.java
b/core/src/test/java/org/apache/iceberg/TestTransaction.java
index 7f371625d4..8fa2658e3a 100644
--- a/core/src/test/java/org/apache/iceberg/TestTransaction.java
+++ b/core/src/test/java/org/apache/iceberg/TestTransaction.java
@@ -35,6 +35,7 @@ import java.util.stream.Collectors;
import org.apache.iceberg.ManifestEntry.Status;
import org.apache.iceberg.exceptions.CommitFailedException;
import org.apache.iceberg.exceptions.CommitStateUnknownException;
+import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -681,6 +682,116 @@ public class TestTransaction extends TestBase {
assertThat(table.currentSnapshot().allManifests(table.io())).hasSize(1);
}
+ @TestTemplate
+ public void testTransactionCleanupReadsOnlyOwnManifestLists() throws
IOException {
+ File location = java.nio.file.Files.createTempDirectory(temp,
"junit").toFile();
+ String tableName = "txnCleanupOwnManifestListsTest";
+ // record the manifest lists opened after the last successful commit
+ List<String> manifestListsRead = Lists.newArrayList();
+ AtomicInteger injectedFailures = new AtomicInteger(0);
+ TestTables.LocalFileIO trackingFileIO =
+ new TestTables.LocalFileIO() {
+ @Override
+ public InputFile newInputFile(String path) {
+ if (path.contains("snap-")) {
+ manifestListsRead.add(path);
+ }
+ return super.newInputFile(path);
+ }
+ };
+ TestTables.TestTableOperations ops =
+ new TestTables.TestTableOperations(tableName, location,
trackingFileIO) {
+ @Override
+ public void commit(TableMetadata base, TableMetadata
updatedMetadata) {
+ if (injectedFailures.getAndDecrement() > 0) {
+ // another writer commits between the failed attempt and the
retry
+ TestTables.load(location,
tableName).newFastAppend().appendFile(FILE_A).commit();
+ throw new CommitFailedException("Injected failure");
+ }
+ super.commit(base, updatedMetadata);
+ manifestListsRead.clear();
+ }
+ };
+ TestTables.TestTable txnTable =
+ TestTables.create(
+ location, tableName, SCHEMA, SPEC, SortOrder.unsorted(),
formatVersion, ops);
+ // each table operations instance numbers snapshots on its own, so use
random ids
+ txnTable.updateProperties().set("random-snapshot-ids", "true").commit();
+
+ Transaction txn = txnTable.newTransaction();
+ txn.newFastAppend().appendFile(FILE_A).commit();
+ String firstManifestList =
txn.table().currentSnapshot().manifestListLocation();
+
+ // round trip through JSON so the snapshots the transaction loads have no
cached manifest lists
+ TestTables.TestTableOperations concurrentOps =
+ new TestTables.TestTableOperations(tableName, location) {
+ @Override
+ public void commit(TableMetadata base, TableMetadata
updatedMetadata) {
+ super.commit(
+ base,
TableMetadataParser.fromJson(TableMetadataParser.toJson(updatedMetadata)));
+ }
+ };
+ Table concurrentTable = new BaseTable(concurrentOps, tableName);
+ concurrentTable.newFastAppend().appendFile(FILE_B).commit();
+ concurrentTable.newFastAppend().appendFile(FILE_C).commit();
+ concurrentTable.newFastAppend().appendFile(FILE_D).commit();
+
+ injectedFailures.set(1);
+ txn.commitTransaction();
+
+ Table committedTable = TestTables.load(location, tableName);
+ assertThat(committedTable.snapshots()).hasSize(5);
+ String committedManifestList =
committedTable.currentSnapshot().manifestListLocation();
+ assertThat(manifestListsRead).containsExactly(committedManifestList);
+ assertThat(new File(committedManifestList)).exists();
+ assertThat(((BaseTransaction)
txn).deletedFiles()).contains(firstManifestList);
+ assertThat(new File(firstManifestList)).doesNotExist();
+ }
+
+ @TestTemplate
+ public void testTransactionRetryKeepsManifestReusedByRetry() throws
IOException {
+ // use only one retry, merge any two manifests, and keep appended
manifests as they are
+ table
+ .updateProperties()
+ .set(TableProperties.COMMIT_NUM_RETRIES, "1")
+ .set(TableProperties.MANIFEST_MIN_MERGE_COUNT, "2")
+ .set(TableProperties.SNAPSHOT_ID_INHERITANCE_ENABLED, "true")
+ .commit();
+
+ table.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
+ assertThat(table.currentSnapshot().allManifests(table.io())).hasSize(1);
+ ManifestFile baseManifest =
table.currentSnapshot().allManifests(table.io()).get(0);
+
+ ManifestFile appendManifest = writeManifestWithName("input.m0", FILE_D);
+
+ // size the merge bins so two manifests merge and three do not
+ long targetSize = (baseManifest.length() + appendManifest.length()) * 5 /
4;
+ table
+ .updateProperties()
+ .set(TableProperties.MANIFEST_TARGET_SIZE_BYTES,
String.valueOf(targetSize))
+ .commit();
+
+ Transaction txn = table.newTransaction();
+ txn.newAppend().appendManifest(appendManifest).commit();
+
+ // the first attempt merges the appended manifest and enqueues it for
deletion
+
assertThat(txn.table().currentSnapshot().allManifests(table.io())).hasSize(1);
+ ManifestFile mergedManifest =
txn.table().currentSnapshot().allManifests(table.io()).get(0);
+ assertThat(((BaseTransaction)
txn).deletedFiles()).contains(appendManifest.path());
+
+ // a third manifest overflows the bin, so the retry keeps the appended
manifest as is
+ table.newFastAppend().appendFile(FILE_C).commit();
+
+ txn.commitTransaction();
+
+ assertThat(table.currentSnapshot().allManifests(table.io()))
+ .extracting(ManifestFile::path)
+ .contains(appendManifest.path())
+ .doesNotContain(mergedManifest.path());
+ assertThat(new File(appendManifest.path())).exists();
+ assertThat(new File(mergedManifest.path())).doesNotExist();
+ }
+
@TestTemplate
public void testTransactionNoCustomDeleteFunc() {
assertThatThrownBy(
