| // 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 <brpc/controller.h> | |
| #include <bvar/window.h> | |
| #include <fmt/core.h> | |
| #include <gen_cpp/cloud.pb.h> | |
| #include <gen_cpp/olap_file.pb.h> | |
| #include <google/protobuf/repeated_field.h> | |
| #include <gtest/gtest.h> | |
| #include <atomic> | |
| #include <condition_variable> | |
| #include <cstdint> | |
| #include <memory> | |
| #include <random> | |
| #include <string> | |
| #include <thread> | |
| #include "common/config.h" | |
| #include "common/logging.h" | |
| #include "common/util.h" | |
| #include "cpp/sync_point.h" | |
| #include "meta-service/meta_service.h" | |
| #include "meta-service/meta_service_helper.h" | |
| #include "meta-store/document_message.h" | |
| #include "meta-store/keys.h" | |
| #include "meta-store/mem_txn_kv.h" | |
| #include "meta-store/meta_reader.h" | |
| #include "meta-store/txn_kv.h" | |
| #include "meta-store/txn_kv_error.h" | |
| #include "meta-store/versioned_value.h" | |
| #include "mock_resource_manager.h" | |
| #include "rate-limiter/rate_limiter.h" | |
| #include "recycler/util.h" | |
| #include "resource-manager/resource_manager.h" | |
| namespace doris::cloud { | |
| // External functions from meta_service_test.cpp | |
| extern std::unique_ptr<MetaServiceProxy> get_meta_service(); | |
| extern std::unique_ptr<MetaServiceProxy> get_meta_service(bool mock_resource_mgr); | |
| extern void create_tablet(MetaServiceProxy* meta_service, int64_t table_id, int64_t index_id, | |
| int64_t partition_id, int64_t tablet_id); | |
| extern doris::RowsetMetaCloudPB create_rowset(int64_t txn_id, int64_t tablet_id, int partition_id, | |
| int64_t version, int num_rows); | |
| extern void prepare_rowset(MetaServiceProxy* meta_service, const doris::RowsetMetaCloudPB& rowset, | |
| CreateRowsetResponse& res); | |
| extern void commit_rowset(MetaServiceProxy* meta_service, const doris::RowsetMetaCloudPB& rowset, | |
| CreateRowsetResponse& res); | |
| extern void insert_rowset(MetaServiceProxy* meta_service, int64_t db_id, const std::string& label, | |
| int64_t table_id, int64_t partition_id, int64_t tablet_id); | |
| extern void add_tablet(CreateTabletsRequest& req, int64_t table_id, int64_t index_id, | |
| int64_t partition_id, int64_t tablet_id); | |
| extern void get_tablet_stats(MetaServiceProxy* meta_service, int64_t table_id, int64_t index_id, | |
| int64_t partition_id, int64_t tablet_id, GetTabletStatsResponse& res); | |
| extern void create_and_commit_rowset(MetaServiceProxy* meta_service, int64_t table_id, | |
| int64_t index_id, int64_t partition_id, int64_t tablet_id, | |
| int64_t txn_id); | |
| void insert_compact_rowset(Transaction* txn, std::string instance_id, int64_t tablet_id, | |
| int64_t partition_id, int64_t start_version, int64_t end_version, | |
| int num_rows) { | |
| doris::RowsetMetaCloudPB compact_rowset = | |
| create_rowset(1, tablet_id, partition_id, start_version, num_rows); | |
| compact_rowset.set_end_version(end_version); | |
| std::string key = versioned::meta_rowset_compact_key({instance_id, tablet_id, end_version}); | |
| ASSERT_TRUE(versioned::document_put(txn, key, std::move(compact_rowset))); | |
| } | |
| void update_tablet_compact_stats(Transaction* txn, std::string instance_id, int64_t tablet_id, | |
| int start_version, int end_version, int num_rows) { | |
| MetaReader reader(instance_id, nullptr); | |
| std::vector<RowsetMetaCloudPB> rowset_metas; | |
| ASSERT_EQ(reader.get_rowset_metas(txn, tablet_id, start_version, end_version, &rowset_metas), | |
| TxnErrorCode::TXN_OK); | |
| int input_data_size = 0; | |
| int input_num_rows = 0; | |
| int input_num_rowsets = 0; | |
| int input_num_segments = 0; | |
| int input_index_size = 0; | |
| int input_segment_size = 0; | |
| for (const auto& rowset_meta : rowset_metas) { | |
| input_data_size += rowset_meta.total_disk_size(); | |
| input_num_rows += rowset_meta.num_rows(); | |
| input_num_rowsets += 1; // Each rowset is considered as one rowset | |
| input_num_segments += rowset_meta.num_segments(); | |
| input_index_size += rowset_meta.index_disk_size(); | |
| input_segment_size += rowset_meta.data_disk_size(); | |
| } | |
| std::string key = versioned::tablet_compact_stats_key({instance_id, tablet_id}); | |
| TabletStatsPB stats; | |
| ASSERT_EQ(versioned::document_get(txn, key, &stats, nullptr), TxnErrorCode::TXN_OK); | |
| // See create_rowset() for the calculation of data size | |
| stats.set_data_size(stats.data_size() + num_rows * 110 - input_data_size); | |
| stats.set_num_rows(stats.num_rows() + num_rows - input_num_rows); | |
| stats.set_num_rowsets(stats.num_rowsets() + 1 - input_num_rowsets); | |
| stats.set_num_segments(stats.num_segments() + 1 - input_num_segments); | |
| stats.set_index_size(stats.index_size() + num_rows * 10 - input_index_size); | |
| stats.set_segment_size(stats.segment_size() + num_rows * 100 - input_segment_size); | |
| ASSERT_TRUE(versioned::document_put(txn, key, std::move(stats))); | |
| } | |
| void compact_rowset(TxnKv* txn_kv, std::string instance_id, int64_t tablet_id, int64_t partition_id, | |
| int64_t start_version, int64_t end_version, int num_rows) { | |
| std::unique_ptr<Transaction> txn; | |
| ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); | |
| insert_compact_rowset(txn.get(), instance_id, tablet_id, partition_id, start_version, | |
| end_version, num_rows); | |
| update_tablet_compact_stats(txn.get(), instance_id, tablet_id, start_version, end_version, | |
| num_rows); | |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); | |
| } | |
| static void get_table_version(MetaServiceProxy* meta_service, int64_t db_id, int64_t table_id, | |
| int64_t& version) { | |
| brpc::Controller ctrl; | |
| GetVersionRequest req; | |
| req.set_db_id(db_id); | |
| req.set_table_id(table_id); | |
| req.set_is_table_version(true); | |
| GetVersionResponse resp; | |
| meta_service->get_version(&ctrl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK) | |
| << ", get table version res=" << resp.ShortDebugString(); | |
| version = resp.version(); | |
| } | |
| // Create a MULTI_VERSION_READ_WRITE instance and refresh the resource manager. | |
| static void create_and_refresh_instance(MetaServiceProxy* service, std::string instance_id) { | |
| // write instance | |
| InstanceInfoPB instance_info; | |
| instance_info.set_instance_id(instance_id); | |
| instance_info.set_multi_version_status(MULTI_VERSION_READ_WRITE); | |
| std::unique_ptr<Transaction> txn; | |
| ASSERT_EQ(service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK); | |
| txn->put(instance_key(instance_id), instance_info.SerializeAsString()); | |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); | |
| service->resource_mgr()->refresh_instance(instance_id); | |
| ASSERT_TRUE(service->resource_mgr()->is_version_write_enabled(instance_id)); | |
| } | |
| #define MOCK_GET_INSTANCE_ID(instance_id) \ | |
| DORIS_CLOUD_DEFER { \ | |
| SyncPoint::get_instance()->clear_all_call_backs(); \ | |
| }; \ | |
| SyncPoint::get_instance()->set_call_back("get_instance_id", [&](auto&& args) { \ | |
| auto* ret = try_any_cast_ret<std::string>(args); \ | |
| ret->first = instance_id; \ | |
| ret->second = true; \ | |
| }); \ | |
| SyncPoint::get_instance()->enable_processing(); | |
| TEST(MetaServiceVersionedReadTest, CommitTxn) { | |
| auto meta_service = get_meta_service(false); | |
| std::string instance_id = "test_cloud_instance_id"; | |
| std::string cloud_unique_id = "1:test_cloud_unique_id:1"; | |
| MOCK_GET_INSTANCE_ID(instance_id); | |
| create_and_refresh_instance(meta_service.get(), instance_id); | |
| int64_t db_id = 666; | |
| int64_t table_id = 1234; | |
| int64_t index_id = 1235; | |
| int64_t partition_id = 1236; | |
| // case: first version of rowset | |
| { | |
| int64_t txn_id = -1; | |
| // begin txn | |
| { | |
| brpc::Controller cntl; | |
| BeginTxnRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| TxnInfoPB txn_info_pb; | |
| txn_info_pb.set_db_id(db_id); | |
| txn_info_pb.set_label("test_label"); | |
| txn_info_pb.add_table_ids(table_id); | |
| txn_info_pb.set_timeout_ms(36000); | |
| req.mutable_txn_info()->CopyFrom(txn_info_pb); | |
| BeginTxnResponse res; | |
| meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), | |
| &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK); | |
| txn_id = res.txn_id(); | |
| } | |
| // mock rowset and tablet | |
| int64_t tablet_id_base = 1103; | |
| for (int i = 0; i < 5; ++i) { | |
| create_tablet(meta_service.get(), table_id, index_id, partition_id, tablet_id_base + i); | |
| auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, partition_id, -1, 100); | |
| CreateRowsetResponse res; | |
| prepare_rowset(meta_service.get(), tmp_rowset, res); | |
| commit_rowset(meta_service.get(), tmp_rowset, res); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK); | |
| } | |
| // precommit txn | |
| { | |
| brpc::Controller cntl; | |
| PrecommitTxnRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| req.set_db_id(db_id); | |
| req.set_txn_id(txn_id); | |
| req.set_precommit_timeout_ms(36000); | |
| PrecommitTxnResponse res; | |
| meta_service->precommit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), | |
| &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK); | |
| } | |
| // commit txn | |
| { | |
| brpc::Controller cntl; | |
| CommitTxnRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| req.set_db_id(db_id); | |
| req.set_txn_id(txn_id); | |
| CommitTxnResponse res; | |
| meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), | |
| &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK); | |
| ASSERT_EQ(res.table_stats().size(), 1); | |
| int64_t table_version = 0; | |
| get_table_version(meta_service.get(), db_id, table_id, table_version); | |
| ASSERT_EQ(res.table_stats()[0].table_version(), table_version); | |
| } | |
| // doubly commit txn | |
| { | |
| brpc::Controller cntl; | |
| CommitTxnRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| req.set_db_id(db_id); | |
| req.set_txn_id(txn_id); | |
| CommitTxnResponse res; | |
| meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), | |
| &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK); | |
| auto found = res.status().msg().find(fmt::format( | |
| "transaction is already visible: db_id={} txn_id={}", db_id, txn_id)); | |
| ASSERT_TRUE(found != std::string::npos); | |
| } | |
| // doubly commit txn(2pc) | |
| { | |
| brpc::Controller cntl; | |
| CommitTxnRequest req; | |
| req.set_cloud_unique_id("test_cloud_unique_id"); | |
| req.set_db_id(db_id); | |
| req.set_txn_id(txn_id); | |
| req.set_is_2pc(true); | |
| CommitTxnResponse res; | |
| meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), | |
| &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::TXN_ALREADY_VISIBLE); | |
| auto found = res.status().msg().find( | |
| fmt::format("transaction [{}] is already visible, not pre-committed.", txn_id)); | |
| ASSERT_TRUE(found != std::string::npos); | |
| } | |
| } | |
| { | |
| // Get the partition versions | |
| brpc::Controller cntl; | |
| GetVersionRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| req.set_db_id(db_id); | |
| req.set_table_id(table_id); | |
| req.set_partition_id(partition_id); | |
| GetVersionResponse res; | |
| meta_service->get_version(&cntl, &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK); | |
| ASSERT_EQ(res.version(), 2); | |
| } | |
| } | |
| TEST(MetaServiceVersionedReadTest, CommitTxnWithSubTxnTest) { | |
| auto meta_service = get_meta_service(false); | |
| const std::string instance_id = "test_cloud_instance_id"; | |
| const std::string cloud_unique_id = "1:test_cloud_unique_id:1"; | |
| MOCK_GET_INSTANCE_ID(instance_id); | |
| create_and_refresh_instance(meta_service.get(), instance_id); | |
| int64_t db_id = 98131; | |
| int64_t txn_id = -1; | |
| int64_t t1 = 10; | |
| int64_t t1_index = 100; | |
| int64_t t1_p1 = 11; | |
| int64_t t1_p1_t1 = 12; | |
| int64_t t1_p1_t2 = 13; | |
| int64_t t1_p2 = 14; | |
| int64_t t1_p2_t1 = 15; | |
| int64_t t2 = 16; | |
| int64_t t2_index = 101; | |
| int64_t t2_p3 = 17; | |
| int64_t t2_p3_t1 = 18; | |
| [[maybe_unused]] int64_t t2_p4 = 19; | |
| [[maybe_unused]] int64_t t2_p4_t1 = 20; | |
| std::string label = "test_label"; | |
| std::string label2 = "test_label_0"; | |
| // begin txn | |
| { | |
| brpc::Controller cntl; | |
| BeginTxnRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| TxnInfoPB txn_info_pb; | |
| txn_info_pb.set_db_id(db_id); | |
| txn_info_pb.set_label(label); | |
| txn_info_pb.add_table_ids(t1); | |
| txn_info_pb.set_timeout_ms(36000); | |
| req.mutable_txn_info()->CopyFrom(txn_info_pb); | |
| BeginTxnResponse res; | |
| meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, | |
| &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK); | |
| txn_id = res.txn_id(); | |
| } | |
| // mock rowset and tablet: for sub_txn1 | |
| int64_t sub_txn_id1 = txn_id; | |
| create_and_commit_rowset(meta_service.get(), t1, t1_index, t1_p1, t1_p1_t1, sub_txn_id1); | |
| create_and_commit_rowset(meta_service.get(), t1, t1_index, t1_p1, t1_p1_t2, sub_txn_id1); | |
| create_and_commit_rowset(meta_service.get(), t1, t1_index, t1_p2, t1_p2_t1, sub_txn_id1); | |
| // begin_sub_txn2 | |
| int64_t sub_txn_id2 = -1; | |
| { | |
| brpc::Controller cntl; | |
| BeginSubTxnRequest req; | |
| req.set_cloud_unique_id("test_cloud_unique_id"); | |
| req.set_txn_id(txn_id); | |
| req.set_sub_txn_num(0); | |
| req.set_db_id(db_id); | |
| req.set_label(label2); | |
| req.mutable_table_ids()->Add(t1); | |
| req.mutable_table_ids()->Add(t2); | |
| BeginSubTxnResponse res; | |
| meta_service->begin_sub_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), | |
| &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK); | |
| ASSERT_EQ(res.txn_info().table_ids().size(), 2); | |
| ASSERT_EQ(res.txn_info().sub_txn_ids().size(), 1); | |
| ASSERT_TRUE(res.has_sub_txn_id()); | |
| sub_txn_id2 = res.sub_txn_id(); | |
| ASSERT_EQ(sub_txn_id2, res.txn_info().sub_txn_ids()[0]); | |
| } | |
| // mock rowset and tablet: for sub_txn3 | |
| create_and_commit_rowset(meta_service.get(), t2, t2_index, t2_p3, t2_p3_t1, sub_txn_id2); | |
| // begin_sub_txn3 | |
| int64_t sub_txn_id3 = -1; | |
| { | |
| brpc::Controller cntl; | |
| BeginSubTxnRequest req; | |
| req.set_cloud_unique_id("test_cloud_unique_id"); | |
| req.set_txn_id(txn_id); | |
| req.set_sub_txn_num(1); | |
| req.set_db_id(db_id); | |
| req.set_label("test_label_1"); | |
| req.mutable_table_ids()->Add(t1); | |
| req.mutable_table_ids()->Add(t2); | |
| req.mutable_table_ids()->Add(t1); | |
| BeginSubTxnResponse res; | |
| meta_service->begin_sub_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), | |
| &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK); | |
| ASSERT_EQ(res.txn_info().table_ids().size(), 3); | |
| ASSERT_EQ(res.txn_info().sub_txn_ids().size(), 2); | |
| ASSERT_TRUE(res.has_sub_txn_id()); | |
| sub_txn_id3 = res.sub_txn_id(); | |
| ASSERT_EQ(sub_txn_id3, res.txn_info().sub_txn_ids()[1]); | |
| } | |
| // mock rowset and tablet: for sub_txn3 | |
| create_and_commit_rowset(meta_service.get(), t1, t1_index, t1_p1, t1_p1_t1, sub_txn_id3); | |
| create_and_commit_rowset(meta_service.get(), t1, t1_index, t1_p1, t1_p1_t2, sub_txn_id3); | |
| // commit txn | |
| CommitTxnRequest req; | |
| { | |
| brpc::Controller cntl; | |
| req.set_cloud_unique_id("test_cloud_unique_id"); | |
| req.set_db_id(666); | |
| req.set_txn_id(txn_id); | |
| req.set_is_txn_load(true); | |
| SubTxnInfo sub_txn_info1; | |
| sub_txn_info1.set_sub_txn_id(sub_txn_id1); | |
| sub_txn_info1.set_table_id(t1); | |
| sub_txn_info1.mutable_base_tablet_ids()->Add(t1_p1_t1); | |
| sub_txn_info1.mutable_base_tablet_ids()->Add(t1_p1_t2); | |
| sub_txn_info1.mutable_base_tablet_ids()->Add(t1_p2_t1); | |
| SubTxnInfo sub_txn_info2; | |
| sub_txn_info2.set_sub_txn_id(sub_txn_id2); | |
| sub_txn_info2.set_table_id(t2); | |
| sub_txn_info2.mutable_base_tablet_ids()->Add(t2_p3_t1); | |
| SubTxnInfo sub_txn_info3; | |
| sub_txn_info3.set_sub_txn_id(sub_txn_id3); | |
| sub_txn_info3.set_table_id(t1); | |
| sub_txn_info3.mutable_base_tablet_ids()->Add(t1_p1_t1); | |
| sub_txn_info3.mutable_base_tablet_ids()->Add(t1_p1_t2); | |
| req.mutable_sub_txn_infos()->Add(std::move(sub_txn_info1)); | |
| req.mutable_sub_txn_infos()->Add(std::move(sub_txn_info2)); | |
| req.mutable_sub_txn_infos()->Add(std::move(sub_txn_info3)); | |
| CommitTxnResponse res; | |
| meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, | |
| &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK); | |
| // std::cout << res.DebugString() << std::endl; | |
| ASSERT_EQ(res.table_ids().size(), 3); | |
| for (int i = 0; i < 3; ++i) { | |
| if (res.partition_ids(i) == t2_p3) { | |
| ASSERT_EQ(res.table_ids(i), t2); | |
| ASSERT_EQ(res.versions(i), 2); | |
| } else if (res.partition_ids(i) == t1_p2) { | |
| ASSERT_EQ(res.table_ids(i), t1); | |
| ASSERT_EQ(res.versions(i), 2); | |
| } else if (res.partition_ids(i) == t1_p1) { | |
| ASSERT_EQ(res.table_ids(i), t1); | |
| ASSERT_EQ(res.versions(i), 3); | |
| } else { | |
| ASSERT_TRUE(false) << "unknown partition: " << res.partition_ids(i) | |
| << ", res=" << res.DebugString(); | |
| } | |
| } | |
| ASSERT_EQ(res.table_stats().size(), 2); | |
| int64_t table_version = 0; | |
| get_table_version(meta_service.get(), db_id, t2, table_version); | |
| ASSERT_EQ(res.table_stats()[0].table_version(), table_version); | |
| table_version = 0; | |
| get_table_version(meta_service.get(), db_id, t1, table_version); | |
| ASSERT_EQ(res.table_stats()[1].table_version(), table_version); | |
| } | |
| // doubly commit txn | |
| { | |
| brpc::Controller cntl; | |
| CommitTxnResponse res; | |
| meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, | |
| &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK); | |
| auto found = res.status().msg().find( | |
| fmt::format("transaction is already visible: db_id={} txn_id={}", db_id, txn_id)); | |
| ASSERT_TRUE(found != std::string::npos); | |
| } | |
| // Verify the partition versions | |
| { | |
| brpc::Controller ctrl; | |
| GetVersionRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| req.set_batch_mode(true); | |
| req.add_db_ids(db_id); | |
| req.add_db_ids(db_id); | |
| req.add_db_ids(db_id); | |
| req.add_table_ids(t1); | |
| req.add_table_ids(t1); | |
| req.add_table_ids(t2); | |
| req.add_partition_ids(t1_p1); | |
| req.add_partition_ids(t1_p2); | |
| req.add_partition_ids(t2_p3); | |
| GetVersionResponse resp; | |
| meta_service->get_version(&ctrl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK) | |
| << " status is " << resp.status().ShortDebugString(); | |
| ASSERT_EQ(resp.versions().size(), 3); | |
| ASSERT_EQ(resp.versions()[0], 3); // t1_p1 | |
| ASSERT_EQ(resp.versions()[1], 2); // t1_p2 | |
| ASSERT_EQ(resp.versions()[2], 2); // t2_p3 | |
| } | |
| } | |
| TEST(MetaServiceVersionedReadTest, GetVersion) { | |
| auto service = get_meta_service(false); | |
| int64_t table_id = 1; | |
| int64_t partition_id = 1; | |
| int64_t tablet_id = 1; | |
| std::string instance_id = "test_cloud_instance_id"; | |
| std::string cloud_unique_id = fmt::format("1:{}:1", instance_id); | |
| MOCK_GET_INSTANCE_ID(instance_id); | |
| create_and_refresh_instance(service.get(), instance_id); | |
| // INVALID_ARGUMENT | |
| { | |
| brpc::Controller ctrl; | |
| GetVersionRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| req.set_table_id(table_id); | |
| req.set_partition_id(partition_id); | |
| GetVersionResponse resp; | |
| service->get_version(&ctrl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::INVALID_ARGUMENT) | |
| << " status is " << resp.status().DebugString(); | |
| } | |
| { | |
| brpc::Controller ctrl; | |
| GetVersionRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| req.set_db_id(1); | |
| req.set_table_id(table_id); | |
| req.set_partition_id(partition_id); | |
| GetVersionResponse resp; | |
| service->get_version(&ctrl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::VERSION_NOT_FOUND) | |
| << " status is " << resp.status().DebugString(); | |
| } | |
| { | |
| brpc::Controller ctrl; | |
| GetVersionRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| req.set_db_id(1); | |
| req.set_table_id(table_id); | |
| req.set_partition_id(partition_id); | |
| req.set_is_table_version(true); | |
| GetVersionResponse resp; | |
| service->get_version(&ctrl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::VERSION_NOT_FOUND) | |
| << " status is " << resp.status().DebugString(); | |
| } | |
| create_tablet(service.get(), table_id, 1, partition_id, tablet_id); | |
| insert_rowset(service.get(), 1, "get_version_label_1", table_id, partition_id, tablet_id); | |
| { | |
| brpc::Controller ctrl; | |
| GetVersionRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| req.set_db_id(1); | |
| req.set_table_id(table_id); | |
| req.set_partition_id(partition_id); | |
| GetVersionResponse resp; | |
| service->get_version(&ctrl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK) | |
| << " status is " << resp.status().DebugString(); | |
| ASSERT_EQ(resp.version(), 2); | |
| } | |
| { | |
| brpc::Controller ctrl; | |
| GetVersionRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| req.set_db_id(1); | |
| req.set_table_id(table_id); | |
| req.set_partition_id(partition_id); | |
| req.set_is_table_version(true); | |
| GetVersionResponse resp; | |
| service->get_version(&ctrl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK) | |
| << " status is " << resp.status().DebugString(); | |
| ASSERT_GT(resp.version(), 2); | |
| } | |
| } | |
| TEST(MetaServiceVersionedReadTest, BatchGetVersion) { | |
| struct TestCase { | |
| std::vector<int64_t> table_ids; | |
| std::vector<int64_t> partition_ids; | |
| std::vector<int64_t> expected_versions; | |
| std::vector< | |
| std::tuple<int64_t /*table_id*/, int64_t /*partition_id*/, int64_t /*tablet_id*/>> | |
| insert_rowsets; | |
| }; | |
| // table ids: 2, 3, 4, 5 | |
| // partition ids: 6, 7, 8, 9 | |
| std::vector<TestCase> cases = { | |
| // all version are missing | |
| {{1, 2, 3, 4}, {6, 7, 8, 9}, {-1, -1, -1, -1}, {}}, | |
| // update table 1, partition 6 | |
| {{1, 2, 3, 4}, {6, 7, 8, 9}, {2, -1, -1, -1}, {{1, 6, 1}}}, | |
| // update table 2, partition 6 | |
| // update table 3, partition 7 | |
| {{1, 2, 3, 4}, {6, 7, 8, 9}, {2, -1, 2, 2}, {{3, 8, 3}, {4, 9, 4}}}, | |
| // update table 1, partition 7 twice | |
| {{1, 2, 3, 4}, {6, 7, 8, 9}, {2, 3, 2, 2}, {{2, 7, 2}, {2, 7, 2}}}, | |
| }; | |
| auto service = get_meta_service(false); | |
| std::string instance_id = "test_cloud_instance_id"; | |
| std::string cloud_unique_id = fmt::format("1:{}:1", instance_id); | |
| MOCK_GET_INSTANCE_ID(instance_id); | |
| create_and_refresh_instance(service.get(), instance_id); | |
| create_tablet(service.get(), 1, 1, 6, 1); | |
| create_tablet(service.get(), 2, 1, 7, 2); | |
| create_tablet(service.get(), 3, 1, 8, 3); | |
| create_tablet(service.get(), 4, 1, 9, 4); | |
| size_t num_cases = cases.size(); | |
| size_t label_index = 0; | |
| for (size_t i = 0; i < num_cases; ++i) { | |
| auto& [table_ids, partition_ids, expected_versions, insert_rowsets] = cases[i]; | |
| for (auto [table_id, partition_id, tablet_id] : insert_rowsets) { | |
| LOG(INFO) << "insert rowset for table " << table_id << " partition " << partition_id | |
| << " table_id " << tablet_id; | |
| insert_rowset(service.get(), 1, std::to_string(++label_index), table_id, partition_id, | |
| tablet_id); | |
| } | |
| brpc::Controller ctrl; | |
| GetVersionRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| req.set_db_id(-1); | |
| req.set_table_id(-1); | |
| req.set_partition_id(-1); | |
| req.set_batch_mode(true); | |
| for (size_t i = 0; i < table_ids.size(); ++i) req.add_db_ids(1); | |
| std::copy(table_ids.begin(), table_ids.end(), | |
| google::protobuf::RepeatedFieldBackInserter(req.mutable_table_ids())); | |
| std::copy(partition_ids.begin(), partition_ids.end(), | |
| google::protobuf::RepeatedFieldBackInserter(req.mutable_partition_ids())); | |
| GetVersionResponse resp; | |
| service->get_version(&ctrl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK) | |
| << "case " << i << " status is " << resp.status().msg() | |
| << ", code=" << resp.status().code(); | |
| std::vector<int64_t> versions(resp.versions().begin(), resp.versions().end()); | |
| EXPECT_EQ(versions, expected_versions) << "case " << i; | |
| // Batch get table versions | |
| req.set_is_table_version(true); | |
| resp.Clear(); | |
| service->get_version(&ctrl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK) | |
| << "case " << i << " status is " << resp.status().msg() | |
| << ", code=" << resp.status().code(); | |
| for (size_t j = 0; j < resp.versions_size(); ++j) { | |
| if (expected_versions[j] == -1) { | |
| ASSERT_LT(resp.versions(j), 0) | |
| << "case " << i << ", j=" << j << ", resp=" << resp.DebugString(); | |
| } else { | |
| ASSERT_GT(resp.versions(j), 0) | |
| << "case " << i << ", j=" << j << ", resp=" << resp.DebugString(); | |
| } | |
| } | |
| } | |
| // INVALID_ARGUMENT | |
| { | |
| brpc::Controller ctrl; | |
| GetVersionRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| req.set_batch_mode(true); | |
| GetVersionResponse resp; | |
| service->get_version(&ctrl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::INVALID_ARGUMENT) | |
| << " status is " << resp.status().msg() << ", code=" << resp.status().code(); | |
| } | |
| } | |
| TEST(MetaServiceVersionedReadTest, BatchGetVersionFallback) { | |
| constexpr size_t N = 100; | |
| size_t i = 0; | |
| auto sp = SyncPoint::get_instance(); | |
| DORIS_CLOUD_DEFER { | |
| SyncPoint::get_instance()->clear_all_call_backs(); | |
| }; | |
| sp->set_call_back("batch_get_version_err", [&](auto&& args) { | |
| if (i++ == N / 10) { | |
| *try_any_cast<TxnErrorCode*>(args) = TxnErrorCode::TXN_TOO_OLD; | |
| } | |
| }); | |
| sp->enable_processing(); | |
| auto service = get_meta_service(false); | |
| std::string instance_id = "test_cloud_instance_id"; | |
| std::string cloud_unique_id = fmt::format("1:{}:1", instance_id); | |
| MOCK_GET_INSTANCE_ID(instance_id); | |
| create_and_refresh_instance(service.get(), instance_id); | |
| for (int64_t i = 1; i <= N; ++i) { | |
| create_tablet(service.get(), 1, 1, i, i); | |
| insert_rowset(service.get(), 1, std::to_string(i), 1, i, i); | |
| } | |
| brpc::Controller ctrl; | |
| GetVersionRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| req.set_db_id(-1); | |
| req.set_table_id(-1); | |
| req.set_partition_id(-1); | |
| req.set_batch_mode(true); | |
| for (size_t i = 1; i <= N; ++i) { | |
| req.add_db_ids(1); | |
| req.add_table_ids(1); | |
| req.add_partition_ids(i); | |
| } | |
| GetVersionResponse resp; | |
| service->get_version(&ctrl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK) | |
| << "case " << i << " status is " << resp.status().msg() | |
| << ", code=" << resp.status().code(); | |
| ASSERT_EQ(resp.versions_size(), N); | |
| } | |
| TEST(MetaServiceVersionedReadTest, GetTablet) { | |
| auto service = get_meta_service(false); | |
| std::string instance_id = "test_cloud_instance_id"; | |
| MOCK_GET_INSTANCE_ID(instance_id); | |
| create_and_refresh_instance(service.get(), instance_id); | |
| int64_t table_id = 2; | |
| int64_t index_id = 3; | |
| int64_t partition_id = 4; | |
| int64_t tablet_id = 5; | |
| // Create tablet | |
| create_tablet(service.get(), table_id, index_id, partition_id, tablet_id); | |
| { | |
| // Get the tablet meta | |
| brpc::Controller cntl; | |
| GetTabletRequest req; | |
| req.set_cloud_unique_id(fmt::format("1:{}:1", instance_id)); | |
| req.set_tablet_id(tablet_id); | |
| GetTabletResponse resp; | |
| service->get_tablet(&cntl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK) | |
| << " status is " << resp.status().DebugString(); | |
| ASSERT_EQ(resp.tablet_meta().table_id(), table_id); | |
| ASSERT_EQ(resp.tablet_meta().index_id(), index_id); | |
| ASSERT_EQ(resp.tablet_meta().partition_id(), partition_id); | |
| ASSERT_EQ(resp.tablet_meta().tablet_id(), tablet_id); | |
| // Verify the tablet schema | |
| ASSERT_TRUE(resp.tablet_meta().has_schema()); | |
| } | |
| } | |
| TEST(MetaServiceVersionedReadTest, GetTabletStats) { | |
| auto meta_service = get_meta_service(false); | |
| std::string instance_id = "test_cloud_instance_id"; | |
| MOCK_GET_INSTANCE_ID(instance_id); | |
| create_and_refresh_instance(meta_service.get(), instance_id); | |
| constexpr auto table_id = 10001, index_id = 10002, partition_id = 10003, tablet_id = 10004; | |
| ASSERT_NO_FATAL_FAILURE( | |
| create_tablet(meta_service.get(), table_id, index_id, partition_id, tablet_id)); | |
| { | |
| GetTabletStatsResponse res; | |
| get_tablet_stats(meta_service.get(), table_id, index_id, partition_id, tablet_id, res); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK); | |
| ASSERT_EQ(res.tablet_stats_size(), 1); | |
| EXPECT_EQ(res.tablet_stats(0).data_size(), 0); | |
| EXPECT_EQ(res.tablet_stats(0).num_rows(), 0); | |
| EXPECT_EQ(res.tablet_stats(0).num_rowsets(), 1); | |
| EXPECT_EQ(res.tablet_stats(0).num_segments(), 0); | |
| EXPECT_EQ(res.tablet_stats(0).index_size(), 0); | |
| EXPECT_EQ(res.tablet_stats(0).segment_size(), 0); | |
| // The tablet idx should be set. | |
| auto tablet_stats = res.tablet_stats(0); | |
| EXPECT_EQ(tablet_stats.idx().tablet_id(), tablet_id); | |
| EXPECT_EQ(tablet_stats.idx().index_id(), index_id); | |
| EXPECT_EQ(tablet_stats.idx().partition_id(), partition_id); | |
| EXPECT_EQ(tablet_stats.idx().table_id(), table_id); | |
| } | |
| { | |
| // Insert rowset | |
| config::split_tablet_stats = false; | |
| ASSERT_NO_FATAL_FAILURE(insert_rowset(meta_service.get(), 10000, "label1", table_id, | |
| partition_id, tablet_id)); | |
| ASSERT_NO_FATAL_FAILURE(insert_rowset(meta_service.get(), 10000, "label2", table_id, | |
| partition_id, tablet_id)); | |
| config::split_tablet_stats = true; | |
| ASSERT_NO_FATAL_FAILURE(insert_rowset(meta_service.get(), 10000, "label3", table_id, | |
| partition_id, tablet_id)); | |
| ASSERT_NO_FATAL_FAILURE(insert_rowset(meta_service.get(), 10000, "label4", table_id, | |
| partition_id, tablet_id)); | |
| ASSERT_NO_FATAL_FAILURE(insert_rowset(meta_service.get(), 10000, "label5", table_id, | |
| partition_id, tablet_id)); | |
| ASSERT_NO_FATAL_FAILURE(insert_rowset(meta_service.get(), 10000, "label6", table_id, | |
| partition_id, tablet_id)); | |
| ASSERT_NO_FATAL_FAILURE(insert_rowset(meta_service.get(), 10000, "label7", table_id, | |
| partition_id, tablet_id)); | |
| } | |
| { | |
| GetTabletStatsResponse res; | |
| get_tablet_stats(meta_service.get(), table_id, index_id, partition_id, tablet_id, res); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK); | |
| ASSERT_EQ(res.tablet_stats_size(), 1); | |
| EXPECT_EQ(res.tablet_stats(0).data_size(), 77000); | |
| EXPECT_EQ(res.tablet_stats(0).num_rows(), 700); | |
| EXPECT_EQ(res.tablet_stats(0).num_rowsets(), 8); | |
| EXPECT_EQ(res.tablet_stats(0).num_segments(), 7); | |
| EXPECT_EQ(res.tablet_stats(0).index_size(), 7000); | |
| EXPECT_EQ(res.tablet_stats(0).segment_size(), 70000); | |
| } | |
| { | |
| // Compact some rowsets, the rows is not changed | |
| auto txn_kv = meta_service->txn_kv(); | |
| compact_rowset(txn_kv.get(), instance_id, tablet_id, partition_id, 2, 4, 300); | |
| compact_rowset(txn_kv.get(), instance_id, tablet_id, partition_id, 6, 7, 200); | |
| } | |
| { | |
| // rowset.set_num_segments(1); | |
| // rowset.set_num_rows(num_rows); | |
| // rowset.set_data_disk_size(num_rows * 100); | |
| // rowset.set_index_disk_size(num_rows * 10); | |
| // rowset.set_total_disk_size(num_rows * 110); | |
| GetTabletStatsResponse res; | |
| get_tablet_stats(meta_service.get(), table_id, index_id, partition_id, tablet_id, res); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK); | |
| ASSERT_EQ(res.tablet_stats_size(), 1); | |
| EXPECT_EQ(res.tablet_stats(0).data_size(), 77000); | |
| EXPECT_EQ(res.tablet_stats(0).num_rows(), 700); | |
| EXPECT_EQ(res.tablet_stats(0).num_rowsets(), 5); // 0-1, 2-4, 5, 6-7, 8 | |
| EXPECT_EQ(res.tablet_stats(0).num_segments(), 4); | |
| EXPECT_EQ(res.tablet_stats(0).index_size(), 7000); | |
| EXPECT_EQ(res.tablet_stats(0).segment_size(), 70000); | |
| } | |
| } | |
| TEST(MetaServiceVersionedReadTest, GetRowsetMetas) { | |
| auto service = get_meta_service(false); | |
| std::string instance_id = "test_cloud_instance_id"; | |
| MOCK_GET_INSTANCE_ID(instance_id); | |
| create_and_refresh_instance(service.get(), instance_id); | |
| int64_t db_id = 1; | |
| int64_t table_id = 2; | |
| int64_t index_id = 3; | |
| int64_t partition_id = 4; | |
| int64_t tablet_id = 5; | |
| // Create tablet | |
| create_tablet(service.get(), table_id, index_id, partition_id, tablet_id); | |
| // Helper function to get rowsets using MetaService get_rowset interface | |
| auto get_rowsets = [&](int64_t start_version, int64_t end_version) -> GetRowsetResponse { | |
| brpc::Controller cntl; | |
| GetRowsetRequest req; | |
| auto* tablet_idx = req.mutable_idx(); | |
| tablet_idx->set_db_id(db_id); | |
| tablet_idx->set_table_id(table_id); | |
| tablet_idx->set_index_id(index_id); | |
| tablet_idx->set_partition_id(partition_id); | |
| tablet_idx->set_tablet_id(tablet_id); | |
| req.set_start_version(start_version); | |
| req.set_end_version(end_version); | |
| req.set_cumulative_compaction_cnt(0); | |
| req.set_base_compaction_cnt(0); | |
| req.set_cumulative_point(2); | |
| GetRowsetResponse resp; | |
| service->get_rowset(&cntl, &req, &resp, nullptr); | |
| return resp; | |
| }; | |
| // Test 1: Contains the first rowset | |
| { | |
| auto resp = get_rowsets(1, 10); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK) << resp.status().DebugString(); | |
| ASSERT_EQ(resp.rowset_meta_size(), 1); | |
| EXPECT_TRUE(resp.has_partition_max_version()); | |
| EXPECT_EQ(resp.partition_max_version(), -1); // The first version is special | |
| } | |
| // Step 1: Insert loading rowsets with versions 2, 3, 4, 5, 6, 7, 8, 9, 10 | |
| LOG(INFO) << "Inserting loading rowsets for versions 2-10"; | |
| for (int64_t version = 2; version <= 10; ++version) { | |
| insert_rowset(service.get(), 1, fmt::format("load_label_{}", version), table_id, | |
| partition_id, tablet_id); | |
| } | |
| // Test 2: Get all loading rowsets | |
| { | |
| auto resp = get_rowsets(1, 10); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK) << resp.status().DebugString(); | |
| ASSERT_EQ(resp.rowset_meta_size(), 10); | |
| EXPECT_TRUE(resp.has_partition_max_version()); | |
| EXPECT_EQ(resp.partition_max_version(), 10); | |
| // Verify rowsets are sorted by version | |
| for (int i = 0; i < resp.rowset_meta_size(); ++i) { | |
| const auto& meta = resp.rowset_meta(i); | |
| ASSERT_EQ(meta.end_version(), i + 1) << meta.DebugString(); | |
| } | |
| } | |
| // Step 2: Insert compact rowsets [3-5] and [9-10] | |
| LOG(INFO) << "Inserting compact rowsets [3-5] and [9-10]"; | |
| // Create compact rowset [3-4] | |
| { | |
| std::unique_ptr<Transaction> txn; | |
| ASSERT_EQ(service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK); | |
| insert_compact_rowset(txn.get(), instance_id, tablet_id, partition_id, 3, 4, 300); | |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); | |
| } | |
| // Create compact rowset [3-5] | |
| { | |
| std::unique_ptr<Transaction> txn; | |
| ASSERT_EQ(service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK); | |
| insert_compact_rowset(txn.get(), instance_id, tablet_id, partition_id, 3, 5, 300); | |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); | |
| } | |
| // Create compact rowset [9-10] | |
| { | |
| std::unique_ptr<Transaction> txn; | |
| ASSERT_EQ(service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK); | |
| insert_compact_rowset(txn.get(), instance_id, tablet_id, partition_id, 9, 10, 190); | |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); | |
| } | |
| // Test 3: Verify compact rowsets override load rowsets | |
| { | |
| auto resp = get_rowsets(1, 10); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK); | |
| // 1, 2, [3-5], 6, 7, 8, [9-10] | |
| ASSERT_EQ(resp.rowset_meta_size(), 7) << resp.ShortDebugString(); | |
| // Expected sequence: v1, v2, [3-5], v6, v7, v8, [9-10] | |
| std::vector<std::pair<int64_t, int64_t>> expected_versions = { | |
| {0, 1}, {2, 2}, {3, 5}, {6, 6}, {7, 7}, {8, 8}, {9, 10}}; | |
| for (int i = 0; i < resp.rowset_meta_size(); ++i) { | |
| const auto& meta = resp.rowset_meta(i); | |
| ASSERT_EQ(meta.start_version(), expected_versions[i].first) | |
| << "Rowset " << i << " start_version mismatch"; | |
| ASSERT_EQ(meta.end_version(), expected_versions[i].second) | |
| << "Rowset " << i << " end_version mismatch"; | |
| } | |
| } | |
| // Test 4: Boundary cases - query ranges that test boundaries | |
| struct TestCase { | |
| int64_t start_version; | |
| int64_t end_version; | |
| std::vector<std::pair<int64_t, int64_t>> expected_ranges; | |
| std::string description; | |
| }; | |
| std::vector<TestCase> test_cases = { | |
| // Test exact version match | |
| {1, 1, {{0, 1}}, "Single version 1"}, | |
| {2, 2, {{2, 2}}, "Single version 2"}, | |
| // Test range before compact | |
| {1, 2, {{0, 1}, {2, 2}}, "Range [1-2] before compact"}, | |
| // Test range ending at compact boundary | |
| {2, 5, {{2, 2}, {3, 5}}, "Range [2-5] ending at compact boundary"}, | |
| // Test range starting at compact boundary | |
| {3, 6, {{3, 5}, {6, 6}}, "Range [3-6] starting at compact boundary"}, | |
| // Test exact compact range | |
| {3, 5, {{3, 5}}, "Exact compact range [3-5]"}, | |
| {9, 10, {{9, 10}}, "Exact compact range [9-10]"}, | |
| // Test range spanning compact | |
| {2, 6, {{2, 2}, {3, 5}, {6, 6}}, "Range [2-6] spanning compact"}, | |
| // Test range between compacts | |
| {6, 8, {{6, 6}, {7, 7}, {8, 8}}, "Range [6-8] between compacts"}, | |
| // Test range crossing second compact | |
| {8, 10, {{8, 8}, {9, 10}}, "Range [8-10] crossing second compact"}, | |
| // Test partial compact overlap | |
| {4, 6, {{3, 5}, {6, 6}}, "Range [4-6] partial compact overlap"}, | |
| {9, 10, {{9, 10}}, "Range [9-10] contains last version"}, | |
| // Test edge cases | |
| {5, | |
| 10, | |
| {{3, 5}, {6, 6}, {7, 7}, {8, 8}, {9, 10}}, | |
| "Range [5-10] crossing both compacts"}, | |
| {5, 9, {{3, 5}, {6, 6}, {7, 7}, {8, 8}, {9, 9}}, "Range [5-9] includes first compact"}, | |
| {3, 4, {{3, 4}}, "Range [3-4] within first compact"}, | |
| }; | |
| for (const auto& test_case : test_cases) { | |
| LOG(INFO) << "Testing: " << test_case.description; | |
| auto resp = get_rowsets(test_case.start_version, test_case.end_version); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK) | |
| << "Failed for " << test_case.description; | |
| ASSERT_EQ(resp.rowset_meta_size(), test_case.expected_ranges.size()) | |
| << "Rowset count mismatch for " << test_case.description << ". Got " | |
| << resp.rowset_meta_size() << " rowsets, expected " | |
| << test_case.expected_ranges.size(); | |
| for (int i = 0; i < resp.rowset_meta_size(); ++i) { | |
| const auto& meta = resp.rowset_meta(i); | |
| const auto& expected = test_case.expected_ranges[i]; | |
| ASSERT_EQ(meta.start_version(), expected.first) | |
| << "Start version mismatch for " << test_case.description << " at index " << i; | |
| ASSERT_EQ(meta.end_version(), expected.second) | |
| << "End version mismatch for " << test_case.description << " at index " << i | |
| << " \nRowsetMeta: " << meta.ShortDebugString() | |
| << " \nResponse: " << resp.ShortDebugString(); | |
| } | |
| } | |
| // Test 5: Test empty ranges | |
| { | |
| auto resp = get_rowsets(11, 15); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK); | |
| ASSERT_EQ(resp.rowset_meta_size(), 0) << "Should return empty for non-existent versions"; | |
| } | |
| LOG(INFO) << "GetRowsetMetas test completed successfully"; | |
| } | |
| TEST(MetaServiceVersionedReadTest, UpdateTablet) { | |
| auto service = get_meta_service(false); | |
| std::string instance_id = "test_cloud_instance_id"; | |
| std::string cloud_unique_id = fmt::format("1:{}:1", instance_id); | |
| MOCK_GET_INSTANCE_ID(instance_id); | |
| create_and_refresh_instance(service.get(), instance_id); | |
| constexpr auto table_id = 11231, index_id = 11232, partition_id = 11233, tablet_id1 = 11234, | |
| tablet_id2 = 21234; | |
| ASSERT_NO_FATAL_FAILURE( | |
| create_tablet(service.get(), table_id, index_id, partition_id, tablet_id1)); | |
| ASSERT_NO_FATAL_FAILURE( | |
| create_tablet(service.get(), table_id, index_id, partition_id, tablet_id2)); | |
| auto get_and_check_tablet_meta = [&](int tablet_id, int64_t ttl_seconds, bool in_memory, | |
| bool is_persistent) { | |
| brpc::Controller cntl; | |
| GetTabletRequest req; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| req.set_tablet_id(tablet_id); | |
| GetTabletResponse resp; | |
| service->get_tablet(&cntl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK) << tablet_id; | |
| EXPECT_EQ(resp.tablet_meta().ttl_seconds(), ttl_seconds); | |
| EXPECT_EQ(resp.tablet_meta().is_in_memory(), in_memory); | |
| EXPECT_EQ(resp.tablet_meta().is_persistent(), is_persistent); | |
| }; | |
| get_and_check_tablet_meta(tablet_id1, 0, false, false); | |
| get_and_check_tablet_meta(tablet_id2, 0, false, false); | |
| { | |
| brpc::Controller cntl; | |
| UpdateTabletRequest req; | |
| UpdateTabletResponse resp; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| TabletMetaInfoPB* tablet_meta_info = req.add_tablet_meta_infos(); | |
| tablet_meta_info->set_tablet_id(tablet_id1); | |
| tablet_meta_info->set_ttl_seconds(300); | |
| tablet_meta_info = req.add_tablet_meta_infos(); | |
| tablet_meta_info->set_tablet_id(tablet_id2); | |
| tablet_meta_info->set_ttl_seconds(3000); | |
| service->update_tablet(&cntl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK); | |
| } | |
| get_and_check_tablet_meta(tablet_id1, 300, false, false); | |
| get_and_check_tablet_meta(tablet_id2, 3000, false, false); | |
| { | |
| brpc::Controller cntl; | |
| UpdateTabletRequest req; | |
| UpdateTabletResponse resp; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| TabletMetaInfoPB* tablet_meta_info = req.add_tablet_meta_infos(); | |
| tablet_meta_info->set_tablet_id(tablet_id1); | |
| tablet_meta_info->set_is_in_memory(true); | |
| service->update_tablet(&cntl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK); | |
| } | |
| { | |
| brpc::Controller cntl; | |
| UpdateTabletRequest req; | |
| UpdateTabletResponse resp; | |
| req.set_cloud_unique_id(cloud_unique_id); | |
| TabletMetaInfoPB* tablet_meta_info = req.add_tablet_meta_infos(); | |
| tablet_meta_info->set_tablet_id(tablet_id1); | |
| tablet_meta_info->set_is_persistent(true); | |
| service->update_tablet(&cntl, &req, &resp, nullptr); | |
| ASSERT_EQ(resp.status().code(), MetaServiceCode::OK); | |
| } | |
| get_and_check_tablet_meta(tablet_id1, 300, true, true); | |
| } | |
| TEST(MetaServiceVersionedReadTest, IndexRequest) { | |
| auto meta_service = get_meta_service(false); | |
| std::string instance_id = "commit_index"; | |
| MOCK_GET_INSTANCE_ID(instance_id); | |
| create_and_refresh_instance(meta_service.get(), instance_id); | |
| constexpr int64_t db_id = 123; | |
| constexpr int64_t table_id = 10001; | |
| constexpr int64_t index_id = 10002; | |
| { | |
| // Prepare index | |
| brpc::Controller ctrl; | |
| IndexRequest req; | |
| IndexResponse res; | |
| req.set_db_id(db_id); | |
| req.set_table_id(table_id); | |
| req.add_index_ids(index_id); | |
| meta_service->prepare_index(&ctrl, &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().DebugString(); | |
| } | |
| { | |
| // Commit index | |
| brpc::Controller ctrl; | |
| IndexRequest req; | |
| IndexResponse res; | |
| req.set_db_id(db_id); | |
| req.set_table_id(table_id); | |
| req.add_index_ids(index_id); | |
| req.set_is_new_table(true); | |
| meta_service->commit_index(&ctrl, &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().DebugString(); | |
| ASSERT_TRUE(res.has_table_version()); | |
| int64_t table_version = 0; | |
| get_table_version(meta_service.get(), db_id, table_id, table_version); | |
| ASSERT_EQ(table_version, res.table_version()); | |
| } | |
| { | |
| // Prepare index again | |
| brpc::Controller ctrl; | |
| IndexRequest req; | |
| IndexResponse res; | |
| req.set_db_id(db_id); | |
| req.set_table_id(table_id); | |
| req.add_index_ids(index_id); | |
| meta_service->prepare_index(&ctrl, &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::ALREADY_EXISTED) | |
| << res.status().DebugString(); | |
| } | |
| { | |
| // Commit index again | |
| brpc::Controller ctrl; | |
| IndexRequest req; | |
| IndexResponse res; | |
| req.set_db_id(db_id); | |
| req.set_table_id(table_id); | |
| req.add_index_ids(index_id); | |
| req.set_is_new_table(true); | |
| meta_service->commit_index(&ctrl, &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().DebugString(); | |
| ASSERT_FALSE(res.has_table_version()); | |
| } | |
| } | |
| TEST(MetaServiceVersionedReadTest, PartitionRequest) { | |
| auto meta_service = get_meta_service(false); | |
| std::string instance_id = "PartitionRequest"; | |
| MOCK_GET_INSTANCE_ID(instance_id); | |
| create_and_refresh_instance(meta_service.get(), instance_id); | |
| constexpr int64_t db_id = 1; | |
| constexpr int64_t table_id = 10001; | |
| constexpr int64_t index_id = 10002; | |
| constexpr int64_t partition_id = 10003; | |
| { | |
| // Prepare transaction | |
| brpc::Controller ctrl; | |
| PartitionRequest req; | |
| PartitionResponse res; | |
| req.set_db_id(db_id); | |
| req.set_table_id(table_id); | |
| req.add_index_ids(index_id); | |
| req.add_partition_ids(partition_id); | |
| meta_service->prepare_partition(&ctrl, &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().DebugString(); | |
| } | |
| { | |
| // Commit partition | |
| brpc::Controller ctrl; | |
| PartitionRequest req; | |
| PartitionResponse res; | |
| req.set_db_id(db_id); | |
| req.set_table_id(table_id); | |
| req.add_index_ids(index_id); | |
| req.add_partition_ids(partition_id); | |
| meta_service->commit_partition(&ctrl, &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().DebugString(); | |
| ASSERT_TRUE(res.has_table_version()); | |
| int64_t table_version = 0; | |
| get_table_version(meta_service.get(), db_id, table_id, table_version); | |
| ASSERT_EQ(table_version, res.table_version()); | |
| } | |
| { | |
| // Prepare partition again | |
| brpc::Controller ctrl; | |
| PartitionRequest req; | |
| PartitionResponse res; | |
| req.set_db_id(db_id); | |
| req.set_table_id(table_id); | |
| req.add_index_ids(index_id); | |
| req.add_partition_ids(partition_id); | |
| meta_service->prepare_partition(&ctrl, &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::ALREADY_EXISTED) | |
| << res.status().DebugString(); | |
| } | |
| { | |
| // Commit partition again | |
| brpc::Controller ctrl; | |
| PartitionRequest req; | |
| PartitionResponse res; | |
| req.set_db_id(db_id); | |
| req.set_table_id(table_id); | |
| req.add_index_ids(index_id); | |
| req.add_partition_ids(partition_id); | |
| meta_service->commit_partition(&ctrl, &req, &res, nullptr); | |
| ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().DebugString(); | |
| } | |
| } | |
| } // namespace doris::cloud |