nprasad2077/doris
1
0 Fork 0
mirror of https://github.com/apache/doris.git synced 2026-11-04 15:39:47 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
master
doris/cloud/test/meta_reader_test.cpp
T
Luwei 0557668f40 [fix](cloud) Exclude covered rowsets from compaction minimum timestamps (#67617) 
Problem Summary:

Versioned rowset reads add load metadata versions to the reader minimum
before compact rowsets remove covered loads from the returned set. With
loads L2[2,2]@80 and L3[3,3]@90 covered by A[2,3]@150, plus L4[4,4]@170,
reading [2,4] returns A and L4 but records a minimum of 80. Later
compaction logs can consequently retain successive compacted rowsets for
a snapshot that only needs L2/L3.

Keep each candidate's metadata Versionstamp alongside its rowset, then
merge only the final returned candidates into the existing minimum after
both scans succeed. The returned set, scan/coverage rules, snapshot
bounds and error propagation are unchanged. Previous stats and other
real dependencies remain tracked, and persisted historical logs are
unchanged.

The regression exercises two real MetaService compactions and the
Recycler reference checker using MemTxnKv: snapshot@100 still protects
the first log containing L2/L3, while the later log has
min_timestamp=150 and is no longer protected by that snapshot.

### Release note

Fix unnecessary retention of later compacted rowsets when snapshots
protect older load rowsets already covered by compaction.
2026-09-09 20:42:12 +08:00

3196 lines
134 KiB
C++
Raw Permalink Blame History

// 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);
}
}
Reference in new issue View Git Blame Copy Permalink