/*
 * 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 <tuple>

#include "arrow/type.h"
#include "fmt/format.h"
#include "gtest/gtest.h"
#include "paimon/common/factories/io_hook.h"
#include "paimon/common/io/cache/lru_cache.h"
#include "paimon/common/table/special_fields.h"
#include "paimon/common/types/data_field.h"
#include "paimon/common/utils/date_time_utils.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/table/source/data_split_impl.h"
#include "paimon/defs.h"
#include "paimon/format/mosaic/mosaic_format_defs.h"
#include "paimon/fs/file_system.h"
#include "paimon/global_index/bitmap_global_index_result.h"
#include "paimon/global_index/indexed_split.h"
#include "paimon/predicate/literal.h"
#include "paimon/predicate/predicate_builder.h"
#include "paimon/result.h"
#include "paimon/status.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"
namespace paimon::test {
using DataEvolutionTableParam = std::tuple<std::string, bool>;

// This is a sdk end-to-end test for data evolution
class DataEvolutionTableTest : public ::testing::Test,
                               public ::testing::WithParamInterface<DataEvolutionTableParam> {
    void SetUp() override {
        dir_ = UniqueTestDirectory::Create("local");
        int64_t seed = DateTimeUtils::GetCurrentUTCTimeUs();
        std::srand(seed);
    }
    void TearDown() override {
        dir_.reset();
    }

    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) const {
        CreateTable(fields_, partition_keys, options);
    }

    void CreateTable(const std::vector<std::string>& partition_keys) const {
        std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                      {Options::FILE_FORMAT, FileFormat()},
                                                      {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>>> WriteArrays(
        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;
    }

    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::shared_ptr<arrow::Array>& write_array) const {
        return WriteArrays(table_path, partition, write_cols, {write_array});
    }

    Result<std::vector<std::shared_ptr<CommitMessage>>> WriteArray(
        const std::string& table_path, const std::vector<std::string>& write_cols,
        const std::shared_ptr<arrow::Array>& write_array) const {
        return WriteArray(table_path, /*partition=*/{}, write_cols, write_array);
    }

    /// Stamps `reset_first_row_id` on the metas the caller holds. The commit assigns row ids
    /// onto its own copies, so a caller that later looks a file up by row id range, to find a
    /// row range group's anchor, has to mirror the assignment here.
    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);
    }

    /// Creates the default (f0, f1, f2) data evolution table and returns the options it was
    /// created with, which a test that later evolves the schema has to pass along.
    /// `extra_options` is merged in for tests that also pin a read batch or split target size.
    std::map<std::string, std::string> CreateDataEvolutionTable(
        bool deletion_vectors_enabled,
        const std::map<std::string, std::string>& extra_options = {}) const {
        std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                      {Options::FILE_FORMAT, FileFormat()},
                                                      {Options::FILE_SYSTEM, "local"},
                                                      {Options::ROW_TRACKING_ENABLED, "true"},
                                                      {Options::DATA_EVOLUTION_ENABLED, "true"}};
        if (deletion_vectors_enabled) {
            options.emplace(Options::DELETION_VECTORS_ENABLED, "true");
        }
        options.insert(extra_options.begin(), extra_options.end());
        CreateTable(/*partition_keys=*/{}, options);
        return options;
    }

    /// 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=*/FileFormat(),
                base_commit_msg, deleted_positions_by_anchor, GetDefaultPool(),
                replaced_commit_msg));
        PAIMON_RETURN_NOT_OK(Commit(table_path, {commit_message}));
        return commit_message;
    }

    /// Writes one row range group holding `f0_values`, starting at `first_row_id`: a full-row
    /// write plus a partial f2 write over the same rows, so the group's split merges columns
    /// from two files and is not raw convertible. When `partition` is set the files land in that
    /// partition and every row carries its value in f1, the partition key.
    ///
    /// The f2 write starts only once the full-row write is committed, which is what makes the
    /// f2 file the newer of the pair. Writing both before either commit left the column merge
    /// free to serve f2 from the full-row write instead.
    Result<std::vector<std::shared_ptr<CommitMessage>>> WriteAndCommitGroup(
        const std::string& table_path, int64_t first_row_id, const std::vector<int32_t>& f0_values,
        const std::map<std::string, std::string>& partition = {}) const {
        auto partition_f1 = partition.find("f1");
        std::string base_json = "[";
        std::string f2_json = "[";
        for (size_t i = 0; i < f0_values.size(); i++) {
            if (i > 0) {
                base_json += ", ";
                f2_json += ", ";
            }
            std::string f1_value =
                partition_f1 == partition.end() ? fmt::format("a{}", i) : partition_f1->second;
            base_json += fmt::format(R"([{}, "{}", "x{}"])", f0_values[i], f1_value, i);
            f2_json += fmt::format(R"(["y{}"])", i);
        }
        base_json += "]";
        f2_json += "]";

        auto base_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), base_json)
                .ValueOrDie());
        PAIMON_ASSIGN_OR_RAISE(
            std::vector<std::shared_ptr<CommitMessage>> base_msgs,
            WriteArray(table_path, partition, arrow::schema(fields_)->field_names(), base_array));
        // both writes are stamped with the same first row id, so their files cover the same row
        // id range and form one row range group
        SetFirstRowId(first_row_id, base_msgs);
        PAIMON_RETURN_NOT_OK(Commit(table_path, base_msgs));

        arrow::FieldVector f2_fields = {fields_[2]};
        auto f2_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(f2_fields), f2_json)
                .ValueOrDie());
        PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<CommitMessage>> f2_msgs,
                               WriteArray(table_path, partition, {"f2"}, f2_array));
        SetFirstRowId(first_row_id, f2_msgs);
        PAIMON_RETURN_NOT_OK(Commit(table_path, f2_msgs));

        std::vector<std::shared_ptr<CommitMessage>> group_msgs;
        group_msgs.insert(group_msgs.end(), base_msgs.begin(), base_msgs.end());
        group_msgs.insert(group_msgs.end(), f2_msgs.begin(), f2_msgs.end());
        return group_msgs;
    }

    /// Flattens the f0 column of a read result, looked up by name so the row kind column the
    /// reader prepends does not shift it.
    static Result<std::vector<int32_t>> CollectF0Values(
        const std::shared_ptr<arrow::ChunkedArray>& rows) {
        std::vector<int32_t> values;
        if (!rows) {
            return values;
        }
        for (const std::shared_ptr<arrow::Array>& chunk : rows->chunks()) {
            auto struct_array = std::dynamic_pointer_cast<arrow::StructArray>(chunk);
            if (!struct_array) {
                return Status::Invalid("read result chunk is not a struct array");
            }
            auto f0_array =
                std::dynamic_pointer_cast<arrow::Int32Array>(struct_array->GetFieldByName("f0"));
            if (!f0_array) {
                return Status::Invalid("read result has no int32 f0 column");
            }
            for (int64_t i = 0; i < f0_array->length(); i++) {
                values.push_back(f0_array->Value(i));
            }
        }
        return values;
    }

    struct LimitScanResult {
        /// The splits the limit push down kept in the plan.
        std::vector<std::shared_ptr<Split>> splits;
        /// The rows reading those splits produced, null when the read returned nothing. The
        /// read does not truncate to the limit, so this is everything the kept splits hold, or
        /// everything that survives the predicate when `enable_predicate_filter` is set.
        std::shared_ptr<arrow::ChunkedArray> rows;
    };

    /// Scans with a pushed-down row limit and reads the planned splits back. The push down only
    /// prunes splits, so a correct plan must still expose at least `limit` rows to the read.
    ///
    /// `enable_predicate_filter` turns the predicate into a row filter while reading. It is off
    /// by default, matching the read context default: a predicate then only prunes while
    /// planning, and the read returns every row the kept splits hold.
    Result<LimitScanResult> ScanAndReadWithLimit(
        const std::string& table_path, const std::vector<std::string>& read_schema, int32_t limit,
        const std::shared_ptr<Predicate>& predicate = nullptr,
        const std::vector<Range>& row_ranges = {}, bool enable_predicate_filter = false) const {
        ScanContextBuilder scan_context_builder(table_path);
        scan_context_builder.SetLimit(limit);
        scan_context_builder.SetPredicate(predicate);
        if (!row_ranges.empty()) {
            scan_context_builder.SetGlobalIndexResult(
                BitmapGlobalIndexResult::FromRanges(row_ranges));
        }
        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ScanContext> scan_context,
                               FinishScanContext(scan_context_builder));
        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<TableScan> table_scan,
                               TableScan::Create(std::move(scan_context)));
        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Plan> result_plan, table_scan->CreatePlan());

        LimitScanResult result;
        result.splits = result_plan->Splits();
        ReadContextBuilder read_context_builder(table_path);
        read_context_builder.SetReadFieldNames(read_schema)
            .SetPredicate(predicate)
            .EnablePredicateFilter(enable_predicate_filter);
        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReadContext> read_context,
                               read_context_builder.Finish());
        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<TableRead> table_read,
                               TableRead::Create(std::move(read_context)));
        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BatchReader> batch_reader,
                               table_read->CreateReader(result.splits));
        PAIMON_ASSIGN_OR_RAISE(result.rows,
                               ReadResultCollector::CollectResult(std::move(batch_reader)));
        return result;
    }

    /// Plans the table without any push down and returns the planned splits, so a test can
    /// assert what the scan handed the read: which data file each deletion file landed on, and
    /// the row count derived from them. Reading the splits back is ScanAndReadWithLimit's job,
    /// which verifies the returned rows after their reader has been destroyed.
    Result<std::vector<std::shared_ptr<Split>>> PlanSplits(const std::string& table_path) const {
        ScanContextBuilder scan_context_builder(table_path);
        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ScanContext> scan_context,
                               FinishScanContext(scan_context_builder));
        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<TableScan> table_scan,
                               TableScan::Create(std::move(scan_context)));
        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Plan> plan, table_scan->CreatePlan());
        return plan->Splits();
    }

    /// Anchor file name of every row range group of the table, ordered by ascending first row
    /// id, derived from the planned splits.
    ///
    /// Deriving them from the same metas the read groups keeps the two in step. Deriving them
    /// from the metas a write hands back uses pre-commit copies, and a vector keyed by a file
    /// the read does not consider the anchor is never found, silently leaving rows undeleted.
    Result<std::vector<std::string>> PlannedAnchorFileNames(const std::string& table_path) const {
        PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<Split>> splits, PlanSplits(table_path));
        std::vector<std::shared_ptr<DataFileMeta>> data_files;
        for (const std::shared_ptr<Split>& split : splits) {
            auto split_impl = std::dynamic_pointer_cast<DataSplitImpl>(split);
            if (!split_impl) {
                return Status::Invalid("split cannot cast to DataSplitImpl");
            }
            for (const std::shared_ptr<DataFileMeta>& file : split_impl->DataFiles()) {
                data_files.push_back(file);
            }
        }
        return DeletionVectorTestHelper::RetrieveAnchorFileNames(data_files);
    }

    /// PlannedAnchorFileNames for a table holding a single row range group.
    Result<std::string> PlannedAnchorFileName(const std::string& table_path) const {
        PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> anchor_file_names,
                               PlannedAnchorFileNames(table_path));
        if (anchor_file_names.size() != 1) {
            return Status::Invalid("expected exactly one row range group in the table");
        }
        return anchor_file_names[0];
    }

    using DeletionCardinalityMap = std::map<std::string, int64_t>;

    /// Maps every data file of `split` that carries a deletion file to that file's cardinality.
    static Result<DeletionCardinalityMap> DeletionCardinalityByDataFile(
        const std::shared_ptr<Split>& split) {
        auto split_impl = std::dynamic_pointer_cast<DataSplitImpl>(split);
        if (!split_impl) {
            return Status::Invalid("split cannot cast to DataSplitImpl");
        }
        const std::vector<std::shared_ptr<DataFileMeta>>& data_files = split_impl->DataFiles();
        const std::vector<std::optional<DeletionFile>>& deletion_files =
            split_impl->DeletionFiles();
        if (!deletion_files.empty() && deletion_files.size() != data_files.size()) {
            return Status::Invalid("deletion files are not aligned with data files");
        }
        DeletionCardinalityMap cardinality_by_file;
        for (size_t i = 0; i < deletion_files.size(); i++) {
            if (deletion_files[i] == std::nullopt) {
                continue;
            }
            if (deletion_files[i].value().cardinality == std::nullopt) {
                return Status::Invalid("deletion file is missing its cardinality");
            }
            cardinality_by_file[data_files[i]->file_name] =
                deletion_files[i].value().cardinality.value();
        }
        return cardinality_by_file;
    }

    Status CommitWithRowIdCheckFromSnapshot(
        const std::string& table_path,
        const std::vector<std::shared_ptr<CommitMessage>>& commit_msgs,
        std::optional<int64_t> row_id_check_from_snapshot) const {
        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)));
        file_store_commit->RowIdCheckConflict(row_id_check_from_snapshot);
        return file_store_commit->Commit(commit_msgs);
    }

    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 = {},
                       bool check_scan_plan_when_empty_result = true,
                       bool apply_predicate_to_scan = true) const {
        // scan
        ScanContextBuilder scan_context_builder(table_path);
        scan_context_builder.SetPredicate(apply_predicate_to_scan ? predicate : nullptr);
        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, FinishScanContext(scan_context_builder));
        PAIMON_ASSIGN_OR_RAISE(auto table_scan, TableScan::Create(std::move(scan_context)));
        PAIMON_ASSIGN_OR_RAISE(auto result_plan, table_scan->CreatePlan());
        if (!expected_array && check_scan_plan_when_empty_result) {
            if (!result_plan->Splits().empty()) {
                return Status::Invalid("check_scan_plan_when_empty_result but plan is not empty");
            }
        }

        // read
        auto splits = result_plan->Splits();
        ReadContextBuilder read_context_builder(table_path);
        read_context_builder.SetReadFieldNames(read_schema).SetPredicate(predicate);
        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(std::move(batch_reader)));

        if (!expected_array) {
            if (read_result) {
                return Status::Invalid("expected array is empty, but read result is not empty");
            }
            return Status::OK();
        }
        // add row kind array for expected 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, expected_array->length()));

        arrow::ArrayVector expected_with_row_kind_fields = expected_array->fields();
        std::vector<std::string> expected_with_row_kind_field_names =
            arrow::schema(expected_array->type()->fields())->field_names();
        expected_with_row_kind_fields.insert(expected_with_row_kind_fields.begin(), row_kind_array);
        expected_with_row_kind_field_names.insert(expected_with_row_kind_field_names.begin(),
                                                  "_VALUE_KIND");

        // check read result
        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
            auto expected_with_row_kind_array,
            arrow::StructArray::Make(expected_with_row_kind_fields,
                                     expected_with_row_kind_field_names));
        auto expected_chunk_array =
            std::make_shared<arrow::ChunkedArray>(expected_with_row_kind_array);
        if (!expected_chunk_array->Equals(read_result)) {
            std::cout << "result=" << read_result->ToString() << std::endl
                      << "expected=" << expected_chunk_array->ToString() << std::endl;
            return Status::Invalid("expected array and result array not equal");
        }
        return Status::OK();
    }

    void CheckScanResult(const std::string& table_path, const std::shared_ptr<Predicate>& predicate,
                         const std::vector<Range>& row_ranges,
                         const std::vector<std::optional<int64_t>>& expected_first_row_ids,
                         const std::vector<int64_t>& expected_row_counts) {
        ASSERT_EQ(expected_first_row_ids.size(), expected_row_counts.size());
        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);
        }
        ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder));
        ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context)));
        ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan());
        const auto& result_splits = result_plan->Splits();
        if (expected_first_row_ids.empty()) {
            ASSERT_EQ(result_splits.size(), 0);
            return;
        }
        ASSERT_EQ(result_splits.size(), 1);
        std::shared_ptr<DataSplitImpl> data_split;
        if (auto indexed_split = std::dynamic_pointer_cast<IndexedSplit>(result_splits[0])) {
            data_split = std::dynamic_pointer_cast<DataSplitImpl>(indexed_split->GetDataSplit());
        } else {
            data_split = std::dynamic_pointer_cast<DataSplitImpl>(result_splits[0]);
        }
        ASSERT_TRUE(data_split);
        std::vector<std::optional<int64_t>> result_first_row_ids;
        std::vector<int64_t> result_row_counts;
        for (const auto& meta : data_split->DataFiles()) {
            result_first_row_ids.push_back(meta->first_row_id);
            result_row_counts.push_back(meta->row_count);
        }
        ASSERT_EQ(result_first_row_ids, expected_first_row_ids);
        ASSERT_EQ(result_row_counts, expected_row_counts);
    }

    Result<std::unique_ptr<ScanContext>> FinishScanContext(ScanContextBuilder& builder) const {
        if (EnableSnapshotLiveManifestCache()) {
            if (!snapshot_live_manifest_cache_) {
                snapshot_live_manifest_cache_ =
                    std::make_shared<LruCache>(/*max_weight=*/64 * 1024 * 1024);
            }
            builder.AddOption(Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS, "3")
                .WithCache(snapshot_live_manifest_cache_);
        }
        return builder.Finish();
    }

    std::string FileFormat() const {
        return std::get<0>(GetParam());
    }

    bool EnableSnapshotLiveManifestCache() const {
        return std::get<1>(GetParam());
    }

    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());
    }

 private:
    std::unique_ptr<UniqueTestDirectory> dir_;
    mutable std::shared_ptr<Cache> snapshot_live_manifest_cache_;
    arrow::FieldVector fields_ = {
        arrow::field("f0", arrow::int32()),
        arrow::field("f1", arrow::utf8()),
        arrow::field("f2", arrow::utf8()),
    };
};

TEST_P(DataEvolutionTableTest, 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: 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));
    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));

    // 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"([
        ["a", 1, 2, 0, "c"]
    ])")
            .ValueOrDie());

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

    // read score but not indexed split
    ASSERT_NOK_WITH_MSG(
        ScanAndRead(table_path, {"f0", "f1", "_INDEX_SCORE"}, expected_row_tracking_array,
                    /*predicate=*/nullptr,
                    /*row_ranges=*/{}),
        "Invalid read schema, read _INDEX_SCORE while split cannot cast to IndexedSplit");
}

TEST_P(DataEvolutionTableTest, TestCommitConflictOnOverlappedRowIdAndWriteColumns) {
    CreateTable();
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    // Snapshot 1: initialize row id range [0, 0].
    std::vector<std::string> init_write_cols = {"f0", "f1", "f2"};
    auto init_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "a", "b"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto init_msgs, WriteArray(table_path, init_write_cols, init_array));
    ASSERT_OK(Commit(table_path, init_msgs));

    // Snapshot 2: update f2 at row id 0.
    std::vector<std::string> write_cols = {"f2"};
    auto src_array_1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
        ["c"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs_1, WriteArray(table_path, write_cols, src_array_1));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs_1);
    ASSERT_OK(Commit(table_path, commit_msgs_1));

    // Snapshot 3 attempt: update f2 at row id 0 again, and check history from snapshot 1.
    // This should conflict with snapshot 2 because row-id range and write columns overlap.
    auto src_array_2 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
        ["d"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs_2, WriteArray(table_path, write_cols, src_array_2));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs_2);
    ASSERT_NOK_WITH_MSG(
        CommitWithRowIdCheckFromSnapshot(table_path, commit_msgs_2,
                                         /*row_id_check_from_snapshot=*/1),
        "multiple 'MERGE INTO' operations have encountered conflicts, updating the same file");
}

TEST_P(DataEvolutionTableTest, 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));
    }
    {
        std::vector<Range> row_ranges = {Range(0l, 0l), Range(11l, 11l)};
        // test with row ids
        CheckScanResult(table_path, /*predicate=*/nullptr,
                        /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{0, 11, 11}, /*expected_row_counts=*/{10, 1, 1});
        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,
                              /*predicate=*/nullptr,
                              /*row_ranges=*/row_ranges));
    }
    // 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(DataEvolutionTableTest, TestOnlySomeColumns) {
    CreateTable();
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    // write field: f0
    std::vector<std::string> write_cols0 = {"f0"};
    auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0]}), R"([
        [1]
    ])")
            .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
    std::vector<std::string> write_cols1 = {"f1"};
    auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1]}), R"([
        ["a"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, write_cols1, src_array1));
    SetFirstRowId(0, commit_msgs1);
    ASSERT_OK(Commit(table_path, 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);
    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", "b"]
    ])")
            .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, 3]
    ])")
            .ValueOrDie());

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

TEST_P(DataEvolutionTableTest, TestMultipleSharedShreddingMapsPartialOverwrite) {
    if (FileFormat() == "mosaic") {
        return;
    }
    if (FileFormat() == "avro") {
        return;
    }

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

    std::vector<std::string> write_cols0 = {"id", "map1"};
    auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[0], fields[1]}), R"([
        [1, [["a", 10], ["b", 20]]],
        [11, [["a", 11], ["b", 21]]]
    ])")
            .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> write_cols1 = {"id", "map2"};
    auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[0], fields[2]}), R"([
        [2, [["c", 30], ["d", 40]]],
        [12, [["c", 31], ["d", 41]]]
    ])")
            .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));

    std::vector<std::string> write_cols2 = {"map1"};
    auto src_array2 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[1]}), R"([
        [[["b", 200], ["a", 100]]],
        [[["b", 201], ["a", 101]]]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs2, WriteArray(table_path, write_cols2, src_array2));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs2);
    ASSERT_OK(Commit(table_path, commit_msgs2));

    // Read all columns and merge values from all partial files.
    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
        [2, [["a", 100], ["b", 200]], [["c", 30], ["d", 40]]],
        [12, [["a", 101], ["b", 201]], [["c", 31], ["d", 41]]]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));

    // Read a subset of columns and recall only the requested shared-shredding MAP column.
    auto expected_column_pruned_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[0], fields[2]}), R"([
        [2, [["c", 30], ["d", 40]]],
        [12, [["c", 31], ["d", 41]]]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"id", "map2"}, expected_column_pruned_array));

    // Read selected keys from both shared-shredding MAP columns after partial overwrite merge.
    {
        auto map1_selected_keys =
            arrow::KeyValueMetadata::Make({DataField::MAP_SELECTED_KEYS}, {"b"});
        auto map2_selected_keys =
            arrow::KeyValueMetadata::Make({DataField::MAP_SELECTED_KEYS}, {"d"});
        auto read_schema = arrow::schema({
            fields[0],
            fields[1]->WithMetadata(map1_selected_keys),
            fields[2]->WithMetadata(map2_selected_keys),
        });
        auto c_schema = std::make_unique<ArrowSchema>();
        ASSERT_TRUE(arrow::ExportSchema(*read_schema, c_schema.get()).ok());

        ScanContextBuilder scan_context_builder(table_path);
        ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder));
        ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context)));
        ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan());

        ReadContextBuilder read_context_builder(table_path);
        read_context_builder.SetReadSchema(std::move(c_schema));
        ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context,
                             read_context_builder.Finish());
        ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context)));
        ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits()));
        ASSERT_OK_AND_ASSIGN(auto actual,
                             ReadResultCollector::CollectResult(std::move(batch_reader)));

        auto expected_type = arrow::struct_({
            SpecialFields::ValueKind().field_,
            fields[0],
            fields[1],
            fields[2],
        });
        auto expected = arrow::ipc::internal::json::ArrayFromJSON(expected_type, R"([
            [0, 2, [["b", 200]], [["d", 40]]],
            [0, 12, [["b", 201]], [["d", 41]]]
        ])")
                            .ValueOrDie();
        auto expected_chunked = std::make_shared<arrow::ChunkedArray>(expected);
        ASSERT_TRUE(expected_chunked->Equals(actual))
            << "actual=" << actual->ToString() << "\nexpected=" << expected_chunked->ToString();
    }

    // Read a subset of rows after merging values from all partial files.
    auto expected_partial_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
        [12, [["a", 101], ["b", 201]], [["c", 31], ["d", 41]]]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_partial_array,
                          /*predicate=*/nullptr,
                          /*row_ranges=*/{Range(1l, 1l)}));

    // Read row tracking fields and verify the latest partial overwrite sequence number.
    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"([
        [2, [["a", 100], ["b", 200]], [["c", 30], ["d", 40]], 0, 3],
        [12, [["a", 101], ["b", 201]], [["c", 31], ["d", 41]], 1, 3]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"id", "map1", "map2", "_ROW_ID", "_SEQUENCE_NUMBER"},
                          expected_row_tracking_array));
}

TEST_P(DataEvolutionTableTest, TestNullValues) {
    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"};
    auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
        [1, null]
    ])")
            .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));

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

    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, null, "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_[0],
                                                      fields_[1],
                                                      fields_[2],
                                                      SpecialFields::RowId().field_,
                                                      SpecialFields::SequenceNumber().field_,
                                                  }),
                                                  R"([
        [1, null, "c", 0, 2]
    ])")
            .ValueOrDie());

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

TEST_P(DataEvolutionTableTest, 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(DataEvolutionTableTest, TestMoreData) {
    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(DataEvolutionTableTest, TestOnlyRowTrackingEnabled) {
    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, FileFormat()},
        {Options::FILE_SYSTEM, "local"},
        {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "false"},
    };
    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_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"],
        [2, "c", "d"]
    ])")
            .ValueOrDie());

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

    // 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"([
        ["a", 1, 1, 0, "b"],
        ["c", 2, 1, 1, "d"]
    ])")
            .ValueOrDie());

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

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

    // write field: f0, f1, f2 for f1=2024
    std::map<std::string, std::string> partition0 = {{"f1", "2024"}};
    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, "2024", "b1"],
        [2, "2024", "b2"],
        [3, "2024", "b3"]
    ])")
            .ValueOrDie());

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

    // write field: f2 for f1=2024
    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"([
        ["c1"],
        ["c2"],
        ["c3"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(commit_msgs, WriteArray(table_path, partition0, write_cols1, src_array1));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs);
    ASSERT_OK(Commit(table_path, commit_msgs));

    // write fields: f1, f2 for f1=2025
    std::map<std::string, std::string> partition2 = {{"f1", "2025"}};
    std::vector<std::string> write_cols2 = {"f1", "f2"};
    auto src_array2 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1], fields_[2]}), R"([
        ["2025", "d1"],
        ["2025", "d2"],
        ["2025", "d3"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(commit_msgs, WriteArray(table_path, partition2, write_cols2, src_array2));
    SetFirstRowId(/*reset_first_row_id=*/3, commit_msgs);
    ASSERT_OK(Commit(table_path, commit_msgs));

    // test read all fields
    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "2024", "c1"],
        [2, "2024", "c2"],
        [3, "2024", "c3"],
        [null, "2025", "d1"],
        [null, "2025", "d2"],
        [null, "2025", "d3"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));

    // test only read partition fields
    auto expected_array_only_partition = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1]}), R"([
        ["2024"],
        ["2024"],
        ["2024"],
        ["2025"],
        ["2025"],
        ["2025"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"f1"}, expected_array_only_partition));

    // 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, "2024", "c1", 0, 2],
        [2, "2024", "c2", 1, 2],
        [3, "2024", "c3", 2, 2],
        [null, "2025", "d1", 3, 3],
        [null, "2025", "d2", 4, 3],
        [null, "2025", "d3", 5, 3]
    ])")
            .ValueOrDie());

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

    // read only read partition fields and row tracking
    auto expected_partition_row_tracking_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(
            arrow::struct_({fields_[1], SpecialFields::RowId().field_,
                            SpecialFields::SequenceNumber().field_}),
            R"([
        ["2024", 0, 2],
        ["2024", 1, 2],
        ["2024", 2, 2],
        ["2025", 3, 3],
        ["2025", 4, 3],
        ["2025", 5, 3]
    ])")
            .ValueOrDie());

    ASSERT_OK(ScanAndRead(table_path, {"f1", "_ROW_ID", "_SEQUENCE_NUMBER"},
                          expected_partition_row_tracking_array));
}

TEST_P(DataEvolutionTableTest, TestWithPartitionWithoutPartitionFieldsInFile) {
    std::vector<std::string> partition_keys = {"f1"};
    CreateTable(partition_keys);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    // write field: f0, f2 for f1=2024
    std::map<std::string, std::string> partition0 = {{"f1", "2024"}};
    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, "b1"],
        [2, "b2"],
        [3, "b3"]
    ])")
            .ValueOrDie());

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

    // write field: f2 for f1=2024
    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"([
        ["c1"],
        ["c2"],
        ["c3"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(commit_msgs, WriteArray(table_path, partition0, 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, "2024", "c1"],
        [2, "2024", "c2"],
        [3, "2024", "c3"]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
    }
    {
        // test with row ids
        std::vector<Range> row_ranges = {Range(1l, 1l)};
        CheckScanResult(table_path, /*predicate=*/nullptr, /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{0, 0}, /*expected_row_counts=*/{3, 3});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [2, "2024", "c2"]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
                              /*predicate=*/nullptr,
                              /*row_ranges=*/row_ranges));
    }

    // 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, "2024", "c1", 0, 2],
        [2, "2024", "c2", 1, 2],
        [3, "2024", "c3", 2, 2]
    ])")
            .ValueOrDie());

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

TEST_P(DataEvolutionTableTest, TestPartitionWithPredicate) {
    auto file_format = FileFormat();
    if (file_format == "avro") {
        return;
    }
    std::vector<std::string> partition_keys = {"f1"};
    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},         {Options::FILE_FORMAT, FileFormat()},
        {Options::FILE_SYSTEM, "local"},           {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "true"}, {"parquet.write.max-row-group-length", "1"}};
    if (file_format == "mosaic") {
        options.emplace(mosaic::MOSAIC_STATS_COLUMNS, "f0");
    }

    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 f1 = "2024"
    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"],
        [12, "2024"],
        [13, "2024"]
    ])")
            .ValueOrDie());

    ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
                         WriteArray(table_path, {{"f1", "2024"}}, write_cols0, src_array0));

    // write field: f2 for partition f1 = "2024"
    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, {{"f1", "2024"}}, 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 f1 = "2025"
    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"([
        [21, "2025"],
        [22, "2025"],
        [23, "2025"]
    ])")
            .ValueOrDie());

    ASSERT_OK_AND_ASSIGN(auto commit_msgs3,
                         WriteArray(table_path, {{"f1", "2025"}}, write_cols3, src_array3));
    SetFirstRowId(3, commit_msgs3);
    ASSERT_OK(Commit(table_path, commit_msgs3));
    {
        // only set data field predicate, predicate only takes effective in file skip
        // read will not push down
        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"],
        [12, "2024", "b"],
        [13, "2024", "c"]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array, equal));
    }
    {
        // only set partition predicate
        auto equal =
            PredicateBuilder::Equal(/*field_index=*/1, /*field_name=*/"f1", FieldType::STRING,
                                    Literal(FieldType::STRING, "2024", 4));
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [11, "2024", "a"],
        [12, "2024", "b"],
        [13, "2024", "c"]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array, equal));
    }
    {
        // set partition predicate and data field predicate
        auto equal =
            PredicateBuilder::Equal(/*field_index=*/1, /*field_name=*/"f1", FieldType::STRING,
                                    Literal(FieldType::STRING, "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_OK(
            ScanAndRead(table_path, schema->field_names(), /*expected_array=*/nullptr, predicate));
    }
    {
        // read with row tracking
        auto equal =
            PredicateBuilder::Equal(/*field_index=*/1, /*field_name=*/"f1", FieldType::STRING,
                                    Literal(FieldType::STRING, "2024", 4));

        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],
        [12, "2024", "b", 1, 1],
        [13, "2024", "c", 2, 1]
    ])")
                .ValueOrDie());

        ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2", "_ROW_ID", "_SEQUENCE_NUMBER"},
                              expected_row_tracking_array, equal));
    }
    {
        // test with row ids
        std::vector<Range> row_ranges = {Range(1l, 1l)};
        CheckScanResult(table_path, /*predicate=*/nullptr, /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{0, 0}, /*expected_row_counts=*/{3, 3});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [12, "2024", "b"]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
                              /*predicate=*/nullptr,
                              /*row_ranges=*/row_ranges));
    }
    {
        std::vector<Range> row_ranges = {Range(1l, 1l)};
        // test with row ids and partition predicate
        auto equal =
            PredicateBuilder::Equal(/*field_index=*/1, /*field_name=*/"f1", FieldType::STRING,
                                    Literal(FieldType::STRING, "2025", 4));
        CheckScanResult(table_path, /*predicate=*/equal, /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{}, /*expected_row_counts=*/{});
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), /*expected_array=*/nullptr,
                              /*predicate=*/equal,
                              /*row_ranges=*/row_ranges));
    }
    {
        std::vector<Range> row_ranges = {Range(1l, 1l)};
        // test with row ids and non-partition predicate
        auto equal = PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::INT,
                                             Literal(11));
        CheckScanResult(table_path, /*predicate=*/equal, /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{0, 0}, /*expected_row_counts=*/{3, 3});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [12, "2024", "b"]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
                              /*predicate=*/equal,
                              /*row_ranges=*/row_ranges));
    }
    {
        std::vector<Range> row_ranges = {Range(1l, 1l)};
        // test with row ids and data predicate
        auto equal = PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::INT,
                                             Literal(50));
        CheckScanResult(table_path, /*predicate=*/equal, /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{}, /*expected_row_counts=*/{});
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), /*expected_array=*/nullptr,
                              /*predicate=*/equal,
                              /*row_ranges=*/row_ranges));
    }
}

TEST_P(DataEvolutionTableTest, TestAlterTable) {
    auto file_format = FileFormat();
    if (file_format == "mosaic") {
        return;
    }
    if (file_format == "avro") {
        return;
    }
    std::string table_path = paimon::test::GetDataDir() + file_format +
                             "/append_table_alter_table_with_cast_with_data_evolution.db/"
                             "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, arrow::field("f6", arrow::int32())),
        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"([
["1970-01-05T00:00", 0, 1, 100, "2024-11-26 06:38:56.001000001", "0.020", true, null, 0, 1],
["1969-11-18T00:00", 0, 1, 110, "2024-11-26 06:38:56.011000011", "11.120", true, null, 1, 1],
["1971-03-21T00:00", 0, 1, 120, "2024-11-26 06:38:56.021000021", "22.220", false, null, 2, 1],
["1957-11-01T00:00", 0, 1, 130, "2024-11-26 06:38:56.031000031", "333.320", false, null, 3, 1],
["2091-09-07T00:00", 0, 1, 140, "2024-11-26 06:38:56.041000041", "444.420", true, null, 4, 1],
["2024-11-26T06:38:56.054000154", 0, 1, 150, "2024-11-26 15:28:31", "55.002", true, 56, 5, 3],
["2024-11-26T06:38:56.064000164", 0, 1, 160, "2024-11-26 15:28:41", "666.012", false, 66, 6, 3],
["2024-11-26T06:38:56.074000174", 0, 1, 170, "2024-11-26 15:28:51", "-77.022", true, 76, 7, 3],
["2024-11-26T06:38:56.084000184", 0, 1, 180, "2024-11-26 15:29:01", "8.032", true, -86, 8, 3],
["2024-11-26T06:38:56.094000194", 0, 1, 190, "I'm strange", "-999.420", false, 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, null, 0, 1],
["1969-11-18T00:00", 0, 1, 110, "2024-11-26 06:38:56.011000011", "11.120", true, null, 1, 1],
["1971-03-21T00:00", 0, 1, 120, "2024-11-26 06:38:56.021000021", "22.220", false, null, 2, 1],
["1957-11-01T00:00", 0, 1, 130, "2024-11-26 06:38:56.031000031", "333.320", false, null, 3, 1],
["2091-09-07T00:00", 0, 1, 140, "2024-11-26 06:38:56.041000041", "444.420", true, 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=*/7, /*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, null, 0, 1],
["1969-11-18T00:00", 0, 1, 110, "2024-11-26 06:38:56.011000011", "11.120", true, null, 1, 1],
["1971-03-21T00:00", 0, 1, 120, "2024-11-26 06:38:56.021000021", "22.220", false, null, 2, 1],
["1957-11-01T00:00", 0, 1, 130, "2024-11-26 06:38:56.031000031", "333.320", false, null, 3, 1],
["2091-09-07T00:00", 0, 1, 140, "2024-11-26 06:38:56.041000041", "444.420", true, null, 4, 1],
["2024-11-26T06:38:56.054000154", 0, 1, 150, "2024-11-26 15:28:31", "55.002", true, 56, 5, 3],
["2024-11-26T06:38:56.064000164", 0, 1, 160, "2024-11-26 15:28:41", "666.012", false, 66, 6, 3],
["2024-11-26T06:38:56.074000174", 0, 1, 170, "2024-11-26 15:28:51", "-77.022", true, 76, 7, 3],
["2024-11-26T06:38:56.084000184", 0, 1, 180, "2024-11-26 15:29:01", "8.032", true, -86, 8, 3],
["2024-11-26T06:38:56.094000194", 0, 1, 190, "I'm strange", "-999.420", false, 96, 9, 3]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array, predicate));
    }
    {
        std::vector<Range> row_ranges = {Range(1l, 1l)};
        // test with row ids
        CheckScanResult(table_path, /*predicate=*/nullptr, /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{0, 0}, /*expected_row_counts=*/{5, 5});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
["1969-11-18T00:00", 0, 1, 110, "2024-11-26 06:38:56.011000011", "11.120", true, null, 1, 1]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array,
                              /*predicate=*/nullptr,
                              /*row_ranges=*/row_ranges));
    }
}

TEST_P(DataEvolutionTableTest, TestReadCompactFiles) {
    if (FileFormat() == "mosaic") {
        return;
    }
    auto file_format = FileFormat();
    if (file_format == "avro") {
        return;
    }
    std::string table_path =
        paimon::test::GetDataDir() + file_format +
        "/append_table_row_tracking_with_compact.db/append_table_row_tracking_with_compact";
    std::vector<DataField> read_fields = {DataField(0, arrow::field("f0", arrow::utf8())),
                                          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, 12, 2.1, 0, 1],
        ["Alice", 3, 13, 3.1, 1, 1],
        ["Bob", 4, 14, 4.1, 2, 2],
        ["David", 5, 15, 5.1, 3, 2]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                          expected_array));
}

TEST_P(DataEvolutionTableTest, TestReadTableWithDenseStats) {
    if (FileFormat() == "mosaic") {
        return;
    }
    auto file_format = FileFormat();
    if (file_format == "avro") {
        return;
    }
    std::string table_path = paimon::test::GetDataDir() + file_format +
                             "/data_evolution_with_dense_stats.db/data_evolution_with_dense_stats";
    std::vector<DataField> read_fields = {DataField(0, arrow::field("f0", arrow::utf8())),
                                          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=*/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 row id with predicate
        std::vector<Range> row_ranges = {Range(0l, 0l)};
        auto predicate = PredicateBuilder::Equal(/*field_index=*/1, /*field_name=*/"f1",
                                                 FieldType::INT, Literal(5));
        CheckScanResult(table_path, /*predicate=*/predicate, /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{}, /*expected_row_counts=*/{});
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              /*expected_array=*/nullptr, predicate,
                              /*row_ranges=*/row_ranges));
    }
    {
        // test row id with predicate
        std::vector<Range> row_ranges = {Range(0l, 0l)};
        auto predicate = PredicateBuilder::GreaterThan(/*field_index=*/3, /*field_name=*/"f3",
                                                       FieldType::DOUBLE, Literal(5.1));
        CheckScanResult(table_path, /*predicate=*/predicate, /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{0, 0}, /*expected_row_counts=*/{2, 2});
        auto expected_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
        ["Lily", 2, 102, 2.1, 0, 2]
    ])")
                .ValueOrDie());

        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array0, predicate,
                              /*row_ranges=*/row_ranges));
    }
}

TEST_P(DataEvolutionTableTest, TestScanAndReadWithIndex) {
    if (FileFormat() == "mosaic") {
        return;
    }
    auto file_format = FileFormat();
    if (file_format == "avro") {
        return;
    }
    // only f2 has index
    std::string table_path = paimon::test::GetDataDir() + file_format +
                             "/data_evolution_with_index.db/data_evolution_with_index";
    std::vector<DataField> read_fields = {DataField(0, arrow::field("f0", arrow::utf8())),
                                          DataField(1, arrow::field("f1", arrow::int32())),
                                          DataField(2, arrow::field("f2", arrow::int32())),
                                          DataField(3, arrow::field("f3", arrow::float64()))};

    std::shared_ptr<arrow::DataType> arrow_data_type =
        DataField::ConvertDataFieldsToArrowStructType(read_fields);
    {
        // first 4 records are two file with the same first row id with data evolution
        // last 2 rows only in one file
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
        ["Lily", 2, 102, 2.1],
        ["Alice", 4, 104, 3.1],
        ["Bob", 6, 106, 4.1],
        ["David", 8, 108, 5.1],
        [null, null, 202, 6.1],
        [null, null, 204, 7.1]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array));
    }
    {
        // The old file's f2 index does not contain 102, but the newer file owns f2.
        auto predicate = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2",
                                                 FieldType::INT, Literal(102));
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
        ["Lily", 2, 102, 2.1],
        ["Alice", 4, 104, 3.1],
        ["Bob", 6, 106, 4.1],
        ["David", 8, 108, 5.1]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array, predicate,
                              /*row_ranges=*/{},
                              /*check_scan_plan_when_empty_result=*/true,
                              /*apply_predicate_to_scan=*/false));
    }
    {
        // The bitmap proves that neither row range group contains f2 = 103.
        auto predicate = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2",
                                                 FieldType::INT, Literal(103));
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              /*expected_array=*/nullptr, predicate,
                              /*row_ranges=*/{},
                              /*check_scan_plan_when_empty_result=*/false,
                              /*apply_predicate_to_scan=*/false));
    }
    {
        // Scan planning keeps the split, but reader-side indexes skip both row range groups.
        auto predicate = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2",
                                                 FieldType::INT, Literal(203));
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              /*expected_array=*/nullptr, predicate,
                              /*row_ranges=*/{},
                              /*check_scan_plan_when_empty_result=*/false,
                              /*apply_predicate_to_scan=*/false));
    }
    {
        // A single-file group applies the exact bitmap row selection.
        auto predicate = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2",
                                                 FieldType::INT, Literal(202));
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
        [null, null, 202, 6.1]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array, predicate,
                              /*row_ranges=*/{},
                              /*check_scan_plan_when_empty_result=*/true,
                              /*apply_predicate_to_scan=*/false));
    }
    {
        auto predicate =
            PredicateBuilder::IsNull(/*field_index=*/0, /*field_name=*/"f0", FieldType::STRING);
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
        [null, null, 202, 6.1],
        [null, null, 204, 7.1]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array, predicate));
    }
    {
        // test row id with predicate
        std::vector<Range> row_ranges = {Range(0l, 2l)};
        // A merged group keeps all selected row ids to preserve column alignment.
        auto predicate = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2",
                                                 FieldType::INT, Literal(106));
        CheckScanResult(table_path, /*predicate=*/predicate, /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{0, 0}, /*expected_row_counts=*/{4, 4});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
        ["Lily", 2, 102, 2.1],
        ["Alice", 4, 104, 3.1],
        ["Bob", 6, 106, 4.1]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array, predicate,
                              /*row_ranges=*/row_ranges,
                              /*check_scan_plan_when_empty_result=*/true,
                              /*apply_predicate_to_scan=*/false));
    }
    {
        // test row id with predicate
        std::vector<Range> row_ranges = {Range(4l, 5l)};
        // The single-file bitmap selection is intersected with the row-id selection.
        auto predicate = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2",
                                                 FieldType::INT, Literal(204));
        CheckScanResult(table_path, /*predicate=*/predicate, /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{4}, /*expected_row_counts=*/{2});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
        [null, null, 204, 7.1]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
                              expected_array, predicate,
                              /*row_ranges=*/row_ranges,
                              /*check_scan_plan_when_empty_result=*/true,
                              /*apply_predicate_to_scan=*/false));
    }
}

TEST_P(DataEvolutionTableTest, TestDataEvolutionPredicatePushDownBoundaries) {
    if (FileFormat() == "mosaic") {
        return;
    }
    auto file_format = FileFormat();
    if (file_format == "avro") {
        return;
    }
    std::string table_path = paimon::test::GetDataDir() + file_format +
                             "/data_evolution_with_index.db/data_evolution_with_index";

    {
        // A file without f0 must not interpret the predicate as f0 = null.
        auto predicate = PredicateBuilder::Equal(
            /*field_index=*/0, /*field_name=*/"f0", FieldType::STRING,
            Literal(FieldType::STRING, "Lily", 4));
        auto read_type =
            arrow::struct_({arrow::field("f0", arrow::utf8()), arrow::field("f2", arrow::int32())});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(read_type, R"([
        ["Lily", 102],
        ["Alice", 104],
        ["Bob", 106],
        ["David", 108],
        [null, 202],
        [null, 204]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, {"f0", "f2"}, expected_array, predicate,
                              /*row_ranges=*/{},
                              /*check_scan_plan_when_empty_result=*/true,
                              /*apply_predicate_to_scan=*/false));
    }
    {
        // System fields are completed after reading and cannot be pushed into data files.
        auto predicate = PredicateBuilder::Equal(/*field_index=*/1, /*field_name=*/"_ROW_ID",
                                                 FieldType::BIGINT, Literal(99l));
        auto read_type =
            arrow::struct_({arrow::field("f2", arrow::int32()), SpecialFields::RowId().field_});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(read_type, R"([
        [102, 0],
        [104, 1],
        [106, 2],
        [108, 3],
        [202, 4],
        [204, 5]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, {"f2", "_ROW_ID"}, expected_array, predicate,
                              /*row_ranges=*/{},
                              /*check_scan_plan_when_empty_result=*/true,
                              /*apply_predicate_to_scan=*/false));
    }
    {
        // Dropping a system-field conjunct must not drop a pushable data conjunct.
        auto data_predicate = PredicateBuilder::Equal(
            /*field_index=*/0, /*field_name=*/"f2", FieldType::INT, Literal(103));
        auto system_predicate = PredicateBuilder::Equal(
            /*field_index=*/1, /*field_name=*/"_ROW_ID", FieldType::BIGINT, Literal(0l));
        ASSERT_OK_AND_ASSIGN(auto predicate,
                             PredicateBuilder::And({data_predicate, system_predicate}));
        ASSERT_OK(ScanAndRead(table_path, {"f2", "_ROW_ID"}, /*expected_array=*/nullptr, predicate,
                              /*row_ranges=*/{},
                              /*check_scan_plan_when_empty_result=*/false,
                              /*apply_predicate_to_scan=*/false));
    }
    {
        // Bitmap positions compose with row ranges without changing the physical row id.
        auto predicate = PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f2",
                                                 FieldType::INT, Literal(204));
        auto read_type =
            arrow::struct_({arrow::field("f2", arrow::int32()), SpecialFields::RowId().field_});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(read_type, R"([
        [204, 5]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, {"f2", "_ROW_ID"}, expected_array, predicate,
                              /*row_ranges=*/{Range(4l, 5l)},
                              /*check_scan_plan_when_empty_result=*/true,
                              /*apply_predicate_to_scan=*/false));
    }
    {
        // The bitmap selects row id 5 while the global-index selection keeps only row id 4.
        auto predicate = PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f2",
                                                 FieldType::INT, Literal(204));
        ASSERT_OK(ScanAndRead(table_path, {"f2", "_ROW_ID"}, /*expected_array=*/nullptr, predicate,
                              /*row_ranges=*/{Range(4l, 4l)},
                              /*check_scan_plan_when_empty_result=*/false,
                              /*apply_predicate_to_scan=*/false));
    }
    {
        // The predicate keeps row ids {4, 5}; the global-index selection keeps {0, 1, 2, 3, 4}.
        auto equal_202 = PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f2",
                                                 FieldType::INT, Literal(202));
        auto equal_204 = PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f2",
                                                 FieldType::INT, Literal(204));
        ASSERT_OK_AND_ASSIGN(auto predicate, PredicateBuilder::Or({equal_202, equal_204}));
        auto read_type =
            arrow::struct_({arrow::field("f2", arrow::int32()), SpecialFields::RowId().field_});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(read_type, R"([
        [202, 4]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, {"f2", "_ROW_ID"}, expected_array, predicate,
                              /*row_ranges=*/{Range(0l, 4l)},
                              /*check_scan_plan_when_empty_result=*/true,
                              /*apply_predicate_to_scan=*/false));
    }
}

TEST_P(DataEvolutionTableTest, TestFormatPredicatePushDownWithoutFileIndex) {
    if (FileFormat() == "avro") {
        return;
    }

    std::map<std::string, std::string> options = {{Options::FILE_INDEX_READ_ENABLED, "false"},
                                                  {Options::WRITE_BATCH_SIZE, "1"},
                                                  {"parquet.page.size", "1"},
                                                  {"parquet.enable-dictionary", "false"},
                                                  {"parquet.write.enable-page-index", "true"},
                                                  {"parquet.read.enable-page-index-filter", "true"},
                                                  {"orc.stripe.size", "1"},
                                                  {"orc.row.index.stride", "1"}};
    if (FileFormat() == "mosaic") {
        options.emplace(Options::FILE_BLOCK_SIZE, "1 B");
        options.emplace(mosaic::MOSAIC_STATS_COLUMNS, "f0");
    }
    CreateDataEvolutionTable(/*deletion_vectors_enabled=*/false, options);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    auto input = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "a", "x"],
        [2, "b", "y"],
        [3, "c", "z"],
        [4, "d", "w"]
    ])")
            .ValueOrDie());
    std::vector<std::shared_ptr<arrow::Array>> write_arrays;
    for (int64_t i = 0; i < input->length(); i++) {
        arrow::ArrayVector children;
        for (const auto& child : input->fields()) {
            children.push_back(child->Slice(i, 1));
        }
        write_arrays.push_back(arrow::StructArray::Make(children, fields_).ValueOrDie());
    }
    ASSERT_OK_AND_ASSIGN(auto commit_messages, WriteArrays(table_path, /*partition=*/{},
                                                           {"f0", "f1", "f2"}, write_arrays));
    ASSERT_OK(Commit(table_path, commit_messages));

    auto predicate =
        PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::INT, Literal(3));
    auto expected = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [3, "c", "z"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2"}, expected, predicate,
                          /*row_ranges=*/{},
                          /*check_scan_plan_when_empty_result=*/true,
                          /*apply_predicate_to_scan=*/true));
}

TEST_P(DataEvolutionTableTest, TestPredicate) {
    if (FileFormat() == "avro") {
        // Avro does not have stats.
        return;
    }
    if (FileFormat() == "mosaic") {
        CreateTable(/*partition_keys=*/{}, {{Options::MANIFEST_FORMAT, "orc"},
                                            {Options::FILE_FORMAT, FileFormat()},
                                            {Options::FILE_SYSTEM, "local"},
                                            {Options::ROW_TRACKING_ENABLED, "true"},
                                            {Options::DATA_EVOLUTION_ENABLED, "true"},
                                            {mosaic::MOSAIC_STATS_COLUMNS, "f0,f1,f2"}});
    } else {
        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_OK(
            ScanAndRead(table_path, schema->field_names(), /*expected_array=*/nullptr, 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(DataEvolutionTableTest, 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);
        SetFirstRowId(/*reset_first_row_id=*/0,
                      const_cast<std::vector<std::shared_ptr<paimon::CommitMessage>>&>(
                          commit_msgs1_result.value()));
        CHECK_HOOK_STATUS(Commit(table_path, commit_msgs1_result.value()), 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(DataEvolutionTableTest, TestWithRowIds) {
    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
                                                  {Options::FILE_FORMAT, FileFormat()},
                                                  {Options::FILE_SYSTEM, "local"},
                                                  {Options::ROW_TRACKING_ENABLED, "true"},
                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
    if (FileFormat() == "mosaic") {
        options.emplace(mosaic::MOSAIC_STATS_COLUMNS, "f0,f1");
    }
    CreateTable(/*partition_keys=*/{}, options);

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

    // turn 0: write field: f0, f1
    std::vector<std::string> write_cols = {"f0", "f1"};
    auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
        [0, "a"],
        [1, "b"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs0, WriteArray(table_path, write_cols, src_array0));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs0);
    ASSERT_OK(Commit(table_path, commit_msgs0));

    // turn 1: write field: f0, f1
    auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
        [2, "c"],
        [3, "d"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, write_cols, src_array1));
    SetFirstRowId(/*reset_first_row_id=*/2, commit_msgs1);
    ASSERT_OK(Commit(table_path, commit_msgs1));

    // turn 2: write field: f0, f1
    auto src_array2 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
        [4, "e"],
        [5, "f"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs2, WriteArray(table_path, write_cols, src_array2));
    SetFirstRowId(/*reset_first_row_id=*/4, commit_msgs2);
    ASSERT_OK(Commit(table_path, commit_msgs2));

    {
        // test without row ids
        CheckScanResult(table_path, /*predicate=*/nullptr, /*row_ranges=*/{},
                        /*expected_first_row_ids=*/{0, 2, 4}, /*expected_row_counts=*/{2, 2, 2});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [0, "a", null],
        [1, "b", null],
        [2, "c", null],
        [3, "d", null],
        [4, "e", null],
        [5, "f", null]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
    }
    {
        // test row ids in first file
        std::vector<Range> row_ranges = {Range(0l, 1l)};
        CheckScanResult(table_path, /*predicate=*/nullptr, /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{0}, /*expected_row_counts=*/{2});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [0, "a", null],
        [1, "b", null]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
                              /*predicate=*/nullptr, /*row_ranges=*/row_ranges));
    }
    {
        // test row ids in last file
        std::vector<Range> row_ranges = {Range(4l, 4l)};
        CheckScanResult(table_path, /*predicate=*/nullptr, /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{4}, /*expected_row_counts=*/{2});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [4, "e", null]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
                              /*predicate=*/nullptr, /*row_ranges=*/row_ranges));
    }
    {
        // test row ids in multiple files
        std::vector<Range> row_ranges = {Range(1l, 1l), Range(4l, 4l)};
        CheckScanResult(table_path, /*predicate=*/nullptr,
                        /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{0, 4}, /*expected_row_counts=*/{2, 2});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "b", null],
        [4, "e", null]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
                              /*predicate=*/nullptr,
                              /*row_ranges=*/row_ranges));
    }
    {
        // test all row ids
        std::vector<Range> row_ranges = {Range(0l, 5l)};
        CheckScanResult(table_path, /*predicate=*/nullptr, /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{0, 2, 4}, /*expected_row_counts=*/{2, 2, 2});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [0, "a", null],
        [1, "b", null],
        [2, "c", null],
        [3, "d", null],
        [4, "e", null],
        [5, "f", null]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
                              /*predicate=*/nullptr, /*row_ranges=*/row_ranges));
    }
    {
        // test unordered row ids
        std::vector<Range> row_ranges = {Range(5l, 5l), Range(3l, 3l)};
        CheckScanResult(table_path, /*predicate=*/nullptr,
                        /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{2, 4}, /*expected_row_counts=*/{2, 2});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [3, "d", null],
        [5, "f", null]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
                              /*predicate=*/nullptr,
                              /*row_ranges=*/row_ranges));
    }
    {
        // test row ids which partially exist
        std::vector<Range> row_ranges = {Range(100l, 100l), Range(5l, 5l), Range(3l, 3l)};
        CheckScanResult(table_path, /*predicate=*/nullptr,
                        /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{2, 4}, /*expected_row_counts=*/{2, 2});
        auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
            arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [3, "d", null],
        [5, "f", null]
    ])")
                .ValueOrDie());
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
                              /*predicate=*/nullptr,
                              /*row_ranges=*/row_ranges));
    }
    {
        // test row ids which do not exist
        std::vector<Range> row_ranges = {Range(100l, 100l), Range(200l, 200l)};
        CheckScanResult(table_path, /*predicate=*/nullptr,
                        /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{}, /*expected_row_counts=*/{});
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), /*expected_array=*/nullptr,
                              /*predicate=*/nullptr,
                              /*row_ranges=*/row_ranges));
    }
    if (FileFormat() == "avro") {
        // Avro does not support stats.
        return;
    }
    {
        // test row id with predicate
        // row ids {0, 1, 5} and predicate f0 = 2
        // scan will filter out all files
        std::vector<Range> row_ranges = {Range(0l, 0l), Range(1l, 1l), Range(5l, 5l)};
        auto predicate = PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0",
                                                 FieldType::INT, Literal(2));
        CheckScanResult(table_path, /*predicate=*/predicate,
                        /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{}, /*expected_row_counts=*/{});
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), /*expected_array=*/nullptr,
                              predicate,
                              /*row_ranges=*/row_ranges));
    }
    {
        // test row id with predicate
        // row ids {0, 1, 5} and predicate f0 = 5
        // scan will filter out file0 and file1
        std::vector<Range> row_ranges = {Range(0l, 0l), Range(1l, 1l), Range(5l, 5l)};
        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"([
        [5, "f", null]
    ])")
                .ValueOrDie());
        CheckScanResult(table_path, /*predicate=*/predicate,
                        /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{4}, /*expected_row_counts=*/{2});
        ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array, predicate,
                              /*row_ranges=*/row_ranges));
    }
    {
        // test with row tracking fields
        std::vector<Range> row_ranges = {Range(0l, 0l), Range(1l, 1l), Range(5l, 5l)};
        CheckScanResult(table_path, /*predicate=*/nullptr,
                        /*row_ranges=*/row_ranges,
                        /*expected_first_row_ids=*/{0, 4}, /*expected_row_counts=*/{2, 2});

        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", 0, 1, 0, null],
        ["b", 1, 1, 1, null],
        ["f", 5, 3, 5, null]
    ])")
                .ValueOrDie());

        ASSERT_OK(ScanAndRead(table_path, {"f1", "f0", "_SEQUENCE_NUMBER", "_ROW_ID", "f2"},
                              expected_array, /*predicate=*/nullptr,
                              /*row_ranges=*/row_ranges));
    }
}

TEST_P(DataEvolutionTableTest, TestReadWithDeletionVectors) {
    CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    // full-row write assigns row ids 0-3, producing the anchor file of the row range group
    auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "a", "x"],
        [2, "b", "y"],
        [3, "c", "z"],
        [4, "d", "w"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
                         WriteArray(table_path, schema->field_names(), src_array));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs0);
    ASSERT_OK(Commit(table_path, commit_msgs0));

    // partial write of f2 over the same row range: the group merges columns from two files
    arrow::FieldVector f2_fields = {fields_[2]};
    auto update_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(f2_fields), R"([
        ["x2"],
        ["y2"],
        ["z2"],
        ["w2"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {"f2"}, update_array));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
    ASSERT_OK(Commit(table_path, commit_msgs1));

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

    ASSERT_OK_AND_ASSIGN(std::string anchor_file_name, PlannedAnchorFileName(table_path));
    ASSERT_OK(CommitDeletionVectors(table_path, commit_msgs0[0],
                                    {{anchor_file_name, /*deleted_positions=*/{1, 3}}}));

    // both files of the group must drop the same rows to keep the column merge aligned
    auto expected_deleted = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "a", "x2"],
        [3, "c", "z2"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_deleted));

    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", "x2", 0],
        [3, "c", "z2", 2]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2", "_ROW_ID"}, expected_with_row_id));

    // a row-range selection composes with the deletion vector: rows {1, 2} minus deleted {1}
    auto expected_selected = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [3, "c", "z2"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_selected,
                          /*predicate=*/nullptr, /*row_ranges=*/{Range(1, 2)}));
}

TEST_P(DataEvolutionTableTest, TestReadWithDeletionVectorsAcrossReadBatches) {
    // the 12 rows below span several read batches, so the deletion vector empties a whole
    // batch of every file of the group: each file reader then skips that batch entirely and
    // the column merge has to stay aligned on the surviving row count alone
    CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true, {{Options::READ_BATCH_SIZE, "4"}});
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> group_msgs,
                         WriteAndCommitGroup(table_path, /*first_row_id=*/0,
                                             /*f0_values=*/{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11}));

    // positions 4-7 cover a whole read batch, position 9 only part of the next one
    ASSERT_OK_AND_ASSIGN(std::string anchor_file_name, PlannedAnchorFileName(table_path));
    ASSERT_OK(CommitDeletionVectors(table_path, group_msgs[0],
                                    {{anchor_file_name, /*deleted_positions=*/{4, 5, 6, 7, 9}}}));

    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [0, "a0", "y0"],
        [1, "a1", "y1"],
        [2, "a2", "y2"],
        [3, "a3", "y3"],
        [8, "a8", "y8"],
        [10, "a10", "y10"],
        [11, "a11", "y11"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));

    arrow::FieldVector row_id_fields = {SpecialFields::RowId().field_};
    auto expected_row_ids = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(row_id_fields), R"([
        [0], [1], [2], [3], [8], [10], [11]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"_ROW_ID"}, expected_row_ids));

    // a row-range selection that spans the fully deleted batch composes with it
    auto expected_selected = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [3, "a3", "y3"],
        [8, "a8", "y8"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_selected,
                          /*predicate=*/nullptr, /*row_ranges=*/{Range(3, 8)}));
}

TEST_P(DataEvolutionTableTest, TestReadWithDeletionVectorsOnPartOfRowRangeGroups) {
    // one split per row range group, so a group's deletion file must not reach the other
    CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true,
                             {{Options::SOURCE_SPLIT_TARGET_SIZE, "1"}});
    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", "x"],
        [2, "b", "y"],
        [3, "c", "z"],
        [4, "d", "w"]
    ])")
            .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));

    auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [5, "e", "v"],
        [6, "f", "u"],
        [7, "g", "t"],
        [8, "h", "s"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs1,
                         WriteArray(table_path, schema->field_names(), src_array1));
    SetFirstRowId(/*reset_first_row_id=*/4, commit_msgs1);
    ASSERT_OK(Commit(table_path, commit_msgs1));

    ASSERT_OK_AND_ASSIGN(std::vector<std::string> anchor_file_names,
                         PlannedAnchorFileNames(table_path));
    ASSERT_EQ(anchor_file_names.size(), 2);
    ASSERT_OK(CommitDeletionVectors(table_path, commit_msgs0[0],
                                    {{anchor_file_names[0], /*deleted_positions=*/{1, 3}}}));

    // the deletion vector applies to its own group only, the other group keeps every row
    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "a", "x"],
        [3, "c", "z"],
        [5, "e", "v"],
        [6, "f", "u"],
        [7, "g", "t"],
        [8, "h", "s"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));

    // a projection-only read drops the same rows
    arrow::FieldVector f0_fields = {fields_[0]};
    auto expected_f0 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(f0_fields), R"([
        [1], [3], [5], [6], [7], [8]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"f0"}, expected_f0));

    arrow::FieldVector row_id_fields = {SpecialFields::RowId().field_};
    auto expected_row_ids = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(row_id_fields), R"([
        [0], [2], [4], [5], [6], [7]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"_ROW_ID"}, expected_row_ids));
}

TEST_P(DataEvolutionTableTest, TestReadWithDeletionVectorsOnEveryRowRangeGroup) {
    CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    // one split (the default target size keeps both groups together), so a single split
    // deletion vector factory serves two groups anchored at different row ids
    ASSERT_OK_AND_ASSIGN(
        std::vector<std::shared_ptr<CommitMessage>> group_msgs0,
        WriteAndCommitGroup(table_path, /*first_row_id=*/0, /*f0_values=*/{0, 1, 2, 3}));
    ASSERT_OK(WriteAndCommitGroup(table_path, /*first_row_id=*/4, /*f0_values=*/{4, 5, 6, 7}));

    ASSERT_OK_AND_ASSIGN(std::vector<std::string> anchor_file_names,
                         PlannedAnchorFileNames(table_path));
    ASSERT_EQ(anchor_file_names.size(), 2);

    // Positions are anchor-relative, so the groups deliberately delete different ones: group 0
    // drops {1, 3} of row ids 0-3, group 1 drops {0, 2} of row ids 4-7. Reading a group with the
    // other group's vector, or with its anchor range as the shift base, cannot match below.
    ASSERT_OK(CommitDeletionVectors(table_path, group_msgs0[0],
                                    {{anchor_file_names[0], /*deleted_positions=*/{1, 3}},
                                     {anchor_file_names[1], /*deleted_positions=*/{0, 2}}}));

    // The read looks a group's deletion vector up by its anchor file name, so the scan has to
    // hand it exactly that. Asserting it separates a scan-side mix-up from a read-side one.
    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> planned_splits,
                         PlanSplits(table_path));
    ASSERT_EQ(planned_splits.size(), 1);
    auto planned_split_impl = std::dynamic_pointer_cast<DataSplitImpl>(planned_splits[0]);
    ASSERT_TRUE(planned_split_impl);

    ASSERT_OK_AND_ASSIGN(DeletionCardinalityMap cardinality_by_file,
                         DeletionCardinalityByDataFile(planned_splits[0]));
    DeletionCardinalityMap expected_cardinalities = {{anchor_file_names[0], 2},
                                                     {anchor_file_names[1], 2}};
    ASSERT_EQ(cardinality_by_file, expected_cardinalities);

    // the same split reports the surviving row count the limit push down prunes on: the two
    // groups hold 4 rows each and each deletion vector drops 2 of them
    ASSERT_OK_AND_ASSIGN(std::optional<int64_t> merged_row_count,
                         planned_split_impl->MergedRowCount());
    ASSERT_EQ(std::optional<int64_t>(4), merged_row_count);

    // a count query answers from that metadata alone, never reading a row, so the deletion
    // vectors have to reach it too: without them it reports the 8 rows the files hold
    ReadContextBuilder count_context_builder(table_path);
    count_context_builder.SetReadFieldNames(schema->field_names());
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> count_context,
                         count_context_builder.Finish());
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> count_table_read,
                         TableRead::Create(std::move(count_context)));
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<CountReader> count_reader,
                         count_table_read->CreateCountReader(planned_splits));
    ASSERT_OK_AND_ASSIGN(int64_t counted_rows, count_reader->CountRows());
    ASSERT_EQ(counted_rows, 4);

    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [0, "a0", "y0"],
        [2, "a2", "y2"],
        [5, "a1", "y1"],
        [7, "a3", "y3"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));

    arrow::FieldVector row_id_fields = {SpecialFields::RowId().field_};
    auto expected_row_ids = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(row_id_fields), R"([
        [0], [2], [5], [7]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"_ROW_ID"}, expected_row_ids));

    // a row-range selection straddling the group boundary composes with both deletion vectors:
    // row ids {2, 3, 4, 5} minus the deleted {3, 4}
    auto expected_selected = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [2, "a2", "y2"],
        [5, "a1", "y1"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_selected,
                          /*predicate=*/nullptr, /*row_ranges=*/{Range(2, 5)}));
}

TEST_P(DataEvolutionTableTest, TestReadWithFullyDeletedRowRangeGroup) {
    CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    // one split (the default target size keeps both groups together), so the emptied group's
    // readers are concatenated with the surviving group's
    ASSERT_OK_AND_ASSIGN(
        std::vector<std::shared_ptr<CommitMessage>> group_msgs0,
        WriteAndCommitGroup(table_path, /*first_row_id=*/0, /*f0_values=*/{0, 1, 2, 3}));
    ASSERT_OK(WriteAndCommitGroup(table_path, /*first_row_id=*/4, /*f0_values=*/{4, 5, 6, 7}));

    // both file readers of the first group then yield nothing, and its column merge must
    // produce no rows at all instead of misaligning
    ASSERT_OK_AND_ASSIGN(std::vector<std::string> anchor_file_names,
                         PlannedAnchorFileNames(table_path));
    ASSERT_EQ(anchor_file_names.size(), 2);
    ASSERT_OK(CommitDeletionVectors(table_path, group_msgs0[0],
                                    {{anchor_file_names[0], /*deleted_positions=*/{0, 1, 2, 3}}}));

    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [4, "a0", "y0"],
        [5, "a1", "y1"],
        [6, "a2", "y2"],
        [7, "a3", "y3"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));

    arrow::FieldVector row_id_fields = {SpecialFields::RowId().field_};
    auto expected_row_ids = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(row_id_fields), R"([
        [4], [5], [6], [7]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"_ROW_ID"}, expected_row_ids));

    // a row-range selection spanning both groups keeps only what survives in the second one
    auto expected_selected = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [4, "a0", "y0"],
        [5, "a1", "y1"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_selected,
                          /*predicate=*/nullptr, /*row_ranges=*/{Range(0, 5)}));

    // a selection covering only deleted rows returns nothing. The plan is asserted non-empty
    // too: the scan cannot prune the split on row ids alone, so the emptiness comes from the
    // deletion vector rather than from a plan with nothing to read.
    ASSERT_OK_AND_ASSIGN(LimitScanResult only_deleted,
                         ScanAndReadWithLimit(table_path, schema->field_names(), /*limit=*/100,
                                              /*predicate=*/nullptr,
                                              /*row_ranges=*/{Range(0, 3)}));
    ASSERT_FALSE(only_deleted.splits.empty());
    ASSERT_FALSE(only_deleted.rows);
}

TEST_P(DataEvolutionTableTest, TestReadAfterUpdatingDeletionVectors) {
    CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
    auto schema = arrow::schema(fields_);

    auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "a", "x"],
        [2, "b", "y"],
        [3, "c", "z"],
        [4, "d", "w"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(table_path, schema->field_names(), src_array));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs);
    ASSERT_OK(Commit(table_path, commit_msgs));

    ASSERT_OK_AND_ASSIGN(std::string anchor_file_name, PlannedAnchorFileName(table_path));
    ASSERT_OK_AND_ASSIGN(std::shared_ptr<CommitMessage> first_dv_msg,
                         CommitDeletionVectors(table_path, commit_msgs[0],
                                               {{anchor_file_name, /*deleted_positions=*/{1}}}));

    auto expected_after_first = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "a", "x"],
        [3, "c", "z"],
        [4, "d", "w"]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_after_first));

    // a second deletion vector replaces the first one instead of both staying live
    ASSERT_OK(CommitDeletionVectors(table_path, commit_msgs[0],
                                    {{anchor_file_name, /*deleted_positions=*/{1, 3}}},
                                    /*replaced_commit_msg=*/first_dv_msg));

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

TEST_P(DataEvolutionTableTest, TestReadWithDeletionVectorsAfterAddingColumn) {
    if (FileFormat() == "avro") {
        GTEST_SKIP() << "Avro has no stats, which the added column's scan pruning relies on";
    }
    std::map<std::string, std::string> options =
        CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
        [1, "a", "x"],
        [2, "b", "y"],
        [3, "c", "z"],
        [4, "d", "w"]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
                         WriteArray(table_path, arrow::schema(fields_)->field_names(), src_array));
    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs);
    ASSERT_OK(Commit(table_path, commit_msgs));

    // add column f3, then fill it for the same row range: the group merges columns from two
    // files written under different schema ids
    auto f3 = arrow::field("f3", arrow::int64());
    ASSERT_OK(TestHelper::WriteNextSchema(dir_->GetFileSystem(), table_path,
                                          {DataField(0, fields_[0]), DataField(1, fields_[1]),
                                           DataField(2, fields_[2]), DataField(3, f3)},
                                          /*highest_field_id=*/3, options));

    auto f3_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({f3}), R"([
        [10], [20], [30], [40]
    ])")
            .ValueOrDie());
    ASSERT_OK_AND_ASSIGN(auto f3_commit_msgs, WriteArray(table_path, {"f3"}, f3_array));
    SetFirstRowId(/*reset_first_row_id=*/0, f3_commit_msgs);
    ASSERT_OK(Commit(table_path, f3_commit_msgs));

    // the deletion vector is still anchored on the oldest normal file, written before the
    // column was added
    ASSERT_OK_AND_ASSIGN(std::string anchor_file_name, PlannedAnchorFileName(table_path));
    ASSERT_OK(CommitDeletionVectors(table_path, commit_msgs[0],
                                    {{anchor_file_name, /*deleted_positions=*/{1, 3}}}));

    arrow::FieldVector evolved_fields = {fields_[0], fields_[1], fields_[2], f3};
    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(evolved_fields), R"([
        [1, "a", "x", 10],
        [3, "c", "z", 30]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2", "f3"}, expected_array));

    // projecting only the added column keeps the same surviving rows
    auto expected_f3 = std::dynamic_pointer_cast<arrow::StructArray>(
        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({f3}), R"([
        [10], [30]
    ])")
            .ValueOrDie());
    ASSERT_OK(ScanAndRead(table_path, {"f3"}, expected_f3));
}

TEST_P(DataEvolutionTableTest, TestLimitPushDownWithHeavilyDeletedFirstRowRangeGroup) {
    // one split per row range group, so the limit has to span both to be satisfied
    CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true,
                             {{Options::SOURCE_SPLIT_TARGET_SIZE, "1"}});
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    ASSERT_OK_AND_ASSIGN(
        std::vector<std::shared_ptr<CommitMessage>> group_msgs0,
        WriteAndCommitGroup(table_path, /*first_row_id=*/0, /*f0_values=*/{0, 1, 2, 3}));
    ASSERT_OK(WriteAndCommitGroup(table_path, /*first_row_id=*/4, /*f0_values=*/{4, 5, 6, 7}));

    // the first group keeps a single surviving row
    ASSERT_OK_AND_ASSIGN(std::vector<std::string> anchor_file_names,
                         PlannedAnchorFileNames(table_path));
    ASSERT_EQ(anchor_file_names.size(), 2);
    ASSERT_OK(CommitDeletionVectors(table_path, group_msgs0[0],
                                    {{anchor_file_names[0], /*deleted_positions=*/{0, 1, 2}}}));

    // the first split alone satisfies a limit of 1: it still holds the one row that survived
    // the deletion vector
    ASSERT_OK_AND_ASSIGN(LimitScanResult limit_1,
                         ScanAndReadWithLimit(table_path, {"f0"}, /*limit=*/1));
    ASSERT_EQ(limit_1.splits.size(), 1);
    ASSERT_OK_AND_ASSIGN(std::vector<int32_t> limit_1_values, CollectF0Values(limit_1.rows));
    ASSERT_EQ(limit_1_values, (std::vector<int32_t>{3}));

    // the first split contributes only one surviving row, so a limit of 3 needs the second one
    ASSERT_OK_AND_ASSIGN(LimitScanResult limit_3,
                         ScanAndReadWithLimit(table_path, {"f0"}, /*limit=*/3));
    ASSERT_EQ(limit_3.splits.size(), 2);
    ASSERT_OK_AND_ASSIGN(std::vector<int32_t> limit_3_values, CollectF0Values(limit_3.rows));
    ASSERT_EQ(limit_3_values, (std::vector<int32_t>{3, 4, 5, 6, 7}));
}

TEST_P(DataEvolutionTableTest, TestLimitPushDownDisabledByNonPartitionFilter) {
    // one split per row range group, so the plan can drop the group holding the matches
    CreateDataEvolutionTable(/*deletion_vectors_enabled=*/false,
                             {{Options::SOURCE_SPLIT_TARGET_SIZE, "1"}});
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    // First row range group: no f0 value lies in [100, 200], but the 300 keeps the group's stats
    // range straddling the filter so the scan cannot prune it. It therefore reaches the plan
    // reporting four rows, and contributes none of them to the result.
    ASSERT_OK(WriteAndCommitGroup(table_path, /*first_row_id=*/0, /*f0_values=*/{0, 1, 2, 300}));
    // second row range group: every f0 value matches
    ASSERT_OK(WriteAndCommitGroup(table_path, /*first_row_id=*/4,
                                  /*f0_values=*/{100, 101, 102, 103}));

    // the metadata row count of the first split alone satisfies the limit, but the filter runs
    // while reading and drops every one of its rows. Pruning the plan on the metadata count
    // would return nothing, so the push down has to be skipped and both splits kept.
    auto at_least_100 = PredicateBuilder::GreaterOrEqual(/*field_index=*/0, /*field_name=*/"f0",
                                                         FieldType::INT, Literal(100));
    auto at_most_200 = PredicateBuilder::LessOrEqual(/*field_index=*/0, /*field_name=*/"f0",
                                                     FieldType::INT, Literal(200));
    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> predicate,
                         PredicateBuilder::And({at_least_100, at_most_200}));
    // filtering while reading is what makes the metadata count an upper bound, so the read has
    // to opt into it for the rows below to show what the push down would have thrown away
    ASSERT_OK_AND_ASSIGN(LimitScanResult limited,
                         ScanAndReadWithLimit(table_path, {"f0"}, /*limit=*/2, predicate,
                                              /*row_ranges=*/{},
                                              /*enable_predicate_filter=*/true));
    ASSERT_EQ(limited.splits.size(), 2);
    ASSERT_OK_AND_ASSIGN(std::vector<int32_t> values, CollectF0Values(limited.rows));
    ASSERT_EQ(values, (std::vector<int32_t>{100, 101, 102, 103}));
}

TEST_P(DataEvolutionTableTest, TestLimitPushDownKeptByPartitionFilter) {
    std::vector<std::string> partition_keys = {"f1"};
    CreateTable(partition_keys);
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    // one row range group per partition, each holding four rows with its own f0 range. A split
    // never spans partitions, so the plan holds one split per partition for the push down to
    // prune.
    ASSERT_OK(WriteAndCommitGroup(table_path, /*first_row_id=*/0, /*f0_values=*/{0, 1, 2, 3},
                                  /*partition=*/{{"f1", "p0"}}));
    ASSERT_OK(WriteAndCommitGroup(table_path, /*first_row_id=*/4,
                                  /*f0_values=*/{10, 11, 12, 13}, /*partition=*/{{"f1", "p1"}}));
    ASSERT_OK(WriteAndCommitGroup(table_path, /*first_row_id=*/8,
                                  /*f0_values=*/{20, 21, 22, 23}, /*partition=*/{{"f1", "p2"}}));

    // a predicate on the partition key alone is evaluated while planning, never while reading,
    // so every row a surviving split reports is actually returned and the push down stays on
    auto not_p0 =
        PredicateBuilder::NotEqual(/*field_index=*/1, /*field_name=*/"f1", FieldType::STRING,
                                   Literal(FieldType::STRING, "p0", 2));

    // baseline: a limit no split combination can reach prunes nothing, so this only shows which
    // splits the partition filter itself leaves behind
    ASSERT_OK_AND_ASSIGN(LimitScanResult unpruned,
                         ScanAndReadWithLimit(table_path, {"f0", "f1"}, /*limit=*/100, not_p0));
    ASSERT_EQ(unpruned.splits.size(), 2);
    ASSERT_OK_AND_ASSIGN(std::vector<int32_t> unpruned_values, CollectF0Values(unpruned.rows));
    std::sort(unpruned_values.begin(), unpruned_values.end());
    ASSERT_EQ(unpruned_values, (std::vector<int32_t>{10, 11, 12, 13, 20, 21, 22, 23}));

    // the first matching split already holds four rows, so a limit of 2 drops the second one.
    // Treating the partition filter as a read-time filter would skip the push down and keep both.
    ASSERT_OK_AND_ASSIGN(LimitScanResult limited,
                         ScanAndReadWithLimit(table_path, {"f0", "f1"}, /*limit=*/2, not_p0));
    ASSERT_EQ(limited.splits.size(), 1);
    ASSERT_OK_AND_ASSIGN(std::vector<int32_t> limited_values, CollectF0Values(limited.rows));
    std::sort(limited_values.begin(), limited_values.end());
    ASSERT_TRUE(limited_values == (std::vector<int32_t>{10, 11, 12, 13}) ||
                limited_values == (std::vector<int32_t>{20, 21, 22, 23}))
        << "unexpected kept rows for limit 2";

    // one non-partition conjunct added to the same partition filter: the predicate is no longer
    // settled while planning, so the push down is skipped and the split the limit would have
    // dropped stays
    auto at_least_10 = PredicateBuilder::GreaterOrEqual(/*field_index=*/0, /*field_name=*/"f0",
                                                        FieldType::INT, Literal(10));
    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> mixed_predicate,
                         PredicateBuilder::And({not_p0, at_least_10}));
    ASSERT_OK_AND_ASSIGN(LimitScanResult mixed, ScanAndReadWithLimit(table_path, {"f0", "f1"},
                                                                     /*limit=*/2, mixed_predicate));
    ASSERT_EQ(mixed.splits.size(), 2);
    ASSERT_OK_AND_ASSIGN(std::vector<int32_t> mixed_values, CollectF0Values(mixed.rows));
    std::sort(mixed_values.begin(), mixed_values.end());
    ASSERT_EQ(mixed_values, (std::vector<int32_t>{10, 11, 12, 13, 20, 21, 22, 23}));
}

TEST_P(DataEvolutionTableTest, TestLimitPushDownDisabledByRowRangeIndex) {
    // one split per row range group, so the plan can drop the group holding the selection
    CreateDataEvolutionTable(/*deletion_vectors_enabled=*/false,
                             {{Options::SOURCE_SPLIT_TARGET_SIZE, "1"}});
    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");

    ASSERT_OK(WriteAndCommitGroup(table_path, /*first_row_id=*/0, /*f0_values=*/{0, 1, 2, 3}));
    ASSERT_OK(WriteAndCommitGroup(table_path, /*first_row_id=*/4,
                                  /*f0_values=*/{100, 101, 102, 103}));

    // the index selects one row of the first group and two of the second, but the first split
    // still reports four rows. Pruning on that count would drop the second group and return a
    // single row for a limit of 2, so the push down has to be skipped.
    ASSERT_OK_AND_ASSIGN(LimitScanResult limited,
                         ScanAndReadWithLimit(table_path, {"f0"}, /*limit=*/2,
                                              /*predicate=*/nullptr,
                                              /*row_ranges=*/{Range(3, 3), Range(5, 6)}));
    ASSERT_EQ(limited.splits.size(), 2);
    ASSERT_OK_AND_ASSIGN(std::vector<int32_t> values, CollectF0Values(limited.rows));
    ASSERT_EQ(values, (std::vector<int32_t>{3, 101, 102}));
}

std::vector<DataEvolutionTableParam> GetTestValuesForDataEvolutionTableTest() {
    std::vector<DataEvolutionTableParam> values;
    for (bool enable_snapshot_live_manifest_cache : {false, true}) {
        values.emplace_back("parquet", enable_snapshot_live_manifest_cache);
#ifdef PAIMON_ENABLE_MOSAIC
        values.emplace_back("mosaic", enable_snapshot_live_manifest_cache);
#endif
#ifdef PAIMON_ENABLE_ORC
        values.emplace_back("orc", enable_snapshot_live_manifest_cache);
#endif
#ifdef PAIMON_ENABLE_AVRO
        values.emplace_back("avro", enable_snapshot_live_manifest_cache);
#endif
    }
    return values;
}

INSTANTIATE_TEST_SUITE_P(FileFormat, DataEvolutionTableTest,
                         ::testing::ValuesIn(GetTestValuesForDataEvolutionTableTest()));

}  // namespace paimon::test