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

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

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

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()

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";

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());

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,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,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
@@ -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