// owner: handora.qc // owner group: transaction /** * 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. */ #include #define protected public #define private public #include "env/ob_simple_cluster_test_base.h" #include "rootserver/ob_tenant_balance_service.h" #include "share/balance/ob_balance_job_table_operator.h" #include "mittest/env/ob_simple_server_helper.h" #include "storage/tx/ob_tx_loop_worker.h" namespace oceanbase { static ObTransID global_tx_id; static share::ObLSID global_ls_id; namespace storage { int ObTransferHandler::wait_src_ls_advance_weak_read_ts_( const share::ObTransferTaskInfo &task_info, ObTimeoutCtx &timeout_ctx) { UNUSED(task_info); UNUSED(timeout_ctx); return OB_SUCCESS; } } namespace transaction { int ObTransService::batch_post_rollback_savepoint_msg_(ObTxDesc &tx, ObTxRollbackSPMsg &msg, const ObTxRollbackParts &list, int &post_succ_num) { int ret = OB_SUCCESS; int last_ret = OB_SUCCESS; post_succ_num = 0; const ObTxDesc *msg_tx_ptr = msg.tx_ptr_; ARRAY_FOREACH_NORET(list, idx) { const ObTxExecPart &p = list.at(idx); msg.receiver_ = p.ls_id_; msg.epoch_ = p.exec_epoch_; if (msg.epoch_ > 0) { msg.tx_ptr_ = NULL; } if (p.exec_epoch_ <= 0 && p.transfer_epoch_ > 0) { msg.set_for_transfer(); } ObCStringHelper helper; if (global_ls_id.is_valid() && global_ls_id == msg.receiver_) { fprintf(stdout, "qcc encounter failure %ld, %s\n", global_ls_id.id(), helper.convert(tx)); } if ((!global_ls_id.is_valid() || global_ls_id != msg.receiver_) && OB_FAIL(rpc_->post_msg(msg.receiver_, msg))) { if (OB_LS_IS_DELETED == ret) { ObSpinLockGuard lock(tx.lock_); ObAddr fake_addr; on_sp_rollback_succ_(p, tx, ObTxPart::EPOCH_DEAD, fake_addr); ret = OB_SUCCESS; } else { TRANS_LOG(WARN, "post msg falied", K(ret), K(msg), K(p)); last_ret = ret; } } else { ++post_succ_num; } msg.tx_ptr_ = msg_tx_ptr; } return last_ret; } } namespace unittest { #define EXE_SQL(sql_str) \ ASSERT_EQ(OB_SUCCESS, sql.assign(sql_str)); \ ASSERT_EQ(OB_SUCCESS, sql_proxy.write(sql.ptr(), affected_rows)); #define EXE_SQL_FMT(...) \ ASSERT_EQ(OB_SUCCESS, sql.assign_fmt(__VA_ARGS__)); \ ASSERT_EQ(OB_SUCCESS, sql_proxy.write(sql.ptr(), affected_rows)); #define WRITE_SQL_BY_CONN(conn, sql_str) \ ASSERT_EQ(OB_SUCCESS, sql.assign(sql_str)); \ ASSERT_EQ(OB_SUCCESS, conn->execute_write(OB_SYS_TENANT_ID, sql.ptr(), affected_rows)); #define WRITE_SQL_FMT_BY_CONN(conn, ...) \ ASSERT_EQ(OB_SUCCESS, sql.assign_fmt(__VA_ARGS__)); \ ASSERT_EQ(OB_SUCCESS, conn->execute_write(OB_SYS_TENANT_ID, sql.ptr(), affected_rows)); #define READ_SQL_BY_CONN(conn, sql_str) \ ASSERT_EQ(OB_SUCCESS, sql.assign(sql_str)); \ ASSERT_EQ(OB_SUCCESS, conn->execute_read(OB_SYS_TENANT_ID, sql.ptr(), read_res)); class ObTransferBetweenRollbackTo : public ObSimpleClusterTestBase { public: ObTransferBetweenRollbackTo(): ObSimpleClusterTestBase("test_transfer_between_rollback_to", "200G", "40G") {} void prepare_tenant_env() { common::ObMySQLProxy &sql_proxy = get_curr_simple_server().get_sql_proxy2(); int64_t affected_rows = 0; ObSqlString sql; sqlclient::ObISQLConnection *connection = nullptr; ASSERT_EQ(OB_SUCCESS, sql_proxy.acquire(connection)); ASSERT_NE(nullptr, connection); WRITE_SQL_BY_CONN(connection, "set GLOBAL ob_trx_timeout = 10000000000"); WRITE_SQL_BY_CONN(connection, "set GLOBAL ob_trx_idle_timeout = 10000000000"); WRITE_SQL_BY_CONN(connection, "set GLOBAL ob_query_timeout = 10000000000"); WRITE_SQL_BY_CONN(connection, "alter system set enable_early_lock_release = False;"); WRITE_SQL_BY_CONN(connection, "alter system set undo_retention = 1800;"); sleep(5); } void create_test_tenant(uint64_t &tenant_id) { TRANS_LOG(INFO, "create_tenant start"); ASSERT_EQ(OB_SUCCESS, create_tenant("tt1", "20G", "100G")); fprintf(stdout, "finish sleep\n"); ASSERT_EQ(OB_SUCCESS, get_tenant_id(tenant_id)); ASSERT_EQ(OB_SUCCESS, get_curr_simple_server().init_sql_proxy2()); TRANS_LOG(INFO, "create_tenant end", K(tenant_id)); } int wait_balance_clean(uint64_t tenant_id) { int ret = OB_SUCCESS; while (OB_SUCC(ret)) { bool is_clean = false; MTL_SWITCH(tenant_id) { ObBalanceJob job; int64_t start_time = OB_INVALID_TIMESTAMP, finish_time = OB_INVALID_TIMESTAMP; if (OB_FAIL(ObBalanceJobTableOperator::get_balance_job(tenant_id, false, *GCTX.sql_proxy_, job, start_time, finish_time))) { if (OB_ENTRY_NOT_EXIST == ret) { ret = OB_SUCCESS; is_clean = true; } } else { ob_usleep(200 * 1000); } } if (is_clean) { int64_t transfer_task_count = 0; if (OB_FAIL(SSH::g_select_int64(tenant_id, "select count(*) as val from __all_transfer_task", transfer_task_count))) { } else if (transfer_task_count == 0) { break; } else { ob_usleep(200 * 1000); } } } return ret; } void get_tablet_info_with_table_name(const char *name, int64_t &table_id, int64_t &object_id, int64_t &tablet_id, int64_t &ls_id) { common::ObMySQLProxy &sql_proxy = get_curr_simple_server().get_sql_proxy2(); int ret = OB_SUCCESS; ObSqlString sql; int64_t affected_rows = 0; ASSERT_EQ(OB_SUCCESS, sql.assign_fmt("SELECT table_id, object_id, tablet_id, ls_id FROM oceanbase.DBA_OB_TABLE_LOCATIONS WHERE TABLE_NAME= '%s';", name)); SMART_VAR(ObMySQLProxy::MySQLResult, res) { ASSERT_EQ(OB_SUCCESS, sql_proxy.read(res, sql.ptr())); sqlclient::ObMySQLResult *result = res.get_result(); ASSERT_NE(nullptr, result); ASSERT_EQ(OB_SUCCESS, result->next()); ASSERT_EQ(OB_SUCCESS, result->get_int("table_id", table_id)); ASSERT_EQ(OB_SUCCESS, result->get_int("object_id", object_id)); ASSERT_EQ(OB_SUCCESS, result->get_int("tablet_id", tablet_id)); ASSERT_EQ(OB_SUCCESS, result->get_int("ls_id", ls_id)); } } int do_balance_inner_(uint64_t tenant_id) { int ret = OB_SUCCESS; static std::mutex mutex; mutex.lock(); MTL_SWITCH(tenant_id) { TRANS_LOG(INFO, "worker to do partition_balance"); auto b_svr = MTL(rootserver::ObTenantBalanceService*); b_svr->stop(); b_svr->reset(); int64_t job_cnt = 0; int64_t start_time = OB_INVALID_TIMESTAMP, finish_time = OB_INVALID_TIMESTAMP; ObBalanceJob job; if (OB_FAIL(b_svr->gather_stat_())) { TRANS_LOG(WARN, "failed to gather stat", KR(ret)); } else if (OB_FAIL(b_svr->gather_ls_status_stat(tenant_id, b_svr->ls_array_))) { TRANS_LOG(WARN, "failed to gather ls stat", KR(ret)); } else if (OB_FAIL(ObBalanceJobTableOperator::get_balance_job( tenant_id, false, *GCTX.sql_proxy_, job, start_time, finish_time))) { if (OB_ENTRY_NOT_EXIST == ret) { //NO JOB, need check current ls status ret = OB_SUCCESS; job_cnt = 0; } else { TRANS_LOG(WARN, "failed to get balance job", KR(ret), K(tenant_id)); } } else if (OB_FAIL(b_svr->try_finish_current_job_(job, job_cnt))) { TRANS_LOG(WARN, "failed to finish current job", KR(ret), K(job)); } if (OB_SUCC(ret) && job_cnt == 0 && OB_FAIL(b_svr->partition_balance_(true))) { TRANS_LOG(WARN, "failed to do partition balance", KR(ret)); } } mutex.unlock(); return ret; } int do_balance(uint64_t tenant_id) { int ret = OB_SUCCESS; if (OB_FAIL(do_balance_inner_(tenant_id))) { } else if (OB_FAIL(do_balance_inner_(tenant_id))) { } return ret; } ObLS *get_ls(const int64_t tenant_id, const ObLSID ls_id) { int ret = OB_SUCCESS; ObLS *ls = nullptr; MTL_SWITCH(tenant_id) { ObLSHandle ls_handle; ObLSService *ls_svr = MTL(ObLSService *); OB_ASSERT(OB_SUCCESS == ls_svr->get_ls(ls_id, ls_handle, ObLSGetMod::STORAGE_MOD)); OB_ASSERT(nullptr != (ls = ls_handle.get_ls())); } return ls; } }; TEST_F(ObTransferBetweenRollbackTo, transfer_between_rollback_to) { ObSqlString sql; int64_t affected_rows = 0; global_ls_id.reset(); global_tx_id.reset(); // ============================== Phase1. create tenant ============================== TRANS_LOG(INFO, "create tenant start"); uint64_t tenant_id = 0; create_test_tenant(tenant_id); TRANS_LOG(INFO, "create tenant end"); share::ObTenantSwitchGuard tenant_guard; ASSERT_EQ(OB_SUCCESS, tenant_guard.switch_to(tenant_id)); prepare_tenant_env(); // ============================== Phase2. create new ls ============================== ASSERT_EQ(0, SSH::create_ls(tenant_id, get_curr_observer().self_addr_)); int64_t ls_count = 0; ASSERT_EQ(0, SSH::g_select_int64(tenant_id, "select count(ls_id) as val from __all_ls where ls_id!=1", ls_count)); ASSERT_EQ(2, ls_count); // ============================== Phase3. create new tables ============================== common::ObMySQLProxy &sql_proxy = get_curr_simple_server().get_sql_proxy2(); TRANS_LOG(INFO, "create table qcc1 start"); EXE_SQL("create table qcc1 (a int)"); TRANS_LOG(INFO, "create_table qcc1 end"); TRANS_LOG(INFO, "create table qcc2 start"); EXE_SQL("create table qcc2 (a int)"); TRANS_LOG(INFO, "create_table qcc2 end"); usleep(3 * 1000 * 1000); ObLSID loc1, loc2; ASSERT_EQ(0, SSH::select_table_loc(tenant_id, "qcc1", loc1)); ASSERT_EQ(0, SSH::select_table_loc(tenant_id, "qcc2", loc2)); ASSERT_NE(loc1, loc2); int64_t table1, table2; int64_t object1, object2; int64_t tablet1, tablet2; int64_t ls1, ls2; get_tablet_info_with_table_name("qcc1", table1, object1, tablet1, ls1); get_tablet_info_with_table_name("qcc2", table2, object2, tablet2, ls2); fprintf(stdout, "qcc is created successfully, loc1: %ld, loc2: %ld, table1: %ld, table2: %ld, tablet1: %ld, tablet2: %ld, ls1: %ld, ls2: %ld\n", loc1.id(), loc2.id(), table1, table2, tablet1, tablet2, ls1, ls2); EXE_SQL("create tablegroup tg1 sharding='NONE';"); // ============================== Phase4. wait minor freeze to remove retain ctx ============================== sqlclient::ObISQLConnection *sys_conn = nullptr; common::ObMySQLProxy &sys_proxy = get_curr_simple_server().get_sql_proxy(); ASSERT_EQ(OB_SUCCESS, sys_proxy.acquire(sys_conn)); ASSERT_NE(nullptr, sys_conn); WRITE_SQL_BY_CONN(sys_conn, "alter system minor freeze tenant sys;"); WRITE_SQL_BY_CONN(sys_conn, "alter system minor freeze tenant all_user;"); WRITE_SQL_BY_CONN(sys_conn, "alter system minor freeze tenant all_meta;"); sleep(5); int ret = OB_SUCCESS; HEAP_VAR(ObMySQLProxy::MySQLResult, res_0) { common::sqlclient::ObMySQLResult *result = nullptr; sql.assign( "SELECT count(*) as cnt FROM oceanbase.__all_virtual_trans_stat where tenant_id = 1002 and ls_id = 1001;"); int retry_times = 100; int64_t cnt = 0; while (--retry_times >= 0) { res_0.reuse(); ASSERT_EQ(OB_SUCCESS, sys_conn->execute_read(OB_SYS_TENANT_ID, sql.ptr(), res_0)); result = res_0.mysql_result(); ASSERT_EQ(OB_SUCCESS, result->next()); ASSERT_EQ(OB_SUCCESS, result->get_int("cnt", cnt)); if (0 == cnt) { break; } else { fprintf(stdout, "waitting for tx ctx table mini merge to clear retain ctx ... \n"); sleep(1); } } ASSERT_EQ(0, cnt); } // ============================== Phase5. start the user txn ============================== sqlclient::ObISQLConnection *user_connection = nullptr; ASSERT_EQ(OB_SUCCESS, sql_proxy.acquire(user_connection)); ASSERT_NE(nullptr, user_connection); WRITE_SQL_BY_CONN(user_connection, "set SESSION ob_trx_timeout = 10000000000"); WRITE_SQL_BY_CONN(user_connection, "set SESSION ob_trx_idle_timeout = 10000000000"); WRITE_SQL_BY_CONN(user_connection, "set SESSION ob_query_timeout = 10000000000"); TRANS_LOG(INFO, "start the txn"); WRITE_SQL_BY_CONN(user_connection, "begin;"); WRITE_SQL_FMT_BY_CONN(user_connection, "savepoint qcqc;"); WRITE_SQL_FMT_BY_CONN(user_connection, "insert into qcc2 values(1);"); WRITE_SQL_FMT_BY_CONN(user_connection, "insert into qcc1 values(1);"); // remember the frozen point ASSERT_EQ(0, SSH::find_tx(user_connection, global_tx_id)); global_ls_id = ls2; fprintf(stdout, "txn is created successfully, tx_id: %ld\n", global_tx_id.get_id()); std::thread th([user_connection] () { ObSqlString sql; int64_t affected_rows = 0; fprintf(stdout, "start to rollback to savepoint, tx_id: %ld\n", global_tx_id.get_id()); WRITE_SQL_FMT_BY_CONN(user_connection, "rollback to savepoint qcqc;"); fprintf(stdout, "rollback to savepoint successfully, tx_id: %ld\n", global_tx_id.get_id()); }); // ============================== Phase5. start the transfer ============================== EXE_SQL("alter tablegroup tg1 add qcc1,qcc2;"); usleep(1 * 1000 * 1000); ASSERT_EQ(0, do_balance(tenant_id)); int64_t begin_time = ObTimeUtility::current_time(); while (true) { ASSERT_EQ(0, SSH::select_table_loc(tenant_id, "qcc1", loc1)); ASSERT_EQ(0, SSH::select_table_loc(tenant_id, "qcc2", loc2)); if (loc1 == loc2) { fprintf(stdout, "succeed wait for balancer\n"); break; } else if (ObTimeUtility::current_time() - begin_time > 300 * 1000 * 1000) { fprintf(stdout, "ERROR: fail to wait for balancer\n"); break; } else { usleep(1 * 1000 * 1000); fprintf(stdout, "wait for balancer\n"); } } ASSERT_EQ(loc1, loc2); while (true) { int64_t transfer_task_count = 0; if (OB_FAIL(SSH::g_select_int64(tenant_id, "select count(*) as val from __all_transfer_task", transfer_task_count))) { TRANS_LOG(WARN, "fail to wait for transfer task"); } else if (transfer_task_count == 0) { fprintf(stdout, "succeed wait for balancer v2\n"); break; } else if (ObTimeUtility::current_time() - begin_time > 300 * 1000 * 1000) { fprintf(stdout, "ERROR: fail to wait for balancer\n"); break; } else { fprintf(stdout, "wait for balancer v2\n"); ob_usleep(200 * 1000); } } usleep(1000 * 1000); global_ls_id.reset(); th.join(); ObISQLClient::ReadResult read_res; READ_SQL_BY_CONN(user_connection, "select * from qcc2;"); sqlclient::ObMySQLResult *result = read_res.get_result(); int i = 0; while (OB_SUCC(result->next())) { i++; int64_t a; ASSERT_EQ(OB_SUCCESS, result->get_int("a", a)); ASSERT_EQ(a, 1); } EXPECT_EQ(0, i); } } // namespace unittest } // namespace oceanbase int main(int argc, char **argv) { using namespace oceanbase::unittest; oceanbase::unittest::init_log_and_gtest(argc, argv); OB_LOGGER.set_log_level("info"); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); }