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

[beryllw]

Purpose

Linked issue: close #3314

Lakehouse tiering needs to persist table-level state in Fluss (beyond per-bucket offsets) and share it across tiering rounds and multiple stateless jobs. This PR introduces a generic, opaque carrier for such state on the lake offsets file and the tiering RPCs, so future state (e.g. partition mark-done) can be added without further proto/RPC changes.

The carrier is intentionally decoupled from any consumer: the coordinator moves it as an opaque blob and never interprets it. The first consumer (partition mark-done) will be delivered separately on top of this contract.

Brief change log

• fluss-common: add LakeTieringTableState (@PublicEvolving) , a flat versioned model, and its JsonSerde; it evolves by adding fields and bumping the format version, and a consumer that reads a higher version than it understands must not overwrite the state so it never drops fields written by a newer build. TableBucketOffsets / TableBucketOffsetsJsonSerde carry the state as an embedded JSON object (top-level version stays 1), validated to be a JSON object on write.
• fluss-rpc: add optional bytes tiering_state_json to PbTableOffsets and GetLakeSnapshotResponse; bump GET_LAKE_SNAPSHOT / PREPARE_LAKE_TABLE_SNAPSHOT maxVersion to 1 for capability detection.
• fluss-server: pass the payload through as an opaque blob (ServerRpcMessageUtils, LakeTable, LakeTableHelper, LakeTableSnapshot); on merge the latest value overwrites, and a request without it keeps the
  previous one.
• fluss-client: LakeSnapshot holds the raw bytes and parses them lazily in getLakeTieringTableState(), so a corrupt/incompatible payload never blocks access to the bucket offsets.

Tests

• LakeTieringTableStateJsonSerdeTest: format, round-trip/defaults, and higher-version tolerance (unknown fields ignored).
• TableBucketOffsetsJsonSerdeTest: carrier embed round-trip (via base) and rejection of malformed / non-object payloads.
• ServerRpcMessageUtilsTieringStateTest: PREPARE→GET passthrough round-trip and absent-state handling.
• LakeTableHelperTest: merge overwrite and keep-previous-when-absent.
• LakeSnapshotTest: lazy parse and corrupt-state isolation.

API and Format

• RPC: two new optional fields + maxVersion 0→1. Wire-compatible in both directions (optional field + unknown-field skip); version 1 is only used for capability detection.
• Offsets file format: top-level version unchanged (1); adds an optional embedded tiering_state JSON object. Old readers ignore it; new readers default to absent. Non-partitioned tables are unchanged.

Documentation

No user-facing docs; the carrier is internal plumbing. The first feature built on it (partition mark-done) will document itself.

[Copilot]

Pull request overview

This PR adds an opaque, table-level tiering state payload (JSON bytes) that is persisted in the lake offsets file and transported via tiering-related RPCs, enabling future features (e.g., partition mark-done) without additional proto/RPC evolution.

Changes:

• Introduces LakeTieringTableState plus JSON serde, and embeds optional tiering_state into TableBucketOffsets JSON (offsets file format stays version=1).
• Extends tiering RPCs (PbTableOffsets, GetLakeSnapshotResponse) with optional bytes tiering_state_json and bumps API max versions to 1 for capability detection.
• Threads the payload through server/client plumbing and adds focused tests for serde behavior, merge semantics, RPC passthrough, and lazy parsing.

Reviewed changes

Copilot reviewed 18 out of 18 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
fluss-server/src/test/java/org/apache/fluss/server/zk/data/lake/LakeTableHelperTest.java	Adds merge/overwrite vs keep-previous tests for tiering state during offsets merge.
fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTieringStateTest.java	Verifies PREPARE→GET passthrough and absence behavior for tiering state bytes.
fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshot.java	Extends snapshot metadata to carry optional tiering state bytes.
fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableHelper.java	Implements merge semantics: offsets merge partially; tiering state uses whole-value overwrite/keep behavior.
fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTable.java	Reads tiering state bytes from offsets file into LakeTableSnapshot.
fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java	Passes tiering state bytes from/to RPC messages without interpretation.
fluss-rpc/src/main/proto/FlussApi.proto	Adds optional bytes tiering_state_json to relevant RPC messages.
fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java	Bumps max versions for GET/PREPARE lake snapshot APIs for capability detection.
fluss-common/src/test/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerdeTest.java	Adds tiering_state embed round-trip and bad-payload rejection tests.
fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java	Adds tests for format, defaults, and higher-version tolerance.
fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerde.java	Embeds and reads tiering_state JSON in the offsets serde.
fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsets.java	Extends model to carry optional tiering state bytes.
fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java	Implements JSON serde for the new tiering state model.
fluss-common/src/main/java/org/apache/fluss/utils/json/JsonSerdeUtils.java	Adds readTree(byte[]) helper for pass-through JSON payload validation/embedding.
fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java	Adds the public-evolving, versioned tiering state model with JSON helpers.
fluss-client/src/test/java/org/apache/fluss/client/metadata/LakeSnapshotTest.java	Tests lazy parsing and isolation from corrupt tiering state bytes.
fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java	Passes tiering state bytes from RPC response into LakeSnapshot.
fluss-client/src/main/java/org/apache/fluss/client/metadata/LakeSnapshot.java	Stores tiering state bytes and parses them on demand.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[platinumhamburg]

@beryllw Thanks for adding generic support for table-level tiering state. Having the Coordinator pass the state through without interpreting it keeps the RPC and server code independent of any specific use case, which is a reasonable design choice.

My main reason for requesting changes is that the lifecycle of the state remains somewhat ambiguous. In a normal lake commit, tiered offsets and state may change together; in that case, publishing them atomically as one result is reasonable. The problem arises when the state changes without new lake data. In #3635, a state-only update creates a new offsets file, reuses the previous snapshot ID, and appends another lake snapshot metadata entry. As a result, the same snapshot ID can refer to multiple entries. Latest and by-ID queries may return different state, while the behavior of readable snapshots, retention, and file cleanup also becomes unclear.

Before the first consumer is merged, please clarify whether the state belongs to a lake snapshot or can change independently. If it belongs to a lake snapshot, it should change only when a new snapshot is committed. If it can change independently, a state-only update should not create another lake snapshot record with the same snapshot ID. The API should also define which state GetLakeSnapshot returns for latest, historical, and readable requests. PREPARE and COMMIT should validate tieringEpoch to prevent a writer from an old assignment from overwriting newer state.

For #3660 itself, the current API still has several correctness issues: the V1 path may lose existing state; a PREPARE carrying state does not require API v1, and its response cannot confirm whether the state was accepted; the API does not prevent an older job from using or rewriting a state version it does not support; and known JSON fields are not validated strictly. These issues should be resolved before the API is considered safe for rolling upgrades.

[luoyuxia]

@beryllw Thanks for the pr. The change lgtm overall.

[luoyuxia]

why we need to bump the version?

[luoyuxia]

@wuchong Could you please help review the api changes. Here's a summary:

// RPC proto changes
message PbTableOffsets {
      required int64 table_id = 1;
      repeated PbBucketOffset bucket_offset = 2;
      optional TablePath table_path = 3;

      // New: opaque table-level tiering state.
      optional bytes tiering_state_json = 4;
  }


message GetLakeSnapshotResponse {
      required int64 table_id = 1;
      required int64 snapshot_id = 2;
      repeated PbLakeSnapshotForBucket bucket_snapshot = 3;

      // New: opaque table-level tiering state.
      optional bytes tiering_state_json = 4;
  }


// Common persisted model change

  class TableBucketOffsets {
      long tableId;
      Map<TableBucket, Long> offsets;

      // New: nullable opaque JSON bytes.
      @Nullable byte[] tieringStateJson;

      @Nullable
      byte[] getTieringStateJson();
  }


  // Client API change
  class LakeSnapshot {
      long snapshotId;
      Map<TableBucket, Long> tableBucketsOffset;

      // New: nullable opaque JSON bytes from RPC.
      @Nullable byte[] lakeTieringTableStateJson;

      // New public API: lazy parse typed state.
      @Nullable
      LakeTieringTableState getLakeTieringTableState();
  }


  // New typed state model

  class LakeTieringTableState {
      int version;
      boolean partitionDoneInitialized;
      Map<Long, Long> partitionUpdateTimes;
  }

[beryllw]

The key fact is that mark-done state can change without a new lake commit (a partition is marked done on idle timeout, with no new data), so it's table-level and doesn't naturally belong to a single snapshot. There are two ways to carry it.

Option 1: keep it embedded in the offsets file (current). Simple, no new files, offsets and state published together. But a state-only change needs a new offsets file, which forces reusing the snapshot ID and appending a duplicate entry — the ambiguity you flagged around latest/by-id/readable and retention.

Option 2: split it into a dedicated table-level file with its own pointer. State advances independently, so IDs stay 1:1 and retention/GetLakeSnapshot stay clean, V1 never touches state, and the state path can validate tieringEpoch on its own. Cost is one extra file with its own lifecycle.