Google Git Google Git
Sign in
apache/doris/refs/heads/stack-demo/./cloud/test/clone_chain_reader_test.cpp
blob: 089747faa360954057b4dc7d53bc4754cca06721 [file]
// 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;
std::string source_instance;
ASSERT_EQ(reader.get_partition_version(1001, &version_pb, &partition_version, false,
&source_instance),
TxnErrorCode::TXN_OK);
ASSERT_EQ(version_pb.version(), 100);
ASSERT_EQ(partition_version, Versionstamp(100, 1));
ASSERT_EQ(source_instance, instance_ids_[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 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;
std::string source_instance;
ASSERT_EQ(reader.get_partition_version(1001, &version_pb, &partition_version, false,
&source_instance),
TxnErrorCode::TXN_OK);
ASSERT_EQ(version_pb.version(), 150)
<< "Should return first valid data encountered in chain";
ASSERT_EQ(partition_version, Versionstamp(1500, 1));
ASSERT_EQ(source_instance, instance_ids_[1]);
}
}
TEST_F(CloneChainReaderTest, GetTableStreamOffsets) {
CloneChainReader reader(instance_ids_[2], Versionstamp(snapshot_versions_[2]), txn_kv_.get(),
resource_mgr_.get());
using OffsetMap = std::unordered_map<int64_t, std::unordered_map<int64_t, TableStreamOffsetPB>>;
using OffsetVersionstampMap =
std::unordered_map<int64_t, std::unordered_map<int64_t, Versionstamp>>;
TableStreamIdentityPB identity;
identity.set_base_db_id(2001);
identity.set_base_table_id(2002);
identity.set_stream_db_id(2003);
identity.set_stream_id(2004);
int64_t partition_id = 2005;
auto read_offsets = [&](const std::vector<int64_t>& partition_ids, OffsetMap* offsets,
OffsetVersionstampMap* versionstamps) {
std::unique_ptr<Transaction> txn;
TxnErrorCode err = txn_kv_->create_txn(&txn);
if (err != TxnErrorCode::TXN_OK) {
return err;
}
TableStreamPartitionSetPB binding;
binding.mutable_identity()->CopyFrom(identity);
for (int64_t id : partition_ids) {
binding.add_partition_ids(id);
}
offsets->clear();
versionstamps->clear();
return reader.get_table_stream_offsets(txn.get(), {binding}, offsets, versionstamps, true);
};
OffsetMap offsets;
OffsetVersionstampMap versionstamps;
ASSERT_EQ(read_offsets({partition_id}, &offsets, &versionstamps), TxnErrorCode::TXN_OK);
EXPECT_TRUE(offsets.empty());
auto put_offset = [&](const std::string& instance_id, int64_t target_partition_id,
Versionstamp key_version, int64_t offset_tso) {
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string key = versioned::table_stream_offset_key(
{instance_id, identity.base_db_id(), identity.base_table_id(),
identity.stream_db_id(), identity.stream_id(), target_partition_id});
TableStreamOffsetPB value;
value.set_partition_id(target_partition_id);
value.set_state(TABLE_STREAM_OFFSET_CONSUMED);
value.set_offset_tso(offset_tso);
versioned_put(txn.get(), key, key_version, value.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
};
put_offset(instance_ids_[0], partition_id, Versionstamp(500, 0), 50);
ASSERT_EQ(read_offsets({partition_id}, &offsets, &versionstamps), TxnErrorCode::TXN_OK);
EXPECT_EQ(offsets.at(identity.stream_id()).at(partition_id).offset_tso(), 50);
// This value is newer than B's source snapshot and must not shadow A's value.
put_offset(instance_ids_[1], partition_id, Versionstamp(2100, 0), 210);
ASSERT_EQ(read_offsets({partition_id}, &offsets, &versionstamps), TxnErrorCode::TXN_OK);
EXPECT_EQ(offsets.at(identity.stream_id()).at(partition_id).offset_tso(), 50);
put_offset(instance_ids_[1], partition_id, Versionstamp(1500, 0), 150);
ASSERT_EQ(read_offsets({partition_id}, &offsets, &versionstamps), TxnErrorCode::TXN_OK);
EXPECT_EQ(offsets.at(identity.stream_id()).at(partition_id).offset_tso(), 150);
put_offset(instance_ids_[2], partition_id, Versionstamp(2500, 0), 250);
ASSERT_EQ(read_offsets({partition_id}, &offsets, &versionstamps), TxnErrorCode::TXN_OK);
EXPECT_EQ(offsets.at(identity.stream_id()).at(partition_id).offset_tso(), 250);
int64_t parent_partition_id = partition_id + 1;
int64_t missing_partition_id = partition_id + 2;
put_offset(instance_ids_[0], parent_partition_id, Versionstamp(600, 0), 60);
ASSERT_EQ(read_offsets({partition_id, parent_partition_id, missing_partition_id}, &offsets,
&versionstamps),
TxnErrorCode::TXN_OK);
ASSERT_EQ(offsets.at(identity.stream_id()).size(), 2);
EXPECT_EQ(offsets.at(identity.stream_id()).at(partition_id).offset_tso(), 250);
EXPECT_EQ(offsets.at(identity.stream_id()).at(parent_partition_id).offset_tso(), 60);
EXPECT_FALSE(offsets.at(identity.stream_id()).contains(missing_partition_id));
}
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});
TabletStatsPB compact_stats;
compact_stats.set_data_size(800);
compact_stats.set_num_rows(80);
compact_stats.set_num_segments(8);
std::string compact_key =
encode_versioned_key(tablet_compact_stats_key, Versionstamp(2500, 1));
txn->put(compact_key, compact_stats.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 3: Should still return data from A, ignoring large version from B
{
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)
<< "Should not read data with version > snapshot"; // 1000 + 500
ASSERT_EQ(tablet_stats.num_rows(), 150); // 100 + 50
ASSERT_EQ(versionstamp, Versionstamp(100, 1));
}
// Case 4: Insert both load and compact stats in instance B (within snapshot version)
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
// Insert load stats with valid version
std::string tablet_load_stats_key =
versioned::tablet_load_stats_key({instance_ids_[1], 9001});
TabletStatsPB load_stats;
load_stats.set_data_size(1200);
load_stats.set_num_rows(120);
load_stats.set_num_rowsets(12);
std::string load_key = encode_versioned_key(tablet_load_stats_key, Versionstamp(1500, 1));
txn->put(load_key, load_stats.SerializeAsString());
// Insert compact stats with valid version
std::string tablet_compact_stats_key =
versioned::tablet_compact_stats_key({instance_ids_[1], 9001});
TabletStatsPB compact_stats;
compact_stats.set_data_size(600);
compact_stats.set_num_rows(60);
compact_stats.set_num_segments(6);
std::string compact_key =
encode_versioned_key(tablet_compact_stats_key, Versionstamp(1500, 1));
txn->put(compact_key, compact_stats.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 4: Should return first valid merged stats (from B)
{
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(), 1800)
<< "Should return first valid data encountered"; // 1200 + 600
ASSERT_EQ(tablet_stats.num_rows(), 180); // 120 + 60
ASSERT_EQ(tablet_stats.num_rowsets(), 12); // load stats only
ASSERT_EQ(tablet_stats.num_segments(), 6); // compact stats only
ASSERT_EQ(versionstamp, Versionstamp(1500, 1));
}
}
TEST_F(CloneChainReaderTest, GetPartitionVersions) {
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 maps
{
std::vector<int64_t> partition_ids = {10001, 10002, 10003};
std::unordered_map<int64_t, VersionPB> versions;
std::unordered_map<int64_t, Versionstamp> versionstamps;
std::unordered_map<int64_t, std::string> source_instances;
ASSERT_EQ(reader.get_partition_versions(partition_ids, &versions, &versionstamps, false,
&source_instances),
TxnErrorCode::TXN_OK);
ASSERT_TRUE(versions.empty());
ASSERT_TRUE(versionstamps.empty());
}
// Case 2: Insert partition versions in instance A
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
// Insert partitions 10001 and 10003, skip 10002
{
std::string partition_version_key =
versioned::partition_version_key({instance_ids_[0], 10001});
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());
}
{
std::string partition_version_key =
versioned::partition_version_key({instance_ids_[0], 10003});
VersionPB version_pb;
version_pb.set_version(300);
std::string key = encode_versioned_key(partition_version_key, Versionstamp(300, 1));
txn->put(key, version_pb.SerializeAsString());
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 2: Read from instance C should find partial data from A
{
std::vector<int64_t> partition_ids = {10001, 10002, 10003};
std::unordered_map<int64_t, VersionPB> versions;
std::unordered_map<int64_t, Versionstamp> versionstamps;
std::unordered_map<int64_t, std::string> source_instances;
ASSERT_EQ(reader.get_partition_versions(partition_ids, &versions, &versionstamps, false,
&source_instances),
TxnErrorCode::TXN_OK);
ASSERT_EQ(versions.size(), 2);
ASSERT_EQ(versionstamps.size(), 2);
ASSERT_EQ(versions[10001].version(), 100);
ASSERT_EQ(versions[10003].version(), 300);
ASSERT_EQ(versionstamps[10001], Versionstamp(100, 1));
ASSERT_EQ(versionstamps[10003], Versionstamp(300, 1));
ASSERT_EQ(versions.count(10002), 0);
ASSERT_EQ(versionstamps.count(10002), 0);
ASSERT_EQ(source_instances[10001], instance_ids_[0]);
ASSERT_EQ(source_instances[10003], instance_ids_[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 partition_version_key =
versioned::partition_version_key({instance_ids_[1], 10001});
VersionPB version_pb;
version_pb.set_version(250);
// 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, ignoring large version from B
{
std::vector<int64_t> partition_ids = {10001, 10002, 10003};
std::unordered_map<int64_t, VersionPB> versions;
std::unordered_map<int64_t, Versionstamp> versionstamps;
std::unordered_map<int64_t, std::string> source_instances;
ASSERT_EQ(reader.get_partition_versions(partition_ids, &versions, &versionstamps, false,
&source_instances),
TxnErrorCode::TXN_OK);
ASSERT_EQ(versions.size(), 2);
ASSERT_EQ(versions[10001].version(), 100) << "Should not read data with version > snapshot";
ASSERT_EQ(versions[10003].version(), 300);
}
// 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], 10001});
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());
}
// Add missing partition 10002 in instance B
{
std::string partition_version_key =
versioned::partition_version_key({instance_ids_[1], 10002});
VersionPB version_pb;
version_pb.set_version(200);
std::string key = encode_versioned_key(partition_version_key, Versionstamp(1600, 1));
txn->put(key, version_pb.SerializeAsString());
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 4: Should return first encountered data (from B for 10001&10002, from A for 10003)
{
std::vector<int64_t> partition_ids = {10001, 10002, 10003};
std::unordered_map<int64_t, VersionPB> versions;
std::unordered_map<int64_t, Versionstamp> versionstamps;
std::unordered_map<int64_t, std::string> source_instances;
ASSERT_EQ(reader.get_partition_versions(partition_ids, &versions, &versionstamps, false,
&source_instances),
TxnErrorCode::TXN_OK);
ASSERT_EQ(versions.size(), 3);
ASSERT_EQ(versions[10001].version(), 150) << "Should return first valid data encountered";
ASSERT_EQ(versions[10002].version(), 200) << "Should find missing data in clone chain";
ASSERT_EQ(versions[10003].version(), 300)
<< "Should return data from A when B doesn't have it";
ASSERT_EQ(versionstamps[10001], Versionstamp(1500, 1));
ASSERT_EQ(versionstamps[10002], Versionstamp(1600, 1));
ASSERT_EQ(versionstamps[10003], Versionstamp(300, 1));
ASSERT_EQ(source_instances[10001], instance_ids_[1]);
ASSERT_EQ(source_instances[10002], instance_ids_[1]);
ASSERT_EQ(source_instances[10003], instance_ids_[0]);
}
}
TEST_F(CloneChainReaderTest, GetTabletIndexes) {
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> tablet_ids = {11001, 11002, 11003};
std::unordered_map<int64_t, TabletIndexPB> tablet_indexes;
ASSERT_EQ(reader.get_tablet_indexes(tablet_ids, &tablet_indexes), TxnErrorCode::TXN_OK);
ASSERT_TRUE(tablet_indexes.empty());
}
// Case 2: Insert tablet indexes in instance A
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
// Insert tablets 11001 and 11003, skip 11002
{
std::string tablet_index_key = versioned::tablet_index_key({instance_ids_[0], 11001});
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());
}
{
std::string tablet_index_key = versioned::tablet_index_key({instance_ids_[0], 11003});
TabletIndexPB tablet_index;
tablet_index.set_table_id(1003);
tablet_index.set_index_id(2003);
tablet_index.set_partition_id(3003);
txn->put(tablet_index_key, tablet_index.SerializeAsString());
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 2: Read from instance C should find partial data from A
{
std::vector<int64_t> tablet_ids = {11001, 11002, 11003};
std::unordered_map<int64_t, TabletIndexPB> tablet_indexes;
ASSERT_EQ(reader.get_tablet_indexes(tablet_ids, &tablet_indexes), TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_indexes.size(), 2);
ASSERT_EQ(tablet_indexes[11001].table_id(), 1001);
ASSERT_EQ(tablet_indexes[11001].index_id(), 2001);
ASSERT_EQ(tablet_indexes[11001].partition_id(), 3001);
ASSERT_EQ(tablet_indexes[11003].table_id(), 1003);
ASSERT_EQ(tablet_indexes[11003].index_id(), 2003);
ASSERT_EQ(tablet_indexes[11003].partition_id(), 3003);
ASSERT_EQ(tablet_indexes.count(11002), 0);
}
// Case 3: Insert valid data in instance B (within snapshot version)
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
// Update tablet 11001 and add missing tablet 11002 in instance B
{
std::string tablet_index_key = versioned::tablet_index_key({instance_ids_[1], 11001});
TabletIndexPB tablet_index;
tablet_index.set_table_id(1001);
tablet_index.set_index_id(2011); // Different index_id
tablet_index.set_partition_id(3011); // Different partition_id
txn->put(tablet_index_key, tablet_index.SerializeAsString());
}
{
std::string tablet_index_key = versioned::tablet_index_key({instance_ids_[1], 11002});
TabletIndexPB tablet_index;
tablet_index.set_table_id(1002);
tablet_index.set_index_id(2002);
tablet_index.set_partition_id(3002);
txn->put(tablet_index_key, tablet_index.SerializeAsString());
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 3: Should return first encountered data (from B for 11001&11002, from A for 11003)
{
std::vector<int64_t> tablet_ids = {11001, 11002, 11003};
std::unordered_map<int64_t, TabletIndexPB> tablet_indexes;
ASSERT_EQ(reader.get_tablet_indexes(tablet_ids, &tablet_indexes), TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_indexes.size(), 3);
ASSERT_EQ(tablet_indexes[11001].index_id(), 2011)
<< "Should return first valid data encountered";
ASSERT_EQ(tablet_indexes[11001].partition_id(), 3011);
ASSERT_EQ(tablet_indexes[11002].table_id(), 1002)
<< "Should find missing data in clone chain";
ASSERT_EQ(tablet_indexes[11002].index_id(), 2002);
ASSERT_EQ(tablet_indexes[11003].index_id(), 2003)
<< "Should return data from A when B doesn't have it";
ASSERT_EQ(tablet_indexes[11003].partition_id(), 3003);
}
}
TEST_F(CloneChainReaderTest, GetTabletCompactStatsBatch) {
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 maps
{
std::vector<int64_t> tablet_ids = {12001, 12002, 12003};
std::unordered_map<int64_t, TabletStatsPB> tablet_stats;
std::unordered_map<int64_t, Versionstamp> versionstamps;
ASSERT_EQ(reader.get_tablet_compact_stats(tablet_ids, &tablet_stats, &versionstamps),
TxnErrorCode::TXN_OK);
ASSERT_TRUE(tablet_stats.empty());
ASSERT_TRUE(versionstamps.empty());
}
// Case 2: Insert tablet compact stats in instance A
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
// Insert tablets 12001 and 12003, skip 12002
{
std::string tablet_compact_stats_key =
versioned::tablet_compact_stats_key({instance_ids_[0], 12001});
TabletStatsPB tablet_stats;
tablet_stats.set_data_size(1000);
tablet_stats.set_num_rows(100);
tablet_stats.set_num_segments(10);
std::string key = encode_versioned_key(tablet_compact_stats_key, Versionstamp(100, 1));
txn->put(key, tablet_stats.SerializeAsString());
}
{
std::string tablet_compact_stats_key =
versioned::tablet_compact_stats_key({instance_ids_[0], 12003});
TabletStatsPB tablet_stats;
tablet_stats.set_data_size(3000);
tablet_stats.set_num_rows(300);
tablet_stats.set_num_segments(30);
std::string key = encode_versioned_key(tablet_compact_stats_key, Versionstamp(300, 1));
txn->put(key, tablet_stats.SerializeAsString());
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 2: Read from instance C should find partial data from A
{
std::vector<int64_t> tablet_ids = {12001, 12002, 12003};
std::unordered_map<int64_t, TabletStatsPB> tablet_stats;
std::unordered_map<int64_t, Versionstamp> versionstamps;
ASSERT_EQ(reader.get_tablet_compact_stats(tablet_ids, &tablet_stats, &versionstamps),
TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_stats.size(), 2);
ASSERT_EQ(versionstamps.size(), 2);
ASSERT_EQ(tablet_stats[12001].data_size(), 1000);
ASSERT_EQ(tablet_stats[12001].num_rows(), 100);
ASSERT_EQ(tablet_stats[12001].num_segments(), 10);
ASSERT_EQ(tablet_stats[12003].data_size(), 3000);
ASSERT_EQ(tablet_stats[12003].num_rows(), 300);
ASSERT_EQ(tablet_stats[12003].num_segments(), 30);
ASSERT_EQ(versionstamps[12001], Versionstamp(100, 1));
ASSERT_EQ(versionstamps[12003], Versionstamp(300, 1));
ASSERT_EQ(tablet_stats.count(12002), 0);
ASSERT_EQ(versionstamps.count(12002), 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 tablet_compact_stats_key =
versioned::tablet_compact_stats_key({instance_ids_[1], 12001});
TabletStatsPB tablet_stats;
tablet_stats.set_data_size(2000);
tablet_stats.set_num_rows(200);
tablet_stats.set_num_segments(20);
// 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, ignoring large version from B
{
std::vector<int64_t> tablet_ids = {12001, 12002, 12003};
std::unordered_map<int64_t, TabletStatsPB> tablet_stats;
std::unordered_map<int64_t, Versionstamp> versionstamps;
ASSERT_EQ(reader.get_tablet_compact_stats(tablet_ids, &tablet_stats, &versionstamps),
TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_stats.size(), 2);
ASSERT_EQ(tablet_stats[12001].data_size(), 1000)
<< "Should not read data with version > snapshot";
ASSERT_EQ(tablet_stats[12001].num_rows(), 100);
ASSERT_EQ(tablet_stats[12003].data_size(), 3000);
}
// 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], 12001});
TabletStatsPB tablet_stats;
tablet_stats.set_data_size(1500);
tablet_stats.set_num_rows(150);
tablet_stats.set_num_segments(15);
std::string key = encode_versioned_key(tablet_compact_stats_key, Versionstamp(1500, 1));
txn->put(key, tablet_stats.SerializeAsString());
}
// Add missing tablet 12002 in instance B
{
std::string tablet_compact_stats_key =
versioned::tablet_compact_stats_key({instance_ids_[1], 12002});
TabletStatsPB tablet_stats;
tablet_stats.set_data_size(2000);
tablet_stats.set_num_rows(200);
tablet_stats.set_num_segments(20);
std::string key = encode_versioned_key(tablet_compact_stats_key, Versionstamp(1600, 1));
txn->put(key, tablet_stats.SerializeAsString());
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 4: Should return first encountered data (from B for 12001&12002, from A for 12003)
{
std::vector<int64_t> tablet_ids = {12001, 12002, 12003};
std::unordered_map<int64_t, TabletStatsPB> tablet_stats;
std::unordered_map<int64_t, Versionstamp> versionstamps;
ASSERT_EQ(reader.get_tablet_compact_stats(tablet_ids, &tablet_stats, &versionstamps),
TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_stats.size(), 3);
ASSERT_EQ(tablet_stats[12001].data_size(), 1500)
<< "Should return first valid data encountered";
ASSERT_EQ(tablet_stats[12001].num_rows(), 150);
ASSERT_EQ(tablet_stats[12002].data_size(), 2000)
<< "Should find missing data in clone chain";
ASSERT_EQ(tablet_stats[12002].num_rows(), 200);
ASSERT_EQ(tablet_stats[12003].data_size(), 3000)
<< "Should return data from A when B doesn't have it";
ASSERT_EQ(tablet_stats[12003].num_rows(), 300);
ASSERT_EQ(versionstamps[12001], Versionstamp(1500, 1));
ASSERT_EQ(versionstamps[12002], Versionstamp(1600, 1));
ASSERT_EQ(versionstamps[12003], Versionstamp(300, 1));
}
}
TEST_F(CloneChainReaderTest, GetLoadRowsetMeta) {
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 tablet_id = 14001;
int64_t version = 5;
// Case 1: No data exists anywhere - should return NOT_FOUND
{
doris::RowsetMetaCloudPB rowset_meta;
ASSERT_EQ(reader.get_load_rowset_meta(tablet_id, version, &rowset_meta),
TxnErrorCode::TXN_KEY_NOT_FOUND);
}
// Case 2: Insert load rowset meta in instance A
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string rowset_load_key =
versioned::meta_rowset_load_key({instance_ids_[0], tablet_id, version});
doris::RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("load_rowset_5");
rowset_meta.set_start_version(version);
rowset_meta.set_end_version(version);
rowset_meta.set_num_rows(500);
rowset_meta.set_tablet_id(tablet_id);
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_load_key, std::move(rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 2: Read from instance C should find data from A through clone chain
{
doris::RowsetMetaCloudPB rowset_meta;
ASSERT_EQ(reader.get_load_rowset_meta(tablet_id, version, &rowset_meta),
TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_meta.rowset_id_v2(), "load_rowset_5");
ASSERT_EQ(rowset_meta.start_version(), version);
ASSERT_EQ(rowset_meta.end_version(), version);
ASSERT_EQ(rowset_meta.num_rows(), 500);
ASSERT_EQ(rowset_meta.tablet_id(), tablet_id);
}
// 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 rowset_load_key =
versioned::meta_rowset_load_key({instance_ids_[1], tablet_id, version});
doris::RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("load_rowset_5_large_version");
rowset_meta.set_start_version(version);
rowset_meta.set_end_version(version);
rowset_meta.set_num_rows(600);
rowset_meta.set_tablet_id(tablet_id);
std::string key = encode_versioned_key(rowset_load_key, Versionstamp(2500, 1));
txn->put(key, rowset_meta.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 3: Should still return data from A, ignoring large version from B
{
doris::RowsetMetaCloudPB rowset_meta;
ASSERT_EQ(reader.get_load_rowset_meta(tablet_id, version, &rowset_meta),
TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_meta.rowset_id_v2(), "load_rowset_5")
<< "Should not read data with version > snapshot";
ASSERT_EQ(rowset_meta.num_rows(), 500);
}
// 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 rowset_load_key =
versioned::meta_rowset_load_key({instance_ids_[1], tablet_id, version});
doris::RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("load_rowset_5_updated");
rowset_meta.set_start_version(version);
rowset_meta.set_end_version(version);
rowset_meta.set_num_rows(550);
rowset_meta.set_tablet_id(tablet_id);
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_load_key, std::move(rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 4: Should return the first encountered data (from B)
{
doris::RowsetMetaCloudPB rowset_meta;
ASSERT_EQ(reader.get_load_rowset_meta(tablet_id, version, &rowset_meta),
TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_meta.rowset_id_v2(), "load_rowset_5_updated")
<< "Should return first valid data encountered in chain";
ASSERT_EQ(rowset_meta.num_rows(), 550);
}
}
TEST_F(CloneChainReaderTest, GetCompactRowsetMeta) {
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 tablet_id = 14001;
int64_t version = 5;
// Case 1: No data exists anywhere - should return NOT_FOUND
{
doris::RowsetMetaCloudPB rowset_meta;
ASSERT_EQ(reader.get_compact_rowset_meta(tablet_id, version, &rowset_meta),
TxnErrorCode::TXN_KEY_NOT_FOUND);
}
// Case 2: Insert load rowset meta in instance A
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string rowset_compact_key =
versioned::meta_rowset_compact_key({instance_ids_[0], tablet_id, version});
doris::RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("compact_rowset_5");
rowset_meta.set_start_version(version);
rowset_meta.set_end_version(version);
rowset_meta.set_num_rows(500);
rowset_meta.set_tablet_id(tablet_id);
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_compact_key, std::move(rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 2: Read from instance C should find data from A through clone chain
{
doris::RowsetMetaCloudPB rowset_meta;
ASSERT_EQ(reader.get_compact_rowset_meta(tablet_id, version, &rowset_meta),
TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_meta.rowset_id_v2(), "compact_rowset_5");
ASSERT_EQ(rowset_meta.start_version(), version);
ASSERT_EQ(rowset_meta.end_version(), version);
ASSERT_EQ(rowset_meta.num_rows(), 500);
ASSERT_EQ(rowset_meta.tablet_id(), tablet_id);
}
// 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 rowset_compact_key =
versioned::meta_rowset_compact_key({instance_ids_[1], tablet_id, version});
doris::RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("compact_rowset_5_large_version");
rowset_meta.set_start_version(version);
rowset_meta.set_end_version(version);
rowset_meta.set_num_rows(600);
rowset_meta.set_tablet_id(tablet_id);
std::string key = encode_versioned_key(rowset_compact_key, Versionstamp(2500, 1));
txn->put(key, rowset_meta.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 3: Should still return data from A, ignoring large version from B
{
doris::RowsetMetaCloudPB rowset_meta;
ASSERT_EQ(reader.get_compact_rowset_meta(tablet_id, version, &rowset_meta),
TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_meta.rowset_id_v2(), "compact_rowset_5")
<< "Should not read data with version > snapshot";
ASSERT_EQ(rowset_meta.num_rows(), 500);
}
// 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 rowset_compact_key =
versioned::meta_rowset_compact_key({instance_ids_[1], tablet_id, version});
doris::RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("compact_rowset_5_updated");
rowset_meta.set_start_version(version);
rowset_meta.set_end_version(version);
rowset_meta.set_num_rows(550);
rowset_meta.set_tablet_id(tablet_id);
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_compact_key, std::move(rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 4: Should return the first encountered data (from B)
{
doris::RowsetMetaCloudPB rowset_meta;
ASSERT_EQ(reader.get_compact_rowset_meta(tablet_id, version, &rowset_meta),
TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_meta.rowset_id_v2(), "compact_rowset_5_updated")
<< "Should return first valid data encountered in chain";
ASSERT_EQ(rowset_meta.num_rows(), 550);
}
}
TEST_F(CloneChainReaderTest, GetRowsetMeta) {
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 tablet_id = 16001;
int64_t version = 5;
// Case 1: No data exists anywhere - should return NOT_FOUND
{
doris::RowsetMetaCloudPB rowset_meta;
ASSERT_EQ(reader.get_rowset_meta(tablet_id, version, &rowset_meta),
TxnErrorCode::TXN_KEY_NOT_FOUND);
}
// Case 2: Insert only load rowset meta in instance A
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string rowset_load_key =
versioned::meta_rowset_load_key({instance_ids_[0], tablet_id, version});
doris::RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("load_rowset_" + std::to_string(version));
rowset_meta.set_start_version(version);
rowset_meta.set_end_version(version);
rowset_meta.set_num_rows(500);
rowset_meta.set_tablet_id(tablet_id);
// Use a specific versionstamp
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_load_key, Versionstamp(100, 1),
std::move(rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 2: Should return load rowset meta
{
doris::RowsetMetaCloudPB rowset_meta;
ASSERT_EQ(reader.get_rowset_meta(tablet_id, version, &rowset_meta), TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_meta.rowset_id_v2(), "load_rowset_" + std::to_string(version));
ASSERT_EQ(rowset_meta.num_rows(), 500);
}
// Case 3: Insert compact rowset meta with earlier versionstamp in instance A
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string rowset_compact_key =
versioned::meta_rowset_compact_key({instance_ids_[0], tablet_id, version});
doris::RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("compact_rowset_" + std::to_string(version));
rowset_meta.set_start_version(version);
rowset_meta.set_end_version(version);
rowset_meta.set_num_rows(600);
rowset_meta.set_tablet_id(tablet_id);
// Earlier versionstamp
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_compact_key, Versionstamp(50, 1),
std::move(rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 3: Should return load rowset meta (higher versionstamp: 100 > 50)
{
doris::RowsetMetaCloudPB rowset_meta;
ASSERT_EQ(reader.get_rowset_meta(tablet_id, version, &rowset_meta), TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_meta.rowset_id_v2(), "load_rowset_" + std::to_string(version))
<< "Should return rowset with higher versionstamp (100 > 50)";
ASSERT_EQ(rowset_meta.num_rows(), 500);
}
// Case 4: Insert compact rowset meta with later versionstamp in instance B
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string rowset_compact_key =
versioned::meta_rowset_compact_key({instance_ids_[1], tablet_id, version});
doris::RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("compact_rowset_updated_" + std::to_string(version));
rowset_meta.set_start_version(version);
rowset_meta.set_end_version(version);
rowset_meta.set_num_rows(700);
rowset_meta.set_tablet_id(tablet_id);
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_compact_key, Versionstamp(1500, 1),
std::move(rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 4: Should return compact rowset meta from B (found first in clone chain)
{
doris::RowsetMetaCloudPB rowset_meta;
ASSERT_EQ(reader.get_rowset_meta(tablet_id, version, &rowset_meta), TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_meta.rowset_id_v2(), "compact_rowset_updated_" + std::to_string(version))
<< "Should return first valid data encountered in chain";
ASSERT_EQ(rowset_meta.num_rows(), 700);
}
// Case 5: Test with version beyond snapshot - should be ignored
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
int64_t another_version = 6;
std::string rowset_load_key =
versioned::meta_rowset_load_key({instance_ids_[1], tablet_id, another_version});
doris::RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("load_rowset_large_version_" +
std::to_string(another_version));
rowset_meta.set_start_version(another_version);
rowset_meta.set_end_version(another_version);
rowset_meta.set_num_rows(800);
rowset_meta.set_tablet_id(tablet_id);
// Use version larger than snapshot_versions_[1] (2000)
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_load_key, Versionstamp(2500, 1),
std::move(rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 5: Should not find the large version rowset
{
doris::RowsetMetaCloudPB rowset_meta;
ASSERT_EQ(reader.get_rowset_meta(tablet_id, 6, &rowset_meta),
TxnErrorCode::TXN_KEY_NOT_FOUND)
<< "Should not read data with version > snapshot";
}
// Case 6: Test only compact rowset meta exists
{
int64_t another_version = 7;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string rowset_compact_key =
versioned::meta_rowset_compact_key({instance_ids_[0], tablet_id, another_version});
doris::RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("compact_only_" + std::to_string(another_version));
rowset_meta.set_start_version(another_version);
rowset_meta.set_end_version(another_version);
rowset_meta.set_num_rows(900);
rowset_meta.set_tablet_id(tablet_id);
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_compact_key, Versionstamp(100, 1),
std::move(rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 6: Should return compact rowset meta when only compact exists
{
doris::RowsetMetaCloudPB rowset_meta;
ASSERT_EQ(reader.get_rowset_meta(tablet_id, 7, &rowset_meta), TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_meta.rowset_id_v2(), "compact_only_7");
ASSERT_EQ(rowset_meta.num_rows(), 900);
}
// Case 7: Test versionstamp comparison - compact rowset with higher versionstamp should win
{
int64_t another_version = 8;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
// Insert load rowset with lower versionstamp
std::string rowset_load_key =
versioned::meta_rowset_load_key({instance_ids_[0], tablet_id, another_version});
doris::RowsetMetaCloudPB load_rowset_meta;
load_rowset_meta.set_rowset_id(0);
load_rowset_meta.set_rowset_id_v2("load_rowset_" + std::to_string(another_version));
load_rowset_meta.set_start_version(another_version);
load_rowset_meta.set_end_version(another_version);
load_rowset_meta.set_num_rows(1000);
load_rowset_meta.set_tablet_id(tablet_id);
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_load_key, Versionstamp(100, 1),
std::move(load_rowset_meta)));
// Insert compact rowset with higher versionstamp
std::string rowset_compact_key =
versioned::meta_rowset_compact_key({instance_ids_[0], tablet_id, another_version});
doris::RowsetMetaCloudPB compact_rowset_meta;
compact_rowset_meta.set_rowset_id(0);
compact_rowset_meta.set_rowset_id_v2("compact_rowset_" + std::to_string(another_version));
compact_rowset_meta.set_start_version(another_version);
compact_rowset_meta.set_end_version(another_version);
compact_rowset_meta.set_num_rows(1100);
compact_rowset_meta.set_tablet_id(tablet_id);
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_compact_key, Versionstamp(200, 1),
std::move(compact_rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 7: Should return compact rowset meta (higher versionstamp)
{
doris::RowsetMetaCloudPB rowset_meta;
ASSERT_EQ(reader.get_rowset_meta(tablet_id, 8, &rowset_meta), TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_meta.rowset_id_v2(), "compact_rowset_8")
<< "Should return rowset with higher versionstamp";
ASSERT_EQ(rowset_meta.num_rows(), 1100);
}
}
TEST_F(CloneChainReaderTest, GetRowsetMetasMinReadVersionstamp) {
constexpr int64_t tablet_id = 15001;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
auto put_rowset = [&](const std::string& instance_id, const std::string& id, int64_t start,
int64_t end, Versionstamp version, bool compact) {
doris::RowsetMetaCloudPB rowset;
rowset.set_rowset_id(0);
rowset.set_rowset_id_v2(id);
rowset.set_tablet_id(tablet_id);
rowset.set_start_version(start);
rowset.set_end_version(end);
auto key = compact ? versioned::meta_rowset_compact_key({instance_id, tablet_id, end})
: versioned::meta_rowset_load_key({instance_id, tablet_id, end});
ASSERT_TRUE(versioned::document_put(txn.get(), key, version, std::move(rowset)));
};
put_rowset("A", "L1", 1, 1, Versionstamp(70, 1), false);
put_rowset("A", "L2", 2, 2, Versionstamp(80, 1), false);
put_rowset("A", "L3", 3, 3, Versionstamp(90, 1), false);
put_rowset("A", "compact", 2, 3, Versionstamp(150, 2), true);
// This source rowset is newer than A's source snapshot (1000) and is invisible to C.
put_rowset("A", "too_new", 1, 3, Versionstamp(1000, 1), true);
put_rowset("C", "L4", 4, 4, Versionstamp(2170, 1), false);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
CloneChainReader reader(instance_ids_[2], Versionstamp(3000), txn_kv_.get(),
resource_mgr_.get());
std::vector<doris::RowsetMetaCloudPB> rowsets;
ASSERT_EQ(reader.get_rowset_metas(tablet_id, 2, 4, &rowsets), TxnErrorCode::TXN_OK);
ASSERT_EQ(rowsets.size(), 2);
EXPECT_EQ(rowsets[0].rowset_id_v2(), "compact");
EXPECT_EQ(rowsets[0].start_version(), 2);
EXPECT_EQ(rowsets[0].end_version(), 3);
EXPECT_EQ(rowsets[0].reference_instance_id(), "A");
EXPECT_EQ(rowsets[1].rowset_id_v2(), "L4");
EXPECT_EQ(rowsets[1].start_version(), 4);
EXPECT_EQ(rowsets[1].end_version(), 4);
EXPECT_EQ(rowsets[1].reference_instance_id(), "C");
EXPECT_EQ(reader.min_read_versionstamp(), Versionstamp(150, 2));
ASSERT_EQ(reader.get_rowset_metas(tablet_id, 1, 1, &rowsets), TxnErrorCode::TXN_OK);
ASSERT_EQ(rowsets.size(), 1);
EXPECT_EQ(rowsets[0].rowset_id_v2(), "L1");
EXPECT_EQ(rowsets[0].reference_instance_id(), "A");
EXPECT_EQ(reader.min_read_versionstamp(), Versionstamp(70, 1));
ASSERT_EQ(reader.get_rowset_metas(tablet_id, 2, 4, &rowsets), TxnErrorCode::TXN_OK);
EXPECT_EQ(reader.min_read_versionstamp(), Versionstamp(70, 1));
ASSERT_EQ(reader.get_rowset_metas(tablet_id, 5, 6, &rowsets), TxnErrorCode::TXN_OK);
EXPECT_TRUE(rowsets.empty());
EXPECT_EQ(reader.min_read_versionstamp(), Versionstamp(70, 1));
}
TEST_F(CloneChainReaderTest, GetRowsetMetas) {
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 tablet_id = 15001;
int64_t start_version = 1;
int64_t end_version = 5;
// Case 1: No data exists anywhere - should return NOT_FOUND
{
std::vector<doris::RowsetMetaCloudPB> rowset_metas;
ASSERT_EQ(reader.get_rowset_metas(tablet_id, start_version, end_version, &rowset_metas),
TxnErrorCode::TXN_OK);
ASSERT_TRUE(rowset_metas.empty());
}
// Case 2: Insert rowset metas in instance A
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
for (int64_t version = start_version; version <= end_version; ++version) {
std::string rowset_key =
versioned::meta_rowset_load_key({instance_ids_[0], tablet_id, version});
doris::RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("rowset_" + std::to_string(version));
rowset_meta.set_start_version(version);
rowset_meta.set_end_version(version);
rowset_meta.set_num_rows(100 * version);
rowset_meta.set_tablet_id(tablet_id);
Versionstamp versionstamp(snapshot_versions_[0] - 1, 1);
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_key, versionstamp,
std::move(rowset_meta)));
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 2: Read from instance C should find data from A through clone chain
{
std::vector<doris::RowsetMetaCloudPB> rowset_metas;
ASSERT_EQ(reader.get_rowset_metas(tablet_id, start_version, end_version, &rowset_metas),
TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_metas.size(), 5);
for (size_t i = 0; i < rowset_metas.size(); ++i) {
int64_t version = start_version + i;
ASSERT_EQ(rowset_metas[i].rowset_id_v2(), "rowset_" + std::to_string(version));
ASSERT_EQ(rowset_metas[i].start_version(), version);
ASSERT_EQ(rowset_metas[i].end_version(), version);
ASSERT_EQ(rowset_metas[i].num_rows(), 100 * version);
}
}
// Case 3: Insert data with version > snapshot_version in instance B (should be ignored)
{
// [1, 1], [2, 5]
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
int64_t version = 1;
std::string rowset_key =
versioned::meta_rowset_load_key({instance_ids_[1], tablet_id, version});
doris::RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("large_rowset_" + std::to_string(version));
rowset_meta.set_start_version(version);
rowset_meta.set_end_version(version);
rowset_meta.set_num_rows(200 * version);
rowset_meta.set_tablet_id(tablet_id);
Versionstamp versionstamp(snapshot_versions_[1], 1);
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_key, versionstamp,
std::move(rowset_meta)));
rowset_key = versioned::meta_rowset_compact_key({instance_ids_[1], tablet_id, end_version});
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("large_rowset_" + std::to_string(version));
rowset_meta.set_start_version(version);
rowset_meta.set_end_version(end_version);
rowset_meta.set_num_rows(200 * version);
rowset_meta.set_tablet_id(tablet_id);
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_key, versionstamp,
std::move(rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 3: Should still return data from A, ignoring large version from B
{
std::vector<doris::RowsetMetaCloudPB> rowset_metas;
ASSERT_EQ(reader.get_rowset_metas(tablet_id, start_version, end_version, &rowset_metas),
TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_metas.size(), 5);
for (size_t i = 0; i < rowset_metas.size(); ++i) {
int64_t version = start_version + i;
ASSERT_EQ(rowset_metas[i].rowset_id_v2(), "rowset_" + std::to_string(version))
<< "Should not read data with version > snapshot";
ASSERT_EQ(rowset_metas[i].num_rows(), 100 * version);
}
}
// 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);
int64_t version = 1;
std::string rowset_key =
versioned::meta_rowset_load_key({instance_ids_[1], tablet_id, version});
doris::RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("large_rowset_" + std::to_string(version));
rowset_meta.set_start_version(version);
rowset_meta.set_end_version(version);
rowset_meta.set_num_rows(200 * version);
rowset_meta.set_tablet_id(tablet_id);
Versionstamp versionstamp(snapshot_versions_[1] - 1, 1);
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_key, versionstamp,
std::move(rowset_meta)));
rowset_key = versioned::meta_rowset_compact_key({instance_ids_[1], tablet_id, end_version});
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("large_rowset_" + std::to_string(end_version));
rowset_meta.set_start_version(version + 1);
rowset_meta.set_end_version(end_version);
rowset_meta.set_num_rows(200 * version);
rowset_meta.set_tablet_id(tablet_id);
ASSERT_TRUE(versioned::document_put(txn.get(), rowset_key, versionstamp,
std::move(rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 4: Should return the first encountered data (from B)
{
std::vector<doris::RowsetMetaCloudPB> rowset_metas;
ASSERT_EQ(reader.get_rowset_metas(tablet_id, start_version, end_version, &rowset_metas),
TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_metas.size(), 2);
ASSERT_EQ(rowset_metas[0].rowset_id_v2(), "large_rowset_1")
<< "Should return first valid data encountered in chain";
ASSERT_EQ(rowset_metas[0].num_rows(), 200 * 1);
ASSERT_EQ(rowset_metas[0].start_version(), 1);
ASSERT_EQ(rowset_metas[0].end_version(), 1);
ASSERT_EQ(rowset_metas[1].rowset_id_v2(), "large_rowset_5")
<< "Should return first valid data encountered in chain";
ASSERT_EQ(rowset_metas[1].num_rows(), 200 * 1);
ASSERT_EQ(rowset_metas[1].start_version(), 2);
ASSERT_EQ(rowset_metas[1].end_version(), 5);
}
}
TEST_F(CloneChainReaderTest, GetTabletMeta) {
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 tablet_id = 16001;
// Case 1: No data exists anywhere - should return NOT_FOUND
{
doris::TabletMetaCloudPB tablet_meta;
Versionstamp versionstamp;
ASSERT_EQ(reader.get_tablet_meta(tablet_id, &tablet_meta, &versionstamp),
TxnErrorCode::TXN_KEY_NOT_FOUND);
}
// Case 2: Insert tablet meta in instance A
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string tablet_meta_key = versioned::meta_tablet_key({instance_ids_[0], tablet_id});
doris::TabletMetaCloudPB tablet_meta;
tablet_meta.set_tablet_id(tablet_id);
tablet_meta.set_table_id(1001);
tablet_meta.set_partition_id(2001);
tablet_meta.set_index_id(3001);
tablet_meta.set_schema_version(1);
ASSERT_TRUE(versioned::document_put(txn.get(), tablet_meta_key, std::move(tablet_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 2: Read from instance C should find data from A through clone chain
{
doris::TabletMetaCloudPB tablet_meta;
Versionstamp versionstamp;
ASSERT_EQ(reader.get_tablet_meta(tablet_id, &tablet_meta, &versionstamp),
TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_meta.tablet_id(), tablet_id);
ASSERT_EQ(tablet_meta.table_id(), 1001);
ASSERT_EQ(tablet_meta.partition_id(), 2001);
ASSERT_EQ(tablet_meta.index_id(), 3001);
ASSERT_EQ(tablet_meta.schema_version(), 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_meta_key = versioned::meta_tablet_key({instance_ids_[1], tablet_id});
doris::TabletMetaCloudPB tablet_meta;
tablet_meta.set_tablet_id(tablet_id);
tablet_meta.set_table_id(1002);
tablet_meta.set_partition_id(2002);
tablet_meta.set_index_id(3002);
tablet_meta.set_schema_version(2);
std::string key = encode_versioned_key(tablet_meta_key, Versionstamp(2500, 1));
txn->put(key, tablet_meta.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 3: Should still return data from A, ignoring large version from B
{
doris::TabletMetaCloudPB tablet_meta;
Versionstamp versionstamp;
ASSERT_EQ(reader.get_tablet_meta(tablet_id, &tablet_meta, &versionstamp),
TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_meta.table_id(), 1001) << "Should not read data with version > snapshot";
ASSERT_EQ(tablet_meta.partition_id(), 2001);
ASSERT_EQ(tablet_meta.schema_version(), 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_meta_key = versioned::meta_tablet_key({instance_ids_[1], tablet_id});
doris::TabletMetaCloudPB tablet_meta;
tablet_meta.set_tablet_id(tablet_id);
tablet_meta.set_table_id(1003);
tablet_meta.set_partition_id(2003);
tablet_meta.set_index_id(3003);
tablet_meta.set_schema_version(3);
ASSERT_TRUE(versioned::document_put(txn.get(), tablet_meta_key, std::move(tablet_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 4: Should return the first encountered data (from B)
{
doris::TabletMetaCloudPB tablet_meta;
Versionstamp versionstamp;
ASSERT_EQ(reader.get_tablet_meta(tablet_id, &tablet_meta, &versionstamp),
TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_meta.table_id(), 1003)
<< "Should return first valid data encountered in chain";
ASSERT_EQ(tablet_meta.partition_id(), 2003);
ASSERT_EQ(tablet_meta.schema_version(), 3);
}
}
TEST_F(CloneChainReaderTest, GetTabletMetaThroughDeletedIntermediateClone) {
ResourceManager resource_mgr(txn_kv_);
InstanceInfoPB instance_b;
instance_b.set_instance_id(instance_ids_[1]);
instance_b.set_status(InstanceInfoPB::NORMAL);
instance_b.set_source_instance_id(instance_ids_[0]);
instance_b.set_source_snapshot_id(Versionstamp(snapshot_versions_[0]).to_string());
resource_mgr.refresh_instance(instance_ids_[1], instance_b);
InstanceInfoPB instance_c;
instance_c.set_instance_id(instance_ids_[2]);
instance_c.set_status(InstanceInfoPB::NORMAL);
instance_c.set_source_instance_id(instance_ids_[1]);
instance_c.set_source_snapshot_id(Versionstamp(snapshot_versions_[1]).to_string());
resource_mgr.refresh_instance(instance_ids_[2], instance_c);
constexpr int64_t tablet_id = 16002;
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string tablet_meta_key = versioned::meta_tablet_key({instance_ids_[0], tablet_id});
doris::TabletMetaCloudPB tablet_meta;
tablet_meta.set_tablet_id(tablet_id);
tablet_meta.set_table_id(1001);
ASSERT_TRUE(versioned::document_put(txn.get(), tablet_meta_key, std::move(tablet_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Single-level clone lookup remains unchanged.
{
CloneChainReader reader(instance_ids_[1], Versionstamp(snapshot_versions_[1]),
txn_kv_.get(), &resource_mgr);
doris::TabletMetaCloudPB tablet_meta;
Versionstamp versionstamp;
ASSERT_EQ(reader.get_tablet_meta(tablet_id, &tablet_meta, &versionstamp),
TxnErrorCode::TXN_OK);
EXPECT_EQ(tablet_meta.table_id(), 1001);
}
instance_b.set_status(InstanceInfoPB::DELETED);
resource_mgr.refresh_instance(instance_ids_[1], instance_b);
// B's tombstone still carries B -> A lineage, so C can continue to A.
CloneChainReader reader(instance_ids_[2], Versionstamp(snapshot_versions_[2]), txn_kv_.get(),
&resource_mgr);
doris::TabletMetaCloudPB tablet_meta;
Versionstamp versionstamp;
ASSERT_EQ(reader.get_tablet_meta(tablet_id, &tablet_meta, &versionstamp), TxnErrorCode::TXN_OK);
EXPECT_EQ(tablet_meta.table_id(), 1001);
// Decoupling the deleted intermediate clone clears both lineage fields and its mapping.
instance_b.clear_source_instance_id();
instance_b.clear_source_snapshot_id();
resource_mgr.refresh_instance(instance_ids_[1], instance_b);
std::string source_instance_id;
Versionstamp source_snapshot_version;
EXPECT_FALSE(resource_mgr.get_source_snapshot_info(instance_ids_[1], &source_instance_id,
&source_snapshot_version));
}
TEST_F(CloneChainReaderTest, GetTabletSchema) {
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 index_id = 17001;
int64_t schema_version = 5;
// Case 1: No data exists anywhere - should return NOT_FOUND
{
doris::TabletSchemaCloudPB tablet_schema;
ASSERT_EQ(reader.get_tablet_schema(index_id, schema_version, &tablet_schema),
TxnErrorCode::TXN_KEY_NOT_FOUND);
}
// Case 2: Insert tablet schema in instance A
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string tablet_schema_key =
versioned::meta_schema_key({instance_ids_[0], index_id, schema_version});
doris::TabletSchemaCloudPB tablet_schema;
tablet_schema.set_schema_version(schema_version);
tablet_schema.set_keys_type(doris::DUP_KEYS);
ASSERT_TRUE(document_put(txn.get(), tablet_schema_key, std::move(tablet_schema)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 2: Read from instance C should find data from A through clone chain
{
doris::TabletSchemaCloudPB tablet_schema;
ASSERT_EQ(reader.get_tablet_schema(index_id, schema_version, &tablet_schema),
TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_schema.schema_version(), schema_version);
ASSERT_EQ(tablet_schema.keys_type(), doris::DUP_KEYS);
}
// Case 4: Insert valid data in instance B
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string tablet_schema_key =
versioned::meta_schema_key({instance_ids_[1], index_id, schema_version});
doris::TabletSchemaCloudPB tablet_schema;
tablet_schema.set_schema_version(schema_version);
tablet_schema.set_keys_type(doris::AGG_KEYS);
ASSERT_TRUE(document_put(txn.get(), tablet_schema_key, std::move(tablet_schema)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 4: Should return the first encountered data (from B)
{
doris::TabletSchemaCloudPB tablet_schema;
ASSERT_EQ(reader.get_tablet_schema(index_id, schema_version, &tablet_schema),
TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_schema.keys_type(), doris::AGG_KEYS)
<< "Should return first valid data encountered in chain";
}
}
TEST_F(CloneChainReaderTest, GetPartitionPendingTxnId) {
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 partition_id = 18001;
// Case 1: No data exists anywhere - should set first_txn_id to -1
{
int64_t first_txn_id;
int64_t partition_version;
ASSERT_EQ(reader.get_partition_pending_txn_id(partition_id, &first_txn_id,
&partition_version),
TxnErrorCode::TXN_OK);
ASSERT_EQ(first_txn_id, -1);
}
// Case 2: Insert partition pending txn data 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], partition_id});
// Create PartitionPendingTxnInfo
doris::cloud::VersionPB partition_version;
partition_version.add_pending_txn_ids(1001);
partition_version.set_version(100);
versioned_put(txn.get(), partition_version_key, Versionstamp(100, 0),
partition_version.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 2: Read from instance C should find data from A through clone chain
{
int64_t first_txn_id;
int64_t partition_version;
ASSERT_EQ(reader.get_partition_pending_txn_id(partition_id, &first_txn_id,
&partition_version),
TxnErrorCode::TXN_OK);
ASSERT_EQ(first_txn_id, 1001);
ASSERT_EQ(partition_version, 100);
}
// 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_pending_key =
versioned::partition_version_key({instance_ids_[1], partition_id});
VersionPB version;
version.set_version(200);
version.add_pending_txn_ids(2001);
std::string key = encode_versioned_key(partition_pending_key, Versionstamp(2500, 1));
versioned_put(txn.get(), key, Versionstamp(snapshot_versions_[1], 100),
version.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 3: Should still return data from A, ignoring large version from B
{
int64_t first_txn_id;
int64_t partition_version;
ASSERT_EQ(reader.get_partition_pending_txn_id(partition_id, &first_txn_id,
&partition_version),
TxnErrorCode::TXN_OK);
ASSERT_EQ(first_txn_id, 1001) << "Should not read data with version > snapshot";
ASSERT_EQ(partition_version, 100);
}
// 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], partition_id});
VersionPB version;
version.set_version(150);
version.add_pending_txn_ids(1500);
versioned_put(txn.get(), partition_version_key, Versionstamp(snapshot_versions_[1] - 1, 0),
version.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 4: Should return the first encountered data (from B)
{
int64_t first_txn_id;
int64_t partition_version;
ASSERT_EQ(reader.get_partition_pending_txn_id(partition_id, &first_txn_id,
&partition_version),
TxnErrorCode::TXN_OK);
ASSERT_EQ(first_txn_id, 1500) << "Should return first valid data encountered in chain";
ASSERT_EQ(partition_version, 150);
}
}
TEST_F(CloneChainReaderTest, GetDeleteBitmapV2) {
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 tablet_id = 19001;
std::string rowset_id = "rowset_1";
// Case 1: No data exists anywhere - should return NOT_FOUND
{
DeleteBitmapStoragePB delete_bitmap;
ASSERT_EQ(reader.get_delete_bitmap_v2(tablet_id, rowset_id, &delete_bitmap),
TxnErrorCode::TXN_KEY_NOT_FOUND);
}
// Case 2: Insert delete bitmap in instance A
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string delete_bitmap_key =
versioned::meta_delete_bitmap_key({instance_ids_[0], tablet_id, rowset_id});
DeleteBitmapStoragePB delete_bitmap;
delete_bitmap.set_store_in_fdb(true);
blob_put(txn.get(), delete_bitmap_key, delete_bitmap, 0);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Case 2: Read from instance C should find data from A through clone chain
{
DeleteBitmapStoragePB delete_bitmap;
ASSERT_EQ(reader.get_delete_bitmap_v2(tablet_id, rowset_id, &delete_bitmap),
TxnErrorCode::TXN_OK)
<< dump_range(txn_kv_.get());
}
}