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 fe081c02af [flink] Avoid redundant HTTP BLOB existence probes (#9181)
fe081c02af is described below
commit fe081c02afaf48e7942886b11888194f782fee15
Author: wangwj <[email protected]>
AuthorDate: Thu Aug 13 22:02:07 2026 +0800
[flink] Avoid redundant HTTP BLOB existence probes (#9181)
---
.../org/apache/paimon/flink/FlinkRowWrapper.java | 38 +++-
.../apache/paimon/flink/sink/FlinkSinkBuilder.java | 27 ++-
.../org/apache/paimon/flink/BlobTableITCase.java | 211 ++++++++++++++++++++-
.../apache/paimon/flink/FlinkRowWrapperTest.java | 91 +++++++--
4 files changed, 337 insertions(+), 30 deletions(-)
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 079a192e7d..74487659a0 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
@@ -63,6 +63,7 @@ public class FlinkRowWrapper implements InternalRow {
private final boolean checkBlobDescriptorExists;
private final boolean writeNullOnFetchFailure;
private final Set<Integer> blobFields;
+ private final Set<Integer> materializedBlobFields;
public FlinkRowWrapper(org.apache.flink.table.data.RowData row) {
this(row, null);
@@ -111,7 +112,8 @@ public class FlinkRowWrapper implements InternalRow {
new UriReaderFactory(catalogContext),
checkBlobDescriptorExists,
writeNullOnFetchFailure,
- blobFields);
+ blobFields,
+ Collections.emptySet());
}
public static FlinkRowWrapper fromUriReaderFactory(
@@ -120,12 +122,29 @@ public class FlinkRowWrapper implements InternalRow {
boolean checkBlobDescriptorExists,
boolean writeNullOnFetchFailure,
Set<Integer> blobFields) {
+ return fromUriReaderFactory(
+ row,
+ uriReaderFactory,
+ checkBlobDescriptorExists,
+ writeNullOnFetchFailure,
+ blobFields,
+ Collections.emptySet());
+ }
+
+ public static FlinkRowWrapper fromUriReaderFactory(
+ org.apache.flink.table.data.RowData row,
+ UriReaderFactory uriReaderFactory,
+ boolean checkBlobDescriptorExists,
+ boolean writeNullOnFetchFailure,
+ Set<Integer> blobFields,
+ Set<Integer> materializedBlobFields) {
return new FlinkRowWrapper(
row,
uriReaderFactory,
checkBlobDescriptorExists,
writeNullOnFetchFailure,
- blobFields);
+ blobFields,
+ materializedBlobFields);
}
private FlinkRowWrapper(
@@ -133,12 +152,14 @@ public class FlinkRowWrapper implements InternalRow {
UriReaderFactory uriReaderFactory,
boolean checkBlobDescriptorExists,
boolean writeNullOnFetchFailure,
- Set<Integer> blobFields) {
+ Set<Integer> blobFields,
+ Set<Integer> materializedBlobFields) {
this.row = row;
this.uriReaderFactory = uriReaderFactory;
this.checkBlobDescriptorExists = checkBlobDescriptorExists;
this.writeNullOnFetchFailure = writeNullOnFetchFailure;
this.blobFields = blobFields;
+ this.materializedBlobFields = materializedBlobFields;
}
public static Set<Integer>
blobFieldIndexes(org.apache.paimon.types.RowType rowType) {
@@ -253,6 +274,14 @@ public class FlinkRowWrapper implements InternalRow {
}
BlobDescriptor descriptor = BlobDescriptor.deserialize(bytes);
+ // Materialized BLOB fields are copied into managed blob files. Their
writer has to open
+ // HTTP resources and already maps HTTP 404 and other open failures to
NULL according to
+ // the two write-null options. Avoid a redundant HEAD / range-GET
existence check before
+ // that required GET. Inline descriptor and view fields keep the
existence check because
+ // they have no later writer fetch.
+ if (materializedBlobFields.contains(pos) &&
isHttpUri(descriptor.uri())) {
+ return false;
+ }
return !descriptorFileExists(pos, descriptor);
}
@@ -301,7 +330,8 @@ public class FlinkRowWrapper implements InternalRow {
}
private static boolean isHttpUri(String uri) {
- return uri.startsWith("http://";) || uri.startsWith("https://";);
+ return uri.regionMatches(true, 0, "http://";, 0, "http://".length())
+ || uri.regionMatches(true, 0, "https://";, 0,
"https://".length());
}
/**
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 656a68db6e..8bea0b5acf 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
@@ -219,13 +219,21 @@ public class FlinkSinkBuilder {
UriReaderFactory readerFactoryForDescriptor =
BlobDescriptorReaderFactory.create(table);
blobDescriptorReaderFactory = readerFactoryForDescriptor;
+ // Primary-key tables externalize BLOBs after merging records, and
that path does not apply
+ // the write-null fallback while fetching descriptors. Retain the
existence preflight there.
+ Set<Integer> materializedBlobFields =
+ table.schema().primaryKeys().isEmpty()
+ ? materializedBlobFieldIndexes(
+ table.rowType(),
table.coreOptions().blobInlineField())
+ : Collections.emptySet();
DataStream<InternalRow> input =
mapToInternalRowWithUriReaderFactory(
this.input,
table.rowType(),
readerFactoryForDescriptor,
table.coreOptions().blobWriteNullOnMissingFile(),
- table.coreOptions().blobWriteNullOnFetchFailure());
+ table.coreOptions().blobWriteNullOnFetchFailure(),
+ materializedBlobFields);
if (table.coreOptions().localMergeEnabled() &&
table.schema().primaryKeys().size() > 0) {
SingleOutputStreamOperator<InternalRow> newInput =
input.forward()
@@ -280,7 +288,8 @@ public class FlinkSinkBuilder {
rowType,
new UriReaderFactory(catalogContext),
checkBlobDescriptorExists,
- writeNullOnFetchFailure);
+ writeNullOnFetchFailure,
+ Collections.emptySet());
}
private static DataStream<InternalRow>
mapToInternalRowWithUriReaderFactory(
@@ -288,7 +297,8 @@ public class FlinkSinkBuilder {
org.apache.paimon.types.RowType rowType,
UriReaderFactory uriReaderFactory,
boolean checkBlobDescriptorExists,
- boolean writeNullOnFetchFailure) {
+ boolean writeNullOnFetchFailure,
+ Set<Integer> materializedBlobFields) {
Set<Integer> blobFields =
checkBlobDescriptorExists
? FlinkRowWrapper.blobFieldIndexes(rowType)
@@ -302,7 +312,8 @@ public class FlinkSinkBuilder {
uriReaderFactory,
checkBlobDescriptorExists,
writeNullOnFetchFailure,
- blobFields))
+ blobFields,
+
materializedBlobFields))
.returns(
org.apache.paimon.flink.utils.InternalTypeInfo.fromRowType(
rowType));
@@ -310,6 +321,14 @@ public class FlinkSinkBuilder {
return result;
}
+ private static Set<Integer> materializedBlobFieldIndexes(
+ org.apache.paimon.types.RowType rowType, Set<String>
inlineBlobFields) {
+ Set<Integer> materializedBlobFields =
FlinkRowWrapper.blobFieldIndexes(rowType);
+ materializedBlobFields.removeIf(
+ pos ->
inlineBlobFields.contains(rowType.getFields().get(pos).name()));
+ return materializedBlobFields;
+ }
+
protected DataStreamSink<?> buildDynamicBucketSink(
DataStream<InternalRow> input, boolean globalIndex) {
if (compactSink && !globalIndex) {
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 37dcf37bd3..839483f970 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
@@ -37,10 +37,14 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.UriReader;
import org.apache.paimon.utils.UriReaderFactory;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.RecordedRequest;
import org.apache.flink.table.planner.factories.TestValuesTableFactory;
import org.apache.flink.types.Row;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.OutputStream;
import java.net.URI;
@@ -50,6 +54,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
+import java.util.concurrent.TimeUnit;
import static org.apache.paimon.flink.LogicalTypeConversion.toLogicalType;
import static org.assertj.core.api.Assertions.assertThat;
@@ -1166,6 +1171,113 @@ public class BlobTableITCase extends CatalogITCaseBase {
assertThat(result.get(0).getField(2)).isNull();
}
+ @Test
+ public void testWriteHttpNotFoundWithMissingFileUsesSingleGet() throws
Exception {
+ TestHttpWebServer httpServer = new TestHttpWebServer("/missing_blob");
+ httpServer.start();
+ try {
+ String httpUrl = httpServer.getBaseUrl();
+ httpServer.enqueueResponse("", 404);
+
+ tEnv.executeSql(
+ "CREATE TABLE missing_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')");
+
+ batchSql(
+ "INSERT INTO missing_blob_table VALUES"
+ + " (1, 'missing', sys.path_to_descriptor('"
+ + httpUrl
+ + "'))");
+
+ List<Row> result = batchSql("SELECT * FROM missing_blob_table");
+ assertThat(result).containsExactly(Row.of(1, "missing", null));
+ assertOnlyFullGetRequests(httpServer, 1);
+ } finally {
+ httpServer.stop();
+ }
+ }
+
+ @Test
+ public void
testPrimaryKeyWriteHttpNotFoundWithMissingFileRetainsPreflight() throws
Exception {
+ TestHttpWebServer httpServer = new
TestHttpWebServer("/missing_pk_blob");
+ httpServer.start();
+ try {
+ String httpUrl = httpServer.getBaseUrl();
+ httpServer.enqueueResponse("", 404);
+ httpServer.enqueueResponse("", 404);
+
+ tEnv.executeSql(
+ "CREATE TABLE missing_pk_blob_table ("
+ + "id INT, picture BYTES, PRIMARY KEY (id) NOT
ENFORCED)"
+ + " WITH ('bucket'='1',"
+ + " 'blob-field'='picture',"
+ + " 'blob-as-descriptor'='true',"
+ + " 'blob-write-null-on-missing-file'='true')");
+
+ batchSql(
+ "INSERT INTO missing_pk_blob_table VALUES"
+ + " (1, sys.path_to_descriptor('"
+ + httpUrl
+ + "'))");
+
+ assertThat(batchSql("SELECT * FROM missing_pk_blob_table"))
+ .containsExactly(Row.of(1, null));
+
+ RecordedRequest headRequest = httpServer.takeRequest(1,
TimeUnit.SECONDS);
+ assertThat(headRequest).isNotNull();
+ assertThat(headRequest.getMethod()).isEqualTo("HEAD");
+
+ RecordedRequest rangeRequest = httpServer.takeRequest(1,
TimeUnit.SECONDS);
+ assertThat(rangeRequest).isNotNull();
+ assertThat(rangeRequest.getMethod()).isEqualTo("GET");
+ assertThat(rangeRequest.getHeader("Range")).isEqualTo("bytes=0-0");
+ assertThat(httpServer.takeRequest(100,
TimeUnit.MILLISECONDS)).isNull();
+ } finally {
+ httpServer.stop();
+ }
+ }
+
+ @Test
+ public void testWriteExistingHttpBlobWithMissingFileUsesSingleGet() throws
Exception {
+ TestHttpWebServer httpServer = new TestHttpWebServer("/existing_blob");
+ httpServer.start();
+ try {
+ String blobContent = "hello-http-blob";
+ String httpUrl = httpServer.getBaseUrl();
+ httpServer.enqueueResponse(blobContent, 200);
+
+ tEnv.executeSql(
+ "CREATE TABLE existing_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')");
+
+ batchSql(
+ "INSERT INTO existing_blob_table VALUES"
+ + " (1, 'existing', sys.path_to_descriptor('"
+ + httpUrl
+ + "'))");
+
+ batchSql("ALTER TABLE existing_blob_table SET
('blob-as-descriptor'='false')");
+ List<Row> result = batchSql("SELECT * FROM existing_blob_table");
+ assertThat(result)
+ .containsExactly(
+ Row.of(
+ 1,
+ "existing",
+
blobContent.getBytes(java.nio.charset.StandardCharsets.UTF_8)));
+ assertOnlyFullGetRequests(httpServer, 1);
+ } finally {
+ httpServer.stop();
+ }
+ }
+
@Test
public void
testWriteHttpBadRequestWritesNullWithMissingFileAndFetchFailure() throws
Exception {
TestHttpWebServer httpServer = new
TestHttpWebServer("/combined_bad_request_blob");
@@ -1173,7 +1285,6 @@ public class BlobTableITCase extends CatalogITCaseBase {
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)"
@@ -1195,11 +1306,109 @@ public class BlobTableITCase extends CatalogITCaseBase
{
assertThat(result.get(0).getField(0)).isEqualTo(1);
assertThat(result.get(0).getField(1)).isEqualTo("bad-request");
assertThat(result.get(0).getField(2)).isNull();
+ assertOnlyFullGetRequests(httpServer, 1);
+ } finally {
+ httpServer.stop();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {429, 503})
+ public void testWriteRetryableHttpStatusUsesOnlyFullGet(int statusCode)
throws Exception {
+ TestHttpWebServer httpServer = new TestHttpWebServer("/retryable_" +
statusCode + "_blob");
+ httpServer.start();
+ try {
+ String httpUrl = httpServer.getBaseUrl();
+ MockResponse retryableResponse =
+ httpServer.generateMockResponse("",
statusCode).addHeader("Retry-After", "1");
+ httpServer.enqueueResponse(retryableResponse);
+ String blobContent = "retried-http-blob";
+ httpServer.enqueueResponse(blobContent, 200);
+
+ String tableName = "retryable_" + statusCode + "_blob_table";
+ tEnv.executeSql(
+ "CREATE TABLE "
+ + tableName
+ + " (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 "
+ + tableName
+ + " VALUES (1, 'retryable',
sys.path_to_descriptor('"
+ + httpUrl
+ + "'))");
+
+ batchSql("ALTER TABLE " + tableName + " SET
('blob-as-descriptor'='false')");
+ List<Row> result = batchSql("SELECT * FROM " + tableName);
+ assertThat(result)
+ .containsExactly(
+ Row.of(
+ 1,
+ "retryable",
+
blobContent.getBytes(java.nio.charset.StandardCharsets.UTF_8)));
+ assertOnlyFullGetRequests(httpServer, 2);
} finally {
httpServer.stop();
}
}
+ @Test
+ public void testInlineHttpDescriptorRetainsMissingFileCheck() throws
Exception {
+ TestHttpWebServer httpServer = new
TestHttpWebServer("/missing_inline_blob");
+ httpServer.start();
+ try {
+ String httpUrl = httpServer.getBaseUrl();
+ httpServer.enqueueResponse("", 404);
+ httpServer.enqueueResponse("", 404);
+
+ tEnv.executeSql(
+ "CREATE TABLE missing_inline_blob_table (id INT, picture
BYTES)"
+ + " WITH ('row-tracking.enabled'='true',"
+ + " 'data-evolution.enabled'='true',"
+ + " 'blob-descriptor-field'='picture',"
+ + " 'blob-as-descriptor'='true',"
+ + " 'blob-write-null-on-missing-file'='true')");
+
+ batchSql(
+ "INSERT INTO missing_inline_blob_table VALUES"
+ + " (1, sys.path_to_descriptor('"
+ + httpUrl
+ + "'))");
+
+ List<Row> result = batchSql("SELECT * FROM
missing_inline_blob_table");
+ assertThat(result).containsExactly(Row.of(1, null));
+
+ RecordedRequest headRequest = httpServer.takeRequest(1,
TimeUnit.SECONDS);
+ assertThat(headRequest).isNotNull();
+ assertThat(headRequest.getMethod()).isEqualTo("HEAD");
+
+ RecordedRequest rangeRequest = httpServer.takeRequest(1,
TimeUnit.SECONDS);
+ assertThat(rangeRequest).isNotNull();
+ assertThat(rangeRequest.getMethod()).isEqualTo("GET");
+ assertThat(rangeRequest.getHeader("Range")).isEqualTo("bytes=0-0");
+ assertThat(httpServer.takeRequest(100,
TimeUnit.MILLISECONDS)).isNull();
+ } finally {
+ httpServer.stop();
+ }
+ }
+
+ private static void assertOnlyFullGetRequests(
+ TestHttpWebServer httpServer, int expectedRequestCount) throws
Exception {
+ for (int i = 0; i < expectedRequestCount; i++) {
+ RecordedRequest request = httpServer.takeRequest(1,
TimeUnit.SECONDS);
+ assertThat(request).isNotNull();
+ assertThat(request.getMethod()).isEqualTo("GET");
+ assertThat(request.getHeader("Range")).isNull();
+ }
+ assertThat(httpServer.takeRequest(100,
TimeUnit.MILLISECONDS)).isNull();
+ }
+
@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 060c90d309..e1fa2a7a92 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,9 +39,9 @@ import java.net.InetSocketAddress;
import java.nio.file.Files;
import java.util.Collections;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link FlinkRowWrapper}. */
public class FlinkRowWrapperTest {
@@ -50,6 +50,7 @@ public class FlinkRowWrapperTest {
private HttpServer httpServer;
private int httpPort;
+ private final AtomicInteger httpRequestCount = new AtomicInteger();
@BeforeEach
public void setUpHttpServer() throws Exception {
@@ -113,10 +114,11 @@ public class FlinkRowWrapperTest {
}
@Test
- public void testMissingHttpBlobDescriptorWithNonBlobColumnBefore() throws
Exception {
+ public void
testHttpBlobDescriptorWithNonBlobColumnBeforeSkipsExistsCheck() throws
Exception {
httpServer.createContext(
"/missing.jpg",
exchange -> {
+ httpRequestCount.incrementAndGet();
sendResponse(exchange, 404, new byte[0]);
});
GenericRowData row =
@@ -129,29 +131,33 @@ public class FlinkRowWrapperTest {
assertThat(wrapper.isNullAt(0)).isFalse();
assertThat(wrapper.getInt(0)).isEqualTo(1);
- assertThat(wrapper.isNullAt(1)).isTrue();
+ assertThat(wrapper.isNullAt(1)).isFalse();
+ assertThat(httpRequestCount).hasValue(0);
}
@Test
- public void testMissingHttpBlobDescriptorIsNullWhenCheckingEnabled()
throws Exception {
+ public void testMissingHttpBlobDescriptorIsDeferredToWriter() throws
Exception {
httpServer.createContext(
"/missing.jpg",
exchange -> {
+ httpRequestCount.incrementAndGet();
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();
+ assertThat(wrapper.isNullAt(0)).isFalse();
+ assertThat(httpRequestCount).hasValue(0);
}
@Test
- public void testExistingHttpBlobDescriptorIsReadableWhenCheckingEnabled()
throws Exception {
+ public void testExistingHttpBlobDescriptorSkipsExistsCheck() throws
Exception {
byte[] bytes = new byte[] {1, 2, 3};
httpServer.createContext(
"/ok.jpg",
exchange -> {
+ httpRequestCount.incrementAndGet();
sendResponse(exchange, 200, bytes);
});
GenericRowData row =
@@ -160,6 +166,41 @@ public class FlinkRowWrapperTest {
FlinkRowWrapper wrapper = wrapper(row, true);
assertThat(wrapper.isNullAt(0)).isFalse();
+ assertThat(httpRequestCount).hasValue(0);
+ }
+
+ @Test
+ public void testUppercaseHttpSchemeSkipsExistsCheck() throws Exception {
+ httpServer.createContext(
+ "/missing.jpg",
+ exchange -> {
+ httpRequestCount.incrementAndGet();
+ 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)).isFalse();
+ assertThat(httpRequestCount).hasValue(0);
+ }
+
+ @Test
+ public void testInlineHttpBlobDescriptorRetainsExistsCheck() throws
Exception {
+ httpServer.createContext(
+ "/missing-inline.jpg",
+ exchange -> {
+ httpRequestCount.incrementAndGet();
+ sendResponse(exchange, 404, new byte[0]);
+ });
+ GenericRowData row =
+ descriptorRow("http://127.0.0.1:"; + httpPort +
"/missing-inline.jpg", 1);
+
+ FlinkRowWrapper wrapper =
+ wrapper(row, true, false, Collections.singleton(0),
Collections.emptySet());
+
+ assertThat(wrapper.isNullAt(0)).isTrue();
+ assertThat(httpRequestCount).hasValue(2);
}
@Test
@@ -185,23 +226,21 @@ public class FlinkRowWrapperTest {
}
@Test
- public void testInvalidUriThrowsWhenFetchFailureDisabled() {
+ public void testInvalidHttpUriIsDeferredToWriterWhenFetchFailureDisabled()
{
GenericRowData row =
descriptorRow("https://img.alicdn.com/imgextra/##1304008055350781673";, 1);
FlinkRowWrapper wrapper = wrapper(row, true, false);
- assertThatThrownBy(() -> wrapper.isNullAt(0))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("Invalid URI")
- .hasMessageNotContaining("1304008055350781673");
+ assertThat(wrapper.isNullAt(0)).isFalse();
}
@Test
- public void testHttpBadRequestDefersExistsCheckWhenFetchFailureEnabled()
throws Exception {
+ public void testHttpBadRequestSkipsExistsCheckWhenFetchFailureEnabled()
throws Exception {
httpServer.createContext(
"/bad.jpg",
exchange -> {
+ httpRequestCount.incrementAndGet();
sendResponse(exchange, 400, new byte[0]);
});
GenericRowData row = descriptorRow("http://127.0.0.1:"; + httpPort +
"/bad.jpg", 1);
@@ -209,24 +248,23 @@ public class FlinkRowWrapperTest {
FlinkRowWrapper wrapper = wrapper(row, true, true);
assertThat(wrapper.isNullAt(0)).isFalse();
+ assertThat(httpRequestCount).hasValue(0);
}
@Test
- public void testHttpBadRequestThrowsWhenFetchFailureDisabled() throws
Exception {
+ public void testHttpBadRequestIsDeferredToWriterWhenFetchFailureDisabled()
throws Exception {
httpServer.createContext(
"/bad.jpg",
exchange -> {
+ httpRequestCount.incrementAndGet();
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");
+ assertThat(wrapper.isNullAt(0)).isFalse();
+ assertThat(httpRequestCount).hasValue(0);
}
private GenericRowData descriptorRow(java.nio.file.Path path, long length)
{
@@ -273,11 +311,22 @@ public class FlinkRowWrapperTest {
boolean checkBlobDescriptorExists,
boolean writeNullOnFetchFailure,
Set<Integer> blobFields) {
- return new FlinkRowWrapper(
+ return wrapper(
+ row, checkBlobDescriptorExists, writeNullOnFetchFailure,
blobFields, blobFields);
+ }
+
+ private FlinkRowWrapper wrapper(
+ GenericRowData row,
+ boolean checkBlobDescriptorExists,
+ boolean writeNullOnFetchFailure,
+ Set<Integer> blobFields,
+ Set<Integer> materializedBlobFields) {
+ return FlinkRowWrapper.fromUriReaderFactory(
row,
- CatalogContext.create(new Options()),
+ new UriReaderFactory(CatalogContext.create(new Options())),
checkBlobDescriptorExists,
writeNullOnFetchFailure,
- blobFields);
+ blobFields,
+ materializedBlobFields);
}
}
