From 865c7006a6ae5381d4a34ae53bc247ed02855d8f Mon Sep 17 00:00:00 2001 From: ls0 Date: Thu, 3 Mar 2022 14:46:39 +0800 Subject: [PATCH] adjust set op to reentrant --- src/sql/engine/set/ob_hash_except_op.cpp | 7 ++++++- src/sql/engine/set/ob_hash_intersect_op.cpp | 7 ++++++- src/sql/engine/set/ob_hash_set_op.cpp | 4 +++- src/sql/engine/set/ob_hash_set_op.h | 1 + src/sql/engine/set/ob_hash_union_op.cpp | 5 +++++ src/sql/engine/set/ob_merge_except_op.cpp | 5 +++++ src/sql/engine/set/ob_merge_intersect_op.cpp | 5 +++++ src/sql/engine/set/ob_merge_set_op.cpp | 4 +++- src/sql/engine/set/ob_merge_set_op.h | 1 + src/sql/engine/set/ob_merge_union_op.cpp | 4 ++++ 10 files changed, 39 insertions(+), 4 deletions(-) diff --git a/src/sql/engine/set/ob_hash_except_op.cpp b/src/sql/engine/set/ob_hash_except_op.cpp index f21223ff0b..66ae454768 100644 --- a/src/sql/engine/set/ob_hash_except_op.cpp +++ b/src/sql/engine/set/ob_hash_except_op.cpp @@ -97,7 +97,9 @@ int ObHashExceptOp::inner_get_next_row() const ObChunkDatumStore::StoredRow* store_row = nullptr; const common::ObIArray* cur_exprs = nullptr; clear_evaluated_flag(); - if (first_get_left_) { + if (iter_end_) { + ret = OB_ITER_END; + } else if (first_get_left_) { if (OB_FAIL(is_left_has_row(left_has_row))) { LOG_WARN("failed to judge left has row", K(ret)); } else if (!left_has_row) { @@ -178,6 +180,9 @@ int ObHashExceptOp::inner_get_next_row() LOG_WARN("copy current row failed", K(ret)); } } + if (OB_ITER_END == ret) { + iter_end_ = true; + } return ret; } diff --git a/src/sql/engine/set/ob_hash_intersect_op.cpp b/src/sql/engine/set/ob_hash_intersect_op.cpp index 29050f2f36..6c02fac1e1 100644 --- a/src/sql/engine/set/ob_hash_intersect_op.cpp +++ b/src/sql/engine/set/ob_hash_intersect_op.cpp @@ -84,7 +84,9 @@ int ObHashIntersectOp::inner_get_next_row() const ObHashPartCols* part_cols = nullptr; const common::ObIArray* cur_exprs = nullptr; clear_evaluated_flag(); - if (first_get_left_) { + if (iter_end_) { + ret = OB_ITER_END; + } else if (first_get_left_) { if (OB_FAIL(is_left_has_row(left_has_row))) { LOG_WARN("failed to judge left has row", K(ret)); } else if (!left_has_row) { @@ -164,6 +166,9 @@ int ObHashIntersectOp::inner_get_next_row() LOG_WARN("copy current row failed", K(ret)); } } + if (OB_ITER_END == ret) { + iter_end_ = true; + } return ret; } diff --git a/src/sql/engine/set/ob_hash_set_op.cpp b/src/sql/engine/set/ob_hash_set_op.cpp index 8f242b8a78..83f651e00b 100644 --- a/src/sql/engine/set/ob_hash_set_op.cpp +++ b/src/sql/engine/set/ob_hash_set_op.cpp @@ -32,7 +32,8 @@ ObHashSetOp::ObHashSetOp(ObExecContext& exec_ctx, const ObOpSpec& spec, ObOpInpu has_got_part_(false), profile_(ObSqlWorkAreaType::HASH_WORK_AREA), sql_mem_processor_(profile_), - hp_infras_() + hp_infras_(), + iter_end_(false) {} int ObHashSetOp::inner_open() @@ -52,6 +53,7 @@ void ObHashSetOp::reset() first_get_left_ = true; has_got_part_ = false; hp_infras_.reset(); + iter_end_ = false; } int ObHashSetOp::inner_close() diff --git a/src/sql/engine/set/ob_hash_set_op.h b/src/sql/engine/set/ob_hash_set_op.h index ac4a700368..eb64dfaf31 100644 --- a/src/sql/engine/set/ob_hash_set_op.h +++ b/src/sql/engine/set/ob_hash_set_op.h @@ -58,6 +58,7 @@ protected: ObSqlWorkAreaProfile profile_; ObSqlMemMgrProcessor sql_mem_processor_; ObHashPartInfrastructure hp_infras_; + bool iter_end_; }; } // end namespace sql diff --git a/src/sql/engine/set/ob_hash_union_op.cpp b/src/sql/engine/set/ob_hash_union_op.cpp index 6994a4da69..3e988757c2 100644 --- a/src/sql/engine/set/ob_hash_union_op.cpp +++ b/src/sql/engine/set/ob_hash_union_op.cpp @@ -91,6 +91,8 @@ int ObHashUnionOp::inner_get_next_row() LOG_WARN("failed to get next row", K(ret)); } first_get_left_ = false; + } else if (iter_end_) { + ret = OB_ITER_END; } bool got_row = false; bool has_exists = false; @@ -145,6 +147,9 @@ int ObHashUnionOp::inner_get_next_row() LOG_WARN("copy current row failed", K(ret)); } } + if (OB_ITER_END == ret) { + iter_end_ = true; + } return ret; } diff --git a/src/sql/engine/set/ob_merge_except_op.cpp b/src/sql/engine/set/ob_merge_except_op.cpp index 5ed1401d88..b68ca338aa 100644 --- a/src/sql/engine/set/ob_merge_except_op.cpp +++ b/src/sql/engine/set/ob_merge_except_op.cpp @@ -72,6 +72,9 @@ int ObMergeExceptOp::inner_get_next_row() bool break_outer_loop = false; const ObIArray* left_row = NULL; clear_evaluated_flag(); + if (iter_end_) { + ret = OB_ITER_END; + } while (OB_SUCC(ret) && OB_SUCC(do_strict_distinct(*left_, last_row_.store_row_, left_row))) { break_outer_loop = right_iter_end_; while (OB_SUCC(ret) && !right_iter_end_) { @@ -120,6 +123,8 @@ int ObMergeExceptOp::inner_get_next_row() } else if (OB_FAIL(last_row_.save_store_row(*left_row, eval_ctx_, 0))) { LOG_WARN("failed to save right row", K(ret)); } + } else if (OB_ITER_END == ret) { + iter_end_ = true; } return ret; } diff --git a/src/sql/engine/set/ob_merge_intersect_op.cpp b/src/sql/engine/set/ob_merge_intersect_op.cpp index 373a756001..5577f2a28f 100644 --- a/src/sql/engine/set/ob_merge_intersect_op.cpp +++ b/src/sql/engine/set/ob_merge_intersect_op.cpp @@ -71,6 +71,9 @@ int ObMergeIntersectOp::inner_get_next_row() int cmp = 0; bool break_outer_loop = false; const ObIArray* left_row = NULL; + if (iter_end_) { + ret = OB_ITER_END; + } clear_evaluated_flag(); while (OB_SUCC(ret) && OB_SUCC(do_strict_distinct(*left_, last_row_.store_row_, left_row))) { while (OB_SUCC(ret) && !right_iter_end_) { @@ -111,6 +114,8 @@ int ObMergeIntersectOp::inner_get_next_row() } else if (OB_FAIL(last_row_.save_store_row(*left_row, eval_ctx_, 0))) { LOG_WARN("failed to save right row", K(ret)); } + } else if (OB_ITER_END == ret) { + iter_end_ = true; } return ret; } diff --git a/src/sql/engine/set/ob_merge_set_op.cpp b/src/sql/engine/set/ob_merge_set_op.cpp index bb1f6dd3a5..4cbaa8ef6f 100644 --- a/src/sql/engine/set/ob_merge_set_op.cpp +++ b/src/sql/engine/set/ob_merge_set_op.cpp @@ -28,7 +28,8 @@ ObMergeSetOp::ObMergeSetOp(ObExecContext& exec_ctx, const ObOpSpec& spec, ObOpIn alloc_(ObModIds::OB_SQL_MERGE_GROUPBY, OB_MALLOC_NORMAL_BLOCK_SIZE, OB_SERVER_TENANT_ID, ObCtxIds::WORK_AREA), last_row_(alloc_), cmp_(), - need_skip_init_row_(false) + need_skip_init_row_(false), + iter_end_(false) {} int ObMergeSetOp::inner_open() @@ -57,6 +58,7 @@ int ObMergeSetOp::rescan() last_row_.reset(); alloc_.reset(); need_skip_init_row_ = false; + iter_end_ = false; if (OB_FAIL(ObOperator::rescan())) { LOG_WARN("failed to rescan", K(ret)); } diff --git a/src/sql/engine/set/ob_merge_set_op.h b/src/sql/engine/set/ob_merge_set_op.h index 9c2ea18f99..34fbd27b8d 100644 --- a/src/sql/engine/set/ob_merge_set_op.h +++ b/src/sql/engine/set/ob_merge_set_op.h @@ -73,6 +73,7 @@ protected: ObChunkDatumStore::LastStoredRow<> last_row_; Compare cmp_; bool need_skip_init_row_; + bool iter_end_; }; template diff --git a/src/sql/engine/set/ob_merge_union_op.cpp b/src/sql/engine/set/ob_merge_union_op.cpp index eff2827beb..0df530d401 100644 --- a/src/sql/engine/set/ob_merge_union_op.cpp +++ b/src/sql/engine/set/ob_merge_union_op.cpp @@ -282,9 +282,13 @@ int ObMergeUnionOp::inner_get_next_row() if (OB_ISNULL(get_next_row_func_)) { ret = OB_ERR_UNEXPECTED; LOG_ERROR("get_next_row_func is NULL", K(ret)); + } else if (iter_end_) { + ret = OB_ITER_END; } else if (OB_FAIL((this->*get_next_row_func_)())) { if (OB_ITER_END != ret) { LOG_WARN("get next row failed", K(ret)); + } else { + iter_end_ = true; } } return ret; -- GitLab