[lake] Support auto snapshot expiration for lake table

LiebingYu · LiebingYu · commit ce821a5d3742 · 2025-12-18T15:07:23.000+08:00
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -1381,6 +1381,13 @@ public class ConfigOptions {
                     .withDescription(
                             "If true, compaction will be triggered automatically when tiering service writes to the datalake. It is disabled by default.");
 
+    public static final ConfigOption<Boolean> TABLE_DATALAKE_AUTO_SNAPSHOT_EXPIRATION =
+            key("table.datalake.auto-snapshot-expiration")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "If true, snapshot expiration will be triggered automatically when tiering service writes to the datalake. It is disabled by default.");
+
     public static final ConfigOption<MergeEngineType> TABLE_MERGE_ENGINE =
             key("table.merge-engine")
                     .enumType(MergeEngineType.class)
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java
@@ -100,6 +100,11 @@ public boolean isDataLakeAutoCompaction() {
         return config.get(ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION);
     }
 
+    /** Whether auto snapshot expiration is enabled. */
+    public boolean isDataLakeAutoSnapshotExpiration() {
+        return config.get(ConfigOptions.TABLE_DATALAKE_AUTO_SNAPSHOT_EXPIRATION);
+    }
+
     /** Gets the optional merge engine type of the table. */
     public Optional<MergeEngineType> getMergeEngineType() {
         return config.getOptional(ConfigOptions.TABLE_MERGE_ENGINE);
diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/CommitterInitContext.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/CommitterInitContext.java
@@ -18,11 +18,12 @@
 package org.apache.fluss.lake.committer;
 
 import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
 
 /**
  * The CommitterInitContext interface provides the context needed to create a LakeCommitter. It
- * includes methods to obtain the table path.
+ * includes methods to obtain the table path and table info.
  *
  * @since 0.7
  */
@@ -35,4 +36,11 @@ public interface CommitterInitContext {
      * @return the table path
      */
     TablePath tablePath();
+
+    /**
+     * Returns the Fluss table info.
+     *
+     * @return the Fluss table info
+     */
+    TableInfo tableInfo();
 }
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java
@@ -205,7 +205,8 @@ private Committable commitWriteResults(
         }
         try (LakeCommitter<WriteResult, Committable> lakeCommitter =
                 lakeTieringFactory.createLakeCommitter(
-                        new TieringCommitterInitContext(tablePath))) {
+                        new TieringCommitterInitContext(
+                                tablePath, admin.getTableInfo(tablePath).get()))) {
             List<WriteResult> writeResults =
                     committableWriteResults.stream()
                             .map(TableBucketWriteResult::writeResult)
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitterInitContext.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitterInitContext.java
@@ -19,19 +19,27 @@
 
 import org.apache.fluss.lake.committer.CommitterInitContext;
 import org.apache.fluss.lake.committer.LakeCommitter;
+import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
 
 /** The {@link CommitterInitContext} implementation for {@link LakeCommitter}. */
 public class TieringCommitterInitContext implements CommitterInitContext {
 
     private final TablePath tablePath;
+    private final TableInfo tableInfo;
 
-    public TieringCommitterInitContext(TablePath tablePath) {
+    public TieringCommitterInitContext(TablePath tablePath, TableInfo tableInfo) {
         this.tablePath = tablePath;
+        this.tableInfo = tableInfo;
     }
 
     @Override
     public TablePath tablePath() {
         return tablePath;
     }
+
+    @Override
+    public TableInfo tableInfo() {
+        return tableInfo;
+    }
 }
diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java
@@ -19,6 +19,7 @@
 
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.lake.committer.CommitterInitContext;
 import org.apache.fluss.lake.committer.LakeCommitter;
 import org.apache.fluss.lake.serializer.SimpleVersionedSerializer;
 import org.apache.fluss.lake.writer.LakeWriter;
@@ -185,7 +186,7 @@ void testTieringWriteTable(boolean isPrimaryKeyTable, boolean isPartitionedTable
 
         // second, commit data
         try (LakeCommitter<IcebergWriteResult, IcebergCommittable> lakeCommitter =
-                createLakeCommitter(tablePath)) {
+                createLakeCommitter(tablePath, tableInfo)) {
             // serialize/deserialize committable
             IcebergCommittable icebergCommittable =
                     lakeCommitter.toCommittable(icebergWriteResults);
@@ -246,8 +247,19 @@ public TableInfo tableInfo() {
     }
 
     private LakeCommitter<IcebergWriteResult, IcebergCommittable> createLakeCommitter(
-            TablePath tablePath) throws IOException {
-        return icebergLakeTieringFactory.createLakeCommitter(() -> tablePath);
+            TablePath tablePath, TableInfo tableInfo) throws IOException {
+        return icebergLakeTieringFactory.createLakeCommitter(
+                new CommitterInitContext() {
+                    @Override
+                    public TablePath tablePath() {
+                        return tablePath;
+                    }
+
+                    @Override
+                    public TableInfo tableInfo() {
+                        return tableInfo;
+                    }
+                });
     }
 
     private Tuple2<List<LogRecord>, List<LogRecord>> genLogTableRecords(
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeCommitter.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeCommitter.java
@@ -19,6 +19,7 @@
 
 import org.apache.fluss.lake.committer.BucketOffset;
 import org.apache.fluss.lake.committer.CommittedLakeSnapshot;
+import org.apache.fluss.lake.committer.CommitterInitContext;
 import org.apache.fluss.lake.committer.LakeCommitter;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
@@ -32,15 +33,15 @@
 import org.apache.paimon.manifest.ManifestCommittable;
 import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.manifest.SimpleFileEntry;
-import org.apache.paimon.operation.FileStoreCommit;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.sink.CommitCallback;
+import org.apache.paimon.table.sink.TableCommitImpl;
 import org.apache.paimon.utils.SnapshotManager;
 
 import javax.annotation.Nullable;
 
 import java.io.IOException;
-import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -55,14 +56,21 @@ public class PaimonLakeCommitter implements LakeCommitter<PaimonWriteResult, Pai
 
     private final Catalog paimonCatalog;
     private final FileStoreTable fileStoreTable;
-    private FileStoreCommit fileStoreCommit;
+    private TableCommitImpl tableCommit;
     private static final ThreadLocal<Long> currentCommitSnapshotId = new ThreadLocal<>();
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
-    public PaimonLakeCommitter(PaimonCatalogProvider paimonCatalogProvider, TablePath tablePath)
+    public PaimonLakeCommitter(
+            PaimonCatalogProvider paimonCatalogProvider, CommitterInitContext committerInitContext)
             throws IOException {
         this.paimonCatalog = paimonCatalogProvider.get();
-        this.fileStoreTable = getTable(tablePath);
+        this.fileStoreTable =
+                getTable(
+                        committerInitContext.tablePath(),
+                        committerInitContext
+                                .tableInfo()
+                                .getTableConfig()
+                                .isDataLakeAutoSnapshotExpiration());
     }
 
     @Override
@@ -82,29 +90,26 @@ public long commit(PaimonCommittable committable, Map<String, String> snapshotPr
         snapshotProperties.forEach(manifestCommittable::addProperty);
 
         try {
-            fileStoreCommit =
-                    fileStoreTable
-                            .store()
-                            .newCommit(FLUSS_LAKE_TIERING_COMMIT_USER, fileStoreTable);
-            fileStoreCommit.commit(manifestCommittable, false);
+            tableCommit = fileStoreTable.newCommit(FLUSS_LAKE_TIERING_COMMIT_USER);
+            tableCommit.commit(manifestCommittable);
+
             Long commitSnapshotId = currentCommitSnapshotId.get();
             currentCommitSnapshotId.remove();
 
             return checkNotNull(commitSnapshotId, "Paimon committed snapshot id must be non-null.");
         } catch (Throwable t) {
-            if (fileStoreCommit != null) {
+            if (tableCommit != null) {
                 // if any error happen while commit, abort the commit to clean committable
-                fileStoreCommit.abort(manifestCommittable.fileCommittables());
+                tableCommit.abort(manifestCommittable.fileCommittables());
             }
             throw new IOException(t);
         }
     }
 
     @Override
     public void abort(PaimonCommittable committable) throws IOException {
-        fileStoreCommit =
-                fileStoreTable.store().newCommit(FLUSS_LAKE_TIERING_COMMIT_USER, fileStoreTable);
-        fileStoreCommit.abort(committable.manifestCommittable().fileCommittables());
+        tableCommit = fileStoreTable.newCommit(FLUSS_LAKE_TIERING_COMMIT_USER);
+        tableCommit.abort(committable.manifestCommittable().fileCommittables());
     }
 
     @Nullable
@@ -191,8 +196,8 @@ private Snapshot getCommittedLatestSnapshotOfLake(String commitUser) throws IOEx
     @Override
     public void close() throws Exception {
         try {
-            if (fileStoreCommit != null) {
-                fileStoreCommit.close();
+            if (tableCommit != null) {
+                tableCommit.close();
             }
             if (paimonCatalog != null) {
                 paimonCatalog.close();
@@ -202,19 +207,20 @@ public void close() throws Exception {
         }
     }
 
-    private FileStoreTable getTable(TablePath tablePath) throws IOException {
+    private FileStoreTable getTable(TablePath tablePath, boolean isAutoSnapshotExpiration)
+            throws IOException {
         try {
-            FileStoreTable table =
-                    (FileStoreTable)
-                            paimonCatalog
-                                    .getTable(toPaimon(tablePath))
-                                    .copy(
-                                            Collections.singletonMap(
-                                                    CoreOptions.COMMIT_CALLBACKS.key(),
-                                                    PaimonLakeCommitter.PaimonCommitCallback.class
-                                                            .getName()));
-
-            return table;
+            FileStoreTable table = (FileStoreTable) paimonCatalog.getTable(toPaimon(tablePath));
+
+            Map<String, String> dynamicOptions = new HashMap<>();
+            dynamicOptions.put(
+                    CoreOptions.COMMIT_CALLBACKS.key(),
+                    PaimonLakeCommitter.PaimonCommitCallback.class.getName());
+            dynamicOptions.put(
+                    CoreOptions.WRITE_ONLY.key(),
+                    isAutoSnapshotExpiration ? Boolean.FALSE.toString() : Boolean.TRUE.toString());
+
+            return table.copy(dynamicOptions);
         } catch (Exception e) {
             throw new IOException("Failed to get table " + tablePath + " in Paimon.", e);
         }
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeTieringFactory.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeTieringFactory.java
@@ -53,7 +53,7 @@ public SimpleVersionedSerializer<PaimonWriteResult> getWriteResultSerializer() {
     @Override
     public LakeCommitter<PaimonWriteResult, PaimonCommittable> createLakeCommitter(
             CommitterInitContext committerInitContext) throws IOException {
-        return new PaimonLakeCommitter(paimonCatalogProvider, committerInitContext.tablePath());
+        return new PaimonLakeCommitter(paimonCatalogProvider, committerInitContext);
     }
 
     @Override
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java
diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md

Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,7 @@ public SimpleVersionedSerializer<PaimonWriteResult> getWriteResultSerializer() {`
`53`	`53`	`@Override`
`54`	`54`	`public LakeCommitter<PaimonWriteResult, PaimonCommittable> createLakeCommitter(`
`55`	`55`	`CommitterInitContext committerInitContext) throws IOException {`
`56`		`- return new PaimonLakeCommitter(paimonCatalogProvider, committerInitContext.tablePath());`
	`56`	`+ return new PaimonLakeCommitter(paimonCatalogProvider, committerInitContext);`
`57`	`57`	`}`
`58`	`58`
`59`	`59`	`@Override`