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

[Fix] commit ddl column checksum in follow replicas

Browse Source
This commit is contained in:
obdev
2022-12-24 04:38:02 +00:00
committed by ob-robot
parent caead7fcaa
commit d1c6e3e1d1
21 changed files with 486 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
src/storage/ddl/ob_tablet_ddl_kv_mgr.cpp
View File

@ -148,10 +148,12 @@ int ObTabletDDLKvMgr::ddl_start(const ObITable::TableKey &table_key,
return ret;
}
int ObTabletDDLKvMgr::ddl_prepare(const SCN &start_scn,
const SCN &prepare_scn,
const uint64_t table_id,
const int64_t ddl_task_id)
int ObTabletDDLKvMgr::ddl_prepare(
const SCN &start_scn,
const SCN &prepare_scn,
const ObIArray<uint64_t> &column_ids,
const uint64_t table_id,
const int64_t ddl_task_id)
{
int ret = OB_SUCCESS;
ObDDLKVHandle kv_handle;
@ -180,6 +182,9 @@ int ObTabletDDLKvMgr::ddl_prepare(const SCN &start_scn,
param.execution_id_ = execution_id_;
param.ddl_task_id_ = ddl_task_id_;
const int64_t start_ts = ObTimeUtility::fast_current_time();
if (OB_FAIL(param.column_ids_.assign(column_ids))) {
LOG_WARN("fail to assign columns ids", K(ret), K(column_ids));
}
while (OB_SUCC(ret) && is_started()) {
if (OB_FAIL(compaction::ObScheduleDagFunc::schedule_ddl_table_merge_dag(param))) {
if (OB_SIZE_OVERFLOW != ret && OB_EAGAIN != ret) {
@ -202,7 +207,13 @@ int ObTabletDDLKvMgr::ddl_prepare(const SCN &start_scn,
return ret;
}
int ObTabletDDLKvMgr::ddl_commit(const SCN &start_scn, const SCN &prepare_scn, const bool is_replay)
int ObTabletDDLKvMgr::ddl_commit(
const SCN &start_scn,
const SCN &prepare_scn,
const bool is_replay,
const uint64_t table_id,
const int64_t ddl_task_id,
const ObIArray<uint64_t> &column_ids)
{
int ret = OB_SUCCESS;
ObLSHandle ls_handle;
@ -217,6 +228,9 @@ int ObTabletDDLKvMgr::ddl_commit(const SCN &start_scn, const SCN &prepare_scn, c
} else if (OB_FAIL(MTL(ObLSService *)->get_ls(ls_id_, ls_handle, ObLSGetMod::DDL_MOD))) {
LOG_WARN("failed to get log stream", K(ret), K(ls_id_));
} else {
table_id_ = table_id;
ddl_task_id_ = ddl_task_id;
ObDDLTableMergeDagParam param;
param.ls_id_ = ls_id_;
param.tablet_id_ = tablet_id_;
@ -226,8 +240,9 @@ int ObTabletDDLKvMgr::ddl_commit(const SCN &start_scn, const SCN &prepare_scn, c
param.table_id_ = table_id_;
param.execution_id_ = execution_id_;
param.ddl_task_id_ = ddl_task_id_;
// retry submit dag in case of the previous dag failed
if (OB_FAIL(compaction::ObScheduleDagFunc::schedule_ddl_table_merge_dag(param))) {
if (OB_FAIL(param.column_ids_.assign(column_ids))) {
LOG_WARN("fail to assign columns ids", K(ret), K(column_ids));
} else if (OB_FAIL(compaction::ObScheduleDagFunc::schedule_ddl_table_merge_dag(param))) { // retry submit dag in case of the previous dag failed
if (OB_SIZE_OVERFLOW == ret || OB_EAGAIN == ret) {
ret = OB_EAGAIN;
} else {
@ -250,15 +265,21 @@ int ObTabletDDLKvMgr::ddl_commit(const SCN &start_scn, const SCN &prepare_scn, c
return ret;
}
int ObTabletDDLKvMgr::wait_ddl_commit(const SCN &start_scn, const SCN &prepare_scn)
int ObTabletDDLKvMgr::wait_ddl_commit(
const SCN &start_scn,
const SCN &prepare_scn,
const uint64_t table_id,
const int64_t ddl_task_id,
const ObIArray<uint64_t> &column_ids)
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(!is_inited_)) {
ret = OB_NOT_INIT;
LOG_WARN("not init", K(ret), K(is_inited_));
} else if (OB_UNLIKELY(!start_scn.is_valid_and_not_min() || !prepare_scn.is_valid_and_not_min())) {
} else if (OB_UNLIKELY(!start_scn.is_valid_and_not_min() || !prepare_scn.is_valid_and_not_min() ||
table_id <= 0 || ddl_task_id <= 0 || column_ids.count() <= 0)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(start_scn), K(prepare_scn));
LOG_WARN("invalid argument", K(ret), K(start_scn), K(prepare_scn), K(table_id), K(ddl_task_id), K(column_ids));
} else if (!is_started()) {
ret = OB_STATE_NOT_MATCH;
LOG_WARN("ddl not started", K(ret));
@ -268,12 +289,12 @@ int ObTabletDDLKvMgr::wait_ddl_commit(const SCN &start_scn, const SCN &prepare_s
} else {
const int64_t wait_start_ts = ObTimeUtility::fast_current_time();
while (OB_SUCC(ret)) {
if (OB_FAIL(ddl_commit(start_scn, prepare_scn, false/*is_replay*/))) {
if (OB_FAIL(ddl_commit(start_scn, prepare_scn, false/*is_replay*/, table_id, ddl_task_id, column_ids))) {
if (OB_EAGAIN == ret) {
ob_usleep(10L * 1000L);
ret = OB_SUCCESS; // retry
} else {
LOG_WARN("commit ddl log failed", K(ret), K(start_scn), K(prepare_scn), K(ls_id_), K(tablet_id_));
LOG_WARN("commit ddl log failed", K(ret), K(start_scn), K(prepare_scn), K(ls_id_), K(tablet_id_), K(column_ids));
}
} else {
break;