// 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 <gen_cpp/AgentService_types.h>
#include <gen_cpp/olap_file.pb.h>
#include <gtest/gtest-message.h>
#include <gtest/gtest-test-part.h>
#include <gtest/gtest.h>

#include <memory>

#include "cloud/cloud_base_compaction.h"
#include "cloud/cloud_cluster_info.h"
#include "cloud/cloud_storage_engine.h"
#include "cloud/cloud_tablet.h"
#include "cloud/cloud_tablet_mgr.h"
#include "json2pb/json_to_pb.h"
#include "olap/olap_common.h"
#include "olap/rowset/rowset_factory.h"
#include "olap/rowset/rowset_meta.h"
#include "olap/storage_policy.h"
#include "olap/tablet_meta.h"
#include "util/uid_util.h"

namespace doris {
class TabletMap;

class CloudCompactionTest : public testing::Test {
    CloudCompactionTest() : _engine(CloudStorageEngine(EngineOptions {})) {}
    void SetUp() override {
        config::compaction_promotion_size_mbytes = 1024;
        config::compaction_promotion_ratio = 0.05;
        config::compaction_promotion_min_size_mbytes = 64;
        config::compaction_min_size_mbytes = 64;

        _tablet_meta.reset(new TabletMeta(1, 2, 15673, 15674, 4, 5, TTabletSchema(), 6, {{7, 8}},
                                          UniqueId(9, 10), TTabletType::TABLET_TYPE_DISK,
                                          TCompressionType::LZ4F));

        _json_rowset_meta = R"({
            "rowset_id": 540081,
            "tablet_id": 15673,
            "txn_id": 4042,
            "tablet_schema_hash": 567997577,
            "rowset_type": "BETA_ROWSET",
            "rowset_state": "VISIBLE",
            "start_version": 2,
            "end_version": 2,
            "num_rows": 3929,
            "total_disk_size": 41,
            "data_disk_size": 41,
            "index_disk_size": 235,
            "empty": false,
            "load_id": {
                "hi": -5350970832824939812,
                "lo": -6717994719194512122
            },
            "creation_time": 1553765670,
            "num_segments": 3
        })";
        _cluster_info = std::make_shared<CloudClusterInfo>();
        _cluster_info->_is_in_standby = false;
        ExecEnv::GetInstance()->_cluster_info = _cluster_info.get();
    }
    void TearDown() override {}

    void init_rs_meta(RowsetMetaSharedPtr& pb1, int64_t start, int64_t end) {
        RowsetMetaPB rowset_meta_pb;
        json2pb::JsonToProtoMessage(_json_rowset_meta, &rowset_meta_pb);
        rowset_meta_pb.set_start_version(start);
        rowset_meta_pb.set_end_version(end);
        rowset_meta_pb.set_creation_time(10000);

        pb1->init_from_pb(rowset_meta_pb);
        pb1->set_total_disk_size(41);
        pb1->set_tablet_schema(_tablet_meta->tablet_schema());
    }

    void init_rs_meta_small_base(std::vector<RowsetMetaSharedPtr>* rs_metas) {
        RowsetMetaSharedPtr ptr1(new RowsetMeta());
        init_rs_meta(ptr1, 0, 0);
        rs_metas->push_back(ptr1);

        RowsetMetaSharedPtr ptr2(new RowsetMeta());
        init_rs_meta(ptr2, 1, 1);
        rs_metas->push_back(ptr2);

        RowsetMetaSharedPtr ptr3(new RowsetMeta());
        init_rs_meta(ptr3, 2, 2);
        rs_metas->push_back(ptr3);

        RowsetMetaSharedPtr ptr4(new RowsetMeta());
        init_rs_meta(ptr4, 3, 3);
        rs_metas->push_back(ptr4);

        RowsetMetaSharedPtr ptr5(new RowsetMeta());
        init_rs_meta(ptr5, 4, 4);
        rs_metas->push_back(ptr5);
    }

protected:
    std::string _json_rowset_meta;
    TabletMetaSharedPtr _tablet_meta;

public:
    CloudStorageEngine _engine;
    std::shared_ptr<CloudClusterInfo> _cluster_info;
};

TEST_F(CloudCompactionTest, failure_base_compaction_tablet_sleep_test) {
    auto filter_out = [](CloudTablet* t) { return false; };
    CloudTabletMgr mgr(_engine);

    std::vector<RowsetMetaSharedPtr> rs_metas;
    init_rs_meta_small_base(&rs_metas);

    CloudTabletSPtr tablet1 = std::make_shared<CloudTablet>(_engine, _tablet_meta);
    for (auto& rs_meta : rs_metas) {
        static_cast<void>(_tablet_meta->add_rs_meta(rs_meta));
    }
    tablet1->tablet_meta()->_tablet_id = 10000;
    tablet1->set_last_base_compaction_failure_time(
            duration_cast<std::chrono::milliseconds>(
                    std::chrono::system_clock::now().time_since_epoch())
                    .count() -
            100000);
    tablet1->set_last_base_compaction_failure_time(0);
    tablet1->tablet_meta()->tablet_schema()->set_disable_auto_compaction(false);
    tablet1->_approximate_num_rowsets = 10;
    mgr.put_tablet_for_UT(tablet1);

    int64_t max_score;
    std::vector<std::shared_ptr<CloudTablet>> tablets {};
    Status st = mgr.get_topn_tablets_to_compact(1, CompactionType::BASE_COMPACTION, filter_out,
                                                &tablets, &max_score);
    ASSERT_EQ(st, Status::OK());
    ASSERT_EQ(tablets.size(), 1);

    tablet1->set_last_base_compaction_failure_time(
            duration_cast<std::chrono::milliseconds>(
                    std::chrono::system_clock::now().time_since_epoch())
                    .count());
    st = mgr.get_topn_tablets_to_compact(1, CompactionType::BASE_COMPACTION, filter_out, &tablets,
                                         &max_score);
    ASSERT_EQ(st, Status::OK());
    ASSERT_EQ(tablets.size(), 0);
}

TEST_F(CloudCompactionTest, failure_cumu_compaction_tablet_sleep_test) {
    auto filter_out = [](CloudTablet* t) { return false; };
    CloudTabletMgr mgr(_engine);

    std::vector<RowsetMetaSharedPtr> rs_metas;
    init_rs_meta_small_base(&rs_metas);

    CloudTabletSPtr tablet1 = std::make_shared<CloudTablet>(_engine, _tablet_meta);
    for (auto& rs_meta : rs_metas) {
        static_cast<void>(_tablet_meta->add_rs_meta(rs_meta));
    }
    tablet1->tablet_meta()->_tablet_id = 10000;
    tablet1->set_last_cumu_compaction_failure_time(
            duration_cast<std::chrono::milliseconds>(
                    std::chrono::system_clock::now().time_since_epoch())
                    .count() -
            100000);
    tablet1->set_last_cumu_compaction_failure_time(0);
    tablet1->tablet_meta()->tablet_schema()->set_disable_auto_compaction(false);
    tablet1->_approximate_cumu_num_deltas = 10;
    mgr.put_tablet_for_UT(tablet1);

    int64_t max_score;
    std::vector<std::shared_ptr<CloudTablet>> tablets {};
    Status st = mgr.get_topn_tablets_to_compact(1, CompactionType::CUMULATIVE_COMPACTION,
                                                filter_out, &tablets, &max_score);
    ASSERT_EQ(st, Status::OK());
    ASSERT_EQ(tablets.size(), 1);

    tablet1->set_last_cumu_compaction_failure_time(
            duration_cast<std::chrono::milliseconds>(
                    std::chrono::system_clock::now().time_since_epoch())
                    .count());
    st = mgr.get_topn_tablets_to_compact(1, CompactionType::BASE_COMPACTION, filter_out, &tablets,
                                         &max_score);
    ASSERT_EQ(st, Status::OK());
    ASSERT_EQ(tablets.size(), 0);
}

static RowsetSharedPtr create_rowset(Version version, int num_segments, bool overlapping,
                                     int data_size) {
    auto rs_meta = std::make_shared<RowsetMeta>();
    rs_meta->set_rowset_type(BETA_ROWSET); // important
    rs_meta->_rowset_meta_pb.set_start_version(version.first);
    rs_meta->_rowset_meta_pb.set_end_version(version.second);
    rs_meta->set_num_segments(num_segments);
    rs_meta->set_segments_overlap(overlapping ? OVERLAPPING : NONOVERLAPPING);
    rs_meta->set_total_disk_size(data_size);
    RowsetSharedPtr rowset;
    Status st = RowsetFactory::create_rowset(nullptr, "", rs_meta, &rowset);
    if (!st.ok()) {
        return nullptr;
    }
    return rowset;
}

class TestableCloudCompaction : public CloudCompactionMixin {
public:
    TestableCloudCompaction(CloudStorageEngine& engine, CloudTabletSPtr tablet)
            : CloudCompactionMixin(engine, tablet, "test_compaction") {}

    // Set input rowsets for testing
    void set_input_rowsets(const std::vector<RowsetSharedPtr>& rowsets) {
        _input_rowsets = rowsets;
    }

    // Get input rowsets for verification
    const std::vector<RowsetSharedPtr>& get_input_rowsets() const { return _input_rowsets; }

    // Expose the protected method for testing
    size_t test_apply_txn_size_truncation_and_log(const std::string& compaction_name) {
        return apply_txn_size_truncation_and_log(compaction_name);
    }

    Status prepare_compact() override { return Status::OK(); }

    ReaderType compaction_type() const override { return ReaderType::READER_CUMULATIVE_COMPACTION; }

    std::string_view compaction_name() const override { return "test_compaction"; }
};

TEST_F(CloudCompactionTest, test_set_storage_resource_from_input_rowsets) {
    S3Conf s3_conf {.bucket = "bucket",
                    .prefix = "prefix",
                    .client_conf = {
                            .endpoint = "endpoint",
                            .region = "region",
                            .ak = "ak",
                            .sk = "sk",
                            .token = "",
                            .bucket = "",
                            .role_arn = "",
                            .external_id = "",
                    }};
    std::string resource_id = "10000";
    auto res = io::S3FileSystem::create(std::move(s3_conf), resource_id);
    ASSERT_TRUE(res.has_value()) << res.error();
    auto fs = res.value();
    StorageResource storage_resource(fs);

    CloudTabletSPtr tablet = std::make_shared<CloudTablet>(_engine, _tablet_meta);
    TestableCloudCompaction compaction(_engine, tablet);

    // Test case 1: All rowsets are empty (num_segments = 0) - should succeed
    {
        std::vector<RowsetSharedPtr> rowsets;

        RowsetSharedPtr rowset1 = create_rowset(Version(2, 2), 0, false, 41);
        ASSERT_TRUE(rowset1 != nullptr);
        rowset1->set_hole_rowset(true); // Mark as hole rowset since num_segments=0
        rowsets.push_back(rowset1);

        RowsetSharedPtr rowset2 = create_rowset(Version(3, 3), 0, false, 41);
        ASSERT_TRUE(rowset2 != nullptr);
        rowset2->set_hole_rowset(true); // Mark as hole rowset since num_segments=0
        rowsets.push_back(rowset2);

        compaction.set_input_rowsets(rowsets);

        RowsetWriterContext ctx;
        Status st = compaction.set_storage_resource_from_input_rowsets(ctx);
        ASSERT_TRUE(st.ok()) << st.to_string();
        // No storage resource should be set since no rowset has resource_id
        ASSERT_FALSE(ctx.storage_resource.has_value());
    }

    // Test case 2: Backward iteration - last rowset has resource_id
    {
        std::vector<RowsetSharedPtr> rowsets;

        // First rowset: empty, no resource_id
        RowsetSharedPtr rowset1 = create_rowset(Version(2, 2), 0, false, 41);
        ASSERT_TRUE(rowset1 != nullptr);
        rowset1->set_hole_rowset(true);
        rowsets.push_back(rowset1);

        // Second rowset: empty, no resource_id
        RowsetSharedPtr rowset2 = create_rowset(Version(3, 3), 0, false, 41);
        ASSERT_TRUE(rowset2 != nullptr);
        rowset2->set_hole_rowset(true);
        rowsets.push_back(rowset2);

        // Third rowset: has resource_id (should be found during backward iteration)
        RowsetSharedPtr rowset3 = create_rowset(Version(4, 4), 1, false, 41);
        ASSERT_TRUE(rowset3 != nullptr);
        rowset3->rowset_meta()->set_remote_storage_resource(storage_resource);
        rowsets.push_back(rowset3);

        compaction.set_input_rowsets(rowsets);

        RowsetWriterContext ctx;
        Status st = compaction.set_storage_resource_from_input_rowsets(ctx);
        ASSERT_TRUE(st.ok()) << st.to_string();
        // Storage resource should be set from rowset3
        ASSERT_TRUE(ctx.storage_resource.has_value());
    }

    // Test case 3: Multiple rowsets with resource_id - should use the last one (backward iteration)
    {
        std::vector<RowsetSharedPtr> rowsets;

        // First rowset: has resource_id
        RowsetSharedPtr rowset1 = create_rowset(Version(2, 2), 1, false, 41);
        ASSERT_TRUE(rowset1 != nullptr);
        StorageResource first_resource(fs);
        rowset1->rowset_meta()->set_remote_storage_resource(first_resource);
        rowsets.push_back(rowset1);

        // Second rowset: empty, no resource_id
        RowsetSharedPtr rowset2 = create_rowset(Version(3, 3), 0, false, 41);
        ASSERT_TRUE(rowset2 != nullptr);
        rowset2->set_hole_rowset(true);
        rowsets.push_back(rowset2);

        // Third rowset: has different resource_id (should be used due to backward iteration)
        RowsetSharedPtr rowset3 = create_rowset(Version(4, 4), 1, false, 41);
        ASSERT_TRUE(rowset3 != nullptr);
        rowset3->rowset_meta()->set_remote_storage_resource(storage_resource);
        rowsets.push_back(rowset3);

        compaction.set_input_rowsets(rowsets);

        RowsetWriterContext ctx;
        Status st = compaction.set_storage_resource_from_input_rowsets(ctx);
        ASSERT_TRUE(st.ok()) << st.to_string();
        // Storage resource should be set from rowset3 (last one with resource_id)
        ASSERT_TRUE(ctx.storage_resource.has_value());
    }

    // Test case 4: Non-empty rowset in the middle without resource_id - should fail
    {
        std::vector<RowsetSharedPtr> rowsets;

        // First rowset: has resource_id
        RowsetSharedPtr rowset1 = create_rowset(Version(2, 2), 1, false, 41);
        ASSERT_TRUE(rowset1 != nullptr);
        rowset1->rowset_meta()->set_remote_storage_resource(storage_resource);
        rowsets.push_back(rowset1);

        // Second rowset: non-empty but no resource_id (invalid)
        RowsetSharedPtr rowset2 = create_rowset(Version(3, 3), 2, false, 41);
        ASSERT_TRUE(rowset2 != nullptr);
        // Intentionally don't set resource_id
        rowsets.push_back(rowset2);

        // Third rowset: empty, no resource_id
        RowsetSharedPtr rowset3 = create_rowset(Version(4, 4), 0, false, 41);
        ASSERT_TRUE(rowset3 != nullptr);
        rowset3->set_hole_rowset(true); // Mark as hole rowset since num_segments=0
        rowsets.push_back(rowset3);

        compaction.set_input_rowsets(rowsets);

        RowsetWriterContext ctx;
        Status st = compaction.set_storage_resource_from_input_rowsets(ctx);
        ASSERT_TRUE(st.is<ErrorCode::INTERNAL_ERROR>());
        ASSERT_TRUE(st.to_string().find("Non-empty rowset must have valid resource_id") !=
                    std::string::npos)
                << st.to_string();
    }

    // Test case 5: Empty input rowsets - should succeed
    {
        std::vector<RowsetSharedPtr> rowsets; // Empty vector

        compaction.set_input_rowsets(rowsets);

        RowsetWriterContext ctx;
        Status st = compaction.set_storage_resource_from_input_rowsets(ctx);
        ASSERT_TRUE(st.ok()) << st.to_string();
        // No storage resource should be set
        ASSERT_FALSE(ctx.storage_resource.has_value());
    }
}
TEST_F(CloudCompactionTest, should_cache_compaction_output) {
    CloudTabletSPtr tablet = std::make_shared<CloudTablet>(_engine, std::make_shared<TabletMeta>());
    CloudBaseCompaction cloud_base_compaction(_engine, tablet);
    cloud_base_compaction._input_rowsets_total_size = 0;
    cloud_base_compaction._input_rowsets_cached_data_size = 0;
    cloud_base_compaction._input_rowsets_cached_index_size = 0;
    ASSERT_EQ(cloud_base_compaction.should_cache_compaction_output(), false);

    cloud_base_compaction._input_rowsets_total_size = 100;
    cloud_base_compaction._input_rowsets_cached_data_size = 0;
    cloud_base_compaction._input_rowsets_cached_index_size = 0;
    ASSERT_EQ(cloud_base_compaction.should_cache_compaction_output(), false);

    cloud_base_compaction._input_rowsets_total_size = 100;
    cloud_base_compaction._input_rowsets_cached_data_size = 70;
    cloud_base_compaction._input_rowsets_cached_index_size = 0;
    ASSERT_EQ(cloud_base_compaction.should_cache_compaction_output(), false);

    cloud_base_compaction._input_rowsets_total_size = 100;
    cloud_base_compaction._input_rowsets_cached_data_size = 0;
    cloud_base_compaction._input_rowsets_cached_index_size = 70;
    ASSERT_EQ(cloud_base_compaction.should_cache_compaction_output(), false);

    cloud_base_compaction._input_rowsets_total_size = 100;
    cloud_base_compaction._input_rowsets_cached_data_size = 0;
    cloud_base_compaction._input_rowsets_cached_index_size = 70;
    ASSERT_EQ(cloud_base_compaction.should_cache_compaction_output(), false);

    cloud_base_compaction._input_rowsets_total_size = 100;
    cloud_base_compaction._input_rowsets_cached_data_size = 80;
    cloud_base_compaction._input_rowsets_cached_index_size = 0;
    ASSERT_EQ(cloud_base_compaction.should_cache_compaction_output(), true);

    cloud_base_compaction._input_rowsets_total_size = 100;
    cloud_base_compaction._input_rowsets_cached_data_size = 0;
    cloud_base_compaction._input_rowsets_cached_index_size = 80;
    ASSERT_EQ(cloud_base_compaction.should_cache_compaction_output(), true);

    cloud_base_compaction._input_rowsets_total_size = 100;
    cloud_base_compaction._input_rowsets_cached_data_size = 50;
    cloud_base_compaction._input_rowsets_cached_index_size = 50;
    ASSERT_EQ(cloud_base_compaction.should_cache_compaction_output(), true);
    LOG(INFO) << "should_cache_compaction_output done";
}

TEST_F(CloudCompactionTest, test_truncate_rowsets_by_txn_size_empty_input) {
    std::vector<RowsetSharedPtr> rowsets;
    int64_t kept_size = 100;
    int64_t truncated_size = 50;

    size_t truncated = cloud::truncate_rowsets_by_txn_size(rowsets, kept_size, truncated_size);

    ASSERT_EQ(truncated, 0);
    ASSERT_EQ(kept_size, 0);
    ASSERT_EQ(truncated_size, 0);
    ASSERT_EQ(rowsets.size(), 0);
}

TEST_F(CloudCompactionTest, test_truncate_rowsets_by_txn_size_single_rowset_under_limit) {
    // Create a single rowset
    std::vector<RowsetSharedPtr> rowsets;
    RowsetSharedPtr rowset1 = create_rowset(Version(2, 2), 1, false, 1024);
    ASSERT_TRUE(rowset1 != nullptr);
    rowsets.push_back(rowset1);

    // Set a large max size
    config::compaction_txn_max_size_bytes = 1024 * 1024 * 1024; // 1GB

    int64_t kept_size = 0;
    int64_t truncated_size = 0;

    size_t truncated = cloud::truncate_rowsets_by_txn_size(rowsets, kept_size, truncated_size);

    ASSERT_EQ(truncated, 0);
    ASSERT_EQ(rowsets.size(), 1);
    ASSERT_GT(kept_size, 0);
    ASSERT_EQ(truncated_size, 0);
}

TEST_F(CloudCompactionTest, test_truncate_rowsets_by_txn_size_multiple_rowsets_all_fit) {
    std::vector<RowsetSharedPtr> rowsets;
    for (int i = 0; i < 5; i++) {
        RowsetSharedPtr rowset = create_rowset(Version(i, i), 1, false, 1024);
        ASSERT_TRUE(rowset != nullptr);
        rowsets.push_back(rowset);
    }

    config::compaction_txn_max_size_bytes = 1024 * 1024 * 1024; // 1GB

    int64_t kept_size = 0;
    int64_t truncated_size = 0;

    size_t truncated = cloud::truncate_rowsets_by_txn_size(rowsets, kept_size, truncated_size);

    ASSERT_EQ(truncated, 0);
    ASSERT_EQ(rowsets.size(), 5);
    ASSERT_GT(kept_size, 0);
    ASSERT_EQ(truncated_size, 0);
}

TEST_F(CloudCompactionTest, test_truncate_rowsets_by_txn_size_exceeds_limit) {
    std::vector<RowsetSharedPtr> rowsets;
    for (int i = 0; i < 10; i++) {
        RowsetSharedPtr rowset = create_rowset(Version(i, i), 1, false, 1024);
        ASSERT_TRUE(rowset != nullptr);
        rowsets.push_back(rowset);
    }

    // Set a very small max size to force truncation
    config::compaction_txn_max_size_bytes = 50; // 50 bytes, should keep only a few rowsets

    int64_t kept_size = 0;
    int64_t truncated_size = 0;

    size_t truncated = cloud::truncate_rowsets_by_txn_size(rowsets, kept_size, truncated_size);

    // Should truncate some rowsets
    ASSERT_GT(truncated, 0);
    ASSERT_LT(rowsets.size(), 10);
    ASSERT_GT(rowsets.size(), 0); // At least 1 rowset kept
    ASSERT_GT(truncated_size, 0);
}

TEST_F(CloudCompactionTest, test_truncate_rowsets_by_txn_size_first_rowset_exceeds_limit) {
    std::vector<RowsetSharedPtr> rowsets;
    RowsetSharedPtr rowset1 = create_rowset(Version(0, 0), 1, false, 1024);
    ASSERT_TRUE(rowset1 != nullptr);
    rowsets.push_back(rowset1);

    // Set max size smaller than the first rowset's metadata size
    config::compaction_txn_max_size_bytes = 1; // 1 byte

    int64_t kept_size = 0;
    int64_t truncated_size = 0;

    size_t truncated = cloud::truncate_rowsets_by_txn_size(rowsets, kept_size, truncated_size);

    // Should keep at least 1 rowset even if it exceeds the limit
    ASSERT_EQ(truncated, 0);
    ASSERT_EQ(rowsets.size(), 1);
    ASSERT_GT(kept_size, config::compaction_txn_max_size_bytes);
    ASSERT_EQ(truncated_size, 0);
}

TEST_F(CloudCompactionTest, test_truncate_rowsets_by_txn_size_exact_boundary) {
    std::vector<RowsetSharedPtr> rowsets;
    RowsetSharedPtr rowset1 = create_rowset(Version(0, 0), 1, false, 1024);
    ASSERT_TRUE(rowset1 != nullptr);
    rowsets.push_back(rowset1);

    // Get the actual size of the first rowset
    int64_t first_kept_size = 0;
    int64_t first_truncated_size = 0;

    std::vector<RowsetSharedPtr> temp_rowsets = {rowset1};
    cloud::truncate_rowsets_by_txn_size(temp_rowsets, first_kept_size, first_truncated_size);

    // Add more rowsets
    for (int i = 1; i < 5; i++) {
        RowsetSharedPtr rowset = create_rowset(Version(i, i), 1, false, 1024);
        ASSERT_TRUE(rowset != nullptr);
        rowsets.push_back(rowset);
    }

    // Set max size to exactly the size of first rowset
    config::compaction_txn_max_size_bytes = first_kept_size;

    int64_t kept_size = 0;
    int64_t truncated_size = 0;

    size_t truncated = cloud::truncate_rowsets_by_txn_size(rowsets, kept_size, truncated_size);

    // Should keep only 1 rowset at the boundary
    ASSERT_EQ(rowsets.size(), 1);
    ASSERT_EQ(truncated, 4);
    ASSERT_EQ(kept_size, first_kept_size);
    ASSERT_GT(truncated_size, 0);
}

TEST_F(CloudCompactionTest, test_truncate_rowsets_by_txn_size_output_parameters) {
    std::vector<RowsetSharedPtr> rowsets;
    for (int i = 0; i < 3; i++) {
        RowsetSharedPtr rowset = create_rowset(Version(i, i), 1, false, 1024);
        ASSERT_TRUE(rowset != nullptr);
        rowsets.push_back(rowset);
    }

    config::compaction_txn_max_size_bytes = 1024 * 1024;

    int64_t kept_size = 0;
    int64_t truncated_size = 0;

    size_t truncated = cloud::truncate_rowsets_by_txn_size(rowsets, kept_size, truncated_size);

    // Verify output parameters are set correctly
    ASSERT_EQ(truncated, 0); // All rowsets fit
    ASSERT_EQ(rowsets.size(), 3);
    ASSERT_GT(kept_size, 0);
    ASSERT_EQ(truncated_size, 0);
}

TEST_F(CloudCompactionTest, test_apply_txn_size_truncation_and_log_empty_input) {
    CloudTabletSPtr tablet = std::make_shared<CloudTablet>(_engine, _tablet_meta);
    TestableCloudCompaction compaction(_engine, tablet);

    // Test with empty input rowsets
    std::vector<RowsetSharedPtr> empty_rowsets;
    compaction.set_input_rowsets(empty_rowsets);

    size_t truncated = compaction.test_apply_txn_size_truncation_and_log("test_compaction");

    ASSERT_EQ(truncated, 0);
    ASSERT_EQ(compaction.get_input_rowsets().size(), 0);
}

TEST_F(CloudCompactionTest, test_apply_txn_size_truncation_and_log_no_truncation) {
    CloudTabletSPtr tablet = std::make_shared<CloudTablet>(_engine, _tablet_meta);
    TestableCloudCompaction compaction(_engine, tablet);

    // Create rowsets that fit within the limit
    std::vector<RowsetSharedPtr> rowsets;
    for (int i = 0; i < 3; i++) {
        RowsetSharedPtr rowset = create_rowset(Version(i, i), 1, false, 1024);
        ASSERT_TRUE(rowset != nullptr);
        rowsets.push_back(rowset);
    }

    compaction.set_input_rowsets(rowsets);

    // Set a large max size
    config::compaction_txn_max_size_bytes = 1024 * 1024 * 1024; // 1GB

    size_t truncated = compaction.test_apply_txn_size_truncation_and_log("test_compaction");

    ASSERT_EQ(truncated, 0);
    ASSERT_EQ(compaction.get_input_rowsets().size(), 3);
}

TEST_F(CloudCompactionTest, test_apply_txn_size_truncation_and_log_with_truncation) {
    CloudTabletSPtr tablet = std::make_shared<CloudTablet>(_engine, _tablet_meta);
    TestableCloudCompaction compaction(_engine, tablet);

    // Create multiple rowsets
    std::vector<RowsetSharedPtr> rowsets;
    for (int i = 0; i < 10; i++) {
        RowsetSharedPtr rowset = create_rowset(Version(i, i), 1, false, 1024);
        ASSERT_TRUE(rowset != nullptr);
        rowsets.push_back(rowset);
    }

    compaction.set_input_rowsets(rowsets);

    // Set a small max size to force truncation
    config::compaction_txn_max_size_bytes = 100; // Very small to force truncation

    size_t truncated = compaction.test_apply_txn_size_truncation_and_log("test_compaction");

    // Should have truncated some rowsets
    ASSERT_GT(truncated, 0);
    ASSERT_LT(compaction.get_input_rowsets().size(), 10);
    ASSERT_GT(compaction.get_input_rowsets().size(), 0);
}

TEST_F(CloudCompactionTest, test_apply_txn_size_truncation_and_log_version_range) {
    CloudTabletSPtr tablet = std::make_shared<CloudTablet>(_engine, _tablet_meta);
    TestableCloudCompaction compaction(_engine, tablet);

    // Create rowsets with consecutive versions
    std::vector<RowsetSharedPtr> rowsets;
    for (int i = 10; i < 20; i++) {
        RowsetSharedPtr rowset = create_rowset(Version(i, i), 1, false, 1024);
        ASSERT_TRUE(rowset != nullptr);
        rowsets.push_back(rowset);
    }

    compaction.set_input_rowsets(rowsets);

    // Set a size that will keep first 5 rowsets
    config::compaction_txn_max_size_bytes = 100; // Small enough to truncate

    size_t truncated = compaction.test_apply_txn_size_truncation_and_log("base_compaction");

    if (truncated > 0) {
        // Verify that the version range is adjusted correctly
        ASSERT_GT(compaction.get_input_rowsets().size(), 0);
        // First rowset should still start at version 10
        ASSERT_EQ(compaction.get_input_rowsets().front()->start_version(), 10);
        // Last rowset version should be less than 19
        ASSERT_LT(compaction.get_input_rowsets().back()->end_version(), 20);
    }
}

TEST_F(CloudCompactionTest, test_apply_txn_size_truncation_and_log_single_large_rowset) {
    CloudTabletSPtr tablet = std::make_shared<CloudTablet>(_engine, _tablet_meta);
    TestableCloudCompaction compaction(_engine, tablet);

    // Create a single large rowset
    std::vector<RowsetSharedPtr> rowsets;
    RowsetSharedPtr rowset = create_rowset(Version(0, 0), 1, false, 1024 * 1024);
    ASSERT_TRUE(rowset != nullptr);
    rowsets.push_back(rowset);

    compaction.set_input_rowsets(rowsets);

    // Set max size smaller than the rowset's metadata size
    config::compaction_txn_max_size_bytes = 1;

    size_t truncated = compaction.test_apply_txn_size_truncation_and_log("cumu_compaction");

    // Should keep at least 1 rowset even if it exceeds the limit
    ASSERT_EQ(truncated, 0);
    ASSERT_EQ(compaction.get_input_rowsets().size(), 1);
}
} // namespace doris