/**

* 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_hash_join_op.h"

#include "sql/engine/px/ob_px_util.h"

#include "observer/omt/ob_tenant_config_mgr.h"

#include "sql/engine/ob_exec_context.h"

#include "sql/session/ob_sql_session_info.h"

#include "observer/omt/ob_tenant_config_mgr.h"

#include "sql/engine/px/ob_px_util.h"

#include "share/diagnosis/ob_sql_monitor_statname.h"

namespace oceanbase

{

using namespace omt;

using namespace common;

namespace sql

{

OB_SERIALIZE_MEMBER(ObHashJoinInput, shared_hj_info_);

// The ctx is owned by thread

// sync_event is shared

int ObHashJoinInput::sync_wait(ObExecContext &ctx, int64_t &sync_event, EventPred pred, bool ignore_interrupt, bool is_open)

{

int ret = OB_SUCCESS;

ObHashTableSharedTableInfo *shared_hj_info = reinterpret_cast<ObHashTableSharedTableInfo *>(shared_hj_info_);

if (OB_ISNULL(shared_hj_info)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected status: shared hash join info is null", K(ret));

} else if (OB_ISNULL(pred)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected status: pred is null", K(ret));

} else {

bool has_process = false;

int64_t loop = 0;

int64_t exit_cnt = shared_hj_info->sqc_thread_count_ *

(1 + sync_event / shared_hj_info->sqc_thread_count_);

while (OB_SUCC(ret)) {

++loop;

if (!has_process) {

ObSpinLockGuard guard(shared_hj_info->lock_);

// got lock to process and only one thread can process pred()

// it may enter by multiple threads

pred(ATOMIC_LOAD(&sync_event) % shared_hj_info->sqc_thread_count_);

// it must be here to guarantee next wait loop to get lock for one thread

has_process = true;

if (ATOMIC_LOAD(&sync_event) + 1 >= exit_cnt) {

// last thread, it will singal and exit by self

ATOMIC_INC(&sync_event);

shared_hj_info->cond_.signal();

LOG_DEBUG("debug sync event", K(ret), K(lbt()), K(sync_event));

break;

}

ATOMIC_INC(&sync_event);

LOG_DEBUG("debug sync event", K(ret), K(lbt()), K(sync_event));

}

if (!ignore_interrupt && OB_SUCCESS != shared_hj_info->ret_) {

// the thread already return error

ret = shared_hj_info->ret_;

} else if (!ignore_interrupt && 0 == loop % 8 && OB_UNLIKELY(IS_INTERRUPTED())) {

// 中断错误处理

// overwrite ret

ObInterruptCode code = GET_INTERRUPT_CODE();

ret = code.code_;

LOG_WARN("received a interrupt", K(code), K(ret));

} else if (!ignore_interrupt && 0 == loop % 16 && OB_FAIL(ctx.fast_check_status())) {

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

} else if (ATOMIC_LOAD(&sync_event) >= exit_cnt) {

// timeout, and signal has done

LOG_DEBUG("debug sync event", K(ret), K(lbt()), K(sync_event),

K(loop), K(exit_cnt));

break;

} else if (ignore_interrupt /*close stage*/ && OB_SUCCESS != ATOMIC_LOAD(&shared_hj_info->open_ret_)) {

LOG_WARN("some op have failed in open stage", K(ATOMIC_LOAD(&shared_hj_info->open_ret_)));

break;

} else {

auto key = shared_hj_info->cond_.get_key();

// wait one time per 1000 us

shared_hj_info->cond_.wait(key, 1000);

}

} // end while

if (OB_FAIL(ret) && is_open) {

set_open_ret(ret);

}

}

return ret;

}

ObHashJoinSpec::ObHashJoinSpec(common::ObIAllocator &alloc, const ObPhyOperatorType type)

: ObJoinSpec(alloc, type),

equal_join_conds_(alloc),

all_join_keys_(alloc),

all_hash_funcs_(alloc),

can_prob_opt_(false),

is_naaj_(false),

is_sna_(false),

is_shared_ht_(false),

is_ns_equal_cond_(alloc)

{

}

OB_SERIALIZE_MEMBER((ObHashJoinSpec, ObJoinSpec),

equal_join_conds_,

all_join_keys_,

all_hash_funcs_,

can_prob_opt_,

is_naaj_,

is_sna_,

is_shared_ht_,

is_ns_equal_cond_);

int ObHashJoinOp::PartHashJoinTable::init(ObIAllocator &alloc)

{

int ret = OB_SUCCESS;

if (!inited_) {

void *alloc_buf = alloc.alloc(sizeof(ModulePageAllocator));

void *bucket_buf = alloc.alloc(sizeof(BucketArray));

if (OB_ISNULL(bucket_buf) || OB_ISNULL(alloc_buf)) {

ret = OB_ALLOCATE_MEMORY_FAILED;

if (OB_NOT_NULL(bucket_buf)) {

alloc.free(bucket_buf);

}

if (OB_NOT_NULL(alloc_buf)) {

alloc.free(alloc_buf);

}

LOG_WARN("failed to alloc memory", K(ret));

} else {

ht_alloc_ = new (alloc_buf) ModulePageAllocator(alloc);

ht_alloc_->set_label("HtOpAlloc");

buckets_ = new (bucket_buf) BucketArray(*ht_alloc_);

magic_ = MAGIC_CODE;

bit_cnt_ = __builtin_ctz(BucketArray::BLOCK_CAPACITY);

}

}

return ret;

}

ObHashJoinOp::ObHashJoinOp(ObExecContext &ctx_, const ObOpSpec &spec, ObOpInput *input)

: ObJoinOp(ctx_, spec, input),

hj_state_(INIT),

hj_processor_(NONE),

buf_mgr_(NULL),

batch_mgr_(NULL),

left_batch_(NULL),

right_batch_(NULL),

tmp_hash_funcs_(ctx_.get_allocator()),

left_hash_funcs_(),

right_hash_funcs_(),

left_join_keys_(),

right_join_keys_(),

going_func_(nullptr),

end_func_(nullptr),

part_level_(0),

part_shift_(MAX_PART_LEVEL << 3),

part_count_(0),

force_hash_join_spill_(false),

hash_join_processor_(7),

tenant_id_(-1),

input_size_(0),

total_extra_size_(0),

predict_row_cnt_(1024),

profile_(ObSqlWorkAreaType::HASH_WORK_AREA),

sql_mem_processor_(profile_, op_monitor_info_),

state_(JS_READ_RIGHT),

cur_right_hash_value_(0),

right_has_matched_(false),

tuple_need_join_(false),

first_get_row_(true),

drain_mode_(HashJoinDrainMode::NONE_DRAIN),

cur_bkid_(0),

remain_data_memory_size_(0),

nth_nest_loop_(0),

cur_nth_row_(0),

dumped_fixed_mem_size_(0),

hash_table_(),

cur_hash_table_(nullptr),

cur_tuple_(NULL),

cur_left_row_(),

mem_context_(nullptr),

alloc_(nullptr),

bloom_filter_alloc_(nullptr),

bloom_filter_(nullptr),

nth_right_row_(-1),

ltb_size_(INIT_LTB_SIZE),

l2_cache_size_(INIT_L2_CACHE_SIZE),

price_per_row_(48),

max_partition_count_per_level_(ltb_size_ << 1),

cur_dumped_partition_(max_partition_count_per_level_),

batch_round_(1),

nest_loop_state_(HJLoopState::LOOP_START),

is_shared_(false),

is_last_chunk_(false),

has_right_bitset_(false),

hj_part_array_(NULL),

right_hj_part_array_(NULL),

left_read_row_(NULL),

right_read_row_(NULL),

postprocessed_left_(false),

has_fill_right_row_(false),

has_fill_left_row_(false),

right_last_row_(),

need_return_(false),

iter_end_(false),

opt_cache_aware_(false),

has_right_material_data_(false),

enable_bloom_filter_(false),

part_histograms_(nullptr),

cur_full_right_partition_(INT64_MAX),

right_iter_end_(false),

cur_bucket_idx_(0),

max_bucket_idx_(0),

enable_batch_(false),

level1_bit_(0),

level1_part_count_(0),

level2_part_count_(0),

right_splitter_(),

cur_left_hist_(nullptr),

cur_right_hist_(nullptr),

cur_probe_row_idx_(0),

max_right_bucket_idx_(0),

probe_cnt_(0),

bitset_filter_cnt_(0),

hash_link_cnt_(0),

hash_equal_cnt_(0),

max_output_cnt_(0),

hj_part_stored_rows_(NULL),

hash_vals_(NULL),

right_brs_(NULL),

right_hj_part_stored_rows_(NULL),

right_read_from_stored_(false),

right_hash_vals_(NULL),

cur_tuples_(NULL),

right_batch_traverse_cnt_(0),

hj_part_added_rows_(NULL),

part_selectors_(NULL),

part_selector_sizes_(NULL),

right_selector_(NULL),

right_selector_cnt_(0),

read_null_in_naaj_(false),

get_next_right_row_func_(nullptr),

get_next_left_row_func_(nullptr),

get_next_right_batch_func_(nullptr),

get_next_left_batch_func_(nullptr),

non_preserved_side_is_not_empty_(false),

null_random_hash_value_(0),

skip_left_null_(false),

skip_right_null_(false)

{

/*

read_left_row -> build_hash_table

state

--------------->--------------- --JS_READ_RIGHT

| |

| ----------------|---------------------

| | |

| succ -------iter_end------------

| | | | |

| func [FT_ITER_GOING] anti need_lef_join else

| | | | |

| cal_right_hash_value |-->JS_LEFT_ANTI_SEMI JS_FILL_LEFT JS_JOIN_END

| | | | | | |

| -->--------READ_HASH_ROW | succ iter_end succ iter_end

| | | | | | | | |

| | succ iter_end ---<-going JS_JOIN_END | |

| | | | going JS_JOIN_END

| -----<----going JS_ERAD_RIGHT |

| | it's too hard to draw cycle

| | |

----------------<--------- JS_FILL_LEFT

*/

state_operation_func_[JS_JOIN_END] = &ObHashJoinOp::join_end_operate;

state_function_func_[JS_JOIN_END][FT_ITER_GOING] = &ObHashJoinOp::join_end_operate;

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

//open时已对left table的数据做了hash操作，这里对右表进行扫描和计算hash值

state_operation_func_[JS_READ_RIGHT] = &ObHashJoinOp::read_right_operate;

state_function_func_[JS_READ_RIGHT][FT_ITER_GOING] = &ObHashJoinOp::calc_right_hash_value;

state_function_func_[JS_READ_RIGHT][FT_ITER_END] = &ObHashJoinOp::read_right_func_end;

// 根据hash value去扫描buckets,如果得到匹配的row,组合row并返回。

// 桶里可能会有多个元組匹配，需要依次扫描。

state_operation_func_[JS_READ_HASH_ROW] = &ObHashJoinOp::read_hashrow;

state_function_func_[JS_READ_HASH_ROW][FT_ITER_GOING] = &ObHashJoinOp::read_hashrow_func_going;

state_function_func_[JS_READ_HASH_ROW][FT_ITER_END] = &ObHashJoinOp::read_hashrow_func_end;

//for anti

state_operation_func_[JS_LEFT_ANTI_SEMI] = &ObHashJoinOp::left_anti_semi_operate;

state_function_func_[JS_LEFT_ANTI_SEMI][FT_ITER_GOING] = &ObHashJoinOp::left_anti_semi_going;

state_function_func_[JS_LEFT_ANTI_SEMI][FT_ITER_END] = &ObHashJoinOp::left_anti_semi_end;

//for unmatched left row, for left-outer,full-outer,left-semi

state_operation_func_[JS_FILL_LEFT] = &ObHashJoinOp::fill_left_operate;

state_function_func_[JS_FILL_LEFT][FT_ITER_GOING] = &ObHashJoinOp::fill_left_going;

state_function_func_[JS_FILL_LEFT][FT_ITER_END] = &ObHashJoinOp::fill_left_end;

get_next_right_row_func_ = &ObHashJoinOp::get_next_right_row;

get_next_left_row_func_ = &ObHashJoinOp::get_next_left_row;

get_next_right_batch_func_ = &ObHashJoinOp::get_next_right_batch;

get_next_left_batch_func_ = &ObHashJoinOp::get_next_left_row_batch;

}

int ObHashJoinOp::set_hash_function(int8_t hash_join_hasher)

{

int ret = OB_SUCCESS;

const common::ObIArray<common::ObHashFunc> *cur_hash_funcs = &MY_SPEC.all_hash_funcs_;

if (DEFAULT_MURMUR_HASH != (hash_join_hasher & HASH_FUNCTION_MASK)) {

ObHashFunc hash_func;

if (OB_FAIL(tmp_hash_funcs_.init(MY_SPEC.all_hash_funcs_.count()))) {

LOG_WARN("failed to init tmp hash func", K(ret));

} else {

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

if (ENABLE_WY_HASH == (hash_join_hasher & HASH_FUNCTION_MASK)) {

hash_func.hash_func_ = MY_SPEC.all_join_keys_.at(i)->basic_funcs_->wy_hash_;

} else if (ENABLE_XXHASH64 == (hash_join_hasher & HASH_FUNCTION_MASK)) {

hash_func.hash_func_ = MY_SPEC.all_join_keys_.at(i)->basic_funcs_->xx_hash_;

} else {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected status: hash join hasher is invalid", K(ret), K(hash_join_hasher));

}

if (OB_FAIL(ret)) {

} else if (OB_FAIL(tmp_hash_funcs_.push_back(hash_func))) {

LOG_WARN("failed to push back wy hash func", K(ret));

} else {

LOG_DEBUG("debug hash function", K(hash_func), K(i), K(*MY_SPEC.all_join_keys_.at(i)));

}

}

cur_hash_funcs = &tmp_hash_funcs_;

}

}

if (OB_SUCC(ret)) {

int64_t all_cnt = cur_hash_funcs->count();

left_hash_funcs_.init(

all_cnt / 2, const_cast<ObHashFunc*>(&cur_hash_funcs->at(0)), all_cnt / 2);

right_hash_funcs_.init(all_cnt / 2,

const_cast<ObHashFunc*>(&cur_hash_funcs->at(0) + left_hash_funcs_.count()),

cur_hash_funcs->count() - left_hash_funcs_.count());

if (left_hash_funcs_.count() != right_hash_funcs_.count()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected status: hash func is not match", K(ret), K(cur_hash_funcs->count()),

K(left_hash_funcs_.count()), K(right_hash_funcs_.count()));

} else if (MY_SPEC.all_join_keys_.count() != cur_hash_funcs->count()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected status: hash func is not match", K(ret), K(cur_hash_funcs->count()),

K(MY_SPEC.all_join_keys_.count()));

}

}

return ret;

}

uint64_t sum_alloc_size() { return 0; }

template <typename X, typename Y, typename ...TS>

uint64_t sum_alloc_size(const X &, const Y &y, const TS &...args)

{

return y + sum_alloc_size(args...);

}

void assign_alloc_ptr(void *) { return; }

template<typename X, typename Y, typename ...TS>

void assign_alloc_ptr(void *ptr, const X &x, const Y &y, const TS &...args)

{

const_cast<X &>(x) = static_cast<X>(ptr);

assign_alloc_ptr(static_cast<char *>(ptr) + y, args...);

}

// allocate pointers which passed <ptr, size> paires

template <typename ...TS>

int alloc_ptrs(ObIAllocator &alloc, const TS &...args)

{

int ret = OB_SUCCESS;

int64_t size = sum_alloc_size(args...);

void *ptr = alloc.alloc(size);

if (NULL == ptr) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("allocate memory failed", K(ret), K(size));

} else {

MEMSET(ptr, 0, size);

assign_alloc_ptr(ptr, args...);

}

return ret;

}

int ObHashJoinOp::inner_open()

{

int ret = OB_SUCCESS;

ObSQLSessionInfo *session = NULL;

if (OB_FAIL(set_shared_info())) {

LOG_WARN("failed to set shared info", K(ret));

} else if (is_shared_ && OB_FAIL(sync_wait_open())) {

is_shared_ = false;

LOG_WARN("failed to sync open for shared hj", K(ret));

} else if ((OB_UNLIKELY(MY_SPEC.all_join_keys_.count() <= 0

|| MY_SPEC.all_join_keys_.count() != MY_SPEC.all_hash_funcs_.count()

|| OB_ISNULL(left_)))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("no equal join conds or left op is null", K(ret));

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

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

} else if (OB_ISNULL(session = ctx_.get_my_session())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("fail to get my session", K(ret));

} else if (OB_FAIL(init_mem_context(session->get_effective_tenant_id()))) {

LOG_WARN("fail to init base join ctx", K(ret));

} else if (OB_FAIL(hash_table_.init(*alloc_))) {

LOG_WARN("fail to init hash table", K(ret));

} else {

init_system_parameters();

tenant_id_ = session->get_effective_tenant_id();

first_get_row_ = true;

ObTenantConfigGuard tenant_config(TENANT_CONF(session->get_effective_tenant_id()));

if (tenant_config.is_valid()) {

force_hash_join_spill_ = tenant_config->_force_hash_join_spill;

hash_join_processor_ = tenant_config->_enable_hash_join_processor;

if (0 == (hash_join_processor_ & HJ_PROCESSOR_MASK)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpect hash join processor", K(ret), K(hash_join_processor_));

} else if (OB_FAIL(set_hash_function(tenant_config->_enable_hash_join_hasher))) {

LOG_WARN("unexpect hash join function", K(ret));

}

} else {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("invalid tenant config", K(ret));

}

if (is_vectorized()) {

if (INNER_JOIN == MY_SPEC.join_type_) {

going_func_ = &ObHashJoinOp::inner_join_read_hashrow_going_batch;

end_func_ = &ObHashJoinOp::inner_join_read_hashrow_end_batch;

} else if (IS_LEFT_SEMI_ANTI_JOIN(MY_SPEC.join_type_)) {

going_func_ = &ObHashJoinOp::left_anti_semi_read_hashrow_going_batch;

end_func_ = &ObHashJoinOp::left_anti_semi_read_hashrow_end_batch;

} else if (IS_RIGHT_SEMI_ANTI_JOIN(MY_SPEC.join_type_)) {

going_func_ = &ObHashJoinOp::right_anti_semi_read_hashrow_going_batch;

end_func_ = &ObHashJoinOp::right_anti_semi_read_hashrow_end_batch;

} else if (IS_OUTER_JOIN(MY_SPEC.join_type_)) {

going_func_ = &ObHashJoinOp::outer_join_read_hashrow_going_batch;

end_func_ = &ObHashJoinOp::outer_join_read_hashrow_end_batch;

}

} else {

if (INNER_JOIN == MY_SPEC.join_type_) {

going_func_ = &ObHashJoinOp::inner_join_read_hashrow_func_going;

end_func_ = &ObHashJoinOp::inner_join_read_hashrow_func_end;

} else {

going_func_ = &ObHashJoinOp::other_join_read_hashrow_func_going;

end_func_ = &ObHashJoinOp::other_join_read_hashrow_func_end;

}

}

if (MY_SPEC.is_naaj_) {

state_function_func_[JS_LEFT_ANTI_SEMI][FT_ITER_GOING] = &ObHashJoinOp::left_anti_naaj_going;

get_next_right_row_func_ = &ObHashJoinOp::get_next_right_row_na;

get_next_left_row_func_ = &ObHashJoinOp::get_next_left_row_na;

get_next_right_batch_func_ = &ObHashJoinOp::get_next_right_batch_na;

get_next_left_batch_func_ = &ObHashJoinOp::get_next_left_row_batch_na;

}

skip_left_null_ = (INNER_JOIN == MY_SPEC.join_type_) || (LEFT_SEMI_JOIN == MY_SPEC.join_type_)

|| (RIGHT_SEMI_JOIN == MY_SPEC.join_type_) || (RIGHT_OUTER_JOIN == MY_SPEC.join_type_);

skip_right_null_ = (INNER_JOIN == MY_SPEC.join_type_) || (LEFT_SEMI_JOIN == MY_SPEC.join_type_)

|| (RIGHT_SEMI_JOIN == MY_SPEC.join_type_) || (LEFT_OUTER_JOIN == MY_SPEC.join_type_);

}

if (OB_SUCC(ret)) {

int64_t all_cnt = MY_SPEC.all_join_keys_.count();

left_join_keys_.init(

all_cnt / 2, const_cast<ObExpr**>(&MY_SPEC.all_join_keys_.at(0)), all_cnt / 2);

right_join_keys_.init(all_cnt - left_join_keys_.count(),

const_cast<ObExpr**>(&MY_SPEC.all_join_keys_.at(0) + left_join_keys_.count()),

all_cnt - left_join_keys_.count());

if (left_join_keys_.count() != right_join_keys_.count()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected status: left and right join is not match", K(ret),

K(left_join_keys_.count()), K(right_join_keys_.count()));

} else if (MY_SPEC.is_naaj_ && 1 != left_join_keys_.count()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("null aware anti join only support 1 join key", K(ret));

}

LOG_DEBUG("trace join keys", K(left_join_keys_), K(right_join_keys_),

K(left_join_keys_.count()), K(right_join_keys_.count()));

}

if (OB_SUCC(ret)) {

part_count_ = 0;

hj_state_ = ObHashJoinOp::INIT;

void* buf = alloc_->alloc(sizeof(ObHashJoinBufMgr));

if (NULL == buf) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("fail to alloc mem");

} else {

//at least one page for each l/r part

buf_mgr_ = new (buf) ObHashJoinBufMgr();

}

}

if (OB_SUCC(ret)) {

void* buf = alloc_->alloc(sizeof(ObHashJoinBatchMgr));

if (NULL == buf) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("fail to alloc mem");

} else {

batch_mgr_ = new (buf) ObHashJoinBatchMgr(*alloc_, buf_mgr_, tenant_id_);

}

}

if (OB_SUCC(ret)) {

if (OB_FAIL(right_last_row_.init(*alloc_, right_->get_spec().output_.count()))) {

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

}

}

if (OB_SUCC(ret) && is_vectorized()) {

const int64_t batch_size = MY_SPEC.max_batch_size_;

OZ(alloc_ptrs(mem_context_->get_arena_allocator(),

hj_part_stored_rows_, sizeof(*hj_part_stored_rows_) * batch_size,

hash_vals_, sizeof(*hash_vals_) * batch_size,

right_hj_part_stored_rows_, sizeof(*right_hj_part_stored_rows_) * batch_size,

right_hash_vals_, sizeof(*right_hash_vals_) * batch_size,

cur_tuples_, sizeof(*cur_tuples_) * batch_size,

child_brs_.skip_, ObBitVector::memory_size(batch_size),

hj_part_added_rows_, sizeof(hj_part_added_rows_) * batch_size,

right_selector_, sizeof(*right_selector_) * batch_size));

}

cur_hash_table_ = &hash_table_;

return ret;

}

int ObHashJoinOp::set_shared_info()

{

int ret = OB_SUCCESS;

ObHashJoinInput *hj_input = static_cast<ObHashJoinInput*>(input_);

if (!MY_SPEC.is_shared_ht_) {

// none shared, nothing to do

} else if (OB_ISNULL(hj_input)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected status: hash join input is null", K(ret));

} else if (0 == hj_input->shared_hj_info_ || OB_ISNULL(hj_input->get_shared_hj_info())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected status: hash join shared info is null", K(ret));

} else {

// only more thant one thread, use shared hash join

is_shared_ = 1 < hj_input->get_sqc_thread_count();

if (is_shared_) {

if (IS_LEFT_STYLE_JOIN(MY_SPEC.join_type_)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("failed shared hash join not support", K(ret), K(MY_SPEC.join_type_));

} else if (hj_input->task_id_ >= hj_input->get_sqc_thread_count()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("task_id is more than thread count", K(ret),

K(hj_input->task_id_), K(hj_input->get_sqc_thread_count()));

} else {

LOG_TRACE("debug enable shared hash join", K(ret), K(spec_.id_));

}

}

}

return ret;

}

void ObHashJoinOp::reset_base()

{

drain_mode_ = HashJoinDrainMode::NONE_DRAIN;

hj_state_ = INIT;

hj_processor_ = NONE;

part_level_ = 0;

part_shift_ = MAX_PART_LEVEL << 3;

part_count_ = 0;

input_size_ = 0;

predict_row_cnt_ = 1024;

left_batch_ = nullptr;

right_batch_ = nullptr;

dumped_fixed_mem_size_ = 0;

}

void ObHashJoinOp::reset()

{

free_bloom_filter();

clean_batch_mgr();

part_rescan();

reset_base();

}

void ObHashJoinOp::part_rescan()

{

state_ = JS_READ_RIGHT;

cur_right_hash_value_ = 0;

right_has_matched_ = false;

tuple_need_join_ = false;

first_get_row_ = true;

cur_bkid_ = 0;

hash_table_.reset();

cur_tuple_ = NULL;

if (nullptr != bloom_filter_) {

bloom_filter_->reset();

}

right_bit_set_.reset();

nest_loop_state_ = HJLoopState::LOOP_START;

is_last_chunk_ = false;

has_right_bitset_ = false;

remain_data_memory_size_ = 0;

reset_nest_loop();

hj_processor_ = NONE;

cur_dumped_partition_ = max_partition_count_per_level_;

reset_statistics();

predict_row_cnt_ = 1024;

if (hj_part_array_ != NULL) {

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

hj_part_array_[i].~ObHashJoinPartition();

}

alloc_->free(hj_part_array_);

hj_part_array_ = NULL;

}

if (right_hj_part_array_ != NULL) {

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

right_hj_part_array_[i].~ObHashJoinPartition();

}

alloc_->free(right_hj_part_array_);

right_hj_part_array_ = NULL;

}

int64_t tmp_part_count = 0 < level2_part_count_ ?

level1_part_count_ * level2_part_count_ :

level1_part_count_;

if (OB_NOT_NULL(part_histograms_)) {

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

part_histograms_[i].~HashJoinHistogram();

}

alloc_->free(part_histograms_);

part_histograms_ = NULL;

}

if (OB_NOT_NULL(part_selectors_)) {

alloc_->free(part_selectors_);

part_selectors_ = nullptr;

part_selector_sizes_ = nullptr;

}

left_read_row_ = NULL;

right_read_row_ = NULL;

postprocessed_left_ = false;

has_fill_right_row_ = false;

has_fill_left_row_ = false;

need_return_ = false;

opt_cache_aware_ = false;

has_right_material_data_ = false;

cur_full_right_partition_ = INT64_MAX;

right_iter_end_ = false;

cur_bucket_idx_ = 0;

max_bucket_idx_ = 0;

enable_batch_ = false;

level1_bit_ = 0;

level1_part_count_ = 0;

level2_part_count_ = 0;

cur_left_hist_ = nullptr;

cur_right_hist_ = nullptr;

right_splitter_.reset();

cur_probe_row_idx_ = 0;

max_right_bucket_idx_ = 0;

right_read_from_stored_ = false;

right_batch_traverse_cnt_ = 0;

null_random_hash_value_ = 0;

}

int ObHashJoinOp::part_rescan(bool reset_all)

{

int ret = OB_SUCCESS;

ObSQLSessionInfo *session = NULL;

if (OB_ISNULL(session = ctx_.get_my_session())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("fail to get my session", K(ret));

} else {

if (reset_all) {

reset();

part_count_ = 0;

} else {

part_rescan();

}

if (OB_SUCC(ret)) {

// reset join ctx

left_row_joined_ = false;

}

}

return ret;

}

int ObHashJoinOp::inner_rescan()

{

int ret = OB_SUCCESS;

if (OB_FAIL(part_rescan(true))) {

LOG_WARN("part rescan failed", K(ret));

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

LOG_WARN("join rescan failed", K(ret));

} else {

batch_round_ = 1;

iter_end_ = false;

read_null_in_naaj_ = false;

non_preserved_side_is_not_empty_ = false;

}

LOG_TRACE("hash join rescan", K(ret));

return ret;

}

int ObHashJoinOp::do_sync_wait_all()

{

int ret = OB_SUCCESS;

if (!batch_reach_end_ && !row_reach_end_ && is_shared_) {

// In shared hash join, it only has two state: one is that it don't build hash table,

// and one is that it return result after probe

// so we only do:

// 1. sync to build hash table

// 2. set join_end and sync to do next_batch

// don't get any right data, we think the right is iter_end

if (ObHashJoinOp::HJState::INIT == hj_state_) {

// don't build hash table

// it must build hash table to guarantee to hold all data

drain_mode_ = HashJoinDrainMode::BUILD_HT;

} else if (ObHashJoinOp::HJState::NORMAL == hj_state_) {

// set state_ = JOIN_END and going to get_next_row/batch_row

state_ = JS_JOIN_END;

drain_mode_ = HashJoinDrainMode::RIGHT_DRAIN;

if (OB_FAIL(right_->drain_exch())) {

LOG_WARN("drain exch failed", K(ret));

}

} else {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected status: hash join state", K(state_), K(hj_state_));

}

if (OB_SUCC(ret)) {

if (!is_vectorized()) {

while (OB_SUCC(ret)) {

if (OB_FAIL(get_next_row())) {

if (OB_ITER_END == ret) {

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

}

}

} // end while

} else {

// sync others thread

const ObBatchRows *child_brs = &child_brs_;

while (OB_SUCC(ret)) {

if (OB_FAIL(get_next_batch(MY_SPEC.max_batch_size_, child_brs))) {

if (OB_ITER_END == ret) {

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

}

} else if (brs_.end_) {

break;

}

} // end while

}

if (OB_ITER_END == ret) {

ret = OB_SUCCESS;

}

}

LOG_TRACE("debug shared hash join wait all to drain", K(ret));

}

return ret;

}

// need add sync to wait exit for shared hash join

int ObHashJoinOp::inner_drain_exch()

{

return do_sync_wait_all();

}

int ObHashJoinOp::next()

{

int ret = OB_SUCCESS;

state_operation_func_type state_operation = NULL;

state_function_func_type state_function = NULL;

ObJoinState &state = state_;

int func = -1;

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

state_operation = this->ObHashJoinOp::state_operation_func_[state];

if (OB_ISNULL(state_operation)) {

ret = OB_BAD_NULL_ERROR;

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

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

func = FT_ITER_END;

ret = OB_SUCCESS;

} else if (OB_FAIL(ret)) {

LOG_WARN("failed state operation", K(ret), K(state));

} else {

func = FT_ITER_GOING;

}

if (OB_SUCC(ret)) {

state_function = this->ObHashJoinOp::state_function_func_[state][func];

if (OB_FAIL((this->*state_function)())) {

if (OB_ITER_END == ret) {

//iter_end ,break;

} else {

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

}

}

}

} //while end

return ret;

}

// 这里有几个逻辑需要理下：

// 1）当返回一行时，如果left对应bucket没有元素，即cur_tuple为nullpr，会继续get_right_row，所以不需要save

// 2）如果不会拿下一个right，则需要restore上次保存的right，继续看是否匹配left其他行

// 这里有两种情况：

// a）如果没有dump，恢复output

// b）如果dump，则仅恢复right_read_row_，至于是否放入到right output，看下一次是否返回避免多余copy

int ObHashJoinOp::inner_get_next_row()

{

int ret = OB_SUCCESS;

bool exit_while = false;

need_return_ = false;

if (is_vectorized()) {

if (OB_UNLIKELY(iter_end_)) {

brs_.size_ = 0;

brs_.end_ = true;

exit_while = true;

}

} else {

has_fill_left_row_ = false;

}

clear_evaluated_flag();

if (OB_UNLIKELY(iter_end_)) {

ret = OB_ITER_END;

}

while (OB_SUCCESS == ret && !exit_while) {

switch (hj_state_) {

case ObHashJoinOp::HJState::INIT: {

hj_state_ = ObHashJoinOp::HJState::NORMAL;

break;

}

case ObHashJoinOp::HJState::NORMAL: {

ret = next();

if (OB_ITER_END == ret) {

hj_state_ = ObHashJoinOp::HJState::NEXT_BATCH;

ret = OB_SUCCESS;

} else if (OB_SUCCESS == ret) {

exit_while = true;

} else {

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

}

break;

}

case ObHashJoinOp::HJState::NEXT_BATCH: {

// It must firstly sync wait, and then remove undumped batch

if (is_shared_ && OB_FAIL(sync_wait_fetch_next_batch())) {

LOG_WARN("failed to sync wait fetch next batch", K(ret));

} else {

batch_mgr_->remove_undumped_batch(is_shared_ ? cur_dumped_partition_ : INT64_MAX, batch_round_);

if (left_batch_ != NULL) {

left_batch_->close();

batch_mgr_->free(left_batch_);

left_batch_ = NULL;

}

if (right_batch_ != NULL) {

right_batch_->close();

batch_mgr_->free(right_batch_);

right_batch_ = NULL;

}

}

ObHashJoinBatchPair batch_pair;

if (OB_FAIL(ret)) {

} else if (OB_FAIL(part_rescan(false))) {

LOG_WARN("fail to reopen hj", K(ret));

} else if (read_null_in_naaj_) {

// if read null in naaj, return the empty set directly

ret = OB_ITER_END;

} else if (OB_FAIL(batch_mgr_->next_batch(batch_pair))) {

} else if (0 != buf_mgr_->get_total_alloc_size()) {

LOG_WARN("expect memory count is ok", K(ret), K(buf_mgr_->get_total_alloc_size()));

}

dumped_fixed_mem_size_ = get_cur_mem_used();

if (OB_ITER_END == ret) {

exit_while = true;

// free resource like memory

iter_end_ = true;

part_rescan(true);

if (is_vectorized()) {

brs_.size_ = 0;

brs_.end_ = true;

ret = OB_SUCCESS;

}

LOG_TRACE("debug iter end", K(spec_.id_));

} else if (OB_SUCCESS == ret) {

++batch_round_;

left_batch_ = batch_pair.left_;

right_batch_ = batch_pair.right_;

part_level_ = batch_pair.left_->get_part_level() + 1;

part_shift_ = batch_pair.left_->get_part_shift();

// asynchronously wait to write while only read the dumped partition

if (OB_FAIL(left_batch_->get_chunk_row_store().finish_add_row(true))) {

LOG_WARN("finish dump failed", K(ret));

} else if (OB_FAIL(right_batch_->get_chunk_row_store().finish_add_row(true))) {

LOG_WARN("finish dump failed", K(ret));

}

batch_pair.left_->open();

batch_pair.right_->open();

if (sizeof(uint64_t) * CHAR_BIT <= part_shift_) {

// avoid loop recursively

// 至多MAX_PART_LEVEL,最后一层要么nest loop，要么in-memory

// hash join dumped too many times, the part level is greater than 32 bit

// we report 4013 instead of 4016, and remind user to increase memory

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("too deep part level", K(ret), K(part_level_), K(part_shift_));

} else {

hj_state_ = ObHashJoinOp::HJState::NORMAL;

}

LOG_DEBUG("trace batch", K(batch_pair.left_->get_batchno()),

K(batch_pair.right_->get_batchno()), K(part_level_), K(batch_round_));

} else {

LOG_WARN("fail get next batch", K(ret));

}

break;

}

}

}

clear_evaluated_flag();

return ret;

}

void ObHashJoinOp::destroy()

{

sql_mem_processor_.unregister_profile_if_necessary();

if (OB_LIKELY(nullptr != alloc_)) {

alloc_ = nullptr;

}

if (OB_LIKELY(NULL != mem_context_)) {

DESTROY_CONTEXT(mem_context_);

mem_context_ = NULL;

}

ObJoinOp::destroy();

}

int ObHashJoinOp::inner_close()

{

int ret = OB_SUCCESS;

sql_mem_processor_.unregister_profile();

if (is_shared_) {

IGNORE_RETURN sync_wait_close();

}

reset();

tmp_hash_funcs_.reset();

if (batch_mgr_ != NULL) {

batch_mgr_->~ObHashJoinBatchMgr();

if (OB_NOT_NULL(alloc_)) {

alloc_->free(batch_mgr_);

}

batch_mgr_ = NULL;

}

if (buf_mgr_ != NULL) {

buf_mgr_->~ObHashJoinBufMgr();

if (OB_NOT_NULL(alloc_)) {

alloc_->free(buf_mgr_);

}

buf_mgr_ = NULL;

}

right_last_row_.reset();

if (nullptr != alloc_) {

hash_table_.free(alloc_);

}

if (nullptr != mem_context_) {

mem_context_->reuse();

}

if (OB_FAIL(ObJoinOp::inner_close())) {

}

return ret;

}

int ObHashJoinOp::join_end_operate()

{

return OB_ITER_END;

}

int ObHashJoinOp::join_end_func_end()

{

int ret = OB_SUCCESS;

LOG_TRACE("trace hash join probe statistics", K(bitset_filter_cnt_), K(probe_cnt_),

K(hash_equal_cnt_), K(hash_link_cnt_));

// nest loop process one block, and need next block

if (HJProcessor::NEST_LOOP == hj_processor_

&& HJLoopState::LOOP_GOING == nest_loop_state_ && !read_null_in_naaj_ && !is_shared_) {

state_ = JS_READ_RIGHT;

first_get_row_ = true;

} else {

ret = OB_ITER_END;

}

return ret;

}

int ObHashJoinOp::get_next_left_row()

{

int ret = common::OB_SUCCESS;

left_row_joined_ = false;

if (left_batch_ == NULL) {

if (OB_FAIL(OB_I(t1) left_->get_next_row())) {

if (OB_ITER_END != ret) {

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

}

}

} else {

if (OB_FAIL(try_check_status())) {

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