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 9ea8120590 [core][flink] Add blob-write-null-on-fetch-failure for BLOB
descriptor writes. (#8412)
9ea8120590 is described below
commit 9ea8120590944a7a31af0a046ad7ec0db0ae5021
Author: wangwj <[email protected]>
AuthorDate: Sat Jul 4 22:29:58 2026 +0800
[core][flink] Add blob-write-null-on-fetch-failure for BLOB descriptor
writes. (#8412)
Add `blob-write-null-on-fetch-failure` for Flink BLOB descriptor writes,
mirroring the scope of `blob-write-null-on-missing-file`.
---
docs/generated/core_configuration.html | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 15 +
.../org/apache/paimon/rest/HttpClientUtils.java | 73 +++-
.../apache/paimon/rest/HttpClientUtilsTest.java | 93 +++++
.../append/DedicatedFormatRollingFileWriter.java | 4 +-
.../paimon/append/MultipleBlobFileWriter.java | 6 +-
.../apache/paimon/operation/BlobFileContext.java | 24 +-
.../org/apache/paimon/flink/FlinkRowWrapper.java | 64 +++-
.../apache/paimon/flink/sink/FlinkSinkBuilder.java | 15 +-
.../org/apache/paimon/flink/BlobTableITCase.java | 143 ++++++++
.../apache/paimon/flink/FlinkRowWrapperTest.java | 79 ++++-
.../apache/paimon/format/blob/BlobFileFormat.java | 13 +-
.../paimon/format/blob/BlobFormatWriter.java | 120 ++++++-
.../paimon/format/blob/BlobFormatWriterTest.java | 377 +++++++++++++++++++++
14 files changed, 1002 insertions(+), 30 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index e31bd111b1..2bb0099a49 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -92,6 +92,12 @@ under the License.
<td>Boolean</td>
<td>Whether to resolve blob-view-field values from upstream tables
at read time. Set to false to preserve BlobViewStruct references when
forwarding blob view values to another blob-view table.</td>
</tr>
+ <tr>
+ <td><h5>blob-write-null-on-fetch-failure</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to write NULL for a descriptor BLOB value when the
referenced resource cannot be fetched during Flink writes (e.g. invalid URI or
HTTP errors other than 404). HTTP 404 is handled by
'blob-write-null-on-missing-file'. When false, the write fails when the
descriptor is read.</td>
+ </tr>
<tr>
<td><h5>blob-write-null-on-missing-file</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 23721c1f44..0302d10995 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2556,6 +2556,17 @@ public class CoreOptions implements Serializable {
+ "writes. When false, the write fails
when the descriptor is "
+ "read.");
+ public static final ConfigOption<Boolean> BLOB_WRITE_NULL_ON_FETCH_FAILURE
=
+ key("blob-write-null-on-fetch-failure")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to write NULL for a descriptor BLOB value
when the "
+ + "referenced resource cannot be fetched
during Flink writes "
+ + "(e.g. invalid URI or HTTP errors other
than 404). "
+ + "HTTP 404 is handled by
'blob-write-null-on-missing-file'. "
+ + "When false, the write fails when the
descriptor is read.");
+
public static final ConfigOption<Boolean> COMMIT_DISCARD_DUPLICATE_FILES =
key("commit.discard-duplicate-files")
.booleanType()
@@ -4133,6 +4144,10 @@ public class CoreOptions implements Serializable {
return options.get(BLOB_WRITE_NULL_ON_MISSING_FILE);
}
+ public boolean blobWriteNullOnFetchFailure() {
+ return options.get(BLOB_WRITE_NULL_ON_FETCH_FAILURE);
+ }
+
public boolean postponeBatchWriteFixedBucket() {
return options.get(POSTPONE_BATCH_WRITE_FIXED_BUCKET);
}
diff --git
a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java
b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java
index f5a4a1ec0d..f54ef7044d 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java
@@ -88,8 +88,13 @@ public class HttpClientUtils {
public static InputStream getAsInputStream(String uri) throws IOException {
HttpGet httpGet = new HttpGet(uri);
CloseableHttpResponse response = DEFAULT_HTTP_CLIENT.execute(httpGet);
- if (response.getCode() != 200) {
- throw new RuntimeException("HTTP error code: " +
response.getCode());
+ int statusCode = response.getCode();
+ if (statusCode != HttpStatus.SC_OK) {
+ try {
+ throw httpError(statusCode);
+ } finally {
+ response.close();
+ }
}
return response.getEntity().getContent();
}
@@ -98,7 +103,7 @@ public class HttpClientUtils {
* Checks whether an HTTP resource exists. HEAD is attempted first; when
HEAD does not return
* 200, a lightweight GET with {@code Range: bytes=0-0} is used to verify
readability. This
* avoids treating signed or GET-only URLs as missing when HEAD is
rejected or returns a
- * different status than GET. HTTP 416 from the range probe indicates a
zero-length resource.
+ * different status than GET.
*/
public static boolean exists(String uri) throws IOException {
int headStatusCode = headStatusCode(uri);
@@ -118,6 +123,64 @@ public class HttpClientUtils {
"Unexpected HTTP status code: " + rangeStatusCode + " for uri:
" + uri);
}
+ public static boolean isNotFoundError(Throwable throwable) {
+ Integer statusCode = getHttpStatusCode(throwable);
+ return statusCode != null && statusCode == HttpStatus.SC_NOT_FOUND;
+ }
+
+ public static boolean isInvalidUriException(Throwable throwable) {
+ Throwable current = throwable;
+ while (current != null) {
+ if (current instanceof java.net.URISyntaxException) {
+ return true;
+ }
+ if (current instanceof IllegalArgumentException
+ && current.getMessage() != null
+ && current.getMessage().contains("Illegal character")) {
+ return true;
+ }
+ current = current.getCause();
+ }
+ return false;
+ }
+
+ public static Integer getHttpStatusCode(Throwable throwable) {
+ Throwable current = throwable;
+ while (current != null) {
+ if (current.getMessage() != null) {
+ Integer statusCode = parseHttpStatusCode(current.getMessage());
+ if (statusCode != null) {
+ return statusCode;
+ }
+ }
+ current = current.getCause();
+ }
+ return null;
+ }
+
+ private static Integer parseHttpStatusCode(String message) {
+ if (message.startsWith("HTTP error code: ")) {
+ return parseStatusCodeSuffix(message.substring("HTTP error code:
".length()));
+ }
+ if (message.startsWith("Unexpected HTTP status code: ")) {
+ int end = message.indexOf(' ', "Unexpected HTTP status code:
".length());
+ String statusText =
+ end < 0
+ ? message.substring("Unexpected HTTP status code:
".length())
+ : message.substring("Unexpected HTTP status code:
".length(), end);
+ return parseStatusCodeSuffix(statusText);
+ }
+ return null;
+ }
+
+ private static Integer parseStatusCodeSuffix(String statusText) {
+ try {
+ return Integer.parseInt(statusText.trim());
+ } catch (NumberFormatException e) {
+ return null;
+ }
+ }
+
private static int headStatusCode(String uri) throws IOException {
HttpHead httpHead = new HttpHead(uri);
try (CloseableHttpResponse response =
DEFAULT_HTTP_CLIENT.execute(httpHead)) {
@@ -132,4 +195,8 @@ public class HttpClientUtils {
return response.getCode();
}
}
+
+ private static RuntimeException httpError(int statusCode) {
+ return new RuntimeException("HTTP error code: " + statusCode);
+ }
}
diff --git
a/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java
b/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java
index 064ebe232f..d92cb2b501 100644
--- a/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java
+++ b/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java
@@ -26,10 +26,12 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link HttpClientUtils}. */
public class HttpClientUtilsTest {
@@ -163,6 +165,97 @@ public class HttpClientUtilsTest {
assertThat(HttpClientUtils.exists(url("/head-404-get-404"))).isFalse();
}
+ @Test
+ public void testGetAsInputStreamThrowsForNotFound() {
+ registerHandler(
+ "/get-missing",
+ exchange -> {
+ respond(exchange, 404, new byte[0]);
+ });
+
+ assertThatThrownBy(() ->
HttpClientUtils.getAsInputStream(url("/get-missing")))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("HTTP error code: 404");
+ }
+
+ @Test
+ public void testGetAsInputStreamDoesNotLeakConnectionsOnRepeatedNotFound()
throws Exception {
+ registerHandler(
+ "/missing",
+ exchange -> {
+ respond(exchange, 404, new byte[0]);
+ });
+ registerHandler(
+ "/ok",
+ exchange -> {
+ respond(exchange, 200, "x".getBytes());
+ });
+
+ for (int i = 0; i < 120; i++) {
+ assertThatThrownBy(() ->
HttpClientUtils.getAsInputStream(url("/missing")))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("HTTP error code: 404");
+ }
+
+ try (InputStream in = HttpClientUtils.getAsInputStream(url("/ok"))) {
+ assertThat(in.read()).isEqualTo('x');
+ }
+ }
+
+ @Test
+ public void testIsNotFoundError() {
+ RuntimeException exception =
+ new RuntimeException("wrapper", new RuntimeException("HTTP
error code: 404"));
+ assertThat(HttpClientUtils.isNotFoundError(exception)).isTrue();
+ assertThat(HttpClientUtils.isNotFoundError(new RuntimeException("HTTP
error code: 500")))
+ .isFalse();
+ }
+
+ @Test
+ public void testGetHttpStatusCodeFromUnexpectedStatusIOException() {
+ IOException exception =
+ new IOException("Unexpected HTTP status code: 400 for uri:
http://127.0.0.1/test";);
+
assertThat(HttpClientUtils.getHttpStatusCode(exception)).isEqualTo(400);
+ }
+
+ @Test
+ public void testIsInvalidUriException() {
+ assertThat(
+ HttpClientUtils.isInvalidUriException(
+ new IllegalArgumentException("Illegal
character in path")))
+ .isTrue();
+ assertThat(
+ HttpClientUtils.isInvalidUriException(
+ new RuntimeException("HTTP error code: 404")))
+ .isFalse();
+ }
+
+ @Test
+ public void testExistsThrowsForBadRequest() {
+ registerHandler(
+ "/bad-request",
+ exchange -> {
+ respond(exchange, 400, new byte[0]);
+ });
+
+ assertThatThrownBy(() -> HttpClientUtils.exists(url("/bad-request")))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Unexpected HTTP status code: 400");
+ }
+
+ @Test
+ public void testExistsThrowsForRateLimitOnHead() {
+ registerHandler(
+ "/rate-limit",
+ exchange -> {
+ respond(exchange, 420, new byte[0]);
+ });
+
+ assertThatThrownBy(() -> HttpClientUtils.exists(url("/rate-limit")))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Unexpected HTTP status code: 420");
+ }
+
private void registerHandler(String path, HttpHandler handler) {
server.createContext(path, handler);
}
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 870cb2dd35..e9ef1e15ee 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
@@ -198,7 +198,9 @@ public class DedicatedFormatRollingFileWriter
statsDenseStore,
blobTargetFileSize,
context.blobConsumer(),
- context.blobInlineFields());
+ context.blobInlineFields(),
+ context.writeNullOnMissingFile(),
+ context.writeNullOnFetchFailure());
} 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 045af6ff75..b8d2d1e6c5 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
@@ -63,12 +63,16 @@ public class MultipleBlobFileWriter implements Closeable {
boolean statsDenseStore,
long targetFileSize,
@Nullable BlobConsumer blobConsumer,
- Set<String> blobInlineFields) {
+ Set<String> blobInlineFields,
+ boolean writeNullOnMissingFile,
+ boolean writeNullOnFetchFailure) {
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);
blobWriters.add(
new BlobProjectedFileWriter(
() ->
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 8e98ef3e8a..dd475da637 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
@@ -35,12 +35,20 @@ public class BlobFileContext {
private final Set<String> blobDescriptorFields;
private final Set<String> blobInlineFields;
+ private final boolean writeNullOnMissingFile;
+ private final boolean writeNullOnFetchFailure;
private @Nullable BlobConsumer blobConsumer;
- private BlobFileContext(Set<String> blobDescriptorFields, Set<String>
blobInlineFields) {
+ private BlobFileContext(
+ Set<String> blobDescriptorFields,
+ Set<String> blobInlineFields,
+ boolean writeNullOnMissingFile,
+ boolean writeNullOnFetchFailure) {
this.blobDescriptorFields = blobDescriptorFields;
this.blobInlineFields = blobInlineFields;
+ this.writeNullOnMissingFile = writeNullOnMissingFile;
+ this.writeNullOnFetchFailure = writeNullOnFetchFailure;
}
@Nullable
@@ -61,7 +69,11 @@ public class BlobFileContext {
if (!requireBlobFile) {
return null;
}
- return new BlobFileContext(descriptorFields, inlineFields);
+ return new BlobFileContext(
+ descriptorFields,
+ inlineFields,
+ options.blobWriteNullOnMissingFile(),
+ options.blobWriteNullOnFetchFailure());
}
public BlobFileContext withBlobConsumer(BlobConsumer blobConsumer) {
@@ -88,4 +100,12 @@ public class BlobFileContext {
public BlobConsumer blobConsumer() {
return blobConsumer;
}
+
+ public boolean writeNullOnMissingFile() {
+ return writeNullOnMissingFile;
+ }
+
+ public boolean writeNullOnFetchFailure() {
+ return writeNullOnFetchFailure;
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowWrapper.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowWrapper.java
index cd9d56ee65..3838d4f4ee 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowWrapper.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowWrapper.java
@@ -30,6 +30,7 @@ import org.apache.paimon.data.InternalVector;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.data.variant.GenericVariant;
import org.apache.paimon.data.variant.Variant;
+import org.apache.paimon.rest.HttpClientUtils;
import org.apache.paimon.types.DataTypeRoot;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.utils.UriReaderFactory;
@@ -57,6 +58,7 @@ public class FlinkRowWrapper implements InternalRow {
private final org.apache.flink.table.data.RowData row;
private final UriReaderFactory uriReaderFactory;
private final boolean checkBlobDescriptorExists;
+ private final boolean writeNullOnFetchFailure;
private final Set<Integer> blobFields;
public FlinkRowWrapper(org.apache.flink.table.data.RowData row) {
@@ -71,17 +73,40 @@ public class FlinkRowWrapper implements InternalRow {
org.apache.flink.table.data.RowData row,
CatalogContext catalogContext,
boolean checkBlobDescriptorExists) {
- this(row, catalogContext, checkBlobDescriptorExists,
Collections.emptySet());
+ this(row, catalogContext, checkBlobDescriptorExists, false,
Collections.emptySet());
}
public FlinkRowWrapper(
org.apache.flink.table.data.RowData row,
CatalogContext catalogContext,
boolean checkBlobDescriptorExists,
+ boolean writeNullOnFetchFailure) {
+ this(
+ row,
+ catalogContext,
+ checkBlobDescriptorExists,
+ writeNullOnFetchFailure,
+ Collections.emptySet());
+ }
+
+ public FlinkRowWrapper(
+ org.apache.flink.table.data.RowData row,
+ CatalogContext catalogContext,
+ boolean checkBlobDescriptorExists,
+ Set<Integer> blobFields) {
+ this(row, catalogContext, checkBlobDescriptorExists, false,
blobFields);
+ }
+
+ public FlinkRowWrapper(
+ org.apache.flink.table.data.RowData row,
+ CatalogContext catalogContext,
+ boolean checkBlobDescriptorExists,
+ boolean writeNullOnFetchFailure,
Set<Integer> blobFields) {
this.row = row;
this.uriReaderFactory = new UriReaderFactory(catalogContext);
this.checkBlobDescriptorExists = checkBlobDescriptorExists;
+ this.writeNullOnFetchFailure = writeNullOnFetchFailure;
this.blobFields = blobFields;
}
@@ -204,13 +229,13 @@ public class FlinkRowWrapper implements InternalRow {
try {
boolean exists = uriReaderFactory.exists(descriptor.uri());
if (!exists) {
- LOG.warn(
- "Blob descriptor file {} does not exist, returning
NULL for BLOB field at position {}.",
- descriptor.uri(),
- pos);
+ logMissingDescriptor(pos, descriptor);
}
return exists;
} catch (IOException e) {
+ if (deferExistsCheckFailure(e)) {
+ return true;
+ }
LOG.warn(
"Failed to check blob descriptor file {} for BLOB field at
position {}.",
descriptor.uri(),
@@ -218,6 +243,9 @@ public class FlinkRowWrapper implements InternalRow {
e);
throw new RuntimeException(e);
} catch (RuntimeException e) {
+ if (deferExistsCheckFailure(e)) {
+ return true;
+ }
LOG.warn(
"Failed to check blob descriptor file {} for BLOB field at
position {}.",
descriptor.uri(),
@@ -227,6 +255,32 @@ public class FlinkRowWrapper implements InternalRow {
}
}
+ private void logMissingDescriptor(int pos, BlobDescriptor descriptor) {
+ if (isHttpUri(descriptor.uri())) {
+ LOG.warn(
+ "Blob descriptor file {} returned HTTP 404, returning NULL
for BLOB field at position {}.",
+ descriptor.uri(),
+ pos);
+ } else {
+ LOG.warn(
+ "Blob descriptor file {} does not exist, returning NULL
for BLOB field at position {}.",
+ descriptor.uri(),
+ pos);
+ }
+ }
+
+ private static boolean isHttpUri(String uri) {
+ return uri.startsWith("http://";) || uri.startsWith("https://";);
+ }
+
+ /**
+ * When fetch-failure null write is enabled, non-404 errors during exists
pre-check are deferred
+ * to the writer fetch path, so they can be logged and written as NULL.
+ */
+ private boolean deferExistsCheckFailure(Throwable failure) {
+ return writeNullOnFetchFailure &&
!HttpClientUtils.isNotFoundError(failure);
+ }
+
@Override
public InternalArray getArray(int pos) {
return new FlinkArrayWrapper(row.getArray(pos));
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
index a3d0dd3b90..c381864747 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
@@ -223,7 +223,8 @@ public class FlinkSinkBuilder {
this.input,
table.rowType(),
contextForDescriptor,
- table.coreOptions().blobWriteNullOnMissingFile());
+ table.coreOptions().blobWriteNullOnMissingFile(),
+ table.coreOptions().blobWriteNullOnFetchFailure());
if (table.coreOptions().localMergeEnabled() &&
table.schema().primaryKeys().size() > 0) {
SingleOutputStreamOperator<InternalRow> newInput =
input.forward()
@@ -256,7 +257,7 @@ public class FlinkSinkBuilder {
DataStream<RowData> input,
org.apache.paimon.types.RowType rowType,
CatalogContext catalogContext) {
- return mapToInternalRow(input, rowType, catalogContext, false);
+ return mapToInternalRow(input, rowType, catalogContext, false, false);
}
public static DataStream<InternalRow> mapToInternalRow(
@@ -264,6 +265,15 @@ public class FlinkSinkBuilder {
org.apache.paimon.types.RowType rowType,
CatalogContext catalogContext,
boolean checkBlobDescriptorExists) {
+ return mapToInternalRow(input, rowType, catalogContext,
checkBlobDescriptorExists, false);
+ }
+
+ public static DataStream<InternalRow> mapToInternalRow(
+ DataStream<RowData> input,
+ org.apache.paimon.types.RowType rowType,
+ CatalogContext catalogContext,
+ boolean checkBlobDescriptorExists,
+ boolean writeNullOnFetchFailure) {
Set<Integer> blobFields =
checkBlobDescriptorExists
? FlinkRowWrapper.blobFieldIndexes(rowType)
@@ -276,6 +286,7 @@ public class FlinkSinkBuilder {
r,
catalogContext,
checkBlobDescriptorExists,
+
writeNullOnFetchFailure,
blobFields))
.returns(
org.apache.paimon.flink.utils.InternalTypeInfo.fromRowType(
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
index 8cb4e1fbd4..d09ccedb16 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
@@ -537,6 +537,149 @@ public class BlobTableITCase extends CatalogITCaseBase {
}
}
+ @Test
+ public void testWriteFetchFailureDescriptorWritesNull() throws Exception {
+ tEnv.executeSql(
+ "CREATE TABLE fetch_failure_blob_table (id INT, data STRING,
picture BYTES)"
+ + " WITH ('row-tracking.enabled'='true',"
+ + " 'data-evolution.enabled'='true',"
+ + " 'blob-field'='picture',"
+ + " 'blob-as-descriptor'='true',"
+ + " 'blob-write-null-on-fetch-failure'='true')");
+
+ batchSql(
+ "INSERT INTO fetch_failure_blob_table VALUES"
+ + " (1, 'invalid-uri',
sys.path_to_descriptor('https://img.alicdn.com/imgextra/##1304008055350781673'))");
+
+ List<Row> result = batchSql("SELECT * FROM fetch_failure_blob_table");
+ assertThat(result).hasSize(1);
+ assertThat(result.get(0).getField(0)).isEqualTo(1);
+ assertThat(result.get(0).getField(1)).isEqualTo("invalid-uri");
+ assertThat(result.get(0).getField(2)).isNull();
+ }
+
+ @Test
+ public void testWriteHttpBadRequestWritesNullWithFetchFailure() throws
Exception {
+ TestHttpWebServer httpServer = new
TestHttpWebServer("/bad_request_blob");
+ httpServer.start();
+ try {
+ String httpUrl = httpServer.getBaseUrl();
+ httpServer.enqueueResponse("", 400);
+ httpServer.enqueueResponse("", 400);
+
+ tEnv.executeSql(
+ "CREATE TABLE bad_request_blob_table (id INT, data STRING,
picture BYTES)"
+ + " WITH ('row-tracking.enabled'='true',"
+ + " 'data-evolution.enabled'='true',"
+ + " 'blob-field'='picture',"
+ + " 'blob-as-descriptor'='true',"
+ + " 'blob-write-null-on-fetch-failure'='true')");
+
+ batchSql(
+ "INSERT INTO bad_request_blob_table VALUES"
+ + " (1, 'bad-request', sys.path_to_descriptor('"
+ + httpUrl
+ + "'))");
+
+ List<Row> result = batchSql("SELECT * FROM
bad_request_blob_table");
+ assertThat(result).hasSize(1);
+ assertThat(result.get(0).getField(0)).isEqualTo(1);
+ assertThat(result.get(0).getField(1)).isEqualTo("bad-request");
+ assertThat(result.get(0).getField(2)).isNull();
+ } finally {
+ httpServer.stop();
+ }
+ }
+
+ @Test
+ public void testWriteHttpRateLimitWritesNullWithFetchFailure() throws
Exception {
+ TestHttpWebServer httpServer = new
TestHttpWebServer("/rate_limit_blob");
+ httpServer.start();
+ try {
+ String httpUrl = httpServer.getBaseUrl();
+ httpServer.enqueueResponse("", 420);
+ httpServer.enqueueResponse("", 420);
+
+ tEnv.executeSql(
+ "CREATE TABLE rate_limit_blob_table (id INT, data STRING,
picture BYTES)"
+ + " WITH ('row-tracking.enabled'='true',"
+ + " 'data-evolution.enabled'='true',"
+ + " 'blob-field'='picture',"
+ + " 'blob-as-descriptor'='true',"
+ + " 'blob-write-null-on-fetch-failure'='true')");
+
+ batchSql(
+ "INSERT INTO rate_limit_blob_table VALUES"
+ + " (1, 'rate-limit', sys.path_to_descriptor('"
+ + httpUrl
+ + "'))");
+
+ List<Row> result = batchSql("SELECT * FROM rate_limit_blob_table");
+ assertThat(result).hasSize(1);
+ assertThat(result.get(0).getField(0)).isEqualTo(1);
+ assertThat(result.get(0).getField(1)).isEqualTo("rate-limit");
+ assertThat(result.get(0).getField(2)).isNull();
+ } finally {
+ httpServer.stop();
+ }
+ }
+
+ @Test
+ public void testWriteInvalidUriWritesNullWithMissingFileAndFetchFailure()
throws Exception {
+ tEnv.executeSql(
+ "CREATE TABLE combined_null_blob_table (id INT, data STRING,
picture BYTES)"
+ + " WITH ('row-tracking.enabled'='true',"
+ + " 'data-evolution.enabled'='true',"
+ + " 'blob-field'='picture',"
+ + " 'blob-as-descriptor'='true',"
+ + " 'blob-write-null-on-missing-file'='true',"
+ + " 'blob-write-null-on-fetch-failure'='true')");
+
+ batchSql(
+ "INSERT INTO combined_null_blob_table VALUES"
+ + " (1, 'invalid-uri',
sys.path_to_descriptor('https://img.alicdn.com/imgextra/##1304008055350781673'))");
+
+ List<Row> result = batchSql("SELECT * FROM combined_null_blob_table");
+ assertThat(result).hasSize(1);
+ assertThat(result.get(0).getField(0)).isEqualTo(1);
+ assertThat(result.get(0).getField(1)).isEqualTo("invalid-uri");
+ assertThat(result.get(0).getField(2)).isNull();
+ }
+
+ @Test
+ public void
testWriteHttpBadRequestWritesNullWithMissingFileAndFetchFailure() throws
Exception {
+ TestHttpWebServer httpServer = new
TestHttpWebServer("/combined_bad_request_blob");
+ httpServer.start();
+ try {
+ String httpUrl = httpServer.getBaseUrl();
+ httpServer.enqueueResponse("", 400);
+ httpServer.enqueueResponse("", 400);
+
+ tEnv.executeSql(
+ "CREATE TABLE combined_bad_request_blob_table (id INT,
data STRING, picture BYTES)"
+ + " WITH ('row-tracking.enabled'='true',"
+ + " 'data-evolution.enabled'='true',"
+ + " 'blob-field'='picture',"
+ + " 'blob-as-descriptor'='true',"
+ + " 'blob-write-null-on-missing-file'='true',"
+ + " 'blob-write-null-on-fetch-failure'='true')");
+
+ batchSql(
+ "INSERT INTO combined_bad_request_blob_table VALUES"
+ + " (1, 'bad-request', sys.path_to_descriptor('"
+ + httpUrl
+ + "'))");
+
+ List<Row> result = batchSql("SELECT * FROM
combined_bad_request_blob_table");
+ assertThat(result).hasSize(1);
+ assertThat(result.get(0).getField(0)).isEqualTo(1);
+ assertThat(result.get(0).getField(1)).isEqualTo("bad-request");
+ assertThat(result.get(0).getField(2)).isNull();
+ } finally {
+ httpServer.stop();
+ }
+ }
+
@Test
public void testBlobTypeSchemaEquals() throws Exception {
// Step 1: Create a Paimon table with blob field via Flink SQL
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkRowWrapperTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkRowWrapperTest.java
index c53adecec1..517917e4c0 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkRowWrapperTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkRowWrapperTest.java
@@ -39,6 +39,7 @@ import java.util.Collections;
import java.util.Set;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link FlinkRowWrapper}. */
public class FlinkRowWrapperTest {
@@ -147,6 +148,60 @@ public class FlinkRowWrapperTest {
assertThat(blob).isNotNull();
}
+ @Test
+ public void testInvalidUriDefersExistsCheckWhenFetchFailureEnabled() {
+ GenericRowData row =
+
descriptorRow("https://img.alicdn.com/imgextra/##1304008055350781673";, 1);
+
+ FlinkRowWrapper wrapper = wrapper(row, true, true);
+
+ assertThat(wrapper.isNullAt(0)).isFalse();
+ }
+
+ @Test
+ public void testInvalidUriThrowsWhenFetchFailureDisabled() {
+ GenericRowData row =
+
descriptorRow("https://img.alicdn.com/imgextra/##1304008055350781673";, 1);
+
+ FlinkRowWrapper wrapper = wrapper(row, true, false);
+
+ assertThatThrownBy(() -> wrapper.isNullAt(0))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Illegal character");
+ }
+
+ @Test
+ public void testHttpBadRequestDefersExistsCheckWhenFetchFailureEnabled()
throws Exception {
+ httpServer.createContext(
+ "/bad.jpg",
+ exchange -> {
+ sendResponse(exchange, 400, new byte[0]);
+ });
+ GenericRowData row = descriptorRow("http://127.0.0.1:"; + httpPort +
"/bad.jpg", 1);
+
+ FlinkRowWrapper wrapper = wrapper(row, true, true);
+
+ assertThat(wrapper.isNullAt(0)).isFalse();
+ }
+
+ @Test
+ public void testHttpBadRequestThrowsWhenFetchFailureDisabled() throws
Exception {
+ httpServer.createContext(
+ "/bad.jpg",
+ exchange -> {
+ sendResponse(exchange, 400, new byte[0]);
+ });
+ GenericRowData row = descriptorRow("http://127.0.0.1:"; + httpPort +
"/bad.jpg", 1);
+
+ FlinkRowWrapper wrapper = wrapper(row, true, false);
+
+ assertThatThrownBy(() -> wrapper.isNullAt(0))
+ .isInstanceOf(RuntimeException.class)
+ .hasRootCauseInstanceOf(IOException.class)
+ .rootCause()
+ .hasMessageContaining("Unexpected HTTP status code: 400");
+ }
+
private GenericRowData descriptorRow(java.nio.file.Path path, long length)
{
return descriptorRow(path.toUri().toString(), length);
}
@@ -170,12 +225,32 @@ public class FlinkRowWrapperTest {
}
private FlinkRowWrapper wrapper(GenericRowData row, boolean
checkBlobDescriptorExists) {
- return wrapper(row, checkBlobDescriptorExists,
Collections.singleton(0));
+ return wrapper(row, checkBlobDescriptorExists, false);
+ }
+
+ private FlinkRowWrapper wrapper(
+ GenericRowData row,
+ boolean checkBlobDescriptorExists,
+ boolean writeNullOnFetchFailure) {
+ return wrapper(
+ row, checkBlobDescriptorExists, writeNullOnFetchFailure,
Collections.singleton(0));
}
private FlinkRowWrapper wrapper(
GenericRowData row, boolean checkBlobDescriptorExists,
Set<Integer> blobFields) {
+ return wrapper(row, checkBlobDescriptorExists, false, blobFields);
+ }
+
+ private FlinkRowWrapper wrapper(
+ GenericRowData row,
+ boolean checkBlobDescriptorExists,
+ boolean writeNullOnFetchFailure,
+ Set<Integer> blobFields) {
return new FlinkRowWrapper(
- row, CatalogContext.create(new Options()),
checkBlobDescriptorExists, blobFields);
+ row,
+ CatalogContext.create(new Options()),
+ checkBlobDescriptorExists,
+ writeNullOnFetchFailure,
+ blobFields);
}
}
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 c65f76eeaa..66798c4bf3 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
@@ -50,6 +50,8 @@ import static
org.apache.paimon.utils.Preconditions.checkArgument;
public class BlobFileFormat extends FileFormat {
private final boolean blobAsDescriptor;
+ private boolean writeNullOnMissingFile;
+ private boolean writeNullOnFetchFailure;
@Nullable public BlobConsumer writeConsumer;
@@ -66,6 +68,14 @@ public class BlobFileFormat extends FileFormat {
return fileName.endsWith("." + BlobFileFormatFactory.IDENTIFIER);
}
+ public void setWriteNullOnMissingFile(boolean writeNullOnMissingFile) {
+ this.writeNullOnMissingFile = writeNullOnMissingFile;
+ }
+
+ public void setWriteNullOnFetchFailure(boolean writeNullOnFetchFailure) {
+ this.writeNullOnFetchFailure = writeNullOnFetchFailure;
+ }
+
public void setWriteConsumer(@Nullable BlobConsumer writeConsumer) {
this.writeConsumer = writeConsumer;
}
@@ -108,7 +118,8 @@ public class BlobFileFormat extends FileFormat {
@Override
public FormatWriter create(PositionOutputStream out, String
compression) {
- return new BlobFormatWriter(out, writeConsumer, type);
+ return new BlobFormatWriter(
+ out, writeConsumer, type, writeNullOnMissingFile,
writeNullOnFetchFailure);
}
}
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 f97158c171..5b3e50537d 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
@@ -22,16 +22,21 @@ import org.apache.paimon.data.Blob;
import org.apache.paimon.data.BlobConsumer;
import org.apache.paimon.data.BlobDescriptor;
import org.apache.paimon.data.BlobPlaceholder;
+import org.apache.paimon.data.BlobRef;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.format.FileAwareFormatWriter;
import org.apache.paimon.format.FormatWriter;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.rest.HttpClientUtils;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.DeltaVarintCompressor;
import org.apache.paimon.utils.LongArrayList;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import javax.annotation.Nullable;
import java.io.IOException;
@@ -44,6 +49,8 @@ import static
org.apache.paimon.utils.StreamUtils.longToLittleEndian;
/** {@link FormatWriter} for blob file. */
public class BlobFormatWriter implements FileAwareFormatWriter {
+ private static final Logger LOG =
LoggerFactory.getLogger(BlobFormatWriter.class);
+
public static final byte VERSION = 1;
public static final int MAGIC_NUMBER = 1481511375;
public static final byte[] MAGIC_NUMBER_BYTES =
intToLittleEndian(MAGIC_NUMBER);
@@ -53,6 +60,8 @@ public class BlobFormatWriter implements
FileAwareFormatWriter {
private final PositionOutputStream out;
@Nullable private final BlobConsumer writeConsumer;
private final String blobFieldName;
+ private final boolean writeNullOnMissingFile;
+ private final boolean writeNullOnFetchFailure;
private final CRC32 crc32;
private final byte[] tmpBuffer;
private final LongArrayList lengths;
@@ -61,8 +70,27 @@ public class BlobFormatWriter implements
FileAwareFormatWriter {
public BlobFormatWriter(
PositionOutputStream out, @Nullable BlobConsumer writeConsumer,
RowType type) {
+ this(out, writeConsumer, type, false, false);
+ }
+
+ public BlobFormatWriter(
+ PositionOutputStream out,
+ @Nullable BlobConsumer writeConsumer,
+ RowType type,
+ boolean writeNullOnMissingFile) {
+ this(out, writeConsumer, type, writeNullOnMissingFile, false);
+ }
+
+ public BlobFormatWriter(
+ PositionOutputStream out,
+ @Nullable BlobConsumer writeConsumer,
+ RowType type,
+ boolean writeNullOnMissingFile,
+ boolean writeNullOnFetchFailure) {
this.out = out;
this.writeConsumer = writeConsumer;
+ this.writeNullOnMissingFile = writeNullOnMissingFile;
+ this.writeNullOnFetchFailure = writeNullOnFetchFailure;
checkArgument(type.getFieldCount() == 1, "BlobFormatWriter only
support one field.");
this.blobFieldName = type.getFieldNames().get(0);
this.crc32 = new CRC32();
@@ -84,34 +112,56 @@ 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)) {
- lengths.add(NULL_LENGTH);
- if (writeConsumer != null) {
- writeConsumer.accept(blobFieldName, null);
- }
+ writeNullElement();
return;
}
- Blob blob = element.getBlob(0);
+ Blob blob;
+ try {
+ blob = element.getBlob(0);
+ } catch (RuntimeException e) {
+ if (tryWriteNullOnFetchFailure(e, null)) {
+ return;
+ }
+ throw e;
+ }
if (blob == BlobPlaceholder.INSTANCE) {
lengths.add(PLACE_HOLDER_LENGTH);
return;
}
- long previousPos = out.getPos();
- crc32.reset();
+ SeekableInputStream in;
+ try {
+ in = blob.newInputStream();
+ } catch (IOException | RuntimeException e) {
+ if (writeNullOnMissingFile && HttpClientUtils.isNotFoundError(e)) {
+ LOG.warn(
+ "Failed to open blob from {} (HTTP 404), writing NULL
for BLOB field {}.",
+ blobUri(blob),
+ blobFieldName,
+ e);
+ writeNullElement();
+ return;
+ }
+ if (tryWriteNullOnFetchFailure(e, blob)) {
+ return;
+ }
+ throw e;
+ }
+ crc32.reset();
write(MAGIC_NUMBER_BYTES);
-
long blobPos = out.getPos();
- try (SeekableInputStream in = blob.newInputStream()) {
- int bytesRead = in.read(tmpBuffer);
+ long blobLength = 0;
+ try (SeekableInputStream stream = in) {
+ int bytesRead = stream.read(tmpBuffer);
while (bytesRead >= 0) {
write(tmpBuffer, bytesRead);
- bytesRead = in.read(tmpBuffer);
+ blobLength += bytesRead;
+ bytesRead = stream.read(tmpBuffer);
}
}
- long blobLength = out.getPos() - blobPos;
- long binLength = out.getPos() - previousPos + 12;
+ long binLength = blobLength + MAGIC_NUMBER_BYTES.length + 12;
lengths.add(binLength);
byte[] lenBytes = longToLittleEndian(binLength);
write(lenBytes);
@@ -127,6 +177,50 @@ public class BlobFormatWriter implements
FileAwareFormatWriter {
}
}
+ private boolean tryWriteNullOnFetchFailure(Throwable e, @Nullable Blob
blob)
+ throws IOException {
+ if (!writeNullOnFetchFailure || HttpClientUtils.isNotFoundError(e)) {
+ return false;
+ }
+ Integer statusCode = HttpClientUtils.getHttpStatusCode(e);
+ if (statusCode != null) {
+ LOG.warn(
+ "Failed to open blob from {} (HTTP {}), writing NULL for
BLOB field {}.",
+ blobUri(blob),
+ statusCode,
+ blobFieldName,
+ e);
+ } else if (HttpClientUtils.isInvalidUriException(e)) {
+ LOG.warn(
+ "Invalid blob URI {} while opening blob, writing NULL for
BLOB field {}.",
+ blobUri(blob),
+ blobFieldName,
+ e);
+ } else {
+ LOG.warn(
+ "Failed to open blob from {} due to fetch failure, writing
NULL for BLOB field {}.",
+ blobUri(blob),
+ blobFieldName,
+ e);
+ }
+ writeNullElement();
+ return true;
+ }
+
+ private void writeNullElement() throws IOException {
+ lengths.add(NULL_LENGTH);
+ if (writeConsumer != null) {
+ writeConsumer.accept(blobFieldName, null);
+ }
+ }
+
+ private static String blobUri(@Nullable Blob blob) {
+ if (blob instanceof BlobRef) {
+ return ((BlobRef) blob).toDescriptor().uri();
+ }
+ return "unknown";
+ }
+
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
new file mode 100644
index 0000000000..90ded82a19
--- /dev/null
+++
b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java
@@ -0,0 +1,377 @@
+/*
+ * 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.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.BlobRef;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.FileRecordIterator;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.UriReader;
+import org.apache.paimon.utils.UriReaderFactory;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Files;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link BlobFormatWriter}. */
+public class BlobFormatWriterTest {
+
+ @Test
+ public void testTwoConsecutiveBlobsPreserveReadback(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ RowType rowType = RowType.of(DataTypes.BLOB());
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+ byte[] firstPayload = "first-blob".getBytes();
+ byte[] secondPayload = "second-blob-payload".getBytes();
+
+ BlobFormatWriter writer =
+ new BlobFormatWriter(
+ new
LocalFileIO.LocalPositionOutputStream(outputFile.toFile()),
+ null,
+ rowType);
+ writer.addElement(GenericRow.of(Blob.fromData(firstPayload)));
+ writer.addElement(GenericRow.of(Blob.fromData(secondPayload)));
+ writer.close();
+
+ LocalFileIO fileIO = new LocalFileIO();
+ Path filePath = new Path(outputFile.toUri());
+ long fileSize = Files.size(outputFile);
+ try (SeekableInputStream in = fileIO.newInputStream(filePath)) {
+ BlobFileMeta fileMeta = new BlobFileMeta(in, fileSize, null);
+ assertThat(fileMeta.recordNumber()).isEqualTo(2);
+
+ BlobFormatReader reader = new BlobFormatReader(fileIO, filePath,
fileMeta, in, 1, 0);
+ FileRecordIterator<InternalRow> iterator = reader.readBatch();
+ assertBlobPayload(iterator.next().getBlob(0), firstPayload);
+ assertBlobPayload(iterator.next().getBlob(0), secondPayload);
+ }
+ }
+
+ @Test
+ public void testWriteNullOnFetchFailureFallbackForHttpBadRequest(
+ @TempDir java.nio.file.Path tempDir) throws Exception {
+ RowType rowType = RowType.of(DataTypes.BLOB());
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+ BlobFormatWriter writer =
+ new BlobFormatWriter(
+ new
LocalFileIO.LocalPositionOutputStream(outputFile.toFile()),
+ null,
+ rowType,
+ false,
+ true);
+
+ writer.addElement(
+ GenericRow.of(
+ new BlobRef(
+ failingHttpReader(400),
+ new
BlobDescriptor("https://example.com/bad-request.jpg";, 0, -1))));
+
+ writer.close();
+
+ assertThat(outputFile.toFile()).exists();
+ }
+
+ @Test
+ public void testHttpRateLimitWritesNullWhenFetchFailureEnabled(
+ @TempDir java.nio.file.Path tempDir) throws Exception {
+ RowType rowType = RowType.of(DataTypes.BLOB());
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+ BlobFormatWriter writer =
+ new BlobFormatWriter(
+ new
LocalFileIO.LocalPositionOutputStream(outputFile.toFile()),
+ null,
+ rowType,
+ false,
+ true);
+
+ writer.addElement(
+ GenericRow.of(
+ new BlobRef(
+ failingHttpReader(420),
+ new
BlobDescriptor("https://example.com/rate-limit.jpg";, 0, -1))));
+
+ writer.close();
+
+ assertThat(outputFile.toFile()).exists();
+ }
+
+ @Test
+ public void testHttpRateLimitFailsWhenFetchFailureDisabled(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ RowType rowType = RowType.of(DataTypes.BLOB());
+ BlobFormatWriter writer =
+ new BlobFormatWriter(
+ new LocalFileIO.LocalPositionOutputStream(
+ tempDir.resolve("blob.out").toFile()),
+ null,
+ rowType,
+ false,
+ false);
+
+ assertThatThrownBy(
+ () ->
+ writer.addElement(
+ GenericRow.of(
+ new BlobRef(
+ failingHttpReader(420),
+ new BlobDescriptor(
+
"https://example.com/rate-limit.jpg";,
+ 0,
+ -1)))))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("HTTP error code: 420");
+ }
+
+ @Test
+ public void testHttpNotFoundPropagatesWhenFetchFailureDisabled(
+ @TempDir java.nio.file.Path tempDir) throws Exception {
+ RowType rowType = RowType.of(DataTypes.BLOB());
+ BlobFormatWriter writer =
+ new BlobFormatWriter(
+ new LocalFileIO.LocalPositionOutputStream(
+ tempDir.resolve("blob.out").toFile()),
+ null,
+ rowType,
+ false,
+ false);
+
+ assertThatThrownBy(
+ () ->
+ writer.addElement(
+ GenericRow.of(
+ new BlobRef(
+ failingHttpReader(404),
+ new BlobDescriptor(
+
"https://example.com/missing.jpg";,
+ 0,
+ -1)))))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("HTTP error code: 404");
+ }
+
+ @Test
+ public void testWriteNullOnFetchFailureForInvalidUriDescriptor(
+ @TempDir java.nio.file.Path tempDir) throws Exception {
+ RowType rowType = RowType.of(DataTypes.BLOB());
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+ UriReaderFactory uriReaderFactory =
+ new UriReaderFactory(CatalogContext.create(new Options()));
+ byte[] descriptorBytes =
+ new
BlobDescriptor("https://img.alicdn.com/imgextra/##1304008055350781673";, 0, -1)
+ .serialize();
+
+ BlobFormatWriter writer =
+ new BlobFormatWriter(
+ new
LocalFileIO.LocalPositionOutputStream(outputFile.toFile()),
+ null,
+ rowType,
+ false,
+ true);
+
+ writer.addElement(new DescriptorBytesRow(descriptorBytes,
uriReaderFactory));
+ writer.close();
+
+ LocalFileIO fileIO = new LocalFileIO();
+ Path filePath = new Path(outputFile.toUri());
+ long fileSize = Files.size(outputFile);
+ try (SeekableInputStream in = fileIO.newInputStream(filePath)) {
+ BlobFileMeta fileMeta = new BlobFileMeta(in, fileSize, null);
+ assertThat(fileMeta.recordNumber()).isEqualTo(1);
+ assertThat(fileMeta.isNull(0)).isTrue();
+ }
+ }
+
+ @Test
+ public void testHttpNotFoundWritesNullWhenMissingFileEnabled(
+ @TempDir java.nio.file.Path tempDir) throws Exception {
+ RowType rowType = RowType.of(DataTypes.BLOB());
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+ BlobFormatWriter writer =
+ new BlobFormatWriter(
+ new
LocalFileIO.LocalPositionOutputStream(outputFile.toFile()),
+ null,
+ rowType,
+ true,
+ false);
+
+ writer.addElement(
+ GenericRow.of(
+ new BlobRef(
+ failingHttpReader(404),
+ new
BlobDescriptor("https://example.com/missing.jpg";, 0, -1))));
+ writer.close();
+
+ LocalFileIO fileIO = new LocalFileIO();
+ Path filePath = new Path(outputFile.toUri());
+ long fileSize = Files.size(outputFile);
+ try (SeekableInputStream in = fileIO.newInputStream(filePath)) {
+ BlobFileMeta fileMeta = new BlobFileMeta(in, fileSize, null);
+ assertThat(fileMeta.recordNumber()).isEqualTo(1);
+ assertThat(fileMeta.isNull(0)).isTrue();
+ }
+ }
+
+ private static void assertBlobPayload(Blob blob, byte[] expected) throws
Exception {
+ try (SeekableInputStream blobIn = blob.newInputStream()) {
+ byte[] actual = new byte[expected.length];
+ org.apache.paimon.utils.IOUtils.readFully(blobIn, actual);
+ assertThat(actual).isEqualTo(expected);
+ }
+ }
+
+ private static UriReader failingHttpReader(int statusCode) {
+ return new UriReader() {
+ @Override
+ public SeekableInputStream newInputStream(String uri) {
+ throw new RuntimeException("HTTP error code: " + statusCode);
+ }
+ };
+ }
+
+ private static final class DescriptorBytesRow implements InternalRow {
+ private final byte[] descriptorBytes;
+ private final UriReaderFactory uriReaderFactory;
+
+ private DescriptorBytesRow(byte[] descriptorBytes, UriReaderFactory
uriReaderFactory) {
+ this.descriptorBytes = descriptorBytes;
+ this.uriReaderFactory = uriReaderFactory;
+ }
+
+ @Override
+ public int getFieldCount() {
+ return 1;
+ }
+
+ @Override
+ public RowKind getRowKind() {
+ return RowKind.INSERT;
+ }
+
+ @Override
+ public void setRowKind(RowKind kind) {}
+
+ @Override
+ public boolean isNullAt(int pos) {
+ return false;
+ }
+
+ @Override
+ public Blob getBlob(int pos) {
+ return Blob.fromBytes(descriptorBytes, uriReaderFactory, null);
+ }
+
+ private UnsupportedOperationException unsupported() {
+ return new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean getBoolean(int pos) {
+ throw unsupported();
+ }
+
+ @Override
+ public byte getByte(int pos) {
+ throw unsupported();
+ }
+
+ @Override
+ public short getShort(int pos) {
+ throw unsupported();
+ }
+
+ @Override
+ public int getInt(int pos) {
+ throw unsupported();
+ }
+
+ @Override
+ public long getLong(int pos) {
+ throw unsupported();
+ }
+
+ @Override
+ public float getFloat(int pos) {
+ throw unsupported();
+ }
+
+ @Override
+ public double getDouble(int pos) {
+ throw unsupported();
+ }
+
+ @Override
+ public org.apache.paimon.data.BinaryString getString(int pos) {
+ throw unsupported();
+ }
+
+ @Override
+ public org.apache.paimon.data.Decimal getDecimal(int pos, int
precision, int scale) {
+ throw unsupported();
+ }
+
+ @Override
+ public org.apache.paimon.data.Timestamp getTimestamp(int pos, int
precision) {
+ throw unsupported();
+ }
+
+ @Override
+ public byte[] getBinary(int pos) {
+ throw unsupported();
+ }
+
+ @Override
+ public org.apache.paimon.data.variant.Variant getVariant(int pos) {
+ throw unsupported();
+ }
+
+ @Override
+ public org.apache.paimon.data.InternalArray getArray(int pos) {
+ throw unsupported();
+ }
+
+ @Override
+ public org.apache.paimon.data.InternalVector getVector(int pos) {
+ throw unsupported();
+ }
+
+ @Override
+ public org.apache.paimon.data.InternalMap getMap(int pos) {
+ throw unsupported();
+ }
+
+ @Override
+ public InternalRow getRow(int pos, int numFields) {
+ throw unsupported();
+ }
+ }
+}
![https://example.com/bad-request.jpg"](https://example.com/bad-request.jpg"){kind=link}
![https://example.com/rate-limit.jpg"](https://example.com/rate-limit.jpg"){kind=link}
![https://example.com/missing.jpg"](https://example.com/missing.jpg"){kind=link}