cloud/test/txn_lazy_commit_test.cpp - doris - Git at Google

 // 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/keys.h"
 #include "meta-service/mem_txn_kv.h"
 #include "meta-service/meta_service.h"
 #include "meta-service/meta_service_helper.h"
 #include "meta-service/txn_kv_error.h"
 #include "mock_resource_manager.h"
 #include "rate-limiter/rate_limiter.h"
 #include "resource-manager/resource_manager.h"

 int main(int argc, char** argv) {
     const std::string conf_file = "doris_cloud.conf";
     if (!doris::cloud::config::init(conf_file.c_str(), true)) {
         std::cerr << "failed to init config file, conf=" << conf_file << std::endl;
         return -1;
     }

     config::enable_cloud_txn_lazy_commit = true;
     config::txn_lazy_commit_rowsets_thresold = 1;
     config::txn_lazy_max_rowsets_per_batch = 1;
     config::txn_lazy_commit_num_threads = 2;

     if (!doris::cloud::init_glog("txn_lazy_commit_test")) {
         std::cerr << "failed to init glog" << std::endl;
         return -1;
     }
     ::testing::InitGoogleTest(&argc, argv);
     return RUN_ALL_TESTS();
 }
 namespace doris::cloud {

 std::unique_ptr<MetaServiceProxy> get_meta_service(std::shared_ptr<TxnKv> txn_kv,
                                                    bool mock_resource_mgr) {
     std::unique_ptr<Transaction> txn;
     EXPECT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
     txn->remove("\x00", "\xfe"); // This is dangerous if the fdb is not correctly set
     EXPECT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

     auto rs = mock_resource_mgr ? std::make_shared<MockResourceManager>(txn_kv)
                                 : std::make_shared<ResourceManager>(txn_kv);
     auto rl = std::make_shared<RateLimiter>();
     auto meta_service = std::make_unique<MetaServiceImpl>(txn_kv, rs, rl);
     return std::make_unique<MetaServiceProxy>(std::move(meta_service));
 }

 static std::string next_rowset_id() {
     static int cnt = 0;
     return std::to_string(++cnt);
 }

 static void add_tablet(CreateTabletsRequest& req, int64_t table_id, int64_t index_id,
                        int64_t partition_id, int64_t tablet_id) {
     auto tablet = req.add_tablet_metas();
     tablet->set_table_id(table_id);
     tablet->set_index_id(index_id);
     tablet->set_partition_id(partition_id);
     tablet->set_tablet_id(tablet_id);
     auto schema = tablet->mutable_schema();
     schema->set_schema_version(0);
     auto first_rowset = tablet->add_rs_metas();
     first_rowset->set_rowset_id(0); // required
     first_rowset->set_rowset_id_v2(next_rowset_id());
     first_rowset->set_start_version(0);
     first_rowset->set_end_version(1);
     first_rowset->mutable_tablet_schema()->CopyFrom(*schema);
 }

 static void create_tablet(MetaServiceProxy* meta_service, int64_t table_id, int64_t index_id,
                           int64_t partition_id, int64_t tablet_id) {
     brpc::Controller cntl;
     CreateTabletsRequest req;
     CreateTabletsResponse res;
     add_tablet(req, table_id, index_id, partition_id, tablet_id);
     meta_service->create_tablets(&cntl, &req, &res, nullptr);
     ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << tablet_id;
 }

 static doris::RowsetMetaCloudPB create_rowset(int64_t txn_id, int64_t tablet_id,
                                               int partition_id = 10, int64_t version = -1,
                                               int num_rows = 100) {
     doris::RowsetMetaCloudPB rowset;
     rowset.set_rowset_id(0); // required
     rowset.set_rowset_id_v2(next_rowset_id());
     rowset.set_tablet_id(tablet_id);
     rowset.set_partition_id(partition_id);
     rowset.set_txn_id(txn_id);
     if (version > 0) {
         rowset.set_start_version(version);
         rowset.set_end_version(version);
     }
     rowset.set_num_segments(1);
     rowset.set_num_rows(num_rows);
     rowset.set_data_disk_size(num_rows * 100);
     rowset.mutable_tablet_schema()->set_schema_version(0);
     rowset.set_txn_expiration(::time(nullptr)); // Required by DCHECK
     return rowset;
 }

 static void commit_rowset(MetaServiceProxy* meta_service, const doris::RowsetMetaCloudPB& rowset,
                           CreateRowsetResponse& res) {
     brpc::Controller cntl;
     auto arena = res.GetArena();
     auto req = google::protobuf::Arena::CreateMessage<CreateRowsetRequest>(arena);
     req->mutable_rowset_meta()->CopyFrom(rowset);
     meta_service->commit_rowset(&cntl, req, &res, nullptr);
     if (!arena) {
         delete req;
     }
 }

 TEST(TxnLazyCommitTest, CommitTxnEventuallyTest) {
     int ret = 0;
     // MemKv
     auto txn_kv = std::dynamic_pointer_cast<TxnKv>(std::make_shared<MemTxnKv>());
     if (txn_kv != nullptr) {
         ret = txn_kv->init();
         [&] { ASSERT_EQ(ret, 0); }();
     }
     [&] { ASSERT_NE(txn_kv.get(), nullptr); }();

     int64_t db_id = 3213235;
     int64_t table_id = 93213131;

     auto meta_service = get_meta_service(txn_kv, true);
     brpc::Controller cntl;
     BeginTxnRequest req;
     req.set_cloud_unique_id("test_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);
     int64_t 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, 1235, 1236, tablet_id_base + i);
         auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i);
         CreateRowsetResponse res;
         commit_rowset(meta_service.get(), tmp_rowset, res);
         ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
     }

     {
         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(false);
         req.set_enable_txn_lazy_commit(true);
         CommitTxnResponse res;
         meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                  &res, nullptr);
         ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
     }

     {
         std::unique_ptr<Transaction> txn;
         ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
         std::string mock_instance = "test_instance";
         int64_t tablet_id_base = 1103;
         for (int i = 0; i < 5; ++i) {
             int64_t tablet_id = tablet_id_base + i;
             std::string key = meta_tablet_idx_key({mock_instance, tablet_id});
             std::string val;
             ASSERT_EQ(txn->get(key, &val), TxnErrorCode::TXN_OK);
             TabletIndexPB tablet_idx_pb;
             tablet_idx_pb.ParseFromString(val);
             ASSERT_EQ(tablet_idx_pb.db_id(), db_id);
         }
     }
 }
 } // namespace doris::cloud
	// 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/keys.h"
	#include "meta-service/mem_txn_kv.h"
	#include "meta-service/meta_service.h"
	#include "meta-service/meta_service_helper.h"
	#include "meta-service/txn_kv_error.h"
	#include "mock_resource_manager.h"
	#include "rate-limiter/rate_limiter.h"
	#include "resource-manager/resource_manager.h"

	int main(int argc, char** argv) {
	const std::string conf_file = "doris_cloud.conf";
	if (!doris::cloud::config::init(conf_file.c_str(), true)) {
	std::cerr << "failed to init config file, conf=" << conf_file << std::endl;
	return -1;
	}

	config::enable_cloud_txn_lazy_commit = true;
	config::txn_lazy_commit_rowsets_thresold = 1;
	config::txn_lazy_max_rowsets_per_batch = 1;
	config::txn_lazy_commit_num_threads = 2;

	if (!doris::cloud::init_glog("txn_lazy_commit_test")) {
	std::cerr << "failed to init glog" << std::endl;
	return -1;
	}
	::testing::InitGoogleTest(&argc, argv);
	return RUN_ALL_TESTS();
	}
	namespace doris::cloud {

	std::unique_ptr<MetaServiceProxy> get_meta_service(std::shared_ptr<TxnKv> txn_kv,
	bool mock_resource_mgr) {
	std::unique_ptr<Transaction> txn;
	EXPECT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
	txn->remove("\x00", "\xfe"); // This is dangerous if the fdb is not correctly set
	EXPECT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

	auto rs = mock_resource_mgr ? std::make_shared<MockResourceManager>(txn_kv)
	: std::make_shared<ResourceManager>(txn_kv);
	auto rl = std::make_shared<RateLimiter>();
	auto meta_service = std::make_unique<MetaServiceImpl>(txn_kv, rs, rl);
	return std::make_unique<MetaServiceProxy>(std::move(meta_service));
	}

	static std::string next_rowset_id() {
	static int cnt = 0;
	return std::to_string(++cnt);
	}

	static void add_tablet(CreateTabletsRequest& req, int64_t table_id, int64_t index_id,
	int64_t partition_id, int64_t tablet_id) {
	auto tablet = req.add_tablet_metas();
	tablet->set_table_id(table_id);
	tablet->set_index_id(index_id);
	tablet->set_partition_id(partition_id);
	tablet->set_tablet_id(tablet_id);
	auto schema = tablet->mutable_schema();
	schema->set_schema_version(0);
	auto first_rowset = tablet->add_rs_metas();
	first_rowset->set_rowset_id(0); // required
	first_rowset->set_rowset_id_v2(next_rowset_id());
	first_rowset->set_start_version(0);
	first_rowset->set_end_version(1);
	first_rowset->mutable_tablet_schema()->CopyFrom(*schema);
	}

	static void create_tablet(MetaServiceProxy* meta_service, int64_t table_id, int64_t index_id,
	int64_t partition_id, int64_t tablet_id) {
	brpc::Controller cntl;
	CreateTabletsRequest req;
	CreateTabletsResponse res;
	add_tablet(req, table_id, index_id, partition_id, tablet_id);
	meta_service->create_tablets(&cntl, &req, &res, nullptr);
	ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << tablet_id;
	}

	static doris::RowsetMetaCloudPB create_rowset(int64_t txn_id, int64_t tablet_id,
	int partition_id = 10, int64_t version = -1,
	int num_rows = 100) {
	doris::RowsetMetaCloudPB rowset;
	rowset.set_rowset_id(0); // required
	rowset.set_rowset_id_v2(next_rowset_id());
	rowset.set_tablet_id(tablet_id);
	rowset.set_partition_id(partition_id);
	rowset.set_txn_id(txn_id);
	if (version > 0) {
	rowset.set_start_version(version);
	rowset.set_end_version(version);
	}
	rowset.set_num_segments(1);
	rowset.set_num_rows(num_rows);
	rowset.set_data_disk_size(num_rows * 100);
	rowset.mutable_tablet_schema()->set_schema_version(0);
	rowset.set_txn_expiration(::time(nullptr)); // Required by DCHECK
	return rowset;
	}

	static void commit_rowset(MetaServiceProxy* meta_service, const doris::RowsetMetaCloudPB& rowset,
	CreateRowsetResponse& res) {
	brpc::Controller cntl;
	auto arena = res.GetArena();
	auto req = google::protobuf::Arena::CreateMessage<CreateRowsetRequest>(arena);
	req->mutable_rowset_meta()->CopyFrom(rowset);
	meta_service->commit_rowset(&cntl, req, &res, nullptr);
	if (!arena) {
	delete req;
	}
	}

	TEST(TxnLazyCommitTest, CommitTxnEventuallyTest) {
	int ret = 0;
	// MemKv
	auto txn_kv = std::dynamic_pointer_cast<TxnKv>(std::make_shared<MemTxnKv>());
	if (txn_kv != nullptr) {
	ret = txn_kv->init();
	[&] { ASSERT_EQ(ret, 0); }();
	}
	[&] { ASSERT_NE(txn_kv.get(), nullptr); }();

	int64_t db_id = 3213235;
	int64_t table_id = 93213131;

	auto meta_service = get_meta_service(txn_kv, true);
	brpc::Controller cntl;
	BeginTxnRequest req;
	req.set_cloud_unique_id("test_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);
	int64_t 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, 1235, 1236, tablet_id_base + i);
	auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i);
	CreateRowsetResponse res;
	commit_rowset(meta_service.get(), tmp_rowset, res);
	ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
	}

	{
	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(false);
	req.set_enable_txn_lazy_commit(true);
	CommitTxnResponse res;
	meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
	&res, nullptr);
	ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
	}

	{
	std::unique_ptr<Transaction> txn;
	ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
	std::string mock_instance = "test_instance";
	int64_t tablet_id_base = 1103;
	for (int i = 0; i < 5; ++i) {
	int64_t tablet_id = tablet_id_base + i;
	std::string key = meta_tablet_idx_key({mock_instance, tablet_id});
	std::string val;
	ASSERT_EQ(txn->get(key, &val), TxnErrorCode::TXN_OK);
	TabletIndexPB tablet_idx_pb;
	tablet_idx_pb.ParseFromString(val);
	ASSERT_EQ(tablet_idx_pb.db_id(), db_id);
	}
	}
	}
	} // namespace doris::cloud