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 78ef07218e [Flink][Blob] Support blob-write-null-on-missing-file for
HTTP URLs. (#8219)
78ef07218e is described below
commit 78ef07218ef9a1dd1e9a3cbaf58b54093c2b4888
Author: wangwj <[email protected]>
AuthorDate: Wed Jul 1 13:38:19 2026 +0800
[Flink][Blob] Support blob-write-null-on-missing-file for HTTP URLs. (#8219)
---
docs/generated/core_configuration.html | 26 +--
.../main/java/org/apache/paimon/CoreOptions.java | 5 +-
.../org/apache/paimon/rest/HttpClientUtils.java | 41 +++++
.../apache/paimon/rest/HttpClientUtilsTest.java | 187 +++++++++++++++++++++
.../java/org/apache/paimon/utils/UriReader.java | 4 +
.../org/apache/paimon/utils/UriReaderFactory.java | 9 +-
.../apache/paimon/utils/UriReaderFactoryTest.java | 67 +++++++-
.../org/apache/paimon/flink/FlinkRowWrapper.java | 29 +++-
.../apache/paimon/flink/sink/FlinkSinkBuilder.java | 9 +-
.../apache/paimon/flink/FlinkRowWrapperTest.java | 104 +++++++++++-
10 files changed, 457 insertions(+), 24 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 34fce08b25..33d038ba43 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -96,7 +96,7 @@ under the License.
<td><h5>blob-write-null-on-missing-file</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 file does not exist during Flink writes. When false, the write fails
when the descriptor is read.</td>
+ <td>Whether to write NULL for a descriptor BLOB value when the
referenced file or HTTP resource does not exist during Flink writes. When
false, the write fails when the descriptor is read.</td>
</tr>
<tr>
<td><h5>blob.split-by-file-size</h5></td>
@@ -464,6 +464,18 @@ under the License.
<td>Boolean</td>
<td>Whether enable data evolution for row tracking table.</td>
</tr>
+ <tr>
+ <td><h5>data-evolution.merge-into.file-pruning</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>If true, enables the file-level pruning step for MergeInto
partial column update on data-evolution tables. Set this to false when most
files in the target partition are expected to be updated, so that the overhead
of collecting touched file IDs outweighs the benefit of pruning untouched
files.</td>
+ </tr>
+ <tr>
+ <td><h5>data-evolution.merge-into.source-persist</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to persist source when process merge into action on
data evolution table.</td>
+ </tr>
<tr>
<td><h5>data-evolution.row-sidecar.enabled</h5></td>
<td style="word-wrap: break-word;">false</td>
@@ -482,18 +494,6 @@ under the License.
<td>Double</td>
<td>Maximum selected row ratio for reading a row-store sidecar
file. The value must be in (0, 1]. The sidecar is used only when the selected
row ratio is no more than this value and the selected row count is no more than
data-evolution.row-sidecar.max-selected-rows.</td>
</tr>
- <tr>
- <td><h5>data-evolution.merge-into.file-pruning</h5></td>
- <td style="word-wrap: break-word;">true</td>
- <td>Boolean</td>
- <td>If true, enables the file-level pruning step for MergeInto
partial column update on data-evolution tables. Set this to false when most
files in the target partition are expected to be updated, so that the overhead
of collecting touched file IDs outweighs the benefit of pruning untouched
files.</td>
- </tr>
- <tr>
- <td><h5>data-evolution.merge-into.source-persist</h5></td>
- <td style="word-wrap: break-word;">false</td>
- <td>Boolean</td>
- <td>Whether to persist source when process merge into action on
data evolution table.</td>
- </tr>
<tr>
<td><h5>data-file.external-paths</h5></td>
<td style="word-wrap: break-word;">(none)</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 979db0cfcd..55c25b409b 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2527,8 +2527,9 @@ public class CoreOptions implements Serializable {
.defaultValue(false)
.withDescription(
"Whether to write NULL for a descriptor BLOB value
when the "
- + "referenced file does not exist during
Flink writes. When "
- + "false, the write fails when the
descriptor is read.");
+ + "referenced file or HTTP resource does
not exist during Flink "
+ + "writes. When false, the write fails
when the descriptor is "
+ + "read.");
public static final ConfigOption<Boolean> COMMIT_DISCARD_DUPLICATE_FILES =
key("commit.discard-duplicate-files")
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 9447f24171..f5a4a1ec0d 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
@@ -22,6 +22,7 @@ import org.apache.paimon.rest.interceptor.LoggingInterceptor;
import org.apache.paimon.rest.interceptor.TimingInterceptor;
import org.apache.hc.client5.http.classic.methods.HttpGet;
+import org.apache.hc.client5.http.classic.methods.HttpHead;
import org.apache.hc.client5.http.config.RequestConfig;
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
@@ -31,6 +32,7 @@ import
org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuil
import org.apache.hc.client5.http.io.HttpClientConnectionManager;
import org.apache.hc.client5.http.ssl.DefaultClientTlsStrategy;
import org.apache.hc.client5.http.ssl.HttpsSupport;
+import org.apache.hc.core5.http.HttpStatus;
import org.apache.hc.core5.reactor.ssl.SSLBufferMode;
import org.apache.hc.core5.ssl.SSLContexts;
import org.apache.hc.core5.util.Timeout;
@@ -91,4 +93,43 @@ public class HttpClientUtils {
}
return response.getEntity().getContent();
}
+
+ /**
+ * 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.
+ */
+ public static boolean exists(String uri) throws IOException {
+ int headStatusCode = headStatusCode(uri);
+ if (headStatusCode == HttpStatus.SC_OK) {
+ return true;
+ }
+ int rangeStatusCode = getRangeStatusCode(uri);
+ if (rangeStatusCode == HttpStatus.SC_OK
+ || rangeStatusCode == HttpStatus.SC_PARTIAL_CONTENT
+ || rangeStatusCode ==
HttpStatus.SC_REQUESTED_RANGE_NOT_SATISFIABLE) {
+ return true;
+ }
+ if (rangeStatusCode == HttpStatus.SC_NOT_FOUND) {
+ return false;
+ }
+ throw new IOException(
+ "Unexpected HTTP status code: " + rangeStatusCode + " for uri:
" + uri);
+ }
+
+ private static int headStatusCode(String uri) throws IOException {
+ HttpHead httpHead = new HttpHead(uri);
+ try (CloseableHttpResponse response =
DEFAULT_HTTP_CLIENT.execute(httpHead)) {
+ return response.getCode();
+ }
+ }
+
+ private static int getRangeStatusCode(String uri) throws IOException {
+ HttpGet httpGet = new HttpGet(uri);
+ httpGet.addHeader("Range", "bytes=0-0");
+ try (CloseableHttpResponse response =
DEFAULT_HTTP_CLIENT.execute(httpGet)) {
+ return response.getCode();
+ }
+ }
}
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
new file mode 100644
index 0000000000..064ebe232f
--- /dev/null
+++ b/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java
@@ -0,0 +1,187 @@
+/*
+ * 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.rest;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link HttpClientUtils}. */
+public class HttpClientUtilsTest {
+
+ private HttpServer server;
+ private int port;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ server = HttpServer.create(new InetSocketAddress(0), 0);
+ port = server.getAddress().getPort();
+ server.start();
+ }
+
+ @AfterEach
+ public void tearDown() {
+ if (server != null) {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ public void testExistsReturnsTrueForAvailableResource() throws Exception {
+ registerHandler(
+ "/ok",
+ exchange -> {
+ respond(exchange, 200, "abc".getBytes());
+ });
+
+ assertThat(HttpClientUtils.exists(url("/ok"))).isTrue();
+ }
+
+ @Test
+ public void testExistsReturnsFalseForMissingResource() throws Exception {
+ registerHandler(
+ "/missing",
+ exchange -> {
+ respond(exchange, 404, new byte[0]);
+ });
+
+ assertThat(HttpClientUtils.exists(url("/missing"))).isFalse();
+ }
+
+ @Test
+ public void testExistsFallsBackToRangeGetWhenHeadNotAllowed() throws
Exception {
+ registerHandler(
+ "/no-head",
+ exchange -> {
+ if ("HEAD".equals(exchange.getRequestMethod())) {
+ respond(exchange, 405, new byte[0]);
+ return;
+ }
+ respond(exchange, 200, "abc".getBytes());
+ });
+
+ assertThat(HttpClientUtils.exists(url("/no-head"))).isTrue();
+ }
+
+ @Test
+ public void testExistsFallsBackToRangeGetWhenHeadReturnsNotFound() throws
Exception {
+ registerHandler(
+ "/head-404-get-ok",
+ exchange -> {
+ if ("HEAD".equals(exchange.getRequestMethod())) {
+ respond(exchange, 404, new byte[0]);
+ return;
+ }
+ if ("GET".equals(exchange.getRequestMethod())
+ && exchange.getRequestHeaders().getFirst("Range")
!= null) {
+ respond(exchange, 206, "abc".getBytes());
+ return;
+ }
+ respond(exchange, 404, new byte[0]);
+ });
+
+ assertThat(HttpClientUtils.exists(url("/head-404-get-ok"))).isTrue();
+ }
+
+ @Test
+ public void testExistsFallsBackToRangeGetWhenHeadReturnsForbidden() throws
Exception {
+ registerHandler(
+ "/head-403-get-ok",
+ exchange -> {
+ if ("HEAD".equals(exchange.getRequestMethod())) {
+ respond(exchange, 403, new byte[0]);
+ return;
+ }
+ if ("GET".equals(exchange.getRequestMethod())
+ && exchange.getRequestHeaders().getFirst("Range")
!= null) {
+ respond(exchange, 200, "abc".getBytes());
+ return;
+ }
+ respond(exchange, 403, new byte[0]);
+ });
+
+ assertThat(HttpClientUtils.exists(url("/head-403-get-ok"))).isTrue();
+ }
+
+ @Test
+ public void testExistsTreatsEmptyResourceAsExistingWhenRangeReturns416()
throws Exception {
+ registerHandler(
+ "/empty-no-head",
+ exchange -> {
+ if ("HEAD".equals(exchange.getRequestMethod())) {
+ respond(exchange, 405, new byte[0]);
+ return;
+ }
+ if ("GET".equals(exchange.getRequestMethod())
+ && exchange.getRequestHeaders().getFirst("Range")
!= null) {
+ respond(exchange, 416, new byte[0]);
+ return;
+ }
+ respond(exchange, 200, new byte[0]);
+ });
+
+ assertThat(HttpClientUtils.exists(url("/empty-no-head"))).isTrue();
+ }
+
+ @Test
+ public void testExistsReturnsFalseOnlyWhenRangeGetAlsoNotFound() throws
Exception {
+ registerHandler(
+ "/head-404-get-404",
+ exchange -> {
+ if ("HEAD".equals(exchange.getRequestMethod())) {
+ respond(exchange, 404, new byte[0]);
+ return;
+ }
+ respond(exchange, 404, new byte[0]);
+ });
+
+ assertThat(HttpClientUtils.exists(url("/head-404-get-404"))).isFalse();
+ }
+
+ private void registerHandler(String path, HttpHandler handler) {
+ server.createContext(path, handler);
+ }
+
+ private String url(String path) {
+ return "http://127.0.0.1:"; + port + path;
+ }
+
+ private static void respond(HttpExchange exchange, int statusCode, byte[]
body)
+ throws IOException {
+ boolean headRequest = "HEAD".equals(exchange.getRequestMethod());
+ long responseLength = headRequest ? -1 : body.length;
+ exchange.sendResponseHeaders(statusCode, responseLength);
+ if (!headRequest && body.length > 0) {
+ try (OutputStream outputStream = exchange.getResponseBody()) {
+ outputStream.write(body);
+ }
+ } else {
+ exchange.close();
+ }
+ }
+}
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/UriReader.java
b/paimon-common/src/main/java/org/apache/paimon/utils/UriReader.java
index 6c0b97a25c..236bd6b6f6 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/UriReader.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/UriReader.java
@@ -64,5 +64,9 @@ public interface UriReader {
public SeekableInputStream newInputStream(String uri) throws
IOException {
return
SeekableInputStream.wrap(HttpClientUtils.getAsInputStream(uri));
}
+
+ public boolean exists(String uri) throws IOException {
+ return HttpClientUtils.exists(uri);
+ }
}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/utils/UriReaderFactory.java
b/paimon-common/src/main/java/org/apache/paimon/utils/UriReaderFactory.java
index 83c16351e0..a2fd2d4ad3 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/UriReaderFactory.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/UriReaderFactory.java
@@ -51,8 +51,13 @@ public class UriReaderFactory implements Serializable {
public boolean exists(String input) throws IOException {
UriReader reader = create(input);
- return !(reader instanceof UriReader.FileUriReader)
- || ((UriReader.FileUriReader) reader).exists(input);
+ if (reader instanceof UriReader.FileUriReader) {
+ return ((UriReader.FileUriReader) reader).exists(input);
+ }
+ if (reader instanceof UriReader.HttpUriReader) {
+ return ((UriReader.HttpUriReader) reader).exists(input);
+ }
+ return true;
}
private void readObject(ObjectInputStream in) throws IOException,
ClassNotFoundException {
diff --git
a/paimon-common/src/test/java/org/apache/paimon/utils/UriReaderFactoryTest.java
b/paimon-common/src/test/java/org/apache/paimon/utils/UriReaderFactoryTest.java
index 290b7f976e..3cf1cbaf3e 100644
---
a/paimon-common/src/test/java/org/apache/paimon/utils/UriReaderFactoryTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/utils/UriReaderFactoryTest.java
@@ -23,9 +23,16 @@ import org.apache.paimon.options.Options;
import org.apache.paimon.utils.UriReader.FileUriReader;
import org.apache.paimon.utils.UriReader.HttpUriReader;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
import java.nio.file.Files;
import static org.assertj.core.api.Assertions.assertThat;
@@ -38,6 +45,23 @@ public class UriReaderFactoryTest {
@TempDir java.nio.file.Path tempPath;
+ private HttpServer httpServer;
+ private int httpPort;
+
+ @BeforeEach
+ public void setUpHttpServer() throws Exception {
+ httpServer = HttpServer.create(new InetSocketAddress(0), 0);
+ httpPort = httpServer.getAddress().getPort();
+ httpServer.start();
+ }
+
+ @AfterEach
+ public void tearDownHttpServer() {
+ if (httpServer != null) {
+ httpServer.stop(0);
+ }
+ }
+
@Test
public void testCreateHttpUriReader() {
UriReader reader = factory.create("http://example.com/file.txt";);
@@ -93,8 +117,25 @@ public class UriReaderFactoryTest {
}
@Test
- public void testExistsSkipsHttpUriReader() throws Exception {
- assertThat(factory.exists("https://example.com/missing.txt";)).isTrue();
+ public void testExistsReturnsFalseForMissingHttpResource() throws
Exception {
+ registerHttpHandler(
+ "/missing.txt",
+ exchange -> {
+ sendResponse(exchange, 404, new byte[0]);
+ });
+
+ assertThat(factory.exists(httpUrl("/missing.txt"))).isFalse();
+ }
+
+ @Test
+ public void testExistsReturnsTrueForAvailableHttpResource() throws
Exception {
+ registerHttpHandler(
+ "/ok.txt",
+ exchange -> {
+ sendResponse(exchange, 200, "ok".getBytes());
+ });
+
+ assertThat(factory.exists(httpUrl("/ok.txt"))).isTrue();
}
@Test
@@ -104,4 +145,26 @@ public class UriReaderFactoryTest {
UriReader reader2 =
deserializedFactory.create("http://my_bucket/path/to/file2.txt";);
assertThat(reader1).isSameAs(reader2);
}
+
+ private void registerHttpHandler(String path,
com.sun.net.httpserver.HttpHandler handler) {
+ httpServer.createContext(path, handler);
+ }
+
+ private String httpUrl(String path) {
+ return "http://127.0.0.1:"; + httpPort + path;
+ }
+
+ private static void sendResponse(HttpExchange exchange, int statusCode,
byte[] body)
+ throws IOException {
+ boolean headRequest = "HEAD".equals(exchange.getRequestMethod());
+ long responseLength = headRequest ? -1 : body.length;
+ exchange.sendResponseHeaders(statusCode, responseLength);
+ if (!headRequest && body.length > 0) {
+ try (OutputStream outputStream = exchange.getResponseBody()) {
+ outputStream.write(body);
+ }
+ } else {
+ exchange.close();
+ }
+ }
}
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 26def0f56f..cd9d56ee65 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.types.DataTypeRoot;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.utils.UriReaderFactory;
@@ -41,6 +42,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
import static org.apache.paimon.flink.FlinkRowData.toFlinkRowKind;
import static org.apache.paimon.flink.LogicalTypeConversion.toDataType;
@@ -53,6 +57,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 Set<Integer> blobFields;
public FlinkRowWrapper(org.apache.flink.table.data.RowData row) {
this(row, null);
@@ -66,9 +71,28 @@ public class FlinkRowWrapper implements InternalRow {
org.apache.flink.table.data.RowData row,
CatalogContext catalogContext,
boolean checkBlobDescriptorExists) {
+ this(row, catalogContext, checkBlobDescriptorExists,
Collections.emptySet());
+ }
+
+ public FlinkRowWrapper(
+ org.apache.flink.table.data.RowData row,
+ CatalogContext catalogContext,
+ boolean checkBlobDescriptorExists,
+ Set<Integer> blobFields) {
this.row = row;
this.uriReaderFactory = new UriReaderFactory(catalogContext);
this.checkBlobDescriptorExists = checkBlobDescriptorExists;
+ this.blobFields = blobFields;
+ }
+
+ public static Set<Integer>
blobFieldIndexes(org.apache.paimon.types.RowType rowType) {
+ Set<Integer> result = new HashSet<>();
+ for (int i = 0; i < rowType.getFieldCount(); i++) {
+ if (rowType.getTypeAt(i).getTypeRoot() == DataTypeRoot.BLOB) {
+ result.add(i);
+ }
+ }
+ return result;
}
@Override
@@ -91,7 +115,10 @@ public class FlinkRowWrapper implements InternalRow {
if (row.isNullAt(pos)) {
return true;
}
- return checkBlobDescriptorExists && isMissingBlobDescriptor(pos,
row.getBinary(pos));
+ if (!checkBlobDescriptorExists || !blobFields.contains(pos)) {
+ return false;
+ }
+ return isMissingBlobDescriptor(pos, row.getBinary(pos));
}
@Override
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 ae8013b7e7..a3d0dd3b90 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
@@ -56,10 +56,12 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import static org.apache.paimon.CoreOptions.clusteringStrategy;
import static
org.apache.paimon.flink.FlinkConnectorOptions.CLUSTERING_SAMPLE_FACTOR;
@@ -262,6 +264,10 @@ public class FlinkSinkBuilder {
org.apache.paimon.types.RowType rowType,
CatalogContext catalogContext,
boolean checkBlobDescriptorExists) {
+ Set<Integer> blobFields =
+ checkBlobDescriptorExists
+ ? FlinkRowWrapper.blobFieldIndexes(rowType)
+ : Collections.emptySet();
SingleOutputStreamOperator<InternalRow> result =
input.map(
(MapFunction<RowData, InternalRow>)
@@ -269,7 +275,8 @@ public class FlinkSinkBuilder {
new FlinkRowWrapper(
r,
catalogContext,
-
checkBlobDescriptorExists))
+
checkBlobDescriptorExists,
+ blobFields))
.returns(
org.apache.paimon.flink.utils.InternalTypeInfo.fromRowType(
rowType));
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 1ca0ed0422..c53adecec1 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
@@ -23,11 +23,20 @@ import org.apache.paimon.data.Blob;
import org.apache.paimon.data.BlobDescriptor;
import org.apache.paimon.options.Options;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
import org.apache.flink.table.data.GenericRowData;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
import java.nio.file.Files;
+import java.util.Collections;
+import java.util.Set;
import static org.assertj.core.api.Assertions.assertThat;
@@ -36,6 +45,23 @@ public class FlinkRowWrapperTest {
@TempDir java.nio.file.Path tempPath;
+ private HttpServer httpServer;
+ private int httpPort;
+
+ @BeforeEach
+ public void setUpHttpServer() throws Exception {
+ httpServer = HttpServer.create(new InetSocketAddress(0), 0);
+ httpPort = httpServer.getAddress().getPort();
+ httpServer.start();
+ }
+
+ @AfterEach
+ public void tearDownHttpServer() {
+ if (httpServer != null) {
+ httpServer.stop(0);
+ }
+ }
+
@Test
public void testMissingBlobDescriptorIsNullWhenCheckingEnabled() {
java.nio.file.Path missing = tempPath.resolve("missing.blob");
@@ -59,6 +85,56 @@ public class FlinkRowWrapperTest {
assertThat(wrapper.getBlob(0).toData()).isEqualTo(bytes);
}
+ @Test
+ public void testMissingHttpBlobDescriptorWithNonBlobColumnBefore() throws
Exception {
+ httpServer.createContext(
+ "/missing.jpg",
+ exchange -> {
+ sendResponse(exchange, 404, new byte[0]);
+ });
+ GenericRowData row =
+ GenericRowData.of(
+ 1,
+ new BlobDescriptor("http://127.0.0.1:"; + httpPort +
"/missing.jpg", 0, 1)
+ .serialize());
+
+ FlinkRowWrapper wrapper = wrapper(row, true, Collections.singleton(1));
+
+ assertThat(wrapper.isNullAt(0)).isFalse();
+ assertThat(wrapper.getInt(0)).isEqualTo(1);
+ assertThat(wrapper.isNullAt(1)).isTrue();
+ }
+
+ @Test
+ public void testMissingHttpBlobDescriptorIsNullWhenCheckingEnabled()
throws Exception {
+ httpServer.createContext(
+ "/missing.jpg",
+ exchange -> {
+ sendResponse(exchange, 404, new byte[0]);
+ });
+ GenericRowData row = descriptorRow("http://127.0.0.1:"; + httpPort +
"/missing.jpg", 1);
+
+ FlinkRowWrapper wrapper = wrapper(row, true);
+
+ assertThat(wrapper.isNullAt(0)).isTrue();
+ }
+
+ @Test
+ public void testExistingHttpBlobDescriptorIsReadableWhenCheckingEnabled()
throws Exception {
+ byte[] bytes = new byte[] {1, 2, 3};
+ httpServer.createContext(
+ "/ok.jpg",
+ exchange -> {
+ sendResponse(exchange, 200, bytes);
+ });
+ GenericRowData row =
+ descriptorRow("http://127.0.0.1:"; + httpPort + "/ok.jpg",
bytes.length);
+
+ FlinkRowWrapper wrapper = wrapper(row, true);
+
+ assertThat(wrapper.isNullAt(0)).isFalse();
+ }
+
@Test
public void testMissingBlobDescriptorUsesDefaultBehaviorWithoutChecking() {
java.nio.file.Path missing = tempPath.resolve("missing.blob");
@@ -72,12 +148,34 @@ public class FlinkRowWrapperTest {
}
private GenericRowData descriptorRow(java.nio.file.Path path, long length)
{
- return GenericRowData.of(
- new BlobDescriptor(path.toUri().toString(), 0,
length).serialize());
+ return descriptorRow(path.toUri().toString(), length);
+ }
+
+ private GenericRowData descriptorRow(String uri, long length) {
+ return GenericRowData.of(new BlobDescriptor(uri, 0,
length).serialize());
+ }
+
+ private static void sendResponse(HttpExchange exchange, int statusCode,
byte[] body)
+ throws IOException {
+ boolean headRequest = "HEAD".equals(exchange.getRequestMethod());
+ long responseLength = headRequest ? -1 : body.length;
+ exchange.sendResponseHeaders(statusCode, responseLength);
+ if (!headRequest && body.length > 0) {
+ try (OutputStream outputStream = exchange.getResponseBody()) {
+ outputStream.write(body);
+ }
+ } else {
+ exchange.close();
+ }
}
private FlinkRowWrapper wrapper(GenericRowData row, boolean
checkBlobDescriptorExists) {
+ return wrapper(row, checkBlobDescriptorExists,
Collections.singleton(0));
+ }
+
+ private FlinkRowWrapper wrapper(
+ GenericRowData row, boolean checkBlobDescriptorExists,
Set<Integer> blobFields) {
return new FlinkRowWrapper(
- row, CatalogContext.create(new Options()),
checkBlobDescriptorExists);
+ row, CatalogContext.create(new Options()),
checkBlobDescriptorExists, blobFields);
}
}
