[lake] Support partition mark-done for lakehouse tiering (processing-time)

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -2520,6 +2520,15 @@ public class ConfigOptions {
                                     + ConfigOptions.TABLE_DATALAKE_AUTO_EXPIRE_SNAPSHOT
                                     + " is false.");
 
+    public static final ConfigOption<Boolean> LAKE_TIERING_PARTITION_MARK_DONE_ENABLED =
+            key("lake.tiering.partition.mark-done.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether the tiering service marks idle partitions of tiered partitioned tables as done. "
+                                    + "Disabled by default. When enabled, a table opts in via its lake-format prefixed "
+                                    + "mark-done custom properties (e.g. 'paimon.partition.idle-time-to-done' for Paimon).");
+
     public static final ConfigOption<String> LAKE_TIERING_IO_TMP_DIRS =
             key("lake.tiering.io.tmp.dirs")
                     .stringType()

fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionMarkDoneMaintainer.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.committer;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.utils.function.SupplierWithException;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * An optional capability that a {@link LakeCommitter} may implement to persist the partition
+ * mark-done state for a tiering round without any data to commit.
+ */
+@Internal
+public interface PartitionMarkDoneMaintainer {
+
+    /**
+     * Performs partition mark-done maintenance for a tiering round without any data to commit.
+     *
+     * @param offsetsFileProvider provides a freshly prepared bucket offsets file for the
+     *     maintenance snapshot; it is only invoked when a snapshot will actually be created. Every
+     *     snapshot must carry its own offsets file since offsets files are deleted along with their
+     *     snapshot metadata and thus must not be shared across snapshots.
+     * @return the properties-only lake snapshot created to persist the mark-done state, or null if
+     *     no snapshot was created (feature disabled or state unchanged)
+     * @throws IOException if an I/O error occurs
+     */
+    @Nullable
+    CommittedLakeSnapshot commitMarkDoneMaintenance(
+            SupplierWithException<String, IOException> offsetsFileProvider) throws IOException;
+}

fluss-common/src/main/java/org/apache/fluss/lake/writer/PartitionMarkDoneEnabler.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.writer;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.metadata.TableInfo;
+
+/**
+ * An optional capability that a {@link LakeTieringFactory} may implement to support marking idle
+ * partitions of a tiered table as done. Lake formats that do not implement this interface never
+ * trigger mark-done maintenance.
+ */
+@Internal
+public interface PartitionMarkDoneEnabler {
+
+    /**
+     * Whether partition mark-done is enabled for the given table, used to decide whether an empty
+     * tiering round should still reach the committer to run mark-done maintenance.
+     *
+     * <p>The switch only recognizes the Fluss table metadata (e.g. custom properties): mark-done
+     * options configured on the lake table directly are deliberately not honored, so the check is
+     * cheap (no lake access) and there is a single source of truth.
+     */
+    boolean isPartitionMarkDoneEnabled(TableInfo tableInfo);
+}

fluss-common/src/main/java/org/apache/fluss/utils/PartitionUtils.java

@@ -439,7 +439,7 @@ public String toString() {
     }
 
     /** Returns the time string format pattern for the given time unit. */
-    private static String getPartitionTimeFormat(
+    public static String getPartitionTimeFormat(
             AutoPartitionTimeUnit timeUnit, AutoPartitionStrategy autoPartitionStrategy) {
         String timeFormat = autoPartitionStrategy.timeFormat();
         if (timeFormat != null) {

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java

@@ -21,6 +21,7 @@
 import org.apache.fluss.client.ConnectionFactory;
 import org.apache.fluss.client.admin.Admin;
 import org.apache.fluss.client.metadata.LakeSnapshot;
+import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.LakeTableSnapshotNotExistException;
 import org.apache.fluss.flink.tiering.event.FailedTieringEvent;
@@ -30,9 +31,11 @@
 import org.apache.fluss.lake.committer.CommittedLakeSnapshot;
 import org.apache.fluss.lake.committer.LakeCommitResult;
 import org.apache.fluss.lake.committer.LakeCommitter;
+import org.apache.fluss.lake.committer.PartitionMarkDoneMaintainer;
 import org.apache.fluss.lake.committer.TieringStats;
 import org.apache.fluss.lake.writer.LakeTieringFactory;
 import org.apache.fluss.lake.writer.LakeWriter;
+import org.apache.fluss.lake.writer.PartitionMarkDoneEnabler;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
@@ -219,6 +222,8 @@ private CommitResult commitWriteResults(
                     "Commit tiering write results is empty for table {}, table path {}",
                     tableId,
                     tablePath);
+            // still run partition mark-done maintenance for the empty round
+            maybeCommitMarkDoneMaintenance(tableId, tablePath);
             return new CommitResult(null, null);
         }
 
@@ -290,6 +295,66 @@ private CommitResult commitWriteResults(
         }
     }
 
+    /**
+     * Runs partition mark-done maintenance for an empty tiering round, and commits the resulting
+     * properties-only lake snapshot (if any) to Fluss.
+     */
+    private void maybeCommitMarkDoneMaintenance(long tableId, TablePath tablePath)
+            throws Exception {
+        if (!lakeTieringConfig.get(ConfigOptions.LAKE_TIERING_PARTITION_MARK_DONE_ENABLED)
+                || !(lakeTieringFactory instanceof PartitionMarkDoneEnabler)) {
+            return;
+        }
+        TableInfo tableInfo = admin.getTableInfo(tablePath).get();
+        if (tableInfo.getTableId() != tableId
+                || !((PartitionMarkDoneEnabler) lakeTieringFactory)
+                        .isPartitionMarkDoneEnabled(tableInfo)) {
+            return;
+        }
+        try (LakeCommitter<WriteResult, Committable> lakeCommitter =
+                lakeTieringFactory.createLakeCommitter(
+                        new TieringCommitterInitContext(
+                                tablePath, tableInfo, lakeTieringConfig, flussConfig))) {
+            if (!(lakeCommitter instanceof PartitionMarkDoneMaintainer)) {
+                return;
+            }
+            // first bring Fluss up to date in case a previous round committed to the lake but
+            // failed to commit to Fluss; otherwise the maintenance would see an unchanged
+            // state, keep returning null and Fluss would stay on the old snapshot forever
+            LakeSnapshot flussCurrentLakeSnapshot = getLatestLakeSnapshot(tablePath);
+            CommittedLakeSnapshot missingCommittedSnapshot =
+                    lakeCommitter.getMissingLakeSnapshot(
+                            flussCurrentLakeSnapshot == null
+                                    ? null
+                                    : flussCurrentLakeSnapshot.getSnapshotId());
+            if (missingCommittedSnapshot != null) {
+                commitMissingLakeSnapshotToFluss(tablePath, tableId, missingCommittedSnapshot);
+            }
+            CommittedLakeSnapshot maintenanceSnapshot =
+                    ((PartitionMarkDoneMaintainer) lakeCommitter)
+                            .commitMarkDoneMaintenance(
+                                    // a fresh offsets file for the maintenance snapshot, since
+                                    // offsets files are deleted along with their snapshot
+                                    // metadata and must not be shared across snapshots
+                                    () ->
+                                            flussTableLakeSnapshotCommitter.prepareLakeSnapshot(
+                                                    tableId, tablePath, Collections.emptyMap()));
+            if (maintenanceSnapshot != null) {
+                flussTableLakeSnapshotCommitter.commit(
+                        tableId,
+                        maintenanceSnapshot.getLakeSnapshotId(),
+                        maintenanceSnapshot
+                                .getSnapshotProperties()
+                                .get(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY),
+                        null,
+                        // no data was written in this round
+                        Collections.emptyMap(),
+                        Collections.emptyMap(),
+                        LakeCommitResult.KEEP_ALL_PREVIOUS);
+            }
+        }
+    }
+
     @Nullable
     private LakeSnapshot getLatestLakeSnapshot(TablePath tablePath) throws Exception {
         LakeSnapshot flussCurrentLakeSnapshot;
@@ -323,48 +388,8 @@ private void checkFlussNotMissingLakeSnapshot(
         // known lake snapshot, which means the data already has been committed to lake,
         // not to commit to lake to avoid data duplicated
         if (missingCommittedSnapshot != null) {
-            String lakeSnapshotOffsetPath =
-                    missingCommittedSnapshot
-                            .getSnapshotProperties()
-                            .get(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY);
-
-            // should only will happen in v0.7 which won't put offsets info
-            // to properties
-            if (lakeSnapshotOffsetPath == null) {
-                throw new IllegalStateException(
-                        String.format(
-                                "Can't find %s field from snapshot property.",
-                                FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY));
-            }
-
-            // the fluss-offsets will be a json string if it's tiered by v0.8,
-            // since this code path should be rare, we do not consider backward compatibility
-            // and throw IllegalStateException directly
-            String trimmedPath = lakeSnapshotOffsetPath.trim();
-            if (trimmedPath.contains("{")) {
-                throw new IllegalStateException(
-                        String.format(
-                                "The %s field in snapshot property is a JSON string (tiered by v0.8), "
-                                        + "which is not supported to restore. Snapshot ID: %d, Table: {tablePath=%s, tableId=%d}.",
-                                FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,
-                                missingCommittedSnapshot.getLakeSnapshotId(),
-                                tablePath,
-                                tableId));
-            }
-
             // commit this missing snapshot to fluss
-            flussTableLakeSnapshotCommitter.commit(
-                    tableId,
-                    missingCommittedSnapshot.getLakeSnapshotId(),
-                    lakeSnapshotOffsetPath,
-                    // don't care readable snapshot and offsets,
-                    null,
-                    // use empty log offsets, log max timestamp, since we can't know that
-                    // in last tiering, it doesn't matter for they are just used to
-                    // report metrics
-                    Collections.emptyMap(),
-                    Collections.emptyMap(),
-                    LakeCommitResult.KEEP_ALL_PREVIOUS);
+            commitMissingLakeSnapshotToFluss(tablePath, tableId, missingCommittedSnapshot);
             // abort this committable to delete the written files
             lakeCommitter.abort(committable);
             throw new IllegalStateException(
@@ -380,6 +405,53 @@ private void checkFlussNotMissingLakeSnapshot(
         }
     }
 
+    /** Commits a lake snapshot committed by Fluss but unknown to Fluss back to Fluss. */
+    private void commitMissingLakeSnapshotToFluss(
+            TablePath tablePath, long tableId, CommittedLakeSnapshot missingCommittedSnapshot)
+            throws Exception {
+        String lakeSnapshotOffsetPath =
+                missingCommittedSnapshot
+                        .getSnapshotProperties()
+                        .get(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY);
+
+        // should only will happen in v0.7 which won't put offsets info
+        // to properties
+        if (lakeSnapshotOffsetPath == null) {
+            throw new IllegalStateException(
+                    String.format(
+                            "Can't find %s field from snapshot property.",
+                            FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY));
+        }
+
+        // the fluss-offsets will be a json string if it's tiered by v0.8,
+        // since this code path should be rare, we do not consider backward compatibility
+        // and throw IllegalStateException directly
+        String trimmedPath = lakeSnapshotOffsetPath.trim();
+        if (trimmedPath.contains("{")) {
+            throw new IllegalStateException(
+                    String.format(
+                            "The %s field in snapshot property is a JSON string (tiered by v0.8), "
+                                    + "which is not supported to restore. Snapshot ID: %d, Table: {tablePath=%s, tableId=%d}.",
+                            FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,
+                            missingCommittedSnapshot.getLakeSnapshotId(),
+                            tablePath,
+                            tableId));
+        }
+
+        flussTableLakeSnapshotCommitter.commit(
+                tableId,
+                missingCommittedSnapshot.getLakeSnapshotId(),
+                lakeSnapshotOffsetPath,
+                // don't care readable snapshot and offsets,
+                null,
+                // use empty log offsets, log max timestamp, since we can't know that
+                // in last tiering, it doesn't matter for they are just used to
+                // report metrics
+                Collections.emptyMap(),
+                Collections.emptyMap(),
+                LakeCommitResult.KEEP_ALL_PREVIOUS);
+    }
+
     private void registerTableBucketWriteResult(
             long tableId, TableBucketWriteResult<WriteResult> tableBucketWriteResult) {
         collectedTableBucketWriteResults

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java

@@ -88,7 +88,11 @@ public Boundedness getBoundedness() {
     public SplitEnumerator<TieringSplit, TieringSourceEnumeratorState> createEnumerator(
             SplitEnumeratorContext<TieringSplit> splitEnumeratorContext) {
         return new TieringSourceEnumerator(
-                flussConf, splitEnumeratorContext, lakeTieringFactory, pollTieringTableIntervalMs);
+                flussConf,
+                lakeTieringConfig,
+                splitEnumeratorContext,
+                lakeTieringFactory,
+                pollTieringTableIntervalMs);
     }
 
     @Override
@@ -97,7 +101,11 @@ public SplitEnumerator<TieringSplit, TieringSourceEnumeratorState> restoreEnumer
             TieringSourceEnumeratorState tieringSourceEnumeratorState) {
         // stateless operator
         return new TieringSourceEnumerator(
-                flussConf, splitEnumeratorContext, lakeTieringFactory, pollTieringTableIntervalMs);
+                flussConf,
+                lakeTieringConfig,
+                splitEnumeratorContext,
+                lakeTieringFactory,
+                pollTieringTableIntervalMs);
     }
 
     @Override