// 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 <gen_cpp/olap_file.pb.h>
#include <google/protobuf/arena.h>
#include <gtest/gtest.h>

#include <random>

#include "common/config.h"
#include "common/defer.h"
#include "cpp/sync_point.h"
#include "meta-service/meta_service.h"
#include "meta-service/meta_service_schema.h"
#include "meta-store/blob_message.h"
#include "meta-store/codec.h"
#include "meta-store/document_message.h"
#include "meta-store/keys.h"
#include "meta-store/txn_kv.h"
#include "meta-store/txn_kv_error.h"

static std::string instance_id = "schema_kv_test";

namespace doris::cloud {
extern std::unique_ptr<MetaServiceProxy> get_meta_service();

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

static void fill_schema(doris::TabletSchemaCloudPB* schema, int32_t schema_version) {
    schema->set_schema_version(schema_version);
    for (int i = 0; i < 10; ++i) {
        auto column = schema->add_column();
        column->set_unique_id(20000 + i);
        column->set_type("INT");
    }
}

static void add_tablet(CreateTabletsRequest& req, int64_t table_id, int64_t index_id,
                       int64_t partition_id, int64_t tablet_id, const std::string& rowset_id,
                       int32_t schema_version) {
    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();
    fill_schema(schema, schema_version);
    auto first_rowset = tablet->add_rs_metas();
    first_rowset->set_rowset_id(0); // required
    first_rowset->set_rowset_id_v2(rowset_id);
    first_rowset->set_start_version(0);
    first_rowset->set_end_version(1);
    first_rowset->mutable_tablet_schema()->CopyFrom(*schema);
}

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

static void get_rowset(MetaServiceProxy* meta_service, int64_t table_id, int64_t index_id,
                       int64_t partition_id, int64_t tablet_id, GetRowsetResponse& res) {
    brpc::Controller cntl;
    GetRowsetRequest req;
    auto tablet_idx = req.mutable_idx();
    tablet_idx->set_table_id(table_id);
    tablet_idx->set_index_id(index_id);
    tablet_idx->set_partition_id(partition_id);
    tablet_idx->set_tablet_id(tablet_id);
    req.set_start_version(0);
    req.set_end_version(-1);
    req.set_cumulative_compaction_cnt(0);
    req.set_base_compaction_cnt(0);
    req.set_cumulative_point(2);
    meta_service->get_rowset(&cntl, &req, &res, nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << tablet_id;
}

static void check_get_tablet(MetaServiceProxy* meta_service, int64_t tablet_id,
                             int32_t schema_version) {
    brpc::Controller cntl;
    GetTabletRequest req;
    GetTabletResponse res;
    req.set_tablet_id(tablet_id);
    meta_service->get_tablet(&cntl, &req, &res, nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << tablet_id;
    ASSERT_TRUE(res.has_tablet_meta()) << tablet_id;
    EXPECT_TRUE(res.tablet_meta().has_schema()) << tablet_id;
    EXPECT_EQ(res.tablet_meta().schema_version(), schema_version) << tablet_id;
};

TEST(DetachSchemaKVTest, TabletTest) {
    auto meta_service = get_meta_service();
    // meta_service->resource_mgr().reset(); // Do not use resource manager

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

    // new MS write with write_schema_kv=false, old MS read
    {
        constexpr auto db_id = 1, table_id = 10001, index_id = 10002, partition_id = 10003,
                       tablet_id = 10004;
        config::write_schema_kv = false;
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
                                              partition_id, tablet_id, next_rowset_id(), 1));
        // check saved values in txn_kv
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string tablet_key, tablet_val;
        meta_tablet_key({instance_id, table_id, index_id, partition_id, tablet_id}, &tablet_key);
        ASSERT_EQ(txn->get(tablet_key, &tablet_val), TxnErrorCode::TXN_OK);
        doris::TabletMetaCloudPB saved_tablet;
        ASSERT_TRUE(saved_tablet.ParseFromString(tablet_val));
        EXPECT_TRUE(saved_tablet.has_schema());
        std::string rowset_key, rowset_val;
        meta_rowset_key({instance_id, tablet_id, 1}, &rowset_key);
        ASSERT_EQ(txn->get(rowset_key, &rowset_val), TxnErrorCode::TXN_OK);
        doris::RowsetMetaCloudPB saved_rowset;
        ASSERT_TRUE(saved_rowset.ParseFromString(rowset_val));
        EXPECT_TRUE(saved_rowset.has_tablet_schema());
    }

    // old MS write, new MS read
    {
        constexpr auto table_id = 10011, index_id = 10012, partition_id = 10013, tablet_id = 10014;
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        doris::TabletMetaCloudPB saved_tablet;
        saved_tablet.set_table_id(table_id);
        saved_tablet.set_index_id(index_id);
        saved_tablet.set_partition_id(partition_id);
        saved_tablet.set_tablet_id(tablet_id);
        saved_tablet.mutable_schema()->set_schema_version(1);
        auto column = saved_tablet.mutable_schema()->add_column();
        column->set_unique_id(30001);
        column->set_type("INT");
        std::string tablet_key, tablet_val;
        meta_tablet_key({instance_id, table_id, index_id, partition_id, tablet_id}, &tablet_key);
        ASSERT_TRUE(saved_tablet.SerializeToString(&tablet_val));
        txn->put(tablet_key, tablet_val);
        TabletIndexPB saved_tablet_idx;
        saved_tablet_idx.set_table_id(table_id);
        saved_tablet_idx.set_index_id(index_id);
        saved_tablet_idx.set_partition_id(partition_id);
        saved_tablet_idx.set_tablet_id(tablet_id);
        std::string tablet_idx_key, tablet_idx_val;
        meta_tablet_idx_key({instance_id, tablet_id}, &tablet_idx_key);
        ASSERT_TRUE(saved_tablet_idx.SerializeToString(&tablet_idx_val));
        txn->put(tablet_idx_key, tablet_idx_val);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
        // check get tablet response
        check_get_tablet(meta_service.get(), tablet_id, 1);
    }

    auto check_new_saved_tablet_val = [](Transaction* txn, int64_t table_id, int64_t index_id,
                                         int64_t partition_id, int64_t tablet_id,
                                         int32_t schema_version) {
        std::string tablet_key, tablet_val;
        meta_tablet_key({instance_id, table_id, index_id, partition_id, tablet_id}, &tablet_key);
        ASSERT_EQ(txn->get(tablet_key, &tablet_val), TxnErrorCode::TXN_OK);
        doris::TabletMetaCloudPB saved_tablet;
        ASSERT_TRUE(saved_tablet.ParseFromString(tablet_val));
        EXPECT_FALSE(saved_tablet.has_schema()) << tablet_id;
        EXPECT_EQ(saved_tablet.schema_version(), schema_version) << tablet_id;
    };

    // new MS write with write_schema_kv=true, new MS read
    {
        constexpr auto db_id = 1, table_id = 10021, index_id = 10022, partition_id = 10023,
                       tablet_id = 10024;
        config::write_schema_kv = true;
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
                                              partition_id, tablet_id, next_rowset_id(), 1));
        // check saved values in txn_kv
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        check_new_saved_tablet_val(txn.get(), table_id, index_id, partition_id, tablet_id, 1);
        std::string rowset_key, rowset_val;
        meta_rowset_key({instance_id, tablet_id, 1}, &rowset_key);
        ASSERT_EQ(txn->get(rowset_key, &rowset_val), TxnErrorCode::TXN_OK);
        doris::RowsetMetaCloudPB saved_rowset;
        ASSERT_TRUE(saved_rowset.ParseFromString(rowset_val));
        EXPECT_FALSE(saved_rowset.has_tablet_schema());
        EXPECT_EQ(saved_rowset.index_id(), index_id);
        EXPECT_EQ(saved_rowset.schema_version(), 1);
        // check get tablet response
        check_get_tablet(meta_service.get(), tablet_id, 1);
        // check get rowset response
        GetRowsetResponse get_rowset_res;
        get_rowset(meta_service.get(), table_id, index_id, partition_id, tablet_id, get_rowset_res);
        ASSERT_EQ(get_rowset_res.rowset_meta_size(), 1);
        EXPECT_TRUE(get_rowset_res.rowset_meta(0).has_tablet_schema());
        EXPECT_EQ(get_rowset_res.rowset_meta(0).index_id(), index_id);
        EXPECT_EQ(get_rowset_res.rowset_meta(0).schema_version(), 1);
        ASSERT_TRUE(get_rowset_res.has_stats());
        EXPECT_EQ(get_rowset_res.stats().num_rows(), 0);
        EXPECT_EQ(get_rowset_res.stats().num_rowsets(), 1);
        EXPECT_EQ(get_rowset_res.stats().num_segments(), 0);
        EXPECT_EQ(get_rowset_res.stats().data_size(), 0);
        EXPECT_EQ(get_rowset_res.stats().index_size(), 0);
        EXPECT_EQ(get_rowset_res.stats().segment_size(), 0);
    }

    // new MS batch create tablets with write_schema_kv=true
    {
        config::write_schema_kv = true;
        brpc::Controller cntl;
        CreateTabletsRequest req;
        CreateTabletsResponse res;
        req.set_db_id(1);
        add_tablet(req, 10031, 10032, 10033, 100031, next_rowset_id(), 1);
        add_tablet(req, 10031, 10032, 10033, 100032, next_rowset_id(), 2);
        add_tablet(req, 10031, 10032, 10033, 100033, next_rowset_id(), 2);
        add_tablet(req, 10031, 10032, 10033, 100034, next_rowset_id(), 3);
        add_tablet(req, 10031, 10032, 10033, 100035, next_rowset_id(), 3);
        add_tablet(req, 10031, 10032, 10033, 100036, next_rowset_id(), 3);
        add_tablet(req, 10031, 10034, 10033, 100037, next_rowset_id(), 1);
        add_tablet(req, 10031, 10034, 10033, 100038, next_rowset_id(), 2);
        add_tablet(req, 10031, 10034, 10033, 100039, next_rowset_id(), 2);
        meta_service->create_tablets(&cntl, &req, &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        // check saved values in txn_kv
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        check_new_saved_tablet_val(txn.get(), 10031, 10032, 10033, 100031, 1);
        check_new_saved_tablet_val(txn.get(), 10031, 10032, 10033, 100032, 2);
        check_new_saved_tablet_val(txn.get(), 10031, 10032, 10033, 100033, 2);
        check_new_saved_tablet_val(txn.get(), 10031, 10032, 10033, 100034, 3);
        check_new_saved_tablet_val(txn.get(), 10031, 10032, 10033, 100035, 3);
        check_new_saved_tablet_val(txn.get(), 10031, 10032, 10033, 100036, 3);
        check_new_saved_tablet_val(txn.get(), 10031, 10034, 10033, 100037, 1);
        check_new_saved_tablet_val(txn.get(), 10031, 10034, 10033, 100038, 2);
        check_new_saved_tablet_val(txn.get(), 10031, 10034, 10033, 100039, 2);
        // check get tablet response
        check_get_tablet(meta_service.get(), 100031, 1);
        check_get_tablet(meta_service.get(), 100032, 2);
        check_get_tablet(meta_service.get(), 100033, 2);
        check_get_tablet(meta_service.get(), 100034, 3);
        check_get_tablet(meta_service.get(), 100035, 3);
        check_get_tablet(meta_service.get(), 100036, 3);
        check_get_tablet(meta_service.get(), 100037, 1);
        check_get_tablet(meta_service.get(), 100038, 2);
        check_get_tablet(meta_service.get(), 100039, 2);
    }
}

TEST(DetachSchemaKVTest, PutSchemaKvTest) {
    config::meta_schema_value_version = 1;
    auto meta_service = get_meta_service();

    int64_t index_id = 14221;
    int64_t schema_version = 0;
    std::string key = meta_schema_key({instance_id, index_id, schema_version});
    std::string versioned_key = versioned::meta_schema_key({instance_id, index_id, schema_version});
    doris::TabletSchemaCloudPB schema;
    fill_schema(&schema, schema_version);

    std::unique_ptr<Transaction> txn;
    MetaServiceCode code = MetaServiceCode::OK;
    std::string msg;
    for (int i = 0; i < 2; ++i) {
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        put_schema_kv(code, msg, txn.get(), key, schema);
        ASSERT_EQ(code, MetaServiceCode::OK);
        put_versioned_schema_kv(code, msg, txn.get(), versioned_key, schema);
        ASSERT_EQ(code, MetaServiceCode::OK);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

        // verify the tablet schema is written
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        doris::TabletSchemaCloudPB saved_schema;
        ValueBuf buf;
        ASSERT_EQ(cloud::blob_get(txn.get(), key, &buf), TxnErrorCode::TXN_OK);

        // verify the versioned tablet schema is written
        ASSERT_EQ(document_get(txn.get(), versioned_key, &saved_schema), TxnErrorCode::TXN_OK);
        EXPECT_EQ(saved_schema.schema_version(), schema_version);
    }

    {
        // put new schema version
        schema_version = 1;
        schema.set_schema_version(schema_version);
        key = meta_schema_key({instance_id, index_id, schema_version});
        versioned_key = versioned::meta_schema_key({instance_id, index_id, schema_version});

        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        put_schema_kv(code, msg, txn.get(), key, schema);
        ASSERT_EQ(code, MetaServiceCode::OK);
        put_versioned_schema_kv(code, msg, txn.get(), versioned_key, schema);
        ASSERT_EQ(code, MetaServiceCode::OK);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

        // verify the tablet schema is written
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        doris::TabletSchemaCloudPB saved_schema;
        ValueBuf buf;
        ASSERT_EQ(cloud::blob_get(txn.get(), key, &buf), TxnErrorCode::TXN_OK);

        // verify the versioned tablet schema is written
        ASSERT_EQ(document_get(txn.get(), versioned_key, &saved_schema), TxnErrorCode::TXN_OK);
        EXPECT_EQ(saved_schema.schema_version(), schema_version);
    }
}

static void begin_txn(MetaServiceProxy* meta_service, int64_t db_id, const std::string& label,
                      int64_t table_id, int64_t& txn_id) {
    brpc::Controller cntl;
    BeginTxnRequest req;
    BeginTxnResponse res;
    auto txn_info = req.mutable_txn_info();
    txn_info->set_db_id(db_id);
    txn_info->set_label(label);
    txn_info->add_table_ids(table_id);
    txn_info->set_timeout_ms(36000);
    meta_service->begin_txn(&cntl, &req, &res, nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << label;
    ASSERT_TRUE(res.has_txn_id()) << label;
    txn_id = res.txn_id();
}

static void commit_txn(MetaServiceProxy* meta_service, int64_t db_id, int64_t txn_id,
                       const std::string& label) {
    brpc::Controller cntl;
    CommitTxnRequest req;
    CommitTxnResponse res;
    req.set_db_id(db_id);
    req.set_txn_id(txn_id);
    meta_service->commit_txn(&cntl, &req, &res, nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << label;
}

static doris::RowsetMetaCloudPB create_rowset(int64_t txn_id, int64_t tablet_id,
                                              const std::string& rowset_id, int32_t schema_version,
                                              int64_t version = -1,
                                              const TabletSchemaCloudPB* schema = nullptr) {
    doris::RowsetMetaCloudPB rowset;
    rowset.set_rowset_id(0); // required
    rowset.set_rowset_id_v2(rowset_id);
    rowset.set_tablet_id(tablet_id);
    rowset.set_txn_id(txn_id);
    rowset.set_num_rows(100);
    rowset.set_num_segments(1);
    rowset.set_data_disk_size(10000);
    rowset.set_index_disk_size(1000);
    rowset.set_total_disk_size(11000);
    if (version > 0) {
        rowset.set_start_version(version);
        rowset.set_end_version(version);
    }
    rowset.mutable_tablet_schema()->set_schema_version(schema_version);
    rowset.set_txn_expiration(::time(nullptr)); // Required by DCHECK
    if (schema != nullptr) {
        rowset.mutable_tablet_schema()->CopyFrom(*schema);
        rowset.mutable_tablet_schema()->set_schema_version(schema_version);
    }
    return rowset;
}

static void prepare_rowset(MetaServiceProxy* meta_service, const doris::RowsetMetaCloudPB& rowset,
                           CreateRowsetResponse& res) {
    brpc::Controller cntl;
    auto arena = res.GetArena();
    auto req = google::protobuf::Arena::CreateMessage<CreateRowsetRequest>(arena);
    req->mutable_rowset_meta()->CopyFrom(rowset);
    meta_service->prepare_rowset(&cntl, req, &res, nullptr);
    if (!arena) delete req;
}

static void commit_rowset(MetaServiceProxy* meta_service, const doris::RowsetMetaCloudPB& rowset,
                          CreateRowsetResponse& res) {
    brpc::Controller cntl;
    auto arena = res.GetArena();
    auto req = google::protobuf::Arena::CreateMessage<CreateRowsetRequest>(arena);
    req->mutable_rowset_meta()->CopyFrom(rowset);
    meta_service->commit_rowset(&cntl, req, &res, nullptr);
    if (!arena) delete req;
}

static void insert_rowset(MetaServiceProxy* meta_service, int64_t db_id, const std::string& label,
                          int64_t table_id, int64_t tablet_id, int32_t schema_version,
                          const TabletSchemaCloudPB* schema = nullptr) {
    int64_t txn_id = 0;
    ASSERT_NO_FATAL_FAILURE(begin_txn(meta_service, db_id, label, table_id, txn_id));
    CreateRowsetResponse res;
    auto rowset = create_rowset(txn_id, tablet_id, next_rowset_id(), schema_version, -1, schema);
    rowset.set_has_variant_type_in_schema(schema != nullptr);
    prepare_rowset(meta_service, rowset, res);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << label;
    res.Clear();
    ASSERT_NO_FATAL_FAILURE(commit_rowset(meta_service, rowset, res));
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << label << ", msg=" << res.status().msg();
    commit_txn(meta_service, db_id, txn_id, label);
}

static TabletSchemaCloudPB getVariantSchema() {
    TabletSchemaCloudPB schema;
    schema.set_schema_version(10086);
    // columns
    ColumnPB var;
    var.set_type("VARIANT");
    var.set_unique_id(10);
    ColumnPB var_sub1;
    var_sub1.set_type("INT");
    var_sub1.set_unique_id(-1);
    schema.add_column()->CopyFrom(var_sub1);
    ColumnPB var_sub2;
    var_sub2.set_type("DOUBLE");
    var_sub2.set_unique_id(-1);
    schema.add_column()->CopyFrom(var_sub2);
    ColumnPB var_sparse_sub1;
    var_sparse_sub1.set_type("DOUBLE");
    var_sparse_sub1.set_unique_id(-1);
    var.add_sparse_columns()->CopyFrom(var_sparse_sub1);
    schema.add_column()->CopyFrom(var);

    // indexes
    doris::TabletIndexPB index1;
    index1.set_index_id(111);
    index1.set_index_suffix_name("aaabbbccc");
    schema.add_index()->CopyFrom(index1);

    doris::TabletIndexPB index2;
    index2.set_index_id(222);
    schema.add_index()->CopyFrom(index2);
    return schema;
}

TEST(DetachSchemaKVTest, RowsetTest) {
    auto meta_service = get_meta_service();
    // meta_service->resource_mgr().reset(); // Do not use resource manager

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

    constexpr int64_t db_id = 10000;

    // new MS write with write_schema_kv=false, old MS read
    {
        constexpr auto table_id = 10001, index_id = 10002, partition_id = 10003, tablet_id = 10004;
        config::write_schema_kv = false;
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
                                              partition_id, tablet_id, next_rowset_id(), 1));
        ASSERT_NO_FATAL_FAILURE(
                insert_rowset(meta_service.get(), db_id, "101", table_id, tablet_id, 2)); // [2-2]
        // check saved values in txn_kv
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string rowset_key, rowset_val;
        meta_rowset_key({instance_id, tablet_id, 2}, &rowset_key); // [2-2]
        ASSERT_EQ(txn->get(rowset_key, &rowset_val), TxnErrorCode::TXN_OK);
        doris::RowsetMetaCloudPB saved_rowset;
        ASSERT_TRUE(saved_rowset.ParseFromString(rowset_val));
        ASSERT_TRUE(saved_rowset.has_tablet_schema());
        EXPECT_EQ(saved_rowset.tablet_schema().schema_version(), 2);
    }

    // old MS write, new MS read
    {
        constexpr auto table_id = 10011, index_id = 10012, partition_id = 10013, tablet_id = 10014;
        config::write_schema_kv = false;
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
                                              partition_id, tablet_id, next_rowset_id(), 1));
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        auto saved_rowset = create_rowset(10015, tablet_id, next_rowset_id(), 2, 2);
        std::string rowset_key, rowset_val;
        meta_rowset_key({instance_id, tablet_id, 2}, &rowset_key); // version=[2-2]
        ASSERT_TRUE(saved_rowset.SerializeToString(&rowset_val));
        txn->put(rowset_key, rowset_val);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
        // check get rowset response
        GetRowsetResponse get_rowset_res;
        get_rowset(meta_service.get(), table_id, index_id, partition_id, tablet_id, get_rowset_res);
        ASSERT_EQ(get_rowset_res.rowset_meta_size(), 2);
        ASSERT_EQ(get_rowset_res.rowset_meta(0).end_version(), 1); // [0-1]
        ASSERT_TRUE(get_rowset_res.rowset_meta(0).has_tablet_schema());
        EXPECT_EQ(get_rowset_res.rowset_meta(0).tablet_schema().schema_version(), 1);
        EXPECT_EQ(get_rowset_res.rowset_meta(0).index_id(), index_id);
        EXPECT_EQ(get_rowset_res.rowset_meta(0).schema_version(), 1);
        ASSERT_EQ(get_rowset_res.rowset_meta(1).end_version(), 2); // [2-2]
        ASSERT_TRUE(get_rowset_res.rowset_meta(1).has_tablet_schema());
        EXPECT_EQ(get_rowset_res.rowset_meta(1).tablet_schema().schema_version(), 2);
        EXPECT_EQ(get_rowset_res.rowset_meta(1).index_id(), index_id);
        EXPECT_EQ(get_rowset_res.rowset_meta(1).schema_version(), 2);
    }

    // new MS write with write_schema_kv=true, new MS read
    {
        constexpr auto table_id = 10021, index_id = 10022, partition_id = 10023, tablet_id = 10024;
        config::write_schema_kv = true;
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
                                              partition_id, tablet_id, next_rowset_id(), 1));
        ASSERT_NO_FATAL_FAILURE(
                insert_rowset(meta_service.get(), db_id, "201", table_id, tablet_id, 2)); // [2-2]
        // check saved values in txn_kv
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string rowset_key, rowset_val;
        meta_rowset_key({instance_id, 10024, 2}, &rowset_key); // [2-2]
        ASSERT_EQ(txn->get(rowset_key, &rowset_val), TxnErrorCode::TXN_OK);
        doris::RowsetMetaCloudPB saved_rowset;
        ASSERT_TRUE(saved_rowset.ParseFromString(rowset_val));
        EXPECT_FALSE(saved_rowset.has_tablet_schema());
        EXPECT_EQ(saved_rowset.index_id(), index_id);
        EXPECT_EQ(saved_rowset.schema_version(), 2);
        // check get rowset response
        GetRowsetResponse get_rowset_res;
        get_rowset(meta_service.get(), table_id, index_id, partition_id, tablet_id, get_rowset_res);
        ASSERT_EQ(get_rowset_res.rowset_meta_size(), 2);
        ASSERT_EQ(get_rowset_res.rowset_meta(0).end_version(), 1); // [0-1]
        ASSERT_TRUE(get_rowset_res.rowset_meta(0).has_tablet_schema());
        EXPECT_EQ(get_rowset_res.rowset_meta(0).tablet_schema().schema_version(), 1);
        EXPECT_EQ(get_rowset_res.rowset_meta(0).index_id(), index_id);
        EXPECT_EQ(get_rowset_res.rowset_meta(0).schema_version(), 1);
        ASSERT_EQ(get_rowset_res.rowset_meta(1).end_version(), 2); // [2-2]
        ASSERT_TRUE(get_rowset_res.rowset_meta(1).has_tablet_schema());
        EXPECT_EQ(get_rowset_res.rowset_meta(1).tablet_schema().schema_version(), 2);
        EXPECT_EQ(get_rowset_res.rowset_meta(1).index_id(), index_id);
        EXPECT_EQ(get_rowset_res.rowset_meta(1).schema_version(), 2);
        ASSERT_TRUE(get_rowset_res.has_stats());
        EXPECT_EQ(get_rowset_res.stats().num_rows(), 100);
        EXPECT_EQ(get_rowset_res.stats().num_rowsets(), 2);
        EXPECT_EQ(get_rowset_res.stats().num_segments(), 1);
        EXPECT_EQ(get_rowset_res.stats().data_size(), 11000);
        EXPECT_EQ(get_rowset_res.stats().index_size(), 1000);
        EXPECT_EQ(get_rowset_res.stats().segment_size(), 10000);
    }

    // new MS read rowsets committed by both old and new MS
    auto insert_and_get_rowset = [&meta_service](int64_t table_id, int64_t index_id,
                                                 int64_t partition_id, int64_t tablet_id,
                                                 int label_base,
                                                 google::protobuf::Arena* arena = nullptr,
                                                 const TabletSchemaCloudPB* schema = nullptr) {
        config::write_schema_kv = false;
        std::mt19937 rng(std::chrono::system_clock::now().time_since_epoch().count());
        std::uniform_int_distribution<int> dist1(1, 4);
        std::uniform_int_distribution<int> dist2(2, 7);
        std::vector<int> schema_versions {1};
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
                                              partition_id, tablet_id, next_rowset_id(),
                                              schema_versions[0]));
        for (int i = 0; i < 10; ++i) {
            schema_versions.push_back(dist1(rng));
            ASSERT_NO_FATAL_FAILURE(insert_rowset(meta_service.get(), db_id,
                                                  std::to_string(++label_base), table_id, tablet_id,
                                                  schema_versions.back(), schema));
        }
        config::write_schema_kv = true;
        for (int i = 0; i < 15; ++i) {
            schema_versions.push_back(dist2(rng));
            ASSERT_NO_FATAL_FAILURE(insert_rowset(meta_service.get(), db_id,
                                                  std::to_string(++label_base), table_id, tablet_id,
                                                  schema_versions.back(), schema));
        }
        // check get rowset response
        auto get_rowset_res = google::protobuf::Arena::CreateMessage<GetRowsetResponse>(arena);
        DORIS_CLOUD_DEFER {
            if (!arena) delete get_rowset_res;
        };
        get_rowset(meta_service.get(), table_id, index_id, partition_id, tablet_id,
                   *get_rowset_res);
        ASSERT_EQ(get_rowset_res->rowset_meta_size(), schema_versions.size());
        for (int i = 0; i < schema_versions.size(); ++i) {
            auto& rowset = get_rowset_res->rowset_meta(i);
            ASSERT_EQ(rowset.end_version(), i + 1);
            ASSERT_TRUE(rowset.has_tablet_schema());
            EXPECT_EQ(rowset.tablet_schema().schema_version(), schema_versions[i]);
            EXPECT_EQ(rowset.index_id(), index_id);
            EXPECT_EQ(rowset.schema_version(), schema_versions[i]);
        }
        ASSERT_TRUE(get_rowset_res->has_stats());
        EXPECT_EQ(get_rowset_res->stats().num_rows(), 2500);
        EXPECT_EQ(get_rowset_res->stats().num_rowsets(), 26);
        EXPECT_EQ(get_rowset_res->stats().num_segments(), 25);
        EXPECT_EQ(get_rowset_res->stats().data_size(), 275000);
        EXPECT_EQ(get_rowset_res->stats().index_size(), 25000);
        EXPECT_EQ(get_rowset_res->stats().segment_size(), 250000);
        if (schema != nullptr) {
            auto schema_version = get_rowset_res->rowset_meta(10).schema_version();
            get_rowset_res->mutable_rowset_meta(10)->mutable_tablet_schema()->set_schema_version(
                    10086);
            std::cout << get_rowset_res->rowset_meta(10).tablet_schema().ShortDebugString()
                      << std::endl;
            std::cout << schema->ShortDebugString() << std::endl;
            EXPECT_EQ(get_rowset_res->rowset_meta(10).tablet_schema().column(2).type(),
                      schema->column(2).type());
            EXPECT_EQ(get_rowset_res->rowset_meta(10).tablet_schema().index(0).index_suffix_name(),
                      schema->index(0).index_suffix_name());
            EXPECT_EQ(get_rowset_res->rowset_meta(10).tablet_schema().index(1).index_id(),
                      schema->index(1).index_id());
            get_rowset_res->mutable_rowset_meta(10)->mutable_tablet_schema()->set_schema_version(
                    schema_version);
        }
    };
    insert_and_get_rowset(10031, 10032, 10033, 10034, 300);
    // use arena
    google::protobuf::Arena arena;
    insert_and_get_rowset(10041, 10042, 10043, 10044, 400, &arena);
    TabletSchemaCloudPB schema = getVariantSchema();
    insert_and_get_rowset(10051, 10052, 10053, 10054, 500, &arena, &schema);
}

TEST(DetachSchemaKVTest, InsertExistedRowsetTest) {
    auto meta_service = get_meta_service();
    // meta_service->resource_mgr().reset(); // Do not use resource manager

    auto sp = SyncPoint::get_instance();
    DORIS_CLOUD_DEFER {
        SyncPoint::get_instance()->clear_all_call_backs();
        config::enable_recycle_delete_rowset_key_check = true;
    };
    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->enable_processing();
    config::enable_recycle_delete_rowset_key_check = false;

    int64_t db_id = 1000;

    // old MS commit rowset, new MS commit rowset again
    {
        constexpr auto table_id = 10001, index_id = 10002, partition_id = 10003, tablet_id = 10004;
        config::write_schema_kv = false;
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
                                              partition_id, tablet_id, next_rowset_id(), 1));

        int64_t txn_id = -1;
        std::string label = "test_abort_txn_label";
        {
            brpc::Controller cntl;
            BeginTxnRequest req;
            BeginTxnResponse res;
            req.set_cloud_unique_id("test_cloud_unique_id");
            auto* txn_info = req.mutable_txn_info();
            txn_info->set_db_id(db_id);
            txn_info->set_label(label);
            txn_info->add_table_ids(table_id);
            txn_info->set_timeout_ms(36000);
            meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                    &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            txn_id = res.txn_id();
            LOG(INFO) << "Step 1: Transaction started, txn_id=" << txn_id;
        }
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        auto committed_rowset = create_rowset(txn_id, tablet_id, next_rowset_id(), 2, 2);
        std::string tmp_rowset_key, tmp_rowset_val;
        // 0:instance_id  1:txn_id  2:tablet_id
        meta_rowset_tmp_key({instance_id, txn_id, tablet_id}, &tmp_rowset_key);
        ASSERT_TRUE(committed_rowset.SerializeToString(&tmp_rowset_val));
        txn->put(tmp_rowset_key, tmp_rowset_val);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
        CreateRowsetResponse res;
        auto new_rowset = create_rowset(txn_id, tablet_id, next_rowset_id(), 2, 2);
        prepare_rowset(meta_service.get(), new_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::ALREADY_EXISTED);
        ASSERT_TRUE(res.has_existed_rowset_meta());
        EXPECT_EQ(res.existed_rowset_meta().rowset_id_v2(), committed_rowset.rowset_id_v2());
        EXPECT_EQ(res.existed_rowset_meta().index_id(), index_id);
        EXPECT_EQ(res.existed_rowset_meta().schema_version(), 2);
        ASSERT_TRUE(res.existed_rowset_meta().has_tablet_schema());
        EXPECT_EQ(res.existed_rowset_meta().tablet_schema().schema_version(), 2);
        res.Clear();
        commit_rowset(meta_service.get(), new_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::ALREADY_EXISTED);
        ASSERT_TRUE(res.has_existed_rowset_meta());
        EXPECT_EQ(res.existed_rowset_meta().rowset_id_v2(), committed_rowset.rowset_id_v2());
        EXPECT_EQ(res.existed_rowset_meta().index_id(), index_id);
        EXPECT_EQ(res.existed_rowset_meta().schema_version(), 2);
        ASSERT_TRUE(res.existed_rowset_meta().has_tablet_schema());
        EXPECT_EQ(res.existed_rowset_meta().tablet_schema().schema_version(), 2);
    }

    // new MS commit rowset, new MS commit rowset again
    auto insert_existed_rowset = [&meta_service, db_id](int64_t table_id, int64_t index_id,
                                                        int64_t partition_id, int64_t tablet_id,
                                                        int64_t txn_id,
                                                        google::protobuf::Arena* arena = nullptr) {
        config::write_schema_kv = true;
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
                                              partition_id, tablet_id, next_rowset_id(), 1));
        auto committed_rowset = create_rowset(txn_id, tablet_id, next_rowset_id(), 2, 2);
        auto res = google::protobuf::Arena::CreateMessage<CreateRowsetResponse>(arena);
        DORIS_CLOUD_DEFER {
            if (!arena) delete res;
        };
        prepare_rowset(meta_service.get(), committed_rowset, *res);
        ASSERT_EQ(res->status().code(), MetaServiceCode::OK);
        res->Clear();
        commit_rowset(meta_service.get(), committed_rowset, *res);
        ASSERT_EQ(res->status().code(), MetaServiceCode::OK);
        res->Clear();
        auto new_rowset = create_rowset(txn_id, tablet_id, next_rowset_id(), 2, 2);
        prepare_rowset(meta_service.get(), new_rowset, *res);
        ASSERT_EQ(res->status().code(), MetaServiceCode::ALREADY_EXISTED);
        ASSERT_TRUE(res->has_existed_rowset_meta());
        EXPECT_EQ(res->existed_rowset_meta().rowset_id_v2(), committed_rowset.rowset_id_v2());
        EXPECT_EQ(res->existed_rowset_meta().index_id(), index_id);
        EXPECT_EQ(res->existed_rowset_meta().schema_version(), 2);
        ASSERT_TRUE(res->existed_rowset_meta().has_tablet_schema());
        EXPECT_EQ(res->existed_rowset_meta().tablet_schema().schema_version(), 2);
        res->Clear();
        commit_rowset(meta_service.get(), new_rowset, *res);
        ASSERT_EQ(res->status().code(), MetaServiceCode::ALREADY_EXISTED);
        ASSERT_TRUE(res->has_existed_rowset_meta());
        EXPECT_EQ(res->existed_rowset_meta().rowset_id_v2(), committed_rowset.rowset_id_v2());
        EXPECT_EQ(res->existed_rowset_meta().index_id(), index_id);
        EXPECT_EQ(res->existed_rowset_meta().schema_version(), 2);
        ASSERT_TRUE(res->existed_rowset_meta().has_tablet_schema());
        EXPECT_EQ(res->existed_rowset_meta().tablet_schema().schema_version(), 2);
    };
    insert_existed_rowset(10011, 10012, 10013, 10014, 10015);
    google::protobuf::Arena arena;
    insert_existed_rowset(10021, 10022, 10023, 10024, 10025, &arena);
}

TEST(SchemaKVTest, InsertExistedRowsetTest) {
    auto meta_service = get_meta_service();
    // meta_service->resource_mgr().reset(); // Do not use resource manager

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

    constexpr int64_t db_id = 2000;

    config::write_schema_kv = true;
    config::meta_schema_value_version = 0;
    ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10004,
                                          next_rowset_id(), 1));
    check_get_tablet(meta_service.get(), 10004, 1);

    config::meta_schema_value_version = 1;
    ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10005,
                                          next_rowset_id(), 2));
    check_get_tablet(meta_service.get(), 10005, 2);
}

static void check_schema(MetaServiceProxy* meta_service, int64_t tablet_id, int32_t schema_version,
                         bool expected_disable_auto_compaction = false) {
    brpc::Controller cntl;
    GetTabletRequest req;
    GetTabletResponse res;
    req.set_tablet_id(tablet_id);
    meta_service->get_tablet(&cntl, &req, &res, nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << tablet_id;
    ASSERT_TRUE(res.has_tablet_meta()) << tablet_id;
    EXPECT_TRUE(res.tablet_meta().has_schema()) << tablet_id;
    EXPECT_EQ(res.tablet_meta().schema_version(), schema_version) << tablet_id;
    EXPECT_EQ(res.tablet_meta().schema().column_size(), 10) << tablet_id;
    EXPECT_EQ(res.tablet_meta().schema().disable_auto_compaction(),
              expected_disable_auto_compaction)
            << tablet_id;
};

static void update_tablet(MetaServiceProxy* meta_service, int64_t tablet_id) {
    brpc::Controller cntl;
    UpdateTabletRequest req;
    UpdateTabletResponse res;

    auto meta_info = req.add_tablet_meta_infos();
    meta_info->set_disable_auto_compaction(true);
    meta_info->set_tablet_id(tablet_id);

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

TEST(AlterSchemaKVTest, AlterDisableAutoCompactionTest) {
    //case 1 config::write_schema_kv = true;
    constexpr int64_t db_id = 3000;
    {
        auto meta_service = get_meta_service();
        config::write_schema_kv = true;
        //config::meta_schema_value_version = 0;
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10004,
                                              next_rowset_id(), 0));
        check_get_tablet(meta_service.get(), 10004, 0);
        check_schema(meta_service.get(), 10004, 0);

        //config::meta_schema_value_version = 1;
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10005,
                                              next_rowset_id(), 2));
        check_get_tablet(meta_service.get(), 10005, 2);

        update_tablet(meta_service.get(), 10005);
        check_schema(meta_service.get(), 10005, 2, true);
    }

    //case 2 config::write_schema_kv = false;
    {
        auto meta_service = get_meta_service();
        config::write_schema_kv = false;
        auto defer1 =
                std::make_unique<std::function<void()>>([]() { config::write_schema_kv = true; });

        //config::meta_schema_value_version = 0;
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10004,
                                              next_rowset_id(), 0));
        check_get_tablet(meta_service.get(), 10004, 0);
        check_schema(meta_service.get(), 10004, 0);

        //config::meta_schema_value_version = 1;
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10005,
                                              next_rowset_id(), 2));
        check_get_tablet(meta_service.get(), 10005, 2);

        update_tablet(meta_service.get(), 10005);
        check_schema(meta_service.get(), 10005, 2, true);
    }

    //case 3 config::write_schema_kv = false, create tablet, config::write_schema_kv = true;
    {
        auto meta_service = get_meta_service();
        config::write_schema_kv = false;
        auto defer1 =
                std::make_unique<std::function<void()>>([]() { config::write_schema_kv = true; });

        //config::meta_schema_value_version = 0;
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10004,
                                              next_rowset_id(), 0));
        check_get_tablet(meta_service.get(), 10004, 0);
        check_schema(meta_service.get(), 10004, 0);

        //config::meta_schema_value_version = 1;
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10005,
                                              next_rowset_id(), 2));
        check_get_tablet(meta_service.get(), 10005, 2);
        config::write_schema_kv = true;
        update_tablet(meta_service.get(), 10005);
        check_schema(meta_service.get(), 10005, 2, true);
    }

    //case 4 config::write_schema_kv = false, create tablet, config::write_schema_kv = true;
    //       meta_schema_value_version = 0, meta_schema_value_version = 1
    {
        auto meta_service = get_meta_service();
        config::write_schema_kv = false;
        auto defer1 = std::make_unique<std::function<void()>>([]() {
            config::write_schema_kv = true;
            config::meta_schema_value_version = 1;
        });

        config::meta_schema_value_version = 0;
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10004,
                                              next_rowset_id(), 0));
        check_get_tablet(meta_service.get(), 10004, 0);
        check_schema(meta_service.get(), 10004, 0);

        config::meta_schema_value_version = 1;
        ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10005,
                                              next_rowset_id(), 2));
        check_get_tablet(meta_service.get(), 10005, 2);
        config::write_schema_kv = true;
        update_tablet(meta_service.get(), 10005);
        check_schema(meta_service.get(), 10005, 2, true);
    }
}

// Helper: count the number of blob chunks for a given base key.
static int count_blob_chunks(Transaction* txn, const std::string& base_key) {
    std::string begin_key = base_key;
    std::string end_key = base_key;
    encode_int64(INT64_MAX, &end_key);
    std::unique_ptr<RangeGetIterator> iter;
    auto err = txn->get(begin_key, end_key, &iter);
    EXPECT_EQ(err, TxnErrorCode::TXN_OK);
    if (err != TxnErrorCode::TXN_OK) return -1;
    int count = 0;
    while (iter->has_next()) {
        iter->next();
        ++count;
    }
    return count;
}

// Verify that ValueBuf::remove() before blob_put is necessary to clean up
// orphaned blob chunks when the number of chunks decreases.
TEST(AlterSchemaKVTest, BlobRemoveBeforeOverwriteTest) {
    auto meta_service = get_meta_service();

    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 = instance_id;
        ret->second = true;
    });
    sp->enable_processing();

    constexpr int64_t db_id = 4000;
    constexpr int64_t table_id = 40001;
    constexpr int64_t index_id = 40002;
    constexpr int64_t partition_id = 40003;
    constexpr int64_t tablet_id = 40004;
    constexpr int32_t schema_version = 0;

    config::write_schema_kv = true;
    config::meta_schema_value_version = 1;

    // Create a tablet with a normal 10-column schema.
    ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
                                          partition_id, tablet_id, next_rowset_id(),
                                          schema_version));
    check_get_tablet(meta_service.get(), tablet_id, schema_version);

    // Compute the schema KV key.
    auto schema_key = meta_schema_key({instance_id, index_id, schema_version});

    // Overwrite the schema KV with a small split_size to create 3 blob chunks.
    // The normal schema (~120 bytes) with split_size=50 yields 3 chunks.
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        // First remove existing blob chunks
        ValueBuf old_val;
        ASSERT_EQ(cloud::blob_get(txn.get(), schema_key, &old_val), TxnErrorCode::TXN_OK);
        old_val.remove(txn.get());
        // Write with small split_size to create 3 chunks
        doris::TabletSchemaCloudPB schema;
        ASSERT_TRUE(old_val.to_pb(&schema));
        std::string serialized = schema.SerializeAsString();
        size_t split_size = (serialized.size() / 3) + 1; // ensures exactly 3 chunks
        cloud::blob_put(txn.get(), schema_key, schema, config::meta_schema_value_version,
                        split_size);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }

    // Verify we now have 3 blob chunks.
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        EXPECT_EQ(count_blob_chunks(txn.get(), schema_key), 3);
    }

    // Case 1: update_tablet WITH remove (default behavior).
    // The remove cleans up old 3 chunks, then blob_put writes 1 chunk
    // (with default split_size=90KB, the ~120 byte schema fits in 1 chunk).
    update_tablet(meta_service.get(), tablet_id);
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        EXPECT_EQ(count_blob_chunks(txn.get(), schema_key), 1)
                << "With remove: should have 1 chunk (old 3 cleaned up)";
        // Verify schema is correct
        ValueBuf val;
        ASSERT_EQ(cloud::blob_get(txn.get(), schema_key, &val), TxnErrorCode::TXN_OK);
        doris::TabletSchemaCloudPB schema;
        ASSERT_TRUE(val.to_pb(&schema));
        EXPECT_TRUE(schema.disable_auto_compaction());
    }
    check_schema(meta_service.get(), tablet_id, schema_version, true);

    // Reset: overwrite schema KV again with 3 blob chunks and
    // disable_auto_compaction=false so update_tablet will modify it again.
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        ValueBuf old_val;
        ASSERT_EQ(cloud::blob_get(txn.get(), schema_key, &old_val), TxnErrorCode::TXN_OK);
        old_val.remove(txn.get());
        doris::TabletSchemaCloudPB schema;
        ASSERT_TRUE(old_val.to_pb(&schema));
        schema.set_disable_auto_compaction(false);
        std::string serialized = schema.SerializeAsString();
        size_t split_size = (serialized.size() / 3) + 1;
        cloud::blob_put(txn.get(), schema_key, schema, config::meta_schema_value_version,
                        split_size);
        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
    }
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        EXPECT_EQ(count_blob_chunks(txn.get(), schema_key), 3);
    }

    // Case 2: update_tablet WITHOUT remove (inject skip via sync point).
    // This simulates what would happen if we didn't have the remove logic.
    sp->set_call_back("update_tablet::skip_schema_remove",
                      [](auto&& args) { *try_any_cast<bool*>(args[0]) = true; });
    update_tablet(meta_service.get(), tablet_id);
    sp->clear_call_back("update_tablet::skip_schema_remove");

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
        int chunks = count_blob_chunks(txn.get(), schema_key);
        EXPECT_EQ(chunks, 3) << "Without remove: should have 3 chunks (2 orphaned + 1 overwritten)";
        // blob_get reads all 3 chunks and concatenates them, producing
        // corrupted data that cannot be parsed correctly.
        ValueBuf val;
        ASSERT_EQ(cloud::blob_get(txn.get(), schema_key, &val), TxnErrorCode::TXN_OK);
        doris::TabletSchemaCloudPB schema;
        // The deserialized schema may parse but contain garbage from
        // orphaned chunks, or the column count will be wrong.
        if (val.to_pb(&schema)) {
            // If it parses, the data is corrupted - the column count or
            // content won't match the expected schema.
            LOG(INFO) << "Without remove: parsed schema has " << schema.column_size()
                      << " columns (expected 10), disable_auto_compaction="
                      << schema.disable_auto_compaction();
        } else {
            LOG(INFO) << "Without remove: schema parse failed as expected (corrupted data)";
        }
    }
}

} // namespace doris::cloud