/**

* 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/rc/ob_tenant_base.h"

#define USING_LOG_PREFIX SQL_ENG

//#define TEST_MODE

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

#include <math.h>

#include "observer/omt/ob_multi_tenant.h"

#include "lib/oblog/ob_log_module.h"

#include "lib/string/ob_sql_string.h"

#include "storage/access/ob_dml_param.h"

#include "sql/parser/ob_parser.h"

#include "sql/resolver/ob_resolver.h"

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

#include "sql/plan_cache/ob_sql_parameterization.h"

#include "sql/code_generator/ob_expr_generator_impl.h"

#include "sql/code_generator/ob_code_generator.h"

#include "sql/engine/ob_exec_context.h"

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

#include "sql/engine/ob_physical_plan_ctx.h"

#include "sql/resolver/expr/ob_raw_expr_util.h"

#include "sql/das/ob_das_location_router.h"

#include "share/ob_tenant_mgr.h"

#include "share/ob_tenant_memstore_info_operator.h"

#include "sql/resolver/ob_schema_checker.h"

#include "observer/ob_inner_sql_connection_pool.h"

#include "observer/ob_inner_sql_result.h"

#include "share/ob_device_manager.h"

#include "share/backup/ob_backup_io_adapter.h"

#include "storage/tx_storage/ob_tenant_freezer.h"

#include "sql/rewrite/ob_transform_utils.h"

#include "observer/omt/ob_tenant_timezone_mgr.h"

using namespace oceanbase::sql;

using namespace oceanbase::common;

using namespace oceanbase::share;

using namespace oceanbase::share::schema;

using namespace oceanbase::storage;

using namespace oceanbase::observer;

namespace oceanbase

{

namespace sql

{

#ifdef TEST_MODE

static const int64_t INSERT_TASK_DROP_RATE = 1;

static void delay_process_by_probability(int64_t percentage) {

if (OB_UNLIKELY(ObRandom::rand(1, 100) <= percentage)) {

ob_usleep(RPC_BATCH_INSERT_TIMEOUT_US);

}

}

#endif

#define OW(statement) \

do {\

int inner_ret = statement;\

if (OB_UNLIKELY(OB_SUCCESS != inner_ret)) {\

LOG_WARN("fail to exec"#statement, K(inner_ret));\

if (OB_SUCC(ret)) { ret = inner_ret; }\

}\

} while (0)

const char *ObLoadDataBase::SERVER_TENANT_MEMORY_EXAMINE_SQL =

"SELECT case when memstore_used < freeze_trigger * 1.02 then false else true end"

" as need_wait_freeze"

" FROM oceanbase.__all_virtual_tenant_memstore_info WHERE tenant_id = %ld"

" and svr_ip = '%s' and svr_port = %d";

const char *log_file_column_names = "\nBatchId\tLineNum\tType\tErrCode\tErrMsg\t\n";

const char *log_file_row_fmt = "%ld\t%ld\t%s\t%d\t%.*s\t\n";

static const int64_t WAIT_INTERVAL_US = 1 * 1000 * 1000; //1s

int ObLoadDataBase::generate_fake_field_strs(ObIArray<ObString> &file_col_values,

ObIAllocator &allocator,

const char id_char)

{

int ret = OB_SUCCESS;

char *buf = NULL;

/* to generate string like "F1F2F3...F99....Fn" into buf

* maxn = 512

*/

int64_t buf_len = 6 * file_col_values.count();

int64_t pos = 0;

if (OB_ISNULL(buf = static_cast<char*>(allocator.alloc(buf_len * sizeof(char))))) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("alloc memory failed", K(ret));

} else {

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

int64_t pos_bak = pos;

if (OB_FAIL(databuff_printf(buf, buf_len, pos, "%c%ld", id_char, i))) { //F1 F2 ..

LOG_WARN("generate str failed", K(ret), K(pos), K(buf_len));

} else {

file_col_values.at(i).assign_ptr(buf + pos_bak, pos - pos_bak);

}

}

LOG_DEBUG("generate fake field result", K(file_col_values));

}

return ret;

}

int ObLoadDataBase::construct_insert_sql(ObSqlString &insert_sql,

const ObString &q_name,

ObIArray<ObLoadTableColumnDesc> &desc,

ObIArray<ObString> &insert_values,

int64_t num_rows)

{

int ret = OB_SUCCESS;

insert_sql.reuse();

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

if (OB_UNLIKELY(q_name.empty())

|| OB_UNLIKELY(desc.count() * num_rows != insert_values.count())) {

ret = OB_INVALID_ARGUMENT;

}

OX (ret = insert_sql.assign("INSERT INTO "));

OX (ret = insert_sql.append(q_name));

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

OX (ret = insert_sql.append(0 == i ? "(" : ","));

OX (ret = insert_sql.append_fmt("%c%.*s%c", q,

desc.at(i).column_name_.length(), desc.at(i).column_name_.ptr(),

q));

}

OX (ret = insert_sql.append(") VALUES "));

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

OX (ret = insert_sql.append(i % desc.count() == 0 ? (i == 0 ? "(" : "),(") : ","));

OX (ret = insert_sql.append_fmt("'%.*s'",

insert_values.at(i).length(), insert_values.at(i).ptr()));

}

OX (ret = insert_sql.append(")"));

if (OB_FAIL(ret)) {

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

} else {

LOG_DEBUG("insert sql generated", K(insert_sql));

}

return ret;

}

int ObLoadDataBase::make_parameterize_stmt(ObExecContext &ctx,

ObSqlString &insertsql,

ParamStore &param_store,

ObInsertStmt *&insert_stmt)

{

int ret = OB_SUCCESS;

ObSQLSessionInfo *session = NULL;

if (OB_ISNULL(session = ctx.get_my_session())) {

ret = OB_ERR_UNEXPECTED;

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

} else if (OB_ISNULL(ctx.get_sql_ctx())

|| OB_ISNULL(ctx.get_sql_ctx()->schema_guard_)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("sql ctx is null", K(ret));

} else {

ObParser parser(ctx.get_allocator(), session->get_sql_mode());

ParseResult parse_result;

SqlInfo not_param_info;

bool is_transform_outline = false;

ObMaxConcurrentParam::FixParamStore fixed_param_store(OB_MALLOC_NORMAL_BLOCK_SIZE,

ObWrapperAllocator(&ctx.get_allocator()));

if (OB_FAIL(parser.parse(insertsql.string(), parse_result))) {

LOG_WARN("parser template insert sql failed", K(ret));

} else if (OB_FAIL(ObSqlParameterization::transform_syntax_tree(ctx.get_allocator(),

*session,

NULL,

parse_result.result_tree_,

not_param_info,

param_store,

NULL,

fixed_param_store,

is_transform_outline))) {

LOG_WARN("parameterize parser tree failed", K(ret));

} else {

SMART_VAR(ObResolverParams, resolver_ctx) {

ObSchemaChecker schema_checker;

schema_checker.init(*(ctx.get_sql_ctx()->schema_guard_));

resolver_ctx.allocator_ = &ctx.get_allocator();

resolver_ctx.schema_checker_ = &schema_checker;

resolver_ctx.session_info_ = session;

resolver_ctx.param_list_ = &param_store;

resolver_ctx.database_id_ = session->get_database_id();

resolver_ctx.disable_privilege_check_ = PRIV_CHECK_FLAG_DISABLE;

resolver_ctx.expr_factory_ = ctx.get_expr_factory();

resolver_ctx.stmt_factory_ = ctx.get_stmt_factory();

if (OB_ISNULL(ctx.get_stmt_factory())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("invalid argument", K(ret), KP(ctx.get_stmt_factory()));

} else if (OB_ISNULL(ctx.get_stmt_factory()->get_query_ctx())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("invalid argument", K(ret), KP(ctx.get_stmt_factory()->get_query_ctx()));

} else {

resolver_ctx.query_ctx_ = ctx.get_stmt_factory()->get_query_ctx();

resolver_ctx.query_ctx_->question_marks_count_ = param_store.count();

resolver_ctx.query_ctx_->sql_schema_guard_.set_schema_guard(ctx.get_sql_ctx()->schema_guard_);

ObResolver resolver(resolver_ctx);

ObStmt *astmt = NULL;

ParseNode *stmt_tree = parse_result.result_tree_->children_[0];

if (OB_ISNULL(stmt_tree) || OB_ISNULL(ctx.get_stmt_factory()->get_query_ctx())) {

ret = OB_ERR_UNEXPECTED;

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

} else if (OB_FAIL(resolver.resolve(ObResolver::IS_NOT_PREPARED_STMT,

*stmt_tree,

astmt))) {

LOG_WARN("resolve sql failed", K(ret), K(insertsql));

} else {

insert_stmt = static_cast<ObInsertStmt*>(astmt);

ctx.get_stmt_factory()->get_query_ctx()->reset();

}

}

}

}

}

return ret;

}

int ObLoadDataBase::memory_check_remote(uint64_t tenant_id, bool &need_wait_minor_freeze)

{

int ret = OB_SUCCESS;

MTL_SWITCH(tenant_id) {

storage::ObTenantFreezer *freezer = nullptr;

if (FALSE_IT(freezer = MTL(storage::ObTenantFreezer *))) {

} else {

int64_t active_memstore_used = 0;

int64_t total_memstore_used = 0;

int64_t major_freeze_trigger = 0;

int64_t memstore_limit = 0;

int64_t freeze_cnt = 0;

if (OB_FAIL(freezer->get_tenant_memstore_cond(active_memstore_used,

total_memstore_used,

major_freeze_trigger,

memstore_limit,

freeze_cnt))) {

LOG_WARN("fail to get memstore used", K(ret));

} else {

if (total_memstore_used > (memstore_limit - major_freeze_trigger)/2 + major_freeze_trigger) {

need_wait_minor_freeze = true;

} else {

need_wait_minor_freeze = false;

}

}

LOG_DEBUG("load data check tenant memory usage", K(active_memstore_used),

K(total_memstore_used),

K(major_freeze_trigger),

K(memstore_limit),

K(freeze_cnt),

K(need_wait_minor_freeze));

}

} else {

LOG_ERROR("switch tenant failed", K(tenant_id), K(ret));

}

return ret;

}

/*

* if param_a != param_b: this variable is from a field of data file,

* calc the corresponding field index via param string value

* return the index

*/

int ObLoadDataBase::calc_param_offset(const ObObj &param_a,

const ObObj &param_b,

int64_t &idx)

{

int ret = OB_SUCCESS;

if (!param_a.is_varchar_or_char() || !param_b.is_varchar_or_char()) {

idx = OB_INVALID_INDEX_INT64;

} else if (param_a.get_string().compare(param_b.get_string()) != 0) {

const ObObj &value = param_a;

const char *value_ptr = value.get_string_ptr();

/* 这里处理的insert模板中的数据，是自己造的，

一定是string类型，字符串内容是 "[F|f][0-9]+"，因此长度大于等于2 */

if (value.get_string_len() < 2 || NULL == value_ptr) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("no possible, the values are changed", K(ret));

} else {

int64_t temp_idx = 0;

for (int32_t j = 1; OB_SUCC(ret) && j < value.get_string_len(); ++j) {

char cur_char = *(value_ptr + j);

if (cur_char > '9' || cur_char < '0') {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("no possible, the values are changed", K(ret));

} else {

temp_idx *= 10;

temp_idx += cur_char - '0';

}

}

idx = temp_idx;

}

}

return ret;

}

int ObLoadDataBase::memory_wait_local(ObExecContext &ctx,

const ObTabletID &tablet_id,

ObAddr &server_addr,

int64_t &total_wait_secs,

bool &is_leader_changed)

{

int ret = OB_SUCCESS;

static const int64_t WAIT_INTERVAL_US = 1 * 1000 * 1000; //1s

ObSQLSessionInfo *session = NULL;

ObMySQLProxy *sql_proxy_ = NULL;

SMART_VAR(ObMySQLProxy::MySQLResult, res) {

sqlclient::ObMySQLResult *result = NULL;

ObSqlString sql;

int64_t start_wait_ts = ObTimeUtil::current_time();

int64_t wait_timeout_ts = 0;

uint64_t tenant_id = OB_INVALID_TENANT_ID;

if (OB_UNLIKELY(!tablet_id.is_valid())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("invalid server addr", K(ret), K(tablet_id));

} else if (OB_ISNULL((sql_proxy_ = GCTX.sql_proxy_))) {

ret = OB_ERR_UNEXPECTED;

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

} else if (OB_ISNULL(session = ctx.get_my_session())) {

ret = OB_ERR_UNEXPECTED;

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

} else {

session->get_query_timeout(wait_timeout_ts);

tenant_id = session->get_effective_tenant_id();

//print info

LOG_INFO("LOAD DATA is suspended until the memory is available",

K(tablet_id), K(server_addr), K(total_wait_secs));

}

bool need_wait_freeze = true;

ObAddr leader_addr;

ObDASLocationRouter &loc_router = DAS_CTX(ctx).get_location_router();

while (OB_SUCC(ret) && need_wait_freeze) {

ob_usleep(WAIT_INTERVAL_US);

leader_addr.reset();

res.reuse();

char leader_ip_str[MAX_IP_ADDR_LENGTH];

const int64_t retry_us = 200 * 1000;

//Try to use the results in the cache as much as possible, without forcing a cache refresh.

const int64_t expire_renew_time = 0;

if (OB_FAIL(ObLoadDataUtils::check_session_status(*session))) {

LOG_WARN("session is not valid during wait", K(ret));

} else if (OB_FAIL(loc_router.get_leader(tenant_id, tablet_id, leader_addr, expire_renew_time))) {

LOG_WARN("failed to get location", K(ret));

ob_usleep(retry_us);

} else {

LOG_DEBUG("get participants", K(tablet_id), K(leader_addr));

}

if (OB_FAIL(ret)) {

} else if (!leader_addr.ip_to_string(leader_ip_str, sizeof(leader_ip_str))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("format leader ip failed", K(ret), K(leader_addr));

} else if (OB_FAIL(sql.assign_fmt(SERVER_TENANT_MEMORY_EXAMINE_SQL,

tenant_id,

leader_ip_str,

leader_addr.get_port()))) {

LOG_WARN("fail to append sql", K(ret), K(tenant_id), K(leader_addr));

} else if (OB_FAIL(sql_proxy_->read(res, OB_SYS_TENANT_ID, sql.ptr()))) {

LOG_WARN("fail to execute sql", K(ret), K(sql));

} else if (NULL == (result = res.get_result())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("fail to get sql result", K(ret));

} else if (OB_FAIL(result->next())) {

LOG_WARN("fail to get result, force renew location", K(ret), K(leader_addr));

if (OB_ITER_END == ret) {

ret = OB_SUCCESS;

}

} else {

EXTRACT_BOOL_FIELD_MYSQL(*result, "need_wait_freeze", need_wait_freeze);

//LOG_INFO("LOAD DATA is waiting for tenant memory available",

//K(waited_seconds), K(total_wait_secs), K(tenant_id));

}

//if it is location exception, refresh location cache with block interface

//because load data can only local retry

loc_router.refresh_location_cache_by_errno(false, ret);

}

//print info

if (OB_SUCC(ret)) {

int64_t wait_secs = (ObTimeUtil::current_time() - start_wait_ts) / 1000000;

total_wait_secs += wait_secs;

if (leader_addr != server_addr) {

LOG_INFO("LOAD DATA location change",

"old_addr", server_addr,

"new_addr", leader_addr);

server_addr = leader_addr;

is_leader_changed = true;

} else {

is_leader_changed = false;

}

LOG_INFO("LOAD DATA is resumed",

"waited_seconds", wait_secs,

K(total_wait_secs));

}

}

return ret;

}

int ObLoadDataBase::pre_parse_lines(ObLoadFileBuffer &buffer,

ObCSVGeneralParser &parser,

bool is_last_buf,

int64_t &valid_len,

int64_t &line_count)

{

int ret = OB_SUCCESS;

if (OB_UNLIKELY(!buffer.is_valid())) {

ret = OB_INVALID_ARGUMENT;

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

} else if (parser.get_opt_params().is_simple_format_) {

const ObCSVGeneralFormat &format = parser.get_format();

char *cur_pos = buffer.begin_ptr();

int64_t cur_lines = 0;

for (char *p = buffer.begin_ptr(); p < buffer.current_ptr(); ++p) {

char cur_char = *p;

if (format.field_escaped_char_ == cur_char && p + 1 < buffer.current_ptr()) {

p++;

} else if (parser.get_opt_params().line_term_c_ == cur_char) {

cur_lines++;

cur_pos = p + 1;

if (cur_lines >= line_count) {

break;

}

}

}

if (is_last_buf && cur_lines < line_count && buffer.current_ptr() > cur_pos) {

cur_lines++;

cur_pos = buffer.current_ptr();

}

valid_len = cur_pos - buffer.begin_ptr();

line_count = cur_lines;

} else {

ObSEArray<ObCSVGeneralParser::LineErrRec, 128> err_records;

const char *ptr = buffer.begin_ptr();

const char *end = ptr + buffer.get_data_len();

auto unused_handler = [](ObIArray<ObCSVGeneralParser::FieldValue> &fields_per_line) -> int {

UNUSED(fields_per_line);

return OB_SUCCESS;

};

if (OB_FAIL(parser.scan(ptr, end, line_count, NULL, NULL, unused_handler, err_records, is_last_buf))) {

LOG_WARN("fail to scan buf", K(ret));

} else {

valid_len = ptr - buffer.begin_ptr();

}

}

return ret;

}

int ObInsertValueGenerator::fill_field_expr(ObIArray<ObCSVGeneralParser::FieldValue> &field_values,

const ObBitSet<> &string_values)

{

int ret = OB_SUCCESS;

if (OB_UNLIKELY(field_values.count() != field_exprs_.count())) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid input", K(ret), K(field_values), K(field_exprs_));

} else {

for (int i = 0; i < field_values.count(); ++i) {

auto expr = static_cast<ObConstRawExpr *>(field_exprs_.at(i));

ObLoadDataBase::field_to_obj(expr->get_value(),

field_values.at(i),

cs_type_,

string_values.has_member(i));

}

}

return ret;

}

int ObInsertValueGenerator::gen_insert_values(ObIArray<ObString> &insert_values,

ObStringBuf &str_buf)

{

int ret = OB_SUCCESS;

for (int i = 0; OB_SUCC(ret) && i < insert_exprs_.count(); ++i) {

auto expr = insert_exprs_.at(i);

ObString store_value;

data_buffer_->reset();

ObConstRawExpr *const_expr = NULL;

if (expr->get_expr_type() == T_DEFAULT) {

OZ (str_buf.write_string("DEFAULT", &store_value));

} else if (expr->is_const_raw_expr()

&& (const_expr = static_cast<ObConstRawExpr *>(expr))->get_value().is_string_type()) {

ObString const_string = const_expr->get_value().get_string();

ObCollationType coll_type = const_expr->get_value().get_collation_type();

uint32_t pos = 0;

if (ObCharset::charset_type_by_coll(coll_type) != CHARSET_UTF8MB4) {

if (OB_FAIL(ObCharset::charset_convert(

coll_type, const_string.ptr(), const_string.length(),

CS_TYPE_UTF8MB4_BIN, data_buffer_->begin_ptr(), data_buffer_->get_remain_len(), pos, false))) {

LOG_WARN("fail to convert charset", K(ret));

} else {

const_string.assign_ptr(data_buffer_->begin_ptr(), pos);

data_buffer_->update_pos(pos);

}

}

if (OB_SUCC(ret)) {

ObHexEscapeSqlStr escape_str(const_string, !!(SMO_NO_BACKSLASH_ESCAPES & sql_mode_));

int64_t len = escape_str.to_string(data_buffer_->current_ptr() + 1, data_buffer_->get_remain_len() - 1);

if (OB_UNLIKELY(len + 2 >= data_buffer_->get_remain_len())) {

ret = OB_SIZE_OVERFLOW;

LOG_WARN("fail to print string", K(ret), K(len), K(data_buffer_->get_remain_len()));

} else {

*data_buffer_->current_ptr() = '\'';

*(data_buffer_->current_ptr() + 1 + len) = '\'';

OZ (str_buf.write_string(ObString(static_cast<int32_t>(len + 2),

data_buffer_->current_ptr()), &store_value));

}

}

} else {

OZ (expr_printer_.do_print(expr, T_NONE_SCOPE));

OZ (str_buf.write_string(ObString(static_cast<int32_t>(data_buffer_->get_data_len()),

data_buffer_->begin_ptr()), &store_value));

}

OX (insert_values.at(i) = store_value);

//OZ (insert_values.push_back(store_value));

}

LOG_DEBUG("LOAD DATA insert values generated", K(insert_values));

return ret;

}

int ObInsertValueGenerator::gen_insert_sql(ObSqlString &insert_sql)

{

int ret = OB_SUCCESS;

OZ (insert_sql.append(insert_header_));

OZ (insert_sql.append(" VALUES("));

for (int i = 0; OB_SUCC(ret) && i < insert_exprs_.count(); ++i) {

auto expr = insert_exprs_.at(i);

if (i != 0) {

OZ (insert_sql.append(","));

}

/*if (expr->is_const_raw_expr() && static_cast<ObConstRawExpr *>(expr)->get_value().is_string_type()) {

auto const_expr = static_cast<ObConstRawExpr *>(expr);

OZ (insert_sql.append_fmt("'%.*s'",

const_expr->get_value().get_string_len(),

const_expr->get_value().get_string_ptr()));

} else {

*/

data_buffer_->reset();

OZ (expr_printer_.do_print(expr, T_NONE_SCOPE));

OZ (insert_sql.append(ObString(data_buffer_->get_data_len(), data_buffer_->begin_ptr())));

//}

}

OZ (insert_sql.append(")"));

return ret;

}

int ObInsertValueGenerator::set_params(ObString &insert_header, ObCollationType cs_type, int64_t sql_mode)

{

insert_header_ = insert_header;

cs_type_ = cs_type;

sql_mode_ = sql_mode;

return OB_SUCCESS;

}

int ObInsertValueGenerator::init(ObSQLSessionInfo &session,

ObLoadFileBuffer *data_buffer,

ObSchemaGetterGuard *schema_guard)

{

ObObjPrintParams param = session.create_obj_print_params();

param.cs_type_ = CS_TYPE_UTF8MB4_BIN;

expr_printer_.init(data_buffer->begin_ptr(),

data_buffer->get_buffer_size(),

data_buffer->get_pos(),

schema_guard,

param);

data_buffer_ = data_buffer;

return OB_SUCCESS;

}

int ObLoadDataSPImpl::gen_insert_columns_names_buff(ObExecContext &ctx,

const ObLoadArgument &load_args,

ObIArray<ObLoadTableColumnDesc> &insert_infos,

ObString &data_buff,

bool need_online_osg)

{

int ret = OB_SUCCESS;

ObSqlString insert_stmt;

ObSEArray<ObString, 16> insert_column_names;

if (OB_FAIL(insert_column_names.reserve(insert_infos.count()))) {

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

}

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

if (OB_FAIL(insert_column_names.push_back(insert_infos.at(i).column_name_))) {

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

}

}

/*

if (OB_SUCC(ret)) {

int64_t len = 0;

char *buf = 0;

OB_UNIS_ADD_LEN(insert_column_names);

if (OB_ISNULL(buf = static_cast<char *>(ctx.get_allocator().alloc(len)))) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("alloc memory failed", K(ret));

} else {

data_buff.set_data(buf, len);

int64_t buf_len = len;

int64_t pos = 0;

OB_UNIS_ENCODE(insert_column_names);

}

}

*/

if (OB_SUCC(ret)) {

if (OB_FAIL(ObLoadDataUtils::build_insert_sql_string_head(load_args.dupl_action_,

load_args.combined_name_,

insert_column_names,

insert_stmt,

need_online_osg))) {

LOG_WARN("gen insert sql column_names failed", K(ret));

} else if (OB_FAIL(ob_write_string(ctx.get_allocator(), insert_stmt.string(), data_buff))) {

LOG_WARN("fail to write string", K(ret));

}

}

return ret;

}

class ReplaceVariables : public ObIRawExprReplacer

{

public:

ReplaceVariables(ObExecContext &ctx,

ObLoadDataStmt &stmt,

ObIArray<ObRawExpr *> &fields)

: ctx_(ctx), load_stmt_(stmt), field_exprs_(fields) {}

int generate_new_expr(ObRawExprFactory &expr_factory, ObRawExpr *raw_expr, ObRawExpr *&new_expr)

{

int ret = OB_SUCCESS;

UNUSED(expr_factory);

ObSQLSessionInfo *session = NULL;

if (OB_ISNULL(raw_expr)) {

ret = OB_INVALID_ARGUMENT;

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

} else if (OB_ISNULL(session = ctx_.get_my_session())) {

ret = OB_ERR_UNEXPECTED;

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

} else if (raw_expr->get_expr_type() == T_REF_COLUMN

|| raw_expr->get_expr_type() == T_OP_GET_USER_VAR) {

ObRawExpr *orig_expr = raw_expr;

bool is_user_variable = false;

//1. get variable name

ObString ref_name;

if (raw_expr->get_expr_type() == T_REF_COLUMN) {

ObColumnRefRawExpr *column_ref = static_cast<ObColumnRefRawExpr*>(raw_expr);

ref_name = column_ref->get_column_name();

} else {

is_user_variable = true;

ObSysFunRawExpr *func_expr = static_cast<ObSysFunRawExpr*>(raw_expr);

if (func_expr->get_children_count() != 1) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("sys func expr child num is not correct", K(ret));

} else {

ObConstRawExpr *c_expr = static_cast<ObConstRawExpr*>(func_expr->get_param_expr(0));

if (c_expr->get_value().get_type() != ObVarcharType) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("const expr child type is not correct", K(ret));

} else {

ref_name = c_expr->get_value().get_string();

}

}

}

//2. find and replace

int64_t idx = OB_INVALID_INDEX;

if (OB_SUCC(ret)) {

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

if (0 == load_stmt_.get_field_or_var_list().at(i).field_or_var_name_.compare(ref_name)) {

idx = i;

break;

}

}

if (OB_INVALID_INDEX != idx) {

new_expr = field_exprs_.at(idx);

} else {

if (!is_user_variable) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unknown column name in set right expr, do nothing", K(ret), K(ref_name));

} else {

ObConstRawExpr *c_expr = NULL;

//find the real value from session

if (OB_ISNULL(c_expr = OB_NEWx(ObConstRawExpr, (&ctx_.get_allocator())))) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("allocate const raw expr failed", K(ret));

} else {

ObObj var_obj;

ObSessionVariable user_var;

if (OB_FAIL(session->get_user_variable(ref_name, user_var))) {

LOG_WARN("get user variable failed", K(ret), K(ref_name));

} else {

var_obj = user_var.value_;

var_obj.set_meta_type(user_var.meta_);

c_expr->set_value(var_obj);

new_expr = c_expr;

}

}

}

}

}

/*

if (OB_SUCC(ret) && need_replaced_to_loaded_data_from_file) {

raw_expr = c_expr;

ObLoadDataReplacedExprInfo varable_info;

varable_info.replaced_expr = c_expr;

varable_info.correspond_file_field_idx = idx;

if (OB_FAIL(generator.add_file_column_replace_info(varable_info))) {

LOG_WARN("push back replaced variable infos array failed", K(ret));

}

}

*/

LOG_DEBUG("replace variable name to field value",

K(ref_name), K(idx), KPC(orig_expr), KPC(raw_expr), KPC(new_expr));

}

return ret;

}

ObExecContext &ctx_;

ObLoadDataStmt &load_stmt_;

ObIArray<ObRawExpr *> &field_exprs_;

};

int ObLoadDataSPImpl::copy_exprs_for_shuffle_task(ObExecContext &ctx,

ObLoadDataStmt &load_stmt,

ObIArray<ObLoadTableColumnDesc> &insert_infos,

ObIArray<ObRawExpr *> &field_exprs,

ObIArray<ObRawExpr *> &insert_exprs)

{

int ret = OB_SUCCESS;

if (OB_ISNULL(ctx.get_expr_factory())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("expr factory is null", K(ret));

}

OZ (field_exprs.reserve(load_stmt.get_field_or_var_list().count()));

for (int i = 0; OB_SUCC(ret) && i < load_stmt.get_field_or_var_list().count(); ++i) {

ObConstRawExpr *field_expr = NULL;

OZ (ObRawExprUtils::build_const_string_expr(*ctx.get_expr_factory(),

ObVarcharType,

ObString(),

load_stmt.get_load_arguments().file_cs_type_,

field_expr));

OZ (field_exprs.push_back(field_expr));

}

OZ (insert_exprs.reserve(insert_infos.count()));

if (OB_SUCC(ret)) {

ObRawExprCopier copier(*ctx.get_expr_factory());

ReplaceVariables replacer(ctx, load_stmt, field_exprs);

for (int i = 0; OB_SUCC(ret) && i < insert_infos.count(); ++i) {

ObRawExpr *insert_expr = nullptr;

ObLoadTableColumnDesc &desc = insert_infos.at(i);

if (OB_NOT_NULL(desc.expr_value_)) {

OZ (copier.copy_on_replace(desc.expr_value_, insert_expr, &replacer));

} else {

insert_expr = field_exprs.at(desc.array_ref_idx_);

}

OZ (insert_exprs.push_back(insert_expr));

LOG_DEBUG("push final insert expr", KPC(insert_expr));

}

}

return ret;

}

int ObLoadDataSPImpl::gen_load_table_column_desc(ObExecContext &ctx,

ObLoadDataStmt &load_stmt,

ObIArray<ObLoadTableColumnDesc> &insert_infos)

{

UNUSED(ctx);

int ret = OB_SUCCESS;

//e.g. general stmt like "INTO TABLE t1 (c1, c2, @a, @b) SET c3 = @a + @b"

// step 1: add c1 and c2

// the first column of file will be written to t1.c1, so c1 will be added to the generator

// similarly, the second column to t1.c2 which also will be added to the generator

// step 2: add c3 (calced by the first assign)

// @a, @b is not match column name, but their data will produce c3 by the "SET" clause,

// in result, c3 will be added

// in addition, replace expr @a with a const string expr which refer to a column from file

// do the same replace to @b

//step 1

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

ObLoadDataStmt::FieldOrVarStruct &item = load_stmt.get_field_or_var_list().at(i);

if (item.is_table_column_) {

ObLoadTableColumnDesc tmp_info;

tmp_info.is_set_values_ = false;

tmp_info.column_name_ = item.field_or_var_name_;

tmp_info.column_id_ = item.column_id_;

tmp_info.column_type_ = item.column_type_;

tmp_info.array_ref_idx_ = i; //array offset

tmp_info.expr_value_ = NULL;

if (OB_FAIL(insert_infos.push_back(tmp_info))) {

LOG_WARN("push str failed", K(ret));

}

} else {

//do nothing

//ignore variables temporarily

}

}

//step 2

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

const ObAssignment &assignment = load_stmt.get_table_assignment().at(i);

ObColumnRefRawExpr *left = assignment.column_expr_;

ObRawExpr *right = assignment.expr_;

if (OB_ISNULL(left)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("set assign expr is null", K(ret));

} /*else if (OB_FAIL(ObRawExprUtils::copy_expr(*ctx.get_expr_factory(),

assignment.expr_,

right,

COPY_REF_SHARED))) {

LOG_WARN("fail to copy expr", K(ret));

} */else {

int64_t found_index = OB_INVALID_INDEX_INT64;

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

if (insert_infos.at(j).column_id_ == left->get_column_id()) {

found_index = j;

break;

}

}

if (found_index != OB_INVALID_INDEX_INT64) {

//overwrite

ObLoadTableColumnDesc &tmp_info = insert_infos.at(found_index);

tmp_info.is_set_values_ = true;

tmp_info.array_ref_idx_ = OB_INVALID_INDEX_INT64;

tmp_info.expr_value_ = right;

} else {

//a new insert column is defined by set expr

ObLoadTableColumnDesc tmp_info;

tmp_info.column_name_ = left->get_column_name();

tmp_info.column_id_ = left->get_column_id();

tmp_info.column_type_ = left->get_result_type().get_type();

tmp_info.is_set_values_ = true;

tmp_info.expr_value_ = right;

if (OB_FAIL(insert_infos.push_back(tmp_info))) {

LOG_WARN("push str failed", K(ret));

}

}

}

}

LOG_DEBUG("generate insert info", K(insert_infos));

return ret;

}

void ObCSVFormats::init(const ObDataInFileStruct &file_formats)

{

field_term_char_ = file_formats.field_term_str_.empty() ?

INT64_MAX : file_formats.field_term_str_[0];

line_term_char_ = file_formats.line_term_str_.empty() ?

INT64_MAX : file_formats.line_term_str_[0];

enclose_char_ = file_formats.field_enclosed_char_;

escape_char_ = file_formats.field_escaped_char_;

null_column_fill_zero_string_ = lib::is_mysql_mode();

if (!file_formats.field_term_str_.empty()

&& file_formats.line_term_str_.empty()) {

is_line_term_by_counting_field_ = true;

line_term_char_ = field_term_char_;

}

is_simple_format_ =

!is_line_term_by_counting_field_

&& (field_term_char_ != INT64_MAX)

&& (line_term_char_ != INT64_MAX)

&& (field_term_char_ != line_term_char_)

&& (enclose_char_ == INT64_MAX);

}

ObShuffleTaskHandle::ObShuffleTaskHandle(ObDataFragMgr &main_datafrag_mgr,

ObBitSet<> &main_string_values,

uint64_t tenant_id)

: exec_ctx(allocator, GCTX.session_mgr_),

data_buffer(NULL),

escape_buffer(NULL),

calc_tablet_id_expr(NULL),

datafrag_mgr(main_datafrag_mgr),

string_values(main_string_values)

{

const int64_t FILE_BUFFER_SIZE = ObLoadFileBuffer::MAX_BUFFER_SIZE;

ObMemAttr attr(tenant_id, ObModIds::OB_SQL_LOAD_DATA);

void *buf = NULL;

buf = ob_malloc(FILE_BUFFER_SIZE, attr);

if (OB_NOT_NULL(buf)) {

data_buffer = new(buf) ObLoadFileBuffer(

FILE_BUFFER_SIZE - sizeof(ObLoadFileBuffer));

}

buf = ob_malloc(FILE_BUFFER_SIZE, attr);

if (OB_NOT_NULL(buf)) {

escape_buffer = new(buf) ObLoadFileBuffer(

FILE_BUFFER_SIZE - sizeof(ObLoadFileBuffer));

}

}

ObShuffleTaskHandle::~ObShuffleTaskHandle()

{

if (OB_NOT_NULL(data_buffer)) {

ob_free(data_buffer);

}

if (OB_NOT_NULL(escape_buffer)) {

ob_free(escape_buffer);

}

}

int ObLoadDataSPImpl::exec_shuffle(int64_t task_id, ObShuffleTaskHandle *handle)

{

int ret = OB_SUCCESS;

int64_t tenant_id = OB_INVALID_TENANT_ID;

void *expr_buf = NULL;

ObLoadFileBuffer *expr_buffer = NULL;

ObArrayHashMap<ObTabletID, ObDataFrag *> part_buf_mgr;

ObSEArray<ObString, 32> insert_values;

int64_t parsed_line_num = 0;

ObStringBuf str_buf("LoadDataStrBuf", OB_MALLOC_MIDDLE_BLOCK_SIZE);

//为了调用 part_buf_mgr.for_each，使用了匿名函数, &引用了外部的 frag_mgr

auto save_frag = [&] (ObTabletID tablet_id, ObDataFrag *frag) -> bool

{

//将存满数据的frag按照分区放入frag_mgr

int ret = OB_SUCCESS;

ObPartDataFragMgr *part_datafrag_mgr = NULL;

if (OB_FAIL(handle->datafrag_mgr.get_part_datafrag(tablet_id,

part_datafrag_mgr))) {

LOG_WARN("fail to get part datafrag", K(ret), K(tablet_id));

} else if (OB_ISNULL(part_datafrag_mgr)) {

ret = OB_ERR_UNEXPECTED;

} else if (OB_FAIL(part_datafrag_mgr->queue_.push(frag))) {

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

} else {

ATOMIC_AAF(&(part_datafrag_mgr->total_row_proceduced_), frag->row_cnt);

LOG_DEBUG("saving frag", K(tablet_id), K(*frag));

}

return OB_SUCCESS == ret;

};

auto free_frag = [&] (ObTabletID tablet_id, ObDataFrag *frag) -> bool

{

if (OB_NOT_NULL(frag)) {

handle->datafrag_mgr.distory_datafrag(frag);

}

return true;

};

if (OB_ISNULL(handle)

|| OB_ISNULL(handle->data_buffer)

|| OB_ISNULL(handle->escape_buffer)

|| OB_ISNULL(handle->exec_ctx.get_my_session())

|| OB_ISNULL(handle->exec_ctx.get_sql_ctx())) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", KP(handle));

// } else if (FALSE_IT(handle->exec_ctx.get_allocator().reuse())) {

} else if (FALSE_IT(tenant_id = handle->exec_ctx.get_my_session()->get_effective_tenant_id())) {

} else if (OB_FAIL(part_buf_mgr.init(ObMemAttr(tenant_id, ObModIds::OB_SQL_LOAD_DATA),

handle->datafrag_mgr.get_total_part_cnt()))) {

LOG_WARN("fail to init part buf mgr", K(ret));

} else if (OB_FAIL(insert_values.prepare_allocate(

handle->generator.get_insert_exprs().count()))) {

LOG_WARN("fail to prealloc", K(ret),

"insert values count", handle->generator.get_insert_exprs().count());