[lake] Use file to store lake snapshot

[luoyuxia]

Purpose

Linked issue: close #2060

Brief change log

• Change to store lake snapshot info to a file and record the file handle in zk node instead of the full lake snapshot
• Change commit lake snapshot to be asynchronously since it'll have heavy io

Tests

API and Format

Documentation

[Copilot]

Pull request overview

This PR refactors lake table snapshot storage to use a file-based approach instead of storing complete snapshot data in ZooKeeper. The change introduces a version 2 storage format where ZooKeeper stores only the metadata file path while the actual snapshot data is persisted in remote storage, improving scalability and reducing ZK load.

• Introduced LakeTable wrapper class supporting both legacy (version 1: full data in ZK) and new (version 2: file path in ZK) storage formats
• Moved LakeTableSnapshot and LakeTableSnapshotJsonSerde to new lake package for better organization
• Updated upsertLakeTableSnapshot method signature to require TablePath for generating remote file paths

Reviewed changes

Copilot reviewed 20 out of 20 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTable.java	New wrapper class for lake table snapshot information supporting both version 1 (legacy) and version 2 (file-based) storage formats
fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableJsonSerde.java	New JSON serializer/deserializer for LakeTable with automatic version detection and backward compatibility
fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableUtils.java	Utility class for storing lake table snapshots to remote files
fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshot.java	Moved from parent package to lake subpackage; license header formatting updated
fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshotJsonSerde.java	Moved from parent package to lake subpackage; added static helper methods toJson/fromJson
fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkData.java	Updated LakeTableZNode to encode/decode LakeTable instead of LakeTableSnapshot
fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java	Updated upsertLakeTableSnapshot to accept tablePath parameter and store snapshots to files; added getLakeTable method
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java	Updated to retrieve and pass tablePath when upserting lake table snapshots
fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java	Added remoteLakeTableSnapshotPath method for generating remote lake snapshot file paths
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java	Updated import to use new lake package location
fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java	Updated import to use new lake package location
fluss-server/src/main/java/org/apache/fluss/server/entity/CommitLakeTableSnapshotData.java	Updated import to use new lake package location
fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java	Updated import to use new lake package location
fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java	Added integration test verifying backward compatibility when upgrading from version 1 to version 2 format
fluss-server/src/test/java/org/apache/fluss/server/zk/data/LakeTableSnapshotJsonSerdeTest.java	Updated imports to use new lake package location
fluss-server/src/test/java/org/apache/fluss/server/replica/CommitLakeTableSnapshotITCase.java	Updated import to use new lake package location
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperatorTest.java	Updated expected snapshot IDs to reflect multiple commits with incremental IDs
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java	Fixed snapshot ID management to use incremental IDs for each partition commit
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TestingLakeTieringFactory.java	Modified testingLakeCommitter field to be mutable to support lazy initialization

---

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

[wuchong]

1. Since thegetLatestLakeSnapshot RPC adds one more remote storage IO, it may blocks TabletService/CoordinatorService requests queue. I suggested to having an ioExecutor in TabletServer as well, and move the IO operation into the ioExecutor and return the future.
2. The current serialized json data of LakeTableSnapshot is not compacted. We should design a new serialization format to reduce the size (avoid duplicated information in the json).
3. I suggest it's best to consider the DV table issue together, it should involve modifying the same laketable znode format.

[Copilot]

Pull request overview

Copilot reviewed 27 out of 27 changed files in this pull request and generated 9 comments.

---

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

[Copilot]

Pull request overview

Copilot reviewed 45 out of 45 changed files in this pull request and generated 12 comments.

---

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

[luoyuxia]

@wuchong Hi, I already addressed your comment.

1. Remove unnecessary information in LakeTableSnapshot to reduce store size
2. Store a list of snapshot in zk node, each snapshot records the tiered offset and readable offset. In this pr, we always set tiered offset to equal to readable offset, and only keep one snapshot since if not consider dv fix, we only need one. But the dv fix issue, I'll extend it to contains multiple lake snapshot.
3. For comment [lake] Use file to store lake snapshot #2037 (comment), maybe we can address it in another pr?