[lake] Carry opaque table-level tiering state in lake offsets file and RPC

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

@@ -18,14 +18,22 @@
 package org.apache.fluss.client.metadata;
 
 import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.lake.committer.PartitionMarkDoneState;
+import org.apache.fluss.lake.committer.TieringStateEntry;
 import org.apache.fluss.metadata.TableBucket;
 
+import javax.annotation.Nullable;
+
 import java.util.Collections;
+import java.util.List;
 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 keyed tiering-state entries, exposed raw via {@link #getTieringState(String)} or parsed
+ *     via typed accessors like {@link #getPartitionMarkDoneState()}. Empty when talking to an old
+ *     coordinator that does not report them.
  *
  * @since 0.3
  */
@@ -37,9 +45,20 @@ public class LakeSnapshot {
     // the specific log offset of the snapshot
     private final Map<TableBucket, Long> tableBucketsOffset;
 
+    // the keyed tiering-state entries; payloads parsed lazily by the typed accessors.
+    private final List<TieringStateEntry> tieringStates;
+
     public LakeSnapshot(long snapshotId, Map<TableBucket, Long> tableBucketsOffset) {
+        this(snapshotId, tableBucketsOffset, Collections.emptyList());
+    }
+
+    public LakeSnapshot(
+            long snapshotId,
+            Map<TableBucket, Long> tableBucketsOffset,
+            List<TieringStateEntry> tieringStates) {
         this.snapshotId = snapshotId;
         this.tableBucketsOffset = tableBucketsOffset;
+        this.tieringStates = tieringStates;
     }
 
     public long getSnapshotId() {
@@ -50,13 +69,38 @@ public Map<TableBucket, Long> getTableBucketsOffset() {
         return Collections.unmodifiableMap(tableBucketsOffset);
     }
 
+    /** Returns the raw tiering-state entry for the given key, or {@code null} if absent. */
+    @Nullable
+    public TieringStateEntry getTieringState(String stateKey) {
+        for (TieringStateEntry entry : tieringStates) {
+            if (entry.getStateKey().equals(stateKey)) {
+                return entry;
+            }
+        }
+        return null;
+    }
+
+    /**
+     * Parses and returns the partition mark-done state, or {@code null} if absent.
+     *
+     * @throws IllegalArgumentException if the entry has an unsupported (newer) version or a corrupt
+     *     payload; the raw entry stays available via {@link #getTieringState(String)}
+     */
+    @Nullable
+    public PartitionMarkDoneState getPartitionMarkDoneState() {
+        TieringStateEntry entry = getTieringState(PartitionMarkDoneState.STATE_KEY);
+        return entry == null ? null : PartitionMarkDoneState.fromStateEntry(entry);
+    }
+
     @Override
     public String toString() {
         return "LakeSnapshot{"
                 + "snapshotId="
                 + snapshotId
                 + ", tableBucketsOffset="
                 + tableBucketsOffset
+                + ", tieringStates="
+                + tieringStates
                 + '}';
     }
 }

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.TieringStateEntry;
 import org.apache.fluss.metadata.AggFunction;
 import org.apache.fluss.metadata.DatabaseChange;
 import org.apache.fluss.metadata.DatabaseSummary;
@@ -97,6 +98,7 @@
 import org.apache.fluss.rpc.messages.PbRenameColumn;
 import org.apache.fluss.rpc.messages.PbTableBucket;
 import org.apache.fluss.rpc.messages.PbTableStatsReqForBucket;
+import org.apache.fluss.rpc.messages.PbTieringStateEntry;
 import org.apache.fluss.rpc.messages.PrefixLookupRequest;
 import org.apache.fluss.rpc.messages.ProduceLogRequest;
 import org.apache.fluss.rpc.messages.PutKvRequest;
@@ -286,7 +288,18 @@ public static LakeSnapshot toLakeTableSnapshotInfo(GetLakeSnapshotResponse respo
                     new TableBucket(tableId, partitionId, pbLakeSnapshotForBucket.getBucketId());
             tableBucketsOffset.put(tableBucket, pbLakeSnapshotForBucket.getLogOffset());
         }
-        return new LakeSnapshot(snapshotId, tableBucketsOffset);
+
+        // pass through the tiering-state entries unparsed (typed parsing happens lazily in
+        // LakeSnapshot).
+        List<TieringStateEntry> tieringStates = new ArrayList<>();
+        for (PbTieringStateEntry pbEntry : response.getTieringStatesList()) {
+            tieringStates.add(
+                    new TieringStateEntry(
+                            pbEntry.getStateKey(),
+                            pbEntry.getStateVersion(),
+                            pbEntry.getPayload()));
+        }
+        return new LakeSnapshot(snapshotId, tableBucketsOffset, tieringStates);
     }
 
     public static List<FsPathAndFileName> toFsPathAndFileName(

fluss-client/src/test/java/org/apache/fluss/client/metadata/LakeSnapshotTest.java

@@ -0,0 +1,96 @@
+/*
+ * 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.client.metadata;
+
+import org.apache.fluss.lake.committer.PartitionMarkDoneState;
+import org.apache.fluss.lake.committer.PartitionMarkDoneState.PartitionState;
+import org.apache.fluss.lake.committer.TieringStateEntry;
+import org.apache.fluss.metadata.TableBucket;
+
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link LakeSnapshot}, in particular the keyed tiering-state accessors. */
+class LakeSnapshotTest {
+
+    @Test
+    void testKeyedStateAccess() {
+        // absent -> null (raw and typed).
+        LakeSnapshot absent = new LakeSnapshot(1L, Collections.emptyMap());
+        assertThat(absent.getTieringState(PartitionMarkDoneState.STATE_KEY)).isNull();
+        assertThat(absent.getPartitionMarkDoneState()).isNull();
+
+        // present -> raw entry exposed, typed accessor parses lazily; unrelated keys untouched.
+        PartitionMarkDoneState state =
+                new PartitionMarkDoneState(
+                        Collections.singletonMap(
+                                5L, new PartitionState(1000L, PartitionMarkDoneState.NOT_DONE)));
+        TieringStateEntry otherEntry =
+                new TieringStateEntry("other-key", 3, "{}".getBytes(StandardCharsets.UTF_8));
+        LakeSnapshot present =
+                new LakeSnapshot(
+                        1L,
+                        Collections.emptyMap(),
+                        Arrays.asList(otherEntry, state.toStateEntry()));
+        assertThat(present.getPartitionMarkDoneState()).isEqualTo(state);
+        assertThat(present.getTieringState("other-key")).isEqualTo(otherEntry);
+        assertThat(present.getTieringState("unknown-key")).isNull();
+    }
+
+    @Test
+    void testNewerVersionExposesRawEntryForPassthrough() {
+        TieringStateEntry newerEntry =
+                new TieringStateEntry(
+                        PartitionMarkDoneState.STATE_KEY,
+                        PartitionMarkDoneState.CURRENT_VERSION + 1,
+                        "{}".getBytes(StandardCharsets.UTF_8));
+        LakeSnapshot snapshot =
+                new LakeSnapshot(1L, Collections.emptyMap(), Collections.singletonList(newerEntry));
+        // unreadable here: parsing fails so the caller passes the raw entry through unchanged.
+        assertThatThrownBy(snapshot::getPartitionMarkDoneState)
+                .isInstanceOf(IllegalArgumentException.class);
+        assertThat(snapshot.getTieringState(PartitionMarkDoneState.STATE_KEY))
+                .isEqualTo(newerEntry);
+    }
+
+    @Test
+    void testCorruptStateDoesNotBlockBucketOffsets() {
+        TableBucket bucket = new TableBucket(1L, 0);
+        Map<TableBucket, Long> offsets = Collections.singletonMap(bucket, 100L);
+        TieringStateEntry corruptEntry =
+                new TieringStateEntry(
+                        PartitionMarkDoneState.STATE_KEY,
+                        1,
+                        "{not-json".getBytes(StandardCharsets.UTF_8));
+        LakeSnapshot snapshot =
+                new LakeSnapshot(1L, offsets, Collections.singletonList(corruptEntry));
+
+        // bucket offsets remain accessible even though the state payload is corrupt.
+        assertThat(snapshot.getTableBucketsOffset()).containsEntry(bucket, 100L);
+        // only the typed accessor surfaces the parse failure.
+        assertThatThrownBy(snapshot::getPartitionMarkDoneState)
+                .isInstanceOf(RuntimeException.class);
+    }
+}

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

@@ -0,0 +1,171 @@
+/*
+ * 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.PartitionMarkDoneStateJsonSerde;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+
+/**
+ * The partition mark-done state, persisted as the {@link TieringStateEntry} keyed by {@link
+ * #STATE_KEY} in the lake offsets file. Each tracked partition records its last update time and
+ * done time ({@link #NOT_DONE} until first marked done); a partition absent from the map has never
+ * been tracked. Entries of dropped partitions are removed by omitting them from the next payload.
+ *
+ * <p>The payload schema version lives in the entry envelope. Evolve the payload by bumping {@link
+ * #CURRENT_VERSION}; {@link #fromStateEntry} rejects a higher version, and the caller must pass the
+ * entry through unchanged so a newer build's state is never dropped.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public class PartitionMarkDoneState {
+
+    /** The tiering-state key of the partition mark-done state. */
+    public static final String STATE_KEY = "fluss.partition-mark-done";
+
+    public static final int CURRENT_VERSION = 1;
+
+    /** Sentinel done time of a partition that has not been marked done yet. */
+    public static final long NOT_DONE = -1L;
+
+    private final Map<Long, PartitionState> partitionStates;
+
+    public PartitionMarkDoneState(Map<Long, PartitionState> partitionStates) {
+        this.partitionStates =
+                partitionStates == null ? Collections.emptyMap() : new HashMap<>(partitionStates);
+    }
+
+    public Map<Long, PartitionState> getPartitionStates() {
+        return Collections.unmodifiableMap(partitionStates);
+    }
+
+    /** Wraps this state into its {@link TieringStateEntry} at {@link #CURRENT_VERSION}. */
+    public TieringStateEntry toStateEntry() {
+        return new TieringStateEntry(
+                STATE_KEY,
+                CURRENT_VERSION,
+                JsonSerdeUtils.writeValueAsBytes(this, PartitionMarkDoneStateJsonSerde.INSTANCE));
+    }
+
+    /**
+     * Parses the state from its {@link TieringStateEntry}.
+     *
+     * @throws IllegalArgumentException if the entry has a different key, an unsupported (newer)
+     *     version, or a corrupt payload
+     */
+    public static PartitionMarkDoneState fromStateEntry(TieringStateEntry entry) {
+        checkArgument(
+                STATE_KEY.equals(entry.getStateKey()),
+                "Expected state key %s but got %s.",
+                STATE_KEY,
+                entry.getStateKey());
+        if (entry.getStateVersion() > CURRENT_VERSION) {
+            throw new IllegalArgumentException(
+                    "Unsupported partition mark-done state version "
+                            + entry.getStateVersion()
+                            + " > "
+                            + CURRENT_VERSION
+                            + "; pass the entry through unchanged.");
+        }
+        return JsonSerdeUtils.readValue(
+                entry.getPayload(), PartitionMarkDoneStateJsonSerde.INSTANCE);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        PartitionMarkDoneState that = (PartitionMarkDoneState) o;
+        return Objects.equals(partitionStates, that.partitionStates);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(partitionStates);
+    }
+
+    @Override
+    public String toString() {
+        return "PartitionMarkDoneState{" + "partitionStates=" + partitionStates + '}';
+    }
+
+    /** The mark-done state of a single partition. */
+    public static class PartitionState {
+
+        private final long updateTime;
+        private final long doneTime;
+
+        public PartitionState(long updateTime, long doneTime) {
+            checkArgument(
+                    updateTime >= 0, "updateTime must be non-negative but got %s.", updateTime);
+            checkArgument(
+                    doneTime == NOT_DONE || doneTime > 0,
+                    "doneTime must be %s (not done) or positive but got %s.",
+                    NOT_DONE,
+                    doneTime);
+            this.updateTime = updateTime;
+            this.doneTime = doneTime;
+        }
+
+        public long getUpdateTime() {
+            return updateTime;
+        }
+
+        public long getDoneTime() {
+            return doneTime;
+        }
+
+        public boolean isDone() {
+            return doneTime != NOT_DONE;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) {
+                return true;
+            }
+            if (o == null || getClass() != o.getClass()) {
+                return false;
+            }
+            PartitionState that = (PartitionState) o;
+            return updateTime == that.updateTime && doneTime == that.doneTime;
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(updateTime, doneTime);
+        }
+
+        @Override
+        public String toString() {
+            return "PartitionState{" + "updateTime=" + updateTime + ", doneTime=" + doneTime + '}';
+        }
+    }
+}