diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java index ab731753969e..6734ef57058b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java @@ -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; @@ -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); @@ -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 expiredManifestLists = new HashSet<>(); Set expiredManifestFileMetas = new HashSet<>(); Iterator it = - pathFactory.getAllMetadataPathBefore(table.fileIO(), snapshotId).iterator(); + pathFactory.getAllMetadataPathBefore(table.fileIO(), earliestMetadataId).iterator(); while (it.hasNext()) { Path path = it.next(); @@ -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 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 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 diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java index 84a987ba5503..cc2b87a75035 100644 --- a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java @@ -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 = @@ -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 =