hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
4c7cfd912030cff30fdba7404781dfe6c668c960
oceanbase/src/sql/engine/cmd/ob_load_data_rpc.cpp
wangzelin.wzl 93a1074b0c patch 4.0
2022-10-24 17:57:12 +08:00

757 lines
27 KiB
C++
Raw Blame History

/**
* 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 SQL_ENG
#include "sql/engine/cmd/ob_load_data_rpc.h"
#include "observer/ob_server_struct.h"
#include "lib/oblog/ob_log_module.h"
#include "lib/string/ob_sql_string.h"
#include "lib/utility/serialization.h"
#include "share/ob_tenant_mgr.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_impl.h"
#include "lib/string/ob_string.h"
#include "storage/tx_storage/ob_tenant_freezer.h"
using namespace oceanbase::sql;
using namespace oceanbase::common;
using namespace oceanbase::share;
using namespace oceanbase::share::schema;
using namespace oceanbase::storage;
namespace oceanbase
{
namespace sql
{
int ObLoadbuffer::deep_copy_str(const ObString &src, ObString &dest)
{
int ret = OB_SUCCESS;
char *buf = NULL;
int64_t len = src.length() + 1;
if (OB_UNLIKELY(len <= 0)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("src string length is invalid", K(ret), K(src), K(len));
} else if (NULL == (buf = static_cast<char*>(field_data_allocator_.alloc(len)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_ERROR("fail to allocate memory, ", K(ret), K(src), K(len));
} else {
MEMCPY(buf, src.ptr(), len - 1);
buf[len - 1] = '\0';
dest.assign_ptr(buf, static_cast<ObString::obstr_size_t>(len - 1));
}
return ret;
}
int ObLoadbuffer::store_row(const ObIArray<ObString> &row_strs,
int64_t cur_line_number)
{
int ret = OB_SUCCESS;
for (int64_t i = 0; OB_SUCC(ret) && i < insert_column_num_; ++i) {
const ObString &src_str = row_strs.at(i);
ObString &dst_str = insert_values_.at(stored_pos_);
if (OB_FAIL(deep_copy_str(src_str, dst_str))) {
LOG_WARN("deep copy string failed", K(ret));
}
stored_pos_++;
}
file_line_number_.at(stored_row_cnt_) = cur_line_number;
stored_row_cnt_++;
return ret;
}
int ObLoadbuffer::set_allocator_tenant()
{
int ret = OB_SUCCESS;
if (OB_INVALID_ID == tenant_id_) {
ret = OB_NOT_INIT;
LOG_WARN("tenant_id must be init first", K(tenant_id_));
} else {
field_data_allocator_.set_tenant_id(tenant_id_);
array_allocator_.set_tenant_id(tenant_id_);
}
return ret;
}
int ObLoadbuffer::prepare_insert_info(const ObIArray<ObString> &column_names, ObExprValueBitSet &expr_value_bitset)
{
int ret = OB_SUCCESS;
if (OB_FAIL(expr_bitset_.add_members(expr_value_bitset))) {
LOG_WARN("prepare expr bitset failed", K(ret));
}
for (int64_t i = 0; OB_SUCC(ret) && i < column_names.count(); ++i) {
if (OB_FAIL(insert_column_names_.push_back(column_names.at(i)))) {
LOG_WARN("insert column names push failed", K(ret));
}
}
return ret;
}
int ObLoadbuffer::init_array()
{
int ret = OB_SUCCESS;
for (int64_t i = 0; OB_SUCC(ret) && i < LOAD_BUFFER_MAX_ROW_COUNT * insert_column_num_; ++i) {
if (OB_FAIL(insert_values_.push_back(ObString::make_empty_string()))) {
LOG_WARN("insert values push empty string failed", K(ret));
}
}
for (int64_t i = 0; OB_SUCC(ret) && i < LOAD_BUFFER_MAX_ROW_COUNT; ++i) {
if (OB_FAIL(file_line_number_.push_back(-1))) {
LOG_WARN("insert values push empty string failed", K(ret));
}
}
return ret;
}
int ObRpcLoadDataShuffleTaskExecuteP::process()
{
UNUSED(gctx_);
int ret = OB_SUCCESS;
ObShuffleTask &task = arg_;
//ObShuffleResult &result = result_;
ObShuffleTaskHandle *handle = nullptr;
ObLoadDataStat *job_status = nullptr;
LOG_DEBUG("LOAD DATA receiving shuffle task", "task_id", task.task_id_);
if (OB_FAIL(ObGlobalLoadDataStatMap::getInstance()->get_job_status(task.gid_, job_status))) {
LOG_WARN("fail to get job, main thread has already quit", K(ret), K(task));
} else if (OB_ISNULL(job_status)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("job status is null", K(ret));
} else {
if (OB_FAIL(task.shuffle_task_handle_.get_arg(handle))) { //check identifier
LOG_ERROR("fail to get arg", K(ret));
} else if (OB_ISNULL(handle)) {
ret = OB_ERR_UNEXPECTED;
LOG_ERROR("handle is null", K(ret));
} else {
if (OB_UNLIKELY(THIS_WORKER.is_timeout())) {
ret = OB_TIMEOUT;
LOG_WARN("LOAD DATA shuffle task timeout", K(ret), K(task));
} else if (OB_FAIL(ObLoadDataSPImpl::exec_shuffle(task.task_id_, handle))) {
LOG_WARN("fail to exec shuffle task", K(ret));
}
handle->result.exec_ret_ = ret;
}
MEM_BARRIER(); //use handle ptr before release job ref
job_status->release();
}
return OB_SUCCESS;
}
int ObRpcLoadDataShuffleTaskCallBack::release_resouce()
{
int ret = OB_SUCCESS;
int ret1 = complete_task_list_.push_back(handle_);
MEM_BARRIER();
int ret2 = task_controller_.on_task_finished();
if (OB_FAIL(ret1) || OB_FAIL(ret2)) {
LOG_ERROR("shuffle call back release resource failed", K(ret1), K(ret2));
}
return ret;
}
int ObRpcLoadDataShuffleTaskCallBack::process() {
//int ret = OB_SUCCESS;
//ObShuffleResult &result = result_;
LOG_DEBUG("LOAD DATA shuffle task callback process");
handle_->result.process_us_ = ObTimeUtil::current_time() - get_send_ts();
//handle_->result = result;
release_resouce();
return OB_SUCCESS;
}
void ObRpcLoadDataShuffleTaskCallBack::on_timeout() {
LOG_WARN("LOAD DATA main thread shuffle task rpc timeout");
if (OB_NOT_NULL(handle_)) {
handle_->result.flags_.set_bit(ObTaskResFlag::RPC_TIMEOUT);
}
release_resouce();
}
void ObRpcLoadDataShuffleTaskCallBack::set_args(const Request &arg)
{
UNUSED(arg);
}
int ObRpcLoadDataInsertTaskCallBack::release_resouce()
{
int ret = OB_SUCCESS;
int ret1 = complete_task_list_.push_back(insert_task_);
MEM_BARRIER();
int ret2 = task_controller_.on_task_finished();
if (OB_FAIL(ret1) || OB_FAIL(ret2)) {
LOG_ERROR("push back buffer failed", K(ret1), K(ret2));
}
return ret;
}
int ObRpcLoadDataInsertTaskCallBack::process() {
int ret = OB_SUCCESS;
LOG_DEBUG("LOAD DATA insert task callback process", K(result_));
if (OB_NOT_NULL(insert_task_)) {
if (OB_FAIL(insert_task_->result_.assign(result_))) {
LOG_WARN("fail to assign result", K(ret));
} else {
insert_task_->result_recv_ts_ = ObTimeUtil::current_time();
insert_task_->process_us_ = insert_task_->result_recv_ts_ - get_send_ts();
}
}
release_resouce();
return OB_SUCCESS;
}
void ObRpcLoadDataInsertTaskCallBack::set_args(const Request &arg)
{
UNUSED(arg);
}
void ObRpcLoadDataInsertTaskCallBack::on_timeout() {
int64_t task_id = -2; //undefined
if (OB_ISNULL(insert_task_)) {
LOG_ERROR("insert task is null on timeout");
} else {
task_id = insert_task_->task_id_;
insert_task_->result_.flags_.set_bit(ObTaskResFlag::RPC_TIMEOUT);
}
LOG_WARN("LOAD DATA main thread insert task rpc timeout", K(task_id));
release_resouce();
}
int ObRpcLoadDataTaskExecuteP::process()
{
int ret = OB_SUCCESS;
ObLoadbuffer &buffer = arg_;
ObLoadResult &result = result_;
ObIArray<ObString> &column_names = buffer.get_insert_keys();
ObIArray<ObString> &values = buffer.get_insert_values();
ObString &table_name = buffer.get_table_name();
ObExprValueBitSet &expr_bitset = buffer.get_expr_bitset();
uint64_t tenant_id = buffer.get_tenant_id();
int64_t task_id = buffer.get_task_id();
ObTabletID tablet_id = buffer.get_tablet_id();
ObLoadDupActionType insert_mode = buffer.get_load_mode();
int64_t insert_column_number = column_names.count();
int64_t total_values_number = buffer.get_stored_pos(); //values.count() will always be 1000
int64_t total_row_number = buffer.get_stored_row_count();
int64_t succ_row_count = 0;
int need_wait_minor_freeze = false;
if (OB_UNLIKELY(total_values_number <= 0)
|| OB_UNLIKELY(insert_column_number <= 0)
|| OB_UNLIKELY(total_values_number != insert_column_number * (total_values_number / insert_column_number))
|| OB_UNLIKELY(total_row_number != total_values_number / insert_column_number)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("insert values are invalid", K(ret), K(total_values_number),
K(insert_column_number), K(total_row_number), K(buffer));
} else if (!escape_data_buffer_.set_data(str_buf_, OB_MAX_DEFAULT_VALUE_LENGTH)) { //for escaped value
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid escape_buf", K(ret));
} else {
//1. try insert as one stmt
ObSqlString batch_insert_sql;
int exec_ret = OB_SUCCESS;
int64_t affected_rows = 0;
if (OB_FAIL(ObLoadDataUtils::build_insert_sql_string_head(insert_mode,
table_name,
column_names,
batch_insert_sql))) {
LOG_WARN("gen insert sql column_names failed", K(ret));
} else if (OB_FAIL(ObLoadDataUtils::append_values_in_remote_process(insert_column_number,
total_values_number,
expr_bitset,
values,
batch_insert_sql,
escape_data_buffer_))) {
LOG_WARN("append values failed", K(ret));
} else if (OB_UNLIKELY(OB_SUCCESS != (exec_ret = gctx_.sql_proxy_->write(tenant_id,
batch_insert_sql.ptr(),
affected_rows,
get_compatibility_mode())))) {
LOG_WARN("fail to execute insert sql in batch", K(exec_ret), K(affected_rows), K(task_id), K(get_compatibility_mode()));
} else {
//succ!
succ_row_count = affected_rows;
}
//2. if failed, try insert seperately
if (OB_SUCC(ret) && OB_SUCCESS != exec_ret) {
ObSqlString seperate_insert_sql_head;
ObSqlString seperate_insert_sql;
if (OB_FAIL(ObLoadDataUtils::build_insert_sql_string_head(insert_mode,
table_name,
column_names,
seperate_insert_sql_head))) {
LOG_WARN("gen insert sql column_names failed", K(ret));
}
for (int64_t row_idx = 0; OB_SUCC(ret) && row_idx < total_row_number; ++row_idx) {
if (OB_FAIL(seperate_insert_sql.assign(seperate_insert_sql_head.string()))) {
LOG_WARN("assign insert column_names failed", K(ret));
} else if (OB_FAIL(ObLoadDataUtils::append_values_in_remote_process(insert_column_number,
insert_column_number,
expr_bitset,
values,
seperate_insert_sql,
escape_data_buffer_,
row_idx))) {
LOG_WARN("append values failed", K(ret));
} else {
exec_ret = OB_SUCCESS;
affected_rows = 0;
if (OB_UNLIKELY(OB_SUCCESS != (exec_ret = gctx_.sql_proxy_->write(tenant_id,
seperate_insert_sql.ptr(),
affected_rows,
get_compatibility_mode())))) {
LOG_WARN("LOAD DATA row insert failed in remote process", K(exec_ret), K(seperate_insert_sql), K(get_compatibility_mode()));
if (OB_FAIL(result.row_number_.push_back(static_cast<int16_t>(row_idx)))) {
LOG_WARN("push back row number failed", K(ret));
} else if (OB_FAIL(result.row_err_code_.push_back(exec_ret))) {
LOG_WARN("push back row err code failed", K(ret));
}
} else {
//succ!
succ_row_count++;
}
}
}
}
}
//check memory
int memory_check_ret = OB_SUCCESS;
MAKE_TENANT_SWITCH_SCOPE_GUARD(guard);
storage::ObTenantFreezer *freezer = nullptr;
if (tenant_id != MTL_ID()) {
memory_check_ret = guard.switch_to(tenant_id);
}
if (OB_UNLIKELY(OB_SUCCESS != memory_check_ret)) {
ret = OB_ERR_UNEXPECTED;
LOG_ERROR("switch tenant failed", K(tenant_id), K(memory_check_ret));
} else 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_UNLIKELY(OB_SUCCESS != (memory_check_ret =
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(memory_check_ret));
} else {
if (static_cast<double>(total_memstore_used)
> static_cast<double>(major_freeze_trigger) * 1.02) {
need_wait_minor_freeze = true;
}
}
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));
}
if (OB_SUCC(ret) && OB_SUCCESS != memory_check_ret) {
ret = memory_check_ret; //cover ret
}
//sumarize return value
if (OB_FAIL(ret)) {
ObLoadDataUtils::set_flag(result.task_flags_,
static_cast<int64_t>(ObLoadTaskResultFlag::RPC_REMOTE_PROCESS_ERROR));
}
if (need_wait_minor_freeze) {
ObLoadDataUtils::set_flag(result.task_flags_,
static_cast<int64_t>(ObLoadTaskResultFlag::NEED_WAIT_MINOR_FREEZE));
}
if (succ_row_count != total_row_number) {
ObLoadDataUtils::set_flag(result.task_flags_,
static_cast<int64_t>(ObLoadTaskResultFlag::HAS_FAILED_ROW));
}
if (succ_row_count == 0) {
ObLoadDataUtils::set_flag(result.task_flags_,
static_cast<int64_t>(ObLoadTaskResultFlag::ALL_ROWS_FAILED));
}
result.affected_rows_ = succ_row_count;
result.failed_rows_ = total_row_number - succ_row_count;
result.task_id_ = task_id;
result.tablet_id_ = tablet_id;
LOG_DEBUG("load data in remote process!", K(task_id), K(succ_row_count),
K(total_row_number), K(need_wait_minor_freeze), K(lib::is_oracle_mode()));
return OB_SUCCESS;
}
OB_INLINE void ObRpcLoadDataTaskCallBack::set_args(const Request &arg)
{
UNUSED(arg);
}
void ObRpcLoadDataTaskCallBack::on_timeout()
{
int ret = OB_SUCCESS;
ObLoadbuffer *buffer = static_cast<ObLoadbuffer*>(request_buffer_ptr_);
if (OB_ISNULL(buffer)) {
ret = OB_ERR_UNEXPECTED;
LOG_ERROR("buffer is null", K(ret));
} else {
buffer->set_returned_timestamp(ObTimeUtility::current_time());
ObLoadDataUtils::set_flag(buffer->get_task_status(),
static_cast<int64_t>(ObLoadTaskResultFlag::TIMEOUT));
ObLoadDataUtils::set_flag(buffer->get_task_status(),
static_cast<int64_t>(ObLoadTaskResultFlag::HAS_FAILED_ROW));
ObLoadDataUtils::set_flag(buffer->get_task_status(),
static_cast<int64_t>(ObLoadTaskResultFlag::ALL_ROWS_FAILED));
//TODO wjh: check task real status, is all rows truely failed?
if (OB_FAIL(complete_task_list_.push_back(buffer))) {
LOG_ERROR("push back buffer failed", K(ret));
}
MEM_BARRIER();
if (OB_FAIL(task_controller_.on_task_finished())) {
LOG_ERROR("error on task finish", K(ret));
}
LOG_WARN("load data task is timeout", K(ret), KPC(buffer));
}
}
int ObRpcLoadDataTaskCallBack::process()
{
int ret = OB_SUCCESS;
ObLoadResult &result = result_;
ObLoadbuffer *buffer = static_cast<ObLoadbuffer*>(request_buffer_ptr_);
if (OB_ISNULL(buffer)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("task buffer is null", K(ret));
} else {
buffer->set_returned_timestamp(ObTimeUtility::current_time());
buffer->set_task_status(result.task_flags_);
if (OB_UNLIKELY(ObLoadDataUtils::has_flag(buffer->get_task_status(),
static_cast<int64_t>(ObLoadTaskResultFlag::HAS_FAILED_ROW)))) {
if (OB_UNLIKELY(result.failed_rows_ != result.row_err_code_.count()
|| result.failed_rows_ != result.row_number_.count())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("size of failed rows is not the same, impossible");
} else {
for (int64_t i = 0; OB_SUCC(ret) && i < result.failed_rows_; ++i) {
if (OB_FAIL(buffer->get_failed_row_idx().push_back(result.row_number_.at(i)))) {
LOG_WARN("push back failed", K(ret));
} else if (OB_FAIL(buffer->get_error_code_array().push_back(result.row_err_code_.at(i)))) {
LOG_WARN("push back failed", K(ret));
}
}
}
}
if (OB_FAIL(ret)) {
ObLoadDataUtils::set_flag(buffer->get_task_status(),
static_cast<int64_t>(ObLoadTaskResultFlag::RPC_CALLBACK_PROCESS_ERROR));
}
//whatever ret is, the task has finished
int second_ret = OB_SUCCESS;
if (OB_SUCCESS != (second_ret = complete_task_list_.push_back(buffer))) {
//buffer will be auto released by the allocator
if (OB_SUCCESS == ret) {
ret = second_ret;
}
LOG_ERROR("push to complete list failed", K(second_ret));
}
MEM_BARRIER();
if (OB_SUCCESS != (second_ret = task_controller_.on_task_finished())) {
//signal main thread failed, fetal error. TODO wjh: handle this error
if (OB_SUCCESS == ret) {
ret = second_ret;
}
LOG_ERROR("on task finished failed", K(ret));
}
}
LOG_DEBUG("in local cb process!", K(result), K(ret));
return OB_SUCCESS;
}
int ObParallelTaskController::init(int64_t max_parallelism)
{
int ret = OB_SUCCESS;
max_parallelism_ = max_parallelism;
if (OB_FAIL(vacant_cond_.init(common::ObWaitEventIds::ASYNC_RPC_PROXY_COND_WAIT))) {
LOG_WARN("init vacant cond failed", K(ret));
}
return ret;
}
int ObParallelTaskController::on_next_task()
{
int ret = OB_SUCCESS;
ObThreadCondGuard guard(vacant_cond_);
if (ATOMIC_AAF(&processing_cnt_, 1) > max_parallelism_) {
ret = vacant_cond_.wait();
}
return ret;
}
void ObParallelTaskController::wait_all_task_finish(const char *task_name, int64_t until_ts)
{
int64_t wait_duration_ms = 0;
int64_t begin_ts = ObTimeUtil::current_time();
int64_t processing_count = 0;
bool is_too_long = false;
LOG_DEBUG("start wait_all_task_finish", K(task_name));
while ((processing_count = get_processing_task_cnt()) > 0) {
ob_usleep(1000 * 10); //wait 10m
wait_duration_ms += 10;
if (0 == wait_duration_ms % 1000) {
int64_t current_ts = ObTimeUtil::current_time();
if (current_ts > until_ts) {
LOG_ERROR("waiting load data task too long and exceed max waiting timestamp",
K(begin_ts), K(until_ts), K(current_ts));
}
}
if (!is_too_long && wait_duration_ms > 10 * 1000) {
is_too_long = true;
LOG_WARN("LOAD DATA, waiting task finish too long",
K(task_name), K(processing_count), K(wait_duration_ms), K(until_ts));
}
}
if (is_too_long) {
LOG_WARN("LOAD DATA finish waitting long task", K(wait_duration_ms));
}
}
int ObParallelTaskController::on_task_finished()
{
int ret = OB_SUCCESS;
if(max_parallelism_ == ATOMIC_AAF(&processing_cnt_, -1)) {
ObThreadCondGuard guard(vacant_cond_);
ret = vacant_cond_.signal();
}
return ret;
}
int ObRpcLoadDataInsertTaskExecuteP::process()
{
UNUSED(gctx_);
int ret = OB_SUCCESS;
ObInsertTask &task = arg_;
ObInsertResult &result = result_;
ObWarningBuffer *warning_buf = NULL;
bool need_wait_freeze = false;
LOG_DEBUG("LOAD DATA receiving insert task", "task_id", task.task_id_);
if (OB_UNLIKELY(THIS_WORKER.is_timeout())) {
ret = OB_TIMEOUT;
LOG_WARN("LOAD DATA shuffle task timeout", K(ret), K(task));
} else if (OB_FAIL(ObLoadDataSPImpl::exec_insert(task, result))) {
LOG_WARN("fail to exec insert", K(ret));
}
result.exec_ret_ = ret;
if (OB_FAIL(ret) && OB_NOT_NULL(warning_buf = ob_get_tsi_warning_buffer())) {
int ret_backup = ret;
result.err_line_no_ = warning_buf->get_error_line();
OZ (ob_write_string(result.allocator_, warning_buf->get_err_msg(), result.err_msg_));
ret = ret_backup;
LOG_DEBUG("check warning buffer", "msg", warning_buf->get_err_msg(),
"line", warning_buf->get_error_line(),
"column", warning_buf->get_error_column(),
"code", warning_buf->get_err_code());
}
int temp_ret = ObLoadDataBase::memory_check_remote(task.tenant_id_, need_wait_freeze);
if (OB_SUCCESS != temp_ret) {
LOG_WARN("LOAD DATA remote memory check failed", K(temp_ret), K(task.tenant_id_));
}
result.flags_.reset();
if (need_wait_freeze) {
result.flags_.set_bit(ObTaskResFlag::NEED_WAIT_MINOR_FREEZE);
}
return OB_SUCCESS;
}
int ObInsertResult::assign(const ObInsertResult &other)
{
int ret = OB_SUCCESS;
flags_ = other.flags_;
exec_ret_ = other.exec_ret_;
err_line_no_ = other.err_line_no_;
if (OB_FAIL(ob_write_string(allocator_, other.err_msg_, err_msg_))) {
LOG_WARN("fail to write string", K(ret));
}
return ret;
}
OB_DEF_SERIALIZE(ObLoadbuffer)
{
int ret = OB_SUCCESS;
LST_DO_CODE(OB_UNIS_ENCODE,
task_id_,
tablet_id_,
table_id_,
tenant_id_,
stored_pos_,
stored_row_cnt_,
insert_mode_,
insert_column_num_);
if (OB_FAIL(ret)) {
} else if (OB_FAIL(table_name_.serialize(buf, buf_len, pos))) {
LOG_WARN("serialize error", K(ret));
} else if (OB_FAIL(insert_column_names_.serialize(buf, buf_len, pos))) {
LOG_WARN("serialize column names error", K(ret));
} else if (OB_FAIL(insert_values_.serialize(buf, buf_len, pos))) {
LOG_WARN("serialize row store error", K(ret));
} else if (OB_FAIL(expr_bitset_.serialize(buf, buf_len, pos))) {
LOG_WARN("serialize expr bitset error", K(ret));
}
return ret;
}
OB_DEF_DESERIALIZE(ObLoadbuffer)
{
int ret = OB_SUCCESS;
LST_DO_CODE(OB_UNIS_DECODE,
task_id_,
tablet_id_,
table_id_,
tenant_id_,
stored_pos_,
stored_row_cnt_,
insert_mode_,
insert_column_num_);
if (OB_FAIL(ret)) {
} else if (OB_FAIL(table_name_.deserialize(buf, data_len, pos))) {
LOG_WARN("deserialize error", K(ret));
} else if (OB_FAIL(insert_column_names_.deserialize(buf, data_len, pos))) {
LOG_WARN("deserialize row store error", K(ret));
} else if (OB_FAIL(insert_values_.deserialize(buf, data_len, pos))) {
LOG_WARN("deserialize row store error", K(ret));
} else if (OB_FAIL(expr_bitset_.deserialize(buf, data_len, pos))) {
LOG_WARN("deserialize expr bitset error", K(ret));
}
return ret;
}
OB_DEF_SERIALIZE_SIZE(ObLoadbuffer)
{
int64_t len = 0;
LST_DO_CODE(OB_UNIS_ADD_LEN,
task_id_,
tablet_id_,
table_id_,
tenant_id_,
stored_pos_,
stored_row_cnt_,
insert_mode_,
insert_column_num_);
len += table_name_.get_serialize_size();
len += insert_column_names_.get_serialize_size();
len += insert_values_.get_serialize_size();
len += expr_bitset_.get_serialize_size();
return len;
}
OB_DEF_SERIALIZE(ObLoadResult)
{
int ret = OB_SUCCESS;
LST_DO_CODE(OB_UNIS_ENCODE,
task_id_,
tablet_id_,
affected_rows_,
failed_rows_,
task_flags_);
if (OB_FAIL(ret)) {
} else if (OB_FAIL(row_number_.serialize(buf, buf_len, pos))) {
LOG_WARN("serialize row_number_ error", K(ret));
} else if (OB_FAIL(row_err_code_.serialize(buf, buf_len, pos))) {
LOG_WARN("serialize row_err_code_ error", K(ret));
}
return ret;
}
OB_DEF_DESERIALIZE(ObLoadResult)
{
int ret = OB_SUCCESS;
LST_DO_CODE(OB_UNIS_DECODE,
task_id_,
tablet_id_,
affected_rows_,
failed_rows_,
task_flags_);
if (OB_FAIL(ret)) {
} else if (OB_FAIL(row_number_.deserialize(buf, data_len, pos))) {
LOG_WARN("deserialize row_number_ error", K(ret));
} else if (OB_FAIL(row_err_code_.deserialize(buf, data_len, pos))) {
LOG_WARN("deserialize row_err_code_ error", K(ret));
}
return ret;
}
OB_DEF_SERIALIZE_SIZE(ObLoadResult)
{
int64_t len = 0;
LST_DO_CODE(OB_UNIS_ADD_LEN,
task_id_,
tablet_id_,
affected_rows_,
failed_rows_,
task_flags_);
len += row_number_.get_serialize_size();
len += row_err_code_.get_serialize_size();
return len;
}
OB_SERIALIZE_MEMBER(ObShuffleTask,
task_id_,
shuffle_task_handle_,
gid_);
OB_SERIALIZE_MEMBER(ObShuffleResult,
task_id_,
flags_,
exec_ret_,
row_cnt_);
OB_SERIALIZE_MEMBER(ObInsertTask,
tenant_id_,
task_id_,
row_count_,
column_count_,
insert_stmt_head_,
insert_value_data_);
OB_SERIALIZE_MEMBER(ObInsertResult,
flags_,
exec_ret_,
err_line_no_,
err_msg_);
}
}
Reference in New Issue View Git Blame Copy Permalink