/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <algorithm>
#include <cstdint>
#include <cstdlib>
#include <filesystem>
#include <functional>
#include <limits>
#include <map>
#include <memory>
#include <optional>
#include <set>
#include <string>
#include <tuple>
#include <utility>
#include <vector>

#include "arrow/api.h"
#include "arrow/array/builder_binary.h"
#include "arrow/array/builder_nested.h"
#include "arrow/array/builder_primitive.h"
#include "arrow/c/abi.h"
#include "arrow/c/bridge.h"
#include "arrow/ipc/json_simple.h"
#include "fmt/format.h"
#include "gtest/gtest.h"
#include "paimon/catalog/catalog.h"
#include "paimon/catalog/identifier.h"
#include "paimon/commit_context.h"
#include "paimon/common/factories/io_hook.h"
#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/date_time_utils.h"
#include "paimon/common/utils/path_util.h"
#include "paimon/core/manifest/manifest_file_meta.h"
#include "paimon/core/manifest/manifest_list.h"
#include "paimon/core/operation/file_store_commit_impl.h"
#include "paimon/core/snapshot.h"
#include "paimon/core/utils/snapshot_manager.h"
#include "paimon/defs.h"
#include "paimon/file_store_commit.h"
#include "paimon/file_store_write.h"
#include "paimon/fs/file_system.h"
#include "paimon/fs/local/local_file_system.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/orphan_files_cleaner.h"
#include "paimon/record_batch.h"
#include "paimon/result.h"
#include "paimon/status.h"
#include "paimon/testing/utils/testharness.h"
#include "paimon/write_context.h"

namespace arrow {
class Array;
}  // namespace arrow
namespace paimon {
class CommitMessage;
}  // namespace paimon

namespace paimon::test {
class CleanInteTest : public testing::Test {
 public:
    void SetUp() override {
        file_system_ = std::make_shared<LocalFileSystem>();
        pool_ = GetDefaultPool();
        int64_t seed = DateTimeUtils::GetCurrentUTCTimeUs();
        std::srand(seed);
    }

    Result<std::string> CreateTestTable(const std::string& base_path, const std::string& db_name,
                                        const std::string& table_name, ::ArrowSchema* schema,
                                        const std::vector<std::string>& partition_keys,
                                        const std::vector<std::string>& primary_keys,
                                        const std::map<std::string, std::string>& options) const {
        PAIMON_ASSIGN_OR_RAISE(auto catalog, Catalog::Create(base_path, options));
        PAIMON_RETURN_NOT_OK(catalog->CreateDatabase(db_name, options, /*ignore_if_exists=*/false));
        Identifier table_id(db_name, table_name);
        PAIMON_RETURN_NOT_OK(catalog->CreateTable(table_id, schema, partition_keys, primary_keys,
                                                  options, /*ignore_if_exists=*/false));
        return PathUtil::JoinPath(base_path, db_name + ".db/" + table_name);
    }

    using ValueType = std::tuple<std::string, int32_t, int32_t, double>;
    Result<std::unique_ptr<RecordBatch>> MakeRecordBatch(
        const std::vector<ValueType>& raw_data,
        const std::map<std::string, std::string>& partition_map, int32_t bucket) const {
        ::ArrowArray arrow_array;
        std::shared_ptr<arrow::Array> array = GenerateArrowArray(raw_data);
        PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &arrow_array));
        RecordBatchBuilder batch_builder(&arrow_array);
        return batch_builder.SetPartition(partition_map).SetBucket(bucket).Finish();
    }

    Result<std::unique_ptr<RecordBatch>> MakeRecordBatch(
        const std::shared_ptr<arrow::DataType>& data_type, const std::string& data_str,
        const std::map<std::string, std::string>& partition_map, int32_t bucket,
        const std::vector<RecordBatch::RowKind>& row_kinds) const {
        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
            auto array, arrow::ipc::internal::json::ArrayFromJSON(data_type, data_str));
        ::ArrowArray arrow_array;
        PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &arrow_array));
        RecordBatchBuilder batch_builder(&arrow_array);
        return batch_builder.SetPartition(partition_map)
            .SetBucket(bucket)
            .SetRowKinds(row_kinds)
            .Finish();
    }

    std::shared_ptr<arrow::Array> GenerateArrowArray(const std::vector<ValueType>& raw_data,
                                                     bool exist_null_value = false) const {
        auto string_field = arrow::field("f0", arrow::utf8());
        auto int_field = arrow::field("f1", arrow::int32());
        auto int_field1 = arrow::field("f2", arrow::int32());
        auto double_field = arrow::field("f3", arrow::float64());
        auto struct_type = arrow::struct_({string_field, int_field, int_field1, double_field});
        auto schema =
            arrow::schema(arrow::FieldVector({string_field, int_field, int_field1, double_field}));

        arrow::StructBuilder struct_builder(
            struct_type, arrow::default_memory_pool(),
            {std::make_shared<arrow::StringBuilder>(), std::make_shared<arrow::Int32Builder>(),
             std::make_shared<arrow::Int32Builder>(), std::make_shared<arrow::DoubleBuilder>()});
        auto string_builder = static_cast<arrow::StringBuilder*>(struct_builder.field_builder(0));
        auto int_builder = static_cast<arrow::Int32Builder*>(struct_builder.field_builder(1));
        auto int_builder1 = static_cast<arrow::Int32Builder*>(struct_builder.field_builder(2));
        auto double_builder = static_cast<arrow::DoubleBuilder*>(struct_builder.field_builder(3));

        for (const auto& d : raw_data) {
            EXPECT_TRUE(struct_builder.Append().ok());
            EXPECT_TRUE(string_builder->Append(std::get<0>(d)).ok());
            EXPECT_TRUE(int_builder->Append(std::get<1>(d)).ok());
            EXPECT_TRUE(int_builder1->Append(std::get<2>(d)).ok());
            if (exist_null_value) {
                EXPECT_TRUE(double_builder->AppendNull().ok());
            } else {
                EXPECT_TRUE(double_builder->Append(std::get<3>(d)).ok());
            }
        }
        std::shared_ptr<arrow::Array> array;
        EXPECT_TRUE(struct_builder.Finish(&array).ok());
        return array;
    }

    bool HitIOHook(const Status& status, size_t io_count) const {
        if (status.ToString().find(fmt::format("io hook triggered io error at position {}",
                                               io_count)) != std::string::npos) {
            return true;
        }
        return false;
    }

    bool HitIOHookInCommitHint(const Status& status) const {
        if (status.ToString().find("io hook triggered io error at position") != std::string::npos &&
            (status.ToString().find("snapshot/LATEST") != std::string::npos ||
             status.ToString().find("snapshot/EARLIEST") != std::string::npos)) {
            return true;
        }
        return false;
    }

    bool CheckEqual(const std::set<std::string>& cleaned_paths,
                    const std::set<std::string>& expected_names) {
        std::set<std::string> file_names;
        for (const auto& file_path : cleaned_paths) {
            file_names.insert(PathUtil::GetName(file_path));
        }
        EXPECT_EQ(file_names, expected_names);
        return file_names == expected_names;
    }

 private:
    std::shared_ptr<FileSystem> file_system_;
    std::shared_ptr<MemoryPool> pool_;
};

TEST_F(CleanInteTest, TestExpireSnapshotFailover) {
    std::string test_data_path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09/";
    std::map<std::string, std::string> clean_options = {
        {Options::MANIFEST_TARGET_FILE_SIZE, "8mb"},
        {Options::FILE_SYSTEM, "local"},
        {Options::SNAPSHOT_NUM_RETAINED_MAX, "2"},
        {Options::SNAPSHOT_NUM_RETAINED_MIN, "1"},
        {Options::SNAPSHOT_TIME_RETAINED, "1ms"},
        {Options::MANIFEST_MERGE_MIN_COUNT, "2"},
        {Options::SNAPSHOT_CLEAN_EMPTY_DIRECTORIES, "true"}};
    // no failover
    {
        auto dir = UniqueTestDirectory::Create();
        std::string table_path = dir->Str();
        ASSERT_TRUE(TestUtil::CopyDirectory(test_data_path, table_path));
        CommitContextBuilder commit_context_builder(table_path, "commit_user_1");
        ASSERT_OK_AND_ASSIGN(
            std::unique_ptr<CommitContext> commit_context,
            commit_context_builder.SetOptions(clean_options).IgnoreEmptyCommit(false).Finish());
        ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context)));
        ASSERT_OK_AND_ASSIGN(int32_t expired_items, commit->Expire());
        ASSERT_EQ(expired_items, 4);
    }
    // failover with non-exist manifest
    {
        auto dir = UniqueTestDirectory::Create();
        std::string table_path = dir->Str();
        ASSERT_TRUE(TestUtil::CopyDirectory(test_data_path, table_path));
        ASSERT_OK(file_system_->Delete(PathUtil::JoinPath(
            table_path, "manifest/manifest-list-616d1847-a02c-495f-9cca-2c8b7def0fec-1")));
        CommitContextBuilder commit_context_builder(table_path, "commit_user_1");
        ASSERT_OK_AND_ASSIGN(
            std::unique_ptr<CommitContext> commit_context,
            commit_context_builder.SetOptions(clean_options).IgnoreEmptyCommit(false).Finish());
        ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context)));
        ASSERT_OK_AND_ASSIGN(int32_t expired_items, commit->Expire());
        ASSERT_EQ(expired_items, 4);
    }
    // failover with non-exist data file
    {
        auto dir = UniqueTestDirectory::Create();
        std::string table_path = dir->Str();
        ASSERT_TRUE(TestUtil::CopyDirectory(test_data_path, table_path));
        ASSERT_OK(file_system_->Delete(PathUtil::JoinPath(
            table_path, "f1=10/bucket-1/data-10b9eea8-241d-4e4b-8ab8-2a82d72d79a2-0.orc")));
        CommitContextBuilder commit_context_builder(table_path, "commit_user_1");
        ASSERT_OK_AND_ASSIGN(
            std::unique_ptr<CommitContext> commit_context,
            commit_context_builder.SetOptions(clean_options).IgnoreEmptyCommit(false).Finish());
        ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context)));
        ASSERT_OK_AND_ASSIGN(int32_t expired_items, commit->Expire());
        ASSERT_EQ(expired_items, 4);
    }
    // failover with discontinuous snapshot
    {
        auto dir = UniqueTestDirectory::Create();
        std::string table_path = dir->Str();
        ASSERT_TRUE(TestUtil::CopyDirectory(test_data_path, table_path));
        ASSERT_OK(file_system_->Delete(PathUtil::JoinPath(table_path, "snapshot/snapshot-3")));
        CommitContextBuilder commit_context_builder(table_path, "commit_user_1");
        ASSERT_OK_AND_ASSIGN(
            std::unique_ptr<CommitContext> commit_context,
            commit_context_builder.SetOptions(clean_options).IgnoreEmptyCommit(false).Finish());
        ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context)));
        ASSERT_OK_AND_ASSIGN(int32_t expired_items, commit->Expire());
        ASSERT_EQ(expired_items, 1);
    }
}

TEST_F(CleanInteTest, TestDropPartitionAndExpireSnapshot) {
    auto string_field = arrow::field("f0", arrow::utf8());
    auto int_field = arrow::field("f1", arrow::int32());
    auto int_field1 = arrow::field("f2", arrow::int32());
    auto double_field = arrow::field("f3", arrow::float64());
    auto schema =
        arrow::schema(arrow::FieldVector({string_field, int_field, int_field1, double_field}));

    ::ArrowSchema arrow_schema;
    ASSERT_TRUE(arrow::ExportSchema(*schema, &arrow_schema).ok());
    auto dir = UniqueTestDirectory::Create();
    ASSERT_TRUE(dir);

    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, "orc"},
        {Options::TARGET_FILE_SIZE, "1024"},
        {Options::FILE_SYSTEM, "local"},
        {Options::BUCKET, "2"},
        {Options::BUCKET_KEY, "f2"},
        {Options::SNAPSHOT_CLEAN_EMPTY_DIRECTORIES, "true"},
    };

    ASSERT_OK_AND_ASSIGN(
        std::string table_path,
        CreateTestTable(dir->Str(), /*db_name=*/"foo", /*table_name=*/"bar", &arrow_schema,
                        /*partition_keys=*/{"f1"},
                        /*primary_keys=*/{}, options));
    WriteContextBuilder context_builder(table_path, "commit_user_1");
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<WriteContext> write_context,
                         context_builder.SetOptions(options).WithStreamingMode(true).Finish());

    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> file_store_write,
                         FileStoreWrite::Create(std::move(write_context)));

    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_0,
                         MakeRecordBatch({{"Alice", 10, 1, 11.1}}, {{"f1", "10"}}, 0));
    ASSERT_OK(file_store_write->Write(std::move(batch_0)));
    ASSERT_OK_AND_ASSIGN(
        std::unique_ptr<RecordBatch> batch_1,
        MakeRecordBatch({{"Bob", 10, 0, 12.1}, {"Emily", 10, 0, 13.1}, {"Tony", 10, 0, 14.1}},
                        {{"f1", "10"}}, 1));
    ASSERT_OK(file_store_write->Write(std::move(batch_1)));
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_2,
                         MakeRecordBatch({{"Lucy", 20, 1, 14.1}}, {{"f1", "20"}}, 0));
    ASSERT_OK(file_store_write->Write(std::move(batch_2)));
    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results,
                         file_store_write->PrepareCommit(/*wait_compaction=*/false, 0));
    ASSERT_EQ(3u, results.size());
    CommitContextBuilder commit_context_builder(table_path, "commit_user_1");
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<CommitContext> commit_context,
                         commit_context_builder.AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb")
                             .WithFileSystem(file_system_)
                             .AddOption(Options::SNAPSHOT_NUM_RETAINED_MAX, "2")
                             .AddOption(Options::SNAPSHOT_NUM_RETAINED_MIN, "1")
                             .AddOption(Options::SNAPSHOT_TIME_RETAINED, "1ms")
                             .AddOption(Options::MANIFEST_MERGE_MIN_COUNT, "2")
                             .AddOption(Options::SNAPSHOT_CLEAN_EMPTY_DIRECTORIES, "true")
                             .IgnoreEmptyCommit(false)
                             .Finish());
    ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context)));
    ASSERT_OK(commit->Commit(results, 0, 10));

    ASSERT_OK_AND_ASSIGN(
        std::unique_ptr<RecordBatch> batch_3,
        MakeRecordBatch({{"Emily", 10, 0, 15.1}, {"Bob", 10, 0, 12.1}, {"Alex", 10, 0, 16.1}},
                        {{"f1", "10"}}, 1));
    ASSERT_OK(file_store_write->Write(std::move(batch_3)));
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_4,
                         MakeRecordBatch({{"Paul", 20, 1, 0}}, {{"f1", "20"}}, 0));
    ASSERT_OK(file_store_write->Write(std::move(batch_4)));

    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results2,
                         file_store_write->PrepareCommit(/*wait_compaction=*/false, 1));
    ASSERT_EQ(3u, results2.size());
    ASSERT_OK(commit->Commit(results2, 1, 30));
    ASSERT_OK(commit->DropPartition({{{"f1", "10"}}}, 2));
    auto commit_impl = dynamic_cast<FileStoreCommitImpl*>(commit.get());
    ASSERT_TRUE(commit_impl);
    ASSERT_OK_AND_ASSIGN(bool snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(1));
    ASSERT_TRUE(snapshot_exist);
    ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(2));
    ASSERT_TRUE(snapshot_exist);
    ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(3));
    ASSERT_TRUE(snapshot_exist);
    ASSERT_OK_AND_ASSIGN(Snapshot snapshot_3, commit_impl->snapshot_manager_->LoadSnapshot(3));
    ASSERT_EQ(30, snapshot_3.Watermark().value());
    ASSERT_EQ(-7, snapshot_3.DeltaRecordCount().value());
    ASSERT_EQ(2, snapshot_3.TotalRecordCount().value());
    ASSERT_EQ(Snapshot::CommitKind::Overwrite(), snapshot_3.GetCommitKind());
    ASSERT_EQ(2, snapshot_3.CommitIdentifier());
    ASSERT_OK_AND_ASSIGN(bool f1_10_bucket_0_exist,
                         file_system_->Exists(PathUtil::JoinPath(table_path, "f1=10/bucket-0")));
    ASSERT_TRUE(f1_10_bucket_0_exist);
    ASSERT_OK_AND_ASSIGN(bool f1_10_bucket_1_exist,
                         file_system_->Exists(PathUtil::JoinPath(table_path, "f1=10/bucket-1")));
    ASSERT_TRUE(f1_10_bucket_1_exist);
    ASSERT_OK_AND_ASSIGN(bool f1_20_bucket_0_exist,
                         file_system_->Exists(PathUtil::JoinPath(table_path, "f1=20/bucket-0")));
    ASSERT_TRUE(f1_20_bucket_0_exist);
    ASSERT_OK(commit->Expire());
    ASSERT_OK_AND_ASSIGN(f1_10_bucket_0_exist,
                         file_system_->Exists(PathUtil::JoinPath(table_path, "f1=10/bucket-0")));
    ASSERT_FALSE(f1_10_bucket_0_exist);
    ASSERT_OK_AND_ASSIGN(f1_10_bucket_1_exist,
                         file_system_->Exists(PathUtil::JoinPath(table_path, "f1=10/bucket-1")));
    ASSERT_FALSE(f1_10_bucket_1_exist);
    ASSERT_OK_AND_ASSIGN(f1_20_bucket_0_exist,
                         file_system_->Exists(PathUtil::JoinPath(table_path, "f1=20/bucket-0")));
    ASSERT_TRUE(f1_20_bucket_0_exist);
    ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(1));
    ASSERT_FALSE(snapshot_exist);
    ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(2));
    ASSERT_FALSE(snapshot_exist);
    ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(3));
    ASSERT_TRUE(snapshot_exist);
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_5,
                         MakeRecordBatch({{"Lucas", 20, 2, 22.2}}, {{"f1", "20"}}, 0));
    ASSERT_OK(file_store_write->Write(std::move(batch_5)));
    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results3,
                         file_store_write->PrepareCommit(/*wait_compaction=*/false, 3));
    ASSERT_EQ(3u, results3.size());
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_6,
                         MakeRecordBatch({{"Ken", 20, 3, 33.3}}, {{"f1", "20"}}, 0));
    ASSERT_OK(file_store_write->Write(std::move(batch_6)));
    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results4,
                         file_store_write->PrepareCommit(/*wait_compaction=*/false, 4));
    ASSERT_EQ(1u, results4.size());
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_7,
                         MakeRecordBatch({{"Dennis", 10, 4, 44.4}}, {{"f1", "10"}}, 0));
    ASSERT_OK(file_store_write->Write(std::move(batch_7)));
    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results5,
                         file_store_write->PrepareCommit(/*wait_compaction=*/false, 5));
    ASSERT_EQ(2u, results5.size());

    results3.insert(results3.end(), results4.begin(), results4.end());
    results3.insert(results3.end(), results5.begin(), results5.end());
    ASSERT_OK(commit->Commit(results3, 6, 40));
    ASSERT_OK(commit->Expire());
    ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(4));
    ASSERT_TRUE(snapshot_exist);
    ASSERT_OK_AND_ASSIGN(Snapshot snapshot_4, commit_impl->snapshot_manager_->LoadSnapshot(4));
    ASSERT_EQ(40, snapshot_4.Watermark().value());
    std::vector<ManifestFileMeta> manifests;
    ASSERT_OK(commit_impl->manifest_list_->ReadDataManifests(snapshot_4, &manifests));
    ASSERT_EQ(2u, manifests.size());
    ASSERT_EQ(2u, manifests[0].NumAddedFiles());
    ASSERT_EQ(3u, manifests[1].NumAddedFiles());
}

TEST_F(CleanInteTest, TestDropPartitionAndExpireSnapshotWithIOException) {
    auto string_field = arrow::field("f0", arrow::utf8());
    auto int_field = arrow::field("f1", arrow::int32());
    auto int_field1 = arrow::field("f2", arrow::int32());
    auto double_field = arrow::field("f3", arrow::float64());
    auto schema =
        arrow::schema(arrow::FieldVector({string_field, int_field, int_field1, double_field}));
    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, "orc"},
        {Options::TARGET_FILE_SIZE, "1024"},
        {Options::FILE_SYSTEM, "local"},
        {Options::BUCKET, "2"},
        {Options::BUCKET_KEY, "f2"},
        {Options::SNAPSHOT_CLEAN_EMPTY_DIRECTORIES, "true"},
    };

    bool drop_partition_scanned_all_io_hook = false;
    bool expire_snapshot_scanned_all_io_hook = false;
    auto io_hook = IOHook::GetInstance();
    for (size_t i = 0; i < 1000; i += paimon::test::RandomNumber(5, 15)) {
        ::ArrowSchema arrow_schema;
        ASSERT_TRUE(arrow::ExportSchema(*schema, &arrow_schema).ok());
        auto dir = UniqueTestDirectory::Create();
        ASSERT_TRUE(dir);
        ASSERT_OK_AND_ASSIGN(std::string table_path,
                             CreateTestTable(dir->Str(), /*db_name=*/"foo",
                                             /*table_name=*/"bar", &arrow_schema,
                                             /*partition_keys=*/{"f1"},
                                             /*primary_keys=*/{}, options));
        WriteContextBuilder context_builder(table_path, "commit_user_1");
        ASSERT_OK_AND_ASSIGN(std::unique_ptr<WriteContext> write_context,
                             context_builder.SetOptions(options).WithStreamingMode(true).Finish());

        ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> file_store_write,
                             FileStoreWrite::Create(std::move(write_context)));

        ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_0,
                             MakeRecordBatch({{"Alice", 10, 1, 11.1}}, {{"f1", "10"}}, 0));
        ASSERT_OK(file_store_write->Write(std::move(batch_0)));
        ASSERT_OK_AND_ASSIGN(
            std::unique_ptr<RecordBatch> batch_1,
            MakeRecordBatch({{"Bob", 10, 0, 12.1}, {"Emily", 10, 0, 13.1}, {"Tony", 10, 0, 14.1}},
                            {{"f1", "10"}}, 1));
        ASSERT_OK(file_store_write->Write(std::move(batch_1)));
        ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_2,
                             MakeRecordBatch({{"Lucy", 20, 1, 14.1}}, {{"f1", "20"}}, 0));
        ASSERT_OK(file_store_write->Write(std::move(batch_2)));
        ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results,
                             file_store_write->PrepareCommit(/*wait_compaction=*/false, 0));
        ASSERT_EQ(3u, results.size());
        CommitContextBuilder commit_context_builder(table_path, "commit_user_1");
        ASSERT_OK_AND_ASSIGN(
            std::unique_ptr<CommitContext> commit_context,
            commit_context_builder.AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb")
                .WithFileSystem(file_system_)
                .AddOption(Options::SNAPSHOT_NUM_RETAINED_MAX, "2")
                .AddOption(Options::SNAPSHOT_NUM_RETAINED_MIN, "1")
                .AddOption(Options::SNAPSHOT_TIME_RETAINED, "1ms")
                .AddOption(Options::MANIFEST_MERGE_MIN_COUNT, "2")
                .AddOption(Options::SNAPSHOT_CLEAN_EMPTY_DIRECTORIES, "true")
                .IgnoreEmptyCommit(false)
                .Finish());
        ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context)));
        ASSERT_OK(commit->Commit(results, 0));

        ASSERT_OK_AND_ASSIGN(
            std::unique_ptr<RecordBatch> batch_3,
            MakeRecordBatch({{"Emily", 10, 0, 15.1}, {"Bob", 10, 0, 12.1}, {"Alex", 10, 0, 16.1}},
                            {{"f1", "10"}}, 1));
        ASSERT_OK(file_store_write->Write(std::move(batch_3)));
        ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_4,
                             MakeRecordBatch({{"Paul", 20, 1, 0}}, {{"f1", "20"}}, 0));
        ASSERT_OK(file_store_write->Write(std::move(batch_4)));

        ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results2,
                             file_store_write->PrepareCommit(/*wait_compaction=*/false, 1));
        ASSERT_EQ(3u, results2.size());
        ASSERT_OK(commit->Commit(results2, 1));
        io_hook->Reset(i, IOHook::Mode::RETURN_ERROR);
        auto status = commit->DropPartition({{{"f1", "10"}}}, 2);
        io_hook->Clear();
        if (HitIOHook(status, i)) {
            if (HitIOHookInCommitHint(status)) {
                continue;
            }
            ASSERT_OK(commit->DropPartition({{{"f1", "10"}}}, 2));
        } else {
            ASSERT_OK(status);
            drop_partition_scanned_all_io_hook = true;
        }
        auto commit_impl = dynamic_cast<FileStoreCommitImpl*>(commit.get());
        ASSERT_TRUE(commit_impl);
        ASSERT_OK_AND_ASSIGN(bool snapshot_exist,
                             commit_impl->snapshot_manager_->SnapshotExists(1));
        ASSERT_TRUE(snapshot_exist);
        ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(2));
        ASSERT_TRUE(snapshot_exist);
        ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(3));
        ASSERT_TRUE(snapshot_exist);
        ASSERT_OK_AND_ASSIGN(Snapshot snapshot_3, commit_impl->snapshot_manager_->LoadSnapshot(3));
        ASSERT_EQ(-7, snapshot_3.DeltaRecordCount().value());
        ASSERT_EQ(2, snapshot_3.TotalRecordCount().value());
        ASSERT_EQ(Snapshot::CommitKind::Overwrite(), snapshot_3.GetCommitKind());
        ASSERT_EQ(2, snapshot_3.CommitIdentifier());
        io_hook->Reset(i, IOHook::Mode::RETURN_ERROR);
        Result<int32_t> expire_result = commit->Expire();
        io_hook->Clear();
        if (HitIOHook(expire_result.status(), i)) {
            if (HitIOHookInCommitHint(expire_result.status())) {
                continue;
            }
            ASSERT_OK(commit->Expire());
        } else {
            ASSERT_OK(expire_result.status());
            expire_snapshot_scanned_all_io_hook = true;
        }
        ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(1));
        ASSERT_FALSE(snapshot_exist);
        ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(2));
        ASSERT_FALSE(snapshot_exist);
        ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(3));
        ASSERT_TRUE(snapshot_exist);
        ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_5,
                             MakeRecordBatch({{"Lucas", 20, 2, 22.2}}, {{"f1", "20"}}, 0));
        ASSERT_OK(file_store_write->Write(std::move(batch_5)));
        ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results3,
                             file_store_write->PrepareCommit(/*wait_compaction=*/false, 3));
        ASSERT_EQ(3u, results3.size());
        ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_6,
                             MakeRecordBatch({{"Ken", 20, 3, 33.3}}, {{"f1", "20"}}, 0));
        ASSERT_OK(file_store_write->Write(std::move(batch_6)));
        ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results4,
                             file_store_write->PrepareCommit(/*wait_compaction=*/false, 4));
        ASSERT_EQ(1u, results4.size());
        ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_7,
                             MakeRecordBatch({{"Dennis", 10, 4, 44.4}}, {{"f1", "10"}}, 0));
        ASSERT_OK(file_store_write->Write(std::move(batch_7)));
        ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results5,
                             file_store_write->PrepareCommit(/*wait_compaction=*/false, 5));
        ASSERT_EQ(2u, results5.size());

        results3.insert(results3.end(), results4.begin(), results4.end());
        results3.insert(results3.end(), results5.begin(), results5.end());
        ASSERT_OK(commit->Commit(results3, 6));
        ASSERT_OK(commit->Expire());
        ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(4));
        ASSERT_TRUE(snapshot_exist);
        ASSERT_OK_AND_ASSIGN(Snapshot snapshot_4, commit_impl->snapshot_manager_->LoadSnapshot(4));
        std::vector<ManifestFileMeta> manifests;
        ASSERT_OK(commit_impl->manifest_list_->ReadDataManifests(snapshot_4, &manifests));
        ASSERT_EQ(2u, manifests.size());
        ASSERT_EQ(2u, manifests[0].NumAddedFiles());
        ASSERT_EQ(3u, manifests[1].NumAddedFiles());
        if (drop_partition_scanned_all_io_hook && expire_snapshot_scanned_all_io_hook) {
            break;
        }
    }
    ASSERT_TRUE(drop_partition_scanned_all_io_hook);
    ASSERT_TRUE(expire_snapshot_scanned_all_io_hook);
}

TEST_F(CleanInteTest, TestOrphanFilesClean) {
    auto string_field = arrow::field("f0", arrow::utf8());
    auto int_field = arrow::field("f1", arrow::int32());
    auto int_field1 = arrow::field("f2", arrow::int32());
    auto double_field = arrow::field("f3", arrow::float64());
    auto schema =
        arrow::schema(arrow::FieldVector({string_field, int_field, int_field1, double_field}));

    std::string commit_user = "commit_user_1";
    ::ArrowSchema arrow_schema;
    ASSERT_TRUE(arrow::ExportSchema(*schema, &arrow_schema).ok());

    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, "orc"},
        {Options::TARGET_FILE_SIZE, "1024"},
        {Options::FILE_SYSTEM, "local"},
        {Options::BUCKET, "2"},
        {Options::BUCKET_KEY, "f3"},
        {Options::SNAPSHOT_CLEAN_EMPTY_DIRECTORIES, "true"},
    };

    auto dir = UniqueTestDirectory::Create();
    ASSERT_TRUE(dir);
    ASSERT_OK_AND_ASSIGN(std::string table_path,
                         CreateTestTable(dir->Str(), /*db_name=*/"foo",
                                         /*table_name=*/"bar", &arrow_schema,
                                         /*partition_keys=*/{"f1", "f2"},
                                         /*primary_keys=*/{}, options));

    WriteContextBuilder context_builder(table_path, commit_user);
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<WriteContext> write_context,
                         context_builder.SetOptions(options).WithStreamingMode(true).Finish());

    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> file_store_write,
                         FileStoreWrite::Create(std::move(write_context)));

    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_0,
                         MakeRecordBatch({{"Alice", 10, 1, 11.1}}, {{"f1", "10"}, {"f2", "1"}}, 0));
    ASSERT_OK(file_store_write->Write(std::move(batch_0)));
    ASSERT_OK_AND_ASSIGN(
        std::unique_ptr<RecordBatch> batch_1,
        MakeRecordBatch({{"Bob", 10, 0, 12.1}, {"Emily", 10, 0, 13.1}, {"Tony", 10, 0, 14.1}},
                        {{"f1", "10"}, {"f2", "0"}}, 1));
    ASSERT_OK(file_store_write->Write(std::move(batch_1)));
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_2,
                         MakeRecordBatch({{"Lucy", 20, 1, 14.1}}, {{"f1", "20"}, {"f2", "1"}}, 0));
    ASSERT_OK(file_store_write->Write(std::move(batch_2)));
    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results,
                         file_store_write->PrepareCommit(/*wait_compaction=*/false, 0));
    ASSERT_EQ(3u, results.size());
    CommitContextBuilder commit_context_builder(table_path, commit_user);
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<CommitContext> commit_context,
                         commit_context_builder.AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb")
                             .WithFileSystem(file_system_)
                             .IgnoreEmptyCommit(false)
                             .Finish());
    ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context)));
    ASSERT_OK(commit->Commit(results, 0));

    ASSERT_OK_AND_ASSIGN(
        std::unique_ptr<RecordBatch> batch_3,
        MakeRecordBatch({{"Emily", 10, 0, 15.1}, {"Bob", 10, 0, 12.1}, {"Alex", 10, 0, 16.1}},
                        {{"f1", "10"}, {"f2", "0"}}, 1));
    ASSERT_OK(file_store_write->Write(std::move(batch_3)));
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_4,
                         MakeRecordBatch({{"Paul", 20, 1, 0}}, {{"f1", "20"}, {"f2", "1"}}, 0));
    ASSERT_OK(file_store_write->Write(std::move(batch_4)));

    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results2,
                         file_store_write->PrepareCommit(/*wait_compaction=*/false, 1));
    ASSERT_EQ(3u, results2.size());
    ASSERT_OK(commit->Commit(results2, 1));

    ASSERT_OK(file_system_->WriteFile(PathUtil::JoinPath(table_path, "f1=10/f2=0/data-orphan1.orc"),
                                      "orphan", true));
    ASSERT_OK(file_system_->Mkdirs(PathUtil::JoinPath(table_path, "f1=20/f2=0/bucket-999")));
    ASSERT_OK(file_system_->WriteFile(
        PathUtil::JoinPath(table_path, "f1=10/f2=1/bucket-0/data-orphan2.orc"), "orphan", true));
    ASSERT_OK(file_system_->WriteFile(
        PathUtil::JoinPath(table_path, "f1=10/f2=1/bucket-0/data-orphan2.orc.index"), "orphan",
        true));
    ASSERT_OK(file_system_->WriteFile(
        PathUtil::JoinPath(table_path, "f1=10/f2=0/bucket-0/orphan3.test"), "orphan", true));
    ASSERT_OK(file_system_->WriteFile(
        PathUtil::JoinPath(table_path, "f1=10/f2=1/bucket-0/data-orphan4"), "orphan", true));
    ASSERT_OK(file_system_->WriteFile(PathUtil::JoinPath(table_path, "manifest/orphan5.test"),
                                      "orphan", true));
    ASSERT_OK(file_system_->WriteFile(PathUtil::JoinPath(table_path, "manifest/manifest-orphan"),
                                      "orphan", true));
    ASSERT_OK(file_system_->WriteFile(
        PathUtil::JoinPath(table_path, "manifest/.manifest-orphan.uuid.tmp"), "orphan", true));
    ASSERT_OK(file_system_->WriteFile(PathUtil::JoinPath(table_path, "snapshot/orphan6.test"),
                                      "orphan", true));
    ASSERT_OK(file_system_->WriteFile(PathUtil::JoinPath(table_path, "data-orphan7.parquet"),
                                      "orphan", true));

    {
        CleanContextBuilder clean_context_builder(table_path);
        ASSERT_OK_AND_ASSIGN(std::unique_ptr<CleanContext> clean_context,
                             clean_context_builder.WithFileSystem(file_system_).Finish());
        ASSERT_OK_AND_ASSIGN(auto cleaner, OrphanFilesCleaner::Create(std::move(clean_context)));
        ASSERT_OK_AND_ASSIGN(std::set<std::string> cleaned_paths, cleaner->Clean());
        ASSERT_TRUE(cleaned_paths.empty());
    }
    {
        CleanContextBuilder clean_context_builder(table_path);
        ASSERT_OK_AND_ASSIGN(std::unique_ptr<CleanContext> clean_context,
                             clean_context_builder.WithFileSystem(file_system_)
                                 .WithOlderThanMs(std::numeric_limits<int64_t>::max())
                                 .Finish());
        ASSERT_OK_AND_ASSIGN(auto cleaner, OrphanFilesCleaner::Create(std::move(clean_context)));
        ASSERT_OK_AND_ASSIGN(std::set<std::string> cleaned_paths, cleaner->Clean());
        ASSERT_TRUE(CheckEqual(
            cleaned_paths, {"data-orphan2.orc", "manifest-orphan", ".manifest-orphan.uuid.tmp"}));
    }
}

TEST_F(CleanInteTest, TestOrphanFilesCleanWithFileRetainCondition) {
    auto string_field = arrow::field("f0", arrow::utf8());
    auto int_field = arrow::field("f1", arrow::int32());
    auto int_field1 = arrow::field("f2", arrow::int32());
    auto double_field = arrow::field("f3", arrow::float64());
    auto schema =
        arrow::schema(arrow::FieldVector({string_field, int_field, int_field1, double_field}));

    std::string commit_user = "commit_user_1";
    ::ArrowSchema arrow_schema;
    ASSERT_TRUE(arrow::ExportSchema(*schema, &arrow_schema).ok());

    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, "orc"},
        {Options::TARGET_FILE_SIZE, "1024"},
        {Options::FILE_SYSTEM, "local"},
        {Options::BUCKET, "2"},
        {Options::BUCKET_KEY, "f3"},
        {Options::SNAPSHOT_CLEAN_EMPTY_DIRECTORIES, "true"},
    };

    auto dir = UniqueTestDirectory::Create();
    ASSERT_TRUE(dir);
    ASSERT_OK_AND_ASSIGN(std::string table_path,
                         CreateTestTable(dir->Str(), /*db_name=*/"foo",
                                         /*table_name=*/"bar", &arrow_schema,
                                         /*partition_keys=*/{"f1", "f2"},
                                         /*primary_keys=*/{}, options));

    WriteContextBuilder context_builder(table_path, commit_user);
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<WriteContext> write_context,
                         context_builder.SetOptions(options).WithStreamingMode(true).Finish());

    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> file_store_write,
                         FileStoreWrite::Create(std::move(write_context)));

    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_0,
                         MakeRecordBatch({{"Alice", 10, 1, 11.1}}, {{"f1", "10"}, {"f2", "1"}}, 0));
    ASSERT_OK(file_store_write->Write(std::move(batch_0)));
    ASSERT_OK_AND_ASSIGN(
        std::unique_ptr<RecordBatch> batch_1,
        MakeRecordBatch({{"Bob", 10, 0, 12.1}, {"Emily", 10, 0, 13.1}, {"Tony", 10, 0, 14.1}},
                        {{"f1", "10"}, {"f2", "0"}}, 1));
    ASSERT_OK(file_store_write->Write(std::move(batch_1)));
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_2,
                         MakeRecordBatch({{"Lucy", 20, 1, 14.1}}, {{"f1", "20"}, {"f2", "1"}}, 0));
    ASSERT_OK(file_store_write->Write(std::move(batch_2)));
    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results,
                         file_store_write->PrepareCommit(/*wait_compaction=*/false, 0));
    ASSERT_EQ(3u, results.size());
    CommitContextBuilder commit_context_builder(table_path, commit_user);
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<CommitContext> commit_context,
                         commit_context_builder.AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb")
                             .WithFileSystem(file_system_)
                             .IgnoreEmptyCommit(false)
                             .Finish());
    ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context)));
    ASSERT_OK(commit->Commit(results, 0));

    ASSERT_OK_AND_ASSIGN(
        std::unique_ptr<RecordBatch> batch_3,
        MakeRecordBatch({{"Emily", 10, 0, 15.1}, {"Bob", 10, 0, 12.1}, {"Alex", 10, 0, 16.1}},
                        {{"f1", "10"}, {"f2", "0"}}, 1));
    ASSERT_OK(file_store_write->Write(std::move(batch_3)));
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_4,
                         MakeRecordBatch({{"Paul", 20, 1, 0}}, {{"f1", "20"}, {"f2", "1"}}, 0));
    ASSERT_OK(file_store_write->Write(std::move(batch_4)));

    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results2,
                         file_store_write->PrepareCommit(/*wait_compaction=*/false, 1));
    ASSERT_EQ(3u, results2.size());
    ASSERT_OK(commit->Commit(results2, 1));

    ASSERT_OK(file_system_->WriteFile(PathUtil::JoinPath(table_path, "f1=10/f2=0/data-orphan1.orc"),
                                      "orphan", true));
    ASSERT_OK(file_system_->Mkdirs(PathUtil::JoinPath(table_path, "f1=20/f2=0/bucket-999")));
    ASSERT_OK(file_system_->WriteFile(
        PathUtil::JoinPath(table_path, "f1=10/f2=1/bucket-0/data-orphan2.orc"), "orphan", true));
    ASSERT_OK(file_system_->WriteFile(
        PathUtil::JoinPath(table_path, "f1=10/f2=1/bucket-0/data-orphan2.orc.index"), "orphan",
        true));
    ASSERT_OK(file_system_->WriteFile(
        PathUtil::JoinPath(table_path, "f1=10/f2=0/bucket-0/orphan3.test"), "orphan", true));
    ASSERT_OK(file_system_->WriteFile(
        PathUtil::JoinPath(table_path, "f1=10/f2=1/bucket-0/data-orphan4"), "orphan", true));
    ASSERT_OK(file_system_->WriteFile(PathUtil::JoinPath(table_path, "manifest/orphan5.test"),
                                      "orphan", true));
    ASSERT_OK(file_system_->WriteFile(PathUtil::JoinPath(table_path, "manifest/manifest-orphan"),
                                      "orphan", true));
    ASSERT_OK(file_system_->WriteFile(
        PathUtil::JoinPath(table_path, "manifest/.manifest-orphan.uuid.tmp"), "orphan", true));
    ASSERT_OK(file_system_->WriteFile(PathUtil::JoinPath(table_path, "snapshot/orphan6.test"),
                                      "orphan", true));
    ASSERT_OK(file_system_->WriteFile(PathUtil::JoinPath(table_path, "data-orphan7.parquet"),
                                      "orphan", true));

    CleanContextBuilder clean_context_builder(table_path);
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<CleanContext> clean_context,
                         clean_context_builder.WithFileSystem(file_system_)
                             .WithOlderThanMs(std::numeric_limits<int64_t>::max())
                             .WithFileRetainCondition([](const std::string& file_name) -> bool {
                                 if (file_name == "data-orphan2.orc") {
                                     return true;
                                 }
                                 return false;
                             })
                             .Finish());
    ASSERT_OK_AND_ASSIGN(auto cleaner, OrphanFilesCleaner::Create(std::move(clean_context)));
    ASSERT_OK_AND_ASSIGN(std::set<std::string> cleaned_paths, cleaner->Clean());
    ASSERT_TRUE(CheckEqual(cleaned_paths, {"manifest-orphan", ".manifest-orphan.uuid.tmp"}));
}

TEST_F(CleanInteTest, TestOrphanFilesCleanWithIOException) {
    auto string_field = arrow::field("f0", arrow::utf8());
    auto int_field = arrow::field("f1", arrow::int32());
    auto int_field1 = arrow::field("f2", arrow::int32());
    auto double_field = arrow::field("f3", arrow::float64());
    auto schema =
        arrow::schema(arrow::FieldVector({string_field, int_field, int_field1, double_field}));

    std::string commit_user = "commit_user_1";
    ::ArrowSchema arrow_schema;
    ASSERT_TRUE(arrow::ExportSchema(*schema, &arrow_schema).ok());

    auto dir = UniqueTestDirectory::Create();
    ASSERT_TRUE(dir);

    std::map<std::string, std::string> options = {
        {Options::MANIFEST_FORMAT, "orc"},
        {Options::FILE_FORMAT, "orc"},
        {Options::TARGET_FILE_SIZE, "1024"},
        {Options::FILE_SYSTEM, "local"},
        {Options::BUCKET, "2"},
        {Options::BUCKET_KEY, "f2"},
        {Options::SNAPSHOT_CLEAN_EMPTY_DIRECTORIES, "true"}};

    ASSERT_OK_AND_ASSIGN(std::string table_path,
                         CreateTestTable(dir->Str(), /*db_name=*/"foo",
                                         /*table_name=*/"bar", &arrow_schema,
                                         /*partition_keys=*/{"f1"},
                                         /*primary_keys=*/{}, options));

    WriteContextBuilder context_builder(table_path, commit_user);
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<WriteContext> write_context,
                         context_builder.SetOptions(options).WithStreamingMode(true).Finish());
    std::string root_path = write_context->GetRootPath();
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> file_store_write,
                         FileStoreWrite::Create(std::move(write_context)));

    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_0,
                         MakeRecordBatch({{"Alice", 10, 1, 11.1}}, {{"f1", "10"}}, 0));
    ASSERT_OK(file_store_write->Write(std::move(batch_0)));
    ASSERT_OK_AND_ASSIGN(
        std::unique_ptr<RecordBatch> batch_1,
        MakeRecordBatch({{"Bob", 10, 0, 12.1}, {"Emily", 10, 0, 13.1}, {"Tony", 10, 0, 14.1}},
                        {{"f1", "10"}}, 1));
    ASSERT_OK(file_store_write->Write(std::move(batch_1)));
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_2,
                         MakeRecordBatch({{"Lucy", 20, 1, 14.1}}, {{"f1", "20"}}, 0));
    ASSERT_OK(file_store_write->Write(std::move(batch_2)));
    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results,
                         file_store_write->PrepareCommit(/*wait_compaction=*/false, 0));
    ASSERT_EQ(3u, results.size());
    CommitContextBuilder commit_context_builder(root_path, commit_user);
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<CommitContext> commit_context,
                         commit_context_builder.AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb")
                             .WithFileSystem(file_system_)
                             .AddOption(Options::BUCKET, "2")
                             .IgnoreEmptyCommit(false)
                             .Finish());
    ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context)));
    ASSERT_OK(commit->Commit(results, 0));

    ASSERT_OK_AND_ASSIGN(
        std::unique_ptr<RecordBatch> batch_3,
        MakeRecordBatch({{"Emily", 10, 0, 15.1}, {"Bob", 10, 0, 12.1}, {"Alex", 10, 0, 16.1}},
                        {{"f1", "10"}}, 1));
    ASSERT_OK(file_store_write->Write(std::move(batch_3)));
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_4,
                         MakeRecordBatch({{"Paul", 20, 1, 0}}, {{"f1", "20"}}, 0));
    ASSERT_OK(file_store_write->Write(std::move(batch_4)));

    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results2,
                         file_store_write->PrepareCommit(/*wait_compaction=*/false, 1));
    ASSERT_EQ(3u, results2.size());
    ASSERT_OK(commit->Commit(results2, 1));

    auto write_orphan_files = [this](const std::string& root_path) {
        ASSERT_OK(file_system_->WriteFile(PathUtil::JoinPath(root_path, "f1=10/data-orphan1.orc"),
                                          "orphan", true));
        ASSERT_OK(file_system_->Mkdirs(PathUtil::JoinPath(root_path, "f1=20/bucket-999")));
        ASSERT_OK(file_system_->WriteFile(
            PathUtil::JoinPath(root_path, "f1=10/bucket-0/data-orphan2.orc"), "orphan", true));
        ASSERT_OK(file_system_->WriteFile(
            PathUtil::JoinPath(root_path, "f1=10/bucket-0/data-orphan2.orc.index"), "orphan",
            true));
        ASSERT_OK(file_system_->WriteFile(
            PathUtil::JoinPath(root_path, "f1=10/bucket-0/orphan3.test"), "orphan", true));
        ASSERT_OK(file_system_->WriteFile(
            PathUtil::JoinPath(root_path, "f1=10/bucket-0/data-orphan4"), "orphan", true));
        ASSERT_OK(file_system_->WriteFile(PathUtil::JoinPath(root_path, "manifest/orphan5.test"),
                                          "orphan", true));
        ASSERT_OK(file_system_->WriteFile(PathUtil::JoinPath(root_path, "manifest/manifest-orphan"),
                                          "orphan", true));
        ASSERT_OK(file_system_->WriteFile(
            PathUtil::JoinPath(root_path, "manifest/.manifest-orphan.uuid.tmp"), "orphan", true));
        ASSERT_OK(file_system_->WriteFile(PathUtil::JoinPath(root_path, "snapshot/orphan6.test"),
                                          "orphan", true));
        ASSERT_OK(file_system_->WriteFile(PathUtil::JoinPath(root_path, "data-orphan7.parquet"),
                                          "orphan", true));
    };

    bool scanned_all_io_hook = true;
    auto io_hook = IOHook::GetInstance();
    for (size_t i = 0; i < 500; i += paimon::test::RandomNumber(5, 15)) {
        scanned_all_io_hook = true;
        write_orphan_files(root_path);
        {
            CleanContextBuilder clean_context_builder(root_path);
            ASSERT_OK_AND_ASSIGN(std::unique_ptr<CleanContext> clean_context,
                                 clean_context_builder.WithFileSystem(file_system_).Finish());
            ASSERT_OK_AND_ASSIGN(auto cleaner,
                                 OrphanFilesCleaner::Create(std::move(clean_context)));
            io_hook->Reset(i, IOHook::Mode::RETURN_ERROR);
            auto clean_result = cleaner->Clean();
            io_hook->Clear();
            if (HitIOHook(clean_result.status(), i)) {
                scanned_all_io_hook = false;
                clean_result = cleaner->Clean();
            }
            ASSERT_OK(clean_result);
            ASSERT_TRUE(clean_result.value().empty());
        }
        {
            CleanContextBuilder clean_context_builder(root_path);
            ASSERT_OK_AND_ASSIGN(std::unique_ptr<CleanContext> clean_context,
                                 clean_context_builder.WithFileSystem(file_system_)
                                     .WithOlderThanMs(std::numeric_limits<int64_t>::max())
                                     .Finish());

            ASSERT_OK_AND_ASSIGN(auto cleaner,
                                 OrphanFilesCleaner::Create(std::move(clean_context)));

            io_hook->Reset(i, IOHook::Mode::RETURN_ERROR);
            auto clean_result = cleaner->Clean();  // first clean
            io_hook->Clear();
            if (HitIOHook(clean_result.status(), i)) {
                scanned_all_io_hook = false;
                clean_result = cleaner->Clean();  // second clean
            }
            ASSERT_OK(clean_result);

            std::set<std::string> expected_orphans = {"data-orphan2.orc", "manifest-orphan",
                                                      ".manifest-orphan.uuid.tmp"};
            // In the first clean, IO errors may already have been triggered in
            // TryBestListingDirs or MinimalTryBestListingDirs. Those errors are handled quietly,
            // which means the cleaner may build an incomplete view of the full file set.
            // As a result, the subsequent delete phase may only run on part of the expected orphan
            // files. So here we only require the cleaned result to be a subset of
            // expected_orphans instead of an exact match.
            std::set<std::string> cleaned_file_names;
            for (const auto& file_path : clean_result.value()) {
                cleaned_file_names.insert(PathUtil::GetName(file_path));
            }
            ASSERT_TRUE(std::includes(expected_orphans.begin(), expected_orphans.end(),
                                      cleaned_file_names.begin(), cleaned_file_names.end()));
            // because clean is quietly delete, if touch IO exception in Delete() in the first
            // clean, Clean() will return OK, but files may not be deleted, so need to call Clean()
            // again. If the third call Clean() return empty, it means all files are deleted and
            // already scanned all io hooks.
            clean_result = cleaner->Clean();  // third clean
            ASSERT_OK(clean_result);
            if (!clean_result.value().empty()) {
                scanned_all_io_hook = false;
            }
        }
        if (scanned_all_io_hook) {
            break;
        }
    }
    ASSERT_TRUE(scanned_all_io_hook);
}

}  // namespace paimon::test