danny0405 commented on issue #20014: URL: https://github.com/apache/hudi/issues/20014#issuecomment-5755291287
This looks like a valid optional optimization. The current memory-pressure path flushes an inactive bucket to reclaim pages, and native MOR writes finalize each mini-batch into at most one data file and one delete file. If a bucket is flushed repeatedly and receives more records before the checkpoint, local spilling could consolidate those batches into fewer, larger LSM log files. The clearest gains would be fewer remote file creations, fewer input streams for readers/compaction, and potentially less downstream compaction work at the same sort-memory budget. However, lower write amplification and higher ingestion throughput need measurements: spilling adds local writes/reads and merge CPU, while the initial remote payload is largely unchanged unless more records can be combined. Small files caused by low per-bucket volume between checkpoints would remain. A few points seem worth addressing in the design: 1. **The preemption hook alone does not cover a single hot bucket.** [`preemptMemory()`](https://github.com/apache/hudi/blob/8c648ec36a657b57de2cf8eeb46fe38517d701ff/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java#L385-L425) excludes the active bucket because it can be in the middle of serializing a row. With no inactive victim, the current recovery path still flushes the failed bucket. Supporting this case needs a safe rollover/retry path outside the allocation callback, preserving committed records and replacing the failed buffer rather than reusing it. 2. **The downstream path also needs to remain memory-bounded.** When pre-combine is enabled, [`FlinkWriteHelper.deduplicateRecords()`](https://github.com/apache/hudi/blob/8c648ec36a657b57de2cf8eeb46fe38517d701ff/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/FlinkWriteHelper.java#L94-L115) collects the whole batch into a map of record lists. A larger spilled batch can therefore be materialized back into heap at flush time. Streaming deduplication with the existing merge semantics, or an explicit supported limit, would be needed for that configuration. 3. **Fallback guarantees should distinguish recoverable spill failures from lost/unreadable runs.** Falling back to remote flush is safe while all records remain available. If a previously completed spill becomes unreadable after its memory pages have been released, failing the task and recovering is necessary. Correctness guards and failure tests should accompany the first enabled spill path rather than arrive later. 4. **Resource accounting and checkpoint behavior need explicit bounds.** Resident/reclaimable pages, total buffered data, and spill bytes should be tracked separately. Victim selection should use reclaimable memory; spill/merge needs reserved working memory and bounded fan-in. Since checkpoints flush the remaining buckets, deferring remote writes could increase checkpoint latency even if file counts improve. One clarification on the motivation: RFC-103 requires file-level ordering, not target-sized files, and already acknowledges small-file proliferation. Early flushing preserves that ordering guarantee. I would frame this as improving file-size efficiency under memory pressure, which is compatible with the RFC. Could the proposal include spill-on/off benchmarks at identical memory budgets for many active buckets, a single hot bucket, and short/long checkpoint intervals? Actual file counts/sizes, throughput, checkpoint p95/p99, local spill I/O, remote bytes including compaction, and read latency would make the tradeoff clear. Failure/recovery, duplicate/update/delete semantics, and page/spill-file cleanup would also be important acceptance tests. -- 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]
