[AMORO-4166] [Improvement]: Implement heap-based flush mechanism for SortedPosDeleteWriter to prevent OOM.

[slfan1989]

Why are the changes needed?

Close #4166.

SortedPosDeleteWriter currently only flushes buffered position deletes based on a record count threshold, which can cause OutOfMemoryError (OOM) in memory-constrained environments when processing large-scale delete operations.

There was a TODO comment in the code indicating the need for a heap memory-based flush policy:

// TODO Flush buffer based on the policy that checking whether whole heap memory size exceed the
// threshold.

This PR implements the heap-based flush mechanism to prevent OOM issues by monitoring JVM heap usage and triggering flush when memory pressure is detected, while maintaining backward compatibility.

Brief change log

Core Implementation:

• SortedPosDeleteWriter:
  • Added HeapUsageProvider interface to monitor JVM heap memory (max/used)
  • Implemented shouldFlushByHeap() method with safety guards (minimum records, valid ratio check)
  • Modified delete() logic to flush when records >= recordsNumThreshold || shouldFlushByHeap()
  • Added multiple constructor overloads to support new parameters while maintaining backward compatibility

Configuration:

• TableProperties: Added three new table properties:
  • pos-delete.flush.heap.ratio (default: 0.8) - Heap usage ratio threshold
  • pos-delete.flush.records (default: Long.MAX_VALUE) - Record count threshold
  • pos-delete.flush.heap.min-records (default: 1000) - Minimum records before heap flush

How was this patch tested?

• Add some test cases that check the changes thoroughly including negative and positive cases if possible
  • Added TestSortedPosDeleteWriterHeapFlush with mock HeapUsageProvider to test heap-based flush logic
  • Test case 1: Verify flush is triggered when heap usage exceeds threshold
  • Test case 2: Verify heap flush is disabled when ratio is set to 0

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? JavaDocs

[slfan1989]

@zhoujinsong Could you please help review this PR when available? Thank you very much!

[j1wonpark]

Thanks for picking up this TODO — the buffer in SortedPosDeleteWriter really is unbounded today, so the concern is valid.

Before merging I'd like to see the two blocking comments addressed: the JVM-wide heap ratio isn't a usable signal for this writer's buffer, and the feature shouldn't be on by default.

That said, I think there is a simpler fix that removes the need for any flush heuristic. Every caller passes row = null, so the buffer only holds (path, position) pairs — and Iceberg already has a bitmap-based writer for that: org.apache.iceberg.deletes.SortingPositionOnlyDeleteWriter (since 1.4.0). A roaring bitmap per data file instead of List<PosRow> cuts memory by orders of magnitude and makes sorting free, so nothing needs to be flushed early. It wraps the same PositionDeleteWriter we already create, and DeleteGranularity.PARTITION keeps the one-file-per-tree-node layout, so callers and output paths stay unchanged. Our reader side (CombinedDeleteFilter) already holds positions in Roaring64Bitmap, so this would just align the writer with it.

If you're up for it, I'd suggest turning this PR into that replacement: swap the internal buffer for SortingPositionOnlyDeleteWriter and drop the heap logic, the three new properties and the extra constructors. The diff would end up smaller than the current one. If you'd rather keep this PR focused, I'm also happy to open a separate issue for the bitmap change and take it myself — just let me know which you prefer.

One question for context: do you have a case where this buffer was the actual cause of an OOM (heap dump, delete counts)? In our deployments the pos-delete buffer has stayed small relative to the equality-delete sets the reader loads, so I'd like to understand the scale we're designing for.

Side note: the Spark-3.4 CI job timed out at 6h, which looks unrelated; a re-run should clear it.

[j1wonpark]

(blocking) totalMemory() - freeMemory() counts garbage that has not been collected yet, and it is a JVM-wide number shared by every rewrite task running in the optimizer. With G1 the heap routinely sits above 80% between collections even when live data is small, so this writer would flush every 1,000 records regardless of how much it actually buffers, and one task's memory use would fragment every other task's pos-delete output. I don't think a process-wide heap ratio can tell us anything about this writer's own buffer.

[j1wonpark]

(blocking) Since MixedIcebergRewriteExecutor / MixedHiveRewriteExecutor now pass table.properties(), the 0.8 default enables this for every mixed-format optimizing task on upgrade. Combined with the point above, existing users would go from one pos-delete file per tree node to many small ones without opting in. Whatever flush policy we land on should be opt-in.