Google Git Google Git
Sign in
apache/doris/refs/heads/stack-demo/./cloud/test/meta_reader_test.cpp
blob: d14ea18f1a813802dd9f62e34312c262914bdff6 [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/meta_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 <tuple>
#include "common/config.h"
#include "common/logging.h"
#include "common/util.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"
using namespace doris::cloud;
int main(int argc, char** argv) {
config::log_dir = "./log/";
if (!doris::cloud::init_glog("meta_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.
static 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;
}
static 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;
}
TEST(MetaReaderTest, GetTableVersion) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
int64_t table_id = 1001;
{
// NOT FOUND
MetaReader meta_reader(instance_id, txn_kv.get());
Versionstamp table_version;
TxnErrorCode err = meta_reader.get_table_version(table_id, &table_version);
ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
}
{
// Put a table version
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_id, table_id});
versioned_put(txn.get(), table_version_key, "");
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
Versionstamp version1;
{
MetaReader meta_reader(instance_id, txn_kv.get());
TxnErrorCode err = meta_reader.get_table_version(table_id, &version1);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(meta_reader.min_read_versionstamp(), version1);
}
{
// Put a table version
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_id, table_id});
versioned_put(txn.get(), table_version_key, "");
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
Versionstamp version2;
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
TxnErrorCode err = meta_reader.get_table_version(txn.get(), table_id, &version2);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(meta_reader.min_read_versionstamp(), version2);
}
ASSERT_LT(version1, version2);
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string begin_key = versioned::table_version_key({instance_id, table_id});
std::string end_key = versioned::table_version_key({instance_id, table_id + 1});
FullRangeGetOptions options;
std::unique_ptr<FullRangeGetIterator> iter =
txn->full_range_get(begin_key, end_key, std::move(options));
int num = 1;
for (auto kvp = iter->next(); kvp.has_value(); kvp = iter->next()) {
auto&& [key, _] = *kvp;
int64_t decoded_id;
Versionstamp decoded_version;
std::string_view key_view(key);
ASSERT_TRUE(
versioned::decode_table_version_key(&key_view, &decoded_id, &decoded_version));
ASSERT_EQ(decoded_id, table_id);
ASSERT_EQ(decoded_version, num == 1 ? version1 : version2);
num++;
}
}
}
TEST(MetaReaderTest, GetTableStreamOffset) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "table_stream_instance";
TableStreamIdentityPB identity;
identity.set_base_db_id(1001);
identity.set_base_table_id(1002);
identity.set_stream_db_id(1003);
identity.set_stream_id(1004);
int64_t partition_id = 1005;
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(), partition_id});
TableStreamOffsetPB offset;
Versionstamp versionstamp;
MetaReader latest_reader(instance_id, txn_kv.get());
std::unique_ptr<Transaction> missing_read_txn;
ASSERT_EQ(txn_kv->create_txn(&missing_read_txn), TxnErrorCode::TXN_OK);
ASSERT_EQ(latest_reader.get_table_stream_offset(missing_read_txn.get(), identity, partition_id,
&offset, &versionstamp),
TxnErrorCode::TXN_KEY_NOT_FOUND);
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
TableStreamOffsetPB first;
first.set_partition_id(partition_id);
first.set_state(TABLE_STREAM_OFFSET_CONSUMED);
first.set_offset_tso(101);
versioned_put(txn.get(), key, Versionstamp(100, 1), first.SerializeAsString());
TableStreamOffsetPB second = first;
second.set_offset_tso(202);
versioned_put(txn.get(), key, Versionstamp(200, 1), second.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
std::unique_ptr<Transaction> latest_read_txn;
ASSERT_EQ(txn_kv->create_txn(&latest_read_txn), TxnErrorCode::TXN_OK);
ASSERT_EQ(latest_reader.get_table_stream_offset(latest_read_txn.get(), identity, partition_id,
&offset, &versionstamp),
TxnErrorCode::TXN_OK);
EXPECT_EQ(offset.offset_tso(), 202);
EXPECT_EQ(versionstamp, Versionstamp(200, 1));
MetaReader snapshot_reader(instance_id, txn_kv.get(), Versionstamp(150, 0));
std::unique_ptr<Transaction> snapshot_read_txn;
ASSERT_EQ(txn_kv->create_txn(&snapshot_read_txn), TxnErrorCode::TXN_OK);
ASSERT_EQ(snapshot_reader.get_table_stream_offset(snapshot_read_txn.get(), identity,
partition_id, &offset, &versionstamp),
TxnErrorCode::TXN_OK);
EXPECT_EQ(offset.offset_tso(), 101);
EXPECT_EQ(versionstamp, Versionstamp(100, 1));
TableStreamPartitionSetPB binding;
binding.mutable_identity()->CopyFrom(identity);
binding.add_partition_ids(partition_id);
binding.add_partition_ids(partition_id + 1);
TableStreamOffsetMap offsets;
TableStreamOffsetVersionstampMap versionstamps;
ASSERT_EQ(snapshot_reader.get_table_stream_offsets(snapshot_read_txn.get(), {binding}, &offsets,
&versionstamps, true),
TxnErrorCode::TXN_OK);
ASSERT_EQ(offsets.at(identity.stream_id()).size(), 1);
EXPECT_EQ(offsets.at(identity.stream_id()).at(partition_id).offset_tso(), 101);
EXPECT_EQ(versionstamps.at(identity.stream_id()).at(partition_id), Versionstamp(100, 1));
EXPECT_FALSE(offsets.at(identity.stream_id()).contains(partition_id + 1));
}
TEST(MetaReaderTest, BatchGetTableVersion) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
std::vector<int64_t> table_ids = {1001, 1002, 1003, 1004};
{
// Test empty input
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<int64_t> empty_ids;
std::unordered_map<int64_t, Versionstamp> table_versions;
TxnErrorCode err = meta_reader.get_table_versions(empty_ids, &table_versions);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(table_versions.empty());
}
{
// Test all keys not found
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, Versionstamp> table_versions;
TxnErrorCode err = meta_reader.get_table_versions(table_ids, &table_versions);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(table_versions.empty());
}
{
// Put some table versions (skip table_ids[1] to test partial results)
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
for (size_t i = 0; i < table_ids.size(); ++i) {
if (i == 1) continue; // Skip table_ids[1]
std::string table_version_key =
versioned::table_version_key({instance_id, table_ids[i]});
versioned_put(txn.get(), table_version_key, "");
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test partial results
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, Versionstamp> table_versions;
TxnErrorCode err = meta_reader.get_table_versions(table_ids, &table_versions);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(table_versions.size(), 3); // All except table_ids[1]
// Check min_read_version
Versionstamp min_expected = Versionstamp::max();
for (const auto& [table_id, version] : table_versions) {
min_expected = std::min(min_expected, version);
}
ASSERT_EQ(meta_reader.min_read_versionstamp(), min_expected);
for (size_t i = 0; i < table_ids.size(); ++i) {
if (i == 1) {
ASSERT_EQ(table_versions.find(table_ids[i]), table_versions.end());
} else {
ASSERT_NE(table_versions.find(table_ids[i]), table_versions.end());
}
}
}
{
// Put the missing table version
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_id, table_ids[1]});
versioned_put(txn.get(), table_version_key, "");
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test all keys found
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, Versionstamp> table_versions;
TxnErrorCode err = meta_reader.get_table_versions(txn.get(), table_ids, &table_versions);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(table_versions.size(), table_ids.size());
// Check min_read_version
Versionstamp min_expected = Versionstamp::max();
for (const auto& [table_id, version] : table_versions) {
min_expected = std::min(min_expected, version);
}
ASSERT_EQ(meta_reader.min_read_versionstamp(), min_expected);
for (int64_t table_id : table_ids) {
ASSERT_NE(table_versions.find(table_id), table_versions.end());
}
}
}
TEST(MetaReaderTest, GetPartitionVersion) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
int64_t partition_id = 2001;
{
// NOT FOUND
MetaReader meta_reader(instance_id, txn_kv.get());
VersionPB version_pb;
Versionstamp partition_version;
TxnErrorCode err =
meta_reader.get_partition_version(partition_id, &version_pb, &partition_version);
ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
}
{
// Put a partition 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_id, partition_id});
VersionPB version_pb;
version_pb.set_version(100);
versioned_put(txn.get(), partition_version_key, version_pb.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
VersionPB version_pb1;
Versionstamp partition_version1;
{
MetaReader meta_reader(instance_id, txn_kv.get());
TxnErrorCode err =
meta_reader.get_partition_version(partition_id, &version_pb1, &partition_version1);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(version_pb1.version(), 100);
ASSERT_EQ(meta_reader.min_read_versionstamp(), partition_version1);
}
{
// Put another partition 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_id, partition_id});
VersionPB version_pb;
version_pb.set_version(200);
versioned_put(txn.get(), partition_version_key, version_pb.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
VersionPB version_pb2;
Versionstamp partition_version2;
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
TxnErrorCode err = meta_reader.get_partition_version(txn.get(), partition_id, &version_pb2,
&partition_version2);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(version_pb2.version(), 200);
ASSERT_EQ(meta_reader.min_read_versionstamp(), partition_version2);
}
ASSERT_LT(partition_version1, partition_version2);
}
TEST(MetaReaderTest, BatchGetPartitionVersion) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
std::vector<int64_t> partition_ids = {2001, 2002, 2003, 2004};
{
// Test empty input
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<int64_t> empty_ids;
std::unordered_map<int64_t, VersionPB> versions;
std::unordered_map<int64_t, Versionstamp> versionstamps;
TxnErrorCode err = meta_reader.get_partition_versions(empty_ids, &versions, &versionstamps);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(versions.empty());
ASSERT_TRUE(versionstamps.empty());
}
{
// Put some partition versions (skip partition_ids[1] to test partial results)
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
for (size_t i = 0; i < partition_ids.size(); ++i) {
if (i == 1) continue; // Skip partition_ids[1]
std::string partition_version_key =
versioned::partition_version_key({instance_id, partition_ids[i]});
VersionPB version_pb;
version_pb.set_version(100 + i * 10); // Different versions: 100, 120, 130
versioned_put(txn.get(), partition_version_key, version_pb.SerializeAsString());
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test partial results
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, VersionPB> versions;
std::unordered_map<int64_t, Versionstamp> versionstamps;
TxnErrorCode err =
meta_reader.get_partition_versions(partition_ids, &versions, &versionstamps);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(versions.size(), 3); // All except partition_ids[1]
ASSERT_EQ(versionstamps.size(), 3);
// Check min_read_version
Versionstamp min_expected = Versionstamp::max();
for (const auto& [partition_id, version] : versionstamps) {
min_expected = std::min(min_expected, version);
}
ASSERT_EQ(meta_reader.min_read_versionstamp(), min_expected);
for (size_t i = 0; i < partition_ids.size(); ++i) {
if (i == 1) {
ASSERT_EQ(versions.find(partition_ids[i]), versions.end());
ASSERT_EQ(versionstamps.find(partition_ids[i]), versionstamps.end());
} else {
ASSERT_NE(versions.find(partition_ids[i]), versions.end());
ASSERT_NE(versionstamps.find(partition_ids[i]), versionstamps.end());
ASSERT_EQ(versions[partition_ids[i]].version(), 100 + i * 10);
}
}
}
{
// Put the missing partition 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_id, partition_ids[1]});
VersionPB version_pb;
version_pb.set_version(110);
versioned_put(txn.get(), partition_version_key, version_pb.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test all keys found
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, VersionPB> versions;
std::unordered_map<int64_t, Versionstamp> versionstamps;
TxnErrorCode err = meta_reader.get_partition_versions(txn.get(), partition_ids, &versions,
&versionstamps);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(versions.size(), partition_ids.size());
ASSERT_EQ(versionstamps.size(), partition_ids.size());
// Check min_read_version
Versionstamp min_expected = Versionstamp::max();
for (const auto& [partition_id, version] : versionstamps) {
min_expected = std::min(min_expected, version);
}
ASSERT_EQ(meta_reader.min_read_versionstamp(), min_expected);
for (size_t i = 0; i < partition_ids.size(); ++i) {
int64_t partition_id = partition_ids[i];
ASSERT_NE(versions.find(partition_id), versions.end());
ASSERT_NE(versionstamps.find(partition_id), versionstamps.end());
int32_t expected_version = (i == 1) ? 110 : 100 + i * 10;
ASSERT_EQ(versions[partition_id].version(), expected_version);
}
}
{
// Test only versionstamps (versions = nullptr)
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, Versionstamp> versionstamps;
TxnErrorCode err =
meta_reader.get_partition_versions(partition_ids, nullptr, &versionstamps);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(versionstamps.size(), partition_ids.size());
// Check min_read_version
Versionstamp min_expected = Versionstamp::max();
for (const auto& [partition_id, version] : versionstamps) {
min_expected = std::min(min_expected, version);
}
ASSERT_EQ(meta_reader.min_read_versionstamp(), min_expected);
}
{
// Test only versions (versionstamps = nullptr)
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, VersionPB> versions;
TxnErrorCode err = meta_reader.get_partition_versions(partition_ids, &versions, nullptr);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(versions.size(), partition_ids.size());
// For this case, min_read_version should still be updated even though versionstamps is nullptr
ASSERT_NE(meta_reader.min_read_versionstamp(), Versionstamp::max());
}
}
TEST(MetaReaderTest, GetPartitionVersionsWithPendingTxn) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
std::vector<int64_t> partition_ids = {2101, 2102, 2103, 2104};
{
// Test empty input
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<int64_t> empty_ids;
std::unordered_map<int64_t, int64_t> versions;
int64_t last_pending_txn_id = -1;
TxnErrorCode err = meta_reader.get_partition_versions(txn.get(), empty_ids, &versions,
&last_pending_txn_id);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(versions.empty());
ASSERT_EQ(last_pending_txn_id, -1);
}
{
// Put some partition versions (skip partition_ids[1] to test partial results)
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
for (size_t i = 0; i < partition_ids.size(); ++i) {
if (i == 1) continue; // Skip partition_ids[1]
std::string partition_version_key =
versioned::partition_version_key({instance_id, partition_ids[i]});
VersionPB version_pb;
version_pb.set_version(100 + i * 10); // Different versions: 100, 120, 130
// Add pending transaction for partition_ids[2]
if (i == 2) {
version_pb.add_pending_txn_ids(3001);
version_pb.add_pending_txn_ids(3002);
}
versioned_put(txn.get(), partition_version_key, version_pb.SerializeAsString());
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test partial results - partition not found should be set to 1
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, int64_t> versions;
int64_t last_pending_txn_id = -1;
TxnErrorCode err = meta_reader.get_partition_versions(txn.get(), partition_ids, &versions,
&last_pending_txn_id);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(versions.size(), 4); // All partition_ids, missing one should be set to 1
for (size_t i = 0; i < partition_ids.size(); ++i) {
int64_t partition_id = partition_ids[i];
ASSERT_NE(versions.find(partition_id), versions.end());
if (i == 1) {
// Missing partition should be set to 1
ASSERT_EQ(versions[partition_id], 1);
} else {
ASSERT_EQ(versions[partition_id], 100 + i * 10);
}
}
// Should return first pending transaction ID from partition_ids[2]
ASSERT_EQ(last_pending_txn_id, 3001);
}
{
// Put the missing partition version without pending transaction
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_id, partition_ids[1]});
VersionPB version_pb;
version_pb.set_version(110);
versioned_put(txn.get(), partition_version_key, version_pb.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test all keys found with no pending transactions
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, int64_t> versions;
int64_t last_pending_txn_id = -1;
TxnErrorCode err = meta_reader.get_partition_versions(txn.get(), partition_ids, &versions,
&last_pending_txn_id);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(versions.size(), partition_ids.size());
for (size_t i = 0; i < partition_ids.size(); ++i) {
int64_t partition_id = partition_ids[i];
ASSERT_NE(versions.find(partition_id), versions.end());
int32_t expected_version = (i == 1) ? 110 : 100 + i * 10;
ASSERT_EQ(versions[partition_id], expected_version);
}
// Should still return first pending transaction ID from partition_ids[2]
ASSERT_EQ(last_pending_txn_id, 3001);
}
{
// Remove pending transactions from partition_ids[2]
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_id, partition_ids[2]});
VersionPB version_pb;
version_pb.set_version(120);
// No pending transactions this time
versioned_put(txn.get(), partition_version_key, version_pb.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test with no pending transactions
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, int64_t> versions;
int64_t last_pending_txn_id = -1;
TxnErrorCode err = meta_reader.get_partition_versions(txn.get(), partition_ids, &versions,
&last_pending_txn_id);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(versions.size(), partition_ids.size());
for (size_t i = 0; i < partition_ids.size(); ++i) {
int64_t partition_id = partition_ids[i];
ASSERT_NE(versions.find(partition_id), versions.end());
int32_t expected_version = (i == 1) ? 110 : (i == 2) ? 120 : 100 + i * 10;
ASSERT_EQ(versions[partition_id], expected_version);
}
// No pending transactions, so last_pending_txn_id should remain -1
ASSERT_EQ(last_pending_txn_id, -1);
}
{
// Test using the convenience method (without snapshot parameter)
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, int64_t> versions;
int64_t last_pending_txn_id = -1;
TxnErrorCode err = meta_reader.get_partition_versions(txn.get(), partition_ids, &versions,
&last_pending_txn_id);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(versions.size(), partition_ids.size());
// Verify same results as the explicit snapshot=false test
for (size_t i = 0; i < partition_ids.size(); ++i) {
int64_t partition_id = partition_ids[i];
ASSERT_NE(versions.find(partition_id), versions.end());
int32_t expected_version = (i == 1) ? 110 : (i == 2) ? 120 : 100 + i * 10;
ASSERT_EQ(versions[partition_id], expected_version);
}
ASSERT_EQ(last_pending_txn_id, -1);
}
}
TEST(MetaReaderTest, GetTabletLoadStats) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
int64_t tablet_id = 3001;
{
// NOT FOUND
MetaReader meta_reader(instance_id, txn_kv.get());
TabletStatsPB tablet_stats;
Versionstamp tablet_stats_version;
TxnErrorCode err =
meta_reader.get_tablet_load_stats(tablet_id, &tablet_stats, &tablet_stats_version);
ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
}
{
// Put a tablet load stats
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_id, tablet_id});
TabletStatsPB tablet_stats;
tablet_stats.set_num_rows(1000);
tablet_stats.set_data_size(500000);
versioned_put(txn.get(), tablet_load_stats_key, tablet_stats.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
TabletStatsPB tablet_stats1;
Versionstamp tablet_stats_version1;
{
MetaReader meta_reader(instance_id, txn_kv.get());
TxnErrorCode err = meta_reader.get_tablet_load_stats(tablet_id, &tablet_stats1,
&tablet_stats_version1);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_stats1.num_rows(), 1000);
ASSERT_EQ(tablet_stats1.data_size(), 500000);
ASSERT_EQ(meta_reader.min_read_versionstamp(), tablet_stats_version1);
}
{
// Put another tablet load stats
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_id, tablet_id});
TabletStatsPB tablet_stats;
tablet_stats.set_num_rows(2000);
tablet_stats.set_data_size(1000000);
versioned_put(txn.get(), tablet_load_stats_key, tablet_stats.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
TabletStatsPB tablet_stats2;
Versionstamp tablet_stats_version2;
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
TxnErrorCode err = meta_reader.get_tablet_load_stats(txn.get(), tablet_id, &tablet_stats2,
&tablet_stats_version2);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_stats2.num_rows(), 2000);
ASSERT_EQ(tablet_stats2.data_size(), 1000000);
ASSERT_EQ(meta_reader.min_read_versionstamp(), tablet_stats_version2);
}
ASSERT_LT(tablet_stats_version1, tablet_stats_version2);
}
TEST(MetaReaderTest, BatchGetTabletLoadStats) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
std::vector<int64_t> tablet_ids = {3001, 3002, 3003, 3004};
{
// Test empty input
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<int64_t> empty_ids;
std::unordered_map<int64_t, TabletStatsPB> tablet_stats;
std::unordered_map<int64_t, Versionstamp> versionstamps;
TxnErrorCode err =
meta_reader.get_tablet_load_stats(empty_ids, &tablet_stats, &versionstamps);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(tablet_stats.empty());
ASSERT_TRUE(versionstamps.empty());
}
{
// Test all keys not found
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, TabletStatsPB> tablet_stats;
std::unordered_map<int64_t, Versionstamp> versionstamps;
TxnErrorCode err =
meta_reader.get_tablet_load_stats(tablet_ids, &tablet_stats, &versionstamps);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(tablet_stats.empty());
ASSERT_TRUE(versionstamps.empty());
}
{
// Put some tablet load stats (skip tablet_ids[1] to test partial results)
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_id, tablet_ids[i]});
TabletStatsPB tablet_stats;
tablet_stats.set_num_rows(1000 * (i + 1));
tablet_stats.set_data_size(500000 * (i + 1));
versioned_put(txn.get(), tablet_load_stats_key, tablet_stats.SerializeAsString());
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test partial results
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, TabletStatsPB> tablet_stats;
std::unordered_map<int64_t, Versionstamp> versionstamps;
TxnErrorCode err =
meta_reader.get_tablet_load_stats(tablet_ids, &tablet_stats, &versionstamps);
ASSERT_EQ(err, 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(meta_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]).num_rows(), 1000 * (i + 1));
ASSERT_EQ(tablet_stats.at(tablet_ids[i]).data_size(), 500000 * (i + 1));
}
}
}
{
// Put the missing tablet load stats
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_id, tablet_ids[1]});
TabletStatsPB tablet_stats;
tablet_stats.set_num_rows(2000);
tablet_stats.set_data_size(1000000);
versioned_put(txn.get(), tablet_load_stats_key, tablet_stats.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test all keys found
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, TabletStatsPB> tablet_stats;
std::unordered_map<int64_t, Versionstamp> versionstamps;
TxnErrorCode err = meta_reader.get_tablet_load_stats(txn.get(), tablet_ids, &tablet_stats,
&versionstamps);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_stats.size(), tablet_ids.size());
ASSERT_EQ(versionstamps.size(), tablet_ids.size());
// 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(meta_reader.min_read_versionstamp(), min_expected);
for (size_t i = 0; i < tablet_ids.size(); ++i) {
ASSERT_EQ(tablet_stats.count(tablet_ids[i]), 1);
ASSERT_EQ(versionstamps.count(tablet_ids[i]), 1);
if (i == 1) {
ASSERT_EQ(tablet_stats.at(tablet_ids[i]).num_rows(), 2000);
ASSERT_EQ(tablet_stats.at(tablet_ids[i]).data_size(), 1000000);
} else {
ASSERT_EQ(tablet_stats.at(tablet_ids[i]).num_rows(), 1000 * (i + 1));
ASSERT_EQ(tablet_stats.at(tablet_ids[i]).data_size(), 500000 * (i + 1));
}
}
}
}
TEST(MetaReaderTest, GetTabletCompactStats) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
int64_t tablet_id = 3001;
{
// Test key not found when no compact stats exist
MetaReader meta_reader(instance_id, txn_kv.get());
TabletStatsPB tablet_stats;
Versionstamp tablet_stats_version;
TxnErrorCode err = meta_reader.get_tablet_compact_stats(tablet_id, &tablet_stats,
&tablet_stats_version);
ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
}
{
// Put a comprehensive tablet compact stats
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_id, tablet_id});
TabletStatsPB tablet_stats;
// Set all compaction-related fields
tablet_stats.set_base_compaction_cnt(5);
tablet_stats.set_cumulative_compaction_cnt(10);
tablet_stats.set_cumulative_point(100);
tablet_stats.set_last_base_compaction_time_ms(1234567890);
tablet_stats.set_last_cumu_compaction_time_ms(2345678901);
tablet_stats.set_full_compaction_cnt(2);
tablet_stats.set_last_full_compaction_time_ms(3456789012);
// Set data-related fields
tablet_stats.set_num_rows(500);
tablet_stats.set_num_rowsets(8);
tablet_stats.set_num_segments(25);
tablet_stats.set_data_size(250000);
tablet_stats.set_index_size(25000);
tablet_stats.set_segment_size(300000);
versioned_put(txn.get(), tablet_compact_stats_key, tablet_stats.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
TabletStatsPB tablet_stats1;
Versionstamp tablet_stats_version1;
{
// Test successful get with created transaction
MetaReader meta_reader(instance_id, txn_kv.get());
TxnErrorCode err = meta_reader.get_tablet_compact_stats(tablet_id, &tablet_stats1,
&tablet_stats_version1);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Verify compaction-related fields
ASSERT_EQ(tablet_stats1.base_compaction_cnt(), 5);
ASSERT_EQ(tablet_stats1.cumulative_compaction_cnt(), 10);
ASSERT_EQ(tablet_stats1.cumulative_point(), 100);
ASSERT_EQ(tablet_stats1.last_base_compaction_time_ms(), 1234567890);
ASSERT_EQ(tablet_stats1.last_cumu_compaction_time_ms(), 2345678901);
ASSERT_EQ(tablet_stats1.full_compaction_cnt(), 2);
ASSERT_EQ(tablet_stats1.last_full_compaction_time_ms(), 3456789012);
// Verify data-related fields
ASSERT_EQ(tablet_stats1.num_rows(), 500);
ASSERT_EQ(tablet_stats1.num_rowsets(), 8);
ASSERT_EQ(tablet_stats1.num_segments(), 25);
ASSERT_EQ(tablet_stats1.data_size(), 250000);
ASSERT_EQ(tablet_stats1.index_size(), 25000);
ASSERT_EQ(tablet_stats1.segment_size(), 300000);
ASSERT_EQ(meta_reader.min_read_versionstamp(), tablet_stats_version1);
}
{
// Update tablet compact stats with new values
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_id, tablet_id});
TabletStatsPB tablet_stats;
// Update compaction-related fields
tablet_stats.set_base_compaction_cnt(8);
tablet_stats.set_cumulative_compaction_cnt(15);
tablet_stats.set_cumulative_point(200);
tablet_stats.set_last_base_compaction_time_ms(1234567890 + 10000);
tablet_stats.set_last_cumu_compaction_time_ms(2345678901 + 10000);
tablet_stats.set_full_compaction_cnt(3);
tablet_stats.set_last_full_compaction_time_ms(3456789012 + 10000);
// Update data-related fields
tablet_stats.set_num_rows(1000);
tablet_stats.set_num_rowsets(12);
tablet_stats.set_num_segments(40);
tablet_stats.set_data_size(500000);
tablet_stats.set_index_size(50000);
tablet_stats.set_segment_size(600000);
versioned_put(txn.get(), tablet_compact_stats_key, tablet_stats.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
TabletStatsPB tablet_stats2;
Versionstamp tablet_stats_version2;
{
// Test successful get with provided transaction
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
TxnErrorCode err = meta_reader.get_tablet_compact_stats(
txn.get(), tablet_id, &tablet_stats2, &tablet_stats_version2);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Verify updated compaction-related fields
ASSERT_EQ(tablet_stats2.base_compaction_cnt(), 8);
ASSERT_EQ(tablet_stats2.cumulative_compaction_cnt(), 15);
ASSERT_EQ(tablet_stats2.cumulative_point(), 200);
ASSERT_EQ(tablet_stats2.last_base_compaction_time_ms(), 1234567890 + 10000);
ASSERT_EQ(tablet_stats2.last_cumu_compaction_time_ms(), 2345678901 + 10000);
ASSERT_EQ(tablet_stats2.full_compaction_cnt(), 3);
ASSERT_EQ(tablet_stats2.last_full_compaction_time_ms(), 3456789012 + 10000);
// Verify updated data-related fields
ASSERT_EQ(tablet_stats2.num_rows(), 1000);
ASSERT_EQ(tablet_stats2.num_rowsets(), 12);
ASSERT_EQ(tablet_stats2.num_segments(), 40);
ASSERT_EQ(tablet_stats2.data_size(), 500000);
ASSERT_EQ(tablet_stats2.index_size(), 50000);
ASSERT_EQ(tablet_stats2.segment_size(), 600000);
ASSERT_EQ(meta_reader.min_read_versionstamp(), tablet_stats_version2);
}
// Verify version ordering
ASSERT_LT(tablet_stats_version1, tablet_stats_version2);
// Test with snapshot version functionality
Versionstamp snapshot_version;
{
// Get current snapshot version
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
int64_t version = 0;
ASSERT_EQ(txn->get_read_version(&version), TxnErrorCode::TXN_OK);
snapshot_version = Versionstamp(version, 1);
}
{
// Update compact stats again
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_id, tablet_id});
TabletStatsPB tablet_stats;
tablet_stats.set_num_rows(2000); // New value
tablet_stats.set_data_size(1000000); // New value
tablet_stats.set_base_compaction_cnt(10); // New value
versioned_put(txn.get(), tablet_compact_stats_key, tablet_stats.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test reading with snapshot version - should get old data
MetaReader meta_reader(instance_id, txn_kv.get(), snapshot_version);
TabletStatsPB tablet_stats;
Versionstamp tablet_stats_version;
TxnErrorCode err = meta_reader.get_tablet_compact_stats(tablet_id, &tablet_stats,
&tablet_stats_version, true);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Should get previous values (before the latest update)
ASSERT_EQ(tablet_stats.num_rows(), 1000); // Old value, not 2000
ASSERT_EQ(tablet_stats.data_size(), 500000); // Old value, not 1000000
ASSERT_EQ(tablet_stats.base_compaction_cnt(), 8); // Old value, not 10
}
{
// Test reading without snapshot - should get new data
MetaReader meta_reader(instance_id, txn_kv.get());
TabletStatsPB tablet_stats;
Versionstamp tablet_stats_version;
TxnErrorCode err = meta_reader.get_tablet_compact_stats(tablet_id, &tablet_stats,
&tablet_stats_version);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Should get latest values
ASSERT_EQ(tablet_stats.num_rows(), 2000); // Updated value
ASSERT_EQ(tablet_stats.data_size(), 1000000); // Updated value
ASSERT_EQ(tablet_stats.base_compaction_cnt(), 10); // Updated value
}
{
// Test with snapshot flag but no snapshot version
MetaReader meta_reader(instance_id, txn_kv.get());
TabletStatsPB tablet_stats;
Versionstamp tablet_stats_version;
TxnErrorCode err = meta_reader.get_tablet_compact_stats(tablet_id, &tablet_stats,
&tablet_stats_version, true);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Should get current values since snapshot flag is set but no snapshot version
ASSERT_EQ(tablet_stats.num_rows(), 2000);
ASSERT_EQ(tablet_stats.data_size(), 1000000);
}
{
// Test with nullptr parameters
MetaReader meta_reader(instance_id, txn_kv.get());
TxnErrorCode err = meta_reader.get_tablet_compact_stats(tablet_id, nullptr, nullptr);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
}
{
// Test getting non-existent tablet
MetaReader meta_reader(instance_id, txn_kv.get());
TabletStatsPB tablet_stats;
Versionstamp tablet_stats_version;
TxnErrorCode err = meta_reader.get_tablet_compact_stats(tablet_id + 1, &tablet_stats,
&tablet_stats_version);
ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
}
}
TEST(MetaReaderTest, GetTabletMergedStats) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
int64_t tablet_id = 3001;
{
// Test when both load and compact stats are not found
MetaReader meta_reader(instance_id, txn_kv.get());
TabletStatsPB tablet_stats;
Versionstamp tablet_stats_version;
TxnErrorCode err = meta_reader.get_tablet_merged_stats(tablet_id, &tablet_stats,
&tablet_stats_version);
ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
}
{
// Put tablet load stats
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_id, tablet_id});
TabletStatsPB load_stats;
load_stats.set_num_rows(1000);
load_stats.set_num_rowsets(10);
load_stats.set_num_segments(20);
load_stats.set_data_size(500000);
load_stats.set_index_size(50000);
load_stats.set_segment_size(600000);
versioned_put(txn.get(), tablet_load_stats_key, load_stats.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test when only load stats exist (compact stats not found)
MetaReader meta_reader(instance_id, txn_kv.get());
TabletStatsPB tablet_stats;
Versionstamp tablet_stats_version;
TxnErrorCode err = meta_reader.get_tablet_merged_stats(tablet_id, &tablet_stats,
&tablet_stats_version);
ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
}
Versionstamp load_version, compact_version;
{
// Put tablet compact stats
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_id, tablet_id});
TabletStatsPB compact_stats;
compact_stats.set_base_compaction_cnt(5);
compact_stats.set_cumulative_compaction_cnt(10);
compact_stats.set_cumulative_point(100);
compact_stats.set_last_base_compaction_time_ms(1234567890);
compact_stats.set_last_cumu_compaction_time_ms(2345678901);
compact_stats.set_full_compaction_cnt(2);
compact_stats.set_last_full_compaction_time_ms(3456789012);
compact_stats.set_num_rows(500);
compact_stats.set_num_rowsets(5);
compact_stats.set_num_segments(15);
compact_stats.set_data_size(250000);
compact_stats.set_index_size(25000);
compact_stats.set_segment_size(300000);
versioned_put(txn.get(), tablet_compact_stats_key, compact_stats.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
TabletStatsPB merged_stats;
Versionstamp merged_version;
{
// Test merged stats
MetaReader meta_reader(instance_id, txn_kv.get());
TxnErrorCode err =
meta_reader.get_tablet_merged_stats(tablet_id, &merged_stats, &merged_version);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Check compaction-related fields (copied from compact stats)
EXPECT_EQ(merged_stats.base_compaction_cnt(), 5);
EXPECT_EQ(merged_stats.cumulative_compaction_cnt(), 10);
EXPECT_EQ(merged_stats.cumulative_point(), 100);
EXPECT_EQ(merged_stats.last_base_compaction_time_ms(), 1234567890);
EXPECT_EQ(merged_stats.last_cumu_compaction_time_ms(), 2345678901);
EXPECT_EQ(merged_stats.full_compaction_cnt(), 2);
EXPECT_EQ(merged_stats.last_full_compaction_time_ms(), 3456789012);
// Check data-related fields (sum of load stats and compact stats)
EXPECT_EQ(merged_stats.num_rows(), 1500); // 1000 + 500
EXPECT_EQ(merged_stats.num_rowsets(), 15); // 10 + 5
EXPECT_EQ(merged_stats.num_segments(), 35); // 20 + 15
EXPECT_EQ(merged_stats.data_size(), 750000); // 500000 + 250000
EXPECT_EQ(merged_stats.index_size(), 75000); // 50000 + 25000
EXPECT_EQ(merged_stats.segment_size(), 900000); // 600000 + 300000
// Check min_read_version - should be updated after reading both load and compact stats
ASSERT_NE(meta_reader.min_read_versionstamp(), Versionstamp::max());
}
{
// Get individual versions for comparison
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
TabletStatsPB load_stats, compact_stats;
TxnErrorCode err1 =
meta_reader.get_tablet_load_stats(txn.get(), tablet_id, &load_stats, &load_version);
TxnErrorCode err2 = meta_reader.get_tablet_compact_stats(txn.get(), tablet_id,
&compact_stats, &compact_version);
ASSERT_EQ(err1, TxnErrorCode::TXN_OK);
ASSERT_EQ(err2, TxnErrorCode::TXN_OK);
// Merged version should be the min of load and compact versions
if (load_version < compact_version) {
ASSERT_EQ(merged_version, load_version);
} else {
ASSERT_EQ(merged_version, compact_version);
}
}
{
// Update load stats and test version update
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_id, tablet_id});
TabletStatsPB updated_load_stats;
updated_load_stats.set_num_rows(2000);
updated_load_stats.set_num_rowsets(20);
updated_load_stats.set_num_segments(40);
updated_load_stats.set_data_size(1000000);
updated_load_stats.set_index_size(100000);
updated_load_stats.set_segment_size(1200000);
versioned_put(txn.get(), tablet_load_stats_key, updated_load_stats.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
TabletStatsPB updated_merged_stats;
Versionstamp updated_merged_version;
{
// Test updated merged stats
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
TxnErrorCode err = meta_reader.get_tablet_merged_stats(
txn.get(), tablet_id, &updated_merged_stats, &updated_merged_version);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Check updated data-related fields (sum of updated load stats and compact stats)
EXPECT_EQ(updated_merged_stats.num_rows(), 2500); // 2000 + 500
EXPECT_EQ(updated_merged_stats.num_rowsets(), 25); // 20 + 5
EXPECT_EQ(updated_merged_stats.num_segments(), 55); // 40 + 15
EXPECT_EQ(updated_merged_stats.data_size(), 1250000); // 1000000 + 250000
EXPECT_EQ(updated_merged_stats.index_size(), 125000); // 100000 + 25000
EXPECT_EQ(updated_merged_stats.segment_size(), 1500000); // 1200000 + 300000
// Compaction fields should remain the same
EXPECT_EQ(updated_merged_stats.base_compaction_cnt(), 5);
EXPECT_EQ(updated_merged_stats.cumulative_compaction_cnt(), 10);
}
// Merged version should be updated
ASSERT_LT(merged_version, updated_merged_version);
}
TEST(MetaReaderTest, GetTabletIndex) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
int64_t tablet_id = 5001;
{
// NOT FOUND
MetaReader meta_reader(instance_id, txn_kv.get());
TabletIndexPB tablet_index;
TxnErrorCode err = meta_reader.get_tablet_index(tablet_id, &tablet_index);
ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
}
{
// Put a tablet index
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_id, tablet_id});
TabletIndexPB tablet_index;
tablet_index.set_db_id(1001);
tablet_index.set_table_id(2001);
tablet_index.set_index_id(3001);
tablet_index.set_partition_id(4001);
tablet_index.set_tablet_id(tablet_id);
txn->put(tablet_index_key, tablet_index.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
TabletIndexPB tablet_index1;
{
MetaReader meta_reader(instance_id, txn_kv.get());
TxnErrorCode err = meta_reader.get_tablet_index(tablet_id, &tablet_index1);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_index1.db_id(), 1001);
ASSERT_EQ(tablet_index1.table_id(), 2001);
ASSERT_EQ(tablet_index1.index_id(), 3001);
ASSERT_EQ(tablet_index1.partition_id(), 4001);
ASSERT_EQ(tablet_index1.tablet_id(), tablet_id);
}
{
// Put another tablet index (update)
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_id, tablet_id});
TabletIndexPB tablet_index;
tablet_index.set_db_id(1002);
tablet_index.set_table_id(2002);
tablet_index.set_index_id(3002);
tablet_index.set_partition_id(4002);
tablet_index.set_tablet_id(tablet_id);
txn->put(tablet_index_key, tablet_index.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
TabletIndexPB tablet_index2;
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
TxnErrorCode err = meta_reader.get_tablet_index(txn.get(), tablet_id, &tablet_index2);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_index2.db_id(), 1002);
ASSERT_EQ(tablet_index2.table_id(), 2002);
ASSERT_EQ(tablet_index2.index_id(), 3002);
ASSERT_EQ(tablet_index2.partition_id(), 4002);
ASSERT_EQ(tablet_index2.tablet_id(), tablet_id);
}
}
TEST(MetaReaderTest, BatchGetTabletIndexes) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
std::vector<int64_t> tablet_ids = {5001, 5002, 5003, 5004};
{
// Test empty input
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<int64_t> empty_ids;
std::unordered_map<int64_t, TabletIndexPB> tablet_indexes;
TxnErrorCode err = meta_reader.get_tablet_indexes(empty_ids, &tablet_indexes);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(tablet_indexes.empty());
}
{
// Put some tablet indexes (skip tablet_ids[1] to test partial results)
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_index_key =
versioned::tablet_index_key({instance_id, tablet_ids[i]});
TabletIndexPB tablet_index;
tablet_index.set_db_id(1000 + i);
tablet_index.set_table_id(2000 + i);
tablet_index.set_index_id(3000 + i);
tablet_index.set_partition_id(4000 + i);
tablet_index.set_tablet_id(tablet_ids[i]);
txn->put(tablet_index_key, tablet_index.SerializeAsString());
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test partial results
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, TabletIndexPB> tablet_indexes;
TxnErrorCode err = meta_reader.get_tablet_indexes(tablet_ids, &tablet_indexes);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_indexes.size(), 3); // All except tablet_ids[1]
for (size_t i = 0; i < tablet_ids.size(); ++i) {
if (i == 1) {
ASSERT_EQ(tablet_indexes.find(tablet_ids[i]), tablet_indexes.end());
} else {
ASSERT_NE(tablet_indexes.find(tablet_ids[i]), tablet_indexes.end());
ASSERT_EQ(tablet_indexes[tablet_ids[i]].db_id(), 1000 + i);
ASSERT_EQ(tablet_indexes[tablet_ids[i]].table_id(), 2000 + i);
ASSERT_EQ(tablet_indexes[tablet_ids[i]].index_id(), 3000 + i);
ASSERT_EQ(tablet_indexes[tablet_ids[i]].partition_id(), 4000 + i);
ASSERT_EQ(tablet_indexes[tablet_ids[i]].tablet_id(), tablet_ids[i]);
}
}
}
{
// Put the missing tablet index
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_id, tablet_ids[1]});
TabletIndexPB tablet_index;
tablet_index.set_db_id(1001);
tablet_index.set_table_id(2001);
tablet_index.set_index_id(3001);
tablet_index.set_partition_id(4001);
tablet_index.set_tablet_id(tablet_ids[1]);
txn->put(tablet_index_key, tablet_index.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test all keys found
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, TabletIndexPB> tablet_indexes;
TxnErrorCode err = meta_reader.get_tablet_indexes(txn.get(), tablet_ids, &tablet_indexes);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_indexes.size(), tablet_ids.size());
for (size_t i = 0; i < tablet_ids.size(); ++i) {
int64_t tablet_id = tablet_ids[i];
ASSERT_NE(tablet_indexes.find(tablet_id), tablet_indexes.end());
int64_t expected_db_id = (i == 1) ? 1001 : 1000 + i;
int64_t expected_table_id = (i == 1) ? 2001 : 2000 + i;
int64_t expected_index_id = (i == 1) ? 3001 : 3000 + i;
int64_t expected_partition_id = (i == 1) ? 4001 : 4000 + i;
ASSERT_EQ(tablet_indexes[tablet_id].db_id(), expected_db_id);
ASSERT_EQ(tablet_indexes[tablet_id].table_id(), expected_table_id);
ASSERT_EQ(tablet_indexes[tablet_id].index_id(), expected_index_id);
ASSERT_EQ(tablet_indexes[tablet_id].partition_id(), expected_partition_id);
ASSERT_EQ(tablet_indexes[tablet_id].tablet_id(), tablet_id);
}
}
}
TEST(MetaReaderTest, GetRowsetMetas) {
using doris::RowsetMetaCloudPB;
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
int64_t tablet_id = 4001;
int64_t start_version = 1;
int64_t end_version = 10;
{
// Test empty result when no rowsets exist
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<RowsetMetaCloudPB> rowset_metas;
TxnErrorCode err =
meta_reader.get_rowset_metas(tablet_id, start_version, end_version, &rowset_metas);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(rowset_metas.empty());
}
// Create some load rowsets (import scenario)
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
// Create load rowsets with versions 2, 3, 4, 5
for (int64_t version = 2; version <= 5; ++version) {
std::string load_key =
versioned::meta_rowset_load_key({instance_id, tablet_id, version});
RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2(fmt::format("load_rowset_{}", 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);
ASSERT_TRUE(versioned::document_put(txn.get(), load_key, std::move(rowset_meta)));
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test getting load rowsets
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<RowsetMetaCloudPB> rowset_metas;
TxnErrorCode err = meta_reader.get_rowset_metas(tablet_id, 2, 5, &rowset_metas);
ASSERT_EQ(err, TxnErrorCode::TXN_OK) << dump_range(txn_kv.get());
ASSERT_EQ(rowset_metas.size(), 4) << [&] {
std::ostringstream oss;
for (const auto& meta : rowset_metas) {
oss << meta.rowset_id_v2() << "[" << meta.start_version() << ", "
<< meta.end_version() << "]\n";
}
return oss.str();
}() << dump_range(txn_kv.get());
// Verify rowsets are sorted by end_version
for (size_t i = 0; i < rowset_metas.size(); ++i) {
ASSERT_EQ(rowset_metas[i].end_version(), 2 + i);
ASSERT_EQ(rowset_metas[i].start_version(), 2 + i);
ASSERT_EQ(rowset_metas[i].num_rows(), 100 * (2 + i));
}
}
// Create compact rowset that covers versions 3-4
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string compact_key = versioned::meta_rowset_compact_key({instance_id, tablet_id, 4});
RowsetMetaCloudPB compact_rowset;
compact_rowset.set_rowset_id(0);
compact_rowset.set_rowset_id_v2("compact_rowset_3_4");
compact_rowset.set_start_version(3);
compact_rowset.set_end_version(4);
compact_rowset.set_num_rows(700); // 300 + 400 = 700
compact_rowset.set_tablet_id(tablet_id);
ASSERT_TRUE(versioned::document_put(txn.get(), compact_key, std::move(compact_rowset)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test getting rowsets after compaction - compact should override load rowsets
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<RowsetMetaCloudPB> rowset_metas;
TxnErrorCode err = meta_reader.get_rowset_metas(tablet_id, 2, 5, &rowset_metas);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_metas.size(), 3); // version 2, compact(3-4), version 5
// Check first rowset (version 2)
ASSERT_EQ(rowset_metas[0].end_version(), 2);
ASSERT_EQ(rowset_metas[0].start_version(), 2);
ASSERT_EQ(rowset_metas[0].rowset_id_v2(), "load_rowset_2");
// Check compact rowset (versions 3-4)
ASSERT_EQ(rowset_metas[1].end_version(), 4);
ASSERT_EQ(rowset_metas[1].start_version(), 3);
ASSERT_EQ(rowset_metas[1].rowset_id_v2(), "compact_rowset_3_4");
ASSERT_EQ(rowset_metas[1].num_rows(), 700);
// Check last rowset (version 5)
ASSERT_EQ(rowset_metas[2].end_version(), 5);
ASSERT_EQ(rowset_metas[2].start_version(), 5);
ASSERT_EQ(rowset_metas[2].rowset_id_v2(), "load_rowset_5");
}
// Test range query that only includes part of the rowsets
{
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<RowsetMetaCloudPB> rowset_metas;
TxnErrorCode err = meta_reader.get_rowset_metas(tablet_id, 3, 4, &rowset_metas);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_metas.size(), 1); // Only the compact rowset
ASSERT_EQ(rowset_metas[0].rowset_id_v2(), "compact_rowset_3_4");
}
// Test with snapshot version functionality
Versionstamp snapshot_version;
{
// Get current snapshot version
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<RowsetMetaCloudPB> rowset_metas;
TxnErrorCode err = meta_reader.get_rowset_metas(txn.get(), tablet_id, 2, 5, &rowset_metas);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Get version from the transaction's read version
int64_t version = 0;
ASSERT_EQ(txn->get_read_version(&version), TxnErrorCode::TXN_OK);
snapshot_version = Versionstamp(version, 1);
LOG(INFO) << "Snapshot version: " << snapshot_version.version();
LOG(INFO) << "Snapshot version: " << snapshot_version.to_string();
}
// Add another compact rowset that covers version 5
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string compact_key = versioned::meta_rowset_compact_key({instance_id, tablet_id, 5});
RowsetMetaCloudPB compact_rowset;
compact_rowset.set_rowset_id(0);
compact_rowset.set_rowset_id_v2("compact_rowset_5");
compact_rowset.set_start_version(5);
compact_rowset.set_end_version(5);
compact_rowset.set_num_rows(500);
compact_rowset.set_tablet_id(tablet_id);
ASSERT_TRUE(versioned::document_put(txn.get(), compact_key, std::move(compact_rowset)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test reading with snapshot version - should get old data
LOG(INFO) << "Reading with snapshot version: " << snapshot_version.version();
MetaReader meta_reader(instance_id, txn_kv.get(), snapshot_version);
std::vector<RowsetMetaCloudPB> rowset_metas;
TxnErrorCode err = meta_reader.get_rowset_metas(tablet_id, 2, 5, &rowset_metas);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_metas.size(), 3) << [&]() {
std::ostringstream oss;
for (const auto& meta : rowset_metas) {
oss << meta.rowset_id_v2() << "[" << meta.start_version() << ", "
<< meta.end_version() << "]\n";
}
return oss.str();
}() << dump_range(txn_kv.get()); // Should still see load_rowset_5, not compact_rowset_5
ASSERT_EQ(rowset_metas[2].rowset_id_v2(), "load_rowset_5") << dump_range(txn_kv.get());
}
{
// Test reading without snapshot - should get new data
LOG(INFO) << "Reading without snapshot version";
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<RowsetMetaCloudPB> rowset_metas;
TxnErrorCode err = meta_reader.get_rowset_metas(tablet_id, 2, 5, &rowset_metas);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_metas.size(), 3); // version 2, compact(3-4), compact(5)
ASSERT_EQ(rowset_metas[2].rowset_id_v2(), "compact_rowset_5");
}
// Add a compact rowset cover [3-5]
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string compact_key = versioned::meta_rowset_compact_key({instance_id, tablet_id, 5});
RowsetMetaCloudPB compact_rowset;
compact_rowset.set_rowset_id(0);
compact_rowset.set_rowset_id_v2("compact_rowset_3_5");
compact_rowset.set_start_version(3);
compact_rowset.set_end_version(5);
compact_rowset.set_num_rows(1200); // 300 + 400 + 500
compact_rowset.set_tablet_id(tablet_id);
ASSERT_TRUE(versioned::document_put(txn.get(), compact_key, std::move(compact_rowset)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test getting rowsets after new compaction
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<RowsetMetaCloudPB> rowset_metas;
TxnErrorCode err = meta_reader.get_rowset_metas(tablet_id, 2, 5, &rowset_metas);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_metas.size(), 2); // version 2, compact(3-5)
// Check first rowset (version 2)
ASSERT_EQ(rowset_metas[0].end_version(), 2);
ASSERT_EQ(rowset_metas[0].start_version(), 2);
ASSERT_EQ(rowset_metas[0].rowset_id_v2(), "load_rowset_2");
// Check new compact rowset (versions 3-5)
ASSERT_EQ(rowset_metas[1].end_version(), 5);
ASSERT_EQ(rowset_metas[1].start_version(), 3);
ASSERT_EQ(rowset_metas[1].rowset_id_v2(), "compact_rowset_3_5");
}
{
// Test getting rowset with old snapshot version
LOG(INFO) << "Reading with old snapshot version: " << snapshot_version.version();
MetaReader meta_reader(instance_id, txn_kv.get(), snapshot_version);
std::vector<RowsetMetaCloudPB> rowset_metas;
TxnErrorCode err = meta_reader.get_rowset_metas(tablet_id, 2, 5, &rowset_metas);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_metas.size(), 3) << [&]() {
std::ostringstream oss;
for (const auto& meta : rowset_metas) {
oss << meta.rowset_id_v2() << "[" << meta.start_version() << ", "
<< meta.end_version() << "]\n";
}
return oss.str();
}() << dump_range(txn_kv.get());
ASSERT_EQ(rowset_metas[2].rowset_id_v2(), "load_rowset_5") << dump_range(txn_kv.get());
// Should still see load_rowset_5, not compact_rowset_3_5
ASSERT_EQ(rowset_metas[2].start_version(), 5);
ASSERT_EQ(rowset_metas[2].end_version(), 5);
ASSERT_EQ(rowset_metas[2].num_rows(), 500) << dump_range(txn_kv.get());
}
}
TEST(MetaReaderTest, GetRowsetMetasMinReadVersionstamp) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
const std::string instance_id = "rowset_min_version";
constexpr int64_t tablet_id = 4001;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
auto put_rowset = [&](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("L1", 1, 1, Versionstamp(70, 1), false);
put_rowset("L2", 2, 2, Versionstamp(80, 1), false);
put_rowset("L3", 3, 3, Versionstamp(90, 1), false);
put_rowset("A", 2, 3, Versionstamp(150, 2), true);
put_rowset("L4", 4, 4, Versionstamp(170, 1), false);
// Compact keys are scanned by descending end version: A also hides this older compact.
put_rowset("old_compact", 2, 2, Versionstamp(110, 1), true);
put_rowset("old_A", 2, 3, Versionstamp(120, 1), true);
TabletStatsPB stats;
stats.set_num_rowsets(1);
versioned_put(txn.get(), versioned::tablet_load_stats_key({instance_id, tablet_id}),
Versionstamp(70, 2), stats.SerializeAsString());
versioned_put(txn.get(), versioned::tablet_compact_stats_key({instance_id, tablet_id}),
Versionstamp(75, 1), stats.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
using Rowset = std::tuple<std::string, int64_t, int64_t>;
auto check_rowsets = [&](MetaReader& reader, int64_t start, int64_t end,
const std::vector<Rowset>& expected, Versionstamp expected_min) {
SCOPED_TRACE(fmt::format("rowset range [{}, {}]", start, end));
std::vector<doris::RowsetMetaCloudPB> rowsets;
ASSERT_EQ(reader.get_rowset_metas(tablet_id, start, end, &rowsets), TxnErrorCode::TXN_OK);
std::vector<Rowset> actual;
for (const auto& rowset : rowsets) {
EXPECT_EQ(rowset.tablet_id(), tablet_id);
actual.emplace_back(rowset.rowset_id_v2(), rowset.start_version(),
rowset.end_version());
}
EXPECT_EQ(actual, expected);
EXPECT_EQ(reader.min_read_versionstamp(), expected_min)
<< "actual metadata version: " << reader.min_read_version()
<< ", expected: " << expected_min.version();
};
{
MetaReader reader(instance_id, txn_kv.get());
// Covered L2@80/L3@90 must not lower the minimum of A@150/L4@170.
check_rowsets(reader, 2, 4, {{"A", 2, 3}, {"L4", 4, 4}}, Versionstamp(150, 2));
}
{
MetaReader reader(instance_id, txn_kv.get());
check_rowsets(reader, 2, 3, {{"A", 2, 3}}, Versionstamp(150, 2));
}
{
MetaReader reader(instance_id, txn_kv.get());
// A selected old load still contributes its full metadata versionstamp.
check_rowsets(reader, 1, 4, {{"L1", 1, 1}, {"A", 2, 3}, {"L4", 4, 4}}, Versionstamp(70, 1));
}
{
MetaReader reader(instance_id, txn_kv.get(), Versionstamp(100, 0));
check_rowsets(reader, 2, 3, {{"L2", 2, 2}, {"L3", 3, 3}}, Versionstamp(80, 1));
}
for (bool load_stats : {true, false}) {
SCOPED_TRACE(load_stats ? "prior load stats" : "prior compact stats");
MetaReader reader(instance_id, txn_kv.get());
auto err = load_stats ? reader.get_tablet_load_stats(tablet_id, &stats, nullptr)
: reader.get_tablet_compact_stats(tablet_id, &stats, nullptr);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
auto previous_min = load_stats ? Versionstamp(70, 2) : Versionstamp(75, 1);
EXPECT_EQ(reader.min_read_versionstamp(), previous_min);
check_rowsets(reader, 2, 4, {{"A", 2, 3}, {"L4", 4, 4}}, previous_min);
check_rowsets(reader, 5, 6, {}, previous_min);
}
{
MetaReader reader(instance_id, txn_kv.get());
check_rowsets(reader, 5, 6, {}, Versionstamp::max());
check_rowsets(reader, 2, 3, {{"A", 2, 3}}, Versionstamp(150, 2));
check_rowsets(reader, 4, 4, {{"L4", 4, 4}}, Versionstamp(150, 2));
check_rowsets(reader, 1, 1, {{"L1", 1, 1}}, Versionstamp(70, 1));
check_rowsets(reader, 2, 4, {{"A", 2, 3}, {"L4", 4, 4}}, Versionstamp(70, 1));
check_rowsets(reader, 5, 6, {}, Versionstamp(70, 1));
}
{
// Equal transaction versions must still compare the two-byte order component.
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
put_rowset("L5", 5, 5, Versionstamp(150, 1), false);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
MetaReader reader(instance_id, txn_kv.get());
check_rowsets(reader, 2, 5, {{"A", 2, 3}, {"L4", 4, 4}, {"L5", 5, 5}},
Versionstamp(150, 1));
}
}
TEST(MetaReaderTest, GetRowsetMetasScanFailure) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
const std::string instance_id = "rowset_scan_failure";
constexpr int64_t tablet_id = 4001;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
doris::RowsetMetaCloudPB load;
load.set_rowset_id(0);
load.set_start_version(2);
load.set_end_version(2);
ASSERT_TRUE(versioned::document_put(
txn.get(), versioned::meta_rowset_load_key({instance_id, tablet_id, 2}),
Versionstamp(80, 1), std::move(load)));
// A malformed compact document fails after the load scan has succeeded.
versioned_put(txn.get(), versioned::meta_rowset_compact_key({instance_id, tablet_id, 3}),
Versionstamp(150, 1), "invalid protobuf");
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
MetaReader reader(instance_id, txn_kv.get());
std::vector<doris::RowsetMetaCloudPB> rowsets(1);
rowsets[0].set_rowset_id_v2("unchanged");
ASSERT_EQ(reader.get_rowset_metas(tablet_id, 2, 3, &rowsets), TxnErrorCode::TXN_INVALID_DATA);
ASSERT_EQ(rowsets.size(), 1);
EXPECT_EQ(rowsets[0].rowset_id_v2(), "unchanged");
EXPECT_EQ(reader.min_read_versionstamp(), Versionstamp::max());
}
TEST(MetaReaderTest, GetPartitionPendingTxnId) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
int64_t partition_id = 3001;
{
// Test with non-existent partition
MetaReader meta_reader(instance_id, txn_kv.get());
int64_t first_txn_id;
TxnErrorCode err = meta_reader.get_partition_pending_txn_id(partition_id, &first_txn_id);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
EXPECT_EQ(first_txn_id, -1) << "Expected -1 for non-existent partition";
}
{
// Put a partition version without pending transactions
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_id, partition_id});
VersionPB version_pb;
version_pb.set_version(100);
versioned_put(txn.get(), partition_version_key, version_pb.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test with no pending transactions - should return -1
MetaReader meta_reader(instance_id, txn_kv.get());
int64_t first_txn_id;
TxnErrorCode err = meta_reader.get_partition_pending_txn_id(partition_id, &first_txn_id);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(first_txn_id, -1);
}
{
// Put a partition version with pending transactions
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_id, partition_id});
VersionPB version_pb;
version_pb.set_version(200);
version_pb.add_pending_txn_ids(1001);
version_pb.add_pending_txn_ids(1002);
version_pb.add_pending_txn_ids(1003);
versioned_put(txn.get(), partition_version_key, version_pb.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test with pending transactions - should return first txn_id
MetaReader meta_reader(instance_id, txn_kv.get());
int64_t first_txn_id;
TxnErrorCode err = meta_reader.get_partition_pending_txn_id(partition_id, &first_txn_id);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(first_txn_id, 1001);
}
{
// Test with transaction parameter
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
int64_t first_txn_id;
TxnErrorCode err =
meta_reader.get_partition_pending_txn_id(txn.get(), partition_id, &first_txn_id);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(first_txn_id, 1001);
}
{
// Put a partition version with single pending transaction
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_id, partition_id});
VersionPB version_pb;
version_pb.set_version(300);
version_pb.add_pending_txn_ids(2001);
versioned_put(txn.get(), partition_version_key, version_pb.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test with single pending transaction
MetaReader meta_reader(instance_id, txn_kv.get());
int64_t first_txn_id;
TxnErrorCode err = meta_reader.get_partition_pending_txn_id(partition_id, &first_txn_id);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(first_txn_id, 2001);
}
{
// Test with snapshot functionality
// First get the current snapshot version
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
VersionPB current_version;
Versionstamp current_versionstamp;
MetaReader current_reader(instance_id, txn_kv.get());
TxnErrorCode err = current_reader.get_partition_version(
txn.get(), partition_id, &current_version, &current_versionstamp);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Create a MetaReader with snapshot
Versionstamp snapshot_version(current_version.version() + 1, 0);
MetaReader snapshot_reader(instance_id, txn_kv.get(), snapshot_version);
int64_t first_txn_id;
err = snapshot_reader.get_partition_pending_txn_id(partition_id, &first_txn_id, true);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(first_txn_id, 2001);
}
}
TEST(MetaReaderTest, GetIndexIndex) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
int64_t index_id = 3001;
{
// NOT FOUND
MetaReader meta_reader(instance_id, txn_kv.get());
IndexIndexPB index;
TxnErrorCode err = meta_reader.get_index_index(index_id, &index);
ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
}
{
// Put an index index
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_id, index_id});
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);
}
{
MetaReader meta_reader(instance_id, txn_kv.get());
IndexIndexPB index;
TxnErrorCode err = meta_reader.get_index_index(index_id, &index);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(index.db_id(), 1001);
ASSERT_EQ(index.table_id(), 2001);
}
}
TEST(MetaReaderTest, GetPartitionIndex) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
int64_t partition_id = 4001;
{
// NOT FOUND
MetaReader meta_reader(instance_id, txn_kv.get());
PartitionIndexPB partition_index;
TxnErrorCode err = meta_reader.get_partition_index(partition_id, &partition_index);
ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
}
{
// Put a partition index
PartitionIndexPB partition_index_pb;
partition_index_pb.set_db_id(100);
partition_index_pb.set_table_id(200);
std::string partition_index_value;
ASSERT_TRUE(partition_index_pb.SerializeToString(&partition_index_value));
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_id, partition_id});
txn->put(partition_index_key, partition_index_value);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
MetaReader meta_reader(instance_id, txn_kv.get());
PartitionIndexPB partition_index;
TxnErrorCode err = meta_reader.get_partition_index(partition_id, &partition_index);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(partition_index.db_id(), 100);
ASSERT_EQ(partition_index.table_id(), 200);
}
}
TEST(MetaReaderTest, GetLoadRowsetMeta) {
using doris::RowsetMetaCloudPB;
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
int64_t tablet_id = 5001;
int64_t version = 10;
{
// Test key not found when no rowset exists
MetaReader meta_reader(instance_id, txn_kv.get());
RowsetMetaCloudPB rowset_meta;
TxnErrorCode err = meta_reader.get_load_rowset_meta(tablet_id, version, &rowset_meta);
ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
}
// Create a load rowset
RowsetMetaCloudPB expected_rowset_meta;
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string load_key = versioned::meta_rowset_load_key({instance_id, tablet_id, version});
expected_rowset_meta.set_rowset_id(0);
expected_rowset_meta.set_rowset_id_v2(fmt::format("test_load_rowset_{}", version));
expected_rowset_meta.set_start_version(version);
expected_rowset_meta.set_end_version(version);
expected_rowset_meta.set_num_rows(1000);
expected_rowset_meta.set_tablet_id(tablet_id);
ASSERT_TRUE(versioned::document_put(txn.get(), load_key, std::move(expected_rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test successful get with created transaction
MetaReader meta_reader(instance_id, txn_kv.get());
RowsetMetaCloudPB rowset_meta;
TxnErrorCode err = meta_reader.get_load_rowset_meta(tablet_id, version, &rowset_meta);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Verify fields match
ASSERT_EQ(rowset_meta.rowset_id_v2(), expected_rowset_meta.rowset_id_v2());
ASSERT_EQ(rowset_meta.start_version(), expected_rowset_meta.start_version());
ASSERT_EQ(rowset_meta.end_version(), expected_rowset_meta.end_version());
ASSERT_EQ(rowset_meta.num_rows(), expected_rowset_meta.num_rows());
ASSERT_EQ(rowset_meta.tablet_id(), expected_rowset_meta.tablet_id());
}
{
// Test successful get with provided transaction
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
RowsetMetaCloudPB rowset_meta;
TxnErrorCode err =
meta_reader.get_load_rowset_meta(txn.get(), tablet_id, version, &rowset_meta);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Verify key fields match
ASSERT_EQ(rowset_meta.rowset_id_v2(), expected_rowset_meta.rowset_id_v2());
ASSERT_EQ(rowset_meta.start_version(), expected_rowset_meta.start_version());
ASSERT_EQ(rowset_meta.end_version(), expected_rowset_meta.end_version());
ASSERT_EQ(rowset_meta.tablet_id(), expected_rowset_meta.tablet_id());
}
// Test with snapshot version functionality
Versionstamp snapshot_version;
{
// Get current snapshot version
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
int64_t version_value = 0;
ASSERT_EQ(txn->get_read_version(&version_value), TxnErrorCode::TXN_OK);
snapshot_version = Versionstamp(version_value, 1);
}
// Update the rowset
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string load_key = versioned::meta_rowset_load_key({instance_id, tablet_id, version});
RowsetMetaCloudPB updated_rowset_meta = expected_rowset_meta;
updated_rowset_meta.set_num_rows(2000); // Update row count
ASSERT_TRUE(versioned::document_put(txn.get(), load_key, std::move(updated_rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test reading with snapshot version - should get old data
MetaReader meta_reader(instance_id, txn_kv.get(), snapshot_version);
RowsetMetaCloudPB rowset_meta;
TxnErrorCode err = meta_reader.get_load_rowset_meta(tablet_id, version, &rowset_meta);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Should get original values
ASSERT_EQ(rowset_meta.num_rows(), 1000);
}
{
// Test reading without snapshot - should get new data
MetaReader meta_reader(instance_id, txn_kv.get());
RowsetMetaCloudPB rowset_meta;
TxnErrorCode err = meta_reader.get_load_rowset_meta(tablet_id, version, &rowset_meta);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Should get updated values
ASSERT_EQ(rowset_meta.num_rows(), 2000);
}
{
// Test with snapshot flag
MetaReader meta_reader(instance_id, txn_kv.get());
RowsetMetaCloudPB rowset_meta;
TxnErrorCode err = meta_reader.get_load_rowset_meta(tablet_id, version, &rowset_meta, true);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Should get current values since snapshot flag is set but no snapshot version
ASSERT_EQ(rowset_meta.num_rows(), 2000);
}
{
// Test getting non-existent version
MetaReader meta_reader(instance_id, txn_kv.get());
RowsetMetaCloudPB rowset_meta;
TxnErrorCode err = meta_reader.get_load_rowset_meta(tablet_id, version + 1, &rowset_meta);
ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
}
}
TEST(MetaReaderTest, GetCompactRowsetMeta) {
using doris::RowsetMetaCloudPB;
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
int64_t tablet_id = 5001;
int64_t version = 10;
{
// Test key not found when no rowset exists
MetaReader meta_reader(instance_id, txn_kv.get());
RowsetMetaCloudPB rowset_meta;
TxnErrorCode err = meta_reader.get_compact_rowset_meta(tablet_id, version, &rowset_meta);
ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
}
// Create a load rowset
RowsetMetaCloudPB expected_rowset_meta;
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string load_key =
versioned::meta_rowset_compact_key({instance_id, tablet_id, version});
expected_rowset_meta.set_rowset_id(0);
expected_rowset_meta.set_rowset_id_v2(fmt::format("test_compact_rowset_{}", version));
expected_rowset_meta.set_start_version(version);
expected_rowset_meta.set_end_version(version);
expected_rowset_meta.set_num_rows(1000);
expected_rowset_meta.set_tablet_id(tablet_id);
ASSERT_TRUE(versioned::document_put(txn.get(), load_key, std::move(expected_rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test successful get with created transaction
MetaReader meta_reader(instance_id, txn_kv.get());
RowsetMetaCloudPB rowset_meta;
TxnErrorCode err = meta_reader.get_compact_rowset_meta(tablet_id, version, &rowset_meta);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Verify fields match
ASSERT_EQ(rowset_meta.rowset_id_v2(), expected_rowset_meta.rowset_id_v2());
ASSERT_EQ(rowset_meta.start_version(), expected_rowset_meta.start_version());
ASSERT_EQ(rowset_meta.end_version(), expected_rowset_meta.end_version());
ASSERT_EQ(rowset_meta.num_rows(), expected_rowset_meta.num_rows());
ASSERT_EQ(rowset_meta.tablet_id(), expected_rowset_meta.tablet_id());
}
{
// Test successful get with provided transaction
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
RowsetMetaCloudPB rowset_meta;
TxnErrorCode err =
meta_reader.get_compact_rowset_meta(txn.get(), tablet_id, version, &rowset_meta);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Verify key fields match
ASSERT_EQ(rowset_meta.rowset_id_v2(), expected_rowset_meta.rowset_id_v2());
ASSERT_EQ(rowset_meta.start_version(), expected_rowset_meta.start_version());
ASSERT_EQ(rowset_meta.end_version(), expected_rowset_meta.end_version());
ASSERT_EQ(rowset_meta.tablet_id(), expected_rowset_meta.tablet_id());
}
// Test with snapshot version functionality
Versionstamp snapshot_version;
{
// Get current snapshot version
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
int64_t version_value = 0;
ASSERT_EQ(txn->get_read_version(&version_value), TxnErrorCode::TXN_OK);
snapshot_version = Versionstamp(version_value, 1);
}
// Update the rowset
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string load_key =
versioned::meta_rowset_compact_key({instance_id, tablet_id, version});
RowsetMetaCloudPB updated_rowset_meta = expected_rowset_meta;
updated_rowset_meta.set_num_rows(2000); // Update row count
ASSERT_TRUE(versioned::document_put(txn.get(), load_key, std::move(updated_rowset_meta)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test reading with snapshot version - should get old data
MetaReader meta_reader(instance_id, txn_kv.get(), snapshot_version);
RowsetMetaCloudPB rowset_meta;
TxnErrorCode err = meta_reader.get_compact_rowset_meta(tablet_id, version, &rowset_meta);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Should get original values
ASSERT_EQ(rowset_meta.num_rows(), 1000);
}
{
// Test reading without snapshot - should get new data
MetaReader meta_reader(instance_id, txn_kv.get());
RowsetMetaCloudPB rowset_meta;
TxnErrorCode err = meta_reader.get_compact_rowset_meta(tablet_id, version, &rowset_meta);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Should get updated values
ASSERT_EQ(rowset_meta.num_rows(), 2000);
}
{
// Test with snapshot flag
MetaReader meta_reader(instance_id, txn_kv.get());
RowsetMetaCloudPB rowset_meta;
TxnErrorCode err =
meta_reader.get_compact_rowset_meta(tablet_id, version, &rowset_meta, true);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
// Should get current values since snapshot flag is set but no snapshot version
ASSERT_EQ(rowset_meta.num_rows(), 2000);
}
{
// Test getting non-existent version
MetaReader meta_reader(instance_id, txn_kv.get());
RowsetMetaCloudPB rowset_meta;
TxnErrorCode err =
meta_reader.get_compact_rowset_meta(tablet_id, version + 1, &rowset_meta);
ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND);
}
}
TEST(MetaReaderTest, BatchGetTabletCompactStats) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
std::vector<int64_t> tablet_ids = {4001, 4002, 4003, 4004};
{
// Test empty input
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<int64_t> empty_ids;
std::unordered_map<int64_t, TabletStatsPB> tablet_stats;
std::unordered_map<int64_t, Versionstamp> versionstamps;
TxnErrorCode err =
meta_reader.get_tablet_compact_stats(empty_ids, &tablet_stats, &versionstamps);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(tablet_stats.empty());
ASSERT_TRUE(versionstamps.empty());
}
{
// Test all keys not found
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, TabletStatsPB> tablet_stats;
std::unordered_map<int64_t, Versionstamp> versionstamps;
TxnErrorCode err =
meta_reader.get_tablet_compact_stats(tablet_ids, &tablet_stats, &versionstamps);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(tablet_stats.empty());
ASSERT_TRUE(versionstamps.empty());
}
{
// Put some tablet compact stats (skip tablet_ids[1] to test partial results)
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_compact_stats_key =
versioned::tablet_compact_stats_key({instance_id, tablet_ids[i]});
TabletStatsPB tablet_stats;
tablet_stats.set_num_rows(1000 * (i + 1));
tablet_stats.set_data_size(500000 * (i + 1));
tablet_stats.set_base_compaction_cnt(5 * (i + 1));
tablet_stats.set_cumulative_compaction_cnt(10 * (i + 1));
tablet_stats.set_cumulative_point(100 * (i + 1));
tablet_stats.set_last_base_compaction_time_ms(1234567890 + i * 1000);
tablet_stats.set_last_cumu_compaction_time_ms(2345678901 + i * 1000);
tablet_stats.set_full_compaction_cnt(2 * (i + 1));
tablet_stats.set_last_full_compaction_time_ms(3456789012 + i * 1000);
tablet_stats.set_num_rowsets(20 * (i + 1));
tablet_stats.set_num_segments(50 * (i + 1));
tablet_stats.set_index_size(50000 * (i + 1));
tablet_stats.set_segment_size(600000 * (i + 1));
versioned_put(txn.get(), tablet_compact_stats_key, tablet_stats.SerializeAsString());
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test partial results
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, TabletStatsPB> tablet_stats;
std::unordered_map<int64_t, Versionstamp> versionstamps;
TxnErrorCode err =
meta_reader.get_tablet_compact_stats(tablet_ids, &tablet_stats, &versionstamps);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_stats.size(), 3); // All except tablet_ids[1]
ASSERT_EQ(versionstamps.size(), 3); // All except tablet_ids[1]
for (size_t i = 0; i < tablet_ids.size(); ++i) {
int64_t tablet_id = tablet_ids[i];
if (i == 1) {
ASSERT_EQ(tablet_stats.find(tablet_id), tablet_stats.end());
ASSERT_EQ(versionstamps.find(tablet_id), versionstamps.end());
} else {
ASSERT_NE(tablet_stats.find(tablet_id), tablet_stats.end());
ASSERT_NE(versionstamps.find(tablet_id), versionstamps.end());
// Verify stats data
const TabletStatsPB& stats = tablet_stats[tablet_id];
ASSERT_EQ(stats.num_rows(), 1000 * (i + 1));
ASSERT_EQ(stats.data_size(), 500000 * (i + 1));
ASSERT_EQ(stats.base_compaction_cnt(), 5 * (i + 1));
ASSERT_EQ(stats.cumulative_compaction_cnt(), 10 * (i + 1));
ASSERT_EQ(stats.cumulative_point(), 100 * (i + 1));
ASSERT_EQ(stats.last_base_compaction_time_ms(), 1234567890 + i * 1000);
ASSERT_EQ(stats.last_cumu_compaction_time_ms(), 2345678901 + i * 1000);
ASSERT_EQ(stats.full_compaction_cnt(), 2 * (i + 1));
ASSERT_EQ(stats.last_full_compaction_time_ms(), 3456789012 + i * 1000);
ASSERT_EQ(stats.num_rowsets(), 20 * (i + 1));
ASSERT_EQ(stats.num_segments(), 50 * (i + 1));
ASSERT_EQ(stats.index_size(), 50000 * (i + 1));
ASSERT_EQ(stats.segment_size(), 600000 * (i + 1));
}
}
}
{
// Put the missing tablet compact stats
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_id, tablet_ids[1]});
TabletStatsPB tablet_stats;
tablet_stats.set_num_rows(2000);
tablet_stats.set_data_size(1000000);
tablet_stats.set_base_compaction_cnt(10);
tablet_stats.set_cumulative_compaction_cnt(20);
tablet_stats.set_cumulative_point(200);
tablet_stats.set_last_base_compaction_time_ms(1234567890 + 1000);
tablet_stats.set_last_cumu_compaction_time_ms(2345678901 + 1000);
tablet_stats.set_full_compaction_cnt(4);
tablet_stats.set_last_full_compaction_time_ms(3456789012 + 1000);
tablet_stats.set_num_rowsets(40);
tablet_stats.set_num_segments(100);
tablet_stats.set_index_size(100000);
tablet_stats.set_segment_size(1200000);
versioned_put(txn.get(), tablet_compact_stats_key, tablet_stats.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test all keys found
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, TabletStatsPB> tablet_stats;
std::unordered_map<int64_t, Versionstamp> versionstamps;
TxnErrorCode err = meta_reader.get_tablet_compact_stats(txn.get(), tablet_ids,
&tablet_stats, &versionstamps);
ASSERT_EQ(err, 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) {
int64_t tablet_id = tablet_ids[i];
ASSERT_NE(tablet_stats.find(tablet_id), tablet_stats.end());
ASSERT_NE(versionstamps.find(tablet_id), versionstamps.end());
// Verify stats data for all tablets
const TabletStatsPB& stats = tablet_stats[tablet_id];
if (i == 1) {
// Special case for the tablet we added later
ASSERT_EQ(stats.num_rows(), 2000);
ASSERT_EQ(stats.data_size(), 1000000);
ASSERT_EQ(stats.base_compaction_cnt(), 10);
ASSERT_EQ(stats.cumulative_compaction_cnt(), 20);
} else {
ASSERT_EQ(stats.num_rows(), 1000 * (i + 1));
ASSERT_EQ(stats.data_size(), 500000 * (i + 1));
ASSERT_EQ(stats.base_compaction_cnt(), 5 * (i + 1));
ASSERT_EQ(stats.cumulative_compaction_cnt(), 10 * (i + 1));
}
}
}
{
// Test with snapshot version functionality
Versionstamp snapshot_version;
{
// Get current snapshot version
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
int64_t version = 0;
ASSERT_EQ(txn->get_read_version(&version), TxnErrorCode::TXN_OK);
snapshot_version = Versionstamp(version, 1);
}
// Update one of the tablet compact stats
{
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_id, tablet_ids[0]});
TabletStatsPB tablet_stats;
tablet_stats.set_num_rows(9999); // Updated value
tablet_stats.set_data_size(999999); // Updated value
tablet_stats.set_base_compaction_cnt(99);
tablet_stats.set_cumulative_compaction_cnt(199);
versioned_put(txn.get(), tablet_compact_stats_key, tablet_stats.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Test reading with snapshot version - should get old data
{
MetaReader meta_reader(instance_id, txn_kv.get(), snapshot_version);
std::unordered_map<int64_t, TabletStatsPB> tablet_stats;
std::unordered_map<int64_t, Versionstamp> versionstamps;
TxnErrorCode err = meta_reader.get_tablet_compact_stats(tablet_ids, &tablet_stats,
&versionstamps, true);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_stats.size(), tablet_ids.size());
// Should still see old data for tablet_ids[0]
const TabletStatsPB& stats = tablet_stats[tablet_ids[0]];
ASSERT_EQ(stats.num_rows(), 1000); // Old value, not 9999
ASSERT_EQ(stats.data_size(), 500000); // Old value, not 999999
}
// Test reading without snapshot - should get new data
{
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, TabletStatsPB> tablet_stats;
std::unordered_map<int64_t, Versionstamp> versionstamps;
TxnErrorCode err =
meta_reader.get_tablet_compact_stats(tablet_ids, &tablet_stats, &versionstamps);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_stats.size(), tablet_ids.size());
// Should see new data for tablet_ids[0]
const TabletStatsPB& stats = tablet_stats[tablet_ids[0]];
ASSERT_EQ(stats.num_rows(), 9999); // Updated value
ASSERT_EQ(stats.data_size(), 999999); // Updated value
}
}
{
// Test with nullptr parameters
MetaReader meta_reader(instance_id, txn_kv.get());
TxnErrorCode err = meta_reader.get_tablet_compact_stats(tablet_ids, nullptr, nullptr);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
}
{
// Test with only tablet_stats parameter
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, TabletStatsPB> tablet_stats;
TxnErrorCode err = meta_reader.get_tablet_compact_stats(tablet_ids, &tablet_stats, nullptr);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_stats.size(), tablet_ids.size());
}
{
// Test with only versionstamps parameter
MetaReader meta_reader(instance_id, txn_kv.get());
std::unordered_map<int64_t, Versionstamp> versionstamps;
TxnErrorCode err =
meta_reader.get_tablet_compact_stats(tablet_ids, nullptr, &versionstamps);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(versionstamps.size(), tablet_ids.size());
}
}
TEST(MetaReaderTest, GetSnapshots) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
{
// Test empty result when no snapshots exist
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<std::pair<SnapshotPB, Versionstamp>> snapshots;
TxnErrorCode err = meta_reader.get_snapshots(&snapshots);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(snapshots.empty());
}
// Create some snapshots
std::vector<Versionstamp> expected_versionstamps;
{
// Create multiple snapshots with different timestamps
for (int i = 1; i <= 3; ++i) {
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string snapshot_key = versioned::snapshot_full_key({instance_id});
SnapshotPB snapshot_pb;
snapshot_pb.set_label(fmt::format("snapshot_{}", i));
snapshot_pb.set_instance_id(instance_id);
std::string snapshot_value = snapshot_pb.SerializeAsString();
versioned_put(txn.get(), snapshot_key, snapshot_value);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
}
{
// Test getting snapshots with created transaction
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<std::pair<SnapshotPB, Versionstamp>> snapshots;
TxnErrorCode err = meta_reader.get_snapshots(&snapshots);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(snapshots.size(), 3);
// Verify snapshots are returned (order may vary due to versionstamp ordering)
std::set<std::string> snapshot_ids;
for (const auto& [snapshot_pb, versionstamp] : snapshots) {
snapshot_ids.insert(snapshot_pb.label());
ASSERT_EQ(snapshot_pb.instance_id(), instance_id);
}
ASSERT_EQ(snapshot_ids.size(), 3);
ASSERT_TRUE(snapshot_ids.count("snapshot_1"));
ASSERT_TRUE(snapshot_ids.count("snapshot_2"));
ASSERT_TRUE(snapshot_ids.count("snapshot_3"));
}
}
TEST(MetaReaderTest, GetLoadRowsetMetas) {
using doris::RowsetMetaCloudPB;
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
int64_t tablet_id = 5001;
{
// Test with no rowset metas
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<std::pair<RowsetMetaCloudPB, Versionstamp>> rowset_metas;
TxnErrorCode err = meta_reader.get_load_rowset_metas(tablet_id, &rowset_metas);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(rowset_metas.empty());
}
{
// Put some load rowset metas
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
for (int64_t version = 1; version <= 3; ++version) {
std::string load_rowset_key =
versioned::meta_rowset_load_key({instance_id, tablet_id, version});
RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(1);
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(100 * version);
rowset_meta.set_data_disk_size(1000 * version);
ASSERT_TRUE(
versioned::document_put(txn.get(), load_rowset_key, std::move(rowset_meta)));
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test getting load rowset metas
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<std::pair<RowsetMetaCloudPB, Versionstamp>> rowset_metas;
TxnErrorCode err = meta_reader.get_load_rowset_metas(tablet_id, &rowset_metas);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_metas.size(), 3);
// Verify the rowset metas are correct
for (size_t i = 0; i < rowset_metas.size(); ++i) {
const auto& [rowset_meta, versionstamp] = rowset_metas[i];
int64_t expected_version = rowset_meta.start_version();
ASSERT_EQ(rowset_meta.rowset_id_v2(),
"load_rowset_" + std::to_string(expected_version));
ASSERT_EQ(rowset_meta.start_version(), expected_version);
ASSERT_EQ(rowset_meta.end_version(), expected_version);
ASSERT_EQ(rowset_meta.num_rows(), 100 * expected_version);
ASSERT_EQ(rowset_meta.data_disk_size(), 1000 * expected_version);
}
// Check min_read_versionstamp is updated
ASSERT_NE(meta_reader.min_read_versionstamp(), Versionstamp::max());
}
{
// Test with transaction
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<std::pair<RowsetMetaCloudPB, Versionstamp>> rowset_metas;
TxnErrorCode err = meta_reader.get_load_rowset_metas(txn.get(), tablet_id, &rowset_metas);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_metas.size(), 3);
}
}
TEST(MetaReaderTest, GetCompactRowsetMetas) {
using doris::RowsetMetaCloudPB;
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
int64_t tablet_id = 5002;
{
// Test with no rowset metas
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<std::pair<RowsetMetaCloudPB, Versionstamp>> rowset_metas;
TxnErrorCode err = meta_reader.get_compact_rowset_metas(tablet_id, &rowset_metas);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(rowset_metas.empty());
}
{
// Put some compact rowset metas
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
// Create compact rowsets with different version ranges
std::vector<std::pair<int64_t, int64_t>> version_ranges = {{1, 3}, {4, 6}, {7, 9}};
for (size_t i = 0; i < version_ranges.size(); ++i) {
auto [start_version, end_version] = version_ranges[i];
std::string compact_rowset_key =
versioned::meta_rowset_compact_key({instance_id, tablet_id, end_version});
RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2("compact_rowset_" + std::to_string(i + 1));
rowset_meta.set_start_version(start_version);
rowset_meta.set_end_version(end_version);
rowset_meta.set_num_rows(300 * (i + 1));
rowset_meta.set_data_disk_size(3000 * (i + 1));
ASSERT_TRUE(
versioned::document_put(txn.get(), compact_rowset_key, std::move(rowset_meta)));
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test getting compact rowset metas
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<std::pair<RowsetMetaCloudPB, Versionstamp>> rowset_metas;
TxnErrorCode err = meta_reader.get_compact_rowset_metas(tablet_id, &rowset_metas);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_metas.size(), 3);
// Verify the rowset metas are correct
std::vector<std::pair<int64_t, int64_t>> expected_ranges = {{1, 3}, {4, 6}, {7, 9}};
// Sort by end_version for consistent comparison
std::sort(rowset_metas.begin(), rowset_metas.end(), [](const auto& a, const auto& b) {
return a.first.end_version() < b.first.end_version();
});
for (size_t i = 0; i < rowset_metas.size(); ++i) {
const auto& [rowset_meta, versionstamp] = rowset_metas[i];
auto [expected_start, expected_end] = expected_ranges[i];
ASSERT_EQ(rowset_meta.rowset_id_v2(), "compact_rowset_" + std::to_string(i + 1));
ASSERT_EQ(rowset_meta.start_version(), expected_start);
ASSERT_EQ(rowset_meta.end_version(), expected_end);
ASSERT_EQ(rowset_meta.num_rows(), 300 * (i + 1));
ASSERT_EQ(rowset_meta.data_disk_size(), 3000 * (i + 1));
}
// Check min_read_versionstamp is updated
ASSERT_NE(meta_reader.min_read_versionstamp(), Versionstamp::max());
}
{
// Test with transaction
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<std::pair<RowsetMetaCloudPB, Versionstamp>> rowset_metas;
TxnErrorCode err =
meta_reader.get_compact_rowset_metas(txn.get(), tablet_id, &rowset_metas);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(rowset_metas.size(), 3);
}
}
TEST(MetaReaderTest, HasSnapshot) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
{
// Test when snapshot does not exist
MetaReader meta_reader(instance_id, txn_kv.get());
bool has_snapshot = true; // Initialize to true to test the change
TxnErrorCode err = meta_reader.has_snapshot(&has_snapshot);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_FALSE(has_snapshot);
}
{
// Create a snapshot entry
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string snapshot_key = versioned::snapshot_full_key({instance_id});
snapshot_key = encode_versioned_key(snapshot_key, Versionstamp(1, 1));
SnapshotPB snapshot_pb;
snapshot_pb.set_label("test_snapshot");
snapshot_pb.set_instance_id(instance_id);
ASSERT_TRUE(versioned::document_put(txn.get(), snapshot_key, std::move(snapshot_pb)));
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test when snapshot exists
MetaReader meta_reader(instance_id, txn_kv.get());
bool has_snapshot = false; // Initialize to false to test the change
TxnErrorCode err = meta_reader.has_snapshot(&has_snapshot);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(has_snapshot);
}
}
TEST(MetaReaderTest, HasSnapshotReferences) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
Versionstamp snapshot_version(12345, 0);
{
// Test when no snapshot references exist - should return false
MetaReader meta_reader(instance_id, txn_kv.get());
bool has_references = true; // Initialize to true to test the change
TxnErrorCode err = meta_reader.has_snapshot_references(snapshot_version, &has_references);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_FALSE(has_references);
}
{
// Create some snapshot reference entries with empty values as specified
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
// Add snapshot reference entries
std::string ref_key1 = versioned::snapshot_reference_key(
{instance_id, snapshot_version, "ref_instance_1"});
std::string ref_key2 = versioned::snapshot_reference_key(
{instance_id, snapshot_version, "ref_instance_2"});
std::string ref_key3 = versioned::snapshot_reference_key(
{instance_id, snapshot_version, "ref_instance_3"});
// Put empty values for snapshot_reference_key as specified in requirements
txn->put(ref_key1, "");
txn->put(ref_key2, "");
txn->put(ref_key3, "");
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test when snapshot references exist - should return true
MetaReader meta_reader(instance_id, txn_kv.get());
bool has_references = false; // Initialize to false to test the change
TxnErrorCode err = meta_reader.has_snapshot_references(snapshot_version, &has_references);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(has_references);
}
{
// Test with transaction parameter
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
bool has_references = false;
TxnErrorCode err =
meta_reader.has_snapshot_references(txn.get(), snapshot_version, &has_references);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(has_references);
}
{
// Test with different snapshot version that has no references
Versionstamp different_snapshot_version(54321, 0);
MetaReader meta_reader(instance_id, txn_kv.get());
bool has_references = true;
TxnErrorCode err =
meta_reader.has_snapshot_references(different_snapshot_version, &has_references);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_FALSE(has_references);
}
{
// Test snapshot functionality
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
bool has_references = false;
TxnErrorCode err = meta_reader.has_snapshot_references(txn.get(), snapshot_version,
&has_references, true);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(has_references);
}
{
// Test with a range of snapshot references
// Create additional references with different versions to test prefix matching
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
Versionstamp nearby_version1(12344, 0); // One less
Versionstamp nearby_version2(12346, 0); // One more
std::string ref_key_before =
versioned::snapshot_reference_key({instance_id, nearby_version1, "ref_before"});
std::string ref_key_after =
versioned::snapshot_reference_key({instance_id, nearby_version2, "ref_after"});
txn->put(ref_key_before, "");
txn->put(ref_key_after, "");
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test that the method only finds references for the exact snapshot version
MetaReader meta_reader(instance_id, txn_kv.get());
bool has_references = false;
TxnErrorCode err = meta_reader.has_snapshot_references(snapshot_version, &has_references);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(has_references); // Should still find the original references
// Test nearby versions
Versionstamp nearby_version1(12344, 0);
has_references = false;
err = meta_reader.has_snapshot_references(nearby_version1, &has_references);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(has_references); // Should find the reference for this version
Versionstamp nearby_version2(12346, 0);
has_references = false;
err = meta_reader.has_snapshot_references(nearby_version2, &has_references);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(has_references); // Should find the reference for this version
// Test a version with no references
Versionstamp no_ref_version(99999, 0);
has_references = true;
err = meta_reader.has_snapshot_references(no_ref_version, &has_references);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_FALSE(has_references); // Should not find any references
}
}
TEST(MetaReaderTest, HasNoIndexes) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
int64_t db_id = 1001;
int64_t table_id = 2001;
{
// Test when no indexes exist - should return true
MetaReader meta_reader(instance_id, txn_kv.get());
bool no_indexes = false;
TxnErrorCode err = meta_reader.has_no_indexes(db_id, table_id, &no_indexes);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(no_indexes);
}
{
// Insert some index_inverted_key entries with empty values
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
// Add index entries for the table
std::string index_key1 = versioned::index_inverted_key({instance_id, db_id, table_id, 1});
std::string index_key2 = versioned::index_inverted_key({instance_id, db_id, table_id, 2});
std::string index_key3 = versioned::index_inverted_key({instance_id, db_id, table_id, 3});
// Put empty values for index_inverted_key as specified in requirements
txn->put(index_key1, "");
txn->put(index_key2, "");
txn->put(index_key3, "");
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
// Test when indexes exist - should return false
MetaReader meta_reader(instance_id, txn_kv.get());
bool no_indexes = true;
TxnErrorCode err = meta_reader.has_no_indexes(db_id, table_id, &no_indexes);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_FALSE(no_indexes);
}
{
// Test with transaction
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
bool no_indexes = true;
TxnErrorCode err = meta_reader.has_no_indexes(txn.get(), db_id, table_id, &no_indexes);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_FALSE(no_indexes);
}
{
// Test with different table_id that has no indexes
int64_t empty_table_id = 3001;
MetaReader meta_reader(instance_id, txn_kv.get());
bool no_indexes = false;
TxnErrorCode err = meta_reader.has_no_indexes(db_id, empty_table_id, &no_indexes);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(no_indexes);
}
{
// Test with different db_id that has no indexes
int64_t empty_db_id = 2001;
MetaReader meta_reader(instance_id, txn_kv.get());
bool no_indexes = false;
TxnErrorCode err = meta_reader.has_no_indexes(empty_db_id, table_id, &no_indexes);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(no_indexes);
}
{
// Test snapshot functionality
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaReader meta_reader(instance_id, txn_kv.get());
bool no_indexes = true;
TxnErrorCode err =
meta_reader.has_no_indexes(txn.get(), db_id, table_id, &no_indexes, true);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_FALSE(no_indexes);
}
}
TEST(MetaReaderTest, GetAllTabletIds) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
std::string instance_id = "test_instance";
// Prepare test data - multiple tablet metas
struct TestTabletMeta {
int64_t tablet_id;
int64_t table_id;
int64_t partition_id;
int64_t index_id;
};
std::vector<TestTabletMeta> test_tablets = {
{.tablet_id = 1001, .table_id = 2001, .partition_id = 3001, .index_id = 4001},
{.tablet_id = 1002, .table_id = 2001, .partition_id = 3001, .index_id = 4001},
{.tablet_id = 1003, .table_id = 2002, .partition_id = 3002, .index_id = 4002},
{.tablet_id = 1004, .table_id = 2002, .partition_id = 3003, .index_id = 4003},
{.tablet_id = 1005, .table_id = 2003, .partition_id = 3004, .index_id = 4004},
};
// Store original tablet metas for comparison
std::map<int64_t, TabletIndexPB> original_metas;
{
// Create and insert tablet metas
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
for (const auto& test_tablet : test_tablets) {
TabletIndexPB tablet_idx;
tablet_idx.set_tablet_id(test_tablet.tablet_id);
tablet_idx.set_table_id(test_tablet.table_id);
tablet_idx.set_partition_id(test_tablet.partition_id);
tablet_idx.set_index_id(test_tablet.index_id);
std::string tablet_key =
versioned::tablet_index_key({instance_id, test_tablet.tablet_id});
// Store original for comparison
original_metas[test_tablet.tablet_id] = tablet_idx;
txn->put(tablet_key, tablet_idx.SerializeAsString());
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
MetaReader meta_reader(instance_id, txn_kv.get());
std::vector<int64_t> tablet_ids;
TxnErrorCode err = meta_reader.get_all_tablet_ids(&tablet_ids, false);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(tablet_ids.size(), test_tablets.size());
for (const auto& tablet_id : tablet_ids) {
TabletIndexPB tablet_idx;
std::string tablet_key = versioned::tablet_index_key({instance_id, tablet_id});
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string value;
ASSERT_EQ(txn->get(tablet_key, &value), TxnErrorCode::TXN_OK);
ASSERT_TRUE(tablet_idx.ParseFromString(value));
ASSERT_EQ(original_metas[tablet_id].table_id(), tablet_idx.table_id());
ASSERT_EQ(original_metas[tablet_id].partition_id(), tablet_idx.partition_id());
ASSERT_EQ(original_metas[tablet_id].index_id(), tablet_idx.index_id());
ASSERT_EQ(original_metas[tablet_id].tablet_id(), tablet_idx.tablet_id());
}
}
}
TEST(MetaReaderTest, FindDerivedInstanceIdsTest) {
auto txn_kv = std::make_shared<MemTxnKv>();
std::string instance_id = "test_instance";
Versionstamp snapshot_version(0x0102030405060708ULL, 0x090a);
// Create snapshot reference keys for multiple derived instances
std::vector<std::string> expected_derived_ids = {
"derived_instance_1",
"derived_instance_2",
"derived_instance_3",
"derived_instance_4",
};
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
for (const auto& derived_id : expected_derived_ids) {
versioned::SnapshotReferenceKeyInfo ref_key_info {instance_id, snapshot_version,
derived_id};
std::string ref_key = versioned::snapshot_reference_key(ref_key_info);
txn->put(ref_key, ""); // Value is empty for reference keys
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Test find_derived_instance_ids
{
MetaReader reader(instance_id, txn_kv.get());
std::vector<std::string> derived_ids;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
TxnErrorCode err =
reader.find_derived_instance_ids(txn.get(), snapshot_version, &derived_ids, false);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(derived_ids.size(), expected_derived_ids.size());
// Convert to set for comparison (order doesn't matter)
std::unordered_set<std::string> derived_set(derived_ids.begin(), derived_ids.end());
std::unordered_set<std::string> expected_set(expected_derived_ids.begin(),
expected_derived_ids.end());
ASSERT_EQ(derived_set, expected_set);
}
// Test with no references
{
Versionstamp empty_snapshot_version(0x0a0b0c0d0e0f1011ULL, 0x1213);
MetaReader reader(instance_id, txn_kv.get());
std::vector<std::string> derived_ids;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
TxnErrorCode err = reader.find_derived_instance_ids(txn.get(), empty_snapshot_version,
&derived_ids, false);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_TRUE(derived_ids.empty());
}
// Test with duplicate instance IDs (should be deduplicated)
{
Versionstamp dup_snapshot_version(0x1112131415161718ULL, 0x191a);
std::string dup_instance_id = "duplicate_instance";
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
// Create multiple references with the same derived instance ID
// This simulates multiple tables/databases referencing the same snapshot
for (int i = 0; i < 5; ++i) {
versioned::SnapshotReferenceKeyInfo ref_key_info {instance_id, dup_snapshot_version,
dup_instance_id};
std::string ref_key = versioned::snapshot_reference_key(ref_key_info);
txn->put(ref_key, "");
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
MetaReader reader(instance_id, txn_kv.get());
std::vector<std::string> derived_ids;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
TxnErrorCode err = reader.find_derived_instance_ids(txn.get(), dup_snapshot_version,
&derived_ids, false);
ASSERT_EQ(err, TxnErrorCode::TXN_OK);
ASSERT_EQ(derived_ids.size(), 1); // Should be deduplicated
ASSERT_EQ(derived_ids[0], dup_instance_id);
}
}