buvb commented on issue #2021:
URL: https://github.com/apache/fluss/issues/2021#issuecomment-3600291317
@luoyuxia
### Lake Snapshot File Storage Optimization Plan
- Solution
Fully reference the design of KV Snapshot by changing Lake Snapshot to：
1. File System: Stores the complete LakeTableSnapshot object.
2. ZooKeeper: Stores only the lightweight LakeTableSnapshotHandle.
- Detailed Design
New Data Structures
LakeTableSnapshotHandle
public class LakeTableSnapshotHandle {
private final long snapshotId;
private final long tableId;
private final FsPath metadataFilePath;
public LakeTableSnapshot retrieveLakeTableSnapshot() throws IOException {
FSDataInputStream in =
metadataFilePath.getFileSystem().open(metadataFilePath);
byte[] jsonBytes = IOUtils.toByteArray(in);
return LakeTableSnapshotJsonSerde.fromJson(jsonBytes);
}
.............................
..........................
}
- File Storage Path Design
{data.dir}/lake-snapshots/table-{tableId}/snapshot-{snapshotId}.json
Example the general format is as follows:
hdfs://namenode/fluss/lake-snapshots/table-123/snapshot-1.json
hdfs://namenode/fluss/lake-snapshots/table-123/snapshot-2.json
hdfs://namenode/fluss/lake-snapshots/table-456/snapshot-1.json
lake-snapshots/
├── table-123/
│ ├── snapshot-1.json (Old version, will be cleaned)
│ └── snapshot-2.json (Latest version)
├── table-456/
│ └── snapshot-1.json
└── table-789/
└── snapshot-3.json
- write and read:
public void upsertLakeTableSnapshot(long tableId, LakeTableSnapshot
newSnapshot) {
Optional<LakeTableSnapshot> oldSnapshot = getLakeTableSnapshot(tableId);
LakeTableSnapshot mergedSnapshot = oldSnapshot.isPresent()
? mergeSnapshots(oldSnapshot.get(), newSnapshot)
: newSnapshot;
FsPath snapshotPath = getSnapshotPath(tableId,
mergedSnapshot.getSnapshotId());
writeSnapshotToFile(mergedSnapshot, snapshotPath);
LakeTableSnapshotHandle handle = new LakeTableSnapshotHandle(
mergedSnapshot.getSnapshotId(), tableId, snapshotPath);
String zkPath = LakeTableZNode.path(tableId);
byte[] handleBytes = LakeTableZNode.encode(handle);
if (zkClient.checkExists().forPath(zkPath) != null) {
zkClient.setData().forPath(zkPath, handleBytes);
} else {
zkClient.create().creatingParentsIfNeeded().forPath(zkPath,
handleBytes);
}
if (oldSnapshot.isPresent()) {
asyncDeleteOldSnapshot(tableId, oldSnapshot.get().getSnapshotId());
}
...............................
}
public Optional<LakeTableSnapshot> getLakeTableSnapshot(long tableId) {
String path = LakeTableZNode.path(tableId);
Optional<byte[]> data = getOrEmpty(path);
if (data.isEmpty()) {
return Optional.empty();
}
LakeTableSnapshotHandle handle = LakeTableZNode.decode(data.get());
LakeTableSnapshot snapshot = handle.retrieveLakeTableSnapshot();
return Optional.of(snapshot);
..................
}
- Backward Compatibility Design
// New format
example
{
"version": 2,
"snapshot_id": 123,
"table_id": 456,
"metadata_path": "hdfs://.../lake-snapshots/table-456/snapshot-123.json"
}
- In addition, an automatic migration logic and lifecycle management have
also been designed.
As mentioned above, these are the designs. I plan to start my work from
these aspects.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
