fix(blob): identify a missing blob file with FileSystem::Exists

BlobFormatWriter told a missing file apart from a fetch failure by testing
Status::NotExist on a failed descriptor open. That makes the two write-null
options depend on every file system plugin agreeing on which status code a
missing file maps to, which is hard to keep uniform across all plugin
interfaces: supporting it already required mapping JDO_FILE_NOT_FOUND_ERROR to
Status::NotExist in the jindo error macro, and any plugin that reports a
missing file as a plain IO error silently disables
'blob-write-null-on-missing-file'.

Decide it with FileSystem::Exists instead. Java classifies the same way,
through UriReader.FileUriReader.exists -> FileIO.exists; the only error code
its writer tests is HTTP 404, which is protocol-defined rather than
plugin-specific. The check runs only when the two options treat the two cases
differently, so the default path pays nothing for it.

Handle the two failures that were reachable but unclassified:

- A failed existence check leaves it unknown whether the file is there, so it
  is reported as a fetch failure rather than assuming either answer.
- A descriptor that cannot be deserialized is a fetch failure too, which is
  what BLOB_WRITE_NULL_ON_FETCH_FAILURE already documented but the code did
  not do; its status keeps its original code and gains an "invalid blob
  descriptor" prefix so the cause is named.

Create() now rejects a null file system, so that configuration error surfaces
at construction instead of being silently written out as NULL rows.

Report both conversions as blob.write.null-on-missing-file.count and
blob.write.null-on-fetch-failure.count, and name the blob URI, the BLOB field
and the output file in the write-null warnings.

Also drop the ${CMAKE_DL_LIBS} entries listed by hand on targets that never
call dlopen/dlsym/dladdr themselves -- a grep over src/ and include/ finds no
such call. Those entries were carried on behalf of arrow, which declares dl as
a usage requirement on its own interface, so CMake places -ldl after
libarrow.a automatically on linkers that resolve symbols strictly
left-to-right, and every target that dropped an entry links arrow and inherits
it transitively. dl is now declared only on the real providers (arrow, lucene,
jindosdk::nextarch), with the convention recorded next to arrow's declaration
so consumers do not add it back. Follow-up to the review comment on alibaba#421.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: SteNicholas

benchmark/CMakeLists.txt

@@ -57,7 +57,6 @@ if(PAIMON_BUILD_BENCHMARKS)
                          ${PAIMON_BENCHMARK_STATIC_LINK_LIBS}
                          test_utils_static
                          Threads::Threads
-                         ${CMAKE_DL_LIBS}
                          ${PAIMON_BENCHMARK_PLATFORM_LINK_LIBS}
                          ${PAIMON_BENCHMARK_LINK_TOOLCHAIN}
                          EXTRA_INCLUDES

cmake_modules/ThirdpartyToolchain.cmake

@@ -1638,6 +1638,8 @@ macro(build_arrow)
 
     # 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.
+    # This is the single source of truth for dl: targets linking arrow inherit it
+    # transitively and must not list ${CMAKE_DL_LIBS} again themselves.
     target_link_libraries(arrow
                           INTERFACE zstd
                                     snappy

src/paimon/CMakeLists.txt

@@ -392,7 +392,6 @@ add_paimon_lib(paimon
                arrow
                tbb
                glog
-               ${CMAKE_DL_LIBS}
                fmt
                roaring_bitmap
                xxhash

src/paimon/common/file_index/CMakeLists.txt

@@ -42,7 +42,6 @@ add_paimon_lib(paimon_file_index
                arrow
                fmt
                xxhash
-               ${CMAKE_DL_LIBS}
                Threads::Threads
                SHARED_LINK_LIBS
                paimon_shared

src/paimon/common/global_index/CMakeLists.txt

@@ -38,7 +38,6 @@ add_paimon_lib(paimon_global_index
                STATIC_LINK_LIBS
                arrow
                fmt
-               ${CMAKE_DL_LIBS}
                Threads::Threads
                SHARED_LINK_LIBS
                paimon_shared

src/paimon/format/avro/CMakeLists.txt

@@ -38,7 +38,6 @@ if(PAIMON_ENABLE_AVRO)
                    fmt
                    avro
                    tbb
-                   ${CMAKE_DL_LIBS}
                    Threads::Threads
                    SHARED_LINK_LIBS
                    paimon_shared

src/paimon/format/blob/CMakeLists.txt

@@ -25,7 +25,6 @@ add_paimon_lib(paimon_blob_file_format
                STATIC_LINK_LIBS
                arrow
                fmt
-               ${CMAKE_DL_LIBS}
                Threads::Threads
                SHARED_LINK_LIBS
                paimon_shared

src/paimon/format/blob/blob_format_writer.cpp

@@ -30,6 +30,7 @@
 #include "paimon/common/utils/arrow/status_utils.h"
 #include "paimon/common/utils/delta_varint_compressor.h"
 #include "paimon/data/blob.h"
+#include "paimon/fs/file_system.h"
 #include "paimon/io/byte_array_input_stream.h"
 #include "paimon/logging.h"
 
@@ -48,6 +49,8 @@ BlobFormatWriter::BlobFormatWriter(const std::shared_ptr<OutputStream>& out, con
       pool_(pool),
       write_null_on_missing_file_(write_null_on_missing_file),
       write_null_on_fetch_failure_(write_null_on_fetch_failure) {
+    // Create() has already checked that data_type has exactly one BLOB field.
+    blob_field_name_ = data_type_->field(0)->name();
     metrics_ = std::make_shared<MetricsImpl>();
     tmp_buffer_ = Bytes::AllocateBytes(kTmpBufferSize, pool_.get());
     magic_number_bytes_ = IntegerToLittleEndian<int32_t>(BlobDefs::kMagicNumber, pool_);
@@ -67,6 +70,9 @@ Result<std::unique_ptr<BlobFormatWriter>> BlobFormatWriter::Create(
     if (pool == nullptr) {
         return Status::Invalid("blob format writer create failed. pool is nullptr");
     }
+    if (fs == nullptr) {
+        return Status::Invalid("blob format writer create failed. fs is nullptr");
+    }
     if (data_type->num_fields() != 1) {
         return Status::Invalid(
             fmt::format("blob data type field number {} is not 1", data_type->num_fields()));
@@ -117,6 +123,10 @@ Status BlobFormatWriter::AddBatch(ArrowArray* batch) {
 }
 
 Status BlobFormatWriter::Flush() {
+    metrics_->SetCounter(BlobMetrics::WRITE_NULL_ON_MISSING_FILE_COUNT,
+                         null_on_missing_file_count_);
+    metrics_->SetCounter(BlobMetrics::WRITE_NULL_ON_FETCH_FAILURE_COUNT,
+                         null_on_fetch_failure_count_);
     return out_->Flush();
 }
 
@@ -147,22 +157,14 @@ 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) {
-        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;
+        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InputStream> descriptor_in,
+                               OpenDescriptorInputStream(blob_data));
+        // A null stream means a write-null option already converted the failure.
+        if (descriptor_in == nullptr) {
+            bin_lengths_.push_back(BlobDefs::kNullBinLength);
+            return Status::OK();
         }
-        in = std::move(opened).value();
+        in = std::move(descriptor_in);
     } else {
         in = std::make_unique<ByteArrayInputStream>(blob_data.data(), blob_data.size());
     }
@@ -206,10 +208,60 @@ Status BlobFormatWriter::WriteBlob(std::string_view blob_data) {
 }
 
 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_);
+    std::string_view blob_data) {
+    // A descriptor that cannot be deserialized is a fetch failure: the referenced data cannot be
+    // reached. Its URI is inside the unreadable bytes, hence the placeholder; the underlying
+    // status comes from the byte reader and never mentions blobs, hence the added context.
+    Result<std::unique_ptr<Blob>> blob_result =
+        Blob::FromDescriptor(blob_data.data(), blob_data.size());
+    if (!blob_result.ok()) {
+        const Status& status = blob_result.status();
+        return HandleFetchFailure(
+            "<unknown>", status.WithMessage("invalid blob descriptor: ", status.message()));
+    }
+    std::unique_ptr<Blob> blob = std::move(blob_result).value();
+
+    // A missing file is identified by FileSystem::Exists rather than by the status of a failed
+    // open: file system implementations disagree on which status a missing file maps to. Java
+    // classifies the same way and also checks up front, which costs one extra call per row whose
+    // file is present; deferring the check until an open fails would avoid that, but the up-front
+    // structure is kept deliberately. Skipped when neither option is enabled, since the two cases
+    // then get the same treatment anyway.
+    if (write_null_on_missing_file_ || write_null_on_fetch_failure_) {
+        Result<bool> exists = fs_->Exists(blob->Uri());
+        if (!exists.ok()) {
+            // Unknown whether the file is there, so do not assume either answer.
+            return HandleFetchFailure(blob->Uri(), exists.status());
+        }
+        if (!exists.value()) {
+            if (!write_null_on_missing_file_) {
+                return Status::NotExist(fmt::format("File '{}' does not exist", blob->Uri()));
+            }
+            PAIMON_LOG_WARN(logger_,
+                            "Blob file %s does not exist, writing NULL for BLOB field %s into %s",
+                            blob->Uri().c_str(), blob_field_name_.c_str(), uri_.c_str());
+            ++null_on_missing_file_count_;
+            return std::unique_ptr<InputStream>();
+        }
+    }
+
+    Result<std::unique_ptr<InputStream>> opened = blob->NewInputStream(fs_);
+    if (!opened.ok()) {
+        return HandleFetchFailure(blob->Uri(), opened.status());
+    }
+    return std::move(opened).value();
+}
+
+Result<std::unique_ptr<InputStream>> BlobFormatWriter::HandleFetchFailure(
+    const std::string& blob_uri, const Status& status) {
+    if (!write_null_on_fetch_failure_) {
+        return status;
+    }
+    PAIMON_LOG_WARN(logger_, "Failed to fetch blob %s, writing NULL for BLOB field %s into %s: %s",
+                    blob_uri.c_str(), blob_field_name_.c_str(), uri_.c_str(),
+                    status.ToString().c_str());
+    ++null_on_fetch_failure_count_;
+    return std::unique_ptr<InputStream>();
 }
 
 Status BlobFormatWriter::WriteBytes(const char* data, int64_t length) {

src/paimon/format/blob/blob_format_writer.h

@@ -47,14 +47,25 @@ class OutputStream;
 
 namespace paimon::blob {
 
+class BlobMetrics {
+ public:
+    /// Number of rows written as NULL because their referenced file did not exist.
+    static inline const char WRITE_NULL_ON_MISSING_FILE_COUNT[] =
+        "blob.write.null-on-missing-file.count";
+    /// Number of rows written as NULL because their referenced data could not be reached.
+    static inline const char WRITE_NULL_ON_FETCH_FAILURE_COUNT[] =
+        "blob.write.null-on-fetch-failure.count";
+};
+
 // Blob format:
 // 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.
+    /// `write_null_on_missing_file` converts a descriptor whose referenced file does not exist
+    /// (as reported by FileSystem::Exists) to a NULL element, and `write_null_on_fetch_failure`
+    /// converts any other failure to access the referenced data; 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,
@@ -84,8 +95,15 @@ class BlobFormatWriter : public FormatWriter {
     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;
+    /// Returns a null stream when the failure is converted to a NULL element by
+    /// `write_null_on_missing_file_` or `write_null_on_fetch_failure_`.
+    Result<std::unique_ptr<InputStream>> OpenDescriptorInputStream(std::string_view blob_data);
+
+    /// Apply `write_null_on_fetch_failure_` to a failure to reach the referenced data: returns
+    /// `status` when the option is disabled, and a null stream when it converts the failure to a
+    /// NULL element. `blob_uri` is only used for logging.
+    Result<std::unique_ptr<InputStream>> HandleFetchFailure(const std::string& blob_uri,
+                                                            const Status& status);
 
     Status WriteBytes(const char* data, int64_t length);
     Status WriteWithCrc32(const char* data, int64_t length);
@@ -100,15 +118,19 @@ class BlobFormatWriter : public FormatWriter {
     uint32_t crc32_ = 0;
     std::vector<int64_t> bin_lengths_;
     std::shared_ptr<OutputStream> out_;
+    /// Path of the blob file being written, not of any referenced blob.
     std::string uri_;
     PAIMON_UNIQUE_PTR<Bytes> tmp_buffer_;
     PAIMON_UNIQUE_PTR<Bytes> magic_number_bytes_;
     std::shared_ptr<arrow::DataType> data_type_;
+    std::string blob_field_name_;
     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;
+    uint64_t null_on_missing_file_count_ = 0;
+    uint64_t null_on_fetch_failure_count_ = 0;
     std::unique_ptr<Logger> logger_;
 };