Skip to content

Oleksandr.nitavskyi/iceberg metadata base fallback #9943

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
JTaky wants to merge 2 commits into
base: master
Choose a base branch
from
+252 −27
Draft

Oleksandr.nitavskyi/iceberg metadata base fallback #9943

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0859c5d
Fall back to rebuilding Iceberg metadata when base manifest list is m…
JTaky Sep 17, 2026
e13efc8
Align expireAllBefore's retention floor with deleteApplicableMetadata…
JTaky Sep 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
96 changes: 69 additions & 27 deletions paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@
import org.apache.paimon.types.MultisetType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.DataFilePathFactories;
import org.apache.paimon.utils.ExceptionUtils;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.ManifestReadThreadPool;
import org.apache.paimon.utils.Pair;
Expand Down Expand Up @@ -430,19 +431,40 @@ private void createMetadata(
Path baseMetadataPath = pathFactory.toMetadataPath(snapshotId - 1);

if (table.fileIO().exists(baseMetadataPath)) {
createMetadataWithBase(
fileChangesCollector,
indexFiles.stream()
.filter(
index ->
index.indexFile()
.indexType()
.equals(DELETION_VECTORS_INDEX))
.collect(Collectors.toList()),
snapshot,
baseMetadataPath,
abandonedLastColumnId,
abandonedNextRowId);
try {
createMetadataWithBase(
fileChangesCollector,
indexFiles.stream()
.filter(
index ->
index.indexFile()
.indexType()
.equals(DELETION_VECTORS_INDEX))
.collect(Collectors.toList()),
snapshot,
baseMetadataPath,
abandonedLastColumnId,
abandonedNextRowId);
} catch (RuntimeException e) {
if (!ExceptionUtils.findThrowable(e, FileNotFoundException.class).isPresent()) {
throw e;
}
// The base metadata file itself exists, but a manifest or manifest list it
// transitively references (from its historical snapshot chain) has already
// been pruned by unrelated, later retention cleanup, so the base is unusable
// even though it exists. Rebuild from scratch instead of crashing permanently
// on every retry: this loses that snapshot's Iceberg-side history/lineage,
// the same tradeoff already accepted when the base file is simply absent.
LOG.warn(
"Failed to read base Iceberg metadata {} for table {} because a file "
+ "it references is missing. Falling back to recreating "
+ "metadata from scratch.",
baseMetadataPath,
table.fullName(),
e);
createMetadataWithoutBase(
snapshotId, abandonedUuid, abandonedLastColumnId, abandonedNextRowId);
}
} else {
createMetadataWithoutBase(
snapshotId, abandonedUuid, abandonedLastColumnId, abandonedNextRowId);
Expand Down Expand Up @@ -1600,10 +1622,18 @@ private void expireManifestList(String toExpire, String next) {
}

private void expireAllBefore(long snapshotId) throws IOException {
long earliestMetadataId = earliestMetadataIdToDelete(snapshotId);
if (earliestMetadataId <= 0) {
// Nothing should be deleted -- either delete-after-commit is disabled (every
// version retained forever) or there aren't enough versions yet. Deleting any
// manifest here could gut a JSON version that's being kept.
return;
}

Set<String> expiredManifestLists = new HashSet<>();
Set<String> expiredManifestFileMetas = new HashSet<>();
Iterator<Path> it =
pathFactory.getAllMetadataPathBefore(table.fileIO(), snapshotId).iterator();
pathFactory.getAllMetadataPathBefore(table.fileIO(), earliestMetadataId).iterator();

while (it.hasNext()) {
Path path = it.next();
Expand Down Expand Up @@ -1639,23 +1669,35 @@ private void expireAllBefore(long snapshotId) throws IOException {
}

private void deleteApplicableMetadataFiles(long snapshotId) throws IOException {
Options options = new Options(table.options());
if (options.get(IcebergOptions.METADATA_DELETE_AFTER_COMMIT)) {
long earliestMetadataId =
snapshotId - options.get(IcebergOptions.METADATA_PREVIOUS_VERSIONS_MAX);
if (earliestMetadataId > 0) {
Iterator<Path> it =
pathFactory
.getAllMetadataPathBefore(table.fileIO(), earliestMetadataId)
.iterator();
while (it.hasNext()) {
Path path = it.next();
table.fileIO().deleteQuietly(path);
}
long earliestMetadataId = earliestMetadataIdToDelete(snapshotId);
if (earliestMetadataId > 0) {
Iterator<Path> it =
pathFactory
.getAllMetadataPathBefore(table.fileIO(), earliestMetadataId)
.iterator();
while (it.hasNext()) {
Path path = it.next();
table.fileIO().deleteQuietly(path);
}
}
}

/**
* The oldest metadata version id that should still be deleted, or -1 if nothing should be
* deleted (metadata.iceberg.delete-after-commit.enabled is false, meaning every version is
* retained forever). Shared by {@link #expireAllBefore} (manifest/manifest-list cleanup) and
* {@link #deleteApplicableMetadataFiles} (JSON cleanup) so the two never disagree about what
* counts as "still retained" -- a JSON version that survives must never have the manifests it
* references deleted out from under it.
*/
private long earliestMetadataIdToDelete(long snapshotId) {
Options options = new Options(table.options());
if (!options.get(IcebergOptions.METADATA_DELETE_AFTER_COMMIT)) {
return -1;
}
return snapshotId - options.get(IcebergOptions.METADATA_PREVIOUS_VERSIONS_MAX);
}

@Override
public void notifyCreation(String tagName) {
// The base TagCallback API does not carry a snapshot id, but Iceberg refs
Expand Down
183 changes: 183 additions & 0 deletions paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -471,6 +471,52 @@ public void testRetryCreateMetadata() throws Exception {
commit.close();
}

@Test
public void testCreateMetadataFallsBackWhenBaseManifestListIsMissing() throws Exception {
RowType rowType =
RowType.of(
new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"});
FileStoreTable table =
createPaimonTable(
rowType, Collections.emptyList(), Collections.singletonList("k"), 1);

String commitUser = UUID.randomUUID().toString();
TableWriteImpl<?> write = table.newWrite(commitUser);
TableCommitImpl commit = table.newCommit(commitUser);

write.write(GenericRow.of(1, 10));
write.write(GenericRow.of(2, 20));
commit.commit(1, write.prepareCommit(false, 1));
assertThat(getIcebergResult()).containsExactlyInAnyOrder("Record(1, 10)", "Record(2, 20)");

// The next commit will use this metadata file (snapshot 1's) as its base. Simulate
// unrelated, later retention cleanup having already pruned the manifest list that this
// base's own current snapshot points to, even though the base metadata file itself is
// still present and otherwise healthy.
IcebergPathFactory pathFactory =
new IcebergPathFactory(new Path(table.location(), "metadata"));
Path baseMetadataPath = pathFactory.toMetadataPath(1);
assertThat(table.fileIO().exists(baseMetadataPath)).isTrue();
IcebergMetadata baseMetadata = IcebergMetadata.fromPath(table.fileIO(), baseMetadataPath);
Path danglingManifestListPath =
pathFactory.toManifestListPath(baseMetadata.currentSnapshot().manifestList());
assertThat(table.fileIO().exists(danglingManifestListPath)).isTrue();
table.fileIO().deleteQuietly(danglingManifestListPath);

// Committing the next snapshot must not crash: createMetadataWithBase() will fail to
// read the now-missing manifest list, and the fallback must rebuild metadata from
// scratch instead of propagating the failure.
write.write(GenericRow.of(1, 11));
write.write(GenericRow.of(3, 30));
write.compact(BinaryRow.EMPTY_ROW, 0, true);
commit.commit(2, write.prepareCommit(true, 2));
assertThat(getIcebergResult())
.containsExactlyInAnyOrder("Record(1, 11)", "Record(2, 20)", "Record(3, 30)");

write.close();
commit.close();
}

@Test
public void testExpireAllBeforeSkipsAlreadyDeletedManifestList() throws Exception {
RowType rowType =
Expand Down Expand Up @@ -519,6 +565,143 @@ public void testExpireAllBeforeSkipsAlreadyDeletedManifestList() throws Exceptio
commit.close();
}

@Test
public void testExpireAllBeforeRespectsPreviousVersionsMaxRetentionFloor() throws Exception {
RowType rowType =
RowType.of(
new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"});
FileStoreTable table =
createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1)
.copy(
Collections.singletonMap(
IcebergOptions.METADATA_PREVIOUS_VERSIONS_MAX.key(), "2"));

String commitUser = UUID.randomUUID().toString();
TableWriteImpl<?> write = table.newWrite(commitUser);
TableCommitImpl commit = table.newCommit(commitUser);

// Three ordinary commits: each one after the first finds its base metadata present, so
// they take the safe createMetadataWithBase() path, not expireAllBefore(). With
// previous-versions-max = 2, nothing has crossed the retention floor yet
// (earliestMetadataId = 3 - 2 = 1, so only ids below 1 -- none -- would be pruned).
for (int i = 1; i <= 3; i++) {
write.write(GenericRow.of(i, i * 10));
commit.commit(i, write.prepareCommit(false, i));
}

IcebergPathFactory pathFactory =
new IcebergPathFactory(new Path(table.location(), "metadata"));

// v2 is about to cross the retention floor once snapshot 5 is committed
// (earliestMetadataId = 5 - 2 = 3, so ids below 3 are pruned). Capture its own manifest
// list now, before that happens, so we can confirm below that it is still correctly
// cleaned up -- the fix must not turn cleanup into a no-op.
Path expiredMetadataPath = pathFactory.toMetadataPath(2);
IcebergMetadata expiredMetadata =
IcebergMetadata.fromPath(table.fileIO(), expiredMetadataPath);
Path expiredListPath =
pathFactory.toManifestListPath(expiredMetadata.currentSnapshot().manifestList());
assertThat(table.fileIO().exists(expiredListPath)).isTrue();

// One more ordinary commit. v3 becomes the version previous-versions-max = 2 keeps
// around once snapshot 4 is current (earliestMetadataId = 4 - 2 = 2). Capture the
// manifest list its own current snapshot points to -- this is the exact file the bug
// deleted out from under a retained version.
write.write(GenericRow.of(4, 40));
commit.commit(4, write.prepareCommit(false, 4));

Path retainedMetadataPath = pathFactory.toMetadataPath(3);
assertThat(table.fileIO().exists(retainedMetadataPath)).isTrue();
IcebergMetadata retainedMetadata =
IcebergMetadata.fromPath(table.fileIO(), retainedMetadataPath);
Path retainedListPath =
pathFactory.toManifestListPath(retainedMetadata.currentSnapshot().manifestList());
assertThat(table.fileIO().exists(retainedListPath)).isTrue();

// Force the next commit down the from-scratch / expireAllBefore() path by dropping its
// base metadata (v4), simulating the retry/rebuild scenario that triggers the bug.
Path baseMetadataPath = pathFactory.toMetadataPath(4);
assertThat(table.fileIO().exists(baseMetadataPath)).isTrue();
table.fileIO().deleteQuietly(baseMetadataPath);

write.write(GenericRow.of(5, 50));
commit.commit(5, write.prepareCommit(false, 5));

// v3's own JSON and the manifest list its current snapshot points to must both still
// exist -- the bug let the manifest-deletion loop run ahead of the JSON-retention floor
// and delete these out from under a retained version.
assertThat(table.fileIO().exists(retainedMetadataPath)).isTrue();
assertThat(table.fileIO().exists(retainedListPath)).isTrue();

// Regression check: v2 is genuinely outside the retention window (below the new
// earliestMetadataId = 5 - 2 = 3 floor), so both its JSON and its own manifest list are
// still correctly cleaned up by this same expireAllBefore() call -- the fix must not
// turn cleanup into a no-op.
assertThat(table.fileIO().exists(expiredMetadataPath)).isFalse();
assertThat(table.fileIO().exists(expiredListPath)).isFalse();

assertThat(getIcebergResult())
.containsExactlyInAnyOrder(
"Record(1, 10)",
"Record(2, 20)",
"Record(3, 30)",
"Record(4, 40)",
"Record(5, 50)");

write.close();
commit.close();
}

@Test
public void testExpireAllBeforeDeletesNothingWhenDeleteAfterCommitDisabled() throws Exception {
RowType rowType =
RowType.of(
new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"});
FileStoreTable table =
createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1)
.copy(
Collections.singletonMap(
IcebergOptions.METADATA_DELETE_AFTER_COMMIT.key(),
"false"));

String commitUser = UUID.randomUUID().toString();
TableWriteImpl<?> write = table.newWrite(commitUser);
TableCommitImpl commit = table.newCommit(commitUser);

write.write(GenericRow.of(1, 10));
commit.commit(1, write.prepareCommit(false, 1));
write.write(GenericRow.of(2, 20));
commit.commit(2, write.prepareCommit(false, 2));

IcebergPathFactory pathFactory =
new IcebergPathFactory(new Path(table.location(), "metadata"));
Path oldMetadataPath = pathFactory.toMetadataPath(1);
assertThat(table.fileIO().exists(oldMetadataPath)).isTrue();
IcebergMetadata oldMetadata = IcebergMetadata.fromPath(table.fileIO(), oldMetadataPath);
Path oldListPath =
pathFactory.toManifestListPath(oldMetadata.currentSnapshot().manifestList());
assertThat(table.fileIO().exists(oldListPath)).isTrue();

// Force the next commit down the from-scratch / expireAllBefore() path.
Path baseMetadataPath = pathFactory.toMetadataPath(2);
table.fileIO().deleteQuietly(baseMetadataPath);

write.write(GenericRow.of(3, 30));
commit.commit(3, write.prepareCommit(false, 3));

// delete-after-commit is disabled: every metadata JSON is retained forever, so nothing
// expireAllBefore() touches may be deleted either -- v1's JSON and its manifest list
// must both survive.
assertThat(table.fileIO().exists(oldMetadataPath)).isTrue();
assertThat(table.fileIO().exists(oldListPath)).isTrue();

assertThat(getIcebergResult())
.containsExactlyInAnyOrder("Record(1, 10)", "Record(2, 20)", "Record(3, 30)");

write.close();
commit.close();
}

@Test
public void testCommitAfterRollbackDoesNotDuplicateSchemas() throws Exception {
RowType rowType =
Expand Down
Loading