[core] Preserving old Iceberg Metadata files (apache#5228)

Author: junmuz

docs/layouts/shortcodes/generated/iceberg_configuration.html

@@ -30,20 +30,26 @@
             <td><h5>metadata.iceberg.compaction.max.file-num</h5></td>
             <td style="word-wrap: break-word;">50</td>
             <td>Integer</td>
-            <td>If number of small Iceberg metadata files exceeds this limit, always trigger metadata compaction regardless of their total size.</td>
+            <td>If number of small Iceberg manifest metadata files exceeds this limit, always trigger manifest metadata compaction regardless of their total size.</td>
         </tr>
         <tr>
             <td><h5>metadata.iceberg.compaction.min.file-num</h5></td>
             <td style="word-wrap: break-word;">10</td>
             <td>Integer</td>
-            <td>Minimum number of Iceberg metadata files to trigger metadata compaction.</td>
+            <td>Minimum number of Iceberg manifest metadata files to trigger manifest metadata compaction.</td>
         </tr>
         <tr>
             <td><h5>metadata.iceberg.database</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
             <td>String</td>
             <td>Metastore database name for Iceberg Catalog. Set this as an iceberg database alias if using a centralized Catalog.</td>
         </tr>
+        <tr>
+            <td><h5>metadata.iceberg.delete-after-commit.enabled</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>Whether to delete old metadata files after each table commit</td>
+        </tr>
         <tr>
             <td><h5>metadata.iceberg.glue.skip-archive</h5></td>
             <td style="word-wrap: break-word;">false</td>
@@ -80,6 +86,12 @@
             <td>Boolean</td>
             <td>Should use the legacy manifest version to generate Iceberg's 1.4 manifest files.</td>
         </tr>
+        <tr>
+            <td><h5>metadata.iceberg.previous-versions-max</h5></td>
+            <td style="word-wrap: break-word;">0</td>
+            <td>Integer</td>
+            <td>The number of old metadata files to keep after each table commit</td>
+        </tr>
         <tr>
             <td><h5>metadata.iceberg.storage</h5></td>
             <td style="word-wrap: break-word;">disabled</td>

paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java

@@ -424,7 +424,7 @@ private void createMetadataWithBase(
                         new Path(pathFactory.metadataDirectory(), VERSION_HINT_FILENAME),
                         String.valueOf(snapshotId));
 
-        table.fileIO().deleteQuietly(baseMetadataPath);
+        deleteApplicableMetadataFiles(snapshotId);
         for (int i = 0; i + 1 < toExpireExceptLast.size(); i++) {
             expireManifestList(
                     new Path(toExpireExceptLast.get(i).manifestList()).getName(),
@@ -752,7 +752,25 @@ private void expireAllBefore(long snapshotId) throws IOException {
                 }
                 table.fileIO().deleteQuietly(listPath);
             }
-            table.fileIO().deleteQuietly(path);
+            deleteApplicableMetadataFiles(snapshotId);
+        }
+    }
+
+    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);
+                }
+            }
         }
     }
 

paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java

@@ -42,15 +42,29 @@ public class IcebergOptions {
                     .intType()
                     .defaultValue(10)
                     .withDescription(
-                            "Minimum number of Iceberg metadata files to trigger metadata compaction.");
+                            "Minimum number of Iceberg manifest metadata files to trigger manifest metadata compaction.");
 
     public static final ConfigOption<Integer> COMPACT_MAX_FILE_NUM =
             ConfigOptions.key("metadata.iceberg.compaction.max.file-num")
                     .intType()
                     .defaultValue(50)
                     .withDescription(
-                            "If number of small Iceberg metadata files exceeds this limit, "
-                                    + "always trigger metadata compaction regardless of their total size.");
+                            "If number of small Iceberg manifest metadata files exceeds this limit, "
+                                    + "always trigger manifest metadata compaction regardless of their total size.");
+
+    public static final ConfigOption<Boolean> METADATA_DELETE_AFTER_COMMIT =
+            key("metadata.iceberg.delete-after-commit.enabled")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "Whether to delete old metadata files after each table commit");
+
+    public static final ConfigOption<Integer> METADATA_PREVIOUS_VERSIONS_MAX =
+            key("metadata.iceberg.previous-versions-max")
+                    .intType()
+                    .defaultValue(0)
+                    .withDescription(
+                            "The number of old metadata files to keep after each table commit");
 
     public static final ConfigOption<String> URI =
             key("metadata.iceberg.uri")

paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java

@@ -440,15 +440,21 @@ public void testIcebergSnapshotExpire() throws Exception {
         write.close();
         commit.close();
 
-        // The old metadata.json is removed when the new metadata.json is created.
-        for (int i = 1; i <= 4; i++) {
+        // The old metadata.json is removed when the new metadata.json is created
+        // depending on the old metadata retention configuration.
+        for (int i = 1; i <= 3; i++) {
             unusedFiles.add(pathFactory.toMetadataPath(i).toString());
         }
 
         for (String path : unusedFiles) {
             assertThat(fileIO.exists(new Path(path))).isFalse();
         }
 
+        // Check existence of retained Iceberg metadata.json files
+        for (int i = 4; i <= 5; i++) {
+            assertThat(fileIO.exists(new Path(pathFactory.toMetadataPath(i).toString()))).isTrue();
+        }
+
         // Test all existing Iceberg snapshots are valid.
         assertThat(getIcebergResult())
                 .containsExactlyInAnyOrder("Record(1, 11)", "Record(2, 21)", "Record(3, 31)");
@@ -961,6 +967,8 @@ private FileStoreTable createPaimonTable(
         options.set(CoreOptions.TARGET_FILE_SIZE, MemorySize.ofKibiBytes(32));
         options.set(IcebergOptions.COMPACT_MIN_FILE_NUM, 4);
         options.set(IcebergOptions.COMPACT_MIN_FILE_NUM, 8);
+        options.set(IcebergOptions.METADATA_DELETE_AFTER_COMMIT, true);
+        options.set(IcebergOptions.METADATA_PREVIOUS_VERSIONS_MAX, 1);
         options.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE, MemorySize.ofKibiBytes(8));
         Schema schema =
                 new Schema(rowType.getFields(), partitionKeys, primaryKeys, options.toMap(), "");