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

fix switch leader cause map data is deleted bug

Browse Source
This commit is contained in:
obdev
2023-02-24 13:38:50 +00:00
committed by ob-robot
parent e44e4b0bbe
commit f372386816
5 changed files with 206 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
src/rootserver/ddl_task/ob_ddl_scheduler.cpp
View File

@ -289,7 +289,7 @@ int ObDDLTaskQueue::update_task_process_schedulable(const int64_t task_id)
return ret;
}
int ObDDLTaskQueue::abort_task(const int64_t task_id, common::ObMySQLProxy &mysql_proxy)
int ObDDLTaskQueue::abort_task(const int64_t task_id)
{
int ret = OB_SUCCESS;
share::ObTaskId trace_id;
@ -302,20 +302,7 @@ int ObDDLTaskQueue::abort_task(const int64_t task_id, common::ObMySQLProxy &mysq
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid arg", K(ret), K(task_id));
} else if (OB_FAIL(task_id_map_.get_refactored(task_id, ddl_task))) {
if (OB_HASH_NOT_EXIST == ret) {
bool exist = false;
if (OB_FAIL(ObDDLTaskRecordOperator::check_task_id_exist(mysql_proxy, task_id, exist))) {
LOG_WARN("check task id exist fail", K(ret));
} else {
if (exist) {
ret = OB_EAGAIN;
LOG_INFO("entry exist, the ddl scheduler hasn't recovered the task yet", K(ret), K(task_id));
} else {
ret = OB_ENTRY_NOT_EXIST;
LOG_WARN("this task does not exist in the hash table", K(ret), K(task_id));
}
}
}
ret = OB_HASH_NOT_EXIST == ret ? OB_ENTRY_NOT_EXIST : ret;
LOG_WARN("get from task map failed", K(ret), K(task_id));
} else if (OB_ISNULL(ddl_task)) {
ret = OB_ERR_UNEXPECTED;
@ -896,8 +883,22 @@ int ObDDLScheduler::abort_redef_table(const int64_t task_id)
if (OB_UNLIKELY(task_id <= 0)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid arg", K(ret), K(task_id));
} else if (OB_FAIL(task_queue_.abort_task(task_id, root_service_->get_sql_proxy()))) {
LOG_WARN("abort redef table task failed", K(ret));
} else if (OB_FAIL(task_queue_.abort_task(task_id))) {
LOG_WARN("abort redef table task failed", K(ret), K(task_id));
if (OB_ENTRY_NOT_EXIST == ret) {
bool exist = false;
int tmp_ret = OB_SUCCESS;
if (OB_TMP_FAIL(ObDDLTaskRecordOperator::check_task_id_exist(root_service_->get_sql_proxy(), task_id, exist))) {
LOG_WARN("check task id exist fail", K(tmp_ret), K(task_id));
} else {
if (exist) {
ret = OB_EAGAIN;
LOG_INFO("entry exist, the ddl scheduler hasn't recovered the task yet", K(ret), K(task_id));
} else {
LOG_WARN("this task does not exist in the hash table", K(ret), K(task_id));
}
}
}
}
return ret;
}
@ -945,6 +946,18 @@ int ObDDLScheduler::copy_table_dependents(const int64_t task_id,
return ret;
}))) {
LOG_WARN("failed to modify task", K(ret));
if (OB_ENTRY_NOT_EXIST == ret) {
int tmp_ret = OB_SUCCESS;
if (OB_TMP_FAIL(ObDDLTaskRecordOperator::get_ddl_task_record(task_id, root_service_->get_sql_proxy(), allocator, task_record))) {
LOG_WARN("get single ddl task failed", K(tmp_ret), K(task_id));
} else if (OB_TMP_FAIL(schedule_ddl_task(task_record))) {
LOG_WARN("failed to schedule ddl task", K(tmp_ret), K(task_record));
} else {
ret = OB_SUCCESS;
}
}
}
if (OB_FAIL(ret)) {
} else if (OB_FAIL(redefinition_task.init(task_record))) {
LOG_WARN("init table redefinition task failed", K(ret));
} else if (OB_FAIL(redefinition_task.set_trace_id(task_record.trace_id_))) {
@ -977,7 +990,21 @@ int ObDDLScheduler::copy_table_dependents(const int64_t task_id,
is_copy_triggers,
is_copy_foreign_keys,
is_ignore_errors))) {
LOG_WARN("update task process setting failed", K(ret));
if (OB_ENTRY_NOT_EXIST == ret) {
bool exist = false;
int tmp_ret = OB_SUCCESS;
if (OB_TMP_FAIL(ObDDLTaskRecordOperator::check_task_id_exist(root_service_->get_sql_proxy(), task_id, exist))) {
LOG_WARN("check task id exist fail", K(tmp_ret), K(task_id));
} else {
if (exist) {
ret = OB_EAGAIN;
LOG_INFO("entry exist, the ddl scheduler hasn't recovered the task yet", K(ret), K(task_id));
} else {
LOG_WARN("this task does not exist in the hash table", K(ret), K(task_id));
}
}
}
LOG_WARN("update task copy deps setting failed", K(ret), K(task_id));
}
}
}
@ -1025,6 +1052,17 @@ int ObDDLScheduler::finish_redef_table(const int64_t task_id, const uint64_t ten
return ret;
}))) {
LOG_WARN("failed to modify task", K(ret));
if (OB_ENTRY_NOT_EXIST == ret) {
int tmp_ret = OB_SUCCESS;
ObSqlString sql_string;
if (OB_TMP_FAIL(ObDDLTaskRecordOperator::get_ddl_task_record(task_id, root_service_->get_sql_proxy(), allocator, task_record))) {
LOG_WARN("get single ddl task failed", K(tmp_ret), K(task_id));
} else if (OB_TMP_FAIL(schedule_ddl_task(task_record))) {
LOG_WARN("failed to schedule ddl task", K(tmp_ret), K(task_record));
} else {
ret = OB_SUCCESS;
}
}
} else if (OB_FAIL(redefinition_task.init(task_record))) {
LOG_WARN("init table redefinition task failed", K(ret));
} else if (OB_FAIL(redefinition_task.set_trace_id(task_record.trace_id_))) {
@ -1048,7 +1086,21 @@ int ObDDLScheduler::finish_redef_table(const int64_t task_id, const uint64_t ten
}
if (OB_SUCC(ret)) {
if (OB_FAIL(task_queue_.update_task_process_schedulable(task_id))) {
LOG_WARN("update task process setting failed", K(ret));
if (OB_ENTRY_NOT_EXIST == ret) {
bool exist = false;
int tmp_ret = OB_SUCCESS;
if (OB_TMP_FAIL(ObDDLTaskRecordOperator::check_task_id_exist(root_service_->get_sql_proxy(), task_id, exist))) {
LOG_WARN("check task id exist fail", K(tmp_ret), K(task_id));
} else {
if (exist) {
ret = OB_EAGAIN;
LOG_INFO("entry exist, the ddl scheduler hasn't recovered the task yet", K(ret), K(task_id));
} else {
LOG_WARN("this task does not exist in the hash table", K(ret), K(task_id));
}
}
}
LOG_WARN("update task process schedulable failed", K(task_id));
}
}
}
@ -1479,11 +1531,12 @@ int ObDDLScheduler::recover_task()
ret = OB_NOT_INIT;
LOG_WARN("not init", K(ret));
} else {
ObSqlString sql_string;
ObArray<ObDDLTaskRecord> task_records;
ObArenaAllocator allocator(lib::ObLabel("DdlTasRecord"));
share::schema::ObMultiVersionSchemaService &schema_service = root_service_->get_schema_service();
if (OB_FAIL(ObDDLTaskRecordOperator::get_all_record(root_service_->get_sql_proxy(), allocator, task_records))) {
LOG_WARN("get all task records failed", K(ret));
if (OB_FAIL(ObDDLTaskRecordOperator::get_all_ddl_task_record(root_service_->get_sql_proxy(), allocator, task_records))) {
LOG_WARN("get task record failed", K(ret), K(sql_string));
}
LOG_INFO("start processing ddl recovery", K(task_records));
for (int64_t i = 0; OB_SUCC(ret) && i < task_records.count(); ++i) {

2
src/rootserver/ddl_task/ob_ddl_scheduler.h
View File

@ -70,7 +70,7 @@ public:
const bool is_copy_foreign_keys,
const bool is_ignore_errors);
int update_task_process_schedulable(const int64_t task_id);
int abort_task(const int64_t task_id, common::ObMySQLProxy &mysql_proxy);
int abort_task(const int64_t task_id);
int64_t get_task_cnt() const { return task_list_.get_size(); }
void destroy();
private:

82
src/rootserver/ddl_task/ob_ddl_task.cpp
View File

@ -2626,47 +2626,87 @@ int ObDDLTaskRecordOperator::check_has_conflict_ddl(
return ret;
}
int ObDDLTaskRecordOperator::get_all_record(
int ObDDLTaskRecordOperator::get_task_record(const ObSqlString &sql_string,
common::ObMySQLProxy &proxy,
common::ObIAllocator &allocator,
common::ObIArray<ObDDLTaskRecord> &records)
{
int ret = OB_SUCCESS;
records.reset();
if (OB_UNLIKELY(!proxy.is_inited())) {
if (OB_UNLIKELY(!proxy.is_inited() || !sql_string.is_valid())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(proxy.is_inited()));
LOG_WARN("invalid argment", K(ret), K(proxy.is_inited()), K(sql_string));
} else {
ObSqlString sql_string;
records.reset();
SMART_VAR(ObMySQLProxy::MySQLResult, res) {
ObDDLTaskRecord record;
sqlclient::ObMySQLResult *result = NULL;
if (OB_FAIL(sql_string.assign_fmt(" SELECT tenant_id, task_id, object_id, target_object_id, ddl_type, "
"schema_version, parent_task_id, trace_id, status, snapshot_version, task_version, execution_id, "
"UNHEX(ddl_stmt_str) as ddl_stmt_str_unhex, ret_code, UNHEX(message) as message_unhex FROM %s ", OB_ALL_VIRTUAL_DDL_TASK_STATUS_TNAME))) {
LOG_WARN("assign sql string failed", K(ret));
} else if (OB_FAIL(proxy.read(res, sql_string.ptr()))) {
if (OB_FAIL(proxy.read(res, sql_string.ptr()))) {
LOG_WARN("query ddl task record failed", K(ret), K(sql_string));
} else if (OB_ISNULL((result = res.get_result()))) {
} else if (OB_ISNULL(result = res.get_result())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("fail to get sql result", K(ret), KP(result));
} else {
ObDDLTaskRecord task_record;
while (OB_SUCC(ret) && OB_SUCC(result->next())) {
if (OB_FAIL(fill_task_record(result, allocator, task_record))) {
if (OB_FAIL(fill_task_record(result, allocator, record))) {
LOG_WARN("fill index task failed", K(ret), K(result));
} else if (!task_record.is_valid()) {
} else if (!record.is_valid()) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("task record is invalid", K(ret), K(task_record));
} else if (OB_FAIL(records.push_back(task_record))) {
LOG_WARN("push back task record failed", K(ret), K(task_record));
}
}
if (OB_ITER_END == ret) {
ret = OB_SUCCESS;
LOG_WARN("task record is invalid", K(ret), K(record));
} else if (OB_FAIL(records.push_back(record))) {
LOG_WARN("push back failed", K(record));
}
}
}
}
}
return ret;
}
int ObDDLTaskRecordOperator::get_ddl_task_record(const int64_t task_id,
common::ObMySQLProxy &proxy,
common::ObIAllocator &allocator,
ObDDLTaskRecord &record)
{
int ret = OB_SUCCESS;
ObSqlString sql_string;
ObArray<ObDDLTaskRecord> task_records;
if (OB_UNLIKELY(!proxy.is_inited())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(proxy.is_inited()));
} else if (OB_FAIL(sql_string.assign_fmt(" SELECT tenant_id, task_id, object_id, target_object_id, ddl_type, "
"schema_version, parent_task_id, trace_id, status, snapshot_version, task_version, execution_id, "
"UNHEX(ddl_stmt_str) as ddl_stmt_str_unhex, ret_code, UNHEX(message) as message_unhex FROM %s WHERE task_id=%lu", OB_ALL_VIRTUAL_DDL_TASK_STATUS_TNAME, task_id))) {
LOG_WARN("assign sql string failed", K(ret), K(task_id));
} else if (OB_FAIL(get_task_record(sql_string, proxy, allocator, task_records))) {
LOG_WARN("get task record failed", K(ret), K(sql_string));
} else if (task_records.count() != 1) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("one task_id but task_records count() != 1", K(ret), K(task_id), K(task_records));
} else if (OB_FAIL(task_records.at(0, record))) {
LOG_WARN("get task_record failed", K(ret), K(task_id));
} else if (!record.is_valid()) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("task record not valid", K(ret));
}
return ret;
}
int ObDDLTaskRecordOperator::get_all_ddl_task_record(common::ObMySQLProxy &proxy,
common::ObIAllocator &allocator,
common::ObIArray<ObDDLTaskRecord> &records)
{
int ret = OB_SUCCESS;
ObSqlString sql_string;
if (OB_UNLIKELY(!proxy.is_inited())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(proxy.is_inited()));
} else if (OB_FAIL(sql_string.assign_fmt(" SELECT tenant_id, task_id, object_id, target_object_id, ddl_type, "
"schema_version, parent_task_id, trace_id, status, snapshot_version, task_version, execution_id, "
"UNHEX(ddl_stmt_str) as ddl_stmt_str_unhex, ret_code, UNHEX(message) as message_unhex FROM %s ", OB_ALL_VIRTUAL_DDL_TASK_STATUS_TNAME))) {
LOG_WARN("assign sql string failed", K(ret));
} else if (OB_FAIL(get_task_record(sql_string, proxy, allocator, records))) {
LOG_WARN("get task record failed", K(ret), K(sql_string));
}
return ret;
}

15
src/rootserver/ddl_task/ob_ddl_task.h
View File

@ -171,10 +171,16 @@ public:
int64_t &task_status,
int64_t &execution_id);
static int get_all_record(
static int get_ddl_task_record(
const int64_t task_id,
common::ObMySQLProxy &proxy,
common::ObIAllocator &allocator,
ObDDLTaskRecord &record);
static int get_all_ddl_task_record(
common::ObMySQLProxy &proxy,
common::ObIAllocator &allocator,
common::ObIArray<ObDDLTaskRecord> &records);
static int check_task_id_exist(
common::ObMySQLProxy &proxy,
const int64_t task_id,
@ -219,11 +225,16 @@ private:
ObDDLTaskRecord &task_record);
static int64_t get_record_id(share::ObDDLType ddl_type, int64_t origin_id);
static int kill_inner_sql(
common::ObMySQLProxy &proxy,
const uint64_t tenant_id,
const uint64_t session_id);
static int get_task_record(
const ObSqlString &sql_string,
common::ObMySQLProxy &proxy,
common::ObIAllocator &allocator,
common::ObIArray<ObDDLTaskRecord> &records);
};
class ObDDLWaitTransEndCtx

51
src/storage/ddl/ob_ddl_server_client.cpp
View File

@ -89,6 +89,8 @@ int ObDDLServerClient::start_redef_table(const obrpc::ObStartRedefTableArg &arg,
int ObDDLServerClient::copy_table_dependents(const obrpc::ObCopyTableDependentsArg &arg)
{
int ret = OB_SUCCESS;
const int64_t retry_interval = 100 * 1000L;
ObAddr rs_leader_addr;
obrpc::ObCommonRpcProxy *common_rpc_proxy = GCTX.rs_rpc_proxy_;
if (OB_UNLIKELY(!arg.is_valid())) {
ret = OB_INVALID_ARGUMENT;
@ -96,8 +98,26 @@ int ObDDLServerClient::copy_table_dependents(const obrpc::ObCopyTableDependentsA
} else if (OB_ISNULL(common_rpc_proxy)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("common rpc proxy is null", K(ret));
} else if (OB_FAIL(common_rpc_proxy->copy_table_dependents(arg))) {
LOG_WARN("failed to copy table dependents", KR(ret), K(arg));
} else {
while (OB_SUCC(ret)) {
int tmp_ret = OB_SUCCESS;
if (OB_TMP_FAIL(GCTX.rs_mgr_->get_master_root_server(rs_leader_addr))) {
LOG_WARN("fail to rootservice address", K(tmp_ret));
} else if (OB_FAIL(common_rpc_proxy->to(rs_leader_addr).copy_table_dependents(arg))) {
LOG_WARN("copy table dependents failed", K(ret), K(arg));
if (OB_ENTRY_NOT_EXIST == ret) {
LOG_WARN("ddl task not exist", K(ret), K(arg));
break;
} else {
LOG_INFO("ddl task exist, try again", K(arg));
ret = OB_SUCCESS;
ob_usleep(retry_interval);
}
} else {
LOG_INFO("copy table dependents success", K(arg));
break;
}
}
}
return ret;
}
@ -120,9 +140,11 @@ int ObDDLServerClient::abort_redef_table(const obrpc::ObAbortRedefTableArg &arg,
if (OB_TMP_FAIL(GCTX.rs_mgr_->get_master_root_server(rs_leader_addr))) {
LOG_WARN("fail to get rootservice address", K(tmp_ret));
} else if (OB_FAIL(common_rpc_proxy->to(rs_leader_addr).abort_redef_table(arg))) {
LOG_WARN("abort redef table failed", K(ret), K(arg));
if (OB_ENTRY_NOT_EXIST == ret) {
break;
} else {
LOG_INFO("ddl task exist, try again", K(arg));
ret = OB_SUCCESS;
ob_usleep(retry_interval);
}
@ -151,6 +173,8 @@ int ObDDLServerClient::finish_redef_table(const obrpc::ObFinishRedefTableArg &fi
{
int ret = OB_SUCCESS;
int tmp_ret = OB_SUCCESS;
const int64_t retry_interval = 100 * 1000L;
ObAddr rs_leader_addr;
obrpc::ObCommonRpcProxy *common_rpc_proxy = GCTX.rs_rpc_proxy_;
if (OB_UNLIKELY(!finish_redef_arg.is_valid() || !build_single_arg.is_valid())) {
ret = OB_INVALID_ARGUMENT;
@ -158,8 +182,26 @@ int ObDDLServerClient::finish_redef_table(const obrpc::ObFinishRedefTableArg &fi
} else if (OB_ISNULL(common_rpc_proxy)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("common rpc proxy is null", K(ret));
} else if (OB_FAIL(common_rpc_proxy->finish_redef_table(finish_redef_arg))) {
LOG_WARN("failed to finish redef table", K(ret), K(finish_redef_arg));
} else {
while (OB_SUCC(ret)) {
int tmp_ret = OB_SUCCESS;
if (OB_TMP_FAIL(GCTX.rs_mgr_->get_master_root_server(rs_leader_addr))) {
LOG_WARN("fail to rootservice address", K(tmp_ret));
} else if (OB_FAIL(common_rpc_proxy->to(rs_leader_addr).finish_redef_table(finish_redef_arg))) {
LOG_WARN("finish redef table failed", K(ret), K(finish_redef_arg));
if (OB_ENTRY_NOT_EXIST == ret) {
break;
} else {
LOG_INFO("ddl task exist, try again", K(finish_redef_arg));
ret = OB_SUCCESS;
ob_usleep(retry_interval);
}
} else {
LOG_INFO("finish redef table success", K(finish_redef_arg));
break;
}
}
if (OB_FAIL(ret)) {
} else if (OB_FAIL(build_ddl_single_replica_response(build_single_arg))) {
LOG_WARN("build ddl single replica response", K(ret), K(build_single_arg));
} else if (OB_FAIL(sql::ObDDLExecutorUtil::wait_ddl_finish(finish_redef_arg.tenant_id_, finish_redef_arg.task_id_, session, common_rpc_proxy))) {
@ -168,6 +210,7 @@ int ObDDLServerClient::finish_redef_table(const obrpc::ObFinishRedefTableArg &fi
if (OB_TMP_FAIL(heart_beat_clear(finish_redef_arg.task_id_))) {
LOG_WARN("heart beat clear failed", K(tmp_ret), K(finish_redef_arg.task_id_));
}
}
return ret;
}