[format] Safely write NULL for BLOB body fetch failures

docs/generated/core_configuration.html

@@ -102,7 +102,7 @@
             <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>
+            <td>Whether to write NULL for a descriptor BLOB value when the referenced resource cannot be opened or fully read during Flink writes (e.g. invalid URI, HTTP errors other than 404, or a response body which fails before it is complete). When enabled, the fetched value is staged before it is appended to the managed BLOB file so that a body read failure cannot leave partial bytes in that file. HTTP 404 is handled by 'blob-write-null-on-missing-file'. Task cancellation and staging or final-output failures still fail the write. When false, the write fails when the descriptor is read.</td>
         </tr>
         <tr>
             <td><h5>blob-write-null-on-missing-file</h5></td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -2730,9 +2730,15 @@ public String toString() {
                     .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). "
+                                    + "referenced resource cannot be opened or fully read during "
+                                    + "Flink writes (e.g. invalid URI, HTTP errors other than 404, "
+                                    + "or a response body which fails before it is complete). When "
+                                    + "enabled, the fetched value is staged before it is appended to "
+                                    + "the managed BLOB file so that a body read failure cannot leave "
+                                    + "partial bytes in that file. "
                                     + "HTTP 404 is handled by 'blob-write-null-on-missing-file'. "
+                                    + "Task cancellation and staging or final-output failures still "
+                                    + "fail the write. "
                                     + "When false, the write fails when the descriptor is read.");
 
     public static final ConfigOption<Boolean> COMMIT_DISCARD_DUPLICATE_FILES =

paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java

@@ -55,6 +55,7 @@
 
 import javax.annotation.Nullable;
 
+import java.io.File;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -386,6 +387,13 @@ private RollingFileWriter<InternalRow, DataFileMeta> createRollingRowWriter() {
                 blobContext != null
                         || !fieldsInVectorFile(writeSchema, vectorFileFormat != null).isEmpty();
         if (hasDedicatedFields) {
+            @Nullable
+            File blobStagingTempDirectory =
+                    blobContext != null
+                                    && blobContext.writeNullOnFetchFailure()
+                                    && ioManager != null
+                            ? new File(ioManager.pickTempDir())
+                            : null;
             return new DedicatedFormatRollingFileWriter(
                     fileIO,
                     schemaId,
@@ -403,7 +411,8 @@ private RollingFileWriter<InternalRow, DataFileMeta> createRollingRowWriter() {
                     fileIndexOptions,
                     fileSource,
                     statsDenseStore,
-                    blobContext);
+                    blobContext,
+                    blobStagingTempDirectory);
         }
         return new RowDataRollingFileWriter(
                 fileIO,

paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java

@@ -44,6 +44,7 @@
 
 import javax.annotation.Nullable;
 
+import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -143,6 +144,46 @@ public DedicatedFormatRollingFileWriter(
             FileSource fileSource,
             boolean statsDenseStore,
             @Nullable BlobFileContext context) {
+        this(
+                fileIO,
+                schemaId,
+                fileFormat,
+                vectorFileFormat,
+                targetFileSize,
+                blobTargetFileSize,
+                vectorTargetFileSize,
+                targetFileRowNum,
+                writeSchema,
+                pathFactory,
+                seqNumCounterSupplier,
+                fileCompression,
+                statsCollectorFactories,
+                fileIndexOptions,
+                fileSource,
+                statsDenseStore,
+                context,
+                null);
+    }
+
+    DedicatedFormatRollingFileWriter(
+            FileIO fileIO,
+            long schemaId,
+            FileFormat fileFormat,
+            @Nullable FileFormat vectorFileFormat,
+            long targetFileSize,
+            long blobTargetFileSize,
+            long vectorTargetFileSize,
+            long targetFileRowNum,
+            RowType writeSchema,
+            DataFilePathFactory pathFactory,
+            Supplier<LongCounter> seqNumCounterSupplier,
+            String fileCompression,
+            StatsCollectorFactories statsCollectorFactories,
+            FileIndexOptions fileIndexOptions,
+            FileSource fileSource,
+            boolean statsDenseStore,
+            @Nullable BlobFileContext context,
+            @Nullable File blobStagingTempDirectory) {
         // Initialize basic fields
         Preconditions.checkArgument(
                 targetFileRowNum > 0,
@@ -210,7 +251,8 @@ public DedicatedFormatRollingFileWriter(
                                     context.writeNullOnMissingFile(),
                                     context.writeNullOnFetchFailure(),
                                     context.blobFetchMetricReporter(),
-                                    context.copyBufferSize());
+                                    context.copyBufferSize(),
+                                    blobStagingTempDirectory);
         } else {
             this.blobWriterFactory = null;
         }

paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java

@@ -40,6 +40,7 @@
 import javax.annotation.Nullable;
 
 import java.io.Closeable;
+import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
@@ -70,13 +71,50 @@ public MultipleBlobFileWriter(
             boolean writeNullOnFetchFailure,
             BlobFetchMetricReporter blobFetchMetricReporter,
             int copyBufferSize) {
+        this(
+                fileIO,
+                schemaId,
+                writeSchema,
+                pathFactory,
+                seqNumCounterSupplier,
+                fileSource,
+                asyncFileWrite,
+                statsDenseStore,
+                targetFileSize,
+                blobConsumer,
+                blobInlineFields,
+                writeNullOnMissingFile,
+                writeNullOnFetchFailure,
+                blobFetchMetricReporter,
+                copyBufferSize,
+                null);
+    }
+
+    MultipleBlobFileWriter(
+            FileIO fileIO,
+            long schemaId,
+            RowType writeSchema,
+            DataFilePathFactory pathFactory,
+            Supplier<LongCounter> seqNumCounterSupplier,
+            FileSource fileSource,
+            boolean asyncFileWrite,
+            boolean statsDenseStore,
+            long targetFileSize,
+            @Nullable BlobConsumer blobConsumer,
+            Set<String> blobInlineFields,
+            boolean writeNullOnMissingFile,
+            boolean writeNullOnFetchFailure,
+            BlobFetchMetricReporter blobFetchMetricReporter,
+            int copyBufferSize,
+            @Nullable File blobStagingTempDirectory) {
         RowType blobRowType = new RowType(fieldsInBlobFile(writeSchema, blobInlineFields));
         this.blobWriters = new ArrayList<>();
         for (String blobFieldName : blobRowType.getFieldNames()) {
             BlobFileFormat blobFileFormat = new BlobFileFormat(false, copyBufferSize);
             blobFileFormat.setWriteConsumer(blobConsumer);
             blobFileFormat.setWriteNullOnMissingFile(writeNullOnMissingFile);
             blobFileFormat.setWriteNullOnFetchFailure(writeNullOnFetchFailure);
+            blobFileFormat.setBlobStagingTempDirectory(blobStagingTempDirectory);
             blobFileFormat.setBlobFetchMetricReporter(blobFetchMetricReporter);
             blobWriters.add(
                     new BlobProjectedFileWriter(

paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java

@@ -26,6 +26,8 @@
 import org.apache.paimon.data.BinaryString;
 import org.apache.paimon.data.BinaryVector;
 import org.apache.paimon.data.BlobData;
+import org.apache.paimon.data.BlobDescriptor;
+import org.apache.paimon.data.BlobRef;
 import org.apache.paimon.data.GenericMap;
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.data.InternalMap;
@@ -41,6 +43,7 @@
 import org.apache.paimon.format.FormatReaderContext;
 import org.apache.paimon.format.SimpleColStats;
 import org.apache.paimon.format.SupportsFieldMetadata;
+import org.apache.paimon.fs.ByteArraySeekableStream;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.local.LocalFileIO;
 import org.apache.paimon.io.DataFileMeta;
@@ -95,7 +98,9 @@
 import java.util.TreeSet;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static org.apache.paimon.io.DataFileMeta.getMaxSequenceNumber;
 import static org.apache.paimon.stats.SimpleStats.EMPTY_STATS;
@@ -706,6 +711,77 @@ public void testSharedShreddingMapWithBlob(String fileFormat) throws Exception {
                                 1));
     }
 
+    @Test
+    public void testBlobStagingUsesIoManagerTempDirectory() throws Exception {
+        java.nio.file.Path stagingDirectory =
+                Files.createDirectory(tempDir.resolve("blob-staging"));
+        byte[] payload = new byte[2 * 1024 * 1024];
+        AtomicBoolean sawSpillFile = new AtomicBoolean();
+        BlobRef blob =
+                new BlobRef(
+                        ignored ->
+                                new ByteArraySeekableStream(payload) {
+                                    @Override
+                                    public int read(byte[] bytes, int offset, int length)
+                                            throws IOException {
+                                        if (containsRegularFile(stagingDirectory)) {
+                                            sawSpillFile.set(true);
+                                        }
+                                        return super.read(bytes, offset, length);
+                                    }
+                                },
+                        new BlobDescriptor("mem://io-manager-staging", 0, -1));
+
+        RowType writeType =
+                DataTypes.ROW(
+                        DataTypes.FIELD(0, "id", DataTypes.INT()),
+                        DataTypes.FIELD(1, "payload", DataTypes.BLOB()));
+        Options rawOptions = new Options();
+        rawOptions.set(CoreOptions.BLOB_WRITE_NULL_ON_FETCH_FAILURE, true);
+        CoreOptions options = new CoreOptions(rawOptions);
+
+        try (IOManager ioManager = IOManager.create(stagingDirectory.toString())) {
+            AppendOnlyWriter writer =
+                    new AppendOnlyWriter(
+                            LocalFileIO.create(),
+                            ioManager,
+                            SCHEMA_ID,
+                            FileFormat.fromIdentifier(AVRO, rawOptions),
+                            null,
+                            1024 * 1024L,
+                            1024 * 1024L,
+                            1024 * 1024L,
+                            Long.MAX_VALUE,
+                            writeType,
+                            null,
+                            -1L,
+                            new NoopCompactManager(),
+                            null,
+                            false,
+                            pathFactory,
+                            null,
+                            false,
+                            false,
+                            CoreOptions.FILE_COMPRESSION.defaultValue(),
+                            CompressOptions.defaultOptions(),
+                            new StatsCollectorFactories(options),
+                            MemorySize.MAX_VALUE,
+                            new FileIndexOptions(),
+                            true,
+                            false,
+                            options.dataEvolutionEnabled(),
+                            null,
+                            BlobFileContext.create(writeType, options));
+
+            writer.write(GenericRow.of(1, blob));
+            writer.prepareCommit(true);
+            writer.close();
+        }
+
+        assertThat(sawSpillFile).isTrue();
+        assertThat(containsRegularFile(stagingDirectory)).isFalse();
+    }
+
     @ParameterizedTest(name = "{0}")
     @ValueSource(strings = {CoreOptions.FILE_FORMAT_PARQUET, CoreOptions.FILE_FORMAT_ORC})
     public void testSharedShreddingMapAllowsForceBufferSpill(String fileFormat) throws Exception {
@@ -1012,6 +1088,12 @@ private DataFilePathFactory createPathFactory() {
         return createPathFactory(CoreOptions.FILE_FORMAT_AVRO);
     }
 
+    private static boolean containsRegularFile(java.nio.file.Path directory) throws IOException {
+        try (Stream<java.nio.file.Path> files = Files.list(directory)) {
+            return files.anyMatch(Files::isRegularFile);
+        }
+    }
+
     private DataFilePathFactory createPathFactory(String fileFormat) {
         return new DataFilePathFactory(
                 new Path(tempDir + "/dt=" + PART + "/bucket-0"),