// 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 <brpc/controller.h>
#include <gen_cpp/cloud.pb.h>
#include <gtest/gtest.h>

#include <cstdint>
#include <initializer_list>
#include <limits>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <vector>

#include "common/config.h"
#include "common/defer.h"
#include "common/lexical_util.h"
#include "cpp/sync_point.h"
#include "meta-service/meta_service.h"
#include "meta-store/keys.h"
#include "meta-store/txn_kv.h"
#include "meta-store/txn_kv_error.h"
#include "meta-store/versioned_value.h"
#include "recycler/recycler.h"
#include "resource-manager/resource_manager.h"
#include "snapshot/snapshot_manager.h"

namespace doris::cloud {

extern std::unique_ptr<MetaServiceProxy> get_meta_service(bool mock_resource_mgr);
extern void add_tablet(CreateTabletsRequest& request, int64_t table_id, int64_t index_id,
                       int64_t partition_id, int64_t tablet_id);
extern doris::RowsetMetaCloudPB create_rowset(int64_t txn_id, int64_t tablet_id, int partition_id,
                                              int64_t version, int num_rows);
extern TxnErrorCode read_operation_log(Transaction* txn, std::string_view log_key,
                                       Versionstamp* log_version, OperationLogPB* operation_log);

class MetaServiceTableStreamTest : public ::testing::Test {
protected:
    void SetUp() override {
        service_ = get_meta_service(false);
        instance_id_ = "table_stream_read_state_instance";
        cloud_unique_id_ = "1:" + instance_id_ + ":test";
        identity_.set_base_db_id(1001);
        identity_.set_base_table_id(1002);
        identity_.set_stream_db_id(1003);
        identity_.set_stream_id(1004);
    }

    void set_multi_version_status(MultiVersionStatus mode) {
        multi_version_status_ = mode;
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        InstanceInfoPB instance;
        instance.set_instance_id(instance_id_);
        instance.set_multi_version_status(mode);
        txn->put(instance_key({instance_id_}), instance.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
        auto [code, message] = service_->resource_mgr()->refresh_instance(instance_id_);
        ASSERT_EQ(code, MetaServiceCode::OK) << message;
    }

    void put_partition_mapping(Transaction* txn, int64_t partition_id) {
        PartitionIndexPB partition;
        partition.set_db_id(identity_.base_db_id());
        partition.set_table_id(identity_.base_table_id());
        txn->put(versioned::partition_index_key({instance_id_, partition_id}),
                 partition.SerializeAsString());
    }

    void put_latest_partition_state(Transaction* txn, int64_t partition_id, int64_t visible_version,
                                    int64_t commit_tso, std::optional<int64_t> offset_tso) {
        VersionPB version;
        version.set_version(visible_version);
        version.set_commit_tso(commit_tso);
        txn->put(partition_version_key({instance_id_, identity_.base_db_id(),
                                        identity_.base_table_id(), partition_id}),
                 version.SerializeAsString());
        if (offset_tso.has_value()) {
            TableStreamOffsetPB offset;
            offset.set_partition_id(partition_id);
            offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
            offset.set_offset_tso(*offset_tso);
            offset.set_last_consumption_time_ms(1234);
            txn->put(table_stream_offset_key({instance_id_, identity_.base_db_id(),
                                              identity_.base_table_id(), identity_.stream_db_id(),
                                              identity_.stream_id(), partition_id}),
                     offset.SerializeAsString());
        }
    }

    void put_versioned_partition_state(Transaction* txn, int64_t partition_id,
                                       int64_t visible_version, int64_t commit_tso,
                                       std::optional<int64_t> offset_tso) {
        put_partition_mapping(txn, partition_id);
        versioned_put(txn, versioned::meta_partition_key({instance_id_, partition_id}),
                      Versionstamp(41, 0), "");

        VersionPB version;
        version.set_version(visible_version);
        version.set_commit_tso(commit_tso);
        versioned_put(txn, versioned::partition_version_key({instance_id_, partition_id}),
                      Versionstamp(42, 0), version.SerializeAsString());
        if (offset_tso.has_value()) {
            TableStreamOffsetPB offset;
            offset.set_partition_id(partition_id);
            offset.set_state(TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING);
            offset.set_offset_tso(*offset_tso);
            versioned_put(txn,
                          versioned::table_stream_offset_key(
                                  {instance_id_, identity_.base_db_id(), identity_.base_table_id(),
                                   identity_.stream_db_id(), identity_.stream_id(), partition_id}),
                          Versionstamp(43, 0), offset.SerializeAsString());
        }
    }

    GetTableStreamOffsetResponse get_read_state(const std::vector<int64_t>& partitions) {
        GetTableStreamOffsetRequest request;
        request.set_cloud_unique_id(cloud_unique_id_);
        auto* binding = request.add_bindings();
        binding->mutable_identity()->CopyFrom(identity_);
        for (int64_t partition_id : partitions) {
            binding->add_partition_ids(partition_id);
        }
        GetTableStreamOffsetResponse response;
        brpc::Controller controller;
        service_->get_table_stream_offset(&controller, &request, &response, nullptr);
        return response;
    }

    GetTableStreamOffsetResponse get_read_state(std::initializer_list<int64_t> partitions) {
        return get_read_state(std::vector<int64_t>(partitions));
    }

    int64_t begin_target_transaction(const std::string& label) {
        return begin_transaction(cloud_unique_id_, label);
    }

    int64_t begin_transaction(const std::string& cloud_unique_id, const std::string& label) {
        BeginTxnRequest request;
        request.set_cloud_unique_id(cloud_unique_id);
        TxnInfoPB* txn_info = request.mutable_txn_info();
        txn_info->set_db_id(3001);
        txn_info->set_label(label);
        txn_info->add_table_ids(3002);
        txn_info->set_timeout_ms(36000);
        BeginTxnResponse response;
        brpc::Controller controller;
        service_->begin_txn(&controller, &request, &response, nullptr);
        EXPECT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
        return response.txn_id();
    }

    void register_instance(const std::string& instance_id, MultiVersionStatus mode) {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        InstanceInfoPB instance;
        instance.set_instance_id(instance_id);
        instance.set_multi_version_status(mode);
        txn->put(instance_key({instance_id}), instance.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
        auto [code, message] = service_->resource_mgr()->refresh_instance(instance_id);
        ASSERT_EQ(code, MetaServiceCode::OK) << message;
    }

    void mark_transaction_visible(const std::string& instance_id, int64_t txn_id,
                                  Transaction* txn) {
        std::string txn_info_value;
        ASSERT_EQ(txn->get(txn_info_key({instance_id, 3001, txn_id}), &txn_info_value),
                  TxnErrorCode::TXN_OK);
        TxnInfoPB txn_info;
        ASSERT_TRUE(txn_info.ParseFromString(txn_info_value));
        txn_info.set_status(TXN_STATUS_VISIBLE);
        txn->put(txn_info_key({instance_id, 3001, txn_id}), txn_info.SerializeAsString());
    }

    CommitTxnRequest make_consume_request(int64_t txn_id, int64_t partition_id,
                                          TableStreamOffsetStatePB expected_state,
                                          int64_t expected_tso, int64_t next_tso) {
        CommitTxnRequest request;
        request.set_cloud_unique_id(cloud_unique_id_);
        request.set_db_id(3001);
        request.set_txn_id(txn_id);
        TableStreamUpdatePB* stream_update = request.add_table_stream_updates();
        stream_update->mutable_identity()->CopyFrom(identity_);
        TableStreamPartitionUpdatePB* partition_update = stream_update->add_partition_updates();
        partition_update->set_partition_id(partition_id);
        partition_update->set_expected_state(expected_state);
        if (expected_state != TABLE_STREAM_OFFSET_UNKNOWN) {
            partition_update->set_expected_offset_tso(expected_tso);
        }
        partition_update->set_next_offset_tso(next_tso);
        return request;
    }

    CommitTxnResponse commit_transaction(const CommitTxnRequest& request) {
        CommitTxnResponse response;
        brpc::Controller controller;
        service_->commit_txn(&controller, &request, &response, nullptr);
        return response;
    }

    CommitTxnResponse consume_partition(int64_t txn_id, int64_t partition_id,
                                        TableStreamOffsetStatePB expected_state,
                                        int64_t expected_tso, int64_t next_tso) {
        return commit_transaction(
                make_consume_request(txn_id, partition_id, expected_state, expected_tso, next_tso));
    }

    void create_target_tablet(int64_t db_id, int64_t table_id, int64_t index_id,
                              int64_t partition_id, int64_t tablet_id) {
        CreateTabletsRequest request;
        request.set_cloud_unique_id(cloud_unique_id_);
        request.set_db_id(db_id);
        add_tablet(request, table_id, index_id, partition_id, tablet_id);
        CreateTabletsResponse response;
        brpc::Controller controller;
        service_->create_tablets(&controller, &request, &response, nullptr);
        ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();

        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        VersionPB version;
        version.set_version(1);
        txn->put(partition_version_key({instance_id_, db_id, table_id, partition_id}),
                 version.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    void stage_target_rowset(int64_t txn_id, int64_t partition_id, int64_t tablet_id) {
        const doris::RowsetMetaCloudPB rowset =
                create_rowset(txn_id, tablet_id, partition_id, -1, 100);
        auto call = [&](bool prepare) {
            CreateRowsetRequest request;
            request.set_cloud_unique_id(cloud_unique_id_);
            request.mutable_rowset_meta()->CopyFrom(rowset);
            CreateRowsetResponse response;
            brpc::Controller controller;
            if (prepare) {
                service_->prepare_rowset(&controller, &request, &response, nullptr);
            } else {
                service_->commit_rowset(&controller, &request, &response, nullptr);
            }
            ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
        };
        ASSERT_NO_FATAL_FAILURE(call(true));
        ASSERT_NO_FATAL_FAILURE(call(false));
    }

    TableStreamOffsetPB get_latest_offset(const TableStreamIdentityPB& identity,
                                          int64_t partition_id) {
        std::unique_ptr<Transaction> txn;
        EXPECT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string value;
        EXPECT_EQ(txn->get(table_stream_offset_key(
                                   {instance_id_, identity.base_db_id(), identity.base_table_id(),
                                    identity.stream_db_id(), identity.stream_id(), partition_id}),
                           &value),
                  TxnErrorCode::TXN_OK);
        TableStreamOffsetPB offset;
        EXPECT_TRUE(offset.ParseFromString(value));
        return offset;
    }

    TableStreamOffsetPB get_latest_offset(int64_t partition_id) {
        return get_latest_offset(identity_, partition_id);
    }

    void set_clone_source(const std::string& source_instance_id, Versionstamp snapshot_version) {
        multi_version_status_ = MULTI_VERSION_READ_WRITE;
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        InstanceInfoPB instance;
        instance.set_instance_id(instance_id_);
        instance.set_multi_version_status(MULTI_VERSION_READ_WRITE);
        instance.set_source_instance_id(source_instance_id);
        instance.set_source_snapshot_id(SnapshotManager::serialize_snapshot_id(snapshot_version));
        txn->put(instance_key({instance_id_}), instance.SerializeAsString());
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
        auto [code, message] = service_->resource_mgr()->refresh_instance(instance_id_);
        ASSERT_EQ(code, MetaServiceCode::OK) << message;
    }

    Versionstamp put_auto_versioned_offset(const std::string& target_instance_id,
                                           const TableStreamIdentityPB& identity,
                                           int64_t partition_id, int64_t offset_tso,
                                           bool write_latest = false) {
        std::unique_ptr<Transaction> txn;
        EXPECT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        TableStreamOffsetPB offset;
        offset.set_partition_id(partition_id);
        offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
        offset.set_offset_tso(offset_tso);
        const TableStreamOffsetKeyInfo key_info {target_instance_id,       identity.base_db_id(),
                                                 identity.base_table_id(), identity.stream_db_id(),
                                                 identity.stream_id(),     partition_id};
        const std::string value = offset.SerializeAsString();
        if (write_latest) {
            txn->put(table_stream_offset_key(key_info), value);
        }
        txn->enable_get_versionstamp();
        versioned_put(txn.get(), versioned::table_stream_offset_key(key_info), value);
        EXPECT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
        Versionstamp version;
        EXPECT_EQ(txn->get_versionstamp(&version), TxnErrorCode::TXN_OK);
        return version;
    }

    std::unique_ptr<MetaServiceProxy> service_;
    std::string instance_id_;
    std::string cloud_unique_id_;
    TableStreamIdentityPB identity_;
    MultiVersionStatus multi_version_status_ = MULTI_VERSION_DISABLED;
};

TEST_F(MetaServiceTableStreamTest, ReadLatestAndUnknownOffsets) {
    set_multi_version_status(MULTI_VERSION_DISABLED);
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
    put_latest_partition_state(txn.get(), 2002, 9, 140, std::nullopt);
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    GetTableStreamOffsetResponse response = get_read_state({2001, 2002});
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    ASSERT_EQ(response.bindings_size(), 1);
    ASSERT_EQ(response.bindings(0).partition_states_size(), 2);

    const auto& consumed = response.bindings(0).partition_states(0);
    EXPECT_EQ(consumed.partition_id(), 2001);
    EXPECT_EQ(consumed.offset_state(), TABLE_STREAM_OFFSET_CONSUMED);
    EXPECT_EQ(consumed.offset_tso(), 100);
    EXPECT_EQ(consumed.end_tso(), 130);
    EXPECT_EQ(consumed.visible_version(), 8);
    EXPECT_EQ(consumed.last_consumption_time_ms(), 1234);

    const auto& unknown = response.bindings(0).partition_states(1);
    EXPECT_EQ(unknown.partition_id(), 2002);
    EXPECT_EQ(unknown.offset_state(), TABLE_STREAM_OFFSET_UNKNOWN);
    EXPECT_FALSE(unknown.has_offset_tso());
    EXPECT_EQ(unknown.end_tso(), 140);
    EXPECT_EQ(unknown.visible_version(), 9);
}

TEST_F(MetaServiceTableStreamTest, ReadVersionedState) {
    set_multi_version_status(MULTI_VERSION_READ_WRITE);
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    put_versioned_partition_state(txn.get(), 2001, 18, 230, 200);
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    GetTableStreamOffsetResponse response = get_read_state({2001});
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    ASSERT_EQ(response.bindings_size(), 1);
    ASSERT_EQ(response.bindings(0).partition_states_size(), 1);
    const auto& state = response.bindings(0).partition_states(0);
    EXPECT_EQ(state.offset_state(), TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING);
    EXPECT_EQ(state.offset_tso(), 200);
    EXPECT_EQ(state.end_tso(), 230);
    EXPECT_EQ(state.visible_version(), 18);
}

TEST_F(MetaServiceTableStreamTest, ReadVersionedStateFromCloneChainInBatch) {
    const std::string source_instance_id = "table_stream_batch_read_source";
    set_clone_source(source_instance_id, Versionstamp(100, 0));
    const std::vector<int64_t> partition_ids = {2001, 2002, 2003};

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    for (size_t i = 0; i < partition_ids.size(); ++i) {
        int64_t partition_id = partition_ids[i];
        PartitionIndexPB partition;
        partition.set_db_id(identity_.base_db_id());
        partition.set_table_id(identity_.base_table_id());
        txn->put(versioned::partition_index_key({source_instance_id, partition_id}),
                 partition.SerializeAsString());
        versioned_put(txn.get(), versioned::meta_partition_key({source_instance_id, partition_id}),
                      Versionstamp(51 + i, 0), "");

        VersionPB version;
        version.set_version(10 + i);
        version.set_commit_tso(1000 + i);
        versioned_put(txn.get(),
                      versioned::partition_version_key({source_instance_id, partition_id}),
                      Versionstamp(60 + i, 0), version.SerializeAsString());
    }

    auto put_offset = [&](const std::string& target_instance_id, int64_t partition_id,
                          Versionstamp versionstamp, int64_t offset_tso) {
        TableStreamOffsetPB offset;
        offset.set_partition_id(partition_id);
        offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
        offset.set_offset_tso(offset_tso);
        versioned_put(
                txn.get(),
                versioned::table_stream_offset_key(
                        {target_instance_id, identity_.base_db_id(), identity_.base_table_id(),
                         identity_.stream_db_id(), identity_.stream_id(), partition_id}),
                versionstamp, offset.SerializeAsString());
    };
    put_offset(source_instance_id, 2001, Versionstamp(70, 0), 701);
    put_offset(source_instance_id, 2002, Versionstamp(71, 0), 702);
    put_offset(instance_id_, 2002, Versionstamp(200, 0), 802);
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    GetTableStreamOffsetResponse response = get_read_state(partition_ids);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    ASSERT_EQ(response.bindings(0).partition_states_size(), partition_ids.size());
    EXPECT_EQ(response.bindings(0).partition_states(0).offset_tso(), 701);
    EXPECT_EQ(response.bindings(0).partition_states(1).offset_tso(), 802);
    EXPECT_EQ(response.bindings(0).partition_states(2).offset_state(), TABLE_STREAM_OFFSET_UNKNOWN);
}

TEST_F(MetaServiceTableStreamTest, ReadLargePartitionBatch) {
    set_multi_version_status(MULTI_VERSION_DISABLED);
    constexpr int kPartitionCount = 1201;
    std::vector<int64_t> partition_ids;
    partition_ids.reserve(kPartitionCount);

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    for (int i = 0; i < kPartitionCount; ++i) {
        int64_t partition_id = 10000 + i;
        partition_ids.push_back(partition_id);
        put_latest_partition_state(txn.get(), partition_id, i + 1, 100000 + i, 90000 + i);
    }
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    GetTableStreamOffsetResponse response = get_read_state(partition_ids);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    ASSERT_EQ(response.bindings(0).partition_states_size(), kPartitionCount);
    EXPECT_EQ(response.bindings(0).partition_states(0).partition_id(), partition_ids.front());
    EXPECT_EQ(response.bindings(0).partition_states(0).offset_tso(), 90000);
    EXPECT_EQ(response.bindings(0).partition_states(kPartitionCount - 1).partition_id(),
              partition_ids.back());
    EXPECT_EQ(response.bindings(0).partition_states(kPartitionCount - 1).offset_tso(),
              90000 + kPartitionCount - 1);
}

TEST_F(MetaServiceTableStreamTest, ReadMultipleBindingsInBatch) {
    set_multi_version_status(MULTI_VERSION_READ_WRITE);
    constexpr int kBindingCount = 101;
    constexpr int64_t kPartitionId = 2001;

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    PartitionIndexPB partition;
    partition.set_db_id(identity_.base_db_id());
    partition.set_table_id(identity_.base_table_id());
    txn->put(versioned::partition_index_key({instance_id_, kPartitionId}),
             partition.SerializeAsString());
    versioned_put(txn.get(), versioned::meta_partition_key({instance_id_, kPartitionId}),
                  Versionstamp(20, 0), "");
    VersionPB version;
    version.set_version(8);
    version.set_commit_tso(130);
    versioned_put(txn.get(), versioned::partition_version_key({instance_id_, kPartitionId}),
                  Versionstamp(21, 0), version.SerializeAsString());

    GetTableStreamOffsetRequest request;
    request.set_cloud_unique_id(cloud_unique_id_);
    for (int i = 0; i < kBindingCount; ++i) {
        int64_t stream_id = identity_.stream_id() + i;
        TableStreamPartitionSetPB* binding = request.add_bindings();
        binding->mutable_identity()->CopyFrom(identity_);
        binding->mutable_identity()->set_stream_id(stream_id);
        binding->add_partition_ids(kPartitionId);
    }
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    GetTableStreamOffsetResponse response;
    brpc::Controller controller;
    service_->get_table_stream_offset(&controller, &request, &response, nullptr);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    ASSERT_EQ(response.bindings_size(), kBindingCount);
    for (const TableStreamReadBindingResultPB& binding : response.bindings()) {
        ASSERT_EQ(binding.partition_states_size(), 1);
        EXPECT_EQ(binding.partition_states(0).offset_state(), TABLE_STREAM_OFFSET_UNKNOWN);
    }
}

TEST_F(MetaServiceTableStreamTest, ReadWriteOnlyState) {
    set_multi_version_status(MULTI_VERSION_WRITE_ONLY);
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    put_latest_partition_state(txn.get(), 2001, 28, 330, 300);
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    GetTableStreamOffsetResponse response = get_read_state({2001});
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    ASSERT_EQ(response.bindings_size(), 1);
    ASSERT_EQ(response.bindings(0).partition_states_size(), 1);
    const auto& state = response.bindings(0).partition_states(0);
    EXPECT_EQ(state.offset_state(), TABLE_STREAM_OFFSET_CONSUMED);
    EXPECT_EQ(state.offset_tso(), 300);
    EXPECT_EQ(state.end_tso(), 330);
    EXPECT_EQ(state.visible_version(), 28);
}

TEST_F(MetaServiceTableStreamTest, WaitsForPendingSourceTxnAndRereadsSnapshot) {
    set_multi_version_status(MULTI_VERSION_DISABLED);
    const int64_t pending_txn_id = begin_target_transaction("pending-source-version");

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    std::string txn_info_value;
    ASSERT_EQ(txn->get(txn_info_key({instance_id_, 3001, pending_txn_id}), &txn_info_value),
              TxnErrorCode::TXN_OK);
    TxnInfoPB txn_info;
    ASSERT_TRUE(txn_info.ParseFromString(txn_info_value));
    txn_info.set_status(TXN_STATUS_VISIBLE);
    txn->put(txn_info_key({instance_id_, 3001, pending_txn_id}), txn_info.SerializeAsString());

    VersionPB pending_version;
    pending_version.set_commit_tso(130);
    pending_version.add_pending_txn_ids(pending_txn_id);
    txn->put(partition_version_key(
                     {instance_id_, identity_.base_db_id(), identity_.base_table_id(), 2001}),
             pending_version.SerializeAsString());
    TableStreamOffsetPB offset;
    offset.set_partition_id(2001);
    offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
    offset.set_offset_tso(100);
    txn->put(table_stream_offset_key({instance_id_, identity_.base_db_id(),
                                      identity_.base_table_id(), identity_.stream_db_id(),
                                      identity_.stream_id(), 2001}),
             offset.SerializeAsString());
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    bool after_wait_called = false;
    auto* sync_point = SyncPoint::get_instance();
    DORIS_CLOUD_DEFER {
        sync_point->disable_processing();
        sync_point->clear_all_call_backs();
    };
    sync_point->set_call_back("get_table_stream_offset::after_wait_for_pending_txns", [&](auto&&) {
        std::unique_ptr<Transaction> update_txn;
        ASSERT_EQ(service_->txn_kv()->create_txn(&update_txn), TxnErrorCode::TXN_OK);
        put_latest_partition_state(update_txn.get(), 2001, 2, 130, 120);
        ASSERT_EQ(update_txn->commit(), TxnErrorCode::TXN_OK);
        after_wait_called = true;
    });
    sync_point->enable_processing();

    GetTableStreamOffsetResponse response = get_read_state({2001});
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    ASSERT_TRUE(after_wait_called);
    ASSERT_EQ(response.bindings_size(), 1);
    ASSERT_EQ(response.bindings(0).partition_states_size(), 1);
    const TableStreamPartitionReadStatePB& state = response.bindings(0).partition_states(0);
    EXPECT_EQ(state.visible_version(), 2);
    EXPECT_EQ(state.end_tso(), 130);
    EXPECT_EQ(state.offset_tso(), 120);
}

TEST_F(MetaServiceTableStreamTest, WaitsForPendingVersionFromCloneSourceInstance) {
    const std::string source_instance_id = "table_stream_pending_version_source";
    register_instance(source_instance_id, MULTI_VERSION_READ_WRITE);
    set_clone_source(source_instance_id, Versionstamp(100, 0));
    const int64_t pending_txn_id = begin_transaction(fmt::format("1:{}:test", source_instance_id),
                                                     "pending-clone-source-version");

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    mark_transaction_visible(source_instance_id, pending_txn_id, txn.get());

    PartitionIndexPB partition;
    partition.set_db_id(identity_.base_db_id());
    partition.set_table_id(identity_.base_table_id());
    txn->put(versioned::partition_index_key({source_instance_id, 2001}),
             partition.SerializeAsString());
    versioned_put(txn.get(), versioned::meta_partition_key({source_instance_id, 2001}),
                  Versionstamp(50, 0), "");

    VersionPB pending_version;
    pending_version.set_commit_tso(130);
    pending_version.add_pending_txn_ids(pending_txn_id);
    versioned_put(txn.get(), versioned::partition_version_key({source_instance_id, 2001}),
                  Versionstamp(60, 0), pending_version.SerializeAsString());

    TableStreamOffsetPB offset;
    offset.set_partition_id(2001);
    offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
    offset.set_offset_tso(100);
    versioned_put(txn.get(),
                  versioned::table_stream_offset_key(
                          {source_instance_id, identity_.base_db_id(), identity_.base_table_id(),
                           identity_.stream_db_id(), identity_.stream_id(), 2001}),
                  Versionstamp(70, 0), offset.SerializeAsString());
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    bool after_wait_called = false;
    auto* sync_point = SyncPoint::get_instance();
    DORIS_CLOUD_DEFER {
        sync_point->disable_processing();
        sync_point->clear_all_call_backs();
    };
    sync_point->set_call_back("get_table_stream_offset::after_wait_for_pending_txns", [&](auto&&) {
        std::unique_ptr<Transaction> update_txn;
        ASSERT_EQ(service_->txn_kv()->create_txn(&update_txn), TxnErrorCode::TXN_OK);
        VersionPB committed_version;
        committed_version.set_version(2);
        committed_version.set_commit_tso(130);
        versioned_put(update_txn.get(),
                      versioned::partition_version_key({source_instance_id, 2001}),
                      Versionstamp(60, 0), committed_version.SerializeAsString());
        offset.set_offset_tso(120);
        versioned_put(
                update_txn.get(),
                versioned::table_stream_offset_key(
                        {source_instance_id, identity_.base_db_id(), identity_.base_table_id(),
                         identity_.stream_db_id(), identity_.stream_id(), 2001}),
                Versionstamp(70, 0), offset.SerializeAsString());
        ASSERT_EQ(update_txn->commit(), TxnErrorCode::TXN_OK);
        after_wait_called = true;
    });
    sync_point->enable_processing();

    GetTableStreamOffsetResponse response = get_read_state({2001});
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    ASSERT_TRUE(after_wait_called);
    ASSERT_EQ(response.bindings_size(), 1);
    ASSERT_EQ(response.bindings(0).partition_states_size(), 1);
    const TableStreamPartitionReadStatePB& state = response.bindings(0).partition_states(0);
    EXPECT_EQ(state.visible_version(), 2);
    EXPECT_EQ(state.end_tso(), 130);
    EXPECT_EQ(state.offset_tso(), 120);
}

TEST_F(MetaServiceTableStreamTest, SingleGetVersionWaitsUsingCloneSourceInstance) {
    const std::string source_instance_id = "table_stream_single_pending_version_source";
    register_instance(source_instance_id, MULTI_VERSION_READ_WRITE);
    set_clone_source(source_instance_id, Versionstamp(100, 0));
    const int64_t pending_txn_id = begin_transaction(fmt::format("1:{}:test", source_instance_id),
                                                     "pending-clone-single-version");

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    mark_transaction_visible(source_instance_id, pending_txn_id, txn.get());
    VersionPB pending_version;
    pending_version.set_commit_tso(130);
    pending_version.add_pending_txn_ids(pending_txn_id);
    versioned_put(txn.get(), versioned::partition_version_key({source_instance_id, 2001}),
                  Versionstamp(60, 0), pending_version.SerializeAsString());
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    GetVersionRequest request;
    request.set_cloud_unique_id(cloud_unique_id_);
    request.set_db_id(identity_.base_db_id());
    request.set_table_id(identity_.base_table_id());
    request.set_partition_id(2001);
    request.set_wait_for_pending_txn(true);
    GetVersionResponse response;
    brpc::Controller controller;
    service_->get_version(&controller, &request, &response, nullptr);

    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    EXPECT_EQ(response.version(), 2);
    ASSERT_EQ(response.commit_tsos_size(), 1);
    EXPECT_EQ(response.commit_tsos(0), 130);
}

TEST_F(MetaServiceTableStreamTest, BatchGetVersionWaitsUsingCloneSourceInstance) {
    const std::string source_instance_id = "table_stream_batch_pending_version_source";
    register_instance(source_instance_id, MULTI_VERSION_READ_WRITE);
    set_clone_source(source_instance_id, Versionstamp(100, 0));
    const int64_t pending_txn_id = begin_transaction(fmt::format("1:{}:test", source_instance_id),
                                                     "pending-clone-batch-version");

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    mark_transaction_visible(source_instance_id, pending_txn_id, txn.get());
    PartitionIndexPB partition;
    partition.set_db_id(identity_.base_db_id());
    partition.set_table_id(identity_.base_table_id());
    txn->put(versioned::partition_index_key({source_instance_id, 2001}),
             partition.SerializeAsString());
    VersionPB pending_version;
    pending_version.set_commit_tso(130);
    pending_version.add_pending_txn_ids(pending_txn_id);
    versioned_put(txn.get(), versioned::partition_version_key({source_instance_id, 2001}),
                  Versionstamp(60, 0), pending_version.SerializeAsString());
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    GetVersionRequest request;
    request.set_cloud_unique_id(cloud_unique_id_);
    request.set_batch_mode(true);
    request.set_wait_for_pending_txn(true);
    request.add_db_ids(identity_.base_db_id());
    request.add_table_ids(identity_.base_table_id());
    request.add_partition_ids(2001);
    GetVersionResponse response;
    brpc::Controller controller;
    service_->get_version(&controller, &request, &response, nullptr);

    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    ASSERT_EQ(response.versions_size(), 1);
    ASSERT_EQ(response.commit_tsos_size(), 1);
    EXPECT_EQ(response.versions(0), 2);
    EXPECT_EQ(response.commit_tsos(0), 130);
}

TEST_F(MetaServiceTableStreamTest, RejectDuplicateBindingsAndPartitions) {
    GetTableStreamOffsetRequest request;
    request.set_cloud_unique_id(cloud_unique_id_);
    auto* binding = request.add_bindings();
    binding->mutable_identity()->CopyFrom(identity_);
    binding->add_partition_ids(2001);
    binding->add_partition_ids(2001);
    GetTableStreamOffsetResponse response;
    brpc::Controller controller;
    service_->get_table_stream_offset(&controller, &request, &response, nullptr);
    EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);

    request.mutable_bindings()->Clear();
    for (int i = 0; i < 2; ++i) {
        binding = request.add_bindings();
        binding->mutable_identity()->CopyFrom(identity_);
        binding->add_partition_ids(2001 + i);
    }
    response.Clear();
    brpc::Controller second_controller;
    service_->get_table_stream_offset(&second_controller, &request, &response, nullptr);
    EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
}

TEST_F(MetaServiceTableStreamTest, RejectEnabledModeAndRecyclingStream) {
    set_multi_version_status(MULTI_VERSION_DISABLED);
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    InstanceInfoPB instance;
    instance.set_instance_id(instance_id_);
    instance.set_multi_version_status(MULTI_VERSION_ENABLED);
    txn->put(instance_key({instance_id_}), instance.SerializeAsString());
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    GetTableStreamOffsetResponse response = get_read_state({2001});
    EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);

    set_multi_version_status(MULTI_VERSION_DISABLED);
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
    RecycleIndexPB recycle_index;
    recycle_index.set_table_id(identity_.base_table_id());
    recycle_index.set_state(RecycleIndexPB::RECYCLING);
    recycle_index.set_object_type(TABLE_STREAM);
    txn->put(recycle_index_key({instance_id_, identity_.stream_id()}),
             recycle_index.SerializeAsString());
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    response = get_read_state({2001});
    EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
}

TEST_F(MetaServiceTableStreamTest, CommitUpdatesLatestOffsetAndIsIdempotent) {
    set_multi_version_status(MULTI_VERSION_DISABLED);
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
    put_latest_partition_state(txn.get(), 2002, 9, 140, std::nullopt);
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    int64_t txn_id = begin_target_transaction("consume-existing-offset");
    CommitTxnResponse response =
            consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    TableStreamOffsetPB offset = get_latest_offset(2001);
    EXPECT_EQ(offset.state(), TABLE_STREAM_OFFSET_CONSUMED);
    EXPECT_EQ(offset.offset_tso(), 120);
    EXPECT_GT(offset.last_consumption_time_ms(), 0);

    response = consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    EXPECT_EQ(get_latest_offset(2001).offset_tso(), 120);

    txn_id = begin_target_transaction("consume-unknown-offset");
    response = consume_partition(txn_id, 2002, TABLE_STREAM_OFFSET_UNKNOWN, 0, 140);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    EXPECT_EQ(get_latest_offset(2002).offset_tso(), 140);
}

TEST_F(MetaServiceTableStreamTest, CommitMultipleStreamsAtomically) {
    set_multi_version_status(MULTI_VERSION_DISABLED);
    TableStreamIdentityPB second_identity = identity_;
    second_identity.set_stream_db_id(1005);
    second_identity.set_stream_id(1006);

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
    TableStreamOffsetPB second_offset;
    second_offset.set_partition_id(2001);
    second_offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
    second_offset.set_offset_tso(105);
    second_offset.set_last_consumption_time_ms(1234);
    txn->put(table_stream_offset_key(
                     {instance_id_, second_identity.base_db_id(), second_identity.base_table_id(),
                      second_identity.stream_db_id(), second_identity.stream_id(), 2001}),
             second_offset.SerializeAsString());
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    auto make_request = [&](int64_t txn_id, int64_t second_expected_tso) {
        CommitTxnRequest request =
                make_consume_request(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
        TableStreamUpdatePB* second_stream_update = request.add_table_stream_updates();
        second_stream_update->mutable_identity()->CopyFrom(second_identity);
        TableStreamPartitionUpdatePB* second_partition_update =
                second_stream_update->add_partition_updates();
        second_partition_update->set_partition_id(2001);
        second_partition_update->set_expected_state(TABLE_STREAM_OFFSET_CONSUMED);
        second_partition_update->set_expected_offset_tso(second_expected_tso);
        second_partition_update->set_next_offset_tso(125);
        return request;
    };

    CommitTxnResponse response = commit_transaction(
            make_request(begin_target_transaction("consume-multiple-streams-stale"), 104));
    EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
    EXPECT_EQ(get_latest_offset(identity_, 2001).offset_tso(), 100);
    EXPECT_EQ(get_latest_offset(second_identity, 2001).offset_tso(), 105);

    response = commit_transaction(
            make_request(begin_target_transaction("consume-multiple-streams"), 105));
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    EXPECT_EQ(get_latest_offset(identity_, 2001).offset_tso(), 120);
    EXPECT_EQ(get_latest_offset(second_identity, 2001).offset_tso(), 125);
}

TEST_F(MetaServiceTableStreamTest, CommitTargetRowsetAndOffsetAtomically) {
    set_multi_version_status(MULTI_VERSION_DISABLED);
    constexpr int64_t kSourcePartitionId = 2001;
    constexpr int64_t kTargetDbId = 3001;
    constexpr int64_t kTargetTableId = 3002;
    constexpr int64_t kTargetIndexId = 3003;
    constexpr int64_t kTargetPartitionId = 4001;
    constexpr int64_t kTargetTabletId = 5001;

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    put_latest_partition_state(txn.get(), kSourcePartitionId, 8, 130, 100);
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    create_target_tablet(kTargetDbId, kTargetTableId, kTargetIndexId, kTargetPartitionId,
                         kTargetTabletId);

    int64_t stale_txn_id = begin_target_transaction("consume-target-rowset-stale");
    stage_target_rowset(stale_txn_id, kTargetPartitionId, kTargetTabletId);
    CommitTxnRequest stale_request = make_consume_request(stale_txn_id, kSourcePartitionId,
                                                          TABLE_STREAM_OFFSET_CONSUMED, 99, 120);
    CommitTxnResponse response = commit_transaction(stale_request);
    ASSERT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
    EXPECT_EQ(get_latest_offset(kSourcePartitionId).offset_tso(), 100);

    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    std::string value;
    EXPECT_EQ(txn->get(meta_rowset_tmp_key({instance_id_, stale_txn_id, kTargetTabletId}), &value),
              TxnErrorCode::TXN_OK);
    EXPECT_EQ(txn->get(meta_rowset_key({instance_id_, kTargetTabletId, 2}), &value),
              TxnErrorCode::TXN_KEY_NOT_FOUND);
    ASSERT_EQ(txn->get(partition_version_key(
                               {instance_id_, kTargetDbId, kTargetTableId, kTargetPartitionId}),
                       &value),
              TxnErrorCode::TXN_OK);
    VersionPB target_version;
    ASSERT_TRUE(target_version.ParseFromString(value));
    EXPECT_EQ(target_version.version(), 1);

    int64_t committed_txn_id = begin_target_transaction("consume-target-rowset");
    stage_target_rowset(committed_txn_id, kTargetPartitionId, kTargetTabletId);
    CommitTxnRequest commit_request = make_consume_request(committed_txn_id, kSourcePartitionId,
                                                           TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
    commit_request.set_commit_tso(999);
    response = commit_transaction(commit_request);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    EXPECT_EQ(get_latest_offset(kSourcePartitionId).offset_tso(), 120);

    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    EXPECT_EQ(txn->get(meta_rowset_tmp_key({instance_id_, committed_txn_id, kTargetTabletId}),
                       &value),
              TxnErrorCode::TXN_KEY_NOT_FOUND);
    ASSERT_EQ(txn->get(meta_rowset_key({instance_id_, kTargetTabletId, 2}), &value),
              TxnErrorCode::TXN_OK);
    doris::RowsetMetaCloudPB committed_rowset;
    ASSERT_TRUE(committed_rowset.ParseFromString(value));
    EXPECT_EQ(committed_rowset.txn_id(), committed_txn_id);
    ASSERT_EQ(txn->get(partition_version_key(
                               {instance_id_, kTargetDbId, kTargetTableId, kTargetPartitionId}),
                       &value),
              TxnErrorCode::TXN_OK);
    ASSERT_TRUE(target_version.ParseFromString(value));
    EXPECT_EQ(target_version.version(), 2);
    EXPECT_EQ(target_version.commit_tso(), 999);

    response = commit_transaction(commit_request);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    EXPECT_EQ(get_latest_offset(kSourcePartitionId).offset_tso(), 120);
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    ASSERT_EQ(txn->get(meta_rowset_key({instance_id_, kTargetTabletId, 2}), &value),
              TxnErrorCode::TXN_OK);
    ASSERT_TRUE(committed_rowset.ParseFromString(value));
    EXPECT_EQ(committed_rowset.txn_id(), committed_txn_id);
    ASSERT_EQ(txn->get(partition_version_key(
                               {instance_id_, kTargetDbId, kTargetTableId, kTargetPartitionId}),
                       &value),
              TxnErrorCode::TXN_OK);
    ASSERT_TRUE(target_version.ParseFromString(value));
    EXPECT_EQ(target_version.version(), 2);
    EXPECT_EQ(txn->get(meta_rowset_key({instance_id_, kTargetTabletId, 3}), &value),
              TxnErrorCode::TXN_KEY_NOT_FOUND);
}

TEST_F(MetaServiceTableStreamTest, CommitLargePartitionBatch) {
    set_multi_version_status(MULTI_VERSION_WRITE_ONLY);
    constexpr int kPartitionCount = 1201;

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    for (int i = 0; i < kPartitionCount; ++i) {
        int64_t partition_id = 30000 + i;
        put_latest_partition_state(txn.get(), partition_id, i + 1, 40000 + i, std::nullopt);
    }
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    CommitTxnRequest request;
    request.set_cloud_unique_id(cloud_unique_id_);
    request.set_db_id(3001);
    request.set_txn_id(begin_target_transaction("consume-large-partition-batch"));
    TableStreamUpdatePB* stream_update = request.add_table_stream_updates();
    stream_update->mutable_identity()->CopyFrom(identity_);
    for (int i = 0; i < kPartitionCount; ++i) {
        int64_t partition_id = 30000 + i;
        TableStreamPartitionUpdatePB* update = stream_update->add_partition_updates();
        update->set_partition_id(partition_id);
        update->set_expected_state(TABLE_STREAM_OFFSET_UNKNOWN);
        update->set_next_offset_tso(40000 + i);
    }

    CommitTxnResponse response = commit_transaction(request);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    EXPECT_EQ(get_latest_offset(30000).offset_tso(), 40000);
    EXPECT_EQ(get_latest_offset(30000 + kPartitionCount / 2).offset_tso(),
              40000 + kPartitionCount / 2);
    EXPECT_EQ(get_latest_offset(30000 + kPartitionCount - 1).offset_tso(),
              40000 + kPartitionCount - 1);
}

TEST_F(MetaServiceTableStreamTest, CommitRejectsStaleAndInvalidOffsets) {
    set_multi_version_status(MULTI_VERSION_DISABLED);
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    int64_t txn_id = begin_target_transaction("consume-stale-offset");
    CommitTxnResponse response =
            consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 99, 120);
    EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
    EXPECT_EQ(get_latest_offset(2001).offset_tso(), 100);

    txn_id = begin_target_transaction("consume-backward-offset");
    response = consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 90);
    EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
    EXPECT_EQ(get_latest_offset(2001).offset_tso(), 100);

    txn_id = begin_target_transaction("consume-beyond-visible-tso");
    response = consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 131);
    EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
    EXPECT_EQ(get_latest_offset(2001).offset_tso(), 100);
}

TEST_F(MetaServiceTableStreamTest, CommitValidatesOffsetTsoDomain) {
    set_multi_version_status(MULTI_VERSION_DISABLED);
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    put_latest_partition_state(txn.get(), 2001, 8, 130, -1);
    put_latest_partition_state(txn.get(), 2002, 9, 140, std::nullopt);
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    CommitTxnResponse response =
            consume_partition(begin_target_transaction("consume-empty-boundary-existing"), 2001,
                              TABLE_STREAM_OFFSET_CONSUMED, -1, -1);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    EXPECT_EQ(get_latest_offset(2001).offset_tso(), -1);

    response = consume_partition(begin_target_transaction("consume-empty-boundary-unknown"), 2002,
                                 TABLE_STREAM_OFFSET_UNKNOWN, 0, -1);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    EXPECT_EQ(get_latest_offset(2002).offset_tso(), -1);

    for (int64_t invalid_tso : {0, -2}) {
        response = consume_partition(
                begin_target_transaction("consume-invalid-expected-" + std::to_string(invalid_tso)),
                2001, TABLE_STREAM_OFFSET_CONSUMED, invalid_tso, -1);
        EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
        EXPECT_EQ(get_latest_offset(2001).offset_tso(), -1);

        response = consume_partition(
                begin_target_transaction("consume-invalid-next-" + std::to_string(invalid_tso)),
                2001, TABLE_STREAM_OFFSET_CONSUMED, -1, invalid_tso);
        EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
        EXPECT_EQ(get_latest_offset(2001).offset_tso(), -1);
    }
}

TEST_F(MetaServiceTableStreamTest, CommitUsesCloneEffectiveOffset) {
    const std::string source_instance_id = "table_stream_read_state_source";
    set_clone_source(source_instance_id, Versionstamp(100, 0));

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    PartitionIndexPB partition_index;
    partition_index.set_db_id(identity_.base_db_id());
    partition_index.set_table_id(identity_.base_table_id());
    txn->put(versioned::partition_index_key({source_instance_id, 2001}),
             partition_index.SerializeAsString());
    versioned_put(txn.get(), versioned::meta_partition_key({source_instance_id, 2001}),
                  Versionstamp(41, 0), "");
    VersionPB version;
    version.set_version(18);
    version.set_commit_tso(230);
    versioned_put(txn.get(), versioned::partition_version_key({source_instance_id, 2001}),
                  Versionstamp(42, 0), version.SerializeAsString());
    TableStreamOffsetPB inherited_offset;
    inherited_offset.set_partition_id(2001);
    inherited_offset.set_state(TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING);
    inherited_offset.set_offset_tso(200);
    versioned_put(txn.get(),
                  versioned::table_stream_offset_key(
                          {source_instance_id, identity_.base_db_id(), identity_.base_table_id(),
                           identity_.stream_db_id(), identity_.stream_id(), 2001}),
                  Versionstamp(43, 0), inherited_offset.SerializeAsString());
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    int64_t txn_id = begin_target_transaction("consume-inherited-offset-as-unknown");
    CommitTxnResponse response =
            consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_UNKNOWN, 0, 220);
    EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);

    txn_id = begin_target_transaction("consume-inherited-offset");
    response =
            consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING, 200, 220);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    EXPECT_EQ(get_latest_offset(2001).offset_tso(), 220);

    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    std::string value;
    Versionstamp offset_version;
    ASSERT_EQ(
            versioned_get(txn.get(),
                          versioned::table_stream_offset_key(
                                  {instance_id_, identity_.base_db_id(), identity_.base_table_id(),
                                   identity_.stream_db_id(), identity_.stream_id(), 2001}),
                          &offset_version, &value),
            TxnErrorCode::TXN_OK);
    TableStreamOffsetPB versioned_offset;
    ASSERT_TRUE(versioned_offset.ParseFromString(value));
    EXPECT_EQ(versioned_offset.offset_tso(), 220);

    OperationLogPB operation_log;
    Versionstamp log_version;
    ASSERT_EQ(read_operation_log(txn.get(), versioned::log_key({instance_id_}), &log_version,
                                 &operation_log),
              TxnErrorCode::TXN_OK);
    ASSERT_TRUE(operation_log.has_commit_txn());
    ASSERT_EQ(operation_log.commit_txn().table_stream_offset_gc_size(), 1);
    const TableStreamPartitionSetPB& offset_gc =
            operation_log.commit_txn().table_stream_offset_gc(0);
    EXPECT_EQ(offset_gc.identity().SerializeAsString(), identity_.SerializeAsString());
    ASSERT_EQ(offset_gc.partition_ids_size(), 1);
    EXPECT_EQ(offset_gc.partition_ids(0), 2001);
}

TEST_F(MetaServiceTableStreamTest, CommitRejectsLatestOffsetWithoutVersionedOffset) {
    set_multi_version_status(MULTI_VERSION_READ_WRITE);
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    put_versioned_partition_state(txn.get(), 2001, 8, 130, std::nullopt);
    put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    int64_t txn_id = begin_target_transaction("consume-inconsistent-local-offset");
    CommitTxnResponse response =
            consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_UNKNOWN, 0, 120);
    EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
    EXPECT_EQ(get_latest_offset(2001).offset_tso(), 100);
}

TEST_F(MetaServiceTableStreamTest, CommitRejectsDifferentLatestAndVersionedOffsetValues) {
    set_multi_version_status(MULTI_VERSION_READ_WRITE);
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    put_versioned_partition_state(txn.get(), 2001, 8, 130, std::nullopt);

    TableStreamOffsetPB latest_offset;
    latest_offset.set_partition_id(2001);
    latest_offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
    latest_offset.set_offset_tso(100);
    latest_offset.set_last_consumption_time_ms(1234);
    TableStreamOffsetKeyInfo key_info {instance_id_,
                                       identity_.base_db_id(),
                                       identity_.base_table_id(),
                                       identity_.stream_db_id(),
                                       identity_.stream_id(),
                                       2001};
    txn->put(table_stream_offset_key(key_info), latest_offset.SerializeAsString());
    TableStreamOffsetPB versioned_offset = latest_offset;
    versioned_offset.set_last_consumption_time_ms(1235);
    versioned_put(txn.get(), versioned::table_stream_offset_key(key_info), Versionstamp(43, 0),
                  versioned_offset.SerializeAsString());
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    int64_t txn_id = begin_target_transaction("consume-inconsistent-offset-values");
    CommitTxnResponse response =
            consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
    EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
    EXPECT_EQ(get_latest_offset(2001).last_consumption_time_ms(), 1234);
}

TEST_F(MetaServiceTableStreamTest, CommitRejectsUnsupportedModeAndMissingCommitTso) {
    set_multi_version_status(MULTI_VERSION_DISABLED);
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    VersionPB version;
    version.set_version(8);
    txn->put(partition_version_key(
                     {instance_id_, identity_.base_db_id(), identity_.base_table_id(), 2001}),
             version.SerializeAsString());
    TableStreamOffsetPB offset;
    offset.set_partition_id(2001);
    offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
    offset.set_offset_tso(100);
    txn->put(table_stream_offset_key({instance_id_, identity_.base_db_id(),
                                      identity_.base_table_id(), identity_.stream_db_id(),
                                      identity_.stream_id(), 2001}),
             offset.SerializeAsString());
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    int64_t txn_id = begin_target_transaction("consume-missing-visible-tso");
    CommitTxnResponse response =
            consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 110);
    EXPECT_EQ(response.status().code(), MetaServiceCode::VERSION_NOT_FOUND);
    EXPECT_EQ(get_latest_offset(2001).offset_tso(), 100);

    set_multi_version_status(MULTI_VERSION_ENABLED);
    txn_id = begin_target_transaction("consume-enabled-mode");
    response = consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 110);
    EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
    EXPECT_EQ(get_latest_offset(2001).offset_tso(), 100);
}

TEST_F(MetaServiceTableStreamTest, CommitRejectsUnsupportedTxnModesAndForcesImmediate) {
    set_multi_version_status(MULTI_VERSION_DISABLED);
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    CommitTxnRequest request = make_consume_request(begin_target_transaction("consume-2pc"), 2001,
                                                    TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
    request.set_is_2pc(true);
    EXPECT_EQ(commit_transaction(request).status().code(), MetaServiceCode::INVALID_ARGUMENT);

    request = make_consume_request(begin_target_transaction("consume-txn-load"), 2001,
                                   TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
    request.set_is_txn_load(true);
    EXPECT_EQ(commit_transaction(request).status().code(), MetaServiceCode::INVALID_ARGUMENT);

    request = make_consume_request(begin_target_transaction("consume-sub-txn"), 2001,
                                   TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
    request.add_sub_txn_infos()->set_sub_txn_id(1);
    EXPECT_EQ(commit_transaction(request).status().code(), MetaServiceCode::INVALID_ARGUMENT);

    int old_fuzzy_possibility = config::cloud_txn_lazy_commit_fuzzy_possibility;
    config::cloud_txn_lazy_commit_fuzzy_possibility = 100;
    request = make_consume_request(begin_target_transaction("consume-immediate"), 2001,
                                   TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
    request.set_enable_txn_lazy_commit(true);
    CommitTxnResponse response = commit_transaction(request);
    config::cloud_txn_lazy_commit_fuzzy_possibility = old_fuzzy_possibility;
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    EXPECT_EQ(get_latest_offset(2001).offset_tso(), 120);
}

TEST_F(MetaServiceTableStreamTest, DropWritesTypedRecycleIndexInDisabledMode) {
    set_multi_version_status(MULTI_VERSION_DISABLED);
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    IndexRequest request;
    request.set_cloud_unique_id(cloud_unique_id_);
    request.add_index_ids(identity_.stream_id());
    request.set_db_id(identity_.base_db_id());
    request.set_table_id(identity_.base_table_id());
    request.set_object_type(IndexObjectTypePB::TABLE_STREAM);
    request.set_stream_db_id(identity_.stream_db_id());
    request.set_expiration(0);
    IndexResponse response;
    brpc::Controller controller;
    service_->drop_index(&controller, &request, &response, nullptr);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();

    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    std::string value;
    ASSERT_EQ(txn->get(recycle_index_key({instance_id_, identity_.stream_id()}), &value),
              TxnErrorCode::TXN_OK);
    RecycleIndexPB recycle_index;
    ASSERT_TRUE(recycle_index.ParseFromString(value));
    EXPECT_EQ(recycle_index.state(), RecycleIndexPB::DROPPED);
    EXPECT_EQ(recycle_index.object_type(), IndexObjectTypePB::TABLE_STREAM);
    EXPECT_EQ(recycle_index.db_id(), identity_.base_db_id());
    EXPECT_EQ(recycle_index.table_id(), identity_.base_table_id());
    EXPECT_EQ(recycle_index.stream_db_id(), identity_.stream_db_id());
}

TEST_F(MetaServiceTableStreamTest, DropWritesTypedOperationLogInVersionedMode) {
    set_multi_version_status(MULTI_VERSION_READ_WRITE);
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    IndexRequest request;
    request.set_cloud_unique_id(cloud_unique_id_);
    request.add_index_ids(identity_.stream_id());
    request.set_db_id(identity_.base_db_id());
    request.set_table_id(identity_.base_table_id());
    request.set_object_type(IndexObjectTypePB::TABLE_STREAM);
    request.set_stream_db_id(identity_.stream_db_id());
    request.set_expiration(0);
    IndexResponse response;
    brpc::Controller controller;
    service_->drop_index(&controller, &request, &response, nullptr);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();

    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    OperationLogPB operation_log;
    Versionstamp log_version;
    ASSERT_EQ(read_operation_log(txn.get(), versioned::log_key({instance_id_}), &log_version,
                                 &operation_log),
              TxnErrorCode::TXN_OK);
    ASSERT_TRUE(operation_log.has_drop_index());
    ASSERT_TRUE(operation_log.has_min_timestamp());
    EXPECT_EQ(operation_log.min_timestamp(),
              static_cast<uint64_t>(std::numeric_limits<int64_t>::max()));
    const DropIndexLogPB& drop_index = operation_log.drop_index();
    EXPECT_EQ(drop_index.object_type(), IndexObjectTypePB::TABLE_STREAM);
    EXPECT_EQ(drop_index.db_id(), identity_.base_db_id());
    EXPECT_EQ(drop_index.table_id(), identity_.base_table_id());
    EXPECT_EQ(drop_index.stream_db_id(), identity_.stream_db_id());
    ASSERT_EQ(drop_index.index_ids_size(), 1);
    EXPECT_EQ(drop_index.index_ids(0), identity_.stream_id());

    std::string value;
    EXPECT_EQ(txn->get(recycle_index_key({instance_id_, identity_.stream_id()}), &value, true),
              TxnErrorCode::TXN_KEY_NOT_FOUND);
}

TEST_F(MetaServiceTableStreamTest, DropUsesEarliestLocalVersionedOffset) {
    const std::string source_instance_id = "table_stream_drop_source";
    const Versionstamp source_version =
            put_auto_versioned_offset(source_instance_id, identity_, 2001, 10);
    set_clone_source(source_instance_id, Versionstamp(source_version.version() + 1, 0));
    // Partition 2002 sorts after partition 2001 but contains the earliest local version.
    const Versionstamp min_local_version =
            put_auto_versioned_offset(instance_id_, identity_, 2002, 30);
    put_auto_versioned_offset(instance_id_, identity_, 2001, 10);
    put_auto_versioned_offset(instance_id_, identity_, 2001, 20);

    IndexRequest request;
    request.set_cloud_unique_id(cloud_unique_id_);
    request.add_index_ids(identity_.stream_id());
    request.set_db_id(identity_.base_db_id());
    request.set_table_id(identity_.base_table_id());
    request.set_object_type(IndexObjectTypePB::TABLE_STREAM);
    request.set_stream_db_id(identity_.stream_db_id());
    request.set_expiration(0);
    IndexResponse response;
    brpc::Controller controller;
    service_->drop_index(&controller, &request, &response, nullptr);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    OperationLogPB operation_log;
    Versionstamp log_version;
    ASSERT_EQ(read_operation_log(txn.get(), versioned::log_key({instance_id_}), &log_version,
                                 &operation_log),
              TxnErrorCode::TXN_OK);
    ASSERT_TRUE(operation_log.has_drop_index());
    ASSERT_TRUE(operation_log.has_min_timestamp());
    EXPECT_LT(source_version.version(), min_local_version.version());
    EXPECT_EQ(operation_log.min_timestamp(), min_local_version.version());
    EXPECT_LT(operation_log.min_timestamp(), log_version.version());
}

TEST_F(MetaServiceTableStreamTest, DropFindsMinimumInLargeVersionedOffsetHistory) {
    set_multi_version_status(MULTI_VERSION_READ_WRITE);
    const int default_full_range_batch_size = FullRangeGetOptions().batch_limit;

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    TableStreamOffsetPB offset;
    offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
    offset.set_offset_tso(100);
    for (int i = 0; i < default_full_range_batch_size; ++i) {
        offset.set_partition_id(2001);
        versioned_put(
                txn.get(),
                versioned::table_stream_offset_key(
                        {instance_id_, identity_.base_db_id(), identity_.base_table_id(),
                         identity_.stream_db_id(), identity_.stream_id(), offset.partition_id()}),
                Versionstamp(1000 + i, 0), offset.SerializeAsString());
    }
    // This key sorts after all partition 2001 history but carries the global minimum version.
    offset.set_partition_id(2002);
    versioned_put(txn.get(),
                  versioned::table_stream_offset_key(
                          {instance_id_, identity_.base_db_id(), identity_.base_table_id(),
                           identity_.stream_db_id(), identity_.stream_id(), offset.partition_id()}),
                  Versionstamp(50, 0), offset.SerializeAsString());
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    IndexRequest request;
    request.set_cloud_unique_id(cloud_unique_id_);
    request.add_index_ids(identity_.stream_id());
    request.set_db_id(identity_.base_db_id());
    request.set_table_id(identity_.base_table_id());
    request.set_object_type(IndexObjectTypePB::TABLE_STREAM);
    request.set_stream_db_id(identity_.stream_db_id());
    request.set_expiration(0);
    IndexResponse response;
    brpc::Controller controller;
    service_->drop_index(&controller, &request, &response, nullptr);
    ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();

    ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    OperationLogPB operation_log;
    Versionstamp log_version;
    ASSERT_EQ(read_operation_log(txn.get(), versioned::log_key({instance_id_}), &log_version,
                                 &operation_log),
              TxnErrorCode::TXN_OK);
    ASSERT_TRUE(operation_log.has_drop_index());
    ASSERT_TRUE(operation_log.has_min_timestamp());
    EXPECT_EQ(operation_log.min_timestamp(), 50);
}

TEST_F(MetaServiceTableStreamTest, DropAndRecycleCloneChildOffsets) {
    const bool old_force_immediate_recycle = config::force_immediate_recycle;
    config::force_immediate_recycle = true;
    DORIS_CLOUD_DEFER {
        config::force_immediate_recycle = old_force_immediate_recycle;
    };

    const std::string source_instance_id = "table_stream_drop_recycle_source";
    TableStreamIdentityPB local_identity = identity_;
    TableStreamIdentityPB inherited_identity = identity_;
    inherited_identity.set_stream_id(identity_.stream_id() + 1);
    put_auto_versioned_offset(source_instance_id, local_identity, 2001, 90);
    const Versionstamp source_version =
            put_auto_versioned_offset(source_instance_id, inherited_identity, 2001, 80);
    const Versionstamp source_snapshot = Versionstamp::next(source_version);
    set_clone_source(source_instance_id, source_snapshot);
    put_auto_versioned_offset(instance_id_, local_identity, 2001, 100, true);

    auto drop_stream = [&](const TableStreamIdentityPB& identity) {
        IndexRequest request;
        request.set_cloud_unique_id(cloud_unique_id_);
        request.add_index_ids(identity.stream_id());
        request.set_db_id(identity.base_db_id());
        request.set_table_id(identity.base_table_id());
        request.set_object_type(IndexObjectTypePB::TABLE_STREAM);
        request.set_stream_db_id(identity.stream_db_id());
        request.set_expiration(0);
        IndexResponse response;
        brpc::Controller controller;
        service_->drop_index(&controller, &request, &response, nullptr);
        EXPECT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
    };
    drop_stream(local_identity);
    drop_stream(inherited_identity);

    InstanceInfoPB child_instance;
    child_instance.set_instance_id(instance_id_);
    child_instance.set_status(InstanceInfoPB::NORMAL);
    child_instance.set_multi_version_status(MULTI_VERSION_READ_WRITE);
    child_instance.set_source_instance_id(source_instance_id);
    child_instance.set_source_snapshot_id(SnapshotManager::serialize_snapshot_id(source_snapshot));
    InstanceRecycler recycler(service_->txn_kv(), child_instance, RecyclerThreadPoolGroup {},
                              std::make_shared<TxnLazyCommitter>(service_->txn_kv()));
    ASSERT_EQ(recycler.init(), 0);
    ASSERT_EQ(recycler.recycle_operation_logs(), 0);

    auto key_exists = [&](std::string_view key) {
        std::unique_ptr<Transaction> txn;
        EXPECT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string value;
        return txn->get(key, &value, true) == TxnErrorCode::TXN_OK;
    };
    EXPECT_TRUE(key_exists(recycle_index_key({instance_id_, local_identity.stream_id()})));
    EXPECT_TRUE(key_exists(recycle_index_key({instance_id_, inherited_identity.stream_id()})));

    ASSERT_EQ(recycler.recycle_indexes(), 0);
    const auto offset_count = [&](const std::string& prefix) {
        std::unique_ptr<Transaction> txn;
        EXPECT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::unique_ptr<RangeGetIterator> iter;
        EXPECT_EQ(txn->get(prefix, lexical_end(prefix), &iter, true, 0), TxnErrorCode::TXN_OK);
        return iter->size();
    };
    EXPECT_EQ(offset_count(table_stream_offset_key_prefix(
                      instance_id_, local_identity.base_db_id(), local_identity.base_table_id(),
                      local_identity.stream_db_id(), local_identity.stream_id())),
              0);
    EXPECT_EQ(offset_count(versioned::table_stream_offset_key_prefix(
                      instance_id_, local_identity.base_db_id(), local_identity.base_table_id(),
                      local_identity.stream_db_id(), local_identity.stream_id())),
              0);
    EXPECT_EQ(offset_count(versioned::table_stream_offset_key_prefix(
                      instance_id_, inherited_identity.base_db_id(),
                      inherited_identity.base_table_id(), inherited_identity.stream_db_id(),
                      inherited_identity.stream_id())),
              0);
    EXPECT_FALSE(key_exists(recycle_index_key({instance_id_, local_identity.stream_id()})));
    EXPECT_FALSE(key_exists(recycle_index_key({instance_id_, inherited_identity.stream_id()})));
    EXPECT_GT(
            offset_count(versioned::table_stream_offset_key_prefix(
                    source_instance_id, local_identity.base_db_id(), local_identity.base_table_id(),
                    local_identity.stream_db_id(), local_identity.stream_id())),
            0);
    EXPECT_GT(offset_count(versioned::table_stream_offset_key_prefix(
                      source_instance_id, inherited_identity.base_db_id(),
                      inherited_identity.base_table_id(), inherited_identity.stream_db_id(),
                      inherited_identity.stream_id())),
              0);
}

} // namespace doris::cloud