hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[FEAT MERGE] Improve the inner implementation of tx table

Browse Source 
Co-authored-by: kongfy <njukongfy@gmail.com>
This commit is contained in:
ZenoWang
2023-05-07 07:38:34 +00:00
committed by ob-robot
parent 1a4be45b79
commit ccde264e06
92 changed files with 4834 additions and 1542 deletions
Download Patch File Download Diff File Expand all files Collapse all files

457
mittest/simple_server/fast_commit_report.cpp
View File

@ -17,15 +17,58 @@
#define private public
#include "env/ob_simple_cluster_test_base.h"
#include "storage/compaction/ob_compaction_diagnose.h"
#include "storage/compaction/ob_schedule_dag_func.h"
#include "storage/ls/ob_ls.h"
#include "storage/tx_storage/ob_ls_handle.h"
#include "storage/tx_storage/ob_ls_service.h"
#include "storage/tx/ob_tx_data_functor.h"
#include "storage/tablet/ob_tablet.h"
enum EnumTestMode : uint32_t
{
NORMAL_TEST = 0,
BIANQUE_TEST = 1,
MAX_TEST_MODE
};
enum EnumDataMode : uint32_t
{
ALL_CLEANOUT = 0,
ALL_SAME_TXN_DELAY_CLEANOUT = 1,
ALL_DIFF_TXN_DELAY_CLEANOUT = 2,
MAX_DATA_MODE
};
enum EnumDataLocation : uint32_t
{
DATA_IN_MEMORY = 0,
DATA_ON_DISK = 1,
MAX_DATA_LOCATION
};
enum EnumTxTableMode : uint32_t
{
ALL_IN_MEMORY = 0,
ALL_IN_DISK = 1,
MAX_TX_TABLE_MODE
};
static const char *TEST_FILE_NAME = "fast_commit_report";
const int64_t TOTAL_FC_ROW_COUNT = 1000000;
const int64_t TOTAL_FC_SESSION = 10;
const int64_t PAINTING_FC_ROW_COUNT = 10000;
const int64_t DEFAULT_TOTAL_FC_ROW_COUNT = 100000;
const int64_t DEFAULT_TOTAL_FC_SESSION = 10;
const int64_t DEFAULT_PAINTING_FC_ROW_COUNT = 10000;
EnumTestMode TEST_MODE = EnumTestMode::NORMAL_TEST;
EnumDataMode DATA_MODE = EnumDataMode::ALL_CLEANOUT;
EnumDataLocation DATA_LOCATION = EnumDataLocation::DATA_IN_MEMORY;
EnumTxTableMode TX_TABLE_MODE = EnumTxTableMode::ALL_IN_MEMORY;
int64_t TOTAL_FC_ROW_COUNT = DEFAULT_TOTAL_FC_ROW_COUNT;
int64_t TOTAL_FC_SESSION = DEFAULT_TOTAL_FC_SESSION;
int64_t TOTAL_READ_TIME = 0;
int64_t READ_CNT = 0;
int64_t SELECT_PARALLEL= 1;
int64_t SLEEP_SECONDS = 0;
namespace oceanbase
{
@ -45,6 +88,8 @@ int ObMvccValueIterator::get_next_node(const void *&tnode)
bool is_lock_node = false;
if (NULL == version_iter_) {
ret = OB_ITER_END;
} else if (EnumDataMode::ALL_CLEANOUT == DATA_MODE && OB_FAIL(try_cleanout_tx_node_(version_iter_))) {
TRANS_LOG(WARN, "fail to cleanout tnode", K(ret), K(*version_iter_));
} else if (OB_FAIL(version_iter_->is_lock_node(is_lock_node))) {
TRANS_LOG(WARN, "fail to check is lock node", K(ret), K(*version_iter_));
} else if (!(version_iter_->is_aborted() // skip abort version
@ -61,6 +106,50 @@ int ObMvccValueIterator::get_next_node(const void *&tnode)
return ret;
}
int ObMemtable::flush(share::ObLSID ls_id)
{
if (EnumDataLocation::DATA_IN_MEMORY == DATA_LOCATION) {
return OB_EAGAIN;
}
int ret = OB_SUCCESS;
int64_t cur_time = ObTimeUtility::current_time();
if (is_flushed_) {
ret = OB_NO_NEED_UPDATE;
} else {
if (mt_stat_.create_flush_dag_time_ == 0 &&
mt_stat_.ready_for_flush_time_ != 0 &&
cur_time - mt_stat_.ready_for_flush_time_ > 30 * 1000 * 1000) {
STORAGE_LOG(WARN, "memtable can not create dag successfully for long time",
K(ls_id), K(*this), K(mt_stat_.ready_for_flush_time_));
compaction::ADD_SUSPECT_INFO(MINI_MERGE,
ls_id, get_tablet_id(),
"memtable can not create dag successfully",
"has been ready for flush time:",
cur_time - mt_stat_.ready_for_flush_time_,
"ready for flush time:",
mt_stat_.ready_for_flush_time_);
}
compaction::ObTabletMergeDagParam param;
param.ls_id_ = ls_id;
param.tablet_id_ = key_.tablet_id_;
param.merge_type_ = MINI_MERGE;
param.merge_version_ = ObVersion::MIN_VERSION;
if (OB_FAIL(compaction::ObScheduleDagFunc::schedule_tablet_merge_dag(param))) {
if (OB_EAGAIN != ret && OB_SIZE_OVERFLOW != ret) {
TRANS_LOG(WARN, "failed to schedule tablet merge dag", K(ret));
}
} else {
mt_stat_.create_flush_dag_time_ = cur_time;
TRANS_LOG(INFO, "schedule tablet merge dag successfully", K(ret), K(param), KPC(this));
}
}
return ret;
}
}
namespace storage
@ -104,37 +193,11 @@ int LockForReadFunctor::operator()(const ObTxData &tx_data, ObTxCCCtx *tx_cc_ctx
namespace unittest
{
using namespace oceanbase::transaction;
using namespace oceanbase::storage;
using namespace oceanbase::memtable;
using namespace oceanbase::storage::checkpoint;
enum FastCommitTestMode : int
{
NORMAL_TEST = 0,
BIANQUE_TEST = 1
};
enum FastCommitReportDataMode : int
{
ALL_CLEANOUT = 0,
ALL_SAME_TXN_DELAY_CLEANOUT = 1,
ALL_DIFF_TXN_DELAY_CLEANOUT = 2
};
enum FastCommitReportTxTableMode : int
{
ALL_IN_MEMORY = 0,
ALL_IN_DISK = 1
};
FastCommitTestMode fast_commit_test_mode = FastCommitTestMode::NORMAL_TEST;
FastCommitReportDataMode fast_commit_data_mode = FastCommitReportDataMode::ALL_CLEANOUT;
FastCommitReportTxTableMode fast_commit_tx_table_mode = FastCommitReportTxTableMode::ALL_IN_MEMORY;
int64_t total_fc_row_count = TOTAL_FC_ROW_COUNT;
int64_t total_fc_session = TOTAL_FC_SESSION;
#define EXE_SQL(sql_str) \
ASSERT_EQ(OB_SUCCESS, sql.assign(sql_str)); \
ASSERT_EQ(OB_SUCCESS, sql_proxy.write(sql.ptr(), affected_rows));
@ -170,6 +233,10 @@ public:
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;");
WRITE_SQL_BY_CONN(connection, "alter system set enable_perf_event = true;");
sleep(5);
}
void set_private_buffer_size(const char* size)
@ -188,23 +255,6 @@ public:
EXE_SQL_FMT("alter system set _fast_commit_callback_count = %ld;", count);
}
void set_memstore_limit()
{
common::ObMySQLProxy &sql_proxy = get_curr_simple_server().get_sql_proxy();
int64_t affected_rows = 0;
ObSqlString sql;
EXE_SQL("alter system set memstore_limit_percentage = 80;");
}
void set_freeze_trigger()
{
common::ObMySQLProxy &sql_proxy = get_curr_simple_server().get_sql_proxy2();
int64_t affected_rows = 0;
ObSqlString sql;
EXE_SQL("alter system set writing_throttling_trigger_percentage = 100;");
EXE_SQL("alter system set freeze_trigger_percentage = 80;");
}
void create_test_tenant(uint64_t &tenant_id)
{
TRANS_LOG(INFO, "create_tenant start");
@ -237,6 +287,46 @@ public:
tablet = (uint64_t)tablet_id;
}
void minor_freeze_data_and_wait()
{
common::ObMySQLProxy &sql_proxy = get_curr_simple_server().get_sql_proxy();
sqlclient::ObISQLConnection *connection = nullptr;
ASSERT_EQ(OB_SUCCESS, sql_proxy.acquire(connection));
int ret = OB_SUCCESS;
ObSqlString sql;
int64_t affected_rows = 0;
WRITE_SQL_BY_CONN(connection, "alter system minor freeze tenant all;");
fprintf(stdout, "start flush user data\n");
sleep(10);
int retry_times = 0;
bool freeze_success = false;
HEAP_VAR(ObMySQLProxy::MySQLResult, res)
{
int64_t cnt = -1;
while (++retry_times <= 600) {
ASSERT_EQ(OB_SUCCESS,
connection->execute_read(OB_SYS_TENANT_ID,
"select count(*) as cnt from oceanbase.__all_virtual_table_mgr where "
"table_type=0 and is_active like '%NO%';",
res));
common::sqlclient::ObMySQLResult *result = res.mysql_result();
ASSERT_EQ(OB_SUCCESS, result->next());
ASSERT_EQ(OB_SUCCESS, result->get_int("cnt", cnt));
if (0 == cnt) {
freeze_success = true;
break;
}
}
fprintf(stdout, "waitting for data minor merge. retry times = %d\n", retry_times);
sleep(1);
}
ASSERT_EQ(true, freeze_success);
}
void check_no_minor_freeze()
{
TRANS_LOG(INFO, "check no minor freeze start");
@ -287,8 +377,10 @@ public:
ASSERT_EQ(OB_SUCCESS, tablet->get_active_memtable(handle));
}
void insert_data_single(const int row_count = oceanbase::unittest::total_fc_row_count)
void insert_data_single(const int row_count)
{
fprintf(stdout, "start insert data, data count : %d \n", row_count);
ObSqlString sql;
int64_t affected_rows = 0;
@ -304,45 +396,58 @@ public:
TRANS_LOG(INFO, "insert data start");
const int64_t begin_time = ObTimeUtility::current_time();
if (FastCommitReportDataMode::ALL_CLEANOUT == oceanbase::unittest::fast_commit_data_mode) {
if (EnumDataMode::ALL_CLEANOUT == DATA_MODE) {
WRITE_SQL_BY_CONN(connection, "begin;");
} else if (FastCommitReportDataMode::ALL_SAME_TXN_DELAY_CLEANOUT == oceanbase::unittest::fast_commit_data_mode) {
} else if (EnumDataMode::ALL_SAME_TXN_DELAY_CLEANOUT == DATA_MODE) {
WRITE_SQL_BY_CONN(connection, "begin;");
} else if (FastCommitReportDataMode::ALL_DIFF_TXN_DELAY_CLEANOUT == oceanbase::unittest::fast_commit_data_mode) {
} else if (EnumDataMode::ALL_DIFF_TXN_DELAY_CLEANOUT == DATA_MODE) {
// pass
}
for (int i = 0; i < row_count; i++) {
if (FastCommitReportDataMode::ALL_DIFF_TXN_DELAY_CLEANOUT == oceanbase::unittest::fast_commit_data_mode) {
if (EnumDataMode::ALL_DIFF_TXN_DELAY_CLEANOUT == DATA_MODE) {
WRITE_SQL_BY_CONN(connection, "begin;");
}
// const int64_t single_begin_time = ObTimeUtility::current_time();
WRITE_SQL_FMT_BY_CONN(connection, "insert into test_fast_commit values(1);");
if (FastCommitReportDataMode::ALL_DIFF_TXN_DELAY_CLEANOUT == oceanbase::unittest::fast_commit_data_mode) {
if (EnumDataMode::ALL_DIFF_TXN_DELAY_CLEANOUT == DATA_MODE) {
WRITE_SQL_BY_CONN(connection, "commit;");
}
// const int64_t single_end_time = ObTimeUtility::current_time();
// TRANS_LOG(INFO, "single insert data single cost", K(single_end_time - single_begin_time));
if ((i + 1) % PAINTING_FC_ROW_COUNT == 0) {
if ((i + 1) % DEFAULT_PAINTING_FC_ROW_COUNT == 0) {
TRANS_LOG(INFO, "insert data single pass one round",
K(oceanbase::unittest::fast_commit_data_mode),
K(oceanbase::unittest::fast_commit_tx_table_mode),
K(DATA_MODE),
K(TX_TABLE_MODE),
K(i + 1));
}
}
if (FastCommitReportDataMode::ALL_CLEANOUT == oceanbase::unittest::fast_commit_data_mode) {
if (EnumDataLocation::DATA_ON_DISK == DATA_LOCATION) {
fprintf(stdout, "data location is DATA_ON_DISK\n");
if (EnumDataMode::ALL_CLEANOUT == DATA_MODE) {
WRITE_SQL_BY_CONN(connection, "commit;");
sleep(10);
minor_freeze_data_and_wait();
} else {
minor_freeze_data_and_wait();
WRITE_SQL_BY_CONN(connection, "commit;");
}
} else if (EnumDataMode::ALL_CLEANOUT == DATA_MODE) {
WRITE_SQL_BY_CONN(connection, "commit;");
} else if (FastCommitReportDataMode::ALL_SAME_TXN_DELAY_CLEANOUT == oceanbase::unittest::fast_commit_data_mode) {
} else if (EnumDataMode::ALL_SAME_TXN_DELAY_CLEANOUT == DATA_MODE) {
WRITE_SQL_BY_CONN(connection, "commit;");
} else if (FastCommitReportDataMode::ALL_DIFF_TXN_DELAY_CLEANOUT == oceanbase::unittest::fast_commit_data_mode) {
} else if (EnumDataMode::ALL_DIFF_TXN_DELAY_CLEANOUT == DATA_MODE) {
// pass
}
const int64_t end_time = ObTimeUtility::current_time();
TRANS_LOG(INFO, "insert data single cost", K(end_time - begin_time), K(begin_time), K(end_time));
}
void insert_data_parallel(const int parrallel_num = oceanbase::unittest::total_fc_session,
const int row_count = oceanbase::unittest::total_fc_row_count)
void insert_data_parallel(const int parrallel_num,
const int row_count)
{
int single_row_count = row_count / parrallel_num;
std::thread *threads[parrallel_num];
@ -359,7 +464,55 @@ public:
TRANS_LOG(INFO, "insert data parallel cost", K(end_time - begin_time), K(begin_time), K(end_time));
}
void read_data(const int row_count = oceanbase::unittest::total_fc_row_count)
void insert_data_pdml(const int row_count)
{
ObSqlString sql;
int64_t affected_rows = 0;
int64_t real_inserted_rows = 0;
common::ObMySQLProxy &sql_proxy = get_curr_simple_server().get_sql_proxy2();
sqlclient::ObISQLConnection *connection = nullptr;
ASSERT_EQ(OB_SUCCESS, sql_proxy.acquire(connection));
ASSERT_NE(nullptr, connection);
WRITE_SQL_BY_CONN(connection, "set SESSION ob_trx_timeout = 10000000000");
WRITE_SQL_BY_CONN(connection, "set SESSION ob_trx_idle_timeout = 10000000000");
WRITE_SQL_BY_CONN(connection, "set SESSION ob_query_timeout = 10000000000");
TRANS_LOG(INFO, "pdml insert data start");
const int64_t begin_time = ObTimeUtility::current_time();
WRITE_SQL_BY_CONN(connection, "begin;");
WRITE_SQL_FMT_BY_CONN(connection, "insert into test_fast_commit values(1);");
real_inserted_rows = 1;
while (real_inserted_rows < row_count) {
WRITE_SQL_FMT_BY_CONN(connection, "insert /*+ enable_parallel_dml PARALLEL(32) */ into test_fast_commit select/*+ PARALLEL(32) */ * from test_fast_commit;");
real_inserted_rows <<= 1;
}
fprintf(stdout, "real inserted rows : %ld \n", real_inserted_rows);
if (EnumDataLocation::DATA_ON_DISK == DATA_LOCATION) {
fprintf(stdout, "data location is DATA_ON_DISK\n");
if (EnumDataMode::ALL_CLEANOUT == DATA_MODE) {
WRITE_SQL_BY_CONN(connection, "commit;");
sleep(10);
minor_freeze_data_and_wait();
} else {
minor_freeze_data_and_wait();
WRITE_SQL_BY_CONN(connection, "commit;");
}
} else {
WRITE_SQL_BY_CONN(connection, "commit;");
}
const int64_t end_time = ObTimeUtility::current_time();
TRANS_LOG(INFO, "insert data single cost", K(end_time - begin_time), K(begin_time), K(end_time));
}
void read_data(const int row_count)
{
ObSqlString sql;
int64_t affected_rows = 0;
@ -381,19 +534,25 @@ public:
ObTxTable *tx_table = ls->get_tx_table();
const int64_t begin_time = ObTimeUtility::current_time();
READ_SQL_BY_CONN(connection, "select count(*) as cnt from test_fast_commit");
sql.assign_fmt("SELECT/*+ enable_parallel_dml PARALLEL(%ld) */ * FROM test_fast_commit;", SELECT_PARALLEL);
TRANS_LOG(INFO, "do select sql", K(sql));
ASSERT_EQ(OB_SUCCESS, connection->execute_read(OB_SYS_TENANT_ID, sql.ptr(), read_res));
const int64_t end_time = ObTimeUtility::current_time();
TRANS_LOG(INFO, "read data cost", K(end_time - begin_time), K(begin_time), K(end_time));
std::cout << "read data cost(total_cost=" << (end_time - begin_time)
<< ", begin_time=" << begin_time
<< ", end_time=" << end_time << ")\n";
if (REACH_TIME_INTERVAL(5 * 1000 * 1000)) {
std::cout << "read data cost(total_cost=" << (end_time - begin_time)/1000 << "ms)\n";
}
READ_CNT++;
if (READ_CNT > 100) {
TOTAL_READ_TIME += (end_time - begin_time);
}
sqlclient::ObMySQLResult *result = read_res.get_result();
ASSERT_NE(nullptr, result);
ASSERT_EQ(OB_SUCCESS, result->next());
ASSERT_EQ(OB_SUCCESS, result->get_int("cnt", cnt));
ASSERT_EQ(cnt, row_count);
}
void check_memtable_cleanout(const bool memtable_is_all_cleanout,
@ -509,65 +668,72 @@ TEST_F(ObFastCommitReport, fast_commit_report)
TRANS_LOG(INFO, "create_table end");
prepare_tenant_env();
set_freeze_trigger();
// set_freeze_trigger();
set_private_buffer_size("2M");
if (FastCommitReportDataMode::ALL_CLEANOUT == oceanbase::unittest::fast_commit_data_mode) {
set_fast_commit_count(oceanbase::unittest::total_fc_row_count + 100);
} else if (FastCommitReportDataMode::ALL_SAME_TXN_DELAY_CLEANOUT == oceanbase::unittest::fast_commit_data_mode) {
if (EnumDataMode::ALL_CLEANOUT == DATA_MODE) {
set_fast_commit_count(TOTAL_FC_ROW_COUNT + 100);
} else if (EnumDataMode::ALL_SAME_TXN_DELAY_CLEANOUT == DATA_MODE) {
set_fast_commit_count(0);
} else if (FastCommitReportDataMode::ALL_DIFF_TXN_DELAY_CLEANOUT == oceanbase::unittest::fast_commit_data_mode) {
} else if (EnumDataMode::ALL_DIFF_TXN_DELAY_CLEANOUT == DATA_MODE) {
set_fast_commit_count(0);
}
if (FastCommitReportDataMode::ALL_CLEANOUT == oceanbase::unittest::fast_commit_data_mode) {
insert_data_parallel();
} else if (FastCommitReportDataMode::ALL_SAME_TXN_DELAY_CLEANOUT == oceanbase::unittest::fast_commit_data_mode) {
insert_data_single();
} else if (FastCommitReportDataMode::ALL_DIFF_TXN_DELAY_CLEANOUT == oceanbase::unittest::fast_commit_data_mode) {
insert_data_parallel();
if (EnumDataMode::ALL_CLEANOUT == DATA_MODE) {
// insert_data_single(TOTAL_FC_ROW_COUNT);
insert_data_pdml(TOTAL_FC_ROW_COUNT);
} else if (EnumDataMode::ALL_SAME_TXN_DELAY_CLEANOUT == DATA_MODE) {
// insert_data_single(TOTAL_FC_ROW_COUNT);
insert_data_pdml(TOTAL_FC_ROW_COUNT);
} else if (EnumDataMode::ALL_DIFF_TXN_DELAY_CLEANOUT == DATA_MODE) {
insert_data_parallel(TOTAL_FC_SESSION, TOTAL_FC_ROW_COUNT);
}
check_no_minor_freeze();
if (FastCommitReportDataMode::ALL_CLEANOUT == oceanbase::unittest::fast_commit_data_mode) {
check_memtable_cleanout(true, /*is_all_cleanout*/
false, /*is_all_delay_cleanout*/
oceanbase::unittest::total_fc_row_count);
} else {
check_memtable_cleanout(false, /*is_all_cleanout*/
true, /*is_all_delay_cleanout*/
oceanbase::unittest::total_fc_row_count);
}
if (FastCommitReportTxTableMode::ALL_IN_DISK == oceanbase::unittest::fast_commit_tx_table_mode) {
if (EnumTxTableMode::ALL_IN_DISK == TX_TABLE_MODE) {
minor_freeze_tx_data_memtable();
}
TRANS_LOG(INFO, "fast commit report with arguments",
K(oceanbase::unittest::fast_commit_data_mode),
K(oceanbase::unittest::fast_commit_tx_table_mode));
K(DATA_MODE),
K(TX_TABLE_MODE));
if (oceanbase::unittest::FastCommitTestMode::NORMAL_TEST == oceanbase::unittest::fast_commit_test_mode) {
// check with 3 times
read_data();
read_data();
read_data();
} else if (oceanbase::unittest::FastCommitTestMode::BIANQUE_TEST == oceanbase::unittest::fast_commit_test_mode) {
if (EnumTestMode::BIANQUE_TEST == TEST_MODE) {
std::cout << "Master, you should start enterring a number as you like\n";
int n;
std::cin >> n;
std::cout << "Your faithful servant, qianchen is start reading data within 60 seconds\n";
std::cout << "So prepare for command: sudo perf record -e cycles -c 100000000 -p $(pidof -s test_fast_commit_report) -g -- sleep 30\n";
int64_t base_ts = OB_TSC_TIMESTAMP.current_time();
while (OB_TSC_TIMESTAMP.current_time() - base_ts <= 60 * 1000 * 1000) {
read_data();
}
std::cout << "Data reading has been finished, you can generate it with command: sudo perf script -i perf.data -F ip,sym -f > data.viz\n";
std::cout << "And finally you can graph it with command: `cat data.viz | ./perfdata2graph.py` or our web based perf tool.\n";
}
while (READ_CNT < 400) {
read_data(TOTAL_FC_ROW_COUNT);
}
std::cout << "Data reading has been finished, you can generate it with command: sudo perf script -i perf.data -F "
"ip,sym -f > data.viz\n";
std::cout << "And finally you can graph it with command: `cat data.viz | ./perfdata2graph.py` or our web based perf "
"tool.\n";
fprintf(stdout, "average spend time : %ld ms, total read count : %ld\n", TOTAL_READ_TIME / (READ_CNT-100) / 1000, READ_CNT);
fprintf(stdout,
"argument : "
" test_mode = %d, "
" data_mode = %d, "
" data_location = %d, "
" tx_table_mode = %d, "
" row_cnt = %ld, "
" session_cnt = %ld"
" pdml_select_parallel = %ld\n",
TEST_MODE,
DATA_MODE,
DATA_LOCATION,
TX_TABLE_MODE,
TOTAL_FC_ROW_COUNT,
TOTAL_FC_SESSION,
SELECT_PARALLEL);
sleep(SLEEP_SECONDS);
}
@ -576,77 +742,74 @@ TEST_F(ObFastCommitReport, fast_commit_report)
void tutorial()
{
std::cout << "./mittest/simple_server/test_fast_commit_report -m $1 -d $2 -t $3 -r $4 -s $5\n"
<< "-m(mode): 0 = NORMAL_MODE; 1 = BIANQUE_MODE\n"
std::cout << "./mittest/simple_server/test_fast_commit_report -m $1 -d $2 -l $3 -t $4 -r $5 -s $6\n"
<< "-m(test mode): 0 = NORMAL_MODE; 1 = BIANQUE_MODE\n"
<< "-d(data model): 0 = ALL_CLEANOUT; 1 = ALL_SAME_TXN_DELAY_CLEANOUT; 2 = ALL_DIFF_TXN_DELAY_CLEANOUT\n"
<< "-t(tx data model): 0 = ALL_IN_MEMORY; 1 = ALL_IN_DISK\n"
<< "-l(data location): 0 = DATA_IN_MEMORY; 1 = DATA_ON_DISK\n"
<< "-t(tx table model): 0 = ALL_IN_MEMORY; 1 = ALL_IN_DISK\n"
<< "-r(row count): n = n row that is read during benchmark\n"
<< "-s(session count): n = n session that is used during insert before benchmark\n";
}
int main(int argc, char **argv)
{
using namespace oceanbase::unittest;
int c = 0;
while(EOF != (c = getopt(argc,argv,"h:m:d:t:r:s:"))) {
while(EOF != (c = getopt(argc,argv,":h:m:d:l:t:r:s:p:S:"))) {
switch(c) {
case 'h':
tutorial();
return 0;
case 'm':
fprintf(stdout, "m : %s\n", optarg);
oceanbase::unittest::fast_commit_test_mode = (oceanbase::unittest::FastCommitTestMode)atoi(optarg);
TEST_MODE = (EnumTestMode)atoi(optarg);
break;
case 'd':
oceanbase::unittest::fast_commit_data_mode = (oceanbase::unittest::FastCommitReportDataMode)atoi(optarg);
DATA_MODE = (EnumDataMode)atoi(optarg);
break;
case 'l':
DATA_LOCATION = (EnumDataLocation)atoi(optarg);
break;
case 't':
oceanbase::unittest::fast_commit_tx_table_mode = (oceanbase::unittest::FastCommitReportTxTableMode)atoi(optarg);
TX_TABLE_MODE = (EnumTxTableMode)atoi(optarg);
break;
case 'r':
oceanbase::unittest::total_fc_row_count = (int64_t)atoi(optarg);
TOTAL_FC_ROW_COUNT = (int64_t)atoi(optarg);
break;
case 's':
oceanbase::unittest::total_fc_session = (int64_t)atoi(optarg);
TOTAL_FC_SESSION = (int64_t)atoi(optarg);
break;
case 'p':
SELECT_PARALLEL = (int64_t)atoi(optarg);
break;
case 'S':
SLEEP_SECONDS = (int64_t)atoi(optarg);
break;
default:
break;
tutorial();
return 0;
}
}
if (oceanbase::unittest::fast_commit_test_mode != oceanbase::unittest::FastCommitTestMode::NORMAL_TEST
&& oceanbase::unittest::fast_commit_test_mode != oceanbase::unittest::FastCommitTestMode::BIANQUE_TEST) {
TRANS_LOG_RET(ERROR, OB_ERR_UNEXPECTED, "wrong choice", K(oceanbase::unittest::fast_commit_test_mode));
ob_abort();
if (EnumDataLocation::DATA_ON_DISK == DATA_LOCATION) {
if (EnumDataMode::ALL_DIFF_TXN_DELAY_CLEANOUT == DATA_MODE) {
fprintf(stdout, "Attention!! Not Support Diff Tx!!\n");
DATA_MODE = EnumDataMode::ALL_SAME_TXN_DELAY_CLEANOUT;
}
TOTAL_FC_SESSION = 1;
}
if (oceanbase::unittest::fast_commit_data_mode != oceanbase::unittest::FastCommitReportDataMode::ALL_CLEANOUT
&& oceanbase::unittest::fast_commit_data_mode != oceanbase::unittest::FastCommitReportDataMode::ALL_SAME_TXN_DELAY_CLEANOUT
&& oceanbase::unittest::fast_commit_data_mode != oceanbase::unittest::FastCommitReportDataMode::ALL_DIFF_TXN_DELAY_CLEANOUT) {
TRANS_LOG_RET(ERROR, OB_ERR_UNEXPECTED, "wrong choice", K(oceanbase::unittest::fast_commit_data_mode));
ob_abort();
}
if (oceanbase::unittest::fast_commit_tx_table_mode != oceanbase::unittest::FastCommitReportTxTableMode::ALL_IN_MEMORY
&& oceanbase::unittest::fast_commit_tx_table_mode != oceanbase::unittest::FastCommitReportTxTableMode::ALL_IN_DISK) {
TRANS_LOG_RET(ERROR, OB_ERR_UNEXPECTED, "wrong choice", K(oceanbase::unittest::fast_commit_tx_table_mode));
ob_abort();
}
if (oceanbase::unittest::total_fc_row_count < oceanbase::unittest::total_fc_session) {
TRANS_LOG_RET(ERROR, OB_ERR_UNEXPECTED, "wrong choice", K(oceanbase::unittest::total_fc_row_count), K(oceanbase::unittest::total_fc_session));
ob_abort();
if (TEST_MODE >= EnumTestMode::MAX_TEST_MODE || DATA_MODE >= EnumDataMode::MAX_DATA_MODE ||
TX_TABLE_MODE >= EnumTxTableMode::MAX_TX_TABLE_MODE || DATA_LOCATION >= EnumDataLocation::MAX_DATA_LOCATION ||
SELECT_PARALLEL <= 0 || SELECT_PARALLEL > 128) {
fprintf(stdout, "invalid argument for fast commit report test. ");
tutorial();
return 0;
}
oceanbase::unittest::init_log_and_gtest(argc, argv);
OB_LOGGER.set_log_level("info");
::testing::InitGoogleTest(&argc, argv);
TRANS_LOG(INFO, "fast commit report with arguments",
K(oceanbase::unittest::fast_commit_test_mode),
K(oceanbase::unittest::fast_commit_data_mode),
K(oceanbase::unittest::fast_commit_tx_table_mode),
K(oceanbase::unittest::total_fc_row_count),
K(oceanbase::unittest::total_fc_session));
return RUN_ALL_TESTS();
}

182
mittest/simple_server/rewrite_function_for_test_big_tx_data.cpp
View File

@ -17,7 +17,6 @@
int64_t TEST_TX_ID = 0;
bool DUMP_BIG_TX_DATA = false;
bool LOAD_BIG_TX_DATA = false;
bool READ_TEST_TX_FROM_SSTABLE = false;
int64_t BIGGEST_TX_DATA_SIZE = 0;
namespace oceanbase
@ -27,6 +26,47 @@ using namespace share;
using namespace transaction;
namespace storage
{
int ObTxDataMemtable::estimate_phy_size(const ObStoreRowkey *start_key,
const ObStoreRowkey *end_key,
int64_t &total_bytes,
int64_t &total_rows)
{
int ret = OB_SUCCESS;
total_bytes = 134217728LL * 2LL;
total_rows = 1;
return ret;
}
int ObTxTable::check_with_tx_data(ObReadTxDataArg &read_tx_data_arg, ObITxDataCheckFunctor &fn)
{
int ret = OB_SUCCESS;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
STORAGE_LOG(WARN, "tx table is not init.", KR(ret), K(read_tx_data_arg));
return ret;
}
/**************************************************************************************************/
// 跳过读缓存
/**************************************************************************************************/
// step 3 : read tx data in tx_ctx table and tx_data table
if (OB_FAIL(ret)) {
} else if (OB_FAIL(check_tx_data_in_tables_(read_tx_data_arg, fn))) {
if (OB_TRANS_CTX_NOT_EXIST != ret) {
STORAGE_LOG(WARN, "check tx data in tables failed", KR(ret), K(ls_id_), K(read_tx_data_arg));
}
}
// step 4 : make sure tx table can be read
if (OB_SUCC(ret) || OB_TRANS_CTX_NOT_EXIST == ret) {
check_state_and_epoch_(read_tx_data_arg.tx_id_, read_tx_data_arg.read_epoch_, true /*need_log_error*/, ret);
}
return ret;
}
int ObTxData::add_undo_action(ObTxTable *tx_table,
transaction::ObUndoAction &new_undo_action,
ObUndoStatusNode *undo_node)
@ -141,6 +181,7 @@ int ObTxDataMemtableScanIterator
int ObTxDataSingleRowGetter::deserialize_tx_data_from_store_buffers_(ObTxData &tx_data)
{
// fprintf(stdout, "deserialize tx data from sstable, tx_id = %ld, buffer size = %ld\n", tx_data.tx_id_.get_id(), tx_data_buffers_.count());
int ret = OB_SUCCESS;
int64_t total_buffer_size = 0;
for (int64_t idx = 0; idx < tx_data_buffers_.count(); ++idx) {
@ -169,6 +210,10 @@ int ObTxDataSingleRowGetter::deserialize_tx_data_from_store_buffers_(ObTxData &t
/**************************************************************************************************/
if (tx_data.tx_id_.get_id() == ATOMIC_LOAD(&TEST_TX_ID)) {
if (tx_data_buffers_.count() > 1) {
fprintf(stdout,
"deserialize large tx data from sstable, tx_id = %ld tx_data_buffers_.count() = %ld\n",
tx_data.tx_id_.get_id(),
tx_data_buffers_.count());
ATOMIC_STORE(&LOAD_BIG_TX_DATA, true);
std::cout << "read big tx id from sstable, tx_id:" << ATOMIC_LOAD(&TEST_TX_ID) << ", undo cnt:" << tx_data.undo_status_list_.undo_node_cnt_ << ", buffer cnt:" << tx_data_buffers_.count() << std::endl;
}
@ -181,140 +226,5 @@ int ObTxDataSingleRowGetter::deserialize_tx_data_from_store_buffers_(ObTxData &t
return ret;
}
int ObTxDataTable::check_with_tx_data(const ObTransID tx_id, ObITxDataCheckFunctor &fn)
{
int ret = OB_SUCCESS;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
STORAGE_LOG(WARN, "tx data table is not init.", KR(ret), KP(this), K(tx_id));
} else if (OB_SUCC(check_tx_data_in_memtable_(tx_id, fn))) {
// successfully do check function in memtable, check done
STORAGE_LOG(DEBUG, "tx data table check with tx memtable data succeed", K(tx_id), K(fn));
} else if (OB_TRANS_CTX_NOT_EXIST == ret && OB_SUCC(check_tx_data_in_sstable_(tx_id, fn))) {
/**************************************************************************************************/
if (tx_id.get_id() == ATOMIC_LOAD(&TEST_TX_ID)) {
ATOMIC_STORE(&READ_TEST_TX_FROM_SSTABLE, true);
}
/**************************************************************************************************/
// successfully do check function in sstable
STORAGE_LOG(DEBUG, "tx data table check with tx sstable data succeed", K(tx_id), K(fn));
} else {
STORAGE_LOG(WARN, "check something in tx data fail.", KR(ret), K(tx_id), KP(this),
K(tablet_id_));
}
return ret;
}
int ObParallelMergeCtx::init_parallel_mini_merge(compaction::ObTabletMergeCtx &merge_ctx)
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(MINI_MERGE != merge_ctx.param_.merge_type_)) {
ret = OB_INVALID_ARGUMENT;
STORAGE_LOG(WARN, "Invalid argument to init parallel mini merge", K(ret), K(merge_ctx));
} else {
const int64_t tablet_size = merge_ctx.get_schema()->get_tablet_size();
memtable::ObIMemtable *memtable = nullptr;
if (OB_FAIL(merge_ctx.tables_handle_.get_first_memtable(memtable))) {
STORAGE_LOG(WARN, "failed to get first memtable", K(ret),
"merge tables", merge_ctx.tables_handle_);
} else {
int64_t total_bytes = 0;
int64_t total_rows = 0;
int64_t mini_merge_thread = 0;
if (OB_FAIL(memtable->estimate_phy_size(nullptr, nullptr, total_bytes, total_rows))) {
STORAGE_LOG(WARN, "Failed to get estimate size from memtable", K(ret));
} else if (MTL(ObTenantDagScheduler *)->get_up_limit(ObDagPrio::DAG_PRIO_COMPACTION_HIGH, mini_merge_thread)) {
STORAGE_LOG(WARN, "failed to get uplimit", K(ret), K(mini_merge_thread));
} else {
ObArray<ObStoreRange> store_ranges;
mini_merge_thread = MAX(mini_merge_thread, PARALLEL_MERGE_TARGET_TASK_CNT);
/**************************************************************************************************/
concurrent_cnt_ = 2;
// concurrent_cnt_ = MIN((total_bytes + tablet_size - 1) / tablet_size, mini_merge_thread);
/**************************************************************************************************/
if (concurrent_cnt_ <= 1) {
if (OB_FAIL(init_serial_merge())) {
STORAGE_LOG(WARN, "Failed to init serialize merge", K(ret));
}
} else if (OB_FAIL(memtable->get_split_ranges(nullptr, nullptr, concurrent_cnt_, store_ranges))) {
if (OB_ENTRY_NOT_EXIST == ret) {
if (OB_FAIL(init_serial_merge())) {
STORAGE_LOG(WARN, "Failed to init serialize merge", K(ret));
}
} else {
STORAGE_LOG(WARN, "Failed to get split ranges from memtable", K(ret));
}
} else if (OB_UNLIKELY(store_ranges.count() != concurrent_cnt_)) {
ret = OB_ERR_UNEXPECTED;
STORAGE_LOG(WARN, "Unexpected range array and concurrent_cnt", K(ret), K_(concurrent_cnt),
K(store_ranges));
} else {
for (int64_t i = 0; OB_SUCC(ret) && i < store_ranges.count(); i++) {
ObDatumRange datum_range;
if (OB_FAIL(datum_range.from_range(store_ranges.at(i), allocator_))) {
STORAGE_LOG(WARN, "Failed to transfer store range to datum range", K(ret), K(i), K(store_ranges.at(i)));
} else if (OB_FAIL(range_array_.push_back(datum_range))) {
STORAGE_LOG(WARN, "Failed to push back merge range to array", K(ret), K(datum_range));
}
}
parallel_type_ = PARALLEL_MINI;
STORAGE_LOG(INFO, "Succ to get parallel mini merge ranges", K_(concurrent_cnt), K_(range_array));
}
}
}
}
return ret;
}
int ObParallelMergeCtx::init(compaction::ObTabletMergeCtx &merge_ctx)
{
int ret = OB_SUCCESS;
if (IS_INIT) {
ret = OB_INIT_TWICE;
STORAGE_LOG(WARN, "ObParallelMergeCtx init twice", K(ret));
} else if (OB_UNLIKELY(nullptr == merge_ctx.get_schema() || merge_ctx.tables_handle_.empty())) {
ret = OB_INVALID_ARGUMENT;
STORAGE_LOG(WARN, "Invalid argument to init parallel merge", K(ret), K(merge_ctx));
} else {
int64_t tablet_size = merge_ctx.get_schema()->get_tablet_size();
/**************************************************************************************************/
bool enable_parallel_minor_merge = true;
// bool enable_parallel_minor_merge = false;
// {
// omt::ObTenantConfigGuard tenant_config(TENANT_CONF(MTL_ID()));
// if (tenant_config.is_valid()) {
// enable_parallel_minor_merge = tenant_config->_enable_parallel_minor_merge;
// }
// }
/**************************************************************************************************/
if (enable_parallel_minor_merge && tablet_size > 0 && is_mini_merge(merge_ctx.param_.merge_type_)) {
if (OB_FAIL(init_parallel_mini_merge(merge_ctx))) {
STORAGE_LOG(WARN, "Failed to init parallel setting for mini merge", K(ret));
}
} else if (enable_parallel_minor_merge && tablet_size > 0 && is_minor_merge(merge_ctx.param_.merge_type_)) {
if (OB_FAIL(init_parallel_mini_minor_merge(merge_ctx))) {
STORAGE_LOG(WARN, "Failed to init parallel setting for mini minor merge", K(ret));
}
} else if (tablet_size > 0 && is_major_merge_type(merge_ctx.param_.merge_type_)) {
if (OB_FAIL(init_parallel_major_merge(merge_ctx))) {
STORAGE_LOG(WARN, "Failed to init parallel major merge", K(ret));
}
} else if (OB_FAIL(init_serial_merge())) {
STORAGE_LOG(WARN, "Failed to init serialize merge", K(ret));
}
if (OB_SUCC(ret)) {
is_inited_ = true;
STORAGE_LOG(INFO, "Succ to init parallel merge ctx",
K(enable_parallel_minor_merge), K(tablet_size), K(merge_ctx.param_));
}
}
return ret;
}
}
}

13
mittest/simple_server/test_big_tx_data.cpp
View File

@ -52,7 +52,8 @@ class DoNothingOP : public ObITxDataCheckFunctor
{
virtual int operator()(const ObTxData &tx_data, ObTxCCCtx *tx_cc_ctx = nullptr) {
UNUSED(tx_cc_ctx);
cout << "read tx data:" << tx_data.tx_id_.get_id() << ", undo cnt:" << tx_data.undo_status_list_.undo_node_cnt_ << ", READ_TEST_TX_FROM_SSTABLE:" << ATOMIC_LOAD(&READ_TEST_TX_FROM_SSTABLE) << endl;
cout << "read tx data:" << tx_data.tx_id_.get_id() << ", undo cnt:" << tx_data.undo_status_list_.undo_node_cnt_ << endl;
STORAGE_LOG_RET(INFO, 0, "read tx data", K(tx_data.tx_id_), K(lbt()));
return OB_SUCCESS;
}
};
@ -95,6 +96,7 @@ TEST_F(TestBigTxData, big_tx_data)
result = res.mysql_result();
DO(result->next());
result->get_int("trans_id", tx_id);
ASSERT_EQ(OB_ITER_END, result->next());
ATOMIC_STORE(&TEST_TX_ID, tx_id);
std::cout << "tx_id:" << tx_id << std::endl;
}
@ -118,11 +120,12 @@ TEST_F(TestBigTxData, big_tx_data)
ObLSService *ls_service = MTL(ObLSService*);
ObLSHandle handle;
DO(ls_service->get_ls(ObLSID(1), handle, storage::ObLSGetMod::DEADLOCK_MOD));
STORAGETEST_LOG(INFO, "start read tx data from sstable", K(TEST_TX_ID));
DO(handle.get_ls()->tx_table_.check_with_tx_data(ObTransID(ATOMIC_LOAD(&TEST_TX_ID)), op, 0));
// 7，检查被测事务的tx data已经经过了deserialize以及从sstable读取过
fprintf(stdout, "start read tx data from sstable, test_tx_id = %ld\n", TEST_TX_ID);
ObTxDataMiniCache fake_cache;
ObReadTxDataArg read_arg(ObTransID(ATOMIC_LOAD(&TEST_TX_ID)), 0, fake_cache);
DO(handle.get_ls()->tx_table_.check_with_tx_data(read_arg, op));
// 7，检查被测事务的tx data已经经过了deserialize
ASSERT_EQ(ATOMIC_LOAD(&LOAD_BIG_TX_DATA), true);
ASSERT_EQ(ATOMIC_LOAD(&READ_TEST_TX_FROM_SSTABLE), true);
}
}

166
mittest/simple_server/test_ob_minor_freeze.cpp
View File

@ -153,19 +153,23 @@ public:
void logstream_freeze();
void tablet_freeze();
void tablet_freeze_for_replace_tablet_meta();
void batch_tablet_freeze();
void insert_and_freeze();
void empty_memtable_flush();
void all_virtual_minor_freeze_info();
void check_frozen_memtable();
void create_tables();
private:
static const int64_t OB_DEFAULT_TABLE_COUNT = 3;
private:
int insert_thread_num_ = 10;
int insert_num_ = 200000;
int freeze_num_ = 1000;
int freeze_duration_ = 50 * 1000 * 1000;
int64_t table_id_ = 0;
ObTabletID tablet_id_;
share::ObLSID ls_id_;
ObLSHandle ls_handle_;
ObSEArray<int64_t, OB_DEFAULT_TABLE_COUNT> table_ids_;
ObSEArray<ObTabletID, OB_DEFAULT_TABLE_COUNT> tablet_ids_;
ObSEArray<share::ObLSID, OB_DEFAULT_TABLE_COUNT> ls_ids_;
ObSEArray<ObLSHandle, OB_DEFAULT_TABLE_COUNT> ls_handles_;
};
void ObMinorFreezeTest::get_tablet_id_and_ls_id()
@ -178,9 +182,9 @@ void ObMinorFreezeTest::get_tablet_id_and_ls_id()
int64_t tmp_ls_id = 0;
OB_LOG(INFO, "get table_id");
{
for (int i = 0; i < OB_DEFAULT_TABLE_COUNT; ++i) {
ObSqlString sql;
ASSERT_EQ(OB_SUCCESS, sql.assign_fmt("select table_id from oceanbase.__all_virtual_table where table_name = 't1'"));
ASSERT_EQ(OB_SUCCESS, sql.assign_fmt("select table_id from oceanbase.__all_virtual_table where table_name = 't%d'", i));
SMART_VAR(ObMySQLProxy::MySQLResult, res) {
ASSERT_EQ(OB_SUCCESS, sql_proxy.read(res, sql.ptr()));
sqlclient::ObMySQLResult *result = res.get_result();
@ -188,12 +192,14 @@ void ObMinorFreezeTest::get_tablet_id_and_ls_id()
ASSERT_EQ(OB_SUCCESS, result->next());
ASSERT_EQ(OB_SUCCESS, result->get_int("table_id", tmp_table_id));
}
table_ids_.push_back(tmp_table_id);
OB_LOG(INFO, "tmp_table_id", K(tmp_table_id));
}
OB_LOG(INFO, "get tablet_id");
{
for (int i = 0; i < OB_DEFAULT_TABLE_COUNT; ++i) {
ObSqlString sql;
ASSERT_EQ(OB_SUCCESS, sql.assign_fmt("select tablet_id from oceanbase.__all_virtual_tablet_to_ls where table_id = %ld", tmp_table_id));
ASSERT_EQ(OB_SUCCESS, sql.assign_fmt("select tablet_id from oceanbase.__all_virtual_tablet_to_ls where table_id = %ld", table_ids_.at(i)));
SMART_VAR(ObMySQLProxy::MySQLResult, res) {
ASSERT_EQ(OB_SUCCESS, sql_proxy.read(res, sql.ptr()));
sqlclient::ObMySQLResult *result = res.get_result();
@ -201,12 +207,14 @@ void ObMinorFreezeTest::get_tablet_id_and_ls_id()
ASSERT_EQ(OB_SUCCESS, result->next());
ASSERT_EQ(OB_SUCCESS, result->get_int("tablet_id", tmp_tablet_id));
}
tablet_ids_.push_back(ObTabletID(tmp_tablet_id));
OB_LOG(INFO, "tmp_tablet_id", K(tmp_tablet_id));
}
OB_LOG(INFO, "get ls_id");
{
for (int i = 0; i < OB_DEFAULT_TABLE_COUNT; ++i) {
ObSqlString sql;
ASSERT_EQ(OB_SUCCESS, sql.assign_fmt("select ls_id from oceanbase.__all_virtual_tablet_to_ls where table_id = %ld", tmp_table_id));
ASSERT_EQ(OB_SUCCESS, sql.assign_fmt("select ls_id from oceanbase.__all_virtual_tablet_to_ls where table_id = %ld", table_ids_.at(i)));
SMART_VAR(ObMySQLProxy::MySQLResult, res) {
ASSERT_EQ(OB_SUCCESS, sql_proxy.read(res, sql.ptr()));
sqlclient::ObMySQLResult *result = res.get_result();
@ -214,18 +222,17 @@ void ObMinorFreezeTest::get_tablet_id_and_ls_id()
ASSERT_EQ(OB_SUCCESS, result->next());
ASSERT_EQ(OB_SUCCESS, result->get_int("ls_id", tmp_ls_id));
}
ls_ids_.push_back(share::ObLSID(tmp_ls_id));
OB_LOG(INFO, "tmp_ls_id", K(tmp_ls_id));
}
table_id_ = tmp_table_id;
tablet_id_ = tmp_tablet_id;
ls_id_ = tmp_ls_id;
OB_LOG(INFO, "tmp_table_id", K(tmp_table_id));
OB_LOG(INFO, "tmp_tablet_id", K(tmp_tablet_id));
OB_LOG(INFO, "tmp_ls_id", K(tmp_ls_id));
OB_LOG(INFO, "tablet_id", K(table_id_));
OB_LOG(INFO, "ls_id", K(ls_id_));
ASSERT_EQ(true, tablet_id_.is_valid());
ASSERT_EQ(true, ls_id_.is_valid());
OB_LOG(INFO, "tablet_ids", K(tablet_ids_));
OB_LOG(INFO, "ls_ids", K(ls_ids_));
for (int i = 0; i < OB_DEFAULT_TABLE_COUNT; ++i) {
ASSERT_EQ(true, tablet_ids_.at(i).is_valid());
ASSERT_EQ(true, ls_ids_.at(i).is_valid());
}
}
void ObMinorFreezeTest::all_virtual_minor_freeze_info()
@ -236,13 +243,15 @@ void ObMinorFreezeTest::all_virtual_minor_freeze_info()
OB_LOG(INFO, "test __all_virtual_minor_freeze_info");
while (ObTimeUtility::current_time() - start_time <= freeze_duration_) {
ObSqlString sql;
ASSERT_EQ(OB_SUCCESS, sql.assign_fmt("select * from oceanbase.__all_virtual_minor_freeze_info where ls_id = %ld", ls_id_.id()));
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());
for (int i = 0; i < OB_DEFAULT_TABLE_COUNT; ++i) {
ObSqlString sql;
ASSERT_EQ(OB_SUCCESS, sql.assign_fmt("select * from oceanbase.__all_virtual_minor_freeze_info where ls_id = %ld", ls_ids_.at(i).id()));
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());
}
}
}
@ -290,14 +299,18 @@ void ObMinorFreezeTest::check_frozen_memtable()
void ObMinorFreezeTest::get_ls()
{
OB_LOG(INFO, "get_ls");
share::ObTenantSwitchGuard tenant_guard;
OB_LOG(INFO, "tenant_id", K(RunCtx.tenant_id_));
ASSERT_EQ(OB_SUCCESS, tenant_guard.switch_to(RunCtx.tenant_id_));
ObLSService *ls_srv = MTL(ObLSService *);
ASSERT_NE(nullptr, ls_srv);
OB_LOG(INFO, "ls_id", K(ls_id_));
ASSERT_EQ(OB_SUCCESS, ls_srv->get_ls(ls_id_, ls_handle_, ObLSGetMod::STORAGE_MOD));
ASSERT_EQ(true, ls_handle_.is_valid());
for (int i = 0; i < OB_DEFAULT_TABLE_COUNT; ++i) {
share::ObTenantSwitchGuard tenant_guard;
ObLSHandle ls_handle;
OB_LOG(INFO, "tenant_id", K(RunCtx.tenant_id_));
ASSERT_EQ(OB_SUCCESS, tenant_guard.switch_to(RunCtx.tenant_id_));
ObLSService *ls_srv = MTL(ObLSService *);
ASSERT_NE(nullptr, ls_srv);
OB_LOG(INFO, "ls_id", K(ls_ids_.at(i)));
ASSERT_EQ(OB_SUCCESS, ls_srv->get_ls(ls_ids_.at(i), ls_handle, ObLSGetMod::STORAGE_MOD));
ASSERT_EQ(true, ls_handle.is_valid());
ls_handles_.push_back(ls_handle);
}
}
void ObMinorFreezeTest::insert_data(int start)
@ -309,11 +322,13 @@ void ObMinorFreezeTest::insert_data(int start)
OB_LOG(INFO, "insert data start");
int i = 0;
while (ObTimeUtility::current_time() - start_time <= freeze_duration_) {
ObSqlString sql;
ASSERT_EQ(OB_SUCCESS, sql.assign_fmt("insert into t1 values(%d, %d)", i + start, i + start));
int64_t affected_rows = 0;
ret = sql_proxy.write(sql.ptr(), affected_rows);
ASSERT_EQ(OB_SUCCESS, ret);
for (int j = 0; j < OB_DEFAULT_TABLE_COUNT; ++j) {
ObSqlString sql;
ASSERT_EQ(OB_SUCCESS, sql.assign_fmt("insert into t%d values(%d, %d)", j, i + start, i + start));
int64_t affected_rows = 0;
ret = sql_proxy.write(sql.ptr(), affected_rows);
ASSERT_EQ(OB_SUCCESS, ret);
}
++i;
}
}
@ -343,7 +358,10 @@ void ObMinorFreezeTest::logstream_freeze()
const int64_t start = ObTimeUtility::current_time();
while (ObTimeUtility::current_time() - start <= freeze_duration_) {
ASSERT_EQ(OB_SUCCESS, ls_handle_.get_ls()->logstream_freeze((i % 2 == 0) ? true : false));
for (int j = 0; j < OB_DEFAULT_TABLE_COUNT; ++j) {
ASSERT_EQ(OB_SUCCESS, ls_handles_.at(j).get_ls()->logstream_freeze((i % 2 == 0) ? true : false));
}
i = i + 1;
}
}
@ -358,7 +376,9 @@ void ObMinorFreezeTest::tablet_freeze()
const int64_t start = ObTimeUtility::current_time();
while (ObTimeUtility::current_time() - start <= freeze_duration_) {
ASSERT_EQ(OB_SUCCESS, ls_handle_.get_ls()->tablet_freeze(tablet_id_, (i % 2 == 0) ? true : false));
for (int j = 0; j < OB_DEFAULT_TABLE_COUNT; ++j) {
ASSERT_EQ(OB_SUCCESS, ls_handles_.at(j).get_ls()->tablet_freeze(tablet_ids_.at(j), (i % 2 == 0) ? true : false));
}
i = i + 1;
}
}
@ -371,24 +391,42 @@ void ObMinorFreezeTest::tablet_freeze_for_replace_tablet_meta()
ASSERT_EQ(OB_SUCCESS, tenant_guard.switch_to(RunCtx.tenant_id_));
const int64_t start = ObTimeUtility::current_time();
while (ObTimeUtility::current_time() - start <= freeze_duration_) {
ObTableHandleV2 handle;
int ret = ls_handle_.get_ls()->get_freezer()->tablet_freeze_for_replace_tablet_meta(tablet_id_, handle);
if (OB_EAGAIN == ret || OB_ENTRY_EXIST == ret) {
ret = OB_SUCCESS;
}
ASSERT_EQ(OB_SUCCESS, ret);
if (OB_SUCC(ret)) {
ASSERT_EQ(OB_SUCCESS, ls_handle_.get_ls()->get_freezer()->handle_frozen_memtable_for_replace_tablet_meta(tablet_id_, handle));
for (int j = 0; j < OB_DEFAULT_TABLE_COUNT; ++j) {
ObTableHandleV2 handle;
int ret = ls_handles_.at(j).get_ls()->get_freezer()->tablet_freeze_for_replace_tablet_meta(tablet_ids_.at(j), handle);
if (OB_EAGAIN == ret || OB_ENTRY_EXIST == ret) {
ret = OB_SUCCESS;
}
ASSERT_EQ(OB_SUCCESS, ret);
if (OB_SUCC(ret)) {
ASSERT_EQ(OB_SUCCESS, ls_handles_.at(j).get_ls()->get_freezer()->handle_frozen_memtable_for_replace_tablet_meta(tablet_ids_.at(j), handle));
}
}
}
}
void ObMinorFreezeTest::batch_tablet_freeze()
{
common::ObMySQLProxy &sql_proxy = get_curr_simple_server().get_sql_proxy2();
share::ObTenantSwitchGuard tenant_guard;
OB_LOG(INFO, "tenant_id", K(RunCtx.tenant_id_));
ASSERT_EQ(OB_SUCCESS, tenant_guard.switch_to(RunCtx.tenant_id_));
int64_t i = 0;
const int64_t start = ObTimeUtility::current_time();
while (ObTimeUtility::current_time() - start <= freeze_duration_) {
ASSERT_EQ(OB_SUCCESS, ls_handles_.at(0).get_ls()->batch_tablet_freeze(tablet_ids_, (i % 2 == 0) ? true : false));
i = i + 1;
}
}
void ObMinorFreezeTest::insert_and_freeze()
{
std::thread tenant_freeze_thread([this]() { tenant_freeze(); });
std::thread tablet_freeze_thread([this]() { tablet_freeze(); });
std::thread tablet_freeze_for_replace_tablet_meta_thread([this]() { tablet_freeze_for_replace_tablet_meta(); });
std::thread check_frozen_memtable_thread([this]() { check_frozen_memtable(); });
std::thread batch_tablet_freeze_thread([this]() { batch_tablet_freeze(); });
std::vector<std::thread> insert_threads;
for (int i = 0; i < insert_thread_num_; ++i) {
int start = i * insert_num_;
@ -399,6 +437,7 @@ void ObMinorFreezeTest::insert_and_freeze()
tablet_freeze_thread.join();
tablet_freeze_for_replace_tablet_meta_thread.join();
check_frozen_memtable_thread.join();
batch_tablet_freeze_thread.join();
for (int i = 0; i < insert_thread_num_; ++i) {
insert_threads[i].join();
}
@ -452,6 +491,20 @@ void ObMinorFreezeTest::empty_memtable_flush()
}
}
void ObMinorFreezeTest::create_tables()
{
common::ObMySQLProxy &sql_proxy = get_curr_simple_server().get_sql_proxy2();
for (int i = 0; i < OB_DEFAULT_TABLE_COUNT; ++i) {
OB_LOG(INFO, "create_table: ", K(i));
ObSqlString sql;
sql.assign_fmt("create table if not exists t%d (c1 int, c2 int, primary key(c1))", i);
int64_t affected_rows = 0;
ASSERT_EQ(OB_SUCCESS, sql_proxy.write(sql.ptr(), affected_rows));
OB_LOG(INFO, "create_table succ", K(i));
}
}
TEST_F(ObMinorFreezeTest, observer_start)
{
SERVER_LOG(INFO, "observer_start succ");
@ -470,16 +523,9 @@ TEST_F(ObMinorFreezeTest, add_tenant)
TEST_F(ObMinorFreezeTest, create_table)
{
common::ObMySQLProxy &sql_proxy = get_curr_simple_server().get_sql_proxy2();
{
OB_LOG(INFO, "create_table start");
ObSqlString sql;
sql.assign_fmt("create table if not exists t1 (c1 int, c2 int, primary key(c1))");
int64_t affected_rows = 0;
ASSERT_EQ(OB_SUCCESS, sql_proxy.write(sql.ptr(), affected_rows));
OB_LOG(INFO, "create_table succ");
}
OB_LOG(INFO, "create_table start");
create_tables();
OB_LOG(INFO, "create_table all succ");
}
TEST_F(ObMinorFreezeTest, insert_and_freeze)

187
mittest/simple_server/test_tx_data_table_mit.cpp
View File

@ -52,6 +52,26 @@ using namespace storage;
using namespace palf;
using namespace share;
namespace storage
{
int ObTxDataMemtable::estimate_phy_size(const ObStoreRowkey *start_key,
const ObStoreRowkey *end_key,
int64_t &total_bytes,
int64_t &total_rows)
{
int ret = OB_SUCCESS;
if (MTL_ID() == 1002 && freezer_->get_ls_id().id() == 1001) {
total_bytes = 134217728LL * 2LL;
} else {
total_bytes = 1;
}
total_rows = 1;
return ret;
}
}
namespace unittest
{
@ -73,6 +93,9 @@ public:
void freeze_tx_data(ObTxDataTable *tx_data_table);
void check_start_tx_scn(ObTxDataTable *tx_data_table);
void basic_test();
void parallel_dump_rollback();
void check_minor_merge();
void check_recycle_scn();
private:
void check_tx_data_minor_succeed();
@ -99,7 +122,13 @@ private:
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 DEF_VAL_FOR_SQL \
int ret = OB_SUCCESS; \
ObSqlString sql; \
int64_t affected_rows = 0;
static int64_t VAL = 0;
void ObTxDataTableTest::insert_tx_data()
{
common::ObMySQLProxy &sql_proxy = get_curr_simple_server().get_sql_proxy2();
@ -116,14 +145,13 @@ void ObTxDataTableTest::insert_tx_data()
void ObTxDataTableTest::insert_rollback_tx_data()
{
DEF_VAL_FOR_SQL
common::ObMySQLProxy &sql_proxy = get_curr_simple_server().get_sql_proxy2();
OB_LOG(INFO, "insert rollback tx data start");
int64_t i = 0;
int64_t affected_rows = 0;
int rollback_cnt = 0;
first_rollback_ts_ = 0;
ObSqlString sql;
sqlclient::ObISQLConnection *connection = nullptr;
ASSERT_EQ(OB_SUCCESS, sql_proxy.acquire(connection));
ASSERT_NE(nullptr, connection);
@ -162,7 +190,7 @@ void ObTxDataTableTest::freeze_tx_data(ObTxDataTable *tx_data_table)
::sleep(1);
int64_t affected_rows = 0;
ObSqlString sql;
EXE_SQL("alter system minor freeze tenant tt1;");
EXE_SQL("alter system minor freeze tenant all;");
}
ATOMIC_STORE(&stop, true);
STORAGE_LOG(INFO, "freeze done");
@ -277,6 +305,131 @@ void ObTxDataTableTest::basic_test()
check_tx_data_minor_succeed();
}
void ObTxDataTableTest::parallel_dump_rollback()
{
DEF_VAL_FOR_SQL
common::ObMySQLProxy &sys_proxy = get_curr_simple_server().get_sql_proxy();
sqlclient::ObISQLConnection *sys_conn = nullptr;
ASSERT_EQ(OB_SUCCESS, sys_proxy.acquire(sys_conn));
ASSERT_NE(nullptr, sys_conn);
common::ObMySQLProxy &tt1_proxy = get_curr_simple_server().get_sql_proxy2();
sqlclient::ObISQLConnection *tt1_conn = nullptr;
ASSERT_EQ(OB_SUCCESS, tt1_proxy.acquire(tt1_conn));
ASSERT_NE(nullptr, tt1_conn);
WRITE_SQL_BY_CONN(tt1_conn, "create table if not exists test_parallel_dump_tx_data (c1 int, c2 int);");
WRITE_SQL_BY_CONN(tt1_conn, "set ob_trx_timeout = 3000000000");
WRITE_SQL_BY_CONN(tt1_conn, "set ob_trx_idle_timeout = 3000000000");
WRITE_SQL_BY_CONN(tt1_conn, "set ob_query_timeout = 3000000000");
fprintf(stdout, "start rollback to savepoint\n");
WRITE_SQL_BY_CONN(tt1_conn, "begin");
for (int64_t i = 0; i < 1000; i++) {
WRITE_SQL_BY_CONN(tt1_conn, "savepoint x");
WRITE_SQL_BY_CONN(tt1_conn, "insert into test_parallel_dump_tx_data values(1,1);");
WRITE_SQL_BY_CONN(tt1_conn, "rollback to savepoint x");
WRITE_SQL_BY_CONN(tt1_conn, "insert into test_parallel_dump_tx_data values(1,1);");
if (0 == i % 100) {
sys_conn->execute_write(OB_SYS_TENANT_ID, "alter system minor freeze tenant tt1 ls 1001 tablet_id = 49402;", affected_rows);
}
}
WRITE_SQL_BY_CONN(tt1_conn, "commit");
fprintf(stdout, "finish rollback to savepoint\n");
sleep(5);
sys_conn->execute_write(OB_SYS_TENANT_ID, "alter system minor freeze tenant tt1 ls 1001 tablet_id = 49402;", affected_rows);
sleep(5);
}
void ObTxDataTableTest::check_minor_merge()
{
DEF_VAL_FOR_SQL
common::ObMySQLProxy &sys_proxy = get_curr_simple_server().get_sql_proxy();
sqlclient::ObISQLConnection *sys_conn = nullptr;
ASSERT_EQ(OB_SUCCESS, sys_proxy.acquire(sys_conn));
ASSERT_NE(nullptr, sys_conn);
HEAP_VAR(ObMySQLProxy::MySQLResult, res)
{
int64_t retry_times = 10;
// 确认MINI MERGE 并行转储正确
int64_t cnt = 0;
while (--retry_times >= 0) {
ASSERT_EQ(OB_SUCCESS,
sys_conn->execute_read(
OB_SYS_TENANT_ID,
"select count(*) as cnt from oceanbase.__all_virtual_tablet_compaction_history where tenant_id "
"= 1002 and ls_id = 1001 and tablet_id = 49402 and parallel_degree > 1 and type = 'MINI_MERGE';",
res));
common::sqlclient::ObMySQLResult *result = res.mysql_result();
ASSERT_EQ(OB_SUCCESS, result->next());
ASSERT_EQ(OB_SUCCESS, result->get_int("cnt", cnt));
if (cnt > 0) {
break;
} else {
sleep(1);
}
}
ASSERT_GT(cnt, 0);
// 确认没有未能转储的memtable
retry_times = 10;
while (--retry_times >= 0) {
ASSERT_EQ(OB_SUCCESS,
sys_conn->execute_read(
OB_SYS_TENANT_ID,
"select count(*) as cnt from oceanbase.__all_virtual_tx_data_table where state = 'FROZEN' or state = 'FREEZING';",
res));
common::sqlclient::ObMySQLResult *result = res.mysql_result();
cnt = -1;
ASSERT_EQ(OB_SUCCESS, result->next());
ASSERT_EQ(OB_SUCCESS, result->get_int("cnt", cnt));
if (0 == cnt) {
break;
} else {
sleep(1);
}
}
ASSERT_EQ(0, cnt);
}
}
void ObTxDataTableTest::check_recycle_scn()
{
DEF_VAL_FOR_SQL
common::ObMySQLProxy &sys_proxy = get_curr_simple_server().get_sql_proxy();
sqlclient::ObISQLConnection *sys_conn = nullptr;
ASSERT_EQ(OB_SUCCESS, sys_proxy.acquire(sys_conn));
ASSERT_NE(nullptr, sys_conn);
HEAP_VAR(ObMySQLProxy::MySQLResult, res)
{
ASSERT_EQ(
OB_SUCCESS,
sys_conn->execute_read(
OB_SYS_TENANT_ID,
"select tenant_id, min_tx_log_scn from oceanbase.__all_virtual_tx_data_table where ls_id = 1 and state = 'MINOR';",
res));
common::sqlclient::ObMySQLResult *result = res.mysql_result();
int tenant_cnt = 0;
while (OB_SUCCESS == result->next()) {
int64_t tenant_id = 0;
int64_t min_tx_log_scn = 0;
ASSERT_EQ(OB_SUCCESS, result->get_int("tenant_id", tenant_id));
ASSERT_EQ(OB_SUCCESS, result->get_int("min_tx_log_scn", min_tx_log_scn));
ASSERT_TRUE(1 == tenant_id || 1001 == tenant_id || 1002 == tenant_id);
ASSERT_TRUE(min_tx_log_scn > 1);
tenant_cnt++;
}
ASSERT_EQ(tenant_cnt, 3);
}
}
TEST_F(ObTxDataTableTest, observer_start) { SERVER_LOG(INFO, "observer_start succ"); }
TEST_F(ObTxDataTableTest, add_tenant)
@ -288,6 +441,19 @@ TEST_F(ObTxDataTableTest, add_tenant)
ASSERT_EQ(OB_SUCCESS, get_curr_simple_server().init_sql_proxy2());
}
TEST_F(ObTxDataTableTest, init_config)
{
DEF_VAL_FOR_SQL
common::ObMySQLProxy &sys_proxy = get_curr_simple_server().get_sql_proxy();
sqlclient::ObISQLConnection *sys_conn = nullptr;
ASSERT_EQ(OB_SUCCESS, sys_proxy.acquire(sys_conn));
ASSERT_NE(nullptr, sys_conn);
WRITE_SQL_BY_CONN(sys_conn, "alter system set _private_buffer_size = '1B'");
WRITE_SQL_BY_CONN(sys_conn, "alter system set minor_compact_trigger = 0");
WRITE_SQL_BY_CONN(sys_conn, "alter system set _minor_compaction_amplification_factor = 1");
sleep(2);
}
TEST_F(ObTxDataTableTest, create_table)
{
common::ObMySQLProxy &sql_proxy = get_curr_simple_server().get_sql_proxy2();
@ -304,6 +470,21 @@ TEST_F(ObTxDataTableTest, basic_test)
basic_test();
}
TEST_F(ObTxDataTableTest, parallel_dump_rollback)
{
parallel_dump_rollback();
}
TEST_F(ObTxDataTableTest, check_minor_merge)
{
check_minor_merge();
}
TEST_F(ObTxDataTableTest, check_recycle_scn)
{
check_recycle_scn();
}
class ObTxDataTableRestartTest : public ObSimpleClusterTestBase
{
public: