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. + * + *
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);