From be4bc842225009cabe7a22ca7e6b23200b14f3b9 Mon Sep 17 00:00:00 2001
From: Oleksandr Nitavskyi <oleksandr.nitavskyi@datadoghq.com>
Date: Wed, 16 Sep 2026 12:50:44 +0200
Subject: [PATCH] Add per-table rowsWritten counter to StoreSinkWriteImpl

There is no continuous per-(table, subtask) view of write throughput:
CommitMetrics counters only update when a commit lands, so they go
dark during checkpoint stalls that need diagnosing, and every
writerBuffer gauge except numWriters reports the shared per-subtask
buffer pool rather than anything per-table.

Register a rowsWritten counter on the writer operator's MetricGroup
under a per-table metric group, so it is tagged with both table and
subtask and is polled independently of checkpointing.

Counted on the returned record so rows dropped by the row-kind filter
(ignore-delete) are excluded, and in all three write() overloads since
they do not delegate to each other. Rows are counted entering the
write buffer, making this an upper bound on
---
 .../paimon/flink/sink/StoreSinkWriteImpl.java | 37 +++++++++++++++++--
 1 file changed, 34 insertions(+), 3 deletions(-)

diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWriteImpl.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWriteImpl.java
index 797f6ad86d04..e63495942db8 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWriteImpl.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWriteImpl.java
@@ -24,6 +24,7 @@
 import org.apache.paimon.flink.metrics.FlinkMetricRegistry;
 import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.memory.MemoryPoolFactory;
+import org.apache.paimon.metrics.Counter;
 import org.apache.paimon.operation.FileStoreWrite;
 import org.apache.paimon.operation.WriteRestore;
 import org.apache.paimon.table.FileStoreTable;
@@ -49,6 +50,11 @@ public class StoreSinkWriteImpl implements StoreSinkWrite {
 
     private static final Logger LOG = LoggerFactory.getLogger(StoreSinkWriteImpl.class);
 
+    /** Per-(table, subtask) write throughput, unlike the commit-gated CommitMetrics counters. */
+    private static final String WRITER_METRIC_GROUP = "writer";
+
+    private static final String ROWS_WRITTEN_METRIC = "rowsWritten";
+
     protected final String commitUser;
     protected final StoreSinkWriteState state;
     private final IOManagerImpl paimonIOManager;
@@ -58,6 +64,7 @@ public class StoreSinkWriteImpl implements StoreSinkWrite {
     private final MemoryPoolFactory memoryPoolFactory;
     @Nullable private final MetricGroup metricGroup;
     private final TableWriteFactory tableWriteFactory;
+    @Nullable private final Counter rowsWritten;
 
     @Nullable private UriReaderFactory blobDescriptorReaderFactory;
 
@@ -106,6 +113,14 @@ public StoreSinkWriteImpl(
         this.memoryPoolFactory = memoryPoolFactory;
         this.metricGroup = metricGroup;
         this.tableWriteFactory = tableWriteFactory;
+        // Not derived from newTableWrite: replace() rebuilds the write on schema evolution,
+        // and the counter must survive that, not reset.
+        this.rowsWritten =
+                metricGroup == null
+                        ? null
+                        : new FlinkMetricRegistry(metricGroup)
+                                .createTableMetricGroup(WRITER_METRIC_GROUP, table.name())
+                                .counter(ROWS_WRITTEN_METRIC);
         this.write = newTableWrite(table);
     }
 
@@ -140,19 +155,20 @@ public void setBlobDescriptorReaderFactory(UriReaderFactory uriReaderFactory) {
     @Override
     @Nullable
     public SinkRecord write(InternalRow rowData) throws Exception {
-        return write.writeAndReturn(withBlobDescriptorReader(rowData));
+        return countRow(write.writeAndReturn(withBlobDescriptorReader(rowData)));
     }
 
     @Override
     @Nullable
     public SinkRecord write(InternalRow rowData, int bucket) throws Exception {
-        return write.writeAndReturn(withBlobDescriptorReader(rowData), bucket);
+        return countRow(write.writeAndReturn(withBlobDescriptorReader(rowData), bucket));
     }
 
     @Override
     @Nullable
     public SinkRecord write(InternalRow rowData, int bucket, int totalBuckets) throws Exception {
-        return write.writeAndReturn(withBlobDescriptorReader(rowData), bucket, totalBuckets);
+        return countRow(
+                write.writeAndReturn(withBlobDescriptorReader(rowData), bucket, totalBuckets));
     }
 
     private InternalRow withBlobDescriptorReader(InternalRow rowData) {
@@ -161,6 +177,21 @@ private InternalRow withBlobDescriptorReader(InternalRow rowData) {
                 : new BlobDescriptorResolvingRow(rowData, blobDescriptorReaderFactory);
     }
 
+    /**
+     * Counts accepted rows. On the result, not on entry: writeAndReturn returns null for rows the
+     * row-kind filter drops. In all three overloads because they do not delegate to each other.
+     *
+     * <p>Rows entering the write buffer, so an upper bound on LAST_DELTA_RECORDS_APPENDED --
+     * MergeTreeWriter merges same-key rows on flush.
+     */
+    @Nullable
+    private SinkRecord countRow(@Nullable SinkRecord record) {
+        if (rowsWritten != null && record != null) {
+            rowsWritten.inc();
+        }
+        return record;
+    }
+
     @Override
     public void compact(BinaryRow partition, int bucket, boolean fullCompaction) throws Exception {
         write.compact(partition, bucket, fullCompaction);