fix(blob): address review comments on write-null options

- Reorder the write-null parameters before fs/pool in BlobFormatWriter
  to match the Java constructor order, and document them on Create().
- Map JDO_FILE_NOT_FOUND_ERROR to Status::NotExist directly in
  PAIMON_RETURN_NOT_OK_FROM_JINDO and drop the _WITH_NOT_EXIST variant,
  making all jindo call sites consistent with LocalFileSystem.
- Enable jindo_utils_test in fs_test when PAIMON_ENABLE_JINDO is ON; it
  only checks in-memory status conversion and needs no OSS cluster.
- Add end-to-end write-null tests in blob_table_inte_test.cpp covering
  the NULL element merge with data files and the failure paths.

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

Author: SteNicholas

src/paimon/CMakeLists.txt

@@ -790,19 +790,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
-                    # fs/jindo/jindo_utils_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/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_format_writer.cpp

@@ -37,10 +37,10 @@ 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,
                                    bool write_null_on_missing_file,
-                                   bool write_null_on_fetch_failure)
+                                   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),
@@ -56,8 +56,8 @@ BlobFormatWriter::BlobFormatWriter(const std::shared_ptr<OutputStream>& out, con
 
 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,
-    bool write_null_on_missing_file, bool write_null_on_fetch_failure) {
+    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");
     }
@@ -77,7 +77,7 @@ Result<std::unique_ptr<BlobFormatWriter>> BlobFormatWriter::Create(
     }
     PAIMON_ASSIGN_OR_RAISE(std::string uri, out->GetUri());
     return std::unique_ptr<BlobFormatWriter>(new BlobFormatWriter(
-        out, uri, data_type, fs, pool, write_null_on_missing_file, write_null_on_fetch_failure));
+        out, uri, data_type, write_null_on_missing_file, write_null_on_fetch_failure, fs, pool));
 }
 
 Status BlobFormatWriter::AddBatch(ArrowArray* batch) {

src/paimon/format/blob/blob_format_writer.h

@@ -51,10 +51,14 @@ 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,
-        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);
+        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;
 
@@ -73,8 +77,9 @@ 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,
-                     bool write_null_on_missing_file, bool write_null_on_fetch_failure);
+                     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);