[flink] Spill cdc log to rocksDB for pk batch read in case of OOM.

[loserwang1024]

Purpose

Linked issue: close #3657

Brief change log

Tests

API and Format

Documentation

[loserwang1024]

@naivedogger @beryllw @leonardBang , CC

[naivedogger]

Thanks @loserwang1024 for this PR! Left some comments, PTAL.

[naivedogger]

Could we use a serialized-byte threshold instead of the number of distinct keys here? The goal is to prevent OOM, but memoryRows.size() does not provide an upper bound on memory usage because row sizes can vary significantly. For example, 8192 rows of 32 KiB already retain about 256 MiB of row data, excluding TreeMap and object overhead, so the process may OOM before spilling. It would be safer to track the retained serialized bytes and spill based on a configurable byte-size limit. Also, could you clarify how the default value of 8192 was determined?

[loserwang1024]

It is difficult to accurately estimate memory usage from the in-memory objects. We could use the size of the value after converting it to a byte[], but that would require eagerly serializing the value before spilling, and the resulting size would still not accurately represent the actual heap usage.

For now, we prefer not to expose the spill threshold as a configuration option because this is still an internal implementation detail, and we do not want to introduce a parameter tied to the current implementation prematurely. We can expose an appropriate configuration later if a concrete need arises.

The current threshold of 8192 was chosen with reference to the maximum limit-query value of 1024, with additional headroom. Its original purpose was also to prevent an unexpectedly large amount of data from accumulating in memory.

[naivedogger]

Closing logRows here does not appear to close the RocksDBLogRowsIterator that has already been passed to SortMergeReader.

The same path appears to exist in LakeSnapshotAndLogSplitScanner. Would it make sense for SortMergeReader to implement Closeable, own its input iterators, and be explicitly closed by the scanners before logRows?

[loserwang1024]

Closing logRows here does not appear to close the RocksDBLogRowsIterator that has already been passed to SortMergeReader.

RocksDBLogRowsIterator is passed to org.apache.fluss.client.table.scanner.SortMergeReader.MergeIterator and will be closed here.

[beryllw]

This workload is essentially append-once and scan-once, requiring only a single ordered traversal. Could you clarify why RocksDB was chosen over a lighter spillable external sort?

[beryllw]

RocksDB may invoke custom comparators concurrently, while the KV comparator ultimately shares a KeyEncoder backed by a mutable writer. Could we use order-preserving encoded keys with RocksDB’s bytewise comparator, or otherwise ensure that the entire comparator chain is thread-safe?

[loserwang1024]

Thanks for raising this. RocksDB was chosen primarily to simplify the implementation, rather than because this workload inherently requires a general-purpose KV store. KV snapshot reading already uses RocksDB, so we can reuse the existing RocksDB dependency and lifecycle utilities instead of introducing and maintaining a separate spillable external-sort implementation. SortedLogRows is also an internal implementation detail, so we can replace it with a dedicated spill mechanism later if benchmarks show that RocksDB becomes a bottleneck.

That said, your concern about the JNI comparator overhead is valid. I plan to limit the RocksDB-backed implementation to KvSnapshotAndLogBatchScanner and stop using it in LakeSnapshotAndLogSplitScanner, for two reasons:

1. The default KV snapshot interval is 10 minutes, while the default lake freshness is 3 minutes. Therefore, the changelog range accumulated between KV snapshots is typically larger, making the KV snapshot path more susceptible to OOM.

2. The KV snapshot primary-key order is already defined by an unsigned bytewise comparison of the encoded primary-key bytes, which is consistent with RocksDB’s default bytewise comparator. We can therefore use the encoded primary-key bytes directly as the RocksDB key and remove the custom Java comparator entirely, avoiding the native-to-Java comparator callbacks. In contrast, a lake snapshot’s ordering is provided by its lake-specific SortedRecordReader and is not necessarily consistent with RocksDB’s bytewise ordering.

This keeps the spill optimization focused on the path with the greater OOM risk while also simplifying the RocksDB key and comparator implementation.

[beryllw]

@luoyuxia cc