/**

* 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.

*/

#define USING_LOG_PREFIX RS

#include "ob_root_utils.h"

#include "ob_balance_info.h"

#include "ob_unit_manager.h"

#include "lib/json/ob_json.h"

#include "lib/string/ob_sql_string.h"

#include "lib/hash/ob_hashset.h"

#include "lib/mysqlclient/ob_mysql_result.h"

#include "share/ob_rpc_struct.h"

#include "share/ob_share_util.h"

#include "share/ob_common_rpc_proxy.h"

#include "share/schema/ob_table_schema.h"

#include "share/schema/ob_schema_struct.h"

#include "share/schema/ob_multi_version_schema_service.h" // ObMultiVersionSchemaService

#include "share/schema/ob_schema_getter_guard.h" // ObSchemaGetterGuard

#include "share/inner_table/ob_inner_table_schema_constants.h"

#include "storage/tx/ob_ts_mgr.h"

#include "rootserver/ob_unit_manager.h"

#include "rootserver/ob_root_service.h"

#include "rootserver/ob_ddl_service.h"

#include "observer/ob_server_struct.h"

#include "logservice/palf_handle_guard.h"

#include "logservice/ob_log_service.h"

#include "share/system_variable/ob_system_variable_alias.h"

#include "share/ob_primary_zone_util.h" // ObPrimaryZoneUtil

#include "share/ob_server_table_operator.h"

#include "share/ob_zone_table_operation.h"

#include "rootserver/ob_tenant_balance_service.h" // for ObTenantBalanceService

using namespace oceanbase::rootserver;

using namespace oceanbase::share;

using namespace oceanbase::share::schema;

using namespace oceanbase::common;

using namespace oceanbase::common::hash;

using namespace oceanbase::common::sqlclient;

using namespace oceanbase::obrpc;

int ObTenantUtils::get_tenant_ids(

ObMultiVersionSchemaService *schema_service,

ObIArray<uint64_t> &tenant_ids)

{

int ret = OB_SUCCESS;

ObSchemaGetterGuard schema_guard;

if (OB_ISNULL(schema_service)) {

ret = OB_NOT_INIT;

LOG_WARN("schema service not init", K(ret));

} else if (OB_FAIL(schema_service->get_tenant_schema_guard(OB_SYS_TENANT_ID, schema_guard))) {

LOG_WARN("get_schema_guard failed", K(ret));

} else if (OB_FAIL(schema_guard.get_tenant_ids(tenant_ids))) {

LOG_WARN("get_tenant_ids failed", K(ret));

}

return ret;

}

int ObTenantUtils::get_tenant_ids(

ObMultiVersionSchemaService *schema_service,

int64_t &sys_schema_version,

ObIArray<uint64_t> &tenant_ids)

{

int ret = OB_SUCCESS;

ObSchemaGetterGuard schema_guard;

if (OB_ISNULL(schema_service)) {

ret = OB_NOT_INIT;

LOG_WARN("schema service not init", K(ret));

} else if (OB_FAIL(schema_service->get_tenant_schema_guard(OB_SYS_TENANT_ID, schema_guard))) {

LOG_WARN("get_schema_guard failed", K(ret));

} else if (OB_FAIL(schema_guard.get_tenant_ids(tenant_ids))) {

LOG_WARN("get_tenant_ids failed", K(ret));

} else if (OB_FAIL(schema_guard.get_schema_version(OB_SYS_TENANT_ID, sys_schema_version))) {

LOG_WARN("fail to get schema version", KR(ret));

}

return ret;

}

bool ObTenantUtils::is_balance_target_schema(

const share::schema::ObSimpleTableSchemaV2 &table_schema)

{

return USER_TABLE == table_schema.get_table_type()

|| TMP_TABLE == table_schema.get_table_type()

|| MATERIALIZED_VIEW == table_schema.get_table_type()

|| TMP_TABLE_ORA_SESS == table_schema.get_table_type()

|| TMP_TABLE_ORA_TRX == table_schema.get_table_type()

|| TMP_TABLE_ALL == table_schema.get_table_type()

|| AUX_VERTIAL_PARTITION_TABLE == table_schema.get_table_type()

|| table_schema.is_global_index_table();

}

bool ObRootServiceRoleChecker::is_rootserver()

{

bool bret = false;

int ret = OB_SUCCESS;

const uint64_t tenant_id = OB_SYS_TENANT_ID;

MTL_SWITCH(tenant_id) {

int64_t proposal_id = -1;

ObRole role = FOLLOWER;

palf::PalfHandleGuard palf_handle_guard;

logservice::ObLogService *log_service = nullptr;

if (OB_ISNULL(log_service = MTL(logservice::ObLogService*))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("MTL ObLogService is null", KR(ret), K(tenant_id));

} else if (OB_FAIL(log_service->open_palf(SYS_LS, palf_handle_guard))) {

LOG_WARN("open palf failed", KR(ret), K(tenant_id));

} else if (OB_FAIL(palf_handle_guard.get_role(role, proposal_id))) {

LOG_WARN("get role failed", KR(ret), K(tenant_id));

} else {

bret = (is_strong_leader(role));

LOG_DEBUG("get __all_core_table role", K(role), K(bret));

}

} else {

if (OB_TENANT_NOT_IN_SERVER == ret) {

ret = OB_SUCCESS;

} else {

LOG_WARN("fail to get tenant storage", KR(ret), "tenant_id", OB_SYS_TENANT_ID);

}

}

return bret;

}

const char *ObRootBalanceHelp::BalanceItem[] = {

"ENABLE_REBUILD",

"ENABLE_EMERGENCY_REPLICATE",

"ENABLE_TYPE_TRANSFORM",

"ENABLE_DELETE_REDUNDANT",

"ENABLE_REPLICATE_TO_UNIT",

"ENABLE_SHRINK",

"ENABLE_REPLICATE",

"ENABLE_COORDINATE_PG",

"ENABLE_MIGRATE_TO_UNIT",

"ENABLE_PARTITION_BALANCE",

"ENABLE_UNIT_BALANCE",

"ENABLE_SERVER_BALANCE",

"ENABLE_CANCEL_UNIT_MIGRATION",

"ENABLE_MODIFY_PAXOS_REPLICA_NUMBER",

"ENABLE_STOP_SERVER",

""

};

int ObRootBalanceHelp::parse_balance_info(const ObString &json_str,

ObRootBalanceHelp::BalanceController &switch_info)

{

int ret = OB_SUCCESS;

ObArenaAllocator allocator(ObModIds::OB_JSON_PARSER);

json::Parser parser;

json::Value *data = NULL;

switch_info.reset();

if (json_str.empty()) {

switch_info.init();

} else if (OB_FAIL(parser.init(&allocator))) {

LOG_WARN("json parser init failed", K(ret));

} else if (OB_FAIL(parser.parse(json_str.ptr(), json_str.length(), data))) {

LOG_WARN("parse json failed", K(ret), K(json_str));

} else if (NULL == data) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("no root value", K(ret));

} else if (json::JT_OBJECT != data->get_type()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("error json format", K(ret), K(json_str));

} else {

DLIST_FOREACH_X(it, data->get_object(), OB_SUCC(ret)) {

bool find = false;

for (int64_t i = 0; i < ARRAYSIZEOF(BalanceItem) - 1 && !find && OB_SUCC(ret); i++) {

if (it->name_.case_compare(BalanceItem[i]) == 0) {

find = true;

if (json::JT_STRING != it->value_->get_type()) {

ret = OB_INVALID_CONFIG;

LOG_WARN("invalid config", K(ret), K(json_str));

} else if (it->value_->get_string().case_compare("true") == 0) {

switch_info.set(i, true);

} else if (it->value_->get_string().case_compare("false") == 0) {

switch_info.set(i, false);

} else {

ret = OB_INVALID_CONFIG;

LOG_WARN("invalid config", K(ret), K(json_str));

}

} // if (it->name_.case_compare

} //for (int64_t i = 0;

if (!find) {

ret = OB_ENTRY_NOT_EXIST;

LOG_WARN("get invalid token", K(ret), K(*it));

}

} //DLIST_FOREACH_X(it

}

return ret;

}

int ObLocalityUtil::parse_zone_list_from_locality_str(ObString &locality_str,

ObIArray<ObZone> &zone_list)

{

int ret = OB_SUCCESS;

zone_list.reset();

if (locality_str.empty()) {

//nothing todo

} else {

ObArray<ObString> sub_locality;

ObString trimed_string = locality_str.trim();

if (OB_FAIL(split_on(trimed_string, ',', sub_locality))) {

LOG_WARN("fail to split string", K(ret), "locality", trimed_string);

} else {

ObArray<ObString> zone_infos;

ObZone tmp_zone;

for (int64_t i = 0; i < sub_locality.count() && OB_SUCC(ret); i++) {

zone_infos.reset();

tmp_zone.reset();

ObString sub_trimed_string = sub_locality.at(i).trim();

if (OB_FAIL(split_on(sub_trimed_string, '@', zone_infos))) {

LOG_WARN("fail to split on string", K(ret), "string", sub_trimed_string);

} else if (zone_infos.count() != 2) {

//FULL{1},READONLY{1}@z3;

//nothing todo

} else if (OB_FAIL(tmp_zone.assign(zone_infos.at(1)))) {

LOG_WARN("fail to assign zone", K(ret), K(zone_infos));

} else if (has_exist_in_array(zone_list, tmp_zone)) {

//nothint todo

} else if (OB_FAIL(zone_list.push_back(tmp_zone))) {

LOG_WARN("fail to push back", K(ret));

}

LOG_DEBUG("split sub locality", K(ret), K(sub_trimed_string), K(tmp_zone), K(zone_infos), K(tmp_zone), K(zone_list));

}

}

}

return ret;

}

int ObTenantGroupParser::parse_tenant_groups(

const common::ObString &ttg_str,

common::ObIArray<TenantNameGroup> &tenant_groups)

{

int ret = OB_SUCCESS;

int64_t end = ttg_str.length();

int64_t pos = 0;

while (OB_SUCC(ret) && pos < end) {

if (OB_FAIL(get_next_tenant_group(pos, end, ttg_str, tenant_groups))) {

LOG_WARN("fail to parse single tenant group", K(ret), K(ttg_str));

} else if (OB_FAIL(jump_to_next_ttg(pos, end, ttg_str))) {

LOG_WARN("fail to jump to next", K(ret));

}

}

return ret;

}

int ObTenantGroupParser::get_next_tenant_group(

int64_t &pos,

const int64_t end,

const common::ObString &ttg_str,

common::ObIArray<TenantNameGroup> &tenant_groups)

{

int ret = OB_SUCCESS;

jump_over_space(pos, end, ttg_str);

if (pos >= end) {

// reach to the end

} else if ('(' != ttg_str[pos]) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

} else {

// begin with a left brace

++pos;

jump_over_space(pos, end, ttg_str);

if (pos >= end) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

} else if ('"' == ttg_str[pos]) {

ret = parse_vector_tenant_group(pos, end, ttg_str, tenant_groups);

} else if ('(' == ttg_str[pos]) {

ret = parse_matrix_tenant_group(pos, end, ttg_str, tenant_groups);

} else {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(pos));

}

// end with a right brace

if (OB_SUCC(ret)) {

jump_over_space(pos, end, ttg_str);

if (pos >= end) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(pos));

} else if (')' != ttg_str[pos]) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(pos));

} else {

++pos;

}

}

}

return ret;

}

int ObTenantGroupParser::get_next_tenant_name(

int64_t &pos,

const int64_t end,

const common::ObString &ttg_str,

common::ObString &tenant_name)

{

int ret = OB_SUCCESS;

jump_over_space(pos, end, ttg_str);

if (pos >= end) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

} else if ('"' != ttg_str[pos]) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

} else {

++pos; // jump over the " symbol

jump_over_space(pos, end, ttg_str);

int64_t start = pos;

while (pos < end && !isspace(ttg_str[pos]) && '"' != ttg_str[pos]) {

++pos;

}

if (pos >= end) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

} else {

tenant_name.assign_ptr(ttg_str.ptr() + start, static_cast<int32_t>(pos - start));

}

// end with a " symbol

if (OB_SUCC(ret)) {

jump_over_space(pos, end, ttg_str);

if (pos >= end) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(pos));

} else if ('"' != ttg_str[pos]) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(pos));

} else {

++pos;

}

}

}

return ret;

}

int ObTenantGroupParser::jump_to_next_tenant_name(

int64_t &pos,

const int64_t end,

const common::ObString &ttg_str)

{

int ret = OB_SUCCESS;

jump_over_space(pos, end, ttg_str);

if (pos >= end) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(pos));

} else if (',' == ttg_str[pos]) {

++pos;

jump_over_space(pos, end, ttg_str);

if (pos >= end) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(pos));

} else if ('"' != ttg_str[pos]) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(pos));

} else {} // good, come across a " symbol

} else if (')' == ttg_str[pos]) {

// good, reach the end of this vector

} else {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(pos));

}

return ret;

}

int ObTenantGroupParser::parse_tenant_vector(

int64_t &pos,

const int64_t end,

const common::ObString &ttg_str,

common::ObIArray<common::ObString> &tenant_names)

{

int ret = OB_SUCCESS;

tenant_names.reset();

while (OB_SUCC(ret) && pos < end && ')' != ttg_str[pos]) {

ObString tenant_name;

if (OB_FAIL(get_next_tenant_name(pos, end, ttg_str, tenant_name))) {

LOG_WARN("fail to get next tenant name", K(ret));

} else if (has_exist_in_array(all_tenant_names_, tenant_name)) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

} else if (OB_FAIL(tenant_names.push_back(tenant_name))) {

LOG_WARN("fail to push back", K(ret));

} else if (OB_FAIL(all_tenant_names_.push_back(tenant_name))) {

LOG_WARN("fail to push back", K(ret));

} else if (OB_FAIL(jump_to_next_tenant_name(pos, end, ttg_str))) {

LOG_WARN("fail to jump to next tenant name", K(ret));

} else {} // next loop round

}

return ret;

}

int ObTenantGroupParser::parse_vector_tenant_group(

int64_t &pos,

const int64_t end,

const common::ObString &ttg_str,

common::ObIArray<TenantNameGroup> &tenant_groups)

{

int ret = OB_SUCCESS;

TenantNameGroup tenant_group;

common::ObArray<common::ObString> tenant_names;

if (OB_FAIL(parse_tenant_vector(pos, end, ttg_str, tenant_names))) {

LOG_WARN("fail to parse tenant vector", K(ret));

} else if (OB_FAIL(append(tenant_group.tenants_, tenant_names))) {

LOG_WARN("fail to append", K(ret));

} else {

tenant_group.row_ = 1;

tenant_group.column_ = tenant_group.tenants_.count();

if (OB_FAIL(tenant_groups.push_back(tenant_group))) {

LOG_WARN("fail to push back", K(ret));

}

}

return ret;

}

int ObTenantGroupParser::parse_matrix_tenant_group(

int64_t &pos,

const int64_t end,

const common::ObString &ttg_str,

common::ObIArray<TenantNameGroup> &tenant_groups)

{

int ret = OB_SUCCESS;

TenantNameGroup tenant_group;

common::ObArray<common::ObString> tenant_names;

bool first = true;

// parse the second layer of two layer bracket structure

while (OB_SUCC(ret) && pos < end && ')' != ttg_str[pos]) {

tenant_names.reset();

jump_over_space(pos, end, ttg_str);

if (pos >= end) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

} else if ('(' != ttg_str[pos]) { // start with left brace

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

} else {

++pos; // jump over left brace

}

if (OB_SUCC(ret)) {

if (OB_FAIL(parse_tenant_vector(pos, end, ttg_str, tenant_names))) {

LOG_WARN("fail to parse tenant vector", K(ret));

} else if (first) {

tenant_group.column_ = tenant_names.count();

tenant_group.row_ = 0;

first = false;

} else if (tenant_group.column_ != tenant_names.count()) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

}

}

if (OB_SUCC(ret)) {

if (OB_FAIL(append(tenant_group.tenants_, tenant_names))) {

LOG_WARN("fail to append", K(ret));

} else {

++tenant_group.row_;

}

}

if (OB_SUCC(ret)) {

jump_over_space(pos, end, ttg_str);

if (pos >= end) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

} else if (')' != ttg_str[pos]) { // end up with right brace for a row tenant

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

} else {

++pos;

}

}

if (OB_SUCC(ret)) {

if (OB_FAIL(jump_to_next_tenant_vector(pos, end, ttg_str))) {

LOG_WARN("fail to jump to next tenant vector", K(ret));

}

}

}

if (OB_SUCC(ret)) {

if (tenant_group.row_ <= 0 || tenant_group.column_ <= 0) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

} else if (OB_FAIL(tenant_groups.push_back(tenant_group))) {

LOG_WARN("fail to push back", K(ret));

} else {} // no more to do

}

return ret;

}

int ObTenantGroupParser::jump_to_next_tenant_vector(

int64_t &pos,

const int64_t end,

const common::ObString &ttg_str)

{

int ret = OB_SUCCESS;

while (pos < end && isspace(ttg_str[pos])) {

++pos;

}

if (pos >= end) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

} else if (')' == ttg_str[pos]) {

// reach to the end of this tenant group

} else if (',' == ttg_str[pos]) {

++pos; // good, and jump over the ',' symbol

while (pos < end && isspace(ttg_str[pos])) {

++pos;

}

if (pos >= end) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

} else if ('(' == ttg_str[pos]) {

// the next tenant vector left brace

} else {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

}

} else {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

}

return ret;

}

void ObTenantGroupParser::jump_over_space(

int64_t &pos,

const int64_t end,

const common::ObString &ttg_str)

{

while (pos < end && isspace(ttg_str[pos])) {

++pos;

}

return;

}

int ObTenantGroupParser::jump_to_next_ttg(

int64_t &pos,

const int64_t end,

const common::ObString &ttg_str)

{

int ret = OB_SUCCESS;

while (pos < end && isspace(ttg_str[pos])) {

++pos;

}

if (pos >= end) {

// good, reach end

} else if (',' == ttg_str[pos]) {

++pos; // good, and jump over the ',' symbol

while (pos < end && isspace(ttg_str[pos])) {

++pos;

}

if (pos >= end) {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

} else if ('(' == ttg_str[pos]) {

// good, the next ttg left brace symbol

} else {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

}

} else {

ret = OB_ERR_PARSER_SYNTAX;

LOG_WARN("invalid tenantgroup string syntax", K(ret), K(ttg_str), K(pos));

}

return ret;

}

int ObLocalityTaskHelp::filter_logonly_task(const common::ObIArray<share::ObResourcePoolName> &pools,

ObUnitManager &unit_mgr,

ObIArray<share::ObZoneReplicaNumSet> &zone_locality)

{

int ret = OB_SUCCESS;

ObArray<ObUnitInfo> logonly_unit_infos;

ObArray<ObUnitInfo> unit_infos;

if (pools.count() <= 0) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret), K(pools));

} else if (OB_FAIL(unit_mgr.get_unit_infos(pools, unit_infos))) {

LOG_WARN("fail to get unit infos", K(ret), K(pools));

} else {

for (int64_t i = 0; i < unit_infos.count() && OB_SUCC(ret); ++i) {

if (REPLICA_TYPE_LOGONLY != unit_infos.at(i).unit_.replica_type_) {

// only L unit is counted

} else if (OB_FAIL(logonly_unit_infos.push_back(unit_infos.at(i)))) {

LOG_WARN("fail to push back", K(ret), K(i), K(unit_infos));

}

}

for (int64_t i = 0; i < zone_locality.count() && OB_SUCC(ret); ++i) {

share::ObZoneReplicaAttrSet &zone_replica_attr_set = zone_locality.at(i);

if (zone_replica_attr_set.get_logonly_replica_num()

+ zone_replica_attr_set.get_encryption_logonly_replica_num() <= 0) {

// no L replica : nothing todo

} else if (zone_replica_attr_set.zone_set_.count() <= 0) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("zone set unexpected", K(ret), K(zone_replica_attr_set));

} else {

for (int64_t j = 0; j < logonly_unit_infos.count(); j++) {

const ObUnitInfo &unit_info = logonly_unit_infos.at(j);

if (!has_exist_in_array(zone_replica_attr_set.zone_set_, unit_info.unit_.zone_)) {

// bypass

} else if (zone_replica_attr_set.get_logonly_replica_num()

+ zone_replica_attr_set.get_encryption_logonly_replica_num() <= 0) {

// bypass

} else if (zone_replica_attr_set.get_logonly_replica_num() > 0) {

ret = zone_replica_attr_set.sub_logonly_replica_num(ReplicaAttr(1, 100));

} else {

ret = zone_replica_attr_set.sub_encryption_logonly_replica_num(ReplicaAttr(1, 100));

}

}

}

}

}

return ret;

}

int ObLocalityTaskHelp::get_logonly_task_with_logonly_unit(const uint64_t tenant_id,

ObUnitManager &unit_mgr,

share::schema::ObSchemaGetterGuard &schema_guard,

ObIArray<share::ObZoneReplicaNumSet> &zone_locality)

{

int ret = OB_SUCCESS;

ObArray<ObUnitInfo> logonly_unit_infos;

const ObTenantSchema *tenant_schema = NULL;

zone_locality.reset();

common::ObArray<share::ObZoneReplicaAttrSet> tenant_zone_locality;

if (OB_INVALID_ID == tenant_id) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret), K(tenant_id));

} else if (OB_FAIL(schema_guard.get_tenant_info(tenant_id, tenant_schema))) {

LOG_WARN("fail to get tenant info", K(ret), K(tenant_id));

} else if (OB_ISNULL(tenant_schema)) {

ret = OB_TENANT_NOT_EXIST;

LOG_WARN("get invalid tenant schema", K(ret), K(tenant_schema));

} else if (OB_FAIL(unit_mgr.get_logonly_unit_by_tenant(tenant_id, logonly_unit_infos))) {

LOG_WARN("fail to get logonly unit infos", K(ret), K(tenant_id));

} else if (OB_FAIL(tenant_schema->get_zone_replica_attr_array(tenant_zone_locality))) {

LOG_WARN("fail to get zone replica attr array", K(ret));

} else {

share::ObZoneReplicaNumSet logonly_set;

for (int64_t i = 0; i < logonly_unit_infos.count() && OB_SUCC(ret); i++) {

const ObUnitInfo &unit = logonly_unit_infos.at(i);

for (int64_t j = 0; j < tenant_zone_locality.count(); j++) {

logonly_set.reset();

const ObZoneReplicaNumSet &zone_set = tenant_zone_locality.at(j);

if (zone_set.zone_ == unit.unit_.zone_

&& zone_set.get_logonly_replica_num() == 1) {

logonly_set.zone_ = zone_set.zone_;

if (OB_FAIL(logonly_set.replica_attr_set_.add_logonly_replica_num(ReplicaAttr(1, 100)))) {

LOG_WARN("fail to add logonly replica num", K(ret));

} else if (OB_FAIL(zone_locality.push_back(logonly_set))) {

LOG_WARN("fail to push back", K(ret));

}

} else if (zone_set.zone_ == unit.unit_.zone_

&& zone_set.get_encryption_logonly_replica_num() == 1) {

logonly_set.zone_ = zone_set.zone_;

if (OB_FAIL(logonly_set.replica_attr_set_.add_encryption_logonly_replica_num(ReplicaAttr(1, 100)))) {

LOG_WARN("fail to add logonly replica num", K(ret));

} else if (OB_FAIL(zone_locality.push_back(logonly_set))) {

LOG_WARN("fail to push back", K(ret));

}

}

}

}

}

return ret;

}

int ObLocalityTaskHelp::filter_logonly_task(const uint64_t tenant_id,

ObUnitManager &unit_mgr,

share::schema::ObSchemaGetterGuard &schema_guard,

ObIArray<share::ObZoneReplicaAttrSet> &zone_locality)

{

int ret = OB_SUCCESS;

ObArray<ObUnitInfo> logonly_unit_infos;

if (OB_FAIL(unit_mgr.get_logonly_unit_by_tenant(schema_guard, tenant_id, logonly_unit_infos))) {

LOG_WARN("fail to get loggonly unit by tenant", K(ret), K(tenant_id));

} else {

LOG_DEBUG("get all logonly unit", K(tenant_id), K(logonly_unit_infos), K(zone_locality));

for (int64_t i = 0; i < zone_locality.count() && OB_SUCC(ret); ++i) {

share::ObZoneReplicaAttrSet &zone_replica_attr_set = zone_locality.at(i);

if (zone_replica_attr_set.get_logonly_replica_num()

+ zone_replica_attr_set.get_encryption_logonly_replica_num() <= 0) {

// no L replica : nothing todo

} else if (zone_replica_attr_set.zone_set_.count() <= 0) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("zone set unexpected", K(ret), K(zone_replica_attr_set));

} else {

for (int64_t j = 0; j < logonly_unit_infos.count(); j++) {

const ObUnitInfo &unit_info = logonly_unit_infos.at(j);

if (!has_exist_in_array(zone_replica_attr_set.zone_set_, unit_info.unit_.zone_)) {

// bypass

} else if (zone_replica_attr_set.get_logonly_replica_num()

+ zone_replica_attr_set.get_encryption_logonly_replica_num() <= 0) {

// bypass

} else if (zone_replica_attr_set.get_logonly_replica_num() > 0) {

ret = zone_replica_attr_set.sub_logonly_replica_num(ReplicaAttr(1, 100));

} else {

ret = zone_replica_attr_set.sub_encryption_logonly_replica_num(ReplicaAttr(1, 100));

}

}

}

}

}

return ret;

}

int ObLocalityTaskHelp::alloc_logonly_replica(ObUnitManager &unit_mgr,

const ObIArray<share::ObResourcePoolName> &pools,

const common::ObIArray<ObZoneReplicaNumSet> &zone_locality,

ObPartitionAddr &partition_addr)

{

int ret = OB_SUCCESS;

ObArray<ObUnitInfo> logonly_units;

ObArray<ObUnitInfo> unit_infos;

if (OB_FAIL(unit_mgr.get_unit_infos(pools, unit_infos))) {

LOG_WARN("fail to get unit infos", K(ret), K(pools));

} else {

for (int64_t i = 0; i < unit_infos.count() && OB_SUCC(ret); i++) {

if (REPLICA_TYPE_LOGONLY != unit_infos.at(i).unit_.replica_type_) {

//nothing todo

} else if (OB_FAIL(logonly_units.push_back(unit_infos.at(i)))) {

LOG_WARN("fail to push back", K(ret), K(i), K(unit_infos));

}

}

}

ObReplicaAddr raddr;

for (int64_t i = 0; i < logonly_units.count() && OB_SUCC(ret); i++) {

for (int64_t j = 0; j < zone_locality.count() && OB_SUCC(ret); j++) {

if (zone_locality.at(j).zone_ == logonly_units.at(i).unit_.zone_

&& (zone_locality.at(j).get_logonly_replica_num() == 1

|| zone_locality.at(j).get_encryption_logonly_replica_num() == 1)) {

raddr.reset();

raddr.unit_id_ = logonly_units.at(i).unit_.unit_id_;

raddr.addr_ = logonly_units.at(i).unit_.server_;

raddr.zone_ = logonly_units.at(i).unit_.zone_;

raddr.replica_type_ = zone_locality.at(j).get_logonly_replica_num() == 1

? REPLICA_TYPE_LOGONLY

: REPLICA_TYPE_ENCRYPTION_LOGONLY;

if (OB_FAIL(partition_addr.push_back(raddr))) {

LOG_WARN("fail to push back", K(ret), K(raddr));

} else {

LOG_INFO("alloc partition for logonly replica", K(raddr));

}

}

}

}

return ret;

}

int ObLocalityCheckHelp::calc_paxos_replica_num(

const common::ObIArray<share::ObZoneReplicaNumSet> &zone_locality,

int64_t &paxos_num)

{

int ret = OB_SUCCESS;

paxos_num = 0;

for (int64_t i = 0; OB_SUCC(ret) && i < zone_locality.count(); i++) {

if (0 > zone_locality.at(i).get_paxos_replica_num()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("invalid paxos replica num", K(ret), "zone_locality_set", zone_locality.at(i));

} else {

paxos_num += zone_locality.at(i).get_paxos_replica_num();

}

}

return ret;

}

/*

* the laws of alter locality:

* 1. each locality altering only allows to execute one operation: add paoxs, reduce paxos and paxos type transform(paxos->paxos, F->L).

* paxos->non_paxos is reduce paxos, non_paxos->paxos is add paxos.

* 2. in a locality altering:

* 2.1 if add paxos, need the orig_locality's paxos num >= majority(new_locality's paxos num);

* 2.2 if reduce paxos, need new_locality's paxos num >= majority(orig_locality's paxos num);

* 2.3 if paxos type transform, only one type transform is allowed for a locality altering.

* 3. the laws of type transform:

* 3.1 for L-replica, it is not allowed to transform replica_type other than F->L,

* and it is not allowed to transform L to other type.

* 4. enable paxos num 1->2. must not paxos num 2->1

* 5. there is no restriction on non-paoxs type transform

*/

int ObLocalityCheckHelp::check_alter_locality(

const ObIArray<share::ObZoneReplicaNumSet> &pre_zone_locality,

const ObIArray<share::ObZoneReplicaNumSet> &cur_zone_locality,

ObIArray<AlterPaxosLocalityTask> &alter_paxos_tasks,

bool &non_paxos_locality_modified,

int64_t &pre_paxos_num,

int64_t &cur_paxos_num,

const share::ObArbitrationServiceStatus &arb_service_status)

{

int ret = OB_SUCCESS;

pre_paxos_num = 0;

cur_paxos_num = 0;

alter_paxos_tasks.reset();

if (OB_UNLIKELY(pre_zone_locality.count() <= 0 || cur_zone_locality.count() <= 0)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret),

"pre_cnt", pre_zone_locality.count(),

"cur_cnt", cur_zone_locality.count());

} else if (OB_FAIL(get_alter_paxos_replica_number_replica_task(pre_zone_locality, cur_zone_locality,

alter_paxos_tasks, non_paxos_locality_modified))) {

LOG_WARN("fail to get alter paxos replica task", K(ret));

} else if (OB_FAIL(calc_paxos_replica_num(pre_zone_locality, pre_paxos_num))) {

LOG_WARN("fail to calc paxos replica num", K(ret));

} else if (OB_FAIL(calc_paxos_replica_num(cur_zone_locality, cur_paxos_num))) {

LOG_WARN("fail to calc paxos replica num", K(ret));

} else if (OB_FAIL(check_alter_locality_valid(alter_paxos_tasks,

pre_paxos_num, cur_paxos_num, arb_service_status))) {

LOG_WARN("check alter locality valid failed", K(ret), K(alter_paxos_tasks),

K(pre_paxos_num), K(cur_paxos_num), K(non_paxos_locality_modified), K(arb_service_status));

}

return ret;

}

/*

* Two zone set either are perfect matched or have intersection

*/

int ObLocalityCheckHelp::check_multi_zone_locality_intersect(

const common::ObIArray<share::ObZoneReplicaAttrSet> &multi_pre_zone_locality,

const common::ObIArray<share::ObZoneReplicaAttrSet> &multi_cur_zone_locality,

bool &intersect)

{

int ret = OB_SUCCESS;

intersect = false;

for (int64_t i = 0;

!intersect && OB_SUCC(ret) && i < multi_pre_zone_locality.count();

++i) {

for (int64_t j = 0;

!intersect && OB_SUCC(ret) && j < multi_cur_zone_locality.count();

++j) {

const share::ObZoneReplicaAttrSet &pre_locality = multi_pre_zone_locality.at(i);

const share::ObZoneReplicaAttrSet &cur_locality = multi_cur_zone_locality.at(j);

const common::ObIArray<common::ObZone> &pre_zone_set = pre_locality.zone_set_;

const common::ObIArray<common::ObZone> &cur_zone_set = cur_locality.zone_set_;

if (pre_zone_set.count() <= 1 || cur_zone_set.count() <= 1) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("zone set count unexpected", K(ret));

} else {

intersect = check_zone_set_intersect(pre_zone_set, cur_zone_set);

}

}

}

return ret;

}

bool ObLocalityCheckHelp::check_zone_set_intersect(

const common::ObIArray<common::ObZone> &pre_zone_set,

const common::ObIArray<common::ObZone> &cur_zone_set)

{

bool intersect = false;

if (pre_zone_set.count() == cur_zone_set.count()) {

// either there is no crossover or there is a perfect match

int64_t counter = 0;

for (int64_t i = 0; i < pre_zone_set.count(); ++i) {

const common::ObZone &this_pre_zone = pre_zone_set.at(i);

if (has_exist_in_array(cur_zone_set, this_pre_zone)) {

counter++;

}

}

if (0 == counter || pre_zone_set.count() == counter) {

intersect = false;

} else {

intersect = true;

}

} else {

// must no cross

intersect = false;

for (int64_t i = 0; !intersect && i < pre_zone_set.count(); ++i) {

const common::ObZone &this_pre_zone = pre_zone_set.at(i);

if (has_exist_in_array(cur_zone_set, this_pre_zone)) {

intersect = true;

}

}

}

return intersect;

}

int ObLocalityCheckHelp::split_single_and_multi_zone_locality(

const common::ObIArray<share::ObZoneReplicaAttrSet> &zone_locality,

common::ObIArray<share::ObZoneReplicaAttrSet> &single_zone_locality,

common::ObIArray<share::ObZoneReplicaAttrSet> &multi_zone_locality)

{

int ret = OB_SUCCESS;

if (OB_UNLIKELY(zone_locality.count() <= 0)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret),

"zone_locality_cnt", zone_locality.count());

} else {

single_zone_locality.reset();

multi_zone_locality.reset();

for (int64_t i = 0; OB_SUCC(ret) && i < zone_locality.count(); ++i) {

const share::ObZoneReplicaAttrSet &this_set = zone_locality.at(i);

if (this_set.zone_set_.count() <= 0) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("zone set count unexpected", K(ret), "zone_set_cnt", this_set.zone_set_.count());

} else if (this_set.zone_set_.count() == 1) {

if (OB_FAIL(single_zone_locality.push_back(this_set))) {

LOG_WARN("fail to push back", K(ret));

}

} else {

if (OB_FAIL(multi_zone_locality.push_back(this_set))) {

LOG_WARN("fail to push back", K(ret));

}

}

}

}

return ret;

}

int ObLocalityCheckHelp::single_zone_locality_search(

const share::ObZoneReplicaAttrSet &this_locality,

const common::ObIArray<share::ObZoneReplicaAttrSet> &single_zone_locality,

const common::ObIArray<share::ObZoneReplicaAttrSet> &multi_zone_locality,

SingleZoneLocalitySearch &search_flag,

int64_t &search_index,

const share::ObZoneReplicaAttrSet *&search_zone_locality)

{

int ret = OB_SUCCESS;

if (OB_UNLIKELY(this_locality.zone_set_.count() != 1)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret));

} else {

bool found = false;

const common::ObZone &compare_zone = this_locality.zone_set_.at(0);

for (int64_t i = 0; !found && OB_SUCC(ret) && i < single_zone_locality.count(); ++i) {

const share::ObZoneReplicaAttrSet &this_zone_locality = single_zone_locality.at(i);

if (this_zone_locality.zone_set_.count() != 1) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("this zone set unexpected", K(ret),

"zone_set_count", this_zone_locality.zone_set_.count());

} else if (compare_zone != this_zone_locality.zone_set_.at(0)) {

// go on check next

} else {

found = true;

search_flag = SingleZoneLocalitySearch::SZLS_IN_ASSOC_SINGLE;

search_index = i;

search_zone_locality = &this_zone_locality;

}

}

for (int64_t i = 0; !found && OB_SUCC(ret) && i < multi_zone_locality.count(); ++i) {

const share::ObZoneReplicaAttrSet &this_zone_locality = multi_zone_locality.at(i);

if (this_zone_locality.zone_set_.count() <= 1) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("this zone set unexpected", K(ret),

"zone_set_count", this_zone_locality.zone_set_.count());

} else if (!has_exist_in_array(this_zone_locality.zone_set_, compare_zone)) {

// go on check next

} else {

found = true;

search_flag = SingleZoneLocalitySearch::SZLS_IN_ASSOC_MULTI;

search_index = i;

search_zone_locality = &this_zone_locality;

}

}

if (OB_SUCC(ret)) {

if (!found) {

search_flag = SingleZoneLocalitySearch::SZLS_NOT_FOUND;

search_index = -1;

search_zone_locality = nullptr;

}

}

}

return ret;

}

int ObLocalityCheckHelp::try_add_single_zone_alter_paxos_task(

const share::ObZoneReplicaAttrSet &pre_locality,

const share::ObZoneReplicaAttrSet &cur_locality,

common::ObIArray<AlterPaxosLocalityTask> &alter_paxos_tasks,

bool &non_paxos_locality_modified)

{

int ret = OB_SUCCESS;

if (OB_UNLIKELY(pre_locality.zone_set_.count() != 1)

|| OB_UNLIKELY(cur_locality.zone_set_.count() != 1)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret), K(pre_locality), K(cur_locality));

} else if (OB_FAIL(check_alter_single_zone_locality_valid(

cur_locality, pre_locality, non_paxos_locality_modified))) {

LOG_WARN("alter single zone locality is invalid", K(ret),

K(cur_locality), K(pre_locality), K(non_paxos_locality_modified));

} else {

AlterPaxosLocalityTask alter_paxos_task;

if (pre_locality.get_paxos_replica_num() == cur_locality.get_paxos_replica_num()) {

if (!pre_locality.is_paxos_locality_match(cur_locality)) {

alter_paxos_task.reset();

alter_paxos_task.task_type_ = NOP_PAXOS_REPLICA_NUMBER;

if (cur_locality.get_full_replica_num() > pre_locality.get_full_replica_num()) {

if (OB_FAIL(alter_paxos_task.associated_replica_type_set_.push_back(REPLICA_TYPE_FULL))) {

LOG_WARN("fail to push back", K(ret));

}

}

if (OB_SUCC(ret)

&& cur_locality.get_logonly_replica_num()

> pre_locality.get_logonly_replica_num()) {

if (OB_FAIL(alter_paxos_task.associated_replica_type_set_.push_back(REPLICA_TYPE_LOGONLY))) {

LOG_WARN("fail to push back", K(ret));