// 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 "meta-store/clone_chain_reader.h"

#include <gen_cpp/cloud.pb.h>

#include <gen_cpp/olap_file.pb.h>

#include <gtest/gtest-death-test.h>

#include <gtest/gtest.h>

#include <memory>

#include "common/config.h"

#include "common/logging.h"

#include "common/util.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 "meta-store/versioned_value.h"

#include "resource-manager/resource_manager.h"

using namespace doris::cloud;

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

config::log_dir = "./log/";

if (!doris::cloud::init_glog("clone_chain_reader_test")) {

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

return -1;

}

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

return RUN_ALL_TESTS();

}

// Convert a string to a hex-escaped string.

// A non-displayed character is represented as \xHH where HH is the hexadecimal value of the character.

// A displayed character is represented as itself.

std::string escape_hex(std::string_view data) {

std::string result;

for (char c : data) {

if (isprint(c)) {

result += c;

} else {

result += fmt::format("\\x{:02x}", static_cast<unsigned char>(c));

}

}

return result;

}

std::string dump_range(TxnKv* txn_kv, std::string_view begin = "", std::string_view end = "\xFF") {

std::unique_ptr<Transaction> txn;

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

return "Failed to create dump range transaction";

}

FullRangeGetOptions opts;

opts.txn = txn.get();

auto iter = txn_kv->full_range_get(std::string(begin), std::string(end), std::move(opts));

std::string buffer;

for (auto&& kv = iter->next(); kv.has_value(); kv = iter->next()) {

buffer +=

fmt::format("Key: {}, Value: {}\n", escape_hex(kv->first), escape_hex(kv->second));

}

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

return buffer;

}

class MockResourceManager : public ResourceManager {

public:

using ResourceManager::ResourceManager;

~MockResourceManager() override = default;

void add_instance_source_snapshot_info(const std::string& instance_id,

const std::string& source_instance_id,

const Versionstamp& source_snapshot_version) {

instance_source_snapshot_info_[instance_id] =

std::make_pair(source_instance_id, source_snapshot_version);

}

bool get_source_snapshot_info(const std::string& instance_id, std::string* source_instance_id,

Versionstamp* source_snapshot_version) override {

auto it = instance_source_snapshot_info_.find(instance_id);

if (it == instance_source_snapshot_info_.end()) {

return false;

}

*source_instance_id = it->second.first;

*source_snapshot_version = it->second.second;

return true;

}

std::unordered_map<std::string, std::pair<std::string, Versionstamp>>

instance_source_snapshot_info_;

};

class CloneChainReaderTest : public ::testing::Test {

public:

void SetUp() override {

txn_kv_ = std::make_shared<MemTxnKv>();

auto resource_mgr = std::make_unique<MockResourceManager>(txn_kv_);

// A (1000) -> B (2000) -> C (3000)

snapshot_versions_ = {1000, 2000, 3000};

instance_ids_ = {"A", "B", "C"};

// A has no source

for (size_t i = 1; i < instance_ids_.size(); ++i) {

std::string instance_id = instance_ids_[i];

resource_mgr->add_instance_source_snapshot_info(instance_id, instance_ids_[i - 1],

snapshot_versions_[i - 1]);

}

resource_mgr_ = std::move(resource_mgr);

}

void TearDown() override {}

protected:

std::shared_ptr<TxnKv> txn_kv_;

std::unique_ptr<ResourceManager> resource_mgr_;

std::vector<int64_t> snapshot_versions_;

std::vector<std::string> instance_ids_;

};

TEST_F(CloneChainReaderTest, GetTableVersion) {

std::string instance_id = instance_ids_[2]; // C

Versionstamp snapshot_version = snapshot_versions_[2];

CloneChainReader reader(instance_id, snapshot_version, txn_kv_.get(), resource_mgr_.get());

// Insert table version in instance A

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string table_version_key = versioned::table_version_key({instance_ids_[0], 1});

std::string key = encode_versioned_key(table_version_key, Versionstamp(100, 1));

txn->put(key, "");

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

}

// Get table version from instance C

{

Versionstamp table_version;

ASSERT_EQ(reader.get_table_version(1, &table_version), TxnErrorCode::TXN_OK);

ASSERT_EQ(table_version, Versionstamp(100, 1));

}

// Get non-existing table version

{

Versionstamp table_version;

ASSERT_EQ(reader.get_table_version(2, &table_version), TxnErrorCode::TXN_KEY_NOT_FOUND);

}

// Insert a large version in instance B

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string table_version_key = versioned::table_version_key({instance_ids_[1], 1});

std::string key = encode_versioned_key(

table_version_key, Versionstamp(snapshot_versions_[1], 1)); // a large version

txn->put(key, "");

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

}

// Get table version from instance C

{

Versionstamp table_version;

ASSERT_EQ(reader.get_table_version(1, &table_version), TxnErrorCode::TXN_OK);

ASSERT_EQ(table_version, Versionstamp(100, 1))

<< "expect: 100, 1, but got: " << table_version.version() << ", "

<< table_version.order();

}

}

TEST_F(CloneChainReaderTest, GetPartitionVersion) {

std::string instance_id = instance_ids_[2]; // C

Versionstamp snapshot_version = snapshot_versions_[2];

CloneChainReader reader(instance_id, snapshot_version, txn_kv_.get(), resource_mgr_.get());

// Case 1: No data exists anywhere - should return NOT_FOUND

{

VersionPB version_pb;

Versionstamp partition_version;

ASSERT_EQ(reader.get_partition_version(1001, &version_pb, &partition_version),

TxnErrorCode::TXN_KEY_NOT_FOUND);

}

// Case 2: Insert partition version in instance A

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string partition_version_key =

versioned::partition_version_key({instance_ids_[0], 1001});

VersionPB version_pb;

version_pb.set_version(100);

std::string key = encode_versioned_key(partition_version_key, Versionstamp(100, 1));

txn->put(key, version_pb.SerializeAsString());

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

}

// Case 2: Read from instance C should find data from A through clone chain

{

VersionPB version_pb;

Versionstamp partition_version;

ASSERT_EQ(reader.get_partition_version(1001, &version_pb, &partition_version),

TxnErrorCode::TXN_OK);

ASSERT_EQ(version_pb.version(), 100);

ASSERT_EQ(partition_version, Versionstamp(100, 1));

}

// Case 3: Insert data with version > snapshot_version in instance B (should be ignored)

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string partition_version_key =

versioned::partition_version_key({instance_ids_[1], 1001});

VersionPB version_pb;

version_pb.set_version(200);

// Use a version larger than snapshot_versions_[1] (2000)

std::string key = encode_versioned_key(partition_version_key, Versionstamp(2500, 1));

txn->put(key, version_pb.SerializeAsString());

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

}

// Case 3: Should still return data from A, not the large version from B

{

VersionPB version_pb;

Versionstamp partition_version;

ASSERT_EQ(reader.get_partition_version(1001, &version_pb, &partition_version),

TxnErrorCode::TXN_OK);

ASSERT_EQ(version_pb.version(), 100) << "Should not read data with version > snapshot";

ASSERT_EQ(partition_version, Versionstamp(100, 1));

}

// Case 4: Insert valid data in instance B (within snapshot version)

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string partition_version_key =

versioned::partition_version_key({instance_ids_[1], 1001});

VersionPB version_pb;

version_pb.set_version(150);

std::string key = encode_versioned_key(partition_version_key, Versionstamp(1500, 1));

txn->put(key, version_pb.SerializeAsString());

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

}

// Case 4: Should return the first encountered data (from B)

{

VersionPB version_pb;

Versionstamp partition_version;

ASSERT_EQ(reader.get_partition_version(1001, &version_pb, &partition_version),

TxnErrorCode::TXN_OK);

ASSERT_EQ(version_pb.version(), 150)

<< "Should return first valid data encountered in chain";

ASSERT_EQ(partition_version, Versionstamp(1500, 1));

}

}

TEST_F(CloneChainReaderTest, GetTabletLoadStats) {

std::string instance_id = instance_ids_[2]; // C

Versionstamp snapshot_version = snapshot_versions_[2];

CloneChainReader reader(instance_id, snapshot_version, txn_kv_.get(), resource_mgr_.get());

// Case 1: No data exists anywhere - should return NOT_FOUND

{

TabletStatsPB tablet_stats;

Versionstamp versionstamp;

ASSERT_EQ(reader.get_tablet_load_stats(2001, &tablet_stats, &versionstamp),

TxnErrorCode::TXN_KEY_NOT_FOUND);

}

// Case 2: Insert tablet load stats in instance A

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string tablet_load_stats_key =

versioned::tablet_load_stats_key({instance_ids_[0], 2001});

TabletStatsPB tablet_stats;

tablet_stats.set_data_size(1000);

tablet_stats.set_num_rows(100);

std::string key = encode_versioned_key(tablet_load_stats_key, Versionstamp(100, 1));

txn->put(key, tablet_stats.SerializeAsString());

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

}

// Case 2: Read from instance C should find data from A through clone chain

{

TabletStatsPB tablet_stats;

Versionstamp versionstamp;

ASSERT_EQ(reader.get_tablet_load_stats(2001, &tablet_stats, &versionstamp),

TxnErrorCode::TXN_OK);

ASSERT_EQ(tablet_stats.data_size(), 1000);

ASSERT_EQ(tablet_stats.num_rows(), 100);

ASSERT_EQ(versionstamp, Versionstamp(100, 1));

}

// Case 3: Insert data with version > snapshot_version in instance B (should be ignored)

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string tablet_load_stats_key =

versioned::tablet_load_stats_key({instance_ids_[1], 2001});

TabletStatsPB tablet_stats;

tablet_stats.set_data_size(2000);

tablet_stats.set_num_rows(200);

// Use a version larger than snapshot_versions_[1] (2000)

std::string key = encode_versioned_key(tablet_load_stats_key, Versionstamp(2500, 1));

txn->put(key, tablet_stats.SerializeAsString());

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

}

// Case 3: Should still return data from A, not the large version from B

{

TabletStatsPB tablet_stats;

Versionstamp versionstamp;

ASSERT_EQ(reader.get_tablet_load_stats(2001, &tablet_stats, &versionstamp),

TxnErrorCode::TXN_OK);

ASSERT_EQ(tablet_stats.data_size(), 1000) << "Should not read data with version > snapshot";

ASSERT_EQ(tablet_stats.num_rows(), 100);

ASSERT_EQ(versionstamp, Versionstamp(100, 1));

}

// Case 4: Insert valid data in instance B (within snapshot version)

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string tablet_load_stats_key =

versioned::tablet_load_stats_key({instance_ids_[1], 2001});

TabletStatsPB tablet_stats;

tablet_stats.set_data_size(1500);

tablet_stats.set_num_rows(150);

std::string key = encode_versioned_key(tablet_load_stats_key, Versionstamp(1500, 1));

txn->put(key, tablet_stats.SerializeAsString());

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

}

// Case 4: Should return the first encountered data (from B)

{

TabletStatsPB tablet_stats;

Versionstamp versionstamp;

ASSERT_EQ(reader.get_tablet_load_stats(2001, &tablet_stats, &versionstamp),

TxnErrorCode::TXN_OK);

ASSERT_EQ(tablet_stats.data_size(), 1500)

<< "Should return first valid data encountered in chain";

ASSERT_EQ(tablet_stats.num_rows(), 150);

ASSERT_EQ(versionstamp, Versionstamp(1500, 1));

}

}

TEST_F(CloneChainReaderTest, BatchGetTabletLoadStats) {

std::string instance_id = instance_ids_[2]; // C

Versionstamp snapshot_version = snapshot_versions_[2];

CloneChainReader reader(instance_id, snapshot_version, txn_kv_.get(), resource_mgr_.get());

std::vector<int64_t> tablet_ids = {2001, 2002, 2003, 2004};

// Case 1: Test empty input

{

std::vector<int64_t> empty_ids;

std::unordered_map<int64_t, TabletStatsPB> tablet_stats;

std::unordered_map<int64_t, Versionstamp> versionstamps;

ASSERT_EQ(reader.get_tablet_load_stats(empty_ids, &tablet_stats, &versionstamps),

TxnErrorCode::TXN_OK);

ASSERT_TRUE(tablet_stats.empty());

ASSERT_TRUE(versionstamps.empty());

}

// Case 2: Test all keys not found

{

std::unordered_map<int64_t, TabletStatsPB> tablet_stats;

std::unordered_map<int64_t, Versionstamp> versionstamps;

ASSERT_EQ(reader.get_tablet_load_stats(tablet_ids, &tablet_stats, &versionstamps),

TxnErrorCode::TXN_OK);

ASSERT_TRUE(tablet_stats.empty());

ASSERT_TRUE(versionstamps.empty());

}

// Case 3: Insert some tablet load stats in instance A (skip tablet_ids[1])

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

for (size_t i = 0; i < tablet_ids.size(); ++i) {

if (i == 1) continue; // Skip tablet_ids[1]

std::string tablet_load_stats_key =

versioned::tablet_load_stats_key({instance_ids_[0], tablet_ids[i]});

TabletStatsPB tablet_stats;

tablet_stats.set_data_size(1000 * (i + 1));

tablet_stats.set_num_rows(100 * (i + 1));

std::string key = encode_versioned_key(tablet_load_stats_key, Versionstamp(100 + i, 1));

txn->put(key, tablet_stats.SerializeAsString());

}

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

}

// Case 3: Read from instance C should find partial data from A through clone chain

{

std::unordered_map<int64_t, TabletStatsPB> tablet_stats;

std::unordered_map<int64_t, Versionstamp> versionstamps;

ASSERT_EQ(reader.get_tablet_load_stats(tablet_ids, &tablet_stats, &versionstamps),

TxnErrorCode::TXN_OK);

ASSERT_EQ(tablet_stats.size(), 3); // All except tablet_ids[1]

ASSERT_EQ(versionstamps.size(), 3);

// Check min_read_version

Versionstamp min_expected = Versionstamp::max();

for (const auto& [tablet_id, versionstamp] : versionstamps) {

min_expected = std::min(min_expected, versionstamp);

}

ASSERT_EQ(reader.min_read_versionstamp(), min_expected);

for (size_t i = 0; i < tablet_ids.size(); ++i) {

if (i == 1) {

ASSERT_EQ(tablet_stats.count(tablet_ids[i]), 0);

ASSERT_EQ(versionstamps.count(tablet_ids[i]), 0);

} else {

ASSERT_EQ(tablet_stats.count(tablet_ids[i]), 1);

ASSERT_EQ(versionstamps.count(tablet_ids[i]), 1);

ASSERT_EQ(tablet_stats.at(tablet_ids[i]).data_size(), 1000 * (i + 1));

ASSERT_EQ(tablet_stats.at(tablet_ids[i]).num_rows(), 100 * (i + 1));

ASSERT_EQ(versionstamps.at(tablet_ids[i]), Versionstamp(100 + i, 1));

}

}

}

// Case 4: Insert data with version > snapshot_version in instance B (should be ignored)

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

for (int64_t tablet_id : tablet_ids) {

std::string tablet_load_stats_key =

versioned::tablet_load_stats_key({instance_ids_[1], tablet_id});

TabletStatsPB tablet_stats;

tablet_stats.set_data_size(9999);

tablet_stats.set_num_rows(999);

// Use a version larger than snapshot_versions_[1] (2000)

std::string key = encode_versioned_key(tablet_load_stats_key, Versionstamp(2500, 1));

txn->put(key, tablet_stats.SerializeAsString());

}

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

}

// Case 4: Should still return data from A, ignoring the large version from B

{

std::unordered_map<int64_t, TabletStatsPB> tablet_stats;

std::unordered_map<int64_t, Versionstamp> versionstamps;

ASSERT_EQ(reader.get_tablet_load_stats(tablet_ids, &tablet_stats, &versionstamps),

TxnErrorCode::TXN_OK);

ASSERT_EQ(tablet_stats.size(), 3); // Still missing tablet_ids[1]

for (size_t i = 0; i < tablet_ids.size(); ++i) {

if (i == 1) continue;

ASSERT_EQ(tablet_stats.at(tablet_ids[i]).data_size(), 1000 * (i + 1))

<< "Should not read data with version > snapshot";

ASSERT_EQ(tablet_stats.at(tablet_ids[i]).num_rows(), 100 * (i + 1));

}

}

// Case 5: Insert valid data in instance B (within snapshot version), including the missing one

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

for (size_t i = 0; i < tablet_ids.size(); ++i) {

std::string tablet_load_stats_key =

versioned::tablet_load_stats_key({instance_ids_[1], tablet_ids[i]});

TabletStatsPB tablet_stats;

tablet_stats.set_data_size(1500 * (i + 1));

tablet_stats.set_num_rows(150 * (i + 1));

std::string key =

encode_versioned_key(tablet_load_stats_key, Versionstamp(1500 + i, 1));

txn->put(key, tablet_stats.SerializeAsString());

}

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

}

// Case 5: Should return the first encountered data (from B for all tablets)

{

std::unordered_map<int64_t, TabletStatsPB> tablet_stats;

std::unordered_map<int64_t, Versionstamp> versionstamps;

ASSERT_EQ(reader.get_tablet_load_stats(tablet_ids, &tablet_stats, &versionstamps),

TxnErrorCode::TXN_OK);

ASSERT_EQ(tablet_stats.size(), tablet_ids.size());

ASSERT_EQ(versionstamps.size(), tablet_ids.size());

for (size_t i = 0; i < tablet_ids.size(); ++i) {

ASSERT_EQ(tablet_stats.count(tablet_ids[i]), 1);

ASSERT_EQ(tablet_stats.at(tablet_ids[i]).data_size(), 1500 * (i + 1))

<< "Should return first valid data encountered in chain";

ASSERT_EQ(tablet_stats.at(tablet_ids[i]).num_rows(), 150 * (i + 1));

ASSERT_EQ(versionstamps.at(tablet_ids[i]), Versionstamp(1500 + i, 1));

}

}

}

TEST_F(CloneChainReaderTest, GetTabletCompactStats) {

std::string instance_id = instance_ids_[2]; // C

Versionstamp snapshot_version = snapshot_versions_[2];

CloneChainReader reader(instance_id, snapshot_version, txn_kv_.get(), resource_mgr_.get());

// Case 1: No data exists anywhere - should return NOT_FOUND

{

TabletStatsPB tablet_stats;

Versionstamp versionstamp;

ASSERT_EQ(reader.get_tablet_compact_stats(3001, &tablet_stats, &versionstamp),

TxnErrorCode::TXN_KEY_NOT_FOUND);

}

// Case 2: Insert tablet compact stats in instance A

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string tablet_compact_stats_key =

versioned::tablet_compact_stats_key({instance_ids_[0], 3001});

TabletStatsPB tablet_stats;

tablet_stats.set_data_size(500);

tablet_stats.set_num_rows(50);

std::string key = encode_versioned_key(tablet_compact_stats_key, Versionstamp(100, 1));

txn->put(key, tablet_stats.SerializeAsString());

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

}

// Case 2: Read from instance C should find data from A through clone chain

{

TabletStatsPB tablet_stats;

Versionstamp versionstamp;

ASSERT_EQ(reader.get_tablet_compact_stats(3001, &tablet_stats, &versionstamp),

TxnErrorCode::TXN_OK);

ASSERT_EQ(tablet_stats.data_size(), 500);

ASSERT_EQ(tablet_stats.num_rows(), 50);

ASSERT_EQ(versionstamp, Versionstamp(100, 1));

}

// Case 3: Insert data with version > snapshot_version in instance B (should be ignored)

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string tablet_compact_stats_key =

versioned::tablet_compact_stats_key({instance_ids_[1], 3001});

TabletStatsPB tablet_stats;

tablet_stats.set_data_size(1000);

tablet_stats.set_num_rows(100);

// Use a version larger than snapshot_versions_[1] (2000)

std::string key = encode_versioned_key(tablet_compact_stats_key, Versionstamp(2500, 1));

txn->put(key, tablet_stats.SerializeAsString());

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

}

// Case 3: Should still return data from A, not the large version from B

{

TabletStatsPB tablet_stats;

Versionstamp versionstamp;

ASSERT_EQ(reader.get_tablet_compact_stats(3001, &tablet_stats, &versionstamp),

TxnErrorCode::TXN_OK);

ASSERT_EQ(tablet_stats.data_size(), 500) << "Should not read data with version > snapshot";

ASSERT_EQ(tablet_stats.num_rows(), 50);

ASSERT_EQ(versionstamp, Versionstamp(100, 1));

}

// Case 4: Insert valid data in instance B (within snapshot version)

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string tablet_compact_stats_key =

versioned::tablet_compact_stats_key({instance_ids_[1], 3001});

TabletStatsPB tablet_stats;

tablet_stats.set_data_size(750);

tablet_stats.set_num_rows(75);

std::string key = encode_versioned_key(tablet_compact_stats_key, Versionstamp(1500, 1));

txn->put(key, tablet_stats.SerializeAsString());

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

}

// Case 4: Should return the first encountered data (from B)

{

TabletStatsPB tablet_stats;

Versionstamp versionstamp;

ASSERT_EQ(reader.get_tablet_compact_stats(3001, &tablet_stats, &versionstamp),

TxnErrorCode::TXN_OK);

ASSERT_EQ(tablet_stats.data_size(), 750)

<< "Should return first valid data encountered in chain";

ASSERT_EQ(tablet_stats.num_rows(), 75);

ASSERT_EQ(versionstamp, Versionstamp(1500, 1));

}

}

TEST_F(CloneChainReaderTest, GetTabletIndex) {

std::string instance_id = instance_ids_[2]; // C

Versionstamp snapshot_version = snapshot_versions_[2];

CloneChainReader reader(instance_id, snapshot_version, txn_kv_.get(), resource_mgr_.get());

// Case 1: No data exists anywhere - should return NOT_FOUND

{

TabletIndexPB tablet_index;

ASSERT_EQ(reader.get_tablet_index(4001, &tablet_index), TxnErrorCode::TXN_KEY_NOT_FOUND);

}

// Case 2: Insert tablet index in instance A

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string tablet_index_key = versioned::tablet_index_key({instance_ids_[0], 4001});

TabletIndexPB tablet_index;

tablet_index.set_table_id(1001);

tablet_index.set_index_id(2001);

tablet_index.set_partition_id(3001);

txn->put(tablet_index_key, tablet_index.SerializeAsString());

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

}

// Case 2: Read from instance C should find data from A through clone chain

{

TabletIndexPB tablet_index;

ASSERT_EQ(reader.get_tablet_index(4001, &tablet_index), TxnErrorCode::TXN_OK);

ASSERT_EQ(tablet_index.table_id(), 1001);

ASSERT_EQ(tablet_index.index_id(), 2001);

ASSERT_EQ(tablet_index.partition_id(), 3001);

}

}

TEST_F(CloneChainReaderTest, GetPartitionIndex) {

std::string instance_id = instance_ids_[2]; // C

Versionstamp snapshot_version = snapshot_versions_[2];

CloneChainReader reader(instance_id, snapshot_version, txn_kv_.get(), resource_mgr_.get());

// Case 1: No data exists anywhere - should return NOT_FOUND

{

PartitionIndexPB partition_index;

ASSERT_EQ(reader.get_partition_index(5001, &partition_index),

TxnErrorCode::TXN_KEY_NOT_FOUND);

}

// Case 2: Insert partition index in instance A

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string partition_index_key = versioned::partition_index_key({instance_ids_[0], 5001});

PartitionIndexPB partition_index;

partition_index.set_db_id(1001);

partition_index.set_table_id(2001);

txn->put(partition_index_key, partition_index.SerializeAsString());

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

}

// Case 2: Read from instance C should find data from A through clone chain

{

PartitionIndexPB partition_index;

ASSERT_EQ(reader.get_partition_index(5001, &partition_index), TxnErrorCode::TXN_OK);

ASSERT_EQ(partition_index.db_id(), 1001);

ASSERT_EQ(partition_index.table_id(), 2001);

}

}

TEST_F(CloneChainReaderTest, GetIndexIndex) {

std::string instance_id = instance_ids_[2]; // C

Versionstamp snapshot_version = snapshot_versions_[2];

CloneChainReader reader(instance_id, snapshot_version, txn_kv_.get(), resource_mgr_.get());

// Case 1: No data exists anywhere - should return NOT_FOUND

{

IndexIndexPB index_index;

ASSERT_EQ(reader.get_index_index(6001, &index_index), TxnErrorCode::TXN_KEY_NOT_FOUND);

}

// Case 2: Insert index index in instance A

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string index_index_key = versioned::index_index_key({instance_ids_[0], 6001});

IndexIndexPB index_index;

index_index.set_db_id(1001);

index_index.set_table_id(2001);

txn->put(index_index_key, index_index.SerializeAsString());

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

}

// Case 2: Read from instance C should find data from A through clone chain

{

IndexIndexPB index_index;

ASSERT_EQ(reader.get_index_index(6001, &index_index), TxnErrorCode::TXN_OK);

ASSERT_EQ(index_index.db_id(), 1001);

ASSERT_EQ(index_index.table_id(), 2001);

}

}

TEST_F(CloneChainReaderTest, GetTableVersions) {

std::string instance_id = instance_ids_[2]; // C

Versionstamp snapshot_version = snapshot_versions_[2];

CloneChainReader reader(instance_id, snapshot_version, txn_kv_.get(), resource_mgr_.get());

// Case 1: No data exists anywhere - should return empty map

{

std::vector<int64_t> table_ids = {1, 2, 3};

std::unordered_map<int64_t, Versionstamp> table_versions;

ASSERT_EQ(reader.get_table_versions(table_ids, &table_versions), TxnErrorCode::TXN_OK);

ASSERT_TRUE(table_versions.empty());

}

// Case 2: Insert table versions in instance A

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

// Insert table 1 and 3, skip table 2

std::string table_version_key1 = versioned::table_version_key({instance_ids_[0], 1});

std::string key1 = encode_versioned_key(table_version_key1, Versionstamp(100, 1));

txn->put(key1, "");

std::string table_version_key3 = versioned::table_version_key({instance_ids_[0], 3});

std::string key3 = encode_versioned_key(table_version_key3, Versionstamp(300, 1));

txn->put(key3, "");

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

}

// Case 2: Read from instance C should find partial data from A

{

std::vector<int64_t> table_ids = {1, 2, 3};

std::unordered_map<int64_t, Versionstamp> table_versions;

ASSERT_EQ(reader.get_table_versions(table_ids, &table_versions), TxnErrorCode::TXN_OK);

ASSERT_EQ(table_versions.size(), 2);

ASSERT_EQ(table_versions[1], Versionstamp(100, 1));

ASSERT_EQ(table_versions[3], Versionstamp(300, 1));

ASSERT_EQ(table_versions.count(2), 0);

}

// Case 3: Insert data with version > snapshot_version in instance B (should be ignored)

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string table_version_key1 = versioned::table_version_key({instance_ids_[1], 1});

std::string key1 = encode_versioned_key(table_version_key1, Versionstamp(2500, 1));

txn->put(key1, "");

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

}

// Case 3: Should still return data from A, ignoring large version from B

{

std::vector<int64_t> table_ids = {1, 2, 3};

std::unordered_map<int64_t, Versionstamp> table_versions;

ASSERT_EQ(reader.get_table_versions(table_ids, &table_versions), TxnErrorCode::TXN_OK);

ASSERT_EQ(table_versions.size(), 2);

ASSERT_EQ(table_versions[1], Versionstamp(100, 1))

<< "Should not read data with version > snapshot";

ASSERT_EQ(table_versions[3], Versionstamp(300, 1));

}

// Case 4: Insert valid data in instance B (within snapshot version)

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string table_version_key1 = versioned::table_version_key({instance_ids_[1], 1});

std::string key1 = encode_versioned_key(table_version_key1, Versionstamp(150, 1));

txn->put(key1, "");

// Add missing table 2 in instance B

std::string table_version_key2 = versioned::table_version_key({instance_ids_[1], 2});

std::string key2 = encode_versioned_key(table_version_key2, Versionstamp(200, 1));

txn->put(key2, "");

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

}

// Case 4: Should return first encountered data (from B for tables 1&2, from A for table 3)

{

std::vector<int64_t> table_ids = {1, 2, 3};

std::unordered_map<int64_t, Versionstamp> table_versions;

ASSERT_EQ(reader.get_table_versions(table_ids, &table_versions), TxnErrorCode::TXN_OK);

ASSERT_EQ(table_versions.size(), 3);

ASSERT_EQ(table_versions[1], Versionstamp(150, 1))

<< "Should return first valid data encountered";

ASSERT_EQ(table_versions[2], Versionstamp(200, 1))

<< "Should find missing data in clone chain";

ASSERT_EQ(table_versions[3], Versionstamp(300, 1))

<< "Should return data from A when B doesn't have it";

}

}

TEST_F(CloneChainReaderTest, IsIndexExists) {

std::string instance_id = instance_ids_[2]; // C

Versionstamp snapshot_version = snapshot_versions_[2];

CloneChainReader reader(instance_id, snapshot_version, txn_kv_.get(), resource_mgr_.get());

// Case 1: No data exists anywhere - should return NOT_FOUND

ASSERT_EQ(reader.is_index_exists(7001), TxnErrorCode::TXN_KEY_NOT_FOUND);

// Case 2: Insert index in instance A

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string index_key = versioned::meta_index_key({instance_ids_[0], 7001});

std::string key = encode_versioned_key(index_key, Versionstamp(100, 1));

txn->put(key, "");

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

}

// Case 2: Read from instance C should find index from A through clone chain

ASSERT_EQ(reader.is_index_exists(7001), TxnErrorCode::TXN_OK);

// Case 3: Insert data with version > snapshot_version in instance B (should be ignored)

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string index_key = versioned::meta_index_key({instance_ids_[1], 7002});

// Use a version larger than snapshot_versions_[1] (2000)

std::string key = encode_versioned_key(index_key, Versionstamp(2500, 1));

txn->put(key, "");

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

}

// Case 3: Should not find index with large version from B

ASSERT_EQ(reader.is_index_exists(7002), TxnErrorCode::TXN_KEY_NOT_FOUND)

<< "Should not read data with version > snapshot";

// Case 4: Insert valid data in instance B (within snapshot version)

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string index_key = versioned::meta_index_key({instance_ids_[1], 7003});

std::string key = encode_versioned_key(index_key, Versionstamp(1500, 1));

txn->put(key, "");

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

}

// Case 4: Should find the first encountered data (from B)

ASSERT_EQ(reader.is_index_exists(7003), TxnErrorCode::TXN_OK)

<< "Should return first valid data encountered in chain";

}

TEST_F(CloneChainReaderTest, IsPartitionExists) {

std::string instance_id = instance_ids_[2]; // C

Versionstamp snapshot_version = snapshot_versions_[2];

CloneChainReader reader(instance_id, snapshot_version, txn_kv_.get(), resource_mgr_.get());

// Case 1: No data exists anywhere - should return NOT_FOUND

ASSERT_EQ(reader.is_partition_exists(8001), TxnErrorCode::TXN_KEY_NOT_FOUND);

// Case 2: Insert partition in instance A

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string partition_key = versioned::meta_partition_key({instance_ids_[0], 8001});

std::string key = encode_versioned_key(partition_key, Versionstamp(100, 1));

txn->put(key, "");

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

}

// Case 2: Read from instance C should find partition from A through clone chain

ASSERT_EQ(reader.is_partition_exists(8001), TxnErrorCode::TXN_OK);

// Case 3: Insert data with version > snapshot_version in instance B (should be ignored)

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string partition_key = versioned::meta_partition_key({instance_ids_[1], 8002});

// Use a version larger than snapshot_versions_[1] (2000)

std::string key = encode_versioned_key(partition_key, Versionstamp(2500, 1));

txn->put(key, "");

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

}

// Case 3: Should not find partition with large version from B

ASSERT_EQ(reader.is_partition_exists(8002), TxnErrorCode::TXN_KEY_NOT_FOUND)

<< "Should not read data with version > snapshot";

// Case 4: Insert valid data in instance B (within snapshot version)

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string partition_key = versioned::meta_partition_key({instance_ids_[1], 8003});

std::string key = encode_versioned_key(partition_key, Versionstamp(1500, 1));

txn->put(key, "");

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

}

// Case 4: Should find the first encountered data (from B)

ASSERT_EQ(reader.is_partition_exists(8003), TxnErrorCode::TXN_OK)

<< "Should return first valid data encountered in chain";

}

TEST_F(CloneChainReaderTest, HasNoIndexes) {

std::string instance_id = instance_ids_[2]; // C

Versionstamp snapshot_version = snapshot_versions_[2];

CloneChainReader reader(instance_id, snapshot_version, txn_kv_.get(), resource_mgr_.get());

int64_t db_id = 1001;

int64_t table_id = 2001;

// Case 1: No index data exists anywhere - should return true (has no indexes)

{

bool no_indexes = false;

ASSERT_EQ(reader.has_no_indexes(db_id, table_id, &no_indexes), TxnErrorCode::TXN_OK);

ASSERT_TRUE(no_indexes) << "Table should have no indexes when no index data exists";

}

// Case 2: Insert index data in instance A

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

std::string index_inverted_key =

versioned::index_inverted_key({instance_ids_[0], db_id, table_id, 3001});

txn->put(index_inverted_key, "");

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

}

// Case 2: Read from instance C should find index from A through clone chain

{

bool no_indexes = true;

ASSERT_EQ(reader.has_no_indexes(db_id, table_id, &no_indexes), TxnErrorCode::TXN_OK);

ASSERT_FALSE(no_indexes)

<< "Table should have indexes when index data exists in clone chain";

}

}

TEST_F(CloneChainReaderTest, GetTabletMergedStats) {

std::string instance_id = instance_ids_[2]; // C

Versionstamp snapshot_version = snapshot_versions_[2];

CloneChainReader reader(instance_id, snapshot_version, txn_kv_.get(), resource_mgr_.get());

// Case 1: No data exists anywhere - should return NOT_FOUND

{

TabletStatsPB tablet_stats;

Versionstamp versionstamp;

ASSERT_EQ(reader.get_tablet_merged_stats(9001, &tablet_stats, &versionstamp),

TxnErrorCode::TXN_KEY_NOT_FOUND);

}

// Case 2: Insert both load and compact stats in instance A

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

// Insert load stats

std::string tablet_load_stats_key =

versioned::tablet_load_stats_key({instance_ids_[0], 9001});

TabletStatsPB load_stats;

load_stats.set_data_size(1000);

load_stats.set_num_rows(100);

load_stats.set_num_rowsets(10);

std::string load_key = encode_versioned_key(tablet_load_stats_key, Versionstamp(100, 1));

txn->put(load_key, load_stats.SerializeAsString());

// Insert compact stats

std::string tablet_compact_stats_key =

versioned::tablet_compact_stats_key({instance_ids_[0], 9001});

TabletStatsPB compact_stats;

compact_stats.set_data_size(500);

compact_stats.set_num_rows(50);

compact_stats.set_num_segments(5);

std::string compact_key =

encode_versioned_key(tablet_compact_stats_key, Versionstamp(100, 1));

txn->put(compact_key, compact_stats.SerializeAsString());

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

}

// Case 2: Read merged stats should return combined stats

{

TabletStatsPB tablet_stats;

Versionstamp versionstamp;

ASSERT_EQ(reader.get_tablet_merged_stats(9001, &tablet_stats, &versionstamp),

TxnErrorCode::TXN_OK);

ASSERT_EQ(tablet_stats.data_size(), 1500); // 1000 + 500

ASSERT_EQ(tablet_stats.num_rows(), 150); // 100 + 50

ASSERT_EQ(tablet_stats.num_rowsets(), 10); // load stats only

ASSERT_EQ(tablet_stats.num_segments(), 5); // compact stats only

ASSERT_EQ(versionstamp, Versionstamp(100, 1));

}

// Case 3: Insert both load and compact stats in instance B (with large version to be ignored)

{

std::unique_ptr<Transaction> txn;

ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);

// Insert load stats with large version (should be ignored)

std::string tablet_load_stats_key =

versioned::tablet_load_stats_key({instance_ids_[1], 9001});

TabletStatsPB load_stats;

load_stats.set_data_size(2000);

load_stats.set_num_rows(200);

load_stats.set_num_rowsets(20);

std::string load_key = encode_versioned_key(tablet_load_stats_key, Versionstamp(2500, 1));

txn->put(load_key, load_stats.SerializeAsString());

// Insert compact stats with large version (should be ignored)

std::string tablet_compact_stats_key =

versioned::tablet_compact_stats_key({instance_ids_[1], 9001});