hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
v4.1
oceanbase/mittest/mtlenv/storage/test_trans.cpp

377 lines
13 KiB
C++
Raw Permalink Blame History

/**
* Copyright (c) 2021 OceanBase
* OceanBase CE is licensed under Mulan PubL v2.
* You can use this software according to the terms and conditions of the Mulan PubL v2.
* You may obtain a copy of Mulan PubL v2 at:
* http://license.coscl.org.cn/MulanPubL-2.0
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
* See the Mulan PubL v2 for more details.
*/
#define USING_LOG_PREFIX STORAGE
#include <gtest/gtest.h>
#include "mtlenv/mock_tenant_module_env.h"
#include "storage/mockcontainer/mock_ob_iterator.h"
#include "storage/mockcontainer/mock_ob_end_trans_callback.h"
#include "storage/init_basic_struct.h"
#include "storage/test_tablet_helper.h"
#include "share/schema/ob_table_dml_param.h"
#include "observer/ob_safe_destroy_thread.h"
#include "storage/ob_dml_running_ctx.h"
#include "storage/tx/ob_trans_part_ctx.h"
namespace oceanbase
{
namespace storage
{
int ObDMLRunningCtx::init(
const common::ObIArray<uint64_t> *column_ids,
const common::ObIArray<uint64_t> *upd_col_ids,
ObMultiVersionSchemaService *schema_service,
ObTabletHandle &tablet_handle)
{
int ret = OB_SUCCESS;
if (IS_INIT) {
ret = OB_INIT_TWICE;
LOG_WARN("init twice", K(ret));
} else if (OB_UNLIKELY(!store_ctx_.is_valid())
|| OB_UNLIKELY(!dml_param_.is_valid())
|| OB_ISNULL(dml_param_.table_param_)
|| OB_ISNULL(schema_service)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(store_ctx_),
K(dml_param_), KP(schema_service));
} else {
const uint64_t tenant_id = MTL_ID();
const uint64_t table_id = dml_param_.table_param_->get_data_table().get_table_id();
const int64_t version = dml_param_.schema_version_;
const int64_t tenant_schema_version = dml_param_.tenant_schema_version_;
}
if (OB_FAIL(ret)) {
// do nothing
} else if (OB_FAIL(prepare_relative_table(
dml_param_.table_param_->get_data_table(),
tablet_handle,
store_ctx_.mvcc_acc_ctx_.get_snapshot_version()))) {
LOG_WARN("failed to get relative table", K(ret), K(dml_param_));
} else if (NULL != column_ids && OB_FAIL(prepare_column_info(*column_ids))) {
LOG_WARN("fail to get column descriptions and column map", K(ret), K(*column_ids));
} else {
store_ctx_.table_version_ = dml_param_.schema_version_;
column_ids_ = column_ids;
is_inited_ = true;
}
if (IS_NOT_INIT) {
relative_table_.destroy();
}
return ret;
}
}
namespace transaction
{
int ObLocationAdapter::get_leader_(const int64_t cluster_id,
const int64_t tenant_id,
const ObLSID &ls_id,
common::ObAddr &leader,
const bool is_sync)
{
UNUSED(cluster_id);
UNUSED(tenant_id);
UNUSED(ls_id);
UNUSED(is_sync);
leader = GCTX.self_addr();
return OB_SUCCESS;
}
}
using namespace storage;
class TestTrans : public ::testing::Test
{
public:
static void SetUpTestCase()
{
LOG_INFO("SetUpTestCase");
ASSERT_EQ(OB_SUCCESS, MockTenantModuleEnv::get_instance().init());
SAFE_DESTROY_INSTANCE.init();
SAFE_DESTROY_INSTANCE.start();
ObServerCheckpointSlogHandler::get_instance().is_started_ = true;
}
static void TearDownTestCase()
{
LOG_INFO("TearDownTestCase");
SAFE_DESTROY_INSTANCE.stop();
SAFE_DESTROY_INSTANCE.wait();
SAFE_DESTROY_INSTANCE.destroy();
MockTenantModuleEnv::get_instance().destroy();
}
void create_ls(uint64_t tenant_id, ObLSID &ls_id, ObLS *&ls);
void insert_rows(ObLSID &ls_id, ObTabletID &tablet_id, ObTxDesc &tx_desc, ObTxReadSnapshot snapshot, const char* in_str);
void prepare_tx_desc(ObTxDesc *&tx_desc, ObTxReadSnapshot &snapshot);
private:
static share::schema::ObTableSchema table_schema_;
};
share::schema::ObTableSchema TestTrans::table_schema_;
void TestTrans::create_ls(uint64_t tenant_id, ObLSID &ls_id, ObLS *&ls)
{
LOG_INFO("create log stream");
ObCreateLSArg arg;
ASSERT_EQ(OB_SUCCESS, gen_create_ls_arg(tenant_id, ls_id, arg));
ObLSService* ls_svr = MTL(ObLSService*);
ASSERT_EQ(OB_SUCCESS, ls_svr->create_ls(arg));
// set member list
LOG_INFO("set member list");
ObLSHandle handle;
ASSERT_EQ(OB_SUCCESS, ls_svr->get_ls(ls_id, handle, ObLSGetMod::STORAGE_MOD));
ls = handle.get_ls();
ObMemberList member_list;
int64_t paxos_replica_num = 1;
(void) member_list.add_server(MockTenantModuleEnv::get_instance().self_addr_);
ASSERT_EQ(OB_SUCCESS, ls->set_initial_member_list(member_list,
paxos_replica_num));
// check leader
LOG_INFO("check leader");
for (int i = 0; i < 15; i++) {
ObRole role;
int64_t leader_epoch = 0;
ASSERT_EQ(OB_SUCCESS, ls->get_log_handler()->get_role(role, leader_epoch));
if (role == ObRole::LEADER) {
break;
}
::sleep(1);
}
}
void TestTrans::insert_rows(ObLSID &ls_id, ObTabletID &tablet_id, ObTxDesc &tx_desc, ObTxReadSnapshot snapshot, const char* ins_str)
{
int64_t affected_rows = 0;
ObMockNewRowIterator ins_iter;
ASSERT_EQ(OB_SUCCESS, ins_iter.from(ins_str));
ObArenaAllocator allocator;
share::schema::ObTableDMLParam table_dml_param(allocator);
ObSEArray<const ObTableSchema *, 1> index_schema_array;
ObSEArray<uint64_t, 1> column_ids;
column_ids.push_back(16); // pk
ASSERT_EQ(OB_SUCCESS, table_dml_param.convert(&table_schema_, 1000, column_ids));
ObDMLBaseParam dml_param;
dml_param.timeout_ = ObTimeUtility::current_time() + 100000000;
dml_param.schema_version_ = 1000;
dml_param.table_param_ = &table_dml_param;
dml_param.snapshot_ = snapshot;
auto das = MTL(ObAccessService*);
LOG_INFO("storage access by dml");
ASSERT_EQ(OB_SUCCESS, das->insert_rows(ls_id,
tablet_id,
tx_desc,
dml_param,
column_ids,
&ins_iter,
affected_rows));
}
void TestTrans::prepare_tx_desc(ObTxDesc *&tx_desc, ObTxReadSnapshot &snapshot)
{
ObTxParam tx_p;
tx_p.timeout_us_ = ObTimeUtility::current_time() + 100000000; // us
tx_p.access_mode_ = transaction::ObTxAccessMode::RW;
tx_p.isolation_ = transaction::ObTxIsolationLevel::RC;
tx_p.cluster_id_ = 1;
ObTransService *tx_service = MTL(ObTransService*);
// alloc tx_desc
ASSERT_EQ(OB_SUCCESS, tx_service->acquire_tx(tx_desc, 100));
// init tx_desc and put in map
ASSERT_EQ(OB_SUCCESS, tx_service->start_tx(*tx_desc, tx_p));
LOG_INFO("get snapshot for dml");
{
int64_t expire_ts = ObTimeUtility::current_time() + 100000000;
ObTxIsolationLevel isolation = ObTxIsolationLevel::RC;
ASSERT_EQ(OB_SUCCESS, tx_service->get_read_snapshot(*tx_desc, isolation, expire_ts, snapshot));
}
}
TEST_F(TestTrans, create_ls_and_tablet)
{
int ret = OB_SUCCESS;
ASSERT_EQ(OB_SUCCESS, ObCurTraceId::get_trace_id()->set("Y1-1111111111111111"));
ObLSID ls_id(100);
ObLS *ls = nullptr;
uint64_t tenant_id = MTL_ID();
create_ls(tenant_id, ls_id, ls);
LOG_INFO("create tablet");
obrpc::ObBatchCreateTabletArg create_tablet_arg;
ObTabletID tablet_id(1001);
ASSERT_EQ(OB_SUCCESS, gen_create_tablet_arg(tenant_id, ls_id, tablet_id, create_tablet_arg, 1, &table_schema_));
ObLSTabletService *ls_tablet_svr = ls->get_tablet_svr();
ASSERT_EQ(OB_SUCCESS, TestTabletHelper::create_tablet(*ls_tablet_svr, create_tablet_arg));
}
TEST_F(TestTrans, basic)
{
int ret = OB_SUCCESS;
uint64_t tenant_id = MTL_ID();
// create log stream
ObLSID ls_id(100);
ObTabletID tablet_id(1001);
LOG_INFO("start transaction");
ObTxDesc *tx_desc = NULL;
ObTxReadSnapshot snapshot;
prepare_tx_desc(tx_desc, snapshot);
// prepare insert param
const char *ins_str =
"bigint dml \n"
"50 T_DML_INSERT \n";
insert_rows(ls_id, tablet_id, *tx_desc, snapshot, ins_str);
ObTransService *tx_service = MTL(ObTransService*);
ObPartTransCtx *part_ctx;
ASSERT_EQ(OB_SUCCESS, tx_service->tx_ctx_mgr_.get_tx_ctx(ls_id, tx_desc->tx_id_, false, part_ctx));
ASSERT_EQ(OB_SUCCESS, tx_service->tx_ctx_mgr_.revert_tx_ctx(part_ctx));
ObLSHandle ls_handle;
ASSERT_EQ(OB_SUCCESS, MTL(ObLSService*)->get_ls(ls_id, ls_handle, ObLSGetMod::STORAGE_MOD));
ObTabletHandle tablet_handle;
ASSERT_EQ(OB_SUCCESS, ls_handle.get_ls()->get_tablet_svr()->get_tablet(tablet_id, tablet_handle, 0));
ObIMemtableMgr *mt_mgr = tablet_handle.get_obj()->get_memtable_mgr();
ASSERT_EQ(1, mt_mgr->get_memtable_count_());
ObTableHandleV2 mt_handle;
ASSERT_EQ(OB_SUCCESS, mt_mgr->get_active_memtable(mt_handle));
memtable::ObMemtable *mt;
ASSERT_EQ(OB_SUCCESS, mt_handle.get_data_memtable(mt));
printf("mt size=%ld occ_size=%ld bree_item_count=%ld rec_scn=%s end_scn=%s\n", mt->get_size(), mt->get_occupied_size(), mt->get_btree_item_count(), to_cstring(mt->get_rec_scn()),to_cstring(mt->get_max_end_scn()));
/*
ObStoreRow store_row;
store_row.flag_.set_flag(ObDmlFlag::DF_INSERT);
ObNewRow *new_row;
ASSERT_EQ(OB_SUCCESS, ins_iter.get_next_row(new_row));
store_row.row_val_ = *new_row;
ObStoreRowkey tmp_key;
memtable::ObMemtableKey mtk;
ASSERT_EQ(OB_SUCCESS, tmp_key.assign(store_row.row_val_.cells_, 1));
ASSERT_EQ(OB_SUCCESS, mtk.encode(columns, &tmp_key));
mt->get_query_engine().get();
*/
memtable::ObTransCallbackMgr &trans_mgr = part_ctx->mt_ctx_.trans_mgr_;
printf("part_ctx trans_mgr:%ld\n",trans_mgr.callback_list_.get_length());
auto cb = trans_mgr.callback_list_.head_.next_;
for (int i = 0; i < trans_mgr.callback_list_.get_length(); i++) {
memtable::ObMvccRowCallback *row_cb = dynamic_cast<memtable::ObMvccRowCallback*>(cb);
if (row_cb != nullptr) {
printf("mvcc row: ptr=%p node_cnt=%ld\n", &row_cb->value_, row_cb->value_.get_total_trans_node_cnt());
} else {
ObOBJLockCallback *tl_cb = dynamic_cast<ObOBJLockCallback*>(cb);
if (tl_cb != nullptr) {
printf("table lock node: lock_id=%ld op_type=%c\n", tl_cb->lock_op_->lock_op_.lock_id_.obj_id_, tl_cb->lock_op_->lock_op_.op_type_);
} else {
abort();
}
}
cb = cb->next_;
}
LOG_INFO("commit transaction");
ASSERT_EQ(OB_SUCCESS, tx_service->commit_tx(*tx_desc, ObTimeUtility::current_time() + 100000000));
LOG_INFO("release transaction");
tx_service->release_tx(*tx_desc);
//ASSERT_EQ(OB_SUCCESS, MTL(ObLSService*)->remove_ls(ls_id, false));
}
TEST_F(TestTrans, dist_trans)
{
ASSERT_EQ(OB_SUCCESS, ObCurTraceId::get_trace_id()->set("Y2-2222222222222222222222"));
uint64_t tenant_id = MTL_ID();
ObLSID ls_id(100);
ObTabletID tablet_id(1001);
ObLSID ls_id2(101);
ObLS *ls2 = nullptr;
create_ls(tenant_id, ls_id2, ls2);
// create tablet
LOG_INFO("create tablet");
obrpc::ObBatchCreateTabletArg create_tablet_arg;
ObTabletID tablet_id2(1002);
ASSERT_EQ(OB_SUCCESS, gen_create_tablet_arg(tenant_id, ls_id2, tablet_id2, create_tablet_arg, 1, &table_schema_));
ObLSTabletService *ls_tablet_svr = ls2->get_tablet_svr();
ASSERT_EQ(OB_SUCCESS, TestTabletHelper::create_tablet(*ls_tablet_svr, create_tablet_arg));
LOG_INFO("start transaction");
ObTxDesc *tx_desc = NULL;
ObTxReadSnapshot snapshot;
prepare_tx_desc(tx_desc, snapshot);
const char *ins_str =
"bigint dml \n"
"100 T_DML_INSERT \n";
insert_rows(ls_id, tablet_id, *tx_desc, snapshot, ins_str);
const char *ins_str2 =
"bigint dml \n"
"200 T_DML_INSERT \n";
insert_rows(ls_id2, tablet_id2, *tx_desc, snapshot, ins_str2);
ObTransService *tx_service = MTL(ObTransService*);
LOG_INFO("commit transaction");
ASSERT_EQ(OB_SUCCESS, tx_service->commit_tx(*tx_desc, ObTimeUtility::current_time() + 10000000000));
LOG_INFO("release transaction");
tx_service->release_tx(*tx_desc);
}
/*
TEST_F(TestTrans, freeze)
{
ObLSID ls_id(100);
ObLSHandle ls_handle;
ObLS *ls;
ASSERT_EQ(OB_SUCCESS, MTL(ObLSService*)->get_ls(ls_id, ls_handle, ObLSGetMod::STORAGE_MOD));
ls = ls_handle.get_ls();
ASSERT_EQ(OB_SUCCESS, ls->logstream_freeze());
}
*/
TEST_F(TestTrans, remove_ls)
{
ObLSID ls_id(100);
ObLSID ls_id2(101);
ASSERT_EQ(OB_SUCCESS, MTL(ObLSService*)->remove_ls(ls_id, false));
ASSERT_EQ(OB_SUCCESS, MTL(ObLSService*)->remove_ls(ls_id2, false));
}
} // end oceanbase
int main(int argc, char **argv)
{
system("rm -rf test_trans.log*");
OB_LOGGER.set_file_name("test_trans.log",true, false, "test_trans.log", "test_trans.log");
OB_LOGGER.set_log_level("INFO");
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
Reference in New Issue View Git Blame Copy Permalink