BaselineJoinHashTableBuilder Class Reference

#include <BaselineHashTableBuilder.h>

Public Member Functions
	BaselineJoinHashTableBuilder ()=default
template<class KEY_HANDLER >
int	initHashTableOnCpu (KEY_HANDLER *key_handler, const CompositeKeyInfo &composite_key_info, const std::vector< JoinColumn > &join_columns, const std::vector< JoinColumnTypeInfo > &join_column_types, const std::vector< JoinBucketInfo > &join_bucket_info, const StrProxyTranslationMapsPtrsAndOffsets &str_proxy_translation_maps_ptrs_and_offsets, const size_t keyspace_entry_count, const size_t keys_for_all_rows, const HashType layout, const JoinType join_type, const size_t key_component_width, const size_t key_component_count, const RegisteredQueryHint &query_hint)
void	allocateDeviceMemory (const HashType layout, const size_t key_component_width, const size_t key_component_count, const size_t keyspace_entry_count, const size_t emitted_keys_count, const int device_id, const Executor *executor, const RegisteredQueryHint &query_hint)
template<class KEY_HANDLER >
int	initHashTableOnGpu (KEY_HANDLER *key_handler, const std::vector< JoinColumn > &join_columns, const HashType layout, const JoinType join_type, const size_t key_component_width, const size_t key_component_count, const size_t keyspace_entry_count, const size_t emitted_keys_count, const int device_id, const Executor *executor, const RegisteredQueryHint &query_hint)
std::unique_ptr < BaselineHashTable >	getHashTable ()
void	setHashLayout (HashType layout)
HashType	getHashLayout () const

Private Attributes
std::unique_ptr < BaselineHashTable >	hash_table_
HashType	layout_

Detailed Description

Definition at line 261 of file BaselineHashTableBuilder.h.

Constructor & Destructor Documentation

BaselineJoinHashTableBuilder::BaselineJoinHashTableBuilder ( )

default

Member Function Documentation

void BaselineJoinHashTableBuilder::allocateDeviceMemory ( const HashType  layout, const size_t  key_component_width, const size_t  key_component_count, const size_t  keyspace_entry_count, const size_t  emitted_keys_count, const int  device_id, const Executor *  executor, const RegisteredQueryHint &  query_hint  )

inline

Definition at line 503 of file BaselineHashTableBuilder.h.

References HashJoin::getHashTypeString(), hash_table_, RegisteredQueryHint::isHintRegistered(), kMaxJoinHashTableSize, HashJoin::layoutRequiresAdditionalBuffers(), RegisteredQueryHint::max_join_hash_table_size, OneToOne, UNREACHABLE, and VLOG.

Referenced by BaselineJoinHashTable::copyCpuHashTableToGpu(), and initHashTableOnGpu().

                                                                   {
#ifdef HAVE_CUDA
    const auto num_hash_entries =
        (key_component_count + (layout == HashType::OneToOne ? 1 : 0));
    const auto entry_size = num_hash_entries * key_component_width;
    const size_t one_to_many_hash_entries =
        HashJoin::layoutRequiresAdditionalBuffers(layout)
            ? 2 * keyspace_entry_count + emitted_keys_count
            : 0;
    const size_t hash_table_size =
        entry_size * keyspace_entry_count + one_to_many_hash_entries * sizeof(int32_t);

    if (query_hint.isHintRegistered(QueryHint::kMaxJoinHashTableSize) &&
        hash_table_size > query_hint.max_join_hash_table_size) {
      throw JoinHashTableTooBig(hash_table_size, query_hint.max_join_hash_table_size);
    }

    // We can't allocate more than 2GB contiguous memory on GPU and each entry is 4 bytes.
    if (hash_table_size > static_cast<size_t>(std::numeric_limits<int32_t>::max())) {
      throw TooManyHashEntries(
          "Hash tables for GPU requiring larger than 2GB contigious memory not supported "
          "yet");
    }

    VLOG(1) << "Initialize a GPU baseline hash table for device " << device_id
            << " with join type " << HashJoin::getHashTypeString(layout)
            << ", hash table size: " << hash_table_size << " Bytes"
            << ", # hash entries: " << num_hash_entries << ", entry_size: " << entry_size
            << ", # entries in the payload buffer: " << one_to_many_hash_entries
            << " (# non-null hash entries: " << key_component_count
            << ", # entries stored in the payload buffer: " << emitted_keys_count << ")";

    hash_table_ = std::make_unique<BaselineHashTable>(executor->getDataMgr(),
                                                      layout,
                                                      keyspace_entry_count,
                                                      emitted_keys_count,
                                                      hash_table_size,
                                                      device_id);
#else
    UNREACHABLE();
#endif
  }

Here is the call graph for this function:

Here is the caller graph for this function:

HashType BaselineJoinHashTableBuilder::getHashLayout ( ) const

inline

Definition at line 694 of file BaselineHashTableBuilder.h.

References layout_.

{ return layout_; }

std::unique_ptr<BaselineHashTable> BaselineJoinHashTableBuilder::getHashTable ( )

inline

Definition at line 690 of file BaselineHashTableBuilder.h.

References hash_table_.

Referenced by BaselineJoinHashTable::copyCpuHashTableToGpu(), and BaselineJoinHashTable::initHashTableForDevice().

{ return std::move(hash_table_); }

Here is the caller graph for this function:

template<class KEY_HANDLER >

int BaselineJoinHashTableBuilder::initHashTableOnCpu ( KEY_HANDLER *  key_handler, const CompositeKeyInfo &  composite_key_info, const std::vector< JoinColumn > &  join_columns, const std::vector< JoinColumnTypeInfo > &  join_column_types, const std::vector< JoinBucketInfo > &  join_bucket_info, const StrProxyTranslationMapsPtrsAndOffsets &  str_proxy_translation_maps_ptrs_and_offsets, const size_t  keyspace_entry_count, const size_t  keys_for_all_rows, const HashType  layout, const JoinType  join_type, const size_t  key_component_width, const size_t  key_component_count, const RegisteredQueryHint &  query_hint  )

inline

Definition at line 265 of file BaselineHashTableBuilder.h.

References ANTI, threading_serial::async(), CHECK, cpu_threads(), DEBUG_TIMER, DEBUG_TIMER_NEW_THREAD, fill_one_to_many_baseline_hash_table_32(), fill_one_to_many_baseline_hash_table_64(), HashJoin::getHashTypeString(), hash_table_, init_baseline_hash_join_buff_32(), init_baseline_hash_join_buff_64(), init_hash_join_buff(), RegisteredQueryHint::isHintRegistered(), kMaxJoinHashTableSize, HashJoin::layoutRequiresAdditionalBuffers(), RegisteredQueryHint::max_join_hash_table_size, OneToOne, SEMI, setHashLayout(), logger::thread_local_ids(), UNREACHABLE, VLOG, and WINDOW_FUNCTION_FRAMING.

Referenced by BaselineJoinHashTable::initHashTableForDevice(), and OverlapsJoinHashTable::initHashTableOnCpu().

                                                                {
    auto timer = DEBUG_TIMER(__func__);
    auto const entry_cnt = (key_component_count + (layout == HashType::OneToOne ? 1 : 0));
    auto const entry_size = entry_cnt * key_component_width;
    size_t const one_to_many_hash_entries =
        HashJoin::layoutRequiresAdditionalBuffers(layout)
            ? 2 * keyspace_entry_count +
                  (keys_for_all_rows *
                   (1 + (join_type == JoinType::WINDOW_FUNCTION_FRAMING)))
            : 0;
    size_t const hash_table_size =
        entry_size * keyspace_entry_count + one_to_many_hash_entries * sizeof(int32_t);

    if (query_hint.isHintRegistered(QueryHint::kMaxJoinHashTableSize) &&
        hash_table_size > query_hint.max_join_hash_table_size) {
      throw JoinHashTableTooBig(hash_table_size, query_hint.max_join_hash_table_size);
    }

    // We can't allocate more than 2GB contiguous memory on GPU and each entry is 4 bytes.
    if (hash_table_size > static_cast<size_t>(std::numeric_limits<int32_t>::max())) {
      throw TooManyHashEntries(
          "Hash tables for GPU requiring larger than 2GB contigious memory not supported "
          "yet");
    }
    const bool for_semi_join =
        (join_type == JoinType::SEMI || join_type == JoinType::ANTI) &&
        layout == HashType::OneToOne;

    hash_table_ = std::make_unique<BaselineHashTable>(
        layout, keyspace_entry_count, keys_for_all_rows, hash_table_size);
    VLOG(1) << "Initialize a CPU baseline hash table for join type "
            << HashJoin::getHashTypeString(layout)
            << ", hash table size: " << hash_table_size << " Bytes"
            << ", # hash entries: " << entry_cnt << ", entry_size: " << entry_size
            << ", # entries in the payload buffer: " << one_to_many_hash_entries
            << " (# non-null hash entries: " << keyspace_entry_count
            << ", # entries stored in the payload buffer: " << keys_for_all_rows << ")";
    auto cpu_hash_table_ptr = hash_table_->getCpuBuffer();
    int thread_count = cpu_threads();
    std::vector<std::future<void>> init_cpu_buff_threads;
    setHashLayout(layout);
    {
      auto timer_init = DEBUG_TIMER("CPU Baseline-Hash: init_baseline_hash_join_buff_32");
#ifdef HAVE_TBB
      switch (key_component_width) {
        case 4:
          init_baseline_hash_join_buff_tbb_32(cpu_hash_table_ptr,
                                              keyspace_entry_count,
                                              key_component_count,
                                              layout == HashType::OneToOne,
                                              -1);
          break;
        case 8:
          init_baseline_hash_join_buff_tbb_64(cpu_hash_table_ptr,
                                              keyspace_entry_count,
                                              key_component_count,
                                              layout == HashType::OneToOne,
                                              -1);
          break;
        default:
          CHECK(false);
      }
#else   // #ifdef HAVE_TBB
      for (int thread_idx = 0; thread_idx < thread_count; ++thread_idx) {
        init_cpu_buff_threads.emplace_back(std::async(
            std::launch::async,
            [keyspace_entry_count,
             key_component_count,
             key_component_width,
             thread_idx,
             thread_count,
             cpu_hash_table_ptr,
             layout,
             parent_thread_local_ids = logger::thread_local_ids()] {
              logger::LocalIdsScopeGuard lisg = parent_thread_local_ids.setNewThreadId();
              DEBUG_TIMER_NEW_THREAD(parent_thread_local_ids.thread_id_);
              switch (key_component_width) {
                case 4:
                  init_baseline_hash_join_buff_32(cpu_hash_table_ptr,
                                                  keyspace_entry_count,
                                                  key_component_count,
                                                  layout == HashType::OneToOne,
                                                  -1,
                                                  thread_idx,
                                                  thread_count);
                  break;
                case 8:
                  init_baseline_hash_join_buff_64(cpu_hash_table_ptr,
                                                  keyspace_entry_count,
                                                  key_component_count,
                                                  layout == HashType::OneToOne,
                                                  -1,
                                                  thread_idx,
                                                  thread_count);
                  break;
                default:
                  UNREACHABLE();
              }
            }));
      }
      for (auto& child : init_cpu_buff_threads) {
        child.get();
      }
#endif  // !HAVE_TBB
    }
    std::vector<std::future<int>> fill_cpu_buff_threads;
    for (int thread_idx = 0; thread_idx < thread_count; ++thread_idx) {
      fill_cpu_buff_threads.emplace_back(std::async(
          std::launch::async,
          [key_handler,
           keyspace_entry_count,
           &join_columns,
           key_component_count,
           key_component_width,
           layout,
           thread_idx,
           cpu_hash_table_ptr,
           thread_count,
           for_semi_join,
           parent_thread_local_ids = logger::thread_local_ids()] {
            logger::LocalIdsScopeGuard lisg = parent_thread_local_ids.setNewThreadId();
            DEBUG_TIMER_NEW_THREAD(parent_thread_local_ids.thread_id_);
            switch (key_component_width) {
              case 4: {
                return fill_baseline_hash_join_buff<int32_t>(cpu_hash_table_ptr,
                                                             keyspace_entry_count,
                                                             -1,
                                                             for_semi_join,
                                                             key_component_count,
                                                             layout == HashType::OneToOne,
                                                             key_handler,
                                                             join_columns[0].num_elems,
                                                             thread_idx,
                                                             thread_count);
                break;
              }
              case 8: {
                return fill_baseline_hash_join_buff<int64_t>(cpu_hash_table_ptr,
                                                             keyspace_entry_count,
                                                             -1,
                                                             for_semi_join,
                                                             key_component_count,
                                                             layout == HashType::OneToOne,
                                                             key_handler,
                                                             join_columns[0].num_elems,
                                                             thread_idx,
                                                             thread_count);
                break;
              }
              default:
                CHECK(false);
            }
            return -1;
          }));
    }
    int err = 0;
    for (auto& child : fill_cpu_buff_threads) {
      int partial_err = child.get();
      if (partial_err) {
        err = partial_err;
      }
    }
    if (err) {
      return err;
    }
    if (HashJoin::layoutRequiresAdditionalBuffers(layout)) {
      auto one_to_many_buff = reinterpret_cast<int32_t*>(
          cpu_hash_table_ptr + keyspace_entry_count * entry_size);
      {
        auto timer_init_additional_buffers =
            DEBUG_TIMER("CPU Baseline-Hash: Additional Buffers init_hash_join_buff");
        init_hash_join_buff(one_to_many_buff, keyspace_entry_count, -1, 0, 1);
      }
      bool is_geo_compressed = false;
      if constexpr (std::is_same_v<KEY_HANDLER, RangeKeyHandler>) {
        if (const auto range_handler =
                reinterpret_cast<const RangeKeyHandler*>(key_handler)) {
          is_geo_compressed = range_handler->is_compressed_;
        }
      }
      setHashLayout(layout);
      switch (key_component_width) {
        case 4: {
          const auto composite_key_dict = reinterpret_cast<int32_t*>(cpu_hash_table_ptr);
          fill_one_to_many_baseline_hash_table_32(
              one_to_many_buff,
              composite_key_dict,
              keyspace_entry_count,
              key_component_count,
              join_columns,
              join_column_types,
              join_bucket_info,
              str_proxy_translation_maps_ptrs_and_offsets.first,
              str_proxy_translation_maps_ptrs_and_offsets.second,
              thread_count,
              std::is_same_v<KEY_HANDLER, RangeKeyHandler>,
              is_geo_compressed,
              join_type == JoinType::WINDOW_FUNCTION_FRAMING);
          break;
        }
        case 8: {
          const auto composite_key_dict = reinterpret_cast<int64_t*>(cpu_hash_table_ptr);
          fill_one_to_many_baseline_hash_table_64(
              one_to_many_buff,
              composite_key_dict,
              keyspace_entry_count,
              key_component_count,
              join_columns,
              join_column_types,
              join_bucket_info,
              str_proxy_translation_maps_ptrs_and_offsets.first,
              str_proxy_translation_maps_ptrs_and_offsets.second,
              thread_count,
              std::is_same_v<KEY_HANDLER, RangeKeyHandler>,
              is_geo_compressed,
              join_type == JoinType::WINDOW_FUNCTION_FRAMING);
          break;
        }
        default:
          CHECK(false);
      }
    }
    return err;
  }

Here is the call graph for this function:

Here is the caller graph for this function:

template<class KEY_HANDLER >

int BaselineJoinHashTableBuilder::initHashTableOnGpu ( KEY_HANDLER *  key_handler, const std::vector< JoinColumn > &  join_columns, const HashType  layout, const JoinType  join_type, const size_t  key_component_width, const size_t  key_component_count, const size_t  keyspace_entry_count, const size_t  emitted_keys_count, const int  device_id, const Executor *  executor, const RegisteredQueryHint &  query_hint  )

inline

Definition at line 554 of file BaselineHashTableBuilder.h.

References allocateDeviceMemory(), ANTI, CHECK, DEBUG_TIMER, getQueryEngineCudaStreamForDevice(), hash_table_, init_baseline_hash_join_buff_on_device_32(), init_baseline_hash_join_buff_on_device_64(), init_hash_join_buff_on_device(), HashJoin::layoutRequiresAdditionalBuffers(), OneToOne, SEMI, setHashLayout(), transfer_flat_object_to_gpu(), UNREACHABLE, and WINDOW_FUNCTION_FRAMING.

Referenced by BaselineJoinHashTable::initHashTableForDevice().

                                                                {
    auto timer = DEBUG_TIMER(__func__);
    int err = 0;
#ifdef HAVE_CUDA
    allocateDeviceMemory(layout,
                         key_component_width,
                         key_component_count,
                         keyspace_entry_count,
                         emitted_keys_count,
                         device_id,
                         executor,
                         query_hint);
    if (!keyspace_entry_count) {
      // need to "allocate" the empty hash table first
      CHECK(!emitted_keys_count);
      return 0;
    }
    auto data_mgr = executor->getDataMgr();
    auto allocator = std::make_unique<CudaAllocator>(
        data_mgr, device_id, getQueryEngineCudaStreamForDevice(device_id));
    auto dev_err_buff = allocator->alloc(sizeof(int));

    allocator->copyToDevice(dev_err_buff, &err, sizeof(err));
    auto gpu_hash_table_buff = hash_table_->getGpuBuffer();
    CHECK(gpu_hash_table_buff);
    const bool for_semi_join =
        (join_type == JoinType::SEMI || join_type == JoinType::ANTI) &&
        layout == HashType::OneToOne;
    setHashLayout(layout);
    const auto key_handler_gpu = transfer_flat_object_to_gpu(*key_handler, *allocator);
    switch (key_component_width) {
      case 4:
        init_baseline_hash_join_buff_on_device_32(gpu_hash_table_buff,
                                                  keyspace_entry_count,
                                                  key_component_count,
                                                  layout == HashType::OneToOne,
                                                  -1);
        break;
      case 8:
        init_baseline_hash_join_buff_on_device_64(gpu_hash_table_buff,
                                                  keyspace_entry_count,
                                                  key_component_count,
                                                  layout == HashType::OneToOne,
                                                  -1);
        break;
      default:
        UNREACHABLE();
    }
    switch (key_component_width) {
      case 4: {
        fill_baseline_hash_join_buff_on_device<int32_t>(
            gpu_hash_table_buff,
            keyspace_entry_count,
            -1,
            for_semi_join,
            key_component_count,
            layout == HashType::OneToOne,
            reinterpret_cast<int*>(dev_err_buff),
            key_handler_gpu,
            join_columns.front().num_elems);
        allocator->copyFromDevice(&err, dev_err_buff, sizeof(err));
        break;
      }
      case 8: {
        fill_baseline_hash_join_buff_on_device<int64_t>(
            gpu_hash_table_buff,
            keyspace_entry_count,
            -1,
            for_semi_join,
            key_component_count,
            layout == HashType::OneToOne,
            reinterpret_cast<int*>(dev_err_buff),
            key_handler_gpu,
            join_columns.front().num_elems);
        allocator->copyFromDevice(&err, dev_err_buff, sizeof(err));
        break;
      }
      default:
        UNREACHABLE();
    }
    if (err) {
      return err;
    }
    if (HashJoin::layoutRequiresAdditionalBuffers(layout)) {
      const auto entry_size = key_component_count * key_component_width;
      auto one_to_many_buff = reinterpret_cast<int32_t*>(
          gpu_hash_table_buff + keyspace_entry_count * entry_size);
      init_hash_join_buff_on_device(one_to_many_buff, keyspace_entry_count, -1);
      setHashLayout(layout);
      switch (key_component_width) {
        case 4: {
          const auto composite_key_dict = reinterpret_cast<int32_t*>(gpu_hash_table_buff);
          fill_one_to_many_baseline_hash_table_on_device<int32_t>(
              one_to_many_buff,
              composite_key_dict,
              keyspace_entry_count,
              key_component_count,
              key_handler_gpu,
              join_columns.front().num_elems,
              join_type == JoinType::WINDOW_FUNCTION_FRAMING);

          break;
        }
        case 8: {
          const auto composite_key_dict = reinterpret_cast<int64_t*>(gpu_hash_table_buff);
          fill_one_to_many_baseline_hash_table_on_device<int64_t>(
              one_to_many_buff,
              composite_key_dict,
              keyspace_entry_count,
              key_component_count,
              key_handler_gpu,
              join_columns.front().num_elems,
              join_type == JoinType::WINDOW_FUNCTION_FRAMING);

          break;
        }
        default:
          UNREACHABLE();
      }
    }
#else
    UNREACHABLE();
#endif
    return err;
  }

Here is the call graph for this function:

Here is the caller graph for this function:

void BaselineJoinHashTableBuilder::setHashLayout ( HashType  layout )

inline

Definition at line 692 of file BaselineHashTableBuilder.h.

References layout_.

Referenced by initHashTableOnCpu(), and initHashTableOnGpu().

{ layout_ = layout; }

Here is the caller graph for this function:

Member Data Documentation

std::unique_ptr<BaselineHashTable> BaselineJoinHashTableBuilder::hash_table_

private

Definition at line 697 of file BaselineHashTableBuilder.h.

Referenced by allocateDeviceMemory(), getHashTable(), initHashTableOnCpu(), and initHashTableOnGpu().

HashType BaselineJoinHashTableBuilder::layout_

private

Definition at line 698 of file BaselineHashTableBuilder.h.

Referenced by getHashLayout(), and setHashLayout().

---

The documentation for this class was generated from the following file:

• /home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/JoinHashTable/Builders/BaselineHashTableBuilder.h