[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.LakeTieringTableState;
 import org.apache.fluss.metadata.TableBucket;
 
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
 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, exposed parsed via {@link #getLakeTieringTableState()} or raw via {@link
+ *     #getRawTieringStateJson()}. 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 +45,21 @@ public class LakeSnapshot {
     // the specific log offset of the snapshot
     private final Map<TableBucket, Long> tableBucketsOffset;
 
+    // opaque tiering-state JSON bytes; null when absent. parsed lazily (see
+    // getLakeTieringTableState).
+    @Nullable private final byte[] lakeTieringTableStateJson;
+
     public LakeSnapshot(long snapshotId, Map<TableBucket, Long> tableBucketsOffset) {
+        this(snapshotId, tableBucketsOffset, null);
+    }
+
+    public LakeSnapshot(
+            long snapshotId,
+            Map<TableBucket, Long> tableBucketsOffset,
+            @Nullable byte[] lakeTieringTableStateJson) {
         this.snapshotId = snapshotId;
         this.tableBucketsOffset = tableBucketsOffset;
+        this.lakeTieringTableStateJson = lakeTieringTableStateJson;
     }
 
     public long getSnapshotId() {
@@ -50,13 +70,32 @@ public Map<TableBucket, Long> getTableBucketsOffset() {
         return Collections.unmodifiableMap(tableBucketsOffset);
     }
 
+    /** Parses and returns the tiering state (lazily), or {@code null} if absent. */
+    @Nullable
+    public LakeTieringTableState getLakeTieringTableState() {
+        return lakeTieringTableStateJson == null
+                ? null
+                : LakeTieringTableState.fromJsonBytes(lakeTieringTableStateJson);
+    }
+
+    /**
+     * Returns the raw, unparsed tiering-state JSON bytes ({@code null} if absent), for passing a
+     * newer, unreadable state through unchanged.
+     */
+    @Nullable
+    public byte[] getRawTieringStateJson() {
+        return lakeTieringTableStateJson;
+    }
+
     @Override
     public String toString() {
         return "LakeSnapshot{"
                 + "snapshotId="
                 + snapshotId
                 + ", tableBucketsOffset="
                 + tableBucketsOffset
+                + ", lakeTieringTableStateJson="
+                + Arrays.toString(lakeTieringTableStateJson)
                 + '}';
     }
 }

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

@@ -286,7 +286,11 @@ 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 opaque tiering-state bytes (parsed lazily by LakeSnapshot).
+        byte[] lakeTieringTableStateJson =
+                response.hasTieringStateJson() ? response.getTieringStateJson() : null;
+        return new LakeSnapshot(snapshotId, tableBucketsOffset, lakeTieringTableStateJson);
     }
 
     public static List<FsPathAndFileName> toFsPathAndFileName(

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

@@ -0,0 +1,75 @@
+/*
+ * 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.LakeTieringTableState;
+import org.apache.fluss.metadata.TableBucket;
+
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+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 lazy parsing of the tiering state. */
+class LakeSnapshotTest {
+
+    @Test
+    void testLazyParse() {
+        // absent -> null (parsed and raw).
+        LakeSnapshot absent = new LakeSnapshot(1L, Collections.emptyMap());
+        assertThat(absent.getLakeTieringTableState()).isNull();
+        assertThat(absent.getRawTieringStateJson()).isNull();
+
+        // present -> parsed lazily; raw bytes also exposed.
+        byte[] json =
+                new LakeTieringTableState(true, Collections.singletonMap(5L, 1000L)).toJsonBytes();
+        LakeSnapshot present = new LakeSnapshot(1L, Collections.emptyMap(), json);
+        LakeTieringTableState state = present.getLakeTieringTableState();
+        assertThat(state).isNotNull();
+        assertThat(state.isPartitionDoneInitialized()).isTrue();
+        assertThat(state.getPartitionUpdateTimes()).containsEntry(5L, 1000L);
+        assertThat(present.getRawTieringStateJson()).isEqualTo(json);
+    }
+
+    @Test
+    void testHigherVersionExposesRawBytesForPassthrough() {
+        byte[] json = "{\"version\":99}".getBytes(StandardCharsets.UTF_8);
+        LakeSnapshot snapshot = new LakeSnapshot(1L, Collections.emptyMap(), json);
+        // unreadable here: parsing fails so the caller passes the raw bytes through unchanged.
+        assertThatThrownBy(snapshot::getLakeTieringTableState)
+                .isInstanceOf(IllegalArgumentException.class);
+        assertThat(snapshot.getRawTieringStateJson()).isEqualTo(json);
+    }
+
+    @Test
+    void testCorruptTieringStateDoesNotBlockBucketOffsets() {
+        TableBucket bucket = new TableBucket(1L, 0);
+        Map<TableBucket, Long> offsets = Collections.singletonMap(bucket, 100L);
+        LakeSnapshot snapshot =
+                new LakeSnapshot(1L, offsets, "{not-json".getBytes(StandardCharsets.UTF_8));
+
+        // bucket offsets remain accessible even though the tiering state is corrupt.
+        assertThat(snapshot.getTableBucketsOffset()).containsEntry(bucket, 100L);
+        // only the explicit state accessor surfaces the parse failure.
+        assertThatThrownBy(snapshot::getLakeTieringTableState).isInstanceOf(RuntimeException.class);
+    }
+}

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

@@ -0,0 +1,127 @@
+/*
+ * 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>
+ *
+ * <p>This is a flat, versioned model. Evolve it by adding fields and bumping {@link
+ * #CURRENT_VERSION}. When a newer and an older tiering job run against the same table across
+ * rounds, a build that reads a {@code version} higher than its own {@link #CURRENT_VERSION} cannot
+ * interpret it: {@link #fromJsonBytes} raises {@link IllegalArgumentException}, and the caller must
+ * pass the raw bytes through unchanged so the newer build's state is never dropped.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public class LakeTieringTableState {
+
+    public static final int VERSION_1 = 1;
+
+    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);
+    }
+
+    public int getVersion() {
+        return version;
+    }
+
+    public boolean isPartitionDoneInitialized() {
+        return partitionDoneInitialized;
+    }
+
+    public Map<Long, Long> getPartitionUpdateTimes() {
+        return Collections.unmodifiableMap(partitionUpdateTimes);
+    }
+
+    public byte[] toJsonBytes() {
+        return JsonSerdeUtils.writeValueAsBytes(this, LakeTieringTableStateJsonSerde.INSTANCE);
+    }
+
+    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/utils/json/JsonSerdeUtils.java

@@ -73,5 +73,13 @@ public static <T> T readValue(byte[] json, JsonDeserializer<T> deserializer) {
         }
     }
 
+    /**
+     * Parse the given JSON bytes into a tree node using the shared object mapper. Useful for
+     * opaque/pass-through JSON payloads whose concrete type is not known to the caller.
+     */
+    public static JsonNode readTree(byte[] json) throws IOException {
+        return OBJECT_MAPPER_INSTANCE.readTree(json);
+    }
+
     private JsonSerdeUtils() {}
 }