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

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

@@ -1643,6 +1643,9 @@ 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.
+    # Every library that uses dl itself (arrow here; also lucene and jindosdk::nextarch)
+    # declares it on its own interface the same way; consumers inherit it transitively
+    # and must not list ${CMAKE_DL_LIBS} again themselves.
     target_link_libraries(arrow
                           INTERFACE zstd
                                     snappy

include/paimon/defs.h

@@ -440,13 +440,14 @@ struct PAIMON_EXPORT Options {
     /// 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".
+    /// the referenced file does not exist at write time. When false, a missing file is treated
+    /// like any other fetch failure, following "blob-write-null-on-fetch-failure". 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".
+    /// offset). A missing file is handled by "blob-write-null-on-missing-file" when that option is
+    /// enabled. 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[];

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();
 }
 
@@ -140,29 +150,20 @@ Status BlobFormatWriter::Finish() {
 Status BlobFormatWriter::WriteBlob(std::string_view blob_data) {
     // 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.
+    // Whether blob_data is a serialized BlobDescriptor is detected by its magic header rather
+    // than taken from a blob_as_descriptor option, so each row may hold either form.
     std::unique_ptr<InputStream> in;
     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 +207,73 @@ 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, since file system implementations disagree on which status a missing file maps to.
+    // The check runs only when `write_null_on_missing_file_` needs the classification; otherwise
+    // a missing file gets the same treatment as any other failed open.
+    if (write_null_on_missing_file_) {
+        Result<bool> exists = fs_->Exists(blob->Uri());
+        if (exists.ok()) {
+            if (!exists.value()) {
+                return HandleMissingFile(blob->Uri());
+            }
+        } else if (!write_null_on_fetch_failure_) {
+            // The check cannot answer whether the file is there; with no fetch-failure handling
+            // to defer to, fail rather than assume either answer.
+            const Status& status = exists.status();
+            return status.WithMessage("failed to check existence of blob file '", blob->Uri(),
+                                      "': ", status.message());
+        }
+        // A failed check is otherwise deferred to the open below, which can still succeed.
+    }
+
+    Result<std::unique_ptr<InputStream>> opened = blob->NewInputStream(fs_);
+    if (!opened.ok()) {
+        // The file can be deleted between the check above and this open. Classifying that from
+        // `opened.status()` would reintroduce the plugin-specific status codes this writer avoids,
+        // so ask FileSystem::Exists once more. This narrows the window rather than closing it; a
+        // check that cannot answer falls through to the open failure.
+        if (write_null_on_missing_file_) {
+            Result<bool> exists = fs_->Exists(blob->Uri());
+            if (exists.ok() && !exists.value()) {
+                return HandleMissingFile(blob->Uri());
+            }
+        }
+        return HandleFetchFailure(blob->Uri(), opened.status());
+    }
+    return std::move(opened).value();
+}
+
+std::unique_ptr<InputStream> BlobFormatWriter::HandleMissingFile(const std::string& 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>> 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,27 @@ 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. The existence check runs only when
+    /// `write_null_on_missing_file` is enabled; otherwise a missing file follows
+    /// `write_null_on_fetch_failure` like any other failed open.
+    /// 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 +97,19 @@ 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);
+
+    /// Convert a file that FileSystem::Exists reported as absent to a NULL element: count it and
+    /// return a null stream. Only reached under `write_null_on_missing_file_`, which callers check.
+    std::unique_ptr<InputStream> HandleMissingFile(const std::string& blob_uri);
+
+    /// 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 +124,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_;
 };