diff --git a/cmake_modules/ThirdpartyToolchain.cmake b/cmake_modules/ThirdpartyToolchain.cmake
index ba46ca185..9d5bf29fc 100644
--- a/cmake_modules/ThirdpartyToolchain.cmake
+++ b/cmake_modules/ThirdpartyToolchain.cmake
@@ -1636,13 +1636,16 @@ macro(build_arrow)
 
     target_link_libraries(arrow_dataset INTERFACE arrow_acero)
 
+    # libarrow.a calls dlsym; keep ${CMAKE_DL_LIBS} in the interface so -ldl is placed
+    # after libarrow.a on linkers that resolve symbols strictly left-to-right.
     target_link_libraries(arrow
                           INTERFACE zstd
                                     snappy
                                     lz4
                                     zlib
                                     re2::re2
-                                    arrow_bundled_dependencies)
+                                    arrow_bundled_dependencies
+                                    ${CMAKE_DL_LIBS})
 
     target_link_libraries(parquet
                           INTERFACE zstd
diff --git a/include/paimon/defs.h b/include/paimon/defs.h
index 772dfee5a..8b4fa8c8e 100644
--- a/include/paimon/defs.h
+++ b/include/paimon/defs.h
@@ -439,6 +439,15 @@ struct PAIMON_EXPORT Options {
     /// Blob View is enabled, cpp paimon cannot automatically obtain the upstream table warehouse
     /// path and requires manual configuration by the user. No default value.
     static const char BLOB_VIEW_UPSTREAM_WAREHOUSE[];
+    /// "blob-write-null-on-missing-file" - Whether to write NULL for a descriptor BLOB value when
+    /// the referenced file does not exist at write time. When false, the write fails when the
+    /// descriptor is read. Default value is "false".
+    static const char BLOB_WRITE_NULL_ON_MISSING_FILE[];
+    /// "blob-write-null-on-fetch-failure" - Whether to write NULL for a descriptor BLOB value when
+    /// the referenced data cannot be fetched at write time (e.g. invalid descriptor or invalid
+    /// offset). A missing file is handled by "blob-write-null-on-missing-file". When false, the
+    /// write fails when the descriptor is read. Default value is "false".
+    static const char BLOB_WRITE_NULL_ON_FETCH_FAILURE[];
     /// "global-index.enabled" - Whether to enable global index for scan. Default value is "true".
     static const char GLOBAL_INDEX_ENABLED[];
     /// "global-index.thread-num" - The maximum number of concurrent scanner for global index. No
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index 3841728e5..46cc13854 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -791,18 +791,29 @@ if(PAIMON_BUILD_TESTS)
                     ${TEST_STATIC_LINK_LIBS}
                     ${GTEST_LINK_TOOLCHAIN})
 
+    # jindo_utils_test only checks the in-memory status conversion and does not need an
+    # OSS cluster, so it runs whenever jindo is built. The other jindo tests need real
+    # OSS access and stay disabled.
+    set(FS_TEST_JINDO_SOURCES)
+    if(PAIMON_ENABLE_JINDO)
+        list(APPEND FS_TEST_JINDO_SOURCES fs/jindo/jindo_utils_test.cpp)
+    endif()
+
     add_paimon_test(fs_test
                     SOURCES
                     common/fs/file_system_test.cpp
                     common/fs/resolving_file_system_test.cpp
                     fs/local/local_file_test.cpp
+                    ${FS_TEST_JINDO_SOURCES}
                     # fs/jindo/jindo_file_system_factory_test.cpp
                     # fs/jindo/jindo_file_system_test.cpp
                     STATIC_LINK_LIBS
                     paimon_shared
                     ${PAIMON_LOCAL_FILE_SYSTEM_STATIC_LINK_LIBS}
-                    # ${PAIMON_JINDO_FILE_SYSTEM_STATIC_LINK_LIBS}
+                    ${PAIMON_JINDO_FILE_SYSTEM_STATIC_LINK_LIBS}
                     test_utils_static
-                    ${GTEST_LINK_TOOLCHAIN})
+                    ${GTEST_LINK_TOOLCHAIN}
+                    EXTRA_INCLUDES
+                    ${JINDOSDK_INCLUDE_DIR})
 
 endif()
diff --git a/src/paimon/common/defs.cpp b/src/paimon/common/defs.cpp
index 501068f32..cf182a53c 100644
--- a/src/paimon/common/defs.cpp
+++ b/src/paimon/common/defs.cpp
@@ -108,6 +108,8 @@ const char Options::FALLBACK_BLOB_DESCRIPTOR_FIELD[] = "blob.stored-descriptor-f
 const char Options::BLOB_VIEW_FIELD[] = "blob-view-field";
 const char Options::BLOB_VIEW_RESOLVE_ENABLED[] = "blob-view.resolve.enabled";
 const char Options::BLOB_VIEW_UPSTREAM_WAREHOUSE[] = "blob-view-upstream-warehouse";
+const char Options::BLOB_WRITE_NULL_ON_MISSING_FILE[] = "blob-write-null-on-missing-file";
+const char Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE[] = "blob-write-null-on-fetch-failure";
 const char Options::GLOBAL_INDEX_ENABLED[] = "global-index.enabled";
 const char Options::GLOBAL_INDEX_THREAD_NUM[] = "global-index.thread-num";
 const char Options::GLOBAL_INDEX_EXTERNAL_PATH[] = "global-index.external-path";
diff --git a/src/paimon/format/blob/blob_file_batch_reader_test.cpp b/src/paimon/format/blob/blob_file_batch_reader_test.cpp
index 2a06b0b35..4e53dfb4c 100644
--- a/src/paimon/format/blob/blob_file_batch_reader_test.cpp
+++ b/src/paimon/format/blob/blob_file_batch_reader_test.cpp
@@ -234,8 +234,10 @@ TEST_P(BlobFileBatchReaderTest, EmptyFile) {
                          file_system->Create(dir->Str() + "/file.blob", /*overwrite=*/true));
     std::shared_ptr<arrow::Field> blob_field = BlobUtils::ToArrowField("blob_col");
     auto struct_type = arrow::struct_({blob_field});
-    ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer,
-                         BlobFormatWriter::Create(output_stream, struct_type, file_system, pool_));
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<BlobFormatWriter> writer,
+        BlobFormatWriter::Create(output_stream, struct_type, /*write_null_on_missing_file=*/false,
+                                 /*write_null_on_fetch_failure=*/false, file_system, pool_));
 
     ASSERT_OK(writer->Flush());
     ASSERT_OK(writer->Finish());
diff --git a/src/paimon/format/blob/blob_file_format_factory_test.cpp b/src/paimon/format/blob/blob_file_format_factory_test.cpp
index 6409ea85a..f1e7a84ae 100644
--- a/src/paimon/format/blob/blob_file_format_factory_test.cpp
+++ b/src/paimon/format/blob/blob_file_format_factory_test.cpp
@@ -16,8 +16,24 @@
 
 #include "paimon/format/blob/blob_file_format_factory.h"
 
+#include <map>
+#include <memory>
+#include <string>
+
+#include "arrow/api.h"
+#include "arrow/c/bridge.h"
 #include "gtest/gtest.h"
+#include "paimon/common/data/blob_utils.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/data/blob.h"
+#include "paimon/defs.h"
+#include "paimon/format/file_format.h"
+#include "paimon/format/file_format_factory.h"
+#include "paimon/format/format_writer.h"
+#include "paimon/format/writer_builder.h"
+#include "paimon/fs/local/local_file_system.h"
 #include "paimon/status.h"
+#include "paimon/testing/utils/test_helper.h"
 #include "paimon/testing/utils/testharness.h"
 
 namespace paimon::blob::test {
@@ -29,4 +45,46 @@ TEST(BlobFileFormatFactoryTest, TestIdentifier) {
     ASSERT_EQ(file_format->Identifier(), "blob");
 }
 
+TEST(BlobFileFormatFactoryTest, TestWriteNullOptionPropagation) {
+    // Verifies the option flows through the production path
+    // FileFormatFactory::Get -> BlobFileFormat -> BlobWriterBuilder -> BlobFormatWriter.
+    std::unique_ptr<paimon::test::UniqueTestDirectory> dir =
+        paimon::test::UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    std::shared_ptr<FileSystem> fs = std::make_shared<LocalFileSystem>();
+    auto struct_type = arrow::struct_({BlobUtils::ToArrowField("blob_col", true)});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Blob> missing_blob,
+                         Blob::FromPath(dir->Str() + "/not_exist_file", /*offset=*/0,
+                                        /*length=*/10));
+
+    auto write_once = [&](const std::map<std::string, std::string>& options,
+                          const std::string& file_name) -> Status {
+        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileFormat> format,
+                               FileFormatFactory::Get("blob", options));
+        auto schema = arrow::schema(struct_type->fields());
+        ::ArrowSchema c_schema;
+        PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &c_schema));
+        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<WriterBuilder> writer_builder,
+                               format->CreateWriterBuilder(&c_schema, /*batch_size=*/1024));
+        // The blob writer builder is a SpecificFSWriterBuilder by construction.
+        static_cast<SpecificFSWriterBuilder*>(writer_builder.get())->WithFileSystem(fs);
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<OutputStream> out,
+                               fs->Create(dir->Str() + "/" + file_name, /*overwrite=*/true));
+        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FormatWriter> writer,
+                               writer_builder->Build(out, "none"));
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> array,
+                               paimon::test::TestHelper::MakeBlobDescriptorArray(
+                                   struct_type, missing_blob, GetDefaultPool()));
+        ArrowArray c_array;
+        PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &c_array));
+        PAIMON_RETURN_NOT_OK(writer->AddBatch(&c_array));
+        return writer->Finish();
+    };
+
+    // Without the option, writing the missing descriptor fails.
+    ASSERT_NOK_WITH_MSG(write_once({}, "no_option.blob"), "not exists");
+    // The option set in the format options map reaches the writer.
+    ASSERT_OK(write_once({{Options::BLOB_WRITE_NULL_ON_MISSING_FILE, "true"}}, "with_option.blob"));
+}
+
 }  // namespace paimon::blob::test
diff --git a/src/paimon/format/blob/blob_format_writer.cpp b/src/paimon/format/blob/blob_format_writer.cpp
index 146ffebf4..03cd36220 100644
--- a/src/paimon/format/blob/blob_format_writer.cpp
+++ b/src/paimon/format/blob/blob_format_writer.cpp
@@ -31,21 +31,32 @@
 #include "paimon/common/utils/delta_varint_compressor.h"
 #include "paimon/data/blob.h"
 #include "paimon/io/byte_array_input_stream.h"
+#include "paimon/logging.h"
 
 namespace paimon::blob {
 
 BlobFormatWriter::BlobFormatWriter(const std::shared_ptr<OutputStream>& out, const std::string& uri,
                                    const std::shared_ptr<arrow::DataType>& data_type,
+                                   bool write_null_on_missing_file,
+                                   bool write_null_on_fetch_failure,
                                    const std::shared_ptr<FileSystem>& fs,
                                    const std::shared_ptr<MemoryPool>& pool)
-    : out_(out), uri_(uri), data_type_(data_type), fs_(fs), pool_(pool) {
+    : out_(out),
+      uri_(uri),
+      data_type_(data_type),
+      fs_(fs),
+      pool_(pool),
+      write_null_on_missing_file_(write_null_on_missing_file),
+      write_null_on_fetch_failure_(write_null_on_fetch_failure) {
     metrics_ = std::make_shared<MetricsImpl>();
     tmp_buffer_ = Bytes::AllocateBytes(kTmpBufferSize, pool_.get());
     magic_number_bytes_ = IntegerToLittleEndian<int32_t>(BlobDefs::kMagicNumber, pool_);
+    logger_ = Logger::GetLogger("BlobFormatWriter");
 }
 
 Result<std::unique_ptr<BlobFormatWriter>> BlobFormatWriter::Create(
     const std::shared_ptr<OutputStream>& out, const std::shared_ptr<arrow::DataType>& data_type,
+    bool write_null_on_missing_file, bool write_null_on_fetch_failure,
     const std::shared_ptr<FileSystem>& fs, const std::shared_ptr<MemoryPool>& pool) {
     if (out == nullptr) {
         return Status::Invalid("blob format writer create failed. out is nullptr");
@@ -65,7 +76,8 @@ Result<std::unique_ptr<BlobFormatWriter>> BlobFormatWriter::Create(
             fmt::format("field {} is not BLOB", data_type->field(0)->ToString()));
     }
     PAIMON_ASSIGN_OR_RAISE(std::string uri, out->GetUri());
-    return std::unique_ptr<BlobFormatWriter>(new BlobFormatWriter(out, uri, data_type, fs, pool));
+    return std::unique_ptr<BlobFormatWriter>(new BlobFormatWriter(
+        out, uri, data_type, write_null_on_missing_file, write_null_on_fetch_failure, fs, pool));
 }
 
 Status BlobFormatWriter::AddBatch(ArrowArray* batch) {
@@ -126,13 +138,8 @@ Status BlobFormatWriter::Finish() {
 }
 
 Status BlobFormatWriter::WriteBlob(std::string_view blob_data) {
-    crc32_ = 0;
-    PAIMON_ASSIGN_OR_RAISE(int64_t previous_pos, out_->GetPos());
-
-    // write magic number
-    PAIMON_RETURN_NOT_OK(WriteWithCrc32(magic_number_bytes_->data(), magic_number_bytes_->size()));
-
-    // write blob content
+    // Open the blob input stream before writing any bytes, so that a failed fetch can be
+    // converted to a NULL element without leaving partial data in the output stream.
     // Dynamically check whether blob_data is a serialized BlobDescriptor (by magic header)
     // rather than relying on blob_as_descriptor_ config. This is consistent with Java behavior:
     // at write time, the input bytes are auto-detected as descriptor or raw data.
@@ -140,13 +147,32 @@ Status BlobFormatWriter::WriteBlob(std::string_view blob_data) {
     PAIMON_ASSIGN_OR_RAISE(bool is_descriptor,
                            BlobDescriptor::IsBlobDescriptor(blob_data.data(), blob_data.size()));
     if (is_descriptor) {
-        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<Blob> blob,
-                               Blob::FromDescriptor(blob_data.data(), blob_data.size()));
-        PAIMON_ASSIGN_OR_RAISE(in, blob->NewInputStream(fs_));
+        Result<std::unique_ptr<InputStream>> opened = OpenDescriptorInputStream(blob_data);
+        if (!opened.ok()) {
+            const Status& status = opened.status();
+            // A missing file is only handled by 'blob-write-null-on-missing-file'; other fetch
+            // failures are only handled by 'blob-write-null-on-fetch-failure' (aligned with Java).
+            bool write_null =
+                status.IsNotExist() ? write_null_on_missing_file_ : write_null_on_fetch_failure_;
+            if (write_null) {
+                PAIMON_LOG_WARN(logger_, "Failed to open blob, writing NULL for BLOB field: %s",
+                                status.ToString().c_str());
+                bin_lengths_.push_back(BlobDefs::kNullBinLength);
+                return Status::OK();
+            }
+            return status;
+        }
+        in = std::move(opened).value();
     } else {
         in = std::make_unique<ByteArrayInputStream>(blob_data.data(), blob_data.size());
     }
     PAIMON_ASSIGN_OR_RAISE(int64_t file_length, in->Length());
+
+    crc32_ = 0;
+    PAIMON_ASSIGN_OR_RAISE(int64_t previous_pos, out_->GetPos());
+
+    // write magic number
+    PAIMON_RETURN_NOT_OK(WriteWithCrc32(magic_number_bytes_->data(), magic_number_bytes_->size()));
     int64_t total_read_length = 0;
     int64_t read_len = std::min(file_length, static_cast<int64_t>(tmp_buffer_->size()));
     while (read_len > 0) {
@@ -179,6 +205,13 @@ Status BlobFormatWriter::WriteBlob(std::string_view blob_data) {
     return Status::OK();
 }
 
+Result<std::unique_ptr<InputStream>> BlobFormatWriter::OpenDescriptorInputStream(
+    std::string_view blob_data) const {
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<Blob> blob,
+                           Blob::FromDescriptor(blob_data.data(), blob_data.size()));
+    return blob->NewInputStream(fs_);
+}
+
 Status BlobFormatWriter::WriteBytes(const char* data, int64_t length) {
     PAIMON_ASSIGN_OR_RAISE(int64_t actual, out_->Write(data, length));
     if (actual != length) {
diff --git a/src/paimon/format/blob/blob_format_writer.h b/src/paimon/format/blob/blob_format_writer.h
index f62c49746..ad90539b7 100644
--- a/src/paimon/format/blob/blob_format_writer.h
+++ b/src/paimon/format/blob/blob_format_writer.h
@@ -26,6 +26,7 @@
 #include "arrow/api.h"
 #include "arrow/util/crc32.h"
 #include "paimon/format/format_writer.h"
+#include "paimon/logging.h"
 #include "paimon/memory/bytes.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/result.h"
@@ -39,6 +40,7 @@ struct ArrowArray;
 namespace paimon {
 class Blob;
 class FileSystem;
+class InputStream;
 class Metrics;
 class OutputStream;
 }  // namespace paimon
@@ -49,8 +51,13 @@ namespace paimon::blob {
 // https://cwiki.apache.org/confluence/display/PAIMON/PIP-35%3A+Introduce+Blob+to+store+multimodal+data
 class BlobFormatWriter : public FormatWriter {
  public:
+    /// When opening a descriptor input fails, `write_null_on_missing_file` converts a
+    /// missing file (Status::NotExist) to a NULL element and `write_null_on_fetch_failure`
+    /// converts any other open failure; failures during the streaming copy always fail the
+    /// write. See Options::BLOB_WRITE_NULL_ON_MISSING_FILE / BLOB_WRITE_NULL_ON_FETCH_FAILURE.
     static Result<std::unique_ptr<BlobFormatWriter>> Create(
         const std::shared_ptr<OutputStream>& out, const std::shared_ptr<arrow::DataType>& data_type,
+        bool write_null_on_missing_file, bool write_null_on_fetch_failure,
         const std::shared_ptr<FileSystem>& fs, const std::shared_ptr<MemoryPool>& pool);
 
     Status AddBatch(ArrowArray* batch) override;
@@ -70,11 +77,16 @@ class BlobFormatWriter : public FormatWriter {
  private:
     BlobFormatWriter(const std::shared_ptr<OutputStream>& out, const std::string& uri,
                      const std::shared_ptr<arrow::DataType>& data_type,
+                     bool write_null_on_missing_file, bool write_null_on_fetch_failure,
                      const std::shared_ptr<FileSystem>& fs,
                      const std::shared_ptr<MemoryPool>& pool);
 
     Status WriteBlob(std::string_view blob_data);
 
+    /// Deserialize the descriptor and open an input stream on the referenced data.
+    Result<std::unique_ptr<InputStream>> OpenDescriptorInputStream(
+        std::string_view blob_data) const;
+
     Status WriteBytes(const char* data, int64_t length);
     Status WriteWithCrc32(const char* data, int64_t length);
 
@@ -95,6 +107,9 @@ class BlobFormatWriter : public FormatWriter {
     std::shared_ptr<FileSystem> fs_;
     std::shared_ptr<MemoryPool> pool_;
     std::shared_ptr<Metrics> metrics_;
+    bool write_null_on_missing_file_ = false;
+    bool write_null_on_fetch_failure_ = false;
+    std::unique_ptr<Logger> logger_;
 };
 
 }  // namespace paimon::blob
diff --git a/src/paimon/format/blob/blob_format_writer_test.cpp b/src/paimon/format/blob/blob_format_writer_test.cpp
index b3f5cf825..42d69c4f6 100644
--- a/src/paimon/format/blob/blob_format_writer_test.cpp
+++ b/src/paimon/format/blob/blob_format_writer_test.cpp
@@ -32,10 +32,24 @@
 #include "paimon/testing/utils/testharness.h"
 
 namespace paimon::blob::test {
-class BlobFormatWriterTest : public ::testing::Test, public ::testing::WithParamInterface<bool> {
+
+/// A file system whose Open() always fails with the configured status, for verifying how the
+/// writer classifies open failures by status code.
+class OpenFailFileSystem : public LocalFileSystem {
+ public:
+    explicit OpenFailFileSystem(Status open_status) : open_status_(std::move(open_status)) {}
+
+    Result<std::unique_ptr<InputStream>> Open(const std::string& path) const override {
+        return open_status_;
+    }
+
+ private:
+    Status open_status_;
+};
+
+class BlobFormatWriterTestBase : public ::testing::Test {
  public:
     void SetUp() override {
-        blob_as_descriptor_ = GetParam();
         pool_ = GetDefaultPool();
         dir_ = paimon::test::UniqueTestDirectory::Create();
         ASSERT_TRUE(dir_);
@@ -50,24 +64,11 @@ class BlobFormatWriterTest : public ::testing::Test, public ::testing::WithParam
         ASSERT_OK(output_stream_->Close());
     }
 
-    Result<std::shared_ptr<arrow::Array>> PrepareBlobArray(
-        const std::shared_ptr<Blob>& blob) const {
-        arrow::StructBuilder struct_builder(struct_type_, arrow::default_memory_pool(),
-                                            {std::make_shared<arrow::LargeBinaryBuilder>()});
-        auto blob_builder =
-            static_cast<arrow::LargeBinaryBuilder*>(struct_builder.field_builder(0));
-        PAIMON_RETURN_NOT_OK_FROM_ARROW(struct_builder.Append());
-        if (blob_as_descriptor_) {
-            PAIMON_RETURN_NOT_OK_FROM_ARROW(blob_builder->Append(
-                blob->ToDescriptor(pool_)->data(), blob->ToDescriptor(pool_)->size()));
-        } else {
-            PAIMON_ASSIGN_OR_RAISE(auto blob_data, blob->ToData(file_system_, pool_));
-            PAIMON_RETURN_NOT_OK_FROM_ARROW(
-                blob_builder->Append(blob_data->data(), blob_data->size()));
-        }
-        std::shared_ptr<arrow::Array> array;
-        PAIMON_RETURN_NOT_OK_FROM_ARROW(struct_builder.Finish(&array));
-        return array;
+    /// Create a writer on output_stream_ with both write-null options disabled.
+    Result<std::unique_ptr<BlobFormatWriter>> CreateDefaultWriter() const {
+        return BlobFormatWriter::Create(output_stream_, struct_type_,
+                                        /*write_null_on_missing_file=*/false,
+                                        /*write_null_on_fetch_failure=*/false, file_system_, pool_);
     }
 
     Status AddBatchOnce(const std::shared_ptr<BlobFormatWriter>& format_writer,
@@ -77,8 +78,30 @@ class BlobFormatWriterTest : public ::testing::Test, public ::testing::WithParam
         return format_writer->AddBatch(c_array.get());
     }
 
- private:
-    bool blob_as_descriptor_;
+    Result<std::shared_ptr<arrow::Array>> PrepareDescriptorArray(
+        const std::shared_ptr<Blob>& blob) const {
+        return paimon::test::TestHelper::MakeBlobDescriptorArray(struct_type_, blob, pool_);
+    }
+
+    Result<std::shared_ptr<arrow::StructArray>> ReadBackAsData() const {
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
+                               file_system_->Open(dir_->Str() + "/file.blob"));
+        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BlobFileBatchReader> reader,
+                               BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024,
+                                                           /*blob_as_descriptor=*/false, pool_));
+        auto schema = arrow::schema(struct_type_->fields());
+        ::ArrowSchema c_schema;
+        PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &c_schema));
+        PAIMON_RETURN_NOT_OK(reader->SetReadSchema(&c_schema, /*predicate=*/nullptr,
+                                                   /*selection_bitmap=*/std::nullopt));
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::ChunkedArray> chunked_array,
+                               paimon::test::ReadResultCollector::CollectResult(reader.get()));
+        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> concat_array,
+                                          arrow::Concatenate(chunked_array->chunks()));
+        return arrow::internal::checked_pointer_cast<arrow::StructArray>(concat_array);
+    }
+
+ protected:
     std::shared_ptr<MemoryPool> pool_;
     std::unique_ptr<paimon::test::UniqueTestDirectory> dir_;
     std::shared_ptr<OutputStream> output_stream_;
@@ -86,13 +109,45 @@ class BlobFormatWriterTest : public ::testing::Test, public ::testing::WithParam
     std::shared_ptr<arrow::DataType> struct_type_;
 };
 
+class BlobFormatWriterTest : public BlobFormatWriterTestBase,
+                             public ::testing::WithParamInterface<bool> {
+ public:
+    void SetUp() override {
+        blob_as_descriptor_ = GetParam();
+        BlobFormatWriterTestBase::SetUp();
+    }
+
+    Result<std::shared_ptr<arrow::Array>> PrepareBlobArray(
+        const std::shared_ptr<Blob>& blob) const {
+        if (blob_as_descriptor_) {
+            return PrepareDescriptorArray(blob);
+        }
+        arrow::StructBuilder struct_builder(struct_type_, arrow::default_memory_pool(),
+                                            {std::make_shared<arrow::LargeBinaryBuilder>()});
+        auto blob_builder =
+            static_cast<arrow::LargeBinaryBuilder*>(struct_builder.field_builder(0));
+        PAIMON_RETURN_NOT_OK_FROM_ARROW(struct_builder.Append());
+        PAIMON_ASSIGN_OR_RAISE(PAIMON_UNIQUE_PTR<Bytes> blob_data,
+                               blob->ToData(file_system_, pool_));
+        PAIMON_RETURN_NOT_OK_FROM_ARROW(blob_builder->Append(blob_data->data(), blob_data->size()));
+        std::shared_ptr<arrow::Array> array;
+        PAIMON_RETURN_NOT_OK_FROM_ARROW(struct_builder.Finish(&array));
+        return array;
+    }
+
+ private:
+    bool blob_as_descriptor_;
+};
+
+/// The write-null tests always feed descriptor bytes, so they do not depend on the
+/// blob_as_descriptor_ parameter and run once on the non-parameterized fixture.
+using BlobFormatWriterWriteNullTest = BlobFormatWriterTestBase;
+
 INSTANTIATE_TEST_SUITE_P(BlobAsDescriptor, BlobFormatWriterTest, ::testing::Values(false, true));
 
 TEST_P(BlobFormatWriterTest, TestSimple) {
     // write
-    ASSERT_OK_AND_ASSIGN(
-        std::shared_ptr<BlobFormatWriter> writer,
-        BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer, CreateDefaultWriter());
 
     std::vector<std::shared_ptr<Blob>> expected_blobs;
     std::string file1 = paimon::test::GetDataDir() + "/avro/data/avro_with_null";
@@ -152,37 +207,42 @@ TEST_P(BlobFormatWriterTest, TestSimple) {
 
 TEST_P(BlobFormatWriterTest, TestCreateWithInvalidParameters) {
     // Test with nullptr output stream
-    ASSERT_NOK_WITH_MSG(BlobFormatWriter::Create(nullptr, struct_type_, file_system_, pool_),
-                        "blob format writer create failed. out is nullptr");
+    ASSERT_NOK_WITH_MSG(
+        BlobFormatWriter::Create(nullptr, struct_type_, /*write_null_on_missing_file=*/false,
+                                 /*write_null_on_fetch_failure=*/false, file_system_, pool_),
+        "blob format writer create failed. out is nullptr");
 
     // Test with nullptr data type
-    ASSERT_NOK_WITH_MSG(BlobFormatWriter::Create(output_stream_, nullptr, file_system_, pool_),
-                        "blob format writer create failed. data_type is nullptr");
+    ASSERT_NOK_WITH_MSG(
+        BlobFormatWriter::Create(output_stream_, nullptr, /*write_null_on_missing_file=*/false,
+                                 /*write_null_on_fetch_failure=*/false, file_system_, pool_),
+        "blob format writer create failed. data_type is nullptr");
 
     // Test with nullptr memory pool
     ASSERT_NOK_WITH_MSG(
-        BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, nullptr),
+        BlobFormatWriter::Create(output_stream_, struct_type_, /*write_null_on_missing_file=*/false,
+                                 /*write_null_on_fetch_failure=*/false, file_system_, nullptr),
         "blob format writer create failed. pool is nullptr");
 
     // Test with invalid field count (more than 1 field)
     auto multi_field_type = arrow::struct_(
         {arrow::field("blob_col1", arrow::binary()), arrow::field("blob_col2", arrow::binary())});
-    ASSERT_NOK_WITH_MSG(
-        BlobFormatWriter::Create(output_stream_, multi_field_type, file_system_, pool_),
-        "blob data type field number 2 is not 1");
+    ASSERT_NOK_WITH_MSG(BlobFormatWriter::Create(
+                            output_stream_, multi_field_type, /*write_null_on_missing_file=*/false,
+                            /*write_null_on_fetch_failure=*/false, file_system_, pool_),
+                        "blob data type field number 2 is not 1");
 
     // Test with non-blob field (missing blob metadata)
     auto non_blob_field = arrow::field("regular_col", arrow::binary());
     auto non_blob_type = arrow::struct_({non_blob_field});
-    ASSERT_NOK_WITH_MSG(
-        BlobFormatWriter::Create(output_stream_, non_blob_type, file_system_, pool_),
-        "field regular_col: binary is not BLOB");
+    ASSERT_NOK_WITH_MSG(BlobFormatWriter::Create(
+                            output_stream_, non_blob_type, /*write_null_on_missing_file=*/false,
+                            /*write_null_on_fetch_failure=*/false, file_system_, pool_),
+                        "field regular_col: binary is not BLOB");
 }
 
 TEST_P(BlobFormatWriterTest, TestInvalidCase) {
-    ASSERT_OK_AND_ASSIGN(
-        std::shared_ptr<BlobFormatWriter> writer,
-        BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer, CreateDefaultWriter());
 
     // Test nullptr batch
     ASSERT_NOK_WITH_MSG(writer->AddBatch(nullptr),
@@ -199,9 +259,7 @@ TEST_P(BlobFormatWriterTest, TestInvalidCase) {
 }
 
 TEST_P(BlobFormatWriterTest, TestAddBatchWithInvalidBatchLength) {
-    ASSERT_OK_AND_ASSIGN(
-        std::shared_ptr<BlobFormatWriter> writer,
-        BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer, CreateDefaultWriter());
 
     // Test batch with wrong length (not 1)
     arrow::StructBuilder struct_builder(struct_type_, arrow::default_memory_pool(),
@@ -227,9 +285,7 @@ TEST_P(BlobFormatWriterTest, TestAddBatchWithInvalidBatchLength) {
 }
 
 TEST_P(BlobFormatWriterTest, TestReachTargetSize) {
-    ASSERT_OK_AND_ASSIGN(
-        std::shared_ptr<BlobFormatWriter> writer,
-        BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer, CreateDefaultWriter());
 
     // Initially should not reach target size
     ASSERT_OK_AND_ASSIGN(bool reached, writer->ReachTargetSize(true, 1000));
@@ -252,9 +308,7 @@ TEST_P(BlobFormatWriterTest, TestReachTargetSize) {
 }
 
 TEST_P(BlobFormatWriterTest, TestGetWriterMetrics) {
-    ASSERT_OK_AND_ASSIGN(
-        std::shared_ptr<BlobFormatWriter> writer,
-        BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer, CreateDefaultWriter());
 
     auto metrics = writer->GetWriterMetrics();
     ASSERT_TRUE(metrics);
@@ -262,9 +316,7 @@ TEST_P(BlobFormatWriterTest, TestGetWriterMetrics) {
 
 TEST_P(BlobFormatWriterTest, TestEmptyWriter) {
     // Test creating a writer and finishing without adding any data
-    ASSERT_OK_AND_ASSIGN(
-        std::shared_ptr<BlobFormatWriter> writer,
-        BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer, CreateDefaultWriter());
 
     ASSERT_OK(writer->Flush());
     ASSERT_OK(writer->Finish());
@@ -283,9 +335,7 @@ TEST_P(BlobFormatWriterTest, TestEmptyWriter) {
 }
 
 TEST_P(BlobFormatWriterTest, TestLargeBlob) {
-    ASSERT_OK_AND_ASSIGN(
-        std::shared_ptr<BlobFormatWriter> writer,
-        BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer, CreateDefaultWriter());
 
     // Create a temporary large file for testing
     std::string large_file_path = dir_->Str() + "/large_test_file.bin";
@@ -338,9 +388,7 @@ TEST_P(BlobFormatWriterTest, TestLargeBlob) {
 }
 
 TEST_P(BlobFormatWriterTest, TestAddBatchWithNullValues) {
-    ASSERT_OK_AND_ASSIGN(
-        std::shared_ptr<BlobFormatWriter> writer,
-        BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer, CreateDefaultWriter());
 
     // Write one row with child-level null blob
     arrow::StructBuilder struct_builder(struct_type_, arrow::default_memory_pool(),
@@ -386,18 +434,159 @@ TEST_P(BlobFormatWriterTest, TestAddBatchWithNullValues) {
     ASSERT_TRUE(struct_builder2.Finish(&null_struct_array).ok());
     auto null_c_array = std::make_unique<ArrowArray>();
     ASSERT_TRUE(arrow::ExportArray(*null_struct_array, null_c_array.get()).ok());
-    ASSERT_OK_AND_ASSIGN(
-        std::shared_ptr<BlobFormatWriter> writer2,
-        BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer2, CreateDefaultWriter());
     ASSERT_NOK_WITH_MSG(writer2->AddBatch(null_c_array.get()),
                         "BlobFormatWriter does not support struct-level null.");
     ArrowArrayRelease(null_c_array.get());
 }
 
-TEST_P(BlobFormatWriterTest, TestAddBatchWithZeroLengthBlob) {
+TEST_F(BlobFormatWriterWriteNullTest, TestWriteNullOnMissingFile) {
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<BlobFormatWriter> writer,
+        BlobFormatWriter::Create(output_stream_, struct_type_, /*write_null_on_missing_file=*/true,
+                                 /*write_null_on_fetch_failure=*/false, file_system_, pool_));
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Blob> missing_blob,
+                         Blob::FromPath(dir_->Str() + "/not_exist_file", /*offset=*/0,
+                                        /*length=*/10));
+    ASSERT_OK_AND_ASSIGN(auto missing_array, PrepareDescriptorArray(missing_blob));
+    ASSERT_OK(AddBatchOnce(writer, missing_array));
+
+    // A fetch failure is not converted to NULL by write_null_on_missing_file alone (aligned
+    // with Java); the rejected row leaves the writer usable.
+    std::string file = paimon::test::GetDataDir() + "/xxhash.data";
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Blob> bad_offset_blob,
+                         Blob::FromPath(file, /*offset=*/1 << 20, /*length=*/10));
+    ASSERT_OK_AND_ASSIGN(auto bad_offset_array, PrepareDescriptorArray(bad_offset_blob));
+    ASSERT_NOK_WITH_MSG(AddBatchOnce(writer, bad_offset_array), "exceed total length");
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Blob> blob, Blob::FromPath(file));
+    ASSERT_OK_AND_ASSIGN(auto array, PrepareDescriptorArray(blob));
+    ASSERT_OK(AddBatchOnce(writer, array));
+
+    ASSERT_OK(writer->Flush());
+    ASSERT_OK(writer->Finish());
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::StructArray> result_struct, ReadBackAsData());
+    ASSERT_EQ(result_struct->length(), 2);
+    ASSERT_TRUE(result_struct->field(0)->IsNull(0));
+    ASSERT_FALSE(result_struct->field(0)->IsNull(1));
+    auto binary_array =
+        arrow::internal::checked_pointer_cast<arrow::LargeBinaryArray>(result_struct->field(0));
+    ASSERT_OK_AND_ASSIGN(auto expected_data, blob->ToData(file_system_, pool_));
+    ASSERT_EQ(binary_array->GetView(1),
+              std::string_view(expected_data->data(), expected_data->size()));
+}
+
+TEST_F(BlobFormatWriterWriteNullTest, TestWriteNullOnFetchFailure) {
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<BlobFormatWriter> writer,
+        BlobFormatWriter::Create(output_stream_, struct_type_, /*write_null_on_missing_file=*/false,
+                                 /*write_null_on_fetch_failure=*/true, file_system_, pool_));
+
+    std::string file = paimon::test::GetDataDir() + "/xxhash.data";
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Blob> bad_offset_blob,
+                         Blob::FromPath(file, /*offset=*/1 << 20, /*length=*/10));
+    ASSERT_OK_AND_ASSIGN(auto bad_offset_array, PrepareDescriptorArray(bad_offset_blob));
+    ASSERT_OK(AddBatchOnce(writer, bad_offset_array));
+
+    // A missing file is not converted to NULL by write_null_on_fetch_failure alone (aligned
+    // with Java); the rejected row leaves the writer usable.
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Blob> missing_blob,
+                         Blob::FromPath(dir_->Str() + "/not_exist_file", /*offset=*/0,
+                                        /*length=*/10));
+    ASSERT_OK_AND_ASSIGN(auto missing_array, PrepareDescriptorArray(missing_blob));
+    ASSERT_NOK_WITH_MSG(AddBatchOnce(writer, missing_array), "not exists");
+
+    ASSERT_OK(writer->Flush());
+    ASSERT_OK(writer->Finish());
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::StructArray> result_struct, ReadBackAsData());
+    ASSERT_EQ(result_struct->length(), 1);
+    ASSERT_TRUE(result_struct->field(0)->IsNull(0));
+}
+
+TEST_F(BlobFormatWriterWriteNullTest, TestWriteNullOnBothOptionsEnabled) {
     ASSERT_OK_AND_ASSIGN(
         std::shared_ptr<BlobFormatWriter> writer,
-        BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_));
+        BlobFormatWriter::Create(output_stream_, struct_type_, /*write_null_on_missing_file=*/true,
+                                 /*write_null_on_fetch_failure=*/true, file_system_, pool_));
+
+    // Row 0: missing file -> NULL.
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Blob> missing_blob,
+                         Blob::FromPath(dir_->Str() + "/not_exist_file", /*offset=*/0,
+                                        /*length=*/10));
+    ASSERT_OK_AND_ASSIGN(auto missing_array, PrepareDescriptorArray(missing_blob));
+    ASSERT_OK(AddBatchOnce(writer, missing_array));
+
+    // Row 1: fetch failure (offset beyond EOF) -> NULL.
+    std::string file = paimon::test::GetDataDir() + "/xxhash.data";
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Blob> bad_offset_blob,
+                         Blob::FromPath(file, /*offset=*/1 << 20, /*length=*/10));
+    ASSERT_OK_AND_ASSIGN(auto bad_offset_array, PrepareDescriptorArray(bad_offset_blob));
+    ASSERT_OK(AddBatchOnce(writer, bad_offset_array));
+
+    // Row 2: valid blob.
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Blob> blob, Blob::FromPath(file));
+    ASSERT_OK_AND_ASSIGN(auto array, PrepareDescriptorArray(blob));
+    ASSERT_OK(AddBatchOnce(writer, array));
+
+    ASSERT_OK(writer->Flush());
+    ASSERT_OK(writer->Finish());
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::StructArray> result_struct, ReadBackAsData());
+    ASSERT_EQ(result_struct->length(), 3);
+    ASSERT_TRUE(result_struct->field(0)->IsNull(0));
+    ASSERT_TRUE(result_struct->field(0)->IsNull(1));
+    ASSERT_FALSE(result_struct->field(0)->IsNull(2));
+    auto binary_array =
+        arrow::internal::checked_pointer_cast<arrow::LargeBinaryArray>(result_struct->field(0));
+    ASSERT_OK_AND_ASSIGN(auto expected_data, blob->ToData(file_system_, pool_));
+    ASSERT_EQ(binary_array->GetView(2),
+              std::string_view(expected_data->data(), expected_data->size()));
+}
+
+TEST_F(BlobFormatWriterWriteNullTest, TestWriteNullClassifiesByStatusCode) {
+    // The missing-file vs fetch-failure split keys on the open status code (NotExist <-> missing
+    // file), independent of the file system implementation.
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Blob> blob,
+                         Blob::FromPath(dir_->Str() + "/any_file", /*offset=*/0, /*length=*/10));
+    ASSERT_OK_AND_ASSIGN(auto array, PrepareDescriptorArray(blob));
+    auto not_exist_fs = std::make_shared<OpenFailFileSystem>(Status::NotExist("mock not exist"));
+    auto io_error_fs = std::make_shared<OpenFailFileSystem>(Status::IOError("mock io error"));
+
+    {
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer,
+                             BlobFormatWriter::Create(
+                                 output_stream_, struct_type_, /*write_null_on_missing_file=*/true,
+                                 /*write_null_on_fetch_failure=*/false, not_exist_fs, pool_));
+        ASSERT_OK(AddBatchOnce(writer, array));
+    }
+    {
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer,
+                             BlobFormatWriter::Create(
+                                 output_stream_, struct_type_, /*write_null_on_missing_file=*/true,
+                                 /*write_null_on_fetch_failure=*/false, io_error_fs, pool_));
+        ASSERT_NOK_WITH_MSG(AddBatchOnce(writer, array), "mock io error");
+    }
+    {
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer,
+                             BlobFormatWriter::Create(
+                                 output_stream_, struct_type_, /*write_null_on_missing_file=*/false,
+                                 /*write_null_on_fetch_failure=*/true, io_error_fs, pool_));
+        ASSERT_OK(AddBatchOnce(writer, array));
+    }
+    {
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer,
+                             BlobFormatWriter::Create(
+                                 output_stream_, struct_type_, /*write_null_on_missing_file=*/false,
+                                 /*write_null_on_fetch_failure=*/true, not_exist_fs, pool_));
+        ASSERT_NOK_WITH_MSG(AddBatchOnce(writer, array), "mock not exist");
+    }
+}
+
+TEST_P(BlobFormatWriterTest, TestAddBatchWithZeroLengthBlob) {
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer, CreateDefaultWriter());
 
     // Create a zero-length file
     std::string zero_file_path = dir_->Str() + "/zero_length_file.bin";
diff --git a/src/paimon/format/blob/blob_writer_builder.h b/src/paimon/format/blob/blob_writer_builder.h
index 4f59315f7..12caa0e16 100644
--- a/src/paimon/format/blob/blob_writer_builder.h
+++ b/src/paimon/format/blob/blob_writer_builder.h
@@ -24,6 +24,8 @@
 #include <utility>
 
 #include "arrow/api.h"
+#include "paimon/common/utils/options_utils.h"
+#include "paimon/defs.h"
 #include "paimon/format/blob/blob_format_writer.h"
 #include "paimon/format/format_writer.h"
 #include "paimon/format/writer_builder.h"
@@ -65,7 +67,14 @@ class BlobWriterBuilder : public SpecificFSWriterBuilder {
         if (fs_ == nullptr) {
             return Status::Invalid("File system is nullptr. Please call WithFileSystem() first.");
         }
-        return BlobFormatWriter::Create(out, data_type_, fs_, pool_);
+        PAIMON_ASSIGN_OR_RAISE(bool write_null_on_missing_file,
+                               OptionsUtils::GetValueFromMap<bool>(
+                                   options_, Options::BLOB_WRITE_NULL_ON_MISSING_FILE, false));
+        PAIMON_ASSIGN_OR_RAISE(bool write_null_on_fetch_failure,
+                               OptionsUtils::GetValueFromMap<bool>(
+                                   options_, Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, false));
+        return BlobFormatWriter::Create(out, data_type_, write_null_on_missing_file,
+                                        write_null_on_fetch_failure, fs_, pool_);
     }
 
  private:
diff --git a/src/paimon/format/blob/blob_writer_builder_test.cpp b/src/paimon/format/blob/blob_writer_builder_test.cpp
index 01a077f96..d823c28b3 100644
--- a/src/paimon/format/blob/blob_writer_builder_test.cpp
+++ b/src/paimon/format/blob/blob_writer_builder_test.cpp
@@ -17,11 +17,15 @@
 #include "paimon/format/blob/blob_writer_builder.h"
 
 #include "arrow/api.h"
+#include "arrow/c/bridge.h"
 #include "gtest/gtest.h"
 #include "paimon/common/data/blob_utils.h"
 #include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/data/blob.h"
+#include "paimon/defs.h"
 #include "paimon/fs/file_system.h"
 #include "paimon/fs/local/local_file_system.h"
+#include "paimon/testing/utils/test_helper.h"
 #include "paimon/testing/utils/testharness.h"
 
 namespace paimon::blob::test {
@@ -33,7 +37,7 @@ class BlobWriterBuilderTest : public ::testing::Test {
         file_system_ = std::make_shared<LocalFileSystem>();
         ASSERT_OK_AND_ASSIGN(output_stream_,
                              file_system_->Create(dir_->Str() + "/file.blob", /*overwrite=*/true));
-        struct_type_ = arrow::struct_({BlobUtils::ToArrowField("blob_col", false)});
+        struct_type_ = arrow::struct_({BlobUtils::ToArrowField("blob_col", true)});
     }
     void TearDown() override {}
 
@@ -53,4 +57,88 @@ TEST_F(BlobWriterBuilderTest, TestSimple) {
     ASSERT_OK(builder.Build(output_stream_, "none"));
 }
 
+TEST_F(BlobWriterBuilderTest, TestWriteNullOptions) {
+    auto prepare_batch = [&](const std::shared_ptr<Blob>& blob, ArrowArray* c_array) -> Status {
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> array,
+                               paimon::test::TestHelper::MakeBlobDescriptorArray(struct_type_, blob,
+                                                                                 GetDefaultPool()));
+        PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, c_array));
+        return Status::OK();
+    };
+    // Each writer gets its own output file so that a Finish() in one scenario cannot leak
+    // bytes into the next.
+    int32_t file_index = 0;
+    auto build_writer = [&](const std::map<std::string, std::string>& options)
+        -> Result<std::unique_ptr<FormatWriter>> {
+        PAIMON_ASSIGN_OR_RAISE(
+            std::shared_ptr<OutputStream> out,
+            file_system_->Create(dir_->Str() + "/file" + std::to_string(file_index++) + ".blob",
+                                 /*overwrite=*/true));
+        BlobWriterBuilder builder(struct_type_, options);
+        builder.WithFileSystem(file_system_);
+        return builder.Build(out, "none");
+    };
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Blob> missing_blob,
+                         Blob::FromPath(dir_->Str() + "/not_exist_file", /*offset=*/0,
+                                        /*length=*/10));
+    std::string data_file = dir_->Str() + "/data_file.bin";
+    ASSERT_OK_AND_ASSIGN(auto data_file_stream,
+                         file_system_->Create(data_file, /*overwrite=*/true));
+    ASSERT_OK_AND_ASSIGN(int64_t written, data_file_stream->Write("blob data", 9));
+    ASSERT_EQ(written, 9);
+    ASSERT_OK(data_file_stream->Flush());
+    ASSERT_OK(data_file_stream->Close());
+    // The offset beyond the end of the 9-byte file makes the descriptor a fetch failure.
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Blob> bad_offset_blob,
+                         Blob::FromPath(data_file, /*offset=*/100, /*length=*/10));
+
+    // Both options default to false.
+    {
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<FormatWriter> writer, build_writer({}));
+        ArrowArray c_array;
+        ASSERT_OK(prepare_batch(missing_blob, &c_array));
+        ASSERT_NOK_WITH_MSG(writer->AddBatch(&c_array), "not exists");
+    }
+    {
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<FormatWriter> writer, build_writer({}));
+        ArrowArray c_array;
+        ASSERT_OK(prepare_batch(bad_offset_blob, &c_array));
+        ASSERT_NOK_WITH_MSG(writer->AddBatch(&c_array), "exceed total length");
+    }
+
+    {
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<FormatWriter> writer,
+                             build_writer({{Options::BLOB_WRITE_NULL_ON_MISSING_FILE, "true"}}));
+        ArrowArray c_array;
+        ASSERT_OK(prepare_batch(missing_blob, &c_array));
+        ASSERT_OK(writer->AddBatch(&c_array));
+        ASSERT_OK(writer->Finish());
+    }
+
+    {
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<FormatWriter> writer,
+                             build_writer({{Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, "true"}}));
+        ArrowArray c_array;
+        ASSERT_OK(prepare_batch(bad_offset_blob, &c_array));
+        ASSERT_OK(writer->AddBatch(&c_array));
+        ASSERT_OK(writer->Finish());
+    }
+
+    // An explicit "false" exercises value parsing, not just the absent-key default.
+    {
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<FormatWriter> writer,
+                             build_writer({{Options::BLOB_WRITE_NULL_ON_MISSING_FILE, "false"},
+                                           {Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, "false"}}));
+        ArrowArray c_array;
+        ASSERT_OK(prepare_batch(missing_blob, &c_array));
+        ASSERT_NOK_WITH_MSG(writer->AddBatch(&c_array), "not exists");
+    }
+
+    {
+        ASSERT_NOK_WITH_MSG(build_writer({{Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, "invalid"}}),
+                            "convert key blob-write-null-on-fetch-failure");
+    }
+}
+
 }  // namespace paimon::blob::test
diff --git a/src/paimon/fs/jindo/jindo_utils.h b/src/paimon/fs/jindo/jindo_utils.h
index 3f9f72ba5..d3c596886 100644
--- a/src/paimon/fs/jindo/jindo_utils.h
+++ b/src/paimon/fs/jindo/jindo_utils.h
@@ -17,15 +17,21 @@
 #pragma once
 
 #include "JdoStatus.hpp"  // NOLINT(build/include_subdir)
+#include "jdo_error.h"    // NOLINT(build/include_subdir)
 #include "paimon/status.h"
 
 namespace paimon::jindo {
-#define PAIMON_RETURN_NOT_OK_FROM_JINDO(JINDO_STATUS) \
-    do {                                              \
-        auto __s = (JINDO_STATUS);                    \
-        if (PAIMON_UNLIKELY(!(__s).ok())) {           \
-            return Status::IOError(__s.errMsg());     \
-        }                                             \
+/// Maps a jindo file-not-found error to Status::NotExist so that callers can distinguish
+/// a missing file from other IO errors, consistent with LocalFileSystem.
+#define PAIMON_RETURN_NOT_OK_FROM_JINDO(JINDO_STATUS)             \
+    do {                                                          \
+        auto __s = (JINDO_STATUS);                                \
+        if (PAIMON_UNLIKELY(!(__s).ok())) {                       \
+            if ((__s).getErrCode() == JDO_FILE_NOT_FOUND_ERROR) { \
+                return Status::NotExist(__s.errMsg());            \
+            }                                                     \
+            return Status::IOError(__s.errMsg());                 \
+        }                                                         \
     } while (false)
 
 }  // namespace paimon::jindo
diff --git a/src/paimon/fs/jindo/jindo_utils_test.cpp b/src/paimon/fs/jindo/jindo_utils_test.cpp
new file mode 100644
index 000000000..041876e67
--- /dev/null
+++ b/src/paimon/fs/jindo/jindo_utils_test.cpp
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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.
+ */
+
+#include "paimon/fs/jindo/jindo_utils.h"
+
+#include "gtest/gtest.h"
+#include "jdo_error.h"  // NOLINT(build/include_subdir)
+#include "paimon/status.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::jindo::test {
+
+namespace {
+
+Status Convert(const JdoStatus& jdo_status) {
+    PAIMON_RETURN_NOT_OK_FROM_JINDO(jdo_status);
+    return Status::OK();
+}
+
+}  // namespace
+
+TEST(JindoUtilsTest, TestMacroMapsStatusByErrorCode) {
+    ASSERT_OK(Convert(JdoStatus()));
+
+    Status not_found = Convert(JdoStatus(JDO_FILE_NOT_FOUND_ERROR, "file not found"));
+    ASSERT_TRUE(not_found.IsNotExist());
+    ASSERT_NOK_WITH_MSG(not_found, "file not found");
+
+    Status other = Convert(JdoStatus(JDO_FILE_NOT_FOUND_ERROR + 1, "some other error"));
+    ASSERT_TRUE(other.IsIOError());
+    ASSERT_NOK_WITH_MSG(other, "some other error");
+}
+
+}  // namespace paimon::jindo::test
diff --git a/src/paimon/testing/utils/test_helper.h b/src/paimon/testing/utils/test_helper.h
index d3348a5c7..7b710d6d9 100644
--- a/src/paimon/testing/utils/test_helper.h
+++ b/src/paimon/testing/utils/test_helper.h
@@ -22,6 +22,7 @@
 #include <utility>
 #include <vector>
 
+#include "arrow/api.h"
 #include "arrow/c/bridge.h"
 #include "arrow/ipc/api.h"
 #include "paimon/api.h"
@@ -175,6 +176,27 @@ class TestHelper {
         return result_plan->Splits();
     }
 
+    /// Builds a one-row struct array holding the serialized descriptor of the blob.
+    static Result<std::shared_ptr<arrow::Array>> MakeBlobDescriptorArray(
+        const std::shared_ptr<arrow::DataType>& struct_type, const std::shared_ptr<Blob>& blob,
+        const std::shared_ptr<MemoryPool>& pool) {
+        if (struct_type->num_fields() != 1 ||
+            struct_type->field(0)->type()->id() != arrow::Type::LARGE_BINARY) {
+            return Status::Invalid("struct_type must have a single large binary field");
+        }
+        arrow::StructBuilder struct_builder(struct_type, arrow::default_memory_pool(),
+                                            {std::make_shared<arrow::LargeBinaryBuilder>()});
+        auto blob_builder =
+            static_cast<arrow::LargeBinaryBuilder*>(struct_builder.field_builder(0));
+        PAIMON_RETURN_NOT_OK_FROM_ARROW(struct_builder.Append());
+        PAIMON_UNIQUE_PTR<Bytes> descriptor = blob->ToDescriptor(pool);
+        PAIMON_RETURN_NOT_OK_FROM_ARROW(
+            blob_builder->Append(descriptor->data(), descriptor->size()));
+        std::shared_ptr<arrow::Array> array;
+        PAIMON_RETURN_NOT_OK_FROM_ARROW(struct_builder.Finish(&array));
+        return array;
+    }
+
     static Result<bool> CheckBlobsEqual(const std::vector<std::shared_ptr<Blob>>& result_blobs,
                                         const std::vector<std::shared_ptr<Blob>>& expected_blobs,
                                         const std::shared_ptr<FileSystem>& fs) {
diff --git a/test/inte/blob_table_inte_test.cpp b/test/inte/blob_table_inte_test.cpp
index 75fee6ef3..d0e834b8c 100644
--- a/test/inte/blob_table_inte_test.cpp
+++ b/test/inte/blob_table_inte_test.cpp
@@ -24,6 +24,7 @@
 #include <optional>
 #include <set>
 #include <string>
+#include <string_view>
 #include <utility>
 #include <vector>
 
@@ -370,6 +371,50 @@ class BlobTableInteTest : public testing::Test, public ::testing::WithParamInter
             });
     }
 
+    /// Delete the file referenced by the serialized BlobDescriptor at `row` of blob field
+    /// `field_name`, so that a subsequent table write observes a missing file.
+    Status DeleteDescriptorTarget(const std::shared_ptr<arrow::StructArray>& desc_array,
+                                  const std::string& field_name, int64_t row) const {
+        const auto& blob_col = arrow::internal::checked_cast<const arrow::LargeBinaryArray&>(
+            *desc_array->GetFieldByName(field_name));
+        std::string_view descriptor_bytes = blob_col.GetView(row);
+        PAIMON_ASSIGN_OR_RAISE(
+            std::unique_ptr<BlobDescriptor> descriptor,
+            BlobDescriptor::Deserialize(descriptor_bytes.data(), descriptor_bytes.size()));
+        LocalFileSystem fs;
+        return fs.Delete(descriptor->Uri(), /*recursive=*/false);
+    }
+
+    /// Rewrite the serialized BlobDescriptor at `row` of blob field `field_name` to reference
+    /// an offset beyond the end of the target file, so that a subsequent table write observes
+    /// a fetch failure that is not a missing file. `row` counts non-null rows.
+    Result<std::shared_ptr<arrow::StructArray>> CorruptDescriptorOffset(
+        const std::shared_ptr<arrow::StructArray>& desc_array, const std::string& field_name,
+        int64_t row) const {
+        int64_t current_row = 0;
+        return TransformBlobFields(
+            desc_array, {field_name},
+            [&](const std::string_view& descriptor_bytes,
+                arrow::LargeBinaryBuilder* builder) -> Status {
+                if (current_row++ != row) {
+                    PAIMON_RETURN_NOT_OK_FROM_ARROW(
+                        builder->Append(descriptor_bytes.data(), descriptor_bytes.size()));
+                    return Status::OK();
+                }
+                PAIMON_ASSIGN_OR_RAISE(
+                    std::unique_ptr<BlobDescriptor> descriptor,
+                    BlobDescriptor::Deserialize(descriptor_bytes.data(), descriptor_bytes.size()));
+                PAIMON_ASSIGN_OR_RAISE(
+                    std::unique_ptr<BlobDescriptor> corrupted,
+                    BlobDescriptor::Create(descriptor->Version(), descriptor->Uri(),
+                                           /*offset=*/1 << 20, descriptor->Length()));
+                auto corrupted_bytes = corrupted->Serialize(pool_);
+                PAIMON_RETURN_NOT_OK_FROM_ARROW(
+                    builder->Append(corrupted_bytes->data(), corrupted_bytes->size()));
+                return Status::OK();
+            });
+    }
+
     struct BlobDescriptorPathRewrite {
         std::string table_path;
         std::vector<std::string> table_relative_blob_dirs;
@@ -561,6 +606,200 @@ TEST_P(BlobTableInteTest, TestAppendTableWriteWithBlobAsDescriptorFalse) {
     ASSERT_OK(ScanAndRead(table_path, schema->field_names(), write_array));
 }
 
+TEST_P(BlobTableInteTest, TestWriteNullOnMissingFile) {
+    // blob-write-null-on-missing-file=true: a descriptor whose file is gone at write time
+    // becomes a NULL blob element, while the row itself is still written and merged with
+    // the data file on read.
+    arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
+                                 arrow::field("f1", arrow::int32()),
+                                 BlobUtils::ToArrowField("blob", true)};
+
+    std::map<std::string, std::string> options = {
+        {Options::MANIFEST_FORMAT, "orc"},       {Options::FILE_FORMAT, GetParam()},
+        {Options::TARGET_FILE_SIZE, "700"},      {Options::BUCKET, "-1"},
+        {Options::ROW_TRACKING_ENABLED, "true"}, {Options::DATA_EVOLUTION_ENABLED, "true"},
+        {Options::BLOB_AS_DESCRIPTOR, "true"},   {Options::BLOB_WRITE_NULL_ON_MISSING_FILE, "true"},
+        {Options::FILE_SYSTEM, "local"}};
+    CreateTable(fields, /*partition_keys=*/{}, options);
+    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+    std::string raw_json = R"([
+        ["str_0", 0, "blob_data_0"],
+        ["str_1", 1, "blob_data_1"],
+        ["str_2", 2, "blob_data_2"]
+    ])";
+    auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"}));
+
+    // Remove the file behind row 1's descriptor so the write observes a missing file.
+    ASSERT_OK(DeleteDescriptorTarget(desc_array, "blob", /*row=*/1));
+
+    auto schema = arrow::schema(fields);
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
+                         WriteArray(table_path, {}, schema->field_names(), {desc_array}));
+    ASSERT_OK(Commit(table_path, commit_msgs));
+
+    // Both the main data file and the .blob file keep the full row count; the read path
+    // merges them back into rows where row 1's blob is NULL.
+    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
+    VerifyDataFileMetas(plan, /*expected_file_count=*/2, /*expected_row_counts=*/{3, 3},
+                        /*expected_min_seqs=*/{1, 1}, /*expected_max_seqs=*/{1, 1},
+                        /*expected_first_row_ids=*/{0, 0},
+                        /*expected_write_cols=*/
+                        {std::vector<std::string>{"f0", "f1"}, std::vector<std::string>{"blob"}});
+
+    ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), plan));
+    ASSERT_TRUE(result.chunked_array);
+    auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie();
+    auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
+    ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"blob"}));
+
+    std::string expected_json = R"([
+        ["str_0", 0, "blob_data_0"],
+        ["str_1", 1, null],
+        ["str_2", 2, "blob_data_2"]
+    ])";
+    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
+            .ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_array));
+    ASSERT_TRUE(resolved->Equals(expected_with_rk))
+        << "result:" << resolved->ToString() << std::endl
+        << "expected:" << expected_with_rk->ToString();
+}
+
+TEST_P(BlobTableInteTest, TestMissingFileFailsWriteWhenWriteNullDisabled) {
+    // Without blob-write-null-on-missing-file, a missing descriptor file fails the write.
+    arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
+                                 arrow::field("f1", arrow::int32()),
+                                 BlobUtils::ToArrowField("blob", true)};
+
+    std::map<std::string, std::string> options = {
+        {Options::MANIFEST_FORMAT, "orc"},       {Options::FILE_FORMAT, GetParam()},
+        {Options::TARGET_FILE_SIZE, "700"},      {Options::BUCKET, "-1"},
+        {Options::ROW_TRACKING_ENABLED, "true"}, {Options::DATA_EVOLUTION_ENABLED, "true"},
+        {Options::BLOB_AS_DESCRIPTOR, "true"},   {Options::FILE_SYSTEM, "local"}};
+    CreateTable(fields, /*partition_keys=*/{}, options);
+    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+    std::string raw_json = R"([
+        ["str_0", 0, "blob_data_0"],
+        ["str_1", 1, "blob_data_1"]
+    ])";
+    auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"}));
+    ASSERT_OK(DeleteDescriptorTarget(desc_array, "blob", /*row=*/1));
+
+    auto schema = arrow::schema(fields);
+    ASSERT_NOK_WITH_MSG(WriteArray(table_path, {}, schema->field_names(), {desc_array}),
+                        "not exists");
+}
+
+TEST_P(BlobTableInteTest, TestWriteNullOnFetchFailure) {
+    // blob-write-null-on-fetch-failure=true: a descriptor whose data cannot be fetched for a
+    // reason other than a missing file (here: offset beyond the end of the file) becomes a
+    // NULL blob element, while the row itself is still written and merged with the data file
+    // on read.
+    arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
+                                 arrow::field("f1", arrow::int32()),
+                                 BlobUtils::ToArrowField("blob", true)};
+
+    std::map<std::string, std::string> options = {
+        {Options::MANIFEST_FORMAT, "orc"},
+        {Options::FILE_FORMAT, GetParam()},
+        {Options::TARGET_FILE_SIZE, "700"},
+        {Options::BUCKET, "-1"},
+        {Options::ROW_TRACKING_ENABLED, "true"},
+        {Options::DATA_EVOLUTION_ENABLED, "true"},
+        {Options::BLOB_AS_DESCRIPTOR, "true"},
+        {Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, "true"},
+        {Options::FILE_SYSTEM, "local"}};
+    CreateTable(fields, /*partition_keys=*/{}, options);
+    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+    std::string raw_json = R"([
+        ["str_0", 0, "blob_data_0"],
+        ["str_1", 1, "blob_data_1"],
+        ["str_2", 2, "blob_data_2"]
+    ])";
+    auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"}));
+
+    // Point row 1's descriptor at an offset beyond the end of its file so the write observes
+    // a fetch failure that is not a missing file.
+    ASSERT_OK_AND_ASSIGN(desc_array, CorruptDescriptorOffset(desc_array, "blob", /*row=*/1));
+
+    auto schema = arrow::schema(fields);
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
+                         WriteArray(table_path, {}, schema->field_names(), {desc_array}));
+    ASSERT_OK(Commit(table_path, commit_msgs));
+
+    // Both the main data file and the .blob file keep the full row count; the read path
+    // merges them back into rows where row 1's blob is NULL.
+    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
+    VerifyDataFileMetas(plan, /*expected_file_count=*/2, /*expected_row_counts=*/{3, 3},
+                        /*expected_min_seqs=*/{1, 1}, /*expected_max_seqs=*/{1, 1},
+                        /*expected_first_row_ids=*/{0, 0},
+                        /*expected_write_cols=*/
+                        {std::vector<std::string>{"f0", "f1"}, std::vector<std::string>{"blob"}});
+
+    ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), plan));
+    ASSERT_TRUE(result.chunked_array);
+    auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie();
+    auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
+    ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"blob"}));
+
+    std::string expected_json = R"([
+        ["str_0", 0, "blob_data_0"],
+        ["str_1", 1, null],
+        ["str_2", 2, "blob_data_2"]
+    ])";
+    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
+            .ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_array));
+    ASSERT_TRUE(resolved->Equals(expected_with_rk))
+        << "result:" << resolved->ToString() << std::endl
+        << "expected:" << expected_with_rk->ToString();
+}
+
+TEST_P(BlobTableInteTest, TestWriteNullOnFetchFailureKeepsMissingFileFailing) {
+    // blob-write-null-on-fetch-failure only converts non-NotExist open failures; a missing
+    // descriptor file still fails the write when only this option is enabled.
+    arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
+                                 arrow::field("f1", arrow::int32()),
+                                 BlobUtils::ToArrowField("blob", true)};
+
+    std::map<std::string, std::string> options = {
+        {Options::MANIFEST_FORMAT, "orc"},
+        {Options::FILE_FORMAT, GetParam()},
+        {Options::TARGET_FILE_SIZE, "700"},
+        {Options::BUCKET, "-1"},
+        {Options::ROW_TRACKING_ENABLED, "true"},
+        {Options::DATA_EVOLUTION_ENABLED, "true"},
+        {Options::BLOB_AS_DESCRIPTOR, "true"},
+        {Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, "true"},
+        {Options::FILE_SYSTEM, "local"}};
+    CreateTable(fields, /*partition_keys=*/{}, options);
+    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+    std::string raw_json = R"([
+        ["str_0", 0, "blob_data_0"],
+        ["str_1", 1, "blob_data_1"]
+    ])";
+    auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"}));
+    ASSERT_OK(DeleteDescriptorTarget(desc_array, "blob", /*row=*/1));
+
+    auto schema = arrow::schema(fields);
+    ASSERT_NOK_WITH_MSG(WriteArray(table_path, {}, schema->field_names(), {desc_array}),
+                        "not exists");
+}
+
 TEST_P(BlobTableInteTest, TestBasic) {
     CreateTable();
     std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");