loserwang1024 opened a new issue, #3657: URL: https://github.com/apache/fluss/issues/3657
### 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<InternalRow, KeyValueRow>`: ```java // LakeSnapshotAndLogSplitScanner.java, line 67 private Map<InternalRow, KeyValueRow> 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! -- 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]
