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

amoro-format-iceberg/src/main/java/org/apache/amoro/io/writer/GenericTaskWriters.java

@@ -170,6 +170,21 @@ public SortedPosDeleteWriter<Record> buildBasePosDeleteWriter(
           org.apache.iceberg.TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX
               + MetadataColumns.DELETE_FILE_POS.name(),
           MetricsModes.Full.get().toString());
+      double heapUsageRatioThreshold =
+          PropertyUtil.propertyAsDouble(
+              table.properties(),
+              TableProperties.POS_DELETE_FLUSH_HEAP_RATIO,
+              TableProperties.POS_DELETE_FLUSH_HEAP_RATIO_DEFAULT);
+      long recordsNumThreshold =
+          PropertyUtil.propertyAsLong(
+              table.properties(),
+              TableProperties.POS_DELETE_FLUSH_RECORDS,
+              TableProperties.POS_DELETE_FLUSH_RECORDS_DEFAULT);
+      int heapFlushMinRecords =
+          PropertyUtil.propertyAsInt(
+              table.properties(),
+              TableProperties.POS_DELETE_FLUSH_HEAP_MIN_RECORDS,
+              TableProperties.POS_DELETE_FLUSH_HEAP_MIN_RECORDS_DEFAULT);
       return new SortedPosDeleteWriter<>(
           appenderFactory,
           new CommonOutputFileFactory(
@@ -185,7 +200,10 @@ public SortedPosDeleteWriter<Record> buildBasePosDeleteWriter(
           fileFormat,
           mask,
           index,
-          partitionKey);
+          partitionKey,
+          recordsNumThreshold,
+          heapUsageRatioThreshold,
+          heapFlushMinRecords);
     }
 
     public GenericChangeTaskWriter buildChangeWriter() {

amoro-format-iceberg/src/main/java/org/apache/amoro/io/writer/MixedTreeNodePosDeleteWriter.java

@@ -21,6 +21,7 @@
 import org.apache.amoro.data.DataTreeNode;
 import org.apache.amoro.io.AuthenticatedFileIO;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
+import org.apache.amoro.table.TableProperties;
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.PartitionSpec;
@@ -30,9 +31,11 @@
 import org.apache.iceberg.io.DeleteWriteResult;
 import org.apache.iceberg.io.FileAppenderFactory;
 import org.apache.iceberg.io.FileWriter;
+import org.apache.iceberg.util.PropertyUtil;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -62,6 +65,9 @@ public class MixedTreeNodePosDeleteWriter<T>
   private final String location;
 
   private final PartitionSpec spec;
+  private final double heapUsageRatioThreshold;
+  private final long recordsNumThreshold;
+  private final int heapFlushMinRecords;
 
   public MixedTreeNodePosDeleteWriter(
       FileAppenderFactory<T> appenderFactory,
@@ -71,7 +77,9 @@ public MixedTreeNodePosDeleteWriter(
       EncryptionManager encryptionManager,
       Long transactionId,
       String location,
-      PartitionSpec spec) {
+      PartitionSpec spec,
+      Map<String, String> properties) {
+    Map<String, String> safeProperties = properties == null ? Collections.emptyMap() : properties;
     this.appenderFactory = appenderFactory;
     this.format = format;
     this.partition = partition;
@@ -80,6 +88,42 @@ public MixedTreeNodePosDeleteWriter(
     this.transactionId = transactionId;
     this.location = location;
     this.spec = spec;
+    this.heapUsageRatioThreshold =
+        PropertyUtil.propertyAsDouble(
+            safeProperties,
+            TableProperties.POS_DELETE_FLUSH_HEAP_RATIO,
+            TableProperties.POS_DELETE_FLUSH_HEAP_RATIO_DEFAULT);
+    this.recordsNumThreshold =
+        PropertyUtil.propertyAsLong(
+            safeProperties,
+            TableProperties.POS_DELETE_FLUSH_RECORDS,
+            TableProperties.POS_DELETE_FLUSH_RECORDS_DEFAULT);
+    this.heapFlushMinRecords =
+        PropertyUtil.propertyAsInt(
+            safeProperties,
+            TableProperties.POS_DELETE_FLUSH_HEAP_MIN_RECORDS,
+            TableProperties.POS_DELETE_FLUSH_HEAP_MIN_RECORDS_DEFAULT);
+  }
+
+  public MixedTreeNodePosDeleteWriter(
+      FileAppenderFactory<T> appenderFactory,
+      FileFormat format,
+      StructLike partition,
+      AuthenticatedFileIO fileIO,
+      EncryptionManager encryptionManager,
+      Long transactionId,
+      String location,
+      PartitionSpec spec) {
+    this(
+        appenderFactory,
+        format,
+        partition,
+        fileIO,
+        encryptionManager,
+        transactionId,
+        location,
+        spec,
+        Collections.emptyMap());
   }
 
   @Override
@@ -109,7 +153,10 @@ private SortedPosDeleteWriter<T> generatePosDelete(DataTreeNode treeNode) {
         format,
         treeNode.mask(),
         treeNode.index(),
-        partition);
+        partition,
+        recordsNumThreshold,
+        heapUsageRatioThreshold,
+        heapFlushMinRecords);
   }
 
   public List<DeleteFile> complete() throws IOException {

amoro-format-iceberg/src/main/java/org/apache/amoro/io/writer/SortedPosDeleteWriter.java

@@ -49,6 +49,27 @@
  */
 public class SortedPosDeleteWriter<T> implements Closeable {
   private static final long DEFAULT_RECORDS_NUM_THRESHOLD = Long.MAX_VALUE;
+  private static final Runtime RUNTIME = Runtime.getRuntime();
+  // Note: totalMemory/freeMemory only reflect the currently allocated heap, not max heap.
+  // When the heap has not fully expanded, usedMemory() can be underestimated and delay flushes.
+  private static final HeapUsageProvider RUNTIME_HEAP_USAGE =
+      new HeapUsageProvider() {
+        @Override
+        public long maxMemory() {
+          return RUNTIME.maxMemory();
+        }
+
+        @Override
+        public long usedMemory() {
+          return RUNTIME.totalMemory() - RUNTIME.freeMemory();
+        }
+      };
+
+  interface HeapUsageProvider {
+    long maxMemory();
+
+    long usedMemory();
+  }
 
   private final Map<CharSequenceWrapper, List<PosRow<T>>> posDeletes = Maps.newHashMap();
   private final List<DeleteFile> completedFiles = Lists.newArrayList();
@@ -61,6 +82,9 @@ public class SortedPosDeleteWriter<T> implements Closeable {
   private final FileFormat format;
   private final TaskWriterKey writerKey;
   private final long recordsNumThreshold;
+  private final double heapUsageRatioThreshold;
+  private final int heapFlushMinRecords;
+  private final HeapUsageProvider heapUsageProvider;
 
   private int records = 0;
 
@@ -72,14 +96,67 @@ public SortedPosDeleteWriter(
       long mask,
       long index,
       StructLike partitionKey,
-      long recordsNumThreshold) {
+      long recordsNumThreshold,
+      double heapUsageRatioThreshold,
+      int heapFlushMinRecords) {
+    this(
+        appenderFactory,
+        fileFactory,
+        io,
+        format,
+        mask,
+        index,
+        partitionKey,
+        recordsNumThreshold,
+        heapUsageRatioThreshold,
+        heapFlushMinRecords,
+        RUNTIME_HEAP_USAGE);
+  }
+
+  SortedPosDeleteWriter(
+      FileAppenderFactory<T> appenderFactory,
+      OutputFileFactory fileFactory,
+      AuthenticatedFileIO io,
+      FileFormat format,
+      long mask,
+      long index,
+      StructLike partitionKey,
+      long recordsNumThreshold,
+      double heapUsageRatioThreshold,
+      int heapFlushMinRecords,
+      HeapUsageProvider heapUsageProvider) {
     this.appenderFactory = appenderFactory;
     this.fileFactory = fileFactory;
     this.io = io;
     this.format = format;
     this.writerKey =
         new TaskWriterKey(partitionKey, DataTreeNode.of(mask, index), DataFileType.POS_DELETE_FILE);
     this.recordsNumThreshold = recordsNumThreshold;
+    this.heapUsageRatioThreshold = heapUsageRatioThreshold;
+    this.heapFlushMinRecords = heapFlushMinRecords;
+    this.heapUsageProvider = heapUsageProvider;
+  }
+
+  public SortedPosDeleteWriter(
+      FileAppenderFactory<T> appenderFactory,
+      OutputFileFactory fileFactory,
+      AuthenticatedFileIO io,
+      FileFormat format,
+      long mask,
+      long index,
+      StructLike partitionKey,
+      long recordsNumThreshold) {
+    this(
+        appenderFactory,
+        fileFactory,
+        io,
+        format,
+        mask,
+        index,
+        partitionKey,
+        recordsNumThreshold,
+        0d,
+        0);
   }
 
   public SortedPosDeleteWriter(
@@ -98,7 +175,9 @@ public SortedPosDeleteWriter(
         mask,
         index,
         partitionKey,
-        DEFAULT_RECORDS_NUM_THRESHOLD);
+        DEFAULT_RECORDS_NUM_THRESHOLD,
+        0d,
+        0);
   }
 
   public SortedPosDeleteWriter(
@@ -115,7 +194,31 @@ public SortedPosDeleteWriter(
         0,
         0,
         partitionKey,
-        DEFAULT_RECORDS_NUM_THRESHOLD);
+        DEFAULT_RECORDS_NUM_THRESHOLD,
+        0d,
+        0);
+  }
+
+  public SortedPosDeleteWriter(
+      FileAppenderFactory<T> appenderFactory,
+      OutputFileFactory fileFactory,
+      AuthenticatedFileIO io,
+      FileFormat format,
+      StructLike partitionKey,
+      long recordsNumThreshold,
+      double heapUsageRatioThreshold,
+      int heapFlushMinRecords) {
+    this(
+        appenderFactory,
+        fileFactory,
+        io,
+        format,
+        0,
+        0,
+        partitionKey,
+        recordsNumThreshold,
+        heapUsageRatioThreshold,
+        heapFlushMinRecords);
   }
 
   public void delete(CharSequence path, long pos) {
@@ -132,9 +235,17 @@ public void delete(CharSequence path, long pos, T row) {
 
     records += 1;
 
-    // TODO Flush buffer based on the policy that checking whether whole heap memory size exceed the
-    // threshold.
-    if (records >= recordsNumThreshold) {
+    // Avoid querying JVM memory on every delete (may be costly on some JDKs).
+    // Sample heap usage every heapFlushMinRecords records at most.
+    boolean flushByHeap = false;
+    if (heapUsageRatioThreshold > 0 && heapUsageRatioThreshold < 1) {
+      int checkInterval = Math.max(1, heapFlushMinRecords);
+      if (records % checkInterval == 0) {
+        flushByHeap = shouldFlushByHeap();
+      }
+    }
+
+    if (records >= recordsNumThreshold || flushByHeap) {
       flushDeletes();
     }
   }
@@ -208,6 +319,24 @@ private void flushDeletes() {
     completedFiles.add(writer.toDeleteFile());
   }
 
+  private boolean shouldFlushByHeap() {
+    // Guard: allow disabling heap-based flushing by setting an invalid ratio.
+    if (heapUsageRatioThreshold <= 0 || heapUsageRatioThreshold >= 1) {
+      return false;
+    }
+    // Guard: avoid flushing too frequently on small buffers.
+    if (records < Math.max(1, heapFlushMinRecords)) {
+      return false;
+    }
+    long max = heapUsageProvider.maxMemory();
+    if (max <= 0) {
+      return false;
+    }
+    // Approximate current heap usage; do not force GC here.
+    long used = heapUsageProvider.usedMemory();
+    return used >= (long) (max * heapUsageRatioThreshold);
+  }
+
   private static class PosRow<R> {
     private final long pos;
     private final R row;

amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/MixedIcebergRewriteExecutor.java

@@ -55,7 +55,8 @@ protected FileWriter<PositionDelete<Record>, DeleteWriteResult> posWriter() {
         encryptionManager(),
         getTransactionId(input.rePosDeletedDataFilesForMixed()),
         baseLocation(),
-        table.spec());
+        table.spec(),
+        table.properties());
   }
 
   @Override

amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java

@@ -300,6 +300,16 @@ private TableProperties() {}
       org.apache.iceberg.TableProperties.WRITE_TARGET_FILE_SIZE_BYTES;
   public static final long WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT = 134217728; // 128 MB
 
+  public static final String POS_DELETE_FLUSH_HEAP_RATIO = "pos-delete.flush.heap.ratio";
+  public static final double POS_DELETE_FLUSH_HEAP_RATIO_DEFAULT = 0.8d;
+
+  public static final String POS_DELETE_FLUSH_RECORDS = "pos-delete.flush.records";
+  public static final long POS_DELETE_FLUSH_RECORDS_DEFAULT = Long.MAX_VALUE;
+
+  public static final String POS_DELETE_FLUSH_HEAP_MIN_RECORDS =
+      "pos-delete.flush.heap.min-records";
+  public static final int POS_DELETE_FLUSH_HEAP_MIN_RECORDS_DEFAULT = 1000;
+
   public static final String UPSERT_ENABLED = "write.upsert.enabled";
   public static final boolean UPSERT_ENABLED_DEFAULT = false;
 

amoro-format-iceberg/src/test/java/org/apache/amoro/io/TestMixedTreeNodePosDeleteWriter.java

@@ -75,7 +75,8 @@ public void test() throws IOException {
             table.encryption(),
             1L,
             table.location(),
-            table.spec());
+            table.spec(),
+            table.properties());
 
     writer.setTreeNode(DataTreeNode.ofId(4));
     writer.delete("a", 0);