io metrics

Author: zhuanshenbsj1

docs/layouts/shortcodes/generated/core_configuration.html

@@ -996,6 +996,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-common/pom.xml

@@ -198,6 +198,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/CoreOptions.java

@@ -796,6 +796,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()
@@ -2372,6 +2378,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/src/main/java/org/apache/paimon/fs/MetricsFileIO.java

@@ -0,0 +1,99 @@
+/*
+ * 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.catalog.CatalogContext;
+import org.apache.paimon.fs.metrics.InputMetrics;
+import org.apache.paimon.fs.metrics.OutputMetrics;
+
+import java.io.IOException;
+
+/**
+ * FileIOWrapper is a wrapper class for the {@link FileIO}.
+ *
+ * <p>It allows users to monitor and track file I/O operations (e.g., read, write, delete, rename).
+ */
+public class MetricsFileIO implements FileIO {
+
+    protected final FileIO fileIO;
+    protected InputMetrics inputMetrics = null;
+    protected OutputMetrics outputMetrics = null;
+
+    public MetricsFileIO(FileIO fileIO) {
+        this.fileIO = fileIO;
+    }
+
+    public MetricsFileIO withMetrics(InputMetrics inputMetrics, OutputMetrics outputMetrics) {
+        this.inputMetrics = inputMetrics;
+        this.outputMetrics = outputMetrics;
+        return this;
+    }
+
+    @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.inputMetrics);
+    }
+
+    @Override
+    public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws IOException {
+        PositionOutputStream outputStream = fileIO.newOutputStream(path, overwrite);
+        return new PositionOutputStreamIOWrapper(outputStream, outputMetrics);
+    }
+
+    @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.OutputMetrics;
+
+import java.io.IOException;
+
+/** Wrap a {@link PositionOutputStream}. */
+public class PositionOutputStreamIOWrapper extends PositionOutputStream {
+
+    protected final PositionOutputStream out;
+    private OutputMetrics metrics;
+
+    public PositionOutputStreamIOWrapper(PositionOutputStream out, OutputMetrics outputMetrics) {
+        this.out = out;
+        this.metrics = outputMetrics;
+    }
+
+    @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,76 @@
+/*
+ * 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.InputMetrics;
+
+import java.io.IOException;
+
+/** Wrap a {@link SeekableInputStream}. */
+public class SeekableInputStreamIOWrapper extends SeekableInputStream {
+
+    protected final SeekableInputStream in;
+    private InputMetrics metrics;
+
+    public SeekableInputStreamIOWrapper(SeekableInputStream in, InputMetrics 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();
+    }
+}

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

@@ -0,0 +1,60 @@
+/*
+ * 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.metrics;
+
+import org.apache.paimon.metrics.MetricGroup;
+import org.apache.paimon.metrics.MetricRegistry;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/** Collects and monitors input stream metrics. */
+public class InputMetrics {
+
+    public static final String GROUP_NAME = "source";
+    private final MetricGroup metricGroup;
+
+    public static final String READ_BYTES = "write.bytes";
+    public static final String READ_OPERATIONS = "write.operations";
+
+    private final AtomicLong readBytes = new AtomicLong(0);
+    private final AtomicLong readOperations = new AtomicLong(0);
+
+    public InputMetrics(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);
+    }
+
+    public void recordReadEvent(long bytes) {
+        readBytes.addAndGet(bytes);
+        readOperations.incrementAndGet();
+    }
+
+    public long getReadBytes() {
+        return readBytes.get();
+    }
+
+    public long getReadOperations() {
+        return readOperations.get();
+    }
+}