[Blob] Add some metric when write blob from http.

docs/docs/maintenance/metrics.md

@@ -37,7 +37,7 @@ Paimon has supported built-in metrics to measure operations of **commits**, **sc
 
 ## Metrics List
 
-Below is lists of Paimon built-in metrics. They are summarized into types of scan metrics, commit metrics, write metrics, write buffer metrics and compaction metrics.
+Below is lists of Paimon built-in metrics. They are summarized into types of scan metrics, commit metrics, write metrics, write buffer metrics, blob fetch metrics and compaction metrics.
 
 ### Scan Metrics
 
@@ -226,6 +226,85 @@ Below is lists of Paimon built-in metrics. They are summarized into types of sca
     </tbody>
 </table>
 
+### Blob Fetch Metrics
+
+<table class="table table-bordered">
+    <thead>
+    <tr>
+      <th class="text-left" style="width: 225pt">Metrics Name</th>
+      <th class="text-left" style="width: 70pt">Type</th>
+      <th class="text-left" style="width: 300pt">Description</th>
+    </tr>
+    </thead>
+    <tbody>
+        <tr>
+            <td>blobFetchTotal</td>
+            <td>Counter</td>
+            <td>Total number of blob fetch attempts during write, including successful fetches, failures, and failures written as NULL.</td>
+        </tr>
+        <tr>
+            <td>blobFetchSuccess</td>
+            <td>Counter</td>
+            <td>Total number of blob fetches that completed successfully.</td>
+        </tr>
+        <tr>
+            <td>blobFetchSuccessBytes</td>
+            <td>Counter</td>
+            <td>Total bytes written by successful blob fetches.</td>
+        </tr>
+        <tr>
+            <td>blobFetchNullWritten</td>
+            <td>Counter</td>
+            <td>Total number of NULL values written because a blob resource was missing or failed to fetch.</td>
+        </tr>
+        <tr>
+            <td>blobFetchMissingFileNullWritten</td>
+            <td>Counter</td>
+            <td>Total number of NULL values written because <code>blob-write-null-on-missing-file</code> handled a missing blob resource.</td>
+        </tr>
+        <tr>
+            <td>blobFetchFailureNullWritten</td>
+            <td>Counter</td>
+            <td>Total number of NULL values written because <code>blob-write-null-on-fetch-failure</code> handled a non-404 blob fetch failure.</td>
+        </tr>
+        <tr>
+            <td>blobFetchFailure</td>
+            <td>Counter</td>
+            <td>Total number of blob fetch failures that were not written as NULL.</td>
+        </tr>
+        <tr>
+            <td>blobFetchHttpNotFound</td>
+            <td>Counter</td>
+            <td>Total number of HTTP 404 blob fetch failures.</td>
+        </tr>
+        <tr>
+            <td>blobFetchHttpClientError</td>
+            <td>Counter</td>
+            <td>Total number of non-404 HTTP 4xx blob fetch failures.</td>
+        </tr>
+        <tr>
+            <td>blobFetchHttpServerError</td>
+            <td>Counter</td>
+            <td>Total number of HTTP 5xx blob fetch failures.</td>
+        </tr>
+        <tr>
+            <td>blobFetchHttpOtherError</td>
+            <td>Counter</td>
+            <td>Total number of blob fetch failures with HTTP status codes outside 4xx and 5xx.</td>
+        </tr>
+        <tr>
+            <td>blobFetchInvalidUri</td>
+            <td>Counter</td>
+            <td>Total number of blob fetch failures caused by invalid URIs.</td>
+        </tr>
+        <tr>
+            <td>blobFetchOtherError</td>
+            <td>Counter</td>
+            <td>Total number of blob fetch failures that are not classified as HTTP status or invalid URI failures.</td>
+        </tr>
+    </tbody>
+</table>
+
 ### Compaction Metrics
 
 <table class="table table-bordered">
@@ -375,6 +454,11 @@ From Flink Web-UI, go to the committer operator's metrics, it's shown as:
             <td>&lt;host&gt;.taskmanager.&lt;tm_id&gt;.&lt;job_name&gt;.&lt;writer_operator_name&gt;.&lt;subtask_index&gt;</td>
             <td>paimon.table.&lt;table_name&gt;.writeBuffer</td>
         </tr>
+        <tr>
+            <td>Blob Fetch Metrics</td>
+            <td>&lt;host&gt;.taskmanager.&lt;tm_id&gt;.&lt;job_name&gt;.&lt;writer_operator_name&gt;.&lt;subtask_index&gt;</td>
+            <td>paimon.table.&lt;table_name&gt;.blobFetch</td>
+        </tr>
         <tr>
             <td>Compaction Metrics</td>
             <td>&lt;host&gt;.taskmanager.&lt;tm_id&gt;.&lt;job_name&gt;.&lt;writer_operator_name&gt;.&lt;subtask_index&gt;</td>

paimon-common/src/main/java/org/apache/paimon/data/BlobFetchMetricReporter.java

@@ -0,0 +1,46 @@
+/*
+ * 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.data;
+
+/** Reporter for blob fetch metrics. */
+public interface BlobFetchMetricReporter {
+
+    BlobFetchMetricReporter NOOP =
+            new BlobFetchMetricReporter() {
+                @Override
+                public void recordSuccess(long bytes) {}
+
+                @Override
+                public void recordMissingFileNullWritten(boolean httpNotFound) {}
+
+                @Override
+                public void recordFetchFailureNullWritten(Throwable throwable) {}
+
+                @Override
+                public void recordFetchFailure(Throwable throwable) {}
+            };
+
+    void recordSuccess(long bytes);
+
+    void recordMissingFileNullWritten(boolean httpNotFound);
+
+    void recordFetchFailureNullWritten(Throwable throwable);
+
+    void recordFetchFailure(Throwable throwable);
+}

paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java

@@ -200,7 +200,8 @@ public DedicatedFormatRollingFileWriter(
                                     context.blobConsumer(),
                                     context.blobInlineFields(),
                                     context.writeNullOnMissingFile(),
-                                    context.writeNullOnFetchFailure());
+                                    context.writeNullOnFetchFailure(),
+                                    context.blobFetchMetricReporter());
         } else {
             this.blobWriterFactory = null;
         }

paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.append;
 
 import org.apache.paimon.data.BlobConsumer;
+import org.apache.paimon.data.BlobFetchMetricReporter;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.fileindex.FileIndexOptions;
 import org.apache.paimon.format.blob.BlobFileFormat;
@@ -65,14 +66,16 @@ public MultipleBlobFileWriter(
             @Nullable BlobConsumer blobConsumer,
             Set<String> blobInlineFields,
             boolean writeNullOnMissingFile,
-            boolean writeNullOnFetchFailure) {
+            boolean writeNullOnFetchFailure,
+            BlobFetchMetricReporter blobFetchMetricReporter) {
         RowType blobRowType = new RowType(fieldsInBlobFile(writeSchema, blobInlineFields));
         this.blobWriters = new ArrayList<>();
         for (String blobFieldName : blobRowType.getFieldNames()) {
             BlobFileFormat blobFileFormat = new BlobFileFormat();
             blobFileFormat.setWriteConsumer(blobConsumer);
             blobFileFormat.setWriteNullOnMissingFile(writeNullOnMissingFile);
             blobFileFormat.setWriteNullOnFetchFailure(writeNullOnFetchFailure);
+            blobFileFormat.setBlobFetchMetricReporter(blobFetchMetricReporter);
             blobWriters.add(
                     new BlobProjectedFileWriter(
                             () ->

paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java

@@ -35,6 +35,8 @@
 import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.io.RowDataRollingFileWriter;
 import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.metrics.MetricRegistry;
+import org.apache.paimon.operation.metrics.BlobFetchMetrics;
 import org.apache.paimon.reader.RecordReaderIterator;
 import org.apache.paimon.statistics.SimpleColStatsCollector;
 import org.apache.paimon.types.RowType;
@@ -80,6 +82,7 @@ public abstract class BaseAppendFileStoreWrite extends MemoryFileStoreWrite<Inte
     private final RowType rowType;
 
     private @Nullable BlobFileContext blobContext;
+    private @Nullable BlobFetchMetrics blobFetchMetrics;
     private RowType writeType;
     private @Nullable List<String> writeCols;
     private boolean forceBufferSpill = false;
@@ -117,6 +120,16 @@ public BaseAppendFileStoreWrite withBlobConsumer(BlobConsumer blobConsumer) {
         return this;
     }
 
+    @Override
+    public BaseAppendFileStoreWrite withMetricRegistry(MetricRegistry metricRegistry) {
+        super.withMetricRegistry(metricRegistry);
+        if (blobContext != null) {
+            blobFetchMetrics = new BlobFetchMetrics(metricRegistry, tableName);
+            blobContext = blobContext.withBlobFetchMetricReporter(blobFetchMetrics);
+        }
+        return this;
+    }
+
     @Override
     protected RecordWriter<InternalRow> createWriter(
             BinaryRow partition,
@@ -179,6 +192,14 @@ private SimpleColStatsCollector.Factory[] statsCollectors() {
         return createStatsFactories(options.statsMode(), options, writeType.getFieldNames());
     }
 
+    @Override
+    public void close() throws Exception {
+        super.close();
+        if (blobFetchMetrics != null) {
+            blobFetchMetrics.close();
+        }
+    }
+
     protected abstract CompactManager getCompactManager(
             BinaryRow partition,
             int bucket,

paimon-core/src/main/java/org/apache/paimon/operation/BlobFileContext.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.data.BlobConsumer;
+import org.apache.paimon.data.BlobFetchMetricReporter;
 import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.DataTypeRoot;
 import org.apache.paimon.types.RowType;
@@ -39,6 +40,7 @@ public class BlobFileContext {
     private final boolean writeNullOnFetchFailure;
 
     private @Nullable BlobConsumer blobConsumer;
+    private BlobFetchMetricReporter blobFetchMetricReporter = BlobFetchMetricReporter.NOOP;
 
     private BlobFileContext(
             Set<String> blobDescriptorFields,
@@ -81,6 +83,12 @@ public BlobFileContext withBlobConsumer(BlobConsumer blobConsumer) {
         return this;
     }
 
+    public BlobFileContext withBlobFetchMetricReporter(
+            BlobFetchMetricReporter blobFetchMetricReporter) {
+        this.blobFetchMetricReporter = blobFetchMetricReporter;
+        return this;
+    }
+
     public BlobFileContext withWriteType(RowType writeType) {
         if (writeType.getFieldTypes().stream().noneMatch(t -> t.is(BLOB))) {
             return null;
@@ -108,4 +116,8 @@ public boolean writeNullOnMissingFile() {
     public boolean writeNullOnFetchFailure() {
         return writeNullOnFetchFailure;
     }
+
+    public BlobFetchMetricReporter blobFetchMetricReporter() {
+        return blobFetchMetricReporter;
+    }
 }