apache / apache/fluss

Optimize log data buffering using RocksDB for deduplication and sorting

Open
#3,657 0 comments 0 reactions 1 assignee Claimed by @loserwang1024 View on GitHub
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.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.