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

[FEAT MERGE] Support exchanging partition between partitioned table A and non-partitioned table B.

Browse Source 
Co-authored-by: fkuner <784819644@qq.com>
This commit is contained in:
791065426@qq.com 2024-04-16 07:19:48 +00:00 committed by ob-robot
parent 2c510c6538
commit 1ee64365ea
46 changed files with 4105 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
deps/oblib/src/common/rowkey/ob_rowkey_info.cpp vendored
View File

@ -70,6 +70,14 @@ DEFINE_GET_SERIALIZE_SIZE(ObRowkeyColumn)
len += serialization::encoded_length_vi32(order_);
return len;
}
bool ObRowkeyColumn::is_equal_except_column_id(const ObRowkeyColumn &other) const
{
return length_ == other.length_ &&
type_ == other.type_ &&
order_ == other.order_ &&
fulltext_flag_ == other.fulltext_flag_ &&
spatial_flag_ == other.spatial_flag_;
}
ObRowkeyColumn& ObRowkeyColumn::operator=(const ObRowkeyColumn &other)
{
@ -85,12 +93,8 @@ ObRowkeyColumn& ObRowkeyColumn::operator=(const ObRowkeyColumn &other)
bool ObRowkeyColumn::operator==(const ObRowkeyColumn &other) const
{
return
this->length_ == other.length_ &&
this->column_id_ == other.column_id_ &&
this->type_ == other.type_ &&
this->order_ == other.order_ &&
this->fulltext_flag_ == other.fulltext_flag_ &&
this->spatial_flag_ == other.spatial_flag_;
this->is_equal_except_column_id(other);
}
ObRowkeyInfo::ObRowkeyInfo()

1
deps/oblib/src/common/rowkey/ob_rowkey_info.h vendored
View File

@ -36,6 +36,7 @@ struct ObRowkeyColumn
return (ObURowIDType == type_.get_type() || length_ >= 0) && common::OB_INVALID_ID != column_id_
&& common::ob_is_valid_obj_type(static_cast<ObObjType>(type_.get_type()));
}
bool is_equal_except_column_id(const ObRowkeyColumn &other) const;
ObRowkeyColumn &operator=(const ObRowkeyColumn &other);
bool operator==(const ObRowkeyColumn &other) const;
const ObObjMeta get_meta_type() const { return type_; }

3
src/logservice/libobcdc/src/ob_cdc_multi_data_source_info.h
View File

@ -53,7 +53,8 @@ public:
inline bool is_tablet_change_node() const
{
return transaction::ObTxDataSourceType::CREATE_TABLET_NEW_MDS == tx_buf_node_.get_data_source_type()
|| transaction::ObTxDataSourceType::DELETE_TABLET_NEW_MDS == tx_buf_node_.get_data_source_type();
|| transaction::ObTxDataSourceType::DELETE_TABLET_NEW_MDS == tx_buf_node_.get_data_source_type()
|| transaction::ObTxDataSourceType::CHANGE_TABLET_TO_TABLE_MDS == tx_buf_node_.get_data_source_type();
}
inline bool is_ddl_trans_node() const

139
src/logservice/libobcdc/src/ob_cdc_tablet_to_table_info.cpp
View File

@ -15,7 +15,8 @@
#define USING_LOG_PREFIX OBLOG_PARSER
#include "ob_cdc_tablet_to_table_info.h"
#include "ob_log_hbase_mode.h"
#include "rootserver/ob_partition_exchange.h"
#include "share/ob_errno.h"
namespace oceanbase
@ -55,8 +56,33 @@ bool DeleteTabletOp::is_valid() const
return tablet_id_.is_valid();
}
void ExchangeTabletOp::reset()
{
tablet_ids_.reset();
table_ids_.reset();
}
bool ExchangeTabletOp::is_valid() const
{
return !tablet_ids_.empty() && !table_ids_.empty();
}
int ExchangeTabletOp::push_back_tablet_to_table(const common::ObTabletID &tablet_id, const uint64_t table_id)
{
int ret = OB_SUCCESS;
if (OB_FAIL(tablet_ids_.push_back(tablet_id))) {
LOG_ERROR("tablet_ids push_back fail", KR(ret), K_(tablet_ids), "count", tablet_ids_.count());
} else if (OB_FAIL(table_ids_.push_back(table_id))) {
LOG_ERROR("table_ids push_back fail", KR(ret), K_(table_ids), "count", table_ids_.count());
} else {
// succ
}
return ret;
}
ObCDCTabletChangeInfo::ObCDCTabletChangeInfo() :
cmd_(TabletChangeCmd::CMD_UNKNOWN), create_tablet_op_arr_(), delete_tablet_op_arr_()
cmd_(TabletChangeCmd::CMD_UNKNOWN), create_tablet_op_arr_(), delete_tablet_op_arr_(), exchange_tablet_op_arr_()
{}
void ObCDCTabletChangeInfo::reset()
@ -64,6 +90,7 @@ void ObCDCTabletChangeInfo::reset()
cmd_ = TabletChangeCmd::CMD_MAX;
create_tablet_op_arr_.reset();
delete_tablet_op_arr_.reset();
exchange_tablet_op_arr_.reset();
}
void ObCDCTabletChangeInfo::reset(const TabletChangeCmd cmd)
@ -112,6 +139,19 @@ int ObCDCTabletChangeInfo::parse_from_multi_data_source_buf(
}
break;
}
case transaction::ObTxDataSourceType::CHANGE_TABLET_TO_TABLE_MDS:
{
rootserver::ObChangeTabletToTableArg exchange_tablet_arg;
if (OB_FAIL(exchange_tablet_arg.deserialize(buf, buf_len, pos))) {
LOG_ERROR("deserialize exchange_tablet_arg failed", KR(ret), K(tls_id), K(multi_data_source_node),
K(exchange_tablet_arg), K(buf_len), K(pos));
} else if (OB_FAIL(parse_exchange_tablet_op_(tls_id, exchange_tablet_arg))) {
LOG_ERROR("parse_exchange_tablet_op_ failed", KR(ret), K(tls_id), K(multi_data_source_node),
K(exchange_tablet_arg), KPC(this));
}
break;
}
default:
{
ret = OB_NOT_SUPPORTED;
@ -219,6 +259,42 @@ int ObCDCTabletChangeInfo::parse_remove_tablet_op_(
return ret;
}
int ObCDCTabletChangeInfo::parse_exchange_tablet_op_(
const logservice::TenantLSID &tls_id,
const rootserver::ObChangeTabletToTableArg &exchange_tablet_arg)
{
int ret = OB_SUCCESS;
cmd_ = TabletChangeCmd::CMD_EXCHANGE;
if (OB_UNLIKELY(! exchange_tablet_arg.is_valid())
|| OB_UNLIKELY(tls_id.get_ls_id() != exchange_tablet_arg.ls_id_)) {
ret = OB_ERR_UNEXPECTED;
LOG_ERROR("ObChangeTabletToTableArg is invalid", KR(ret), K(tls_id), K(exchange_tablet_arg));
} else {
const common::ObSArray<common::ObTabletID> &tablet_ids = exchange_tablet_arg.tablet_ids_;
const common::ObSArray<uint64_t> &table_ids = exchange_tablet_arg.table_ids_;
ExchangeTabletOp exchange_tablet_op;
ARRAY_FOREACH_N(tablet_ids, tablet_id_idx, count) {
const common::ObTabletID &tablet_id = tablet_ids.at(tablet_id_idx);
const uint64_t table_id = table_ids.at(tablet_id_idx);
if (OB_FAIL(exchange_tablet_op.push_back_tablet_to_table(tablet_id, table_id))) {
LOG_ERROR("exchange_tablet_op push back failed", KR(ret), K(tablet_id), K(table_id));
}
}
if (OB_SUCC(ret)) {
if (OB_FAIL(push_exchange_tablet_op_(exchange_tablet_op))) {
LOG_ERROR("push_exchange_tablet_op failed", KR(ret), K(tls_id), K(exchange_tablet_op), KPC(this));
} else {
LOG_DEBUG("[EXCHANGE_TABLET_INFO]", K(tls_id), K(exchange_tablet_op));
}
}
}
return ret;
}
int ObCDCTabletChangeInfo::push_create_tablet_op_(const CreateTabletOp &create_tablet_op)
{
int ret = OB_SUCCESS;
@ -253,12 +329,31 @@ int ObCDCTabletChangeInfo::push_delete_tablet_op_(const DeleteTabletOp &delete_t
return ret;
}
int ObCDCTabletChangeInfo::push_exchange_tablet_op_(const ExchangeTabletOp &exchange_tablet_op)
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(!exchange_tablet_op.is_valid())) {
ret = OB_INVALID_ARGUMENT;
LOG_ERROR("invalid arg for push_exchange_tablet_op", KR(ret), K(exchange_tablet_op));
} else if (OB_FAIL(exchange_tablet_op_arr_.push_back(exchange_tablet_op))) {
LOG_ERROR("push_exchange_tablet_op into exchange_tablet_op_arr_ failed",
KR(ret), K(exchange_tablet_op), K_(exchange_tablet_op_arr));
} else {
// success
}
return ret;
}
void ObCDCTabletChangeInfo::print_detail_for_debug() const
{
if (is_create_tablet_op()) {
LOG_DEBUG("tablet_change_info", "create_cnt", create_tablet_op_arr_.count(), K_(create_tablet_op_arr));
} else if (is_delete_tablet_op()) {
LOG_DEBUG("tablet_change_info", "delete_cnt", delete_tablet_op_arr_.count(), K_(delete_tablet_op_arr));
} else if (is_exchange_tablet_op()) {
LOG_DEBUG("tablet_change_info", K_(exchange_tablet_op_arr));
} else {
LOG_DEBUG("tablet_change_info: None");
}
@ -360,6 +455,46 @@ int TabletToTableInfo::insert_tablet_table_info(const common::ObTabletID &tablet
return ret;
}
int TabletToTableInfo::exchange_tablet_table_info(const common::ObSArray<common::ObTabletID> &tablet_ids, const common::ObSArray<uint64_t> &table_ids)
{
int ret = OB_SUCCESS;
common::ObLinearHashMap<TableID, ObCDCTableInfo> table_to_table_info_map;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
LOG_ERROR("TabletIDToTableIDInfo is not inited", KR(ret), K_(is_inited));
} else if (OB_FAIL(table_to_table_info_map.init("TableToTable"))) {
LOG_ERROR("table_to_table_info_map init failed", KR(ret));
} else {
ARRAY_FOREACH_N(tablet_ids, idx, count) {
const common::ObTabletID &tablet_id = tablet_ids.at(idx);
ObCDCTableInfo tmp_table_info;
if (OB_FAIL(tablet_to_table_map_.get(tablet_id, tmp_table_info))) {
LOG_ERROR("tablet_to_table_map_ get failed", KR(ret), K(tablet_id));
} else if (OB_FAIL(table_to_table_info_map.insert(TableID(tmp_table_info.get_table_id()), tmp_table_info))) {
LOG_ERROR("table_to_table_info_map insert failed", KR(ret), K(tablet_id), K(tmp_table_info));
} else {
LOG_INFO("table_to_table_info_map insert success", K(tablet_id), K(tmp_table_info));
}
}
ARRAY_FOREACH_N(tablet_ids, idx, count) {
const common::ObTabletID &tablet_id = tablet_ids.at(idx);
const uint64_t table_id = table_ids.at(idx);
ObCDCTableInfo tmp_table_info;
if (OB_FAIL(table_to_table_info_map.get(TableID(table_id), tmp_table_info))) {
LOG_ERROR("table_to_table_info_map get failed", KR(ret), K(table_id));
} else if (OB_FAIL(tablet_to_table_map_.insert_or_update(tablet_id, tmp_table_info))) {
LOG_ERROR("tablet_to_table_map_ update failed", KR(ret), K(tablet_id), K(tmp_table_info));
} else {
LOG_INFO("tablet_to_table_map_ update success", K(tablet_id), K(table_id), K(tmp_table_info));
}
}
}
return ret;
}
int TabletToTableInfo::remove_tablet_table_info(const common::ObTabletID &tablet_id)
{
int ret = OB_SUCCESS;

68
src/logservice/libobcdc/src/ob_cdc_tablet_to_table_info.h
View File

@ -21,11 +21,16 @@
#include "storage/tx/ob_multi_data_source.h" // ObTxBufferNode
#include "rootserver/ob_tablet_creator.h" // ObBatchCreateTabletArg
#include "rootserver/ob_tablet_drop.h" // ObBatchRemoveTabletArg
#include "logservice/common_util/ob_log_ls_define.h"
namespace oceanbase
{
namespace rootserver
{
class ObChangeTabletToTableArg;
}
namespace libobcdc
{
// TabletChangeOp Type
@ -35,6 +40,7 @@ enum TabletChangeCmd
CMD_CREATE,
CMD_DELETE,
CMD_TRANSFER,
CMD_EXCHANGE,
CMD_MAX
};
@ -118,6 +124,30 @@ private:
common::ObTabletID tablet_id_;
};
// data used for ExchangeTablet
class ExchangeTabletOp
{
public:
ExchangeTabletOp() { reset(); }
~ExchangeTabletOp() { reset(); }
void reset();
public:
bool is_valid() const;
int push_back_tablet_to_table(const common::ObTabletID &tablet_id, const uint64_t table_id);
const common::ObSArray<common::ObTabletID> &get_tablet_ids() const { return tablet_ids_; }
const common::ObSArray<uint64_t> &get_table_ids() const { return table_ids_; }
public:
TO_STRING_KV(
K_(tablet_ids),
"tablet_ids_count", tablet_ids_.count(),
K_(table_ids),
"table_id_count", table_ids_.count());
private:
// use array to also store related table, such as lob aux table
common::ObSArray<common::ObTabletID> tablet_ids_;
common::ObSArray<uint64_t> table_ids_;
};
// struct store the TabletChangeOp that get from LS_MEMTABLE mulit_data_source trans
class ObCDCTabletChangeInfo
{
@ -133,18 +163,22 @@ public:
public:
inline bool is_valid() const
{
return (TabletChangeCmd::CMD_CREATE == cmd_ && 0 >= delete_tablet_op_arr_.count()) // create_tablet_op_arr_ may empty after filter tablet.
|| (TabletChangeCmd::CMD_DELETE == cmd_ && 0 >= create_tablet_op_arr_.count()); // delete_tablet_op_arri_ may empty.
return (TabletChangeCmd::CMD_CREATE == cmd_ && 0 >= delete_tablet_op_arr_.count() && 0 >= exchange_tablet_op_arr_.count()) // create_tablet_op_arr_ may empty after filter tablet.
|| (TabletChangeCmd::CMD_DELETE == cmd_ && 0 >= create_tablet_op_arr_.count() && 0 >= exchange_tablet_op_arr_.count()) // delete_tablet_op_arr_ may empty.
|| (TabletChangeCmd::CMD_EXCHANGE == cmd_ && 0 >= create_tablet_op_arr_.count() && 0 >= delete_tablet_op_arr_.count()); // exchange_tablet_op_arr_ may empty.
}
inline bool is_create_tablet_op() const { return TabletChangeCmd::CMD_CREATE == cmd_; }
inline bool is_delete_tablet_op() const { return TabletChangeCmd::CMD_DELETE == cmd_; }
const ObArray<CreateTabletOp> &get_create_tablet_op_arr() const { return create_tablet_op_arr_; }
const ObArray<DeleteTabletOp> &get_delete_tablet_op_arr() const { return delete_tablet_op_arr_; }
inline bool is_exchange_tablet_op() const { return TabletChangeCmd::CMD_EXCHANGE == cmd_; }
const ObArray<CreateTabletOp> &get_create_tablet_op_arr() const { return create_tablet_op_arr_; }
const ObArray<DeleteTabletOp> &get_delete_tablet_op_arr() const { return delete_tablet_op_arr_; }
const ObArray<ExchangeTabletOp> &get_exchange_tablet_op_arr() const { return exchange_tablet_op_arr_; }
public:
void print_detail_for_debug() const;
TO_STRING_KV(K_(cmd),
"create_tablet_cnt", create_tablet_op_arr_.count(),
"delete_tablet_cnt", delete_tablet_op_arr_.count());
"create_tablet_cnt", create_tablet_op_arr_.count(),
"delete_tablet_cnt", delete_tablet_op_arr_.count(),
"exchange_tablet_cnt", exchange_tablet_op_arr_.count());
private:
int parse_create_tablet_op_(
const logservice::TenantLSID &tls_id,
@ -152,12 +186,17 @@ private:
int parse_remove_tablet_op_(
const logservice::TenantLSID &tls_id,
const obrpc::ObBatchRemoveTabletArg &remove_tablet_arg);
int parse_exchange_tablet_op_(
const logservice::TenantLSID &tls_id,
const rootserver::ObChangeTabletToTableArg &exchange_tablet_arg);
int push_create_tablet_op_(const CreateTabletOp &create_tablet_op);
int push_delete_tablet_op_(const DeleteTabletOp &delete_tablet_op);
int push_exchange_tablet_op_(const ExchangeTabletOp &exchange_tablet_op);
private:
TabletChangeCmd cmd_;
ObArray<CreateTabletOp> create_tablet_op_arr_;
ObArray<DeleteTabletOp> delete_tablet_op_arr_;
ObArray<CreateTabletOp> create_tablet_op_arr_;
ObArray<DeleteTabletOp> delete_tablet_op_arr_;
ObArray<ExchangeTabletOp> exchange_tablet_op_arr_;
};
typedef common::ObLinearHashMap<common::ObTabletID, ObCDCTableInfo> TabletToTableMap; // Map of TabletID->TableID
@ -196,6 +235,17 @@ public:
/// @retval other ERROR unexpected error.
int insert_tablet_table_info(const common::ObTabletID &tablet_id, const ObCDCTableInfo &table_info);
/// exchange tablet_id->table_info
///
/// @param [in] tablet_id tablet_id to insert
/// @param [in] table_info table_info of tablet_id
///
/// @retval OB_SUCCESS replace success
/// @retval OB_INVALID_ARGUMENT tablet_id is invalid
/// @retval OB_ENTRY_NOT_EXIST the tablet_id is not in tablet_to_table_map_
/// @retval other ERROR unexpected error
int exchange_tablet_table_info(const common::ObSArray<common::ObTabletID> &tablet_ids, const common::ObSArray<uint64_t> &table_ids);
/// remove tablet_id->table_info pair for specified tablet_id
///
/// @param [in] tablet_id tablet_id to remove

32
src/logservice/libobcdc/src/ob_log_part_mgr.cpp
View File

@ -1345,6 +1345,38 @@ int ObLogPartMgr::apply_delete_tablet_change(const ObCDCTabletChangeInfo &tablet
return ret;
}
int ObLogPartMgr::apply_exchange_tablet_change(const ObCDCTabletChangeInfo &tablet_change_info)
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(! tablet_change_info.is_valid())
|| OB_UNLIKELY(! tablet_change_info.is_exchange_tablet_op())) {
ret = OB_INVALID_ARGUMENT;
LOG_ERROR("invalid tablet_change_info for exchange_tablet_op", KR(ret), K(tablet_change_info));
} else {
const ObArray<ExchangeTabletOp> &exchange_tablet_op_arr = tablet_change_info.get_exchange_tablet_op_arr();
ARRAY_FOREACH_N(exchange_tablet_op_arr, idx, count) {
const ExchangeTabletOp &exchange_tablet_op = exchange_tablet_op_arr.at(idx);
if (OB_UNLIKELY(! exchange_tablet_op.is_valid())) {
ret = OB_ERR_UNEXPECTED;
LOG_ERROR("exchange_tablet_op is invalid", KR(ret), K(exchange_tablet_op));
} else {
const common::ObSArray<common::ObTabletID> &tablet_ids = exchange_tablet_op.get_tablet_ids();
const common::ObSArray<uint64_t> &table_ids = exchange_tablet_op.get_table_ids();
if (OB_FAIL(tablet_to_table_info_.exchange_tablet_table_info(tablet_ids, table_ids))) {
LOG_ERROR("tablet_to_table_info_ exchange table info failed", K(tablet_ids), K(table_ids));
} else {
LOG_INFO("apply_exchange_tablet_change success", K(tablet_ids), K(table_ids));
}
}
}
}
return ret;
}
// @retval OB_SUCCESS success
// @retval OB_TIMEOUT timeout
// @retval OB_TENANT_HAS_BEEN_DROPPED caller should ignore error code if schema error like tenant/database not exist

2
src/logservice/libobcdc/src/ob_log_part_mgr.h
View File

@ -257,6 +257,7 @@ public:
virtual int insert_table_id_into_cache(const uint64_t table_id, const uint64_t database_id) = 0;
virtual int delete_table_id_from_cache(const uint64_t table_id) = 0;
virtual int delete_db_from_cache(const uint64_t database_id) = 0;
virtual int apply_exchange_tablet_change(const ObCDCTabletChangeInfo &tablet_change_info) = 0;
};
/////////////////////////////////////////////////////////////////////////////
@ -368,6 +369,7 @@ public:
virtual int insert_table_id_into_cache(const uint64_t table_id, const uint64_t database_id);
virtual int delete_table_id_from_cache(const uint64_t table_id);
virtual int delete_db_from_cache(const uint64_t database_id);
virtual int apply_exchange_tablet_change(const ObCDCTabletChangeInfo &tablet_change_info);
private:
template<class TableMeta>

1
src/logservice/libobcdc/src/ob_log_part_trans_task.cpp
View File

@ -2576,6 +2576,7 @@ int PartTransTask::push_multi_data_source_data(
}
case transaction::ObTxDataSourceType::CREATE_TABLET_NEW_MDS:
case transaction::ObTxDataSourceType::DELETE_TABLET_NEW_MDS:
case transaction::ObTxDataSourceType::CHANGE_TABLET_TO_TABLE_MDS:
{
if (! is_commit_log) {
if (OB_FAIL(alloc_and_save_multi_data_source_node_(lsn, mds_buffer_node))) {

10
src/logservice/libobcdc/src/ob_log_sequencer1.cpp
View File

@ -1030,6 +1030,16 @@ int ObLogSequencer::handle_multi_data_source_info_(
} else {
LOG_DEBUG("CDC_DELETE_TABLET", KR(ret), K(tablet_change_info), K(part_trans_task), KPC(part_trans_task), K(tenant));
}
} else if (tablet_change_info.is_exchange_tablet_op()) {
if (OB_FAIL(wait_until_parser_done_("exchange_tablet_op", stop_flag))) {
if (OB_IN_STOP_STATE != ret) {
LOG_ERROR("wait_until_parser_done_ failed", KR(ret), KPC(part_trans_task));
}
} else if (OB_FAIL(part_mgr.apply_exchange_tablet_change(tablet_change_info))) {
LOG_ERROR("apply_exchange_tablet_change failed", KR(ret), K(tablet_change_info), K(tenant), KPC(part_trans_task));
} else {
LOG_INFO("CDC_EXCHANGE_TABLET", KR(ret), K(tablet_change_info), K(part_trans_task), KPC(part_trans_task), K(tenant));
}
}
}
}

2
src/objit/include/objit/common/ob_item_type.h
View File

@ -2428,14 +2428,12 @@ typedef enum ObItemType
T_SHOW_FUNCTION_CODE,
T_CHANGE_EXTERNAL_STORAGE_DEST,
T_ALTER_USER_PROXY,
T_PARALLEL_DAS_DML,
T_DISABLE_PARALLEL_DAS_DML,
T_ENABLE_LOB_PREFETCH,
T_MV_OPTIONS,
T_MV_REWRITE,
T_MV_NO_REWRITE,
// select into outfile
T_INTO_FILE_LIST,
T_SINGLE_OPT,

1
src/observer/ob_srv_xlator_rootserver.cpp
View File

@ -106,6 +106,7 @@ void oceanbase::observer::init_srv_xlator_for_rootserver(ObSrvRpcXlator *xlator)
RPC_PROCESSOR(rootserver::ObRpcUpdateDDLTaskActiveTimeP, *gctx_.root_service_);
RPC_PROCESSOR(rootserver::ObRpcCreateHiddenTableP, *gctx_.root_service_);
RPC_PROCESSOR(rootserver::ObRpcAlterTableP, *gctx_.root_service_);
RPC_PROCESSOR(rootserver::ObRpcExchangePartitionP, *gctx_.root_service_);
RPC_PROCESSOR(rootserver::ObRpcDropTableP, *gctx_.root_service_);
RPC_PROCESSOR(rootserver::ObRpcRenameTableP, *gctx_.root_service_);
RPC_PROCESSOR(rootserver::ObRpcTruncateTableP, *gctx_.root_service_);

1
src/rootserver/CMakeLists.txt
View File

@ -101,6 +101,7 @@ ob_set_subtarget(ob_rootserver common
ob_ls_recovery_stat_handler.cpp
ob_shrink_expand_resource_pool_checker.cpp
ob_transfer_partition_command.cpp
ob_partition_exchange.cpp
)
ob_set_subtarget(ob_rootserver balance

177
src/rootserver/ob_ddl_operator.cpp
View File

@ -12022,6 +12022,183 @@ int ObDDLOperator::try_add_dep_info_for_synonym(const ObSimpleSynonymSchema *syn
return ret;
}
int ObDDLOperator::exchange_table_partitions(const share::schema::ObTableSchema &orig_table_schema,
share::schema::ObTableSchema &inc_table_schema,
share::schema::ObTableSchema &del_table_schema,
common::ObMySQLTransaction &trans)
{
int ret = OB_SUCCESS;
const uint64_t tenant_id = orig_table_schema.get_tenant_id();
int64_t new_schema_version = OB_INVALID_VERSION;
ObSchemaService *schema_service = schema_service_.get_schema_service();
if (OB_ISNULL(schema_service)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("schema_service is NULL", K(ret));
} else if (OB_FAIL(schema_service_.gen_new_schema_version(tenant_id, new_schema_version))) {
LOG_WARN("fail to gen new schema_version", K(ret), K(tenant_id));
} else if (OB_FAIL(schema_service->get_table_sql_service().exchange_part_info(
trans,
orig_table_schema,
inc_table_schema,
del_table_schema,
new_schema_version))) {
LOG_WARN("exchange part info failed", K(ret));
}
return ret;
}
int ObDDLOperator::exchange_table_subpartitions(const share::schema::ObTableSchema &orig_table_schema,
share::schema::ObTableSchema &inc_table_schema,
share::schema::ObTableSchema &del_table_schema,
common::ObMySQLTransaction &trans)
{
int ret = OB_SUCCESS;
const uint64_t tenant_id = orig_table_schema.get_tenant_id();
int64_t new_schema_version = OB_INVALID_VERSION;
ObSchemaService *schema_service = schema_service_.get_schema_service();
if (OB_ISNULL(schema_service)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("schema_service is NULL", K(ret));
} else if (OB_FAIL(schema_service_.gen_new_schema_version(tenant_id, new_schema_version))) {
LOG_WARN("fail to gen new schema_version", K(ret), K(tenant_id));
} else if (OB_FAIL(schema_service->get_table_sql_service().exchange_subpart_info(
trans,
orig_table_schema,
inc_table_schema,
del_table_schema,
new_schema_version))) {
LOG_WARN("delete inc part info failed", K(ret));
}
return ret;
}
int ObDDLOperator::get_target_auto_inc_sequence_value(const uint64_t tenant_id,
const uint64_t table_id,
const uint64_t column_id,
uint64_t &sequence_value,
common::ObMySQLTransaction &trans)
{
int ret = OB_SUCCESS;
sequence_value = OB_INVALID_ID;
if (OB_UNLIKELY(OB_INVALID_TENANT_ID == tenant_id || OB_INVALID_ID == table_id || OB_INVALID_ID == column_id)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(tenant_id), K(table_id), K(column_id));
} else {
ObSqlString sql;
const uint64_t exec_tenant_id = tenant_id;
const char *table_name = OB_ALL_AUTO_INCREMENT_TNAME;
if (OB_FAIL(sql.assign_fmt(" SELECT sequence_value FROM %s WHERE tenant_id = %lu AND sequence_key = %lu"
" AND column_id = %lu FOR UPDATE",
table_name,
ObSchemaUtils::get_extract_tenant_id(exec_tenant_id, tenant_id),
ObSchemaUtils::get_extract_schema_id(exec_tenant_id, table_id),
column_id))) {
LOG_WARN("failed to assign sql", K(ret), K(tenant_id), K(table_id), K(column_id));
} else {
SMART_VAR(ObMySQLProxy::MySQLResult, res) {
ObMySQLResult *result = NULL;
uint64_t sequence_table_id = OB_ALL_AUTO_INCREMENT_TID;
if (OB_FAIL(trans.read(res, exec_tenant_id, sql.ptr()))) {
LOG_WARN("failed to read data", K(ret));
} else if (NULL == (result = res.get_result())) {
LOG_WARN("failed to get result", K(ret));
ret = OB_ERR_UNEXPECTED;
} else if (OB_FAIL(result->next())) {
LOG_WARN("failed to get next", K(ret));
if (OB_ITER_END == ret) {
// auto-increment column has been deleted
ret = OB_SCHEMA_ERROR;
LOG_WARN("failed to get next", K(ret));
}
} else if (OB_FAIL(result->get_uint("sequence_value", sequence_value))) {
LOG_WARN("failed to get int_value.", K(ret));
}
if (OB_SUCC(ret)) {
int tmp_ret = OB_SUCCESS;
if (OB_ITER_END != (tmp_ret = result->next())) {
if (OB_SUCCESS == tmp_ret) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("more than one row", K(ret), K(tenant_id), K(table_id), K(column_id));
} else {
ret = tmp_ret;
LOG_WARN("fail to iter next row", K(ret), K(tenant_id), K(table_id), K(column_id));
}
}
}
}
}
}
return ret;
}
int ObDDLOperator::set_target_auto_inc_sync_value(const uint64_t tenant_id,
const uint64_t table_id,
const uint64_t column_id,
const uint64_t new_sequence_value,
const uint64_t new_sync_value,
common::ObMySQLTransaction &trans)
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(OB_INVALID_TENANT_ID == tenant_id || OB_INVALID_ID == table_id || OB_INVALID_ID == column_id || new_sequence_value < 0 || new_sync_value < 0)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(tenant_id), K(table_id), K(column_id), K(new_sequence_value), K(new_sync_value));
} else {
ObSqlString sql;
int64_t affected_rows = 0;
const char *table_name = OB_ALL_AUTO_INCREMENT_TNAME;
if (OB_FAIL(sql.assign_fmt(
"UPDATE %s SET sequence_value = %lu, sync_value = %lu WHERE tenant_id=%lu AND sequence_key=%lu AND column_id=%lu",
table_name, new_sequence_value, new_sync_value,
ObSchemaUtils::get_extract_tenant_id(tenant_id, tenant_id), ObSchemaUtils::get_extract_schema_id(tenant_id, table_id), column_id))) {
LOG_WARN("failed to assign sql", K(ret), K(tenant_id), K(table_id), K(column_id), K(new_sequence_value), K(new_sync_value));
} else if (OB_FAIL(trans.write(tenant_id, sql.ptr(), affected_rows))) {
LOG_WARN("failed to execute", K(ret), K(sql));
}
}
return ret;
}
int ObDDLOperator::get_target_sequence_sync_value(const uint64_t tenant_id,
const uint64_t sequence_id,
common::ObMySQLTransaction &trans,
ObIAllocator &allocator,
common::number::ObNumber &next_value)
{
int ret = OB_SUCCESS;
next_value.set_zero();
ObSchemaService *schema_service_impl = schema_service_.get_schema_service();
if (OB_UNLIKELY(OB_INVALID_TENANT_ID == tenant_id || OB_INVALID_ID == sequence_id)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(tenant_id), K(sequence_id));
} else if (OB_ISNULL(schema_service_impl)) {
ret = OB_ERR_SYS;
LOG_ERROR("schema_service_impl must not null", K(ret));
} else if (OB_FAIL(schema_service_impl->get_sequence_sql_service().get_sequence_sync_value(tenant_id,
sequence_id,
true,/*is select for update*/
trans,
allocator,
next_value))) {
LOG_WARN("fail to get sequence sync value", K(ret), K(tenant_id), K(sequence_id));
}
return ret;
}
int ObDDLOperator::alter_target_sequence_start_with(const ObSequenceSchema &sequence_schema, common::ObMySQLTransaction &trans)
{
int ret = OB_SUCCESS;
ObSchemaService *schema_service_impl = schema_service_.get_schema_service();
if (OB_UNLIKELY(!sequence_schema.is_valid())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(sequence_schema));
} else if (OB_ISNULL(schema_service_impl)) {
ret = OB_ERR_SYS;
LOG_ERROR("schema_service_impl must not null", K(ret));
} else if (OB_FAIL(schema_service_impl->get_sequence_sql_service().alter_sequence_start_with(sequence_schema, trans))) {
LOG_WARN("fail to alter sequence start with", K(ret), K(sequence_schema));
}
return ret;
}
}//end namespace rootserver
}//end namespace oceanbase

26
src/rootserver/ob_ddl_operator.h
View File

@ -1023,6 +1023,32 @@ public:
const share::schema::ObTableSchema *table);
int try_add_dep_info_for_synonym(const ObSimpleSynonymSchema *synonym_info,
common::ObMySQLTransaction &trans);
int exchange_table_partitions(const share::schema::ObTableSchema &orig_table_schema,
share::schema::ObTableSchema &inc_table_schema,
share::schema::ObTableSchema &del_table_schema,
common::ObMySQLTransaction &trans);
int exchange_table_subpartitions(const share::schema::ObTableSchema &orig_table_schema,
share::schema::ObTableSchema &inc_table_schema,
share::schema::ObTableSchema &del_table_schema,
common::ObMySQLTransaction &trans);
int get_target_auto_inc_sequence_value(const uint64_t tenant_id,
const uint64_t table_id,
const uint64_t column_id,
uint64_t &sequence_value,
common::ObMySQLTransaction &trans);
int set_target_auto_inc_sync_value(const uint64_t tenant_id,
const uint64_t table_id,
const uint64_t column_id,
const uint64_t new_sequence_value,
const uint64_t new_sync_value,
common::ObMySQLTransaction &trans);
int get_target_sequence_sync_value(const uint64_t tenant_id,
const uint64_t sequence_id,
common::ObMySQLTransaction &trans,
ObIAllocator &allocator,
common::number::ObNumber &next_value);
int alter_target_sequence_start_with(const ObSequenceSchema &sequence_schema,
common::ObMySQLTransaction &trans);
private:
virtual int set_need_flush_ora(
share::schema::ObSchemaGetterGuard &schema_guard,

18
src/rootserver/ob_ddl_service.cpp
View File

@ -38648,8 +38648,8 @@ int ObDDLService::check_rename_first(const AlterTableSchema &alter_table_schema,
}
int ObDDLService::fix_local_idx_part_name_(const ObSimpleTableSchemaV2 &ori_data_table_schema,
const ObSimpleTableSchemaV2 &ori_table_schema,
ObSimpleTableSchemaV2 &inc_table_schema)
const ObSimpleTableSchemaV2 &ori_table_schema,
ObSimpleTableSchemaV2 &inc_table_schema)
{
int ret = OB_SUCCESS;
bool ori_oracle_mode = false;
@ -38701,7 +38701,7 @@ int ObDDLService::fix_local_idx_part_name_(const ObSimpleTableSchemaV2 &ori_data
ret = OB_ERR_UNEXPECTED;
LOG_WARN("src part is null", KR(ret));
} else if (FALSE_IT(part_name = ori_part->get_part_name())){
} else if (OB_FAIL(check_same_partition_(ori_oracle_mode, *ori_data_part, *ori_part, ori_part_func_type, is_matched))) {
} else if (OB_FAIL(check_same_partition(ori_oracle_mode, *ori_data_part, *ori_part, ori_part_func_type, is_matched))) {
LOG_WARN("fail to check ori_table_part and ori_aux_part is the same", KR(ret), KPC(ori_data_part), KPC(ori_part), K(ori_part_func_type));
} else if (OB_UNLIKELY(!is_matched)) {
ret = OB_INDEX_INELIGIBLE;
@ -38716,8 +38716,8 @@ int ObDDLService::fix_local_idx_part_name_(const ObSimpleTableSchemaV2 &ori_data
}
int ObDDLService::fix_local_idx_subpart_name_(const ObSimpleTableSchemaV2 &ori_data_table_schema,
const ObSimpleTableSchemaV2 &ori_table_schema,
ObSimpleTableSchemaV2 &inc_table_schema)
const ObSimpleTableSchemaV2 &ori_table_schema,
ObSimpleTableSchemaV2 &inc_table_schema)
{
int ret = OB_SUCCESS;
bool ori_oracle_mode = false;
@ -38782,7 +38782,7 @@ int ObDDLService::fix_local_idx_subpart_name_(const ObSimpleTableSchemaV2 &ori_d
} else if (OB_ISNULL(ori_part)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("src part is null", KR(ret));
} else if (OB_FAIL(check_same_partition_(ori_oracle_mode, *ori_data_part, *ori_part, ori_part_func_type, is_matched))) {
} else if (OB_FAIL(check_same_partition(ori_oracle_mode, *ori_data_part, *ori_part, ori_part_func_type, is_matched))) {
LOG_WARN("fail to check ori_table_part and ori_aux_part is the same", KR(ret), KPC(ori_data_part), KPC(ori_part), K(ori_part_func_type));
} else if (OB_UNLIKELY(!is_matched)) {
ret = OB_INDEX_INELIGIBLE;
@ -38796,7 +38796,7 @@ int ObDDLService::fix_local_idx_subpart_name_(const ObSimpleTableSchemaV2 &ori_d
ret = OB_ERR_UNEXPECTED;
LOG_WARN("fail to get src subpart.", KR(ret));
} else if (FALSE_IT(part_name = ori_subpart->get_part_name())) {
} else if (OB_FAIL(check_same_subpartition_(ori_oracle_mode, *ori_data_subpart, *ori_subpart, ori_subpart_func_type, is_matched))) {
} else if (OB_FAIL(check_same_subpartition(ori_oracle_mode, *ori_data_subpart, *ori_subpart, ori_subpart_func_type, is_matched))) {
LOG_WARN("fail to check ori_table_subpart and ori_aux_subpart is the same", KR(ret), KPC(ori_data_part), KPC(ori_part), K(ori_subpart_func_type));
} else if (OB_UNLIKELY(!is_matched)) {
ret = OB_INDEX_INELIGIBLE;
@ -38912,7 +38912,7 @@ int ObDDLService::fix_local_idx_part_name_for_add_subpart_(const ObSimpleTableSc
return ret;
}
int ObDDLService::check_same_partition_(const bool is_oracle_mode, const ObPartition &l, const ObPartition &r,
int ObDDLService::check_same_partition(const bool is_oracle_mode, const ObPartition &l, const ObPartition &r,
const ObPartitionFuncType part_type, bool &is_matched) const
{
int ret = OB_SUCCESS;
@ -38931,7 +38931,7 @@ int ObDDLService::check_same_partition_(const bool is_oracle_mode, const ObParti
return ret;
}
int ObDDLService::check_same_subpartition_(const bool is_oracle_mode, const ObSubPartition &l, const ObSubPartition &r,
int ObDDLService::check_same_subpartition(const bool is_oracle_mode, const ObSubPartition &l, const ObSubPartition &r,
const ObPartitionFuncType part_type, bool &is_matched) const
{
int ret = OB_SUCCESS;

68
src/rootserver/ob_ddl_service.h
View File

@ -2576,9 +2576,43 @@ private:
int gen_inc_table_schema_for_add_part(
const share::schema::ObTableSchema &orig_table_schema,
share::schema::AlterTableSchema &inc_table_schema);
int gen_inc_table_schema_for_add_subpart(
const share::schema::ObTableSchema &orig_table_schema,
share::schema::AlterTableSchema &inc_table_schema);
int gen_inc_table_schema_for_drop_part(
const share::schema::ObTableSchema &orig_table_schema,
share::schema::AlterTableSchema &inc_table_schema);
int gen_inc_table_schema_for_drop_subpart(
const share::schema::ObTableSchema &orig_table_schema,
share::schema::AlterTableSchema &inc_table_schema);
public:
//not check belong to the same table
int check_same_partition(const bool is_oracle_mode, const ObPartition &l, const ObPartition &r,
const ObPartitionFuncType part_type, bool &is_matched) const;
//not check belong to the same table
int check_same_subpartition(const bool is_oracle_mode, const ObSubPartition &l, const ObSubPartition &r,
const ObPartitionFuncType part_type, bool &is_matched) const;
private:
//After renaming a partition/subpartition, the consistency of the partition name between the data table and aux table is no longer guaranteed.
//Therefore, the partition names in the inc aux table must be synchronized with the ori aux table after assigning the data table's partition
//schema to the inc aux table.
//This function relies on the assumption that the inc table schema has a valid partition name.
int fix_local_idx_part_name_(const ObSimpleTableSchemaV2 &ori_data_table_schema,
const ObSimpleTableSchemaV2 &ori_table_schema,
ObSimpleTableSchemaV2 &inc_table_schema);
//This function relies on the assumption that the inc table schema has a valid subpartition name.
int fix_local_idx_subpart_name_(const ObSimpleTableSchemaV2 &ori_data_table_schema,
const ObSimpleTableSchemaV2 &ori_table_schema,
ObSimpleTableSchemaV2 &inc_table_schema);
//During the process of adding a partition/subpartition, we only check whether the partition schema of the argument is valid.
//It's possible for the inc aux table's partition name to duplicate with an existing partition name if one renames a partition/subpartition
//to another name and then adds a partition/subpartition with the same name.
//In this case, we will generate a name with a part/subpart id to replace the inc part/subpart name to avoid duplication.
int fix_local_idx_part_name_for_add_part_(const ObSimpleTableSchemaV2 &ori_table_schema,
ObSimpleTableSchemaV2 &inc_table_schema);
int fix_local_idx_part_name_for_add_subpart_(const ObSimpleTableSchemaV2 &ori_table_schema,
ObSimpleTableSchemaV2 &inc_table_schema);
int gen_inc_table_schema_for_rename_part_(
const share::schema::ObTableSchema &orig_table_schema,
share::schema::AlterTableSchema &inc_table_schema);
@ -2586,12 +2620,6 @@ private:
const share::schema::ObTableSchema &orig_table_schema,
share::schema::AlterTableSchema &inc_table_schema,
share::schema::AlterTableSchema &del_table_schema);
int gen_inc_table_schema_for_add_subpart(
const share::schema::ObTableSchema &orig_table_schema,
share::schema::AlterTableSchema &inc_table_schema);
int gen_inc_table_schema_for_drop_subpart(
const share::schema::ObTableSchema &orig_table_schema,
share::schema::AlterTableSchema &inc_table_schema);
int gen_inc_table_schema_for_rename_subpart_(
const share::schema::ObTableSchema &orig_table_schema,
share::schema::AlterTableSchema &inc_table_schema);
@ -2637,34 +2665,6 @@ private:
const share::schema::ObTenantSchema &orig_tenant_schema,
const share::schema::ObTenantSchema &new_tenant_schema);
//not check belong to the same table
int check_same_partition_(const bool is_oracle_mode, const ObPartition &l, const ObPartition &r,
const ObPartitionFuncType part_type, bool &is_matched) const;
//not check belong to the same table
int check_same_subpartition_(const bool is_oracle_mode, const ObSubPartition &l, const ObSubPartition &r,
const ObPartitionFuncType part_type, bool &is_matched) const;
//After renaming a partition/subpartition, the consistency of the partition name between the data table and aux table is no longer guaranteed.
//Therefore, the partition names in the inc aux table must be synchronized with the ori aux table after assigning the data table's partition
//schema to the inc aux table.
//This function relies on the assumption that the inc table schema has a valid partition name.
int fix_local_idx_part_name_(const ObSimpleTableSchemaV2 &ori_data_table_schema,
const ObSimpleTableSchemaV2 &ori_table_schema,
ObSimpleTableSchemaV2 &inc_table_schema);
//This function relies on the assumption that the inc table schema has a valid subpartition name.
int fix_local_idx_subpart_name_(const ObSimpleTableSchemaV2 &ori_data_table_schema,
const ObSimpleTableSchemaV2 &ori_table_schema,
ObSimpleTableSchemaV2 &inc_table_schema);
//During the process of adding a partition/subpartition, we only check whether the partition schema of the argument is valid.
//It's possible for the inc aux table's partition name to duplicate with an existing partition name if one renames a partition/subpartition
//to another name and then adds a partition/subpartition with the same name.
//In this case, we will generate a name with a part/subpart id to replace the inc part/subpart name to avoid duplication.
int fix_local_idx_part_name_for_add_part_(const ObSimpleTableSchemaV2 &ori_table_schema,
ObSimpleTableSchemaV2 &inc_table_schema);
int fix_local_idx_part_name_for_add_subpart_(const ObSimpleTableSchemaV2 &ori_table_schema,
ObSimpleTableSchemaV2 &inc_table_schema);
private:
int check_locality_compatible_(ObTenantSchema &schema);
int pre_rename_mysql_columns_online(const ObTableSchema &origin_table_schema,

2670
src/rootserver/ob_partition_exchange.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

296
src/rootserver/ob_partition_exchange.h Normal file
View File

@ -0,0 +1,296 @@
/**
* 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.
*/
#ifndef OCEANBASE_ROOTSERVER_OB_PARTITION_EXCHANGE_H_
#define OCEANBASE_ROOTSERVER_OB_PARTITION_EXCHANGE_H_
#include "lib/container/ob_array.h"
#include "share/ob_ddl_task_executor.h"
#include "share/ob_rpc_struct.h"
#include "share/schema/ob_schema_struct.h"
#include "storage/tablet/ob_tablet_binding_helper.h"
namespace oceanbase
{
namespace share
{
class AutoincParam;
namespace schema
{
class ObTableSchema;
}
}
namespace rootserver
{
class ObDDLService;
class ObDDLSQLTransaction;
class ObPartitionExchange final
{
public:
typedef std::pair<share::ObLSID, common::ObTabletID> LSTabletID;
explicit ObPartitionExchange(ObDDLService &ddl_service);
~ObPartitionExchange();
int check_and_exchange_partition(const obrpc::ObExchangePartitionArg &arg, obrpc::ObAlterTableRes &res, ObSchemaGetterGuard &schema_guard);
private:
int check_partition_exchange_conditions_(const obrpc::ObExchangePartitionArg &arg, const ObTableSchema &base_table_schema, const ObTableSchema &inc_table_schema, const bool is_oracle_mode, ObSchemaGetterGuard &schema_guard);
int do_exchange_partition_(const obrpc::ObExchangePartitionArg &arg, obrpc::ObAlterTableRes &res, const ObTableSchema &base_table_schema, const ObTableSchema &inc_table_schema, const bool is_oracle_mode, ObSchemaGetterGuard &schema_guard);
int lock_exchange_data_table_and_partition_(const uint64_t tenant_id, const ObTableSchema &partitioned_table_schema, const ObTableSchema &non_partitioned_table_schema, const common::ObTabletID &tablet_id, ObDDLSQLTransaction &trans);
int check_data_table_partition_exchange_conditions_(const ObTableSchema &base_table_schema, const ObTableSchema &inc_table_schema, const ObString &exchange_partition_name, const ObPartitionLevel exchange_partition_level,const bool is_oracle_mode);
// table level conditions that need to be checked for partition exchange in mysql mode and oracle mode
int check_table_conditions_in_common_(const ObTableSchema &base_table_schema, const ObTableSchema &inc_table_schema, const ObString &exchange_partition_name, const ObPartitionLevel exchange_partition_level, const bool is_oracle_mode);
// table level conditions that need to be checked for partition exchange in mysql mode
int check_table_conditions_in_mysql_mode_(const ObTableSchema &base_table_schema, const ObTableSchema &inc_table_schema);
// table level conditions that need to be checked for partition exchange in oracle mode
int check_table_conditions_in_oracle_mode_(const ObTableSchema &base_table_schema, const ObTableSchema &inc_table_schema);
int check_partition_and_table_tablespace_(const ObTableSchema &base_table_schema, const ObTableSchema &inc_table_schema, const ObString &exchange_partition_name, const ObPartitionLevel exchange_partition_level, const bool is_oracle_mode);
int check_table_index_infos_(const ObTableSchema &base_table_schema, const ObTableSchema &inc_table_schema, const bool is_oracle_mode);
int check_table_lob_infos_(const ObTableSchema &base_table_schema, const ObTableSchema &inc_table_schema, const bool is_oracle_mode);
int check_table_rowkey_infos_(const ObTableSchema &base_table_schema, const ObTableSchema &inc_table_schema, const bool is_oracle_mode);
int compare_two_rowkey_info_(const common::ObRowkeyInfo &l_rowkey_info, const common::ObRowkeyInfo &r_rowkey_info, bool &is_equal);
int check_table_column_groups_(const ObTableSchema &base_table_schema, const ObTableSchema &inc_table_schema, const bool is_oracle_mode);
int compare_two_column_group_schema_(const ObTableSchema &base_table_schema, const ObTableSchema &inc_table_schema, const ObColumnGroupSchema &base_cg_schema, const ObColumnGroupSchema &inc_cg_schema, const bool is_oracle_mode, bool &is_equal);
// verify if the check constraints of two tables meet the requirements
int check_table_constraints_(const ObTableSchema &base_table_schema, const ObTableSchema &inc_table_schema, const bool is_oracle_mode);
int check_table_all_column_conditions_(const ObTableSchema &base_table_schema, const ObTableSchema &inc_table_schema, const bool is_oracle_mode);
int check_column_level_conditions_(const ObColumnSchemaV2 *base_table_col_schema, const ObColumnSchemaV2 *inc_table_col_schema, const bool is_aux_table_column, const bool is_oracle_mode);
int check_column_conditions_in_common_(const ObColumnSchemaV2 *base_table_col_schema, const ObColumnSchemaV2 *inc_table_col_schema, const bool is_oracle_mode);
int check_column_conditions_in_mysql_mode_(const ObColumnSchemaV2 *base_table_col_schema, const ObColumnSchemaV2 *inc_table_col_schema, const bool is_aux_table_column);
int check_column_conditions_in_oracle_mode_(const ObColumnSchemaV2 *base_table_col_schema, const ObColumnSchemaV2 *inc_table_col_schema, const bool is_aux_table_column);
int check_generate_column_conditions_(const ObColumnSchemaV2 *base_table_col_schema, const ObColumnSchemaV2 *inc_table_col_schema, bool &is_equal);
int check_column_flags_(const ObColumnSchemaV2 *base_table_col_schema, const ObColumnSchemaV2 *inc_table_col_schema, bool &is_equal);
int check_column_default_value_(const ObColumnSchemaV2 *base_table_col_schema, const ObColumnSchemaV2 *inc_table_col_schema, const bool is_oracle_mode, bool &is_equal);
int compare_default_value_(ObObj &l_value, ObObj &r_value, const bool is_oracle_mode, bool &is_equal);
int get_next_pair_column_schema_(ObTableSchema::const_column_iterator &base_iter_begin,
ObTableSchema::const_column_iterator &base_iter_end,
ObTableSchema::const_column_iterator &inc_iter_begin,
ObTableSchema::const_column_iterator &inc_iter_end,
const bool is_oracle_mode,
ObColumnSchemaV2 *&base_table_col_schema,
ObColumnSchemaV2 *&inc_table_col_schema);
int get_next_need_check_column_(ObTableSchema::const_column_iterator &iter_begin, ObTableSchema::const_column_iterator &iter_end, const bool is_oracle_mode, ObColumnSchemaV2 *&table_col_schema);
int set_global_storage_index_unusable_(const uint64_t tenant_id,
const ObTableSchema &partitioned_data_table_schema,
const ObTableSchema &non_partitioned_data_table_schema,
ObDDLOperator &ddl_operator,
ObDDLSQLTransaction &trans,
ObSchemaGetterGuard &schema_guard);
int get_data_partition_and_index_(const ObTableSchema &partitioned_data_table_schema, const ObString &data_part_name, const ObPartition *&data_part, int64_t &data_partition_index);
int get_data_subpartition_and_index_(const ObTableSchema &partitioned_data_table_schema,
const ObString &data_subpart_name,
const ObPartition *&data_part,
const ObSubPartition *&data_subpart,
int64_t &data_partition_index,
int64_t &data_subpartition_index);
int exchange_data_table_partition_(const uint64_t tenant_id,
const ObTableSchema &partitioned_table_schema,
const ObTableSchema &non_partitioned_table_schema,
const ObPartition &part,
const bool is_oracle_mode,
ObDDLOperator &ddl_operator,
ObDDLSQLTransaction &trans,
ObSchemaGetterGuard &schema_guard);
int exchange_data_table_subpartition_(const uint64_t tenant_id,
const ObTableSchema &partitioned_table_schema,
const ObTableSchema &non_partitioned_table_schema,
const ObPartition &part,
const ObSubPartition &subpart,
const bool is_oracle_mode,
ObDDLOperator &ddl_operator,
ObDDLSQLTransaction &trans,
ObSchemaGetterGuard &schema_guard);
int exchange_auxiliary_table_partition_(const uint64_t tenant_id,
const int64_t ori_data_partition_index,
const ObPartition &ori_data_part,
const bool is_oracle_mode,
ObDDLOperator &ddl_operator,
ObDDLSQLTransaction &trans,
ObSchemaGetterGuard &schema_guard);
int exchange_auxiliary_table_subpartition_(const uint64_t tenant_id,
const int64_t ori_data_partition_index,
const int64_t ori_data_subpartition_index,
const ObPartition &ori_data_part,
const ObSubPartition &ori_data_subpart,
const bool is_oracle_mode,
ObDDLOperator &ddl_operator,
ObDDLSQLTransaction &trans,
ObSchemaGetterGuard &schema_guard);
int exchange_partition_map_relationship_(const uint64_t tenant_id,
const ObPartition &part,
const ObTableSchema &partitioned_table_schema,
const ObTableSchema &non_partitioned_table_schema,
const bool is_oracle_mode,
ObDDLOperator &ddl_operator,
ObDDLSQLTransaction &trans,
ObSchemaGetterGuard &schema_guard);
int exchange_subpartition_map_relationship_(const uint64_t tenant_id,
const ObPartition &part,
const ObSubPartition &subpart,
const ObTableSchema &partitioned_table_schema,
const ObTableSchema &non_partitioned_table_schema,
const bool is_oracle_mode,
ObDDLOperator &ddl_operator,
ObDDLSQLTransaction &trans,
ObSchemaGetterGuard &schema_guard);
int update_exchange_table_non_schema_attributes_(const uint64_t tenant_id,
const int64_t old_partition_id,
const int64_t new_partition_id,
const bool is_exchange_subpartition,
const ObTableSchema &partitioned_table_schema,
const ObTableSchema &non_partitioned_table_schema,
const ObIArray<uint64_t> &exchange_table_ids,
const ObIArray<ObTabletID> &exchange_tablet_ids,
const bool is_oracle_mode,
ObDDLOperator &ddl_operator,
ObDDLSQLTransaction &trans,
ObSchemaGetterGuard &schema_guard);
int update_exchange_table_level_attributes_(const uint64_t tenant_id,
const ObIArray<uint64_t> &exchange_table_ids,
const ObIArray<ObTabletID> &exchange_tablet_ids,
ObTableSchema &partitioned_table_schema,
ObTableSchema &non_partitioned_table_schema,
ObDDLSQLTransaction &trans);
int update_table_to_tablet_id_mapping_(const uint64_t tenant_id,
const ObIArray<uint64_t> &table_ids,
const ObIArray<ObTabletID> &tablet_ids,
ObDDLSQLTransaction &trans);
int refresh_table_schema_version_(const uint64_t tenant_id, ObTableSchema &table_schema);
int update_table_attribute_(const ObTableSchema &table_schema,
ObDDLSQLTransaction &trans);
int push_data_table_schema_version_(const uint64_t tenant_id,
const ObTableSchema &table_schema,
const common::ObString *ddl_stmt_str,
const uint64_t exchange_data_table_id,
int64_t &new_schema_version,
ObDDLSQLTransaction &trans);
int get_local_storage_index_and_lob_table_schemas_(const ObTableSchema &table_schema,
const bool is_pt_schema,
const bool is_oracle_mode,
ObIArray<const ObTableSchema*> &table_schemas,
ObSchemaGetterGuard &schema_guard);
int check_auxiliary_schema_conditions_(const ObTableSchema *table_schema, const bool is_oracle_mode);
int compare_column_extended_type_info_(const common::ObIArray<common::ObString> &l_extended_type_info,
const common::ObIArray<common::ObString> &r_extended_type_info,
bool &is_equal);
bool in_supported_table_type_white_list_(const ObTableSchema &table_schema);
// generate corresponding auxiliary table mapping that need to exchange partitions
bool in_find_same_aux_table_retry_white_list_(const int ret_code);
int generate_auxiliary_table_mapping_(const ObTableSchema &partitioned_data_table_schema,
const ObTableSchema &non_partitioned_data_table_schema,
const ObString &exchange_partition_name,
const ObPartitionLevel exchange_partition_level,
const bool is_oracle_mode,
ObSchemaGetterGuard &schema_guard);
int generate_local_storage_index_and_lob_table_mapping_(const ObTableSchema &partitioned_table_schema,
ObIArray<const ObTableSchema*> &non_partitioned_table_schemas,
const ObString &exchange_partition_name,
const ObPartitionLevel exchange_partition_level,
const bool is_oracle_mode,
ObIArray<bool> &used_nt_schema_flag);
int generate_local_storage_index_table_mapping_in_mysql_mode_(const ObTableSchema &partitioned_table_schema,
ObIArray<const ObTableSchema*> &non_partitioned_table_schemas,
const ObString &exchange_partition_name,
const ObPartitionLevel exchange_partition_level,
ObIArray<bool> &used_nt_schema_flag,
bool &find_related_nt_schema);
int generate_local_storage_index_table_mapping_in_oracle_mode_(const ObTableSchema &partitioned_table_schema,
ObIArray<const ObTableSchema*> &non_partitioned_table_schemas,
const ObString &exchange_partition_name,
const ObPartitionLevel exchange_partition_level,
ObIArray<bool> &used_nt_schema_flag,
bool &find_related_nt_schema);
int generate_lob_table_mapping_(const ObTableSchema &partitioned_table_schema,
ObIArray<const ObTableSchema*> &non_partitioned_table_schemas,
const ObString &exchange_partition_name,
const ObPartitionLevel exchange_partition_level,
const bool is_oracle_mode,
ObIArray<bool> &used_nt_schema_flag,
bool &find_related_nt_schema);
int update_index_status_(const uint64_t tenant_id,
const uint64_t table_id,
const share::schema::ObIndexStatus status,
const bool in_offline_ddl_white_list,
ObDDLOperator &ddl_operator,
ObDDLSQLTransaction &trans,
ObSchemaGetterGuard &schema_guard);
// Register MDS for read and write defense verification after single table ddl
int build_single_table_rw_defensive_(const uint64_t tenant_id,
const ObArray<common::ObTabletID> &tablet_ids,
const int64_t schema_version,
ObDDLSQLTransaction &trans);
int build_modify_tablet_binding_args_v1_(const uint64_t tenant_id,
const ObArray<ObTabletID> &tablet_ids,
const int64_t schema_version,
ObIArray<ObBatchUnbindTabletArg> &modify_args,
ObDDLSQLTransaction &trans);
int get_tablets_(const uint64_t tenant_id,
const ObArray<common::ObTabletID> &tablet_ids,
ObIArray<LSTabletID> &tablets,
ObDDLSQLTransaction &trans);
int adapting_cdc_changes_in_exchange_partition_(const uint64_t tenant_id,
const uint64_t partitioned_table_id,
const uint64_t non_partitioned_table_id,
ObDDLSQLTransaction &trans);
int update_autoinc_column_information_(const uint64_t tenant_id,
const ObTableSchema &partitioned_table_schema,
const ObTableSchema &non_partitioned_table_schema,
ObDDLOperator &ddl_operator,
ObDDLSQLTransaction &trans);
int update_identity_column_information_(const uint64_t tenant_id,
const ObTableSchema &partitioned_table_schema,
const ObTableSchema &non_partitioned_table_schema,
const bool is_oracle_mode,
ObDDLOperator &ddl_operator,
ObDDLSQLTransaction &trans,
ObSchemaGetterGuard &schema_guard);
int sync_exchange_partition_stats_info_(const uint64_t tenant_id,
const uint64_t new_table_id,
const uint64_t new_stat_level,
const int64_t old_partition_id,
const int64_t new_partition_id,
const ObTabletID &tablet_id,
const ObTableSchema &orig_table_schema,
ObDDLSQLTransaction &trans);
int update_table_all_monitor_modified_(const uint64_t tenant_id, const uint64_t new_table_id, const ObTabletID &tablet_id, const ObTableSchema &orig_table_schema, ObDDLSQLTransaction &trans);
int get_object_id_from_partition_schema_(ObPartitionSchema &partition_schema, const bool get_subpart_only, int64_t &object_id);
private:
ObDDLService &ddl_service_;
common::hash::ObHashMap<uint64_t, uint64_t> used_pt_nt_id_map_;
common::hash::ObHashMap<uint64_t, ObTabletID> used_table_to_tablet_id_map_;
common::ObSArray<uint64_t> unused_pt_index_id_;
common::ObSArray<uint64_t> unused_nt_index_id_;
private:
DISALLOW_COPY_AND_ASSIGN(ObPartitionExchange);
};
struct ObChangeTabletToTableArg final
{
OB_UNIS_VERSION_V(1);
public:
ObChangeTabletToTableArg() : tenant_id_(OB_INVALID_TENANT_ID), ls_id_(), base_table_id_(OB_INVALID_ID), inc_table_id_(OB_INVALID_ID), tablet_ids_(), table_ids_() {}
~ObChangeTabletToTableArg() {}
bool is_valid() const;
int assign(const ObChangeTabletToTableArg &other);
uint64_t tenant_id_;
share::ObLSID ls_id_;
uint64_t base_table_id_; // PT table, always with the large amount of data.
uint64_t inc_table_id_; // NT table, with the incremental data.
// tablet ids of data table, index table, lob meta/piece table.
common::ObSArray<ObTabletID> tablet_ids_;
// the table ids corresponding to the tablet ids.
common::ObSArray<uint64_t> table_ids_;
TO_STRING_KV(K_(tenant_id), K_(ls_id), K_(base_table_id), K_(inc_table_id), K_(tablet_ids), K_(table_ids));
};
}//end namespace rootserver
}//end namespace oceanbase
#endif //OCEANBASE_ROOTSERVER_OB_PARTITION_EXCHANGE_H_

40
src/rootserver/ob_root_service.cpp
View File

@ -74,6 +74,7 @@
#include "observer/dbms_scheduler/ob_dbms_sched_job_master.h"
#include "rootserver/ob_bootstrap.h"
#include "rootserver/ob_partition_exchange.h"
#include "rootserver/ob_schema2ddl_sql.h"
#include "rootserver/ob_index_builder.h"
#include "rootserver/ob_mlog_builder.h"
@ -4530,6 +4531,45 @@ int ObRootService::alter_table(const obrpc::ObAlterTableArg &arg, obrpc::ObAlter
return ret;
}
int ObRootService::exchange_partition(const obrpc::ObExchangePartitionArg &arg, obrpc::ObAlterTableRes &res)
{
int ret = OB_SUCCESS;
uint64_t compat_version = 0;
ObSchemaGetterGuard schema_guard;
ObPartitionExchange partition_exchange(ddl_service_);
schema_guard.set_session_id(arg.session_id_);
LOG_DEBUG("receive exchange partition arg", K(ret), K(arg));
if (!inited_) {
ret = OB_NOT_INIT;
LOG_WARN("not init", K(ret));
} else if (!arg.is_valid()) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid arg", K(ret), K(arg));
} else if (OB_FAIL(GET_MIN_DATA_VERSION(arg.tenant_id_, compat_version))) {
LOG_WARN("fail to get data version", K(ret), K(arg.tenant_id_));
} else if (compat_version < DATA_VERSION_4_3_1_0) {
ret = OB_NOT_SUPPORTED;
LOG_WARN("data version less than 4.3.1.0 does not support this operation", K(ret), K(compat_version));
} else if (OB_FAIL(ddl_service_.get_tenant_schema_guard_with_version_in_inner_table(arg.tenant_id_, schema_guard))) {
LOG_WARN("get schema guard in inner table failed", K(ret));
} else if (OB_FAIL(check_parallel_ddl_conflict(schema_guard, arg))) {
LOG_WARN("check parallel ddl conflict failed", K(ret));
} else if (OB_FAIL(partition_exchange.check_and_exchange_partition(arg, res, schema_guard))) {
LOG_WARN("fail to check and exchange partition", K(ret), K(arg), K(res));
}
char table_id_buffer[256];
snprintf(table_id_buffer, sizeof(table_id_buffer), "table_id:%ld, exchange_table_id:%ld",
arg.base_table_id_, arg.inc_table_id_);
ROOTSERVICE_EVENT_ADD("ddl scheduler", "alter table",
K(arg.tenant_id_),
"ret", ret,
"trace_id", *ObCurTraceId::get_trace_id(),
"table_id", table_id_buffer,
"schema_version", res.schema_version_);
LOG_INFO("finish alter table ddl", K(ret), K(arg), K(res), "ddl_event_info", ObDDLEventInfo());
return ret;
}
int ObRootService::create_index(const ObCreateIndexArg &arg, obrpc::ObAlterTableRes &res)
{
int ret = OB_SUCCESS;

1
src/rootserver/ob_root_service.h
View File

@ -501,6 +501,7 @@ public:
int rename_table(const obrpc::ObRenameTableArg &arg);
int truncate_table(const obrpc::ObTruncateTableArg &arg, obrpc::ObDDLRes &res);
int truncate_table_v2(const obrpc::ObTruncateTableArg &arg, obrpc::ObDDLRes &res);
int exchange_partition(const obrpc::ObExchangePartitionArg &arg, obrpc::ObAlterTableRes &res);
int create_index(const obrpc::ObCreateIndexArg &arg, obrpc::ObAlterTableRes &res);
int drop_table(const obrpc::ObDropTableArg &arg, obrpc::ObDDLRes &res);
int drop_database(const obrpc::ObDropDatabaseArg &arg, obrpc::ObDropDatabaseRes &drop_database_res);

1
src/rootserver/ob_rs_rpc_processor.h
View File

@ -343,6 +343,7 @@ DEFINE_DDL_RS_RPC_PROCESSOR(obrpc::OB_CREATE_TABLE, ObRpcCreateTableP, create_ta
DEFINE_DDL_RS_RPC_PROCESSOR(obrpc::OB_RECOVER_RESTORE_TABLE_DDL, ObRpcRecoverRestoreTableDDLP, recover_restore_table_ddl(arg_));
DEFINE_DDL_RS_RPC_PROCESSOR(obrpc::OB_PARALLEL_CREATE_TABLE, ObRpcParallelCreateTableP, parallel_create_table(arg_, result_));
DEFINE_DDL_RS_RPC_PROCESSOR(obrpc::OB_ALTER_TABLE, ObRpcAlterTableP, alter_table(arg_, result_));
DEFINE_DDL_RS_RPC_PROCESSOR(obrpc::OB_EXCHANGE_PARTITION, ObRpcExchangePartitionP, exchange_partition(arg_, result_));
DEFINE_DDL_RS_RPC_PROCESSOR(obrpc::OB_DROP_TABLE, ObRpcDropTableP, drop_table(arg_, result_));
DEFINE_DDL_RS_RPC_PROCESSOR(obrpc::OB_RENAME_TABLE, ObRpcRenameTableP, rename_table(arg_));
DEFINE_DDL_RS_RPC_PROCESSOR(obrpc::OB_TRUNCATE_TABLE, ObRpcTruncateTableP, truncate_table(arg_, result_));

1
src/share/ob_common_rpc_proxy.h
View File

@ -95,6 +95,7 @@ public:
RPC_S(PRD execute_ddl_task, obrpc::OB_EXECUTE_DDL_TASK, (ObAlterTableArg), common::ObSArray<uint64_t>);
RPC_S(PRD maintain_obj_dependency_info, obrpc::OB_MAINTAIN_OBJ_DEPENDENCY_INFO, (ObDependencyObjDDLArg));
RPC_S(PRD mview_complete_refresh, obrpc::OB_MVIEW_COMPLETE_REFRESH, (obrpc::ObMViewCompleteRefreshArg), obrpc::ObMViewCompleteRefreshRes);
RPC_S(PRD exchange_partition, obrpc::OB_EXCHANGE_PARTITION, (ObExchangePartitionArg), ObAlterTableRes);
//----Definitions for managing privileges----
RPC_S(PRD create_user, obrpc::OB_CREATE_USER, (ObCreateUserArg), common::ObSArray<int64_t>);

1
src/share/ob_errno.def
View File

@ -2652,7 +2652,6 @@ DEFINE_ERROR_EXT(OB_ERR_ILLEGAL_USER_VAR, -11013, ER_ILLEGAL_USER_VAR, "42000",
DEFINE_ERROR(OB_ERR_FT_COLUMN_NOT_INDEXED, -11014, ER_FT_MATCHING_KEY_NOT_FOUND, "HY000", "Can't find FULLTEXT index matching the column list");
DEFINE_ERROR_EXT(OB_ERR_CANT_UPDATE_TABLE_IN_CREATE_TABLE_SELECT, -11015, ER_CANT_UPDATE_TABLE_IN_CREATE_TABLE_SELECT, "HY000", "Can't update table while ctas is being created.", "Can't update table '%s' while '%s' is being created.");
DEFINE_ERROR_EXT(OB_ERR_PARTITION_EXCHANGE_PART_TABLE, -11017, ER_PARTITION_EXCHANGE_PART_TABLE, "HY000", "Table to exchange with partition is partitioned", "Table to exchange with partition is partitioned: \'%.*s\'");
// 余留位置
////////////////////////////////////////////////////////////////
// SQL扩展区段1：[-11000, -12000)

95
src/share/ob_rpc_struct.cpp
View File

@ -2650,6 +2650,101 @@ OB_DEF_SERIALIZE_SIZE(ObAlterTableArg)
return len;
}
bool ObExchangePartitionArg::is_valid() const
{
return OB_INVALID_ID != session_id_ && OB_INVALID_ID != tenant_id_ && PARTITION_LEVEL_ZERO != exchange_partition_level_ && PARTITION_LEVEL_MAX != exchange_partition_level_ && OB_INVALID_ID != base_table_id_
&& !base_table_part_name_.empty() && OB_INVALID_ID != inc_table_id_;
}
int ObExchangePartitionArg::assign(const ObExchangePartitionArg &other)
{
int ret = OB_SUCCESS;
if (this == &other) {
//do nothing
} else if (OB_FAIL(ObDDLArg::assign(other))) {
LOG_WARN("assign failed", K(ret));
} else {
session_id_ = other.session_id_;
tenant_id_ = other.tenant_id_;
exchange_partition_level_ = other.exchange_partition_level_;
base_table_id_ = other.base_table_id_;
base_table_part_name_ = other.base_table_part_name_;
inc_table_id_ = other.inc_table_id_;
including_indexes_ = other.including_indexes_;
without_validation_ = other.without_validation_;
update_global_indexes_ = other.update_global_indexes_;
}
return ret;
}
OB_DEF_SERIALIZE(ObExchangePartitionArg)
{
int ret = OB_SUCCESS;
BASE_SER((, ObDDLArg));
LST_DO_CODE(OB_UNIS_ENCODE,
session_id_,
tenant_id_,
exchange_partition_level_,
base_table_id_,
base_table_part_name_,
inc_table_id_,
including_indexes_,
without_validation_,
update_global_indexes_);
return ret;
}
OB_DEF_DESERIALIZE(ObExchangePartitionArg)
{
int ret = OB_SUCCESS;
BASE_DESER((, ObDDLArg));
LST_DO_CODE(OB_UNIS_DECODE,
session_id_,
tenant_id_,
exchange_partition_level_,
base_table_id_,
base_table_part_name_,
inc_table_id_,
including_indexes_,
without_validation_,
update_global_indexes_);
return ret;
}
OB_DEF_SERIALIZE_SIZE(ObExchangePartitionArg)
{
int64_t len = ObDDLArg::get_serialize_size();
LST_DO_CODE(OB_UNIS_ADD_LEN,
session_id_,
tenant_id_,
exchange_partition_level_,
base_table_id_,
base_table_part_name_,
inc_table_id_,
including_indexes_,
without_validation_,
update_global_indexes_);
return len;
}
DEF_TO_STRING(ObExchangePartitionArg)
{
int64_t pos = 0;
pos += ObDDLArg::to_string(buf + pos, buf_len - pos);
J_OBJ_START();
J_KV(K_(session_id),
K_(tenant_id),
K_(exchange_partition_level),
K_(base_table_id),
K_(base_table_part_name),
K_(inc_table_id),
K_(including_indexes),
K_(without_validation),
K_(update_global_indexes));
J_OBJ_END();
return pos;
}
bool ObTruncateTableArg::is_valid() const
{
return OB_INVALID_ID != tenant_id_ && !database_name_.empty()

35
src/share/ob_rpc_struct.h
View File

@ -2263,6 +2263,41 @@ public:
int64_t get_index_args_serialize_size() const;
};
struct ObExchangePartitionArg : public ObDDLArg
{
OB_UNIS_VERSION(1);
public:
ObExchangePartitionArg():
ObDDLArg(),
session_id_(common::OB_INVALID_ID),
tenant_id_(common::OB_INVALID_ID),
exchange_partition_level_(PARTITION_LEVEL_MAX),
base_table_id_(common::OB_INVALID_ID),
base_table_part_name_(),
inc_table_id_(common::OB_INVALID_ID),
including_indexes_(true),
without_validation_(true),
update_global_indexes_(false)
{
}
virtual ~ObExchangePartitionArg()
{
}
bool is_valid() const;
int assign(const ObExchangePartitionArg& other);
public:
DECLARE_TO_STRING;
uint64_t session_id_;
uint64_t tenant_id_;
ObPartitionLevel exchange_partition_level_;
uint64_t base_table_id_; // PT table, always contains large amount pf data.
ObString base_table_part_name_;
uint64_t inc_table_id_; // NT table, always contains incremental data.
bool including_indexes_; // default true.
bool without_validation_; // default true.
bool update_global_indexes_; // default false.
};
struct ObTableItem
{
OB_UNIS_VERSION(1);

1
src/share/schema/ob_column_schema.h
View File

@ -300,6 +300,7 @@ int assign(const ObColumnSchemaV2 &src_schema);
inline bool is_enum_or_set() const { return meta_type_.is_enum_or_set(); }
inline static bool is_hidden_pk_column_id(const uint64_t column_id);
inline bool is_unused() const { return column_flags_ & UNUSED_COLUMN_FLAG; }
//other methods
int64_t get_convert_size(void) const;

14
src/share/schema/ob_constraint.cpp
View File

@ -96,6 +96,20 @@ int64_t ObConstraint::get_convert_size() const
convert_size += column_cnt_ * sizeof(uint64_t);
return convert_size;
}
/*The following function is used to determine whether the check constraints of the two partition exchange tables are the same.
To ensure that the check constraints are exactly the same, the following conditions need to be met:
1、rely_flag_、enable_flag_、validate_flag_, these three values are the conditions for determining the constraint state.
2、constraint_type_, This value determines the type of constraint.
3、check_expr_, at present, it is required that the expression of the constraint must be exactly the same of the two partition exchange tables.
*/
bool ObConstraint::is_match_partition_exchange_constraint_conditions(const ObConstraint &r) const
{
return rely_flag_ == r.get_rely_flag() &&
enable_flag_ == r.get_enable_flag() &&
validate_flag_ == r.get_validate_flag() &&
constraint_type_ == r.get_constraint_type() &&
(0 == check_expr_.compare(r.get_check_expr_str()));
}
int ObConstraint::get_not_null_column_name(ObString &cst_col_name) const
{

1
src/share/schema/ob_constraint.h
View File

@ -119,6 +119,7 @@ public:
// other
int64_t get_convert_size() const;
bool is_match_partition_exchange_constraint_conditions(const ObConstraint &r) const;
void reset();
DECLARE_VIRTUAL_TO_STRING;
private:

14
src/share/schema/ob_schema_struct.cpp
View File

@ -14448,6 +14448,20 @@ int ObColumnGroupSchema::get_column_group_type_name(ObString &readable_cg_name)
}
return ret;
}
/*
The following function is used by partition exchange to compare whether cg-level attributes are the same and three attributes are not considered.
1、column_group_name: The same column group in the two tables may have different names specified by the user, so no comparison is required.
2、column_group_id: Adding and deleting column groups multiple times to the same table will cause the column group id to increase, and the rules are similar to column ids. Therefore, in two tables, the same column group may have different column group ids.
3、column id contained in column group: Since in the comparison of columns in two tables, the column id of the same column is not necessarily the same, therefore, the column id in the two column groups cannot distinguish whether they are the same column. You need to use the column id to get the column schema and then compare the column attributes in sequence.
*/
bool ObColumnGroupSchema::has_same_column_group_attributes_for_part_exchange(const ObColumnGroupSchema &other) const
{
return column_group_type_ == other.get_column_group_type() &&
block_size_ == other.get_block_size() &&
compressor_type_ == other.get_compressor_type() &&
row_store_type_ == other.get_row_store_type() &&
column_id_cnt_ == other.get_column_id_count();
}
OB_DEF_SERIALIZE(ObSkipIndexColumnAttr)
{

1
src/share/schema/ob_schema_struct.h
View File

@ -8938,6 +8938,7 @@ public:
int get_column_id(const int64_t idx, uint64_t &column_id) const;
int remove_column_id(const uint64_t column_id);
int get_column_group_type_name(ObString &readable_cg_name) const;
bool has_same_column_group_attributes_for_part_exchange(const ObColumnGroupSchema &other) const;
VIRTUAL_TO_STRING_KV(K_(column_group_id),
K_(column_group_name),

52
src/share/schema/ob_sequence_sql_service.cpp
View File

@ -92,6 +92,7 @@ int ObSequenceSqlService::alter_sequence_start_with(const ObSequenceSchema &sequ
// to get sync value from inner table.
int ObSequenceSqlService::get_sequence_sync_value(const uint64_t tenant_id,
const uint64_t sequence_id,
const bool is_for_update,
common::ObISQLClient &sql_client,
ObIAllocator &allocator,
common::number::ObNumber &next_value)
@ -100,6 +101,7 @@ int ObSequenceSqlService::get_sequence_sync_value(const uint64_t tenant_id,
ObSqlString sql;
common::number::ObNumber tmp;
const char *tname = OB_ALL_SEQUENCE_VALUE_TNAME;
const char *is_for_update_str = "FOR UPDATE";
SMART_VAR(ObMySQLProxy::MySQLResult, res) {
common::sqlclient::ObMySQLResult *result = nullptr;
if (OB_FAIL(sql.assign_fmt(
@ -107,29 +109,36 @@ int ObSequenceSqlService::get_sequence_sync_value(const uint64_t tenant_id,
"WHERE SEQUENCE_ID = %lu",
tname, sequence_id))) {
LOG_WARN("fail to format sql", K(ret));
} else if (OB_FAIL(sql_client.read(res, tenant_id, sql.ptr()))) {
LOG_WARN("fail to execute sql", K(sql), K(ret));
} else if (nullptr == (result = res.get_result())) {
ret = OB_ENTRY_NOT_EXIST;
LOG_WARN("can't find sequence", K(ret), K(tname), K(tenant_id), K(sequence_id));
} else if (OB_FAIL(result->next())) {
if (OB_ITER_END != ret) {
LOG_WARN("fail to get next row", K(ret), K(tname), K(tenant_id), K(sequence_id));
} else {
// OB_ITER_END means there is no record in table,
// thus the sync value is its' start value, and init the table when operate it.
} else if (is_for_update) {
if (OB_FAIL(sql.append_fmt(" %s", is_for_update_str))) {
LOG_WARN("fail to assign sql", K(ret));
}
} else {
EXTRACT_NUMBER_FIELD_MYSQL(*result, NEXT_VALUE, tmp);
if (OB_FAIL(ret)) {
LOG_WARN("fail to get NEXT_VALUE", K(ret));
} else if (OB_FAIL(next_value.from(tmp, allocator))) {
LOG_WARN("fail to deep copy next_val", K(tmp), K(ret));
} else if (OB_ITER_END != (ret = result->next())) {
LOG_WARN("expected OB_ITER_END", K(ret));
ret = (OB_SUCCESS == ret ? OB_ERR_UNEXPECTED : ret);
}
if (OB_SUCC(ret)) {
if (OB_FAIL(sql_client.read(res, tenant_id, sql.ptr()))) {
LOG_WARN("fail to execute sql", K(sql), K(ret));
} else if (nullptr == (result = res.get_result())) {
ret = OB_ENTRY_NOT_EXIST;
LOG_WARN("can't find sequence", K(ret), K(tname), K(tenant_id), K(sequence_id));
} else if (OB_FAIL(result->next())) {
if (OB_ITER_END != ret) {
LOG_WARN("fail to get next row", K(ret), K(tname), K(tenant_id), K(sequence_id));
} else {
// OB_ITER_END means there is no record in table,
// thus the sync value is its' start value, and init the table when operate it.
}
} else {
ret = OB_SUCCESS;
EXTRACT_NUMBER_FIELD_MYSQL(*result, NEXT_VALUE, tmp);
if (OB_FAIL(ret)) {
LOG_WARN("fail to get NEXT_VALUE", K(ret));
} else if (OB_FAIL(next_value.from(tmp, allocator))) {
LOG_WARN("fail to deep copy next_val", K(tmp), K(ret));
} else if (OB_ITER_END != (ret = result->next())) {
LOG_WARN("expected OB_ITER_END", K(ret));
ret = (OB_SUCCESS == ret ? OB_ERR_UNEXPECTED : ret);
} else {
ret = OB_SUCCESS;
}
}
}
}
@ -475,6 +484,7 @@ int ObSequenceSqlService::add_sequence_to_value_table(const uint64_t tenant_id,
common::number::ObNumber next_value;
if (OB_FAIL(get_sequence_sync_value(tenant_id,
old_sequence_id,
false,/*is select for update*/
sql_client,
allocator,
next_value))) {

17
src/share/schema/ob_sequence_sql_service.h
View File

@ -56,6 +56,15 @@ public:
const int64_t new_schema_version,
common::ObISQLClient *sql_client,
const common::ObString *ddl_stmt_str = NULL);
int alter_sequence_start_with(const ObSequenceSchema &sequence_schema,
common::ObISQLClient &sql_client);
int get_sequence_sync_value(const uint64_t tenant_id,
const uint64_t sequence_id,
const bool is_for_update,
common::ObISQLClient &sql_client,
ObIAllocator &allocator,
common::number::ObNumber &next_value);
int clean_sequence_cache(uint64_t tenant_id, uint64_t sequence_id);
private:
int add_sequence(common::ObISQLClient &sql_client, const ObSequenceSchema &sequence_schema,
const bool only_history, const uint64_t *old_sequence_id);
@ -65,14 +74,6 @@ private:
const uint64_t new_sequence_id,
common::ObISQLClient &sql_client,
ObIAllocator &allocator);
int alter_sequence_start_with(const ObSequenceSchema &sequence_schema,
common::ObISQLClient &sql_client);
int get_sequence_sync_value(const uint64_t tenant_id,
const uint64_t sequence_id,
common::ObISQLClient &sql_client,
ObIAllocator &allocator,
common::number::ObNumber &next_value);
int clean_sequence_cache(uint64_t tenant_id, uint64_t sequence_id);
private:
DISALLOW_COPY_AND_ASSIGN(ObSequenceSqlService);
};

55
src/share/schema/ob_table_sql_service.cpp
View File

@ -713,6 +713,61 @@ int ObTableSqlService::truncate_subpart_info(
return ret;
}
int ObTableSqlService::exchange_part_info(
common::ObISQLClient &sql_client,
const ObTableSchema &ori_table,
ObTableSchema &inc_table,
ObTableSchema &del_table,
const int64_t schema_version)
{
int ret = OB_SUCCESS;
bool is_truncate_table = false;
bool is_truncate_partition = true;
if (OB_FAIL(check_ddl_allowed(ori_table))) {
LOG_WARN("check ddl allowd failed", K(ret), K(ori_table));
} else if (OB_FAIL(drop_inc_part_info(sql_client,
ori_table,
del_table,
schema_version,
is_truncate_partition,
is_truncate_table))) {
LOG_WARN("delete inc part info failed", K(ret));
} else if (OB_FAIL(add_inc_partition_info(sql_client,
ori_table,
inc_table,
schema_version,
true/*is_truncate_table*/,
false/*is_subpart*/))) {
LOG_WARN("add inc part info failed", K(ret));
}
return ret;
}
int ObTableSqlService::exchange_subpart_info(
common::ObISQLClient &sql_client,
const ObTableSchema &ori_table,
ObTableSchema &inc_table,
ObTableSchema &del_table,
const int64_t schema_version)
{
int ret = OB_SUCCESS;
const ObPartition *part = NULL;
if (OB_FAIL(check_ddl_allowed(ori_table))) {
LOG_WARN("check ddl allowd failed", K(ret), K(ori_table));
} else if (OB_FAIL(drop_inc_subpart_info(sql_client, ori_table, del_table,
schema_version))) {
LOG_WARN("failed to drop partition", K(ret), K(del_table));
} else if (OB_FAIL(add_inc_partition_info(sql_client,
ori_table,
inc_table,
schema_version,
true/*is_truncate_table*/,
true/*is_subpart*/))) {
LOG_WARN("add partition info failed", K(ret));
}
return ret;
}
int ObTableSqlService::drop_table(const ObTableSchema &table_schema,
const int64_t new_schema_version,
ObISQLClient &sql_client,

14
src/share/schema/ob_table_sql_service.h
View File

@ -208,6 +208,20 @@ public:
ObTableSchema &del_table,
const int64_t schema_version);
int exchange_part_info(
common::ObISQLClient &sql_client,
const ObTableSchema &ori_table,
ObTableSchema &inc_table,
ObTableSchema &del_table,
const int64_t schema_version);
int exchange_subpart_info(
common::ObISQLClient &sql_client,
const ObTableSchema &ori_table,
ObTableSchema &inc_table,
ObTableSchema &del_table,
const int64_t schema_version);
int sync_schema_version_for_history(
common::ObISQLClient &sql_client,
ObTableSchema &schema,

35
src/share/tablet/ob_tablet_to_ls_operator.cpp
View File

@ -288,6 +288,41 @@ int ObTabletToLSTableOperator::batch_update(
}
return ret;
}
int ObTabletToLSTableOperator::update_table_to_tablet_id_mapping(common::ObISQLClient &sql_proxy,
const uint64_t tenant_id,
const uint64_t table_id,
const common::ObTabletID &tablet_id)
{
int ret = OB_SUCCESS;
uint64_t data_version = 0;
if (OB_UNLIKELY(OB_INVALID_TENANT_ID == tenant_id || OB_INVALID_ID == table_id || !tablet_id.is_valid())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(tenant_id), K(tablet_id), K(tablet_id));
} else if (OB_FAIL(GET_MIN_DATA_VERSION(tenant_id, data_version))) {
LOG_WARN("fail to get min data version", KR(ret));
} else if (data_version < DATA_VERSION_4_3_1_0) {
ret = OB_NOT_SUPPORTED;
LOG_WARN("update table id and tablet id mapping when data_version is less than 4.3.1.0 is not supported", K(ret), K(table_id), K(tablet_id));
} else {
ObSqlString sql;
ObDMLSqlSplicer dml_splicer;
int64_t affected_rows = 0;
if (OB_FAIL(dml_splicer.add_pk_column("tablet_id", tablet_id.id()))
|| OB_FAIL(dml_splicer.add_column("table_id", table_id))) {
LOG_WARN("fail to add column", K(ret), K(tablet_id), K(table_id));
} else if (OB_FAIL(dml_splicer.splice_update_sql(OB_ALL_TABLET_TO_LS_TNAME, sql))) {
LOG_WARN("fail to splice batch insert update sql", K(ret), K(sql));
} else if (OB_FAIL(sql_proxy.write(tenant_id, sql.ptr(), affected_rows))) {
LOG_WARN("fail to write sql", K(ret), K(sql), K(affected_rows));
} else if(!is_single_row(affected_rows)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("expect one row", K(ret), K(sql), K(affected_rows));
} else {
LOG_TRACE("update tablet_to_ls success", K(tenant_id), K(affected_rows));
}
}
return ret;
}
int ObTabletToLSTableOperator::inner_batch_update_by_sql_(
common::ObISQLClient &sql_proxy,

5
src/share/tablet/ob_tablet_to_ls_operator.h
View File

@ -133,6 +133,11 @@ public:
const int64_t new_transfer_seq,
const ObLSID &new_ls_id,
const int32_t group_id);
static int update_table_to_tablet_id_mapping(
common::ObISQLClient &sql_proxy,
const uint64_t tenant_id,
const uint64_t table_id,
const common::ObTabletID &tablet_id);
// Get rows from __all_tablet_to_ls according to ObTableIDs
//
// @param [in] sql_proxy, ObMySQLProxy or ObMySQLTransaction

42
src/sql/engine/cmd/ob_table_executor.cpp
View File

@ -900,6 +900,24 @@ int ObAlterTableExecutor::alter_table_rpc_v2(
return ret;
}
int ObAlterTableExecutor::alter_table_exchange_partition_rpc(obrpc::ObExchangePartitionArg &exchange_partition_arg,
obrpc::ObAlterTableRes &res,
obrpc::ObCommonRpcProxy *common_rpc_proxy,
ObSQLSessionInfo *my_session)
{
int ret = OB_SUCCESS;
if (OB_ISNULL(my_session) || OB_ISNULL(common_rpc_proxy) || OB_UNLIKELY(!exchange_partition_arg.is_valid())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(exchange_partition_arg.is_valid()));
} else if (OB_FAIL(common_rpc_proxy->exchange_partition(exchange_partition_arg, res))) {
LOG_WARN("rpc proxy alter table failed", K(ret), "dst", common_rpc_proxy->get_server(), K(exchange_partition_arg));
} else {
// 在回滚时不会重试，也不检查 schema version
exchange_partition_arg.based_schema_object_infos_.reset();
}
return ret;
}
int ObAlterTableExecutor::sort_external_files(ObIArray<ObString> &file_urls,
ObIArray<int64_t> &file_sizes) {
int ret = OB_SUCCESS;
@ -1188,6 +1206,7 @@ int ObAlterTableExecutor::execute(ObExecContext &ctx, ObAlterTableStmt &stmt)
ObTaskExecutorCtx *task_exec_ctx = NULL;
obrpc::ObCommonRpcProxy *common_rpc_proxy = NULL;
obrpc::ObAlterTableArg &alter_table_arg = stmt.get_alter_table_arg();
obrpc::ObExchangePartitionArg &exchange_partition_arg = stmt.get_exchange_partition_arg();
LOG_DEBUG("start of alter table execute", K(alter_table_arg));
ObString first_stmt;
OZ (stmt.get_first_stmt(first_stmt));
@ -1219,6 +1238,7 @@ int ObAlterTableExecutor::execute(ObExecContext &ctx, ObAlterTableStmt &stmt)
LOG_WARN("get first statement failed", K(ret));
} else {
alter_table_arg.ddl_stmt_str_ = first_stmt;
exchange_partition_arg.ddl_stmt_str_ = first_stmt;
my_session = ctx.get_my_session();
if (NULL == my_session) {
ret = OB_ERR_UNEXPECTED;
@ -1262,13 +1282,19 @@ int ObAlterTableExecutor::execute(ObExecContext &ctx, ObAlterTableStmt &stmt)
}
}
if (OB_SUCC(ret)) {
if (OB_FAIL(alter_table_rpc_v2(
alter_table_arg,
res,
allocator,
common_rpc_proxy,
my_session,
is_sync_ddl_user))) {
if (obrpc::ObAlterTableArg::EXCHANGE_PARTITION == alter_table_arg.alter_part_type_) {
if (OB_FAIL(alter_table_exchange_partition_rpc(exchange_partition_arg,
res,
common_rpc_proxy,
my_session))) {
LOG_WARN("Failed to alter table exchange partition rpc", K(ret), K(exchange_partition_arg));
}
} else if (OB_FAIL(alter_table_rpc_v2(alter_table_arg,
res,
allocator,
common_rpc_proxy,
my_session,
is_sync_ddl_user))) {
LOG_WARN("Failed to alter table rpc v2", K(ret));
}
}
@ -1977,6 +2003,8 @@ int ObAlterTableExecutor::check_alter_partition(ObExecContext &ctx,
|| obrpc::ObAlterTableArg::TRUNCATE_PARTITION == arg.alter_part_type_
|| obrpc::ObAlterTableArg::TRUNCATE_SUB_PARTITION == arg.alter_part_type_) {
// do-nothing
} else if (obrpc::ObAlterTableArg::EXCHANGE_PARTITION == arg.alter_part_type_) {
// do-nothing
} else {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("no operation", K(arg.alter_part_type_), K(ret));

6
src/sql/engine/cmd/ob_table_executor.h
View File

@ -143,6 +143,12 @@ private:
ObSQLSessionInfo *my_session,
const bool is_sync_ddl_user);
int alter_table_exchange_partition_rpc(
obrpc::ObExchangePartitionArg &exchange_partition_arg,
obrpc::ObAlterTableRes &res,
obrpc::ObCommonRpcProxy *common_rpc_proxy,
ObSQLSessionInfo *my_session);
int need_check_constraint_validity(obrpc::ObAlterTableArg &alter_table_arg, bool &need_check);
int set_alter_col_nullable_ddl_stmt_str(

2
src/sql/parser/non_reserved_keywords_mysql_mode.c
View File

@ -1023,6 +1023,8 @@ static const NonReservedKeyword Mysql_none_reserved_keywords[] =
{"kv_attributes", KV_ATTRIBUTES},
{"RESOURCE_POOL", RESOURCE_POOL},
{"clone", CLONE},
{"without", WITHOUT},
{"validation", VALIDATION},
{"_st_asmvt", _ST_ASMVT},
};

10
src/sql/parser/sql_parser_mysql_mode.y
View File

@ -363,9 +363,9 @@ END_P SET_VAR DELIMITER
UNUSUAL UPGRADE USE_BLOOM_FILTER UNKNOWN USE_FRM USER USER_RESOURCES UNBOUNDED UP UNLIMITED
VALID VALUE VARIANCE VARIABLES VERBOSE VERIFY VIEW VISIBLE VIRTUAL_COLUMN_ID VALIDATE VAR_POP
VAR_SAMP
VAR_SAMP VALIDATION
WAIT WARNINGS WASH WEEK WEIGHT_STRING WHENEVER WORK WRAPPER WINDOW WEAK WITH_COLUMN_GROUP
WAIT WARNINGS WASH WEEK WEIGHT_STRING WHENEVER WORK WRAPPER WINDOW WEAK WITH_COLUMN_GROUP WITHOUT
X509 XA XML
@ -16078,6 +16078,10 @@ ADD PARTITION opt_partition_range_or_list
merge_nodes($$, result, T_NAME_LIST, $3);
malloc_non_terminal_node($$, result->malloc_pool_, T_ALTER_SUBPARTITION_TRUNCATE, 2, $$, NULL);
}
| EXCHANGE PARTITION relation_name WITH TABLE relation_factor WITHOUT VALIDATION
{
malloc_non_terminal_node($$, result->malloc_pool_, T_ALTER_PARTITION_EXCHANGE, 2, $3, $6);
}
;
opt_partition_range_or_list:
@ -21072,6 +21076,8 @@ ACCOUNT
| OBJECT_ID
| TRANSFER
| SUM_OPNSIZE
| VALIDATION
| WITHOUT
;
unreserved_keyword_special:

149
src/sql/resolver/ddl/ob_alter_table_resolver.cpp
View File

@ -2544,6 +2544,147 @@ int ObAlterTableResolver::resolve_drop_subpartition(const ParseNode &node,
return ret;
}
int ObAlterTableResolver::resolve_exchange_partition(const ParseNode &node,
const ObTableSchema &orig_table_schema)
{
int ret = OB_SUCCESS;
uint64_t tenant_data_version = 0;
ObAlterTableStmt *alter_table_stmt = get_alter_table_stmt();
const ObPartitionLevel part_level = orig_table_schema.get_part_level();
if (OB_UNLIKELY(T_ALTER_PARTITION_EXCHANGE != node.type_ || OB_ISNULL(node.children_))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid parse tree", K(ret), K(node.type_));
} else if (OB_UNLIKELY(2 != node.num_child_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid parse tree, num child != 2", K(ret), K(node.num_child_));
} else if (OB_ISNULL(node.children_[0]) || OB_ISNULL(node.children_[1])) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid parse tree", K(ret));
} else if (OB_ISNULL(session_info_) || OB_ISNULL(alter_table_stmt)) {
ret = OB_NOT_INIT;
LOG_WARN("session info and alter table stmt should not be null", K(ret), KPC(session_info_), KPC(alter_table_stmt));
} else if (OB_FAIL(GET_MIN_DATA_VERSION(session_info_->get_effective_tenant_id(), tenant_data_version))) {
LOG_WARN("get data version failed", K(ret), K(session_info_->get_effective_tenant_id()), K(tenant_data_version));
} else if (OB_UNLIKELY(tenant_data_version < DATA_VERSION_4_3_1_0)) {
ret = OB_NOT_SUPPORTED;
LOG_WARN("data version and feature mismatch", K(ret), K(tenant_data_version));
} else if (OB_UNLIKELY(!orig_table_schema.is_user_table())) {
ret = OB_NOT_SUPPORTED;
LOG_USER_ERROR(OB_NOT_SUPPORTED, "alter table type");
LOG_WARN("unsupport behavior on not user table", K(ret), K(orig_table_schema));
} else if (OB_UNLIKELY(PARTITION_LEVEL_ZERO == part_level)) {
ret = OB_ERR_PARTITION_MGMT_ON_NONPARTITIONED;
LOG_USER_ERROR(OB_ERR_PARTITION_MGMT_ON_NONPARTITIONED);
LOG_WARN("unsupport management on non partitioned table", K(ret), K(part_level));
} else {
const ObPartition *part = nullptr;
const ObSubPartition *subpart = nullptr;
const ObTableSchema *exchange_table_schema = NULL;
ObString exchange_table_name;
ObString exchange_db_name;
ParseNode *second_node = node.children_[1];
ObString origin_partition_name(static_cast<int32_t>(node.children_[0]->str_len_), node.children_[0]->str_value_);
if (OB_FAIL(resolve_table_relation_node(second_node,
exchange_table_name,
exchange_db_name))){
LOG_WARN("failed to resolve exchange table node", K(ret), K(exchange_table_name), K(exchange_db_name));
} else if (lib::is_oracle_mode() && 0 != exchange_db_name.compare(session_info_->get_database_name())) {
ret = OB_TABLE_NOT_EXIST;//compatible with Oracle, reporting error table does not exist
LOG_WARN("Swapping partitions between different database tables is not supported in oracle mode", K(ret), K(lib::is_oracle_mode()), K(exchange_db_name), K(session_info_->get_database_name()));
} else if (0 == exchange_db_name.compare(session_info_->get_database_name()) && 0 == exchange_table_name.compare(orig_table_schema.get_table_name())) {
ret = OB_ERR_NONUNIQ_TABLE;
LOG_USER_ERROR(OB_ERR_NONUNIQ_TABLE, exchange_table_name.length(), exchange_table_name.ptr());
LOG_WARN("Not unique table/alias", K(ret), K(exchange_table_name), K(exchange_db_name), K(session_info_->get_database_name()), K(orig_table_schema.get_table_name()));
} else if (OB_ISNULL(schema_checker_)) {
ret = OB_NOT_INIT;
LOG_WARN("schema_checker should not be null", K(ret));
} else if (OB_FAIL(schema_checker_->get_table_schema(session_info_->get_effective_tenant_id(),
exchange_db_name,
exchange_table_name,
false,
exchange_table_schema))) {
if (OB_TABLE_NOT_EXIST == ret) {
LOG_USER_ERROR(OB_TABLE_NOT_EXIST, to_cstring(exchange_db_name), to_cstring(exchange_table_name));
}
LOG_WARN("fail to get table schema", K(ret), KPC(exchange_table_schema), K(exchange_db_name), K(exchange_table_name));
} else if (OB_ISNULL(exchange_table_schema)) {
ret = OB_TABLE_NOT_EXIST;
LOG_WARN("table not found", K(ret), KPC(exchange_table_schema), K(exchange_db_name), K(exchange_table_name));
} else if (OB_UNLIKELY(!exchange_table_schema->is_user_table())) {
ret = OB_NOT_SUPPORTED;
LOG_USER_ERROR(OB_NOT_SUPPORTED, "alter table type");
LOG_WARN("unsupport behavior on not user table", K(ret), KPC(exchange_table_schema));
} else if (OB_UNLIKELY(exchange_table_schema->is_partitioned_table())) {
ret = OB_ERR_PARTITION_EXCHANGE_PART_TABLE;
LOG_USER_ERROR(OB_ERR_PARTITION_EXCHANGE_PART_TABLE, exchange_table_name.length(), exchange_table_name.ptr());
} else if (share::schema::ObPartitionLevel::PARTITION_LEVEL_ONE == orig_table_schema.get_part_level()) {
if (OB_FAIL(orig_table_schema.get_partition_by_name(origin_partition_name, part))) {
LOG_WARN("fail to get partition", K(ret), K(orig_table_schema), K(origin_partition_name));
} else if (OB_ISNULL(part)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("partition not found", K(ret), K(orig_table_schema), K(origin_partition_name));
} else {
share::schema::ObPartitionFuncType part_type = orig_table_schema.get_part_option().get_part_func_type();
if (OB_UNLIKELY(PARTITION_FUNC_TYPE_RANGE != part_type && PARTITION_FUNC_TYPE_RANGE_COLUMNS != part_type)) {
ret = OB_NOT_SUPPORTED;
LOG_WARN("Only support exchanging range/range columns partitions currently", K(ret), K(part_type));
LOG_USER_ERROR(OB_NOT_SUPPORTED, "Exchange partition except range/range columns");
}
}
} else if (share::schema::ObPartitionLevel::PARTITION_LEVEL_TWO == orig_table_schema.get_part_level()) {
if (OB_FAIL(orig_table_schema.get_partition_by_name(origin_partition_name, part))) {
if (OB_UNKNOWN_PARTITION == ret) {
ret = OB_SUCCESS;
if (OB_FAIL(orig_table_schema.get_subpartition_by_name(origin_partition_name, part, subpart))) {
LOG_WARN("get subpartition by name failed", K(ret), K(orig_table_schema), K(origin_partition_name));
} else if (OB_ISNULL(part) || OB_ISNULL(subpart)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("partition not found", K(ret), K(OB_ISNULL(part)), K(OB_ISNULL(subpart)));
} else {
share::schema::ObPartitionFuncType subpart_type = orig_table_schema.get_sub_part_option().get_part_func_type();
if (OB_UNLIKELY(PARTITION_FUNC_TYPE_RANGE != subpart_type && PARTITION_FUNC_TYPE_RANGE_COLUMNS != subpart_type)) {
ret = OB_NOT_SUPPORTED;
LOG_WARN("Only support exchanging range/range columns partitions currently", K(ret), K(subpart_type));
LOG_USER_ERROR(OB_NOT_SUPPORTED, "Exchange partition except range/range columns");
}
}
} else {
LOG_WARN("fail to get partition by name", K(ret), K(orig_table_schema), K(origin_partition_name));
}
} else {
ret = OB_ERR_EXCHANGE_COMPOSITE_PARTITION;
LOG_WARN("cannot EXCHANGE a composite partition with a non-partitioned table", K(ret), K(orig_table_schema), K(origin_partition_name));
}
} else {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("partition level is invalid", K(ret), K(orig_table_schema.get_part_level()));
}
if (OB_SUCC(ret)) {
obrpc::ObExchangePartitionArg exchange_partition_arg;
if (OB_FAIL(exchange_partition_arg.based_schema_object_infos_.assign(alter_table_stmt->get_alter_table_arg().based_schema_object_infos_))) {
LOG_WARN("fail to assign based_schema_object_infos", K(ret), K(alter_table_stmt->get_alter_table_arg().based_schema_object_infos_));
} else if (OB_FAIL(exchange_partition_arg.based_schema_object_infos_.push_back(ObBasedSchemaObjectInfo(exchange_table_schema->get_table_id(), TABLE_SCHEMA, exchange_table_schema->get_schema_version())))) {
LOG_WARN("failed to add exchange table info", K(ret), KPC(exchange_table_schema));
} else {
exchange_partition_arg.session_id_ = session_info_->get_sessid();
exchange_partition_arg.tenant_id_ = session_info_->get_effective_tenant_id();
exchange_partition_arg.exchange_partition_level_ = orig_table_schema.get_part_level();
exchange_partition_arg.base_table_id_ = orig_table_schema.get_table_id();
exchange_partition_arg.base_table_part_name_ = origin_partition_name;
exchange_partition_arg.inc_table_id_ = exchange_table_schema->get_table_id();
exchange_partition_arg.including_indexes_ = true;
exchange_partition_arg.without_validation_ = true;
exchange_partition_arg.update_global_indexes_ = false;
exchange_partition_arg.exec_tenant_id_ = session_info_->get_effective_tenant_id();
if (OB_FAIL(alter_table_stmt->set_exchange_partition_arg(exchange_partition_arg))) {
LOG_WARN("fail to set exchange_partition_arg", K(ret), K(exchange_partition_arg));
}
}
}
}
return ret;
}
int ObAlterTableResolver::check_subpart_name(const ObPartition &partition,
const ObSubPartition &subpartition)
{
@ -4240,6 +4381,14 @@ int ObAlterTableResolver::resolve_partition_options(const ParseNode &node)
}
break;
}
case T_ALTER_PARTITION_EXCHANGE: {
if (OB_FAIL(resolve_exchange_partition(*partition_node, *table_schema_))) {
LOG_WARN("failed to resolve exchange partition", K(ret));
} else {
alter_table_stmt->get_alter_table_arg().alter_part_type_ = ObAlterTableArg::EXCHANGE_PARTITION;
}
break;
}
default: {
ret = OB_ERR_UNEXPECTED;
SQL_RESV_LOG(WARN, "Unknown alter partition option %d!",

2
src/sql/resolver/ddl/ob_alter_table_resolver.h
View File

@ -145,6 +145,8 @@ private:
const share::schema::ObTableSchema &orig_table_schema);
int resolve_drop_subpartition(const ParseNode &node,
const share::schema::ObTableSchema &orig_table_schema);
int resolve_exchange_partition(const ParseNode &node,
const share::schema::ObTableSchema &orig_table_schema);
int resolve_rename_partition(const ParseNode &node,
const share::schema::ObTableSchema &orig_table_schema);
int resolve_rename_subpartition(const ParseNode &node,

9
src/sql/resolver/ddl/ob_alter_table_stmt.cpp
View File

@ -160,5 +160,14 @@ int ObAlterTableStmt::fill_session_vars(const ObBasicSessionInfo &session) {
return ret;
}
int ObAlterTableStmt::set_exchange_partition_arg(const obrpc::ObExchangePartitionArg &exchange_partition_arg)
{
int ret = OB_SUCCESS;
if (OB_FAIL(exchange_partition_arg_.assign(exchange_partition_arg))) {
SQL_RESV_LOG(WARN, "failed to assign", K(ret), K(exchange_partition_arg));
}
return ret;
}
} //namespace sql
} //namespace oceanbase

3
src/sql/resolver/ddl/ob_alter_table_stmt.h
View File

@ -111,6 +111,8 @@ public:
obrpc::ObAlterTriggerArg &get_tg_arg() { return tg_arg_; }
const ObTableSchema &get_alter_table_schema() const { return alter_table_arg_.alter_table_schema_; }
ObTableSchema &get_alter_table_schema() { return alter_table_arg_.alter_table_schema_; }
obrpc::ObExchangePartitionArg &get_exchange_partition_arg() { return exchange_partition_arg_;}
int set_exchange_partition_arg(const obrpc::ObExchangePartitionArg &exchange_partition_arg);
private:
obrpc::ObAlterTableArg alter_table_arg_;
bool is_comment_table_;
@ -122,6 +124,7 @@ private:
ObRawExpr *transition_expr_;
uint64_t alter_table_action_count_;
int64_t alter_external_table_type_;
obrpc::ObExchangePartitionArg exchange_partition_arg_;
};
inline int ObAlterTableStmt::set_tz_info_wrap(const common::ObTimeZoneInfoWrap &tz_info_wrap)

66
src/storage/ddl/ob_ddl_change_tablet_to_table_helper.h Normal file
View File

@ -0,0 +1,66 @@
/**
* 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.
*/
#ifndef OCEANBASE_STORAGE_OB_CHANGE_TABLET_TO_TABLE_HELPER
#define OCEANBASE_STORAGE_OB_CHANGE_TABLET_TO_TABLE_HELPER
#include "common/ob_tablet_id.h"
#include "lib/container/ob_array.h"
#include "lib/container/ob_array_serialization.h"
#include "lib/ob_define.h"
#include "share/ob_ls_id.h"
namespace oceanbase
{
namespace share
{
class SCN;
}
namespace storage
{
namespace mds
{
struct BufferCtx;
}
class ObChangeTabletToTableHelper final
{
public:
static int on_register(const char* buf,
const int64_t len,
mds::BufferCtx &ctx); // 出参，将对应修改记录在Ctx中
static int on_replay(const char* buf,
const int64_t len,
const share::SCN &scn, // 日志scn
mds::BufferCtx &ctx); // 备机回放
};
inline int ObChangeTabletToTableHelper::on_register(const char* buf,
const int64_t len,
mds::BufferCtx &ctx)
{
int ret = OB_SUCCESS;
return ret;
}
inline int ObChangeTabletToTableHelper::on_replay(const char* buf,
const int64_t len,
const share::SCN &scn, // 日志scn
mds::BufferCtx &ctx)
{
int ret = OB_SUCCESS;
return ret;
}
} // namespace storage
} // namespace oceanbase
#endif

6
src/storage/multi_data_source/compile_utility/mds_register.h
View File

@ -18,6 +18,7 @@
// inlcude those classes definations header file in below MACRO BLOCK
// CAUTION: MAKE SURE your header file is as CLEAN as possible to avoid recursive dependencies!
#if defined (NEED_MDS_REGISTER_DEFINE) && !defined (NEED_GENERATE_MDS_FRAME_CODE_FOR_TRANSACTION)
#include "src/storage/ddl/ob_ddl_change_tablet_to_table_helper.h"
#include "src/storage/multi_data_source/compile_utility/mds_dummy_key.h"
#include "src/storage/multi_data_source/mds_ctx.h"
#include "src/storage/multi_data_source/test/example_user_data_define.h"
@ -118,7 +119,10 @@ _GENERATE_MDS_FRAME_CODE_FOR_TRANSACTION_(HELPER_CLASS, BUFFER_CTX_TYPE, ID, ENU
28,\
TRANSFER_DEST_PREPARE)
// UNBIND_LOB_TABLET: ID = 29 for drop lob tablet when drop column instant.
// CHANGE_TABLET_TO_TABLE_MDS: ID = 30 for exchange tablet to table bind relationship after exchanging partition.
GENERATE_MDS_FRAME_CODE_FOR_TRANSACTION(::oceanbase::storage::ObChangeTabletToTableHelper,\
::oceanbase::storage::mds::MdsCtx,\
30,\
CHANGE_TABLET_TO_TABLE_MDS)
// TABLET_SPLIT: ID = 31 for auto_split.
#undef GENERATE_MDS_FRAME_CODE_FOR_TRANSACTION
#endif