From 7b36b8fa013ff248ff94dd7c16cad3ed92aaaab7 Mon Sep 17 00:00:00 2001 From: Nicholas Jiang Date: Tue, 14 Jul 2026 20:41:35 +0800 Subject: [PATCH 1/4] 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 --- include/paimon/defs.h | 9 + src/paimon/CMakeLists.txt | 1 + src/paimon/common/defs.cpp | 2 + .../blob/blob_file_format_factory_test.cpp | 58 +++++ src/paimon/format/blob/blob_format_writer.cpp | 61 ++++- src/paimon/format/blob/blob_format_writer.h | 16 +- .../format/blob/blob_format_writer_test.cpp | 243 ++++++++++++++++-- src/paimon/format/blob/blob_writer_builder.h | 11 +- .../format/blob/blob_writer_builder_test.cpp | 90 ++++++- src/paimon/fs/jindo/jindo_file_system.cpp | 8 +- src/paimon/fs/jindo/jindo_utils.h | 15 ++ src/paimon/fs/jindo/jindo_utils_test.cpp | 57 ++++ src/paimon/testing/utils/test_helper.h | 22 ++ 13 files changed, 550 insertions(+), 43 deletions(-) create mode 100644 src/paimon/fs/jindo/jindo_utils_test.cpp diff --git a/include/paimon/defs.h b/include/paimon/defs.h index 772dfee5a..8b4fa8c8e 100644 --- a/include/paimon/defs.h +++ b/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 diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index 3841728e5..89735b102 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -798,6 +798,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} diff --git a/src/paimon/common/defs.cpp b/src/paimon/common/defs.cpp index 501068f32..cf182a53c 100644 --- a/src/paimon/common/defs.cpp +++ b/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"; diff --git a/src/paimon/format/blob/blob_file_format_factory_test.cpp b/src/paimon/format/blob/blob_file_format_factory_test.cpp index 6409ea85a..f1e7a84ae 100644 --- a/src/paimon/format/blob/blob_file_format_factory_test.cpp +++ b/src/paimon/format/blob/blob_file_format_factory_test.cpp @@ -16,8 +16,24 @@ #include "paimon/format/blob/blob_file_format_factory.h" +#include +#include +#include + +#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 dir = + paimon::test::UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + std::shared_ptr fs = std::make_shared(); + auto struct_type = arrow::struct_({BlobUtils::ToArrowField("blob_col", true)}); + ASSERT_OK_AND_ASSIGN(std::shared_ptr missing_blob, + Blob::FromPath(dir->Str() + "/not_exist_file", /*offset=*/0, + /*length=*/10)); + + auto write_once = [&](const std::map& options, + const std::string& file_name) -> Status { + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr 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 writer_builder, + format->CreateWriterBuilder(&c_schema, /*batch_size=*/1024)); + // The blob writer builder is a SpecificFSWriterBuilder by construction. + static_cast(writer_builder.get())->WithFileSystem(fs); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr out, + fs->Create(dir->Str() + "/" + file_name, /*overwrite=*/true)); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr writer, + writer_builder->Build(out, "none")); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr 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 diff --git a/src/paimon/format/blob/blob_format_writer.cpp b/src/paimon/format/blob/blob_format_writer.cpp index 146ffebf4..07c7bb883 100644 --- a/src/paimon/format/blob/blob_format_writer.cpp +++ b/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& out, const std::string& uri, const std::shared_ptr& data_type, const std::shared_ptr& fs, - const std::shared_ptr& pool) - : out_(out), uri_(uri), data_type_(data_type), fs_(fs), pool_(pool) { + const std::shared_ptr& 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(); tmp_buffer_ = Bytes::AllocateBytes(kTmpBufferSize, pool_.get()); magic_number_bytes_ = IntegerToLittleEndian(BlobDefs::kMagicNumber, pool_); + logger_ = Logger::GetLogger("BlobFormatWriter"); } Result> BlobFormatWriter::Create( const std::shared_ptr& out, const std::shared_ptr& data_type, - const std::shared_ptr& fs, const std::shared_ptr& pool) { + const std::shared_ptr& fs, const std::shared_ptr& 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> 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(new BlobFormatWriter(out, uri, data_type, fs, pool)); + return std::unique_ptr(new BlobFormatWriter( + out, uri, data_type, fs, pool, write_null_on_missing_file, write_null_on_fetch_failure)); } Status BlobFormatWriter::AddBatch(ArrowArray* batch) { @@ -126,13 +138,8 @@ 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. @@ -140,13 +147,32 @@ 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) { - PAIMON_ASSIGN_OR_RAISE(std::unique_ptr blob, - Blob::FromDescriptor(blob_data.data(), blob_data.size())); - PAIMON_ASSIGN_OR_RAISE(in, blob->NewInputStream(fs_)); + Result> 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(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(tmp_buffer_->size())); while (read_len > 0) { @@ -179,6 +205,13 @@ Status BlobFormatWriter::WriteBlob(std::string_view blob_data) { return Status::OK(); } +Result> BlobFormatWriter::OpenDescriptorInputStream( + std::string_view blob_data) const { + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr 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) { diff --git a/src/paimon/format/blob/blob_format_writer.h b/src/paimon/format/blob/blob_format_writer.h index f62c49746..efa12016d 100644 --- a/src/paimon/format/blob/blob_format_writer.h +++ b/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> Create( const std::shared_ptr& out, const std::shared_ptr& data_type, - const std::shared_ptr& fs, const std::shared_ptr& pool); + const std::shared_ptr& fs, const std::shared_ptr& 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& out, const std::string& uri, const std::shared_ptr& data_type, - const std::shared_ptr& fs, - const std::shared_ptr& pool); + const std::shared_ptr& fs, const std::shared_ptr& 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> 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 fs_; std::shared_ptr pool_; std::shared_ptr metrics_; + bool write_null_on_missing_file_ = false; + bool write_null_on_fetch_failure_ = false; + std::unique_ptr logger_; }; } // namespace paimon::blob diff --git a/src/paimon/format/blob/blob_format_writer_test.cpp b/src/paimon/format/blob/blob_format_writer_test.cpp index b3f5cf825..b5d66bbc6 100644 --- a/src/paimon/format/blob/blob_format_writer_test.cpp +++ b/src/paimon/format/blob/blob_format_writer_test.cpp @@ -32,10 +32,24 @@ #include "paimon/testing/utils/testharness.h" namespace paimon::blob::test { -class BlobFormatWriterTest : public ::testing::Test, public ::testing::WithParamInterface { + +/// A file system whose Open() always fails with the configured status, for verifying how the +/// writer classifies open failures by status code. +class OpenFailFileSystem : public LocalFileSystem { + public: + explicit OpenFailFileSystem(Status open_status) : open_status_(std::move(open_status)) {} + + Result> Open(const std::string& path) const override { + return open_status_; + } + + private: + Status open_status_; +}; + +class BlobFormatWriterTestBase : public ::testing::Test { public: void SetUp() override { - blob_as_descriptor_ = GetParam(); pool_ = GetDefaultPool(); dir_ = paimon::test::UniqueTestDirectory::Create(); ASSERT_TRUE(dir_); @@ -50,42 +64,78 @@ class BlobFormatWriterTest : public ::testing::Test, public ::testing::WithParam ASSERT_OK(output_stream_->Close()); } + Status AddBatchOnce(const std::shared_ptr& format_writer, + const std::shared_ptr& blob_array) const { + auto c_array = std::make_unique(); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*blob_array, c_array.get())); + return format_writer->AddBatch(c_array.get()); + } + + Result> PrepareDescriptorArray( + const std::shared_ptr& blob) const { + return paimon::test::TestHelper::MakeBlobDescriptorArray(struct_type_, blob, pool_); + } + + Result> ReadBackAsData() const { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr input_stream, + file_system_->Open(dir_->Str() + "/file.blob")); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr reader, + BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024, + /*blob_as_descriptor=*/false, pool_)); + auto schema = arrow::schema(struct_type_->fields()); + ::ArrowSchema c_schema; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &c_schema)); + PAIMON_RETURN_NOT_OK(reader->SetReadSchema(&c_schema, /*predicate=*/nullptr, + /*selection_bitmap=*/std::nullopt)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr chunked_array, + paimon::test::ReadResultCollector::CollectResult(reader.get())); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr concat_array, + arrow::Concatenate(chunked_array->chunks())); + return arrow::internal::checked_pointer_cast(concat_array); + } + + protected: + std::shared_ptr pool_; + std::unique_ptr dir_; + std::shared_ptr output_stream_; + std::shared_ptr file_system_; + std::shared_ptr struct_type_; +}; + +class BlobFormatWriterTest : public BlobFormatWriterTestBase, + public ::testing::WithParamInterface { + public: + void SetUp() override { + blob_as_descriptor_ = GetParam(); + BlobFormatWriterTestBase::SetUp(); + } + Result> PrepareBlobArray( const std::shared_ptr& blob) const { + if (blob_as_descriptor_) { + return PrepareDescriptorArray(blob); + } arrow::StructBuilder struct_builder(struct_type_, arrow::default_memory_pool(), {std::make_shared()}); auto blob_builder = static_cast(struct_builder.field_builder(0)); PAIMON_RETURN_NOT_OK_FROM_ARROW(struct_builder.Append()); - if (blob_as_descriptor_) { - PAIMON_RETURN_NOT_OK_FROM_ARROW(blob_builder->Append( - blob->ToDescriptor(pool_)->data(), blob->ToDescriptor(pool_)->size())); - } else { - PAIMON_ASSIGN_OR_RAISE(auto blob_data, blob->ToData(file_system_, pool_)); - PAIMON_RETURN_NOT_OK_FROM_ARROW( - blob_builder->Append(blob_data->data(), blob_data->size())); - } + PAIMON_ASSIGN_OR_RAISE(PAIMON_UNIQUE_PTR blob_data, + blob->ToData(file_system_, pool_)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(blob_builder->Append(blob_data->data(), blob_data->size())); std::shared_ptr array; PAIMON_RETURN_NOT_OK_FROM_ARROW(struct_builder.Finish(&array)); return array; } - Status AddBatchOnce(const std::shared_ptr& format_writer, - const std::shared_ptr& blob_array) const { - auto c_array = std::make_unique(); - PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*blob_array, c_array.get())); - return format_writer->AddBatch(c_array.get()); - } - private: bool blob_as_descriptor_; - std::shared_ptr pool_; - std::unique_ptr dir_; - std::shared_ptr output_stream_; - std::shared_ptr file_system_; - std::shared_ptr struct_type_; }; +/// The write-null tests always feed descriptor bytes, so they do not depend on the +/// blob_as_descriptor_ parameter and run once on the non-parameterized fixture. +using BlobFormatWriterWriteNullTest = BlobFormatWriterTestBase; + INSTANTIATE_TEST_SUITE_P(BlobAsDescriptor, BlobFormatWriterTest, ::testing::Values(false, true)); TEST_P(BlobFormatWriterTest, TestSimple) { @@ -394,6 +444,155 @@ TEST_P(BlobFormatWriterTest, TestAddBatchWithNullValues) { ArrowArrayRelease(null_c_array.get()); } +TEST_F(BlobFormatWriterWriteNullTest, TestWriteNullOnMissingFile) { + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, + BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_, + /*write_null_on_missing_file=*/true, + /*write_null_on_fetch_failure=*/false)); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr missing_blob, + Blob::FromPath(dir_->Str() + "/not_exist_file", /*offset=*/0, + /*length=*/10)); + ASSERT_OK_AND_ASSIGN(auto missing_array, PrepareDescriptorArray(missing_blob)); + ASSERT_OK(AddBatchOnce(writer, missing_array)); + + // A fetch failure is not converted to NULL by write_null_on_missing_file alone (aligned + // with Java); the rejected row leaves the writer usable. + std::string file = paimon::test::GetDataDir() + "/xxhash.data"; + ASSERT_OK_AND_ASSIGN(std::shared_ptr bad_offset_blob, + Blob::FromPath(file, /*offset=*/1 << 20, /*length=*/10)); + ASSERT_OK_AND_ASSIGN(auto bad_offset_array, PrepareDescriptorArray(bad_offset_blob)); + ASSERT_NOK_WITH_MSG(AddBatchOnce(writer, bad_offset_array), "exceed total length"); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr blob, Blob::FromPath(file)); + ASSERT_OK_AND_ASSIGN(auto array, PrepareDescriptorArray(blob)); + ASSERT_OK(AddBatchOnce(writer, array)); + + ASSERT_OK(writer->Flush()); + ASSERT_OK(writer->Finish()); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr result_struct, ReadBackAsData()); + ASSERT_EQ(result_struct->length(), 2); + ASSERT_TRUE(result_struct->field(0)->IsNull(0)); + ASSERT_FALSE(result_struct->field(0)->IsNull(1)); + auto binary_array = + arrow::internal::checked_pointer_cast(result_struct->field(0)); + ASSERT_OK_AND_ASSIGN(auto expected_data, blob->ToData(file_system_, pool_)); + ASSERT_EQ(binary_array->GetView(1), + std::string_view(expected_data->data(), expected_data->size())); +} + +TEST_F(BlobFormatWriterWriteNullTest, TestWriteNullOnFetchFailure) { + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, + BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_, + /*write_null_on_missing_file=*/false, + /*write_null_on_fetch_failure=*/true)); + + std::string file = paimon::test::GetDataDir() + "/xxhash.data"; + ASSERT_OK_AND_ASSIGN(std::shared_ptr bad_offset_blob, + Blob::FromPath(file, /*offset=*/1 << 20, /*length=*/10)); + ASSERT_OK_AND_ASSIGN(auto bad_offset_array, PrepareDescriptorArray(bad_offset_blob)); + ASSERT_OK(AddBatchOnce(writer, bad_offset_array)); + + // A missing file is not converted to NULL by write_null_on_fetch_failure alone (aligned + // with Java); the rejected row leaves the writer usable. + ASSERT_OK_AND_ASSIGN(std::shared_ptr missing_blob, + Blob::FromPath(dir_->Str() + "/not_exist_file", /*offset=*/0, + /*length=*/10)); + ASSERT_OK_AND_ASSIGN(auto missing_array, PrepareDescriptorArray(missing_blob)); + ASSERT_NOK_WITH_MSG(AddBatchOnce(writer, missing_array), "not exists"); + + ASSERT_OK(writer->Flush()); + ASSERT_OK(writer->Finish()); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr result_struct, ReadBackAsData()); + ASSERT_EQ(result_struct->length(), 1); + ASSERT_TRUE(result_struct->field(0)->IsNull(0)); +} + +TEST_F(BlobFormatWriterWriteNullTest, TestWriteNullOnBothOptionsEnabled) { + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, + BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_, + /*write_null_on_missing_file=*/true, + /*write_null_on_fetch_failure=*/true)); + + // Row 0: missing file -> NULL. + ASSERT_OK_AND_ASSIGN(std::shared_ptr missing_blob, + Blob::FromPath(dir_->Str() + "/not_exist_file", /*offset=*/0, + /*length=*/10)); + ASSERT_OK_AND_ASSIGN(auto missing_array, PrepareDescriptorArray(missing_blob)); + ASSERT_OK(AddBatchOnce(writer, missing_array)); + + // Row 1: fetch failure (offset beyond EOF) -> NULL. + std::string file = paimon::test::GetDataDir() + "/xxhash.data"; + ASSERT_OK_AND_ASSIGN(std::shared_ptr bad_offset_blob, + Blob::FromPath(file, /*offset=*/1 << 20, /*length=*/10)); + ASSERT_OK_AND_ASSIGN(auto bad_offset_array, PrepareDescriptorArray(bad_offset_blob)); + ASSERT_OK(AddBatchOnce(writer, bad_offset_array)); + + // Row 2: valid blob. + ASSERT_OK_AND_ASSIGN(std::shared_ptr blob, Blob::FromPath(file)); + ASSERT_OK_AND_ASSIGN(auto array, PrepareDescriptorArray(blob)); + ASSERT_OK(AddBatchOnce(writer, array)); + + ASSERT_OK(writer->Flush()); + ASSERT_OK(writer->Finish()); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr result_struct, ReadBackAsData()); + ASSERT_EQ(result_struct->length(), 3); + ASSERT_TRUE(result_struct->field(0)->IsNull(0)); + ASSERT_TRUE(result_struct->field(0)->IsNull(1)); + ASSERT_FALSE(result_struct->field(0)->IsNull(2)); + auto binary_array = + arrow::internal::checked_pointer_cast(result_struct->field(0)); + ASSERT_OK_AND_ASSIGN(auto expected_data, blob->ToData(file_system_, pool_)); + ASSERT_EQ(binary_array->GetView(2), + std::string_view(expected_data->data(), expected_data->size())); +} + +TEST_F(BlobFormatWriterWriteNullTest, TestWriteNullClassifiesByStatusCode) { + // The missing-file vs fetch-failure split keys on the open status code (NotExist <-> missing + // file), independent of the file system implementation. + ASSERT_OK_AND_ASSIGN(std::shared_ptr blob, + Blob::FromPath(dir_->Str() + "/any_file", /*offset=*/0, /*length=*/10)); + ASSERT_OK_AND_ASSIGN(auto array, PrepareDescriptorArray(blob)); + auto not_exist_fs = std::make_shared(Status::NotExist("mock not exist")); + auto io_error_fs = std::make_shared(Status::IOError("mock io error")); + + { + ASSERT_OK_AND_ASSIGN( + std::shared_ptr writer, + BlobFormatWriter::Create(output_stream_, struct_type_, not_exist_fs, pool_, + /*write_null_on_missing_file=*/true, + /*write_null_on_fetch_failure=*/false)); + ASSERT_OK(AddBatchOnce(writer, array)); + } + { + ASSERT_OK_AND_ASSIGN( + std::shared_ptr writer, + BlobFormatWriter::Create(output_stream_, struct_type_, io_error_fs, pool_, + /*write_null_on_missing_file=*/true, + /*write_null_on_fetch_failure=*/false)); + ASSERT_NOK_WITH_MSG(AddBatchOnce(writer, array), "mock io error"); + } + { + ASSERT_OK_AND_ASSIGN( + std::shared_ptr writer, + BlobFormatWriter::Create(output_stream_, struct_type_, io_error_fs, pool_, + /*write_null_on_missing_file=*/false, + /*write_null_on_fetch_failure=*/true)); + ASSERT_OK(AddBatchOnce(writer, array)); + } + { + ASSERT_OK_AND_ASSIGN( + std::shared_ptr writer, + BlobFormatWriter::Create(output_stream_, struct_type_, not_exist_fs, pool_, + /*write_null_on_missing_file=*/false, + /*write_null_on_fetch_failure=*/true)); + ASSERT_NOK_WITH_MSG(AddBatchOnce(writer, array), "mock not exist"); + } +} + TEST_P(BlobFormatWriterTest, TestAddBatchWithZeroLengthBlob) { ASSERT_OK_AND_ASSIGN( std::shared_ptr writer, diff --git a/src/paimon/format/blob/blob_writer_builder.h b/src/paimon/format/blob/blob_writer_builder.h index 4f59315f7..314885850 100644 --- a/src/paimon/format/blob/blob_writer_builder.h +++ b/src/paimon/format/blob/blob_writer_builder.h @@ -24,6 +24,8 @@ #include #include "arrow/api.h" +#include "paimon/common/utils/options_utils.h" +#include "paimon/defs.h" #include "paimon/format/blob/blob_format_writer.h" #include "paimon/format/format_writer.h" #include "paimon/format/writer_builder.h" @@ -65,7 +67,14 @@ class BlobWriterBuilder : public SpecificFSWriterBuilder { if (fs_ == nullptr) { return Status::Invalid("File system is nullptr. Please call WithFileSystem() first."); } - return BlobFormatWriter::Create(out, data_type_, fs_, pool_); + PAIMON_ASSIGN_OR_RAISE(bool write_null_on_missing_file, + OptionsUtils::GetValueFromMap( + options_, Options::BLOB_WRITE_NULL_ON_MISSING_FILE, false)); + PAIMON_ASSIGN_OR_RAISE(bool write_null_on_fetch_failure, + OptionsUtils::GetValueFromMap( + options_, Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, false)); + return BlobFormatWriter::Create(out, data_type_, fs_, pool_, write_null_on_missing_file, + write_null_on_fetch_failure); } private: diff --git a/src/paimon/format/blob/blob_writer_builder_test.cpp b/src/paimon/format/blob/blob_writer_builder_test.cpp index 01a077f96..d823c28b3 100644 --- a/src/paimon/format/blob/blob_writer_builder_test.cpp +++ b/src/paimon/format/blob/blob_writer_builder_test.cpp @@ -17,11 +17,15 @@ #include "paimon/format/blob/blob_writer_builder.h" #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/fs/file_system.h" #include "paimon/fs/local/local_file_system.h" +#include "paimon/testing/utils/test_helper.h" #include "paimon/testing/utils/testharness.h" namespace paimon::blob::test { @@ -33,7 +37,7 @@ class BlobWriterBuilderTest : public ::testing::Test { file_system_ = std::make_shared(); ASSERT_OK_AND_ASSIGN(output_stream_, file_system_->Create(dir_->Str() + "/file.blob", /*overwrite=*/true)); - struct_type_ = arrow::struct_({BlobUtils::ToArrowField("blob_col", false)}); + struct_type_ = arrow::struct_({BlobUtils::ToArrowField("blob_col", true)}); } void TearDown() override {} @@ -53,4 +57,88 @@ TEST_F(BlobWriterBuilderTest, TestSimple) { ASSERT_OK(builder.Build(output_stream_, "none")); } +TEST_F(BlobWriterBuilderTest, TestWriteNullOptions) { + auto prepare_batch = [&](const std::shared_ptr& blob, ArrowArray* c_array) -> Status { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr array, + paimon::test::TestHelper::MakeBlobDescriptorArray(struct_type_, blob, + GetDefaultPool())); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, c_array)); + return Status::OK(); + }; + // Each writer gets its own output file so that a Finish() in one scenario cannot leak + // bytes into the next. + int32_t file_index = 0; + auto build_writer = [&](const std::map& options) + -> Result> { + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr out, + file_system_->Create(dir_->Str() + "/file" + std::to_string(file_index++) + ".blob", + /*overwrite=*/true)); + BlobWriterBuilder builder(struct_type_, options); + builder.WithFileSystem(file_system_); + return builder.Build(out, "none"); + }; + + ASSERT_OK_AND_ASSIGN(std::shared_ptr missing_blob, + Blob::FromPath(dir_->Str() + "/not_exist_file", /*offset=*/0, + /*length=*/10)); + std::string data_file = dir_->Str() + "/data_file.bin"; + ASSERT_OK_AND_ASSIGN(auto data_file_stream, + file_system_->Create(data_file, /*overwrite=*/true)); + ASSERT_OK_AND_ASSIGN(int64_t written, data_file_stream->Write("blob data", 9)); + ASSERT_EQ(written, 9); + ASSERT_OK(data_file_stream->Flush()); + ASSERT_OK(data_file_stream->Close()); + // The offset beyond the end of the 9-byte file makes the descriptor a fetch failure. + ASSERT_OK_AND_ASSIGN(std::shared_ptr bad_offset_blob, + Blob::FromPath(data_file, /*offset=*/100, /*length=*/10)); + + // Both options default to false. + { + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, build_writer({})); + ArrowArray c_array; + ASSERT_OK(prepare_batch(missing_blob, &c_array)); + ASSERT_NOK_WITH_MSG(writer->AddBatch(&c_array), "not exists"); + } + { + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, build_writer({})); + ArrowArray c_array; + ASSERT_OK(prepare_batch(bad_offset_blob, &c_array)); + ASSERT_NOK_WITH_MSG(writer->AddBatch(&c_array), "exceed total length"); + } + + { + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, + build_writer({{Options::BLOB_WRITE_NULL_ON_MISSING_FILE, "true"}})); + ArrowArray c_array; + ASSERT_OK(prepare_batch(missing_blob, &c_array)); + ASSERT_OK(writer->AddBatch(&c_array)); + ASSERT_OK(writer->Finish()); + } + + { + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, + build_writer({{Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, "true"}})); + ArrowArray c_array; + ASSERT_OK(prepare_batch(bad_offset_blob, &c_array)); + ASSERT_OK(writer->AddBatch(&c_array)); + ASSERT_OK(writer->Finish()); + } + + // An explicit "false" exercises value parsing, not just the absent-key default. + { + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, + build_writer({{Options::BLOB_WRITE_NULL_ON_MISSING_FILE, "false"}, + {Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, "false"}})); + ArrowArray c_array; + ASSERT_OK(prepare_batch(missing_blob, &c_array)); + ASSERT_NOK_WITH_MSG(writer->AddBatch(&c_array), "not exists"); + } + + { + ASSERT_NOK_WITH_MSG(build_writer({{Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, "invalid"}}), + "convert key blob-write-null-on-fetch-failure"); + } +} + } // namespace paimon::blob::test diff --git a/src/paimon/fs/jindo/jindo_file_system.cpp b/src/paimon/fs/jindo/jindo_file_system.cpp index 6246780dd..64bb309c7 100644 --- a/src/paimon/fs/jindo/jindo_file_system.cpp +++ b/src/paimon/fs/jindo/jindo_file_system.cpp @@ -57,7 +57,8 @@ JindoFileSystem::JindoFileSystem(std::unique_ptr&& fs) Result> JindoFileSystem::Open(const std::string& path) const { std::unique_ptr reader; - PAIMON_RETURN_NOT_OK_FROM_JINDO(impl_->GetFileSystem()->openReader(path, &reader)); + PAIMON_RETURN_NOT_OK_FROM_JINDO_WITH_NOT_EXIST( + impl_->GetFileSystem()->openReader(path, &reader)); return std::make_unique(impl_, std::move(reader)); } @@ -208,7 +209,10 @@ Result JindoInputStream::GetPos() const { Result JindoInputStream::Length() const { int64_t len = -1; - PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->getFileLength(len)); + // openReader may succeed lazily for a missing object; keep the not-found mapping here so + // that Blob::NewInputStream(), which queries the length right after Open(), still observes + // Status::NotExist. Read failures keep the plain IOError mapping. + PAIMON_RETURN_NOT_OK_FROM_JINDO_WITH_NOT_EXIST(reader_->getFileLength(len)); PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(len, "jindo input length")); return len; } diff --git a/src/paimon/fs/jindo/jindo_utils.h b/src/paimon/fs/jindo/jindo_utils.h index 3f9f72ba5..692340a9f 100644 --- a/src/paimon/fs/jindo/jindo_utils.h +++ b/src/paimon/fs/jindo/jindo_utils.h @@ -17,6 +17,7 @@ #pragma once #include "JdoStatus.hpp" // NOLINT(build/include_subdir) +#include "jdo_error.h" // NOLINT(build/include_subdir) #include "paimon/status.h" namespace paimon::jindo { @@ -28,4 +29,18 @@ namespace paimon::jindo { } \ } while (false) +/// Like PAIMON_RETURN_NOT_OK_FROM_JINDO, but maps a jindo file-not-found error to +/// Status::NotExist so that callers can distinguish a missing file from other IO errors, +/// consistent with LocalFileSystem::Open. +#define PAIMON_RETURN_NOT_OK_FROM_JINDO_WITH_NOT_EXIST(JINDO_STATUS) \ + do { \ + auto __s = (JINDO_STATUS); \ + if (PAIMON_UNLIKELY(!(__s).ok())) { \ + if ((__s).getErrCode() == JDO_FILE_NOT_FOUND_ERROR) { \ + return Status::NotExist(__s.errMsg()); \ + } \ + return Status::IOError(__s.errMsg()); \ + } \ + } while (false) + } // namespace paimon::jindo diff --git a/src/paimon/fs/jindo/jindo_utils_test.cpp b/src/paimon/fs/jindo/jindo_utils_test.cpp new file mode 100644 index 000000000..d742d0889 --- /dev/null +++ b/src/paimon/fs/jindo/jindo_utils_test.cpp @@ -0,0 +1,57 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "paimon/fs/jindo/jindo_utils.h" + +#include "gtest/gtest.h" +#include "jdo_error.h" // NOLINT(build/include_subdir) +#include "paimon/status.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::jindo::test { + +namespace { + +Status ConvertWithNotExist(const JdoStatus& jdo_status) { + PAIMON_RETURN_NOT_OK_FROM_JINDO_WITH_NOT_EXIST(jdo_status); + return Status::OK(); +} + +Status Convert(const JdoStatus& jdo_status) { + PAIMON_RETURN_NOT_OK_FROM_JINDO(jdo_status); + return Status::OK(); +} + +} // namespace + +TEST(JindoUtilsTest, TestNotExistMacroMapsStatusByErrorCode) { + ASSERT_OK(ConvertWithNotExist(JdoStatus())); + + Status not_found = ConvertWithNotExist(JdoStatus(JDO_FILE_NOT_FOUND_ERROR, "file not found")); + ASSERT_TRUE(not_found.IsNotExist()); + ASSERT_NOK_WITH_MSG(not_found, "file not found"); + + Status other = ConvertWithNotExist(JdoStatus(JDO_FILE_NOT_FOUND_ERROR + 1, "some other error")); + ASSERT_TRUE(other.IsIOError()); + ASSERT_NOK_WITH_MSG(other, "some other error"); +} + +TEST(JindoUtilsTest, TestPlainMacroKeepsIOErrorForNotFound) { + Status status = Convert(JdoStatus(JDO_FILE_NOT_FOUND_ERROR, "file not found")); + ASSERT_TRUE(status.IsIOError()); +} + +} // namespace paimon::jindo::test diff --git a/src/paimon/testing/utils/test_helper.h b/src/paimon/testing/utils/test_helper.h index d3348a5c7..7b710d6d9 100644 --- a/src/paimon/testing/utils/test_helper.h +++ b/src/paimon/testing/utils/test_helper.h @@ -22,6 +22,7 @@ #include #include +#include "arrow/api.h" #include "arrow/c/bridge.h" #include "arrow/ipc/api.h" #include "paimon/api.h" @@ -175,6 +176,27 @@ class TestHelper { return result_plan->Splits(); } + /// Builds a one-row struct array holding the serialized descriptor of the blob. + static Result> MakeBlobDescriptorArray( + const std::shared_ptr& struct_type, const std::shared_ptr& blob, + const std::shared_ptr& pool) { + if (struct_type->num_fields() != 1 || + struct_type->field(0)->type()->id() != arrow::Type::LARGE_BINARY) { + return Status::Invalid("struct_type must have a single large binary field"); + } + arrow::StructBuilder struct_builder(struct_type, arrow::default_memory_pool(), + {std::make_shared()}); + auto blob_builder = + static_cast(struct_builder.field_builder(0)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(struct_builder.Append()); + PAIMON_UNIQUE_PTR descriptor = blob->ToDescriptor(pool); + PAIMON_RETURN_NOT_OK_FROM_ARROW( + blob_builder->Append(descriptor->data(), descriptor->size())); + std::shared_ptr array; + PAIMON_RETURN_NOT_OK_FROM_ARROW(struct_builder.Finish(&array)); + return array; + } + static Result CheckBlobsEqual(const std::vector>& result_blobs, const std::vector>& expected_blobs, const std::shared_ptr& fs) { From 0bc42709c251dd7c6c51b8fc5ed7376ae78d0deb Mon Sep 17 00:00:00 2001 From: Nicholas Jiang Date: Thu, 16 Jul 2026 12:47:52 +0800 Subject: [PATCH 2/4] 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 --- src/paimon/CMakeLists.txt | 16 +- .../blob/blob_file_batch_reader_test.cpp | 6 +- src/paimon/format/blob/blob_format_writer.cpp | 12 +- src/paimon/format/blob/blob_format_writer.h | 13 +- .../format/blob/blob_format_writer_test.cpp | 136 ++++++++--------- src/paimon/format/blob/blob_writer_builder.h | 4 +- src/paimon/fs/jindo/jindo_file_system.cpp | 8 +- src/paimon/fs/jindo/jindo_utils.h | 31 ++-- src/paimon/fs/jindo/jindo_utils_test.cpp | 18 +-- test/inte/blob_table_inte_test.cpp | 140 ++++++++++++++++++ 10 files changed, 254 insertions(+), 130 deletions(-) diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index 89735b102..46cc13854 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -791,19 +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 - # 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() diff --git a/src/paimon/format/blob/blob_file_batch_reader_test.cpp b/src/paimon/format/blob/blob_file_batch_reader_test.cpp index 2a06b0b35..4e53dfb4c 100644 --- a/src/paimon/format/blob/blob_file_batch_reader_test.cpp +++ b/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 blob_field = BlobUtils::ToArrowField("blob_col"); auto struct_type = arrow::struct_({blob_field}); - ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, - BlobFormatWriter::Create(output_stream, struct_type, file_system, pool_)); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr 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()); diff --git a/src/paimon/format/blob/blob_format_writer.cpp b/src/paimon/format/blob/blob_format_writer.cpp index 07c7bb883..03cd36220 100644 --- a/src/paimon/format/blob/blob_format_writer.cpp +++ b/src/paimon/format/blob/blob_format_writer.cpp @@ -37,10 +37,10 @@ namespace paimon::blob { BlobFormatWriter::BlobFormatWriter(const std::shared_ptr& out, const std::string& uri, const std::shared_ptr& data_type, - const std::shared_ptr& fs, - const std::shared_ptr& pool, bool write_null_on_missing_file, - bool write_null_on_fetch_failure) + bool write_null_on_fetch_failure, + const std::shared_ptr& fs, + const std::shared_ptr& pool) : out_(out), uri_(uri), data_type_(data_type), @@ -56,8 +56,8 @@ BlobFormatWriter::BlobFormatWriter(const std::shared_ptr& out, con Result> BlobFormatWriter::Create( const std::shared_ptr& out, const std::shared_ptr& data_type, - const std::shared_ptr& fs, const std::shared_ptr& 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& fs, const std::shared_ptr& pool) { if (out == nullptr) { return Status::Invalid("blob format writer create failed. out is nullptr"); } @@ -77,7 +77,7 @@ Result> BlobFormatWriter::Create( } PAIMON_ASSIGN_OR_RAISE(std::string uri, out->GetUri()); return std::unique_ptr(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) { diff --git a/src/paimon/format/blob/blob_format_writer.h b/src/paimon/format/blob/blob_format_writer.h index efa12016d..ad90539b7 100644 --- a/src/paimon/format/blob/blob_format_writer.h +++ b/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> Create( const std::shared_ptr& out, const std::shared_ptr& data_type, - const std::shared_ptr& fs, const std::shared_ptr& 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& fs, const std::shared_ptr& pool); Status AddBatch(ArrowArray* batch) override; @@ -73,8 +77,9 @@ class BlobFormatWriter : public FormatWriter { private: BlobFormatWriter(const std::shared_ptr& out, const std::string& uri, const std::shared_ptr& data_type, - const std::shared_ptr& fs, const std::shared_ptr& 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& fs, + const std::shared_ptr& pool); Status WriteBlob(std::string_view blob_data); diff --git a/src/paimon/format/blob/blob_format_writer_test.cpp b/src/paimon/format/blob/blob_format_writer_test.cpp index b5d66bbc6..42d69c4f6 100644 --- a/src/paimon/format/blob/blob_format_writer_test.cpp +++ b/src/paimon/format/blob/blob_format_writer_test.cpp @@ -64,6 +64,13 @@ class BlobFormatWriterTestBase : public ::testing::Test { ASSERT_OK(output_stream_->Close()); } + /// Create a writer on output_stream_ with both write-null options disabled. + Result> CreateDefaultWriter() const { + return BlobFormatWriter::Create(output_stream_, struct_type_, + /*write_null_on_missing_file=*/false, + /*write_null_on_fetch_failure=*/false, file_system_, pool_); + } + Status AddBatchOnce(const std::shared_ptr& format_writer, const std::shared_ptr& blob_array) const { auto c_array = std::make_unique(); @@ -140,9 +147,7 @@ INSTANTIATE_TEST_SUITE_P(BlobAsDescriptor, BlobFormatWriterTest, ::testing::Valu TEST_P(BlobFormatWriterTest, TestSimple) { // write - ASSERT_OK_AND_ASSIGN( - std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, CreateDefaultWriter()); std::vector> expected_blobs; std::string file1 = paimon::test::GetDataDir() + "/avro/data/avro_with_null"; @@ -202,37 +207,42 @@ TEST_P(BlobFormatWriterTest, TestSimple) { TEST_P(BlobFormatWriterTest, TestCreateWithInvalidParameters) { // Test with nullptr output stream - ASSERT_NOK_WITH_MSG(BlobFormatWriter::Create(nullptr, struct_type_, file_system_, pool_), - "blob format writer create failed. out is nullptr"); + ASSERT_NOK_WITH_MSG( + BlobFormatWriter::Create(nullptr, struct_type_, /*write_null_on_missing_file=*/false, + /*write_null_on_fetch_failure=*/false, file_system_, pool_), + "blob format writer create failed. out is nullptr"); // Test with nullptr data type - ASSERT_NOK_WITH_MSG(BlobFormatWriter::Create(output_stream_, nullptr, file_system_, pool_), - "blob format writer create failed. data_type is nullptr"); + ASSERT_NOK_WITH_MSG( + BlobFormatWriter::Create(output_stream_, nullptr, /*write_null_on_missing_file=*/false, + /*write_null_on_fetch_failure=*/false, file_system_, pool_), + "blob format writer create failed. data_type is nullptr"); // Test with nullptr memory pool ASSERT_NOK_WITH_MSG( - BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, nullptr), + BlobFormatWriter::Create(output_stream_, struct_type_, /*write_null_on_missing_file=*/false, + /*write_null_on_fetch_failure=*/false, file_system_, nullptr), "blob format writer create failed. pool is nullptr"); // Test with invalid field count (more than 1 field) auto multi_field_type = arrow::struct_( {arrow::field("blob_col1", arrow::binary()), arrow::field("blob_col2", arrow::binary())}); - ASSERT_NOK_WITH_MSG( - BlobFormatWriter::Create(output_stream_, multi_field_type, file_system_, pool_), - "blob data type field number 2 is not 1"); + ASSERT_NOK_WITH_MSG(BlobFormatWriter::Create( + output_stream_, multi_field_type, /*write_null_on_missing_file=*/false, + /*write_null_on_fetch_failure=*/false, file_system_, pool_), + "blob data type field number 2 is not 1"); // Test with non-blob field (missing blob metadata) auto non_blob_field = arrow::field("regular_col", arrow::binary()); auto non_blob_type = arrow::struct_({non_blob_field}); - ASSERT_NOK_WITH_MSG( - BlobFormatWriter::Create(output_stream_, non_blob_type, file_system_, pool_), - "field regular_col: binary is not BLOB"); + ASSERT_NOK_WITH_MSG(BlobFormatWriter::Create( + output_stream_, non_blob_type, /*write_null_on_missing_file=*/false, + /*write_null_on_fetch_failure=*/false, file_system_, pool_), + "field regular_col: binary is not BLOB"); } TEST_P(BlobFormatWriterTest, TestInvalidCase) { - ASSERT_OK_AND_ASSIGN( - std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, CreateDefaultWriter()); // Test nullptr batch ASSERT_NOK_WITH_MSG(writer->AddBatch(nullptr), @@ -249,9 +259,7 @@ TEST_P(BlobFormatWriterTest, TestInvalidCase) { } TEST_P(BlobFormatWriterTest, TestAddBatchWithInvalidBatchLength) { - ASSERT_OK_AND_ASSIGN( - std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, CreateDefaultWriter()); // Test batch with wrong length (not 1) arrow::StructBuilder struct_builder(struct_type_, arrow::default_memory_pool(), @@ -277,9 +285,7 @@ TEST_P(BlobFormatWriterTest, TestAddBatchWithInvalidBatchLength) { } TEST_P(BlobFormatWriterTest, TestReachTargetSize) { - ASSERT_OK_AND_ASSIGN( - std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, CreateDefaultWriter()); // Initially should not reach target size ASSERT_OK_AND_ASSIGN(bool reached, writer->ReachTargetSize(true, 1000)); @@ -302,9 +308,7 @@ TEST_P(BlobFormatWriterTest, TestReachTargetSize) { } TEST_P(BlobFormatWriterTest, TestGetWriterMetrics) { - ASSERT_OK_AND_ASSIGN( - std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, CreateDefaultWriter()); auto metrics = writer->GetWriterMetrics(); ASSERT_TRUE(metrics); @@ -312,9 +316,7 @@ TEST_P(BlobFormatWriterTest, TestGetWriterMetrics) { TEST_P(BlobFormatWriterTest, TestEmptyWriter) { // Test creating a writer and finishing without adding any data - ASSERT_OK_AND_ASSIGN( - std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, CreateDefaultWriter()); ASSERT_OK(writer->Flush()); ASSERT_OK(writer->Finish()); @@ -333,9 +335,7 @@ TEST_P(BlobFormatWriterTest, TestEmptyWriter) { } TEST_P(BlobFormatWriterTest, TestLargeBlob) { - ASSERT_OK_AND_ASSIGN( - std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, CreateDefaultWriter()); // Create a temporary large file for testing std::string large_file_path = dir_->Str() + "/large_test_file.bin"; @@ -388,9 +388,7 @@ TEST_P(BlobFormatWriterTest, TestLargeBlob) { } TEST_P(BlobFormatWriterTest, TestAddBatchWithNullValues) { - ASSERT_OK_AND_ASSIGN( - std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, CreateDefaultWriter()); // Write one row with child-level null blob arrow::StructBuilder struct_builder(struct_type_, arrow::default_memory_pool(), @@ -436,19 +434,17 @@ TEST_P(BlobFormatWriterTest, TestAddBatchWithNullValues) { ASSERT_TRUE(struct_builder2.Finish(&null_struct_array).ok()); auto null_c_array = std::make_unique(); ASSERT_TRUE(arrow::ExportArray(*null_struct_array, null_c_array.get()).ok()); - ASSERT_OK_AND_ASSIGN( - std::shared_ptr writer2, - BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer2, CreateDefaultWriter()); ASSERT_NOK_WITH_MSG(writer2->AddBatch(null_c_array.get()), "BlobFormatWriter does not support struct-level null."); ArrowArrayRelease(null_c_array.get()); } TEST_F(BlobFormatWriterWriteNullTest, TestWriteNullOnMissingFile) { - ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_, - /*write_null_on_missing_file=*/true, - /*write_null_on_fetch_failure=*/false)); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr writer, + BlobFormatWriter::Create(output_stream_, struct_type_, /*write_null_on_missing_file=*/true, + /*write_null_on_fetch_failure=*/false, file_system_, pool_)); ASSERT_OK_AND_ASSIGN(std::shared_ptr missing_blob, Blob::FromPath(dir_->Str() + "/not_exist_file", /*offset=*/0, @@ -483,10 +479,10 @@ TEST_F(BlobFormatWriterWriteNullTest, TestWriteNullOnMissingFile) { } TEST_F(BlobFormatWriterWriteNullTest, TestWriteNullOnFetchFailure) { - ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_, - /*write_null_on_missing_file=*/false, - /*write_null_on_fetch_failure=*/true)); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr writer, + BlobFormatWriter::Create(output_stream_, struct_type_, /*write_null_on_missing_file=*/false, + /*write_null_on_fetch_failure=*/true, file_system_, pool_)); std::string file = paimon::test::GetDataDir() + "/xxhash.data"; ASSERT_OK_AND_ASSIGN(std::shared_ptr bad_offset_blob, @@ -511,10 +507,10 @@ TEST_F(BlobFormatWriterWriteNullTest, TestWriteNullOnFetchFailure) { } TEST_F(BlobFormatWriterWriteNullTest, TestWriteNullOnBothOptionsEnabled) { - ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_, - /*write_null_on_missing_file=*/true, - /*write_null_on_fetch_failure=*/true)); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr writer, + BlobFormatWriter::Create(output_stream_, struct_type_, /*write_null_on_missing_file=*/true, + /*write_null_on_fetch_failure=*/true, file_system_, pool_)); // Row 0: missing file -> NULL. ASSERT_OK_AND_ASSIGN(std::shared_ptr missing_blob, @@ -560,43 +556,37 @@ TEST_F(BlobFormatWriterWriteNullTest, TestWriteNullClassifiesByStatusCode) { auto io_error_fs = std::make_shared(Status::IOError("mock io error")); { - ASSERT_OK_AND_ASSIGN( - std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, not_exist_fs, pool_, - /*write_null_on_missing_file=*/true, - /*write_null_on_fetch_failure=*/false)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, + BlobFormatWriter::Create( + output_stream_, struct_type_, /*write_null_on_missing_file=*/true, + /*write_null_on_fetch_failure=*/false, not_exist_fs, pool_)); ASSERT_OK(AddBatchOnce(writer, array)); } { - ASSERT_OK_AND_ASSIGN( - std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, io_error_fs, pool_, - /*write_null_on_missing_file=*/true, - /*write_null_on_fetch_failure=*/false)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, + BlobFormatWriter::Create( + output_stream_, struct_type_, /*write_null_on_missing_file=*/true, + /*write_null_on_fetch_failure=*/false, io_error_fs, pool_)); ASSERT_NOK_WITH_MSG(AddBatchOnce(writer, array), "mock io error"); } { - ASSERT_OK_AND_ASSIGN( - std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, io_error_fs, pool_, - /*write_null_on_missing_file=*/false, - /*write_null_on_fetch_failure=*/true)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, + BlobFormatWriter::Create( + output_stream_, struct_type_, /*write_null_on_missing_file=*/false, + /*write_null_on_fetch_failure=*/true, io_error_fs, pool_)); ASSERT_OK(AddBatchOnce(writer, array)); } { - ASSERT_OK_AND_ASSIGN( - std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, not_exist_fs, pool_, - /*write_null_on_missing_file=*/false, - /*write_null_on_fetch_failure=*/true)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, + BlobFormatWriter::Create( + output_stream_, struct_type_, /*write_null_on_missing_file=*/false, + /*write_null_on_fetch_failure=*/true, not_exist_fs, pool_)); ASSERT_NOK_WITH_MSG(AddBatchOnce(writer, array), "mock not exist"); } } TEST_P(BlobFormatWriterTest, TestAddBatchWithZeroLengthBlob) { - ASSERT_OK_AND_ASSIGN( - std::shared_ptr writer, - BlobFormatWriter::Create(output_stream_, struct_type_, file_system_, pool_)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, CreateDefaultWriter()); // Create a zero-length file std::string zero_file_path = dir_->Str() + "/zero_length_file.bin"; diff --git a/src/paimon/format/blob/blob_writer_builder.h b/src/paimon/format/blob/blob_writer_builder.h index 314885850..12caa0e16 100644 --- a/src/paimon/format/blob/blob_writer_builder.h +++ b/src/paimon/format/blob/blob_writer_builder.h @@ -73,8 +73,8 @@ class BlobWriterBuilder : public SpecificFSWriterBuilder { PAIMON_ASSIGN_OR_RAISE(bool write_null_on_fetch_failure, OptionsUtils::GetValueFromMap( options_, Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, false)); - return BlobFormatWriter::Create(out, data_type_, fs_, pool_, write_null_on_missing_file, - write_null_on_fetch_failure); + return BlobFormatWriter::Create(out, data_type_, write_null_on_missing_file, + write_null_on_fetch_failure, fs_, pool_); } private: diff --git a/src/paimon/fs/jindo/jindo_file_system.cpp b/src/paimon/fs/jindo/jindo_file_system.cpp index 64bb309c7..6246780dd 100644 --- a/src/paimon/fs/jindo/jindo_file_system.cpp +++ b/src/paimon/fs/jindo/jindo_file_system.cpp @@ -57,8 +57,7 @@ JindoFileSystem::JindoFileSystem(std::unique_ptr&& fs) Result> JindoFileSystem::Open(const std::string& path) const { std::unique_ptr reader; - PAIMON_RETURN_NOT_OK_FROM_JINDO_WITH_NOT_EXIST( - impl_->GetFileSystem()->openReader(path, &reader)); + PAIMON_RETURN_NOT_OK_FROM_JINDO(impl_->GetFileSystem()->openReader(path, &reader)); return std::make_unique(impl_, std::move(reader)); } @@ -209,10 +208,7 @@ Result JindoInputStream::GetPos() const { Result JindoInputStream::Length() const { int64_t len = -1; - // openReader may succeed lazily for a missing object; keep the not-found mapping here so - // that Blob::NewInputStream(), which queries the length right after Open(), still observes - // Status::NotExist. Read failures keep the plain IOError mapping. - PAIMON_RETURN_NOT_OK_FROM_JINDO_WITH_NOT_EXIST(reader_->getFileLength(len)); + PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->getFileLength(len)); PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(len, "jindo input length")); return len; } diff --git a/src/paimon/fs/jindo/jindo_utils.h b/src/paimon/fs/jindo/jindo_utils.h index 692340a9f..d3c596886 100644 --- a/src/paimon/fs/jindo/jindo_utils.h +++ b/src/paimon/fs/jindo/jindo_utils.h @@ -21,26 +21,17 @@ #include "paimon/status.h" namespace paimon::jindo { -#define PAIMON_RETURN_NOT_OK_FROM_JINDO(JINDO_STATUS) \ - do { \ - auto __s = (JINDO_STATUS); \ - if (PAIMON_UNLIKELY(!(__s).ok())) { \ - return Status::IOError(__s.errMsg()); \ - } \ - } while (false) - -/// Like PAIMON_RETURN_NOT_OK_FROM_JINDO, but maps a jindo file-not-found error to -/// Status::NotExist so that callers can distinguish a missing file from other IO errors, -/// consistent with LocalFileSystem::Open. -#define PAIMON_RETURN_NOT_OK_FROM_JINDO_WITH_NOT_EXIST(JINDO_STATUS) \ - do { \ - auto __s = (JINDO_STATUS); \ - if (PAIMON_UNLIKELY(!(__s).ok())) { \ - if ((__s).getErrCode() == JDO_FILE_NOT_FOUND_ERROR) { \ - return Status::NotExist(__s.errMsg()); \ - } \ - return Status::IOError(__s.errMsg()); \ - } \ +/// Maps a jindo file-not-found error to Status::NotExist so that callers can distinguish +/// a missing file from other IO errors, consistent with LocalFileSystem. +#define PAIMON_RETURN_NOT_OK_FROM_JINDO(JINDO_STATUS) \ + do { \ + auto __s = (JINDO_STATUS); \ + if (PAIMON_UNLIKELY(!(__s).ok())) { \ + if ((__s).getErrCode() == JDO_FILE_NOT_FOUND_ERROR) { \ + return Status::NotExist(__s.errMsg()); \ + } \ + return Status::IOError(__s.errMsg()); \ + } \ } while (false) } // namespace paimon::jindo diff --git a/src/paimon/fs/jindo/jindo_utils_test.cpp b/src/paimon/fs/jindo/jindo_utils_test.cpp index d742d0889..041876e67 100644 --- a/src/paimon/fs/jindo/jindo_utils_test.cpp +++ b/src/paimon/fs/jindo/jindo_utils_test.cpp @@ -25,11 +25,6 @@ namespace paimon::jindo::test { namespace { -Status ConvertWithNotExist(const JdoStatus& jdo_status) { - PAIMON_RETURN_NOT_OK_FROM_JINDO_WITH_NOT_EXIST(jdo_status); - return Status::OK(); -} - Status Convert(const JdoStatus& jdo_status) { PAIMON_RETURN_NOT_OK_FROM_JINDO(jdo_status); return Status::OK(); @@ -37,21 +32,16 @@ Status Convert(const JdoStatus& jdo_status) { } // namespace -TEST(JindoUtilsTest, TestNotExistMacroMapsStatusByErrorCode) { - ASSERT_OK(ConvertWithNotExist(JdoStatus())); +TEST(JindoUtilsTest, TestMacroMapsStatusByErrorCode) { + ASSERT_OK(Convert(JdoStatus())); - Status not_found = ConvertWithNotExist(JdoStatus(JDO_FILE_NOT_FOUND_ERROR, "file not found")); + Status not_found = Convert(JdoStatus(JDO_FILE_NOT_FOUND_ERROR, "file not found")); ASSERT_TRUE(not_found.IsNotExist()); ASSERT_NOK_WITH_MSG(not_found, "file not found"); - Status other = ConvertWithNotExist(JdoStatus(JDO_FILE_NOT_FOUND_ERROR + 1, "some other error")); + Status other = Convert(JdoStatus(JDO_FILE_NOT_FOUND_ERROR + 1, "some other error")); ASSERT_TRUE(other.IsIOError()); ASSERT_NOK_WITH_MSG(other, "some other error"); } -TEST(JindoUtilsTest, TestPlainMacroKeepsIOErrorForNotFound) { - Status status = Convert(JdoStatus(JDO_FILE_NOT_FOUND_ERROR, "file not found")); - ASSERT_TRUE(status.IsIOError()); -} - } // namespace paimon::jindo::test diff --git a/test/inte/blob_table_inte_test.cpp b/test/inte/blob_table_inte_test.cpp index 75fee6ef3..7afd81b7b 100644 --- a/test/inte/blob_table_inte_test.cpp +++ b/test/inte/blob_table_inte_test.cpp @@ -24,6 +24,7 @@ #include #include #include +#include #include #include @@ -370,6 +371,20 @@ class BlobTableInteTest : public testing::Test, public ::testing::WithParamInter }); } + /// Delete the file referenced by the serialized BlobDescriptor at `row` of blob field + /// `field_name`, so that a subsequent table write observes a missing file. + Status DeleteDescriptorTarget(const std::shared_ptr& desc_array, + const std::string& field_name, int64_t row) const { + const auto& blob_col = arrow::internal::checked_cast( + *desc_array->GetFieldByName(field_name)); + std::string_view descriptor_bytes = blob_col.GetView(row); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr descriptor, + BlobDescriptor::Deserialize(descriptor_bytes.data(), descriptor_bytes.size())); + LocalFileSystem fs; + return fs.Delete(descriptor->Uri(), /*recursive=*/false); + } + struct BlobDescriptorPathRewrite { std::string table_path; std::vector table_relative_blob_dirs; @@ -561,6 +576,131 @@ TEST_P(BlobTableInteTest, TestAppendTableWriteWithBlobAsDescriptorFalse) { ASSERT_OK(ScanAndRead(table_path, schema->field_names(), write_array)); } +TEST_P(BlobTableInteTest, TestWriteNullOnMissingFile) { + // blob-write-null-on-missing-file=true: a descriptor whose file is gone at write time + // becomes a NULL blob element, while the row itself is still written and merged with + // the data file on read. + arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()), + arrow::field("f1", arrow::int32()), + BlobUtils::ToArrowField("blob", true)}; + + std::map options = { + {Options::MANIFEST_FORMAT, "orc"}, {Options::FILE_FORMAT, GetParam()}, + {Options::TARGET_FILE_SIZE, "700"}, {Options::BUCKET, "-1"}, + {Options::ROW_TRACKING_ENABLED, "true"}, {Options::DATA_EVOLUTION_ENABLED, "true"}, + {Options::BLOB_AS_DESCRIPTOR, "true"}, {Options::BLOB_WRITE_NULL_ON_MISSING_FILE, "true"}, + {Options::FILE_SYSTEM, "local"}}; + CreateTable(fields, /*partition_keys=*/{}, options); + std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); + + std::string raw_json = R"([ + ["str_0", 0, "blob_data_0"], + ["str_1", 1, "blob_data_1"], + ["str_2", 2, "blob_data_2"] + ])"; + auto raw_array = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie()); + ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"})); + + // Remove the file behind row 1's descriptor so the write observes a missing file. + ASSERT_OK(DeleteDescriptorTarget(desc_array, "blob", /*row=*/1)); + + auto schema = arrow::schema(fields); + ASSERT_OK_AND_ASSIGN(auto commit_msgs, + WriteArray(table_path, {}, schema->field_names(), {desc_array})); + ASSERT_OK(Commit(table_path, commit_msgs)); + + // Both the main data file and the .blob file keep the full row count; the read path + // merges them back into rows where row 1's blob is NULL. + ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path)); + VerifyDataFileMetas(plan, /*expected_file_count=*/2, /*expected_row_counts=*/{3, 3}, + /*expected_min_seqs=*/{1, 1}, /*expected_max_seqs=*/{1, 1}, + /*expected_first_row_ids=*/{0, 0}, + /*expected_write_cols=*/ + {std::vector{"f0", "f1"}, std::vector{"blob"}}); + + ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), plan)); + ASSERT_TRUE(result.chunked_array); + auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie(); + auto read_struct = std::dynamic_pointer_cast(read_concat); + ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"blob"})); + + std::string expected_json = R"([ + ["str_0", 0, "blob_data_0"], + ["str_1", 1, null], + ["str_2", 2, "blob_data_2"] + ])"; + auto expected_array = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json) + .ValueOrDie()); + ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_array)); + ASSERT_TRUE(resolved->Equals(expected_with_rk)) + << "result:" << resolved->ToString() << std::endl + << "expected:" << expected_with_rk->ToString(); +} + +TEST_P(BlobTableInteTest, TestMissingFileFailsWriteWhenWriteNullDisabled) { + // Without blob-write-null-on-missing-file, a missing descriptor file fails the write. + arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()), + arrow::field("f1", arrow::int32()), + BlobUtils::ToArrowField("blob", true)}; + + std::map options = { + {Options::MANIFEST_FORMAT, "orc"}, {Options::FILE_FORMAT, GetParam()}, + {Options::TARGET_FILE_SIZE, "700"}, {Options::BUCKET, "-1"}, + {Options::ROW_TRACKING_ENABLED, "true"}, {Options::DATA_EVOLUTION_ENABLED, "true"}, + {Options::BLOB_AS_DESCRIPTOR, "true"}, {Options::FILE_SYSTEM, "local"}}; + CreateTable(fields, /*partition_keys=*/{}, options); + std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); + + std::string raw_json = R"([ + ["str_0", 0, "blob_data_0"], + ["str_1", 1, "blob_data_1"] + ])"; + auto raw_array = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie()); + ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"})); + ASSERT_OK(DeleteDescriptorTarget(desc_array, "blob", /*row=*/1)); + + auto schema = arrow::schema(fields); + ASSERT_NOK_WITH_MSG(WriteArray(table_path, {}, schema->field_names(), {desc_array}), + "not exists"); +} + +TEST_P(BlobTableInteTest, TestWriteNullOnFetchFailureKeepsMissingFileFailing) { + // blob-write-null-on-fetch-failure only converts non-NotExist open failures; a missing + // descriptor file still fails the write when only this option is enabled. + arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()), + arrow::field("f1", arrow::int32()), + BlobUtils::ToArrowField("blob", true)}; + + std::map options = { + {Options::MANIFEST_FORMAT, "orc"}, + {Options::FILE_FORMAT, GetParam()}, + {Options::TARGET_FILE_SIZE, "700"}, + {Options::BUCKET, "-1"}, + {Options::ROW_TRACKING_ENABLED, "true"}, + {Options::DATA_EVOLUTION_ENABLED, "true"}, + {Options::BLOB_AS_DESCRIPTOR, "true"}, + {Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, "true"}, + {Options::FILE_SYSTEM, "local"}}; + CreateTable(fields, /*partition_keys=*/{}, options); + std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); + + std::string raw_json = R"([ + ["str_0", 0, "blob_data_0"], + ["str_1", 1, "blob_data_1"] + ])"; + auto raw_array = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie()); + ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"})); + ASSERT_OK(DeleteDescriptorTarget(desc_array, "blob", /*row=*/1)); + + auto schema = arrow::schema(fields); + ASSERT_NOK_WITH_MSG(WriteArray(table_path, {}, schema->field_names(), {desc_array}), + "not exists"); +} + TEST_P(BlobTableInteTest, TestBasic) { CreateTable(); std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); From 991db0bcda3737ac7d4ec0919c650fb489fe1902 Mon Sep 17 00:00:00 2001 From: Nicholas Jiang Date: Thu, 16 Jul 2026 13:48:50 +0800 Subject: [PATCH 3/4] test(blob): add end-to-end test for write-null-on-fetch-failure with bad offset Add TestWriteNullOnFetchFailure to blob_table_inte_test.cpp: with blob-write-null-on-fetch-failure=true, a descriptor whose offset points beyond the end of the target file (a fetch failure that is not a missing file) is written as a NULL blob element, the row itself is kept in both the data file and the .blob file, and the read path merges them back with the blob column NULL for that row. Introduce a CorruptDescriptorOffset test helper that rewrites the serialized BlobDescriptor of a single row to reference an out-of-range offset, reusing the TransformBlobFields framework. Co-Authored-By: Claude Fable 5 --- test/inte/blob_table_inte_test.cpp | 99 ++++++++++++++++++++++++++++++ 1 file changed, 99 insertions(+) diff --git a/test/inte/blob_table_inte_test.cpp b/test/inte/blob_table_inte_test.cpp index 7afd81b7b..d0e834b8c 100644 --- a/test/inte/blob_table_inte_test.cpp +++ b/test/inte/blob_table_inte_test.cpp @@ -385,6 +385,36 @@ class BlobTableInteTest : public testing::Test, public ::testing::WithParamInter return fs.Delete(descriptor->Uri(), /*recursive=*/false); } + /// Rewrite the serialized BlobDescriptor at `row` of blob field `field_name` to reference + /// an offset beyond the end of the target file, so that a subsequent table write observes + /// a fetch failure that is not a missing file. `row` counts non-null rows. + Result> CorruptDescriptorOffset( + const std::shared_ptr& desc_array, const std::string& field_name, + int64_t row) const { + int64_t current_row = 0; + return TransformBlobFields( + desc_array, {field_name}, + [&](const std::string_view& descriptor_bytes, + arrow::LargeBinaryBuilder* builder) -> Status { + if (current_row++ != row) { + PAIMON_RETURN_NOT_OK_FROM_ARROW( + builder->Append(descriptor_bytes.data(), descriptor_bytes.size())); + return Status::OK(); + } + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr descriptor, + BlobDescriptor::Deserialize(descriptor_bytes.data(), descriptor_bytes.size())); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr corrupted, + BlobDescriptor::Create(descriptor->Version(), descriptor->Uri(), + /*offset=*/1 << 20, descriptor->Length())); + auto corrupted_bytes = corrupted->Serialize(pool_); + PAIMON_RETURN_NOT_OK_FROM_ARROW( + builder->Append(corrupted_bytes->data(), corrupted_bytes->size())); + return Status::OK(); + }); + } + struct BlobDescriptorPathRewrite { std::string table_path; std::vector table_relative_blob_dirs; @@ -667,6 +697,75 @@ TEST_P(BlobTableInteTest, TestMissingFileFailsWriteWhenWriteNullDisabled) { "not exists"); } +TEST_P(BlobTableInteTest, TestWriteNullOnFetchFailure) { + // blob-write-null-on-fetch-failure=true: a descriptor whose data cannot be fetched for a + // reason other than a missing file (here: offset beyond the end of the file) becomes a + // NULL blob element, while the row itself is still written and merged with the data file + // on read. + arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()), + arrow::field("f1", arrow::int32()), + BlobUtils::ToArrowField("blob", true)}; + + std::map options = { + {Options::MANIFEST_FORMAT, "orc"}, + {Options::FILE_FORMAT, GetParam()}, + {Options::TARGET_FILE_SIZE, "700"}, + {Options::BUCKET, "-1"}, + {Options::ROW_TRACKING_ENABLED, "true"}, + {Options::DATA_EVOLUTION_ENABLED, "true"}, + {Options::BLOB_AS_DESCRIPTOR, "true"}, + {Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, "true"}, + {Options::FILE_SYSTEM, "local"}}; + CreateTable(fields, /*partition_keys=*/{}, options); + std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); + + std::string raw_json = R"([ + ["str_0", 0, "blob_data_0"], + ["str_1", 1, "blob_data_1"], + ["str_2", 2, "blob_data_2"] + ])"; + auto raw_array = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie()); + ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"})); + + // Point row 1's descriptor at an offset beyond the end of its file so the write observes + // a fetch failure that is not a missing file. + ASSERT_OK_AND_ASSIGN(desc_array, CorruptDescriptorOffset(desc_array, "blob", /*row=*/1)); + + auto schema = arrow::schema(fields); + ASSERT_OK_AND_ASSIGN(auto commit_msgs, + WriteArray(table_path, {}, schema->field_names(), {desc_array})); + ASSERT_OK(Commit(table_path, commit_msgs)); + + // Both the main data file and the .blob file keep the full row count; the read path + // merges them back into rows where row 1's blob is NULL. + ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path)); + VerifyDataFileMetas(plan, /*expected_file_count=*/2, /*expected_row_counts=*/{3, 3}, + /*expected_min_seqs=*/{1, 1}, /*expected_max_seqs=*/{1, 1}, + /*expected_first_row_ids=*/{0, 0}, + /*expected_write_cols=*/ + {std::vector{"f0", "f1"}, std::vector{"blob"}}); + + ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), plan)); + ASSERT_TRUE(result.chunked_array); + auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie(); + auto read_struct = std::dynamic_pointer_cast(read_concat); + ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"blob"})); + + std::string expected_json = R"([ + ["str_0", 0, "blob_data_0"], + ["str_1", 1, null], + ["str_2", 2, "blob_data_2"] + ])"; + auto expected_array = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json) + .ValueOrDie()); + ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_array)); + ASSERT_TRUE(resolved->Equals(expected_with_rk)) + << "result:" << resolved->ToString() << std::endl + << "expected:" << expected_with_rk->ToString(); +} + TEST_P(BlobTableInteTest, TestWriteNullOnFetchFailureKeepsMissingFileFailing) { // blob-write-null-on-fetch-failure only converts non-NotExist open failures; a missing // descriptor file still fails the write when only this option is enabled. From 67f049d249be187bb698347c94d0d9e26f8d6e31 Mon Sep 17 00:00:00 2001 From: Nicholas Jiang Date: Thu, 16 Jul 2026 14:48:43 +0800 Subject: [PATCH 4/4] fix(cmake): link CMAKE_DL_LIBS after libarrow.a to fix gcc8 fs_test link libarrow.a calls dlsym but the arrow imported target did not declare ${CMAKE_DL_LIBS} in its link interface, so -ldl could end up before libarrow.a in the final link line. Linkers that resolve symbols strictly left-to-right (gcc8-test in CI) then fail linking paimon-fs-test with "undefined reference to dlsym / DSO missing from command line" once fs_test started linking the jindo static libraries. Declare ${CMAKE_DL_LIBS} on the arrow interface, matching the lucene and jindosdk::nextarch targets. Co-Authored-By: Claude Fable 5 --- cmake_modules/ThirdpartyToolchain.cmake | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/cmake_modules/ThirdpartyToolchain.cmake b/cmake_modules/ThirdpartyToolchain.cmake index ba46ca185..9d5bf29fc 100644 --- a/cmake_modules/ThirdpartyToolchain.cmake +++ b/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