/*

* Copyright 2017 MapD Technologies, Inc.

*

* Licensed under the Apache License, Version 2.0 (the "License");

* you may not use this file except in compliance with the License.

* You may obtain a copy of the License at

*

* http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*/

#include "JoinHashTable.h"

#include "ThrustAllocator.h"

#include "Execute.h"

#include "ExpressionRewrite.h"

#include "HashJoinRuntime.h"

#include "RuntimeFunctions.h"

#include <glog/logging.h>

#include <numeric>

#include <thread>

namespace {

std::pair<const Analyzer::ColumnVar*, const Analyzer::ColumnVar*> get_cols(

const std::shared_ptr<Analyzer::BinOper> qual_bin_oper,

const Catalog_Namespace::Catalog& cat,

const TemporaryTables* temporary_tables) {

const auto lhs = qual_bin_oper->get_left_operand();

const auto rhs = qual_bin_oper->get_right_operand();

const auto& lhs_ti = lhs->get_type_info();

const auto& rhs_ti = rhs->get_type_info();

if (lhs_ti.get_type() != rhs_ti.get_type()) {

throw HashJoinFail("Equijoin types must be identical, found: " + lhs_ti.get_type_name() + ", " +

rhs_ti.get_type_name());

}

if (!lhs_ti.is_integer() && !lhs_ti.is_string()) {

throw HashJoinFail("Cannot apply hash join to " + lhs_ti.get_type_name());

}

const auto lhs_cast = dynamic_cast<const Analyzer::UOper*>(lhs);

const auto rhs_cast = dynamic_cast<const Analyzer::UOper*>(rhs);

if (static_cast<bool>(lhs_cast) != static_cast<bool>(rhs_cast) || (lhs_cast && lhs_cast->get_optype() != kCAST) ||

(rhs_cast && rhs_cast->get_optype() != kCAST)) {

throw HashJoinFail("Cannot use hash join for given expression");

}

CHECK_EQ(kENCODING_NONE, lhs_ti.get_compression());

CHECK_EQ(kENCODING_NONE, rhs_ti.get_compression());

const auto lhs_col = lhs_cast ? dynamic_cast<const Analyzer::ColumnVar*>(lhs_cast->get_operand())

: dynamic_cast<const Analyzer::ColumnVar*>(lhs);

const auto rhs_col = rhs_cast ? dynamic_cast<const Analyzer::ColumnVar*>(rhs_cast->get_operand())

: dynamic_cast<const Analyzer::ColumnVar*>(rhs);

if (!lhs_col || !rhs_col) {

throw HashJoinFail("Cannot use hash join for given expression");

}

const Analyzer::ColumnVar* inner_col{nullptr};

const Analyzer::ColumnVar* outer_col{nullptr};

if (lhs_col->get_rte_idx() == 0 && rhs_col->get_rte_idx() == 1) {

inner_col = rhs_col;

outer_col = lhs_col;

} else {

CHECK((lhs_col->get_rte_idx() == 1 && rhs_col->get_rte_idx() == 0));

inner_col = lhs_col;

outer_col = rhs_col;

}

// We need to fetch the actual type information from the catalog since Analyzer

// always reports nullable as true for inner table columns in left joins.

const auto inner_col_cd = get_column_descriptor_maybe(inner_col->get_column_id(), inner_col->get_table_id(), cat);

const auto& inner_col_real_ti =

get_column_type(inner_col->get_column_id(), inner_col->get_table_id(), inner_col_cd, temporary_tables);

const auto& outer_col_ti = outer_col->get_type_info();

if (outer_col_ti.get_notnull() != inner_col_real_ti.get_notnull()) {

throw HashJoinFail("For hash join, both sides must have the same nullability");

}

if (!(inner_col_real_ti.is_integer() ||

(inner_col_real_ti.is_string() && inner_col_real_ti.get_compression() == kENCODING_DICT))) {

throw HashJoinFail("Can only apply hash join to integer-like types and dictionary encoded strings");

}

return {inner_col, outer_col};

}

} // namespace

std::vector<std::pair<JoinHashTable::JoinHashTableCacheKey, std::shared_ptr<std::vector<int32_t>>>>

JoinHashTable::join_hash_table_cache_;

std::mutex JoinHashTable::join_hash_table_cache_mutex_;

size_t get_shard_count(const Analyzer::BinOper* join_condition, const Catalog_Namespace::Catalog& catalog) {

return 0; // Sharded joins aren't enabled yet.

}

std::shared_ptr<JoinHashTable> JoinHashTable::getInstance(

const std::shared_ptr<Analyzer::BinOper> qual_bin_oper,

const Catalog_Namespace::Catalog& cat,

const std::vector<InputTableInfo>& query_infos,

const std::list<std::shared_ptr<const InputColDescriptor>>& input_col_descs,

const Data_Namespace::MemoryLevel memory_level,

const int device_count,

Executor* executor) {

CHECK_EQ(kEQ, qual_bin_oper->get_optype());

const auto redirected_bin_oper =

std::dynamic_pointer_cast<Analyzer::BinOper>(redirect_expr(qual_bin_oper.get(), input_col_descs));

CHECK(redirected_bin_oper);

const auto cols = get_cols(redirected_bin_oper, cat, executor->temporary_tables_);

const auto inner_col = cols.first;

CHECK(inner_col);

const auto& ti = inner_col->get_type_info();

auto col_range = getExpressionRange(ti.is_string() ? cols.second : inner_col, query_infos, executor);

if (col_range.getType() == ExpressionRangeType::Invalid) {

throw HashJoinFail("Could not compute range for the expressions involved in the equijoin");

}

if (ti.is_string()) {

// The nullable info must be the same as the source column.

const auto source_col_range = getExpressionRange(inner_col, query_infos, executor);

if (source_col_range.getType() == ExpressionRangeType::Invalid) {

throw HashJoinFail("Could not compute range for the expressions involved in the equijoin");

}

col_range = ExpressionRange::makeIntRange(std::min(source_col_range.getIntMin(), col_range.getIntMin()),

std::max(source_col_range.getIntMax(), col_range.getIntMax()),

0,

source_col_range.hasNulls());

}

auto join_hash_table = std::shared_ptr<JoinHashTable>(

new JoinHashTable(qual_bin_oper, inner_col, cat, query_infos, memory_level, col_range, executor, device_count));

const int err = join_hash_table->reify(device_count);

if (err) {

#ifndef ENABLE_MULTIFRAG_JOIN

if (err == ERR_MULTI_FRAG) {

const auto cols = get_cols(qual_bin_oper, cat, executor->temporary_tables_);

const auto inner_col = cols.first;

CHECK(inner_col);

const auto& table_info = join_hash_table->getInnerQueryInfo(inner_col);

throw HashJoinFail("Multi-fragment inner table '" + get_table_name_by_id(table_info.table_id, cat) +

"' not supported yet");

}

#endif

if (err == ERR_FAILED_TO_FETCH_COLUMN) {

throw HashJoinFail("Not enough memory for the columns involved in join");

}

if (err == ERR_FAILED_TO_JOIN_ON_VIRTUAL_COLUMN) {

throw HashJoinFail("Cannot join on rowid");

}

throw HashJoinFail("Could not build a 1-to-1 correspondence for columns involved in equijoin");

}

return join_hash_table;

}

std::pair<const int8_t*, size_t> JoinHashTable::getColumnFragment(

const Analyzer::ColumnVar& hash_col,

const Fragmenter_Namespace::FragmentInfo& fragment,

const Data_Namespace::MemoryLevel effective_mem_lvl,

const int device_id,

std::vector<std::shared_ptr<Chunk_NS::Chunk>>& chunks_owner,

std::map<int, std::shared_ptr<const ColumnarResults>>& frags_owner) {

auto chunk_meta_it = fragment.getChunkMetadataMap().find(hash_col.get_column_id());

CHECK(chunk_meta_it != fragment.getChunkMetadataMap().end());

const auto cd = get_column_descriptor_maybe(hash_col.get_column_id(), hash_col.get_table_id(), cat_);

CHECK(!cd || !(cd->isVirtualCol));

const int8_t* col_buff = nullptr;

if (cd) {

ChunkKey chunk_key{

cat_.get_currentDB().dbId, hash_col.get_table_id(), hash_col.get_column_id(), fragment.fragmentId};

const auto chunk = Chunk_NS::Chunk::getChunk(cd,

&cat_.get_dataMgr(),

chunk_key,

effective_mem_lvl,

effective_mem_lvl == Data_Namespace::CPU_LEVEL ? 0 : device_id,

chunk_meta_it->second.numBytes,

chunk_meta_it->second.numElements);

chunks_owner.push_back(chunk);

CHECK(chunk);

auto ab = chunk->get_buffer();

CHECK(ab->getMemoryPtr());

col_buff = reinterpret_cast<int8_t*>(ab->getMemoryPtr());

} else {

const auto frag_id = fragment.fragmentId;

auto frag_it = frags_owner.find(frag_id);

if (frag_it == frags_owner.end()) {

std::shared_ptr<const ColumnarResults> col_frag(

columnarize_result(executor_->row_set_mem_owner_,

get_temporary_table(executor_->temporary_tables_, hash_col.get_table_id()),

frag_id));

auto res = frags_owner.insert(std::make_pair(frag_id, col_frag));

CHECK(res.second);

frag_it = res.first;

}

col_buff = Executor::ExecutionDispatch::getColumn(frag_it->second.get(),

hash_col.get_column_id(),

&cat_.get_dataMgr(),

effective_mem_lvl,

effective_mem_lvl == Data_Namespace::CPU_LEVEL ? 0 : device_id);

}

return {col_buff, fragment.getNumTuples()};

}

std::pair<const int8_t*, size_t> JoinHashTable::getAllColumnFragments(

const Analyzer::ColumnVar& hash_col,

const std::deque<Fragmenter_Namespace::FragmentInfo>& fragments,

std::vector<std::shared_ptr<Chunk_NS::Chunk>>& chunks_owner,

std::map<int, std::shared_ptr<const ColumnarResults>>& frags_owner) {

CHECK(!fragments.empty());

const size_t elem_width = hash_col.get_type_info().get_size();

std::vector<const int8_t*> col_frags;

std::vector<size_t> elem_counts;

for (auto& frag : fragments) {

const int8_t* col_frag = nullptr;

size_t elem_count = 0;

std::tie(col_frag, elem_count) =

getColumnFragment(hash_col, frag, Data_Namespace::CPU_LEVEL, 0, chunks_owner, frags_owner);

CHECK(col_frag != nullptr);

CHECK_NE(elem_count, size_t(0));

col_frags.push_back(col_frag);

elem_counts.push_back(elem_count);

}

CHECK(!col_frags.empty());

CHECK_EQ(col_frags.size(), elem_counts.size());

const auto total_elem_count = std::accumulate(elem_counts.begin(), elem_counts.end(), size_t(0));

auto col_buff = reinterpret_cast<int8_t*>(checked_malloc(total_elem_count * elem_width));

for (size_t i = 0, offset = 0; i < col_frags.size(); ++i) {

memcpy(col_buff + offset, col_frags[i], elem_counts[i] * elem_width);

offset += elem_counts[i] * elem_width;

}

return {col_buff, total_elem_count};

}

int JoinHashTable::reify(const int device_count) {

CHECK_LT(0, device_count);

const auto cols = get_cols(qual_bin_oper_, cat_, executor_->temporary_tables_);

const auto inner_col = cols.first;

CHECK(inner_col);

const auto& query_info = getInnerQueryInfo(inner_col).info;

if (query_info.fragments.empty()) {

return 0;

}

#ifndef ENABLE_MULTIFRAG_JOIN

if (query_info.fragments.size() != 1) { // we don't support multiple fragment inner tables (yet)

return ERR_MULTI_FRAG;

}

#else

const bool has_multi_frag = query_info.fragments.size() > 1;

#endif

const auto cd = get_column_descriptor_maybe(inner_col->get_column_id(), inner_col->get_table_id(), cat_);

if (cd && cd->isVirtualCol) {

return ERR_FAILED_TO_JOIN_ON_VIRTUAL_COLUMN;

}

CHECK(!cd || !(cd->isVirtualCol));

const auto& ti =

get_column_type(inner_col->get_column_id(), inner_col->get_table_id(), cd, executor_->temporary_tables_);

// Since we don't have the string dictionary payloads on the GPU, we'll build

// the join hash table on the CPU and transfer it to the GPU.

const auto effective_memory_level = ti.is_string() ? Data_Namespace::CPU_LEVEL : memory_level_;

#ifdef HAVE_CUDA

gpu_hash_table_buff_.resize(device_count);

#endif

std::vector<int> errors(device_count);

std::vector<std::thread> init_threads;

std::vector<std::shared_ptr<Chunk_NS::Chunk>> chunks_owner;

std::map<int, std::shared_ptr<const ColumnarResults>> frags_owner;

const auto& first_frag = query_info.fragments.front();

ChunkKey chunk_key{cat_.get_currentDB().dbId, inner_col->get_table_id(), inner_col->get_column_id()};

const int8_t* col_buff = nullptr;

size_t elem_count = 0;

#ifdef ENABLE_MULTIFRAG_JOIN

const size_t elem_width = inner_col->get_type_info().get_size();

auto& data_mgr = cat_.get_dataMgr();

RowSetMemoryOwner col_buff_owner;

std::vector<ThrustAllocator> dev_buff_owner;

if (has_multi_frag) {

try {

std::tie(col_buff, elem_count) =

getAllColumnFragments(*inner_col, query_info.fragments, chunks_owner, frags_owner);

col_buff_owner.addColBuffer(col_buff);

} catch (...) {

return ERR_FAILED_TO_FETCH_COLUMN;

}

} else

#endif

{

chunk_key.push_back(first_frag.fragmentId);

}

for (int device_id = 0; device_id < device_count; ++device_id) {

#ifdef ENABLE_MULTIFRAG_JOIN

dev_buff_owner.emplace_back(&data_mgr, device_id);

if (has_multi_frag) {

if (effective_memory_level == Data_Namespace::GPU_LEVEL) {

CHECK(col_buff != nullptr);

CHECK_NE(elem_count, size_t(0));

int8_t* dev_col_buff = nullptr;

dev_col_buff = dev_buff_owner[device_id].allocate(elem_count * elem_width);

copy_to_gpu(

&data_mgr, reinterpret_cast<CUdeviceptr>(dev_col_buff), col_buff, elem_count * elem_width, device_id);

col_buff = dev_col_buff;

}

} else

#endif

{

try {

std::tie(col_buff, elem_count) =

getColumnFragment(*inner_col, first_frag, effective_memory_level, device_id, chunks_owner, frags_owner);

} catch (...) {

return ERR_FAILED_TO_FETCH_COLUMN;

}

}

init_threads.emplace_back(

[&errors, &chunk_key, &cols, elem_count, col_buff, effective_memory_level, device_id, this] {

try {

errors[device_id] =

initHashTableForDevice(chunk_key, col_buff, elem_count, cols, effective_memory_level, device_id);

} catch (...) {

errors[device_id] = -1;

}

});

}

for (auto& init_thread : init_threads) {

init_thread.join();

}

for (const int err : errors) {

if (err) {

return err;

}

}

return 0;

}

int JoinHashTable::initHashTableOnCpu(const int8_t* col_buff,

const size_t num_elements,

const std::pair<const Analyzer::ColumnVar*, const Analyzer::ColumnVar*>& cols,

const int32_t hash_entry_count,

const int32_t hash_join_invalid_val) {

const auto inner_col = cols.first;

CHECK(inner_col);

const auto& ti = inner_col->get_type_info();

int err = 0;

if (!cpu_hash_table_buff_) {

cpu_hash_table_buff_ = std::make_shared<std::vector<int32_t>>(hash_entry_count);

const StringDictionaryProxy* sd_inner_proxy{nullptr};

const StringDictionaryProxy* sd_outer_proxy{nullptr};

if (ti.is_string()) {

CHECK_EQ(kENCODING_DICT, ti.get_compression());

sd_inner_proxy =

executor_->getStringDictionaryProxy(inner_col->get_comp_param(), executor_->row_set_mem_owner_, true);

CHECK(sd_inner_proxy);

sd_outer_proxy =

executor_->getStringDictionaryProxy(cols.second->get_comp_param(), executor_->row_set_mem_owner_, true);

CHECK(sd_outer_proxy);

}

int thread_count = cpu_threads();

std::vector<std::thread> init_cpu_buff_threads;

for (int thread_idx = 0; thread_idx < thread_count; ++thread_idx) {

init_cpu_buff_threads.emplace_back([this, hash_entry_count, hash_join_invalid_val, thread_idx, thread_count] {

init_hash_join_buff(

&(*cpu_hash_table_buff_)[0], hash_entry_count, hash_join_invalid_val, thread_idx, thread_count);

});

}

for (auto& t : init_cpu_buff_threads) {

t.join();

}

init_cpu_buff_threads.clear();

for (int thread_idx = 0; thread_idx < thread_count; ++thread_idx) {

init_cpu_buff_threads.emplace_back([this,

hash_join_invalid_val,

col_buff,

num_elements,

sd_inner_proxy,

sd_outer_proxy,

thread_idx,

thread_count,

&ti,

&err] {

int partial_err = fill_hash_join_buff(&(*cpu_hash_table_buff_)[0],

hash_join_invalid_val,

{col_buff, num_elements},

{static_cast<size_t>(ti.get_size()),

col_range_.getIntMin(),

inline_fixed_encoding_null_val(ti),

col_range_.getIntMax() + 1},

sd_inner_proxy,

sd_outer_proxy,

thread_idx,

thread_count);

__sync_val_compare_and_swap(&err, 0, partial_err);

});

}

for (auto& t : init_cpu_buff_threads) {

t.join();

}

if (err) {

cpu_hash_table_buff_.reset();

}

}

return err;

}

namespace {

#ifdef HAVE_CUDA

// Number of entries per shard, rounded up.

size_t get_entries_per_shard(const size_t total_entry_count, const size_t shard_count) {

CHECK_NE(size_t(0), shard_count);

return (total_entry_count + shard_count - 1) / shard_count;

}

#endif // HAVE_CUDA

// Number of entries required for the given range.

size_t get_hash_entry_count(const ExpressionRange& col_range) {

CHECK_LE(col_range.getIntMin(), col_range.getIntMax());

return col_range.getIntMax() - col_range.getIntMin() + 1 + (col_range.hasNulls() ? 1 : 0);

}

} // namespace

int JoinHashTable::initHashTableForDevice(const ChunkKey& chunk_key,

const int8_t* col_buff,

const size_t num_elements,

const std::pair<const Analyzer::ColumnVar*, const Analyzer::ColumnVar*>& cols,

const Data_Namespace::MemoryLevel effective_memory_level,

const int device_id) {

auto hash_entry_count = get_hash_entry_count(col_range_);

#ifdef HAVE_CUDA

const auto shard_count = get_shard_count(qual_bin_oper_.get(), cat_);

const size_t entries_per_shard{shard_count ? get_entries_per_shard(hash_entry_count, shard_count) : 0};

// Even if we join on dictionary encoded strings, the memory on the GPU is still needed

// once the join hash table has been built on the CPU.

if (memory_level_ == Data_Namespace::GPU_LEVEL) {

auto& data_mgr = cat_.get_dataMgr();

if (shard_count) {

const auto shards_per_device = (shard_count + device_count_ - 1) / device_count_;

CHECK_GT(shards_per_device, 0);

hash_entry_count = entries_per_shard * shards_per_device;

}

gpu_hash_table_buff_[device_id] = alloc_gpu_mem(&data_mgr, hash_entry_count * sizeof(int32_t), device_id, nullptr);

}

#else

CHECK_EQ(Data_Namespace::CPU_LEVEL, effective_memory_level);

#endif

const auto inner_col = cols.first;

CHECK(inner_col);

#ifdef HAVE_CUDA

const auto& ti = inner_col->get_type_info();

#endif

int err = 0;

const int32_t hash_join_invalid_val{-1};

if (effective_memory_level == Data_Namespace::CPU_LEVEL) {

initHashTableOnCpuFromCache(chunk_key, num_elements, cols);

{

std::lock_guard<std::mutex> cpu_hash_table_buff_lock(cpu_hash_table_buff_mutex_);

err = initHashTableOnCpu(col_buff, num_elements, cols, hash_entry_count, hash_join_invalid_val);

}

if (!err && inner_col->get_table_id() > 0) {

putHashTableOnCpuToCache(chunk_key, num_elements, cols);

}

// Transfer the hash table on the GPU if we've only built it on CPU

// but the query runs on GPU (join on dictionary encoded columns).

// Don't transfer the buffer if there was an error since we'll bail anyway.

if (memory_level_ == Data_Namespace::GPU_LEVEL && !err) {

#ifdef HAVE_CUDA

CHECK(ti.is_string());

auto& data_mgr = cat_.get_dataMgr();

copy_to_gpu(&data_mgr,

gpu_hash_table_buff_[device_id],

&(*cpu_hash_table_buff_)[0],

cpu_hash_table_buff_->size() * sizeof((*cpu_hash_table_buff_)[0]),

device_id);

#else

CHECK(false);

#endif

}

} else {

#ifdef HAVE_CUDA

CHECK_EQ(Data_Namespace::GPU_LEVEL, effective_memory_level);

auto& data_mgr = cat_.get_dataMgr();

auto dev_err_buff = alloc_gpu_mem(&data_mgr, sizeof(int), device_id, nullptr);

copy_to_gpu(&data_mgr, dev_err_buff, &err, sizeof(err), device_id);

init_hash_join_buff_on_device(reinterpret_cast<int32_t*>(gpu_hash_table_buff_[device_id]),

hash_entry_count,

hash_join_invalid_val,

executor_->blockSize(),

executor_->gridSize());

JoinColumn join_column{col_buff, num_elements};

JoinColumnTypeInfo type_info{static_cast<size_t>(ti.get_size()),

col_range_.getIntMin(),

inline_fixed_encoding_null_val(ti),

col_range_.getIntMax() + 1};

if (shard_count) {

CHECK_GT(device_count_, 0);

for (size_t shard = device_id; shard < shard_count; shard += device_count_) {

ShardInfo shard_info{shard, entries_per_shard, shard_count, device_count_};

fill_hash_join_buff_on_device_sharded(reinterpret_cast<int32_t*>(gpu_hash_table_buff_[device_id]),

hash_join_invalid_val,

reinterpret_cast<int*>(dev_err_buff),

join_column,

type_info,

shard_info,

executor_->blockSize(),

executor_->gridSize());

}

} else {

fill_hash_join_buff_on_device(reinterpret_cast<int32_t*>(gpu_hash_table_buff_[device_id]),

hash_join_invalid_val,

reinterpret_cast<int*>(dev_err_buff),

join_column,

type_info,

executor_->blockSize(),

executor_->gridSize());

}

copy_from_gpu(&data_mgr, &err, dev_err_buff, sizeof(err), device_id);

#else

CHECK(false);

#endif

}

return err;

}

void JoinHashTable::initHashTableOnCpuFromCache(

const ChunkKey& chunk_key,

const size_t num_elements,

const std::pair<const Analyzer::ColumnVar*, const Analyzer::ColumnVar*>& cols) {

JoinHashTableCacheKey cache_key{col_range_, *cols.first, *cols.second, num_elements, chunk_key};

std::lock_guard<std::mutex> join_hash_table_cache_lock(join_hash_table_cache_mutex_);

for (const auto& kv : join_hash_table_cache_) {

if (kv.first == cache_key) {

cpu_hash_table_buff_ = kv.second;

break;

}

}

}

void JoinHashTable::putHashTableOnCpuToCache(

const ChunkKey& chunk_key,

const size_t num_elements,

const std::pair<const Analyzer::ColumnVar*, const Analyzer::ColumnVar*>& cols) {

JoinHashTableCacheKey cache_key{col_range_, *cols.first, *cols.second, num_elements, chunk_key};

std::lock_guard<std::mutex> join_hash_table_cache_lock(join_hash_table_cache_mutex_);

for (const auto& kv : join_hash_table_cache_) {

if (kv.first == cache_key) {

return;

}

}

join_hash_table_cache_.emplace_back(cache_key, cpu_hash_table_buff_);

}

llvm::Value* JoinHashTable::codegenSlot(const CompilationOptions& co) noexcept {

CHECK(executor_->plan_state_->join_info_.join_impl_type_ == Executor::JoinImplType::HashOneToOne);

const auto cols = get_cols(qual_bin_oper_, cat_, executor_->temporary_tables_);

auto key_col = cols.second;

CHECK(key_col);

auto val_col = cols.first;

CHECK(val_col);

const auto key_lvs = executor_->codegen(key_col, true, co);

CHECK_EQ(size_t(1), key_lvs.size());

CHECK(executor_->plan_state_->join_info_.join_hash_table_);

auto hash_ptr = get_arg_by_name(executor_->cgen_state_->row_func_, "join_hash_table");

std::vector<llvm::Value*> hash_join_idx_args{hash_ptr,

executor_->castToTypeIn(key_lvs.front(), 64),

executor_->ll_int(col_range_.getIntMin()),

executor_->ll_int(col_range_.getIntMax())};

const int shard_count = get_shard_count(qual_bin_oper_.get(), cat_);

if (shard_count) {

const auto hash_entry_count = get_hash_entry_count(col_range_);

const auto entry_count_per_shard = (hash_entry_count + shard_count - 1) / shard_count;

hash_join_idx_args.push_back(executor_->ll_int<uint32_t>(entry_count_per_shard));

hash_join_idx_args.push_back(executor_->ll_int<uint32_t>(shard_count));

hash_join_idx_args.push_back(executor_->ll_int<uint32_t>(device_count_));

}

if (col_range_.hasNulls()) {

hash_join_idx_args.push_back(executor_->ll_int(inline_fixed_encoding_null_val(key_col->get_type_info())));

}

std::string fname{"hash_join_idx"};

if (shard_count) {

fname += "_sharded";

}

if (col_range_.hasNulls()) {

fname += "_nullable";

}

const auto slot_lv = executor_->cgen_state_->emitCall(fname, hash_join_idx_args);

const auto it_ok = executor_->cgen_state_->scan_idx_to_hash_pos_.emplace(val_col->get_rte_idx(), slot_lv);

CHECK(it_ok.second);

const auto slot_valid_lv =

executor_->cgen_state_->ir_builder_.CreateICmp(llvm::ICmpInst::ICMP_SGE, slot_lv, executor_->ll_int(int64_t(0)));

return slot_valid_lv;

}

const InputTableInfo& JoinHashTable::getInnerQueryInfo(const Analyzer::ColumnVar* inner_col) {

ssize_t ti_idx = -1;

for (size_t i = 0; i < query_infos_.size(); ++i) {

if (inner_col->get_table_id() == query_infos_[i].table_id) {

ti_idx = i;

break;

}

}

CHECK_NE(ssize_t(-1), ti_idx);

return query_infos_[ti_idx];

}