feat(blob): support blob-write-null-on-missing-file and blob-write-null-on-fetch-failure options

Introduce two table options for descriptor BLOB writes and thread them
from the table options map through BlobFileFormat/BlobWriterBuilder
into BlobFormatWriter:

- blob-write-null-on-missing-file: write NULL instead of failing the
  write when the file referenced by a blob descriptor does not exist.
- blob-write-null-on-fetch-failure: write NULL when the referenced data
  cannot be fetched for other reasons, e.g. an invalid offset.

BlobFormatWriter::WriteBlob now opens the descriptor input stream
before writing any bytes, so a rejected row leaves no partial data in
the output stream, while failures during the streaming copy still fail
the write. JindoFileSystem maps JDO_FILE_NOT_FOUND_ERROR to
Status::NotExist in Open() and JindoInputStream::Length() so that a
missing OSS object is classified as a missing file rather than a
generic IO error.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: SteNicholas

include/paimon/defs.h

@@ -416,6 +416,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

src/paimon/CMakeLists.txt

@@ -773,6 +773,7 @@ if(PAIMON_BUILD_TESTS)
                     fs/local/local_file_test.cpp
                     # fs/jindo/jindo_file_system_factory_test.cpp
                     # fs/jindo/jindo_file_system_test.cpp
+                    # fs/jindo/jindo_utils_test.cpp
                     STATIC_LINK_LIBS
                     paimon_shared
                     ${PAIMON_LOCAL_FILE_SYSTEM_STATIC_LINK_LIBS}

src/paimon/common/defs.cpp

@@ -104,6 +104,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";

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

src/paimon/format/blob/blob_format_writer.cpp

@@ -31,22 +31,33 @@
 #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,
                                    const std::shared_ptr<FileSystem>& fs,
-                                   const std::shared_ptr<MemoryPool>& pool)
-    : out_(out), uri_(uri), data_type_(data_type), fs_(fs), pool_(pool) {
+                                   const std::shared_ptr<MemoryPool>& pool,
+                                   bool write_null_on_missing_file,
+                                   bool write_null_on_fetch_failure)
+    : 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,
-    const std::shared_ptr<FileSystem>& fs, const std::shared_ptr<MemoryPool>& pool) {
+    const std::shared_ptr<FileSystem>& fs, const std::shared_ptr<MemoryPool>& pool,
+    bool write_null_on_missing_file, bool write_null_on_fetch_failure) {
     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, fs, pool, write_null_on_missing_file, write_null_on_fetch_failure));
 }
 
 Status BlobFormatWriter::AddBatch(ArrowArray* batch) {
@@ -126,27 +138,41 @@ 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.
     std::unique_ptr<InputStream> in;
     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) {

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
@@ -51,7 +53,8 @@ class BlobFormatWriter : public FormatWriter {
  public:
     static Result<std::unique_ptr<BlobFormatWriter>> Create(
         const std::shared_ptr<OutputStream>& out, const std::shared_ptr<arrow::DataType>& data_type,
-        const std::shared_ptr<FileSystem>& fs, const std::shared_ptr<MemoryPool>& pool);
+        const std::shared_ptr<FileSystem>& fs, const std::shared_ptr<MemoryPool>& pool,
+        bool write_null_on_missing_file = false, bool write_null_on_fetch_failure = false);
 
     Status AddBatch(ArrowArray* batch) override;
 
@@ -70,11 +73,15 @@ class BlobFormatWriter : public FormatWriter {
  private:
     BlobFormatWriter(const std::shared_ptr<OutputStream>& out, const std::string& uri,
                      const std::shared_ptr<arrow::DataType>& data_type,
-                     const std::shared_ptr<FileSystem>& fs,
-                     const std::shared_ptr<MemoryPool>& pool);
+                     const std::shared_ptr<FileSystem>& fs, const std::shared_ptr<MemoryPool>& pool,
+                     bool write_null_on_missing_file, bool write_null_on_fetch_failure);
 
     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 +102,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