This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new d7e493341a [Blob] Add some metric when write blob from http. (#8464)
d7e493341a is described below
commit d7e493341a0fe6eff48162346733385c8a58ad32
Author: wangwj <[email protected]>
AuthorDate: Mon Jul 13 11:31:34 2026 +0800
[Blob] Add some metric when write blob from http. (#8464)
For tables with BLOB columns, writers may fetch external HTTP resources
through blob descriptors. After `blob-write-null-on-missing-file` and
`blob-write-null-on-fetch-failure` are available, users can choose to
write NULL instead of failing the write for missing files or non-404
fetch failures. However, it is still difficult to observe how often blob
fetches succeed, fail, or are converted to NULL during write.
---
docs/docs/maintenance/metrics.md | 86 +++++++++++-
.../paimon/data/BlobFetchMetricReporter.java | 46 ++++++
.../append/DedicatedFormatRollingFileWriter.java | 3 +-
.../paimon/append/MultipleBlobFileWriter.java | 5 +-
.../paimon/operation/BaseAppendFileStoreWrite.java | 21 +++
.../apache/paimon/operation/BlobFileContext.java | 12 ++
.../paimon/operation/metrics/BlobFetchMetrics.java | 146 +++++++++++++++++++
.../operation/metrics/BlobFetchMetricsTest.java | 97 +++++++++++++
.../apache/paimon/format/blob/BlobFileFormat.java | 13 +-
.../paimon/format/blob/BlobFormatWriter.java | 44 ++++++
.../paimon/format/blob/BlobFormatWriterTest.java | 154 ++++++++++++++++++++-
11 files changed, 621 insertions(+), 6 deletions(-)
diff --git a/docs/docs/maintenance/metrics.md b/docs/docs/maintenance/metrics.md
index ef764c13e1..1dffec4612 100644
--- a/docs/docs/maintenance/metrics.md
+++ b/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 outcomes handled during write,
including successful fetches, failures, and missing resources 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><host>.taskmanager.<tm_id>.<job_name>.<writer_operator_name>.<subtask_index></td>
<td>paimon.table.<table_name>.writeBuffer</td>
</tr>
+ <tr>
+ <td>Blob Fetch Metrics</td>
+
<td><host>.taskmanager.<tm_id>.<job_name>.<writer_operator_name>.<subtask_index></td>
+ <td>paimon.table.<table_name>.blobFetch</td>
+ </tr>
<tr>
<td>Compaction Metrics</td>
<td><host>.taskmanager.<tm_id>.<job_name>.<writer_operator_name>.<subtask_index></td>
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/BlobFetchMetricReporter.java
b/paimon-common/src/main/java/org/apache/paimon/data/BlobFetchMetricReporter.java
new file mode 100644
index 0000000000..0670c7098c
--- /dev/null
+++
b/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);
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java
index e9ef1e15ee..86df6b10b2 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java
@@ -200,7 +200,8 @@ public class DedicatedFormatRollingFileWriter
context.blobConsumer(),
context.blobInlineFields(),
context.writeNullOnMissingFile(),
- context.writeNullOnFetchFailure());
+ context.writeNullOnFetchFailure(),
+ context.blobFetchMetricReporter());
} else {
this.blobWriterFactory = null;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java
index b8d2d1e6c5..d952018ee1 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java
+++
b/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,7 +66,8 @@ public class MultipleBlobFileWriter implements Closeable {
@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()) {
@@ -73,6 +75,7 @@ public class MultipleBlobFileWriter implements Closeable {
blobFileFormat.setWriteConsumer(blobConsumer);
blobFileFormat.setWriteNullOnMissingFile(writeNullOnMissingFile);
blobFileFormat.setWriteNullOnFetchFailure(writeNullOnFetchFailure);
+ blobFileFormat.setBlobFetchMetricReporter(blobFetchMetricReporter);
blobWriters.add(
new BlobProjectedFileWriter(
() ->
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
index 0e4eeb9e4e..43e83333ee 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
@@ -35,6 +35,8 @@ import org.apache.paimon.io.BundleRecords;
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;
@@ -125,6 +128,16 @@ public abstract class BaseAppendFileStoreWrite extends
MemoryFileStoreWrite<Inte
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,
@@ -187,6 +200,14 @@ public abstract class BaseAppendFileStoreWrite extends
MemoryFileStoreWrite<Inte
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,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/BlobFileContext.java
b/paimon-core/src/main/java/org/apache/paimon/operation/BlobFileContext.java
index dd475da637..dc43824b6c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/BlobFileContext.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/BlobFileContext.java
@@ -20,6 +20,7 @@ package org.apache.paimon.operation;
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 class BlobFileContext {
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 class BlobFileContext {
public boolean writeNullOnFetchFailure() {
return writeNullOnFetchFailure;
}
+
+ public BlobFetchMetricReporter blobFetchMetricReporter() {
+ return blobFetchMetricReporter;
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/metrics/BlobFetchMetrics.java
b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/BlobFetchMetrics.java
new file mode 100644
index 0000000000..3f708a1388
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/BlobFetchMetrics.java
@@ -0,0 +1,146 @@
+/*
+ * 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.operation.metrics;
+
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.data.BlobFetchMetricReporter;
+import org.apache.paimon.metrics.Counter;
+import org.apache.paimon.metrics.MetricGroup;
+import org.apache.paimon.metrics.MetricRegistry;
+import org.apache.paimon.rest.HttpClientUtils;
+
+/** Metrics to measure blob fetches during write. */
+public class BlobFetchMetrics implements BlobFetchMetricReporter {
+
+ public static final String GROUP_NAME = "blobFetch";
+
+ public static final String BLOB_FETCH_TOTAL = "blobFetchTotal";
+ public static final String BLOB_FETCH_SUCCESS = "blobFetchSuccess";
+ public static final String BLOB_FETCH_SUCCESS_BYTES =
"blobFetchSuccessBytes";
+ public static final String BLOB_FETCH_NULL_WRITTEN =
"blobFetchNullWritten";
+ public static final String BLOB_FETCH_MISSING_FILE_NULL_WRITTEN =
+ "blobFetchMissingFileNullWritten";
+ public static final String BLOB_FETCH_FAILURE_NULL_WRITTEN =
"blobFetchFailureNullWritten";
+ public static final String BLOB_FETCH_FAILURE = "blobFetchFailure";
+ public static final String BLOB_FETCH_HTTP_NOT_FOUND =
"blobFetchHttpNotFound";
+ public static final String BLOB_FETCH_HTTP_CLIENT_ERROR =
"blobFetchHttpClientError";
+ public static final String BLOB_FETCH_HTTP_SERVER_ERROR =
"blobFetchHttpServerError";
+ public static final String BLOB_FETCH_HTTP_OTHER_ERROR =
"blobFetchHttpOtherError";
+ public static final String BLOB_FETCH_INVALID_URI = "blobFetchInvalidUri";
+ public static final String BLOB_FETCH_OTHER_ERROR = "blobFetchOtherError";
+
+ private final MetricGroup metricGroup;
+ private final Counter totalCounter;
+ private final Counter successCounter;
+ private final Counter successBytesCounter;
+ private final Counter nullWrittenCounter;
+ private final Counter missingFileNullWrittenCounter;
+ private final Counter fetchFailureNullWrittenCounter;
+ private final Counter failureCounter;
+ private final Counter httpNotFoundCounter;
+ private final Counter httpClientErrorCounter;
+ private final Counter httpServerErrorCounter;
+ private final Counter httpOtherErrorCounter;
+ private final Counter invalidUriCounter;
+ private final Counter otherErrorCounter;
+
+ public BlobFetchMetrics(MetricRegistry registry, String tableName) {
+ this.metricGroup = registry.createTableMetricGroup(GROUP_NAME,
tableName);
+ this.totalCounter = metricGroup.counter(BLOB_FETCH_TOTAL);
+ this.successCounter = metricGroup.counter(BLOB_FETCH_SUCCESS);
+ this.successBytesCounter =
metricGroup.counter(BLOB_FETCH_SUCCESS_BYTES);
+ this.nullWrittenCounter = metricGroup.counter(BLOB_FETCH_NULL_WRITTEN);
+ this.missingFileNullWrittenCounter =
+ metricGroup.counter(BLOB_FETCH_MISSING_FILE_NULL_WRITTEN);
+ this.fetchFailureNullWrittenCounter =
metricGroup.counter(BLOB_FETCH_FAILURE_NULL_WRITTEN);
+ this.failureCounter = metricGroup.counter(BLOB_FETCH_FAILURE);
+ this.httpNotFoundCounter =
metricGroup.counter(BLOB_FETCH_HTTP_NOT_FOUND);
+ this.httpClientErrorCounter =
metricGroup.counter(BLOB_FETCH_HTTP_CLIENT_ERROR);
+ this.httpServerErrorCounter =
metricGroup.counter(BLOB_FETCH_HTTP_SERVER_ERROR);
+ this.httpOtherErrorCounter =
metricGroup.counter(BLOB_FETCH_HTTP_OTHER_ERROR);
+ this.invalidUriCounter = metricGroup.counter(BLOB_FETCH_INVALID_URI);
+ this.otherErrorCounter = metricGroup.counter(BLOB_FETCH_OTHER_ERROR);
+ }
+
+ @VisibleForTesting
+ public MetricGroup getMetricGroup() {
+ return metricGroup;
+ }
+
+ @Override
+ public void recordSuccess(long bytes) {
+ totalCounter.inc();
+ successCounter.inc();
+ if (bytes > 0) {
+ successBytesCounter.inc(bytes);
+ }
+ }
+
+ @Override
+ public void recordMissingFileNullWritten(boolean httpNotFound) {
+ totalCounter.inc();
+ nullWrittenCounter.inc();
+ missingFileNullWrittenCounter.inc();
+ if (httpNotFound) {
+ httpNotFoundCounter.inc();
+ }
+ }
+
+ @Override
+ public void recordFetchFailureNullWritten(Throwable throwable) {
+ totalCounter.inc();
+ nullWrittenCounter.inc();
+ fetchFailureNullWrittenCounter.inc();
+ recordFailureReason(throwable);
+ }
+
+ @Override
+ public void recordFetchFailure(Throwable throwable) {
+ totalCounter.inc();
+ failureCounter.inc();
+ recordFailureReason(throwable);
+ }
+
+ public void close() {
+ metricGroup.close();
+ }
+
+ private void recordFailureReason(Throwable throwable) {
+ Integer statusCode = HttpClientUtils.getHttpStatusCode(throwable);
+ if (statusCode != null) {
+ if (statusCode == 404) {
+ httpNotFoundCounter.inc();
+ } else if (statusCode >= 400 && statusCode < 500) {
+ httpClientErrorCounter.inc();
+ } else if (statusCode >= 500 && statusCode < 600) {
+ httpServerErrorCounter.inc();
+ } else {
+ httpOtherErrorCounter.inc();
+ }
+ return;
+ }
+
+ if (HttpClientUtils.isInvalidUriException(throwable)) {
+ invalidUriCounter.inc();
+ return;
+ }
+
+ otherErrorCounter.inc();
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/metrics/BlobFetchMetricsTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/metrics/BlobFetchMetricsTest.java
new file mode 100644
index 0000000000..7fb53005fc
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/metrics/BlobFetchMetricsTest.java
@@ -0,0 +1,97 @@
+/*
+ * 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.operation.metrics;
+
+import org.apache.paimon.metrics.Counter;
+import org.apache.paimon.metrics.Metric;
+import org.apache.paimon.metrics.MetricGroup;
+import org.apache.paimon.metrics.TestMetricRegistry;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link BlobFetchMetrics}. */
+public class BlobFetchMetricsTest {
+
+ private static final String TABLE_NAME = "myTable";
+
+ @Test
+ public void testMetricRegistration() {
+ BlobFetchMetrics metrics = new BlobFetchMetrics(new
TestMetricRegistry(), TABLE_NAME);
+ MetricGroup metricGroup = metrics.getMetricGroup();
+
+
assertThat(metricGroup.getGroupName()).isEqualTo(BlobFetchMetrics.GROUP_NAME);
+ assertThat(metricGroup.getAllVariables()).containsEntry("table",
TABLE_NAME);
+ assertThat(metricGroup.getMetrics().keySet())
+ .containsExactlyInAnyOrder(
+ BlobFetchMetrics.BLOB_FETCH_TOTAL,
+ BlobFetchMetrics.BLOB_FETCH_SUCCESS,
+ BlobFetchMetrics.BLOB_FETCH_SUCCESS_BYTES,
+ BlobFetchMetrics.BLOB_FETCH_NULL_WRITTEN,
+ BlobFetchMetrics.BLOB_FETCH_MISSING_FILE_NULL_WRITTEN,
+ BlobFetchMetrics.BLOB_FETCH_FAILURE_NULL_WRITTEN,
+ BlobFetchMetrics.BLOB_FETCH_FAILURE,
+ BlobFetchMetrics.BLOB_FETCH_HTTP_NOT_FOUND,
+ BlobFetchMetrics.BLOB_FETCH_HTTP_CLIENT_ERROR,
+ BlobFetchMetrics.BLOB_FETCH_HTTP_SERVER_ERROR,
+ BlobFetchMetrics.BLOB_FETCH_HTTP_OTHER_ERROR,
+ BlobFetchMetrics.BLOB_FETCH_INVALID_URI,
+ BlobFetchMetrics.BLOB_FETCH_OTHER_ERROR);
+ }
+
+ @Test
+ public void testMetricsAreUpdated() {
+ BlobFetchMetrics metrics = new BlobFetchMetrics(new
TestMetricRegistry(), TABLE_NAME);
+
+ metrics.recordSuccess(10);
+ metrics.recordMissingFileNullWritten(true);
+ metrics.recordMissingFileNullWritten(false);
+ metrics.recordFetchFailureNullWritten(new RuntimeException("HTTP error
code: 500"));
+ metrics.recordFetchFailure(new RuntimeException("HTTP error code:
429"));
+ metrics.recordFetchFailure(
+ new IllegalArgumentException("Illegal character in path at
index 1"));
+ metrics.recordFetchFailure(new RuntimeException("boom"));
+
+ assertCounter(metrics, BlobFetchMetrics.BLOB_FETCH_TOTAL, 7);
+ assertCounter(metrics, BlobFetchMetrics.BLOB_FETCH_SUCCESS, 1);
+ assertCounter(metrics, BlobFetchMetrics.BLOB_FETCH_SUCCESS_BYTES, 10);
+ assertCounter(metrics, BlobFetchMetrics.BLOB_FETCH_NULL_WRITTEN, 3);
+ assertCounter(metrics,
BlobFetchMetrics.BLOB_FETCH_MISSING_FILE_NULL_WRITTEN, 2);
+ assertCounter(metrics,
BlobFetchMetrics.BLOB_FETCH_FAILURE_NULL_WRITTEN, 1);
+ assertCounter(metrics, BlobFetchMetrics.BLOB_FETCH_FAILURE, 3);
+ assertCounter(metrics, BlobFetchMetrics.BLOB_FETCH_HTTP_NOT_FOUND, 1);
+ assertCounter(metrics, BlobFetchMetrics.BLOB_FETCH_HTTP_CLIENT_ERROR,
1);
+ assertCounter(metrics, BlobFetchMetrics.BLOB_FETCH_HTTP_SERVER_ERROR,
1);
+ assertCounter(metrics, BlobFetchMetrics.BLOB_FETCH_HTTP_OTHER_ERROR,
0);
+ assertCounter(metrics, BlobFetchMetrics.BLOB_FETCH_INVALID_URI, 1);
+ assertCounter(metrics, BlobFetchMetrics.BLOB_FETCH_OTHER_ERROR, 1);
+ }
+
+ private static void assertCounter(BlobFetchMetrics metrics, String name,
long expected) {
+ assertThat(counter(metrics, name)).isEqualTo(expected);
+ }
+
+ private static long counter(BlobFetchMetrics metrics, String name) {
+ Map<String, Metric> registeredMetrics =
metrics.getMetricGroup().getMetrics();
+ return ((Counter) registeredMetrics.get(name)).getCount();
+ }
+}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java
b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java
index 66798c4bf3..afa8cdf232 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java
@@ -19,6 +19,7 @@
package org.apache.paimon.format.blob;
import org.apache.paimon.data.BlobConsumer;
+import org.apache.paimon.data.BlobFetchMetricReporter;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.format.EmptyStatsExtractor;
import org.apache.paimon.format.FileFormat;
@@ -52,6 +53,7 @@ public class BlobFileFormat extends FileFormat {
private final boolean blobAsDescriptor;
private boolean writeNullOnMissingFile;
private boolean writeNullOnFetchFailure;
+ private BlobFetchMetricReporter blobFetchMetricReporter =
BlobFetchMetricReporter.NOOP;
@Nullable public BlobConsumer writeConsumer;
@@ -80,6 +82,10 @@ public class BlobFileFormat extends FileFormat {
this.writeConsumer = writeConsumer;
}
+ public void setBlobFetchMetricReporter(BlobFetchMetricReporter
blobFetchMetricReporter) {
+ this.blobFetchMetricReporter = blobFetchMetricReporter;
+ }
+
@Override
public FormatReaderFactory createReaderFactory(
RowType dataSchemaRowType,
@@ -119,7 +125,12 @@ public class BlobFileFormat extends FileFormat {
@Override
public FormatWriter create(PositionOutputStream out, String
compression) {
return new BlobFormatWriter(
- out, writeConsumer, type, writeNullOnMissingFile,
writeNullOnFetchFailure);
+ out,
+ writeConsumer,
+ type,
+ writeNullOnMissingFile,
+ writeNullOnFetchFailure,
+ blobFetchMetricReporter);
}
}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java
b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java
index 5b3e50537d..07e98c283e 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java
@@ -21,6 +21,7 @@ package org.apache.paimon.format.blob;
import org.apache.paimon.data.Blob;
import org.apache.paimon.data.BlobConsumer;
import org.apache.paimon.data.BlobDescriptor;
+import org.apache.paimon.data.BlobFetchMetricReporter;
import org.apache.paimon.data.BlobPlaceholder;
import org.apache.paimon.data.BlobRef;
import org.apache.paimon.data.InternalRow;
@@ -59,6 +60,7 @@ public class BlobFormatWriter implements
FileAwareFormatWriter {
private final PositionOutputStream out;
@Nullable private final BlobConsumer writeConsumer;
+ private final BlobFetchMetricReporter blobFetchMetricReporter;
private final String blobFieldName;
private final boolean writeNullOnMissingFile;
private final boolean writeNullOnFetchFailure;
@@ -87,8 +89,25 @@ public class BlobFormatWriter implements
FileAwareFormatWriter {
RowType type,
boolean writeNullOnMissingFile,
boolean writeNullOnFetchFailure) {
+ this(
+ out,
+ writeConsumer,
+ type,
+ writeNullOnMissingFile,
+ writeNullOnFetchFailure,
+ BlobFetchMetricReporter.NOOP);
+ }
+
+ public BlobFormatWriter(
+ PositionOutputStream out,
+ @Nullable BlobConsumer writeConsumer,
+ RowType type,
+ boolean writeNullOnMissingFile,
+ boolean writeNullOnFetchFailure,
+ BlobFetchMetricReporter blobFetchMetricReporter) {
this.out = out;
this.writeConsumer = writeConsumer;
+ this.blobFetchMetricReporter = blobFetchMetricReporter;
this.writeNullOnMissingFile = writeNullOnMissingFile;
this.writeNullOnFetchFailure = writeNullOnFetchFailure;
checkArgument(type.getFieldCount() == 1, "BlobFormatWriter only
support one field.");
@@ -112,6 +131,7 @@ public class BlobFormatWriter implements
FileAwareFormatWriter {
public void addElement(InternalRow element) throws IOException {
checkArgument(element.getFieldCount() == 1, "BlobFormatWriter only
support one field.");
if (element.isNullAt(0)) {
+ recordPreCheckedMissingFileNull(element);
writeNullElement();
return;
}
@@ -122,6 +142,7 @@ public class BlobFormatWriter implements
FileAwareFormatWriter {
if (tryWriteNullOnFetchFailure(e, null)) {
return;
}
+ blobFetchMetricReporter.recordFetchFailure(e);
throw e;
}
if (blob == BlobPlaceholder.INSTANCE) {
@@ -139,12 +160,14 @@ public class BlobFormatWriter implements
FileAwareFormatWriter {
blobUri(blob),
blobFieldName,
e);
+ blobFetchMetricReporter.recordMissingFileNullWritten(true);
writeNullElement();
return;
}
if (tryWriteNullOnFetchFailure(e, blob)) {
return;
}
+ blobFetchMetricReporter.recordFetchFailure(e);
throw e;
}
@@ -159,6 +182,9 @@ public class BlobFormatWriter implements
FileAwareFormatWriter {
blobLength += bytesRead;
bytesRead = stream.read(tmpBuffer);
}
+ } catch (IOException | RuntimeException e) {
+ blobFetchMetricReporter.recordFetchFailure(e);
+ throw e;
}
long binLength = blobLength + MAGIC_NUMBER_BYTES.length + 12;
@@ -175,6 +201,7 @@ public class BlobFormatWriter implements
FileAwareFormatWriter {
out.flush();
}
}
+ blobFetchMetricReporter.recordSuccess(blobLength);
}
private boolean tryWriteNullOnFetchFailure(Throwable e, @Nullable Blob
blob)
@@ -203,10 +230,22 @@ public class BlobFormatWriter implements
FileAwareFormatWriter {
blobFieldName,
e);
}
+ blobFetchMetricReporter.recordFetchFailureNullWritten(e);
writeNullElement();
return true;
}
+ private void recordPreCheckedMissingFileNull(InternalRow element) {
+ if (!writeNullOnMissingFile) {
+ return;
+ }
+ Blob blob = element.getBlob(0);
+ if (blob instanceof BlobRef) {
+ BlobDescriptor descriptor = ((BlobRef) blob).toDescriptor();
+
blobFetchMetricReporter.recordMissingFileNullWritten(isHttpUri(descriptor.uri()));
+ }
+ }
+
private void writeNullElement() throws IOException {
lengths.add(NULL_LENGTH);
if (writeConsumer != null) {
@@ -221,6 +260,11 @@ public class BlobFormatWriter implements
FileAwareFormatWriter {
return "unknown";
}
+ private static boolean isHttpUri(String uri) {
+ return uri.regionMatches(true, 0, "http://";, 0, "http://".length())
+ || uri.regionMatches(true, 0, "https://";, 0,
"https://".length());
+ }
+
private void write(byte[] bytes) throws IOException {
write(bytes, bytes.length);
}
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java
index 90ded82a19..3c363dbae8 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java
@@ -21,6 +21,7 @@ package org.apache.paimon.format.blob;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.data.Blob;
import org.apache.paimon.data.BlobDescriptor;
+import org.apache.paimon.data.BlobFetchMetricReporter;
import org.apache.paimon.data.BlobRef;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
@@ -241,6 +242,114 @@ public class BlobFormatWriterTest {
}
}
+ @Test
+ public void testBlobFetchMetricReporterForSuccessAndNullWritten(
+ @TempDir java.nio.file.Path tempDir) throws Exception {
+ RowType rowType = RowType.of(DataTypes.BLOB());
+ TestingBlobFetchMetricReporter metricReporter = new
TestingBlobFetchMetricReporter();
+ BlobFormatWriter writer =
+ new BlobFormatWriter(
+ new LocalFileIO.LocalPositionOutputStream(
+ tempDir.resolve("blob.out").toFile()),
+ null,
+ rowType,
+ false,
+ true,
+ metricReporter);
+
+ writer.addElement(GenericRow.of(Blob.fromData("image".getBytes())));
+ writer.addElement(
+ GenericRow.of(
+ new BlobRef(
+ failingHttpReader(500),
+ new
BlobDescriptor("https://example.com/error.jpg";, 0, -1))));
+ writer.close();
+
+ assertThat(metricReporter.success).isEqualTo(1);
+ assertThat(metricReporter.successBytes).isEqualTo(5);
+ assertThat(metricReporter.fetchFailureNullWritten).isEqualTo(1);
+ assertThat(metricReporter.failure).isEqualTo(0);
+ }
+
+ @Test
+ public void testBlobFetchMetricReporterForUnhandledFailure(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ RowType rowType = RowType.of(DataTypes.BLOB());
+ TestingBlobFetchMetricReporter metricReporter = new
TestingBlobFetchMetricReporter();
+ BlobFormatWriter writer =
+ new BlobFormatWriter(
+ new LocalFileIO.LocalPositionOutputStream(
+ tempDir.resolve("blob.out").toFile()),
+ null,
+ rowType,
+ false,
+ false,
+ metricReporter);
+
+ assertThatThrownBy(
+ () ->
+ writer.addElement(
+ GenericRow.of(
+ new BlobRef(
+ failingHttpReader(500),
+ new BlobDescriptor(
+
"https://example.com/error.jpg";,
+ 0,
+ -1)))))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("HTTP error code: 500");
+ assertThat(metricReporter.failure).isEqualTo(1);
+ assertThat(metricReporter.fetchFailureNullWritten).isEqualTo(0);
+ }
+
+ @Test
+ public void testBlobFetchMetricReporterForPreCheckedMissingFile(
+ @TempDir java.nio.file.Path tempDir) throws Exception {
+ RowType rowType = RowType.of(DataTypes.BLOB());
+ UriReaderFactory uriReaderFactory =
+ new UriReaderFactory(CatalogContext.create(new Options()));
+ byte[] descriptorBytes =
+ new BlobDescriptor("https://example.com/missing.jpg";, 0,
-1).serialize();
+ TestingBlobFetchMetricReporter metricReporter = new
TestingBlobFetchMetricReporter();
+ BlobFormatWriter writer =
+ new BlobFormatWriter(
+ new LocalFileIO.LocalPositionOutputStream(
+ tempDir.resolve("blob.out").toFile()),
+ null,
+ rowType,
+ true,
+ false,
+ metricReporter);
+
+ writer.addElement(new DescriptorBytesRow(descriptorBytes,
uriReaderFactory, true));
+ writer.close();
+
+ assertThat(metricReporter.missingFileNullWritten).isEqualTo(1);
+ assertThat(metricReporter.httpNotFound).isEqualTo(1);
+ }
+
+ @Test
+ public void testBlobFetchMetricReporterIgnoresUserNull(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ RowType rowType = RowType.of(DataTypes.BLOB());
+ TestingBlobFetchMetricReporter metricReporter = new
TestingBlobFetchMetricReporter();
+ BlobFormatWriter writer =
+ new BlobFormatWriter(
+ new LocalFileIO.LocalPositionOutputStream(
+ tempDir.resolve("blob.out").toFile()),
+ null,
+ rowType,
+ true,
+ false,
+ metricReporter);
+
+ writer.addElement(GenericRow.of((Object) null));
+ writer.close();
+
+ assertThat(metricReporter.missingFileNullWritten).isEqualTo(0);
+ assertThat(metricReporter.httpNotFound).isEqualTo(0);
+ }
+
private static void assertBlobPayload(Blob blob, byte[] expected) throws
Exception {
try (SeekableInputStream blobIn = blob.newInputStream()) {
byte[] actual = new byte[expected.length];
@@ -261,10 +370,17 @@ public class BlobFormatWriterTest {
private static final class DescriptorBytesRow implements InternalRow {
private final byte[] descriptorBytes;
private final UriReaderFactory uriReaderFactory;
+ private final boolean nullAt;
private DescriptorBytesRow(byte[] descriptorBytes, UriReaderFactory
uriReaderFactory) {
+ this(descriptorBytes, uriReaderFactory, false);
+ }
+
+ private DescriptorBytesRow(
+ byte[] descriptorBytes, UriReaderFactory uriReaderFactory,
boolean nullAt) {
this.descriptorBytes = descriptorBytes;
this.uriReaderFactory = uriReaderFactory;
+ this.nullAt = nullAt;
}
@Override
@@ -282,7 +398,7 @@ public class BlobFormatWriterTest {
@Override
public boolean isNullAt(int pos) {
- return false;
+ return nullAt;
}
@Override
@@ -346,7 +462,7 @@ public class BlobFormatWriterTest {
@Override
public byte[] getBinary(int pos) {
- throw unsupported();
+ return descriptorBytes;
}
@Override
@@ -374,4 +490,38 @@ public class BlobFormatWriterTest {
throw unsupported();
}
}
+
+ private static final class TestingBlobFetchMetricReporter implements
BlobFetchMetricReporter {
+
+ private int success;
+ private long successBytes;
+ private int missingFileNullWritten;
+ private int httpNotFound;
+ private int fetchFailureNullWritten;
+ private int failure;
+
+ @Override
+ public void recordSuccess(long bytes) {
+ success++;
+ successBytes += bytes;
+ }
+
+ @Override
+ public void recordMissingFileNullWritten(boolean httpNotFound) {
+ missingFileNullWritten++;
+ if (httpNotFound) {
+ this.httpNotFound++;
+ }
+ }
+
+ @Override
+ public void recordFetchFailureNullWritten(Throwable throwable) {
+ fetchFailureNullWritten++;
+ }
+
+ @Override
+ public void recordFetchFailure(Throwable throwable) {
+ failure++;
+ }
+ }
}
![https://example.com/error.jpg"](https://example.com/error.jpg"){kind=link}
![https://example.com/missing.jpg"](https://example.com/missing.jpg"){kind=link}