fhan688 opened a new issue, #20014: URL: https://github.com/apache/hudi/issues/20014
**Labels:** `type:feature`, `area:storage-format`, `flink` > cc @danny0405 @cshuo @zhangyue19921010 @xushiyan — this is proposed as a follow-up enhancement under the LSM epic #14310 / RFC-103. Could a maintainer attach it as a sub-issue of #14310 Happy to also fold the design into an RFC-103 addendum if that's preferred. --- ### Background RFC-103 (Hudi LSM tree layout) establishes **sorted write** as a core feature: within any written base or log file, records are fully sorted by record key(s). For Flink streaming write, RFC-103's memory strategy is: > **Efficient memory management** — Integrate Flink's built-in **MemorySegmentPool** with **BinaryInMemorySortBuffer** to enable fine-grained memory control and efficient sorting, greatly reduc ing GC pressure and sorting overhead. This is now implemented on master: each bucket buffers into a `BinaryInMemorySortBuffer` backed by a shared `HeapMemorySegmentPool`, and #19728 (`perf(flink): preempt inactive write buckets on memory exhaustion`) added `PreemptiveMemorySegmentPool` so that, when the pool is exhausted, the largest **inactive** bucket is reclaimed via `preemptMemory` → `flushAndDisposeBucket`, freeing its pages for the in-flight write. ### Problem RFC-103's memory section assumes the sort happens **entirely in memory**. It does not specify what happens when the volume of buffered-but-not-yet-flushed sorted data exceeds the configured memory budget. Today the only release valve is `flushAndDisposeBucket`, i.e. **flush the bucket to storage to reclaim pages**. Under memory pressure — large checkpoint intervals, skewed/hot buckets, or tight per-task memory — this forces a bucket to flush **before it has accumulated a full target-sized file**. The consequences work directly against the LSM layout's goals: - **Small-file proliferation** and increased **write amplification**, exactly what LSM + minor compaction is meant to reduce. - More downstream compaction pressure. - It breaks the "accumulate a batch, then write one target-sized sorted file" intent of sorted write. ### Proposal Add an **optional, spillable external sort buffer** as an additional release valve *before* falling back to a premature flush. When the memory pool is exhausted, an eligible bucket can **spill its currently-sorted run to task-local disk** (via Flink's own `IOManager` / `BinaryExternalMerger` / `SpillChannelManager`), return its pages to the pool, and continue buffering. At flush/checkpoint time, the spilled runs plus the remaining in-memory run are **k-way merged** into the final sorted output — preserving RFC-103's file-level ordering guarantee. This is: - **Additive and off by default** — behind a feature flag; when disabled, behavior is byte-for-byte the current path. - **A safe fallback** — any spill failure or configured disk-size/run-count limit cleanly falls back to today's `flushAndDisposeBucket`. - **Complementary to #19728**, not a replacement — it hooks the same `preemptMemory` decision point: *try to spill/reclaim first, flush only if spill isn't possible.* ### Relationship to RFC-103 (why this is not a conflict) RFC-103 commits to in-memory sorting for the common case; this proposal does **not** change that. It only defines the behavior for the case RFC-103 leaves open — *sorted data exceeds the memory budget* — turning "flush early and make small files" into "spill locally and still emit one sorted file." I'd suggest capturing it as a short *addendum to RFC-103's memory-management section rather than a new RFC*. Open to maintainers' preference here. ### Prior art JD has a production implementation of this exact idea on an internal Hudi branch (spillable per-bucket sort buffer with ZSTD-compressed local spill, disk/run limits, record-count conservation guards, and IOManager wiring). It predates and diverged from the community's current LSM write path, so the plan is to **re-implement the concept on top of master's already-merged LSM write path + #19728**, not to port the internal branch as-is. Benchmarks demonstrating small-file / write-amplification reduction under memory pressure will accompany the work (as requested in #14310, on a 1.x baseline). ### Proposed PR breakdown Small, independently reviewable PRs, each gated behind the flag until the path is complete: 1. **Refactor (no behavior change):** introduce a `SortBuffer` abstraction; wrap the existing `BinaryInMemorySortBuffer` as the in-memory implementation; `RowDataBucket` holds the interface. De fault = fully in-memory, identical behavior. 2. **Spill implementation (flag, default off):** `BinaryExternalSortBuffer` reusing Flink's `BinaryExternalMerger`/`SpillChannelManager`; config options (`spillable`, max disk size, max runs, c ompression codec + block size); wire task-local `IOManager` through to the write function. Null `IOManager` ⇒ degrade to in-memory. 3. **Preemption hook:** at the `preemptMemory` decision, attempt spill-based reclaim before `flushAndDisposeBucket`; flush remains the fallback. 4. **Correctness guards + metrics:** record-count conservation checks across spill/merge (fail-fast, no silent data loss) and spill metrics. 5. **Tests:** unit coverage for the external sort buffer + a memory-exhaustion e2e test exercising the spill path. -- 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]
