/**

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

*/

#include "share/ob_cluster_version.h"

#define USING_LOG_PREFIX SQL_ENG

#include "sql/engine/cmd/ob_table_executor.h"

#include "sql/engine/cmd/ob_index_executor.h"

#include "sql/engine/cmd/ob_ddl_executor_util.h"

#include "share/object/ob_obj_cast.h"

#include "lib/mysqlclient/ob_mysql_proxy.h"

#include "lib/utility/ob_tracepoint.h"

#include "share/ob_common_rpc_proxy.h"

#include "share/ob_ddl_error_message_table_operator.h"

#include "sql/resolver/ddl/ob_create_table_stmt.h"

#include "sql/resolver/ddl/ob_alter_table_stmt.h"

#include "sql/resolver/ddl/ob_drop_table_stmt.h"

#include "sql/resolver/ddl/ob_rename_table_stmt.h"

#include "sql/resolver/ddl/ob_truncate_table_stmt.h"

#include "sql/resolver/ddl/ob_create_table_like_stmt.h"

#include "sql/resolver/ddl/ob_flashback_stmt.h"

#include "sql/resolver/ddl/ob_purge_stmt.h"

#include "sql/resolver/ddl/ob_optimize_stmt.h"

#include "sql/resolver/dml/ob_delete_stmt.h"

#include "sql/resolver/dml/ob_delete_resolver.h"

#include "sql/resolver/ob_resolver_utils.h"

#include "sql/engine/ob_exec_context.h"

#include "sql/engine/ob_physical_plan.h"

#include "sql/session/ob_sql_session_info.h"

#include "sql/code_generator/ob_expr_generator_impl.h"

#include "sql/engine/cmd/ob_partition_executor_utils.h"

#include "sql/parser/ob_parser.h"

#include "share/system_variable/ob_sys_var_class_type.h"

#include "sql/ob_select_stmt_printer.h"

#include "observer/ob_server_struct.h"

#include "observer/ob_server.h"

#include "observer/ob_server_event_history_table_operator.h"

#include "lib/worker.h"

#include "share/external_table/ob_external_table_file_mgr.h"

#include "share/external_table/ob_external_table_file_task.h"

#include "share/external_table/ob_external_table_file_rpc_proxy.h"

#include "observer/ob_srv_network_frame.h"

#include "observer/dbms_job/ob_dbms_job_master.h"

#include "observer/ob_inner_sql_connection_pool.h"

#include "share/backup/ob_backup_io_adapter.h"

#include "share/external_table/ob_external_table_file_rpc_processor.h"

#include "share/external_table/ob_external_table_utils.h"

#include "share/ob_debug_sync.h"

#include "share/schema/ob_schema_utils.h"

namespace oceanbase

{

using namespace common;

using namespace share;

using namespace share::schema;

using namespace observer;

namespace sql

{

ObCreateTableExecutor::ObCreateTableExecutor()

{

}

ObCreateTableExecutor::~ObCreateTableExecutor()

{

}

int ObCreateTableExecutor::prepare_stmt(ObCreateTableStmt &stmt,

const ObSQLSessionInfo &my_session,

ObString &create_table_name)

{

int ret = OB_SUCCESS;

ObArenaAllocator allocator("CreateTableExec");

const int64_t buf_len = OB_MAX_SQL_LENGTH;

char *buf = static_cast<char*>(allocator.alloc(buf_len));

int64_t pos = 0;

const int64_t session_id = my_session.get_sessid();

const int64_t timestamp = ObTimeUtility::current_time();

obrpc::ObCreateTableArg &create_table_arg = stmt.get_create_table_arg();

create_table_name = create_table_arg.schema_.get_table_name_str();

if (OB_FAIL(databuff_printf(buf, buf_len, pos, "__ctas_%ld_%ld", session_id, timestamp))) {

LOG_WARN("failed to print tmp table name", K(ret));

} else {

ObString tmp_table_name(pos, buf);

if (OB_FAIL(create_table_arg.schema_.set_table_name(tmp_table_name))) {

LOG_WARN("failed to set tmp table name", K(ret));

}

}

return ret;

}

int ObCreateTableExecutor::ObInsSQLPrinter::inner_print(char *buf, int64_t buf_len, int64_t &res_len)

{

int ret = OB_SUCCESS;

const char sep_char = lib::is_oracle_mode()? '"': '`';

const ObSelectStmt *select_stmt = NULL;

int64_t pos1 = 0;

if (OB_ISNULL(stmt_) || OB_ISNULL(select_stmt= stmt_->get_sub_select())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("null stmt", K(ret));

} else if (OB_FAIL(databuff_printf(buf, buf_len, pos1,

do_osg_

? "insert /*+GATHER_OPTIMIZER_STATISTICS*/ into %c%.*s%c.%c%.*s%c"

: "insert /*+NO_GATHER_OPTIMIZER_STATISTICS*/ into %c%.*s%c.%c%.*s%c",

sep_char,

stmt_->get_database_name().length(),

stmt_->get_database_name().ptr(),

sep_char,

sep_char,

stmt_->get_table_name().length(),

stmt_->get_table_name().ptr(),

sep_char))) {

LOG_WARN("fail to print insert into string", K(ret));

} else if (lib::is_oracle_mode()) {

const ObTableSchema &table_schema = stmt_->get_create_table_arg().schema_;

int64_t used_column_count = 0;

for (int64_t i = 0; OB_SUCC(ret) && i < table_schema.get_column_count(); ++i) {

const ObColumnSchemaV2 *column_schema = table_schema.get_column_schema_by_idx(i);

if (OB_ISNULL(column_schema)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("get null column schema", K(ret));

} else if (column_schema->get_column_id() < OB_END_RESERVED_COLUMN_ID_NUM ||

column_schema->is_udt_hidden_column()) {

// do nothing

} else if (OB_FAIL(databuff_printf(buf, buf_len, pos1, (0 == used_column_count)? "(": ", "))) {

LOG_WARN("failed to print insert into string", K(ret), K(i));

} else if (OB_FAIL(databuff_printf(buf, buf_len, pos1, "%c%.*s%c", sep_char, LEN_AND_PTR(column_schema->get_column_name_str()), sep_char))) {

LOG_WARN("failed to print insert into string", K(ret));

} else {

++ used_column_count;

}

}

} else {

for (int64_t i = 0; OB_SUCC(ret) && i < select_stmt->get_select_item_size(); ++i) {

const SelectItem &select_item = select_stmt->get_select_item(i);

if (OB_FAIL(databuff_printf(buf, buf_len, pos1, (0 == i)? "(": ", "))) {

LOG_WARN("failed to print insert into string", K(ret), K(i));

} else { /* do nothing */ }

if (OB_SUCC(ret)) {

if (!select_item.alias_name_.empty()) {

if (OB_FAIL(databuff_printf(buf, buf_len, pos1, "%c%.*s%c", sep_char, LEN_AND_PTR(select_item.alias_name_), sep_char))) {

LOG_WARN("failed to print insert into string", K(ret));

} else { /* do nothing */ }

} else {

if (OB_FAIL(databuff_printf(buf, buf_len, pos1, "%c%.*s%c", sep_char, LEN_AND_PTR(select_item.expr_name_), sep_char))) {

LOG_WARN("failed to print insert into string", K(ret));

} else { /* do nothing */ }

}

} else { /* do nothing */ }

}

}

if (OB_SUCC(ret)) {

ObSelectStmtPrinter select_stmt_printer(buf, buf_len, &pos1, select_stmt,

schema_guard_,

print_params_,

param_store_,

true);

select_stmt_printer.set_is_first_stmt_for_hint(true); // need print global hint

if (OB_FAIL(databuff_printf(buf, buf_len, pos1, ") "))) {

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

} else if (OB_FAIL(select_stmt_printer.do_print())) {

LOG_WARN("fail to print select stmt", K(ret));

} else {

res_len = pos1;

}

}

return ret;

}

//准备查询插入的脚本

int ObCreateTableExecutor::prepare_ins_arg(ObCreateTableStmt &stmt,

const ObSQLSessionInfo *my_session,

ObSchemaGetterGuard *schema_guard,

const ParamStore *param_store,

ObSqlString &ins_sql) //out, 最终的查询插入语句

{

int ret = OB_SUCCESS;

ObArenaAllocator allocator("CreateTableExec");

char *buf = static_cast<char*>(allocator.alloc(OB_MAX_SQL_LENGTH));

int64_t buf_len = OB_MAX_SQL_LENGTH;

int64_t pos1 = 0;

bool is_oracle_mode = lib::is_oracle_mode();

bool no_osg_hint = false;

bool online_sys_var = false;

ObSelectStmt *select_stmt = stmt.get_sub_select();

ObObjPrintParams obj_print_params(select_stmt->get_query_ctx()->get_timezone_info());

obj_print_params.print_origin_stmt_ = true;

if (OB_ISNULL(buf)) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_ERROR("allocate memory failed");

} else if (OB_ISNULL(select_stmt) || OB_ISNULL(select_stmt->get_query_ctx()) || OB_ISNULL(my_session)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("select stmt should not be null", K(ret));

} else {

//get hint

no_osg_hint = select_stmt->get_query_ctx()->get_global_hint().has_no_gather_opt_stat_hint();

//get system variable

ObObj online_sys_var_obj;

if (OB_FAIL(OB_FAIL(my_session->get_sys_variable(SYS_VAR__OPTIMIZER_GATHER_STATS_ON_LOAD, online_sys_var_obj)))) {

LOG_WARN("fail to get sys var", K(ret));

} else {

online_sys_var = online_sys_var_obj.get_bool();

LOG_DEBUG("online opt stat gather", K(online_sys_var), K(no_osg_hint));

}

}

if (OB_SUCC(ret)) {

ObInsSQLPrinter sql_printer(&stmt, schema_guard, obj_print_params, param_store, !no_osg_hint && online_sys_var);

ObString sql;

if (OB_FAIL(sql_printer.do_print(allocator, sql))) {

LOG_WARN("failed to print", K(ret));

} else if (OB_FAIL(ins_sql.append(sql))){

LOG_WARN("fail to append insert into string", K(ret));

}

}

if (OB_SUCC(ret)) {

ObString converted_sql = ins_sql.string();

if (OB_FAIL(ObSQLUtils::convert_sql_text_from_schema_for_resolve(allocator,

my_session->get_dtc_params(),

converted_sql,

ObCharset::COPY_STRING_ON_SAME_CHARSET))) {

LOG_WARN("fail to convert insert into string to client_cs_type", K(ret));

} else if (OB_FAIL(ins_sql.assign(converted_sql))) {

LOG_WARN("fail to assign converted insert into string", K(ret));

}

}

LOG_DEBUG("ins str preparation complete!", K(ins_sql), K(ret), K(lib::is_oracle_mode()));

return ret;

}

//准备alter table 的参数

int ObCreateTableExecutor::prepare_alter_arg(ObCreateTableStmt &stmt,

const ObSQLSessionInfo *my_session,

const ObString &create_table_name,

obrpc::ObAlterTableArg &alter_table_arg) //out, 最终的alter table arg, set session_id = 0;

{

int ret = OB_SUCCESS;

const obrpc::ObCreateTableArg &create_table_arg = stmt.get_create_table_arg();

ObTableSchema &table_schema = const_cast<obrpc::ObCreateTableArg&>(create_table_arg).schema_;

AlterTableSchema *alter_table_schema = &alter_table_arg.alter_table_schema_;

table_schema.set_session_id(my_session->get_sessid_for_table());

alter_table_arg.session_id_ = my_session->get_sessid_for_table();

alter_table_schema->alter_type_ = OB_DDL_ALTER_TABLE;

//compat for old server

alter_table_arg.tz_info_ = my_session->get_tz_info_wrap().get_tz_info_offset();

alter_table_arg.is_inner_ = my_session->is_inner();

alter_table_arg.exec_tenant_id_ = my_session->get_effective_tenant_id();

if (OB_FAIL(alter_table_arg.tz_info_wrap_.deep_copy(my_session->get_tz_info_wrap()))) {

LOG_WARN("failed to deep_copy tz info wrap", "tz_info_wrap", my_session->get_tz_info_wrap(), K(ret));

} else if (OB_FAIL(alter_table_arg.set_nls_formats(

my_session->get_local_nls_date_format(),

my_session->get_local_nls_timestamp_format(),

my_session->get_local_nls_timestamp_tz_format()))) {

LOG_WARN("failed to set_nls_formats", K(ret));

} else if (OB_FAIL(alter_table_schema->assign(table_schema))) {

LOG_WARN("failed to assign alter table schema", K(ret));

} else if (!table_schema.is_mysql_tmp_table()

&& FALSE_IT(alter_table_schema->set_session_id(0))) {

//impossible

} else if (OB_FAIL(alter_table_schema->set_origin_table_name(stmt.get_table_name()))) {

LOG_WARN("failed to set origin table name", K(ret));

} else if (OB_FAIL(alter_table_schema->set_origin_database_name(stmt.get_database_name()))) {

LOG_WARN("failed to set origin database name", K(ret));

} else if (OB_FAIL(alter_table_schema->set_table_name(create_table_name))) {

LOG_WARN("failed to set table name", K(ret));

} else if (OB_FAIL(alter_table_schema->set_database_name(stmt.get_database_name()))) {

LOG_WARN("failed to set database name", K(ret));

} else if (!table_schema.is_mysql_tmp_table()

&& OB_FAIL(alter_table_schema->alter_option_bitset_.add_member(obrpc::ObAlterTableArg::SESSION_ID))) {

LOG_WARN("failed to add member SESSION_ID for alter table schema", K(ret), K(alter_table_arg));

} else if (OB_FAIL(alter_table_schema->alter_option_bitset_.add_member(obrpc::ObAlterTableArg::TABLE_NAME))) {

LOG_WARN("failed to add member TABLE_NAME for alter table schema", K(ret), K(alter_table_arg));

}

LOG_DEBUG("alter table arg preparation complete!", K(*alter_table_schema), K(ret));

return ret;

}

//准备drop table的参数

int ObCreateTableExecutor::prepare_drop_arg(const ObCreateTableStmt &stmt,

const ObSQLSessionInfo *my_session,

obrpc::ObTableItem &table_item,

obrpc::ObDropTableArg &drop_table_arg) //out, drop table的参数

{

int ret = OB_SUCCESS;

const ObString &db_name = stmt.get_database_name();

const ObString &tab_name = stmt.get_table_name();

drop_table_arg.if_exist_ = true;

drop_table_arg.tenant_id_ = my_session->get_login_tenant_id();

drop_table_arg.to_recyclebin_ = false;

drop_table_arg.table_type_ = USER_TABLE;

drop_table_arg.session_id_ = my_session->get_sessid_for_table();

drop_table_arg.exec_tenant_id_ = my_session->get_effective_tenant_id();

int64_t foreign_key_checks = 0;

my_session->get_foreign_key_checks(foreign_key_checks);

drop_table_arg.foreign_key_checks_ = is_oracle_mode() || (is_mysql_mode() && foreign_key_checks);

table_item.database_name_ = db_name;

table_item.table_name_ = tab_name;

if (OB_FAIL(my_session->get_name_case_mode(table_item.mode_))) {

LOG_WARN("failed to get name case mode!", K(ret));

} else if (OB_FAIL(drop_table_arg.tables_.push_back(table_item))) {

LOG_WARN("failed to add table item!", K(table_item), K(ret));

}

LOG_DEBUG("drop table arg preparation complete!", K(drop_table_arg), K(table_item), K(ret));

return ret;

}

//查询建表的处理, 通过内部session执行查询插入代码参考了 ObTableModify::ObTableModifyCtx::open_inner_conn() 实现

int ObCreateTableExecutor::execute_ctas(ObExecContext &ctx,

ObCreateTableStmt &stmt,

obrpc::ObCommonRpcProxy *common_rpc_proxy)

{

int ret = OB_SUCCESS;

ObString cur_query;

int64_t affected_rows = 0;

ObMySQLProxy *sql_proxy = ctx.get_sql_proxy();

common::ObCommonSqlProxy *user_sql_proxy;

common::ObOracleSqlProxy oracle_sql_proxy;

ObSQLSessionInfo *my_session = ctx.get_my_session();

ObPhysicalPlanCtx *plan_ctx = ctx.get_physical_plan_ctx();

ObArenaAllocator allocator("CreateTableExec");

HEAP_VAR(obrpc::ObAlterTableArg, alter_table_arg) {

obrpc::ObDropTableArg drop_table_arg;

obrpc::ObTableItem table_item;

ObString create_table_name;

ObSqlString ins_sql;

const observer::ObGlobalContext &gctx = observer::ObServer::get_instance().get_gctx();

obrpc::ObCreateTableRes create_table_res;

obrpc::ObCreateTableArg &create_table_arg = stmt.get_create_table_arg();

create_table_arg.is_inner_ = my_session->is_inner();

bool need_clean = true;

CK(OB_NOT_NULL(sql_proxy),

OB_NOT_NULL(my_session),

OB_NOT_NULL(gctx.schema_service_),

OB_NOT_NULL(plan_ctx),

OB_NOT_NULL(common_rpc_proxy),

OB_NOT_NULL(ctx.get_sql_ctx()));

if (OB_SUCC(ret)) {

if (OB_FAIL(ob_write_string(allocator, my_session->get_current_query_string(), cur_query))) {

LOG_WARN("failed to write string to session", K(ret));

}

}

if (OB_SUCC(ret)) {

ObInnerSQLConnectionPool *pool = static_cast<observer::ObInnerSQLConnectionPool*>(sql_proxy->get_pool());

if (OB_ISNULL(pool)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("pool is null", K(ret));

} else if (OB_FAIL(oracle_sql_proxy.init(pool))) {

LOG_WARN("init oracle sql proxy failed", K(ret));

} else if (OB_FAIL(prepare_stmt(stmt, *my_session, create_table_name))) {

LOG_WARN("failed to prepare stmt", K(ret));

} else if (OB_FAIL(prepare_ins_arg(stmt, my_session, ctx.get_sql_ctx()->schema_guard_, &plan_ctx->get_param_store(), ins_sql))) { //1, 参数准备;

LOG_WARN("failed to prepare insert table arg", K(ret));

} else if (OB_FAIL(prepare_alter_arg(stmt, my_session, create_table_name, alter_table_arg))) {

LOG_WARN("failed to prepare alter table arg", K(ret));

} else if (OB_FAIL(prepare_drop_arg(stmt, my_session, table_item, drop_table_arg))) {

LOG_WARN("failed to prepare drop table arg", K(ret));

} else if (OB_FAIL(ctx.get_sql_ctx()->schema_guard_->reset())){

LOG_WARN("schema_guard reset failed", K(ret));

} else if (OB_FAIL(common_rpc_proxy->create_table(create_table_arg, create_table_res))) { //2, 建表;

LOG_WARN("rpc proxy create table failed", K(ret), "dst", common_rpc_proxy->get_server());

} else if (OB_INVALID_ID != create_table_res.table_id_) { //如果表已存在则后续的查询插入不进行

if (OB_INVALID_VERSION == create_table_res.schema_version_) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("get unexpected schema version", K(ret), K(create_table_res));

} else {

uint64_t tenant_id = my_session->get_effective_tenant_id();

if (OB_FAIL(gctx.schema_service_->async_refresh_schema(tenant_id,

create_table_res.schema_version_))) {

LOG_WARN("failed to async refresh schema", K(ret));

}

}

#ifdef ERRSIM

{

int tmp_ret = OB_E(EventTable::EN_CTAS_FAIL_NO_DROP_ERROR) OB_SUCCESS; //错误注入, 使表不能清理

if (OB_FAIL(tmp_ret)) {

ret = tmp_ret;

need_clean = false;

}

}

#else

//do nothing...

#endif

//3, 插入数据

if (OB_SUCC(ret)) {

bool is_mysql_temp_table = stmt.get_create_table_arg().schema_.is_mysql_tmp_table();

bool in_trans = my_session->is_in_transaction();

ObBasicSessionInfo::UserScopeGuard user_scope_guard(my_session->get_sql_scope_flags());

common::sqlclient::ObISQLConnection *conn = NULL;

const uint64_t tenant_id = my_session->get_effective_tenant_id();

if (lib::is_oracle_mode()) {

user_sql_proxy = &oracle_sql_proxy;

} else {

user_sql_proxy = sql_proxy;

}

if (OB_FAIL(pool->acquire(my_session, conn))) {

LOG_WARN("failed to acquire inner connection", K(ret));

} else if (OB_ISNULL(conn)) {

ret = OB_INNER_STAT_ERROR;

LOG_WARN("connection can not be NULL", K(ret));

} else if ((!is_mysql_temp_table || !in_trans)

&& OB_FAIL(conn->start_transaction(tenant_id))) {

LOG_WARN("failed start transaction", K(ret), K(tenant_id));

} else {

if (OB_FAIL(conn->execute_write(tenant_id, ins_sql.ptr(),

affected_rows, true))) {

LOG_WARN("failed to exec sql", K(tenant_id), K(ins_sql), K(ret));

}

// transaction started, must commit or rollback

int tmp_ret = OB_SUCCESS;

if (!is_mysql_temp_table || !in_trans) {

if (OB_LIKELY(OB_SUCCESS == ret)) {

tmp_ret = conn->commit();

} else {

int64_t MIN_ROLLBACK_TIMEOUT = 10 * 1000 * 1000;// 10s

int64_t origin_timeout_ts = THIS_WORKER.get_timeout_ts();

if (INT64_MAX != origin_timeout_ts &&

origin_timeout_ts < ObTimeUtility::current_time() + MIN_ROLLBACK_TIMEOUT) {

THIS_WORKER.set_timeout_ts(ObTimeUtility::current_time() + MIN_ROLLBACK_TIMEOUT);

LOG_INFO("set timeout for rollback", K(origin_timeout_ts),

K(ObTimeUtility::current_time() + MIN_ROLLBACK_TIMEOUT));

}

tmp_ret = conn->rollback();

}

if (OB_UNLIKELY(OB_SUCCESS != tmp_ret)) {

ret = (OB_SUCCESS == ret) ? tmp_ret : ret;

LOG_WARN("fail to end transaction", K(ret), K(tmp_ret));

}

}

}

if (OB_NOT_NULL(conn)) {

user_sql_proxy->close(conn, true);

}

}

DEBUG_SYNC(BEFORE_EXECUTE_CTAS_CLEAR_SESSION_ID);

//4, 刷新schema, 将table的sess id重置为0

if (OB_SUCC(ret)) {

obrpc::ObAlterTableRes res;

alter_table_arg.compat_mode_ = ORACLE_MODE == my_session->get_compatibility_mode() ?

lib::Worker::CompatMode::ORACLE : lib::Worker::CompatMode::MYSQL;

if (OB_FAIL(common_rpc_proxy->alter_table(alter_table_arg, res))) {

LOG_WARN("failed to update table session", K(ret), K(alter_table_arg));

if (alter_table_arg.compat_mode_ == lib::Worker::CompatMode::ORACLE && OB_ERR_TABLE_EXIST == ret) {

ret = OB_ERR_EXIST_OBJECT;

}

}

}

if (OB_FAIL(ret)) { //5, 查询建表失败, 需要清理环境即DROP TABLE

my_session->update_last_active_time();

if (OB_LIKELY(need_clean)) {

int tmp_ret = OB_SUCCESS;

obrpc::ObDDLRes res;

drop_table_arg.compat_mode_ = ORACLE_MODE == my_session->get_compatibility_mode() ?

lib::Worker::CompatMode::ORACLE : lib::Worker::CompatMode::MYSQL;

if (OB_SUCCESS != (tmp_ret = common_rpc_proxy->drop_table(drop_table_arg, res))) {

LOG_WARN("failed to drop table", K(drop_table_arg), K(ret));

} else {

LOG_INFO("table is created and dropped due to error ", K(ret));

}

}

} else {

plan_ctx->set_affected_rows(affected_rows);

LOG_DEBUG("CTAS all done", K(ins_sql), K(affected_rows), K(lib::is_oracle_mode()));

}

if (OB_ERR_TABLE_EXIST == ret && create_table_arg.if_not_exist_) {

ret = OB_SUCCESS;

LOG_DEBUG("table exists, force return success after cleanup", K(create_table_name));

}

} else {

LOG_DEBUG("table exists, no need to CTAS", K(create_table_res.table_id_));

}

if (OB_NOT_NULL(common_rpc_proxy)) {

char table_info_buffer[256];

snprintf(table_info_buffer, sizeof(table_info_buffer), "table_id:%ld, hidden_table_id:%ld",

alter_table_arg.table_id_, alter_table_arg.hidden_table_id_);

SERVER_EVENT_ADD("ddl", "create table as select execute finish",

"tenant_id", MTL_ID(),

"ret", ret,

"trace_id", *ObCurTraceId::get_trace_id(),

"rpc_dst", common_rpc_proxy->get_server(),

"table_info", table_info_buffer,

"schema_version", create_table_res.schema_version_);

}

SQL_ENG_LOG(INFO, "finish create table execute.", K(ret), "ddl_event_info", ObDDLEventInfo(), K(stmt), K(create_table_arg), K(alter_table_arg));

}

OZ(my_session->store_query_string(cur_query));

}

return ret;

}

int ObCreateTableExecutor::execute(ObExecContext &ctx, ObCreateTableStmt &stmt)

{

int ret = OB_SUCCESS;

ObTaskExecutorCtx *task_exec_ctx = NULL;

obrpc::ObCommonRpcProxy *common_rpc_proxy = NULL;

obrpc::ObCreateTableRes res;

obrpc::ObCreateTableArg &create_table_arg = stmt.get_create_table_arg();

ObString first_stmt;

ObSelectStmt *select_stmt = stmt.get_sub_select();

ObTableSchema &table_schema = create_table_arg.schema_;

ObSQLSessionInfo *my_session = ctx.get_my_session();

uint64_t tenant_id = table_schema.get_tenant_id();

uint64_t data_version = 0;

if (OB_ISNULL(my_session)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("session is null", K(ret));

} else if (OB_FAIL(stmt.get_first_stmt(first_stmt))) {

LOG_WARN("get first statement failed", K(ret));

} else if (OB_FAIL(GET_MIN_DATA_VERSION(tenant_id, data_version))) {

LOG_WARN("fail to get data version", KR(ret), K(tenant_id));

} else if (table_schema.is_duplicate_table()) {

bool is_compatible = false;

if (OB_FAIL(ObShareUtil::check_compat_version_for_readonly_replica(tenant_id, is_compatible))) {

LOG_WARN("fail to check data version for duplicate table", KR(ret), K(tenant_id));

} else if (!is_compatible) {

ret = OB_NOT_SUPPORTED;

LOG_WARN("duplicate table is not supported below 4.2", KR(ret), K(table_schema), K(is_compatible));

LOG_USER_ERROR(OB_NOT_SUPPORTED, "create duplicate table below 4.2");

} else if (is_sys_tenant(tenant_id) || is_meta_tenant(tenant_id)) {

// TODO@jingyu_cr: make sure whether sys log stream have to be duplicated

ret = OB_NOT_SUPPORTED;

LOG_USER_ERROR(OB_NOT_SUPPORTED, "create duplicate table under sys or meta tenant");

LOG_WARN("create dup table not supported", KR(ret), K(table_schema));

}

}

if (OB_FAIL(ret)) {

} else {

create_table_arg.is_inner_ = my_session->is_inner();

create_table_arg.consumer_group_id_ = THIS_WORKER.get_group_id();

const_cast<obrpc::ObCreateTableArg&>(create_table_arg).ddl_stmt_str_ = first_stmt;

bool enable_parallel_create_table = false;

{

omt::ObTenantConfigGuard tenant_config(TENANT_CONF(tenant_id));

enable_parallel_create_table = tenant_config.is_valid()

&& tenant_config->_enable_parallel_table_creation;

}

if (OB_ISNULL(task_exec_ctx = GET_TASK_EXECUTOR_CTX(ctx))) {

ret = OB_NOT_INIT;

LOG_WARN("get task executor context failed", K(ret));

} else if (!table_schema.is_external_table() //external table can not define partitions by create table stmt

&& OB_FAIL(ObPartitionExecutorUtils::calc_values_exprs(ctx, stmt))) {

LOG_WARN("compare range parition expr fail", K(ret));

} else if (OB_FAIL(set_index_arg_list(ctx, stmt))) {

LOG_WARN("fail to set index_arg_list", K(ret));

} else if (OB_FAIL(task_exec_ctx->get_common_rpc(common_rpc_proxy))) {

LOG_WARN("get common rpc proxy failed", K(ret));

} else if (OB_ISNULL(common_rpc_proxy)){

ret = OB_ERR_UNEXPECTED;

LOG_WARN("common rpc proxy should not be null", K(ret));

} else if (OB_ISNULL(select_stmt)) { // 普通建表的处理

if (OB_FAIL(ctx.get_sql_ctx()->schema_guard_->reset())){

LOG_WARN("schema_guard reset failed", KR(ret));

} else if (table_schema.is_view_table()

|| data_version < DATA_VERSION_4_2_1_0

|| !enable_parallel_create_table) {

if (OB_FAIL(common_rpc_proxy->create_table(create_table_arg, res))) {

LOG_WARN("rpc proxy create table failed", KR(ret), "dst", common_rpc_proxy->get_server());

}

} else {

DEBUG_SYNC(BEFORE_SEND_PARALLEL_CREATE_TABLE);

int64_t start_time = ObTimeUtility::current_time();

ObTimeoutCtx ctx;

if (OB_FAIL(ctx.set_timeout(common_rpc_proxy->get_timeout()))) {

LOG_WARN("fail to set timeout ctx", K(ret));

} else if (OB_FAIL(common_rpc_proxy->parallel_create_table(create_table_arg, res))) {

LOG_WARN("rpc proxy create table failed", KR(ret), "dst", common_rpc_proxy->get_server());

} else {

int64_t refresh_time = ObTimeUtility::current_time();

if (OB_FAIL(ObSchemaUtils::try_check_parallel_ddl_schema_in_sync(

ctx, tenant_id, res.schema_version_))) {

LOG_WARN("fail to check paralleld ddl schema in sync", KR(ret), K(res));

}

int64_t end_time = ObTimeUtility::current_time();

LOG_INFO("[parallel_create_table]", KR(ret),

"cost", end_time - start_time,

"execute_time", refresh_time - start_time,

"wait_schema", end_time - refresh_time,

"table_name", create_table_arg.schema_.get_table_name());

}

}

if (OB_SUCC(ret) && table_schema.is_external_table()) {

//auto refresh after create external table

OZ (ObAlterTableExecutor::update_external_file_list(

table_schema.get_tenant_id(), res.table_id_,

table_schema.get_external_file_location(),

table_schema.get_external_file_location_access_info(),

table_schema.get_external_file_pattern(),

ctx));

}

} else {

if (table_schema.is_external_table()) {

ret = OB_NOT_SUPPORTED;

LOG_USER_ERROR(OB_NOT_SUPPORTED, "create external table as select");

} else if (OB_FAIL(execute_ctas(ctx, stmt, common_rpc_proxy))){ // 查询建表的处理

LOG_WARN("execute create table as select failed", KR(ret));

}

}

if (OB_NOT_NULL(common_rpc_proxy)) {

SERVER_EVENT_ADD("ddl", "create table execute finish",

"tenant_id", MTL_ID(),

"ret", ret,

"trace_id", *ObCurTraceId::get_trace_id(),

"rpc_dst", common_rpc_proxy->get_server(),

"table_info", res.table_id_,

"schema_version", res.schema_version_);

}

SQL_ENG_LOG(INFO, "finish create table execute.", K(ret), "ddl_event_info", ObDDLEventInfo(), K(stmt), K(create_table_arg));

// only CTAS or create temporary table will make session_id != 0. If such table detected, set

// need ctas cleanup task anyway to do some cleanup jobs

if (0 != table_schema.get_session_id()) {

LOG_TRACE("CTAS or temporary table create detected", K(table_schema));

ATOMIC_STORE(&OBSERVER.need_ctas_cleanup_, true);

}

}

return ret;

}

int ObCreateTableExecutor::set_index_arg_list(ObExecContext &ctx, ObCreateTableStmt &stmt)

{

int ret = OB_SUCCESS;

obrpc::ObCreateTableArg &create_table_arg = const_cast<obrpc::ObCreateTableArg &>(stmt.get_create_table_arg());

if (stmt.get_index_partition_resolve_results().count()

!= stmt.get_index_arg_list().count()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("invalid index resolve result", K(ret), K(stmt));

}

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

HEAP_VAR(ObCreateIndexStmt, index_stmt) {

ObPartitionResolveResult &resolve_result = stmt.get_index_partition_resolve_results().at(i);

index_stmt.get_part_fun_exprs() = resolve_result.get_part_fun_exprs();

index_stmt.get_part_values_exprs() = resolve_result.get_part_values_exprs();

index_stmt.get_subpart_fun_exprs() = resolve_result.get_subpart_fun_exprs();

index_stmt.get_template_subpart_values_exprs() = resolve_result.get_template_subpart_values_exprs();

index_stmt.get_individual_subpart_values_exprs() = resolve_result.get_individual_subpart_values_exprs();

if (OB_FAIL(index_stmt.get_create_index_arg().assign(stmt.get_index_arg_list().at(i)))) {

LOG_WARN("fail to assign index arg", K(ret));

} else if (OB_FAIL(ObPartitionExecutorUtils::calc_values_exprs(ctx, index_stmt))) {

LOG_WARN("fail to compare range partition expr", K(ret));

} else if (OB_FAIL(create_table_arg.index_arg_list_.push_back(index_stmt.get_create_index_arg()))) {

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

}

}

}

return ret;

}

ObAlterTableExecutor::ObAlterTableExecutor()

{

}

ObAlterTableExecutor::~ObAlterTableExecutor()

{

}

int ObAlterTableExecutor::refresh_schema_for_table(

const uint64_t tenant_id)

{

int ret = OB_SUCCESS;

share::schema::ObSchemaGetterGuard schema_guard;

const observer::ObGlobalContext &gctx = observer::ObServer::get_instance().get_gctx();

ObMultiVersionSchemaService *schema_service = gctx.schema_service_;

int64_t local_version = OB_INVALID_VERSION;

int64_t global_version = OB_INVALID_VERSION;

if (OB_ISNULL(schema_service)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("error unexpected, schema service must not be NULL", K(ret));

} else if (OB_FAIL(schema_service->get_tenant_refreshed_schema_version(

tenant_id, local_version))) {

LOG_WARN("fail to get local version", K(ret), "tenant_id", tenant_id);

} else if (OB_FAIL(schema_service->get_tenant_received_broadcast_version(

tenant_id, global_version))) {

LOG_WARN("fail to get global version", K(ret), "tenant_id", tenant_id);

} else if (local_version < global_version) {

LOG_INFO("try to refresh schema", K(local_version), K(global_version));

// force refresh schema最新版本

ObSEArray<uint64_t, 1> tenant_ids;

if (OB_FAIL(tenant_ids.push_back(tenant_id))) {

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

} else if (OB_FAIL(schema_service->refresh_and_add_schema(tenant_ids))) {

LOG_WARN("failed to refresh schema", K(ret));

}

}

return ret;

}

/* 从 3100 开始的版本 alter table 逻辑是将建索引和其他操作放到同一个 rpc 里发到 rs，返回后对每个创建的索引进行同步等，如果一个索引创建失败，则回滚全部索引

mysql 模式下支持 alter table 同时做建索引操作和其他操作，需要保证 rs 在处理 drop index 之后再处理 add index

否则前缀索引会有问题：

*/

int ObAlterTableExecutor::alter_table_rpc_v2(

obrpc::ObAlterTableArg &alter_table_arg,

obrpc::ObAlterTableRes &res,

common::ObIAllocator &allocator,

obrpc::ObCommonRpcProxy *common_rpc_proxy,

ObSQLSessionInfo *my_session,

const bool is_sync_ddl_user)

{

int ret = OB_SUCCESS;

// do not support cancel drop_index_task.

bool is_support_cancel = true;

const ObSArray<obrpc::ObIndexArg *> index_arg_list = alter_table_arg.index_arg_list_;

ObSArray<obrpc::ObIndexArg *> add_index_arg_list;

ObSArray<obrpc::ObIndexArg *> drop_index_args;

alter_table_arg.index_arg_list_.reset();

if (OB_ISNULL(my_session)) {

ret = OB_INVALID_ARGUMENT;

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

} else {

alter_table_arg.compat_mode_ = ORACLE_MODE == my_session->get_compatibility_mode() ?

lib::Worker::CompatMode::ORACLE : lib::Worker::CompatMode::MYSQL;

}

for (int64_t i = 0; OB_SUCC(ret) && i < index_arg_list.size(); ++i) {

obrpc::ObIndexArg *index_arg = index_arg_list.at(i);

if (OB_ISNULL(index_arg)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("index arg should not be null", KR(ret));

} else if (obrpc::ObIndexArg::ADD_INDEX == index_arg->index_action_type_) {

if (OB_FAIL(add_index_arg_list.push_back(index_arg))) {

LOG_WARN("fail to push back to arg_for_adding_index_list", KR(ret));

}

} else if (obrpc::ObIndexArg::DROP_INDEX == index_arg->index_action_type_) {

if (OB_FAIL(drop_index_args.push_back(index_arg))) {

LOG_WARN("push back drop index arg failed", K(ret));

} else if (OB_FAIL(alter_table_arg.index_arg_list_.push_back(index_arg))) {

LOG_WARN("push back index arg failed", K(ret));

} else {

ObDropIndexArg *drop_index_arg = static_cast<ObDropIndexArg *>(index_arg);

drop_index_arg->is_add_to_scheduler_ = true;

is_support_cancel = false;

}

} else { // for rename/drop index action

if (OB_FAIL(alter_table_arg.index_arg_list_.push_back(index_arg))) {

LOG_WARN("fail to push back to arg_for_adding_index_list", KR(ret));

}

}

}

// for add index action

for (int64_t i = 0; OB_SUCC(ret) && i < add_index_arg_list.size(); ++i) {

if (OB_FAIL(alter_table_arg.index_arg_list_.push_back(add_index_arg_list.at(i)))) {

LOG_WARN("fail to push back to arg_for_adding_index_list", KR(ret));

}

}

if (OB_SUCC(ret)) {

if (obrpc::ObAlterTableArg::SET_INTERVAL == alter_table_arg.alter_part_type_

|| obrpc::ObAlterTableArg::INTERVAL_TO_RANGE == alter_table_arg.alter_part_type_) {

alter_table_arg.is_alter_partitions_ = true;

}

if (OB_FAIL(common_rpc_proxy->alter_table(alter_table_arg, res))) {

LOG_WARN("rpc proxy alter table failed", KR(ret), "dst", common_rpc_proxy->get_server(), K(alter_table_arg));

} else {

// 在回滚时不会重试，也不检查 schema version

alter_table_arg.based_schema_object_infos_.reset();

}

}

if (OB_SUCC(ret)) {

ObIArray<obrpc::ObDDLRes> &ddl_ress = res.ddl_res_array_;

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

ObDDLRes &ddl_res = ddl_ress.at(i);

if (OB_FAIL(ObDDLExecutorUtil::wait_ddl_finish(ddl_res.tenant_id_, ddl_res.task_id_, my_session, common_rpc_proxy, is_support_cancel))) {

LOG_WARN("wait drop index finish", K(ret));

}

}

}

if (OB_SUCC(ret)) {

ObCreateIndexExecutor create_index_executor;

uint64_t failed_index_no = OB_INVALID_ID;

// 对drop/truncate分区全局索引的处理

if (!is_sync_ddl_user && alter_table_arg.is_update_global_indexes_

&& (obrpc::ObAlterTableArg::DROP_PARTITION == alter_table_arg.alter_part_type_

|| obrpc::ObAlterTableArg::DROP_SUB_PARTITION == alter_table_arg.alter_part_type_

|| obrpc::ObAlterTableArg::TRUNCATE_PARTITION == alter_table_arg.alter_part_type_

|| obrpc::ObAlterTableArg::TRUNCATE_SUB_PARTITION == alter_table_arg.alter_part_type_)) {

common::ObSArray<ObAlterTableResArg> &res_array = res.res_arg_array_;

for (int64_t i = 0; OB_SUCC(ret) && i < res_array.size(); ++i) {

SMART_VAR(obrpc::ObCreateIndexArg, create_index_arg) {

create_index_arg.index_schema_.set_table_id(res_array.at(i).schema_id_);

create_index_arg.index_schema_.set_schema_version(res_array.at(i).schema_version_);

if (OB_FAIL(create_index_executor.sync_check_index_status(*my_session, *common_rpc_proxy, create_index_arg, res, allocator))) {

LOG_WARN("failed to sync_check_index_status", KR(ret), K(create_index_arg), K(i));

}

}

}

} else if (DDL_CREATE_INDEX == res.ddl_type_ || DDL_NORMAL_TYPE == res.ddl_type_) {

// TODO(shuangcan): alter table create index returns DDL_NORMAL_TYPE now, check if we can fix this later

// 同步等索引建成功

for (int64_t i = 0; OB_SUCC(ret) && i < add_index_arg_list.size(); ++i) {

obrpc::ObIndexArg *index_arg = add_index_arg_list.at(i);

obrpc::ObCreateIndexArg *create_index_arg = NULL;

if (OB_ISNULL(index_arg)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("index arg is null", KR(ret), K(i));

} else if (obrpc::ObIndexArg::ADD_INDEX != index_arg->index_action_type_) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("index action type should be add index", KR(ret), K(i), K(*index_arg));

} else if (OB_ISNULL(create_index_arg = static_cast<obrpc::ObCreateIndexArg *>(index_arg))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("create index arg is null", KR(ret), K(i));

} else if (INDEX_TYPE_PRIMARY == create_index_arg->index_type_) {

// do nothing

} else if (!is_sync_ddl_user) {

// 只考虑非备份恢复时的索引同步检查

create_index_arg->index_schema_.set_table_id(res.res_arg_array_.at(i).schema_id_);

create_index_arg->index_schema_.set_schema_version(res.res_arg_array_.at(i).schema_version_);

// 只考虑非备份恢复时的索引同步检查

if (OB_FAIL(create_index_executor.sync_check_index_status(*my_session, *common_rpc_proxy, *create_index_arg, res, allocator))) {

failed_index_no = i;

LOG_WARN("failed to sync_check_index_status", KR(ret), K(*create_index_arg), K(i));

}

}

}

// 回滚所有已经建立的 index

if (OB_FAIL(ret)) {

int tmp_ret = OB_SUCCESS;

uint64_t tenant_id = OB_INVALID_ID;

for (int64_t i = 0; (OB_SUCCESS == tmp_ret) && (i < add_index_arg_list.size()); ++i) {

if (failed_index_no == i) {

// 同步建索引逻辑里已经把这个失败的删掉了

continue;

} else {

obrpc::ObDropIndexArg drop_index_arg;

obrpc::ObDropIndexRes drop_index_res;

obrpc::ObCreateIndexArg *create_index_arg = static_cast<obrpc::ObCreateIndexArg *>(add_index_arg_list.at(i));

drop_index_arg.tenant_id_ = create_index_arg->tenant_id_;

drop_index_arg.exec_tenant_id_ = create_index_arg->tenant_id_;

drop_index_arg.index_table_id_ = res.res_arg_array_.at(i).schema_id_;

drop_index_arg.session_id_ = create_index_arg->session_id_;

drop_index_arg.index_name_ = create_index_arg->index_name_;

drop_index_arg.table_name_ = create_index_arg->table_name_;

drop_index_arg.database_name_ = create_index_arg->database_name_;

drop_index_arg.index_action_type_ = obrpc::ObIndexArg::DROP_INDEX;

drop_index_arg.is_add_to_scheduler_ = false;

tenant_id = drop_index_arg.tenant_id_;

if (OB_SUCCESS != (tmp_ret = create_index_executor.set_drop_index_stmt_str(drop_index_arg, allocator))) {

LOG_WARN("fail to set drop index ddl_stmt_str", K(tmp_ret));

} else if (OB_SUCCESS != (tmp_ret = common_rpc_proxy->drop_index(drop_index_arg, drop_index_res))) {

LOG_WARN("rpc proxy drop index failed", "dst", common_rpc_proxy->get_server(),

K(tmp_ret),

K(drop_index_arg.table_name_),

K(drop_index_arg.index_name_));

}

}

}

if (OB_SUCCESS != tmp_ret && OB_INVALID_ID != failed_index_no) { // rewrite LOG_USER_ERROR message

uint64_t index_table_id = res.res_arg_array_.at(failed_index_no).schema_id_;

int64_t schema_version = res.res_arg_array_.at(failed_index_no).schema_version_;

bool is_finish = false;

if (OB_SUCCESS != (tmp_ret = ObDDLExecutorUtil::wait_build_index_finish(tenant_id, res.task_id_, is_finish))) {

LOG_WARN("wait build index finish failed", K(tmp_ret), K(tenant_id), K(res.task_id_));

}

}

LOG_INFO("added indexes failed, we rolled back all indexes added in this same alter table sql. But we didn't roll back other actions in this same alter table sql");

}

}

}

return ret;

}

int ObAlterTableExecutor::get_external_file_list(const ObString &location,

ObIArray<ObString> &file_urls,

ObIArray<int64_t> &file_sizes,

const ObString &access_info,

ObIAllocator &allocator,

common::ObStorageType &storage_type)

{

int ret = OB_SUCCESS;

ObExternalDataAccessDriver driver;

if (OB_FAIL(driver.init(location, access_info))) {

LOG_WARN("init external data access driver failed", K(ret));

} else if (OB_FAIL(driver.get_file_list(location, file_urls, allocator))) {

LOG_WARN("get file urls failed", K(ret));

} else if (OB_FAIL(driver.get_file_sizes(location, file_urls, file_sizes))) {

LOG_WARN("get file sizes failed", K(ret));

}

if (driver.is_opened()) {

storage_type = driver.get_storage_type();

driver.close();

}

LOG_DEBUG("show external table files", K(file_urls), K(storage_type), K(access_info));

return ret;

}

int ObAlterTableExecutor::filter_and_sort_external_files(const ObString &pattern,

ObExecContext &exec_ctx,

ObIArray<ObString> &file_urls,

ObIArray<int64_t> &file_sizes) {

int ret = OB_SUCCESS;

const int64_t count = file_urls.count();

ObSEArray<int64_t, 8> tmp_file_sizes;

hash::ObHashMap<ObString, int64_t> file_map;

if (0 == count) {

/* do nothing */

} else if (OB_UNLIKELY(count != file_sizes.count())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("array size error", K(ret));

} else if (OB_FAIL(file_map.create(count, "ExtFileMap", "ExtFileMap"))) {

LOG_WARN("fail to init hashmap", K(ret));

} else {

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

if (OB_FAIL(file_map.set_refactored(file_urls.at(i), file_sizes.at(i)))) {

LOG_WARN("failed to set refactored to file_map", K(ret));

}

}

if (OB_SUCC(ret)) {

if (OB_FAIL(ObExternalTableUtils::filter_external_table_files(pattern, exec_ctx, file_urls))) {

LOG_WARN("failed to filter external table files");

}

}

if (OB_SUCC(ret)) {

std::sort(file_urls.get_data(), file_urls.get_data() + file_urls.count());

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

int64_t file_size = 0;

if (OB_FAIL(file_map.get_refactored(file_urls.at(i), file_size))) {

if (OB_UNLIKELY(OB_HASH_NOT_EXIST == ret)) {

ret = OB_ERR_UNEXPECTED;

}

LOG_WARN("failed to get key meta", K(ret));

} else if (OB_FAIL(tmp_file_sizes.push_back(file_size))) {

LOG_WARN("failed to push back into tmp_file_sizes", K(ret));

}

}

}

if (OB_SUCC(ret)) {

if (OB_FAIL(file_sizes.assign(tmp_file_sizes))) {

LOG_WARN("failed to assign file_sizes", K(ret));

} else if (OB_FAIL(file_map.destroy())) {

LOG_WARN("failed to destory file_map");

}

}

}

LOG_TRACE("after filter external table files", K(ret), K(file_urls));

return ret;

}

int ObAlterTableExecutor::flush_external_file_cache(

const uint64_t tenant_id,

const uint64_t table_id,

const ObIArray<ObAddr> &all_servers)

{

int ret = OB_SUCCESS;

ObArenaAllocator allocator;

ObAsyncRpcTaskWaitContext<ObRpcAsyncFlushExternalTableKVCacheCallBack> context;

int64_t send_task_count = 0;

OZ (context.init());

OZ (context.get_cb_list().reserve(all_servers.count()));

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

ObFlushExternalTableFileCacheReq req;

int64_t timeout = ObExternalTableFileManager::CACHE_EXPIRE_TIME;

req.tenant_id_ = tenant_id;

req.table_id_ = table_id;

req.partition_id_ = 0;

ObRpcAsyncFlushExternalTableKVCacheCallBack* async_cb = nullptr;

if (OB_ISNULL(async_cb = OB_NEWx(ObRpcAsyncFlushExternalTableKVCacheCallBack, (&allocator), (&context)))) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("failed to allocate async cb memory", K(ret));

}

OZ (context.get_cb_list().push_back(async_cb));

OZ (GCTX.external_table_proxy_->to(all_servers.at(i))

.by(tenant_id)

.timeout(timeout)

.flush_file_kvcahce(req, async_cb));

if (OB_SUCC(ret)) {

send_task_count++;

}

}

context.set_task_count(send_task_count);

do {

int temp_ret = context.wait_executing_tasks();

if (OB_SUCCESS != temp_ret) {

LOG_WARN("fail to wait executing task", K(temp_ret));

if (OB_SUCC(ret)) {

ret = temp_ret;

}

}

} while(0);