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

[FEAT MERGE] [CP] transfer partition && cancel transfer partition

Browse Source 
Co-authored-by: linqiucen <lqcgrace@outlook.com>
This commit is contained in:
maosy
2024-03-20 09:46:02 +00:00
committed by ob-robot
parent e69c06c5d4
commit 84c5698b2a
87 changed files with 7195 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files

295
src/sql/resolver/cmd/ob_alter_system_resolver.cpp
View File

@ -5912,6 +5912,28 @@ int ObResetConfigResolver::resolve(const ParseNode &parse_tree)
return ret;
}
int resolve_part_info(const ParseNode &parse_node, uint64_t &table_id, ObObjectID &object_id)
{
int ret = OB_SUCCESS;
table_id = OB_INVALID_ID;
object_id = OB_INVALID_OBJECT_ID;
if (OB_UNLIKELY(T_PARTITION_INFO != parse_node.type_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid parse node, type is not T_PARTITION_INFO", KR(ret), "type",
get_type_name(parse_node.type_));
} else if (2 != parse_node.num_child_
|| OB_ISNULL(parse_node.children_[0])
|| OB_ISNULL(parse_node.children_[1])) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid parse node", KR(ret), "num_child", parse_node.num_child_,
KP(parse_node.children_[0]), KP(parse_node.children_[1]));
} else {
table_id = parse_node.children_[0]->value_;
object_id = parse_node.children_[1]->value_;
}
return ret;
}
int ObAlterSystemResetResolver::resolve(const ParseNode &parse_tree)
{
int ret = OB_SUCCESS;
@ -6129,9 +6151,280 @@ int ObCancelCloneResolver::resolve(const ParseNode &parse_tree)
}
}
}
return ret;
}
int resolve_transfer_partition_to_ls(
const ParseNode &parse_node,
const uint64_t target_tenant_id,
const uint64_t exec_tenant_id,
ObTransferPartitionStmt *stmt)
{
int ret = OB_SUCCESS;
uint64_t table_id = OB_INVALID_ID;
ObObjectID object_id = OB_INVALID_OBJECT_ID;
if (OB_UNLIKELY(T_TRANSFER_PARTITION_TO_LS != parse_node.type_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid parse node, type is not T_TRANSFER_PARTITION_TO_LS", KR(ret), "type",
get_type_name(parse_node.type_));
} else if (OB_ISNULL(stmt)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("stmt is null", KR(ret), KP(stmt));
} else if (OB_UNLIKELY(!is_valid_tenant_id(exec_tenant_id) || !is_valid_tenant_id(target_tenant_id))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid exec_tenant_id or target_tenant_id", KR(ret), K(exec_tenant_id), K(target_tenant_id));
} else if (2 != parse_node.num_child_
|| OB_ISNULL(parse_node.children_[0])
|| OB_ISNULL(parse_node.children_[1])) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid parse node", KR(ret), "num_child", parse_node.num_child_,
KP(parse_node.children_[0]), KP(parse_node.children_[1]));
} else if (OB_FAIL(resolve_part_info(*parse_node.children_[0], table_id, object_id))) {
LOG_WARN("fail to resolve partition info", KR(ret), KP(parse_node.children_[0]));
} else {
int64_t id = parse_node.children_[1]->value_;
ObLSID ls_id(id);
if (OB_FAIL(stmt->get_arg().init_for_transfer_partition_to_ls(
target_tenant_id,
table_id,
object_id,
ls_id))) {
LOG_WARN("fail to init stmt rpc arg", KR(ret), K(target_tenant_id),
K(table_id), K(object_id), K(ls_id));
}
}
return ret;
}
int get_and_verify_tenant_name(
const ParseNode *parse_node,
const uint64_t exec_tenant_id,
uint64_t &target_tenant_id,
const char * const op_str)
{
int ret = OB_SUCCESS;
ObString tenant_name;
ObSchemaGetterGuard schema_guard;
int tmp_ret = OB_SUCCESS;
const int64_t COMMENT_LENGTH = 512;
char comment[COMMENT_LENGTH] = {0};
int64_t pos = 0;
if (OB_UNLIKELY(!is_valid_tenant_id(exec_tenant_id))
|| OB_ISNULL(op_str)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("exec tenant id is invalid", KR(ret), K(exec_tenant_id), K(op_str));
} else if (NULL == parse_node) {
if (OB_SYS_TENANT_ID != exec_tenant_id) {
target_tenant_id = exec_tenant_id;
} else {
ret = OB_NOT_SUPPORTED;
LOG_WARN("no specified tenant in the sys tenant's session", KR(ret));
if (OB_TMP_FAIL(databuff_printf(comment, COMMENT_LENGTH, pos,
"%s of SYS tenant is", op_str))) {
LOG_WARN("failed to printf to comment", KR(ret), KR(tmp_ret), K(op_str));
} else {
LOG_USER_ERROR(OB_NOT_SUPPORTED, comment);
}
}
} else if (OB_FAIL(resolve_tenant_name(parse_node, exec_tenant_id, tenant_name))) {
LOG_WARN("fail to resolve target tenant id", KR(ret));
} else if (OB_ISNULL(GCTX.schema_service_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("GCTX.schema_service_ is null", KR(ret), KP(GCTX.schema_service_));
} else if (OB_FAIL(GCTX.schema_service_->get_tenant_schema_guard(OB_SYS_TENANT_ID, schema_guard))) {
LOG_WARN("failed to get_tenant_schema_guard", KR(ret));
} else if (OB_FAIL(schema_guard.get_tenant_id(tenant_name, target_tenant_id))) {
LOG_WARN("failed to get tenant id from schema guard", KR(ret), K(tenant_name));
if (OB_TENANT_NOT_EXIST == ret || OB_ERR_INVALID_TENANT_NAME == ret) {
ret = OB_TENANT_NOT_EXIST;
LOG_USER_ERROR(OB_TENANT_NOT_EXIST, tenant_name.length(), tenant_name.ptr());
}
} else if (OB_UNLIKELY(!is_user_tenant(target_tenant_id))) {
ret = OB_NOT_SUPPORTED;
LOG_WARN("only support user tenant", KR(ret), K(target_tenant_id));
if (OB_TMP_FAIL(databuff_printf(comment, COMMENT_LENGTH, pos,
"%s of META or SYS tenant is", op_str))) {
LOG_WARN("failed to printf to comment", KR(ret), KR(tmp_ret), K(op_str));
} else {
LOG_USER_ERROR(OB_NOT_SUPPORTED, comment);
}
} else if (OB_SYS_TENANT_ID != exec_tenant_id && target_tenant_id != exec_tenant_id) {
ret = OB_ERR_NO_PRIVILEGE;
LOG_WARN("no support operating other user tenants", KR(ret), K(target_tenant_id), K(exec_tenant_id));
}
return ret;
}
int ObTransferPartitionResolver::resolve_transfer_partition_(const ParseNode &parse_tree)
{
int ret = OB_SUCCESS;
ObTransferPartitionStmt *stmt = create_stmt<ObTransferPartitionStmt>();
uint64_t target_tenant_id = OB_INVALID_TENANT_ID;
if (OB_UNLIKELY(T_TRANSFER_PARTITION != parse_tree.type_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid parse node, type is not T_TRANSFER_PARTITION", KR(ret), "type",
get_type_name(parse_tree.type_));
} else if (OB_ISNULL(stmt)) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("create stmt fail", KR(ret));
} else if (2 != parse_tree.num_child_ || OB_ISNULL(session_info_)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid parse tree or session info", KR(ret), "num_child", parse_tree.num_child_,
KP(session_info_));
} else if (OB_ISNULL(parse_tree.children_[0])) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("parse node is null", KR(ret), KP(parse_tree.children_[0]));
} else if (OB_FAIL(get_and_verify_tenant_name(
parse_tree.children_[1],
session_info_->get_effective_tenant_id(),
target_tenant_id, "Transfer partition"))) {
LOG_WARN("fail to execute get_and_verify_tenant_name", KR(ret),
K(session_info_->get_effective_tenant_id()), KP(parse_tree.children_[1]));
} else {
ParseNode *transfer_partition_node = parse_tree.children_[0];
switch(transfer_partition_node->type_) {
case T_TRANSFER_PARTITION_TO_LS:
if (OB_FAIL(resolve_transfer_partition_to_ls(
*transfer_partition_node,
target_tenant_id,
session_info_->get_effective_tenant_id(),
stmt))) {
LOG_WARN("fail to resolve transfer_partition_to_ls", KR(ret), K(target_tenant_id), K(session_info_->get_effective_tenant_id()));
}
break;
default:
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid transfer partition node type", KR(ret), "type", get_type_name(transfer_partition_node->type_));
}
if (OB_SUCC(ret)) {
stmt_ = stmt;
}
}
return ret;
}
int ObTransferPartitionResolver::resolve(const ParseNode &parse_tree)
{
int ret = OB_SUCCESS;
if (T_TRANSFER_PARTITION == parse_tree.type_) {
if (OB_FAIL(resolve_transfer_partition_(parse_tree))) {
LOG_WARN("failed to reslove transfer partition", KR(ret));
}
} else if (T_CANCEL_TRANSFER_PARTITION == parse_tree.type_) {
if (OB_FAIL(resolve_cancel_transfer_partition_(parse_tree))) {
LOG_WARN("failed to resolve cancel transfer partition", KR(ret));
}
} else if (T_CANCEL_BALANCE_JOB == parse_tree.type_) {
if (OB_FAIL(resolve_cancel_balance_job_(parse_tree))) {
LOG_WARN("failed to resolve cancel balance job", KR(ret));
}
} else {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid parse node, type is not T_TRANSFER_PARTITION", KR(ret), "type",
get_type_name(parse_tree.type_));
}
if (OB_SUCC(ret) && ObSchemaChecker::is_ora_priv_check()) {
if (OB_ISNULL(schema_checker_)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret));
} else if (OB_FAIL(schema_checker_->check_ora_ddl_priv(
session_info_->get_effective_tenant_id(),
session_info_->get_priv_user_id(),
ObString(""),
// why use T_ALTER_SYSTEM_SET_PARAMETER?
// because T_ALTER_SYSTEM_SET_PARAMETER has following traits:
// T_ALTER_SYSTEM_SET_PARAMETER can allow dba to do an operation
// and prohibit other user to do this operation
// so we reuse this.
stmt::T_ALTER_SYSTEM_SET_PARAMETER,
session_info_->get_enable_role_array()))) {
LOG_WARN("failed to check privilege", K(session_info_->get_effective_tenant_id()), K(session_info_->get_user_id()));
}
}
return ret;
}
int ObTransferPartitionResolver::resolve_cancel_transfer_partition_(const ParseNode &parse_tree)
{
int ret = OB_SUCCESS;
ObTransferPartitionStmt *stmt = create_stmt<ObTransferPartitionStmt>();
uint64_t target_tenant_id = OB_INVALID_TENANT_ID;
if (OB_UNLIKELY(T_CANCEL_TRANSFER_PARTITION != parse_tree.type_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid parse node, type is not T_TRANSFER_PARTITION", KR(ret), "type",
get_type_name(parse_tree.type_));
} else if (OB_ISNULL(stmt)) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("create stmt fail", KR(ret));
} else if (2 != parse_tree.num_child_ || OB_ISNULL(session_info_)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid parse tree or session info", KR(ret), "num_child", parse_tree.num_child_,
KP(session_info_));
} else if (OB_ISNULL(parse_tree.children_[0])) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("parse node is null", KR(ret), KP(parse_tree.children_[0]));
} else if (OB_FAIL(get_and_verify_tenant_name(
parse_tree.children_[1],
session_info_->get_effective_tenant_id(),
target_tenant_id, "Cancel transfer partition"))) {
LOG_WARN("fail to execute get_and_verify_tenant_name", KR(ret),
K(session_info_->get_effective_tenant_id()), KP(parse_tree.children_[1]));
} else {
ParseNode *transfer_partition_node = parse_tree.children_[0];
uint64_t table_id = OB_INVALID_ID;
ObObjectID object_id = OB_INVALID_OBJECT_ID;
rootserver::ObTransferPartitionArg::ObTransferPartitionType type = rootserver::ObTransferPartitionArg::INVALID_TYPE;
if (T_PARTITION_INFO == transfer_partition_node->type_) {
type = rootserver::ObTransferPartitionArg::CANCEL_TRANSFER_PARTITION;
if (OB_FAIL(resolve_part_info(*transfer_partition_node, table_id, object_id))) {
LOG_WARN("failed to resolve part info", KR(ret));
}
} else if (T_ALL == transfer_partition_node->type_) {
type = rootserver::ObTransferPartitionArg::CANCEL_TRANSFER_PARTITION_ALL;
} else {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid transfer partition node type", KR(ret), "type", get_type_name(transfer_partition_node->type_));
}
if (FAILEDx(stmt->get_arg().init_for_cancel_transfer_partition(
target_tenant_id, type, table_id, object_id))) {
LOG_WARN("fail to init stmt rpc arg", KR(ret), K(target_tenant_id), K(type),
K(table_id), K(object_id));
} else {
stmt_ = stmt;
}
}
return ret;
}
int ObTransferPartitionResolver::resolve_cancel_balance_job_(const ParseNode &parse_tree)
{
int ret = OB_SUCCESS;
ObTransferPartitionStmt *stmt = create_stmt<ObTransferPartitionStmt>();
uint64_t target_tenant_id = OB_INVALID_TENANT_ID;
if (OB_UNLIKELY(T_CANCEL_BALANCE_JOB != parse_tree.type_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid parse node, type is not T_TRANSFER_PARTITION", KR(ret), "type",
get_type_name(parse_tree.type_));
} else if (OB_ISNULL(stmt)) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("create stmt fail", KR(ret));
} else if (1 != parse_tree.num_child_
|| OB_ISNULL(session_info_)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid parse tree or session info", KR(ret), "num_child", parse_tree.num_child_,
KP(session_info_));
} else if (OB_FAIL(get_and_verify_tenant_name(
parse_tree.children_[0],
session_info_->get_effective_tenant_id(),
target_tenant_id, "Cancel balance job"))) {
LOG_WARN("fail to execute get_and_verify_tenant_name", KR(ret),
K(session_info_->get_effective_tenant_id()), KP(parse_tree.children_[0]));
} else if (OB_FAIL(stmt->get_arg().init_for_cancel_balance_job(
target_tenant_id))) {
LOG_WARN("fail to init stmt rpc arg", KR(ret), K(target_tenant_id));
} else {
stmt_ = stmt;
}
return ret;
}
} // end namespace sql
} // end namespace oceanbase

13
src/sql/resolver/cmd/ob_alter_system_resolver.h
View File

@ -162,6 +162,7 @@ DEF_SIMPLE_CMD_RESOLVER(ObAddRestoreSourceResolver);
DEF_SIMPLE_CMD_RESOLVER(ObClearRestoreSourceResolver);
DEF_SIMPLE_CMD_RESOLVER(ObCheckpointSlogResolver);
int resolve_restore_until(const ParseNode &time_node,
const ObSQLSessionInfo *session_info,
share::SCN &recovery_until_scn,
@ -223,7 +224,17 @@ private:
int check_param_valid(int64_t tenant_id,
const common::ObString &name_node, const common::ObString &value_node);
};
class ObTransferPartitionResolver : public ObSystemCmdResolver
{
public:
ObTransferPartitionResolver(ObResolverParams &params) : ObSystemCmdResolver(params) {}
virtual ~ObTransferPartitionResolver() {}
virtual int resolve(const ParseNode &parse_tree);
private:
int resolve_transfer_partition_(const ParseNode &parse_tree);
int resolve_cancel_transfer_partition_(const ParseNode &parse_tree);
int resolve_cancel_balance_job_(const ParseNode &parse_tree);
};
class ObFreezeResolver : public ObSystemCmdResolver {
public:
ObFreezeResolver(ObResolverParams &params) : ObSystemCmdResolver(params) {}

13
src/sql/resolver/cmd/ob_alter_system_stmt.h
View File

@ -17,6 +17,7 @@
#include "share/ob_rpc_struct.h"
#include "share/scheduler/ob_sys_task_stat.h"
#include "share/backup/ob_backup_clean_struct.h"
#include "rootserver/ob_transfer_partition_command.h"
namespace oceanbase
{
@ -1364,6 +1365,18 @@ public:
private:
common::ObFixedLengthString<common::OB_MAX_TENANT_NAME_LENGTH + 1> clone_tenant_name_;
};
class ObTransferPartitionStmt : public ObSystemCmdStmt
{
public:
ObTransferPartitionStmt()
: ObSystemCmdStmt(stmt::T_TRANSFER_PARTITION),
arg_() {}
virtual ~ObTransferPartitionStmt() {}
rootserver::ObTransferPartitionArg &get_arg() { return arg_; }
private:
rootserver::ObTransferPartitionArg arg_;
};
} // end namespace sql
} // end namespace oceanbase

12
src/sql/resolver/ob_resolver.cpp
View File

@ -1185,6 +1185,18 @@ int ObResolver::resolve(IsPrepared if_prepared, const ParseNode &parse_tree, ObS
REGISTER_STMT_RESOLVER(CancelClone);
break;
}
case T_TRANSFER_PARTITION: {
REGISTER_STMT_RESOLVER(TransferPartition);
break;
}
case T_CANCEL_TRANSFER_PARTITION: {
REGISTER_STMT_RESOLVER(TransferPartition);
break;
}
case T_CANCEL_BALANCE_JOB: {
REGISTER_STMT_RESOLVER(TransferPartition);
break;
}
default: {
ret = OB_NOT_SUPPORTED;
const char *type_name = get_type_name(parse_tree.type_);

2
src/sql/resolver/ob_stmt_type.h
View File

@ -294,8 +294,8 @@ OB_STMT_TYPE_DEF_UNKNOWN_AT(T_CLONE_TENANT, get_sys_tenant_alter_system_priv, 29
OB_STMT_TYPE_DEF_UNKNOWN_AT(T_CANCEL_CLONE, get_sys_tenant_alter_system_priv, 294)
OB_STMT_TYPE_DEF_UNKNOWN_AT(T_CREATE_MLOG, get_create_mlog_stmt_need_privs, 295)
OB_STMT_TYPE_DEF_UNKNOWN_AT(T_DROP_MLOG, get_drop_mlog_stmt_need_privs, 296)
OB_STMT_TYPE_DEF_UNKNOWN_AT(T_TRANSFER_PARTITION, get_sys_tenant_alter_system_priv, 297)
OB_STMT_TYPE_DEF_UNKNOWN_AT(T_FLUSH_PRIVILEGES, no_priv_needed, 298)
// OB_STMT_TYPE_DEF_UNKNOWN_AT(T_TRANSFER_PARTITION, get_sys_tenant_alter_system_priv, 297)
// OB_STMT_TYPE_DEF_UNKNOWN_AT(T_ALTER_LS_REPLICA, get_sys_tenant_alter_system_priv, 299)
OB_STMT_TYPE_DEF_UNKNOWN_AT(T_SHOW_PROCEDURE_CODE, err_stmt_type_priv, 300)
OB_STMT_TYPE_DEF_UNKNOWN_AT(T_SHOW_FUNCTION_CODE, err_stmt_type_priv, 301)