update metrics

Author: zhuanshenbsj1

paimon-common/src/main/java/org/apache/paimon/fs/MetricsFileIO.java

@@ -20,8 +20,7 @@
 
 import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.catalog.CatalogContext;
-import org.apache.paimon.fs.metrics.InputMetrics;
-import org.apache.paimon.fs.metrics.OutputMetrics;
+import org.apache.paimon.fs.metrics.IOMetrics;
 
 import java.io.IOException;
 
@@ -33,16 +32,14 @@
 public class MetricsFileIO implements FileIO {
 
     protected final FileIO fileIO;
-    protected InputMetrics inputMetrics = null;
-    protected OutputMetrics outputMetrics = null;
+    protected IOMetrics ioMetrics = null;
 
     public MetricsFileIO(FileIO fileIO) {
         this.fileIO = fileIO;
     }
 
-    public MetricsFileIO withMetrics(InputMetrics inputMetrics, OutputMetrics outputMetrics) {
-        this.inputMetrics = inputMetrics;
-        this.outputMetrics = outputMetrics;
+    public MetricsFileIO withMetrics(IOMetrics ioMetrics) {
+        this.ioMetrics = ioMetrics;
         return this;
     }
 
@@ -56,7 +53,7 @@ public FileIO getFileIOInternal() {
 
     @VisibleForTesting
     public Boolean isMetricsEnabled() {
-        return inputMetrics != null && outputMetrics != null;
+        return ioMetrics != null;
     }
 
     @Override
@@ -72,13 +69,13 @@ public void configure(CatalogContext context) {
     @Override
     public SeekableInputStream newInputStream(Path path) throws IOException {
         SeekableInputStream inputStream = fileIO.newInputStream(path);
-        return new SeekableInputStreamIOWrapper(inputStream, this.inputMetrics);
+        return new SeekableInputStreamIOWrapper(inputStream, this.ioMetrics);
     }
 
     @Override
     public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws IOException {
         PositionOutputStream outputStream = fileIO.newOutputStream(path, overwrite);
-        return new PositionOutputStreamIOWrapper(outputStream, outputMetrics);
+        return new PositionOutputStreamIOWrapper(outputStream, this.ioMetrics);
     }
 
     @Override

paimon-common/src/main/java/org/apache/paimon/fs/PositionOutputStreamIOWrapper.java

@@ -18,19 +18,19 @@
 
 package org.apache.paimon.fs;
 
-import org.apache.paimon.fs.metrics.OutputMetrics;
+import org.apache.paimon.fs.metrics.IOMetrics;
 
 import java.io.IOException;
 
 /** Wrap a {@link PositionOutputStream}. */
 public class PositionOutputStreamIOWrapper extends PositionOutputStream {
 
     protected final PositionOutputStream out;
-    private OutputMetrics metrics;
+    private IOMetrics metrics;
 
-    public PositionOutputStreamIOWrapper(PositionOutputStream out, OutputMetrics outputMetrics) {
+    public PositionOutputStreamIOWrapper(PositionOutputStream out, IOMetrics metrics) {
         this.out = out;
-        this.metrics = outputMetrics;
+        this.metrics = metrics;
     }
 
     @Override

paimon-common/src/main/java/org/apache/paimon/fs/SeekableInputStreamIOWrapper.java

@@ -18,17 +18,17 @@
 
 package org.apache.paimon.fs;
 
-import org.apache.paimon.fs.metrics.InputMetrics;
+import org.apache.paimon.fs.metrics.IOMetrics;
 
 import java.io.IOException;
 
 /** Wrap a {@link SeekableInputStream}. */
-public class SeekableInputStreamIOWrapper extends SeekableInputStream {
+public class SeekableInputStreamIOWrapper extends SeekableInputStream implements VectoredReadable {
 
     protected final SeekableInputStream in;
-    private InputMetrics metrics;
+    private IOMetrics metrics;
 
-    public SeekableInputStreamIOWrapper(SeekableInputStream in, InputMetrics metrics) {
+    public SeekableInputStreamIOWrapper(SeekableInputStream in, IOMetrics metrics) {
         this.in = in;
         this.metrics = metrics;
     }
@@ -73,4 +73,24 @@ public int read(byte[] b, int off, int len) throws IOException {
     public void close() throws IOException {
         in.close();
     }
+
+    @Override
+    public int pread(long position, byte[] buffer, int offset, int length) throws IOException {
+        int bytesRead = 0;
+        try {
+            if (in instanceof VectoredReadable) {
+                bytesRead = ((VectoredReadable) in).pread(position, buffer, offset, length);
+            } else {
+                long originalPos = in.getPos();
+                in.seek(position);
+                bytesRead = in.read(buffer, offset, length);
+                in.seek(originalPos);
+            }
+        } finally {
+            if (metrics != null && bytesRead != -1) {
+                metrics.recordReadEvent(bytesRead);
+            }
+        }
+        return bytesRead;
+    }
 }

paimon-common/src/main/java/org/apache/paimon/fs/metrics/InputMetrics.java renamed to paimon-common/src/main/java/org/apache/paimon/fs/metrics/IOMetrics.java

@@ -24,25 +24,31 @@
 import java.util.concurrent.atomic.AtomicLong;
 
 /** Collects and monitors input stream metrics. */
-public class InputMetrics {
+public class IOMetrics {
 
-    public static final String GROUP_NAME = "source";
+    public static final String GROUP_NAME = "io";
     private final MetricGroup metricGroup;
 
     public static final String READ_BYTES = "read.bytes";
     public static final String READ_OPERATIONS = "read.operations";
+    public static final String WRITE_BYTES = "write.bytes";
+    public static final String WRITE_OPERATIONS = "write.operations";
 
     private final AtomicLong readBytes = new AtomicLong(0);
     private final AtomicLong readOperations = new AtomicLong(0);
+    private final AtomicLong writeBytes = new AtomicLong(0);
+    private final AtomicLong writeOperations = new AtomicLong(0);
 
-    public InputMetrics(MetricRegistry registry, String tableName) {
+    public IOMetrics(MetricRegistry registry, String tableName) {
         metricGroup = registry.createTableMetricGroup(GROUP_NAME, tableName);
         registerMetrics();
     }
 
     private void registerMetrics() {
         metricGroup.gauge(READ_BYTES, this::getReadBytes);
         metricGroup.gauge(READ_OPERATIONS, this::getReadOperations);
+        metricGroup.gauge(WRITE_BYTES, this::getWriteBytes);
+        metricGroup.gauge(WRITE_OPERATIONS, this::getWriteOperations);
     }
 
     public void recordReadEvent(long bytes) {
@@ -57,4 +63,17 @@ public long getReadBytes() {
     public long getReadOperations() {
         return readOperations.get();
     }
+
+    public void recordWriteEvent(long bytes) {
+        writeBytes.addAndGet(bytes);
+        writeOperations.incrementAndGet();
+    }
+
+    public long getWriteBytes() {
+        return writeBytes.get();
+    }
+
+    public long getWriteOperations() {
+        return writeOperations.get();
+    }
 }

paimon-common/src/main/java/org/apache/paimon/fs/metrics/OutputMetrics.java

@@ -25,8 +25,7 @@
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.MetricsFileIO;
 import org.apache.paimon.fs.Path;
-import org.apache.paimon.fs.metrics.InputMetrics;
-import org.apache.paimon.fs.metrics.OutputMetrics;
+import org.apache.paimon.fs.metrics.IOMetrics;
 import org.apache.paimon.manifest.IndexManifestEntry;
 import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.manifest.ManifestFileMeta;
@@ -94,8 +93,6 @@ abstract class AbstractFileStoreTable implements FileStoreTable {
     private static final String WATERMARK_PREFIX = "watermark-";
 
     protected final MetricsFileIO fileIO;
-    private InputMetrics inputMetrics;
-    private OutputMetrics outputMetrics;
     protected final Path path;
     protected final TableSchema tableSchema;
     protected final CatalogEnvironment catalogEnvironment;
@@ -127,9 +124,8 @@ public AbstractFileStoreTable withMetricRegistry(MetricRegistry registry) {
                     catalogEnvironment.identifier() != null
                             ? catalogEnvironment.identifier().getTableName()
                             : "unknown";
-            this.inputMetrics = new InputMetrics(registry, tableName);
-            this.outputMetrics = new OutputMetrics(registry, tableName);
-            this.fileIO.withMetrics(inputMetrics, outputMetrics);
+            IOMetrics ioMetrics = new IOMetrics(registry, tableName);
+            this.fileIO.withMetrics(ioMetrics);
         }
         return this;
     }

paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java

@@ -55,6 +55,8 @@ public class ReadBuilderImpl implements ReadBuilder {
 
     private boolean dropStats = false;
 
+    private MetricRegistry metricRegistry = null;
+
     public ReadBuilderImpl(InnerTable table) {
         this.table = table;
     }
@@ -66,6 +68,7 @@ public String tableName() {
 
     @Override
     public void withMetricsRegistry(MetricRegistry metricRegistry) {
+        this.metricRegistry = metricRegistry;
         this.table.withMetricRegistry(metricRegistry);
     }
 
@@ -156,7 +159,8 @@ public TableScan newScan() {
 
     @Override
     public StreamTableScan newStreamScan() {
-        return (StreamTableScan) configureScan(table.newStreamScan());
+        return (StreamTableScan)
+                configureScan(table.newStreamScan()).withMetricRegistry(metricRegistry);
     }
 
     private InnerTableScan configureScan(InnerTableScan scan) {
@@ -188,7 +192,8 @@ private InnerTableScan configureScan(InnerTableScan scan) {
 
     @Override
     public TableRead newRead() {
-        InnerTableRead read = table.newRead().withFilter(filter);
+        InnerTableRead read =
+                table.newRead().withFilter(filter).withMetricRegistry(this.metricRegistry);
         if (readType != null) {
             read.withReadType(readType);
         }

paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java

@@ -27,11 +27,11 @@
 import org.apache.paimon.data.BinaryString;
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.MetricsFileIO;
 import org.apache.paimon.function.Function;
 import org.apache.paimon.function.FunctionChange;
 import org.apache.paimon.function.FunctionDefinition;
-import org.apache.paimon.fs.FileIO;
-import org.apache.paimon.fs.MetricsFileIO;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.partition.Partition;
 import org.apache.paimon.partition.PartitionStatistics;

paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java

@@ -25,7 +25,6 @@
 import org.apache.paimon.options.Options;
 import org.apache.paimon.table.BucketMode;
 import org.apache.paimon.table.source.ReadBuilder;
-import org.apache.paimon.table.source.StreamDataTableScan;
 import org.apache.paimon.table.source.StreamTableScan;
 
 import org.apache.flink.api.connector.source.Boundedness;
@@ -83,10 +82,6 @@ public SplitEnumerator<FileStoreSourceSplit, PendingSplitsCheckpoint> restoreEnu
             readBuilder.withMetricsRegistry(new FlinkMetricRegistry(context.metricGroup()));
         }
         StreamTableScan scan = readBuilder.newStreamScan();
-        if (metricGroup(context) != null) {
-            ((StreamDataTableScan) scan)
-                    .withMetricRegistry(new FlinkMetricRegistry(context.metricGroup()));
-        }
         scan.restore(nextSnapshotId);
         return buildEnumerator(context, splits, nextSnapshotId, scan);
     }

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileStoreSource.java

@@ -66,8 +66,7 @@ public SourceReader<RowData, FileStoreSourceSplit> createReader(SourceReaderCont
         FileStoreSourceReaderMetrics sourceReaderMetrics =
                 new FileStoreSourceReaderMetrics(metricGroup);
         this.readBuilder.withMetricsRegistry(new FlinkMetricRegistry(metricGroup));
-        TableRead tableRead =
-                readBuilder.newRead().withMetricRegistry(new FlinkMetricRegistry(metricGroup));
+        TableRead tableRead = readBuilder.newRead();
         return new FileStoreSourceReader(
                 context,
                 tableRead,