cloud/test/recycler_operation_log_test.cpp

// 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 <butil/strings/string_split.h>
#include <bvar/variable.h>
#include <fmt/core.h>
#include <gen_cpp/cloud.pb.h>
#include <gen_cpp/olap_file.pb.h>
#include <gtest/gtest.h>

#include <climits>
#include <cstdint>
#include <memory>
#include <string>

#include "common/bvars.h"
#include "common/config.h"
#include "common/util.h"
#include "cpp/sync_point.h"
#include "meta-service/meta_service.h"
#include "meta-store/blob_message.h"
#include "meta-store/document_message.h"
#include "meta-store/keys.h"
#include "meta-store/mem_txn_kv.h"
#include "meta-store/meta_reader.h"
#include "meta-store/txn_kv.h"
#include "meta-store/txn_kv_error.h"
#include "meta-store/versioned_value.h"
#include "mock_resource_manager.h"
#include "recycler/checker.h"
#include "recycler/recycler.h"
#include "recycler/util.h"

using namespace doris::cloud;

extern std::string instance_id;
extern int64_t current_time;
extern doris::cloud::RecyclerThreadPoolGroup thread_group;

// Convert a string to a hex-escaped string.
// A non-displayed character is represented as \xHH where HH is the hexadecimal value of the character.
// A displayed character is represented as itself.
static std::string escape_hex(std::string_view data) {
    std::string result;
    for (char c : data) {
        if (isprint(c)) {
            result += c;
        } else {
            result += fmt::format("\\x{:02x}", static_cast<unsigned char>(c));
        }
    }
    return result;
}

static size_t count_range(TxnKv* txn_kv, std::string_view begin = "",
                          std::string_view end = "\xFF") {
    std::unique_ptr<Transaction> txn;
    EXPECT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
    if (!txn) {
        return 0; // Failed to create transaction
    }

    FullRangeGetOptions opts;
    opts.txn = txn.get();
    auto iter = txn_kv->full_range_get(std::string(begin), std::string(end), std::move(opts));
    size_t total = 0;
    for (auto&& kvp = iter->next(); kvp.has_value(); kvp = iter->next()) {
        total += 1;
    }

    EXPECT_TRUE(iter->is_valid()); // The iterator should still be valid after the next call.
    return total;
}

static bool is_empty_range(TxnKv* txn_kv, std::string_view begin = "",
                           std::string_view end = "\xFF") {
    return count_range(txn_kv, begin, end) == 0;
}

static std::string dump_range(TxnKv* txn_kv, std::string_view begin = "",
                              std::string_view end = "\xFF") {
    std::unique_ptr<Transaction> txn;
    if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {
        return "Failed to create dump range transaction";
    }
    FullRangeGetOptions opts;
    opts.txn = txn.get();
    auto iter = txn_kv->full_range_get(std::string(begin), std::string(end), std::move(opts));
    std::string buffer;
    for (auto&& kv = iter->next(); kv.has_value(); kv = iter->next()) {
        buffer +=
                fmt::format("Key: {}, Value: {}\n", escape_hex(kv->first), escape_hex(kv->second));
    }
    EXPECT_TRUE(iter->is_valid()); // The iterator should still be valid after the next call.
    return buffer;
}

static void update_instance_info(TxnKv* txn_kv, const InstanceInfoPB& instance) {
    std::string key = instance_key({instance_id});
    std::string value = instance.SerializeAsString();
    ASSERT_FALSE(value.empty()) << "Failed to serialize instance info";
    std::unique_ptr<Transaction> txn;
    TxnErrorCode err = txn_kv->create_txn(&txn);
    ASSERT_EQ(err, TxnErrorCode::TXN_OK) << "Failed to create transaction";
    txn->put(key, value);
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK) << "Failed to commit transaction";
}

static void remove_instance_info(TxnKv* txn_kv) {
    std::string key = instance_key({instance_id});
    std::unique_ptr<Transaction> txn;
    TxnErrorCode err = txn_kv->create_txn(&txn);
    ASSERT_EQ(err, TxnErrorCode::TXN_OK) << "Failed to create transaction";
    txn->remove(key);
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK) << "Failed to commit transaction";
}

// It will get the latest versioned values.
TxnErrorCode read_operation_log(Transaction* txn, std::string_view log_key,
                                Versionstamp* log_version, OperationLogPB* operation_log) {
    std::string begin_key = encode_versioned_key(log_key, Versionstamp::min());
    std::string end_key = encode_versioned_key(log_key, Versionstamp::max());
    auto iter = blob_get_range(txn, begin_key, end_key);
    if (!iter->valid()) {
        TxnErrorCode err = iter->error_code();
        if (err != TxnErrorCode::TXN_OK) {
            return err;
        }
        return TxnErrorCode::TXN_KEY_NOT_FOUND;
    }
    for (; iter->valid(); iter->next()) {
        std::string_view key = iter->key();
        if (!decode_versioned_key(&key, log_version)) {
            return TxnErrorCode::TXN_INVALID_DATA;
        }
        if (!iter->parse_value(operation_log)) {
            return TxnErrorCode::TXN_INVALID_DATA;
        }
    }
    return iter->error_code();
}

TEST(RecycleOperationLogTest, RecycleOneOperationLog) {
    auto txn_kv = std::make_shared<MemTxnKv>();
    txn_kv->update_commit_version(1000);
    ASSERT_EQ(txn_kv->init(), 0);

    InstanceInfoPB instance;
    instance.set_instance_id(instance_id);
    instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
    auto* obj_info = instance.add_obj_info();
    obj_info->set_id("recycle_empty");
    obj_info->set_ak(config::test_s3_ak);
    obj_info->set_sk(config::test_s3_sk);
    obj_info->set_endpoint(config::test_s3_endpoint);
    obj_info->set_region(config::test_s3_region);
    obj_info->set_bucket(config::test_s3_bucket);
    obj_info->set_prefix("recycle_empty");

    update_instance_info(txn_kv.get(), instance);

    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    {
        // Put a empty operation log
        std::string log_key = versioned::log_key(instance_id);
        Versionstamp versionstamp(123, 0);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(versionstamp.version());

        std::unique_ptr<Transaction> txn;
        TxnErrorCode err = txn_kv->create_txn(&txn);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, versionstamp, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Recycle the operation logs.
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // Scan the operation logs to verify that the empty log is removed.
    remove_instance_info(txn_kv.get());
    ASSERT_TRUE(is_empty_range(txn_kv.get())) << dump_range(txn_kv.get());
}

TEST(RecycleOperationLogTest, RecycleCommitPartitionLog) {
    auto txn_kv = std::make_shared<MemTxnKv>();
    txn_kv->update_commit_version(1000);
    ASSERT_EQ(txn_kv->init(), 0);

    InstanceInfoPB instance;
    instance.set_instance_id(instance_id);
    instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
    update_instance_info(txn_kv.get(), instance);

    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    uint64_t db_id = 1;
    uint64_t table_id = 2;
    uint64_t index_id = 3;
    uint64_t partition_id = 4;
    {
        // Update the table version
        std::string ver_key = versioned::table_version_key({instance_id, table_id});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), ver_key, "");
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

        MetaReader meta_reader(instance_id, txn_kv.get());
        Versionstamp table_version;
        TxnErrorCode err = meta_reader.get_table_version(table_id, &table_version);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
    }

    {
        // Put a commit partition log
        std::string log_key = versioned::log_key(instance_id);
        Versionstamp versionstamp(123, 0);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(versionstamp.version());
        auto* commit_partition = operation_log.mutable_commit_partition();
        commit_partition->set_db_id(db_id);
        commit_partition->set_table_id(table_id);
        commit_partition->add_index_ids(index_id);
        commit_partition->add_partition_ids(partition_id);

        std::unique_ptr<Transaction> txn;
        TxnErrorCode err = txn_kv->create_txn(&txn);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Recycle the operation logs.
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // The above table version key should be removed.
    {
        MetaReader meta_reader(instance_id, txn_kv.get());
        Versionstamp table_version;
        TxnErrorCode err = meta_reader.get_table_version(table_id, &table_version);
        ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
    }

    {
        // Put a new commit partition log
        std::string log_key = versioned::log_key(instance_id);
        Versionstamp versionstamp(123, 0);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(versionstamp.version());
        auto* commit_partition = operation_log.mutable_commit_partition();
        commit_partition->set_db_id(db_id);
        commit_partition->set_table_id(table_id);
        commit_partition->add_index_ids(index_id);
        commit_partition->add_partition_ids(partition_id);

        std::unique_ptr<Transaction> txn;
        TxnErrorCode err = txn_kv->create_txn(&txn);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Recycle the operation logs.
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // Scan the operation logs to verify that the commit partition log is removed.
    remove_instance_info(txn_kv.get());
    ASSERT_TRUE(is_empty_range(txn_kv.get())) << dump_range(txn_kv.get());
}

TEST(RecycleOperationLogTest, RecycleDropPartitionLog) {
    auto txn_kv = std::make_shared<MemTxnKv>();
    txn_kv->update_commit_version(1000);
    ASSERT_EQ(txn_kv->init(), 0);

    InstanceInfoPB instance;
    instance.set_instance_id(instance_id);
    instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
    update_instance_info(txn_kv.get(), instance);

    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    uint64_t db_id = 1;
    uint64_t table_id = 2;
    uint64_t index_id = 3;
    uint64_t partition_id = 4;
    uint64_t stream_db_id = 5;
    uint64_t stream_id = 6;
    int64_t expiration = ::time(nullptr) + 3600; // 1 hour from now

    {
        // Update the table version, it will be removed after the drop partition log is recycled
        std::string ver_key = versioned::table_version_key({instance_id, table_id});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), ver_key, "");
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

        MetaReader meta_reader(instance_id, txn_kv.get());
        Versionstamp table_version;
        TxnErrorCode err = meta_reader.get_table_version(table_id, &table_version);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
    }

    {
        // Put a drop partition log
        std::string log_key = versioned::log_key(instance_id);
        Versionstamp versionstamp(123, 0);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(versionstamp.version());
        auto* drop_partition = operation_log.mutable_drop_partition();
        drop_partition->set_db_id(db_id);
        drop_partition->set_table_id(table_id);
        drop_partition->add_index_ids(index_id);
        drop_partition->add_partition_ids(partition_id);
        drop_partition->set_expired_at_s(expiration);
        TableStreamIdentityPB* stream = drop_partition->add_table_streams();
        stream->set_base_db_id(db_id);
        stream->set_base_table_id(table_id);
        stream->set_stream_db_id(stream_db_id);
        stream->set_stream_id(stream_id);
        drop_partition->set_update_table_version(
                true); // Update table version to ensure the table version is removed

        std::unique_ptr<Transaction> txn;
        TxnErrorCode err = txn_kv->create_txn(&txn);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Recycle the operation logs.
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // Verify that the recycle partition record is created
    {
        std::string recycle_key = recycle_partition_key({instance_id, partition_id});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string value;
        TxnErrorCode err = txn->get(recycle_key, &value);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);

        RecyclePartitionPB recycle_partition_pb;
        ASSERT_TRUE(recycle_partition_pb.ParseFromString(value));
        ASSERT_EQ(recycle_partition_pb.db_id(), db_id);
        ASSERT_EQ(recycle_partition_pb.table_id(), table_id);
        ASSERT_EQ(recycle_partition_pb.index_id_size(), 1);
        ASSERT_EQ(recycle_partition_pb.index_id(0), index_id);
        ASSERT_EQ(recycle_partition_pb.state(), RecyclePartitionPB::DROPPED);
        ASSERT_EQ(recycle_partition_pb.expiration(), expiration);
        ASSERT_EQ(recycle_partition_pb.table_streams_size(), 1);
        ASSERT_EQ(recycle_partition_pb.table_streams(0).base_db_id(), db_id);
        ASSERT_EQ(recycle_partition_pb.table_streams(0).base_table_id(), table_id);
        ASSERT_EQ(recycle_partition_pb.table_streams(0).stream_db_id(), stream_db_id);
        ASSERT_EQ(recycle_partition_pb.table_streams(0).stream_id(), stream_id);
    }

    // The table version key should be removed because update_table_version is true
    {
        MetaReader meta_reader(instance_id, txn_kv.get());
        Versionstamp table_version;
        TxnErrorCode err = meta_reader.get_table_version(table_id, &table_version);
        ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
    }

    {
        // Put a new drop partition log without updating table version
        std::string log_key = versioned::log_key(instance_id);
        Versionstamp versionstamp(124, 0);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(versionstamp.version());
        auto* drop_partition = operation_log.mutable_drop_partition();
        drop_partition->set_db_id(db_id);
        drop_partition->set_table_id(table_id);
        drop_partition->add_index_ids(index_id);
        drop_partition->add_partition_ids(partition_id + 1); // Different partition
        drop_partition->set_expired_at_s(expiration);
        drop_partition->set_update_table_version(false); // Don't update table version

        std::unique_ptr<Transaction> txn;
        TxnErrorCode err = txn_kv->create_txn(&txn);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Recycle the operation logs.
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // Verify that the second recycle partition record is created
    {
        std::string recycle_key = recycle_partition_key({instance_id, partition_id + 1});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string value;
        TxnErrorCode err = txn->get(recycle_key, &value);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);

        RecyclePartitionPB recycle_partition_pb;
        ASSERT_TRUE(recycle_partition_pb.ParseFromString(value));
        ASSERT_EQ(recycle_partition_pb.db_id(), db_id);
        ASSERT_EQ(recycle_partition_pb.table_id(), table_id);
        ASSERT_EQ(recycle_partition_pb.index_id_size(), 1);
        ASSERT_EQ(recycle_partition_pb.index_id(0), index_id);
        ASSERT_EQ(recycle_partition_pb.state(), RecyclePartitionPB::DROPPED);
        ASSERT_EQ(recycle_partition_pb.expiration(), expiration);
    }

    // Scan the operation logs to verify that the drop partition logs are removed.
    remove_instance_info(txn_kv.get());

    // Verify that the recycle partition records are created and operation logs are removed
    {
        // Check that the first recycle partition record exists
        std::string recycle_key1 = recycle_partition_key({instance_id, partition_id});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string value;
        TxnErrorCode err = txn->get(recycle_key1, &value);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);

        RecyclePartitionPB recycle_partition_pb;
        ASSERT_TRUE(recycle_partition_pb.ParseFromString(value));
        ASSERT_EQ(recycle_partition_pb.db_id(), db_id);
        ASSERT_EQ(recycle_partition_pb.table_id(), table_id);
        ASSERT_EQ(recycle_partition_pb.index_id_size(), 1);
        ASSERT_EQ(recycle_partition_pb.index_id(0), index_id);
        ASSERT_EQ(recycle_partition_pb.state(), RecyclePartitionPB::DROPPED);
        ASSERT_EQ(recycle_partition_pb.expiration(), expiration);
    }

    {
        // Check that the second recycle partition record exists
        std::string recycle_key2 = recycle_partition_key({instance_id, partition_id + 1});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string value;
        TxnErrorCode err = txn->get(recycle_key2, &value);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);

        RecyclePartitionPB recycle_partition_pb;
        ASSERT_TRUE(recycle_partition_pb.ParseFromString(value));
        ASSERT_EQ(recycle_partition_pb.db_id(), db_id);
        ASSERT_EQ(recycle_partition_pb.table_id(), table_id);
        ASSERT_EQ(recycle_partition_pb.index_id_size(), 1);
        ASSERT_EQ(recycle_partition_pb.index_id(0), index_id);
        ASSERT_EQ(recycle_partition_pb.state(), RecyclePartitionPB::DROPPED);
        ASSERT_EQ(recycle_partition_pb.expiration(), expiration);
    }

    // Verify that operation logs are removed (only recycle partition records should remain)
    ASSERT_EQ(count_range(txn_kv.get()), 2) << "Should only have 2 recycle partition records";
}

TEST(RecycleOperationLogTest, RecycleCommitIndexLog) {
    auto txn_kv = std::make_shared<MemTxnKv>();
    txn_kv->update_commit_version(1000);
    ASSERT_EQ(txn_kv->init(), 0);

    InstanceInfoPB instance;
    instance.set_instance_id(instance_id);
    instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
    update_instance_info(txn_kv.get(), instance);

    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    uint64_t db_id = 1;
    uint64_t table_id = 2;
    uint64_t index_id = 3;
    {
        // Update the table version
        std::string ver_key = versioned::table_version_key({instance_id, table_id});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), ver_key, "");
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

        MetaReader meta_reader(instance_id, txn_kv.get());
        Versionstamp table_version;
        TxnErrorCode err = meta_reader.get_table_version(table_id, &table_version);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
    }

    {
        // Put a commit index log
        std::string log_key = versioned::log_key(instance_id);
        Versionstamp versionstamp(123, 0);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(versionstamp.version());
        auto* commit_index = operation_log.mutable_commit_index();
        commit_index->set_db_id(db_id);
        commit_index->set_table_id(table_id);
        commit_index->add_index_ids(index_id);
        commit_index->set_update_table_version(true);

        std::unique_ptr<Transaction> txn;
        TxnErrorCode err = txn_kv->create_txn(&txn);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Recycle the operation logs.
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // The above table version key should be removed.
    {
        MetaReader meta_reader(instance_id, txn_kv.get());
        Versionstamp table_version;
        TxnErrorCode err = meta_reader.get_table_version(table_id, &table_version);
        ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
    }

    {
        // Put a new commit index log
        std::string log_key = versioned::log_key(instance_id);
        Versionstamp versionstamp(123, 0);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(versionstamp.version());
        auto* commit_index = operation_log.mutable_commit_index();
        commit_index->set_db_id(db_id);
        commit_index->set_table_id(table_id);
        commit_index->add_index_ids(index_id);

        std::unique_ptr<Transaction> txn;
        TxnErrorCode err = txn_kv->create_txn(&txn);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Recycle the operation logs.
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // Scan the operation logs to verify that the commit index log is removed.
    remove_instance_info(txn_kv.get());
    ASSERT_TRUE(is_empty_range(txn_kv.get())) << dump_range(txn_kv.get());
}

TEST(RecycleOperationLogTest, RecycleDropIndexLog) {
    auto txn_kv = std::make_shared<MemTxnKv>();
    txn_kv->update_commit_version(1000);
    ASSERT_EQ(txn_kv->init(), 0);

    InstanceInfoPB instance;
    instance.set_instance_id(instance_id);
    instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
    update_instance_info(txn_kv.get(), instance);

    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    uint64_t db_id = 1;
    uint64_t table_id = 2;
    uint64_t index_id = 3;
    uint64_t stream_db_id = 4;
    int64_t expiration = ::time(nullptr) + 3600; // 1 hour from now

    {
        // Put a drop index log
        std::string log_key = versioned::log_key(instance_id);
        Versionstamp versionstamp(123, 0);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(versionstamp.version());
        auto* drop_index = operation_log.mutable_drop_index();
        drop_index->set_db_id(db_id);
        drop_index->set_table_id(table_id);
        drop_index->add_index_ids(index_id);
        drop_index->set_expiration(expiration);
        drop_index->set_object_type(IndexObjectTypePB::TABLE_STREAM);
        drop_index->set_stream_db_id(stream_db_id);

        std::unique_ptr<Transaction> txn;
        TxnErrorCode err = txn_kv->create_txn(&txn);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Recycle the operation logs.
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // Verify that the recycle index record is created
    {
        std::string recycle_key = recycle_index_key({instance_id, index_id});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string value;
        TxnErrorCode err = txn->get(recycle_key, &value);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);

        RecycleIndexPB recycle_index_pb;
        ASSERT_TRUE(recycle_index_pb.ParseFromString(value));
        ASSERT_EQ(recycle_index_pb.db_id(), db_id);
        ASSERT_EQ(recycle_index_pb.table_id(), table_id);
        ASSERT_EQ(recycle_index_pb.state(), RecycleIndexPB::DROPPED);
        ASSERT_EQ(recycle_index_pb.expiration(), expiration);
        ASSERT_EQ(recycle_index_pb.object_type(), IndexObjectTypePB::TABLE_STREAM);
        ASSERT_EQ(recycle_index_pb.stream_db_id(), stream_db_id);
    }

    // Scan the operation logs to verify that the drop index logs are removed.
    remove_instance_info(txn_kv.get());

    // Verify that the recycle index records are created and operation logs are removed
    {
        // Check that the first recycle index record exists
        std::string recycle_key1 = recycle_index_key({instance_id, index_id});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string value;
        TxnErrorCode err = txn->get(recycle_key1, &value);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);

        RecycleIndexPB recycle_index_pb;
        ASSERT_TRUE(recycle_index_pb.ParseFromString(value));
        ASSERT_EQ(recycle_index_pb.db_id(), db_id);
        ASSERT_EQ(recycle_index_pb.table_id(), table_id);
        ASSERT_EQ(recycle_index_pb.state(), RecycleIndexPB::DROPPED);
        ASSERT_EQ(recycle_index_pb.expiration(), expiration);
        ASSERT_EQ(recycle_index_pb.object_type(), IndexObjectTypePB::TABLE_STREAM);
        ASSERT_EQ(recycle_index_pb.stream_db_id(), stream_db_id);
    }

    // Verify that operation logs are removed (only recycle index records should remain)
    ASSERT_EQ(count_range(txn_kv.get()), 1) << "Should only have one recycle index record";
}

TEST(RecycleOperationLogTest, RecycleLegacyDropIndexLogAsMaterializedIndex) {
    auto txn_kv = std::make_shared<MemTxnKv>();
    txn_kv->update_commit_version(1000);
    ASSERT_EQ(txn_kv->init(), 0);

    InstanceInfoPB instance;
    instance.set_instance_id(instance_id);
    instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
    update_instance_info(txn_kv.get(), instance);

    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    constexpr int64_t db_id = 11;
    constexpr int64_t table_id = 12;
    constexpr int64_t index_id = 13;
    constexpr int64_t expiration = 3600;
    {
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(123);
        auto* drop_index = operation_log.mutable_drop_index();
        drop_index->set_db_id(db_id);
        drop_index->set_table_id(table_id);
        drop_index->add_index_ids(index_id);
        drop_index->set_expiration(expiration);
        ASSERT_FALSE(drop_index->has_object_type());
        ASSERT_EQ(drop_index->object_type(), IndexObjectTypePB::MATERIALIZED_INDEX);

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), versioned::log_key(instance_id), operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
    std::string value;
    ASSERT_EQ(txn->get(recycle_index_key({instance_id, index_id}), &value), TxnErrorCode::TXN_OK);
    RecycleIndexPB recycle_index;
    ASSERT_TRUE(recycle_index.ParseFromString(value));
    EXPECT_EQ(recycle_index.db_id(), db_id);
    EXPECT_EQ(recycle_index.table_id(), table_id);
    EXPECT_EQ(recycle_index.object_type(), IndexObjectTypePB::MATERIALIZED_INDEX);
    EXPECT_FALSE(recycle_index.has_stream_db_id());
}

TEST(RecycleOperationLogTest, RecycleCommitTxnLog) {
    auto txn_kv = std::make_shared<MemTxnKv>();
    txn_kv->update_commit_version(1000);
    ASSERT_EQ(txn_kv->init(), 0);

    InstanceInfoPB instance;
    instance.set_instance_id(instance_id);
    instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
    update_instance_info(txn_kv.get(), instance);

    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    uint64_t db_id = 1;
    uint64_t table_id = 2;
    uint64_t partition_id = 4;
    uint64_t tablet_id = 5;
    uint64_t txn_id = 12345;
    uint64_t stream_db_id = 6;
    uint64_t stream_id = 7;

    // Create TxnInfo with VISIBLE status
    {
        TxnInfoPB txn_info;
        txn_info.set_db_id(db_id);
        txn_info.set_txn_id(txn_id);
        txn_info.set_status(TxnStatusPB::TXN_STATUS_VISIBLE);
        txn_info.set_label("test_label");

        std::string key = txn_info_key({instance_id, db_id, txn_id});
        std::string txn_info_value = txn_info.SerializeAsString();

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        txn->put(key, txn_info_value);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Create previous partition version to be recycled
    {
        std::string partition_version_key =
                versioned::partition_version_key({instance_id, partition_id});
        VersionPB version_pb;
        version_pb.set_version(100);

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), partition_version_key, version_pb.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Create previous tablet load stats to be recycled
    {
        std::string tablet_stats_key = versioned::tablet_load_stats_key({instance_id, tablet_id});
        TabletStatsPB stats_pb;
        // TabletStatsPB typically doesn't have tablet_id field, just create empty stats

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), tablet_stats_key, stats_pb.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Create previous table version to be recycled
    {
        std::string table_version_key = versioned::table_version_key({instance_id, table_id});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), table_version_key, "");
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    TableStreamIdentityPB stream_identity;
    stream_identity.set_base_db_id(db_id);
    stream_identity.set_base_table_id(table_id);
    stream_identity.set_stream_db_id(stream_db_id);
    stream_identity.set_stream_id(stream_id);
    std::string stream_offset_key = versioned::table_stream_offset_key(
            {instance_id, db_id, table_id, stream_db_id, stream_id, partition_id});
    {
        TableStreamOffsetPB offset;
        offset.set_partition_id(partition_id);
        offset.set_state(TableStreamOffsetStatePB::TABLE_STREAM_OFFSET_CONSUMED);
        offset.set_offset_tso(100);
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), stream_offset_key, offset.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Create versioned rowset meta (should NOT be recycled)
    {
        std::string versioned_rowset_key =
                versioned::meta_rowset_load_key({instance_id, tablet_id, 100});
        doris::RowsetMetaCloudPB rowset_meta;
        rowset_meta.set_rowset_id(123456);
        rowset_meta.set_tablet_id(tablet_id);
        rowset_meta.set_partition_id(partition_id);
        rowset_meta.set_start_version(100);
        rowset_meta.set_end_version(100);
        rowset_meta.set_num_rows(1000);
        rowset_meta.set_total_disk_size(1024 * 1024);

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        ASSERT_TRUE(
                versioned::document_put(txn.get(), versioned_rowset_key, std::move(rowset_meta)));
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Put a commit txn log
    {
        std::string log_key = versioned::log_key(instance_id);
        Versionstamp versionstamp(123, 0);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(versionstamp.version());

        auto* commit_txn = operation_log.mutable_commit_txn();
        commit_txn->set_txn_id(txn_id);
        commit_txn->set_db_id(db_id);
        commit_txn->add_table_ids(table_id);

        // Add partition version mapping
        auto& partition_version_map = *commit_txn->mutable_partition_version_map();
        partition_version_map[partition_id] = 101; // New version

        // Add tablet to partition mapping
        auto& tablet_to_partition_map = *commit_txn->mutable_tablet_to_partition_map();
        tablet_to_partition_map[tablet_id] = partition_id;

        // Add recycle txn info
        auto* recycle_txn = commit_txn->mutable_recycle_txn();
        recycle_txn->set_label("test_label");

        TableStreamPartitionSetPB* stream_offset_gc = commit_txn->add_table_stream_offset_gc();
        stream_offset_gc->mutable_identity()->CopyFrom(stream_identity);
        stream_offset_gc->add_partition_ids(partition_id);

        std::unique_ptr<Transaction> txn;
        TxnErrorCode err = txn_kv->create_txn(&txn);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
        TableStreamOffsetPB offset;
        offset.set_partition_id(partition_id);
        offset.set_state(TableStreamOffsetStatePB::TABLE_STREAM_OFFSET_CONSUMED);
        offset.set_offset_tso(110);
        versioned_put(txn.get(), stream_offset_key, offset.SerializeAsString());
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Recycle the operation logs
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // Verify that the recycle txn record is created
    {
        std::string recycle_key = recycle_txn_key({instance_id, db_id, txn_id});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string value;
        TxnErrorCode err = txn->get(recycle_key, &value);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);

        RecycleTxnPB recycle_txn_pb;
        ASSERT_TRUE(recycle_txn_pb.ParseFromString(value));
        ASSERT_EQ(recycle_txn_pb.label(), "test_label");
        ASSERT_GT(recycle_txn_pb.creation_time(), 0);
    }

    // Verify that previous versions are recycled (should not be found with current snapshot)
    {
        MetaReader meta_reader(instance_id, txn_kv.get());
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        // Previous partition version should be marked for removal
        Versionstamp partition_version;
        (void)meta_reader.get_partition_version(partition_id, nullptr, &partition_version);
        // Since we only have one version, it should still exist but the previous one should be scheduled for removal

        // Previous tablet stats should be marked for removal
        Versionstamp tablet_version;
        (void)meta_reader.get_tablet_load_stats(tablet_id, nullptr, &tablet_version);

        // Previous table version should be marked for removal
        Versionstamp table_version;
        (void)meta_reader.get_table_version(table_id, &table_version);

        std::string offset_value;
        Versionstamp offset_version;
        ASSERT_EQ(versioned_get(txn.get(), stream_offset_key, &offset_version, &offset_value),
                  TxnErrorCode::TXN_OK);
        TableStreamOffsetPB offset;
        ASSERT_TRUE(offset.ParseFromString(offset_value));
        EXPECT_EQ(offset.offset_tso(), 110);
        EXPECT_EQ(count_range(txn_kv.get(),
                              encode_versioned_key(stream_offset_key, Versionstamp::min()),
                              encode_versioned_key(stream_offset_key, Versionstamp::max())),
                  1);
    }

    // Scan the operation logs to verify that the commit txn log is removed
    remove_instance_info(txn_kv.get());

    // Should have TxnInfoPB + RecycleTxnPB + RowsetMetaCloudPB + latest offset version = 4 records
    ASSERT_EQ(count_range(txn_kv.get()), 4)
            << "Should retain transaction, recycle, rowset and latest offset records";
}

TEST(RecycleOperationLogTest, StreamOffsetGcKeepsVersionReferencedByCloneSnapshot) {
    auto txn_kv = std::make_shared<MemTxnKv>();
    txn_kv->update_commit_version(1000);
    ASSERT_EQ(txn_kv->init(), 0);

    InstanceInfoPB instance;
    instance.set_instance_id(instance_id);
    instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
    update_instance_info(txn_kv.get(), instance);

    constexpr int64_t db_id = 21;
    constexpr int64_t table_id = 22;
    constexpr int64_t partition_id = 23;
    constexpr int64_t stream_db_id = 24;
    constexpr int64_t stream_id = 25;
    constexpr int64_t txn_id = 26;
    TableStreamIdentityPB identity;
    identity.set_base_db_id(db_id);
    identity.set_base_table_id(table_id);
    identity.set_stream_db_id(stream_db_id);
    identity.set_stream_id(stream_id);
    const std::string offset_key = versioned::table_stream_offset_key(
            {instance_id, db_id, table_id, stream_db_id, stream_id, partition_id});

    TableStreamOffsetPB old_offset;
    old_offset.set_partition_id(partition_id);
    old_offset.set_state(TableStreamOffsetStatePB::TABLE_STREAM_OFFSET_CONSUMED);
    old_offset.set_offset_tso(100);
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), offset_key, old_offset.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    Versionstamp old_offset_version;
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string value;
        ASSERT_EQ(versioned_get(txn.get(), offset_key, &old_offset_version, &value),
                  TxnErrorCode::TXN_OK);
    }

    const std::string snapshot_key = versioned::snapshot_full_key(instance_id);
    {
        SnapshotPB snapshot;
        snapshot.set_status(SnapshotStatus::SNAPSHOT_NORMAL);
        snapshot.set_type(SnapshotType::SNAPSHOT_REFERENCE);
        snapshot.set_instance_id(instance_id);
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), snapshot_key, snapshot.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    Versionstamp snapshot_version;
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string value;
        ASSERT_EQ(versioned_get(txn.get(), snapshot_key, &snapshot_version, &value),
                  TxnErrorCode::TXN_OK);
    }

    {
        TableStreamOffsetPB new_offset = old_offset;
        new_offset.set_offset_tso(110);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(old_offset_version.version());
        auto* commit_txn = operation_log.mutable_commit_txn();
        commit_txn->set_txn_id(txn_id);
        commit_txn->set_db_id(db_id);
        commit_txn->mutable_recycle_txn()->set_label("clone_safe_stream_offset_gc");
        auto* offset_gc = commit_txn->add_table_stream_offset_gc();
        offset_gc->mutable_identity()->CopyFrom(identity);
        offset_gc->add_partition_ids(partition_id);

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), offset_key, new_offset.SerializeAsString());
        versioned::blob_put(txn.get(), versioned::log_key(instance_id), operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // A clone created from this snapshot still reads the old offset. The active snapshot keeps
    // the whole GC operation log pending, so both the inherited version and the latest version
    // remain available.
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
    Versionstamp actual_version;
    std::string value;
    ASSERT_EQ(versioned_get(txn.get(), offset_key, snapshot_version, &actual_version, &value),
              TxnErrorCode::TXN_OK);
    TableStreamOffsetPB snapshot_offset;
    ASSERT_TRUE(snapshot_offset.ParseFromString(value));
    EXPECT_EQ(snapshot_offset.offset_tso(), 100);
    EXPECT_EQ(actual_version, old_offset_version);
    EXPECT_EQ(count_range(txn_kv.get(), encode_versioned_key(offset_key, Versionstamp::min()),
                          encode_versioned_key(offset_key, Versionstamp::max())),
              2);
    EXPECT_EQ(
            count_range(txn_kv.get(),
                        encode_versioned_key(versioned::log_key(instance_id), Versionstamp::min()),
                        encode_versioned_key(versioned::log_key(instance_id), Versionstamp::max())),
            1);
}

TEST(RecycleOperationLogTest, RecycleCommitTxnLogWhenTxnIsNotVisible) {
    auto txn_kv = std::make_shared<MemTxnKv>();
    txn_kv->update_commit_version(1000);
    ASSERT_EQ(txn_kv->init(), 0);

    InstanceInfoPB instance;
    instance.set_instance_id(instance_id);
    instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
    update_instance_info(txn_kv.get(), instance);

    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    uint64_t db_id = 1;
    uint64_t table_id = 2;
    uint64_t partition_id = 4;
    uint64_t tablet_id = 5;
    uint64_t txn_id = 12345;

    // Create TxnInfo with COMMITTED status (not VISIBLE)
    {
        TxnInfoPB txn_info;
        txn_info.set_db_id(db_id);
        txn_info.set_txn_id(txn_id);
        txn_info.set_status(TxnStatusPB::TXN_STATUS_COMMITTED); // Not VISIBLE
        txn_info.set_label("test_label");

        std::string key = txn_info_key({instance_id, db_id, txn_id});
        std::string txn_info_value = txn_info.SerializeAsString();

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        txn->put(key, txn_info_value);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Create previous partition version
    {
        std::string partition_version_key =
                versioned::partition_version_key({instance_id, partition_id});
        VersionPB version_pb;
        version_pb.set_version(100);

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), partition_version_key, version_pb.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Put a commit txn log
    {
        std::string log_key = versioned::log_key(instance_id);
        Versionstamp versionstamp(123, 0);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(versionstamp.version());

        auto* commit_txn = operation_log.mutable_commit_txn();
        commit_txn->set_txn_id(txn_id);
        commit_txn->set_db_id(db_id);
        commit_txn->add_table_ids(table_id);

        // Add partition version mapping
        auto& partition_version_map = *commit_txn->mutable_partition_version_map();
        partition_version_map[partition_id] = 101;

        // Add tablet to partition mapping
        auto& tablet_to_partition_map = *commit_txn->mutable_tablet_to_partition_map();
        tablet_to_partition_map[tablet_id] = partition_id;

        // Add recycle txn info
        auto* recycle_txn = commit_txn->mutable_recycle_txn();
        recycle_txn->set_label("test_label");

        std::unique_ptr<Transaction> txn;
        TxnErrorCode err = txn_kv->create_txn(&txn);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Recycle the operation logs
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // Verify that NO recycle txn record is created (because txn is not VISIBLE)
    {
        std::string recycle_key = recycle_txn_key({instance_id, db_id, txn_id});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string value;
        TxnErrorCode err = txn->get(recycle_key, &value);
        ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND)
                << "Recycle txn record should not exist for non-VISIBLE transactions";
    }

    // Verify that the original partition version still exists (not recycled)
    {
        MetaReader meta_reader(instance_id, txn_kv.get());
        Versionstamp partition_version;
        TxnErrorCode err =
                meta_reader.get_partition_version(partition_id, nullptr, &partition_version);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK)
                << "Partition version should still exist for non-VISIBLE transactions";
    }

    // Scan the operation logs to verify that the commit txn log is removed
    remove_instance_info(txn_kv.get());

    // Should have TxnInfoPB + partition_version + RowsetMetaCloudPB = 3 records (no recycling for non-VISIBLE txn)
    ASSERT_EQ(count_range(txn_kv.get()), 3)
            << "Should have TxnInfoPB, partition version and RowsetMetaCloudPB records";
}

TEST(RecycleOperationLogTest, RecycleUpdateTabletLog) {
    auto txn_kv = std::make_shared<MemTxnKv>();
    txn_kv->update_commit_version(1000);
    ASSERT_EQ(txn_kv->init(), 0);

    InstanceInfoPB instance;
    instance.set_instance_id(instance_id);
    instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
    update_instance_info(txn_kv.get(), instance);

    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    uint64_t tablet_id = 1;
    {
        // Update the tablet meta
        std::string tablet_meta_key = versioned::meta_tablet_key({instance_id, tablet_id});
        doris::TabletMetaCloudPB tablet_meta;
        tablet_meta.set_tablet_id(tablet_id);
        tablet_meta.set_creation_time(::time(nullptr));

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), tablet_meta_key, tablet_meta.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

        MetaReader meta_reader(instance_id, txn_kv.get());
        Versionstamp versionstamp;
        TxnErrorCode err = meta_reader.get_tablet_meta(tablet_id, &tablet_meta, &versionstamp);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
    }

    {
        // Put an update tablet log
        std::string log_key = versioned::log_key(instance_id);
        Versionstamp versionstamp(123, 0);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(versionstamp.version());
        auto* update_tablet = operation_log.mutable_update_tablet();
        update_tablet->add_tablet_ids(tablet_id);

        std::unique_ptr<Transaction> txn;
        TxnErrorCode err = txn_kv->create_txn(&txn);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Recycle the operation logs.
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // The tablet meta should be removed
    {
        MetaReader meta_reader(instance_id, txn_kv.get());
        doris::TabletMetaCloudPB tablet_meta;
        Versionstamp versionstamp;
        TxnErrorCode err = meta_reader.get_tablet_meta(tablet_id, &tablet_meta, &versionstamp);
        ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
    }
    // The operation log should be removed
    remove_instance_info(txn_kv.get());
    ASSERT_TRUE(is_empty_range(txn_kv.get())) << dump_range(txn_kv.get());
}

std::unique_ptr<MetaServiceProxy> get_meta_service(bool mock_resource_mgr) {
    int ret = 0;
    // MemKv
    auto txn_kv = std::dynamic_pointer_cast<TxnKv>(std::make_shared<MemTxnKv>());
    if (txn_kv != nullptr) {
        ret = txn_kv->init();
        [&] { ASSERT_EQ(ret, 0); }();
    }
    [&] { ASSERT_NE(txn_kv.get(), nullptr); }();

    // FdbKv
    //     config::fdb_cluster_file_path = "fdb.cluster";
    //     static auto txn_kv = std::dynamic_pointer_cast<TxnKv>(std::make_shared<FdbTxnKv>());
    //     static std::atomic<bool> init {false};
    //     bool tmp = false;
    //     if (init.compare_exchange_strong(tmp, true)) {
    //         int ret = txn_kv->init();
    //         [&] { ASSERT_EQ(ret, 0); ASSERT_NE(txn_kv.get(), nullptr); }();
    //     }

    std::unique_ptr<Transaction> txn;
    EXPECT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
    txn->remove("\x00", "\xfe"); // This is dangerous if the fdb is not correctly set
    EXPECT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    auto rs = mock_resource_mgr ? std::make_shared<MockResourceManager>(txn_kv)
                                : std::make_shared<ResourceManager>(txn_kv);
    auto rl = std::make_shared<RateLimiter>();
    auto snapshot = std::make_shared<SnapshotManager>(txn_kv);
    auto meta_service = std::make_unique<MetaServiceImpl>(txn_kv, rs, rl, snapshot);
    return std::make_unique<MetaServiceProxy>(std::move(meta_service));
}

namespace doris::cloud {
static std::string next_rowset_id() {
    static int cnt = 0;
    return std::to_string(++cnt);
}

void add_tablet(CreateTabletsRequest& req, int64_t table_id, int64_t index_id, int64_t partition_id,
                int64_t tablet_id) {
    auto tablet = req.add_tablet_metas();
    tablet->set_table_id(table_id);
    tablet->set_index_id(index_id);
    tablet->set_partition_id(partition_id);
    tablet->set_tablet_id(tablet_id);
    auto schema = tablet->mutable_schema();
    schema->set_schema_version(0);
    auto first_rowset = tablet->add_rs_metas();
    first_rowset->set_rowset_id(0); // required
    first_rowset->set_rowset_id_v2(next_rowset_id());
    first_rowset->set_start_version(0);
    first_rowset->set_end_version(1);
    first_rowset->mutable_tablet_schema()->CopyFrom(*schema);
}

void create_tablet(MetaServiceProxy* meta_service, int64_t table_id, int64_t index_id,
                   int64_t partition_id, int64_t tablet_id) {
    brpc::Controller cntl;
    CreateTabletsRequest req;
    CreateTabletsResponse res;
    req.set_db_id(1);
    add_tablet(req, table_id, index_id, partition_id, tablet_id);
    meta_service->create_tablets(&cntl, &req, &res, nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << tablet_id;
}

doris::RowsetMetaCloudPB create_rowset(int64_t txn_id, int64_t tablet_id, int partition_id = 10,
                                       int64_t version = -1, int num_rows = 100) {
    doris::RowsetMetaCloudPB rowset;
    rowset.set_rowset_id(0); // required
    rowset.set_rowset_id_v2(next_rowset_id());
    rowset.set_tablet_id(tablet_id);
    rowset.set_partition_id(partition_id);
    rowset.set_txn_id(txn_id);
    if (version > 0) {
        rowset.set_start_version(version);
        rowset.set_end_version(version);
    }
    rowset.set_num_segments(1);
    rowset.set_num_rows(num_rows);
    rowset.set_data_disk_size(num_rows * 100);
    rowset.set_index_disk_size(num_rows * 10);
    rowset.set_total_disk_size(num_rows * 110);
    rowset.mutable_tablet_schema()->set_schema_version(0);
    auto* key_bounds = rowset.add_segments_key_bounds();
    key_bounds->set_min_key(fmt::format("min-key-{}", rowset.rowset_id_v2()));
    key_bounds->set_max_key(fmt::format("max-key-{}", rowset.rowset_id_v2()));
    rowset.set_segments_key_bounds_truncated(false);
    rowset.set_txn_expiration(::time(nullptr)); // Required by DCHECK
    return rowset;
}

TEST(RecycleOperationLogTest, CoveredLoadDoesNotProtectLaterCompaction) {
    auto meta_service = get_meta_service(false);
    const std::string test_instance_id = "covered_load_compaction";
    auto txn_kv = std::dynamic_pointer_cast<MemTxnKv>(meta_service->txn_kv());
    ASSERT_NE(txn_kv, nullptr);
    auto* sp = SyncPoint::get_instance();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->disable_processing();
    };
    sp->set_call_back("get_instance_id", [&](auto&& args) {
        auto* ret = try_any_cast_ret<std::string>(args);
        ret->first = test_instance_id;
        ret->second = true;
    });
    sp->enable_processing();

    constexpr int64_t table_id = 20001, index_id = 20002, partition_id = 20003, tablet_id = 20004;
    InstanceInfoPB instance;
    instance.set_instance_id(test_instance_id);
    instance.set_multi_version_status(MULTI_VERSION_ENABLED);
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
    txn->put(instance_key(test_instance_id), instance.SerializeAsString());
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    meta_service->resource_mgr()->refresh_instance(test_instance_id);
    ASSERT_TRUE(meta_service->resource_mgr()->is_version_read_enabled(test_instance_id));
    create_tablet(meta_service.get(), table_id, index_id, partition_id, tablet_id);

    auto l2 = create_rowset(102, tablet_id, partition_id, 2, 10);
    auto l3 = create_rowset(103, tablet_id, partition_id, 3, 10);
    auto l4 = create_rowset(104, tablet_id, partition_id, 4, 10);
    ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
    ASSERT_TRUE(versioned::document_put(
            txn.get(), versioned::meta_rowset_load_key({test_instance_id, tablet_id, 2}),
            Versionstamp(80), doris::RowsetMetaCloudPB(l2)));
    ASSERT_TRUE(versioned::document_put(
            txn.get(), versioned::meta_rowset_load_key({test_instance_id, tablet_id, 3}),
            Versionstamp(90), doris::RowsetMetaCloudPB(l3)));
    SnapshotPB snapshot;
    snapshot.set_status(SNAPSHOT_NORMAL);
    versioned_put(txn.get(), versioned::snapshot_full_key(test_instance_id), Versionstamp(100),
                  snapshot.SerializeAsString());
    // Keep both stats dependencies at/after 150 so only the selected inputs decide protection.
    TabletStatsPB compact_stats;
    compact_stats.set_num_rows(20);
    compact_stats.set_num_rowsets(2);
    compact_stats.set_num_segments(2);
    compact_stats.set_data_size(2200);
    compact_stats.set_index_size(200);
    compact_stats.set_segment_size(2000);
    compact_stats.set_cumulative_point(2);
    versioned_put(txn.get(), versioned::tablet_compact_stats_key({test_instance_id, tablet_id}),
                  Versionstamp(150), compact_stats.SerializeAsString());
    versioned_put(txn.get(), versioned::tablet_load_stats_key({test_instance_id, tablet_id}),
                  Versionstamp(150), TabletStatsPB().SerializeAsString());
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    std::vector<std::string> output_rowset_ids;
    auto compact = [&](int64_t end, int64_t commit_version, int base_count, OperationLogPB* log,
                       Versionstamp* log_version) {
        SCOPED_TRACE(commit_version);
        const int64_t txn_id = 30000 + end;
        const int64_t rows = (end - 1) * 10;
        auto output = create_rowset(txn_id, tablet_id, partition_id, 2, rows);
        output.set_end_version(end);
        output_rowset_ids.push_back(output.rowset_id_v2());
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        txn->put(meta_rowset_tmp_key({test_instance_id, txn_id, tablet_id}),
                 output.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

        StartTabletJobRequest start;
        auto* idx = start.mutable_job()->mutable_idx();
        idx->set_table_id(table_id);
        idx->set_index_id(index_id);
        idx->set_partition_id(partition_id);
        idx->set_tablet_id(tablet_id);
        auto* job = start.mutable_job()->add_compaction();
        job->set_id(fmt::format("base_{}", base_count));
        job->set_initiator("test_be");
        job->set_type(TabletCompactionJobPB::BASE);
        job->set_base_compaction_cnt(base_count);
        job->set_cumulative_compaction_cnt(0);
        job->add_input_versions(2);
        job->add_input_versions(end);
        job->set_expiration(time(nullptr) + 3600);
        job->set_lease(time(nullptr) + 3600);
        brpc::Controller cntl;
        StartTabletJobResponse start_response;
        meta_service->start_tablet_job(&cntl, &start, &start_response, nullptr);
        ASSERT_EQ(start_response.status().code(), MetaServiceCode::OK)
                << start_response.status().msg();

        FinishTabletJobRequest finish;
        finish.set_action(FinishTabletJobRequest::COMMIT);
        finish.mutable_job()->CopyFrom(start.job());
        job = finish.mutable_job()->mutable_compaction(0);
        job->add_txn_id(txn_id);
        job->add_output_versions(end);
        job->add_output_rowset_ids(output.rowset_id_v2());
        job->set_output_cumulative_point(2);
        job->set_num_input_rows(rows);
        job->set_num_output_rows(rows);
        job->set_num_input_rowsets(2);
        job->set_num_output_rowsets(1);
        job->set_num_input_segments(2);
        job->set_num_output_segments(1);
        job->set_size_input_rowsets(rows * 110);
        job->set_size_output_rowsets(rows * 110);
        job->set_index_size_input_rowsets(rows * 10);
        job->set_index_size_output_rowsets(rows * 10);
        job->set_segment_size_input_rowsets(rows * 100);
        job->set_segment_size_output_rowsets(rows * 100);
        txn_kv->update_commit_version(commit_version - 1);
        FinishTabletJobResponse finish_response;
        meta_service->finish_tablet_job(&cntl, &finish, &finish_response, nullptr);
        ASSERT_EQ(finish_response.status().code(), MetaServiceCode::OK)
                << finish_response.status().msg();
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        ASSERT_EQ(read_operation_log(txn.get(), versioned::log_key(test_instance_id), log_version,
                                     log),
                  TxnErrorCode::TXN_OK);
        ASSERT_EQ(log_version->version(), commit_version);
        ASSERT_TRUE(log->has_compaction());
        ASSERT_EQ(log->compaction().recycle_rowsets_size(), 2);
    };

    OperationLogPB first_log, second_log;
    Versionstamp first_version, second_version;
    ASSERT_NO_FATAL_FAILURE(compact(3, 150, 0, &first_log, &first_version));
    EXPECT_EQ(first_log.min_timestamp(), 80);
    EXPECT_EQ(first_log.compaction().recycle_rowsets(0).rowset_meta().rowset_id_v2(),
              l2.rowset_id_v2());
    EXPECT_EQ(first_log.compaction().recycle_rowsets(1).rowset_meta().rowset_id_v2(),
              l3.rowset_id_v2());

    ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
    ASSERT_TRUE(versioned::document_put(
            txn.get(), versioned::meta_rowset_load_key({test_instance_id, tablet_id, 4}),
            Versionstamp(170), doris::RowsetMetaCloudPB(l4)));
    TabletStatsPB load_stats;
    load_stats.set_num_rows(10);
    load_stats.set_num_rowsets(1);
    load_stats.set_num_segments(1);
    load_stats.set_data_size(1100);
    load_stats.set_index_size(100);
    load_stats.set_segment_size(1000);
    versioned_put(txn.get(), versioned::tablet_load_stats_key({test_instance_id, tablet_id}),
                  Versionstamp(170), load_stats.SerializeAsString());
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    ASSERT_NO_FATAL_FAILURE(compact(4, 200, 1, &second_log, &second_version));
    EXPECT_EQ(second_log.min_timestamp(), 150);
    const auto& inputs = second_log.compaction().recycle_rowsets();
    EXPECT_EQ(inputs[0].rowset_meta().start_version(), 2);
    EXPECT_EQ(inputs[0].rowset_meta().end_version(), 3);
    EXPECT_EQ(inputs[0].rowset_meta().rowset_id_v2(), output_rowset_ids[0]);
    EXPECT_EQ(inputs[1].rowset_meta().rowset_id_v2(), l4.rowset_id_v2());

    // Exercise the real Recycler snapshot-reference decision with the actual MS logs.
    OperationLogRecycleChecker checker(test_instance_id, txn_kv.get(), instance);
    ASSERT_EQ(checker.init(), 0);
    OperationLogReferenceInfo first_reference, second_reference;
    EXPECT_FALSE(checker.can_recycle(first_version, first_log.min_timestamp(), &first_reference));
    EXPECT_TRUE(first_reference.referenced_by_snapshot);
    EXPECT_EQ(first_reference.referenced_snapshot_timestamp, Versionstamp(100));
    EXPECT_TRUE(checker.can_recycle(second_version, second_log.min_timestamp(), &second_reference));
    EXPECT_FALSE(second_reference.referenced_by_snapshot);

    MetaReader snapshot_reader(test_instance_id, txn_kv.get(), Versionstamp(100));
    std::vector<doris::RowsetMetaCloudPB> snapshot_rowsets;
    ASSERT_EQ(snapshot_reader.get_rowset_metas(tablet_id, 2, 3, &snapshot_rowsets),
              TxnErrorCode::TXN_OK);
    ASSERT_EQ(snapshot_rowsets.size(), 2);
    EXPECT_EQ(snapshot_rowsets[0].rowset_id_v2(), l2.rowset_id_v2());
    EXPECT_EQ(snapshot_rowsets[1].rowset_id_v2(), l3.rowset_id_v2());
    EXPECT_EQ(snapshot_reader.min_read_versionstamp(), Versionstamp(80));
}

TEST(RecycleOperationLogTest, RecycleCompactionLog) {
    // Ensure strip behavior is enabled for this test
    auto old_flag = config::enable_recycle_rowset_strip_key_bounds;
    config::enable_recycle_rowset_strip_key_bounds = true;
    DORIS_CLOUD_DEFER {
        config::enable_recycle_rowset_strip_key_bounds = old_flag;
    };

    auto meta_service = get_meta_service(false);
    std::string test_instance_id = "recycle_compaction_log_test";
    auto* sp = SyncPoint::get_instance();
    DORIS_CLOUD_DEFER {
        SyncPoint::get_instance()->clear_all_call_backs();
    };
    sp->set_call_back("get_instance_id", [&](auto&& args) {
        auto* ret = try_any_cast_ret<std::string>(args);
        ret->first = test_instance_id;
        ret->second = true;
    });
    sp->set_call_back("check_lazy_txn_finished::bypass_check", [&](auto&& args) {
        auto* ret = doris::try_any_cast_ret<bool>(args);
        ret->first = true;
        ret->second = true;
    });
    sp->set_call_back("delete_rowset_data::bypass_check", [&](auto&& args) {
        auto* ret = doris::try_any_cast_ret<bool>(args);
        ret->first = true;
        ret->second = true;
    });
    sp->set_call_back("recycle_tablet::begin", [&](auto&& args) {
        auto* ret = doris::try_any_cast_ret<int>(args);
        ret->first = 0;
        ret->second = true;
    });
    sp->enable_processing();

    constexpr int64_t table_id = 20001;
    constexpr int64_t index_id = 20002;
    constexpr int64_t partition_id = 20003;
    constexpr int64_t tablet_id = 20004;

    {
        // write instance
        InstanceInfoPB instance_info;
        instance_info.set_instance_id(test_instance_id);
        instance_info.set_multi_version_status(MULTI_VERSION_ENABLED);
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        txn->put(instance_key(test_instance_id), instance_info.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

        meta_service->resource_mgr()->refresh_instance(test_instance_id);
        ASSERT_TRUE(meta_service->resource_mgr()->is_version_write_enabled(test_instance_id));
    }

    {
        // Create tablet first
        create_tablet(meta_service.get(), table_id, index_id, partition_id, tablet_id);
    }

    // Create input rowsets for compaction (versions 2-4)
    std::vector<doris::RowsetMetaCloudPB> input_rowsets;
    auto txn_kv = meta_service->txn_kv();
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        for (int i = 0; i < 3; ++i) {
            auto rowset = create_rowset(i + 100, tablet_id, partition_id, i + 2, 50 * (i + 1));
            input_rowsets.push_back(rowset);

            // Put rowset directly to meta storage
            auto rowset_key = meta_rowset_key({test_instance_id, tablet_id, rowset.end_version()});
            auto rowset_val = rowset.SerializeAsString();
            txn->put(rowset_key, rowset_val);
            auto versioned_rowset_key = versioned::meta_rowset_load_key(
                    {test_instance_id, tablet_id, rowset.end_version()});
            versioned::document_put(txn.get(), versioned_rowset_key, std::move(rowset));
        }
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Create output rowset as tmp rowset
    constexpr int64_t txn_id = 30001;
    constexpr int64_t output_start_version = 2;
    constexpr int64_t output_end_version = 4;
    auto output_rowset = create_rowset(200, tablet_id, partition_id, output_start_version, 100);
    output_rowset.set_end_version(output_end_version); // Set end version to create 2-4 range
    output_rowset.set_txn_id(txn_id);

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        // Put tmp rowset
        auto tmp_rowset_key = meta_rowset_tmp_key({test_instance_id, txn_id, tablet_id});
        auto tmp_rowset_val = output_rowset.SerializeAsString();
        txn->put(tmp_rowset_key, tmp_rowset_val);

        // Create initial tablet stats
        TabletStatsPB initial_stats;
        initial_stats.set_num_rows(150); // Total from input rowsets
        initial_stats.set_data_size(150 * 50);
        initial_stats.set_num_rowsets(3);
        initial_stats.set_num_segments(3);
        initial_stats.set_index_size(100);
        initial_stats.set_segment_size(200);
        initial_stats.set_cumulative_point(1);

        auto stats_key =
                stats_tablet_key({test_instance_id, table_id, index_id, partition_id, tablet_id});
        auto stats_val = initial_stats.SerializeAsString();
        txn->put(stats_key, stats_val);
        auto versioned_load_stats_key =
                versioned::tablet_load_stats_key({test_instance_id, tablet_id});
        versioned_put(txn.get(), versioned_load_stats_key, stats_val);

        // Create tablet compact stats for versioned storage
        auto tablet_compact_stats_key =
                versioned::tablet_compact_stats_key({test_instance_id, tablet_id});
        auto tablet_compact_stats_val = initial_stats.SerializeAsString();
        versioned_put(txn.get(), tablet_compact_stats_key, tablet_compact_stats_val);

        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Start compaction job first
    const std::string job_id = "test_compaction_job";
    const std::string initiator = "test_be";

    {
        brpc::Controller cntl;
        StartTabletJobRequest req;
        StartTabletJobResponse res;
        req.mutable_job()->mutable_idx()->set_tablet_id(tablet_id);
        auto compaction = req.mutable_job()->add_compaction();
        compaction->set_id(job_id);
        compaction->set_initiator(initiator);
        compaction->set_type(TabletCompactionJobPB::CUMULATIVE);
        compaction->set_base_compaction_cnt(0);
        compaction->set_cumulative_compaction_cnt(0);
        compaction->add_input_versions(2);
        compaction->add_input_versions(4);
        long now = time(nullptr);
        compaction->set_expiration(now + 12);
        compaction->set_lease(now + 3);
        meta_service->start_tablet_job(&cntl, &req, &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    // Now use real finish_tablet_job to trigger process_compaction_job
    {
        brpc::Controller cntl;
        FinishTabletJobRequest req;
        FinishTabletJobResponse res;

        req.set_action(FinishTabletJobRequest::COMMIT);
        req.mutable_job()->mutable_idx()->set_table_id(table_id);
        req.mutable_job()->mutable_idx()->set_index_id(index_id);
        req.mutable_job()->mutable_idx()->set_partition_id(partition_id);
        req.mutable_job()->mutable_idx()->set_tablet_id(tablet_id);

        auto compaction = req.mutable_job()->add_compaction();
        compaction->set_id(job_id);
        compaction->set_initiator(initiator);
        compaction->set_type(TabletCompactionJobPB::CUMULATIVE);

        // Input versions and rowsets
        compaction->add_input_versions(2);
        compaction->add_input_versions(4);

        // Output information
        compaction->add_txn_id(txn_id);
        compaction->add_output_versions(output_end_version);
        compaction->add_output_rowset_ids(output_rowset.rowset_id_v2());
        compaction->set_output_cumulative_point(5);

        // Compaction stats for updating tablet stats
        compaction->set_size_input_rowsets(150 * 50); // Size of input rowsets
        compaction->set_index_size_input_rowsets(100);
        compaction->set_segment_size_input_rowsets(200);
        compaction->set_num_input_rows(150);
        compaction->set_num_input_rowsets(3);
        compaction->set_num_input_segments(3);

        compaction->set_size_output_rowsets(100 * 50); // Size of output rowset
        compaction->set_index_size_output_rowsets(50);
        compaction->set_segment_size_output_rowsets(100);
        compaction->set_num_output_rows(100);
        compaction->set_num_output_rowsets(1);
        compaction->set_num_output_segments(1);

        // This will trigger process_compaction_job internally and create compaction log
        meta_service->finish_tablet_job(&cntl, &req, &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    // Verify compaction log was created
    Versionstamp log_version;
    OperationLogPB operation_log;
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string log_key = versioned::log_key({test_instance_id});
        ASSERT_EQ(read_operation_log(txn.get(), log_key, &log_version, &operation_log),
                  TxnErrorCode::TXN_OK);

        ASSERT_TRUE(operation_log.has_compaction());
        // Ensure input rowsets carried key bounds before strip
        ASSERT_GT(input_rowsets.front().segments_key_bounds_size(), 0);

        const auto& compaction_log = operation_log.compaction();
        ASSERT_EQ(compaction_log.tablet_id(), tablet_id);
        ASSERT_EQ(compaction_log.start_version(), 2);
        ASSERT_EQ(compaction_log.end_version(), 4);
        ASSERT_EQ(compaction_log.recycle_rowsets_size(), 3);
        for (const auto& recycle_rs : compaction_log.recycle_rowsets()) {
            ASSERT_EQ(recycle_rs.rowset_meta().segments_key_bounds_size(), 0);
            ASSERT_FALSE(recycle_rs.rowset_meta().has_segments_key_bounds_truncated());
        }
    }

    // Set up recycler using the same txn_kv as meta_service
    InstanceInfoPB instance_info;
    instance_info.set_instance_id(test_instance_id);
    instance_info.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);

    InstanceRecycler recycler(txn_kv, instance_info, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    // Now recycle the compaction operation log
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // Verify that input rowsets are converted to recycle rowsets
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        for (const auto& input_rowset : input_rowsets) {
            // Check that recycle rowset was created
            std::string recycle_key =
                    recycle_rowset_key({test_instance_id, tablet_id, input_rowset.rowset_id_v2()});
            std::string recycle_value;
            TxnErrorCode err = txn->get(recycle_key, &recycle_value);
            ASSERT_EQ(err, TxnErrorCode::TXN_OK)
                    << "Recycle rowset should exist for rowset " << input_rowset.rowset_id_v2();

            RecycleRowsetPB recycle_rowset_pb;
            ASSERT_TRUE(recycle_rowset_pb.ParseFromString(recycle_value));
            ASSERT_EQ(recycle_rowset_pb.rowset_meta().rowset_id_v2(), input_rowset.rowset_id_v2());
            ASSERT_EQ(recycle_rowset_pb.rowset_meta().tablet_id(), tablet_id);

            // Check that compact and load keys don't exist (they never existed in this test scenario)
            std::string meta_rowset_compact_key = versioned::meta_rowset_compact_key(
                    {test_instance_id, tablet_id, input_rowset.end_version()});
            std::string meta_rowset_load_key = versioned::meta_rowset_load_key(
                    {test_instance_id, tablet_id, input_rowset.end_version()});

            RowsetMetaCloudPB compact_pb, load_pb;
            Versionstamp compact_vs, load_vs;
            if (input_rowset.end_version() == output_end_version) {
                ASSERT_EQ(versioned::document_get(txn.get(), meta_rowset_compact_key, &compact_pb,
                                                  &compact_vs),
                          TxnErrorCode::TXN_OK)
                        << "Output rowset compact meta should exist";

                // Check that this pb is indeed the 2-4 output rowset
                ASSERT_EQ(compact_pb.start_version(), output_start_version)
                        << "Output rowset should have start_version = " << output_start_version;
                ASSERT_EQ(compact_pb.end_version(), output_end_version)
                        << "Output rowset should have end_version = " << output_end_version;
                ASSERT_EQ(compact_pb.tablet_id(), tablet_id)
                        << "Output rowset should have correct tablet_id";
                ASSERT_EQ(compact_pb.rowset_id_v2(), output_rowset.rowset_id_v2())
                        << "Output rowset should have correct rowset_id_v2";

            } else {
                ASSERT_EQ(versioned::document_get(txn.get(), meta_rowset_compact_key, &compact_pb,
                                                  &compact_vs),
                          TxnErrorCode::TXN_KEY_NOT_FOUND)
                        << "Input rowset compact meta should not exist";
                ASSERT_EQ(versioned::document_get(txn.get(), meta_rowset_load_key, &load_pb,
                                                  &load_vs),
                          TxnErrorCode::TXN_KEY_NOT_FOUND)
                        << "Input rowset load meta should not exist";
            }
        }
    }

    // Verify tablet compact stats still exist (recycling compaction log should not delete them)
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        std::string tablet_compact_stats_key =
                versioned::tablet_compact_stats_key({test_instance_id, tablet_id});
        std::string tablet_compact_stats_value;
        Versionstamp* versionstamp = nullptr;
        TxnErrorCode err = versioned_get(txn.get(), tablet_compact_stats_key, versionstamp,
                                         &tablet_compact_stats_value);
        ASSERT_EQ(err, TxnErrorCode::TXN_OK)
                << "Tablet compact stats should still exist after recycling compaction log";
    }

    // Verify the operation log is removed
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        std::string log_key_with_version =
                encode_versioned_key(versioned::log_key(test_instance_id), log_version);
        std::string log_value;
        TxnErrorCode err = txn->get(log_key_with_version, &log_value);
        ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND)
                << "Operation log should be removed after recycling";
    }

    // Now test second compaction to verify meta_rowset_compact_key deletion
    // Add version 5 rowset for second compaction and update tablet stats
    auto version5_rowset = create_rowset(300, tablet_id, partition_id, 5, 50);
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        // Add version 5 rowset
        auto rowset_key_5 = meta_rowset_key({test_instance_id, tablet_id, 5});
        txn->put(rowset_key_5, version5_rowset.SerializeAsString());

        // Update tablet stats to include version 5 rowset
        TabletStatsPB updated_stats;
        updated_stats.set_num_rows(150);        // 100 from 2-4 + 50 from version 5
        updated_stats.set_data_size(150 * 100); // Updated data size
        updated_stats.set_num_rowsets(2);       // 2-4 rowset + version 5 rowset
        updated_stats.set_num_segments(2);
        updated_stats.set_index_size(150);
        updated_stats.set_segment_size(300);
        updated_stats.set_cumulative_point(5); // Updated cumulative point

        auto stats_key =
                stats_tablet_key({test_instance_id, table_id, index_id, partition_id, tablet_id});
        txn->put(stats_key, updated_stats.SerializeAsString());

        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Create second output rowset as tmp rowset (2-5)
    constexpr int64_t second_txn_id = 30002;
    constexpr int64_t second_output_end_version = 5;
    auto second_output_rowset = create_rowset(400, tablet_id, partition_id, 2, 200);
    second_output_rowset.set_end_version(second_output_end_version);
    second_output_rowset.set_txn_id(second_txn_id);
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        auto tmp_rowset_key = meta_rowset_tmp_key({test_instance_id, second_txn_id, tablet_id});
        txn->put(tmp_rowset_key, second_output_rowset.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Start second compaction job (2-4 + 5-5 -> 2-5)
    const std::string second_job_id = "test_second_compaction_job";
    {
        brpc::Controller cntl;
        StartTabletJobRequest req;
        StartTabletJobResponse res;
        req.mutable_job()->mutable_idx()->set_table_id(table_id);
        req.mutable_job()->mutable_idx()->set_index_id(index_id);
        req.mutable_job()->mutable_idx()->set_partition_id(partition_id);
        req.mutable_job()->mutable_idx()->set_tablet_id(tablet_id);
        auto* compaction = req.mutable_job()->add_compaction();
        compaction->set_id(second_job_id);
        compaction->set_initiator(initiator);
        compaction->set_type(TabletCompactionJobPB::CUMULATIVE);
        compaction->set_base_compaction_cnt(0);
        compaction->set_cumulative_compaction_cnt(1); // Already did one compaction
        compaction->add_input_versions(2);            // 2-4 range
        compaction->add_input_versions(5);            // + version 5
        long now = time(nullptr);
        compaction->set_expiration(now + 12);
        compaction->set_lease(now + 3);
        meta_service->start_tablet_job(&cntl, &req, &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    // Finish second compaction job to create second compaction log
    {
        brpc::Controller cntl;
        FinishTabletJobRequest req;
        FinishTabletJobResponse res;

        req.set_action(FinishTabletJobRequest::COMMIT);
        req.mutable_job()->mutable_idx()->set_table_id(table_id);
        req.mutable_job()->mutable_idx()->set_index_id(index_id);
        req.mutable_job()->mutable_idx()->set_partition_id(partition_id);
        req.mutable_job()->mutable_idx()->set_tablet_id(tablet_id);

        auto* compaction = req.mutable_job()->add_compaction();
        compaction->set_id(second_job_id);
        compaction->set_initiator(initiator);
        compaction->set_type(TabletCompactionJobPB::CUMULATIVE);

        // Input: 2-4 and 5-5
        compaction->add_input_versions(2);
        compaction->add_input_versions(5);

        // Output: 2-5
        compaction->add_txn_id(second_txn_id);
        compaction->add_output_versions(second_output_end_version);
        compaction->add_output_rowset_ids(second_output_rowset.rowset_id_v2());
        compaction->set_output_cumulative_point(6);

        // Compaction stats
        compaction->set_size_input_rowsets(150 * 50);
        compaction->set_index_size_input_rowsets(150);
        compaction->set_segment_size_input_rowsets(300);
        compaction->set_num_input_rows(150);
        compaction->set_num_input_rowsets(2);
        compaction->set_num_input_segments(2);

        compaction->set_size_output_rowsets(200 * 50);
        compaction->set_index_size_output_rowsets(100);
        compaction->set_segment_size_output_rowsets(200);
        compaction->set_num_output_rows(200);
        compaction->set_num_output_rowsets(1);
        compaction->set_num_output_segments(1);

        // This creates the second compaction log
        meta_service->finish_tablet_job(&cntl, &req, &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    // Verify that 2-4 meta_rowset_compact_key exists before recycling
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        auto compact_key_24 = versioned::meta_rowset_compact_key({test_instance_id, tablet_id, 4});
        RowsetMetaCloudPB compact_pb_24;
        Versionstamp compact_vs_24;
        ASSERT_EQ(
                versioned::document_get(txn.get(), compact_key_24, &compact_pb_24, &compact_vs_24),
                TxnErrorCode::TXN_OK)
                << "2-4 rowset compact meta should exist before second recycling";
        // Check that this pb is indeed the 2-4 rowset
        ASSERT_EQ(compact_pb_24.start_version(), 2) << "Should be 2-4 rowset";
        ASSERT_EQ(compact_pb_24.end_version(), 4) << "Should be 2-4 rowset";
        ASSERT_EQ(compact_pb_24.rowset_id_v2(), output_rowset.rowset_id_v2())
                << "Should match output rowset ID";
    }

    // Recycle the second compaction log using the same recycler
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // Verify that 2-4 meta_rowset_compact_key is now deleted
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        auto compact_key_24 = versioned::meta_rowset_compact_key({test_instance_id, tablet_id, 4});
        RowsetMetaCloudPB compact_pb_24;
        Versionstamp compact_vs_24;
        ASSERT_EQ(
                versioned::document_get(txn.get(), compact_key_24, &compact_pb_24, &compact_vs_24),
                TxnErrorCode::TXN_KEY_NOT_FOUND)
                << "2-4 rowset compact meta should be deleted after recycling second compaction "
                   "log";

        // Verify that recycle rowset was created for 2-4
        std::string recycle_key_24 =
                recycle_rowset_key({test_instance_id, tablet_id, output_rowset.rowset_id_v2()});
        std::string recycle_value_24;
        ASSERT_EQ(txn->get(recycle_key_24, &recycle_value_24), TxnErrorCode::TXN_OK)
                << "Recycle rowset should exist for 2-4 rowset";

        RecycleRowsetPB recycle_pb_24;
        ASSERT_TRUE(recycle_pb_24.ParseFromString(recycle_value_24));
        ASSERT_EQ(recycle_pb_24.rowset_meta().start_version(), 2);
        ASSERT_EQ(recycle_pb_24.rowset_meta().end_version(), 4);
        ASSERT_EQ(recycle_pb_24.rowset_meta().rowset_id_v2(), output_rowset.rowset_id_v2());
    }

    // Finally, test tablet recycling to verify tablet compact stats deletion
    // Now recycle tablets to clean up tablet compact stats
    RecyclerMetricsContext ctx(instance_id, "test");
    ASSERT_EQ(recycler.recycle_tablets(table_id, index_id, ctx), 0);

    // Verify tablet compact stats are now deleted (recycling tablets should delete them)
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        std::string tablet_compact_stats_key =
                versioned::tablet_compact_stats_key({test_instance_id, tablet_id});
        std::string tablet_compact_stats_value;
        Versionstamp* versionstamp = nullptr;
        TxnErrorCode err = versioned_get(txn.get(), tablet_compact_stats_key, versionstamp,
                                         &tablet_compact_stats_value);
        ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND)
                << "Tablet compact stats should be deleted after recycling tablets";
    }
}

TEST(RecycleOperationLogTest, RecycleSchemaChangeLog) {
    // =========================================================================
    //                    Recycle Schema Change Operation Log Test
    // =========================================================================
    // This test simulates the complete schema change process and then verifies that:
    // 1. recycler.recycle_operation_logs removes schema change operation logs
    // 2. meta_rowset_compact_key is properly deleted after recycling operation logs
    // 3. recycler.recycle_tablets removes meta tablet key and tablet load stats key
    //
    // Test Scenario (copied from SchemaChangeLog):
    // - Old tablet starts with versions [2-2, 3-3] (20+30=50 rows)
    // - During schema change, both tablets receive versions [4-4, 5-5] (alter version = 5, 40+50=90 rows)
    // - After schema change starts, both tablets receive versions [6-6, 7-7] (60+70=130 rows)
    // - When finishing schema change:
    //   * Old tablet's [2-2, 3-3, 4-4, 5-5] data is converted to tmp rowsets in new tablet
    //   * New tablet deletes its own [4-4, 5-5] rowsets and converts tmp rowsets to permanent
    //   * New tablet keeps [6-6, 7-7] unchanged
    //   * Schema change net effect: add [2-2, 3-3] data to new tablet's tablet_load_stats
    //   * Multi-version keys are properly created for new tablet
    // - Then test recycling to clean up schema change operation logs and keys
    // =========================================================================
    auto old_flag = config::enable_recycle_rowset_strip_key_bounds;
    config::enable_recycle_rowset_strip_key_bounds = true;
    DORIS_CLOUD_DEFER {
        config::enable_recycle_rowset_strip_key_bounds = old_flag;
    };

    // Step 1: Initialize test environment using recycler infrastructure
    auto meta_service = get_meta_service(false);
    std::string instance_id = "recycle_schema_change_test";
    auto* sp = SyncPoint::get_instance();
    DORIS_CLOUD_DEFER {
        SyncPoint::get_instance()->clear_all_call_backs();
    };
    sp->set_call_back("get_instance_id", [&](auto&& args) {
        auto* ret = try_any_cast_ret<std::string>(args);
        ret->first = instance_id;
        ret->second = true;
    });
    sp->set_call_back("check_lazy_txn_finished::bypass_check", [&](auto&& args) {
        auto* ret = doris::try_any_cast_ret<bool>(args);
        ret->first = true;
        ret->second = true;
    });
    sp->set_call_back("delete_rowset_data::bypass_check", [&](auto&& args) {
        auto* ret = doris::try_any_cast_ret<bool>(args);
        ret->first = true;
        ret->second = true;
    });
    sp->set_call_back("recycle_tablet::begin", [&](auto&& args) {
        auto* ret = doris::try_any_cast_ret<int>(args);
        ret->first = 0;
        ret->second = true;
    });
    sp->enable_processing();

    constexpr int64_t table_id = 50001;
    constexpr int64_t index_id = 50002;
    constexpr int64_t partition_id = 50003;
    constexpr int64_t old_tablet_id = 50004;
    constexpr int64_t new_tablet_id = 50005;

    // Create instance with multi-version support
    {
        InstanceInfoPB instance_info;
        instance_info.set_instance_id(instance_id);
        instance_info.set_multi_version_status(MULTI_VERSION_ENABLED);
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        txn->put(instance_key(instance_id), instance_info.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

        meta_service->resource_mgr()->refresh_instance(instance_id);
        ASSERT_TRUE(meta_service->resource_mgr()->is_version_write_enabled(instance_id));
    }

    // Create old tablet
    create_tablet(meta_service.get(), table_id, index_id, partition_id, old_tablet_id);

    auto txn_kv = meta_service->txn_kv();

    // Create initial rowsets for old tablet: versions 2, 3
    std::vector<doris::RowsetMetaCloudPB> initial_rowsets;
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        for (int version = 2; version <= 3; ++version) {
            auto rowset = create_rowset(100 + version, old_tablet_id, partition_id, version,
                                        version * 10);
            initial_rowsets.push_back(rowset);

            auto rowset_key = meta_rowset_key({instance_id, old_tablet_id, version});
            auto rowset_val = rowset.SerializeAsString();
            txn->put(rowset_key, rowset_val);
            auto versioned_rowset_key = versioned::meta_rowset_load_key(
                    {instance_id, old_tablet_id, rowset.end_version()});
            versioned::document_put(txn.get(), versioned_rowset_key, std::move(rowset));
        }
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Step 2: Create new tablet in NOTREADY state (empty, waiting for schema change)
    {
        brpc::Controller cntl;
        CreateTabletsRequest req;
        CreateTabletsResponse res;
        req.set_db_id(1000);
        req.set_cloud_unique_id("test_cloud_unique_id");
        add_tablet(req, table_id, index_id, partition_id, new_tablet_id);

        // Set new tablet state to NOTREADY (for schema change)
        auto tablet_meta = req.mutable_tablet_metas(0);
        tablet_meta->set_tablet_state(doris::TabletStatePB::PB_NOTREADY);

        meta_service->create_tablets(&cntl, &req, &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    // Step 3: Add alter versions [4-4, 5-5] and start schema change job (alter_version = 5)
    const std::string job_id = "recycle_schema_change_job";
    const std::string initiator = "test_be";
    constexpr int64_t alter_version = 5; // Determined when starting the job

    // Add versions 4-5 to old tablet (alter versions)
    std::vector<doris::RowsetMetaCloudPB> alter_rowsets_old;
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        for (int version = 4; version <= 5; ++version) {
            auto rowset = create_rowset(200 + version, old_tablet_id, partition_id, version,
                                        version * 10);
            alter_rowsets_old.push_back(rowset);

            auto rowset_key = meta_rowset_key({instance_id, old_tablet_id, version});
            auto rowset_val = rowset.SerializeAsString();
            txn->put(rowset_key, rowset_val);
            auto versioned_rowset_key = versioned::meta_rowset_load_key(
                    {instance_id, old_tablet_id, rowset.end_version()});
            versioned::document_put(txn.get(), versioned_rowset_key, std::move(rowset));
        }
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Add versions 4-5 to new tablet (alter versions)
    std::vector<doris::RowsetMetaCloudPB> alter_rowsets_new;
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        for (int version = 4; version <= 5; ++version) {
            auto rowset = create_rowset(300 + version, new_tablet_id, partition_id, version,
                                        version * 10);
            alter_rowsets_new.push_back(rowset);

            auto rowset_key = meta_rowset_key({instance_id, new_tablet_id, version});
            auto rowset_val = rowset.SerializeAsString();
            txn->put(rowset_key, rowset_val);
            auto versioned_rowset_key = versioned::meta_rowset_load_key(
                    {instance_id, new_tablet_id, rowset.end_version()});
            versioned::document_put(txn.get(), versioned_rowset_key, std::move(rowset));
        }
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Start schema change job
    {
        brpc::Controller cntl;
        StartTabletJobRequest req;
        StartTabletJobResponse res;
        req.mutable_job()->mutable_idx()->set_tablet_id(old_tablet_id);
        auto* schema_change = req.mutable_job()->mutable_schema_change();
        schema_change->set_id(job_id);
        schema_change->set_initiator(initiator);
        schema_change->mutable_new_tablet_idx()->set_table_id(table_id);
        schema_change->mutable_new_tablet_idx()->set_index_id(index_id);
        schema_change->mutable_new_tablet_idx()->set_partition_id(partition_id);
        schema_change->mutable_new_tablet_idx()->set_tablet_id(new_tablet_id);
        schema_change->set_alter_version(alter_version);

        long now = time(nullptr);
        schema_change->set_expiration(now + 12);

        meta_service->start_tablet_job(&cntl, &req, &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    // Step 4: Simulate ongoing data ingestion - add post-alter versions [6-6, 7-7]
    std::vector<doris::RowsetMetaCloudPB> post_alter_rowsets_old;
    std::vector<doris::RowsetMetaCloudPB> post_alter_rowsets_new;

    // Add versions 6-7 to old tablet
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        for (int version = 6; version <= 7; ++version) {
            auto rowset = create_rowset(400 + version, old_tablet_id, partition_id, version,
                                        version * 10);
            post_alter_rowsets_old.push_back(rowset);

            auto rowset_key = meta_rowset_key({instance_id, old_tablet_id, version});
            auto rowset_val = rowset.SerializeAsString();
            txn->put(rowset_key, rowset_val);
            auto versioned_rowset_key = versioned::meta_rowset_load_key(
                    {instance_id, old_tablet_id, rowset.end_version()});
            versioned::document_put(txn.get(), versioned_rowset_key, std::move(rowset));
        }
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Add versions 6-7 to new tablet
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        for (int version = 6; version <= 7; ++version) {
            auto rowset = create_rowset(500 + version, new_tablet_id, partition_id, version,
                                        version * 10);
            post_alter_rowsets_new.push_back(rowset);

            auto rowset_key = meta_rowset_key({instance_id, new_tablet_id, version});
            auto rowset_val = rowset.SerializeAsString();
            txn->put(rowset_key, rowset_val);
            auto versioned_rowset_key = versioned::meta_rowset_load_key(
                    {instance_id, new_tablet_id, rowset.end_version()});
            versioned::document_put(txn.get(), versioned_rowset_key, std::move(rowset));
        }
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Step 5: Complete schema change - convert old tablet data via tmp rowsets
    // Create tmp rowsets representing old tablet's [2-2, 3-3, 4-4, 5-5] data that will be converted to new tablet
    std::vector<int64_t> tmp_txn_ids;
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        // Create tmp rowsets for versions 2-2, 3-3, 4-4, 5-5 (simulating old tablet data conversion)
        for (int version = 2; version <= 5; ++version) {
            int64_t tmp_txn_id = 40000 + version;
            tmp_txn_ids.push_back(tmp_txn_id);

            // Create tmp rowset with data from old tablet
            auto tmp_rowset = create_rowset(600 + version, new_tablet_id, partition_id, version,
                                            version * 10);
            tmp_rowset.set_txn_id(tmp_txn_id);

            auto tmp_rowset_key = meta_rowset_tmp_key({instance_id, tmp_txn_id, new_tablet_id});
            auto tmp_rowset_val = tmp_rowset.SerializeAsString();
            txn->put(tmp_rowset_key, tmp_rowset_val);
        }

        // Create initial tablet stats for both tablets
        TabletStatsPB old_tablet_stats;
        old_tablet_stats.set_num_rows(270); // sum of versions 2-7: 20+30+40+50+60+70
        old_tablet_stats.set_data_size(270 * 100);
        old_tablet_stats.set_num_rowsets(6);
        old_tablet_stats.set_num_segments(6);
        old_tablet_stats.set_index_size(300);
        old_tablet_stats.set_segment_size(600);
        old_tablet_stats.set_cumulative_point(3);

        auto old_stats_key =
                stats_tablet_key({instance_id, table_id, index_id, partition_id, old_tablet_id});
        txn->put(old_stats_key, old_tablet_stats.SerializeAsString());

        // Create tablet load stats for old tablet (versions 2-2, 3-3, 4-4, 5-5, 6-6, 7-7: all data)
        TabletStatsPB old_tablet_load_stats;
        old_tablet_load_stats.set_num_rows(270); // all versions 2-7: 20+30+40+50+60+70
        old_tablet_load_stats.set_data_size(270 * 100);
        old_tablet_load_stats.set_num_rowsets(6);
        old_tablet_load_stats.set_num_segments(6);
        old_tablet_load_stats.set_index_size(300);
        old_tablet_load_stats.set_segment_size(600);
        old_tablet_load_stats.set_cumulative_point(3);

        auto old_tablet_load_stats_key =
                versioned::tablet_load_stats_key({instance_id, old_tablet_id});
        versioned_put(txn.get(), old_tablet_load_stats_key,
                      old_tablet_load_stats.SerializeAsString());

        TabletStatsPB new_tablet_stats;
        new_tablet_stats.set_num_rows(220); // versions 4-7: 40+50+60+70
        new_tablet_stats.set_data_size(220 * 100);
        new_tablet_stats.set_num_rowsets(4);
        new_tablet_stats.set_num_segments(4);
        new_tablet_stats.set_index_size(200);
        new_tablet_stats.set_segment_size(400);
        new_tablet_stats.set_cumulative_point(3);

        auto new_stats_key =
                stats_tablet_key({instance_id, table_id, index_id, partition_id, new_tablet_id});
        txn->put(new_stats_key, new_tablet_stats.SerializeAsString());

        // Create initial tablet load stats for new tablet (versions 4-4, 5-5, 6-6, 7-7)
        TabletStatsPB new_tablet_load_stats;
        new_tablet_load_stats.set_num_rows(220); // versions 4-7: 40+50+60+70
        new_tablet_load_stats.set_data_size(220 * 100);
        new_tablet_load_stats.set_num_rowsets(4);
        new_tablet_load_stats.set_num_segments(4);
        new_tablet_load_stats.set_index_size(200);
        new_tablet_load_stats.set_segment_size(400);
        new_tablet_load_stats.set_cumulative_point(6);

        auto new_tablet_load_stats_key =
                versioned::tablet_load_stats_key({instance_id, new_tablet_id});
        versioned_put(txn.get(), new_tablet_load_stats_key,
                      new_tablet_load_stats.SerializeAsString());

        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    size_t num_logs_before = count_range(txn_kv.get(), versioned::log_key(instance_id),
                                         versioned::log_key(instance_id) + "\xFF");

    // Call finish_tablet_job to complete schema change (triggers process_schema_change_job)
    {
        brpc::Controller cntl;
        FinishTabletJobRequest req;
        FinishTabletJobResponse res;

        req.set_action(FinishTabletJobRequest::COMMIT);
        req.mutable_job()->mutable_idx()->set_table_id(table_id);
        req.mutable_job()->mutable_idx()->set_index_id(index_id);
        req.mutable_job()->mutable_idx()->set_partition_id(partition_id);
        req.mutable_job()->mutable_idx()->set_tablet_id(old_tablet_id);

        auto* schema_change = req.mutable_job()->mutable_schema_change();
        schema_change->set_id(job_id);
        schema_change->set_initiator(initiator);
        schema_change->mutable_new_tablet_idx()->set_table_id(table_id);
        schema_change->mutable_new_tablet_idx()->set_index_id(index_id);
        schema_change->mutable_new_tablet_idx()->set_partition_id(partition_id);
        schema_change->mutable_new_tablet_idx()->set_tablet_id(new_tablet_id);
        schema_change->set_alter_version(alter_version);

        // Specify tmp txn ids containing converted data from old tablet [2-2, 3-3, 4-4, 5-5]
        for (int64_t tmp_txn_id : tmp_txn_ids) {
            schema_change->add_txn_ids(tmp_txn_id);
        }

        // Add output versions corresponding to tmp rowsets [2-2, 3-3, 4-4, 5-5]
        for (int version = 2; version <= 5; ++version) {
            schema_change->add_output_versions(version);
        }

        // Add schema change statistics
        // Output rowsets [2-2, 3-3, 4-4, 5-5]: 20+30+40+50 = 140
        schema_change->set_num_output_rows(140);
        schema_change->set_num_output_rowsets(4);
        schema_change->set_num_output_segments(4);
        schema_change->set_size_output_rowsets(140 * 100);
        schema_change->set_index_size_output_rowsets(200);
        schema_change->set_segment_size_output_rowsets(400);
        schema_change->set_output_cumulative_point(5);

        meta_service->finish_tablet_job(&cntl, &req, &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    // Verification Step: Check schema change operation log was generated
    size_t num_logs_after = count_range(txn_kv.get(), versioned::log_key(instance_id),
                                        versioned::log_key(instance_id) + "\xFF");
    ASSERT_GT(num_logs_after, num_logs_before)
            << "Expected new schema change operation log, but found no new logs";

    // Verify recycle rowset metas in schema change log have key bounds stripped
    {
        Versionstamp log_version;
        OperationLogPB operation_log;
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string log_key = versioned::log_key({instance_id});
        ASSERT_EQ(read_operation_log(txn.get(), log_key, &log_version, &operation_log),
                  TxnErrorCode::TXN_OK);
        ASSERT_TRUE(operation_log.has_schema_change());
        const auto& schema_change_log = operation_log.schema_change();
        ASSERT_GT(schema_change_log.recycle_rowsets_size(), 0);
        for (const auto& recycle_rs : schema_change_log.recycle_rowsets()) {
            ASSERT_EQ(recycle_rs.rowset_meta().segments_key_bounds_size(), 0);
            ASSERT_FALSE(recycle_rs.rowset_meta().has_segments_key_bounds_truncated());
        }
    }

    // Verify meta_rowset_compact_keys exist before recycling
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        // Check that meta_rowset_compact_keys exist for all converted rowsets [2-2, 3-3, 4-4, 5-5]
        for (int version = 2; version <= 5; ++version) {
            auto meta_rowset_compact_key =
                    versioned::meta_rowset_compact_key({instance_id, new_tablet_id, version});
            doris::RowsetMetaCloudPB compact_rowset_meta;
            Versionstamp versionstamp;
            ASSERT_EQ(versioned::document_get(txn.get(), meta_rowset_compact_key,
                                              &compact_rowset_meta, &versionstamp),
                      TxnErrorCode::TXN_OK)
                    << "Meta rowset compact key should exist for version " << version
                    << " before recycling";
        }
    }

    // Set up recycler using the same txn_kv as meta_service
    InstanceInfoPB instance_info;
    instance_info.set_instance_id(instance_id);
    instance_info.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
    InstanceRecycler recycler(txn_kv, instance_info, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    // Now recycle the schema change operation log
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // Verify schema change recycling results
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        // Check that meta_rowset_keys exist for all versions 2-7 (schema change writes 2-7 to new tablet)
        for (int version = 2; version <= 7; ++version) {
            auto meta_key = meta_rowset_key({instance_id, new_tablet_id, version});
            std::string rowset_value;
            ASSERT_EQ(txn->get(meta_key, &rowset_value), TxnErrorCode::TXN_OK)
                    << "Meta rowset key should exist for version " << version
                    << " after schema change";
        }

        // Check that recycle_rowset_keys exist only for deleted versions 4-5 (schema change deletes new tablet's original 4-5 rowsets)
        for (int version = 4; version <= 5; ++version) {
            int idx = version - 4;
            std::string rowset_id_v2 = alter_rowsets_new[idx].rowset_id_v2();
            std::string recycle_key =
                    recycle_rowset_key({instance_id, new_tablet_id, rowset_id_v2});
            std::string recycle_value;
            ASSERT_EQ(txn->get(recycle_key, &recycle_value), TxnErrorCode::TXN_OK)
                    << "Recycle rowset key should exist for deleted version " << version
                    << " with rowset_id_v2 " << rowset_id_v2;
        }

        // Check that recycle_rowset_keys do NOT exist for versions 2-3 (these are newly created from tmp conversion, not deleted)
        // Note: We cannot easily check this as we don't know the exact rowset_id_v2 for newly created rowsets

        // Check that recycle_rowset_keys do NOT exist for versions 6-7 (these were not deleted by schema change)
        for (int version = 6; version <= 7; ++version) {
            int idx = version - 6;
            std::string rowset_id_v2 = post_alter_rowsets_new[idx].rowset_id_v2();
            std::string recycle_key =
                    recycle_rowset_key({instance_id, new_tablet_id, rowset_id_v2});
            std::string recycle_value;
            ASSERT_EQ(txn->get(recycle_key, &recycle_value), TxnErrorCode::TXN_KEY_NOT_FOUND)
                    << "Recycle rowset key should NOT exist for non-deleted version " << version
                    << " with rowset_id_v2 " << rowset_id_v2;
        }

        // Verify that old tablet's meta_rowset_compact_keys are deleted after recycling operation logs
        for (int version = 2; version <= 5; ++version) {
            auto old_meta_rowset_compact_key =
                    versioned::meta_rowset_compact_key({instance_id, old_tablet_id, version});
            doris::RowsetMetaCloudPB compact_rowset_meta;
            Versionstamp versionstamp;
            ASSERT_EQ(versioned::document_get(txn.get(), old_meta_rowset_compact_key,
                                              &compact_rowset_meta, &versionstamp),
                      TxnErrorCode::TXN_KEY_NOT_FOUND)
                    << "Old tablet meta rowset compact key should be deleted for version "
                    << version << " after recycling operation logs";
        }

        // Verify that new tablet's meta_rowset_compact_keys still exist after recycling operation logs
        for (int version = 2; version <= 5; ++version) {
            auto new_meta_rowset_compact_key =
                    versioned::meta_rowset_compact_key({instance_id, new_tablet_id, version});
            doris::RowsetMetaCloudPB compact_rowset_meta;
            Versionstamp versionstamp;
            ASSERT_EQ(versioned::document_get(txn.get(), new_meta_rowset_compact_key,
                                              &compact_rowset_meta, &versionstamp),
                      TxnErrorCode::TXN_OK)
                    << "New tablet meta rowset compact key should still exist for version "
                    << version << " after recycling operation logs";
        }
    }

    // Verify tablet load stats and meta tablet keys exist before recycling tablets
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        // Check tablet load stats keys exist
        auto old_tablet_load_stats_key =
                versioned::tablet_load_stats_key({instance_id, old_tablet_id});
        auto new_tablet_load_stats_key =
                versioned::tablet_load_stats_key({instance_id, new_tablet_id});
        std::string load_stats_value;
        Versionstamp* versionstamp = nullptr;

        ASSERT_EQ(versioned_get(txn.get(), old_tablet_load_stats_key, versionstamp,
                                &load_stats_value),
                  TxnErrorCode::TXN_OK)
                << "Old tablet load stats should exist before recycling tablets";
        ASSERT_EQ(versioned_get(txn.get(), new_tablet_load_stats_key, versionstamp,
                                &load_stats_value),
                  TxnErrorCode::TXN_OK)
                << "New tablet load stats should exist before recycling tablets";
    }

    // Verify that meta_tablet_keys exist before recycling tablets
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        auto old_tablet_key =
                meta_tablet_key({instance_id, table_id, index_id, partition_id, old_tablet_id});
        auto new_tablet_key =
                meta_tablet_key({instance_id, table_id, index_id, partition_id, new_tablet_id});
        std::string tablet_meta_value;

        LOG(INFO) << "Checking old_tablet_key: " << hex(old_tablet_key);
        TxnErrorCode old_err = txn->get(old_tablet_key, &tablet_meta_value);
        LOG(INFO) << "Old tablet key get result: " << old_err;

        LOG(INFO) << "Checking new_tablet_key: " << hex(new_tablet_key);
        TxnErrorCode new_err = txn->get(new_tablet_key, &tablet_meta_value);
        LOG(INFO) << "New tablet key get result: " << new_err;

        ASSERT_EQ(old_err, TxnErrorCode::TXN_OK)
                << "Old tablet meta key should exist before recycling";
        ASSERT_EQ(new_err, TxnErrorCode::TXN_OK)
                << "New tablet meta key should exist before recycling";
    }

    // Finally, test tablet recycling to verify tablet load stats deletion
    RecyclerMetricsContext ctx(instance_id, "test");
    int recycled = recycler.recycle_tablets(table_id, index_id, ctx);
    LOG(INFO) << "recycle_tablets returned: " << recycled;
    ASSERT_EQ(recycled, 0);

    // Verify tablet load stats are now deleted (recycling tablets should delete them)
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        // Check that tablet load stats keys are deleted
        auto old_tablet_load_stats_key =
                versioned::tablet_load_stats_key({instance_id, old_tablet_id});
        auto new_tablet_load_stats_key =
                versioned::tablet_load_stats_key({instance_id, new_tablet_id});
        std::string load_stats_value;
        Versionstamp* versionstamp = nullptr;

        ASSERT_EQ(versioned_get(txn.get(), old_tablet_load_stats_key, versionstamp,
                                &load_stats_value),
                  TxnErrorCode::TXN_KEY_NOT_FOUND)
                << "Old tablet load stats should be deleted after recycling tablets";
        ASSERT_EQ(versioned_get(txn.get(), new_tablet_load_stats_key, versionstamp,
                                &load_stats_value),
                  TxnErrorCode::TXN_KEY_NOT_FOUND)
                << "New tablet load stats should be deleted after recycling tablets";

        // Check that versioned meta tablet keys are deleted
        auto old_tablet_key = versioned::meta_tablet_key({instance_id, old_tablet_id});
        auto new_tablet_key = versioned::meta_tablet_key({instance_id, new_tablet_id});
        std::string tablet_meta_value;

        ASSERT_EQ(versioned_get(txn.get(), old_tablet_key, versionstamp, &tablet_meta_value),
                  TxnErrorCode::TXN_KEY_NOT_FOUND)
                << "Old versioned meta tablet key should be deleted after recycling tablets";
        ASSERT_EQ(versioned_get(txn.get(), new_tablet_key, versionstamp, &tablet_meta_value),
                  TxnErrorCode::TXN_KEY_NOT_FOUND)
                << "New versioned meta tablet key should be deleted after recycling tablets";

        // Verify that new tablet's meta_rowset_compact_keys still exist after recycling tablets
        // (they are not deleted during tablet recycling)
        for (int version = 2; version <= 5; ++version) {
            auto new_meta_rowset_compact_key =
                    versioned::meta_rowset_compact_key({instance_id, new_tablet_id, version});
            doris::RowsetMetaCloudPB compact_rowset_meta;
            Versionstamp versionstamp;
            ASSERT_EQ(versioned::document_get(txn.get(), new_meta_rowset_compact_key,
                                              &compact_rowset_meta, &versionstamp),
                      TxnErrorCode::TXN_OK)
                    << "New tablet meta rowset compact key should still exist for version "
                    << version << " after recycling tablets";
        }
    }
}

TEST(RecycleOperationLogTest, RecycleSchemaChangeLogKeepKeyBoundsWhenDisabled) {
    auto old_flag = config::enable_recycle_rowset_strip_key_bounds;
    config::enable_recycle_rowset_strip_key_bounds = false;
    DORIS_CLOUD_DEFER {
        config::enable_recycle_rowset_strip_key_bounds = old_flag;
    };

    auto meta_service = get_meta_service(false);
    std::string instance_id = "recycle_schema_change_keep_bounds";
    auto* sp = SyncPoint::get_instance();
    DORIS_CLOUD_DEFER {
        SyncPoint::get_instance()->clear_all_call_backs();
    };
    sp->set_call_back("get_instance_id", [&](auto&& args) {
        auto* ret = try_any_cast_ret<std::string>(args);
        ret->first = instance_id;
        ret->second = true;
    });
    sp->set_call_back("check_lazy_txn_finished::bypass_check", [&](auto&& args) {
        auto* ret = doris::try_any_cast_ret<bool>(args);
        ret->first = true;
        ret->second = true;
    });
    sp->set_call_back("delete_rowset_data::bypass_check", [&](auto&& args) {
        auto* ret = doris::try_any_cast_ret<bool>(args);
        ret->first = true;
        ret->second = true;
    });
    sp->set_call_back("recycle_tablet::bypass_check", [&](auto&& args) {
        auto* ret = doris::try_any_cast_ret<bool>(args);
        ret->first = false;
        ret->second = true;
    });
    sp->enable_processing();

    constexpr int64_t table_id = 52001;
    constexpr int64_t index_id = 52002;
    constexpr int64_t partition_id = 52003;
    constexpr int64_t old_tablet_id = 52004;
    constexpr int64_t new_tablet_id = 52005;
    constexpr int64_t alter_version = 5;
    const std::string job_id = "recycle_schema_change_keep_bounds_job";
    const std::string initiator = "test_be";

    {
        InstanceInfoPB instance_info;
        instance_info.set_instance_id(instance_id);
        instance_info.set_multi_version_status(MULTI_VERSION_ENABLED);
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        txn->put(instance_key(instance_id), instance_info.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

        meta_service->resource_mgr()->refresh_instance(instance_id);
        ASSERT_TRUE(meta_service->resource_mgr()->is_version_write_enabled(instance_id));
    }

    // Create old tablet
    create_tablet(meta_service.get(), table_id, index_id, partition_id, old_tablet_id);
    auto txn_kv = meta_service->txn_kv();

    // Add initial rowsets 2-3 to old tablet
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int version = 2; version <= 3; ++version) {
            auto rowset = create_rowset(1000 + version, old_tablet_id, partition_id, version,
                                        version * 10);
            auto rowset_key = meta_rowset_key({instance_id, old_tablet_id, version});
            txn->put(rowset_key, rowset.SerializeAsString());
            auto versioned_rowset_key = versioned::meta_rowset_load_key(
                    {instance_id, old_tablet_id, rowset.end_version()});
            versioned::document_put(txn.get(), versioned_rowset_key, std::move(rowset));
        }
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Create new tablet in NOTREADY state
    {
        brpc::Controller cntl;
        CreateTabletsRequest req;
        CreateTabletsResponse res;
        req.set_db_id(1);
        req.set_cloud_unique_id("test_cloud_unique_id_keep_bounds");
        add_tablet(req, table_id, index_id, partition_id, new_tablet_id);
        req.mutable_tablet_metas(0)->set_tablet_state(doris::TabletStatePB::PB_NOTREADY);
        meta_service->create_tablets(&cntl, &req, &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    // Add alter versions 4-5 to both old and new tablets
    std::vector<doris::RowsetMetaCloudPB> alter_rowsets_new;
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int version = 4; version <= 5; ++version) {
            auto rowset_old = create_rowset(1100 + version, old_tablet_id, partition_id, version,
                                            version * 10);
            auto rowset_key_old = meta_rowset_key({instance_id, old_tablet_id, version});
            txn->put(rowset_key_old, rowset_old.SerializeAsString());
            auto versioned_rowset_key_old = versioned::meta_rowset_load_key(
                    {instance_id, old_tablet_id, rowset_old.end_version()});
            versioned::document_put(txn.get(), versioned_rowset_key_old, std::move(rowset_old));

            auto rowset_new = create_rowset(1200 + version, new_tablet_id, partition_id, version,
                                            version * 10);
            alter_rowsets_new.push_back(rowset_new);
            auto rowset_key_new = meta_rowset_key({instance_id, new_tablet_id, version});
            txn->put(rowset_key_new, rowset_new.SerializeAsString());
            auto versioned_rowset_key_new = versioned::meta_rowset_load_key(
                    {instance_id, new_tablet_id, rowset_new.end_version()});
            versioned::document_put(txn.get(), versioned_rowset_key_new, std::move(rowset_new));
        }
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Start schema change job
    {
        brpc::Controller cntl;
        StartTabletJobRequest req;
        StartTabletJobResponse res;
        req.mutable_job()->mutable_idx()->set_tablet_id(old_tablet_id);
        auto* schema_change = req.mutable_job()->mutable_schema_change();
        schema_change->set_id(job_id);
        schema_change->set_initiator(initiator);
        schema_change->mutable_new_tablet_idx()->set_table_id(table_id);
        schema_change->mutable_new_tablet_idx()->set_index_id(index_id);
        schema_change->mutable_new_tablet_idx()->set_partition_id(partition_id);
        schema_change->mutable_new_tablet_idx()->set_tablet_id(new_tablet_id);
        schema_change->set_alter_version(alter_version);
        long now = time(nullptr);
        schema_change->set_expiration(now + 12);
        meta_service->start_tablet_job(&cntl, &req, &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    // Create tmp rowsets for versions 2-5 (converted data)
    std::vector<int64_t> tmp_txn_ids;
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int version = 2; version <= 5; ++version) {
            int64_t tmp_txn_id = 520000 + version;
            tmp_txn_ids.push_back(tmp_txn_id);
            auto tmp_rowset = create_rowset(1300 + version, new_tablet_id, partition_id, version,
                                            version * 10);
            tmp_rowset.set_txn_id(tmp_txn_id);
            auto tmp_rowset_key = meta_rowset_tmp_key({instance_id, tmp_txn_id, new_tablet_id});
            txn->put(tmp_rowset_key, tmp_rowset.SerializeAsString());
        }

        // Seed tablet stats for new tablet to satisfy stats fetch
        TabletStatsPB new_tablet_stats;
        new_tablet_stats.set_num_rows(90); // versions 4-5: 40+50
        new_tablet_stats.set_data_size(90 * 100);
        new_tablet_stats.set_num_rowsets(2);
        new_tablet_stats.set_num_segments(2);
        new_tablet_stats.set_index_size(100);
        new_tablet_stats.set_segment_size(200);
        new_tablet_stats.set_cumulative_point(5);

        auto new_stats_key =
                stats_tablet_key({instance_id, table_id, index_id, partition_id, new_tablet_id});
        txn->put(new_stats_key, new_tablet_stats.SerializeAsString());

        auto new_tablet_load_stats_key =
                versioned::tablet_load_stats_key({instance_id, new_tablet_id});
        versioned_put(txn.get(), new_tablet_load_stats_key, new_tablet_stats.SerializeAsString());

        auto new_tablet_compact_stats_key =
                versioned::tablet_compact_stats_key({instance_id, new_tablet_id});
        versioned_put(txn.get(), new_tablet_compact_stats_key,
                      new_tablet_stats.SerializeAsString());

        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Finish schema change job
    {
        brpc::Controller cntl;
        FinishTabletJobRequest req;
        FinishTabletJobResponse res;
        req.set_action(FinishTabletJobRequest::COMMIT);
        req.mutable_job()->mutable_idx()->set_table_id(table_id);
        req.mutable_job()->mutable_idx()->set_index_id(index_id);
        req.mutable_job()->mutable_idx()->set_partition_id(partition_id);
        req.mutable_job()->mutable_idx()->set_tablet_id(old_tablet_id);

        auto* schema_change = req.mutable_job()->mutable_schema_change();
        schema_change->set_id(job_id);
        schema_change->set_initiator(initiator);
        schema_change->mutable_new_tablet_idx()->set_table_id(table_id);
        schema_change->mutable_new_tablet_idx()->set_index_id(index_id);
        schema_change->mutable_new_tablet_idx()->set_partition_id(partition_id);
        schema_change->mutable_new_tablet_idx()->set_tablet_id(new_tablet_id);
        schema_change->set_alter_version(alter_version);
        for (int64_t tmp_txn_id : tmp_txn_ids) {
            schema_change->add_txn_ids(tmp_txn_id);
        }
        for (int version = 2; version <= 5; ++version) {
            schema_change->add_output_versions(version);
        }
        schema_change->set_num_output_rows(140); // 20+30+40+50
        schema_change->set_num_output_rowsets(4);
        schema_change->set_num_output_segments(4);
        schema_change->set_size_output_rowsets(140 * 100);
        schema_change->set_index_size_output_rowsets(200);
        schema_change->set_segment_size_output_rowsets(400);
        schema_change->set_output_cumulative_point(5);

        meta_service->finish_tablet_job(&cntl, &req, &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    // Verify schema change operation log keeps key bounds when stripping is disabled
    {
        Versionstamp log_version;
        OperationLogPB operation_log;
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string log_key = versioned::log_key({instance_id});
        ASSERT_EQ(read_operation_log(txn.get(), log_key, &log_version, &operation_log),
                  TxnErrorCode::TXN_OK);
        ASSERT_TRUE(operation_log.has_schema_change());
        const auto& schema_change_log = operation_log.schema_change();
        ASSERT_EQ(schema_change_log.recycle_rowsets_size(), alter_rowsets_new.size());
        for (const auto& recycle_rs : schema_change_log.recycle_rowsets()) {
            ASSERT_GT(recycle_rs.rowset_meta().segments_key_bounds_size(), 0);
            ASSERT_TRUE(recycle_rs.rowset_meta().has_segments_key_bounds_truncated());
            ASSERT_FALSE(recycle_rs.rowset_meta().segments_key_bounds_truncated());
        }
    }
}

TEST(RecycleOperationLogTest, RecycleDeletedInstance) {
    auto txn_kv = std::make_shared<MemTxnKv>();
    ASSERT_EQ(txn_kv->init(), 0);

    InstanceInfoPB instance;
    instance.set_instance_id(instance_id);
    instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_WRITE_ONLY);
    update_instance_info(txn_kv.get(), instance);

    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    uint64_t tablet_id = 1, partition_id = 2, index_id = 3, table_id = 4, db_id = 5;
    uint64_t stream_db_id = 6, stream_id = 7;
    {
        // Create tablet meta
        std::string tablet_meta_key = versioned::meta_tablet_key({instance_id, tablet_id});
        doris::TabletMetaCloudPB tablet_meta;
        tablet_meta.set_tablet_id(tablet_id);
        tablet_meta.set_creation_time(::time(nullptr));

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), tablet_meta_key, tablet_meta.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    {
        // Put partition version
        std::string partition_version_key =
                versioned::partition_version_key({instance_id, partition_id});
        VersionPB version;
        version.set_version(12345);
        version.set_update_time_ms(0);
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), partition_version_key, version.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    {
        // Put partition index
        std::string partition_index_key =
                versioned::partition_index_key({instance_id, partition_id});
        PartitionIndexPB partition_index;
        partition_index.set_table_id(table_id);
        partition_index.set_db_id(db_id);

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        txn->put(partition_index_key, partition_index.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    {
        // Put rowset meta, tablet stats and data ref count
        // Create rowset meta
        std::string rowset_key = meta_rowset_key({instance_id, tablet_id, 2});
        doris::RowsetMetaCloudPB rowset_meta;
        rowset_meta.set_rowset_id(12345);
        rowset_meta.set_rowset_id_v2("test_rowset_id");
        rowset_meta.set_tablet_id(tablet_id);
        rowset_meta.set_partition_id(partition_id);
        rowset_meta.set_start_version(2);
        rowset_meta.set_end_version(2);
        rowset_meta.set_num_rows(100);
        rowset_meta.set_data_disk_size(10000);
        rowset_meta.set_index_disk_size(1000);
        rowset_meta.set_total_disk_size(11000);

        // Create tablet stats
        std::string tablet_stats_key =
                stats_tablet_key({instance_id, table_id, index_id, partition_id, tablet_id});
        TabletStatsPB tablet_stats;
        tablet_stats.set_num_rows(100);
        tablet_stats.set_data_size(10000);
        tablet_stats.set_num_rowsets(1);
        tablet_stats.set_num_segments(1);
        tablet_stats.set_index_size(1000);
        tablet_stats.set_segment_size(11000);
        tablet_stats.set_cumulative_point(1);

        // Create data ref count
        std::string data_ref_count_key =
                versioned::data_rowset_ref_count_key({instance_id, tablet_id, "test_rowset_id"});
        std::string ref_count = "1";

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned::document_put(txn.get(), rowset_key, std::move(rowset_meta));
        versioned_put(txn.get(), tablet_stats_key, tablet_stats.SerializeAsString());
        versioned_put(txn.get(), data_ref_count_key, ref_count);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    {
        // Put operation logs
        std::string log_key = versioned::log_key(instance_id);
        Versionstamp versionstamp(123, 0);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(versionstamp.version());

        // Create a commit txn log
        auto* commit_txn = operation_log.mutable_commit_txn();
        commit_txn->set_db_id(db_id);
        commit_txn->set_txn_id(100);

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    {
        // Put latest and versioned table stream offsets.
        TableStreamOffsetPB offset;
        offset.set_partition_id(partition_id);
        offset.set_state(TableStreamOffsetStatePB::TABLE_STREAM_OFFSET_CONSUMED);
        offset.set_offset_tso(12345);
        const std::string value = offset.SerializeAsString();
        const std::string latest_key = table_stream_offset_key(
                {instance_id, db_id, table_id, stream_db_id, stream_id, partition_id});
        const std::string versioned_key = versioned::table_stream_offset_key(
                {instance_id, db_id, table_id, stream_db_id, stream_id, partition_id});

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        txn->put(latest_key, value);
        versioned_put(txn.get(), versioned_key, value);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string latest_value;
        EXPECT_EQ(txn->get(latest_key, &latest_value), TxnErrorCode::TXN_OK);
        EXPECT_EQ(latest_value, value);
        EXPECT_EQ(
                count_range(txn_kv.get(), encode_versioned_key(versioned_key, Versionstamp::min()),
                            encode_versioned_key(versioned_key, Versionstamp::max())),
                1);
    }

    {
        // Mark instance deleted.
        InstanceInfoPB deleted_instance = instance;
        deleted_instance.set_status(InstanceInfoPB::DELETED);
        update_instance_info(txn_kv.get(), deleted_instance);
    }

    ASSERT_EQ(recycler.recycle_deleted_instance(), 0);
    ASSERT_EQ(recycler.recycle_deleted_instance(), 0);
    ASSERT_EQ(recycler.recycle_deleted_instance(), 0);

    // Verify all data keys are deleted, keeping the instance status and instance_update keys.
    ASSERT_EQ(count_range(txn_kv.get()), 2) << dump_range(txn_kv.get());
}

// Test OperationLogRecycleChecker class
TEST(OperationLogRecycleCheckerTest, InitAndBasicCheck) {
    auto txn_kv = std::make_shared<MemTxnKv>();
    txn_kv->update_commit_version(1000);
    ASSERT_EQ(txn_kv->init(), 0);

    std::string test_instance_id = "test_operation_log_recycle_checker";
    auto get_current_versionstamp = [&]() -> Versionstamp {
        std::unique_ptr<Transaction> txn;
        EXPECT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        int64_t read_version;
        EXPECT_EQ(txn->get_read_version(&read_version), TxnErrorCode::TXN_OK);
        return Versionstamp(read_version, 0);
    };

    auto insert_empty_value = [&]() {
        std::unique_ptr<Transaction> txn;
        EXPECT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        txn->put("dummy", "");
        EXPECT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    };

    insert_empty_value();

    Versionstamp old_version = get_current_versionstamp();

    insert_empty_value();

    // Test initialization without snapshots
    OperationLogReferenceInfo reference_info;
    {
        InstanceInfoPB instance_info;
        OperationLogRecycleChecker checker(test_instance_id, txn_kv.get(), instance_info);
        ASSERT_EQ(checker.init(), 0);

        // All logs should be recyclable when no snapshots exist
        ASSERT_TRUE(checker.can_recycle(old_version, 1, &reference_info)) << old_version.version();
    }

    {
        // Even a log has no min_timestamp, it can be recycled.
        InstanceInfoPB instance_info;
        OperationLogRecycleChecker checker(test_instance_id, txn_kv.get(), instance_info);
        ASSERT_EQ(checker.init(), 0);

        OperationLogPB op_log;
        ASSERT_TRUE(checker.can_recycle(old_version, op_log.min_timestamp(), &reference_info));
    }

    auto write_snapshot = [&]() {
        // Write snapshot
        SnapshotPB snapshot;
        std::string snapshot_key = versioned::snapshot_full_key(test_instance_id);
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), snapshot_key, snapshot.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
        insert_empty_value();
    };

    write_snapshot();
    Versionstamp version1 = get_current_versionstamp();

    write_snapshot();
    Versionstamp version2 = get_current_versionstamp();

    insert_empty_value();

    {
        InstanceInfoPB instance_info;
        OperationLogRecycleChecker checker(test_instance_id, txn_kv.get(), instance_info);
        ASSERT_EQ(checker.init(), 0);

        // case 1, old operation log can be recycled.
        ASSERT_TRUE(checker.can_recycle(old_version, 1, &reference_info));
        // case 2. snapshot exist in the log range, can not be recycled.
        ASSERT_FALSE(checker.can_recycle(version1, old_version.version(), &reference_info))
                << "version1: " << version1.version() << ", old_version: " << old_version.version();

        Versionstamp version3 = get_current_versionstamp();
        Versionstamp version4(version3.version(), 1);

        // case 3. large operation log can not be recycled.
        ASSERT_FALSE(checker.can_recycle(version4, version2.version(), &reference_info));

        // case 4: [min_version, operation log version)
        ASSERT_TRUE(checker.can_recycle(version1, version1.version(), &reference_info));
    }

    {
        Versionstamp snapshot_versionstamp = get_current_versionstamp();
        InstanceInfoPB instance_info;
        instance_info.set_source_snapshot_id(
                SnapshotManager::serialize_snapshot_id(snapshot_versionstamp));
        OperationLogRecycleChecker checker(test_instance_id, txn_kv.get(), instance_info);
        ASSERT_EQ(checker.init(), 0);

        Versionstamp version5 = get_current_versionstamp();

        ASSERT_FALSE(checker.can_recycle(version5, version2.version(), &reference_info))
                << "version5: " << version5.version() << ", version2: " << version2.version();
    }
}

TEST(RecycleOperationLogTest, RecycleCompactionLogKeepKeyBoundsWhenDisabled) {
    auto old_flag = config::enable_recycle_rowset_strip_key_bounds;
    config::enable_recycle_rowset_strip_key_bounds = false;
    DORIS_CLOUD_DEFER {
        config::enable_recycle_rowset_strip_key_bounds = old_flag;
    };

    auto meta_service = get_meta_service(false);
    std::string test_instance_id = "recycle_compaction_log_keep_bounds";
    auto* sp = SyncPoint::get_instance();
    DORIS_CLOUD_DEFER {
        SyncPoint::get_instance()->clear_all_call_backs();
    };
    sp->set_call_back("get_instance_id", [&](auto&& args) {
        auto* ret = try_any_cast_ret<std::string>(args);
        ret->first = test_instance_id;
        ret->second = true;
    });
    sp->set_call_back("check_lazy_txn_finished::bypass_check", [&](auto&& args) {
        auto* ret = doris::try_any_cast_ret<bool>(args);
        ret->first = true;
        ret->second = true;
    });
    sp->set_call_back("recycle_tablet::bypass_check", [&](auto&& args) {
        auto* ret = doris::try_any_cast_ret<bool>(args);
        ret->first = false;
        ret->second = true;
    });
    sp->enable_processing();

    constexpr int64_t table_id = 21001;
    constexpr int64_t index_id = 21002;
    constexpr int64_t partition_id = 21003;
    constexpr int64_t tablet_id = 21004;

    {
        InstanceInfoPB instance_info;
        instance_info.set_instance_id(test_instance_id);
        instance_info.set_multi_version_status(MULTI_VERSION_ENABLED);
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        txn->put(instance_key(test_instance_id), instance_info.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
        meta_service->resource_mgr()->refresh_instance(test_instance_id);
        ASSERT_TRUE(meta_service->resource_mgr()->is_version_write_enabled(test_instance_id));
    }

    create_tablet(meta_service.get(), table_id, index_id, partition_id, tablet_id);

    std::vector<doris::RowsetMetaCloudPB> input_rowsets;
    auto txn_kv = meta_service->txn_kv();
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < 2; ++i) {
            auto rowset = create_rowset(i + 500, tablet_id, partition_id, i + 2, 20 * (i + 1));
            input_rowsets.push_back(rowset);
            auto rowset_key = meta_rowset_key({test_instance_id, tablet_id, rowset.end_version()});
            txn->put(rowset_key, rowset.SerializeAsString());
            auto versioned_rowset_key = versioned::meta_rowset_load_key(
                    {test_instance_id, tablet_id, rowset.end_version()});
            versioned::document_put(txn.get(), versioned_rowset_key, std::move(rowset));
        }
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    constexpr int64_t txn_id = 60001;
    constexpr int64_t output_end_version = 3;
    auto output_rowset = create_rowset(700, tablet_id, partition_id, 2, 40);
    output_rowset.set_end_version(output_end_version);
    output_rowset.set_txn_id(txn_id);
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        auto tmp_rowset_key = meta_rowset_tmp_key({test_instance_id, txn_id, tablet_id});
        txn->put(tmp_rowset_key, output_rowset.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Seed tablet stats to avoid negative adjustments during compaction
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

        TabletStatsPB initial_stats;
        initial_stats.set_num_rows(60);    // 20 + 40
        initial_stats.set_data_size(6000); // data_disk_size sum
        initial_stats.set_num_rowsets(2);
        initial_stats.set_num_segments(2);
        initial_stats.set_index_size(600);    // index_disk_size sum
        initial_stats.set_segment_size(6000); // data size
        initial_stats.set_cumulative_point(1);

        auto stats_key =
                stats_tablet_key({test_instance_id, table_id, index_id, partition_id, tablet_id});
        txn->put(stats_key, initial_stats.SerializeAsString());

        auto versioned_load_stats_key =
                versioned::tablet_load_stats_key({test_instance_id, tablet_id});
        versioned_put(txn.get(), versioned_load_stats_key, initial_stats.SerializeAsString());

        auto tablet_compact_stats_key =
                versioned::tablet_compact_stats_key({test_instance_id, tablet_id});
        versioned_put(txn.get(), tablet_compact_stats_key, initial_stats.SerializeAsString());

        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    {
        brpc::Controller cntl;
        StartTabletJobRequest req;
        StartTabletJobResponse res;
        req.mutable_job()->mutable_idx()->set_tablet_id(tablet_id);
        auto compaction = req.mutable_job()->add_compaction();
        compaction->set_id("job_keep_bounds");
        compaction->set_initiator("be");
        compaction->set_type(TabletCompactionJobPB::CUMULATIVE);
        compaction->set_base_compaction_cnt(0);
        compaction->set_cumulative_compaction_cnt(0);
        compaction->add_input_versions(2);
        compaction->add_input_versions(3);
        long now = time(nullptr);
        compaction->set_expiration(now + 12);
        compaction->set_lease(now + 3);
        meta_service->start_tablet_job(&cntl, &req, &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    {
        brpc::Controller cntl;
        FinishTabletJobRequest req;
        FinishTabletJobResponse res;
        req.set_action(FinishTabletJobRequest::COMMIT);
        req.mutable_job()->mutable_idx()->set_table_id(table_id);
        req.mutable_job()->mutable_idx()->set_index_id(index_id);
        req.mutable_job()->mutable_idx()->set_partition_id(partition_id);
        req.mutable_job()->mutable_idx()->set_tablet_id(tablet_id);

        auto compaction = req.mutable_job()->add_compaction();
        compaction->set_id("job_keep_bounds");
        compaction->set_initiator("be");
        compaction->set_type(TabletCompactionJobPB::CUMULATIVE);
        compaction->set_base_compaction_cnt(0);
        compaction->set_cumulative_compaction_cnt(0);
        compaction->add_input_versions(2);
        compaction->add_input_versions(3);
        compaction->add_txn_id(txn_id);
        compaction->add_output_versions(output_end_version);
        compaction->add_output_rowset_ids(output_rowset.rowset_id_v2());
        compaction->set_output_cumulative_point(4);
        compaction->set_size_input_rowsets(6600);         // total_disk_size of inputs
        compaction->set_index_size_input_rowsets(600);    // index_disk_size of inputs
        compaction->set_segment_size_input_rowsets(6000); // data_disk_size of inputs
        compaction->set_num_input_rows(60);
        compaction->set_num_input_rowsets(2);
        compaction->set_num_input_segments(2);
        compaction->set_size_output_rowsets(4400);         // total_disk_size of output
        compaction->set_index_size_output_rowsets(400);    // index_disk_size of output
        compaction->set_segment_size_output_rowsets(4000); // data_disk_size of output
        compaction->set_num_output_rows(40);
        compaction->set_num_output_rowsets(1);
        compaction->set_num_output_segments(1);
        meta_service->finish_tablet_job(&cntl, &req, &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    Versionstamp log_version;
    OperationLogPB operation_log;
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string log_key = versioned::log_key({test_instance_id});
        ASSERT_EQ(read_operation_log(txn.get(), log_key, &log_version, &operation_log),
                  TxnErrorCode::TXN_OK);
    }
    ASSERT_TRUE(operation_log.has_compaction());
    const auto& compaction_log = operation_log.compaction();
    ASSERT_EQ(compaction_log.recycle_rowsets_size(), 2);
    for (const auto& recycle_rs : compaction_log.recycle_rowsets()) {
        ASSERT_GT(recycle_rs.rowset_meta().segments_key_bounds_size(), 0);
        ASSERT_TRUE(recycle_rs.rowset_meta().has_segments_key_bounds_truncated());
        ASSERT_FALSE(recycle_rs.rowset_meta().segments_key_bounds_truncated());
    }
}

// ============================================================================
// Tests for OplogRecycleStats and recycle_operation_logs metrics integration
// ============================================================================

TEST(RecycleOperationLogTest, OplogRecycleStatsAtomicOperations) {
    OplogRecycleStats stats;

    // Test initial values
    EXPECT_EQ(stats.total_num.load(), 0);
    EXPECT_EQ(stats.not_recycled_num.load(), 0);
    EXPECT_EQ(stats.failed_num.load(), 0);
    EXPECT_EQ(stats.recycled_commit_partition.load(), 0);
    EXPECT_EQ(stats.recycled_drop_partition.load(), 0);
    EXPECT_EQ(stats.recycled_commit_index.load(), 0);
    EXPECT_EQ(stats.recycled_drop_index.load(), 0);
    EXPECT_EQ(stats.recycled_update_tablet.load(), 0);
    EXPECT_EQ(stats.recycled_compaction.load(), 0);
    EXPECT_EQ(stats.recycled_schema_change.load(), 0);
    EXPECT_EQ(stats.recycled_commit_txn.load(), 0);

    // Test fetch_add
    stats.not_recycled_num.fetch_add(3, std::memory_order_relaxed);
    EXPECT_EQ(stats.not_recycled_num.load(), 3);

    stats.failed_num.fetch_add(1, std::memory_order_relaxed);
    EXPECT_EQ(stats.failed_num.load(), 1);

    stats.recycled_drop_partition.fetch_add(5, std::memory_order_relaxed);
    stats.recycled_drop_index.fetch_add(2, std::memory_order_relaxed);
    stats.recycled_compaction.fetch_add(10, std::memory_order_relaxed);
    stats.recycled_commit_txn.fetch_add(7, std::memory_order_relaxed);
    EXPECT_EQ(stats.recycled_drop_partition.load(), 5);
    EXPECT_EQ(stats.recycled_drop_index.load(), 2);
    EXPECT_EQ(stats.recycled_compaction.load(), 10);
    EXPECT_EQ(stats.recycled_commit_txn.load(), 7);

    // Test total_num
    stats.total_num.fetch_add(10, std::memory_order_relaxed);
    EXPECT_EQ(stats.total_num.load(), 10);
}

TEST(RecycleOperationLogTest, OplogRecycleStatsPerTypeCounters) {
    OplogRecycleStats stats;

    // Simulate multiple oplog recycling events
    stats.recycled_drop_partition.fetch_add(1, std::memory_order_relaxed);
    stats.recycled_drop_partition.fetch_add(1, std::memory_order_relaxed);
    stats.recycled_drop_index.fetch_add(1, std::memory_order_relaxed);
    stats.recycled_compaction.fetch_add(1, std::memory_order_relaxed);
    stats.recycled_compaction.fetch_add(1, std::memory_order_relaxed);
    stats.recycled_compaction.fetch_add(1, std::memory_order_relaxed);
    stats.recycled_commit_txn.fetch_add(1, std::memory_order_relaxed);

    EXPECT_EQ(stats.recycled_commit_partition.load(), 0);
    EXPECT_EQ(stats.recycled_drop_partition.load(), 2);
    EXPECT_EQ(stats.recycled_commit_index.load(), 0);
    EXPECT_EQ(stats.recycled_drop_index.load(), 1);
    EXPECT_EQ(stats.recycled_update_tablet.load(), 0);
    EXPECT_EQ(stats.recycled_compaction.load(), 3);
    EXPECT_EQ(stats.recycled_schema_change.load(), 0);
    EXPECT_EQ(stats.recycled_commit_txn.load(), 1);
}

// Test recycle_operation_logs with stats enabled: drop partition + drop index logs
// Verifies per-type recycled counts and bvar reporting
TEST(RecycleOperationLogTest, RecycleOperationLogsWithStatsEnabled) {
    auto old_flag = config::enable_recycler_stats_metrics;
    config::enable_recycler_stats_metrics = true;
    DORIS_CLOUD_DEFER {
        config::enable_recycler_stats_metrics = old_flag;
    };

    auto txn_kv = std::make_shared<MemTxnKv>();
    txn_kv->update_commit_version(1000);
    ASSERT_EQ(txn_kv->init(), 0);

    InstanceInfoPB instance;
    instance.set_instance_id(instance_id);
    instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
    auto* obj_info = instance.add_obj_info();
    obj_info->set_id("recycle_oplog_stats");
    obj_info->set_ak(config::test_s3_ak);
    obj_info->set_sk(config::test_s3_sk);
    obj_info->set_endpoint(config::test_s3_endpoint);
    obj_info->set_region(config::test_s3_region);
    obj_info->set_bucket(config::test_s3_bucket);
    obj_info->set_prefix("recycle_oplog_stats");
    update_instance_info(txn_kv.get(), instance);

    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    uint64_t db_id = 1;
    uint64_t table_id = 100;
    uint64_t index_id1 = 201;
    uint64_t index_id2 = 202;
    uint64_t partition_id1 = 301;
    uint64_t partition_id2 = 302;
    int64_t expiration = ::time(nullptr) + 3600;

    // Create table version so the recycler can remove it
    {
        std::string ver_key = versioned::table_version_key({instance_id, table_id});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), ver_key, "");
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Put a drop partition log with 2 partitions
    {
        std::string log_key = versioned::log_key(instance_id);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(100);
        auto* drop_partition = operation_log.mutable_drop_partition();
        drop_partition->set_db_id(db_id);
        drop_partition->set_table_id(table_id);
        drop_partition->add_index_ids(index_id1);
        drop_partition->add_partition_ids(partition_id1);
        drop_partition->add_partition_ids(partition_id2);
        drop_partition->set_expired_at_s(expiration);
        drop_partition->set_update_table_version(true);

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Put a drop index log with 2 indexes
    {
        std::string log_key = versioned::log_key(instance_id);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(101);
        auto* drop_index = operation_log.mutable_drop_index();
        drop_index->set_db_id(db_id);
        drop_index->set_table_id(table_id);
        drop_index->add_index_ids(index_id1);
        drop_index->add_index_ids(index_id2);

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Save cumulative baselines before recycling (mBvarIntAdder accumulates across tests)
    int64_t baseline_drop_partition =
            g_bvar_recycler_oplog_recycled_drop_partition_num.get({instance_id});
    int64_t baseline_drop_index = g_bvar_recycler_oplog_recycled_drop_index_num.get({instance_id});

    // Recycle the operation logs - stats should be collected
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // Verify recycle partition records were created
    {
        std::string recycle_key1 = recycle_partition_key({instance_id, partition_id1});
        std::string recycle_key2 = recycle_partition_key({instance_id, partition_id2});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string value;
        ASSERT_EQ(txn->get(recycle_key1, &value), TxnErrorCode::TXN_OK);
        ASSERT_EQ(txn->get(recycle_key2, &value), TxnErrorCode::TXN_OK);
    }

    // Verify recycle index records were created
    {
        std::string recycle_key1 = recycle_index_key({instance_id, index_id1});
        std::string recycle_key2 = recycle_index_key({instance_id, index_id2});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string value;
        ASSERT_EQ(txn->get(recycle_key1, &value), TxnErrorCode::TXN_OK);
        ASSERT_EQ(txn->get(recycle_key2, &value), TxnErrorCode::TXN_OK);
    }

    // Verify total oplog count: 2 logs (1 drop_partition + 1 drop_index)
    {
        int64_t total = g_bvar_recycler_oplog_last_round_total_num.get({instance_id});
        EXPECT_EQ(total, 2) << "Should have scanned 2 operation logs in total";
    }

    // Verify per-type last round recycled counts: 1 drop_partition log and 1 drop_index log
    {
        int last_round_drop_partition =
                g_bvar_recycler_oplog_last_round_recycled_drop_partition_num.get({instance_id});
        EXPECT_EQ(last_round_drop_partition, 1) << "Should have recycled 1 drop_partition log";
    }
    {
        int last_round_drop_index =
                g_bvar_recycler_oplog_last_round_recycled_drop_index_num.get({instance_id});
        EXPECT_EQ(last_round_drop_index, 1) << "Should have recycled 1 drop_index log";
    }

    // Verify per-type cumulative recycled counts (check delta from baseline)
    {
        int64_t recycled_drop_partition =
                g_bvar_recycler_oplog_recycled_drop_partition_num.get({instance_id});
        EXPECT_EQ(recycled_drop_partition - baseline_drop_partition, 1)
                << "Cumulative drop_partition delta should be 1";
    }
    {
        int64_t recycled_drop_index =
                g_bvar_recycler_oplog_recycled_drop_index_num.get({instance_id});
        EXPECT_EQ(recycled_drop_index - baseline_drop_index, 1)
                << "Cumulative drop_index delta should be 1";
    }
}

// Test recycle_operation_logs with snapshot protection: some logs skipped
TEST(RecycleOperationLogTest, RecycleOperationLogsSkippedBySnapshot) {
    auto old_flag = config::enable_recycler_stats_metrics;
    config::enable_recycler_stats_metrics = true;
    DORIS_CLOUD_DEFER {
        config::enable_recycler_stats_metrics = old_flag;
    };

    auto txn_kv = std::make_shared<MemTxnKv>();
    txn_kv->update_commit_version(1000);
    ASSERT_EQ(txn_kv->init(), 0);

    std::string test_instance_id = "test_oplog_not_recycled";

    InstanceInfoPB instance;
    instance.set_instance_id(test_instance_id);
    instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
    auto* obj_info = instance.add_obj_info();
    obj_info->set_id("recycle_snapshot_skip");
    obj_info->set_ak(config::test_s3_ak);
    obj_info->set_sk(config::test_s3_sk);
    obj_info->set_endpoint(config::test_s3_endpoint);
    obj_info->set_region(config::test_s3_region);
    obj_info->set_bucket(config::test_s3_bucket);
    obj_info->set_prefix("recycle_snapshot_skip");

    // Store instance info
    {
        std::string key = instance_key({test_instance_id});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        txn->put(key, instance.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    auto insert_empty_value = [&]() {
        std::unique_ptr<Transaction> txn;
        EXPECT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        txn->put("dummy_snapshot_test", "");
        EXPECT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    };

    // Create table version
    uint64_t table_id = 500;
    {
        std::string ver_key = versioned::table_version_key({test_instance_id, table_id});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), ver_key, "");
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Put an old operation log (will be recyclable)
    {
        std::string log_key = versioned::log_key(test_instance_id);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(100);

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    insert_empty_value();

    // Write a snapshot (this creates a protection boundary)
    {
        SnapshotPB snapshot;
        std::string snapshot_key = versioned::snapshot_full_key(test_instance_id);
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), snapshot_key, snapshot.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    insert_empty_value();

    // Put a new operation log after the snapshot (min_timestamp before snapshot)
    // This log should be skipped by snapshot protection
    {
        std::string log_key = versioned::log_key(test_instance_id);
        OperationLogPB operation_log;
        // min_timestamp is set to a value before the snapshot, so the snapshot
        // falls within [min_timestamp, log_version) and the log can't be recycled
        operation_log.set_min_timestamp(100);

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    insert_empty_value();

    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    // Recycle - some logs should be recycled, some not recycled due to snapshot
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // Verify not_recycled bvar is set
    {
        int64_t not_recycled =
                g_bvar_recycler_oplog_last_round_not_recycled_num.get({test_instance_id});
        EXPECT_GE(not_recycled, 1) << "At least one log should not be recycled";
    }
}

// Test recycle_operation_logs with stats disabled (default)
TEST(RecycleOperationLogTest, RecycleOperationLogsStatsDisabled) {
    auto old_flag = config::enable_recycler_stats_metrics;
    config::enable_recycler_stats_metrics = false;
    DORIS_CLOUD_DEFER {
        config::enable_recycler_stats_metrics = old_flag;
    };

    auto txn_kv = std::make_shared<MemTxnKv>();
    txn_kv->update_commit_version(1000);
    ASSERT_EQ(txn_kv->init(), 0);

    InstanceInfoPB instance;
    instance.set_instance_id(instance_id);
    instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
    auto* obj_info = instance.add_obj_info();
    obj_info->set_id("recycle_stats_disabled");
    obj_info->set_ak(config::test_s3_ak);
    obj_info->set_sk(config::test_s3_sk);
    obj_info->set_endpoint(config::test_s3_endpoint);
    obj_info->set_region(config::test_s3_region);
    obj_info->set_bucket(config::test_s3_bucket);
    obj_info->set_prefix("recycle_stats_disabled");
    update_instance_info(txn_kv.get(), instance);

    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    // Put an empty operation log
    {
        std::string log_key = versioned::log_key(instance_id);
        Versionstamp versionstamp(123, 0);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(versionstamp.version());

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, versionstamp, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Recycle should succeed without stats collection
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // Verify all logs were recycled
    remove_instance_info(txn_kv.get());
    ASSERT_TRUE(is_empty_range(txn_kv.get())) << dump_range(txn_kv.get());
}

// Test recycle_operation_logs compaction log with stats: recycled_compaction tracking
TEST(RecycleOperationLogTest, RecycleCompactionLogWithStats) {
    auto old_flag = config::enable_recycler_stats_metrics;
    config::enable_recycler_stats_metrics = true;
    DORIS_CLOUD_DEFER {
        config::enable_recycler_stats_metrics = old_flag;
    };

    auto txn_kv = std::make_shared<MemTxnKv>();
    txn_kv->update_commit_version(1000);
    ASSERT_EQ(txn_kv->init(), 0);

    std::string test_instance_id = "test_compaction_log_stats";

    InstanceInfoPB instance;
    instance.set_instance_id(test_instance_id);
    instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
    auto* obj_info = instance.add_obj_info();
    obj_info->set_id("compaction_stats");
    obj_info->set_ak(config::test_s3_ak);
    obj_info->set_sk(config::test_s3_sk);
    obj_info->set_endpoint(config::test_s3_endpoint);
    obj_info->set_region(config::test_s3_region);
    obj_info->set_bucket(config::test_s3_bucket);
    obj_info->set_prefix("compaction_stats");

    {
        std::string key = instance_key({test_instance_id});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        txn->put(key, instance.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));
    ASSERT_EQ(recycler.init(), 0);

    int64_t table_id = 600;
    int64_t tablet_id = 603;

    // Create table version
    {
        std::string ver_key = versioned::table_version_key({test_instance_id, table_id});
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned_put(txn.get(), ver_key, "");
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Put a compaction log with 3 recycle rowsets
    {
        std::string log_key = versioned::log_key(test_instance_id);
        OperationLogPB operation_log;
        operation_log.set_min_timestamp(100);
        auto* compaction = operation_log.mutable_compaction();
        compaction->set_tablet_id(tablet_id);

        for (int i = 0; i < 3; i++) {
            auto* recycle_rowset = compaction->add_recycle_rowsets();
            auto* rowset_meta = recycle_rowset->mutable_rowset_meta();
            rowset_meta->set_rowset_id(0);
            rowset_meta->set_tablet_id(tablet_id);
            rowset_meta->set_rowset_id_v2(fmt::format("compaction_rowset_{}", i));
            rowset_meta->set_start_version(i * 10);
            rowset_meta->set_end_version(i * 10 + 9);
        }

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        versioned::blob_put(txn.get(), log_key, operation_log);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Recycle the operation logs
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    // Verify recycle rowset records were created (3 rowsets from compaction log)
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < 3; i++) {
            std::string rowset_id = fmt::format("compaction_rowset_{}", i);
            std::string recycle_key = recycle_rowset_key({test_instance_id, tablet_id, rowset_id});
            std::string value;
            TxnErrorCode err = txn->get(recycle_key, &value);
            EXPECT_EQ(err, TxnErrorCode::TXN_OK)
                    << "Recycle rowset key not found for rowset " << rowset_id;
        }
    }

    // Verify last_round recycled_compaction bvar count (1 compaction log recycled)
    {
        int last_round_compaction =
                g_bvar_recycler_oplog_last_round_recycled_compaction_num.get({test_instance_id});
        EXPECT_EQ(last_round_compaction, 1) << "Last round should have recycled 1 compaction log";
    }
    // Verify cumulative recycled_compaction bvar count
    {
        int recycled_compaction =
                g_bvar_recycler_oplog_recycled_compaction_num.get({test_instance_id});
        EXPECT_EQ(recycled_compaction, 1) << "Cumulative compaction should be 1";
    }
}

} // namespace doris::cloud