/**

* 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/dml/ob_table_modify_op.h"

#include "sql/engine/dml/ob_dml_service.h"

#include "sql/engine/dml/ob_table_insert_op.h"

#include "sql/engine/dml/ob_table_update_op.h"

#include "sql/engine/basic/ob_expr_values_op.h"

#include "sql/engine/expr/ob_expr_column_conv.h"

#include "sql/engine/expr/ob_expr_calc_partition_id.h"

#include "sql/executor/ob_task_spliter.h"

#include "sql/das/ob_das_utils.h"

#include "storage/ob_i_store.h"

#include "lib/mysqlclient/ob_isql_client.h"

#include "observer/ob_inner_sql_connection_pool.h"

#include "lib/worker.h"

#include "share/ob_debug_sync.h"

#include "sql/engine/dml/ob_fk_checker.h"

namespace oceanbase

{

using namespace common;

using namespace common::sqlclient;

using namespace storage;

using namespace share;

using namespace share::schema;

using namespace observer;

namespace sql

{

int ForeignKeyHandle::do_handle(ObTableModifyOp &op,

const ObDMLBaseCtDef &dml_ctdef,

ObDMLBaseRtDef &dml_rtdef)

{

int ret = OB_SUCCESS;

if (op.need_foreign_key_checks()) {

const ObExprPtrIArray &old_row = dml_ctdef.old_row_;

const ObExprPtrIArray &new_row = dml_ctdef.new_row_;

op.get_exec_ctx().get_das_ctx().is_fk_cascading_ = true;

LOG_DEBUG("do foreign_key_handle", K(old_row), K(new_row));

if (OB_FAIL(op.check_stack())) {

LOG_WARN("fail to check stack", K(ret));

}

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

const ObForeignKeyArg &fk_arg = dml_ctdef.fk_args_.at(i);

ObForeignKeyChecker *fk_checker = dml_rtdef.fk_checker_array_.at(i);

if (OB_SUCC(ret) && !new_row.empty()) {

if (ACTION_CHECK_EXIST == fk_arg.ref_action_) {

// insert or update.

bool is_foreign_key_cascade = false;

if (OB_FAIL(op.get_foreign_key_cascade(is_foreign_key_cascade))) {

LOG_WARN("failed to get foreign key cascade", K(ret), K(fk_arg), K(new_row));

} else if (is_foreign_key_cascade) {

// nested update can not check parent row.

LOG_DEBUG("skip foreign_key_check_exist in nested session");

} else if (OB_FAIL(check_exist(op, fk_arg, new_row, fk_checker, false))) {

LOG_WARN("failed to check exist", K(ret), K(fk_arg), K(new_row));

}

}

}

if (OB_SUCC(ret) && !old_row.empty()) {

if (ACTION_RESTRICT == fk_arg.ref_action_ || ACTION_NO_ACTION == fk_arg.ref_action_) {

// update or delete.

bool has_changed = false;

if (OB_FAIL(value_changed(op, fk_arg.columns_, old_row, new_row, has_changed))) {

LOG_WARN("failed to check if foreign key value has changed",

K(ret), K(fk_arg), K(old_row), K(new_row));

} else if (!has_changed) {

// nothing.

} else if (OB_FAIL(check_exist(op, fk_arg, old_row, fk_checker, true))) {

LOG_WARN("failed to check exist", K(ret), K(fk_arg), K(old_row));

}

} else if (ACTION_CASCADE == fk_arg.ref_action_) {

// update or delete.

bool is_self_ref = false;

if (OB_FAIL(is_self_ref_row(op.get_eval_ctx(), old_row, fk_arg, is_self_ref))) {

LOG_WARN("is_self_ref_row failed", K(ret), K(old_row), K(fk_arg));

} else if (new_row.empty() && is_self_ref && op.is_fk_nested_session()) {

// delete self refercnced row should not cascade delete.

} else if (OB_FAIL(check_exist_inner_sql(op, fk_arg, old_row, true, true))) {

LOG_WARN("check exist before cascade failed", K(ret), K(fk_arg), K(old_row));

} else if (OB_FAIL(cascade(op, fk_arg, old_row, new_row))) {

LOG_WARN("failed to cascade", K(ret), K(fk_arg), K(old_row), K(new_row));

} else if (!new_row.empty() && is_self_ref) {

// we got here only when:

// 1. handling update operator and

// 2. foreign key constraint is self reference

// need to change %new_row and %old_row

// (see

// xxx_row_res_info helps to restore row. restore row before get_next_row()

//op.fk_self_ref_row_res_infos_.reset();

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

const int32_t val_idx = fk_arg.columns_.at(i).idx_;

const int32_t name_idx = fk_arg.columns_.at(i).name_idx_;

if (OB_SUCC(ret)) {

const ObDatum &updated_value = new_row.at(val_idx)->

locate_expr_datum(op.get_eval_ctx());

ObDatum &new_row_name_col = new_row.at(name_idx)->

locate_expr_datum(op.get_eval_ctx());

ObDatum &old_row_name_col = old_row.at(name_idx)->

locate_expr_datum(op.get_eval_ctx());

OX(new_row_name_col = updated_value);

OX(old_row_name_col = updated_value);

}

}

}

} else if (ACTION_SET_NULL == fk_arg.ref_action_) {

if (OB_FAIL(check_exist_inner_sql(op, fk_arg, old_row, true, true))) {

LOG_WARN("check exist before cascade failed", K(ret), K(fk_arg), K(old_row));

} else if (OB_FAIL(set_null(op, fk_arg, old_row))) {

LOG_WARN("failed to perform set null for foreign key", K(ret));

}

}

} // if (old_row.is_valid())

} // for

//fk cascading end

op.get_exec_ctx().get_das_ctx().is_fk_cascading_ = false;

} else {

LOG_DEBUG("skip foreign_key_handle");

}

return ret;

}

int ForeignKeyHandle::value_changed(ObTableModifyOp &op,

const ObIArray<ObForeignKeyColumn> &columns,

const ObExprPtrIArray &old_row,

const ObExprPtrIArray &new_row,

bool &has_changed)

{

int ret = OB_SUCCESS;

has_changed = false;

if (!old_row.empty() && !new_row.empty()) {

ObDatum *old_row_col = NULL;

ObDatum *new_row_col = NULL;

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

int64_t col_idx = columns.at(i).idx_;

CK(col_idx < old_row.count());

CK(col_idx < new_row.count());

CK(OB_NOT_NULL(old_row.at(col_idx)));

CK(OB_NOT_NULL(new_row.at(col_idx)));

OZ(old_row.at(col_idx)->eval(op.get_eval_ctx(), old_row_col));

OZ(new_row.at(col_idx)->eval(op.get_eval_ctx(), new_row_col));

OX(has_changed = (false == ObDatum::binary_equal(*old_row_col, *new_row_col)));

}

} else {

has_changed = true;

}

return ret;

}

int ForeignKeyHandle::check_exist(ObTableModifyOp &modify_op, const ObForeignKeyArg &fk_arg,

const ObExprPtrIArray &row, ObForeignKeyChecker *fk_checker, bool expect_zero)

{

int ret = OB_SUCCESS;

DEBUG_SYNC(BEFORE_FOREIGN_KEY_CONSTRAINT_CHECK);

if (!expect_zero) {

ret = check_exist_scan_task(modify_op, fk_arg, row, fk_checker);

} else {

if (OB_FAIL(check_exist_inner_sql(modify_op, fk_arg, row, expect_zero, true))) {

LOG_WARN("check exist and iter uncommmited row meet failed", K(ret));

} else if (OB_FAIL(check_exist_inner_sql(modify_op, fk_arg, row, expect_zero, false))) {

LOG_WARN("check exist and iter commmited row meet failed", K(ret));

}

}

return ret;

}

int ForeignKeyHandle::check_exist_scan_task(ObTableModifyOp &modify_op, const ObForeignKeyArg &fk_arg,

const ObExprPtrIArray &row, ObForeignKeyChecker *fk_checker)

{

int ret = OB_SUCCESS;

bool has_result = false;

if (OB_ISNULL(fk_checker)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("foreign key checker is nullptr", K(ret));

} else if (OB_FAIL(fk_checker->do_fk_check_single_row(fk_arg.columns_, row, has_result))) {

LOG_WARN("failed to perform foreign key check by das scan tasks", K(ret));

} else {

if (!has_result) {

ret = OB_ERR_NO_REFERENCED_ROW;

} else {

ret = OB_SUCCESS;

}

}

return ret;

}

int ForeignKeyHandle::check_exist_inner_sql(ObTableModifyOp &op,

const ObForeignKeyArg &fk_arg,

const ObExprPtrIArray &row,

bool expect_zero,

bool iter_uncommitted_row)

{

int ret = OB_SUCCESS;

static const char *SELECT_FMT_MYSQL =

"select /*+ no_parallel */ 1 from `%.*s`.`%.*s` where %.*s for update";

static const char *SELECT_FMT_ORACLE =

"select /*+ no_parallel */ 1 from \"%.*s\".\"%.*s\" where %.*s for update";

const char *select_fmt = lib::is_mysql_mode() ? SELECT_FMT_MYSQL : SELECT_FMT_ORACLE;

ObArenaAllocator alloc(ObModIds::OB_MODULE_PAGE_ALLOCATOR,

OB_MALLOC_NORMAL_BLOCK_SIZE,

op.get_eval_ctx().exec_ctx_.get_my_session()->get_effective_tenant_id());

char *stmt_buf = nullptr;

int64_t stmt_len = 0;

char *where_buf = nullptr;

int64_t where_len = 0;

int64_t stmt_pos = 0;

int64_t where_pos = 0;

const ObString &database_name = fk_arg.database_name_;

const ObString &table_name = fk_arg.table_name_;

if (row.empty()) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("row is invalid", K(ret));

} else if (OB_FAIL(gen_where(op.get_eval_ctx(), where_buf, where_len, where_pos,

fk_arg.columns_, row, alloc, op.get_obj_print_params()))) {

if (OB_LIKELY(OB_ERR_NULL_VALUE == ret)) {

// skip check exist if any column in foreign key is NULL.

ret = OB_SUCCESS;

stmt_pos = 0;

} else {

LOG_WARN("failed to gen foreign key where", K(ret), K(row), K(fk_arg.columns_));

}

} else if (OB_FAIL(databuff_printf(stmt_buf, stmt_len, stmt_pos, alloc, select_fmt,

database_name.length(), database_name.ptr(),

table_name.length(), table_name.ptr(),

static_cast<int>(where_pos), where_buf))) {

LOG_WARN("failed to print stmt", K(ret), K(table_name), K(where_buf));

} else {

stmt_buf[stmt_pos++] = 0;

}

if (OB_SUCC(ret) && stmt_pos > 0) {

if (iter_uncommitted_row) {

op.get_exec_ctx().get_das_ctx().iter_uncommitted_row_ = true;

}

LOG_DEBUG("foreign_key_check_exist", "stmt", stmt_buf, K(row), K(fk_arg));

SMART_VAR(ObMySQLProxy::MySQLResult, res) {

if (OB_FAIL(op.begin_nested_session(fk_arg.is_self_ref_))) {

LOG_WARN("failed to begin nested session", K(ret), K(stmt_buf));

} else if (OB_FAIL(op.set_foreign_key_check_exist(true))) {

LOG_WARN("failed to set foreign key cascade", K(ret));

} else {

// must call end_nested_session() if begin_nested_session() success.

bool is_zero = false;

if (OB_FAIL(op.execute_read(stmt_buf, res))) {

LOG_WARN("failed to execute stmt", K(ret), K(stmt_buf));

} else {

// must call res.get_result()->close() if execute_read() success.

if (OB_ISNULL(res.get_result())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("result is NULL", K(ret));

} else if (OB_FAIL(res.get_result()->next())) {

if (OB_ITER_END == ret) {

is_zero = true;

ret = OB_SUCCESS;

} else {

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

}

}

if (OB_SUCC(ret)) {

if (is_zero != expect_zero) {

// 如果是自引用外键 新行引用了新行

bool is_self_ref = false;

// oracle for update does not support aggregate functions, so when is_zero is false

// judge whether only one row is affected by whether the second record can be obtained

bool is_affect_only_one = false;

if (!is_zero && OB_FAIL(res.get_result()->next())) {

if (OB_ITER_END == ret) {

is_affect_only_one = true;

ret = OB_SUCCESS;

} else {

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

}

}

/**

* expect_zero is false when fk_arg.ref_action_ is equal to

* ACTION_CHECK_EXIST, if the is_zero != expect_zero condition is

* true, then is_zero is true, need to exclude the case of self reference.

* other cases return OB_ERR_NO_REFERENCED_ROW.

*

* expect_zero is true when fk_arg.ref_action_ is equal to

* ACTION_RESTRICT or ACTION_NO_ACTION, if is_zero != expect_zero condition

* is true, then is_zero is false, need to exclude the case of self reference and

* only affect one row. other cases return OB_ERR_ROW_IS_REFERENCED.

*/

if (!iter_uncommitted_row) {

if (OB_FAIL(is_self_ref_row(op.get_eval_ctx(), row, fk_arg, is_self_ref))) {

LOG_WARN("is_self_ref_row failed", K(ret), K(row), K(fk_arg));

} else if (is_zero && !is_self_ref) {

ret = OB_ERR_NO_REFERENCED_ROW;

LOG_WARN("parent row is not exist", K(ret), K(fk_arg), K(row));

} else if (!is_zero) {

ret = OB_ERR_ROW_IS_REFERENCED;

LOG_WARN("child row is exist", K(ret), K(fk_arg), K(row));

}

}

}

}

int close_ret = res.get_result()->close();

if (OB_SUCCESS != close_ret) {

LOG_WARN("failed to close", K(close_ret));

if (OB_SUCCESS == ret) {

ret = close_ret;

}

}

}

int reset_ret = op.set_foreign_key_check_exist(false);

if (OB_SUCCESS != reset_ret) {

LOG_WARN("failed to reset foreign key cascade", K(reset_ret));

if (OB_SUCCESS == ret) {

ret = reset_ret;

}

}

int end_ret = op.end_nested_session();

if (OB_SUCCESS != end_ret) {

LOG_WARN("failed to end nested session", K(end_ret));

if (OB_SUCCESS == ret) {

ret = end_ret;

}

}

}

}

}

op.get_exec_ctx().get_das_ctx().iter_uncommitted_row_ = false;

return ret;

}

int ForeignKeyHandle::cascade(ObTableModifyOp &op,

const ObForeignKeyArg &fk_arg,

const ObExprPtrIArray &old_row,

const ObExprPtrIArray &new_row)

{

static const char *UPDATE_FMT_MYSQL = "update `%.*s`.`%.*s` set %.*s where %.*s";

static const char *UPDATE_FMT_ORACLE = "update \"%.*s\".\"%.*s\" set %.*s where %.*s";

static const char *DELETE_FMT_MYSQL = "delete from `%.*s`.`%.*s` where %.*s";

static const char *DELETE_FMT_ORACLE = "delete from \"%.*s\".\"%.*s\" where %.*s";

int ret = OB_SUCCESS;

ObArenaAllocator alloc(ObModIds::OB_MODULE_PAGE_ALLOCATOR,

OB_MALLOC_NORMAL_BLOCK_SIZE,

op.get_eval_ctx().exec_ctx_.get_my_session()->get_effective_tenant_id());

char *stmt_buf = nullptr;

int64_t stmt_len = 0;

char *where_buf = nullptr;

int64_t where_len = 0;

int64_t stmt_pos = 0;

int64_t where_pos = 0;

const ObString &database_name = fk_arg.database_name_;

const ObString &table_name = fk_arg.table_name_;

if (OB_FAIL(op.get_exec_ctx().check_status())) {

LOG_WARN("failed check status", K(ret));

} else if (old_row.empty()) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("old row is invalid", K(ret));

} else if (OB_FAIL(gen_where(op.get_eval_ctx(), where_buf, where_len, where_pos,

fk_arg.columns_, old_row, alloc, op.get_obj_print_params()))) {

if (OB_LIKELY(OB_ERR_NULL_VALUE == ret)) {

// skip cascade if any column in foreign key is NULL.

ret = OB_SUCCESS;

stmt_pos = 0;

} else {

LOG_WARN("failed to gen foreign key where", K(ret), K(old_row), K(fk_arg.columns_));

}

} else {

if (!new_row.empty()) {

// update.

const char *update_fmt = lib::is_mysql_mode() ? UPDATE_FMT_MYSQL : UPDATE_FMT_ORACLE;

char *set_buf = nullptr;

int64_t set_len = 0;

int64_t set_pos = 0;

if (OB_FAIL(gen_set(op.get_eval_ctx(), set_buf, set_len, set_pos,

fk_arg.columns_, new_row, alloc, op.get_obj_print_params()))) {

LOG_WARN("failed to gen foreign key set", K(ret), K(new_row), K(fk_arg.columns_));

} else if (OB_FAIL(databuff_printf(stmt_buf, stmt_len, stmt_pos, alloc, update_fmt,

database_name.length(), database_name.ptr(),

table_name.length(), table_name.ptr(),

static_cast<int>(set_pos), set_buf,

static_cast<int>(where_pos), where_buf))) {

LOG_WARN("failed to print stmt", K(ret), K(table_name), K(set_buf), K(where_buf));

} else {

stmt_buf[stmt_pos++] = 0;

}

} else {

// delete.

const char *delete_fmt = lib::is_mysql_mode() ? DELETE_FMT_MYSQL : DELETE_FMT_ORACLE;

if (OB_FAIL(databuff_printf(stmt_buf, stmt_len, stmt_pos, alloc, delete_fmt,

database_name.length(), database_name.ptr(),

table_name.length(), table_name.ptr(),

static_cast<int>(where_pos), where_buf))) {

LOG_WARN("failed to print stmt", K(ret), K(table_name), K(where_buf));

} else {

stmt_buf[stmt_pos++] = 0;

}

}

}

if (OB_SUCC(ret) && stmt_pos > 0) {

LOG_DEBUG("foreign_key_cascade", "stmt", stmt_buf, K(old_row), K(new_row), K(fk_arg));

if (OB_FAIL(op.begin_nested_session(fk_arg.is_self_ref_))) {

LOG_WARN("failed to begin nested session", K(ret));

} else {

// must call end_nested_session() if begin_nested_session() success.

//

// skip modify_ctx.set_foreign_key_cascade when cascade update and self ref.

if (!(fk_arg.is_self_ref_ && !new_row.empty()) &&

OB_FAIL(op.set_foreign_key_cascade(true))) {

LOG_WARN("failed to set foreign key cascade", K(ret));

} else if (OB_FAIL(op.execute_write(stmt_buf))) {

LOG_WARN("failed to execute stmt", K(ret), K(stmt_buf));

}

int reset_ret = op.set_foreign_key_cascade(false);

if (OB_SUCCESS != reset_ret) {

LOG_WARN("failed to reset foreign key cascade", K(reset_ret));

if (OB_SUCCESS == ret) {

ret = reset_ret;

}

}

int end_ret = op.end_nested_session();

if (OB_SUCCESS != end_ret) {

LOG_WARN("failed to end nested session", K(end_ret));

if (OB_SUCCESS == ret) {

ret = end_ret;

}

}

}

}

return ret;

}

int ForeignKeyHandle::set_null(ObTableModifyOp &op,

const ObForeignKeyArg &fk_arg,

const ObExprPtrIArray &old_row)

{

int ret = OB_SUCCESS;

static const char *UPDATE_FMT_MYSQL = "update `%.*s`.`%.*s` set %.*s where %.*s";

static const char *UPDATE_FMT_ORACLE = "update \"%.*s\".\"%.*s\" set %.*s where %.*s";

ObArenaAllocator alloc(ObModIds::OB_MODULE_PAGE_ALLOCATOR,

OB_MALLOC_NORMAL_BLOCK_SIZE,

MTL_ID());

char *stmt_buf = nullptr;

int64_t stmt_len = 0;

char *where_buf = nullptr;

int64_t where_len = 0;

int64_t stmt_pos = 0;

int64_t where_pos = 0;

const ObString &database_name = fk_arg.database_name_;

const ObString &table_name = fk_arg.table_name_;

if (old_row.empty()) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("old row is invalid", K(ret));

} else if (OB_FAIL(gen_where(op.get_eval_ctx(), where_buf, where_len, where_pos,

fk_arg.columns_, old_row, alloc, op.get_obj_print_params()))) {

if (OB_LIKELY(OB_ERR_NULL_VALUE == ret)) {

// skip cascade if any column in foreign key is NULL.

ret = OB_SUCCESS;

stmt_pos = 0;

} else {

LOG_WARN("failed to gen foreign key where", K(ret), K(old_row), K(fk_arg.columns_));

}

} else {

const char *update_fmt = lib::is_mysql_mode() ? UPDATE_FMT_MYSQL : UPDATE_FMT_ORACLE;

char *set_buf = nullptr;

int64_t set_len = 0;

int64_t set_pos = 0;

if (OB_FAIL(gen_column_null_value(op.get_eval_ctx(), set_buf, set_len, set_pos, fk_arg.columns_, alloc, op.get_obj_print_params()))) {

LOG_WARN("failed to gen foreign key set null", K(ret), K(fk_arg.columns_));

} else if (OB_FAIL(databuff_printf(stmt_buf, stmt_len, stmt_pos, alloc, update_fmt,

database_name.length(), database_name.ptr(),

table_name.length(), table_name.ptr(),

static_cast<int>(set_pos), set_buf,

static_cast<int>(where_pos), where_buf))) {

LOG_WARN("failed to print stmt", K(ret), K(table_name), K(set_buf), K(where_buf));

} else {

stmt_buf[stmt_pos++] = 0;

}

}

if (OB_SUCC(ret) && stmt_pos > 0) {

LOG_DEBUG("foreign key cascade set null", "stmt", stmt_buf, K(old_row), K(fk_arg));

if (OB_FAIL(op.begin_nested_session(fk_arg.is_self_ref_))) {

LOG_WARN("failed to begin nested session", K(ret));

} else {

if (OB_FAIL(op.set_foreign_key_cascade(true))) {

LOG_WARN("failed to set foreign key cascade", K(ret));

} else if (OB_FAIL(op.execute_write(stmt_buf))) {

LOG_WARN("failed to execute stmt", K(ret), K(stmt_buf));

}

int reset_ret = op.set_foreign_key_cascade(false);

if (OB_SUCCESS != reset_ret) {

LOG_WARN("failed to reset foreign key cascade", K(reset_ret));

if (OB_SUCCESS == ret) {

ret = reset_ret;

}

}

int end_ret = op.end_nested_session();

if (OB_SUCCESS != end_ret) {

LOG_WARN("failed to end nested session", K(end_ret));

if (OB_SUCCESS == ret) {

ret = end_ret;

}

}

}

}

return ret;

}

int ForeignKeyHandle::gen_set(ObEvalCtx &eval_ctx, char *&buf, int64_t &len, int64_t &pos,

const ObIArray<ObForeignKeyColumn> &columns,

const ObExprPtrIArray &row,

ObIAllocator &alloc,

const ObObjPrintParams &print_params)

{

return gen_column_value(eval_ctx, buf, len, pos, columns, row, ", ", alloc, print_params, false);

}

int ForeignKeyHandle::gen_where(ObEvalCtx &eval_ctx, char *&buf, int64_t &len, int64_t &pos,

const ObIArray<ObForeignKeyColumn> &columns,

const ObExprPtrIArray &row,

ObIAllocator &alloc,

const ObObjPrintParams &print_params)

{

return gen_column_value(eval_ctx, buf, len, pos, columns, row, " and ", alloc, print_params, true);

}

int ForeignKeyHandle::gen_column_value(ObEvalCtx &eval_ctx, char *&buf, int64_t &len, int64_t &pos,

const ObIArray<ObForeignKeyColumn> &columns,

const ObExprPtrIArray &row,

const char *delimiter,

ObIAllocator &alloc,

const ObObjPrintParams &print_params,

bool forbid_null)

{

static const char *COLUMN_FMT_MYSQL = "`%.*s` = ";

static const char *COLUMN_FMT_ORACLE = "\"%.*s\" = ";

const char *column_fmt = lib::is_mysql_mode() ? COLUMN_FMT_MYSQL : COLUMN_FMT_ORACLE;

int ret = OB_SUCCESS;

ObDatum *col_datum = NULL;

if (OB_ISNULL(delimiter) || OB_ISNULL(print_params.tz_info_)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("argument is invalid", K(ret), KP(buf), KP(delimiter), KP(print_params.tz_info_));

} else if (columns.count() <= 0) {

ret= OB_ERR_UNEXPECTED;

LOG_ERROR("columns count of fk is zero or less than zero", K(ret), K(columns.count()), K(columns));

}

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

OZ(row.at(columns.at(i).idx_)->eval(eval_ctx, col_datum));

if (OB_SUCC(ret)) {

const ObString &col_name = columns.at(i).name_;

const ObObjMeta &col_obj_meta = row.at(columns.at(i).idx_)->obj_meta_;

ObObj col_obj;

if (forbid_null && col_datum->is_null()) {

ret = OB_ERR_NULL_VALUE;

// NO LOG.

} else if (OB_FAIL(databuff_printf(buf, len, pos, alloc, column_fmt,

col_name.length(), col_name.ptr()))) {

LOG_WARN("failed to print column name", K(ret), K(col_name));

} else if (OB_FAIL(col_datum->to_obj(col_obj, col_obj_meta))) {

LOG_WARN("to_obj failed", K(ret), K(*col_datum), K(col_obj_meta));

} else if (OB_FAIL(col_obj.print_sql_literal(buf, len, pos, alloc, print_params))) {

LOG_WARN("failed to print column value", K(ret), K(*col_datum), K(col_obj));

} else if (OB_FAIL(databuff_printf(buf, len, pos, alloc, "%s", delimiter))) {

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

}

}

}

if (OB_SUCC(ret)) {

pos -= STRLEN(delimiter);

buf[pos++] = 0;

}

return ret;

}

int ForeignKeyHandle::gen_column_null_value(ObEvalCtx &ctx, char *&buf, int64_t &len, int64_t &pos,

const common::ObIArray<ObForeignKeyColumn> &columns,

common::ObIAllocator &alloc,

const common::ObObjPrintParams &print_params)

{

int ret = OB_SUCCESS;

static const char *COLUMN_FMT_MYSQL = "`%.*s` = ";

static const char *COLUMN_FMT_ORACLE = "\"%.*s\" = ";

const char *column_fmt = lib::is_mysql_mode() ? COLUMN_FMT_MYSQL : COLUMN_FMT_ORACLE;

const char *delimiter = ", ";

if (OB_ISNULL(delimiter) || OB_ISNULL(print_params.tz_info_)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("argument is invalid", K(ret), KP(buf), KP(delimiter), KP(print_params.tz_info_));

} else if (columns.count() <= 0) {

ret= OB_ERR_UNEXPECTED;

LOG_ERROR("columns count of fk is zero or less than zero", K(ret), K(columns.count()), K(columns));

}

ObObj col_obj;

col_obj.set_null();

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

if (OB_SUCC(ret)) {

const ObString &col_name = columns.at(i).name_;

if (OB_FAIL(databuff_printf(buf, len, pos, alloc, column_fmt,

col_name.length(), col_name.ptr()))) {

LOG_WARN("failed to print column name", K(ret), K(col_name));

} else if (OB_FAIL(col_obj.print_sql_literal(buf, len, pos, alloc, print_params))) {

LOG_WARN("failed to print column value", K(ret), K(col_obj));

} else if (OB_FAIL(databuff_printf(buf, len, pos, alloc, "%s", delimiter))) {

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

}

}

}

if (OB_SUCC(ret)) {

pos -= STRLEN(delimiter);

buf[pos++] = 0;

}

return ret;

}

int ForeignKeyHandle::is_self_ref_row(ObEvalCtx &eval_ctx,

const ObExprPtrIArray &row,

const ObForeignKeyArg &fk_arg,

bool &is_self_ref)

{

int ret = OB_SUCCESS;

is_self_ref = fk_arg.is_self_ref_;

ObDatum *name_col = NULL;

ObDatum *val_col = NULL;

for (int64_t i = 0; is_self_ref && i < fk_arg.columns_.count(); i++) {

const int32_t name_idx = fk_arg.columns_.at(i).name_idx_;

const int32_t val_idx = fk_arg.columns_.at(i).idx_;

ObExprCmpFuncType cmp_func = row.at(name_idx)->basic_funcs_->null_first_cmp_;

// TODO qubin.qb: uncomment below block revert the defensive check

//OZ((cmp_func = (row.at(name_idx)->basic_funcs_->null_first_cmp_))

// != row.at(val_idx)->basic_funcs_->null_first_cmp_);

OZ(row.at(name_idx)->eval(eval_ctx, name_col));

OZ(row.at(val_idx)->eval(eval_ctx, val_col));

int cmp_ret = 0;

if (OB_FAIL(cmp_func(*name_col, *val_col, cmp_ret))) {

LOG_WARN("cmp failed", K(ret), K(i));

} else {

is_self_ref = (0 == cmp_ret);

}

}

return ret;

}

// make MY_SPEC macro available.

OB_INLINE static const ObTableModifySpec &get_my_spec(const ObTableModifyOp &op)

{

return op.get_spec();

}

ObTableModifySpec::ObTableModifySpec(common::ObIAllocator &alloc,

const ObPhyOperatorType type)

: ObOpSpec(alloc, type),

expr_frame_info_(NULL),

ab_stmt_id_(nullptr),

flags_(0)

{

}

OB_DEF_SERIALIZE(ObTableModifySpec)

{

int ret = OB_SUCCESS;

BASE_SER((ObTableModifySpec, ObOpSpec));

OB_UNIS_ENCODE(flags_);

OB_UNIS_ENCODE(ab_stmt_id_);

return ret;

}

OB_DEF_DESERIALIZE(ObTableModifySpec)

{

int ret = OB_SUCCESS;

BASE_DESER((ObTableModifySpec, ObOpSpec));

OB_UNIS_DECODE(flags_);

OB_UNIS_DECODE(ab_stmt_id_);

return ret;

}

OB_DEF_SERIALIZE_SIZE(ObTableModifySpec)

{

int64_t len = 0;

BASE_ADD_LEN((ObTableModifySpec, ObOpSpec));

OB_UNIS_ADD_LEN(flags_);

OB_UNIS_ADD_LEN(ab_stmt_id_);

return len;

}

OB_SERIALIZE_MEMBER(ObTableModifyOpInput);

ObTableModifyOp::ObTableModifyOp(ObExecContext &ctx,

const ObOpSpec &spec,

ObOpInput *input)

: ObOperator(ctx, spec, input),

sql_proxy_(NULL),

inner_conn_(NULL),

tenant_id_(0),

saved_conn_(),

foreign_key_checks_(false),

need_close_conn_(false),

iter_end_(false),

dml_rtctx_(eval_ctx_, ctx, *this),

is_error_logging_(false),

execute_single_row_(false),

err_log_rt_def_(),

dml_modify_rows_(ctx.get_allocator()),

saved_session_(NULL)

{

obj_print_params_ = CREATE_OBJ_PRINT_PARAM(ctx_.get_my_session());

obj_print_params_.need_cast_expr_ = true;

// bugfix:

// in NO_BACKSLASH_ESCAPES, obj_print_sql<ObVarcharType> won't escape.

// We use skip_escape_ to indicate this case. It will finally be passed to ObHexEscapeSqlStr.

GET_SQL_MODE_BIT(IS_NO_BACKSLASH_ESCAPES, ctx_.get_my_session()->get_sql_mode(), obj_print_params_.skip_escape_);

}

bool ObTableModifyOp::is_fk_root_session() {

bool ret = false;

if (OB_ISNULL(ctx_.get_parent_ctx())) {

if (this->need_foreign_key_checks()) {

ret = true;

}

} else {

if (!ctx_.get_parent_ctx()->get_das_ctx().is_fk_cascading_ && need_foreign_key_checks()) {

ret = true;

}

}

return ret;

}

int ObTableModifyOp::inner_open()

{

int ret = OB_SUCCESS;

ObSQLSessionInfo *my_session = GET_MY_SESSION(ctx_);

if (OB_FAIL(init_foreign_key_operation())) {

LOG_WARN("failed to init foreign key operation", K(ret));

} else if (MY_SPEC.plan_->has_nested_sql() && OB_FAIL(open_inner_conn())) {

LOG_WARN("failed to open inner conn", K(ret));

} else if (OB_FAIL(calc_single_table_loc())) {

LOG_WARN("calc single table loc failed", K(ret));

} else if (OB_FAIL(check_need_exec_single_row())) {

LOG_WARN("failed to perform single row execution check", K(ret));

} else {

init_das_dml_ctx();

}

return ret;

}

void ObTableModifyOp::clear_dml_evaluated_flag(ObExpr *clear_expr)

{

if (clear_expr->is_batch_result()) {

clear_expr->get_evaluated_flags(eval_ctx_).unset(eval_ctx_.get_batch_idx());

} else {

clear_expr->get_eval_info(eval_ctx_).clear_evaluated_flag();

}

if (0 < clear_expr->parent_cnt_) {

clear_dml_evaluated_flag(clear_expr->parent_cnt_, clear_expr->parents_);

}

}

void ObTableModifyOp::clear_dml_evaluated_flag(int64_t parent_cnt, ObExpr **parent_exprs)

{

for (int64_t i = 0; i < parent_cnt; ++i) {

clear_dml_evaluated_flag(parent_exprs[i]);

}

}

void ObTableModifyOp::clear_dml_evaluated_flag()

{

// Clear the parent expr because the assignment expr may be changed

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

ObExpr *expr = trigger_clear_exprs_.at(i);

clear_dml_evaluated_flag(expr);

}

}

OB_INLINE int ObTableModifyOp::init_das_dml_ctx()

{

ObSQLSessionInfo *session = GET_MY_SESSION(ctx_);

const char *label = nullptr;

switch(MY_SPEC.type_) {

case PHY_INSERT:

label = "InsDASCtx";

break;

case PHY_UPDATE:

label = "UpdDASCtx";

break;

case PHY_DELETE:

label = "DelDASCtx";

break;

case PHY_LOCK:

label = "LockDASCtx";

break;

default:

label = "DmlDASCtx";

break;

}

ObMemAttr memattr(session->get_effective_tenant_id(), label, ObCtxIds::EXECUTE_CTX_ID);

dml_rtctx_.das_ref_.set_expr_frame_info(nullptr != MY_SPEC.expr_frame_info_

? MY_SPEC.expr_frame_info_

: &MY_SPEC.plan_->get_expr_frame_info());

dml_rtctx_.das_ref_.set_mem_attr(memattr);

dml_rtctx_.das_ref_.set_execute_directly(!MY_SPEC.use_dist_das_);

return OB_SUCCESS;

}

int ObTableModifyOp::merge_implict_cursor(int64_t insert_rows,

int64_t update_rows,

int64_t delete_rows,

int64_t found_rows)

{

int ret = OB_SUCCESS;

bool is_ins_val_opt = ctx_.get_sql_ctx()->is_do_insert_batch_opt();

if (MY_SPEC.ab_stmt_id_ != nullptr && !is_ins_val_opt) {

ObDatum *stmt_id_datum = nullptr;

if (OB_FAIL(MY_SPEC.ab_stmt_id_->eval(eval_ctx_, stmt_id_datum))) {

LOG_WARN("eval ab stmt id failed", K(ret));

} else {

ObPhysicalPlanCtx *plan_ctx = GET_PHY_PLAN_CTX(ctx_);

int64_t stmt_id = stmt_id_datum->get_int();

ObImplicitCursorInfo implicit_cursor;

implicit_cursor.stmt_id_ = stmt_id;

implicit_cursor.found_rows_ += found_rows;

implicit_cursor.matched_rows_ += found_rows;

if (insert_rows > 0) {

implicit_cursor.affected_rows_ += insert_rows;

}

if (update_rows > 0) {

ObSQLSessionInfo *session = GET_MY_SESSION(ctx_);

bool client_found_rows = session->get_capability().cap_flags_.OB_CLIENT_FOUND_ROWS;

implicit_cursor.duplicated_rows_ += update_rows;

implicit_cursor.affected_rows_ += client_found_rows ? found_rows : update_rows;

}

if (delete_rows > 0) {

implicit_cursor.affected_rows_ += delete_rows;

}

if (OB_FAIL(plan_ctx->merge_implicit_cursor_info(implicit_cursor))) {

LOG_WARN("merge implicit cursor info to plan ctx failed", K(ret), K(implicit_cursor));

}

LOG_DEBUG("merge implicit cursor", K(ret), K(implicit_cursor));

}

}

return ret;

}

int ObTableModifyOp::inner_switch_iterator()

{

int ret = OB_SUCCESS;

if (OB_FAIL(ObOperator::inner_switch_iterator())) {

LOG_WARN("switch iterator failed", K(ret));

}

return ret;

}

int ObTableModifyOp::inner_close()

{

int ret = OB_SUCCESS;

ObPhysicalPlanCtx *plan_ctx = ctx_.get_physical_plan_ctx();

// release cache_handle for auto-increment

share::ObAutoincrementService &auto_service = share::ObAutoincrementService::get_instance();

ObIArray<share::AutoincParam> &autoinc_params = plan_ctx->get_autoinc_params();

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

if (NULL != autoinc_params.at(i).cache_handle_) {

auto_service.release_handle(autoinc_params.at(i).cache_handle_);

}

}

close_inner_conn();

if (dml_rtctx_.das_ref_.has_task()) {

if (OB_FAIL(dml_rtctx_.das_ref_.close_all_task())) {

LOG_WARN("close all insert das task failed", K(ret));

} else {

dml_rtctx_.das_ref_.reset();

}

}

dml_modify_rows_.clear();

// Release the hash sets created at fk root ctx for delete distinct checks

if (OB_SUCC(ret) && get_exec_ctx().is_fk_root_ctx()) {

DASDelCtxList& del_ctx_list = get_exec_ctx().get_das_ctx().get_das_del_ctx_list();

DASDelCtxList::iterator iter = del_ctx_list.begin();

for (; OB_SUCC(ret)&& iter != del_ctx_list.end(); iter++) {

DmlRowkeyDistCtx del_ctx = *iter;

if (del_ctx.deleted_rows_ != nullptr) {

del_ctx.deleted_rows_->destroy();

del_ctx.deleted_rows_ = nullptr;

}

}

del_ctx_list.destroy();

}

return ret;

}

OB_INLINE int ObTableModifyOp::init_foreign_key_operation()

{

int ret = OB_SUCCESS;

ObPhysicalPlanCtx *phy_plan_ctx = GET_PHY_PLAN_CTX(ctx_);

if (OB_ISNULL(phy_plan_ctx)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("modify_ctx or phy_plan_ctx is NULL", K(ret), KP(phy_plan_ctx));

} else if ((lib::is_mysql_mode() && phy_plan_ctx->need_foreign_key_checks())

|| lib::is_oracle_mode()) {

// set fk check even if ret != OB_SUCCESS. see ObTableModifyOp::inner_close()

set_foreign_key_checks();

}

return ret;

}

int ObTableModifyOp::inner_rescan()

{

int ret = OB_SUCCESS;

if (OB_FAIL(ObOperator::inner_rescan())) {

LOG_WARN("rescan child operator failed", K(ret));

} else {

iter_end_ = false;

if (dml_rtctx_.das_ref_.has_task()) {

if (OB_FAIL(dml_rtctx_.das_ref_.close_all_task())) {

LOG_WARN("close all insert das task failed", K(ret));

} else {

dml_rtctx_.reuse();

}

}

}

if (OB_SUCC(ret)) {

dml_modify_rows_.clear();

if (OB_FAIL(calc_single_table_loc())) {

LOG_WARN("calc single table loc failed", K(ret));

}

}

return ret;

}

int ObTableModifyOp::check_need_exec_single_row() {

int ret = OB_SUCCESS;

if (MY_SPEC.is_returning_ && need_foreign_key_checks()) {

execute_single_row_ = true;

}

return ret;

}

int ObTableModifyOp::get_gi_task()

{

int ret = OB_SUCCESS;

ObGranuleTaskInfo gi_task_info;

GIPrepareTaskMap *gi_prepare_map = nullptr;

ObDASCtx &das_ctx = ctx_.get_das_ctx();

ObTableID table_loc_id = OB_INVALID_ID;

ObTableID ref_table_id = OB_INVALID_ID;

ObTabletID tablet_id;

if (OB_FAIL(MY_SPEC.get_single_table_loc_id(table_loc_id, ref_table_id))) {

LOG_WARN("get single table loc id failed", K(ret));

} else if (OB_FAIL(ctx_.get_gi_task_map(gi_prepare_map))) {

LOG_WARN("Failed to get gi task map", K(ret));

} else if (OB_FAIL(gi_prepare_map->get_refactored(MY_SPEC.id_, gi_task_info))) {

if (ret != OB_HASH_NOT_EXIST) {

LOG_WARN("failed to get prepare gi task", K(ret), K(MY_SPEC.get_id()));

} else {

LOG_DEBUG("no prepared task info, set table modify to end",

K(MY_SPEC.get_id()), K(this), K(lbt()));

// 当前DML算子无法从 GI中获得 task，表示当前DML算子iter end

iter_end_ = true;

ret = OB_SUCCESS;

}

} else if (OB_ISNULL(get_input()->table_loc_) &&

OB_ISNULL(get_input()->table_loc_ = das_ctx.get_table_loc_by_id(table_loc_id, ref_table_id))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("get table location by table id failed", K(ret), K(table_loc_id), K(ref_table_id));

} else {

get_input()->tablet_loc_ = gi_task_info.tablet_loc_;

LOG_DEBUG("DML operator consume a task", K(ret),

K(gi_task_info), KPC(get_input()), K(MY_SPEC.id_), K(get_input()));

}

return ret;

}

int ObTableModifyOp::calc_single_table_loc()

{

int ret = OB_SUCCESS;

if (!MY_SPEC.use_dist_das_) {

ObTableID table_loc_id = OB_INVALID_ID;

ObTableID ref_table_id = OB_INVALID_ID;

ObDASCtx &das_ctx = ctx_.get_das_ctx();

ObDASTableLoc *table_loc = nullptr;

if (OB_UNLIKELY(MY_SPEC.gi_above_)) {

if (OB_FAIL(get_gi_task())) {

LOG_WARN("get gi task failed", K(ret));

}

} else if (OB_FAIL(MY_SPEC.get_single_table_loc_id(table_loc_id, ref_table_id))) {

LOG_WARN("get single table loc id failed", K(ret));

} else if (OB_ISNULL(table_loc = das_ctx.get_table_loc_by_id(table_loc_id, ref_table_id))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("get table location by table id failed", K(ret),

K(table_loc_id), K(ref_table_id), K(das_ctx.get_table_loc_list()));

} else {

get_input()->table_loc_ = table_loc;

get_input()->tablet_loc_ = table_loc->get_first_tablet_loc();

}