/**

* 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/join/ob_merge_join_op.h"

#include "sql/engine/ob_exec_context.h"

namespace oceanbase

{

using namespace common;

namespace sql

{

static const int64_t BATCH_MULTIPLE_TIMES = 10;

OB_SERIALIZE_MEMBER((ObMergeJoinSpec, ObJoinSpec), equal_cond_infos_,

merge_directions_, is_left_unique_,

left_child_fetcher_all_exprs_,

right_child_fetcher_all_exprs_);

OB_SERIALIZE_MEMBER(ObMergeJoinSpec::EqualConditionInfo,

expr_, ser_eval_func_, is_opposite_);

const int64_t ObMergeJoinSpec::MERGE_DIRECTION_ASC = 1;

const int64_t ObMergeJoinSpec::MERGE_DIRECTION_DESC = -1;

ObMergeJoinOp::ObMergeJoinOp(ObExecContext &exec_ctx, const ObOpSpec &spec, ObOpInput *input)

: ObJoinOp(exec_ctx, spec, input),

state_(JS_JOIN_BEGIN),

mem_context_(NULL),

right_cache_(ObModIds::OB_SQL_MERGE_JOIN),

stored_row_(NULL),

stored_row_idx_(-1),

empty_cache_iter_side_(ITER_BOTH),

equal_cmp_(-1),

left_row_matched_(false),

cmp_res_(0),

match_groups_(),

is_last_right_join_output_(false),

output_cache_(),

rj_match_vec_(NULL),

rj_match_vec_size_(0),

group_idx_(0),

left_group_(),

right_group_(),

batch_join_state_(BJS_JOIN_BEGIN),

left_brs_fetcher_(match_groups_, *this, exec_ctx.get_allocator()),

right_brs_fetcher_(match_groups_, *this, exec_ctx.get_allocator()),

profile_(ObSqlWorkAreaType::HASH_WORK_AREA),

sql_mem_processor_(profile_, op_monitor_info_),

left_mem_bound_ratio_(0)

{

state_operation_func_[JS_JOIN_END] = &ObMergeJoinOp::join_end_operate;

state_function_func_[JS_JOIN_END][FT_ROWS_EQUAL] = NULL;

state_function_func_[JS_JOIN_END][FT_ROWS_DIFF] = NULL;

state_function_func_[JS_JOIN_END][FT_ITER_GOING] = NULL;

state_function_func_[JS_JOIN_END][FT_ITER_END] = &ObMergeJoinOp::join_end_func_end;

state_operation_func_[JS_JOIN_BEGIN] = &ObMergeJoinOp::join_begin_operate;

state_function_func_[JS_JOIN_BEGIN][FT_ROWS_EQUAL] = NULL;

state_function_func_[JS_JOIN_BEGIN][FT_ROWS_DIFF] = NULL;

state_function_func_[JS_JOIN_BEGIN][FT_ITER_GOING] = &ObMergeJoinOp::join_begin_func_going;

state_function_func_[JS_JOIN_BEGIN][FT_ITER_END] = &ObMergeJoinOp::join_begin_func_end;

state_operation_func_[JS_LEFT_JOIN] = &ObMergeJoinOp::left_join_operate;

state_function_func_[JS_LEFT_JOIN][FT_ROWS_EQUAL] = NULL;

state_function_func_[JS_LEFT_JOIN][FT_ROWS_DIFF] = NULL;

state_function_func_[JS_LEFT_JOIN][FT_ITER_GOING] = &ObMergeJoinOp::left_join_func_going;

state_function_func_[JS_LEFT_JOIN][FT_ITER_END] = &ObMergeJoinOp::left_join_func_end;

state_operation_func_[JS_RIGHT_JOIN_CACHE] = &ObMergeJoinOp::right_join_cache_operate;

state_function_func_[JS_RIGHT_JOIN_CACHE][FT_ROWS_EQUAL] = NULL;

state_function_func_[JS_RIGHT_JOIN_CACHE][FT_ROWS_DIFF] = NULL;

state_function_func_[JS_RIGHT_JOIN_CACHE][FT_ITER_GOING] = &ObMergeJoinOp::right_join_cache_func_going;

state_function_func_[JS_RIGHT_JOIN_CACHE][FT_ITER_END] = &ObMergeJoinOp::right_join_cache_func_end;

state_operation_func_[JS_RIGHT_JOIN] = &ObMergeJoinOp::right_join_operate;

state_function_func_[JS_RIGHT_JOIN][FT_ROWS_EQUAL] = NULL;

state_function_func_[JS_RIGHT_JOIN][FT_ROWS_DIFF] = NULL;

state_function_func_[JS_RIGHT_JOIN][FT_ITER_GOING] = &ObMergeJoinOp::right_join_func_going;

state_function_func_[JS_RIGHT_JOIN][FT_ITER_END] = &ObMergeJoinOp::right_join_func_end;

state_operation_func_[JS_GOING_END_ONLY] = NULL;

state_function_func_[JS_GOING_END_ONLY][FT_ROWS_EQUAL] = NULL;

state_function_func_[JS_GOING_END_ONLY][FT_ROWS_DIFF] = NULL;

state_function_func_[JS_GOING_END_ONLY][FT_ITER_GOING] = NULL;

state_function_func_[JS_GOING_END_ONLY][FT_ITER_END] = NULL;

state_operation_func_[JS_EMPTY_CACHE] = &ObMergeJoinOp::empty_cache_operate;

state_function_func_[JS_EMPTY_CACHE][FT_ROWS_EQUAL] = &ObMergeJoinOp::empty_cache_func_equal;

state_function_func_[JS_EMPTY_CACHE][FT_ROWS_DIFF] = &ObMergeJoinOp::empty_cache_func_diff;

state_function_func_[JS_EMPTY_CACHE][FT_ITER_GOING] = NULL;

state_function_func_[JS_EMPTY_CACHE][FT_ITER_END] = &ObMergeJoinOp::empty_cache_func_end;

state_operation_func_[JS_FILL_CACHE] = &ObMergeJoinOp::fill_cache_operate;

state_function_func_[JS_FILL_CACHE][FT_ROWS_EQUAL] = &ObMergeJoinOp::fill_cache_func_equal;

state_function_func_[JS_FILL_CACHE][FT_ROWS_DIFF] = &ObMergeJoinOp::fill_cache_func_diff_end;

state_function_func_[JS_FILL_CACHE][FT_ITER_GOING] = NULL;

state_function_func_[JS_FILL_CACHE][FT_ITER_END] = &ObMergeJoinOp::fill_cache_func_diff_end;

state_operation_func_[JS_FULL_CACHE] = &ObMergeJoinOp::full_cache_operate;

state_function_func_[JS_FULL_CACHE][FT_ROWS_EQUAL] = &ObMergeJoinOp::full_cache_func_equal;

state_function_func_[JS_FULL_CACHE][FT_ROWS_DIFF] = &ObMergeJoinOp::full_cache_func_diff;

state_function_func_[JS_FULL_CACHE][FT_ITER_GOING] = NULL;

state_function_func_[JS_FULL_CACHE][FT_ITER_END] = &ObMergeJoinOp::full_cache_func_end;

state_operation_func_[JS_READ_CACHE] = &ObMergeJoinOp::read_cache_operate;

state_function_func_[JS_READ_CACHE][FT_ROWS_EQUAL] = NULL;

state_function_func_[JS_READ_CACHE][FT_ROWS_DIFF] = NULL;

state_function_func_[JS_READ_CACHE][FT_ITER_GOING] = &ObMergeJoinOp::read_cache_func_going;

state_function_func_[JS_READ_CACHE][FT_ITER_END] = &ObMergeJoinOp::read_cache_func_end;

}

int ObMergeJoinOp::inner_open()

{

int ret = OB_SUCCESS;

ObIAllocator &allocator = ctx_.get_allocator();

if (OB_FAIL(ObJoinOp::inner_open())) {

LOG_WARN("failed to open in base class", K(ret));

} else if (OB_FAIL(left_fetcher_.init(*left_, allocator, &left_row_joined_))

|| OB_FAIL(right_fetcher_.init(*right_, allocator, NULL))) {

LOG_WARN("init row fetcher failed", K(ret));

} else if (OB_FAIL(init_mem_context())) {

LOG_WARN("fail to init memory context", K(ret));

} else if (MY_SPEC.is_vectorized()) {

const ExprFixedArray &left_outputs = left_->get_spec().output_;

const ExprFixedArray &right_outputs = right_->get_spec().output_;

const uint64_t tenant_id = ctx_.get_my_session()->get_effective_tenant_id();

match_groups_.set_attr(ObMemAttr(tenant_id, "SqlMJGroups"));

output_cache_.set_attr(ObMemAttr(tenant_id, "SqlMJOutput"));

const ObIArray<ObMergeJoinSpec::EqualConditionInfo> &equal_cond_infos =

MY_SPEC.equal_cond_infos_;

const int64_t left_width = left_->get_spec().width_;

const int64_t right_width = right_->get_spec().width_;

const double width_ratio = ((double)left_width) / ((double)(left_width + right_width));

const double MIN_LEFT_MEM_BOUND_RATIO = 0.2;

const double MAX_LEFT_MEM_BOUND_RATIO = 0.8;

// We prefer more memory to the left, otherwise there may waste memory, so left_mem_bound_ratio_

// is multiplied by a coefficient of 1.2.

left_mem_bound_ratio_ = MAX(MIN(MAX_LEFT_MEM_BOUND_RATIO, 1.2 * width_ratio), MIN_LEFT_MEM_BOUND_RATIO);

const int64_t cache_size = MY_SPEC.max_batch_size_ * BATCH_MULTIPLE_TIMES *

(left_width + right_width);

if (OB_FAIL(sql_mem_processor_.init(&mem_context_->get_malloc_allocator(),

tenant_id,

std::max(2L << 20, cache_size),

MY_SPEC.type_,

MY_SPEC.id_,

&ctx_))) {

LOG_WARN("failed to init sql memory manager processor", K(ret));

} else if (OB_FAIL(left_brs_fetcher_.init(

tenant_id, true, left_, equal_cond_infos,

&(MY_SPEC.left_child_fetcher_all_exprs_)))) {

LOG_WARN("init left batch fetcher failed", K(ret));

} else if (OB_FAIL(right_brs_fetcher_.init(

tenant_id, false, right_, equal_cond_infos,

&(MY_SPEC.right_child_fetcher_all_exprs_)))) {

LOG_WARN("init right batch fetcher failed", K(ret));

}

LOG_TRACE("trace init sql mem mgr for merge join", K(profile_.get_cache_size()),

K(profile_.get_expect_size()));

} else {

// non-vectorized

const uint64_t tenant_id = ctx_.get_my_session()->get_effective_tenant_id();

if (!MY_SPEC.is_skip_cache()) {

if (OB_FAIL(sql_mem_processor_.init(&mem_context_->get_malloc_allocator(),

tenant_id, 256 * right_->get_spec().width_,

MY_SPEC.type_,

MY_SPEC.id_,

&ctx_))) {

LOG_WARN("failed to init sql memory manager processor", K(ret));

} else if (OB_FAIL(right_cache_.init(UINT64_MAX/*mem_limit*/,

tenant_id,

ObCtxIds::WORK_AREA,

ObModIds::OB_SQL_MERGE_JOIN))) {

LOG_WARN("init row store failed", K(ret));

} else {

right_cache_.set_allocator(mem_context_->get_malloc_allocator());

right_cache_.set_callback(&sql_mem_processor_);

right_cache_.set_io_event_observer(&io_event_observer_);

right_cache_.set_dir_id(sql_mem_processor_.get_dir_id());

}

}

}

LOG_TRACE("merge join left unique", K(MY_SPEC.id_), K(MY_SPEC.is_left_unique_));

return ret;

}

int ObMergeJoinOp::inner_switch_iterator()

{

int ret = OB_SUCCESS;

reset();

if (OB_FAIL(ObJoinOp::inner_switch_iterator())) {

if (OB_ITER_END != ret) {

LOG_WARN("failed to rescan ObJoin", K(ret));

}

}

return ret;

}

int ObMergeJoinOp::inner_rescan()

{

int ret = OB_SUCCESS;

reset();

if (OB_FAIL(ObJoinOp::inner_rescan())) {

LOG_WARN("failed to rescan ObJoin", K(ret));

}

return ret;

}

int ObMergeJoinOp::inner_get_next_row()

{

int ret = OB_SUCCESS;

state_operation_func_type state_operation = NULL;

state_function_func_type state_function = NULL;

int func = -1;

output_row_produced_ = false;

while (OB_SUCC(ret) && !output_row_produced_) {

state_operation = this->ObMergeJoinOp::state_operation_func_[state_];

OB_ASSERT(NULL != state_operation);

if (OB_ITER_END == (ret = (this->*state_operation)())) {

func = FT_ITER_END;

ret = OB_SUCCESS;

} else if (OB_FAIL(ret)) {

LOG_WARN("failed to exec state operation", K(ret), K(state_));

} else if (FALSE_IT(clear_evaluated_flag())) {

// do nothing

} else if (state_ < JS_GOING_END_ONLY) { // see comment for JS_GOING_END_ONLY.

func = FT_ITER_GOING;

} else if (OB_FAIL(calc_equal_conds(equal_cmp_))) {

LOG_WARN("failed to calc equal conds", K(ret), K(state_));

} else if (0 == equal_cmp_) {

func = FT_ROWS_EQUAL;

} else {

func = FT_ROWS_DIFF;

}

if (OB_SUCC(ret)) {

state_function = this->ObMergeJoinOp::state_function_func_[state_][func];

if (OB_ISNULL(state_function)) {

ret = OB_BAD_NULL_ERROR;

LOG_WARN("state operation is null ", K(ret));

} else if (OB_FAIL((this->*state_function)()) && OB_ITER_END != ret) {

LOG_WARN("failed state function", K(ret), K(state_), K(func));

}

}

} // while end

return ret;

}

int ObMergeJoinOp::join_end_operate()

{

return OB_ITER_END;

}

int ObMergeJoinOp::join_end_func_end()

{

return OB_ITER_END;

}

int ObMergeJoinOp::join_begin_operate()

{

int ret = OB_SUCCESS;

if (OB_FAIL(left_fetcher_.next()) && OB_ITER_END != ret) {

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

}

return ret;

}

int ObMergeJoinOp::join_begin_func_going()

{

int ret = OB_SUCCESS;

state_ = JS_EMPTY_CACHE;

empty_cache_iter_side_ = ITER_RIGHT;

return ret;

}

int ObMergeJoinOp::join_begin_func_end()

{

if (need_right_join()) {

state_ = JS_RIGHT_JOIN;

} else {

state_ = JS_JOIN_END;

}

return OB_SUCCESS;

}

int ObMergeJoinOp::left_join_operate()

{

int ret = OB_SUCCESS;

if (OB_FAIL(left_fetcher_.next()) && OB_ITER_END != ret) {

LOG_WARN("failed to get left row", K(ret));

}

return ret;

}

int ObMergeJoinOp::left_join_func_going()

{

int ret = OB_SUCCESS;

output_row_produced_ = true;

if (LEFT_ANTI_JOIN != MY_SPEC.join_type_

&& OB_FAIL(blank_right_row())) {

LOG_WARN("fail to blank left row", K(ret));

}

return ret;

}

int ObMergeJoinOp::left_join_func_end()

{

state_ = JS_JOIN_END;

return OB_SUCCESS;

}

int ObMergeJoinOp::right_join_cache_operate()

{

int ret = OB_SUCCESS;

if (OB_FAIL(get_next_right_cache_row()) && OB_ITER_END != ret) {

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

}

return ret;

}

int ObMergeJoinOp::right_join_cache_func_going()

{

int ret = OB_SUCCESS;

if (OB_ISNULL(stored_row_)) {

ret = OB_BAD_NULL_ERROR;

LOG_WARN("stored row is null", K(ret));

} else if (!is_match(stored_row_idx_)) {

output_row_produced_ = true;

if (OB_FAIL(blank_left_row())) {

LOG_WARN("fail to blank right row", K(ret));

}

} else {}

return ret;

}

int ObMergeJoinOp::right_join_cache_func_end()

{

right_cache_.reset();

right_cache_iter_.reset();

stored_row_idx_ = -1;

rj_match_vec_->reset(rj_match_vec_size_);

// full cache diff, 且right reach end时,

// 先将right cache中数据处理, 结束后如果是FULL_OUTER_JOIN,

// 再处理LEFT 数据

if (right_fetcher_.reach_end_) {

if (FULL_OUTER_JOIN == MY_SPEC.join_type_) {

state_ = JS_LEFT_JOIN;

} else {

state_ = JS_JOIN_END;

}

} else {

if (left_fetcher_.reach_end_) {

state_ = JS_RIGHT_JOIN;

} else {

state_ = JS_EMPTY_CACHE;

}

}

return OB_SUCCESS;

}

int ObMergeJoinOp::right_join_operate()

{

int ret = OB_SUCCESS;

if (OB_FAIL(right_fetcher_.next()) && OB_ITER_END != ret) {

LOG_WARN("failed to get right row", K(ret));

}

return ret;

}

int ObMergeJoinOp::right_join_func_going()

{

int ret = OB_SUCCESS;

output_row_produced_ = true;

if (OB_FAIL(blank_left_row())) {

LOG_WARN("fail to blank right row", K(ret));

}

return ret;

}

int ObMergeJoinOp::right_join_func_end()

{

state_ = JS_JOIN_END;

return OB_SUCCESS;

}

int ObMergeJoinOp::read_cache_operate()

{

int ret = OB_SUCCESS;

if ((LEFT_SEMI_JOIN == MY_SPEC.join_type_ || LEFT_ANTI_JOIN == MY_SPEC.join_type_)

&& left_row_matched_) {

//terminate read right rows in advance

ret = OB_ITER_END;

} else if (OB_FAIL(get_next_right_cache_row()) && OB_ITER_END != ret) {

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

}

return ret;

}

int ObMergeJoinOp::read_cache_func_going()

{

int ret = OB_SUCCESS;

if (OB_FAIL(trans_to_read_cache())) {

LOG_WARN("failed to state read cache", K(ret));

}

return ret;

}

int ObMergeJoinOp::read_cache_func_end()

{

int ret = OB_SUCCESS;

clear_evaluated_flag();

if (LEFT_ANTI_JOIN == MY_SPEC.join_type_ && JS_READ_CACHE == state_) {

if (!left_row_matched_) {

output_row_produced_ = true;

}

left_row_matched_ = false;

} else if (LEFT_SEMI_JOIN == MY_SPEC.join_type_ && JS_READ_CACHE == state_) {

if (left_row_matched_) {

output_row_produced_ = true;

}

left_row_matched_ = false;

} else if (need_left_join() && !left_row_joined_) {

output_row_produced_ = true;

if (OB_FAIL(blank_right_row())) {

LOG_WARN("fail to blank left row", K(ret));

}

} else {}

if (OB_SUCC(ret)) {

left_row_matched_ = false;

if (OB_FAIL(right_cache_.begin(right_cache_iter_))) {

LOG_WARN("failed to begin iterator for chunk row store", K(ret));

} else {

stored_row_idx_ = -1;

state_ = JS_FULL_CACHE;

}

}

return ret;

}

int ObMergeJoinOp::full_cache_operate()

{

int ret = OB_SUCCESS;

if (OB_FAIL(left_fetcher_.next())) {

if (OB_ITER_END != ret) {

LOG_WARN("get left row failed", K(ret));

}

} else if (OB_FAIL(get_next_right_cache_row())) {

if (OB_ITER_END != ret) {

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

} else {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("right cache should not be empty", K(ret));

}

}

return ret;

}

int ObMergeJoinOp::full_cache_func_equal()

{

int ret = OB_SUCCESS;

if (OB_FAIL(trans_to_read_cache())) {

LOG_WARN("failed to state read cache", K(ret));

}

return ret;

}

int ObMergeJoinOp::full_cache_func_diff()

{

int ret = OB_SUCCESS;

if (OB_FAIL(left_fetcher_.save_last())) {

LOG_WARN("save last left row failed", K(ret));

} else {

//在右表已迭代结束时semi join和inner join可提前退出

if (right_fetcher_.reach_end_) {

if (LEFT_OUTER_JOIN == MY_SPEC.join_type_ || LEFT_ANTI_JOIN == MY_SPEC.join_type_) {

state_ = JS_LEFT_JOIN;

} else if (need_right_join()) {

// full cache diff, 且right reach end时,

// 先将right cache中数据处理, 结束后如果是FULL_OUTER_JOIN,

// 再处理LEFT 数据

if (OB_FAIL(right_cache_.begin(right_cache_iter_))) {

LOG_WARN("failed to begin iterator for chunk row store", K(ret));

} else {

stored_row_idx_ = -1;

state_ = JS_RIGHT_JOIN_CACHE;

}

} else {

state_ = JS_JOIN_END;

}

} else {

if (need_right_join()) {

if (OB_FAIL(right_cache_.begin(right_cache_iter_))) {

LOG_WARN("failed to begin iterator for chunk row store", K(ret));

} else {

stored_row_idx_ = -1;

state_ = JS_RIGHT_JOIN_CACHE;

}

} else {

right_cache_.reset();

right_cache_iter_.reset();

stored_row_idx_ = -1;

state_ = JS_EMPTY_CACHE;

}

}

}

return ret;

}

// 当左表数据已迭代结束时

int ObMergeJoinOp::full_cache_func_end()

{

int ret = OB_SUCCESS;

if (need_right_join()) {

if (OB_FAIL(right_cache_.begin(right_cache_iter_))) {

LOG_WARN("failed to begin iterator for chunk row store", K(ret));

} else {

stored_row_idx_ = -1;

state_ = JS_RIGHT_JOIN_CACHE;

}

} else {

state_ = JS_JOIN_END;

}

return ret;

}

int ObMergeJoinOp::empty_cache_operate()

{

int ret = OB_SUCCESS;

if (empty_cache_iter_side_ & ITER_LEFT) {

if (OB_FAIL(left_fetcher_.next()) && OB_ITER_END != ret) {

LOG_WARN("get row from left failed", K(ret));

}

// For left join, the right row may be overwrite by blank_right_row(), need to recover here

// if right side has backup row and right will not by iterated.

if (OB_SUCC(ret)

&& !(empty_cache_iter_side_ & ITER_RIGHT)

&& right_fetcher_.has_backup_row_

&& need_left_join()) {

if (OB_FAIL(right_fetcher_.restore())) {

LOG_WARN("restore backup row failed", K(ret));

}

}

}

if (OB_SUCCESS == ret && (empty_cache_iter_side_ & ITER_RIGHT)) {

if (OB_FAIL(right_fetcher_.next()) && OB_ITER_END != ret) {

LOG_WARN("get row from right failed", K(ret));

}

// Same with the left join here, see comment above.

if (OB_SUCC(ret)

&& !(empty_cache_iter_side_ & ITER_LEFT)

&& left_fetcher_.has_backup_row_

&& need_right_join()) {

if (OB_FAIL(left_fetcher_.restore())) {

LOG_WARN("restore backup row failed", K(ret));

}

}

}

return ret;

}

int ObMergeJoinOp::empty_cache_func_equal()

{

int ret = OB_SUCCESS;

if (MY_SPEC.is_skip_cache()) {

// Keep in JS_EMPTY_CACHE state, iterate right side.

empty_cache_iter_side_ = ITER_RIGHT;

} else {

// When reenter JS_EMPTY_CACHE state after JS_FILL_CACHE and JS_FULL_CACHE

// iterate both side.

empty_cache_iter_side_ = ITER_BOTH;

}

if (OB_FAIL(trans_to_fill_cache())) {

LOG_WARN("failed to fill cache", K(ret));

}

return ret;

}

int ObMergeJoinOp::empty_cache_func_diff()

{

int ret = OB_SUCCESS;

if (equal_cmp_ < 0) {

empty_cache_iter_side_ = ITER_LEFT;

if (LEFT_ANTI_JOIN == MY_SPEC.join_type_) {

output_row_produced_ = true;

} else if (need_left_join()) {

output_row_produced_ = true;

if (OB_FAIL(blank_right_row())) {

LOG_WARN("fail to blank left row", K(ret));

}

} else {}

} else if (equal_cmp_ > 0) {

empty_cache_iter_side_ = ITER_RIGHT;;

if (need_right_join()) {

output_row_produced_ = true;

if (OB_FAIL(blank_left_row())) {

LOG_WARN("fail to blank right row", K(ret));

}

} else {}

} else {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("compare result should NOT be equal", K(ret));

}

return ret;

}

int ObMergeJoinOp::empty_cache_func_end()

{

int ret = OB_SUCCESS;

if (right_fetcher_.reach_end_) {

if (need_left_join()

|| LEFT_OUTER_JOIN == MY_SPEC.join_type_

|| LEFT_ANTI_JOIN == MY_SPEC.join_type_) {

state_ = JS_LEFT_JOIN;

if (OB_FAIL(left_fetcher_.save_last())) {

LOG_WARN("save last row failed", K(ret));

}

} else {

state_ = JS_JOIN_END;

}

} else if (left_fetcher_.reach_end_) {

if (need_right_join()) {

state_ = JS_RIGHT_JOIN;

if (OB_FAIL(right_fetcher_.save_last())) {

LOG_WARN("save last row failed", K(ret));

}

} else {

state_ = JS_JOIN_END;

}

} else {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("one side should reach end", K(ret));

}

return ret;

}

int ObMergeJoinOp::fill_cache_operate()

{

int ret = OB_SUCCESS;

// left row is already fetched.

if (OB_FAIL(right_fetcher_.next()) && OB_ITER_END != ret) {

LOG_WARN("get row from right failed", K(ret));

}

return ret;

}

int ObMergeJoinOp::fill_cache_func_equal()

{

int ret = OB_SUCCESS;

if (OB_FAIL(trans_to_fill_cache())) {

LOG_WARN("failed to state fill cache", K(ret));

}

return ret;

}

int ObMergeJoinOp::fill_cache_func_diff_end()

{

int ret = OB_SUCCESS;

clear_evaluated_flag();

if (!right_fetcher_.reach_end_ && OB_FAIL(right_fetcher_.save_last())) {

LOG_WARN("save last row failed", K(ret));

} else if (OB_FAIL(right_cache_.finish_add_row(false))) {

LOG_WARN("failed to finish add row to row store", K(ret));

} else if (OB_FAIL(right_cache_.begin(right_cache_iter_))) {

LOG_WARN("failed to begin iterator for chunk row store", K(ret));

} else {

stored_row_idx_ = -1;

state_ = JS_FULL_CACHE;

//fill cache结束, 如果last_left_row标记为expired_, 则不用输出,

//如果last_left_row没有标记expired_(false), 则表示该行不存在符合条件的行，可输出

if (LEFT_ANTI_JOIN == MY_SPEC.join_type_) {

if (!left_row_matched_) {

output_row_produced_ = true;

}

left_row_matched_ = false;

} else if (LEFT_SEMI_JOIN == MY_SPEC.join_type_) {

if (left_row_matched_) {

output_row_produced_ = true;

}

left_row_matched_ = false;

} else if (need_left_join() && !left_row_joined_) {

output_row_produced_ = true;

if (OB_FAIL(blank_right_row())) {

LOG_WARN("fail to blank left row", K(ret));

}

}

}

return ret;

}

int ObMergeJoinOp::get_next_right_cache_row()

{

int ret = OB_SUCCESS;

if (OB_FAIL(try_check_status())) {

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

} else if (OB_FAIL(right_cache_iter_.get_next_row(right_->get_spec().output_, eval_ctx_,

const_cast<const ObChunkDatumStore::StoredRow **>(&stored_row_)))) {

if (OB_ITER_END != ret) {

LOG_WARN("get next row from right cache failed", K(ret));

}

} else {

++stored_row_idx_;

}

return ret;

}

int ObMergeJoinOp::trans_to_read_cache()

{

int ret = OB_SUCCESS;

bool is_match = false;

if (OB_FAIL(calc_other_conds(is_match))) {

LOG_WARN("failed to compare left and right row on other join conds", K(ret));

} else {

if (is_match) {

if (LEFT_ANTI_JOIN == MY_SPEC.join_type_

|| LEFT_SEMI_JOIN == MY_SPEC.join_type_) {

left_row_matched_ = true;

} else if (FALSE_IT(output_row_produced_ = true)) {

// do nothing

} else if (OB_ISNULL(stored_row_)) {

ret = OB_BAD_NULL_ERROR;

LOG_WARN("stored row is null", K(ret));

} else {

left_row_joined_ = true; // left row joined sign.

if (need_right_join() && OB_FAIL(set_is_match(stored_row_idx_, true))) {

LOG_WARN("fail to set right row joined sign", K(ret), K(stored_row_idx_));

}

}

} else {}

if (OB_SUCC(ret)) {

state_ = JS_READ_CACHE;

}

}

return ret;

}

int ObMergeJoinOp::trans_to_fill_cache()

{

int ret = OB_SUCCESS;

bool is_match = false;

if (OB_FAIL(calc_other_conds(is_match))) {

LOG_WARN("failed to compare left and right row on other join conds", K(ret));

} else if (!MY_SPEC.is_skip_cache()) {

if (OB_FAIL(process_dump())) {

LOG_WARN("failed to process dump", K(ret));

} else if (OB_FAIL(right_cache_.add_row(right_->get_spec().output_, &eval_ctx_, &stored_row_))) {

LOG_WARN("failed to add right row to cache", K(ret));

} else if (OB_ISNULL(stored_row_)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("failed to add stored row", K(stored_row_), K(ret));

} else if (need_right_join() && OB_FAIL(set_is_match(++stored_row_idx_, is_match))) {

LOG_WARN("failed to set right join flags", K(ret), K(stored_row_idx_));

}

}

if (OB_SUCC(ret)) {

if (is_match) {

if (LEFT_ANTI_JOIN == MY_SPEC.join_type_

|| LEFT_SEMI_JOIN == MY_SPEC.join_type_) {

left_row_matched_ = true;

} else if (FALSE_IT(output_row_produced_ = true)) {

// do nothing

} else {

left_row_joined_ = true;

}

} else {}

if (OB_SUCC(ret)) {

if (MY_SPEC.is_skip_cache()) {

state_ = JS_EMPTY_CACHE;

} else {

state_ = JS_FILL_CACHE;

}

}

}

return ret;

}

int ObMergeJoinOp::calc_equal_conds(int64_t &cmp_res)

{

int ret = OB_SUCCESS;

cmp_res = 0;

for (int64_t i = 0;

OB_SUCC(ret) && 0 == cmp_res && i < MY_SPEC.equal_cond_infos_.count();

i++) {

const ObMergeJoinSpec::EqualConditionInfo &equal_cond = MY_SPEC.equal_cond_infos_.at(i);

ObExpr *l_expr = equal_cond.is_opposite_ ? equal_cond.expr_->args_[1]

: equal_cond.expr_->args_[0];

ObExpr *r_expr = equal_cond.is_opposite_ ? equal_cond.expr_->args_[0]

: equal_cond.expr_->args_[1];

ObDatum *l_datum = NULL;

ObDatum *r_datum = NULL;

if (OB_FAIL(l_expr->eval(eval_ctx_, l_datum))

|| OB_FAIL(r_expr->eval(eval_ctx_, r_datum))) {

LOG_WARN("expr evaluate failed", K(ret));

} else {

if (l_datum->is_null() && r_datum->is_null()) {

cmp_res = (T_OP_NSEQ == equal_cond.expr_->type_) ? 0 : -1;

} else {

int cmp_ret = 0;

if (OB_FAIL(equal_cond.ns_cmp_func_(*l_datum, *r_datum, cmp_ret))) {

LOG_WARN("failed to compare", K(ret));

} else if (cmp_ret != 0) {

cmp_res = cmp_ret;

cmp_res *= MY_SPEC.merge_directions_.at(i);

}

}

}

} // for end

return ret;

}

int ObMergeJoinOp::init_mem_context()

{

int ret = OB_SUCCESS;

if (OB_ISNULL(mem_context_)) {

ObSQLSessionInfo *session = ctx_.get_my_session();

uint64_t tenant_id = session->get_effective_tenant_id();

lib::ContextParam param;

param.set_mem_attr(tenant_id,

ObModIds::OB_SQL_MERGE_JOIN,

ObCtxIds::WORK_AREA)

.set_properties(lib::USE_TL_PAGE_OPTIONAL);

if (OB_FAIL(CURRENT_CONTEXT->CREATE_CONTEXT(mem_context_, param))) {

LOG_WARN("create entity failed", K(ret));

} else if (OB_ISNULL(mem_context_)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("null memory entity returned", K(ret));

}

}

return ret;

}

int ObMergeJoinOp::set_is_match(const int64_t idx, const bool is_match)

{

int ret = OB_SUCCESS;

if (OB_FAIL(expand_match_flags_if_necessary(idx + 1, true))) {

LOG_WARN("fail to expand match flags", K(ret));

} else if (is_match) {

rj_match_vec_->set(idx);

}

return ret;

}

int ObMergeJoinOp::process_dump()

{

int ret = OB_SUCCESS;

bool updated = false;

bool dumped = false;

UNUSED(updated);

if (OB_FAIL(sql_mem_processor_.update_max_available_mem_size_periodically(

&mem_context_->get_malloc_allocator(),

[&](int64_t cur_cnt){ return right_cache_.get_row_cnt_in_memory() > cur_cnt; },

updated))) {

LOG_WARN("failed to update max available memory size periodically", K(ret));

} else if (sql_mem_processor_.get_data_size() > sql_mem_processor_.get_mem_bound()

&& GCONF.is_sql_operator_dump_enabled()

&& OB_FAIL(sql_mem_processor_.extend_max_memory_size(

&mem_context_->get_malloc_allocator(),

[&](int64_t max_memory_size) {

return sql_mem_processor_.get_data_size() > max_memory_size;

},

dumped, sql_mem_processor_.get_data_size()))) {

LOG_WARN("failed to extend max memory size", K(ret));

} else if (dumped && OB_FAIL(right_cache_.dump(false, true))) {

LOG_WARN("failed to dump row store", K(ret));

} else {

LOG_TRACE("trace material dump", K(sql_mem_processor_.get_data_size()),

K(right_cache_.get_row_cnt_in_memory()), K(sql_mem_processor_.get_mem_bound()));

}

return ret;

}

int ObMergeJoinOp::ChildBatchFetcher::init(

const uint64_t tenant_id, bool is_left, ObOperator *child,

const ObIArray<ObMergeJoinSpec::EqualConditionInfo> &equal_cond_infos,

const ExprFixedArray *all_exprs)

{

int ret = OB_SUCCESS;

child_ = child;

all_exprs_ = all_exprs;

// child operator's spec batch_size might be zero(vectorization NOT enabled case)

// In that case, child return 1 row at a time, set local batch default 1

batch_size_ =

child->get_spec().max_batch_size_ == 0 ? 1 : child->get_spec().max_batch_size_;

if (OB_FAIL(datum_store_.init(0/*mem_limit*/,

tenant_id,

ObCtxIds::WORK_AREA,

ObModIds::OB_SQL_MERGE_JOIN))) {

LOG_WARN("init datum store failed", K(ret));

} else if (OB_FAIL(brs_holder_.init(*all_exprs_, merge_join_op_.eval_ctx_))) {

LOG_WARN("init brs holder failed", K(ret));

} else if (OB_FAIL(equal_param_idx_.init(equal_cond_infos.count()))) {

LOG_WARN("init equal param idx failed", K(ret));

} else if (OB_ISNULL(merge_join_op_.mem_context_)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("null memory context", K(ret));

} else {

datum_store_.set_allocator(merge_join_op_.mem_context_->get_malloc_allocator());

datum_store_.set_mem_stat(&(merge_join_op_.sql_mem_processor_));

datum_store_.set_io_observer(&(merge_join_op_.io_event_observer_));

datum_store_.set_dir_id(merge_join_op_.sql_mem_processor_.get_dir_id());

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

const ObMergeJoinSpec::EqualConditionInfo &equal_cond = equal_cond_infos.at(i);

ObExpr *param_expr = NULL;

if ((is_left && !equal_cond.is_opposite_) || (!is_left && equal_cond.is_opposite_)) {

param_expr = equal_cond.expr_->args_[0];

} else {

param_expr = equal_cond.expr_->args_[1];

}

int64_t idx = -1;

for (int64_t j = 0; j < all_exprs_->count() && idx < 0; j++) {

if (all_exprs_->at(j) == param_expr) {

idx = j;

}

}

if (OB_UNLIKELY(idx < 0)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("equal cond param not found in child output", K(ret), K(i), K(equal_cond),

KPC(param_expr), K(is_left), K(merge_join_op_.left_->get_spec().output_),

K(merge_join_op_.right_->get_spec().output_));

} else if (OB_FAIL(equal_param_idx_.push_back(idx))) {

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

}

}

LOG_DEBUG("end init equal conds params idx", K(equal_param_idx_));

}

return ret;

}

int ObMergeJoinOp::ChildBatchFetcher::get_next_batch(const int64_t max_row_cnt)

{

int ret = OB_SUCCESS;

const int64_t remain_backup_rows = backup_rows_cnt_ - backup_rows_used_;

if (OB_UNLIKELY(0 != remain_backup_rows)) {

if (OB_UNLIKELY(backup_datums_.count() != all_exprs_->count())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("store datums cnt and child output cnt not equal", K(ret),

K(backup_datums_.count()), K(all_exprs_->count()));

} else {

const int64_t restore_cnt = MIN(max_row_cnt, remain_backup_rows);

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

const ObExpr *expr = all_exprs_->at(i);

if (expr->is_const_expr()) {

continue;

} else {

ObDatum *datum = all_exprs_->at(i)->locate_batch_datums(merge_join_op_.eval_ctx_);

MEMCPY(datum, backup_datums_.at(i) + backup_rows_used_, sizeof(ObDatum) * restore_cnt);

all_exprs_->at(i)->set_evaluated_projected(merge_join_op_.eval_ctx_);

}

}

brs_.size_ = restore_cnt;

brs_.end_ = false;

brs_.skip_ = NULL;

backup_rows_used_ += restore_cnt;

}

} else {

if (brs_.end_) {

// not more fetch if reach end

brs_.size_ = 0;

} else {

const ObBatchRows *child_brs = NULL;

if (OB_FAIL(child_->get_next_batch(max_row_cnt, child_brs))) {

LOG_WARN("get child next batch failed", K(ret));

} else {