Skip to content

O
oceanbase

clmforever / oceanbase
与 Fork 源项目一致

Fork自 oceanbase / oceanbase

通知 1
Star 0
Fork 0
O
oceanbase

体验新版 GitCode，发现更多精彩内容 >>

提交 0b9d1f8c 编写于 作者: O obdev 提交者： wangzelin.wzl
浏览文件
操作

Fix: fix duplicate delete same row induced by fk cascade delete

上级 2542756c
隐藏空白更改
内联 并排
Showing with 316 addition and 51 deletion
src/sql/code_generator/ob_dml_cg_service.cpp
浏览文件 @ 0b9d1f8c
......@@ -2369,6 +2369,9 @@ int ObDmlCgService::generate_fk_arg(ObForeignKeyArg &fk_arg,
} else if (OB_FAIL(fk_arg.columns_.reserve(name_column_ids.count()))) {
LOG_WARN("failed to reserve foreign key columns", K(name_column_ids.count()), K(ret));
}
if ( OB_SUCC(ret) && need_handle) {
fk_arg.table_id_ = name_table_id;
}
for (int64_t i = 0; OB_SUCC(ret) && need_handle && i < name_column_ids.count(); i++) {
ObForeignKeyColumn fk_column;
if (OB_ISNULL(column_schema = (table_schema->get_column_schema(name_column_ids.at(i))))) {
......
src/sql/code_generator/ob_static_engine_cg.cpp
浏览文件 @ 0b9d1f8c
......@@ -2036,6 +2036,23 @@ int ObStaticEngineCG::generate_delete_with_das(ObLogDelete &op, ObTableDeleteSpe
}
} // for index_dml_infos end
} //for table_columns end
for (int64_t i = 0; OB_SUCC(ret) && i < delete_table_list.count(); ++i) {
ObTableDeleteSpec::DelCtDefArray &ctdefs = spec.del_ctdefs_.at(i);
ObDelCtDef &del_ctdef = *ctdefs.at(0);
const uint64_t del_table_id = del_ctdef.das_base_ctdef_.index_tid_;
bool is_dup = false;
for (int j = 0; !is_dup && OB_SUCC(ret) && j < delete_table_list.count(); ++j) {
const uint64_t root_table_id = spec.del_ctdefs_.at(j).at(0)->das_base_ctdef_.index_tid_;
DASTableIdList parent_tables(phy_plan_->get_allocator());
if(OB_FAIL(check_fk_nested_dup_del(del_table_id, root_table_id, parent_tables, is_dup))) {
LOG_WARN("failed to perform nested duplicate table check", K(ret), K(del_table_id), K(root_table_id));
}
}
if (OB_SUCC(ret) && is_dup) {
del_ctdef.distinct_algo_ = T_HASH_DISTINCT;
}
}
return ret;
}
......@@ -2091,6 +2108,15 @@ int ObStaticEngineCG::generate_spec(ObLogInsert &op, ObTableReplaceSpec &spec, c
LOG_WARN("generate conflict_checker failed", K(ret));
} else if (OB_FAIL(mark_expr_self_produced(index_dml_info->column_exprs_))) {
LOG_WARN("mark self expr failed", K(ret));
} else {
bool is_dup = false;
const uint64_t replace_table_id = replace_ctdef->del_ctdef_->das_base_ctdef_.index_tid_;
DASTableIdList parent_tables(phy_plan_->get_allocator());
if(OB_FAIL(check_fk_nested_dup_del(replace_table_id, replace_table_id, parent_tables, is_dup))) {
LOG_WARN("failed to perform nested duplicate table check", K(ret), K(replace_table_id));
} else if (is_dup) {
replace_ctdef->del_ctdef_->distinct_algo_ = T_HASH_DISTINCT;
}
}
}
spec.replace_ctdefs_.at(i) = replace_ctdef;
......@@ -6702,5 +6728,58 @@ int ObStaticEngineCG::check_only_one_unique_key(const ObLogPlan& log_plan,
return ret;
}
bool ObStaticEngineCG::has_cycle_reference(DASTableIdList &parent_tables, const uint64_t table_id)
{
bool ret = false;
if (!parent_tables.empty()) {
DASTableIdList::iterator iter = parent_tables.begin();
for (; !ret && iter != parent_tables.end(); iter++) {
if (*iter == table_id) {
ret = true;
}
}
}
return ret;
}
int ObStaticEngineCG::check_fk_nested_dup_del(const uint64_t table_id,
const uint64_t root_table_id,
DASTableIdList &parent_tables,
bool &is_dup)
{
int ret = OB_SUCCESS;
ObSchemaGetterGuard schema_guard;
const ObTableSchema *table_schema = NULL;
const uint64_t tenant_id = MTL_ID();
if (OB_FAIL(parent_tables.push_back(root_table_id))) {
LOG_WARN("failed to push root_table_id to parent tables list", K(ret), K(root_table_id), K(parent_tables.size()));
} else if (OB_FAIL(GCTX.schema_service_->get_tenant_schema_guard(tenant_id, schema_guard))) {
LOG_WARN("get tenant schema guard failed", K(ret));
} else if (OB_FAIL(schema_guard.get_table_schema(tenant_id, root_table_id, table_schema))) {
LOG_WARN("get table schema failed", K(ret), K(tenant_id), K(root_table_id));
} else if (!OB_ISNULL(table_schema)) {
const common::ObIArray<ObForeignKeyInfo> &foreign_key_infos = table_schema->get_foreign_key_infos();
for (int64_t i = 0; OB_SUCC(ret) && i < foreign_key_infos.count() && !is_dup; ++i) {
const ObForeignKeyInfo &fk_info = foreign_key_infos.at(i);
const uint64_t child_table_id = fk_info.child_table_id_;
const uint64_t parent_table_id = fk_info.parent_table_id_;
ObReferenceAction del_act = fk_info.delete_action_;
if (child_table_id != common::OB_INVALID_ID && del_act == ACTION_CASCADE) {
if (child_table_id == table_id) {
is_dup = true;
} else if (has_cycle_reference(parent_tables, child_table_id)) {
LOG_DEBUG("This schema has a circular foreign key dependencies");
} else if (OB_FAIL(SMART_CALL(check_fk_nested_dup_del(table_id, child_table_id, parent_tables, is_dup)))) {
LOG_WARN("failed deep search nested duplicate delete table", K(ret), K(table_id), K(root_table_id), K(child_table_id));
}
}
}
}
if (OB_SUCC(ret) && OB_FAIL(parent_tables.pop_back())) {
LOG_WARN("failed to pop latest table id", K(ret));
}
return ret;
}
} // end namespace sql
} // end namespace oceanbase
src/sql/code_generator/ob_static_engine_cg.h
浏览文件 @ 0b9d1f8c
......@@ -124,6 +124,7 @@ class ObDuplicatedKeyChecker;
struct ObTableScanCtDef;
struct ObDASScanCtDef;
struct InsertAllTableInfo;
typedef common::ObList<uint64_t, common::ObIAllocator> DASTableIdList;
typedef common::ObSEArray<common::ObSEArray<int64_t, 8, common::ModulePageAllocator, true>,
1, common::ModulePageAllocator, true> RowParamMap;
//
......@@ -461,6 +462,11 @@ private:
|| T_FUN_SUM == expr_type
|| T_FUN_MAX == expr_type
|| T_FUN_MIN == expr_type; }
int check_fk_nested_dup_del(const uint64_t table_id,
const uint64_t root_table_id,
DASTableIdList &parent_tables,
bool &is_dup);
bool has_cycle_reference(DASTableIdList &parent_tables, const uint64_t table_id);
private:
ObPhysicalPlan *phy_plan_;
ObOptimizerContext *opt_ctx_;
......
src/sql/das/ob_das_context.cpp
浏览文件 @ 0b9d1f8c
......@@ -24,6 +24,8 @@ using namespace common;
using namespace share;
namespace sql
{
int ObDASCtx::init(const ObPhysicalPlan &plan, ObExecContext &ctx)
{
int ret = OB_SUCCESS;
......
src/sql/das/ob_das_context.h
浏览文件 @ 0b9d1f8c
......@@ -17,11 +17,25 @@
#include "share/schema/ob_schema_getter_guard.h"
#include "sql/das/ob_das_factory.h"
#include "storage/tx/ob_trans_define.h"
#include "sql/engine/dml/ob_dml_ctx_define.h"
namespace oceanbase
{
namespace sql
{
class ObDASTabletMapper;
struct DmlRowkeyDistCtx
{
public:
DmlRowkeyDistCtx()
: deleted_rows_(nullptr),
table_id_(common::OB_INVALID_ID)
{}
SeRowkeyDistCtx *deleted_rows_;
uint64_t table_id_;
};
typedef common::ObList<DmlRowkeyDistCtx, common::ObIAllocator> DASDelCtxList;
class ObDASCtx
{
OB_UNIS_VERSION(1);
......@@ -36,6 +50,7 @@ public:
self_schema_guard_(false),
snapshot_(),
savepoint_(0),
del_ctx_list_(allocator),
flags_(0)
{
need_check_server_ = 1;
......@@ -52,6 +67,7 @@ public:
int init(const ObPhysicalPlan &plan, ObExecContext &ctx);
ObDASTableLoc *get_table_loc_by_id(uint64_t table_loc_id, uint64_t ref_table_id);
DASTableLocList &get_table_loc_list() { return table_locs_; }
DASDelCtxList& get_das_del_ctx_list() {return del_ctx_list_;}
int extended_tablet_loc(ObDASTableLoc &table_loc,
const common::ObTabletID &tablet_id,
ObDASTabletLoc *&tablet_loc);
......@@ -102,6 +118,7 @@ private:
transaction::ObTxReadSnapshot snapshot_; // Mvcc snapshot
int64_t savepoint_; // DML savepoint
//@todo: save snapshot version
DASDelCtxList del_ctx_list_;
public:
union {
uint64_t flags_;
......
src/sql/das/ob_das_define.h
浏览文件 @ 0b9d1f8c
......@@ -300,6 +300,7 @@ private:
int64_t lookup_cnt_;
};
typedef common::ObList<ObDASTableLoc*, common::ObIAllocator> DASTableLocList;
typedef common::ObList<uint64_t, common::ObIAllocator> DASTableIdList;
typedef common::ObFixedArray<uint64_t, common::ObIAllocator> UIntFixedArray;
typedef common::ObFixedArray<int64_t, common::ObIAllocator> IntFixedArray;
typedef common::ObFixedArray<ObObjectID, common::ObIAllocator> ObjectIDFixedArray;
......
src/sql/engine/dml/ob_dml_ctx_define.h
浏览文件 @ 0b9d1f8c
......@@ -282,6 +282,7 @@ public:
: ref_action_(share::schema::ACTION_INVALID),
database_name_(),
table_name_(),
table_id_(0),
columns_(),
is_self_ref_(false)
{}
......@@ -290,6 +291,7 @@ public:
: ref_action_(share::schema::ACTION_INVALID),
database_name_(),
table_name_(),
table_id_(0),
columns_(alloc),
is_self_ref_(false)
{}
......@@ -298,6 +300,7 @@ public:
ref_action_ = share::schema::ACTION_INVALID;
database_name_.reset();
table_name_.reset();
table_id_ = 0;
columns_.reset();
}
TO_STRING_KV(K_(ref_action), K_(database_name), K_(table_name), K_(columns), K_(is_self_ref));
......@@ -305,6 +308,7 @@ public:
share::schema::ObReferenceAction ref_action_;
common::ObString database_name_;
common::ObString table_name_;
uint64_t table_id_;
common::ObFixedArray<ObForeignKeyColumn, common::ObIAllocator> columns_;
bool is_self_ref_;
};
......@@ -746,7 +750,7 @@ public:
virtual ~ObDelRtDef()
{
if (se_rowkey_dist_ctx_ != nullptr) {
se_rowkey_dist_ctx_->destroy();
// se_rowkey_dist_ctx_->destroy();
se_rowkey_dist_ctx_ = nullptr;
}
}
......@@ -757,7 +761,6 @@ public:
DASDelRtDefArray related_rtdefs_;
SeRowkeyDistCtx *se_rowkey_dist_ctx_;
};
struct ObMergeCtDef
{
OB_UNIS_VERSION(1);
......
src/sql/engine/dml/ob_dml_service.cpp
浏览文件 @ 0b9d1f8c
......@@ -120,43 +120,19 @@ int ObDMLService::check_rowkey_whether_distinct(const ObExprPtrIArray &row,
int64_t estimate_row,
DistinctType distinct_algo,
ObEvalCtx &eval_ctx,
SeRowkeyDistCtx *&rowkey_dist_ctx,
ObExecContext &root_ctx,
SeRowkeyDistCtx *rowkey_dist_ctx,
bool &is_dist)
{
int ret = OB_SUCCESS;
is_dist = true;
if (T_DISTINCT_NONE != distinct_algo) {
if (T_HASH_DISTINCT == distinct_algo) {
ObIAllocator &allocator = eval_ctx.exec_ctx_.get_allocator();
ObIAllocator &allocator = root_ctx.get_allocator();
if (OB_ISNULL(rowkey_dist_ctx)) {
//create rowkey distinct context
void *buf = allocator.alloc(sizeof(SeRowkeyDistCtx));
ObSQLSessionInfo *my_session = eval_ctx.exec_ctx_.get_my_session();
if (OB_ISNULL(buf)) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("allocate memory failed", K(ret), "size", sizeof(SeRowkeyDistCtx));
} else if (OB_ISNULL(my_session)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("my session is null", K(ret));
} else {
rowkey_dist_ctx = new (buf) SeRowkeyDistCtx();
int64_t match_rows = estimate_row > ObDMLBaseCtDef::MIN_ROWKEY_DISTINCT_BUCKET_NUM ?
estimate_row : ObDMLBaseCtDef::MIN_ROWKEY_DISTINCT_BUCKET_NUM;
// https://work.aone.alibaba-inc.com/issue/23348769
// match_rows是优化器估行的结果，如果这个值很大，
// 直接创建有这么多bucket的hashmap会申请
// 不到内存，这里做了限制为64k，防止报内存不足的错误
const int64_t max_bucket_num = match_rows > ObDMLBaseCtDef::MAX_ROWKEY_DISTINCT_BUCKET_NUM ?
ObDMLBaseCtDef::MAX_ROWKEY_DISTINCT_BUCKET_NUM : match_rows;
if (OB_FAIL(rowkey_dist_ctx->create(max_bucket_num,
ObModIds::OB_DML_CHECK_ROWKEY_DISTINCT_BUCKET,
ObModIds::OB_DML_CHECK_ROWKEY_DISTINCT_NODE,
my_session->get_effective_tenant_id()))) {
LOG_WARN("create rowkey distinct context failed", K(ret), "rows", estimate_row);
}
}
}
if (OB_SUCC(ret)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("distinct check hash set is null", K(ret));
} else {
SeRowkeyItem rowkey_item;
if (OB_ISNULL(rowkey_dist_ctx)) {
ret = OB_ERR_UNEXPECTED;
......@@ -188,6 +164,46 @@ int ObDMLService::check_rowkey_whether_distinct(const ObExprPtrIArray &row,
return ret;
}
int ObDMLService::create_rowkey_check_hashset(int64_t estimate_row,
ObExecContext *root_ctx,
SeRowkeyDistCtx *&rowkey_dist_ctx)
{
int ret = OB_SUCCESS;
ObIAllocator &allocator = root_ctx->get_allocator();
if (OB_ISNULL(rowkey_dist_ctx)) {
//create rowkey distinct context
void *buf = allocator.alloc(sizeof(SeRowkeyDistCtx));
ObSQLSessionInfo *my_session = root_ctx->get_my_session();
if (OB_ISNULL(buf)) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("allocate memory failed", K(ret), "size", sizeof(SeRowkeyDistCtx));
} else if (OB_ISNULL(my_session)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("my session is null", K(ret));
} else {
rowkey_dist_ctx = new (buf) SeRowkeyDistCtx();
int64_t match_rows = estimate_row > ObDMLBaseCtDef::MIN_ROWKEY_DISTINCT_BUCKET_NUM ?
estimate_row : ObDMLBaseCtDef::MIN_ROWKEY_DISTINCT_BUCKET_NUM;
// https://work.aone.alibaba-inc.com/issue/23348769
// match_rows是优化器估行的结果，如果这个值很大，
// 直接创建有这么多bucket的hashmap会申请
// 不到内存，这里做了限制为64k，防止报内存不足的错误
const int64_t max_bucket_num = match_rows > ObDMLBaseCtDef::MAX_ROWKEY_DISTINCT_BUCKET_NUM ?
ObDMLBaseCtDef::MAX_ROWKEY_DISTINCT_BUCKET_NUM : match_rows;
if (OB_FAIL(rowkey_dist_ctx->create(max_bucket_num,
ObModIds::OB_DML_CHECK_ROWKEY_DISTINCT_BUCKET,
ObModIds::OB_DML_CHECK_ROWKEY_DISTINCT_NODE,
my_session->get_effective_tenant_id()))) {
LOG_WARN("create rowkey distinct context failed", K(ret), "rows", estimate_row);
}
}
} else {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("Create hash set on a pointer that is not null", K(ret));
}
return ret;
}
int ObDMLService::check_row_whether_changed(const ObUpdCtDef &upd_ctdef,
ObUpdRtDef &upd_rtdef,
ObEvalCtx &eval_ctx)
......@@ -542,21 +558,30 @@ int ObDMLService::process_delete_row(const ObDelCtDef &del_ctdef,
is_skipped = true;
}
}
if (OB_SUCC(ret) && !is_skipped && !has_instead_of_trg) {
if (OB_SUCC(ret) && !is_skipped && !OB_ISNULL(del_rtdef.se_rowkey_dist_ctx_) && !has_instead_of_trg) {
bool is_distinct = false;
if (OB_FAIL(check_rowkey_whether_distinct(del_ctdef.distinct_key_,
del_ctdef.distinct_key_.count(),
dml_op.get_spec().rows_,
del_ctdef.distinct_algo_,
dml_op.get_eval_ctx(),
del_rtdef.se_rowkey_dist_ctx_,
is_distinct))) {
ObExecContext *root_ctx = nullptr;
if (OB_FAIL(dml_op.get_exec_ctx().get_root_ctx(root_ctx))) {
LOG_WARN("get root ExecContext failed", K(ret));
} else if (OB_ISNULL(root_ctx)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("the root ctx of foreign key nested session is null", K(ret));
} else if (OB_FAIL(check_rowkey_whether_distinct(del_ctdef.distinct_key_,
del_ctdef.distinct_key_.count(),
dml_op.get_spec().rows_,
T_HASH_DISTINCT,
dml_op.get_eval_ctx(),
*root_ctx,
del_rtdef.se_rowkey_dist_ctx_,
is_distinct))) {
LOG_WARN("check rowkey whether distinct failed", K(ret),
K(del_ctdef), K(del_rtdef), K(dml_op.get_spec().rows_));
K(del_ctdef), K(del_rtdef), K(dml_op.get_spec().rows_));
} else if (!is_distinct) {
is_skipped = true;
}
}
if (OB_SUCC(ret) && !is_skipped) {
if (!has_instead_of_trg && OB_FAIL(ForeignKeyHandle::do_handle(dml_op, del_ctdef, del_rtdef))) {
LOG_WARN("do handle old row for delete op failed", K(ret), K(del_ctdef), K(del_rtdef));
......@@ -619,6 +644,7 @@ int ObDMLService::process_update_row(const ObUpdCtDef &upd_ctdef,
dml_op.get_spec().rows_,
upd_ctdef.distinct_algo_,
dml_op.get_eval_ctx(),
dml_op.get_exec_ctx(),
upd_rtdef.se_rowkey_dist_ctx_,
is_distinct))) {
LOG_WARN("check rowkey whether distinct failed", K(ret),
......@@ -1058,6 +1084,39 @@ int ObDMLService::init_del_rtdef(ObDMLRtCtx &dml_rtctx,
del_rtdef.das_rtdef_.related_ctdefs_ = &del_ctdef.related_ctdefs_;
del_rtdef.das_rtdef_.related_rtdefs_ = &del_rtdef.related_rtdefs_;
}
if (OB_SUCC(ret)) {
ObTableModifyOp &dml_op = dml_rtctx.op_;
const uint64_t del_table_id = del_ctdef.das_base_ctdef_.index_tid_;
ObExecContext *root_ctx = nullptr;
if (OB_FAIL(dml_op.get_exec_ctx().get_root_ctx(root_ctx))) {
LOG_WARN("failed to get root exec ctx", K(ret));
} else if (OB_ISNULL(root_ctx)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("the root exec ctx is nullptr", K(ret));
} else {
DASDelCtxList& del_ctx_list = root_ctx->get_das_ctx().get_das_del_ctx_list();
if (!ObDMLService::is_nested_dup_table(del_table_id, del_ctx_list) && T_DISTINCT_NONE != del_ctdef.distinct_algo_) {
DmlRowkeyDistCtx del_ctx;
del_ctx.table_id_ = del_table_id;
if (OB_FAIL(ObDMLService::create_rowkey_check_hashset(dml_op.get_spec().rows_, root_ctx, del_ctx.deleted_rows_))) {
LOG_WARN("Failed to create hash set", K(ret));
} else if (OB_FAIL(del_ctx_list.push_back(del_ctx))) {
LOG_WARN("failed to push del ctx to list", K(ret));
} else {
del_rtdef.se_rowkey_dist_ctx_ = del_ctx.deleted_rows_;
}
} else if (T_DISTINCT_NONE != del_ctdef.distinct_algo_ &&
OB_FAIL(ObDMLService::get_nested_dup_table_ctx(del_table_id, del_ctx_list, del_rtdef.se_rowkey_dist_ctx_))) {
LOG_WARN("failed to get nested duplicate delete table ctx for fk nested session", K(ret));
} else if (dml_op.is_fk_nested_session() && OB_FAIL(ObDMLService::get_nested_dup_table_ctx(del_table_id,
del_ctx_list,
del_rtdef.se_rowkey_dist_ctx_))) {
LOG_WARN("failed to get nested duplicate delete table ctx for fk nested session", K(ret));
}
}
}
return ret;
}
......@@ -1114,6 +1173,15 @@ int ObDMLService::init_upd_rtdef(
upd_rtdef.dupd_rtdef_.related_rtdefs_ = &upd_rtdef.related_upd_rtdefs_;
dml_rtctx.get_exec_ctx().set_update_columns(&upd_ctdef.assign_columns_);
}
if (OB_SUCC(ret) && T_DISTINCT_NONE != upd_ctdef.distinct_algo_) {
ObTableModifyOp &dml_op = dml_rtctx.op_;
if (OB_FAIL(create_rowkey_check_hashset(dml_op.get_spec().rows_,
&dml_op.get_exec_ctx(),
upd_rtdef.se_rowkey_dist_ctx_))) {
LOG_WARN("failed to create distinct check hash set", K(ret));
}
}
return ret;
}
......@@ -1657,5 +1725,33 @@ int ObDMLService::convert_exprs_to_row(const ExprFixedArray &exprs,
}
return ret;
}
bool ObDMLService::is_nested_dup_table(const uint64_t table_id, DASDelCtxList& del_ctx_list)
{
bool ret = false;
DASDelCtxList::iterator iter = del_ctx_list.begin();
for (; !ret && iter != del_ctx_list.end(); iter++) {
DmlRowkeyDistCtx del_ctx = *iter;
if (del_ctx.table_id_ == table_id) {
ret = true;
}
}
return ret;
}
int ObDMLService::get_nested_dup_table_ctx(const uint64_t table_id, DASDelCtxList& del_ctx_list, SeRowkeyDistCtx* &rowkey_dist_ctx)
{
int ret = OB_SUCCESS;
bool find = false;
DASDelCtxList::iterator iter = del_ctx_list.begin();
for (; !find && iter != del_ctx_list.end(); iter++) {
DmlRowkeyDistCtx del_ctx = *iter;
if (del_ctx.table_id_ == table_id) {
find = true;
rowkey_dist_ctx = del_ctx.deleted_rows_;
}
}
return ret;
}
} // namespace sql
} // namespace oceanbase
src/sql/engine/dml/ob_dml_service.h
浏览文件 @ 0b9d1f8c
......@@ -13,6 +13,7 @@
#ifndef DEV_SRC_SQL_ENGINE_DML_OB_DML_SERVICE_H_
#define DEV_SRC_SQL_ENGINE_DML_OB_DML_SERVICE_H_
#include "sql/engine/dml/ob_dml_ctx_define.h"
#include "sql/das/ob_das_context.h"
namespace oceanbase
{
namespace sql
......@@ -41,8 +42,13 @@ public:
int64_t estimate_row,
DistinctType distinct_algo,
ObEvalCtx &eval_ctx,
SeRowkeyDistCtx *&rowkey_dist_ctx,
ObExecContext &root_ctx,
SeRowkeyDistCtx *rowkey_dist_ctx,
bool &is_dist);
static int create_rowkey_check_hashset(int64_t estimate_row,
ObExecContext *root_ctx,
SeRowkeyDistCtx *&rowkey_dist_ctx);
static int check_row_whether_changed(const ObUpdCtDef &upd_ctdef, ObUpdRtDef &upd_rtdef, ObEvalCtx &eval_ctx);
static int filter_row_for_check_cst(const ExprFixedArray &cst_exprs,
ObEvalCtx &eval_ctx,
......@@ -200,6 +206,11 @@ public:
const ExprFixedArray &row,
const ObDMLBaseCtDef &dml_ctdef,
ObDMLBaseRtDef &dml_rtdef);
static bool is_nested_dup_table(const uint64_t table_id,DASDelCtxList& del_ctx_list);
static int get_nested_dup_table_ctx(const uint64_t table_id,
DASDelCtxList& del_ctx_list,
SeRowkeyDistCtx* &rowkey_dist_ctx);
private:
template <int N>
static int write_row_to_das_op(const ObDASDMLBaseCtDef &ctdef,
......
src/sql/engine/dml/ob_table_delete_op.cpp
浏览文件 @ 0b9d1f8c
......@@ -11,7 +11,7 @@
*/
#define USING_LOG_PREFIX SQL_ENG
#include "common/ob_smart_call.h"
#include "sql/engine/dml/ob_table_delete_op.h"
#include "sql/session/ob_sql_session_info.h"
#include "sql/executor/ob_task_executor_ctx.h"
......
src/sql/engine/dml/ob_table_merge_op.cpp
浏览文件 @ 0b9d1f8c
......@@ -169,6 +169,8 @@ int ObTableMergeOp::open_table_for_each()
int ret = OB_SUCCESS;
if (OB_FAIL(merge_rtdefs_.allocate_array(ctx_.get_allocator(), MY_SPEC.merge_ctdefs_.count()))) {
LOG_WARN("allocate merge rtdef failed", K(ret), K(MY_SPEC.merge_ctdefs_.count()));
} else if (OB_FAIL(ObDMLService::create_rowkey_check_hashset(get_spec().rows_, &ctx_, merge_rtdefs_.at(0).rowkey_dist_ctx_))) {
LOG_WARN("Failed to create hash set", K(ret));
}
trigger_clear_exprs_.reset();
for (int64_t i = 0; OB_SUCC(ret) && i < MY_SPEC.merge_ctdefs_.count(); ++i) {
......@@ -402,6 +404,7 @@ int ObTableMergeOp::check_is_distinct(bool &conflict)
MY_SPEC.rows_,
DistinctType::T_HASH_DISTINCT,
get_eval_ctx(),
get_exec_ctx(),
merge_rtdefs_.at(0).rowkey_dist_ctx_,
// merge_rtdefs_ length must > 0
is_distinct))) {
......
src/sql/engine/dml/ob_table_modify_op.cpp
浏览文件 @ 0b9d1f8c
......@@ -82,7 +82,7 @@ int ForeignKeyHandle::do_handle(ObTableModifyOp &op,
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_nested_session()) {
} 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(cascade(op, fk_arg, old_row, new_row))) {
LOG_WARN("failed to cascade", K(ret), K(fk_arg), K(old_row), K(new_row));
......@@ -519,7 +519,6 @@ ObTableModifyOp::ObTableModifyOp(ObExecContext &ctx,
inner_conn_(NULL),
tenant_id_(0),
saved_conn_(),
is_nested_session_(false),
foreign_key_checks_(false),
need_close_conn_(false),
iter_end_(false),
......@@ -536,6 +535,20 @@ ObTableModifyOp::ObTableModifyOp(ObExecContext &ctx,
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;
......@@ -679,6 +692,16 @@ int ObTableModifyOp::inner_close()
dml_rtctx_.das_ref_.reset();
}
}
// Release the hash sets created at root ctx for delete distinct check
if (OB_SUCC(ret) && get_exec_ctx().is_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;
del_ctx.deleted_rows_->destroy();
}
del_ctx_list.destroy();
}
return ret;
}
......@@ -813,7 +836,6 @@ int ObTableModifyOp::open_inner_conn()
if (OB_SUCC(ret)) {
inner_conn_ = static_cast<ObInnerSQLConnection *>(session->get_inner_conn());
tenant_id_ = session->get_effective_tenant_id();
is_nested_session_ = ObSQLUtils::is_nested_sql(&ctx_);
}
return ret;
}
......
src/sql/engine/dml/ob_table_modify_op.h
浏览文件 @ 0b9d1f8c
......@@ -194,9 +194,11 @@ public:
int execute_write(const char *sql);
int execute_read(const char *sql, common::ObMySQLProxy::MySQLResult &res);
int check_stack();
bool is_nested_session() { return is_nested_session_; }
bool is_nested_session() { return ObSQLUtils::is_nested_sql(&ctx_); }
bool is_fk_nested_session() { return ObSQLUtils::is_fk_nested_sql(&ctx_); }
void set_foreign_key_checks() { foreign_key_checks_ = true; }
bool need_foreign_key_checks() { return foreign_key_checks_; }
bool is_fk_root_session();
const ObObjPrintParams &get_obj_print_params() { return obj_print_params_; }
int init_foreign_key_operation();
void log_user_error_inner(int ret, int64_t col_idx, int64_t row_num,
......@@ -241,7 +243,6 @@ public:
observer::ObInnerSQLConnection *inner_conn_;
uint64_t tenant_id_;
observer::ObInnerSQLConnection::SavedValue saved_conn_;
bool is_nested_session_;
bool foreign_key_checks_;
bool need_close_conn_;
......
src/sql/engine/dml/ob_table_replace_op.cpp
浏览文件 @ 0b9d1f8c
......@@ -11,6 +11,7 @@
*/
#define USING_LOG_PREFIX SQL_ENG
#include "common/ob_smart_call.h"
#include "sql/engine/dml/ob_table_replace_op.h"
#include "share/ob_autoincrement_service.h"
#include "sql/engine/ob_physical_plan_ctx.h"
......@@ -93,6 +94,7 @@ OB_DEF_SERIALIZE_SIZE(ObTableReplaceSpec)
return len;
}
int ObTableReplaceOp::inner_open()
{
int ret = OB_SUCCESS;
......
src/sql/engine/expr/ob_expr_to_outfile_row.cpp
浏览文件 @ 0b9d1f8c
......@@ -268,9 +268,9 @@ int ObExprToOutfileRow::print_field(char *buf, const int64_t buf_len, int64_t &p
ObString tmp_str(out_info.tmp_buf_len_, tmp_pos, out_info.tmp_buf_);
OZ(ObCharsetUtils::foreach_char(tmp_str, out_info.print_params_.cs_type_, escape_func));
}
if (need_enclose) {
OZ(out_info.enclose_.print_plain_str_literal(buf, buf_len, pos, out_info.print_params_));
}
if (need_enclose) {
OZ(out_info.enclose_.print_plain_str_literal(buf, buf_len, pos, out_info.print_params_));
}
return ret;
}
......
src/sql/engine/ob_exec_context.cpp
浏览文件 @ 0b9d1f8c
......@@ -224,6 +224,21 @@ void ObExecContext::reset_op_env()
}
}
int ObExecContext::get_root_ctx(ObExecContext* &root_ctx)
{
int ret = OB_SUCCESS;
if (OB_ISNULL(this->get_parent_ctx())) {
root_ctx = this;
} else if (OB_ISNULL(get_parent_ctx()->get_pl_stack_ctx())) {
root_ctx = this;
} else if (get_parent_ctx()->get_pl_stack_ctx()->in_autonomous()) {
root_ctx = this;
} else if (OB_FAIL( SMART_CALL(get_parent_ctx()->get_root_ctx(root_ctx)))) {
LOG_WARN("failed to get root ctx", K(ret));
}
return ret;
}
int ObExecContext::init_phy_op(const uint64_t phy_op_size)
{
int ret = OB_SUCCESS;
......
src/sql/engine/ob_exec_context.h
浏览文件 @ 0b9d1f8c
......@@ -207,6 +207,9 @@ public:
inline ObSQLSessionInfo *get_my_session() const;
//get the parent execute context in nested sql
ObExecContext *get_parent_ctx() { return parent_ctx_; }
//get the root execute context in nested sql
int get_root_ctx(ObExecContext* &root_ctx);
bool is_root_ctx() {return parent_ctx_ == nullptr;}
int64_t get_nested_level() const { return nested_level_; }
/**
* @brief set sql proxy
......
src/sql/engine/pdml/static/ob_px_multi_part_delete_op.cpp
浏览文件 @ 0b9d1f8c
......@@ -90,6 +90,7 @@ int ObPxMultiPartDeleteOp::check_rowkey_distinct(const ObExprPtrIArray &row,
MY_SPEC.rows_,
MY_SPEC.del_ctdef_.distinct_algo_,
eval_ctx_,
ctx_,
del_rtdef_.se_rowkey_dist_ctx_,
is_distinct);
} else {
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或