add memory buffer preempt count

paimon-core/src/main/java/org/apache/paimon/memory/MemoryPoolFactory.java

@@ -37,6 +37,8 @@ public class MemoryPoolFactory {
 
     private Iterable<MemoryOwner> owners;
 
+    private long memoryPreemptCount;
+
     public MemoryPoolFactory(MemorySegmentPool innerPool) {
         this.innerPool = innerPool;
         this.totalPages = innerPool.freePages();
@@ -57,6 +59,10 @@ public void notifyNewOwner(MemoryOwner owner) {
         owner.setMemoryPool(createSubPool(owner));
     }
 
+    public long getMemoryPreemptCount() {
+        return memoryPreemptCount;
+    }
+
     @VisibleForTesting
     public Iterable<MemoryOwner> memoryOwners() {
         return owners;
@@ -81,6 +87,7 @@ private void preemptMemory(MemoryOwner owner) {
         if (max != null) {
             try {
                 max.flushMemory();
+                memoryPreemptCount++;
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }

paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java

@@ -74,6 +74,8 @@ public abstract class AbstractFileStoreWrite<T>
     private boolean ignorePreviousFiles = false;
     protected boolean isStreamingMode = false;
 
+    private MemoryPoolFactory memoryPoolFactory;
+
     protected AbstractFileStoreWrite(
             String commitUser,
             SnapshotManager snapshotManager,
@@ -95,9 +97,18 @@ public FileStoreWrite<T> withIOManager(IOManager ioManager) {
 
     @Override
     public FileStoreWrite<T> withMemoryPoolFactory(MemoryPoolFactory memoryPoolFactory) {
+        this.memoryPoolFactory = memoryPoolFactory;
         return this;
     }
 
+    public long getMemoryPreemptCount() {
+        if (memoryPoolFactory != null) {
+            return this.memoryPoolFactory.getMemoryPreemptCount();
+        } else {
+            return -1L;
+        }
+    }
+
     @Override
     public void withIgnorePreviousFiles(boolean ignorePreviousFiles) {
         this.ignorePreviousFiles = ignorePreviousFiles;

paimon-core/src/main/java/org/apache/paimon/operation/MemoryFileStoreWrite.java

@@ -68,6 +68,7 @@ public MemoryFileStoreWrite(
 
     @Override
     public FileStoreWrite<T> withMemoryPoolFactory(MemoryPoolFactory memoryPoolFactory) {
+        super.withMemoryPoolFactory(memoryPoolFactory);
         this.writeBufferPool = memoryPoolFactory.addOwners(this::memoryOwners);
         return this;
     }

paimon-core/src/main/java/org/apache/paimon/table/sink/TableWrite.java

@@ -32,6 +32,7 @@
  */
 @Public
 public interface TableWrite extends AutoCloseable {
+    long getMemoryPreemptCount();
 
     /** With {@link IOManager}, this is needed if 'write-buffer-spillable' is set to true. */
     TableWrite withIOManager(IOManager ioManager);

paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java

@@ -70,6 +70,11 @@ public TableWriteImpl<T> isStreamingMode(boolean isStreamingMode) {
         return this;
     }
 
+    @Override
+    public long getMemoryPreemptCount() {
+        return write.getMemoryPreemptCount();
+    }
+
     @Override
     public TableWriteImpl<T> withIOManager(IOManager ioManager) {
         write.withIOManager(ioManager);

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWrite.java

@@ -40,6 +40,8 @@ public interface StoreSinkWrite {
 
     SinkRecord write(InternalRow rowData) throws Exception;
 
+    long getMemoryPreemptCount();
+
     SinkRecord toLogRecord(SinkRecord record);
 
     void compact(BinaryRow partition, int bucket, boolean fullCompaction) throws Exception;

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWriteImpl.java

@@ -159,6 +159,11 @@ public SinkRecord write(InternalRow rowData) throws Exception {
         return write.writeAndReturn(rowData);
     }
 
+    @Override
+    public long getMemoryPreemptCount() {
+        return write.getMemoryPreemptCount();
+    }
+
     @Override
     public SinkRecord toLogRecord(SinkRecord record) {
         return write.toLogRecord(record);

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/TableWriteOperator.java

@@ -92,6 +92,11 @@ void initStateAndWriter(
         state = new StoreSinkWriteState(context, stateFilter);
 
         write = storeSinkWriteProvider.provide(table, commitUser, state, ioManager, memoryPool);
+        addWriterMetric();
+    }
+
+    void addWriterMetric() {
+        getMetricGroup().gauge("writer_preempt_count", () -> write.getMemoryPreemptCount());
     }
 
     protected abstract boolean containLogSystem();