[flink] Avoid redundant HTTP BLOB existence probes

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 FlinkRowWrapper(
                 new UriReaderFactory(catalogContext),
                 checkBlobDescriptorExists,
                 writeNullOnFetchFailure,
-                blobFields);
+                blobFields,
+                Collections.emptySet());
     }
 
     public static FlinkRowWrapper fromUriReaderFactory(
@@ -120,25 +122,44 @@ public static FlinkRowWrapper fromUriReaderFactory(
             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(
             org.apache.flink.table.data.RowData row,
             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 @@ private boolean isMissingBlobDescriptor(int pos, byte[] bytes) {
         }
 
         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 @@ private void logMissingDescriptor(int pos, BlobDescriptor descriptor) {
     }
 
     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());
     }
 
     /**

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java

@@ -219,13 +219,21 @@ public DataStreamSink<?> build() {
         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,15 +288,17 @@ public static DataStream<InternalRow> mapToInternalRow(
                 rowType,
                 new UriReaderFactory(catalogContext),
                 checkBlobDescriptorExists,
-                writeNullOnFetchFailure);
+                writeNullOnFetchFailure,
+                Collections.emptySet());
     }
 
     private static DataStream<InternalRow> mapToInternalRowWithUriReaderFactory(
             DataStream<RowData> input,
             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,14 +312,23 @@ private static DataStream<InternalRow> mapToInternalRowWithUriReaderFactory(
                                                         uriReaderFactory,
                                                         checkBlobDescriptorExists,
                                                         writeNullOnFetchFailure,
-                                                        blobFields))
+                                                        blobFields,
+                                                        materializedBlobFields))
                         .returns(
                                 org.apache.paimon.flink.utils.InternalTypeInfo.fromRowType(
                                         rowType));
         forwardParallelism(result, input);
         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) {