[Umbrella] Support batch read for PK tables based on SST files + apply logs

[loserwang1024]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

The community is actively working on #1600 (FIP-17): Support Full KV Scan for Primary Key Tables, which implements full PK table scanning via server-side RocksDB iteration. While this is a valuable direction, an alternative approach based on downloading KV snapshot SST files to the client and applying incremental logs remains meaningful for the following reasons:

1. Server pressure: FIP-17's approach requires the server to maintain scanner session context (scannerId / callSeqId) for each active scan. Under heavy concurrent query workloads, the server must manage many concurrent sessions, leading to significant memory and CPU overhead.
2. Backward compatibility: Many deployed server versions do not support Full KV Scan. The SST-file-based approach works against these older clusters without server-side changes.
3. Compute pushdown: Once SST files are downloaded to the client, local RocksDB can be leveraged for iteration, filtering, and projection, reducing server-side resource consumption.

Therefore, we propose supporting both approaches in Fluss, selectable via user configuration:

• Server-side scan (FIP-17): suitable for small tables and low-concurrency scenarios; lower latency
• Client-side SST + log (this proposal): suitable for large tables and high-concurrency scenarios; less server pressure

Solution

┌──────────────────────────────────────────────────────────┐
│                  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!