Optimize log data buffering using RocksDB for deduplication and sorting
- Dominant language
- Java
- Stars
- 2.1k
- Forks
- 625
- Avg merge
- 3d 14h
- Merged PRs (30d)
- 97
Description
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.
### Motivation
In both the `KvSnapshotAndLogBatchScanner` proposed in Sub-Issue 1 and the existing [`LakeSnapshotAndLogSplitScanner`](https://github.com/apache/fluss/blob/main/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/LakeSnapshotAndLogSplitScanner.java), incremental log records are buffered in an in-memory `TreeMap`:
```java
// LakeSnapshotAndLogSplitScanner.java, line 67
private Map logRows;
// ...
logRows = new TreeMap<>(rowComparator);
```
**Problem:** The volume of logs between two KV snapshots can be very large. With the default `kv.snapshot.interval` of 10 minutes, a high-throughput workload (e.g., 100 MB/s) can produce tens of GB of log data within a single interval. Buffering all of this in an in-memory `TreeMap` leads to OOM.
### Solution
Replace the in-memory `TreeMap` with a local RocksDB instance to buffer, deduplicate, and sort log records:
1. **Write:** Each fetched log record is written to a local temporary RocksDB as `key → value`
2. **Deduplication:** Subsequent writes for the same key automatically overwrite prior records (RocksDB's `put` semantics provide natural deduplication)
3. **Deletion:** DELETE and UPDATE_BEFORE records are written as **tombstones** to preserve the delete semantics and prevent data loss
4. **Sorting:** RocksDB stores keys in sorted order; reading via `RocksIterator` yields records sorted by primary key
5. **Merge:** The RocksDB log iterator is passed to `SortMergeReader`, which performs a two-way merge with the snapshot's `SnapshotFilesReader`
```
┌──────────────────────────────────────────────────────────┐
│ Client Side │
│ │
│ ┌─────────────┐ ┌──────────────┐ │
│ │ KV Snapshot │ │ LogScanner │ │
│ │ SST Files │ │ (incremental)│ │
│ │ (download) │ │ (fetch) │ │
│ └──────┬──────┘ └──────┬───────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────────┐ ┌──────────────┐ │
│ │SnapshotFiles│ │ Log Buffer │ │
│ │ Reader │ │ (RocksDB) │ │
│ │ (RocksDB) │ │ (dedup+sort) │ │
│ └──────┬──────┘ └──────┬───────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────────────────────────┐ │
│ │ SortMergeReader │ │
│ │ (snapshot + log merge by PK) │ │
│ └──────────────┬───────────────────┘ │
│ │ │
│ ▼ │
│ Sorted InternalRow stream │
└──────────────────────────────────────────────────────────┘
```
### Anything else?
_No response_
### Willingness to contribute
- [x] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.