writer metric

paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java

@@ -34,6 +34,7 @@
 import org.apache.paimon.io.RowDataRollingFileWriter;
 import org.apache.paimon.memory.MemoryOwner;
 import org.apache.paimon.memory.MemorySegmentPool;
+import org.apache.paimon.operation.metrics.WriterMetrics;
 import org.apache.paimon.statistics.FieldStatsCollector;
 import org.apache.paimon.types.RowKind;
 import org.apache.paimon.types.RowType;
@@ -74,6 +75,8 @@ public class AppendOnlyWriter implements RecordWriter<InternalRow>, MemoryOwner
     private final FieldStatsCollector.Factory[] statsCollectors;
     private final IOManager ioManager;
 
+    private WriterMetrics writerMetrics;
+
     public AppendOnlyWriter(
             FileIO fileIO,
             IOManager ioManager,
@@ -89,7 +92,8 @@ public AppendOnlyWriter(
             boolean useWriteBuffer,
             boolean spillable,
             String fileCompression,
-            FieldStatsCollector.Factory[] statsCollectors) {
+            FieldStatsCollector.Factory[] statsCollectors,
+            WriterMetrics writerMetrics) {
         this.fileIO = fileIO;
         this.schemaId = schemaId;
         this.fileFormat = fileFormat;
@@ -114,10 +118,12 @@ public AppendOnlyWriter(
             compactBefore.addAll(increment.compactIncrement().compactBefore());
             compactAfter.addAll(increment.compactIncrement().compactAfter());
         }
+        this.writerMetrics = writerMetrics;
     }
 
     @Override
     public void write(InternalRow rowData) throws Exception {
+        long start = System.currentTimeMillis();
         Preconditions.checkArgument(
                 rowData.getRowKind() == RowKind.INSERT,
                 "Append-only writer can only accept insert row kind, but current row kind is: %s",
@@ -133,6 +139,11 @@ public void write(InternalRow rowData) throws Exception {
                 throw new RuntimeException("Mem table is too small to hold a single element.");
             }
         }
+
+        if (writerMetrics != null) {
+            writerMetrics.incWriteRecordNum();
+            writerMetrics.updateWriteCostMS(System.currentTimeMillis() - start);
+        }
     }
 
     @Override
@@ -152,20 +163,29 @@ public Collection<DataFileMeta> dataFiles() {
 
     @Override
     public CommitIncrement prepareCommit(boolean waitCompaction) throws Exception {
+        long start = System.currentTimeMillis();
         flush(false, false);
         trySyncLatestCompaction(waitCompaction || forceCompact);
-        return drainIncrement();
+        CommitIncrement increment = drainIncrement();
+        if (writerMetrics != null) {
+            writerMetrics.updatePrepareCommitCostMS(System.currentTimeMillis() - start);
+        }
+        return increment;
     }
 
     private void flush(boolean waitForLatestCompaction, boolean forcedFullCompaction)
             throws Exception {
+        long start = System.currentTimeMillis();
         List<DataFileMeta> flushedFiles = sinkWriter.flush();
 
         // add new generated files
         flushedFiles.forEach(compactManager::addNewFile);
         trySyncLatestCompaction(waitForLatestCompaction);
         compactManager.triggerCompaction(forcedFullCompaction);
         newFiles.addAll(flushedFiles);
+        if (writerMetrics != null) {
+            writerMetrics.updateBufferFlushCostMS(System.currentTimeMillis() - start);
+        }
     }
 
     @Override
@@ -204,13 +224,15 @@ private RowDataRollingFileWriter createRollingRowWriter() {
 
     private void trySyncLatestCompaction(boolean blocking)
             throws ExecutionException, InterruptedException {
+        long start = System.currentTimeMillis();
         compactManager
                 .getCompactionResult(blocking)
                 .ifPresent(
                         result -> {
                             compactBefore.addAll(result.before());
                             compactAfter.addAll(result.after());
                         });
+        writerMetrics.updateSyncLastestCompactionCostMS(System.currentTimeMillis() - start);
     }
 
     private CommitIncrement drainIncrement() {

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

@@ -37,9 +37,13 @@ public class MemoryPoolFactory {
 
     private Iterable<MemoryOwner> owners;
 
+    private final long totalBufferSize;
+    private long bufferPreemptCount;
+
     public MemoryPoolFactory(MemorySegmentPool innerPool) {
         this.innerPool = innerPool;
         this.totalPages = innerPool.freePages();
+        this.totalBufferSize = (long) totalPages * innerPool.pageSize();
     }
 
     public MemoryPoolFactory addOwners(Iterable<MemoryOwner> newOwners) {
@@ -81,12 +85,38 @@ private void preemptMemory(MemoryOwner owner) {
         if (max != null) {
             try {
                 max.flushMemory();
+                ++bufferPreemptCount;
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }
     }
 
+    public BufferStat bufferStat() {
+        return new BufferStat();
+    }
+
+    /** stat for buffer metric. */
+    public class BufferStat {
+        public long bufferPreemptCount() {
+            return bufferPreemptCount;
+        }
+
+        public long usedBufferSize() {
+            long usedBufferSize = 0L;
+            if (owners != null) {
+                for (MemoryOwner owner : owners) {
+                    usedBufferSize += owner.memoryOccupancy();
+                }
+            }
+            return usedBufferSize;
+        }
+
+        public long totalBufferSize() {
+            return totalBufferSize;
+        }
+    }
+
     private class OwnerMemoryPool implements MemorySegmentPool {
 
         private final MemoryOwner owner;

paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java

@@ -33,6 +33,7 @@
 import org.apache.paimon.memory.MemoryOwner;
 import org.apache.paimon.memory.MemorySegmentPool;
 import org.apache.paimon.mergetree.compact.MergeFunction;
+import org.apache.paimon.operation.metrics.WriterMetrics;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.CommitIncrement;
 import org.apache.paimon.utils.RecordWriter;
@@ -74,6 +75,8 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
     private long newSequenceNumber;
     private WriteBuffer writeBuffer;
 
+    private WriterMetrics writerMetrics;
+
     public MergeTreeWriter(
             boolean writeBufferSpillable,
             int sortMaxFan,
@@ -85,7 +88,8 @@ public MergeTreeWriter(
             KeyValueFileWriterFactory writerFactory,
             boolean commitForceCompact,
             ChangelogProducer changelogProducer,
-            @Nullable CommitIncrement increment) {
+            @Nullable CommitIncrement increment,
+            WriterMetrics writerMetrics) {
         this.writeBufferSpillable = writeBufferSpillable;
         this.sortMaxFan = sortMaxFan;
         this.ioManager = ioManager;
@@ -114,6 +118,7 @@ public MergeTreeWriter(
             compactAfter.addAll(increment.compactIncrement().compactAfter());
             compactChangelog.addAll(increment.compactIncrement().changelogFiles());
         }
+        this.writerMetrics = writerMetrics;
     }
 
     private long newSequenceNumber() {
@@ -139,10 +144,12 @@ public void setMemoryPool(MemorySegmentPool memoryPool) {
 
     @Override
     public void write(KeyValue kv) throws Exception {
+        long start = System.currentTimeMillis();
         long sequenceNumber =
                 kv.sequenceNumber() == KeyValue.UNKNOWN_SEQUENCE
                         ? newSequenceNumber()
                         : kv.sequenceNumber();
+
         boolean success = writeBuffer.put(sequenceNumber, kv.valueKind(), kv.key(), kv.value());
         if (!success) {
             flushWriteBuffer(false, false);
@@ -151,6 +158,11 @@ public void write(KeyValue kv) throws Exception {
                 throw new RuntimeException("Mem table is too small to hold a single element.");
             }
         }
+
+        if (writerMetrics != null) {
+            writerMetrics.incWriteRecordNum();
+            writerMetrics.updateWriteCostMS(System.currentTimeMillis() - start);
+        }
     }
 
     @Override
@@ -183,6 +195,7 @@ public void flushMemory() throws Exception {
 
     private void flushWriteBuffer(boolean waitForLatestCompaction, boolean forcedFullCompaction)
             throws Exception {
+        long start = System.currentTimeMillis();
         if (writeBuffer.size() > 0) {
             if (compactManager.shouldWaitForLatestCompaction()) {
                 waitForLatestCompaction = true;
@@ -222,16 +235,26 @@ private void flushWriteBuffer(boolean waitForLatestCompaction, boolean forcedFul
 
         trySyncLatestCompaction(waitForLatestCompaction);
         compactManager.triggerCompaction(forcedFullCompaction);
+        if (writerMetrics != null) {
+            writerMetrics.updateBufferFlushCostMS(System.currentTimeMillis() - start);
+        }
     }
 
     @Override
     public CommitIncrement prepareCommit(boolean waitCompaction) throws Exception {
+        long start = System.currentTimeMillis();
         flushWriteBuffer(waitCompaction, false);
         trySyncLatestCompaction(
                 waitCompaction
                         || commitForceCompact
                         || compactManager.shouldWaitForPreparingCheckpoint());
-        return drainIncrement();
+
+        CommitIncrement increment = drainIncrement();
+
+        if (writerMetrics != null) {
+            writerMetrics.updatePrepareCommitCostMS(System.currentTimeMillis() - start);
+        }
+        return increment;
     }
 
     @Override
@@ -281,8 +304,10 @@ private void updateCompactResult(CompactResult result) {
     }
 
     private void trySyncLatestCompaction(boolean blocking) throws Exception {
+        long start = System.currentTimeMillis();
         Optional<CompactResult> result = compactManager.getCompactionResult(blocking);
         result.ifPresent(this::updateCompactResult);
+        writerMetrics.updateSyncLastestCompactionCostMS(System.currentTimeMillis() - start);
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/metrics/MetricRegistry.java

@@ -27,10 +27,18 @@ public abstract class MetricRegistry {
     private static final String KEY_TABLE = "table";
     private static final String KEY_PARTITION = "partition";
     private static final String KEY_BUCKET = "bucket";
+    private static final String COMMIT_USER = "commit_user";
 
     public MetricGroup tableMetricGroup(String groupName, String tableName) {
+        return tableMetricGroup(groupName, tableName, null);
+    }
+
+    public MetricGroup tableMetricGroup(String groupName, String tableName, String commitUser) {
         Map<String, String> variables = new LinkedHashMap<>();
         variables.put(KEY_TABLE, tableName);
+        if (commitUser != null) {
+            variables.put(COMMIT_USER, commitUser);
+        }
 
         return createMetricGroup(groupName, variables);
     }

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

@@ -31,6 +31,7 @@
 import org.apache.paimon.memory.MemoryPoolFactory;
 import org.apache.paimon.metrics.MetricRegistry;
 import org.apache.paimon.operation.metrics.CompactionMetrics;
+import org.apache.paimon.operation.metrics.WriterMetrics;
 import org.apache.paimon.table.sink.CommitMessage;
 import org.apache.paimon.table.sink.CommitMessageImpl;
 import org.apache.paimon.utils.CommitIncrement;
@@ -77,7 +78,10 @@ public abstract class AbstractFileStoreWrite<T>
     private boolean closeCompactExecutorWhenLeaving = true;
     private boolean ignorePreviousFiles = false;
     protected boolean isStreamingMode = false;
-    private MetricRegistry metricRegistry = null;
+    private MetricRegistry metricRegistry;
+
+    protected WriterMetrics writerMetrics;
+
     private final String tableName;
     private final FileStorePathFactory pathFactory;
 
@@ -369,6 +373,14 @@ public CompactionMetrics getCompactionMetrics(BinaryRow partition, int bucket) {
         return null;
     }
 
+    @Nullable
+    public WriterMetrics getWriterMetrics() {
+        if (metricRegistry != null && writerMetrics == null) {
+            writerMetrics = new WriterMetrics(metricRegistry, tableName, commitUser);
+        }
+        return writerMetrics;
+    }
+
     private String getPartitionString(FileStorePathFactory pathFactory, BinaryRow partition) {
         String partitionStr =
                 pathFactory.getPartitionString(partition).replace(Path.SEPARATOR, "_");

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

@@ -31,6 +31,7 @@
 import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.io.DataFilePathFactory;
 import org.apache.paimon.io.RowDataRollingFileWriter;
+import org.apache.paimon.operation.metrics.WriterMetrics;
 import org.apache.paimon.reader.RecordReaderIterator;
 import org.apache.paimon.statistics.FieldStatsCollector;
 import org.apache.paimon.table.BucketMode;
@@ -124,6 +125,9 @@ protected RecordWriter<InternalRow> createWriter(
                                 targetFileSize,
                                 compactRewriter(partition, bucket),
                                 getCompactionMetrics(partition, bucket));
+        WriterMetrics writerMetrics = getWriterMetrics();
+        registerMemoryPoolMetric(writerMetrics);
+
         return new AppendOnlyWriter(
                 fileIO,
                 ioManager,
@@ -139,7 +143,8 @@ protected RecordWriter<InternalRow> createWriter(
                 useWriteBuffer,
                 spillable,
                 fileCompression,
-                statsCollectors);
+                statsCollectors,
+                writerMetrics);
     }
 
     public AppendOnlyCompactManager.CompactRewriter compactRewriter(

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

@@ -25,6 +25,7 @@
 import org.apache.paimon.memory.MemoryPoolFactory;
 import org.apache.paimon.memory.MemorySegmentPool;
 import org.apache.paimon.metrics.MetricRegistry;
+import org.apache.paimon.operation.metrics.WriterMetrics;
 import org.apache.paimon.table.sink.CommitMessage;
 import org.apache.paimon.table.sink.SinkRecord;
 import org.apache.paimon.utils.RecordWriter;
@@ -125,4 +126,6 @@ List<CommitMessage> prepareCommit(boolean waitCompaction, long commitIdentifier)
 
     /** With metrics to measure compaction. */
     FileStoreWrite<T> withMetricRegistry(MetricRegistry metricRegistry);
+
+    WriterMetrics getWriterMetrics();
 }

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

@@ -51,6 +51,7 @@
 import org.apache.paimon.mergetree.compact.MergeTreeCompactRewriter;
 import org.apache.paimon.mergetree.compact.UniversalCompaction;
 import org.apache.paimon.operation.metrics.CompactionMetrics;
+import org.apache.paimon.operation.metrics.WriterMetrics;
 import org.apache.paimon.schema.KeyValueFieldsExtractor;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.types.RowType;
@@ -170,6 +171,10 @@ protected MergeTreeWriter createWriter(
                         compactExecutor,
                         levels,
                         getCompactionMetrics(partition, bucket));
+
+        WriterMetrics writerMetrics = getWriterMetrics();
+        registerMemoryPoolMetric(writerMetrics);
+
         return new MergeTreeWriter(
                 bufferSpillable(),
                 options.localSortMaxNumFileHandles(),
@@ -181,7 +186,8 @@ protected MergeTreeWriter createWriter(
                 writerFactory,
                 options.commitForceCompact(),
                 options.changelogProducer(),
-                restoreIncrement);
+                restoreIncrement,
+                writerMetrics);
     }
 
     @VisibleForTesting