/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 <algorithm>
#include <cstdint>
#include <cstdlib>
#include <initializer_list>
#include <map>
#include <memory>
#include <numeric>
#include <optional>
#include <set>
#include <string>
#include <string_view>
#include <utility>
#include <vector>

#include "arrow/api.h"
#include "arrow/array/array_base.h"
#include "arrow/array/builder_binary.h"
#include "arrow/array/builder_nested.h"
#include "arrow/array/builder_primitive.h"
#include "arrow/c/abi.h"
#include "arrow/c/bridge.h"
#include "arrow/ipc/json_simple.h"
#include "arrow/type.h"
#include "gtest/gtest.h"
#include "paimon/commit_context.h"
#include "paimon/common/data/binary_array_writer.h"
#include "paimon/common/data/binary_row.h"
#include "paimon/common/data/binary_row_writer.h"
#include "paimon/common/data/blob_defs.h"
#include "paimon/common/data/blob_descriptor.h"
#include "paimon/common/data/blob_utils.h"
#include "paimon/common/data/blob_view_struct.h"
#include "paimon/common/factories/io_hook.h"
#include "paimon/common/reader/reader_utils.h"
#include "paimon/common/table/special_fields.h"
#include "paimon/common/utils/checked_cast.h"
#include "paimon/common/utils/path_util.h"
#include "paimon/common/utils/scope_guard.h"
#include "paimon/core/global_index/indexed_split_impl.h"
#include "paimon/core/snapshot.h"
#include "paimon/core/stats/simple_stats.h"
#include "paimon/core/table/source/data_split_impl.h"
#include "paimon/core/utils/file_utils.h"
#include "paimon/core/utils/snapshot_manager.h"
#include "paimon/data/blob.h"
#include "paimon/defs.h"
#include "paimon/file_store_write.h"
#include "paimon/fs/file_system.h"
#include "paimon/fs/local/local_file_system.h"
#include "paimon/global_index/bitmap_global_index_result.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/predicate/literal.h"
#include "paimon/predicate/predicate_builder.h"
#include "paimon/read_context.h"
#include "paimon/reader/batch_reader.h"
#include "paimon/record_batch.h"
#include "paimon/result.h"
#include "paimon/scan_context.h"
#include "paimon/status.h"
#include "paimon/table/source/data_split.h"
#include "paimon/table/source/startup_mode.h"
#include "paimon/table/source/table_read.h"
#include "paimon/testing/utils/deletion_vector_test_helper.h"
#include "paimon/testing/utils/io_exception_helper.h"
#include "paimon/testing/utils/test_helper.h"
#include "paimon/testing/utils/testharness.h"
#include "paimon/write_context.h"

namespace paimon {
class DataSplit;
class RecordBatch;
}  // namespace paimon

namespace paimon::test {

struct ReadResult {
    std::unique_ptr<BatchReader> batch_reader;
    std::shared_ptr<arrow::ChunkedArray> chunked_array;
};

class BlobTableInteTest : public testing::Test, public ::testing::WithParamInterface<std::string> {
 public:
    void SetUp() override {
        pool_ = GetDefaultPool();
        dir_ = UniqueTestDirectory::Create("local");
        blob_dir_ = UniqueTestDirectory::Create("local");
    }

    void TearDown() override {
        dir_.reset();
    }

    void CreateTable(const std::vector<std::string>& partition_keys,
                     const std::map<std::string, std::string>& options) const {
        CreateTable(fields_, partition_keys, options);
    }

    void CreateTable(const arrow::FieldVector& fields,
                     const std::vector<std::string>& partition_keys,
                     const std::map<std::string, std::string>& options) const {
        auto schema = arrow::schema(fields);
        ::ArrowSchema c_schema;
        ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok());

        ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(dir_->Str(), {}));
        ASSERT_OK(catalog->CreateDatabase("foo", {}, /*ignore_if_exists=*/false));
        ASSERT_OK(catalog->CreateTable(Identifier("foo", "bar"), &c_schema, partition_keys,
                                       /*primary_keys=*/{}, options,
                                       /*ignore_if_exists=*/false));
    }

    void CreateTable(const std::vector<std::string>& partition_keys) const {
        std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                      {Options::FILE_FORMAT, GetParam()},
                                                      {Options::FILE_SYSTEM, "local"},
                                                      {Options::ROW_TRACKING_ENABLED, "true"},
                                                      {Options::DATA_EVOLUTION_ENABLED, "true"}};
        return CreateTable(partition_keys, options);
    }

    void CreateTable() const {
        return CreateTable(/*partition_keys=*/{});
    }

    Result<std::vector<std::shared_ptr<CommitMessage>>> WriteArray(
        const std::string& table_path, const std::map<std::string, std::string>& partition,
        const std::vector<std::string>& write_cols,
        const std::vector<std::shared_ptr<arrow::Array>>& write_arrays) const {
        // write
        WriteContextBuilder write_builder(table_path, "commit_user_1");
        write_builder.WithWriteSchema(write_cols);
        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<WriteContext> write_context, write_builder.Finish());
        PAIMON_ASSIGN_OR_RAISE(auto file_store_write,
                               FileStoreWrite::Create(std::move(write_context)));

        for (const auto& write_array : write_arrays) {
            ArrowArray c_array;
            PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*write_array, &c_array));
            auto record_batch = std::make_unique<RecordBatch>(
                partition, /*bucket=*/0,
                /*row_kinds=*/std::vector<RecordBatch::RowKind>(), &c_array);
            PAIMON_RETURN_NOT_OK(file_store_write->Write(std::move(record_batch)));
        }
        PAIMON_ASSIGN_OR_RAISE(auto commit_msgs,
                               file_store_write->PrepareCommit(
                                   /*wait_compaction=*/false, /*commit_identifier=*/0));
        PAIMON_RETURN_NOT_OK(file_store_write->Close());
        return commit_msgs;
    }

    void SetFirstRowId(int64_t reset_first_row_id,
                       std::vector<std::shared_ptr<CommitMessage>>& commit_msgs) const {
        for (auto& commit_msg : commit_msgs) {
            auto commit_msg_impl = std::dynamic_pointer_cast<CommitMessageImpl>(commit_msg);
            ASSERT_TRUE(commit_msg_impl);
            for (auto& file : commit_msg_impl->data_increment_.new_files_) {
                file->AssignFirstRowId(reset_first_row_id);
            }
        }
    }

    Status Commit(const std::string& table_path,
                  const std::vector<std::shared_ptr<CommitMessage>>& commit_msgs) const {
        // commit
        CommitContextBuilder commit_builder(table_path, "commit_user_1");
        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<CommitContext> commit_context,
                               commit_builder.Finish());
        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreCommit> file_store_commit,
                               FileStoreCommit::Create(std::move(commit_context)));
        return file_store_commit->Commit(commit_msgs);
    }

    /// DeletionVectorTestHelper::CreateDeletionVectorCommitMessage, documented there, plus the
    /// commit of the message it returns.
    Result<std::shared_ptr<CommitMessage>> CommitDeletionVectors(
        const std::string& table_path, const std::shared_ptr<CommitMessage>& base_commit_msg,
        const std::map<std::string, std::vector<int64_t>>& deleted_positions_by_anchor,
        const std::shared_ptr<CommitMessage>& replaced_commit_msg = nullptr) const {
        PAIMON_ASSIGN_OR_RAISE(
            std::shared_ptr<CommitMessage> commit_message,
            DeletionVectorTestHelper::CreateDeletionVectorCommitMessage(
                dir_->GetFileSystem(), table_path, /*file_format_identifier=*/GetParam(),
                base_commit_msg, deleted_positions_by_anchor, pool_, replaced_commit_msg));
        PAIMON_RETURN_NOT_OK(Commit(table_path, {commit_message}));
        return commit_message;
    }

    /// Creates the (f0 INT, view) blob-view table the dangling-reference tests share: unaware
    /// bucket, data evolution on, and `view` resolved against `upstream_warehouse`. Returns the
    /// table's fields, which the callers reuse to build their expected arrays.
    arrow::FieldVector CreateBlobViewTable(const std::string& upstream_warehouse,
                                           bool deletion_vectors_enabled) const {
        arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                     BlobUtils::ToArrowField("view", true)};
        std::map<std::string, std::string> options = {
            {Options::MANIFEST_FORMAT, "orc"},
            {Options::FILE_FORMAT, GetParam()},
            {Options::BUCKET, "-1"},
            {Options::ROW_TRACKING_ENABLED, "true"},
            {Options::DATA_EVOLUTION_ENABLED, "true"},
            {Options::BLOB_VIEW_FIELD, "view"},
            {Options::BLOB_VIEW_UPSTREAM_WAREHOUSE, upstream_warehouse},
            {Options::FILE_SYSTEM, "local"}};
        if (deletion_vectors_enabled) {
            options.emplace(Options::DELETION_VECTORS_ENABLED, "true");
        }
        CreateTable(fields, /*partition_keys=*/{}, options);
        return fields;
    }

    /// Writes and commits the two rows the blob-view skip tests share: row 0 holds a blob view
    /// reference to an upstream table that does not exist, row 1 holds no blob view at all.
    /// Resolving row 0 fails the whole read, so any read the pre-read still reaches it from must
    /// error; only dropping the row through the deletion vector or the row ranges avoids that,
    /// because those are the two filters the pre-read applies. The table must already exist with
    /// fields (f0, view).
    ///
    /// `first_row_id`, when set, is stamped on the returned metas before committing: the commit
    /// assigns row ids onto its own copies, so a caller that needs them (to look up the group's
    /// anchor) has to set them here, matching what the commit would assign.
    Result<std::vector<std::shared_ptr<CommitMessage>>> WriteDanglingBlobViewRows(
        const std::string& table_path,
        const std::optional<int64_t>& first_row_id = std::nullopt) const {
        Identifier upstream_identifier("nonexistent_db", "nonexistent_table");
        BlobViewStruct view_struct(upstream_identifier, /*field_id=*/2, /*row_id=*/0);
        auto serialized = view_struct.Serialize(pool_);
        arrow::LargeBinaryBuilder view_builder;
        PAIMON_RETURN_NOT_OK_FROM_ARROW(view_builder.Append(
            reinterpret_cast<const uint8_t*>(serialized->data()), serialized->size()));
        PAIMON_RETURN_NOT_OK_FROM_ARROW(view_builder.AppendNull());
        std::shared_ptr<arrow::Array> write_view_array;
        PAIMON_RETURN_NOT_OK_FROM_ARROW(view_builder.Finish(&write_view_array));
        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
            std::shared_ptr<arrow::Array> write_f0_array,
            arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), R"([100, 101])"));
        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
            std::shared_ptr<arrow::Array> write_struct,
            arrow::StructArray::Make(arrow::ArrayVector({write_f0_array, write_view_array}),
                                     std::vector<std::string>({"f0", "view"})));

        PAIMON_ASSIGN_OR_RAISE(
            std::vector<std::shared_ptr<CommitMessage>> commit_msgs,
            WriteArray(table_path, /*partition=*/{}, {"f0", "view"}, {write_struct}));
        if (first_row_id) {
            SetFirstRowId(first_row_id.value(), commit_msgs);
        }
        PAIMON_RETURN_NOT_OK(Commit(table_path, commit_msgs));
        return commit_msgs;
    }

    Status WriteNextSchema(const std::string& table_path, const std::vector<DataField>& fields,
                           int32_t highest_field_id,
                           const std::map<std::string, std::string>& options) const {
        return TestHelper::WriteNextSchema(dir_->GetFileSystem(), table_path, fields,
                                           highest_field_id, options);
    }

    /// Scan table and return the plan (without reading data).
    Result<std::shared_ptr<Plan>> ScanTable(const std::string& table_path,
                                            const std::shared_ptr<Predicate>& predicate = nullptr,
                                            const std::vector<Range>& row_ranges = {}) const {
        ScanContextBuilder scan_context_builder(table_path);
        scan_context_builder.SetPredicate(predicate);
        if (!row_ranges.empty()) {
            auto global_index_result = BitmapGlobalIndexResult::FromRanges(row_ranges);
            scan_context_builder.SetGlobalIndexResult(global_index_result);
        }
        PAIMON_ASSIGN_OR_RAISE(auto scan_context, scan_context_builder.Finish());
        PAIMON_ASSIGN_OR_RAISE(auto table_scan, TableScan::Create(std::move(scan_context)));
        PAIMON_ASSIGN_OR_RAISE(auto result_plan, table_scan->CreatePlan());
        return result_plan;
    }

    /// Read from table using a pre-scanned plan, returning the ChunkedArray and batch_reader.
    /// The batch_reader must outlive the returned ChunkedArray (array memory depends on reader).
    Result<ReadResult> ReadTable(const std::string& table_path,
                                 const std::vector<std::string>& read_schema,
                                 const std::shared_ptr<Plan>& plan,
                                 const std::shared_ptr<Predicate>& predicate = nullptr,
                                 const std::map<std::string, std::string>& options = {}) const {
        auto splits = plan->Splits();
        ReadContextBuilder read_context_builder(table_path);
        read_context_builder.SetReadFieldNames(read_schema).SetPredicate(predicate);
        if (!options.empty()) {
            read_context_builder.SetOptions(options);
        }
        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReadContext> read_context,
                               read_context_builder.Finish());
        PAIMON_ASSIGN_OR_RAISE(auto table_read, TableRead::Create(std::move(read_context)));
        PAIMON_ASSIGN_OR_RAISE(auto batch_reader, table_read->CreateReader(splits));
        PAIMON_ASSIGN_OR_RAISE(auto read_result,
                               ReadResultCollector::CollectResult(batch_reader.get()));
        return ReadResult{std::move(batch_reader), std::move(read_result)};
    }

    /// Convenience: scan + read in one call.
    Result<ReadResult> ScanAndReadResult(const std::string& table_path,
                                         const std::vector<std::string>& read_schema,
                                         const std::shared_ptr<Predicate>& predicate = nullptr,
                                         const std::vector<Range>& row_ranges = {}) const {
        PAIMON_ASSIGN_OR_RAISE(auto result_plan, ScanTable(table_path, predicate, row_ranges));
        return ReadTable(table_path, read_schema, result_plan, predicate);
    }

    /// Prepend a _VALUE_KIND (Insert) column to a StructArray.
    static Result<std::shared_ptr<arrow::StructArray>> PrependRowKindColumn(
        const std::shared_ptr<arrow::StructArray>& array) {
        auto row_kind_scalar =
            std::make_shared<arrow::Int8Scalar>(RowKind::Insert()->ToByteValue());
        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
            auto row_kind_array, arrow::MakeArrayFromScalar(*row_kind_scalar, array->length()));
        arrow::ArrayVector fields_with_row_kind = array->fields();
        std::vector<std::string> names_with_row_kind =
            arrow::schema(array->type()->fields())->field_names();
        fields_with_row_kind.insert(fields_with_row_kind.begin(), row_kind_array);
        names_with_row_kind.insert(names_with_row_kind.begin(), "_VALUE_KIND");
        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
            auto result, arrow::StructArray::Make(fields_with_row_kind, names_with_row_kind));
        return std::dynamic_pointer_cast<arrow::StructArray>(result);
    }

    Status ScanAndRead(const std::string& table_path, const std::vector<std::string>& read_schema,
                       const std::shared_ptr<arrow::StructArray>& expected_array,
                       const std::shared_ptr<Predicate>& predicate = nullptr,
                       const std::vector<Range>& row_ranges = {}) const {
        PAIMON_ASSIGN_OR_RAISE(auto scan_read,
                               ScanAndReadResult(table_path, read_schema, predicate, row_ranges));

        if (!expected_array) {
            EXPECT_FALSE(scan_read.chunked_array);
            return Status::OK();
        }
        PAIMON_ASSIGN_OR_RAISE(auto expected_with_row_kind, PrependRowKindColumn(expected_array));
        auto expected_chunk_array = std::make_shared<arrow::ChunkedArray>(expected_with_row_kind);
        EXPECT_TRUE(expected_chunk_array->Equals(scan_read.chunked_array))
            << "result:" << scan_read.chunked_array->ToString() << std::endl
            << "expected:" << expected_chunk_array->ToString();
        return Status::OK();
    }

    std::shared_ptr<arrow::StructArray> PrepareBulkData(
        int32_t write_batch_size, std::function<std::string(int32_t)> data_generator,
        const arrow::FieldVector& fields) const {
        std::string data_str = "[";
        for (int32_t i = 0; i < write_batch_size; i++) {
            data_str.append("[");
            auto row_str = data_generator(i);
            data_str.append(row_str);
            data_str.append("],");
        }
        data_str.pop_back();
        data_str.append("]");
        return std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), data_str)
                .ValueOrDie());
    }

    /// Convert a StructArray with raw blob bytes into a StructArray with serialized
    /// BlobDescriptor bytes. Each raw blob value is written to a temporary file, and
    /// the corresponding cell is replaced with the serialized BlobDescriptor pointing
    /// to that file.
    /// Common framework for transforming blob fields in a StructArray.
    /// Non-blob fields are kept as-is; blob fields are processed row-by-row via `transform_row`.
    /// `transform_row` receives (binary_value_view) and returns the transformed bytes via builder.
    using BlobRowTransform =
        std::function<Status(const std::string_view& value, arrow::LargeBinaryBuilder* builder)>;

    Result<std::shared_ptr<arrow::StructArray>> TransformBlobFields(
        const std::shared_ptr<arrow::StructArray>& input_array,
        const std::set<std::string>& blob_fields, BlobRowTransform transform_row) const {
        auto fields = input_array->type()->fields();
        arrow::ArrayVector child_arrays;

        for (const auto& field : fields) {
            auto col = input_array->GetFieldByName(field->name());
            if (blob_fields.count(field->name()) == 0) {
                child_arrays.push_back(col);
                continue;
            }
            const auto& binary_array = checked_cast<const arrow::LargeBinaryArray&>(*col);
            arrow::LargeBinaryBuilder builder;
            for (int64_t i = 0; i < binary_array.length(); ++i) {
                if (binary_array.IsNull(i)) {
                    PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.AppendNull());
                    continue;
                }
                PAIMON_RETURN_NOT_OK(transform_row(binary_array.GetView(i), &builder));
            }
            std::shared_ptr<arrow::Array> result_col;
            PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&result_col));
            child_arrays.push_back(result_col);
        }

        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(auto result,
                                          arrow::StructArray::Make(child_arrays, fields));
        return result;
    }

    Result<std::shared_ptr<arrow::StructArray>> ConvertRawBlobToDescriptor(
        const std::shared_ptr<arrow::StructArray>& raw_array,
        const std::set<std::string>& blob_fields) {
        auto fs = std::make_shared<LocalFileSystem>();
        return TransformBlobFields(
            raw_array, blob_fields,
            [&](const std::string_view& raw_value, arrow::LargeBinaryBuilder* builder) -> Status {
                std::string file_path =
                    blob_dir_->Str() + "/blob_" + std::to_string(blob_file_counter_++) + ".bin";
                auto raw_size = static_cast<int64_t>(raw_value.size());
                PAIMON_ASSIGN_OR_RAISE(auto out, fs->Create(file_path, /*overwrite=*/true));
                PAIMON_ASSIGN_OR_RAISE(auto written, out->Write(raw_value.data(), raw_size));
                PAIMON_RETURN_NOT_OK(out->Flush());
                PAIMON_RETURN_NOT_OK(out->Close());
                if (written != raw_size) {
                    return Status::Invalid("Short write: expected {}, wrote {}", raw_value.size(),
                                           written);
                }
                PAIMON_ASSIGN_OR_RAISE(auto blob, Blob::FromPath(file_path));
                auto descriptor = blob->ToDescriptor(pool_);
                PAIMON_RETURN_NOT_OK_FROM_ARROW(
                    builder->Append(descriptor->data(), descriptor->size()));
                return Status::OK();
            });
    }

    /// 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<arrow::StructArray>& desc_array,
                                  const std::string& field_name, int64_t row) const {
        const auto& blob_col =
            checked_cast<const arrow::LargeBinaryArray&>(*desc_array->GetFieldByName(field_name));
        std::string_view descriptor_bytes = blob_col.GetView(row);
        PAIMON_ASSIGN_OR_RAISE(
            std::unique_ptr<BlobDescriptor> descriptor,
            BlobDescriptor::Deserialize(descriptor_bytes.data(), descriptor_bytes.size()));
        LocalFileSystem fs;
        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<std::shared_ptr<arrow::StructArray>> CorruptDescriptorOffset(
        const std::shared_ptr<arrow::StructArray>& 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<BlobDescriptor> descriptor,
                    BlobDescriptor::Deserialize(descriptor_bytes.data(), descriptor_bytes.size()));
                PAIMON_ASSIGN_OR_RAISE(
                    std::unique_ptr<BlobDescriptor> 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<std::string> table_relative_blob_dirs;
    };

    static std::optional<std::string> TryRewriteDescriptorUri(
        const std::string& descriptor_uri, const BlobDescriptorPathRewrite& rewrite,
        const std::shared_ptr<LocalFileSystem>& fs) {
        if (rewrite.table_path.empty()) {
            return std::nullopt;
        }

        for (const auto& blob_dir : rewrite.table_relative_blob_dirs) {
            const std::string marker = "/" + blob_dir + "/";
            auto marker_pos = descriptor_uri.find(marker);
            if (marker_pos != std::string::npos) {
                std::string relative_blob_path = descriptor_uri.substr(marker_pos + 1);
                return PathUtil::JoinPath(rewrite.table_path, relative_blob_path);
            }
        }
        return std::nullopt;
    }

    /// Convert a StructArray with serialized BlobDescriptor bytes back to a StructArray
    /// with raw blob bytes. Only blob fields are resolved; other columns (including
    /// _VALUE_KIND) are kept as-is.
    Result<std::shared_ptr<arrow::StructArray>> ConvertDescriptorToRawBlob(
        const std::shared_ptr<arrow::StructArray>& desc_array,
        const std::set<std::string>& blob_fields,
        const BlobDescriptorPathRewrite& rewrite = {}) const {
        auto fs = std::make_shared<LocalFileSystem>();
        return TransformBlobFields(
            desc_array, blob_fields,
            [&](const std::string_view& descriptor_bytes,
                arrow::LargeBinaryBuilder* builder) -> Status {
                PAIMON_ASSIGN_OR_RAISE(
                    auto descriptor,
                    BlobDescriptor::Deserialize(descriptor_bytes.data(), descriptor_bytes.size()));
                std::string descriptor_uri = descriptor->Uri();
                auto rewritten_uri = TryRewriteDescriptorUri(descriptor_uri, rewrite, fs);
                if (rewritten_uri.has_value()) {
                    descriptor_uri = rewritten_uri.value();
                }

                PAIMON_ASSIGN_OR_RAISE(
                    auto rewritten_descriptor,
                    BlobDescriptor::Create(descriptor->Version(), descriptor_uri,
                                           descriptor->Offset(), descriptor->Length()));
                auto rewritten_descriptor_bytes = rewritten_descriptor->Serialize(pool_);
                PAIMON_ASSIGN_OR_RAISE(auto blob,
                                       Blob::FromDescriptor(rewritten_descriptor_bytes->data(),
                                                            rewritten_descriptor_bytes->size()));
                PAIMON_ASSIGN_OR_RAISE(auto data, blob->ToData(fs, pool_));
                PAIMON_RETURN_NOT_OK_FROM_ARROW(builder->Append(data->data(), data->size()));
                return Status::OK();
            });
    }

    /// Verify DataFileMeta properties from a scan plan.
    /// Each vector element corresponds to one expected DataFileMeta (ordered by file index).
    static void VerifyDataFileMetas(
        const std::shared_ptr<Plan>& plan, size_t expected_file_count,
        const std::vector<int64_t>& expected_row_counts,
        const std::vector<int64_t>& expected_min_seqs,
        const std::vector<int64_t>& expected_max_seqs,
        const std::vector<int64_t>& expected_first_row_ids,
        const std::vector<std::optional<std::vector<std::string>>>& expected_write_cols) {
        std::vector<std::shared_ptr<DataFileMeta>> all_files;
        for (const auto& split : plan->Splits()) {
            auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
            ASSERT_TRUE(data_split);
            for (const auto& file : data_split->DataFiles()) {
                all_files.push_back(file);
            }
        }
        ASSERT_EQ(all_files.size(), expected_file_count);
        ASSERT_EQ(expected_row_counts.size(), expected_file_count);
        ASSERT_EQ(expected_min_seqs.size(), expected_file_count);
        ASSERT_EQ(expected_max_seqs.size(), expected_file_count);
        ASSERT_EQ(expected_first_row_ids.size(), expected_file_count);
        ASSERT_EQ(expected_write_cols.size(), expected_file_count);
        for (size_t i = 0; i < all_files.size(); ++i) {
            const auto& file = all_files[i];
            EXPECT_EQ(file->row_count, expected_row_counts[i]);
            EXPECT_EQ(file->min_sequence_number, expected_min_seqs[i]);
            EXPECT_EQ(file->max_sequence_number, expected_max_seqs[i]);
            ASSERT_TRUE(file->first_row_id.has_value());
            EXPECT_EQ(file->first_row_id.value(), expected_first_row_ids[i]);
            EXPECT_EQ(file->write_cols, expected_write_cols[i]);
        }
    }

 private:
    std::shared_ptr<MemoryPool> pool_;
    std::unique_ptr<UniqueTestDirectory> dir_;
    std::unique_ptr<UniqueTestDirectory> blob_dir_;
    int blob_file_counter_ = 0;
    arrow::FieldVector fields_ = {arrow::field("f0", arrow::int32()), BlobUtils::ToArrowField("f1"),
                                  arrow::field("f2", arrow::utf8())};
};

std::vector<std::string> GetTestValuesForBlobTableInteTest() {
    std::vector<std::string> values;
    values.emplace_back("parquet");
#ifdef PAIMON_ENABLE_ORC
    values.emplace_back("orc");
#endif
#ifdef PAIMON_ENABLE_AVRO
    values.emplace_back("avro");
#endif
    return values;
}

INSTANTIATE_TEST_SUITE_P(FileFormat, BlobTableInteTest,
                         ::testing::ValuesIn(GetTestValuesForBlobTableInteTest()));

TEST_P(BlobTableInteTest, TestAppendTableWriteWithBlobAsDescriptorTrue) {
    arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
                                 arrow::field("f1", arrow::int32()),
                                 BlobUtils::ToArrowField("blob", true)};

    std::map<std::string, std::string> 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");

    // prepare data: input uses plain raw blob bytes for readability
    std::string raw_json = R"([
        ["str_0", null, "hello_blob_0"],
        ["str_1", 1,    "blob_data_1"],
        ["str_2", 2,    "blob_data_2"],
        ["str_3", null, "blob_data_3"]
    ])";
    auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"}));

    // write descriptor array
    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));

    // read result contains descriptors pointing to paimon internal blob files
    // resolve descriptors back to raw bytes, then prepend _VALUE_KIND and compare
    ASSERT_OK_AND_ASSIGN(auto result, ScanAndReadResult(table_path, schema->field_names()));
    ASSERT_TRUE(result.chunked_array);
    auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie();
    auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
    ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"blob"}));
    ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(raw_array));
    ASSERT_TRUE(resolved->Equals(expected_with_rk));
}

TEST_P(BlobTableInteTest, TestAppendTableWriteWithBlobAsDescriptorFalse) {
    arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
                                 arrow::field("f1", arrow::int32()),
                                 BlobUtils::ToArrowField("blob", true)};

    std::map<std::string, std::string> 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, "false"},  {Options::FILE_SYSTEM, "local"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    std::string data_json = R"([
        ["str_0", null, "apple"],
        ["str_1", 1,    "banana"],
        ["str_2", 2,    "cat"],
        ["str_3", null, "dog"]
    ])";
    auto write_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), data_json).ValueOrDie());

    auto schema = arrow::schema(fields);
    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(table_path, {}, schema->field_names(), {write_array}));
    ASSERT_OK(Commit(table_path, commit_msgs));

    // BLOB_AS_DESCRIPTOR=false: blob data is stored inline, read result should match input
    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<std::string, std::string> 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::StructArray>(
        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<std::string>{"f0", "f1"}, std::vector<std::string>{"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<arrow::StructArray>(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::StructArray>(
        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<std::string, std::string> 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::StructArray>(
        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, 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<std::string, std::string> 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::StructArray>(
        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<std::string>{"f0", "f1"}, std::vector<std::string>{"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<arrow::StructArray>(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::StructArray>(
        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, TestWriteNullOnFetchFailureCoversMissingFile) {
    // The existence check runs only under blob-write-null-on-missing-file, so with only
    // blob-write-null-on-fetch-failure a missing file is converted as an ordinary fetch failure.
    arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
                                 arrow::field("f1", arrow::int32()),
                                 BlobUtils::ToArrowField("blob", true)};

    std::map<std::string, std::string> 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::StructArray>(
        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_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(table_path, {}, schema->field_names(), {desc_array}));
    ASSERT_OK(Commit(table_path, commit_msgs));

    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<std::string>{"f0", "f1"}, std::vector<std::string>{"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<arrow::StructArray>(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::StructArray>(
        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, TestBasic) {
    CreateTable();
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    // write field: f0, f1, f2
    std::vector<std::string> write_cols0 = schema->field_names();
    auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "a", "b"]
    ])")
            .ValueOrDie());

    ASSERT_OK_AND_ASSIGN(auto commit_msgs, WriteArray(table_path, {}, write_cols0, {src_array0}));
    ASSERT_OK(Commit(table_path, commit_msgs));

    // write field: f1, f2
    std::vector<std::string> write_cols1 = {"f1", "f2"};
    auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1], fields_[2]}), R"([
        ["new_blob", "c"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(commit_msgs, WriteArray(table_path, {}, write_cols1, {src_array1}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs);
    ASSERT_OK(Commit(table_path, commit_msgs));
    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "new_blob", "c"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));

    // read with row tracking
    auto expected_row_tracking_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(
            arrow::struct_({fields_[1], fields_[0], SpecialFields::SequenceNumber().field_,
                            SpecialFields::RowId().field_, fields_[2]}),
            R"([
        ["new_blob", 1, 2, 0, "c"]
    ])")
            .ValueOrDie());

    ASSERT_OK(ScanAndRead(table_path, {"f1", "f0", "_SEQUENCE_NUMBER", "_ROW_ID", "f2"},
                          expected_row_tracking_array));
}

TEST_P(BlobTableInteTest, TestBlobFilesAcrossSchemaIds) {
    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                  {Options::FILE_FORMAT, GetParam()},
                                                  {Options::FILE_SYSTEM, "local"},
                                                  {Options::ROW_TRACKING_ENABLED, "true"},
                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
    CreateTable(/*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    // Simulate the post-compaction layout: one normal file bridges blob files across schema ids.
    std::vector<std::string> write_cols0 = {"f0", "f2"};
    auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[2]}), R"([
        [1, "a"],
        [2, "b"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs0, WriteArray(table_path, {}, write_cols0, {src_array0}));
    ASSERT_OK(Commit(table_path, commit_msgs0));

    std::vector<std::string> blob_write_cols = {"f1"};
    auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1]}), R"([
        ["c"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1,
                         WriteArray(table_path, {}, blob_write_cols, {src_array1}));
    SetFirstRowId(0, commit_msgs1);
    ASSERT_OK(Commit(table_path, commit_msgs1));

    auto f3 = arrow::field("f3", arrow::int64());
    ASSERT_OK(WriteNextSchema(table_path,
                              {DataField(0, fields_[0]), DataField(1, fields_[1]),
                               DataField(2, fields_[2]), DataField(3, f3)},
                              /*highest_field_id=*/3, options));

    auto src_array2 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1]}), R"([
        ["d"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs2,
                         WriteArray(table_path, {}, blob_write_cols, {src_array2}));
    SetFirstRowId(1, commit_msgs2);
    ASSERT_OK(Commit(table_path, commit_msgs2));

    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "c", "a"],
        [2, "d", "b"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, arrow::schema(fields_)->field_names(), expected_array));
}

TEST_P(BlobTableInteTest, TestMultipleAppends) {
    CreateTable();
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    // write field: f0, f1, f2
    std::vector<std::string> write_cols0 = schema->field_names();
    auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs, WriteArray(table_path, {}, write_cols0, {src_array0}));
    ASSERT_OK(Commit(table_path, commit_msgs));

    // write field: f0, f1
    std::vector<std::string> write_cols1 = {"f0", "f1"};
    auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
        [1, "a"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, write_cols1, {src_array1}));
    SetFirstRowId(10, commit_msgs1);

    // write field: f2
    std::vector<std::string> write_cols2 = {"f2"};
    auto src_array2 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
        ["b"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs2, WriteArray(table_path, {}, write_cols2, {src_array2}));
    SetFirstRowId(10, commit_msgs2);

    std::vector<std::shared_ptr<CommitMessage>> total_msgs;
    total_msgs.insert(total_msgs.end(), commit_msgs1.begin(), commit_msgs1.end());
    total_msgs.insert(total_msgs.end(), commit_msgs2.begin(), commit_msgs2.end());
    ASSERT_OK(Commit(table_path, total_msgs));

    // write field: f0, f1
    std::vector<std::string> write_cols3 = {"f0", "f1"};
    auto src_array3 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
        [2, "c"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs3, WriteArray(table_path, {}, write_cols3, {src_array3}));
    SetFirstRowId(11, commit_msgs3);
    ASSERT_OK(Commit(table_path, commit_msgs3));

    // write field: f2
    std::vector<std::string> write_cols4 = {"f2"};
    auto src_array4 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
        ["d"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs4, WriteArray(table_path, {}, write_cols4, {src_array4}));
    SetFirstRowId(11, commit_msgs4);
    ASSERT_OK(Commit(table_path, commit_msgs4));

    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [1, "a", "b"],
        [2, "c", "d"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));

    // read with row tracking
    auto expected_row_tracking_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({
                                                      fields_[0],
                                                      fields_[1],
                                                      fields_[2],
                                                      SpecialFields::RowId().field_,
                                                      SpecialFields::SequenceNumber().field_,
                                                  }),
                                                  R"([
        [1, "a", "b", 0, 1],
        [1, "a", "b", 1, 1],
        [1, "a", "b", 2, 1],
        [1, "a", "b", 3, 1],
        [1, "a", "b", 4, 1],
        [1, "a", "b", 5, 1],
        [1, "a", "b", 6, 1],
        [1, "a", "b", 7, 1],
        [1, "a", "b", 8, 1],
        [1, "a", "b", 9, 1],
        [1, "a", "b", 10, 2],
        [2, "c", "d", 11, 4]
    ])")
            .ValueOrDie());

    ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2", "_ROW_ID", "_SEQUENCE_NUMBER"},
                          expected_row_tracking_array));
}

TEST_P(BlobTableInteTest, TestDataEvolutionBlobOnlyWriteWithFirstRowId) {
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0")};
    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                  {Options::FILE_FORMAT, GetParam()},
                                                  {Options::FILE_SYSTEM, "local"},
                                                  {Options::ROW_TRACKING_ENABLED, "true"},
                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields);

    // Initial full-row write assigns row ids 0 and 1.
    auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
        [1, "a", "old_blob_0"],
        [2, "b", "old_blob_1"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
                         WriteArray(table_path, {}, schema->field_names(), {src_array0}));
    ASSERT_OK(Commit(table_path, commit_msgs0));

    // Update only b0 and align it with the existing row ids.
    std::vector<std::string> blob_write_cols = {"b0"};
    auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[2]}), R"([
        ["new_blob_0"],
        ["new_blob_1"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1,
                         WriteArray(table_path, {}, blob_write_cols, {src_array1}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
    ASSERT_OK(Commit(table_path, commit_msgs1));

    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
        [1, "a", "new_blob_0"],
        [2, "b", "new_blob_1"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));

    auto expected_with_row_id = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(
            arrow::struct_({fields[0], fields[1], fields[2], SpecialFields::RowId().field_}),
            R"([
        [1, "a", "new_blob_0", 0],
        [2, "b", "new_blob_1", 1]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "b0", "_ROW_ID"}, expected_with_row_id));
}

/// Build a single-blob-column StructArray for a data-evolution partial update: "PH" marks a row
/// whose blob is not updated (persisted as a placeholder entry), std::nullopt a null blob.
std::shared_ptr<arrow::StructArray> MakeBlobUpdateArray(
    const std::shared_ptr<arrow::Field>& blob_field,
    const std::vector<std::optional<std::string>>& rows) {
    auto struct_type = arrow::struct_({blob_field});
    arrow::StructBuilder struct_builder(struct_type, arrow::default_memory_pool(),
                                        {std::make_shared<arrow::LargeBinaryBuilder>()});
    auto blob_builder = checked_cast<arrow::LargeBinaryBuilder*>(struct_builder.field_builder(0));
    for (const auto& row : rows) {
        EXPECT_TRUE(struct_builder.Append().ok());
        if (!row) {
            EXPECT_TRUE(blob_builder->AppendNull().ok());
        } else if (*row == "PH") {
            std::string_view sentinel = BlobDefs::PlaceholderSentinelView();
            EXPECT_TRUE(blob_builder->Append(sentinel.data(), sentinel.size()).ok());
        } else {
            EXPECT_TRUE(blob_builder->Append(row->data(), row->size()).ok());
        }
    }
    std::shared_ptr<arrow::Array> array;
    EXPECT_TRUE(struct_builder.Finish(&array).ok());
    return std::dynamic_pointer_cast<arrow::StructArray>(array);
}

TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateFallback) {
    // the blob column is updated to null below, so it must be nullable
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 arrow::field("f1", arrow::utf8()),
                                 BlobUtils::ToArrowField("b0", /*nullable=*/true)};
    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                  {Options::FILE_FORMAT, GetParam()},
                                                  {Options::FILE_SYSTEM, "local"},
                                                  {Options::ROW_TRACKING_ENABLED, "true"},
                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields);

    // Initial full-row write assigns row ids 0-2.
    auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
        [1, "a", "blob_a"],
        [2, "b", "blob_b"],
        [3, "c", "blob_c"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
                         WriteArray(table_path, {}, schema->field_names(), {src_array0}));
    ASSERT_OK(Commit(table_path, commit_msgs0));

    // Partial update: only row 1 gets a new blob, the untouched rows are written as
    // placeholder entries and must fall back to the previous blob file when read.
    auto update_array = MakeBlobUpdateArray(fields[2], {"PH", "updated_b", "PH"});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
    ASSERT_OK(Commit(table_path, commit_msgs1));

    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
        [1, "a", "blob_a"],
        [2, "b", "updated_b"],
        [3, "c", "blob_c"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));

    // updating a blob to null is not a placeholder: the null must win over older layers
    auto null_update_array = MakeBlobUpdateArray(fields[2], {std::nullopt, "PH", "PH"});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs2,
                         WriteArray(table_path, {}, {"b0"}, {null_update_array}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs2);
    ASSERT_OK(Commit(table_path, commit_msgs2));

    auto expected_array2 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
        [1, "a", null],
        [2, "b", "updated_b"],
        [3, "c", "blob_c"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array2));

    // row ids still come from the data files and stay aligned with the fallback result
    auto expected_with_row_id = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(
            arrow::struct_({fields[0], fields[1], fields[2], SpecialFields::RowId().field_}),
            R"([
        [1, "a", null, 0],
        [2, "b", "updated_b", 1],
        [3, "c", "blob_c", 2]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "b0", "_ROW_ID"}, expected_with_row_id));

    // blob_as_descriptor read mode: the fallback-merged values come back as descriptors and
    // must still resolve to the same bytes
    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
    std::map<std::string, std::string> read_options = {{Options::BLOB_AS_DESCRIPTOR, "true"}};
    ASSERT_OK_AND_ASSIGN(auto desc_result, ReadTable(table_path, schema->field_names(), plan,
                                                     /*predicate=*/nullptr, read_options));
    ASSERT_TRUE(desc_result.chunked_array);
    auto desc_concat = arrow::Concatenate(desc_result.chunked_array->chunks()).ValueOrDie();
    auto desc_struct = std::dynamic_pointer_cast<arrow::StructArray>(desc_concat);
    ASSERT_TRUE(desc_struct);
    ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(desc_struct, {"b0"}));
    ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_array2));
    ASSERT_TRUE(resolved->Equals(expected_with_rk))
        << "result:" << resolved->ToString() << "\nexpected:" << expected_with_rk->ToString();
}

TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateMultipleLayers) {
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0")};
    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                  {Options::FILE_FORMAT, GetParam()},
                                                  {Options::FILE_SYSTEM, "local"},
                                                  {Options::ROW_TRACKING_ENABLED, "true"},
                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields);

    auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
        [1, "a", "blob_0"],
        [2, "b", "blob_1"],
        [3, "c", "blob_2"],
        [4, "d", "blob_3"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
                         WriteArray(table_path, {}, schema->field_names(), {src_array0}));
    ASSERT_OK(Commit(table_path, commit_msgs0));

    // second layer updates row 0 within rows [0, 1]
    auto update_array1 = MakeBlobUpdateArray(fields[2], {"update1_0", "PH"});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array1}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
    ASSERT_OK(Commit(table_path, commit_msgs1));

    // third layer updates row 3 within rows [2, 3]; each layer only partially covers the
    // range, the uncovered parts behave as placeholders
    auto update_array2 = MakeBlobUpdateArray(fields[2], {"PH", "update2_3"});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs2, WriteArray(table_path, {}, {"b0"}, {update_array2}));
    SetFirstRowId(/*reset_first_row_id=*/2, commit_msgs2);
    ASSERT_OK(Commit(table_path, commit_msgs2));

    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
        [1, "a", "update1_0"],
        [2, "b", "blob_1"],
        [3, "c", "blob_2"],
        [4, "d", "update2_3"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));

    // row-range pushdown where the selected rows fall in one layer's file and in another
    // layer's uncovered gap: row 1 is a gap row for the third layer, row 2 for the second
    auto expected_middle = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
        [2, "b", "blob_1"],
        [3, "c", "blob_2"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_middle,
                          /*predicate=*/nullptr, /*row_ranges=*/{Range(1, 2)}));

    // per-row reads resolve every row independently
    const std::vector<std::string> expected_rows = {
        R"([[1, "a", "update1_0"]])", R"([[2, "b", "blob_1"]])", R"([[3, "c", "blob_2"]])",
        R"([[4, "d", "update2_3"]])"};
    for (int32_t i = 0; i < static_cast<int32_t>(expected_rows.size()); i++) {
        auto expected_single = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_rows[i])
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_single,
                              /*predicate=*/nullptr, /*row_ranges=*/{Range(i, i)}));
    }
}

TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateWithDeletionVectors) {
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0")};
    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},         {Options::FILE_FORMAT, GetParam()},
        {Options::FILE_SYSTEM, "local"},           {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "true"}, {Options::DELETION_VECTORS_ENABLED, "true"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields);

    auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
        [1, "a", "blob_0"],
        [2, "b", "blob_1"],
        [3, "c", "blob_2"],
        [4, "d", "blob_3"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
                         WriteArray(table_path, {}, schema->field_names(), {src_array0}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs0);
    ASSERT_OK(Commit(table_path, commit_msgs0));

    // second layer updates row 0 within rows [0, 1], rows [2, 3] are its placeholder gap
    auto update_array1 = MakeBlobUpdateArray(fields[2], {"update1_0", "PH"});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array1}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
    ASSERT_OK(Commit(table_path, commit_msgs1));

    // third layer updates row 3 within rows [2, 3], rows [0, 1] are its placeholder gap
    auto update_array2 = MakeBlobUpdateArray(fields[2], {"PH", "update2_3"});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs2, WriteArray(table_path, {}, {"b0"}, {update_array2}));
    SetFirstRowId(/*reset_first_row_id=*/2, commit_msgs2);
    ASSERT_OK(Commit(table_path, commit_msgs2));

    // delete row 1 and row 3: each sits inside one blob layer's file and the other layer's
    // gap, so both blob layers and the normal file must drop the same rows to stay aligned
    ASSERT_OK_AND_ASSIGN(std::string anchor_file_name,
                         DeletionVectorTestHelper::RetrieveAnchorFileName(commit_msgs0));
    ASSERT_OK(CommitDeletionVectors(table_path, commit_msgs0[0],
                                    {{anchor_file_name, /*deleted_positions=*/{1, 3}}}));

    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
        [1, "a", "update1_0"],
        [3, "c", "blob_2"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));

    auto expected_with_row_id = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(
            arrow::struct_({fields[0], fields[1], fields[2], SpecialFields::RowId().field_}),
            R"([
        [1, "a", "update1_0", 0],
        [3, "c", "blob_2", 2]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "b0", "_ROW_ID"}, expected_with_row_id));

    // Projecting only the blob column leaves no data bunch, so the whole read is the blob
    // fallback path: the layer files carry the shifted deletion vector and the placeholder gap
    // segments drop the deleted row ids, with no normal file to anchor the row count.
    auto expected_blob_only = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[2]}), R"([
        ["update1_0"],
        ["blob_2"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"b0"}, expected_blob_only));

    // a row-range selection composes with the deletion vector: rows {0, 1} minus deleted {1}.
    // The range stops before row 2, the other survivor, so it cannot pass by returning what the
    // unrestricted read above already returned.
    auto expected_selected = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
        [1, "a", "update1_0"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_selected,
                          /*predicate=*/nullptr, /*row_ranges=*/{Range(0, 1)}));
}

TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateCompactedLayers) {
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0")};
    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                  {Options::FILE_FORMAT, GetParam()},
                                                  {Options::FILE_SYSTEM, "local"},
                                                  {Options::ROW_TRACKING_ENABLED, "true"},
                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields);

    // Mirrors the layer shapes of Java's BlobUpdateTest.testReadCompactedBlobSequenceGroups
    // (with a single base write, as the C++ write path assigns one sequence per commit):
    //   row id:  0    1    2    3    4    5    6    7    8    9
    //   seq1:   [b0   b1   b2   b3   b4   b5   b6   b7   b8   b9]
    //   seq2:   [u20  P    P    u23  u24] .    .    .    .    .
    //   seq3:    .    .    .    .    .   [P    u46  P    u48  P]
    //   seq4:   [P    u61  P    P    P    P    P    P    P    u69]
    //   result:  u20  u61  b2   u23  u24  b5   u46  b7   u48  u69
    auto base_array = PrepareBulkData(
        10,
        [](int32_t i) {
            return std::to_string(i) + ", \"name_" + std::to_string(i) + "\", \"blob_" +
                   std::to_string(i) + "\"";
        },
        fields);
    ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
                         WriteArray(table_path, {}, schema->field_names(), {base_array}));
    ASSERT_OK(Commit(table_path, commit_msgs0));

    auto update_array1 = MakeBlobUpdateArray(fields[2], {"u20", "PH", "PH", "u23", "u24"});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array1}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
    ASSERT_OK(Commit(table_path, commit_msgs1));

    auto update_array2 = MakeBlobUpdateArray(fields[2], {"PH", "u46", "PH", "u48", "PH"});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs2, WriteArray(table_path, {}, {"b0"}, {update_array2}));
    SetFirstRowId(/*reset_first_row_id=*/5, commit_msgs2);
    ASSERT_OK(Commit(table_path, commit_msgs2));

    auto update_array3 = MakeBlobUpdateArray(
        fields[2], {"PH", "u61", "PH", "PH", "PH", "PH", "PH", "PH", "PH", "u69"});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs3, WriteArray(table_path, {}, {"b0"}, {update_array3}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs3);
    ASSERT_OK(Commit(table_path, commit_msgs3));

    const std::vector<std::string> expected_blobs = {"u20",    "u61", "blob_2", "u23", "u24",
                                                     "blob_5", "u46", "blob_7", "u48", "u69"};
    auto expected_row = [&](int32_t i) {
        return std::to_string(i) + ", \"name_" + std::to_string(i) + "\", \"" + expected_blobs[i] +
               "\"";
    };
    auto expected_full = PrepareBulkData(10, expected_row, fields);
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_full));

    // every row resolves independently under row-range pushdown
    for (int32_t i = 0; i < 10; i++) {
        auto expected_single = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
                                                      "[[" + expected_row(i) + "]]")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_single,
                              /*predicate=*/nullptr, /*row_ranges=*/{Range(i, i)}));
    }
}

TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateWithRowRanges) {
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0")};
    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                  {Options::FILE_FORMAT, GetParam()},
                                                  {Options::FILE_SYSTEM, "local"},
                                                  {Options::ROW_TRACKING_ENABLED, "true"},
                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields);

    auto base_array = PrepareBulkData(
        10,
        [](int32_t i) {
            return std::to_string(i) + ", \"name_" + std::to_string(i) + "\", \"blob_" +
                   std::to_string(i) + "\"";
        },
        fields);
    ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
                         WriteArray(table_path, {}, schema->field_names(), {base_array}));
    ASSERT_OK(Commit(table_path, commit_msgs0));

    // partial update touching rows 1 and 9 only
    auto update_array = MakeBlobUpdateArray(
        fields[2], {"PH", "update_1", "PH", "PH", "PH", "PH", "PH", "PH", "PH", "update_9"});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
    ASSERT_OK(Commit(table_path, commit_msgs1));

    // push down row ranges hitting updated and untouched rows
    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
        [1, "name_1", "update_1"],
        [5, "name_5", "blob_5"],
        [9, "name_9", "update_9"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
                          /*predicate=*/nullptr,
                          /*row_ranges=*/{Range(1, 1), Range(5, 5), Range(9, 9)}));

    // full read still resolves every row
    auto expected_full = PrepareBulkData(
        10,
        [](int32_t i) {
            std::string blob = (i == 1 || i == 9) ? "\"update_" + std::to_string(i) + "\""
                                                  : "\"blob_" + std::to_string(i) + "\"";
            return std::to_string(i) + ", \"name_" + std::to_string(i) + "\", " + blob;
        },
        fields);
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_full));
}

/// A BLOB sequence layer whose physical file covers only a strict subrange of the full row id
/// range: the newest group is internally Gap(2), File([u2, u3]), Gap(6). The table needs a
/// normal column to be creatable, but only the blob column is ever written, so the split
/// contains only blob files and the blob bunch itself carries the row-tracking fields: the
/// fallback reader must keep _ROW_ID correct and report each row's _SEQUENCE_NUMBER from the
/// layer that resolved it, without ever exposing the internal placeholder sentinel.
TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateRowTrackingWithSubrangeLayer) {
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 BlobUtils::ToArrowField("b0", /*nullable=*/true)};
    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                  {Options::FILE_FORMAT, GetParam()},
                                                  {Options::FILE_SYSTEM, "local"},
                                                  {Options::ROW_TRACKING_ENABLED, "true"},
                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    // old layer: rows 0-9 all real
    std::vector<std::optional<std::string>> base_rows;
    for (int32_t i = 0; i < 10; i++) {
        base_rows.emplace_back("b" + std::to_string(i));
    }
    auto base_array = MakeBlobUpdateArray(fields[1], base_rows);
    ASSERT_OK_AND_ASSIGN(auto commit_msgs0, WriteArray(table_path, {}, {"b0"}, {base_array}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs0);
    ASSERT_OK(Commit(table_path, commit_msgs0));

    // new layer: first_row_id=2, row_count=2, covering only rows 2-3
    auto update_array = MakeBlobUpdateArray(fields[1], {"u2", "u3"});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array}));
    SetFirstRowId(/*reset_first_row_id=*/2, commit_msgs1);
    ASSERT_OK(Commit(table_path, commit_msgs1));

    ASSERT_OK_AND_ASSIGN(auto scan_read,
                         ScanAndReadResult(table_path, {"b0", "_ROW_ID", "_SEQUENCE_NUMBER"}));
    ASSERT_TRUE(scan_read.chunked_array);
    auto concat_array = arrow::Concatenate(scan_read.chunked_array->chunks()).ValueOrDie();
    auto struct_array = std::dynamic_pointer_cast<arrow::StructArray>(concat_array);
    ASSERT_TRUE(struct_array);
    ASSERT_EQ(struct_array->length(), 10);
    auto blob_col =
        std::dynamic_pointer_cast<arrow::LargeBinaryArray>(struct_array->GetFieldByName("b0"));
    auto row_id_col = std::dynamic_pointer_cast<arrow::Int64Array>(
        struct_array->GetFieldByName(SpecialFields::RowId().Name()));
    auto seq_col = std::dynamic_pointer_cast<arrow::Int64Array>(
        struct_array->GetFieldByName(SpecialFields::SequenceNumber().Name()));
    ASSERT_TRUE(blob_col && row_id_col && seq_col);
    int64_t old_layer_seq = seq_col->Value(0);
    int64_t new_layer_seq = seq_col->Value(2);
    ASSERT_GT(new_layer_seq, old_layer_seq);
    for (int64_t i = 0; i < 10; i++) {
        ASSERT_FALSE(blob_col->IsNull(i));
        std::string expected_blob =
            (i == 2 || i == 3) ? "u" + std::to_string(i) : "b" + std::to_string(i);
        // the leading and trailing gaps never expose the internal placeholder sentinel
        ASSERT_EQ(blob_col->GetString(i), expected_blob) << "row " << i;
        ASSERT_EQ(row_id_col->Value(i), i);
        ASSERT_EQ(seq_col->Value(i), (i == 2 || i == 3) ? new_layer_seq : old_layer_seq)
            << "row " << i;
    }

    // row-range pushdown with the row-tracking projection: the selection removes rows inside
    // the blob files, so batch positions must still map back to the right file row indexes
    ASSERT_OK_AND_ASSIGN(auto range_read,
                         ScanAndReadResult(table_path, {"b0", "_ROW_ID", "_SEQUENCE_NUMBER"},
                                           /*predicate=*/nullptr,
                                           /*row_ranges=*/{Range(1, 2), Range(8, 8)}));
    ASSERT_TRUE(range_read.chunked_array);
    auto range_concat = arrow::Concatenate(range_read.chunked_array->chunks()).ValueOrDie();
    auto range_struct = std::dynamic_pointer_cast<arrow::StructArray>(range_concat);
    ASSERT_TRUE(range_struct);
    ASSERT_EQ(range_struct->length(), 3);
    auto range_blob_col =
        std::dynamic_pointer_cast<arrow::LargeBinaryArray>(range_struct->GetFieldByName("b0"));
    auto range_row_id_col = std::dynamic_pointer_cast<arrow::Int64Array>(
        range_struct->GetFieldByName(SpecialFields::RowId().Name()));
    auto range_seq_col = std::dynamic_pointer_cast<arrow::Int64Array>(
        range_struct->GetFieldByName(SpecialFields::SequenceNumber().Name()));
    ASSERT_TRUE(range_blob_col && range_row_id_col && range_seq_col);
    const std::vector<int64_t> expected_row_ids = {1, 2, 8};
    const std::vector<std::string> expected_blobs = {"b1", "u2", "b8"};
    for (int64_t i = 0; i < 3; i++) {
        ASSERT_EQ(range_blob_col->GetString(i), expected_blobs[i]) << "row " << i;
        ASSERT_EQ(range_row_id_col->Value(i), expected_row_ids[i]) << "row " << i;
        ASSERT_EQ(range_seq_col->Value(i), expected_row_ids[i] == 2 ? new_layer_seq : old_layer_seq)
            << "row " << i;
    }
}

/// A row that is a placeholder in every BLOB layer degrades to a null blob but keeps its
/// _ROW_ID and reports -1 as its _SEQUENCE_NUMBER. Only the blob column is ever written, so
/// the blob bunch itself carries the row-tracking fields.
TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateAllPlaceholderRowTracking) {
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 BlobUtils::ToArrowField("b0", /*nullable=*/true)};
    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                  {Options::FILE_FORMAT, GetParam()},
                                                  {Options::FILE_SYSTEM, "local"},
                                                  {Options::ROW_TRACKING_ENABLED, "true"},
                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    // the base layer itself holds a placeholder at row 1, so after the second layer also
    // leaves it untouched, row 1 is a placeholder in every layer
    auto base_array = MakeBlobUpdateArray(fields[1], {"blob_a", "PH", "blob_c"});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs0, WriteArray(table_path, {}, {"b0"}, {base_array}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs0);
    ASSERT_OK(Commit(table_path, commit_msgs0));

    auto update_array = MakeBlobUpdateArray(fields[1], {"PH", "PH", "update_c"});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
    ASSERT_OK(Commit(table_path, commit_msgs1));

    ASSERT_OK_AND_ASSIGN(auto scan_read,
                         ScanAndReadResult(table_path, {"b0", "_ROW_ID", "_SEQUENCE_NUMBER"}));
    ASSERT_TRUE(scan_read.chunked_array);
    auto concat_array = arrow::Concatenate(scan_read.chunked_array->chunks()).ValueOrDie();
    auto struct_array = std::dynamic_pointer_cast<arrow::StructArray>(concat_array);
    ASSERT_TRUE(struct_array);
    ASSERT_EQ(struct_array->length(), 3);
    auto blob_col =
        std::dynamic_pointer_cast<arrow::LargeBinaryArray>(struct_array->GetFieldByName("b0"));
    auto row_id_col = std::dynamic_pointer_cast<arrow::Int64Array>(
        struct_array->GetFieldByName(SpecialFields::RowId().Name()));
    auto seq_col = std::dynamic_pointer_cast<arrow::Int64Array>(
        struct_array->GetFieldByName(SpecialFields::SequenceNumber().Name()));
    ASSERT_TRUE(blob_col && row_id_col && seq_col);

    ASSERT_EQ(blob_col->GetString(0), "blob_a");
    ASSERT_EQ(blob_col->GetString(2), "update_c");
    // the all-placeholder row: null blob, row id kept, sequence number -1
    ASSERT_TRUE(blob_col->IsNull(1));
    ASSERT_EQ(row_id_col->Value(1), 1);
    ASSERT_EQ(seq_col->Value(1), -1);
    for (int64_t i : {static_cast<int64_t>(0), static_cast<int64_t>(2)}) {
        ASSERT_EQ(row_id_col->Value(i), i);
        ASSERT_GE(seq_col->Value(i), 0);
    }
    ASSERT_GT(seq_col->Value(2), seq_col->Value(0));
}

/// A normal (full-row) write never interprets blob bytes: a user value whose bytes exactly
/// equal the placeholder sentinel must be stored verbatim (not persisted as a bin_length -2
/// entry) and read back unchanged. The sentinel is reserved only inside the data-evolution
/// partial-update channels (see BlobDefs::kPlaceholderSentinel).
TEST_P(BlobTableInteTest, TestBlobValueEqualToPlaceholderSentinelBytes) {
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 arrow::field("f1", arrow::utf8()),
                                 BlobUtils::ToArrowField("b0", /*nullable=*/true)};
    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                  {Options::FILE_FORMAT, GetParam()},
                                                  {Options::FILE_SYSTEM, "local"},
                                                  {Options::ROW_TRACKING_ENABLED, "true"},
                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields);
    std::string sentinel_bytes = std::string(BlobDefs::PlaceholderSentinelView());

    // full-row write: row 0's blob is exactly the sentinel bytes
    auto struct_type = arrow::struct_(fields);
    arrow::StructBuilder struct_builder(
        struct_type, arrow::default_memory_pool(),
        {std::make_shared<arrow::Int32Builder>(), std::make_shared<arrow::StringBuilder>(),
         std::make_shared<arrow::LargeBinaryBuilder>()});
    auto f0_builder = checked_cast<arrow::Int32Builder*>(struct_builder.field_builder(0));
    auto f1_builder = checked_cast<arrow::StringBuilder*>(struct_builder.field_builder(1));
    auto b0_builder = checked_cast<arrow::LargeBinaryBuilder*>(struct_builder.field_builder(2));
    ASSERT_TRUE(struct_builder.Append().ok());
    ASSERT_TRUE(f0_builder->Append(1).ok());
    ASSERT_TRUE(f1_builder->Append("a").ok());
    ASSERT_TRUE(b0_builder->Append(sentinel_bytes.data(), sentinel_bytes.size()).ok());
    ASSERT_TRUE(struct_builder.Append().ok());
    ASSERT_TRUE(f0_builder->Append(2).ok());
    ASSERT_TRUE(f1_builder->Append("b").ok());
    ASSERT_TRUE(b0_builder->Append("normal", 6).ok());
    std::shared_ptr<arrow::Array> src_array;
    ASSERT_TRUE(struct_builder.Finish(&src_array).ok());
    auto src_struct = std::dynamic_pointer_cast<arrow::StructArray>(src_array);
    ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
                         WriteArray(table_path, {}, schema->field_names(), {src_struct}));
    ASSERT_OK(Commit(table_path, commit_msgs0));

    // read back unchanged: the sentinel-equal bytes are a normal value
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), src_struct));

    // a value that merely starts with the sentinel bytes stays a normal value even through a
    // partial-update fallback (placeholders are identified by exact equality only)
    std::string prefixed_bytes = sentinel_bytes + "suffix";
    auto update_array = MakeBlobUpdateArray(fields[2], {prefixed_bytes, "updated_b"});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
    ASSERT_OK(Commit(table_path, commit_msgs1));

    ASSERT_OK_AND_ASSIGN(auto scan_read, ScanAndReadResult(table_path, schema->field_names()));
    ASSERT_TRUE(scan_read.chunked_array);
    auto concat_array = arrow::Concatenate(scan_read.chunked_array->chunks()).ValueOrDie();
    auto struct_result = std::dynamic_pointer_cast<arrow::StructArray>(concat_array);
    ASSERT_TRUE(struct_result);
    ASSERT_EQ(struct_result->length(), 2);
    auto blob_col =
        std::dynamic_pointer_cast<arrow::LargeBinaryArray>(struct_result->GetFieldByName("b0"));
    ASSERT_TRUE(blob_col);
    ASSERT_FALSE(blob_col->IsNull(0));
    ASSERT_EQ(blob_col->GetString(0), prefixed_bytes);
    ASSERT_EQ(blob_col->GetString(1), "updated_b");
}

TEST_P(BlobTableInteTest, TestBlobSentinelValueInBaseLayerDegradesToNull) {
    // Pins the accepted collision of the byte-identified placeholder protocol (see
    // BlobDefs::kPlaceholderSentinel): the fallback merge byte-compares every layer, so a user
    // blob equal to the sentinel bytes that a later partial update leaves untouched reads as a
    // placeholder in every layer and degrades to a null blob instead of falling back.
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 arrow::field("f1", arrow::utf8()),
                                 BlobUtils::ToArrowField("b0", /*nullable=*/true)};
    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                  {Options::FILE_FORMAT, GetParam()},
                                                  {Options::FILE_SYSTEM, "local"},
                                                  {Options::ROW_TRACKING_ENABLED, "true"},
                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields);
    std::string sentinel_bytes = std::string(BlobDefs::PlaceholderSentinelView());

    // the full-row write stores row 0's sentinel-equal bytes verbatim (the write channel is off)
    std::string src_json =
        std::string(R"([[1, "a", ")") + sentinel_bytes + R"("], [2, "b", "blob_b"]])";
    auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), src_json).ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
                         WriteArray(table_path, {}, schema->field_names(), {src_array}));
    ASSERT_OK(Commit(table_path, commit_msgs0));

    // the partial update leaves row 0 untouched, so its base value collides with the
    // placeholder markers of the newer layer
    auto update_array = MakeBlobUpdateArray(fields[2], {"PH", "updated_b"});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
    ASSERT_OK(Commit(table_path, commit_msgs1));

    // row 0's non-blob fields survive (they come from the data file); only the blob degrades
    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
        [1, "a", null],
        [2, "b", "updated_b"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
}

TEST_P(BlobTableInteTest, TestUserSuppliedInternalPlaceholderOptionsIgnored) {
    // blob.internal.* options are reserved for the data-evolution write and read paths. Set in
    // the user table options they must be ignored: were the write key honored, this full-row
    // write would persist the sentinel-equal user value as a placeholder entry that no older
    // layer can resolve, making the table unreadable.
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 arrow::field("f1", arrow::utf8()),
                                 BlobUtils::ToArrowField("b0", /*nullable=*/true)};
    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                  {Options::FILE_FORMAT, GetParam()},
                                                  {Options::FILE_SYSTEM, "local"},
                                                  {Options::ROW_TRACKING_ENABLED, "true"},
                                                  {Options::DATA_EVOLUTION_ENABLED, "true"},
                                                  {BlobDefs::kWritePlaceholderKey, "true"},
                                                  {BlobDefs::kEmitPlaceholderSentinelKey, "true"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields);
    std::string sentinel_bytes = std::string(BlobDefs::PlaceholderSentinelView());

    std::string src_json =
        std::string(R"([[1, "a", ")") + sentinel_bytes + R"("], [2, "b", "blob_b"]])";
    auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), src_json).ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(table_path, {}, schema->field_names(), {src_array}));
    ASSERT_OK(Commit(table_path, commit_msgs));

    // the sentinel-equal value round-trips verbatim: the user-supplied keys were stripped
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), src_array));
}

TEST_P(BlobTableInteTest, TestDataEvolutionBlobOnlyFirstCommitFailsWithoutFirstRowId) {
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0")};
    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                  {Options::FILE_FORMAT, GetParam()},
                                                  {Options::FILE_SYSTEM, "local"},
                                                  {Options::ROW_TRACKING_ENABLED, "true"},
                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    auto blob_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[2]}), R"([
        ["blob_0"]
    ])")
            .ValueOrDie());
    std::vector<std::string> blob_write_cols = {"b0"};
    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(table_path, {}, blob_write_cols, {blob_array}));

    ASSERT_EQ(commit_msgs.size(), 1);
    auto commit_msg = std::dynamic_pointer_cast<CommitMessageImpl>(commit_msgs[0]);
    ASSERT_TRUE(commit_msg);
    ASSERT_EQ(commit_msg->data_increment_.new_files_.size(), 1);
    const auto& blob_file = commit_msg->data_increment_.new_files_[0];
    ASSERT_TRUE(BlobUtils::IsBlobFile(blob_file->file_name));
    ASSERT_FALSE(blob_file->first_row_id.has_value());

    ASSERT_NOK_WITH_MSG(Commit(table_path, commit_msgs), "blobStart 0 should be less than start 0");
}

TEST_P(BlobTableInteTest, TestMultipleAppendsDifferentFirstRowIds) {
    CreateTable();
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    // First commit, firstRowId = 0
    // write field: f0, f1
    std::vector<std::string> write_cols1 = {"f0", "f1"};
    auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
        [1, "a"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, write_cols1, {src_array1}));
    SetFirstRowId(0, commit_msgs1);

    // write field: f2
    std::vector<std::string> write_cols2 = {"f2"};
    auto src_array2 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
        ["b"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs2, WriteArray(table_path, {}, write_cols2, {src_array2}));
    SetFirstRowId(0, commit_msgs2);

    std::vector<std::shared_ptr<CommitMessage>> total_msgs;
    total_msgs.insert(total_msgs.end(), commit_msgs1.begin(), commit_msgs1.end());
    total_msgs.insert(total_msgs.end(), commit_msgs2.begin(), commit_msgs2.end());
    ASSERT_OK(Commit(table_path, total_msgs));

    // Second commit, firstRowId = 1
    // write field: f0, f1
    std::vector<std::string> write_cols3 = {"f0", "f1"};
    auto src_array3 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
        [2, "c"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs3, WriteArray(table_path, {}, write_cols3, {src_array3}));
    SetFirstRowId(1, commit_msgs3);
    ASSERT_OK(Commit(table_path, commit_msgs3));

    // Third commit
    // write field: f2
    std::vector<std::string> write_cols4 = {"f2"};
    auto src_array4 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
        ["d"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs4, WriteArray(table_path, {}, write_cols4, {src_array4}));
    SetFirstRowId(1, commit_msgs4);
    ASSERT_OK(Commit(table_path, commit_msgs4));

    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "a", "b"],
        [2, "c", "d"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));

    // read with row tracking
    auto expected_row_tracking_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({
                                                      fields_[0],
                                                      fields_[1],
                                                      fields_[2],
                                                      SpecialFields::RowId().field_,
                                                      SpecialFields::SequenceNumber().field_,
                                                  }),
                                                  R"([
        [1, "a", "b", 0, 1],
        [2, "c", "d", 1, 3]
    ])")
            .ValueOrDie());

    ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2", "_ROW_ID", "_SEQUENCE_NUMBER"},
                          expected_row_tracking_array));
}

TEST_P(BlobTableInteTest, TestMoreDataWithDataEvolution) {
    CreateTable();
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    // write field: f0, f1
    std::vector<std::string> write_cols1 = {"f0", "f1"};
    // row0: 0, a0; row1: 1, a1 ...
    auto src_array1 = PrepareBulkData(
        10000, [](int32_t i) { return std::to_string(i) + ", \"a" + std::to_string(i) + "\""; },
        {fields_[0], fields_[1]});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, write_cols1, {src_array1}));
    SetFirstRowId(0, commit_msgs1);

    // write field: f2
    std::vector<std::string> write_cols2 = {"f2"};
    // row0: b0; row1: b1 ...
    auto src_array2 = PrepareBulkData(
        10000, [](int32_t i) { return "\"b" + std::to_string(i) + "\""; }, {fields_[2]});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs2, WriteArray(table_path, {}, write_cols2, {src_array2}));
    SetFirstRowId(0, commit_msgs2);

    std::vector<std::shared_ptr<CommitMessage>> total_msgs;
    total_msgs.insert(total_msgs.end(), commit_msgs1.begin(), commit_msgs1.end());
    total_msgs.insert(total_msgs.end(), commit_msgs2.begin(), commit_msgs2.end());
    ASSERT_OK(Commit(table_path, total_msgs));

    // write field: f2
    std::vector<std::string> write_cols3 = {"f2"};
    // row0: c0; row1: c1 ...
    auto src_array3 = PrepareBulkData(
        10000, [](int32_t i) { return "\"c" + std::to_string(i) + "\""; }, {fields_[2]});
    ASSERT_OK_AND_ASSIGN(auto commit_msgs3, WriteArray(table_path, {}, write_cols3, {src_array3}));
    ASSERT_OK(Commit(table_path, commit_msgs3));

    // row0: 0, a0, c0; row1: 1, a1, c1 ...
    auto expected_array = PrepareBulkData(10000,
                                          [](int32_t i) {
                                              return std::to_string(i) + ", \"a" +
                                                     std::to_string(i) + "\", \"c" +
                                                     std::to_string(i) + "\"";
                                          },
                                          {fields_[0], fields_[1], fields_[2]});
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
}

TEST_P(BlobTableInteTest, TestBlobWriteMultiRound) {
    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},        {Options::FILE_FORMAT, GetParam()},
        {Options::FILE_SYSTEM, "local"},          {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::BLOB_TARGET_FILE_SIZE, "1000"}, {Options::TARGET_FILE_SIZE, "100"},
        {Options::DATA_EVOLUTION_ENABLED, "true"}};
    CreateTable(/*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    // write field: f0, f1, f2
    std::vector<std::string> write_cols = schema->field_names();
    int32_t offset = 0;
    auto array_1 = PrepareBulkData(5000,
                                   [offset](int32_t i) {
                                       return std::to_string(offset + i) + ", \"a" +
                                              std::to_string(offset + i) + "\", \"c" +
                                              std::to_string(offset + i) + "\"";
                                   },
                                   {fields_[0], fields_[1], fields_[2]});

    offset = 5000;
    auto array_2 = PrepareBulkData(5000,
                                   [offset](int32_t i) {
                                       return std::to_string(offset + i) + ", \"a" +
                                              std::to_string(offset + i) + "\", \"c" +
                                              std::to_string(offset + i) + "\"";
                                   },
                                   {fields_[0], fields_[1], fields_[2]});

    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(table_path, {}, write_cols, {array_1, array_2}));
    ASSERT_OK(Commit(table_path, commit_msgs));
    auto concat_array = arrow::Concatenate({array_1, array_2}).ValueOrDie();
    auto expect_array = std::dynamic_pointer_cast<arrow::StructArray>(concat_array);
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expect_array));
}

TEST_P(BlobTableInteTest, TestExternalPath) {
    // create external path dir
    auto external_dir = UniqueTestDirectory::Create("local");
    ASSERT_TRUE(external_dir);
    std::string external_test_dir = external_dir->Str();

    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, GetParam()},
        {Options::FILE_SYSTEM, "local"},
        {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "true"},
        {Options::DATA_FILE_EXTERNAL_PATHS, "FILE://" + external_test_dir}};
    CreateTable(/*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    // write field: f0, f1
    std::vector<std::string> write_cols0 = {"f0", "f1"};
    auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
        [1, "a"],
        [2, "c"]
    ])")
            .ValueOrDie());

    ASSERT_OK_AND_ASSIGN(auto commit_msgs0, WriteArray(table_path, {}, write_cols0, {src_array0}));
    ASSERT_OK(Commit(table_path, commit_msgs0));

    // write field: f0, f2
    std::vector<std::string> write_cols1 = {"f0", "f2"};
    auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[2]}), R"([
        [10, "b"],
        [20, "d"]
    ])")
            .ValueOrDie());

    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, write_cols1, {src_array1}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
    ASSERT_OK(Commit(table_path, commit_msgs1));

    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [10, "a", "b"],
        [20, "c", "d"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));

    // read with row tracking
    auto expected_row_tracking_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(
            arrow::struct_({fields_[1], fields_[0], fields_[2], SpecialFields::RowId().field_,
                            SpecialFields::SequenceNumber().field_}),
            R"([
        ["a", 10, "b", 0, 2],
        ["c", 20, "d", 1, 2]
    ])")
            .ValueOrDie());

    ASSERT_OK(ScanAndRead(table_path, {"f1", "f0", "f2", "_ROW_ID", "_SEQUENCE_NUMBER"},
                          expected_row_tracking_array));
}

TEST_P(BlobTableInteTest, TestPartitionWithPredicate) {
    auto file_format = GetParam();
    std::vector<std::string> partition_keys = {"f0"};
    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},         {Options::FILE_FORMAT, GetParam()},
        {Options::FILE_SYSTEM, "local"},           {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "true"}, {"parquet.write.max-row-group-length", "1"}};

    CreateTable(partition_keys, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    // write field: f0, f1 for partition f0 = "11"
    std::vector<std::string> write_cols0 = {"f0", "f1"};
    auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
        [11, "2024"],
        [11, "2025"],
        [11, "2026"]
    ])")
            .ValueOrDie());

    ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
                         WriteArray(table_path, {{"f0", "11"}}, write_cols0, {src_array0}));

    // write field: f2 for partition f0 = "11"
    std::vector<std::string> write_cols1 = {"f2"};
    auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
        ["a"],
        ["b"],
        ["c"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1,
                         WriteArray(table_path, {{"f0", "11"}}, write_cols1, {src_array1}));
    std::vector<std::shared_ptr<CommitMessage>> total_msgs;
    total_msgs.insert(total_msgs.end(), commit_msgs0.begin(), commit_msgs0.end());
    total_msgs.insert(total_msgs.end(), commit_msgs1.begin(), commit_msgs1.end());
    SetFirstRowId(0, total_msgs);
    ASSERT_OK(Commit(table_path, total_msgs));

    // write field: f0, f1 for partition f0 = "22"
    std::vector<std::string> write_cols3 = {"f0", "f1"};
    auto src_array3 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
        [22, "2027"],
        [22, "2028"],
        [22, "2029"]
    ])")
            .ValueOrDie());

    ASSERT_OK_AND_ASSIGN(auto commit_msgs3,
                         WriteArray(table_path, {{"f0", "22"}}, write_cols3, {src_array3}));
    SetFirstRowId(3, commit_msgs3);
    ASSERT_OK(Commit(table_path, commit_msgs3));
    {
        // only set data field predicate, file with field all null will be returned
        auto equal = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2",
                                             FieldType::STRING, Literal(FieldType::STRING, "a", 1));
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [11, "2024", "a"],
        [11, "2025", "b"],
        [11, "2026", "c"],
        [22, "2027", null],
        [22, "2028", null],
        [22, "2029", null]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array, equal));
    }
    {
        // only set partition predicate
        auto equal = PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::INT,
                                             Literal(11));
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [11, "2024", "a"],
        [11, "2025", "b"],
        [11, "2026", "c"]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array, equal));
    }
    {
        // set partition predicate and data field predicate, blob type not support predicate
        auto equal =
            PredicateBuilder::Equal(/*field_index=*/1, /*field_name=*/"f1", FieldType::STRING,
                                    Literal(FieldType::BLOB, "2024", 4));
        auto greater_than = PredicateBuilder::GreaterThan(/*field_index=*/0, /*field_name=*/"f0",
                                                          FieldType::INT, Literal(100));
        ASSERT_OK_AND_ASSIGN(auto predicate, PredicateBuilder::And({equal, greater_than}));
        ASSERT_NOK_WITH_MSG(
            ScanAndRead(table_path, schema->field_names(), /*expected_array=*/nullptr, predicate),
            "Invalid type large_binary for predicate");
    }
    {
        // read with row tracking
        auto equal = PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::INT,
                                             Literal(11));

        auto expected_row_tracking_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(
                arrow::struct_({fields_[0], fields_[1], fields_[2], SpecialFields::RowId().field_,
                                SpecialFields::SequenceNumber().field_}),
                R"([
        [11, "2024", "a", 0, 1],
        [11, "2025", "b", 1, 1],
        [11, "2026", "c", 2, 1]
    ])")
                .ValueOrDie());

        ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2", "_ROW_ID", "_SEQUENCE_NUMBER"},
                              expected_row_tracking_array, equal));
    }
}

TEST_P(BlobTableInteTest, TestPredicate) {
    if (GetParam() == "avro") {
        // Avro does not have stats.
        return;
    }
    CreateTable();
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    // write field: f0, f1, f2
    std::vector<std::string> write_cols0 = schema->field_names();
    auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "a", "b"]
    ])")
            .ValueOrDie());

    ASSERT_OK_AND_ASSIGN(auto commit_msgs, WriteArray(table_path, {}, write_cols0, {src_array0}));
    ASSERT_OK(Commit(table_path, commit_msgs));

    // write field: f2
    std::vector<std::string> write_cols1 = {"f2"};
    auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
        ["c"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(commit_msgs, WriteArray(table_path, {}, write_cols1, {src_array1}));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs);
    ASSERT_OK(Commit(table_path, commit_msgs));
    {
        // test no predicate
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "a", "c"]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
    }
    {
        // test predicate with f2
        auto predicate =
            PredicateBuilder::NotEqual(/*field_index=*/2, /*field_name=*/"f2", FieldType::STRING,
                                       Literal(FieldType::STRING, "b", 1));
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "a", "c"]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array, predicate));
    }
    {
        // test predicate with f1
        auto predicate =
            PredicateBuilder::NotEqual(/*field_index=*/1, /*field_name=*/"f1", FieldType::STRING,
                                       Literal(FieldType::STRING, "a", 1));
        ASSERT_NOK_WITH_MSG(
            ScanAndRead(table_path, schema->field_names(), /*expected_array=*/nullptr, predicate),
            "Invalid type large_binary for predicate");
    }
    {
        // test predicate with f2
        auto predicate =
            PredicateBuilder::NotEqual(/*field_index=*/2, /*field_name=*/"f2", FieldType::STRING,
                                       Literal(FieldType::STRING, "c", 1));
        ASSERT_OK(
            ScanAndRead(table_path, schema->field_names(), /*expected_array=*/nullptr, predicate));
    }
}

TEST_P(BlobTableInteTest, TestIOException) {
    std::string table_path;
    // write and commit with I/O exception
    bool write_run_complete = false;
    auto io_hook = IOHook::GetInstance();
    for (size_t i = 0; i < 2000; i += paimon::test::RandomNumber(20, 30)) {
        ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
        dir_ = UniqueTestDirectory::Create("local");
        CreateTable();
        table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
        auto schema = arrow::schema(fields_);

        io_hook->Reset(i, IOHook::Mode::RETURN_ERROR);
        // write field: f0, f1, f2
        std::vector<std::string> write_cols0 = schema->field_names();
        auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [10, "a", "b"],
        [20, "aa", "bb"],
        [23, "aaa", "bbb"]
    ])")
                .ValueOrDie());
        auto commit_msgs0_result = WriteArray(table_path, {}, write_cols0, {src_array0});
        CHECK_HOOK_STATUS(commit_msgs0_result.status(), i);
        CHECK_HOOK_STATUS(Commit(table_path, commit_msgs0_result.value()), i);

        // write field: f2, f0
        std::vector<std::string> write_cols1 = {"f2", "f0"};
        auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2], fields_[0]}), R"([
            ["c", 100],
            ["cc", 200],
            ["ccc", 300]
        ])")
                .ValueOrDie());
        auto commit_msgs1_result = WriteArray(table_path, {}, write_cols1, {src_array1});
        CHECK_HOOK_STATUS(commit_msgs1_result.status(), i);
        auto commit_msgs1 = std::move(commit_msgs1_result).value();
        SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
        CHECK_HOOK_STATUS(Commit(table_path, commit_msgs1), i);
        write_run_complete = true;
        break;
    }
    ASSERT_TRUE(write_run_complete);

    // scan and read with I / O exception
    bool read_run_complete = false;
    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(
            arrow::struct_({fields_[1], fields_[0], SpecialFields::SequenceNumber().field_,
                            SpecialFields::RowId().field_, fields_[2]}),
            R"([
        ["a", 100, 2, 0, "c"],
        ["aa", 200, 2, 1, "cc"],
        ["aaa", 300, 2, 2, "ccc"]
    ])")
            .ValueOrDie());

    for (size_t i = 0; i < 2000; i++) {
        ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
        io_hook->Reset(i, IOHook::Mode::RETURN_ERROR);
        CHECK_HOOK_STATUS(ScanAndRead(table_path, {"f1", "f0", "_SEQUENCE_NUMBER", "_ROW_ID", "f2"},
                                      expected_array),
                          i);
        read_run_complete = true;
        break;
    }
    ASSERT_TRUE(read_run_complete);
}

TEST_P(BlobTableInteTest, TestReadTableWithDenseStats) {
    auto file_format = GetParam();
    if (file_format == "avro") {
        return;
    }
    std::string table_path =
        paimon::test::GetDataDir() + file_format +
        "/blob_data_evolution_with_dense_stats.db/blob_data_evolution_with_dense_stats/";
    std::vector<DataField> read_fields = {DataField(0, BlobUtils::ToArrowField("f0")),
                                          DataField(1, arrow::field("f1", arrow::int32())),
                                          DataField(2, arrow::field("f2", arrow::int32())),
                                          DataField(3, arrow::field("f3", arrow::float64())),
                                          SpecialFields::RowId(),
                                          SpecialFields::SequenceNumber()};

    std::shared_ptr<arrow::DataType> arrow_data_type =
        DataField::ConvertDataFieldsToArrowStructType(read_fields);
    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
        ["Lily", 2, 102, 2.1, 0, 2],
        ["Alice", 4, 104, 3.1, 1, 2]
    ])")
            .ValueOrDie());

    {
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array));
    }
    {
        auto predicate =
            PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::BLOB,
                                    Literal(FieldType::BLOB, "Alice", 5));
        ASSERT_NOK_WITH_MSG(
            ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                        expected_array, predicate),
            "Invalid type large_binary for predicate");
    }
    {
        auto predicate = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2",
                                                 FieldType::INT, Literal(102));
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array, predicate));
    }
    {
        auto predicate = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2",
                                                 FieldType::INT, Literal(12));
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              /*expected_array=*/nullptr, predicate));
    }
    {
        auto predicate = PredicateBuilder::GreaterThan(/*field_index=*/1, /*field_name=*/"f1",
                                                       FieldType::INT, Literal(6));
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              /*expected_array=*/nullptr, predicate));
    }
    {
        // f3 does not have stats, therefore data will not be filtered
        auto predicate = PredicateBuilder::GreaterThan(/*field_index=*/3, /*field_name=*/"f3",
                                                       FieldType::DOUBLE, Literal(5.1));
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array, predicate));
    }
}

TEST_P(BlobTableInteTest, TestDataEvolutionAndAlterTable) {
    auto file_format = GetParam();
    if (file_format == "avro") {
        return;
    }
    std::string table_path = paimon::test::GetDataDir() + file_format +
                             "/blob_append_table_alter_table_with_cast_with_data_evolution.db/"
                             "blob_append_table_alter_table_with_cast_with_data_evolution/";
    std::vector<DataField> read_fields = {
        DataField(6, arrow::field("f4", arrow::timestamp(arrow::TimeUnit::NANO))),
        DataField(0, arrow::field("key0", arrow::int32())),
        DataField(1, arrow::field("key1", arrow::int32())),
        DataField(2, arrow::field("f3", arrow::int32())),
        DataField(3, arrow::field("f1", arrow::utf8())),
        DataField(4, arrow::field("f2", arrow::decimal128(6, 3))),
        DataField(5, arrow::field("f0", arrow::boolean())),
        DataField(8, BlobUtils::ToArrowField("blob")),
        DataField(9, arrow::field("f6", arrow::int32())),
        SpecialFields::RowId(),
        SpecialFields::SequenceNumber()};

    std::shared_ptr<arrow::DataType> arrow_data_type =
        DataField::ConvertDataFieldsToArrowStructType(read_fields);

    {
        // only read blob column
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(
                arrow::struct_({BlobUtils::ToArrowField("blob")}), R"([
            ["Lily"],
            ["Alice"],
            ["Bob"],
            ["Cindy"],
            ["Dave"],
            ["Apple"],
            ["Banana"],
            ["Cherry"],
            ["Durian"],
            ["Elderberry"]
        ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, {"blob"}, expected_array));
    }
    {
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
["1970-01-05T00:00", 0, 1, 100, "2024-11-26 06:38:56.001000001", "0.020", true, "Lily", null, 0, 1],
["1969-11-18T00:00", 0, 1, 110, "2024-11-26 06:38:56.011000011", "11.120", true, "Alice", null, 1, 1],
["1971-03-21T00:00", 0, 1, 120, "2024-11-26 06:38:56.021000021", "22.220", false, "Bob", null, 2, 1],
["1957-11-01T00:00", 0, 1, 130, "2024-11-26 06:38:56.031000031", "333.320", false, "Cindy", null, 3, 1],
["2091-09-07T00:00", 0, 1, 140, "2024-11-26 06:38:56.041000041", "444.420", true, "Dave", null, 4, 1],
["2024-11-26T06:38:56.054000154", 0, 1, 150, "2024-11-26 15:28:31", "55.002", true, "Apple", 56, 5, 3],
["2024-11-26T06:38:56.064000164", 0, 1, 160, "2024-11-26 15:28:41", "666.012", false, "Banana", 66, 6, 3],
["2024-11-26T06:38:56.074000174", 0, 1, 170, "2024-11-26 15:28:51", "-77.022", true, "Cherry", 76, 7, 3],
["2024-11-26T06:38:56.084000184", 0, 1, 180, "2024-11-26 15:29:01", "8.032", true, "Durian", -86, 8, 3],
["2024-11-26T06:38:56.094000194", 0, 1, 190, "I'm strange", "-999.420", false, "Elderberry", 96, 9, 3]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array));
    }
    {
        // only files with schema-1 will be skipped, while files with schema-0 will be reserved
        // as type change
        auto predicate = PredicateBuilder::GreaterThan(/*field_index=*/3, /*field_name=*/"f3",
                                                       FieldType::INT, Literal(200));
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
["1970-01-05T00:00", 0, 1, 100, "2024-11-26 06:38:56.001000001", "0.020", true, "Lily", null, 0, 1],
["1969-11-18T00:00", 0, 1, 110, "2024-11-26 06:38:56.011000011", "11.120", true, "Alice", null, 1, 1],
["1971-03-21T00:00", 0, 1, 120, "2024-11-26 06:38:56.021000021", "22.220", false, "Bob", null, 2, 1],
["1957-11-01T00:00", 0, 1, 130, "2024-11-26 06:38:56.031000031", "333.320", false, "Cindy", null, 3, 1],
["2091-09-07T00:00", 0, 1, 140, "2024-11-26 06:38:56.041000041", "444.420", true, "Dave", null, 4, 1]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array, predicate));
    }
    {
        // files with schema-0 will be reserved as f6 does not exist in schema-0 (null count =
        // null), files with schema-1 will also be reserved
        auto predicate =
            PredicateBuilder::IsNotNull(/*field_index=*/8, /*field_name=*/"f6", FieldType::INT);
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
["1970-01-05T00:00", 0, 1, 100, "2024-11-26 06:38:56.001000001", "0.020", true, "Lily", null, 0, 1],
["1969-11-18T00:00", 0, 1, 110, "2024-11-26 06:38:56.011000011", "11.120", true, "Alice", null, 1, 1],
["1971-03-21T00:00", 0, 1, 120, "2024-11-26 06:38:56.021000021", "22.220", false, "Bob", null, 2, 1],
["1957-11-01T00:00", 0, 1, 130, "2024-11-26 06:38:56.031000031", "333.320", false, "Cindy", null, 3, 1],
["2091-09-07T00:00", 0, 1, 140, "2024-11-26 06:38:56.041000041", "444.420", true, "Dave", null, 4, 1],
["2024-11-26T06:38:56.054000154", 0, 1, 150, "2024-11-26 15:28:31", "55.002", true, "Apple", 56, 5, 3],
["2024-11-26T06:38:56.064000164", 0, 1, 160, "2024-11-26 15:28:41", "666.012", false, "Banana", 66, 6, 3],
["2024-11-26T06:38:56.074000174", 0, 1, 170, "2024-11-26 15:28:51", "-77.022", true, "Cherry", 76, 7, 3],
["2024-11-26T06:38:56.084000184", 0, 1, 180, "2024-11-26 15:29:01", "8.032", true, "Durian", -86, 8, 3],
["2024-11-26T06:38:56.094000194", 0, 1, 190, "I'm strange", "-999.420", false, "Elderberry", 96, 9, 3]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array, predicate));
    }
}

TEST_P(BlobTableInteTest, TestWithRowIdsSimple) {
    CreateTable();
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    std::vector<std::string> write_cols = {"f0", "f1", "f2"};
    // row0: 0, "a0", "b0"; row1: 1, "a1", "b1" ...
    auto src_array = PrepareBulkData(
        1000,
        [](int32_t i) {
            return std::to_string(i) + ", \"a" + std::to_string(i) + "\", \"b" + std::to_string(i) +
                   "\"";
        },
        fields_);
    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(table_path, /*partition=*/{}, write_cols, {src_array}));
    ASSERT_OK(Commit(table_path, commit_msgs));

    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [10, "a10", "b10"],
        [999, "a999", "b999"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
                          /*predicate=*/nullptr,
                          /*row_ranges=*/{Range(999l, 999l), Range(10l, 10l)}));
}

TEST_P(BlobTableInteTest, TestWithRowIdsForMultipleBlobFiles) {
    auto file_format = GetParam();
    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                  {Options::FILE_FORMAT, file_format},
                                                  {Options::TARGET_FILE_SIZE, "1000"},
                                                  {Options::BLOB_TARGET_FILE_SIZE, "80"},
                                                  {Options::BUCKET, "-1"},
                                                  {Options::ROW_TRACKING_ENABLED, "true"},
                                                  {Options::DATA_EVOLUTION_ENABLED, "true"},
                                                  {Options::FILE_SYSTEM, "local"}};
    CreateTable(/*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    // target blob size is 80, therefore, each 4 rows in blob will be in one file
    std::vector<std::string> write_cols = {"f0", "f1", "f2"};
    auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [0, "aaa0", "b0"],
        [1, "aaa1", "b1"],
        [2, "aaa2", "b2"],
        [3, "aaa3", "b3"],
        [4, "aaa4", "b4"],
        [5, "aaa5", "b5"],
        [6, "aaa6", "b6"],
        [7, "aaa7", "b7"],
        [8, "aaa8", "b8"],
        [9, "aaa9", "b9"]
    ])")
            .ValueOrDie());

    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(table_path, /*partition=*/{}, write_cols, {src_array}));
    ASSERT_OK(Commit(table_path, commit_msgs));

    {
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "aaa1", "b1"],
        [8, "aaa8", "b8"]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
                              /*predicate=*/nullptr,
                              /*row_ranges=*/{Range(1l, 1l), Range(8l, 8l)}));
    }
    {
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "aaa1", "b1"],
        [5, "aaa5", "b5"]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
                              /*predicate=*/nullptr,
                              /*row_ranges=*/{Range(1l, 1l), Range(5l, 5l)}));
    }
    {
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [5, "aaa5", "b5"],
        [6, "aaa6", "b6"],
        [8, "aaa8", "b8"]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
                              /*predicate=*/nullptr,
                              /*row_ranges=*/{Range(5l, 6l), Range(8l, 8l)}));
    }
    {
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [5, "aaa5", "b5"],
        [7, "aaa7", "b7"]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
                              /*predicate=*/nullptr,
                              /*row_ranges=*/{Range(5l, 5l), Range(7l, 7l)}));
    }
    {
        // f0 0-9 in one file, predicate does not skip any data file
        auto predicate = PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0",
                                                 FieldType::INT, Literal(5));
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "aaa1", "b1"],
        [5, "aaa5", "b5"]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array, predicate,
                              /*row_ranges=*/{Range(1l, 1l), Range(5l, 5l)}));
    }
    {
        // test not read blob field
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1]}), R"([
        ["aaa1"],
        ["aaa8"]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, {"f1"}, expected_array,
                              /*predicate=*/nullptr,
                              /*row_ranges=*/{Range(1l, 1l), Range(8l, 8l)}));
    }
}

TEST_P(BlobTableInteTest, TestAppendTableWriteWithMultipleBlobFields) {
    arrow::FieldVector fields = {
        arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()),
        BlobUtils::ToArrowField("blob1", true), BlobUtils::ToArrowField("blob2", true)};

    std::map<std::string, std::string> 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::FILE_SYSTEM, "local"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    std::string data_json = R"([
        ["str_0", null, "apple",  "red"],
        ["str_1", 1,    "banana", "yellow"],
        ["str_2", 2,    "cat",    "black"]
    ])";
    auto write_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), data_json).ValueOrDie());

    auto schema = arrow::schema(fields);
    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(table_path, {}, schema->field_names(), {write_array}));
    ASSERT_OK(Commit(table_path, commit_msgs));
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), write_array));
}

TEST_P(BlobTableInteTest, TestAppendWriteWithNullBlob) {
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 BlobUtils::ToArrowField("blob", true)};

    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                  {Options::FILE_FORMAT, GetParam()},
                                                  {Options::BUCKET, "-1"},
                                                  {Options::FILE_SYSTEM, "local"},
                                                  {Options::ROW_TRACKING_ENABLED, "true"},
                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    // Write: row 0 non-null blob, row 1 null blob, row 2 non-null blob
    std::string data_json = R"([
        [1, "hello"],
        [2, null],
        [3, "world"]
    ])";
    auto write_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), data_json).ValueOrDie());

    auto schema = arrow::schema(fields);
    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(table_path, {}, schema->field_names(), {write_array}));
    ASSERT_OK(Commit(table_path, commit_msgs));
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), write_array));
}

TEST_P(BlobTableInteTest, TestReadTableWithMultiBlobFields) {
    auto file_format = GetParam();
    if (file_format == "avro") {
        return;
    }
    std::string table_path = paimon::test::GetDataDir() + file_format +
                             "/append_table_with_multi_blob.db/append_table_with_multi_blob";
    std::vector<DataField> read_fields = {
        DataField(5, BlobUtils::ToArrowField("f5")),
        DataField(1, arrow::field("f1", arrow::int32())),
        DataField(2, arrow::field("f2", arrow::int32())),
        DataField(3, arrow::field("f3", arrow::float64())),
        DataField(4, arrow::field("f4", arrow::utf8())),
        SpecialFields::RowId(),
        DataField(6, BlobUtils::ToArrowField("f6")),
    };

    std::shared_ptr<arrow::DataType> arrow_data_type =
        DataField::ConvertDataFieldsToArrowStructType(read_fields);

    auto make_json_row = [&](int32_t i) -> std::string {
        std::string f5_json;
        if (i <= 1) {
            f5_json = "null";
        } else {
            f5_json = "\"" + std::string(1024, static_cast<char>('A' + i)) + "\"";
        }

        std::string f6_json;
        if (i == 0 || i == 2) {
            f6_json = "null";
        } else {
            f6_json = "\"" + std::string(2048, static_cast<char>('a' + i)) + "\"";
        }
        // f1=i, f2=i*10, f3=i+0.5, f4="desc_i", row_id=i
        return "[" + f5_json + ", " + std::to_string(i) + ", " + std::to_string(i * 10) + ", " +
               std::to_string(i + 0.5) + ", \"desc_" + std::to_string(i) + "\", " +
               std::to_string(i) + ", " + f6_json + "]";
    };

    auto build_expected =
        [&](const std::vector<int32_t>& row_indices) -> std::shared_ptr<arrow::StructArray> {
        std::string json_str = "[";
        for (size_t idx = 0; idx < row_indices.size(); idx++) {
            if (idx > 0) {
                json_str += ",";
            }
            json_str += make_json_row(row_indices[idx]);
        }
        json_str += "]";
        return std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, json_str).ValueOrDie());
    };

    // Full scan: all 10 rows
    {
        std::vector<int32_t> all_rows(10);
        std::iota(all_rows.begin(), all_rows.end(), 0);
        auto expected_array = build_expected(all_rows);
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array));
    }

    // Test with row_ranges: select row 0 (both null), row 2 (f5 non-null, f6 null),
    // row 5 (both non-null) to verify null handling under row range filtering.
    {
        auto expected_rr = build_expected({0, 2, 5});
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_rr, /*predicate=*/nullptr,
                              /*row_ranges=*/{Range(0l, 0l), Range(2l, 2l), Range(5l, 5l)}));
    }
}

TEST_P(BlobTableInteTest, TestBlobDescriptorField) {
    // Two blob fields configured via BLOB_DESCRIPTOR_FIELD and stored inline as descriptors.
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 BlobUtils::ToArrowField("b0", true),
                                 BlobUtils::ToArrowField("b1", true)};

    std::map<std::string, std::string> 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_DESCRIPTOR_FIELD, "b0,b1"}, {Options::FILE_SYSTEM, "local"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    // Input uses plain raw bytes for readability
    std::string raw_json = R"([
        [1, "image_data_0", "video_data_0"],
        [2, "image_data_1", "video_data_1"],
        [3, "image_data_2", "video_data_2"]
    ])";
    auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"b0", "b1"}));

    // write descriptor array
    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));

    // Scan and verify DataFileMeta: all blob fields are inline descriptors, so write_cols is unset.
    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
    VerifyDataFileMetas(plan, /*expected_file_count=*/1, /*expected_row_counts=*/{3},
                        /*expected_min_seqs=*/{1}, /*expected_max_seqs=*/{1},
                        /*expected_first_row_ids=*/{0},
                        /*expected_write_cols=*/{std::nullopt});

    // Read and resolve descriptors back to raw bytes
    std::map<std::string, std::string> read_options = {};
    ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), plan,
                                                /*predicate=*/nullptr, read_options));
    ASSERT_TRUE(result.chunked_array);
    auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie();
    auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
    ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"b0", "b1"}));
    ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(raw_array));
    ASSERT_TRUE(resolved->Equals(expected_with_rk));

    // Descriptor bytes should be unchanged (inline, not repacked)
    ASSERT_TRUE(read_struct->GetFieldByName("b0")->Equals(desc_array->GetFieldByName("b0")));
    ASSERT_TRUE(read_struct->GetFieldByName("b1")->Equals(desc_array->GetFieldByName("b1")));
}

TEST_P(BlobTableInteTest, TestBlobDescriptorFieldPartialInline) {
    // 4 blob fields: b0,b1 are inline descriptors; b2,b3 are regular blob fields written to
    // .blob files.
    arrow::FieldVector fields = {
        arrow::field("f0", arrow::int32()), BlobUtils::ToArrowField("b0", true),
        BlobUtils::ToArrowField("b1", true), BlobUtils::ToArrowField("b2", true),
        BlobUtils::ToArrowField("b3", true)};

    std::map<std::string, std::string> 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_DESCRIPTOR_FIELD, "b0,b1"}, {Options::FILE_SYSTEM, "local"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    // Input uses plain raw bytes:
    // b0: all non-null, b1: has nulls, b2: all non-null, b3: has nulls
    std::string raw_json = R"([
        [1, "img_0", null,    "raw_2_0", "raw_3_0"],
        [2, "img_1", "vid_1", "raw_2_1", null      ],
        [3, "img_2", null,    "raw_2_2", "raw_3_2" ]
    ])";
    auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto desc_array,
                         ConvertRawBlobToDescriptor(raw_array, {"b0", "b1", "b2", "b3"}));

    // write: b0,b1 as descriptor bytes; b2,b3 as raw bytes (paimon writes them to .blob files)
    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));

    // Scan and verify DataFileMeta: b2,b3 go to .blob files, "f0", "b0", "b1" go to main files.
    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
    VerifyDataFileMetas(plan, /*expected_file_count=*/3, /*expected_row_counts=*/{3, 3, 3},
                        /*expected_min_seqs=*/{1, 1, 1}, /*expected_max_seqs=*/{1, 1, 1},
                        /*expected_first_row_ids=*/{0, 0, 0},
                        /*expected_write_cols=*/
                        {std::vector<std::string>{"f0", "b0", "b1"}, std::vector<std::string>{"b2"},
                         std::vector<std::string>{"b3"}});

    std::map<std::string, std::string> read_options = {{Options::BLOB_AS_DESCRIPTOR, "true"}};
    ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), plan,
                                                /*predicate=*/nullptr, read_options));
    ASSERT_TRUE(result.chunked_array);
    auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie();
    auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);

    // b0,b1 inline descriptor (not repacked), should match input
    ASSERT_TRUE(read_struct->GetFieldByName("b0")->Equals(desc_array->GetFieldByName("b0")));
    ASSERT_TRUE(read_struct->GetFieldByName("b1")->Equals(desc_array->GetFieldByName("b1")));

    // Resolve b0,b1 descriptors back to raw bytes, then compare full struct
    ASSERT_OK_AND_ASSIGN(auto resolved,
                         ConvertDescriptorToRawBlob(read_struct, {"b0", "b1", "b2", "b3"}));
    ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(raw_array));
    ASSERT_TRUE(resolved->Equals(expected_with_rk));
}

TEST_P(BlobTableInteTest, TestBlobDescriptorMultiCommitAndShuffledReadSchema) {
    // Multiple write+commit rounds with a shuffled read schema: b3, b2, b1, b0, f0.
    arrow::FieldVector fields = {
        arrow::field("f0", arrow::int32()), BlobUtils::ToArrowField("b0", true),
        BlobUtils::ToArrowField("b1", true), BlobUtils::ToArrowField("b2", true),
        BlobUtils::ToArrowField("b3", true)};

    std::map<std::string, std::string> 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_DESCRIPTOR_FIELD, "b0,b1"}, {Options::FILE_SYSTEM, "local"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields);

    // --- First write+commit ---
    std::string raw_json_1 = R"([
        [1, "img_0", null,    "raw_2_0", "raw_3_0"],
        [2, "img_1", "vid_1", "raw_2_1", null      ]
    ])";
    auto raw_array_1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json_1).ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto desc_array_1, ConvertRawBlobToDescriptor(raw_array_1, {"b0", "b1"}));
    ASSERT_OK_AND_ASSIGN(auto commit_msgs_1,
                         WriteArray(table_path, {}, schema->field_names(), {desc_array_1}));
    ASSERT_OK(Commit(table_path, commit_msgs_1));

    // --- Second write+commit ---
    std::string raw_json_2 = R"([
        [3, "img_2", "vid_2", "raw_2_2", "raw_3_2"],
        [4, null,    "vid_3", "raw_2_3", "raw_3_3"]
    ])";
    auto raw_array_2 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json_2).ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto desc_array_2, ConvertRawBlobToDescriptor(raw_array_2, {"b0", "b1"}));
    ASSERT_OK_AND_ASSIGN(auto commit_msgs_2,
                         WriteArray(table_path, {}, schema->field_names(), {desc_array_2}));
    ASSERT_OK(Commit(table_path, commit_msgs_2));

    // --- Third write+commit ---
    std::string raw_json_3 = R"([
        [5, "img_4", null,    "raw_2_4", null     ],
        [6, "img_5", "vid_5", null,      "raw_3_5"]
    ])";
    auto raw_array_3 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json_3).ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto desc_array_3, ConvertRawBlobToDescriptor(raw_array_3, {"b0", "b1"}));
    ASSERT_OK_AND_ASSIGN(auto commit_msgs_3,
                         WriteArray(table_path, {}, schema->field_names(), {desc_array_3}));
    ASSERT_OK(Commit(table_path, commit_msgs_3));

    // test read
    {
        // --- Read with shuffled schema: b3, b2, b1, b0, f0 ---
        std::vector<std::string> shuffled_read_schema = {"b3", "b2", "b1", "b0", "f0"};
        ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));

        std::map<std::string, std::string> read_options = {{Options::BLOB_AS_DESCRIPTOR, "false"}};
        ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, shuffled_read_schema, plan,
                                                    /*predicate=*/nullptr, read_options));
        ASSERT_TRUE(result.chunked_array);
        auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie();
        auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);

        // Build expected array in shuffled order from all 3 batches
        arrow::FieldVector shuffled_fields = {
            BlobUtils::ToArrowField("b3", true), BlobUtils::ToArrowField("b2", true),
            BlobUtils::ToArrowField("b1", true), BlobUtils::ToArrowField("b0", true),
            arrow::field("f0", arrow::int32())};
        std::string expected_json = R"([
        ["raw_3_0", "raw_2_0", null,    "img_0", 1],
        [null,      "raw_2_1", "vid_1", "img_1", 2],
        ["raw_3_2", "raw_2_2", "vid_2", "img_2", 3],
        ["raw_3_3", "raw_2_3", "vid_3", null,    4],
        [null,      "raw_2_4", null,    "img_4", 5],
        ["raw_3_5", null,      "vid_5", "img_5", 6]
    ])";
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(shuffled_fields),
                                                      expected_json)
                .ValueOrDie());

        // Resolve descriptors (b0, b1 are descriptor fields) back to raw bytes
        ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"b0", "b1"}));
        ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_array));
        ASSERT_TRUE(resolved->Equals(expected_with_rk));
    }
    {
        // test scan and read with GlobalIndexResult
        std::vector<std::string> shuffled_read_schema = {"b3", "b2", "b1", "b0", "f0"};
        ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path, /*predicate=*/nullptr,
                                                  /*row_ranges=*/{Range(1, 3), Range(5, 5)}));
        std::map<std::string, std::string> read_options = {{Options::BLOB_AS_DESCRIPTOR, "false"}};
        ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, shuffled_read_schema, plan,
                                                    /*predicate=*/nullptr, read_options));
        ASSERT_TRUE(result.chunked_array);
        auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie();
        auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);

        // Build expected array in shuffled order from all 3 batches
        arrow::FieldVector shuffled_fields = {
            BlobUtils::ToArrowField("b3", true), BlobUtils::ToArrowField("b2", true),
            BlobUtils::ToArrowField("b1", true), BlobUtils::ToArrowField("b0", true),
            arrow::field("f0", arrow::int32())};
        std::string expected_json = R"([
        [null,      "raw_2_1", "vid_1", "img_1", 2],
        ["raw_3_2", "raw_2_2", "vid_2", "img_2", 3],
        ["raw_3_3", "raw_2_3", "vid_3", null,    4],
        ["raw_3_5", null,      "vid_5", "img_5", 6]
    ])";
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(shuffled_fields),
                                                      expected_json)
                .ValueOrDie());

        // Resolve descriptors (b0, b1 are descriptor fields) back to raw bytes
        ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"b0", "b1"}));
        ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_array));
        ASSERT_TRUE(resolved->Equals(expected_with_rk));
    }
}

// The shared-shredding map is read from one main data file while the blob payload is read from a
// separate blob file with the same row-id range.
TEST_P(BlobTableInteTest, TestSharedShreddingWithBlobDataEvolution) {
    if (GetParam() == "avro") {
        return;
    }

    arrow::FieldVector fields = {
        arrow::field("id", arrow::int32()),
        arrow::field("tags", arrow::map(arrow::utf8(), arrow::int64())),
        BlobUtils::ToArrowField("payload"),
    };
    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, GetParam()},
        {Options::BUCKET, "-1"},
        {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "true"},
        {Options::FILE_SYSTEM, "local"},
        {"fields.tags.map.storage-layout", "shared-shredding"},
        {"fields.tags.map.shared-shredding.max-columns", "1"},
    };
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    auto map_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[0], fields[1]}),
                                                  R"([
                [1, [["a", 10], ["z", 11]]],
                [2, [["a", 20]]],
                [3, null]
            ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto map_msgs, WriteArray(table_path, {}, {"id", "tags"}, {map_array}));
    ASSERT_OK(Commit(table_path, map_msgs));

    auto blob_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[0], fields[2]}),
                                                  R"([
                [1, "payload-1"],
                [2, "payload-2"],
                [3, "payload-3"]
            ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto blob_msgs,
                         WriteArray(table_path, {}, {"id", "payload"}, {blob_array}));
    SetFirstRowId(0, blob_msgs);
    ASSERT_OK(Commit(table_path, blob_msgs));

    auto expected = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
                                                  R"([
                [1, [["a", 10], ["z", 11]], "payload-1"],
                [2, [["a", 20]], "payload-2"],
                [3, null, "payload-3"]
            ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, arrow::schema(fields)->field_names(), expected));
}

// Two independent shared-shredding map columns are written into different main files.
TEST_P(BlobTableInteTest, TestMultipleSharedShreddingMapsWithBlobDataEvolution) {
    if (GetParam() == "avro") {
        return;
    }

    arrow::FieldVector fields = {
        arrow::field("id", arrow::int32()),
        arrow::field("f0", arrow::map(arrow::utf8(), arrow::int64())),
        arrow::field("f1", arrow::map(arrow::utf8(), arrow::utf8())),
        BlobUtils::ToArrowField("payload"),
    };
    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, GetParam()},
        {Options::BUCKET, "-1"},
        {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "true"},
        {Options::FILE_SYSTEM, "local"},
        {"fields.f0.map.storage-layout", "shared-shredding"},
        {"fields.f0.map.shared-shredding.max-columns", "1"},
        {"fields.f1.map.storage-layout", "shared-shredding"},
        {"fields.f1.map.shared-shredding.max-columns", "1"},
    };
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    auto f0_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[0], fields[1]}),
                                                  R"([
                [1, [["a", 10], ["z", 11]]],
                [2, [["a", 20]]],
                [3, null]
            ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto f0_msgs, WriteArray(table_path, {}, {"id", "f0"}, {f0_array}));
    ASSERT_OK(Commit(table_path, f0_msgs));

    auto f1_blob_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[2], fields[3]}),
                                                  R"([
                [[["b", "red"], ["y", "blue"]], "payload-1"],
                [[["b", "green"]], "payload-2"],
                [[], "payload-3"]
            ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto f1_blob_msgs,
                         WriteArray(table_path, {}, {"f1", "payload"}, {f1_blob_array}));
    SetFirstRowId(0, f1_blob_msgs);
    ASSERT_OK(Commit(table_path, f1_blob_msgs));

    auto expected = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
                                                  R"([
                [1, [["a", 10], ["z", 11]], [["b", "red"], ["y", "blue"]], "payload-1"],
                [2, [["a", 20]], [["b", "green"]], "payload-2"],
                [3, null, [], "payload-3"]
            ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, arrow::schema(fields)->field_names(), expected));
}

// A newer partial data file rewrites only the shared-shredding map for the same row-id range.
TEST_P(BlobTableInteTest, TestSharedShreddingMapOverrideWithBlobDataEvolution) {
    if (GetParam() == "avro") {
        return;
    }

    arrow::FieldVector fields = {
        arrow::field("id", arrow::int32()),
        arrow::field("tags", arrow::map(arrow::utf8(), arrow::int64())),
        BlobUtils::ToArrowField("payload"),
    };
    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, GetParam()},
        {Options::BUCKET, "-1"},
        {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "true"},
        {Options::FILE_SYSTEM, "local"},
        {"fields.tags.map.storage-layout", "shared-shredding"},
        {"fields.tags.map.shared-shredding.max-columns", "1"},
    };
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    auto old_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
                                                  R"([
                [1, [["a", 10], ["z", 11]], "payload-1"],
                [2, [["a", 20]], "payload-2"],
                [3, [["a", 30]], "payload-3"]
            ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(
        auto old_msgs,
        WriteArray(table_path, {}, arrow::schema(fields)->field_names(), {old_array}));
    ASSERT_OK(Commit(table_path, old_msgs));

    auto new_map_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[1]}),
                                                  R"([
                [[["a", 100], ["z", 101]]],
                [null],
                [[]]
            ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto new_map_msgs, WriteArray(table_path, {}, {"tags"}, {new_map_array}));
    SetFirstRowId(0, new_map_msgs);
    ASSERT_OK(Commit(table_path, new_map_msgs));

    auto expected = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
                                                  R"([
                [1, [["a", 100], ["z", 101]], "payload-1"],
                [2, null, "payload-2"],
                [3, [], "payload-3"]
            ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, arrow::schema(fields)->field_names(), expected));
}

TEST_P(BlobTableInteTest, TestOrcMapStorageLayoutEvolutionWithBlobDataEvolution) {
    if (GetParam() != "orc") {
        return;
    }

    arrow::FieldVector fields = {
        arrow::field("id", arrow::int32()),
        arrow::field("tags", arrow::map(arrow::utf8(), arrow::utf8())),
        BlobUtils::ToArrowField("payload"),
    };
    std::map<std::string, std::string> options_v0 = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, "orc"},
        {Options::BUCKET, "-1"},
        {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "true"},
        {Options::FILE_SYSTEM, "local"},
        {"fields.tags.map.storage-layout", "default"},
        {"orc.read.enable-lazy-decoding", "true"},
        {"orc.dictionary-key-size-threshold", "1"},
    };
    CreateTable(fields, /*partition_keys=*/{}, options_v0);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    auto array_v0 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
                                                  R"([
                [1, [["a", "red"], ["z", "blue"]], "payload-1"],
                [2, [["a", "red"], ["z", "green"]], "payload-2"]
            ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(
        auto msgs_v0, WriteArray(table_path, {}, arrow::schema(fields)->field_names(), {array_v0}));
    ASSERT_OK(Commit(table_path, msgs_v0));

    std::map<std::string, std::string> options_v1 = options_v0;
    options_v1["fields.tags.map.storage-layout"] = "shared-shredding";
    options_v1["fields.tags.map.shared-shredding.max-columns"] = "1";
    ASSERT_OK(WriteNextSchema(
        table_path, {DataField(0, fields[0]), DataField(1, fields[1]), DataField(2, fields[2])},
        /*highest_field_id=*/2, options_v1));

    auto array_v1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
                                                  R"([
                [3, [["a", "red"], ["z", "yellow"]], "payload-3"],
                [4, [["a", "red"]], "payload-4"]
            ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(
        auto msgs_v1, WriteArray(table_path, {}, arrow::schema(fields)->field_names(), {array_v1}));
    ASSERT_OK(Commit(table_path, msgs_v1));

    auto expected = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
                                                  R"([
                [1, [["a", "red"], ["z", "blue"]], "payload-1"],
                [2, [["a", "red"], ["z", "green"]], "payload-2"],
                [3, [["a", "red"], ["z", "yellow"]], "payload-3"],
                [4, [["a", "red"]], "payload-4"]
            ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, arrow::schema(fields)->field_names(), expected));
}

TEST_P(BlobTableInteTest, TestDataEvolutionWithBlobDescriptorField) {
    // Test DataEvolution (split-column write) combined with blob descriptor fields.
    // Schema: f0(int32), b0/b1(blob descriptor inline), b2/b3(blob).
    // Commit 1: file A writes (f0, b2, b3)
    // Commit 2: file B writes (f0, b0, b1) with SetFirstRowId(0)
    // -> merges with commit 1
    // Commit 3: file A writes (f0, b0, b1, b3)
    // Commit 4: file B writes (b0, b1, b3) with SetFirstRowId(3)
    // -> merges with commit 3
    arrow::FieldVector fields = {
        arrow::field("f0", arrow::int32()), BlobUtils::ToArrowField("b0", true),
        BlobUtils::ToArrowField("b1", true), BlobUtils::ToArrowField("b2", true),
        BlobUtils::ToArrowField("b3", true)};

    std::map<std::string, std::string> 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_DESCRIPTOR_FIELD, "b0,b1"}, {Options::FILE_SYSTEM, "local"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    // --- Commit 1: file A (f0, b2, b3), Commit 2: file B (f0, b0, b1) SetFirstRowId(0) ---
    std::string file_a1_json = R"([
        [1, "raw_2_0", "raw_3_0"],
        [2, "raw_2_1", null     ],
        [3, null,      "raw_3_2"]
    ])";
    arrow::FieldVector file_a1_fields = {fields[0], fields[3], fields[4]};
    auto file_a1_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(file_a1_fields), file_a1_json)
            .ValueOrDie());

    std::string file_b1_json = R"([
        [1, "img_0", "vid_0"],
        [2, "img_1", null   ],
        [3, "img_2", "vid_2"]
    ])";
    arrow::FieldVector file_b1_fields = {fields[0], fields[1], fields[2]};
    auto file_b1_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(file_b1_fields), file_b1_json)
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto file_b1_desc,
                         ConvertRawBlobToDescriptor(file_b1_array, {"b0", "b1"}));

    ASSERT_OK_AND_ASSIGN(auto commit_msgs_a1,
                         WriteArray(table_path, {}, {"f0", "b2", "b3"}, {file_a1_array}));
    ASSERT_OK(Commit(table_path, commit_msgs_a1));

    ASSERT_OK_AND_ASSIGN(auto commit_msgs_b1,
                         WriteArray(table_path, {}, {"f0", "b0", "b1"}, {file_b1_desc}));
    SetFirstRowId(0, commit_msgs_b1);
    ASSERT_OK(Commit(table_path, commit_msgs_b1));

    // --- Commit 3: file A (f0, b0, b1, b3), Commit 4: file B (b0, b1, b3) SetFirstRowId(3) ---
    // Duplicate cols b0, b1, b3: file B (commit 4, newer) takes precedence.
    std::string file_a2_json = R"([
        [4, "img_3_old", "vid_3_old", "raw_3_3_old"],
        [5, null,        "vid_4_old", "raw_3_4_old"],
        [6, "img_5_old", null,        null         ]
    ])";
    arrow::FieldVector file_a2_fields = {fields[0], fields[1], fields[2], fields[4]};
    auto file_a2_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(file_a2_fields), file_a2_json)
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto file_a2_desc,
                         ConvertRawBlobToDescriptor(file_a2_array, {"b0", "b1"}));

    std::string file_b2_json = R"([
        ["img_3", "vid_3", "raw_3_3"],
        [null,    "vid_4", "raw_3_4"],
        ["img_5", null,    null     ]
    ])";
    arrow::FieldVector file_b2_fields = {fields[1], fields[2], fields[4]};
    auto file_b2_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(file_b2_fields), file_b2_json)
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto file_b2_desc,
                         ConvertRawBlobToDescriptor(file_b2_array, {"b0", "b1"}));

    ASSERT_OK_AND_ASSIGN(auto commit_msgs_a2,
                         WriteArray(table_path, {}, {"f0", "b0", "b1", "b3"}, {file_a2_desc}));
    ASSERT_OK(Commit(table_path, commit_msgs_a2));

    ASSERT_OK_AND_ASSIGN(auto commit_msgs_b2,
                         WriteArray(table_path, {}, {"b0", "b1", "b3"}, {file_b2_desc}));
    SetFirstRowId(3, commit_msgs_b2);
    ASSERT_OK(Commit(table_path, commit_msgs_b2));

    // --- Read all data with full schema ---
    std::vector<std::string> read_schema = {"f0", "b0", "b1", "b2", "b3"};
    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));

    std::map<std::string, std::string> read_options = {{Options::BLOB_AS_DESCRIPTOR, "false"}};
    ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, read_schema, plan,
                                                /*predicate=*/nullptr, read_options));
    ASSERT_TRUE(result.chunked_array);
    auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie();
    auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
    ASSERT_EQ(read_struct->length(), 6);

    // Expected: round1 all columns present; round2 b2=null, b0/b1/b3 from file B (newer)
    std::string expected_json = R"([
        [1, "img_0", "vid_0", "raw_2_0", "raw_3_0"],
        [2, "img_1", null,    "raw_2_1", null      ],
        [3, "img_2", "vid_2", null,      "raw_3_2" ],
        [4, "img_3", "vid_3", null,      "raw_3_3" ],
        [5, null,    "vid_4", null,      "raw_3_4" ],
        [6, "img_5", null,    null,      null      ]
    ])";
    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
            .ValueOrDie());

    // Resolve descriptors back to raw bytes
    ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"b0", "b1"}));
    ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_array));
    ASSERT_TRUE(resolved->type()->Equals(expected_with_rk->type()));
    ASSERT_TRUE(resolved->Equals(expected_with_rk));
}

TEST_P(BlobTableInteTest, TestBlobDescriptorFieldWriteRawBytesDirectly) {
    // Similar to TestBlobDescriptorField but writes raw bytes directly without converting to
    // descriptor first. Descriptor fields reject values without the descriptor magic header.
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 BlobUtils::ToArrowField("b0", true),
                                 BlobUtils::ToArrowField("b1", true)};

    std::map<std::string, std::string> 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_DESCRIPTOR_FIELD, "b0,b1"}, {Options::FILE_SYSTEM, "local"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    // Write raw bytes directly (no ConvertRawBlobToDescriptor)
    std::string raw_json = R"([
        [1, "image_data_0", "video_data_0"],
        [2, "image_data_1", "video_data_1"],
        [3, "image_data_2", "video_data_2"]
    ])";
    auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());

    auto schema = arrow::schema(fields);
    ASSERT_NOK_WITH_MSG(WriteArray(table_path, {}, schema->field_names(), {raw_array}),
                        "BLOB inline field b0 require values to be set as corresponding type.");
}

TEST_P(BlobTableInteTest, TestBlobViewFieldWithUpstreamTable) {
    auto file_format = GetParam();
    if (file_format == "avro") {
        return;
    }

    const std::string upstream_db_name = "append_table_with_multi_blob";
    const std::string upstream_table_name = "append_table_with_multi_blob";
    std::string src_db_path = paimon::test::GetDataDir() + file_format + "/" + upstream_db_name +
                              ".db/" + upstream_table_name;
    std::string dst_db_path =
        PathUtil::JoinPath(dir_->Str(), upstream_db_name + ".db/" + upstream_table_name);
    ASSERT_TRUE(TestUtil::CopyDirectory(src_db_path, dst_db_path));

    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 BlobUtils::ToArrowField("view", true)};
    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, file_format},
        {Options::BUCKET, "-1"},
        {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "true"},
        {Options::BLOB_VIEW_FIELD, "view"},
        {Options::BLOB_VIEW_UPSTREAM_WAREHOUSE, dir_->Str()},
        {Options::FILE_SYSTEM, "local"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    // src array
    Identifier upstream_identifier(upstream_db_name, upstream_table_name);
    arrow::LargeBinaryBuilder view_builder;
    for (int32_t i = 0; i < 8; ++i) {
        if (i < 6) {
            BlobViewStruct view_struct(upstream_identifier, /*field_id=*/6,
                                       /*row_id=*/static_cast<int64_t>(i));
            auto serialized = view_struct.Serialize(pool_);
            ASSERT_TRUE(view_builder
                            .Append(reinterpret_cast<const uint8_t*>(serialized->data()),
                                    serialized->size())
                            .ok());
        } else {
            ASSERT_TRUE(view_builder.AppendNull().ok());
        }
    }
    std::shared_ptr<arrow::Array> write_view_array;
    ASSERT_TRUE(view_builder.Finish(&write_view_array).ok());
    auto write_f0_array = arrow::ipc::internal::json::ArrayFromJSON(
                              arrow::int32(), R"([100,101,102,103,104,105,106,107])")
                              .ValueOrDie();
    auto write_struct = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::StructArray::Make(arrow::ArrayVector({write_f0_array, write_view_array}),
                                 std::vector<std::string>({"f0", "view"}))
            .ValueOrDie());

    // write & commit
    auto schema = arrow::schema(fields);
    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(table_path, {}, schema->field_names(), {write_struct}));
    ASSERT_OK(Commit(table_path, commit_msgs));

    std::string padding_b(2048, 'b');
    std::string padding_d(2048, 'd');
    std::string padding_e(2048, 'e');
    std::string padding_f(2048, 'f');

    // scan & read
    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
    ASSERT_OK_AND_ASSIGN(auto result,
                         ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr));
    ASSERT_TRUE(result.chunked_array);
    auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie();
    auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
    ASSERT_EQ(read_struct->length(), 8);
    ASSERT_OK_AND_ASSIGN(auto result_array, ConvertDescriptorToRawBlob(read_struct, {"view"}));

    // clang-format off
    std::string expected_json = R"([
[100, null],
[101, ")" + padding_b + R"("],
[102, null],
[103, ")" + padding_d + R"("],
[104, ")" + padding_e + R"("],
[105, ")" + padding_f + R"("],
[106, null],
[107, null]
])";
    // clang-format on
    auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_struct));

    ASSERT_TRUE(result_array->Equals(expected_with_rk))
        << "result_array:" << result_array->ToString() << std::endl
        << "expected:" << expected_with_rk->ToString();

    // Sub-case 1: scan with row_ranges.
    {
        ASSERT_OK_AND_ASSIGN(auto range_plan, ScanTable(table_path, /*predicate=*/nullptr,
                                                        /*row_ranges=*/{Range(1, 3), Range(5, 5)}));
        ASSERT_OK_AND_ASSIGN(auto range_result, ReadTable(table_path, schema->field_names(),
                                                          range_plan, /*predicate=*/nullptr));
        ASSERT_TRUE(range_result.chunked_array);
        auto range_concat = arrow::Concatenate(range_result.chunked_array->chunks()).ValueOrDie();
        auto range_struct = std::dynamic_pointer_cast<arrow::StructArray>(range_concat);
        ASSERT_EQ(range_struct->length(), 4);
        ASSERT_OK_AND_ASSIGN(auto range_resolved,
                             ConvertDescriptorToRawBlob(range_struct, {"view"}));

        // clang-format off
        std::string range_json = R"([
[101, ")" + padding_b + R"("],
[102, null],
[103, ")" + padding_d + R"("],
[105, ")" + padding_f + R"("]
])";
        // clang-format on
        auto range_expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), range_json)
                .ValueOrDie());
        ASSERT_OK_AND_ASSIGN(auto range_expected_with_rk,
                             PrependRowKindColumn(range_expected_struct));
        ASSERT_TRUE(range_resolved->Equals(range_expected_with_rk))
            << "range_resolved:" << range_resolved->ToString() << std::endl
            << "expected:" << range_expected_with_rk->ToString();
    }

    // Sub-case 2: scan with predicate (f0 > 102), data evolution split read will ignore format push
    // down
    {
        auto predicate = PredicateBuilder::GreaterThan(/*field_index=*/0, /*field_name=*/"f0",
                                                       FieldType::INT, Literal(102));
        ASSERT_OK_AND_ASSIGN(auto pred_plan, ScanTable(table_path, predicate, /*row_ranges=*/{}));
        ASSERT_OK_AND_ASSIGN(auto pred_result,
                             ReadTable(table_path, schema->field_names(), pred_plan, predicate));
        ASSERT_TRUE(pred_result.chunked_array);
        auto pred_concat = arrow::Concatenate(pred_result.chunked_array->chunks()).ValueOrDie();
        auto pred_struct = std::dynamic_pointer_cast<arrow::StructArray>(pred_concat);
        ASSERT_EQ(pred_struct->length(), 8);
        ASSERT_OK_AND_ASSIGN(auto pred_resolved, ConvertDescriptorToRawBlob(pred_struct, {"view"}));

        // clang-format off
        std::string pred_json = R"([
[100, null],
[101, ")" + padding_b + R"("],
[102, null],
[103, ")" + padding_d + R"("],
[104, ")" + padding_e + R"("],
[105, ")" + padding_f + R"("],
[106, null],
[107, null]
])";
        // clang-format on
        auto pred_expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), pred_json)
                .ValueOrDie());
        ASSERT_OK_AND_ASSIGN(auto pred_expected_with_rk,
                             PrependRowKindColumn(pred_expected_struct));
        ASSERT_TRUE(pred_resolved->Equals(pred_expected_with_rk))
            << "pred_resolved:" << pred_resolved->ToString() << std::endl
            << "expected:" << pred_expected_with_rk->ToString();
    }
}

TEST_P(BlobTableInteTest, TestForwardBlobViewReference) {
    auto file_format = GetParam();
    if (file_format == "avro") {
        return;
    }

    // Forward blob view references between two blob-view tables: read the source table with
    // resolve dynamically disabled, write the preserved BlobViewStruct bytes into the target
    // table, then verify the target still stores the original upstream references and a
    // default read resolves them to the actual upstream blob values.
    const std::string upstream_db_name = "append_table_with_multi_blob";
    const std::string upstream_table_name = "append_table_with_multi_blob";
    std::string src_db_path = paimon::test::GetDataDir() + file_format + "/" + upstream_db_name +
                              ".db/" + upstream_table_name;
    std::string dst_db_path =
        PathUtil::JoinPath(dir_->Str(), upstream_db_name + ".db/" + upstream_table_name);
    ASSERT_TRUE(TestUtil::CopyDirectory(src_db_path, dst_db_path));

    // The source table has no upstream warehouse configured: only a read with resolve
    // dynamically disabled can succeed on it.
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 BlobUtils::ToArrowField("view", true)};
    std::map<std::string, std::string> source_options = {{Options::MANIFEST_FORMAT, "orc"},
                                                         {Options::FILE_FORMAT, file_format},
                                                         {Options::BUCKET, "-1"},
                                                         {Options::ROW_TRACKING_ENABLED, "true"},
                                                         {Options::DATA_EVOLUTION_ENABLED, "true"},
                                                         {Options::BLOB_VIEW_FIELD, "view"},
                                                         {Options::FILE_SYSTEM, "local"}};
    CreateTable(fields, /*partition_keys=*/{}, source_options);
    std::string source_table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    // The target table configures the upstream warehouse for resolving forwarded references.
    std::map<std::string, std::string> target_options = source_options;
    target_options[Options::BLOB_VIEW_UPSTREAM_WAREHOUSE] = dir_->Str();
    auto schema = arrow::schema(fields);
    ::ArrowSchema c_target_schema;
    ASSERT_TRUE(arrow::ExportSchema(*schema, &c_target_schema).ok());
    ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(dir_->Str(), {}));
    ASSERT_OK(catalog->CreateTable(Identifier("foo", "bar_forward"), &c_target_schema,
                                   /*partition_keys=*/{}, /*primary_keys=*/{}, target_options,
                                   /*ignore_if_exists=*/false));
    std::string target_table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar_forward");

    // src array
    Identifier upstream_identifier(upstream_db_name, upstream_table_name);
    arrow::LargeBinaryBuilder view_builder;
    for (int32_t i = 0; i < 8; ++i) {
        if (i < 6) {
            BlobViewStruct view_struct(upstream_identifier, /*field_id=*/6,
                                       /*row_id=*/static_cast<int64_t>(i));
            auto serialized = view_struct.Serialize(pool_);
            ASSERT_TRUE(view_builder
                            .Append(reinterpret_cast<const uint8_t*>(serialized->data()),
                                    serialized->size())
                            .ok());
        } else {
            ASSERT_TRUE(view_builder.AppendNull().ok());
        }
    }
    std::shared_ptr<arrow::Array> write_view_array;
    ASSERT_TRUE(view_builder.Finish(&write_view_array).ok());
    auto write_f0_array = arrow::ipc::internal::json::ArrayFromJSON(
                              arrow::int32(), R"([100,101,102,103,104,105,106,107])")
                              .ValueOrDie();
    auto write_struct = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::StructArray::Make(arrow::ArrayVector({write_f0_array, write_view_array}),
                                 std::vector<std::string>({"f0", "view"}))
            .ValueOrDie());

    // write & commit into the source table
    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(source_table_path, {}, schema->field_names(), {write_struct}));
    ASSERT_OK(Commit(source_table_path, commit_msgs));

    // A default read fails on the missing upstream warehouse: the pass-through below is
    // enabled by the dynamic option alone.
    ASSERT_OK_AND_ASSIGN(auto source_plan, ScanTable(source_table_path));
    ASSERT_NOK_WITH_MSG(
        ReadTable(source_table_path, schema->field_names(), source_plan, /*predicate=*/nullptr),
        "BLOB_VIEW_UPSTREAM_WAREHOUSE");

    ASSERT_OK_AND_ASSIGN(
        auto source_result,
        ReadTable(source_table_path, schema->field_names(), source_plan, /*predicate=*/nullptr,
                  {{Options::BLOB_VIEW_RESOLVE_ENABLED, "false"}}));
    ASSERT_TRUE(source_result.chunked_array);
    auto source_concat = arrow::Concatenate(source_result.chunked_array->chunks()).ValueOrDie();
    auto source_struct = std::dynamic_pointer_cast<arrow::StructArray>(source_concat);
    ASSERT_EQ(source_struct->length(), 8);
    auto forward_f0_array = source_struct->GetFieldByName("f0");
    ASSERT_TRUE(forward_f0_array);
    auto forward_view_array = source_struct->GetFieldByName("view");
    ASSERT_TRUE(forward_view_array);
    ASSERT_TRUE(forward_view_array->Equals(write_view_array))
        << "source view:" << forward_view_array->ToString() << std::endl
        << "written view:" << write_view_array->ToString();

    // Forward the preserved references into the target blob-view table.
    auto forward_struct = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::StructArray::Make(arrow::ArrayVector({forward_f0_array, forward_view_array}),
                                 std::vector<std::string>({"f0", "view"}))
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(
        auto forward_commit_msgs,
        WriteArray(target_table_path, {}, schema->field_names(), {forward_struct}));
    ASSERT_OK(Commit(target_table_path, forward_commit_msgs));

    // The target table stores the original upstream references byte-identically.
    ASSERT_OK_AND_ASSIGN(auto target_plan, ScanTable(target_table_path));
    ASSERT_OK_AND_ASSIGN(
        auto raw_target_result,
        ReadTable(target_table_path, schema->field_names(), target_plan, /*predicate=*/nullptr,
                  {{Options::BLOB_VIEW_RESOLVE_ENABLED, "false"}}));
    ASSERT_TRUE(raw_target_result.chunked_array);
    auto raw_target_concat =
        arrow::Concatenate(raw_target_result.chunked_array->chunks()).ValueOrDie();
    auto raw_target_struct = std::dynamic_pointer_cast<arrow::StructArray>(raw_target_concat);
    ASSERT_EQ(raw_target_struct->length(), 8);
    auto raw_target_view_array = raw_target_struct->GetFieldByName("view");
    ASSERT_TRUE(raw_target_view_array);
    ASSERT_TRUE(raw_target_view_array->Equals(write_view_array))
        << "target view:" << raw_target_view_array->ToString() << std::endl
        << "written view:" << write_view_array->ToString();

    // A default read of the target table resolves to the actual upstream blob values.
    ASSERT_OK_AND_ASSIGN(auto resolved_result,
                         ReadTable(target_table_path, schema->field_names(), target_plan,
                                   /*predicate=*/nullptr));
    ASSERT_TRUE(resolved_result.chunked_array);
    auto resolved_concat = arrow::Concatenate(resolved_result.chunked_array->chunks()).ValueOrDie();
    auto resolved_struct = std::dynamic_pointer_cast<arrow::StructArray>(resolved_concat);
    ASSERT_EQ(resolved_struct->length(), 8);
    ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(resolved_struct, {"view"}));

    std::string padding_b(2048, 'b');
    std::string padding_d(2048, 'd');
    std::string padding_e(2048, 'e');
    std::string padding_f(2048, 'f');
    // clang-format off
    std::string expected_json = R"([
[100, null],
[101, ")" + padding_b + R"("],
[102, null],
[103, ")" + padding_d + R"("],
[104, ")" + padding_e + R"("],
[105, ")" + padding_f + R"("],
[106, null],
[107, null]
])";
    // clang-format on
    auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_struct));
    ASSERT_TRUE(resolved->Equals(expected_with_rk))
        << "resolved:" << resolved->ToString() << std::endl
        << "expected:" << expected_with_rk->ToString();
}

TEST_P(BlobTableInteTest, TestBlobViewFieldWithUpstreamDescriptorBlob) {
    auto file_format = GetParam();
    // Upstream table has two blob descriptor fields. The downstream view references cells from
    // both b0 (field_id=1) and b1 (field_id=2).
    const std::string upstream_db_name = "upstream_two_blob";
    const std::string upstream_table_name = "upstream_two_blob";
    arrow::FieldVector upstream_fields = {arrow::field("f0", arrow::int32()),
                                          BlobUtils::ToArrowField("b0", true),
                                          BlobUtils::ToArrowField("b1", true)};
    auto upstream_schema = arrow::schema(upstream_fields);
    std::map<std::string, std::string> upstream_options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, file_format},
        {Options::BUCKET, "-1"},
        {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "true"},
        {Options::BLOB_DESCRIPTOR_FIELD, "b0,b1"},
        {Options::FILE_SYSTEM, "local"}};

    ::ArrowSchema upstream_c_schema;
    ASSERT_TRUE(arrow::ExportSchema(*upstream_schema, &upstream_c_schema).ok());
    ASSERT_OK_AND_ASSIGN(auto upstream_catalog,
                         Catalog::Create(dir_->Str(), {{Options::FILE_SYSTEM, "local"}}));
    ASSERT_OK(upstream_catalog->CreateDatabase(upstream_db_name, {}, /*ignore_if_exists=*/true));
    ASSERT_OK(upstream_catalog->CreateTable(
        Identifier(upstream_db_name, upstream_table_name), &upstream_c_schema,
        /*partition_keys=*/{}, /*primary_keys=*/{}, upstream_options,
        /*ignore_if_exists=*/false));
    std::string upstream_table_path =
        PathUtil::JoinPath(dir_->Str(), upstream_db_name + ".db/" + upstream_table_name);

    // Write 4 rows of b0/b1 data into the upstream table.
    std::string upstream_raw_json = R"([
[0, "b0_data_0", "b1_data_0"],
[1, "b0_data_1", "b1_data_1"],
[2, "b0_data_2", "b1_data_2"],
[3, "b0_data_3", "b1_data_3"]
])";
    auto upstream_raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(upstream_fields),
                                                  upstream_raw_json)
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto upstream_desc_array,
                         ConvertRawBlobToDescriptor(upstream_raw_array, {"b0", "b1"}));
    ASSERT_OK_AND_ASSIGN(
        auto upstream_commit_msgs,
        WriteArray(upstream_table_path, {}, upstream_schema->field_names(), {upstream_desc_array}));
    ASSERT_OK(Commit(upstream_table_path, upstream_commit_msgs));

    // Create the downstream blob-view table that references both b0 and b1 of the upstream table.
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 BlobUtils::ToArrowField("view", true)};
    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, file_format},
        {Options::BUCKET, "-1"},
        {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "true"},
        {Options::BLOB_VIEW_FIELD, "view"},
        {Options::BLOB_VIEW_UPSTREAM_WAREHOUSE, dir_->Str()},
        {Options::FILE_SYSTEM, "local"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    // Build the view column mixing references to b0 (field_id=1) and b1 (field_id=2).
    // - row 0: b0 row 0 -> "b0_data_0"
    // - row 1: b1 row 1 -> "b1_data_1"
    // - row 2: b0 row 2 -> "b0_data_2"
    // - row 3: b1 row 3 -> "b1_data_3"
    Identifier upstream_identifier(upstream_db_name, upstream_table_name);
    auto append_view = [&](int32_t field_id, int64_t row_id, arrow::LargeBinaryBuilder* builder) {
        BlobViewStruct view_struct(upstream_identifier, field_id, row_id);
        auto serialized = view_struct.Serialize(pool_);
        ASSERT_TRUE(
            builder
                ->Append(reinterpret_cast<const uint8_t*>(serialized->data()), serialized->size())
                .ok());
    };
    arrow::LargeBinaryBuilder view_builder;
    append_view(/*field_id=*/1, /*row_id=*/0, &view_builder);
    append_view(/*field_id=*/2, /*row_id=*/1, &view_builder);
    append_view(/*field_id=*/1, /*row_id=*/2, &view_builder);
    append_view(/*field_id=*/2, /*row_id=*/3, &view_builder);
    std::shared_ptr<arrow::Array> write_view_array;
    ASSERT_TRUE(view_builder.Finish(&write_view_array).ok());

    auto write_f0_array =
        arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), R"([100,101,102,103])")
            .ValueOrDie();
    auto write_struct = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::StructArray::Make(arrow::ArrayVector({write_f0_array, write_view_array}),
                                 std::vector<std::string>({"f0", "view"}))
            .ValueOrDie());

    auto schema = arrow::schema(fields);
    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(table_path, {}, schema->field_names(), {write_struct}));
    ASSERT_OK(Commit(table_path, commit_msgs));

    // Scan/read the downstream table and verify the resolved view blobs.
    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
    ASSERT_OK_AND_ASSIGN(auto result,
                         ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr));
    ASSERT_TRUE(result.chunked_array);
    auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie();
    auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
    ASSERT_EQ(read_struct->length(), 4);
    ASSERT_OK_AND_ASSIGN(auto result_array, ConvertDescriptorToRawBlob(read_struct, {"view"}));

    std::string expected_json = R"([
[100, "b0_data_0"],
[101, "b1_data_1"],
[102, "b0_data_2"],
[103, "b1_data_3"]
])";
    auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_struct));
    ASSERT_TRUE(result_array->Equals(expected_with_rk))
        << "result_array:" << result_array->ToString() << std::endl
        << "expected:" << expected_with_rk->ToString();
}

TEST_P(BlobTableInteTest, TestBlobViewFieldWithMultipleUpstreamTables) {
    auto file_format = GetParam();
    if (file_format == "avro") {
        return;
    }

    // Upstream table 1: append_table_with_multi_blob, with two blob fields f5 (field_id=5) and
    // f6 (field_id=6).
    const std::string multi_blob_db_name = "append_table_with_multi_blob";
    const std::string multi_blob_table_name = "append_table_with_multi_blob";
    {
        std::string src_db_path = paimon::test::GetDataDir() + file_format + "/" +
                                  multi_blob_db_name + ".db/" + multi_blob_table_name;
        std::string dst_db_path =
            PathUtil::JoinPath(dir_->Str(), multi_blob_db_name + ".db/" + multi_blob_table_name);
        ASSERT_TRUE(TestUtil::CopyDirectory(src_db_path, dst_db_path));
    }

    // Upstream table 2: blob_append_table_alter_table_with_cast_with_data_evolution, with one blob
    // field blob (field_id=8).
    const std::string alter_db_name = "blob_append_table_alter_table_with_cast_with_data_evolution";
    const std::string alter_table_name =
        "blob_append_table_alter_table_with_cast_with_data_evolution";
    {
        std::string src_db_path = paimon::test::GetDataDir() + file_format + "/" + alter_db_name +
                                  ".db/" + alter_table_name;
        std::string dst_db_path =
            PathUtil::JoinPath(dir_->Str(), alter_db_name + ".db/" + alter_table_name);
        ASSERT_TRUE(TestUtil::CopyDirectory(src_db_path, dst_db_path));
    }

    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 BlobUtils::ToArrowField("view1", true),
                                 BlobUtils::ToArrowField("view2", true)};
    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, file_format},
        {Options::BUCKET, "-1"},
        {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "true"},
        {Options::BLOB_VIEW_FIELD, "view1,view2"},
        {Options::BLOB_VIEW_UPSTREAM_WAREHOUSE, dir_->Str()},
        {Options::FILE_SYSTEM, "local"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    Identifier multi_blob_identifier(multi_blob_db_name, multi_blob_table_name);
    Identifier alter_identifier(alter_db_name, alter_table_name);

    auto append_view = [&](const Identifier& identifier, int32_t field_id, int64_t row_id,
                           arrow::LargeBinaryBuilder* builder) {
        BlobViewStruct view_struct(identifier, field_id, row_id);
        auto serialized = view_struct.Serialize(pool_);
        ASSERT_TRUE(
            builder
                ->Append(reinterpret_cast<const uint8_t*>(serialized->data()), serialized->size())
                .ok());
    };

    // Build view1 column. References multi_blob.f5 (field_id=5) and f6 (field_id=6).
    // Some upstream cells are referenced more than once on purpose.
    // - row 0: f5 row 3  -> 'D' * 1024
    // - row 1: f6 row 1  -> 'b' * 2048
    // - row 2: f5 row 3  -> 'D' * 1024  (repeat of row 0)
    // - row 3: f6 row 4  -> 'e' * 2048
    // - row 4: f5 row 3  -> 'D' * 1024  (repeat of row 0)
    // - row 5: f6 row 1  -> 'b' * 2048  (repeat of row 1)
    // - row 6: f5 row 5  -> 'F' * 1024
    // - row 7: f6 row 6  -> 'g' * 2048
    arrow::LargeBinaryBuilder view1_builder;
    append_view(multi_blob_identifier, /*field_id=*/5, /*row_id=*/3, &view1_builder);
    append_view(multi_blob_identifier, /*field_id=*/6, /*row_id=*/1, &view1_builder);
    append_view(multi_blob_identifier, /*field_id=*/5, /*row_id=*/3, &view1_builder);
    append_view(multi_blob_identifier, /*field_id=*/6, /*row_id=*/4, &view1_builder);
    append_view(multi_blob_identifier, /*field_id=*/5, /*row_id=*/3, &view1_builder);
    append_view(multi_blob_identifier, /*field_id=*/6, /*row_id=*/1, &view1_builder);
    append_view(multi_blob_identifier, /*field_id=*/5, /*row_id=*/5, &view1_builder);
    append_view(multi_blob_identifier, /*field_id=*/6, /*row_id=*/6, &view1_builder);
    std::shared_ptr<arrow::Array> write_view1_array;
    ASSERT_TRUE(view1_builder.Finish(&write_view1_array).ok());

    // Build view2 column. References alter.blob (field_id=8).
    // Some upstream cells are referenced more than once on purpose.
    // - row 0: blob row 0  -> "Lily"
    // - row 1: blob row 5  -> "Apple"
    // - row 2: blob row 0  -> "Lily"        (repeat of row 0)
    // - row 3: blob row 2  -> "Bob"
    // - row 4: blob row 5  -> "Apple"       (repeat of row 1)
    // - row 5: blob row 9  -> "Elderberry"
    // - row 6: blob row 0  -> "Lily"        (repeat of row 0)
    // - row 7: blob row 3  -> "Cindy"
    arrow::LargeBinaryBuilder view2_builder;
    append_view(alter_identifier, /*field_id=*/8, /*row_id=*/0, &view2_builder);
    append_view(alter_identifier, /*field_id=*/8, /*row_id=*/5, &view2_builder);
    append_view(alter_identifier, /*field_id=*/8, /*row_id=*/0, &view2_builder);
    append_view(alter_identifier, /*field_id=*/8, /*row_id=*/2, &view2_builder);
    append_view(alter_identifier, /*field_id=*/8, /*row_id=*/5, &view2_builder);
    append_view(alter_identifier, /*field_id=*/8, /*row_id=*/9, &view2_builder);
    append_view(alter_identifier, /*field_id=*/8, /*row_id=*/0, &view2_builder);
    append_view(alter_identifier, /*field_id=*/8, /*row_id=*/3, &view2_builder);
    std::shared_ptr<arrow::Array> write_view2_array;
    ASSERT_TRUE(view2_builder.Finish(&write_view2_array).ok());

    auto write_f0_array = arrow::ipc::internal::json::ArrayFromJSON(
                              arrow::int32(), R"([100,101,102,103,104,105,106,107])")
                              .ValueOrDie();
    auto write_struct = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::StructArray::Make(
            arrow::ArrayVector({write_f0_array, write_view1_array, write_view2_array}),
            std::vector<std::string>({"f0", "view1", "view2"}))
            .ValueOrDie());

    // write & commit
    auto schema = arrow::schema(fields);
    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(table_path, {}, schema->field_names(), {write_struct}));
    ASSERT_OK(Commit(table_path, commit_msgs));

    // Expected blob contents per referenced upstream cell.
    std::string blob_f5_row3(1024, 'D');  // multi_blob.f5 row 3
    std::string blob_f5_row5(1024, 'F');  // multi_blob.f5 row 5
    std::string blob_f6_row1(2048, 'b');  // multi_blob.f6 row 1
    std::string blob_f6_row4(2048, 'e');  // multi_blob.f6 row 4
    std::string blob_f6_row6(2048, 'g');  // multi_blob.f6 row 6

    std::vector<std::string> read_fields = {"view2", "view1", "f0"};
    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
    ASSERT_OK_AND_ASSIGN(auto result,
                         ReadTable(table_path, read_fields, plan, /*predicate=*/nullptr));
    ASSERT_TRUE(result.chunked_array);
    auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie();
    auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
    ASSERT_EQ(read_struct->length(), 8);
    ASSERT_OK_AND_ASSIGN(auto result_array,
                         ConvertDescriptorToRawBlob(read_struct, {"view1", "view2"}));

    // Expected struct follows the requested (shuffled) column order: view2, view1, f0.
    arrow::FieldVector expected_fields = {BlobUtils::ToArrowField("view2", true),
                                          BlobUtils::ToArrowField("view1", true),
                                          arrow::field("f0", arrow::int32())};
    // clang-format off
    std::string expected_json = R"([
["Lily", ")" + blob_f5_row3 + R"(", 100],
["Apple", ")" + blob_f6_row1 + R"(", 101],
["Lily", ")" + blob_f5_row3 + R"(", 102],
["Bob", ")" + blob_f6_row4 + R"(", 103],
["Apple", ")" + blob_f5_row3 + R"(", 104],
["Elderberry", ")" + blob_f6_row1 + R"(", 105],
["Lily", ")" + blob_f5_row5 + R"(", 106],
["Cindy", ")" + blob_f6_row6 + R"(", 107]
])";
    // clang-format on
    auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(expected_fields), expected_json)
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_struct));

    ASSERT_TRUE(result_array->Equals(expected_with_rk))
        << "result_array:" << result_array->ToString() << std::endl
        << "expected:" << expected_with_rk->ToString();
}

TEST_P(BlobTableInteTest, TestBlobViewFailsWhenBothPathsAbsent) {
    auto upstream_dir = UniqueTestDirectory::Create("local");
    arrow::FieldVector fields =
        CreateBlobViewTable(upstream_dir->Str(), /*deletion_vectors_enabled=*/false);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields);

    ASSERT_OK(WriteDanglingBlobViewRows(table_path));

    // reading fails because neither the table path nor the fallback one exists upstream
    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
    ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr),
                        "Ambiguous table path");
}

TEST_P(BlobTableInteTest, TestBlobViewSkipsDanglingReferenceOfDeletedRow) {
    auto upstream_dir = UniqueTestDirectory::Create("local");
    arrow::FieldVector fields =
        CreateBlobViewTable(upstream_dir->Str(), /*deletion_vectors_enabled=*/true);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields);

    // resolving row 0 would fail the whole query, so the pre-read that collects the structs to
    // resolve must honor the deletion vector too
    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> commit_msgs,
                         WriteDanglingBlobViewRows(table_path, /*first_row_id=*/0));

    // without the deletion vector the dangling reference fails the read
    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
    ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr),
                        "Ambiguous table path");

    ASSERT_OK_AND_ASSIGN(std::string anchor_file_name,
                         DeletionVectorTestHelper::RetrieveAnchorFileName(commit_msgs));
    ASSERT_OK(CommitDeletionVectors(table_path, commit_msgs[0],
                                    {{anchor_file_name, /*deleted_positions=*/{0}}}));

    // the deleted row is never resolved, so only the surviving row is returned
    ASSERT_OK_AND_ASSIGN(auto dv_plan, ScanTable(table_path));
    ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), dv_plan,
                                                /*predicate=*/nullptr));
    ASSERT_TRUE(result.chunked_array);
    auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie();
    auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
    ASSERT_EQ(read_struct->length(), 1);

    auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([[101, null]])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_struct));
    ASSERT_TRUE(read_struct->Equals(expected_with_rk))
        << "read_struct:" << read_struct->ToString() << std::endl
        << "expected:" << expected_with_rk->ToString();
}

TEST_P(BlobTableInteTest, TestBlobViewSkipsDanglingReferenceInEveryRowRangeGroup) {
    auto upstream_dir = UniqueTestDirectory::Create("local");
    arrow::FieldVector fields =
        CreateBlobViewTable(upstream_dir->Str(), /*deletion_vectors_enabled=*/true);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields);

    // two commits, so the two pairs of rows form two row range groups. The blob view pre-read
    // walks the groups one by one; a group it drops, or reads with another group's deletion
    // vector, still hands its dangling reference to the resolver and fails the whole read.
    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> commit_msgs0,
                         WriteDanglingBlobViewRows(table_path, /*first_row_id=*/0));
    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> commit_msgs1,
                         WriteDanglingBlobViewRows(table_path, /*first_row_id=*/2));

    // without deletion vectors both groups' dangling references fail the read
    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
    ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr),
                        "Ambiguous table path");

    // each group's anchor gets a deletion vector dropping its row 0, the dangling one
    ASSERT_OK_AND_ASSIGN(std::string anchor_file_name0,
                         DeletionVectorTestHelper::RetrieveAnchorFileName(commit_msgs0));
    ASSERT_OK_AND_ASSIGN(std::string anchor_file_name1,
                         DeletionVectorTestHelper::RetrieveAnchorFileName(commit_msgs1));
    ASSERT_NE(anchor_file_name0, anchor_file_name1);
    ASSERT_OK(CommitDeletionVectors(table_path, commit_msgs0[0],
                                    {{anchor_file_name0, /*deleted_positions=*/{0}},
                                     {anchor_file_name1, /*deleted_positions=*/{0}}}));

    // both groups keep only their row 1, which holds no blob view at all
    auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
                                                  R"([[101, null], [101, null]])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_struct));
}

// Pins one of the limitations documented on CreateBlobViewReader: the pre-read takes no
// predicate, so filtering a row out that way does not stop its blob view reference from being
// resolved. This asserts the current behavior, not a desired one.
TEST_P(BlobTableInteTest, TestBlobViewPreReadHonorsRowRangesNotPredicate) {
    auto upstream_dir = UniqueTestDirectory::Create("local");
    arrow::FieldVector fields =
        CreateBlobViewTable(upstream_dir->Str(), /*deletion_vectors_enabled=*/false);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields);

    // row 0 (f0 = 100) holds the dangling reference, row 1 (f0 = 101) holds no blob view
    ASSERT_OK(WriteDanglingBlobViewRows(table_path));

    // reading the whole split fails on the dangling reference of row 0
    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
    ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr),
                        "Ambiguous table path");

    // a predicate keeping only row 1 does not rescue the query: it is evaluated above the split
    // read, while the pre-read takes none, so row 0's reference is still collected and resolved
    auto keep_row_1 = PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0",
                                              FieldType::INT, Literal(101));
    ASSERT_OK_AND_ASSIGN(auto filtered_plan, ScanTable(table_path, keep_row_1));
    ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), filtered_plan, keep_row_1),
                        "Ambiguous table path");

    // dropping the same row through the row ranges does: the pre-read honors them the same way
    // the main read does, so row 0 is never resolved
    auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([[101, null]])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_struct,
                          /*predicate=*/nullptr, /*row_ranges=*/{Range(1, 1)}));
}

TEST_P(BlobTableInteTest, TestBlobViewWithFallbackPath) {
    auto file_format = GetParam();
    const std::string upstream_db_name = "fallback_db";
    const std::string upstream_table_name = "fallback_table";
    arrow::FieldVector upstream_fields = {arrow::field("f0", arrow::int32()),
                                          BlobUtils::ToArrowField("blob", true)};
    auto upstream_schema = arrow::schema(upstream_fields);
    std::map<std::string, std::string> upstream_options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, file_format},
        {Options::BUCKET, "-1"},
        {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "true"},
        {Options::BLOB_AS_DESCRIPTOR, "true"},
        {Options::FILE_SYSTEM, "local"}};

    // Create the upstream table at the fallback path: <warehouse>/db/table (no .db).
    auto upstream_dir = UniqueTestDirectory::Create("local");
    std::string fallback_table_path =
        PathUtil::JoinPath(upstream_dir->Str(), upstream_db_name + "/" + upstream_table_name);

    // Manually create schema at fallback path so it can be read as a valid paimon table.
    {
        // Use a temporary warehouse with Catalog to build the table data, then copy to fallback.
        auto temp_dir = UniqueTestDirectory::Create("local");
        ::ArrowSchema c_schema;
        ASSERT_TRUE(arrow::ExportSchema(*upstream_schema, &c_schema).ok());
        ASSERT_OK_AND_ASSIGN(auto catalog,
                             Catalog::Create(temp_dir->Str(), {{Options::FILE_SYSTEM, "local"}}));
        ASSERT_OK(catalog->CreateDatabase(upstream_db_name, {}, /*ignore_if_exists=*/true));
        ASSERT_OK(catalog->CreateTable(Identifier(upstream_db_name, upstream_table_name), &c_schema,
                                       /*partition_keys=*/{}, /*primary_keys=*/{}, upstream_options,
                                       /*ignore_if_exists=*/false));
        std::string temp_table_path =
            PathUtil::JoinPath(temp_dir->Str(), upstream_db_name + ".db/" + upstream_table_name);

        // Write data to the temp table.
        std::string raw_json = R"([[0, "hello"], [1, "world"]])";
        auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(upstream_fields), raw_json)
                .ValueOrDie());
        ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"}));
        ASSERT_OK_AND_ASSIGN(
            auto upstream_commit_msgs,
            WriteArray(temp_table_path, {}, upstream_schema->field_names(), {desc_array}));
        ASSERT_OK(Commit(temp_table_path, upstream_commit_msgs));

        // Copy the temp table to the fallback path (without .db).
        ASSERT_TRUE(TestUtil::CopyDirectory(temp_table_path, fallback_table_path));
    }

    // Build the downstream table.
    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
                                 BlobUtils::ToArrowField("view", true)};
    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, file_format},
        {Options::BUCKET, "-1"},
        {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "true"},
        {Options::BLOB_VIEW_FIELD, "view"},
        {Options::BLOB_VIEW_UPSTREAM_WAREHOUSE, upstream_dir->Str()},
        {Options::FILE_SYSTEM, "local"}};
    CreateTable(fields, /*partition_keys=*/{}, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    // Write downstream rows referencing the upstream fallback table.
    Identifier upstream_identifier(upstream_db_name, upstream_table_name);
    arrow::LargeBinaryBuilder view_builder;
    for (int64_t row = 0; row < 2; ++row) {
        BlobViewStruct view_struct(upstream_identifier, /*field_id=*/1, /*row_id=*/row);
        auto serialized = view_struct.Serialize(pool_);
        ASSERT_TRUE(
            view_builder
                .Append(reinterpret_cast<const uint8_t*>(serialized->data()), serialized->size())
                .ok());
    }
    std::shared_ptr<arrow::Array> write_view_array;
    ASSERT_TRUE(view_builder.Finish(&write_view_array).ok());
    auto write_f0_array =
        arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), R"([100, 101])").ValueOrDie();
    auto write_struct = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::StructArray::Make(arrow::ArrayVector({write_f0_array, write_view_array}),
                                 std::vector<std::string>({"f0", "view"}))
            .ValueOrDie());

    auto schema = arrow::schema(fields);
    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(table_path, {}, schema->field_names(), {write_struct}));
    ASSERT_OK(Commit(table_path, commit_msgs));

    // Read and verify
    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
    ASSERT_OK_AND_ASSIGN(auto result,
                         ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr));
    ASSERT_TRUE(result.chunked_array);
    auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie();
    auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
    ASSERT_EQ(read_struct->length(), 2);
    ASSERT_OK_AND_ASSIGN(auto result_array, ConvertDescriptorToRawBlob(read_struct, {"view"}));

    std::string expected_json = R"([[100, "hello"], [101, "world"]])";
    auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_struct));
    ASSERT_TRUE(result_array->Equals(expected_with_rk))
        << "result_array:" << result_array->ToString() << std::endl
        << "expected:" << expected_with_rk->ToString();
}

TEST_P(BlobTableInteTest, TestReadBlobDescriptorFieldFromJava) {
    auto file_format = GetParam();
    if (file_format == "avro") {
        return;
    }
    std::string table_path =
        GetDataDir() + "/" + file_format + "/blob_desc_field.db/blob_desc_field";
    arrow::FieldVector fields = {
        arrow::field("f0", arrow::int32()), BlobUtils::ToArrowField("b0", true),
        BlobUtils::ToArrowField("b1", true), BlobUtils::ToArrowField("b2", true),
        BlobUtils::ToArrowField("b3", true)};
    auto schema = arrow::schema(fields);
    // b0: all non-null, b1: has nulls, b2: all non-null, b3: has nulls
    std::string raw_json = R"([
        [1, "img_0", null,    "raw_2_0", "raw_3_0"],
        [2, "img_1", "vid_1", "raw_2_1", null      ],
        [3, "img_2", null,    "raw_2_2", "raw_3_2" ]
    ])";
    auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());

    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
    std::map<std::string, std::string> read_options = {{Options::BLOB_AS_DESCRIPTOR, "false"}};
    ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), plan,
                                                /*predicate=*/nullptr, read_options));
    ASSERT_TRUE(result.chunked_array);
    auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie();
    auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);

    // After read, b0 and b1 are both descriptor-stored; resolve all back to raw bytes.
    // Java-generated descriptors may contain absolute paths from the generation machine.
    // Rewrite them to the portable blob directories inside the copied table path.
    BlobDescriptorPathRewrite rewrite{table_path, {"raw_blob", "external_blob"}};
    ASSERT_OK_AND_ASSIGN(auto resolved,
                         ConvertDescriptorToRawBlob(read_struct, {"b0", "b1"}, rewrite));
    ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(raw_array));
    ASSERT_TRUE(resolved->Equals(expected_with_rk));
}

}  // namespace paimon::test