// 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 "recycler/recycler.h"

#include <butil/strings/string_split.h>

#include <bvar/variable.h>

#include <fmt/core.h>

#include <gen_cpp/cloud.pb.h>

#include <gen_cpp/olap_file.pb.h>

#include <glog/logging.h>

#include <gtest/gtest.h>

#include <array>

#include <atomic>

#include <chrono>

#include <cstdint>

#include <cstdlib>

#include <exception>

#include <future>

#include <memory>

#include <numeric>

#include <random>

#include <string>

#include <string_view>

#include <thread>

#include "common/bvars.h"

#include "common/config.h"

#include "common/defer.h"

#include "common/logging.h"

#include "common/simple_thread_pool.h"

#include "common/util.h"

#include "cpp/sync_point.h"

#include "meta-service/meta_service.h"

#include "meta-store/blob_message.h"

#include "meta-store/document_message.h"

#include "meta-store/keys.h"

#include "meta-store/mem_txn_kv.h"

#include "meta-store/txn_kv.h"

#include "meta-store/txn_kv_error.h"

#include "mock_accessor.h"

#include "mock_resource_manager.h"

#include "rate-limiter/rate_limiter.h"

#include "recycler/checker.h"

#include "recycler/recycler.cpp"

#include "recycler/storage_vault_accessor.h"

#include "recycler/util.h"

#include "recycler/white_black_list.h"

using namespace doris;

std::string instance_id = "instance_id_recycle_test";

int64_t current_time = 0;

static constexpr int64_t db_id = 1000;

static RecyclerMetricsContext ctx;

namespace {

int64_t read_status_bvar_value(const std::string& module, const std::string& name,

const std::string& tag) {

const std::string bvar_name = fmt::format("{}_{}_{}", module, name, tag);

const std::string value = bvar::Variable::describe_exposed(bvar_name);

if (value.empty()) {

ADD_FAILURE() << "bvar " << bvar_name << " not exposed";

return 0;

}

try {

return std::stoll(value);

} catch (const std::exception& e) {

ADD_FAILURE() << "failed to parse bvar value for " << bvar_name << ": " << e.what()

<< ", raw=\"" << value << "\"";

return 0;

}

}

} // namespace

std::vector<std::string> index_v2_file_path = {

"data/1753202639945/0200000000001a5c92f4e7d9j8f2b4c8a3e6f8b1c9d2e5f8_0.idx",

"data/1753202639947/0200000000001b8d45a74r6c7sf3e9c2b6d4a8e1f7c3d9e2_0.idx",

"data/1753202639951/0200000000001c9e56b8g4f0x8s7g2f0d3c7e5b9f2e8d4f0_0.idx",

"data/1753202639953/0200000000001d0f67c9h5g8a3e6f8b1e4d8f6c0g3f9e5g1_0.idx",

"data/1753202639955/0200000000001e1g78d067c9h5g8i6h2f5e9g7d1h4g0f6h2_0.idx",

"data/1753202639957/0200000000001f2h89e1jg7d1h4g07i3g6f0h8e2i5h1g7i3_0.idx",

"data/1753202639959/020000000000208i90f2k0h8e2i5h8j4h7g1i9f3j6i2h8j4_0.idx",

"data/1753202639961/02000000000021aj01g3l9k5i8h2j8e2i5h8j0g4k7j3i9k5_0.idx",

"data/1753202639963/02000000000022bk12h4m0lk0h8e2i56j9i3k1h5l8k4j0l6_0.idx",

"data/1753202639965/02000000000023cl23i5n1m7g3l9k5i8k0j4l2i6m9l5k1m7_0.idx",

"data/1753202639967/02000000000024dm34j1m7g3l9k6o2n8l1k5m3j7n0m6l2n8_0.idx",

"data/1753202639969/02000000000025en45k7p3o9m2l6n4k34j1m7g38o1n7m3o9_0.idx",

"data/1753202639971/02000000000026fo56l8q4p0n2l6n4k343m7o5l9p2o8n4p0_0.idx",

"data/1753202639973/02000000000027gp67m9r5q8q4p0n2l1o4n8p6m0q3p9o5q1_0.idx",

"data/1753202639975/02000000000028hq78n0s6rm9r5q8q42p5o9q7n1r4q0p6r2_0.idx",

"data/1753202639977/02000000000029ir89o1t7s78n0s6rm3q6p0r8o2s5r1q7s3_0.idx",

"data/1753202639979/0200000000002ajs90p2u8t4m3q6p0r8r7q1s9p3t6s2r8t4_0.idx",

"data/1753202639981/0200000000002bkt01q3v9u2u8t4m3q5s8r2t0q4u7t3s9u5_0.idx",

"data/1753202639983/0200000000002clu12r4w1q3v9u2u0v6t9s3u1r5v8u4t0v6_0.idx",

"data/1753202639985/0200000000002dmv23s5x1w7u0t4t9s3u1r5v2s6w9v5u1w7_0.idx"};

// clang-format off

std::vector<std::string> index_v1_file_path = {

"data/1753202846974/0200000000007864994f6aa97288842758c2e89b03e65682_0_1753202846943.idx",

"data/1753202845724/020000000000786635407b55b72242ac167cf83cd4c598a2_0_1753202841593.idx",

"data/1753202846984/020000000000788bdd40fcf18bcaa1bbd4058ef92606e79a_0_1753202846923.idx",

"data/1753202846986/02000000000078e635407b55b72242ac167cf83cd4c598a0_0_1753202846963.idx",

"data/1753202846986/02000000000078e635407b55b72242ac167cf83cd4c598a1_0_1753202846963.idx",

"data/1753202847030/020000000000791335407b55b72242ac167cf83cd4c598a0_0_1753202844931.idx",

"data/1753202847030/020000000000791335407b55b72242ac167cf83cd4c598a1_0_1753202844931.idx",

"data/1753202847030/020000000000791335407b55b72242ac167cf83cd4c598a2_0_1753202844931.idx",

"data/1753202858558/0200000000007aed994f6aa97288842758c2e89b03e65680_0_1753202843931.idx",

"data/1753202858558/0200000000007aed994f6aa97288842758c2e89b03e65681_0_1753202843931.idx",

"data/1753202858558/0200000000007aed994f6aa97288842758c2e89b03e65682_0_1753202843931.idx",

"data/1753202858458/0200000000007afc994f6aa97288842758c2e89b03e65680_0_1753202824931.idx",

"data/1753202858458/0200000000007afc994f6aa97288842758c2e89b03e65681_0_1753202824931.idx",

"data/1753202858458/0200000000007afc994f6aa97288842758c2e89b03e65682_0_1753202824931.idx"};

// clang-format on

doris::cloud::RecyclerThreadPoolGroup thread_group;

int main(int argc, char** argv) {

auto conf_file = "doris_cloud.conf";

if (!cloud::config::init(conf_file, true)) {

std::cerr << "failed to init config file, conf=" << conf_file << std::endl;

return -1;

}

if (!cloud::init_glog("recycler")) {

std::cerr << "failed to init glog" << std::endl;

return -1;

}

using namespace std::chrono;

current_time = duration_cast<seconds>(system_clock::now().time_since_epoch()).count();

config::recycler_sleep_before_scheduling_seconds = 0; // we dont have to wait in UT

::testing::InitGoogleTest(&argc, argv);

auto s3_producer_pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);

s3_producer_pool->start();

auto recycle_tablet_pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);

recycle_tablet_pool->start();

auto group_recycle_function_pool =

std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);

group_recycle_function_pool->start();

thread_group =

RecyclerThreadPoolGroup(std::move(s3_producer_pool), std::move(recycle_tablet_pool),

std::move(group_recycle_function_pool));

return RUN_ALL_TESTS();

}

namespace doris::cloud {

TEST(RecyclerTest, WhiteBlackList) {

WhiteBlackList filter;

EXPECT_FALSE(filter.filter_out("instance1"));

EXPECT_FALSE(filter.filter_out("instance2"));

filter.reset({}, {"instance1", "instance2"});

EXPECT_TRUE(filter.filter_out("instance1"));

EXPECT_TRUE(filter.filter_out("instance2"));

EXPECT_FALSE(filter.filter_out("instance3"));

filter.reset({"instance1", "instance2"}, {});

EXPECT_FALSE(filter.filter_out("instance1"));

EXPECT_FALSE(filter.filter_out("instance2"));

EXPECT_TRUE(filter.filter_out("instance3"));

filter.reset({"instance1"}, {"instance1"}); // whitelist overrides blacklist

EXPECT_FALSE(filter.filter_out("instance1"));

EXPECT_TRUE(filter.filter_out("instance2"));

}

static std::string next_rowset_id() {

static int64_t cnt = 0;

return std::to_string(++cnt);

}

static int64_t next_small_file_txn_id() {

static std::atomic<int64_t> txn {1000};

return ++txn;

}

static doris::RowsetMetaCloudPB create_rowset(const std::string& resource_id, int64_t tablet_id,

int64_t index_id, int num_segments,

const doris::TabletSchemaCloudPB& schema,

int64_t txn_id = 0) {

doris::RowsetMetaCloudPB rowset;

rowset.set_rowset_id(0); // useless but required

rowset.set_rowset_id_v2(next_rowset_id());

rowset.set_txn_id(txn_id);

rowset.set_num_segments(num_segments);

rowset.set_tablet_id(tablet_id);

rowset.set_index_id(index_id);

rowset.set_resource_id(resource_id);

rowset.set_schema_version(schema.schema_version());

rowset.mutable_tablet_schema()->CopyFrom(schema);

return rowset;

}

static doris::RowsetMetaCloudPB create_rowset(const std::string& resource_id, int64_t tablet_id,

int64_t index_id, int num_segments,

const doris::TabletSchemaCloudPB& schema,

RowsetStatePB rowset_state, int64_t txn_id = 0) {

doris::RowsetMetaCloudPB rowset;

rowset.set_rowset_id(0); // useless but required

rowset.set_rowset_id_v2(next_rowset_id());

rowset.set_txn_id(txn_id);

rowset.set_num_segments(num_segments);

rowset.set_tablet_id(tablet_id);

rowset.set_index_id(index_id);

rowset.set_resource_id(resource_id);

rowset.set_schema_version(schema.schema_version());

rowset.mutable_tablet_schema()->CopyFrom(schema);

rowset.set_rowset_state(rowset_state);

return rowset;

}

static int create_delete_bitmaps_v1(TxnKv* txn_kv, int64_t tablet_id, std::string rowset_id) {

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

auto key = meta_delete_bitmap_key({instance_id, tablet_id, rowset_id, 0, 0});

txn->put(key, "delete_bitmap_data");

if (txn->commit() != TxnErrorCode::TXN_OK) {

return -1;

}

return 0;

}

static int create_delete_bitmaps_v2(TxnKv* txn_kv, StorageVaultAccessor* accessor,

int64_t tablet_id, std::string rowset_id) {

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

DeleteBitmapPB delete_bitmap;

DeleteBitmapStoragePB delete_bitmap_storage;

delete_bitmap_storage.set_store_in_fdb(false);

auto key = versioned::meta_delete_bitmap_key({instance_id, tablet_id, rowset_id});

std::string val;

delete_bitmap_storage.SerializeToString(&val);

txn->put(key, val);

if (txn->commit() != TxnErrorCode::TXN_OK) {

return -1;

}

return accessor->put_file(delete_bitmap_path(tablet_id, rowset_id), "");

}

static int create_recycle_rowset(TxnKv* txn_kv, StorageVaultAccessor* accessor,

const doris::RowsetMetaCloudPB& rowset, RecycleRowsetPB::Type type,

bool write_schema_kv, bool enable_create_delete_bitmaps_v2 = false,

bool enable_create_delete_bitmaps_v1 = false) {

std::string key;

std::string val;

RecycleRowsetKeyInfo key_info {instance_id, rowset.tablet_id(), rowset.rowset_id_v2()};

recycle_rowset_key(key_info, &key);

RecycleRowsetPB rowset_pb;

rowset_pb.set_creation_time(current_time);

if (type != RecycleRowsetPB::UNKNOWN) {

rowset_pb.set_type(type);

rowset_pb.mutable_rowset_meta()->CopyFrom(rowset);

if (write_schema_kv) { // Detach schema

rowset_pb.mutable_rowset_meta()->set_allocated_tablet_schema(nullptr);

}

} else { // old version RecycleRowsetPB

rowset_pb.set_tablet_id(rowset.tablet_id());

rowset_pb.set_resource_id(rowset.resource_id());

}

rowset_pb.SerializeToString(&val);

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

txn->put(key, val);

std::string schema_key, schema_val;

if (write_schema_kv) {

meta_schema_key({instance_id, rowset.index_id(), rowset.schema_version()}, &schema_key);

rowset.tablet_schema().SerializeToString(&schema_val);

txn->put(schema_key, schema_val);

auto versioned_schema_key = versioned::meta_schema_key(

{instance_id, rowset.index_id(), rowset.schema_version()});

doris::TabletSchemaCloudPB tablet_schema(rowset.tablet_schema());

versioned::document_put(txn.get(), versioned_schema_key, std::move(tablet_schema));

}

if (txn->commit() != TxnErrorCode::TXN_OK) {

return -1;

}

for (int i = 0; i < rowset.num_segments(); ++i) {

auto path = segment_path(rowset.tablet_id(), rowset.rowset_id_v2(), i);

accessor->put_file(path, "");

for (auto& index : rowset.tablet_schema().index()) {

auto path = inverted_index_path_v1(rowset.tablet_id(), rowset.rowset_id_v2(), i,

index.index_id(), index.index_suffix_name());

accessor->put_file(path, "");

}

}

if (enable_create_delete_bitmaps_v2) {

auto ret = create_delete_bitmaps_v2(txn_kv, accessor, rowset.tablet_id(),

rowset.rowset_id_v2());

if (ret != 0) {

return ret;

}

}

if (enable_create_delete_bitmaps_v1) {

auto ret = create_delete_bitmaps_v1(txn_kv, rowset.tablet_id(), rowset.rowset_id_v2());

if (ret != 0) {

return ret;

}

}

return 0;

}

static size_t count_recycle_rowsets(TxnKv* txn_kv) {

std::unique_ptr<Transaction> txn;

EXPECT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

std::unique_ptr<RangeGetIterator> it;

auto begin_key = recycle_key_prefix(instance_id);

auto end_key = recycle_key_prefix(instance_id + '\xff');

EXPECT_EQ(txn->get(begin_key, end_key, &it), TxnErrorCode::TXN_OK);

return it->size();

}

static size_t count_recycle_rowsets(TxnKv* txn_kv, int64_t tablet_id) {

std::unique_ptr<Transaction> txn;

EXPECT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);

std::unique_ptr<RangeGetIterator> it;

auto begin_key = recycle_rowset_key({instance_id, tablet_id, ""});

auto end_key = recycle_rowset_key({instance_id, tablet_id, "\xff"});

EXPECT_EQ(txn->get(begin_key, end_key, &it), TxnErrorCode::TXN_OK);

return it->size();

}

static int create_tmp_rowset(TxnKv* txn_kv, StorageVaultAccessor* accessor,

const doris::RowsetMetaCloudPB& rowset, bool write_schema_kv,

bool is_inverted_idx_v2 = false,

bool enable_create_delete_bitmaps_v2 = false) {

std::string key, val;

meta_rowset_tmp_key({instance_id, rowset.txn_id(), rowset.tablet_id()}, &key);

if (write_schema_kv) {

auto rowset_copy = rowset;

rowset_copy.clear_tablet_schema();

rowset_copy.SerializeToString(&val);

} else {

rowset.SerializeToString(&val);

}

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

txn->put(key, val);

std::string schema_key, schema_val;

if (write_schema_kv) {

meta_schema_key({instance_id, rowset.index_id(), rowset.schema_version()}, &schema_key);

rowset.tablet_schema().SerializeToString(&schema_val);

txn->put(schema_key, schema_val);

}

// Add rowset ref count key

std::string rowset_ref_count_key = versioned::data_rowset_ref_count_key(

{instance_id, rowset.tablet_id(), rowset.rowset_id_v2()});

LOG(INFO) << "add rowset ref count key, instance_id=" << instance_id

<< "key=" << hex(rowset_ref_count_key);

txn->atomic_add(rowset_ref_count_key, 1);

if (txn->commit() != TxnErrorCode::TXN_OK) {

return -1;

}

for (int i = 0; i < rowset.num_segments(); ++i) {

auto path = segment_path(rowset.tablet_id(), rowset.rowset_id_v2(), i);

accessor->put_file(path, path);

for (auto& index : rowset.tablet_schema().index()) {

std::string path;

if (!is_inverted_idx_v2) {

path = inverted_index_path_v1(rowset.tablet_id(), rowset.rowset_id_v2(), i,

index.index_id(), index.index_suffix_name());

} else {

path = inverted_index_path_v2(rowset.tablet_id(), rowset.rowset_id_v2(), i);

}

accessor->put_file(path, path);

}

}

if (enable_create_delete_bitmaps_v2) {

return create_delete_bitmaps_v2(txn_kv, accessor, rowset.tablet_id(),

rowset.rowset_id_v2());

}

return 0;

}

static int create_committed_rowset_by_real_index_v2_file(TxnKv* txn_kv,

StorageVaultAccessor* accessor,

const std::string& resource_id,

const std::string& file_path,

int64_t version = 1) {

std::string val;

std::unique_ptr<Transaction> txn;

TxnErrorCode err;

// Parse file path to extract tablet_id and rowset_id

// Expected format: data/{tablet_id}/{rowset_id}_{segment_id}.{ext}

std::vector<std::string> path_parts;

butil::SplitString(file_path, '/', &path_parts);

if (path_parts.size() < 3 || path_parts[0] != "data") {

LOG(WARNING) << "Invalid file path format: " << file_path;

return -1;

}

int64_t tablet_id = std::stoll(path_parts[1]);

std::string filename = path_parts[2];

// Extract rowset_id and segment_id from filename

size_t underscore_pos = filename.find_last_of('_');

size_t dot_pos = filename.find_last_of('.');

if (underscore_pos == std::string::npos || dot_pos == std::string::npos ||

underscore_pos >= dot_pos) {

LOG(WARNING) << "Invalid filename format: " << filename;

return -1;

}

std::string rowset_id = filename.substr(0, underscore_pos);

std::string segment_str = filename.substr(underscore_pos + 1, dot_pos - underscore_pos - 1);

std::string extension = filename.substr(dot_pos + 1);

int64_t segment_id = stoll(segment_str);

int64_t tablet_index_id = tablet_id + 10;

// take the last 4 digits of tablet_id as the unique identifier

int64_t schema_version =

std::atoll(path_parts[1].substr(path_parts[1].size() - 4).c_str()) + version;

// Create rowset meta data

MetaRowsetKeyInfo key_info {instance_id, tablet_id, version};

std::string rowset_meta_key = meta_rowset_key(key_info);

doris::RowsetMetaCloudPB rowset_pb;

rowset_pb.set_rowset_id(0); // useless but required

rowset_pb.set_rowset_id_v2(rowset_id);

rowset_pb.set_num_segments(segment_id + 1); // segment_id is 0-based

rowset_pb.set_tablet_id(tablet_id);

rowset_pb.set_resource_id(resource_id);

rowset_pb.set_creation_time(current_time);

rowset_pb.set_schema_version(schema_version);

rowset_pb.SerializeToString(&val);

err = txn_kv->create_txn(&txn);

EXPECT_EQ(err, TxnErrorCode::TXN_OK) << err;

txn->put(rowset_meta_key, val);

// Create tablet index meta data

TabletIndexPB tablet_index;

tablet_index.set_index_id(tablet_index_id);

tablet_index.set_tablet_id(tablet_id);

std::string tablet_index_key = meta_tablet_idx_key({instance_id, tablet_id});

tablet_index.SerializeToString(&val);

txn->put(tablet_index_key, val);

// Create tablet schema if dealing with index files

if (extension == "idx") {

std::string tablet_schema_key =

meta_schema_key({instance_id, tablet_index_id, schema_version});

doris::TabletSchemaCloudPB tablet_schema;

tablet_schema.set_inverted_index_storage_format(InvertedIndexStorageFormatPB::V2);

tablet_schema.set_schema_version(schema_version);

auto index = tablet_schema.add_index();

index->set_index_id(tablet_schema.index().size());

index->set_index_type(IndexType::INVERTED);

MetaServiceCode code;

std::string msg;

put_schema_kv(code, msg, txn.get(), tablet_schema_key, tablet_schema);

}

err = txn->commit();

EXPECT_EQ(err, TxnErrorCode::TXN_OK) << err;

std::string segment_path = file_path.substr(0, file_path.size() - 4) + ".dat";

accessor->put_file(segment_path, "");

accessor->put_file(file_path, "");

return 0;

}

static int create_committed_rowset_by_real_index_v1_file(TxnKv* txn_kv,

StorageVaultAccessor* accessor,

const std::string& resource_id,

const std::string& file_path,

size_t& version) {

std::string val;

std::unique_ptr<Transaction> txn;

TxnErrorCode err;

// Parse file path to extract tablet_id and rowset_id

// Expected format: data/{tablet_id}/{rowset_id}_{segment_id}_{index_id}{suffix}.idx

std::vector<std::string> path_parts;

butil::SplitString(file_path, '/', &path_parts);

if (path_parts.size() < 3 || path_parts[0] != "data") {

LOG(WARNING) << "Invalid file path format: " << file_path;

return -1;

}

int64_t tablet_id = std::stoll(path_parts[1]);

std::string filename = path_parts[2];

// Extract rowset_id, segment_id, index_id, and suffix from filename

// Format: {rowset_id}_{segment_id}_{index_id}{suffix}.idx

size_t first_underscore_pos = filename.find('_');

size_t second_underscore_pos = filename.find('_', first_underscore_pos + 1);

size_t dot_pos = filename.find_last_of('.');

if (first_underscore_pos == std::string::npos || second_underscore_pos == std::string::npos ||

dot_pos == std::string::npos || first_underscore_pos >= second_underscore_pos ||

second_underscore_pos >= dot_pos) {

LOG(WARNING) << "Invalid filename format: " << filename;

return -1;

}

std::string rowset_id = filename.substr(0, first_underscore_pos);

std::string segment_str = filename.substr(first_underscore_pos + 1,

second_underscore_pos - first_underscore_pos - 1);

std::string remaining =

filename.substr(second_underscore_pos + 1, dot_pos - second_underscore_pos - 1);

std::string extension = filename.substr(dot_pos + 1);

// Parse index_id and suffix from remaining part

// Format: {index_id}{suffix} or just {index_id}

std::string index_id_str = remaining;

std::string index_suffix = "";

int segment_id = stoll(segment_str);

int64_t index_id = std::stoll(index_id_str);

int64_t tablet_index_id = tablet_id + 10;

int64_t schema_version =

std::atoll(path_parts[1].substr(path_parts[1].size() - 4).c_str()) + version;

// Create rowset meta data

MetaRowsetKeyInfo key_info {instance_id, tablet_id, version};

std::string rowset_meta_key = meta_rowset_key(key_info);

doris::RowsetMetaCloudPB rowset_pb;

rowset_pb.set_rowset_id(0); // useless but required

rowset_pb.set_rowset_id_v2(rowset_id);

rowset_pb.set_num_segments(segment_id + 1); // segment_id is 0-based

rowset_pb.set_tablet_id(tablet_id);

rowset_pb.set_resource_id(resource_id);

rowset_pb.set_creation_time(current_time);

rowset_pb.set_schema_version(schema_version);

rowset_pb.SerializeToString(&val);

err = txn_kv->create_txn(&txn);

EXPECT_EQ(err, TxnErrorCode::TXN_OK) << err;

txn->put(rowset_meta_key, val);

// Create tablet index meta data

TabletIndexPB tablet_index;

tablet_index.set_index_id(tablet_index_id);

tablet_index.set_tablet_id(tablet_id);

std::string tablet_index_key = meta_tablet_idx_key({instance_id, tablet_id});

tablet_index.SerializeToString(&val);

txn->put(tablet_index_key, val);

// Create tablet schema if dealing with index files

if (extension == "idx") {

std::string tablet_schema_key =

meta_schema_key({instance_id, tablet_index_id, schema_version});

doris::TabletSchemaCloudPB tablet_schema;

tablet_schema.set_inverted_index_storage_format(InvertedIndexStorageFormatPB::V1);

tablet_schema.set_schema_version(schema_version);

auto index = tablet_schema.add_index();

index->set_index_id(index_id);

index->set_index_type(IndexType::INVERTED);

if (!index_suffix.empty()) {

index->set_index_suffix_name(index_suffix);

}

MetaServiceCode code;

std::string msg;

put_schema_kv(code, msg, txn.get(), tablet_schema_key, tablet_schema);

}

err = txn->commit();

EXPECT_EQ(err, TxnErrorCode::TXN_OK) << err;

std::string segment_path = fmt::format("data/{}/{}_{}.dat", tablet_id, rowset_id, segment_id);

accessor->put_file(segment_path, "");

accessor->put_file(file_path, "");

return 0;

}

static int create_committed_rowset(TxnKv* txn_kv, StorageVaultAccessor* accessor,

const std::string& resource_id, int64_t tablet_id,

int64_t version, int64_t index_id, int num_segments = 1,

int num_inverted_indexes = 1,

bool enable_create_delete_bitmaps_v2 = false) {

std::string key;

std::string val;

int64_t schema_version = tablet_id + version + num_inverted_indexes + 1;

auto rowset_id = next_rowset_id();

MetaRowsetKeyInfo key_info {instance_id, tablet_id, version};

meta_rowset_key(key_info, &key);

doris::RowsetMetaCloudPB rowset_pb;

rowset_pb.set_rowset_id(0); // useless but required

rowset_pb.set_rowset_id_v2(rowset_id);

rowset_pb.set_num_segments(num_segments);

rowset_pb.set_tablet_id(tablet_id);

rowset_pb.set_resource_id(resource_id);

rowset_pb.set_creation_time(current_time);

rowset_pb.set_schema_version(schema_version);

rowset_pb.set_index_id(index_id);

rowset_pb.SerializeToString(&val);

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

txn->put(key, val);

std::string key1;

std::string val1;

MetaTabletIdxKeyInfo key_info1 {instance_id, tablet_id};

meta_tablet_idx_key(key_info1, &key1);

TabletIndexPB tablet_idx_pb;

tablet_idx_pb.set_db_id(db_id);

tablet_idx_pb.set_index_id(index_id);

tablet_idx_pb.set_tablet_id(tablet_id);

if (!tablet_idx_pb.SerializeToString(&val1)) {

return -1;

}

txn->put(key1, val1);

int64_t tablet_schema_version_cnt = tablet_id + version + 1;

for (size_t i = 0; i < num_inverted_indexes; i++) {

std::string tablet_schema_key =

meta_schema_key({instance_id, index_id, ++tablet_schema_version_cnt});

doris::TabletSchemaCloudPB tablet_schema;

tablet_schema.set_inverted_index_storage_format(InvertedIndexStorageFormatPB::V1);

tablet_schema.set_schema_version(tablet_schema_version_cnt);

auto index = tablet_schema.add_index();

index->set_index_id(i);

index->set_index_type(IndexType::INVERTED);

MetaServiceCode code;

std::string msg;

put_schema_kv(code, msg, txn.get(), tablet_schema_key, tablet_schema);

}

if (txn->commit() != TxnErrorCode::TXN_OK) {

return -1;

}

for (int i = 0; i < num_segments; ++i) {

auto path = segment_path(tablet_id, rowset_id, i);

accessor->put_file(path, "");

for (int j = 0; j < num_inverted_indexes; ++j) {

auto path = inverted_index_path_v1(tablet_id, rowset_id, i, j, "");

accessor->put_file(path, "");

}

}

if (enable_create_delete_bitmaps_v2) {

return create_delete_bitmaps_v2(txn_kv, accessor, tablet_id, rowset_id);

}

return 0;

}

static int create_committed_rowset_with_rowset_id(TxnKv* txn_kv, StorageVaultAccessor* accessor,

const std::string& resource_id, int64_t tablet_id,

int64_t start_version, int64_t end_version,

std::string rowset_id, bool segments_overlap,

int num_segments,

int64_t create_time = current_time) {

std::string key;

std::string val;

MetaRowsetKeyInfo key_info {instance_id, tablet_id, end_version};

meta_rowset_key(key_info, &key);

doris::RowsetMetaCloudPB rowset_pb;

rowset_pb.set_rowset_id(0); // useless but required

rowset_pb.set_rowset_id_v2(rowset_id);

rowset_pb.set_num_segments(num_segments);

rowset_pb.set_tablet_id(tablet_id);

rowset_pb.set_resource_id(resource_id);

rowset_pb.set_creation_time(create_time);

rowset_pb.set_start_version(start_version);

rowset_pb.set_end_version(end_version);

rowset_pb.set_segments_overlap_pb(segments_overlap ? OVERLAPPING : NONOVERLAPPING);

rowset_pb.SerializeToString(&val);

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

txn->put(key, val);

if (txn->commit() != TxnErrorCode::TXN_OK) {

return -1;

}

for (int i = 0; i < num_segments; ++i) {

auto path = segment_path(tablet_id, rowset_id, i);

accessor->put_file(path, "");

}

return 0;

}

static int create_restore_job_rowset(TxnKv* txn_kv, StorageVaultAccessor* accessor,

const std::string& resource_id, int64_t tablet_id,

int64_t version, int num_segments = 1,

int num_inverted_indexes = 1) {

std::string key;

std::string val;

auto rowset_id = next_rowset_id();

JobRestoreRowsetKeyInfo key_info {instance_id, tablet_id, version};

job_restore_rowset_key(key_info, &key);

doris::RowsetMetaCloudPB rowset_pb;

rowset_pb.set_rowset_id(0); // useless but required

rowset_pb.set_rowset_id_v2(rowset_id);

rowset_pb.set_num_segments(num_segments);

rowset_pb.set_tablet_id(tablet_id);

rowset_pb.set_resource_id(resource_id);

rowset_pb.set_creation_time(current_time);

if (num_inverted_indexes > 0) {

auto schema = rowset_pb.mutable_tablet_schema();

for (int i = 0; i < num_inverted_indexes; ++i) {

schema->add_index()->set_index_id(i);

}

}

rowset_pb.SerializeToString(&val);

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

txn->put(key, val);

LOG(INFO) << "rowset_pb:" << rowset_pb.DebugString() << " key=" << hex(key);

if (txn->commit() != TxnErrorCode::TXN_OK) {

return -1;

}

for (int i = 0; i < num_segments; ++i) {

auto path = segment_path(tablet_id, rowset_id, i);

accessor->put_file(path, "");

for (int j = 0; j < num_inverted_indexes; ++j) {

auto path = inverted_index_path_v1(tablet_id, rowset_id, i, j, "");

accessor->put_file(path, "");

}

}

return 0;

}

static int create_restore_job_tablet(TxnKv* txn_kv, int64_t tablet_id,

RestoreJobCloudPB::State state) {

std::string key;

std::string val;

JobRestoreTabletKeyInfo key_info {instance_id, tablet_id};

job_restore_tablet_key(key_info, &key);

RestoreJobCloudPB restore_job_pb;

restore_job_pb.set_tablet_id(tablet_id);

restore_job_pb.set_ctime_s(::time(nullptr) - 3600);

restore_job_pb.set_expired_at_s(0);

restore_job_pb.set_state(state);

restore_job_pb.SerializeToString(&val);

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

txn->put(key, val);

LOG(INFO) << "restore_job_pb:" << restore_job_pb.DebugString() << " key=" << hex(key);

if (txn->commit() != TxnErrorCode::TXN_OK) {

return -1;

}

return 0;

}

static void create_delete_bitmaps(Transaction* txn, int64_t tablet_id, std::string rowset_id,

int64_t start_version, int64_t end_version) {

for (int64_t ver {start_version}; ver <= end_version; ver++) {

auto key = meta_delete_bitmap_key({instance_id, tablet_id, rowset_id, ver, 0});

std::string val {"test_data"};

txn->put(key, val);

}

}

static void create_delete_bitmaps(Transaction* txn, int64_t tablet_id, std::string rowset_id,

std::vector<int64_t> versions, int64_t segment_num = 1) {

for (int64_t ver : versions) {

for (int64_t segment_id {0}; segment_id < segment_num; segment_id++) {

auto key = meta_delete_bitmap_key({instance_id, tablet_id, rowset_id, ver, segment_id});

if (segment_id % 2 == 0) {

std::string val {"test_data"};

txn->put(key, val);

} else {

std::string val(1000, 'A');

cloud::blob_put(txn, key, val, 0, 300);

}

}

}

}

static int create_tablet(TxnKv* txn_kv, int64_t table_id, int64_t index_id, int64_t partition_id,

int64_t tablet_id, bool is_mow = false, bool has_sequence_col = false) {

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

doris::TabletMetaCloudPB tablet_meta;

tablet_meta.set_tablet_id(tablet_id);

tablet_meta.set_enable_unique_key_merge_on_write(is_mow);

if (has_sequence_col) {

tablet_meta.mutable_schema()->set_sequence_col_idx(1);

}

auto val = tablet_meta.SerializeAsString();

auto key = meta_tablet_key({instance_id, table_id, index_id, partition_id, tablet_id});

txn->put(key, val);

key = stats_tablet_key({instance_id, table_id, index_id, partition_id, tablet_id});

txn->put(key, val); // val is not necessary

key = job_tablet_key({instance_id, table_id, index_id, partition_id, tablet_id});

txn->put(key, val); // val is not necessary

// mock tablet index

TabletIndexPB tablet_idx_pb;

tablet_idx_pb.set_db_id(db_id);

tablet_idx_pb.set_table_id(table_id);

tablet_idx_pb.set_index_id(index_id);

tablet_idx_pb.set_partition_id(partition_id);

tablet_idx_pb.set_tablet_id(tablet_id);

auto idx_val = tablet_idx_pb.SerializeAsString();

key = meta_tablet_idx_key({instance_id, tablet_id});

txn->put(key, idx_val);

LOG(INFO) << "tablet_idx_pb:" << tablet_idx_pb.DebugString() << " key=" << hex(key);

if (txn->commit() != TxnErrorCode::TXN_OK) {

return -1;

}

return 0;

}

static int create_partition_meta_and_index_keys(TxnKv* txn_kv, int64_t db_id, int64_t table_id,

int64_t partition_id) {

std::string meta_key = versioned::meta_partition_key({instance_id, partition_id});

std::string meta_index_key = versioned::partition_index_key({instance_id, partition_id});

std::string meta_inverted_index_key =

versioned::partition_inverted_index_key({instance_id, db_id, table_id, partition_id});

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

std::string val;

PartitionIndexPB partition_index_pb;

partition_index_pb.set_db_id(db_id);

partition_index_pb.set_table_id(table_id);

partition_index_pb.SerializeToString(&val);

versioned_put(txn.get(), meta_key, "");

txn->put(meta_index_key, partition_index_pb.SerializeAsString());

txn->put(meta_inverted_index_key, "");

if (txn->commit() != TxnErrorCode::TXN_OK) {

return -1;

}

return 0;

}

static int create_recycle_partiton(TxnKv* txn_kv, int64_t table_id, int64_t partition_id,

const std::vector<int64_t>& index_ids) {

std::string key;

std::string val;

RecyclePartKeyInfo key_info {instance_id, partition_id};

recycle_partition_key(key_info, &key);

RecyclePartitionPB partition_pb;

partition_pb.set_db_id(db_id);

partition_pb.set_table_id(table_id);

for (auto index_id : index_ids) {

partition_pb.add_index_id(index_id);

}

partition_pb.set_creation_time(current_time);

partition_pb.set_state(RecyclePartitionPB::DROPPED);

partition_pb.SerializeToString(&val);

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

txn->put(key, val);

if (txn->commit() != TxnErrorCode::TXN_OK) {

return -1;

}

return 0;

}

static int create_partition_version_kv(TxnKv* txn_kv, int64_t table_id, int64_t partition_id) {

auto key = partition_version_key({instance_id, db_id, table_id, partition_id});

VersionPB version;

version.set_version(1);

auto val = version.SerializeAsString();

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

txn->put(key, val);

if (txn->commit() != TxnErrorCode::TXN_OK) {

return -1;

}

return 0;

}

static int create_partition_version_with_pending_txn_kv(TxnKv* txn_kv, int64_t table_id,

int64_t partition_id, int64_t txn_id) {

auto key = partition_version_key({instance_id, db_id, table_id, partition_id});

VersionPB version;

version.set_version(1);

version.add_pending_txn_ids(txn_id);

auto val = version.SerializeAsString();

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

txn->put(key, val);

if (txn->commit() != TxnErrorCode::TXN_OK) {

return -1;

}

return 0;

}

static int create_delete_bitmap_update_lock_kv(TxnKv* txn_kv, int64_t table_id, int64_t lock_id,

int64_t initiator, int64_t expiration) {

auto key = meta_delete_bitmap_update_lock_key({instance_id, table_id, -1});

DeleteBitmapUpdateLockPB lock_info;

lock_info.set_lock_id(lock_id);

auto val = lock_info.SerializeAsString();

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

txn->put(key, val);

std::string tablet_job_key = mow_tablet_job_key({instance_id, table_id, initiator});

std::string tablet_job_val;

MowTabletJobPB mow_tablet_job;

mow_tablet_job.set_expiration(expiration);

mow_tablet_job.SerializeToString(&tablet_job_val);

txn->put(tablet_job_key, tablet_job_val);

if (txn->commit() != TxnErrorCode::TXN_OK) {

return -1;

}

return 0;

}

static int create_table_version_kv(TxnKv* txn_kv, int64_t table_id, int64_t version = 1) {

auto key = table_version_key({instance_id, db_id, table_id});

std::string val(sizeof(int64_t), 0);

*reinterpret_cast<int64_t*>(val.data()) = version;

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

txn->put(key, val);

if (txn->commit() != TxnErrorCode::TXN_OK) {

return -1;

}

return 0;

}

static int create_txn_label_kv(TxnKv* txn_kv, std::string label, int64_t db_id) {

std::string txn_label_key_;

std::string txn_label_val;

auto keyinfo = TxnLabelKeyInfo({instance_id, db_id, label});

txn_label_key(keyinfo, &txn_label_key_);

std::unique_ptr<Transaction> txn;

if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {

return -1;

}

txn->put(txn_label_key_, label);