Optimize log data buffering using RocksDB for deduplication and sorting

[loserwang1024]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

In both the KvSnapshotAndLogBatchScanner proposed in Sub-Issue 1 and the existing LakeSnapshotAndLogSplitScanner, incremental log records are buffered in an in-memory TreeMap<InternalRow, KeyValueRow>:

// 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

• I'm willing to submit a PR!