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

fluss-client/src/main/java/org/apache/fluss/client/metadata/LakeSnapshot.java

@@ -18,14 +18,20 @@
 package org.apache.fluss.client.metadata;
 
 import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.lake.committer.LakeTieringTableState;
 import org.apache.fluss.metadata.TableBucket;
 
+import javax.annotation.Nullable;
+
 import java.util.Collections;
 import java.util.Map;
 
 /**
  * A class representing the lake snapshot information of a table. It contains:
  * <li>The snapshot id and the log offset for each bucket.
+ * <li>The opaque table-level tiering state (see {@link LakeTieringTableState}) for partitioned
+ *     tables, used by e.g. lake partition mark-done. It is {@code null} for non-partitioned tables
+ *     or when talking to an old coordinator that does not report it.
  *
  * @since 0.3
  */
@@ -37,9 +43,21 @@ public class LakeSnapshot {
     // the specific log offset of the snapshot
     private final Map<TableBucket, Long> tableBucketsOffset;
 
+    // table-level tiering state; null when non-partitioned table or when the coordinator does not
+    // report it (old version).
+    @Nullable private final LakeTieringTableState lakeTieringTableState;
+
     public LakeSnapshot(long snapshotId, Map<TableBucket, Long> tableBucketsOffset) {
+        this(snapshotId, tableBucketsOffset, null);
+    }
+
+    public LakeSnapshot(
+            long snapshotId,
+            Map<TableBucket, Long> tableBucketsOffset,
+            @Nullable LakeTieringTableState lakeTieringTableState) {
         this.snapshotId = snapshotId;
         this.tableBucketsOffset = tableBucketsOffset;
+        this.lakeTieringTableState = lakeTieringTableState;
     }
 
     public long getSnapshotId() {
@@ -50,13 +68,21 @@ public Map<TableBucket, Long> getTableBucketsOffset() {
         return Collections.unmodifiableMap(tableBucketsOffset);
     }
 
+    /** Returns the table-level tiering state, or {@code null} if absent. */
+    @Nullable
+    public LakeTieringTableState getLakeTieringTableState() {
+        return lakeTieringTableState;
+    }
+
     @Override
     public String toString() {
         return "LakeSnapshot{"
                 + "snapshotId="
                 + snapshotId
                 + ", tableBucketsOffset="
                 + tableBucketsOffset
+                + ", lakeTieringTableState="
+                + lakeTieringTableState
                 + '}';
     }
 }

fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java

@@ -41,6 +41,7 @@
 import org.apache.fluss.fs.FsPath;
 import org.apache.fluss.fs.FsPathAndFileName;
 import org.apache.fluss.fs.token.ObtainedSecurityToken;
+import org.apache.fluss.lake.committer.LakeTieringTableState;
 import org.apache.fluss.metadata.AggFunction;
 import org.apache.fluss.metadata.DatabaseChange;
 import org.apache.fluss.metadata.DatabaseSummary;
@@ -286,7 +287,14 @@ public static LakeSnapshot toLakeTableSnapshotInfo(GetLakeSnapshotResponse respo
                     new TableBucket(tableId, partitionId, pbLakeSnapshotForBucket.getBucketId());
             tableBucketsOffset.put(tableBucket, pbLakeSnapshotForBucket.getLogOffset());
         }
-        return new LakeSnapshot(snapshotId, tableBucketsOffset);
+
+        // parse the opaque table-level tiering state (null when talking to an old coordinator that
+        // does not report tiering_state_json, or when absent).
+        LakeTieringTableState lakeTieringTableState =
+                response.hasTieringStateJson()
+                        ? LakeTieringTableState.fromJsonBytes(response.getTieringStateJson())
+                        : null;
+        return new LakeSnapshot(snapshotId, tableBucketsOffset, lakeTieringTableState);
     }
 
     public static List<FsPathAndFileName> toFsPathAndFileName(

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

@@ -0,0 +1,136 @@
+/*
+ * 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.PublicEvolving;
+import org.apache.fluss.utils.json.JsonSerdeUtils;
+import org.apache.fluss.utils.json.LakeTieringTableStateJsonSerde;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Table-level tiering state carried by the lake offsets file as an opaque {@code tiering_state}
+ * payload (transported as {@code bytes tiering_state_json} in RPC); owned by the lake side and
+ * passed through by the offsets serde without parsing.
+ *
+ * <p>Partition mark-done is the first consumer (delete-on-done model):
+ *
+ * <ul>
+ *   <li>{@code partitionDoneInitialized}: table-level one-shot flag, {@code true} once the
+ *       full-table cold-start back-fill has completed; disambiguates "all done" from "first run".
+ *   <li>{@code partitionUpdateTimes}: last update time of only the not-yet-done partitions, keyed
+ *       by partitionId. A partition is removed once marked done; no {@code doneTime} is persisted.
+ * </ul>
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public class LakeTieringTableState {
+
+    /** The tiering-state format version 1. */
+    public static final int VERSION_1 = 1;
+
+    /** The current tiering-state format version written by this build. */
+    public static final int CURRENT_VERSION = VERSION_1;
+
+    private final int version;
+    private final boolean partitionDoneInitialized;
+    private final Map<Long, Long> partitionUpdateTimes;
+
+    public LakeTieringTableState(
+            boolean partitionDoneInitialized, Map<Long, Long> partitionUpdateTimes) {
+        this(CURRENT_VERSION, partitionDoneInitialized, partitionUpdateTimes);
+    }
+
+    public LakeTieringTableState(
+            int version, boolean partitionDoneInitialized, Map<Long, Long> partitionUpdateTimes) {
+        this.version = version;
+        this.partitionDoneInitialized = partitionDoneInitialized;
+        this.partitionUpdateTimes =
+                partitionUpdateTimes == null
+                        ? Collections.emptyMap()
+                        : new HashMap<>(partitionUpdateTimes);
+    }
+
+    /** Returns the tiering-state format version. */
+    public int getVersion() {
+        return version;
+    }
+
+    /** Returns whether the full-table cold-start back-fill has completed. */
+    public boolean isPartitionDoneInitialized() {
+        return partitionDoneInitialized;
+    }
+
+    /** Returns the last update time of the not-yet-done partitions, keyed by partitionId. */
+    public Map<Long, Long> getPartitionUpdateTimes() {
+        return Collections.unmodifiableMap(partitionUpdateTimes);
+    }
+
+    /**
+     * Serialize to a JSON byte array.
+     *
+     * @see LakeTieringTableStateJsonSerde
+     */
+    public byte[] toJsonBytes() {
+        return JsonSerdeUtils.writeValueAsBytes(this, LakeTieringTableStateJsonSerde.INSTANCE);
+    }
+
+    /**
+     * Deserialize from a JSON byte array.
+     *
+     * @see LakeTieringTableStateJsonSerde
+     */
+    public static LakeTieringTableState fromJsonBytes(byte[] json) {
+        return JsonSerdeUtils.readValue(json, LakeTieringTableStateJsonSerde.INSTANCE);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        LakeTieringTableState that = (LakeTieringTableState) o;
+        return version == that.version
+                && partitionDoneInitialized == that.partitionDoneInitialized
+                && Objects.equals(partitionUpdateTimes, that.partitionUpdateTimes);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(version, partitionDoneInitialized, partitionUpdateTimes);
+    }
+
+    @Override
+    public String toString() {
+        return "LakeTieringTableState{"
+                + "version="
+                + version
+                + ", partitionDoneInitialized="
+                + partitionDoneInitialized
+                + ", partitionUpdateTimes="
+                + partitionUpdateTimes
+                + '}';
+    }
+}

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

@@ -0,0 +1,84 @@
+/*
+ * 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.PublicEvolving;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * A candidate partition passed to {@link PartitionDoneHandler} for mark-done judgement. Transient,
+ * runtime-only input keyed by {@code partitionName} (not to be confused with the persisted {@link
+ * LakeTieringTableState} keyed by partitionId).
+ *
+ * <p>{@code lastUpdateTime} is the aggregated MAX of bucket max-timestamps (the partition's last
+ * write time in processing-time mode); -1 means unknown.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public class PartitionDoneCandidate implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private final String partitionName;
+    private final long lastUpdateTime;
+
+    public PartitionDoneCandidate(String partitionName, long lastUpdateTime) {
+        this.partitionName = partitionName;
+        this.lastUpdateTime = lastUpdateTime;
+    }
+
+    public String partitionName() {
+        return partitionName;
+    }
+
+    public long lastUpdateTime() {
+        return lastUpdateTime;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        PartitionDoneCandidate that = (PartitionDoneCandidate) o;
+        return lastUpdateTime == that.lastUpdateTime
+                && Objects.equals(partitionName, that.partitionName);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(partitionName, lastUpdateTime);
+    }
+
+    @Override
+    public String toString() {
+        return "PartitionDoneCandidate{"
+                + "partitionName='"
+                + partitionName
+                + '\''
+                + ", lastUpdateTime="
+                + lastUpdateTime
+                + '}';
+    }
+}

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

@@ -0,0 +1,57 @@
+/*
+ * 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.PublicEvolving;
+
+import java.util.List;
+
+/**
+ * The handler that decides which partitions can be marked done and executes the lake-specific
+ * mark-done action (e.g. writing a {@code _SUCCESS} file for Paimon).
+ *
+ * <p>This is a lake-specific SPI created by {@link
+ * org.apache.fluss.lake.writer.LakeTieringFactory#createPartitionDoneHandler}. The tiering commit
+ * operator only feeds partitions that are not yet done (those still tracked in the tiering state,
+ * since done partitions are removed under delete-on-done) and relies on this handler to perform the
+ * idle judgement and the actual action.
+ *
+ * <p>Implementations MUST be idempotent: the same partition may be marked done more than once in
+ * rare failure windows (e.g. action executed but the new tiering state not yet persisted before a
+ * crash), so repeatedly executing the action for an already-done partition must be safe.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public interface PartitionDoneHandler extends AutoCloseable {
+
+    /**
+     * Judges which partitions can be marked done and executes the mark-done action (idempotent),
+     * returning the names of the partitions that were newly marked done in this round.
+     *
+     * @param candidates the candidate partitions that are not yet done
+     * @param currentTime the current time (System.currentTimeMillis() in processing-time mode)
+     * @return the names of partitions newly marked done in this round
+     * @throws Exception if the mark-done action fails
+     */
+    List<String> markDoneIfReady(List<PartitionDoneCandidate> candidates, long currentTime)
+            throws Exception;
+
+    @Override
+    default void close() throws Exception {}
+}