[metrics] Add io-metrics for paimon in Flink

docs/layouts/shortcodes/generated/core_configuration.html

@@ -1014,6 +1014,12 @@
             <td>Boolean</td>
             <td>Whether to read the changes from overwrite in streaming mode. Cannot be set to true when changelog producer is full-compaction or lookup because it will read duplicated changes.</td>
         </tr>
+        <tr>
+            <td><h5>metrics.file-io.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether to enable MetricsFileIO metrics statistics.</td>
+        </tr>
         <tr>
             <td><h5>streaming.read.snapshot.delay</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -801,6 +801,12 @@ public class CoreOptions implements Serializable {
                                     + "Note: Scale-up this parameter will increase memory usage while scanning manifest files. "
                                     + "We can consider downsize it when we encounter an out of memory exception while scanning");
 
+    public static final ConfigOption<Boolean> METRICS_FILE_IO_ENABLED =
+            key("metrics.file-io.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription("Whether to enable MetricsFileIO metrics statistics.");
+
     public static final ConfigOption<Duration> STREAMING_READ_SNAPSHOT_DELAY =
             key("streaming.read.snapshot.delay")
                     .durationType()
@@ -2369,6 +2375,10 @@ public Integer scanManifestParallelism() {
         return options.get(SCAN_MANIFEST_PARALLELISM);
     }
 
+    public boolean isMetricsFileIOEnabled() {
+        return options.get(METRICS_FILE_IO_ENABLED);
+    }
+
     public Duration streamingReadDelay() {
         return options.get(STREAMING_READ_SNAPSHOT_DELAY);
     }

paimon-common/pom.xml

@@ -204,6 +204,12 @@ under the License.
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-math3</artifactId>
+            <version>3.6.1</version>
+        </dependency>
+
         <!-- for the HDFS mini cluster test suite -->
         <dependency>
             <groupId>org.apache.hadoop</groupId>

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

@@ -44,4 +44,17 @@ public static FileIOLoader checkAccess(FileIOLoader fileIO, Path path, CatalogCo
         io.exists(path);
         return fileIO;
     }
+
+    /**
+     * Get MetricsFileIO instance, avoiding multiple wrapping.
+     *
+     * @param fileIO the original FileIO instance
+     * @return MetricsFileIO instance (existing or new)
+     */
+    public static MetricsFileIO getMetricsFileIO(FileIO fileIO) {
+        if (fileIO instanceof MetricsFileIO) {
+            return (MetricsFileIO) fileIO;
+        }
+        return new MetricsFileIO(fileIO);
+    }
 }

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

@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.fs;
+
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.fs.metrics.IOMetrics;
+
+import java.io.IOException;
+
+/**
+ * MetricsFileIO is a wrapper class for the {@link FileIO}.
+ *
+ * <p>It allows users to monitor and track file I/O operations.
+ */
+public class MetricsFileIO implements FileIO {
+
+    protected final FileIO fileIO;
+    protected IOMetrics ioMetrics = null;
+
+    public MetricsFileIO(FileIO fileIO) {
+        this.fileIO = fileIO;
+    }
+
+    public MetricsFileIO withMetrics(IOMetrics ioMetrics) {
+        this.ioMetrics = ioMetrics;
+        return this;
+    }
+
+    public FileIO getFileIOInternal() {
+        FileIO currentFileIO = fileIO;
+        while (currentFileIO instanceof MetricsFileIO) {
+            currentFileIO = ((MetricsFileIO) currentFileIO).fileIO;
+        }
+        return currentFileIO;
+    }
+
+    @VisibleForTesting
+    public Boolean isMetricsEnabled() {
+        return ioMetrics != null;
+    }
+
+    @Override
+    public boolean isObjectStore() {
+        return fileIO.isObjectStore();
+    }
+
+    @Override
+    public void configure(CatalogContext context) {
+        fileIO.configure(context);
+    }
+
+    @Override
+    public SeekableInputStream newInputStream(Path path) throws IOException {
+        SeekableInputStream inputStream = fileIO.newInputStream(path);
+        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, this.ioMetrics);
+    }
+
+    @Override
+    public FileStatus getFileStatus(Path path) throws IOException {
+        return fileIO.getFileStatus(path);
+    }
+
+    @Override
+    public FileStatus[] listStatus(Path path) throws IOException {
+        return fileIO.listStatus(path);
+    }
+
+    @Override
+    public boolean exists(Path path) throws IOException {
+        return fileIO.exists(path);
+    }
+
+    @Override
+    public boolean delete(Path path, boolean recursive) throws IOException {
+        return fileIO.delete(path, recursive);
+    }
+
+    @Override
+    public boolean mkdirs(Path path) throws IOException {
+        return fileIO.mkdirs(path);
+    }
+
+    @Override
+    public boolean rename(Path src, Path dst) throws IOException {
+        return fileIO.rename(src, dst);
+    }
+}

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

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.fs;
+
+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 IOMetrics metrics;
+
+    public PositionOutputStreamIOWrapper(PositionOutputStream out, IOMetrics metrics) {
+        this.out = out;
+        this.metrics = metrics;
+    }
+
+    @Override
+    public long getPos() throws IOException {
+        return out.getPos();
+    }
+
+    @Override
+    public void write(int b) throws IOException {
+        try {
+            out.write(b);
+        } finally {
+            if (metrics != null) {
+                metrics.recordWriteEvent(1);
+            }
+        }
+    }
+
+    @Override
+    public void write(byte[] b) throws IOException {
+        try {
+            out.write(b);
+        } finally {
+            if (metrics != null) {
+                metrics.recordWriteEvent(b.length);
+            }
+        }
+    }
+
+    @Override
+    public void write(byte[] b, int off, int len) throws IOException {
+        try {
+            out.write(b, off, len);
+        } finally {
+            if (metrics != null) {
+                metrics.recordWriteEvent(len);
+            }
+        }
+    }
+
+    @Override
+    public void flush() throws IOException {
+        out.flush();
+    }
+
+    @Override
+    public void close() throws IOException {
+        out.close();
+    }
+}

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

@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.fs;
+
+import org.apache.paimon.fs.metrics.IOMetrics;
+
+import java.io.IOException;
+
+/** Wrap a {@link SeekableInputStream}. */
+public class SeekableInputStreamIOWrapper extends SeekableInputStream implements VectoredReadable {
+
+    protected final SeekableInputStream in;
+    private IOMetrics metrics;
+
+    public SeekableInputStreamIOWrapper(SeekableInputStream in, IOMetrics metrics) {
+        this.in = in;
+        this.metrics = metrics;
+    }
+
+    @Override
+    public void seek(long desired) throws IOException {
+        in.seek(desired);
+    }
+
+    @Override
+    public long getPos() throws IOException {
+        return in.getPos();
+    }
+
+    @Override
+    public int read() throws IOException {
+        int bytesRead = 0;
+        try {
+            bytesRead = in.read();
+        } finally {
+            if (metrics != null && bytesRead != -1) {
+                metrics.recordReadEvent(bytesRead);
+            }
+        }
+        return bytesRead;
+    }
+
+    @Override
+    public int read(byte[] b, int off, int len) throws IOException {
+        int bytesRead = 0;
+        try {
+            bytesRead = in.read(b, off, len);
+        } finally {
+            if (metrics != null && bytesRead != -1) {
+                metrics.recordReadEvent(bytesRead);
+            }
+        }
+        return bytesRead;
+    }
+
+    @Override
+    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;
+    }
+}