Execute.cpp

Go to the documentation of this file.

/*
 * Copyright 2020 OmniSci, 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 "Execute.h"

#include "AggregateUtils.h"
#include "CodeGenerator.h"
#include "ColumnFetcher.h"
#include "Descriptors/QueryCompilationDescriptor.h"
#include "Descriptors/QueryFragmentDescriptor.h"
#include "DynamicWatchdog.h"
#include "EquiJoinCondition.h"
#include "ErrorHandling.h"
#include "ExpressionRewrite.h"
#include "ExternalCacheInvalidators.h"
#include "GpuMemUtils.h"
#include "InPlaceSort.h"
#include "JoinHashTable/BaselineJoinHashTable.h"
#include "JoinHashTable/OverlapsJoinHashTable.h"
#include "JsonAccessors.h"
#include "OutputBufferInitialization.h"
#include "QueryEngine/QueryDispatchQueue.h"
#include "QueryRewrite.h"
#include "QueryTemplateGenerator.h"
#include "ResultSetReductionJIT.h"
#include "RuntimeFunctions.h"
#include "SpeculativeTopN.h"

#include "TableFunctions/TableFunctionCompilationContext.h"
#include "TableFunctions/TableFunctionExecutionContext.h"

#include "CudaMgr/CudaMgr.h"
#include "DataMgr/BufferMgr/BufferMgr.h"
#include "Parser/ParserNode.h"
#include "Shared/SystemParameters.h"
#include "Shared/TypedDataAccessors.h"
#include "Shared/checked_alloc.h"
#include "Shared/measure.h"
#include "Shared/misc.h"
#include "Shared/scope.h"
#include "Shared/shard_key.h"
#include "Shared/threadpool.h"

#include "AggregatedColRange.h"
#include "StringDictionaryGenerations.h"

#include <llvm/Transforms/Utils/BasicBlockUtils.h>
#include <boost/filesystem/operations.hpp>
#include <boost/filesystem/path.hpp>

#ifdef HAVE_CUDA
#include <cuda.h>
#endif  // HAVE_CUDA
#include <chrono>
#include <ctime>
#include <future>
#include <iostream>
#include <memory>
#include <numeric>
#include <set>
#include <thread>

bool g_enable_watchdog{false};
bool g_enable_dynamic_watchdog{false};
bool g_use_tbb_pool{false};
bool g_enable_filter_function{true};
unsigned g_dynamic_watchdog_time_limit{10000};
bool g_allow_cpu_retry{true};
bool g_null_div_by_zero{false};
unsigned g_trivial_loop_join_threshold{1000};
bool g_from_table_reordering{true};
bool g_inner_join_fragment_skipping{true};
extern bool g_enable_smem_group_by;
extern std::unique_ptr<llvm::Module> udf_gpu_module;
extern std::unique_ptr<llvm::Module> udf_cpu_module;
bool g_enable_filter_push_down{false};
float g_filter_push_down_low_frac{-1.0f};
float g_filter_push_down_high_frac{-1.0f};
size_t g_filter_push_down_passing_row_ubound{0};
bool g_enable_columnar_output{false};
bool g_enable_overlaps_hashjoin{false};
bool g_enable_hashjoin_many_to_many{false};
size_t g_overlaps_max_table_size_bytes{1024 * 1024 * 1024};
bool g_strip_join_covered_quals{false};
size_t g_constrained_by_in_threshold{10};
size_t g_big_group_threshold{20000};
bool g_enable_window_functions{true};
bool g_enable_table_functions{false};
size_t g_max_memory_allocation_size{2000000000};  // set to max slab size
size_t g_min_memory_allocation_size{
    256};  // minimum memory allocation required for projection query output buffer
           // without pre-flight count
bool g_enable_bump_allocator{false};
double g_bump_allocator_step_reduction{0.75};
bool g_enable_direct_columnarization{true};
extern bool g_enable_experimental_string_functions;
bool g_enable_runtime_query_interrupt{false};
bool g_use_estimator_result_cache{true};
unsigned g_pending_query_interrupt_freq{1000};
double g_running_query_interrupt_freq{0.5};
size_t g_gpu_smem_threshold{
    4096};  // GPU shared memory threshold (in bytes), if larger
            // buffer sizes are required we do not use GPU shared
            // memory optimizations Setting this to 0 means unlimited
            // (subject to other dynamically calculated caps)
bool g_enable_smem_grouped_non_count_agg{
    true};  // enable use of shared memory when performing group-by with select non-count
            // aggregates
bool g_enable_smem_non_grouped_agg{
    true};  // enable optimizations for using GPU shared memory in implementation of
            // non-grouped aggregates
bool g_is_test_env{false};  // operating under a unit test environment. Currently only
                            // limits the allocation for the output buffer arena

size_t g_approx_quantile_buffer{1000};
size_t g_approx_quantile_centroids{300};

extern bool g_cache_string_hash;

int const Executor::max_gpu_count;

const int32_t Executor::ERR_SINGLE_VALUE_FOUND_MULTIPLE_VALUES;

Executor::Executor(const ExecutorId executor_id,
                   const size_t block_size_x,
                   const size_t grid_size_x,
                   const size_t max_gpu_slab_size,
                   const std::string& debug_dir,
                   const std::string& debug_file)
    : cgen_state_(new CgenState({}, false))
    , cpu_code_cache_(code_cache_size)
    , gpu_code_cache_(code_cache_size)
    , block_size_x_(block_size_x)
    , grid_size_x_(grid_size_x)
    , max_gpu_slab_size_(max_gpu_slab_size)
    , debug_dir_(debug_dir)
    , debug_file_(debug_file)
    , executor_id_(executor_id)
    , catalog_(nullptr)
    , temporary_tables_(nullptr)
    , input_table_info_cache_(this) {}

std::shared_ptr<Executor> Executor::getExecutor(
    const ExecutorId executor_id,
    const std::string& debug_dir,
    const std::string& debug_file,
    const SystemParameters system_parameters) {
  INJECT_TIMER(getExecutor);

  mapd_unique_lock<mapd_shared_mutex> write_lock(executors_cache_mutex_);
  auto it = executors_.find(executor_id);
  if (it != executors_.end()) {
    return it->second;
  }
  auto executor = std::make_shared<Executor>(executor_id,
                                             system_parameters.cuda_block_size,
                                             system_parameters.cuda_grid_size,
                                             system_parameters.max_gpu_slab_size,
                                             debug_dir,
                                             debug_file);
  CHECK(executors_.insert(std::make_pair(executor_id, executor)).second);
  return executor;
}

void Executor::clearMemory(const Data_Namespace::MemoryLevel memory_level) {
  switch (memory_level) {
    case Data_Namespace::MemoryLevel::CPU_LEVEL:
    case Data_Namespace::MemoryLevel::GPU_LEVEL: {
      mapd_unique_lock<mapd_shared_mutex> flush_lock(
          execute_mutex_);  // Don't flush memory while queries are running

      Catalog_Namespace::SysCatalog::instance().getDataMgr().clearMemory(memory_level);
      if (memory_level == Data_Namespace::MemoryLevel::CPU_LEVEL) {
        // The hash table cache uses CPU memory not managed by the buffer manager. In the
        // future, we should manage these allocations with the buffer manager directly.
        // For now, assume the user wants to purge the hash table cache when they clear
        // CPU memory (currently used in ExecuteTest to lower memory pressure)
        JoinHashTableCacheInvalidator::invalidateCaches();
      }
      break;
    }
    default: {
      throw std::runtime_error(
          "Clearing memory levels other than the CPU level or GPU level is not "
          "supported.");
    }
  }
}

size_t Executor::getArenaBlockSize() {
  return g_is_test_env ? 100000000 : (1UL << 32) + kArenaBlockOverhead;
}

StringDictionaryProxy* Executor::getStringDictionaryProxy(
    const int dict_id_in,
    std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner,
    const bool with_generation) const {
  CHECK(row_set_mem_owner);
  std::lock_guard<std::mutex> lock(
      str_dict_mutex_);  // TODO: can we use RowSetMemOwner state mutex here?
  return row_set_mem_owner->getOrAddStringDictProxy(
      dict_id_in, with_generation, catalog_);
}

StringDictionaryProxy* RowSetMemoryOwner::getOrAddStringDictProxy(
    const int dict_id_in,
    const bool with_generation,
    const Catalog_Namespace::Catalog* catalog) {
  const int dict_id{dict_id_in < 0 ? REGULAR_DICT(dict_id_in) : dict_id_in};
  CHECK(catalog);
  const auto dd = catalog->getMetadataForDict(dict_id);
  if (dd) {
    CHECK(dd->stringDict);
    CHECK_LE(dd->dictNBits, 32);
    const int64_t generation =
        with_generation ? string_dictionary_generations_.getGeneration(dict_id) : -1;
    return addStringDict(dd->stringDict, dict_id, generation);
  }
  CHECK_EQ(0, dict_id);
  if (!lit_str_dict_proxy_) {
    std::shared_ptr<StringDictionary> tsd =
        std::make_shared<StringDictionary>("", false, true, g_cache_string_hash);
    lit_str_dict_proxy_.reset(new StringDictionaryProxy(tsd, 0));
  }
  return lit_str_dict_proxy_.get();
}

quantile::TDigest* RowSetMemoryOwner::newTDigest() {
  std::lock_guard<std::mutex> lock(state_mutex_);
  auto& td = t_digests_.emplace_back(std::make_unique<quantile::TDigest>());
  // Separating TDigests from their buffers allows for flexibile memory management.
  // The incoming Buffer is used to collect and sort incoming data,
  // and used as scratch space during its merging into the main Centroids buffer.
  td->setBuffer(t_digest_buffers_.emplace_back(g_approx_quantile_buffer));
  td->setCentroids(t_digest_buffers_.emplace_back(g_approx_quantile_centroids));
  return td.get();
}

bool Executor::isCPUOnly() const {
  CHECK(catalog_);
  return !catalog_->getDataMgr().getCudaMgr();
}

const ColumnDescriptor* Executor::getColumnDescriptor(
    const Analyzer::ColumnVar* col_var) const {
  return get_column_descriptor_maybe(
      col_var->get_column_id(), col_var->get_table_id(), *catalog_);
}

const ColumnDescriptor* Executor::getPhysicalColumnDescriptor(
    const Analyzer::ColumnVar* col_var,
    int n) const {
  const auto cd = getColumnDescriptor(col_var);
  if (!cd || n > cd->columnType.get_physical_cols()) {
    return nullptr;
  }
  return get_column_descriptor_maybe(
      col_var->get_column_id() + n, col_var->get_table_id(), *catalog_);
}

const Catalog_Namespace::Catalog* Executor::getCatalog() const {
  return catalog_;
}

void Executor::setCatalog(const Catalog_Namespace::Catalog* catalog) {
  catalog_ = catalog;
}

const std::shared_ptr<RowSetMemoryOwner> Executor::getRowSetMemoryOwner() const {
  return row_set_mem_owner_;
}

const TemporaryTables* Executor::getTemporaryTables() const {
  return temporary_tables_;
}

Fragmenter_Namespace::TableInfo Executor::getTableInfo(const int table_id) const {
  return input_table_info_cache_.getTableInfo(table_id);
}

const TableGeneration& Executor::getTableGeneration(const int table_id) const {
  return table_generations_.getGeneration(table_id);
}

ExpressionRange Executor::getColRange(const PhysicalInput& phys_input) const {
  return agg_col_range_cache_.getColRange(phys_input);
}

size_t Executor::getNumBytesForFetchedRow(const std::set<int>& table_ids_to_fetch) const {
  size_t num_bytes = 0;
  if (!plan_state_) {
    return 0;
  }
  for (const auto& fetched_col_pair : plan_state_->columns_to_fetch_) {
    if (table_ids_to_fetch.count(fetched_col_pair.first) == 0) {
      continue;
    }

    if (fetched_col_pair.first < 0) {
      num_bytes += 8;
    } else {
      const auto cd =
          catalog_->getMetadataForColumn(fetched_col_pair.first, fetched_col_pair.second);
      const auto& ti = cd->columnType;
      const auto sz = ti.get_type() == kTEXT && ti.get_compression() == kENCODING_DICT
                          ? 4
                          : ti.get_size();
      if (sz < 0) {
        // for varlen types, only account for the pointer/size for each row, for now
        num_bytes += 16;
      } else {
        num_bytes += sz;
      }
    }
  }
  return num_bytes;
}

std::vector<ColumnLazyFetchInfo> Executor::getColLazyFetchInfo(
    const std::vector<Analyzer::Expr*>& target_exprs) const {
  CHECK(plan_state_);
  CHECK(catalog_);
  std::vector<ColumnLazyFetchInfo> col_lazy_fetch_info;
  for (const auto target_expr : target_exprs) {
    if (!plan_state_->isLazyFetchColumn(target_expr)) {
      col_lazy_fetch_info.emplace_back(
          ColumnLazyFetchInfo{false, -1, SQLTypeInfo(kNULLT, false)});
    } else {
      const auto col_var = dynamic_cast<const Analyzer::ColumnVar*>(target_expr);
      CHECK(col_var);
      auto col_id = col_var->get_column_id();
      auto rte_idx = (col_var->get_rte_idx() == -1) ? 0 : col_var->get_rte_idx();
      auto cd = (col_var->get_table_id() > 0)
                    ? get_column_descriptor(col_id, col_var->get_table_id(), *catalog_)
                    : nullptr;
      if (cd && IS_GEO(cd->columnType.get_type())) {
        // Geo coords cols will be processed in sequence. So we only need to track the
        // first coords col in lazy fetch info.
        {
          auto cd0 =
              get_column_descriptor(col_id + 1, col_var->get_table_id(), *catalog_);
          auto col0_ti = cd0->columnType;
          CHECK(!cd0->isVirtualCol);
          auto col0_var = makeExpr<Analyzer::ColumnVar>(
              col0_ti, col_var->get_table_id(), cd0->columnId, rte_idx);
          auto local_col0_id = plan_state_->getLocalColumnId(col0_var.get(), false);
          col_lazy_fetch_info.emplace_back(
              ColumnLazyFetchInfo{true, local_col0_id, col0_ti});
        }
      } else {
        auto local_col_id = plan_state_->getLocalColumnId(col_var, false);
        const auto& col_ti = col_var->get_type_info();
        col_lazy_fetch_info.emplace_back(ColumnLazyFetchInfo{true, local_col_id, col_ti});
      }
    }
  }
  return col_lazy_fetch_info;
}

void Executor::clearMetaInfoCache() {
  input_table_info_cache_.clear();
  agg_col_range_cache_.clear();
  table_generations_.clear();
}

std::vector<int8_t> Executor::serializeLiterals(
    const std::unordered_map<int, CgenState::LiteralValues>& literals,
    const int device_id) {
  if (literals.empty()) {
    return {};
  }
  const auto dev_literals_it = literals.find(device_id);
  CHECK(dev_literals_it != literals.end());
  const auto& dev_literals = dev_literals_it->second;
  size_t lit_buf_size{0};
  std::vector<std::string> real_strings;
  std::vector<std::vector<double>> double_array_literals;
  std::vector<std::vector<int8_t>> align64_int8_array_literals;
  std::vector<std::vector<int32_t>> int32_array_literals;
  std::vector<std::vector<int8_t>> align32_int8_array_literals;
  std::vector<std::vector<int8_t>> int8_array_literals;
  for (const auto& lit : dev_literals) {
    lit_buf_size = CgenState::addAligned(lit_buf_size, CgenState::literalBytes(lit));
    if (lit.which() == 7) {
      const auto p = boost::get<std::string>(&lit);
      CHECK(p);
      real_strings.push_back(*p);
    } else if (lit.which() == 8) {
      const auto p = boost::get<std::vector<double>>(&lit);
      CHECK(p);
      double_array_literals.push_back(*p);
    } else if (lit.which() == 9) {
      const auto p = boost::get<std::vector<int32_t>>(&lit);
      CHECK(p);
      int32_array_literals.push_back(*p);
    } else if (lit.which() == 10) {
      const auto p = boost::get<std::vector<int8_t>>(&lit);
      CHECK(p);
      int8_array_literals.push_back(*p);
    } else if (lit.which() == 11) {
      const auto p = boost::get<std::pair<std::vector<int8_t>, int>>(&lit);
      CHECK(p);
      if (p->second == 64) {
        align64_int8_array_literals.push_back(p->first);
      } else if (p->second == 32) {
        align32_int8_array_literals.push_back(p->first);
      } else {
        CHECK(false);
      }
    }
  }
  if (lit_buf_size > static_cast<size_t>(std::numeric_limits<int16_t>::max())) {
    throw TooManyLiterals();
  }
  int16_t crt_real_str_off = lit_buf_size;
  for (const auto& real_str : real_strings) {
    CHECK_LE(real_str.size(), static_cast<size_t>(std::numeric_limits<int16_t>::max()));
    lit_buf_size += real_str.size();
  }
  if (double_array_literals.size() > 0) {
    lit_buf_size = align(lit_buf_size, sizeof(double));
  }
  int16_t crt_double_arr_lit_off = lit_buf_size;
  for (const auto& double_array_literal : double_array_literals) {
    CHECK_LE(double_array_literal.size(),
             static_cast<size_t>(std::numeric_limits<int16_t>::max()));
    lit_buf_size += double_array_literal.size() * sizeof(double);
  }
  if (align64_int8_array_literals.size() > 0) {
    lit_buf_size = align(lit_buf_size, sizeof(uint64_t));
  }
  int16_t crt_align64_int8_arr_lit_off = lit_buf_size;
  for (const auto& align64_int8_array_literal : align64_int8_array_literals) {
    CHECK_LE(align64_int8_array_literals.size(),
             static_cast<size_t>(std::numeric_limits<int16_t>::max()));
    lit_buf_size += align64_int8_array_literal.size();
  }
  if (int32_array_literals.size() > 0) {
    lit_buf_size = align(lit_buf_size, sizeof(int32_t));
  }
  int16_t crt_int32_arr_lit_off = lit_buf_size;
  for (const auto& int32_array_literal : int32_array_literals) {
    CHECK_LE(int32_array_literal.size(),
             static_cast<size_t>(std::numeric_limits<int16_t>::max()));
    lit_buf_size += int32_array_literal.size() * sizeof(int32_t);
  }
  if (align32_int8_array_literals.size() > 0) {
    lit_buf_size = align(lit_buf_size, sizeof(int32_t));
  }
  int16_t crt_align32_int8_arr_lit_off = lit_buf_size;
  for (const auto& align32_int8_array_literal : align32_int8_array_literals) {
    CHECK_LE(align32_int8_array_literals.size(),
             static_cast<size_t>(std::numeric_limits<int16_t>::max()));
    lit_buf_size += align32_int8_array_literal.size();
  }
  int16_t crt_int8_arr_lit_off = lit_buf_size;
  for (const auto& int8_array_literal : int8_array_literals) {
    CHECK_LE(int8_array_literal.size(),
             static_cast<size_t>(std::numeric_limits<int16_t>::max()));
    lit_buf_size += int8_array_literal.size();
  }
  unsigned crt_real_str_idx = 0;
  unsigned crt_double_arr_lit_idx = 0;
  unsigned crt_align64_int8_arr_lit_idx = 0;
  unsigned crt_int32_arr_lit_idx = 0;
  unsigned crt_align32_int8_arr_lit_idx = 0;
  unsigned crt_int8_arr_lit_idx = 0;
  std::vector<int8_t> serialized(lit_buf_size);
  size_t off{0};
  for (const auto& lit : dev_literals) {
    const auto lit_bytes = CgenState::literalBytes(lit);
    off = CgenState::addAligned(off, lit_bytes);
    switch (lit.which()) {
      case 0: {
        const auto p = boost::get<int8_t>(&lit);
        CHECK(p);
        serialized[off - lit_bytes] = *p;
        break;
      }
      case 1: {
        const auto p = boost::get<int16_t>(&lit);
        CHECK(p);
        memcpy(&serialized[off - lit_bytes], p, lit_bytes);
        break;
      }
      case 2: {
        const auto p = boost::get<int32_t>(&lit);
        CHECK(p);
        memcpy(&serialized[off - lit_bytes], p, lit_bytes);
        break;
      }
      case 3: {
        const auto p = boost::get<int64_t>(&lit);
        CHECK(p);
        memcpy(&serialized[off - lit_bytes], p, lit_bytes);
        break;
      }
      case 4: {
        const auto p = boost::get<float>(&lit);
        CHECK(p);
        memcpy(&serialized[off - lit_bytes], p, lit_bytes);
        break;
      }
      case 5: {
        const auto p = boost::get<double>(&lit);
        CHECK(p);
        memcpy(&serialized[off - lit_bytes], p, lit_bytes);
        break;
      }
      case 6: {
        const auto p = boost::get<std::pair<std::string, int>>(&lit);
        CHECK(p);
        const auto str_id =
            g_enable_experimental_string_functions
                ? getStringDictionaryProxy(p->second, row_set_mem_owner_, true)
                      ->getOrAddTransient(p->first)
                : getStringDictionaryProxy(p->second, row_set_mem_owner_, true)
                      ->getIdOfString(p->first);
        memcpy(&serialized[off - lit_bytes], &str_id, lit_bytes);
        break;
      }
      case 7: {
        const auto p = boost::get<std::string>(&lit);
        CHECK(p);
        int32_t off_and_len = crt_real_str_off << 16;
        const auto& crt_real_str = real_strings[crt_real_str_idx];
        off_and_len |= static_cast<int16_t>(crt_real_str.size());
        memcpy(&serialized[off - lit_bytes], &off_and_len, lit_bytes);
        memcpy(&serialized[crt_real_str_off], crt_real_str.data(), crt_real_str.size());
        ++crt_real_str_idx;
        crt_real_str_off += crt_real_str.size();
        break;
      }
      case 8: {
        const auto p = boost::get<std::vector<double>>(&lit);
        CHECK(p);
        int32_t off_and_len = crt_double_arr_lit_off << 16;
        const auto& crt_double_arr_lit = double_array_literals[crt_double_arr_lit_idx];
        int32_t len = crt_double_arr_lit.size();
        CHECK_EQ((len >> 16), 0);
        off_and_len |= static_cast<int16_t>(len);
        int32_t double_array_bytesize = len * sizeof(double);
        memcpy(&serialized[off - lit_bytes], &off_and_len, lit_bytes);
        memcpy(&serialized[crt_double_arr_lit_off],
               crt_double_arr_lit.data(),
               double_array_bytesize);
        ++crt_double_arr_lit_idx;
        crt_double_arr_lit_off += double_array_bytesize;
        break;
      }
      case 9: {
        const auto p = boost::get<std::vector<int32_t>>(&lit);
        CHECK(p);
        int32_t off_and_len = crt_int32_arr_lit_off << 16;
        const auto& crt_int32_arr_lit = int32_array_literals[crt_int32_arr_lit_idx];
        int32_t len = crt_int32_arr_lit.size();
        CHECK_EQ((len >> 16), 0);
        off_and_len |= static_cast<int16_t>(len);
        int32_t int32_array_bytesize = len * sizeof(int32_t);
        memcpy(&serialized[off - lit_bytes], &off_and_len, lit_bytes);
        memcpy(&serialized[crt_int32_arr_lit_off],
               crt_int32_arr_lit.data(),
               int32_array_bytesize);
        ++crt_int32_arr_lit_idx;
        crt_int32_arr_lit_off += int32_array_bytesize;
        break;
      }
      case 10: {
        const auto p = boost::get<std::vector<int8_t>>(&lit);
        CHECK(p);
        int32_t off_and_len = crt_int8_arr_lit_off << 16;
        const auto& crt_int8_arr_lit = int8_array_literals[crt_int8_arr_lit_idx];
        int32_t len = crt_int8_arr_lit.size();
        CHECK_EQ((len >> 16), 0);
        off_and_len |= static_cast<int16_t>(len);
        int32_t int8_array_bytesize = len;
        memcpy(&serialized[off - lit_bytes], &off_and_len, lit_bytes);
        memcpy(&serialized[crt_int8_arr_lit_off],
               crt_int8_arr_lit.data(),
               int8_array_bytesize);
        ++crt_int8_arr_lit_idx;
        crt_int8_arr_lit_off += int8_array_bytesize;
        break;
      }
      case 11: {
        const auto p = boost::get<std::pair<std::vector<int8_t>, int>>(&lit);
        CHECK(p);
        if (p->second == 64) {
          int32_t off_and_len = crt_align64_int8_arr_lit_off << 16;
          const auto& crt_align64_int8_arr_lit =
              align64_int8_array_literals[crt_align64_int8_arr_lit_idx];
          int32_t len = crt_align64_int8_arr_lit.size();
          CHECK_EQ((len >> 16), 0);
          off_and_len |= static_cast<int16_t>(len);
          int32_t align64_int8_array_bytesize = len;
          memcpy(&serialized[off - lit_bytes], &off_and_len, lit_bytes);
          memcpy(&serialized[crt_align64_int8_arr_lit_off],
                 crt_align64_int8_arr_lit.data(),
                 align64_int8_array_bytesize);
          ++crt_align64_int8_arr_lit_idx;
          crt_align64_int8_arr_lit_off += align64_int8_array_bytesize;
        } else if (p->second == 32) {
          int32_t off_and_len = crt_align32_int8_arr_lit_off << 16;
          const auto& crt_align32_int8_arr_lit =
              align32_int8_array_literals[crt_align32_int8_arr_lit_idx];
          int32_t len = crt_align32_int8_arr_lit.size();
          CHECK_EQ((len >> 16), 0);
          off_and_len |= static_cast<int16_t>(len);
          int32_t align32_int8_array_bytesize = len;
          memcpy(&serialized[off - lit_bytes], &off_and_len, lit_bytes);
          memcpy(&serialized[crt_align32_int8_arr_lit_off],
                 crt_align32_int8_arr_lit.data(),
                 align32_int8_array_bytesize);
          ++crt_align32_int8_arr_lit_idx;
          crt_align32_int8_arr_lit_off += align32_int8_array_bytesize;
        } else {
          CHECK(false);
        }
        break;
      }
      default:
        CHECK(false);
    }
  }
  return serialized;
}

int Executor::deviceCount(const ExecutorDeviceType device_type) const {
  if (device_type == ExecutorDeviceType::GPU) {
    const auto cuda_mgr = catalog_->getDataMgr().getCudaMgr();
    CHECK(cuda_mgr);
    return cuda_mgr->getDeviceCount();
  } else {
    return 1;
  }
}

int Executor::deviceCountForMemoryLevel(
    const Data_Namespace::MemoryLevel memory_level) const {
  return memory_level == GPU_LEVEL ? deviceCount(ExecutorDeviceType::GPU)
                                   : deviceCount(ExecutorDeviceType::CPU);
}

// TODO(alex): remove or split
std::pair<int64_t, int32_t> Executor::reduceResults(const SQLAgg agg,
                                                    const SQLTypeInfo& ti,
                                                    const int64_t agg_init_val,
                                                    const int8_t out_byte_width,
                                                    const int64_t* out_vec,
                                                    const size_t out_vec_sz,
                                                    const bool is_group_by,
                                                    const bool float_argument_input) {
  switch (agg) {
    case kAVG:
    case kSUM:
      if (0 != agg_init_val) {
        if (ti.is_integer() || ti.is_decimal() || ti.is_time() || ti.is_boolean()) {
          int64_t agg_result = agg_init_val;
          for (size_t i = 0; i < out_vec_sz; ++i) {
            agg_sum_skip_val(&agg_result, out_vec[i], agg_init_val);
          }
          return {agg_result, 0};
        } else {
          CHECK(ti.is_fp());
          switch (out_byte_width) {
            case 4: {
              int agg_result = static_cast<int32_t>(agg_init_val);
              for (size_t i = 0; i < out_vec_sz; ++i) {
                agg_sum_float_skip_val(
                    &agg_result,
                    *reinterpret_cast<const float*>(may_alias_ptr(&out_vec[i])),
                    *reinterpret_cast<const float*>(may_alias_ptr(&agg_init_val)));
              }
              const int64_t converted_bin =
                  float_argument_input
                      ? static_cast<int64_t>(agg_result)
                      : float_to_double_bin(static_cast<int32_t>(agg_result), true);
              return {converted_bin, 0};
              break;
            }
            case 8: {
              int64_t agg_result = agg_init_val;
              for (size_t i = 0; i < out_vec_sz; ++i) {
                agg_sum_double_skip_val(
                    &agg_result,
                    *reinterpret_cast<const double*>(may_alias_ptr(&out_vec[i])),
                    *reinterpret_cast<const double*>(may_alias_ptr(&agg_init_val)));
              }
              return {agg_result, 0};
              break;
            }
            default:
              CHECK(false);
          }
        }
      }
      if (ti.is_integer() || ti.is_decimal() || ti.is_time()) {
        int64_t agg_result = 0;
        for (size_t i = 0; i < out_vec_sz; ++i) {
          agg_result += out_vec[i];
        }
        return {agg_result, 0};
      } else {
        CHECK(ti.is_fp());
        switch (out_byte_width) {
          case 4: {
            float r = 0.;
            for (size_t i = 0; i < out_vec_sz; ++i) {
              r += *reinterpret_cast<const float*>(may_alias_ptr(&out_vec[i]));
            }
            const auto float_bin = *reinterpret_cast<const int32_t*>(may_alias_ptr(&r));
            const int64_t converted_bin =
                float_argument_input ? float_bin : float_to_double_bin(float_bin, true);
            return {converted_bin, 0};
          }
          case 8: {
            double r = 0.;
            for (size_t i = 0; i < out_vec_sz; ++i) {
              r += *reinterpret_cast<const double*>(may_alias_ptr(&out_vec[i]));
            }
            return {*reinterpret_cast<const int64_t*>(may_alias_ptr(&r)), 0};
          }
          default:
            CHECK(false);
        }
      }
      break;
    case kCOUNT: {
      uint64_t agg_result = 0;
      for (size_t i = 0; i < out_vec_sz; ++i) {
        const uint64_t out = static_cast<uint64_t>(out_vec[i]);
        agg_result += out;
      }
      return {static_cast<int64_t>(agg_result), 0};
    }
    case kMIN: {
      if (ti.is_integer() || ti.is_decimal() || ti.is_time() || ti.is_boolean()) {
        int64_t agg_result = agg_init_val;
        for (size_t i = 0; i < out_vec_sz; ++i) {
          agg_min_skip_val(&agg_result, out_vec[i], agg_init_val);
        }
        return {agg_result, 0};
      } else {
        switch (out_byte_width) {
          case 4: {
            int32_t agg_result = static_cast<int32_t>(agg_init_val);
            for (size_t i = 0; i < out_vec_sz; ++i) {
              agg_min_float_skip_val(
                  &agg_result,
                  *reinterpret_cast<const float*>(may_alias_ptr(&out_vec[i])),
                  *reinterpret_cast<const float*>(may_alias_ptr(&agg_init_val)));
            }
            const int64_t converted_bin =
                float_argument_input
                    ? static_cast<int64_t>(agg_result)
                    : float_to_double_bin(static_cast<int32_t>(agg_result), true);
            return {converted_bin, 0};
          }
          case 8: {
            int64_t agg_result = agg_init_val;
            for (size_t i = 0; i < out_vec_sz; ++i) {
              agg_min_double_skip_val(
                  &agg_result,
                  *reinterpret_cast<const double*>(may_alias_ptr(&out_vec[i])),
                  *reinterpret_cast<const double*>(may_alias_ptr(&agg_init_val)));
            }
            return {agg_result, 0};
          }
          default:
            CHECK(false);
        }
      }
    }
    case kMAX:
      if (ti.is_integer() || ti.is_decimal() || ti.is_time() || ti.is_boolean()) {
        int64_t agg_result = agg_init_val;
        for (size_t i = 0; i < out_vec_sz; ++i) {
          agg_max_skip_val(&agg_result, out_vec[i], agg_init_val);
        }
        return {agg_result, 0};
      } else {
        switch (out_byte_width) {
          case 4: {
            int32_t agg_result = static_cast<int32_t>(agg_init_val);
            for (size_t i = 0; i < out_vec_sz; ++i) {
              agg_max_float_skip_val(
                  &agg_result,
                  *reinterpret_cast<const float*>(may_alias_ptr(&out_vec[i])),
                  *reinterpret_cast<const float*>(may_alias_ptr(&agg_init_val)));
            }
            const int64_t converted_bin =
                float_argument_input ? static_cast<int64_t>(agg_result)
                                     : float_to_double_bin(agg_result, !ti.get_notnull());
            return {converted_bin, 0};
          }
          case 8: {
            int64_t agg_result = agg_init_val;
            for (size_t i = 0; i < out_vec_sz; ++i) {
              agg_max_double_skip_val(
                  &agg_result,
                  *reinterpret_cast<const double*>(may_alias_ptr(&out_vec[i])),
                  *reinterpret_cast<const double*>(may_alias_ptr(&agg_init_val)));
            }
            return {agg_result, 0};
          }
          default:
            CHECK(false);
        }
      }
    case kSINGLE_VALUE: {
      int64_t agg_result = agg_init_val;
      for (size_t i = 0; i < out_vec_sz; ++i) {
        if (out_vec[i] != agg_init_val) {
          if (agg_result == agg_init_val) {
            agg_result = out_vec[i];
          } else if (out_vec[i] != agg_result) {
            return {agg_result, Executor::ERR_SINGLE_VALUE_FOUND_MULTIPLE_VALUES};
          }
        }
      }
      return {agg_result, 0};
    }
    case kSAMPLE: {
      int64_t agg_result = agg_init_val;
      for (size_t i = 0; i < out_vec_sz; ++i) {
        if (out_vec[i] != agg_init_val) {
          agg_result = out_vec[i];
          break;
        }
      }
      return {agg_result, 0};
    }
    default:
      CHECK(false);
  }
  abort();
}

namespace {

ResultSetPtr get_merged_result(
    std::vector<std::pair<ResultSetPtr, std::vector<size_t>>>& results_per_device) {
  auto& first = results_per_device.front().first;
  CHECK(first);
  for (size_t dev_idx = 1; dev_idx < results_per_device.size(); ++dev_idx) {
    const auto& next = results_per_device[dev_idx].first;
    CHECK(next);
    first->append(*next);
  }
  return std::move(first);
}

}  // namespace

ResultSetPtr Executor::resultsUnion(SharedKernelContext& shared_context,
                                    const RelAlgExecutionUnit& ra_exe_unit) {
  auto& results_per_device = shared_context.getFragmentResults();
  if (results_per_device.empty()) {
    std::vector<TargetInfo> targets;
    for (const auto target_expr : ra_exe_unit.target_exprs) {
      targets.push_back(get_target_info(target_expr, g_bigint_count));
    }
    return std::make_shared<ResultSet>(targets,
                                       ExecutorDeviceType::CPU,
                                       QueryMemoryDescriptor(),
                                       row_set_mem_owner_,
                                       catalog_,
                                       blockSize(),
                                       gridSize());
  }
  using IndexedResultSet = std::pair<ResultSetPtr, std::vector<size_t>>;
  std::sort(results_per_device.begin(),
            results_per_device.end(),
            [](const IndexedResultSet& lhs, const IndexedResultSet& rhs) {
              CHECK_GE(lhs.second.size(), size_t(1));
              CHECK_GE(rhs.second.size(), size_t(1));
              return lhs.second.front() < rhs.second.front();
            });

  return get_merged_result(results_per_device);
}

ResultSetPtr Executor::reduceMultiDeviceResults(
    const RelAlgExecutionUnit& ra_exe_unit,
    std::vector<std::pair<ResultSetPtr, std::vector<size_t>>>& results_per_device,
    std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner,
    const QueryMemoryDescriptor& query_mem_desc) const {
  auto timer = DEBUG_TIMER(__func__);
  if (ra_exe_unit.estimator) {
    return reduce_estimator_results(ra_exe_unit, results_per_device);
  }

  if (results_per_device.empty()) {
    std::vector<TargetInfo> targets;
    for (const auto target_expr : ra_exe_unit.target_exprs) {
      targets.push_back(get_target_info(target_expr, g_bigint_count));
    }
    return std::make_shared<ResultSet>(targets,
                                       ExecutorDeviceType::CPU,
                                       QueryMemoryDescriptor(),
                                       nullptr,
                                       catalog_,
                                       blockSize(),
                                       gridSize());
  }

  return reduceMultiDeviceResultSets(
      results_per_device,
      row_set_mem_owner,
      ResultSet::fixupQueryMemoryDescriptor(query_mem_desc));
}

namespace {

ReductionCode get_reduction_code(
    std::vector<std::pair<ResultSetPtr, std::vector<size_t>>>& results_per_device,
    int64_t* compilation_queue_time) {
  auto clock_begin = timer_start();
  std::lock_guard<std::mutex> compilation_lock(Executor::compilation_mutex_);
  *compilation_queue_time = timer_stop(clock_begin);
  const auto& this_result_set = results_per_device[0].first;
  ResultSetReductionJIT reduction_jit(this_result_set->getQueryMemDesc(),
                                      this_result_set->getTargetInfos(),
                                      this_result_set->getTargetInitVals());
  return reduction_jit.codegen();
};

}  // namespace

ResultSetPtr Executor::reduceMultiDeviceResultSets(
    std::vector<std::pair<ResultSetPtr, std::vector<size_t>>>& results_per_device,
    std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner,
    const QueryMemoryDescriptor& query_mem_desc) const {
  auto timer = DEBUG_TIMER(__func__);
  std::shared_ptr<ResultSet> reduced_results;

  const auto& first = results_per_device.front().first;

  if (query_mem_desc.getQueryDescriptionType() ==
          QueryDescriptionType::GroupByBaselineHash &&
      results_per_device.size() > 1) {
    const auto total_entry_count = std::accumulate(
        results_per_device.begin(),
        results_per_device.end(),
        size_t(0),
        [](const size_t init, const std::pair<ResultSetPtr, std::vector<size_t>>& rs) {
          const auto& r = rs.first;
          return init + r->getQueryMemDesc().getEntryCount();
        });
    CHECK(total_entry_count);
    auto query_mem_desc = first->getQueryMemDesc();
    query_mem_desc.setEntryCount(total_entry_count);
    reduced_results = std::make_shared<ResultSet>(first->getTargetInfos(),
                                                  ExecutorDeviceType::CPU,
                                                  query_mem_desc,
                                                  row_set_mem_owner,
                                                  catalog_,
                                                  blockSize(),
                                                  gridSize());
    auto result_storage = reduced_results->allocateStorage(plan_state_->init_agg_vals_);
    reduced_results->initializeStorage();
    switch (query_mem_desc.getEffectiveKeyWidth()) {
      case 4:
        first->getStorage()->moveEntriesToBuffer<int32_t>(
            result_storage->getUnderlyingBuffer(), query_mem_desc.getEntryCount());
        break;
      case 8:
        first->getStorage()->moveEntriesToBuffer<int64_t>(
            result_storage->getUnderlyingBuffer(), query_mem_desc.getEntryCount());
        break;
      default:
        CHECK(false);
    }
  } else {
    reduced_results = first;
  }

  int64_t compilation_queue_time = 0;
  const auto reduction_code =
      get_reduction_code(results_per_device, &compilation_queue_time);

  for (size_t i = 1; i < results_per_device.size(); ++i) {
    reduced_results->getStorage()->reduce(
        *(results_per_device[i].first->getStorage()), {}, reduction_code);
  }
  reduced_results->addCompilationQueueTime(compilation_queue_time);
  return reduced_results;
}

ResultSetPtr Executor::reduceSpeculativeTopN(
    const RelAlgExecutionUnit& ra_exe_unit,
    std::vector<std::pair<ResultSetPtr, std::vector<size_t>>>& results_per_device,
    std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner,
    const QueryMemoryDescriptor& query_mem_desc) const {
  if (results_per_device.size() == 1) {
    return std::move(results_per_device.front().first);
  }
  const auto top_n = ra_exe_unit.sort_info.limit + ra_exe_unit.sort_info.offset;
  SpeculativeTopNMap m;
  for (const auto& result : results_per_device) {
    auto rows = result.first;
    CHECK(rows);
    if (!rows) {
      continue;
    }
    SpeculativeTopNMap that(
        *rows,
        ra_exe_unit.target_exprs,
        std::max(size_t(10000 * std::max(1, static_cast<int>(log(top_n)))), top_n));
    m.reduce(that);
  }
  CHECK_EQ(size_t(1), ra_exe_unit.sort_info.order_entries.size());
  const auto desc = ra_exe_unit.sort_info.order_entries.front().is_desc;
  return m.asRows(ra_exe_unit, row_set_mem_owner, query_mem_desc, this, top_n, desc);
}

std::unordered_set<int> get_available_gpus(const Catalog_Namespace::Catalog& cat) {
  std::unordered_set<int> available_gpus;
  if (cat.getDataMgr().gpusPresent()) {
    int gpu_count = cat.getDataMgr().getCudaMgr()->getDeviceCount();
    CHECK_GT(gpu_count, 0);
    for (int gpu_id = 0; gpu_id < gpu_count; ++gpu_id) {
      available_gpus.insert(gpu_id);
    }
  }
  return available_gpus;
}

size_t get_context_count(const ExecutorDeviceType device_type,
                         const size_t cpu_count,
                         const size_t gpu_count) {
  return device_type == ExecutorDeviceType::GPU ? gpu_count
                                                : static_cast<size_t>(cpu_count);
}

namespace {

// Compute a very conservative entry count for the output buffer entry count using no
// other information than the number of tuples in each table and multiplying them
// together.
size_t compute_buffer_entry_guess(const std::vector<InputTableInfo>& query_infos) {
  using Fragmenter_Namespace::FragmentInfo;
  // Check for overflows since we're multiplying potentially big table sizes.
  using checked_size_t = boost::multiprecision::number<
      boost::multiprecision::cpp_int_backend<64,
                                             64,
                                             boost::multiprecision::unsigned_magnitude,
                                             boost::multiprecision::checked,
                                             void>>;
  checked_size_t max_groups_buffer_entry_guess = 1;
  for (const auto& query_info : query_infos) {
    CHECK(!query_info.info.fragments.empty());
    auto it = std::max_element(query_info.info.fragments.begin(),
                               query_info.info.fragments.end(),
                               [](const FragmentInfo& f1, const FragmentInfo& f2) {
                                 return f1.getNumTuples() < f2.getNumTuples();
                               });
    max_groups_buffer_entry_guess *= it->getNumTuples();
  }
  // Cap the rough approximation to 100M entries, it's unlikely we can do a great job for
  // baseline group layout with that many entries anyway.
  constexpr size_t max_groups_buffer_entry_guess_cap = 100000000;
  try {
    return std::min(static_cast<size_t>(max_groups_buffer_entry_guess),
                    max_groups_buffer_entry_guess_cap);
  } catch (...) {
    return max_groups_buffer_entry_guess_cap;
  }
}

std::string get_table_name(const InputDescriptor& input_desc,
                           const Catalog_Namespace::Catalog& cat) {
  const auto source_type = input_desc.getSourceType();
  if (source_type == InputSourceType::TABLE) {
    const auto td = cat.getMetadataForTable(input_desc.getTableId());
    CHECK(td);
    return td->tableName;
  } else {
    return "$TEMPORARY_TABLE" + std::to_string(-input_desc.getTableId());
  }
}

inline size_t getDeviceBasedScanLimit(const ExecutorDeviceType device_type,
                                      const int device_count) {
  if (device_type == ExecutorDeviceType::GPU) {
    return device_count * Executor::high_scan_limit;
  }
  return Executor::high_scan_limit;
}

void checkWorkUnitWatchdog(const RelAlgExecutionUnit& ra_exe_unit,
                           const std::vector<InputTableInfo>& table_infos,
                           const Catalog_Namespace::Catalog& cat,
                           const ExecutorDeviceType device_type,
                           const int device_count) {
  for (const auto target_expr : ra_exe_unit.target_exprs) {
    if (dynamic_cast<const Analyzer::AggExpr*>(target_expr)) {
      return;
    }
  }
  if (!ra_exe_unit.scan_limit && table_infos.size() == 1 &&
      table_infos.front().info.getPhysicalNumTuples() < Executor::high_scan_limit) {
    // Allow a query with no scan limit to run on small tables
    return;
  }
  if (ra_exe_unit.use_bump_allocator) {
    // Bump allocator removes the scan limit (and any knowledge of the size of the output
    // relative to the size of the input), so we bypass this check for now
    return;
  }
  if (ra_exe_unit.sort_info.algorithm != SortAlgorithm::StreamingTopN &&
      ra_exe_unit.groupby_exprs.size() == 1 && !ra_exe_unit.groupby_exprs.front() &&
      (!ra_exe_unit.scan_limit ||
       ra_exe_unit.scan_limit > getDeviceBasedScanLimit(device_type, device_count))) {
    std::vector<std::string> table_names;
    const auto& input_descs = ra_exe_unit.input_descs;
    for (const auto& input_desc : input_descs) {
      table_names.push_back(get_table_name(input_desc, cat));
    }
    if (!ra_exe_unit.scan_limit) {
      throw WatchdogException(
          "Projection query would require a scan without a limit on table(s): " +
          boost::algorithm::join(table_names, ", "));
    } else {
      throw WatchdogException(
          "Projection query output result set on table(s): " +
          boost::algorithm::join(table_names, ", ") + "  would contain " +
          std::to_string(ra_exe_unit.scan_limit) +
          " rows, which is more than the current system limit of " +
          std::to_string(getDeviceBasedScanLimit(device_type, device_count)));
    }
  }
}

}  // namespace

bool is_trivial_loop_join(const std::vector<InputTableInfo>& query_infos,
                          const RelAlgExecutionUnit& ra_exe_unit) {
  if (ra_exe_unit.input_descs.size() < 2) {
    return false;
  }

  // We only support loop join at the end of folded joins
  // where ra_exe_unit.input_descs.size() > 2 for now.
  const auto inner_table_id = ra_exe_unit.input_descs.back().getTableId();

  std::optional<size_t> inner_table_idx;
  for (size_t i = 0; i < query_infos.size(); ++i) {
    if (query_infos[i].table_id == inner_table_id) {
      inner_table_idx = i;
      break;
    }
  }
  CHECK(inner_table_idx);
  return query_infos[*inner_table_idx].info.getNumTuples() <=
         g_trivial_loop_join_threshold;
}

namespace {

template <typename T>
std::vector<std::string> expr_container_to_string(const T& expr_container) {
  std::vector<std::string> expr_strs;
  for (const auto& expr : expr_container) {
    if (!expr) {
      expr_strs.emplace_back("NULL");
    } else {
      expr_strs.emplace_back(expr->toString());
    }
  }
  return expr_strs;
}

template <>
std::vector<std::string> expr_container_to_string(
    const std::list<Analyzer::OrderEntry>& expr_container) {
  std::vector<std::string> expr_strs;
  for (const auto& expr : expr_container) {
    expr_strs.emplace_back(expr.toString());
  }
  return expr_strs;
}

std::string join_type_to_string(const JoinType type) {
  switch (type) {
    case JoinType::INNER:
      return "INNER";
    case JoinType::LEFT:
      return "LEFT";
    case JoinType::INVALID:
      return "INVALID";
  }
  UNREACHABLE();
  return "";
}

std::string sort_algorithm_to_string(const SortAlgorithm algorithm) {
  switch (algorithm) {
    case SortAlgorithm::Default:
      return "ResultSet";
    case SortAlgorithm::SpeculativeTopN:
      return "Speculative Top N";
    case SortAlgorithm::StreamingTopN:
      return "Streaming Top N";
  }
  UNREACHABLE();
  return "";
}

}  // namespace

std::string ra_exec_unit_desc_for_caching(const RelAlgExecutionUnit& ra_exe_unit) {
  // todo(yoonmin): replace a cache key as a DAG representation of a query plan
  // instead of ra_exec_unit description if possible
  std::ostringstream os;
  for (const auto& input_col_desc : ra_exe_unit.input_col_descs) {
    const auto& scan_desc = input_col_desc->getScanDesc();
    os << scan_desc.getTableId() << "," << input_col_desc->getColId() << ","
       << scan_desc.getNestLevel();
  }
  if (!ra_exe_unit.simple_quals.empty()) {
    for (const auto& qual : ra_exe_unit.simple_quals) {
      if (qual) {
        os << qual->toString() << ",";
      }
    }
  }
  if (!ra_exe_unit.quals.empty()) {
    for (const auto& qual : ra_exe_unit.quals) {
      if (qual) {
        os << qual->toString() << ",";
      }
    }
  }
  if (!ra_exe_unit.join_quals.empty()) {
    for (size_t i = 0; i < ra_exe_unit.join_quals.size(); i++) {
      const auto& join_condition = ra_exe_unit.join_quals[i];
      os << std::to_string(i) << join_type_to_string(join_condition.type);
      for (const auto& qual : join_condition.quals) {
        if (qual) {
          os << qual->toString() << ",";
        }
      }
    }
  }
  if (!ra_exe_unit.groupby_exprs.empty()) {
    for (const auto& qual : ra_exe_unit.groupby_exprs) {
      if (qual) {
        os << qual->toString() << ",";
      }
    }
  }
  for (const auto& expr : ra_exe_unit.target_exprs) {
    if (expr) {
      os << expr->toString() << ",";
    }
  }
  os << ::toString(ra_exe_unit.estimator == nullptr);
  os << std::to_string(ra_exe_unit.scan_limit);
  return os.str();
}

std::ostream& operator<<(std::ostream& os, const RelAlgExecutionUnit& ra_exe_unit) {
  os << "\n\tTable/Col/Levels: ";
  for (const auto& input_col_desc : ra_exe_unit.input_col_descs) {
    const auto& scan_desc = input_col_desc->getScanDesc();
    os << "(" << scan_desc.getTableId() << ", " << input_col_desc->getColId() << ", "
       << scan_desc.getNestLevel() << ") ";
  }
  if (!ra_exe_unit.simple_quals.empty()) {
    os << "\n\tSimple Quals: "
       << boost::algorithm::join(expr_container_to_string(ra_exe_unit.simple_quals),
                                 ", ");
  }
  if (!ra_exe_unit.quals.empty()) {
    os << "\n\tQuals: "
       << boost::algorithm::join(expr_container_to_string(ra_exe_unit.quals), ", ");
  }
  if (!ra_exe_unit.join_quals.empty()) {
    os << "\n\tJoin Quals: ";
    for (size_t i = 0; i < ra_exe_unit.join_quals.size(); i++) {
      const auto& join_condition = ra_exe_unit.join_quals[i];
      os << "\t\t" << std::to_string(i) << " "
         << join_type_to_string(join_condition.type);
      os << boost::algorithm::join(expr_container_to_string(join_condition.quals), ", ");
    }
  }
  if (!ra_exe_unit.groupby_exprs.empty()) {
    os << "\n\tGroup By: "
       << boost::algorithm::join(expr_container_to_string(ra_exe_unit.groupby_exprs),
                                 ", ");
  }
  os << "\n\tProjected targets: "
     << boost::algorithm::join(expr_container_to_string(ra_exe_unit.target_exprs), ", ");
  os << "\n\tHas Estimator: " << ::toString(ra_exe_unit.estimator == nullptr);
  os << "\n\tSort Info: ";
  const auto& sort_info = ra_exe_unit.sort_info;
  os << "\n\t  Order Entries: "
     << boost::algorithm::join(expr_container_to_string(sort_info.order_entries), ", ");
  os << "\n\t  Algorithm: " << sort_algorithm_to_string(sort_info.algorithm);
  os << "\n\t  Limit: " << std::to_string(sort_info.limit);
  os << "\n\t  Offset: " << std::to_string(sort_info.offset);
  os << "\n\tScan Limit: " << std::to_string(ra_exe_unit.scan_limit);
  os << "\n\tBump Allocator: " << ::toString(ra_exe_unit.use_bump_allocator);
  if (ra_exe_unit.union_all) {
    os << "\n\tUnion: " << std::string(*ra_exe_unit.union_all ? "UNION ALL" : "UNION");
  }
  return os;
}

namespace {

RelAlgExecutionUnit replace_scan_limit(const RelAlgExecutionUnit& ra_exe_unit_in,
                                       const size_t new_scan_limit) {
  return {ra_exe_unit_in.input_descs,
          ra_exe_unit_in.input_col_descs,
          ra_exe_unit_in.simple_quals,
          ra_exe_unit_in.quals,
          ra_exe_unit_in.join_quals,
          ra_exe_unit_in.groupby_exprs,
          ra_exe_unit_in.target_exprs,
          ra_exe_unit_in.estimator,
          ra_exe_unit_in.sort_info,
          new_scan_limit,
          ra_exe_unit_in.query_hint,
          ra_exe_unit_in.use_bump_allocator,
          ra_exe_unit_in.union_all,
          ra_exe_unit_in.query_state};
}

}  // namespace

ResultSetPtr Executor::executeWorkUnit(size_t& max_groups_buffer_entry_guess,
                                       const bool is_agg,
                                       const std::vector<InputTableInfo>& query_infos,
                                       const RelAlgExecutionUnit& ra_exe_unit_in,
                                       const CompilationOptions& co,
                                       const ExecutionOptions& eo,
                                       const Catalog_Namespace::Catalog& cat,
                                       RenderInfo* render_info,
                                       const bool has_cardinality_estimation,
                                       ColumnCacheMap& column_cache) {
  VLOG(1) << "Executor " << executor_id_ << " is executing work unit:" << ra_exe_unit_in;

  ScopeGuard cleanup_post_execution = [this] {
    // cleanup/unpin GPU buffer allocations
    // TODO: separate out this state into a single object
    plan_state_.reset(nullptr);
    if (cgen_state_) {
      cgen_state_->in_values_bitmaps_.clear();
    }
  };

  try {
    auto result = executeWorkUnitImpl(max_groups_buffer_entry_guess,
                                      is_agg,
                                      true,
                                      query_infos,
                                      ra_exe_unit_in,
                                      co,
                                      eo,
                                      cat,
                                      row_set_mem_owner_,
                                      render_info,
                                      has_cardinality_estimation,
                                      column_cache);
    if (result) {
      result->setKernelQueueTime(kernel_queue_time_ms_);
      result->addCompilationQueueTime(compilation_queue_time_ms_);
      if (eo.just_validate) {
        result->setValidationOnlyRes();
      }
    }
    return result;
  } catch (const CompilationRetryNewScanLimit& e) {
    auto result =
        executeWorkUnitImpl(max_groups_buffer_entry_guess,
                            is_agg,
                            false,
                            query_infos,
                            replace_scan_limit(ra_exe_unit_in, e.new_scan_limit_),
                            co,
                            eo,
                            cat,
                            row_set_mem_owner_,
                            render_info,
                            has_cardinality_estimation,
                            column_cache);
    if (result) {
      result->setKernelQueueTime(kernel_queue_time_ms_);
      result->addCompilationQueueTime(compilation_queue_time_ms_);
      if (eo.just_validate) {
        result->setValidationOnlyRes();
      }
    }
    return result;
  }
}

ResultSetPtr Executor::executeWorkUnitImpl(
    size_t& max_groups_buffer_entry_guess,
    const bool is_agg,
    const bool allow_single_frag_table_opt,
    const std::vector<InputTableInfo>& query_infos,
    const RelAlgExecutionUnit& ra_exe_unit_in,
    const CompilationOptions& co,
    const ExecutionOptions& eo,
    const Catalog_Namespace::Catalog& cat,
    std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner,
    RenderInfo* render_info,
    const bool has_cardinality_estimation,
    ColumnCacheMap& column_cache) {
  INJECT_TIMER(Exec_executeWorkUnit);
  const auto [ra_exe_unit, deleted_cols_map] = addDeletedColumn(ra_exe_unit_in, co);
  const auto device_type = getDeviceTypeForTargets(ra_exe_unit, co.device_type);
  CHECK(!query_infos.empty());
  if (!max_groups_buffer_entry_guess) {
    // The query has failed the first execution attempt because of running out
    // of group by slots. Make the conservative choice: allocate fragment size
    // slots and run on the CPU.
    CHECK(device_type == ExecutorDeviceType::CPU);
    max_groups_buffer_entry_guess = compute_buffer_entry_guess(query_infos);
  }

  int8_t crt_min_byte_width{get_min_byte_width()};
  do {
    SharedKernelContext shared_context(query_infos);
    ColumnFetcher column_fetcher(this, column_cache);
    auto query_comp_desc_owned = std::make_unique<QueryCompilationDescriptor>();
    std::unique_ptr<QueryMemoryDescriptor> query_mem_desc_owned;
    if (eo.executor_type == ExecutorType::Native) {
      try {
        INJECT_TIMER(query_step_compilation);
        auto clock_begin = timer_start();
        std::lock_guard<std::mutex> compilation_lock(compilation_mutex_);
        compilation_queue_time_ms_ += timer_stop(clock_begin);

        query_mem_desc_owned =
            query_comp_desc_owned->compile(max_groups_buffer_entry_guess,
                                           crt_min_byte_width,
                                           has_cardinality_estimation,
                                           ra_exe_unit,
                                           query_infos,
                                           deleted_cols_map,
                                           column_fetcher,
                                           {device_type,
                                            co.hoist_literals,
                                            co.opt_level,
                                            co.with_dynamic_watchdog,
                                            co.allow_lazy_fetch,
                                            co.filter_on_deleted_column,
                                            co.explain_type,
                                            co.register_intel_jit_listener},
                                           eo,
                                           render_info,
                                           this);
        CHECK(query_mem_desc_owned);
        crt_min_byte_width = query_comp_desc_owned->getMinByteWidth();
      } catch (CompilationRetryNoCompaction&) {
        crt_min_byte_width = MAX_BYTE_WIDTH_SUPPORTED;
        continue;
      }
    } else {
      plan_state_.reset(new PlanState(false, query_infos, deleted_cols_map, this));
      plan_state_->allocateLocalColumnIds(ra_exe_unit.input_col_descs);
      CHECK(!query_mem_desc_owned);
      query_mem_desc_owned.reset(
          new QueryMemoryDescriptor(this, 0, QueryDescriptionType::Projection, false));
    }
    if (eo.just_explain) {
      return executeExplain(*query_comp_desc_owned);
    }

    for (const auto target_expr : ra_exe_unit.target_exprs) {
      plan_state_->target_exprs_.push_back(target_expr);
    }

    if (!eo.just_validate) {
      int available_cpus = cpu_threads();
      auto available_gpus = get_available_gpus(cat);

      const auto context_count =
          get_context_count(device_type, available_cpus, available_gpus.size());
      try {
        auto kernels = createKernels(shared_context,
                                     ra_exe_unit,
                                     column_fetcher,
                                     query_infos,
                                     eo,
                                     is_agg,
                                     allow_single_frag_table_opt,
                                     context_count,
                                     *query_comp_desc_owned,
                                     *query_mem_desc_owned,
                                     render_info,
                                     available_gpus,
                                     available_cpus);
        if (g_use_tbb_pool) {
#ifdef HAVE_TBB
          VLOG(1) << "Using TBB thread pool for kernel dispatch.";
          launchKernels<threadpool::TbbThreadPool<void>>(shared_context,
                                                         std::move(kernels));
#else
          throw std::runtime_error(
              "This build is not TBB enabled. Restart the server with "
              "\"enable-modern-thread-pool\" disabled.");
#endif
        } else {
          launchKernels<threadpool::FuturesThreadPool<void>>(shared_context,
                                                             std::move(kernels));
        }
      } catch (QueryExecutionError& e) {
        if (eo.with_dynamic_watchdog && interrupted_.load() &&
            e.getErrorCode() == ERR_OUT_OF_TIME) {
          resetInterrupt();
          throw QueryExecutionError(ERR_INTERRUPTED);
        }
        if (eo.allow_runtime_query_interrupt && interrupted_.load()) {
          resetInterrupt();
          throw QueryExecutionError(ERR_INTERRUPTED);
        }
        if (e.getErrorCode() == ERR_OVERFLOW_OR_UNDERFLOW &&
            static_cast<size_t>(crt_min_byte_width << 1) <= sizeof(int64_t)) {
          crt_min_byte_width <<= 1;
          continue;
        }
        throw;
      }
    }
    if (is_agg) {
      try {
        return collectAllDeviceResults(shared_context,
                                       ra_exe_unit,
                                       *query_mem_desc_owned,
                                       query_comp_desc_owned->getDeviceType(),
                                       row_set_mem_owner);
      } catch (ReductionRanOutOfSlots&) {
        throw QueryExecutionError(ERR_OUT_OF_SLOTS);
      } catch (OverflowOrUnderflow&) {
        crt_min_byte_width <<= 1;
        continue;
      }
    }
    return resultsUnion(shared_context, ra_exe_unit);

  } while (static_cast<size_t>(crt_min_byte_width) <= sizeof(int64_t));

  return std::make_shared<ResultSet>(std::vector<TargetInfo>{},
                                     ExecutorDeviceType::CPU,
                                     QueryMemoryDescriptor(),
                                     nullptr,
                                     catalog_,
                                     blockSize(),
                                     gridSize());
}

void Executor::executeWorkUnitPerFragment(const RelAlgExecutionUnit& ra_exe_unit_in,
                                          const InputTableInfo& table_info,
                                          const CompilationOptions& co,
                                          const ExecutionOptions& eo,
                                          const Catalog_Namespace::Catalog& cat,
                                          PerFragmentCallBack& cb) {
  const auto [ra_exe_unit, deleted_cols_map] = addDeletedColumn(ra_exe_unit_in, co);
  ColumnCacheMap column_cache;

  std::vector<InputTableInfo> table_infos{table_info};
  SharedKernelContext kernel_context(table_infos);

  ColumnFetcher column_fetcher(this, column_cache);
  auto query_comp_desc_owned = std::make_unique<QueryCompilationDescriptor>();
  std::unique_ptr<QueryMemoryDescriptor> query_mem_desc_owned;
  {
    auto clock_begin = timer_start();
    std::lock_guard<std::mutex> compilation_lock(compilation_mutex_);
    compilation_queue_time_ms_ += timer_stop(clock_begin);
    query_mem_desc_owned =
        query_comp_desc_owned->compile(0,
                                       8,
                                       /*has_cardinality_estimation=*/false,
                                       ra_exe_unit,
                                       table_infos,
                                       deleted_cols_map,
                                       column_fetcher,
                                       co,
                                       eo,
                                       nullptr,
                                       this);
  }
  CHECK(query_mem_desc_owned);
  CHECK_EQ(size_t(1), ra_exe_unit.input_descs.size());
  const auto table_id = ra_exe_unit.input_descs[0].getTableId();
  const auto& outer_fragments = table_info.info.fragments;

  {
    auto clock_begin = timer_start();
    std::lock_guard<std::mutex> kernel_lock(kernel_mutex_);
    kernel_queue_time_ms_ += timer_stop(clock_begin);

    for (size_t fragment_index = 0; fragment_index < outer_fragments.size();
         ++fragment_index) {
      // We may want to consider in the future allowing this to execute on devices other
      // than CPU
      FragmentsList fragments_list{{table_id, {fragment_index}}};
      ExecutionKernel kernel(ra_exe_unit,
                             co.device_type,
                             /*device_id=*/0,
                             eo,
                             column_fetcher,
                             *query_comp_desc_owned,
                             *query_mem_desc_owned,
                             fragments_list,
                             ExecutorDispatchMode::KernelPerFragment,
                             /*render_info=*/nullptr,
                             /*rowid_lookup_key=*/-1);
      kernel.run(this, kernel_context);
    }
  }

  const auto& all_fragment_results = kernel_context.getFragmentResults();

  for (size_t fragment_index = 0; fragment_index < outer_fragments.size();
       ++fragment_index) {
    const auto fragment_results = all_fragment_results[fragment_index];
    cb(fragment_results.first, outer_fragments[fragment_index]);
  }
}

ResultSetPtr Executor::executeTableFunction(
    const TableFunctionExecutionUnit exe_unit,
    const std::vector<InputTableInfo>& table_infos,
    const CompilationOptions& co,
    const ExecutionOptions& eo,
    const Catalog_Namespace::Catalog& cat) {
  INJECT_TIMER(Exec_executeTableFunction);
  nukeOldState(false, table_infos, PlanState::DeletedColumnsMap{}, nullptr);

  ColumnCacheMap column_cache;  // Note: if we add retries to the table function
                                // framework, we may want to move this up a level

  ColumnFetcher column_fetcher(this, column_cache);
  TableFunctionCompilationContext compilation_context;
  compilation_context.compile(exe_unit, co, this);

  TableFunctionExecutionContext exe_context(getRowSetMemoryOwner());
  return exe_context.execute(
      exe_unit, table_infos, &compilation_context, column_fetcher, co.device_type, this);
}

ResultSetPtr Executor::executeExplain(const QueryCompilationDescriptor& query_comp_desc) {
  return std::make_shared<ResultSet>(query_comp_desc.getIR());
}

ExecutorDeviceType Executor::getDeviceTypeForTargets(
    const RelAlgExecutionUnit& ra_exe_unit,
    const ExecutorDeviceType requested_device_type) {
  for (const auto target_expr : ra_exe_unit.target_exprs) {
    const auto agg_info = get_target_info(target_expr, g_bigint_count);
    if (!ra_exe_unit.groupby_exprs.empty() &&
        !isArchPascalOrLater(requested_device_type)) {
      if ((agg_info.agg_kind == kAVG || agg_info.agg_kind == kSUM) &&
          agg_info.agg_arg_type.get_type() == kDOUBLE) {
        return ExecutorDeviceType::CPU;
      }
    }
    if (dynamic_cast<const Analyzer::RegexpExpr*>(target_expr)) {
      return ExecutorDeviceType::CPU;
    }
  }
  return requested_device_type;
}

namespace {

int64_t inline_null_val(const SQLTypeInfo& ti, const bool float_argument_input) {
  CHECK(ti.is_number() || ti.is_time() || ti.is_boolean() || ti.is_string());
  if (ti.is_fp()) {
    if (float_argument_input && ti.get_type() == kFLOAT) {
      int64_t float_null_val = 0;
      *reinterpret_cast<float*>(may_alias_ptr(&float_null_val)) =
          static_cast<float>(inline_fp_null_val(ti));
      return float_null_val;
    }
    const auto double_null_val = inline_fp_null_val(ti);
    return *reinterpret_cast<const int64_t*>(may_alias_ptr(&double_null_val));
  }
  return inline_int_null_val(ti);
}

void fill_entries_for_empty_input(std::vector<TargetInfo>& target_infos,
                                  std::vector<int64_t>& entry,
                                  const std::vector<Analyzer::Expr*>& target_exprs,
                                  const QueryMemoryDescriptor& query_mem_desc) {
  for (size_t target_idx = 0; target_idx < target_exprs.size(); ++target_idx) {
    const auto target_expr = target_exprs[target_idx];
    const auto agg_info = get_target_info(target_expr, g_bigint_count);
    CHECK(agg_info.is_agg);
    target_infos.push_back(agg_info);
    if (g_cluster) {
      const auto executor = query_mem_desc.getExecutor();
      CHECK(executor);
      auto row_set_mem_owner = executor->getRowSetMemoryOwner();
      CHECK(row_set_mem_owner);
      const auto& count_distinct_desc =
          query_mem_desc.getCountDistinctDescriptor(target_idx);
      if (count_distinct_desc.impl_type_ == CountDistinctImplType::Bitmap) {
        CHECK(row_set_mem_owner);
        auto count_distinct_buffer = row_set_mem_owner->allocateCountDistinctBuffer(
            count_distinct_desc.bitmapPaddedSizeBytes());
        entry.push_back(reinterpret_cast<int64_t>(count_distinct_buffer));
        continue;
      }
      if (count_distinct_desc.impl_type_ == CountDistinctImplType::StdSet) {
        auto count_distinct_set = new std::set<int64_t>();
        CHECK(row_set_mem_owner);
        row_set_mem_owner->addCountDistinctSet(count_distinct_set);
        entry.push_back(reinterpret_cast<int64_t>(count_distinct_set));
        continue;
      }
    }
    const bool float_argument_input = takes_float_argument(agg_info);
    if (agg_info.agg_kind == kCOUNT || agg_info.agg_kind == kAPPROX_COUNT_DISTINCT) {
      entry.push_back(0);
    } else if (agg_info.agg_kind == kAVG) {
      entry.push_back(inline_null_val(agg_info.sql_type, float_argument_input));
      entry.push_back(0);
    } else if (agg_info.agg_kind == kSINGLE_VALUE || agg_info.agg_kind == kSAMPLE) {
      if (agg_info.sql_type.is_geometry()) {
        for (int i = 0; i < agg_info.sql_type.get_physical_coord_cols() * 2; i++) {
          entry.push_back(0);
        }
      } else if (agg_info.sql_type.is_varlen()) {
        entry.push_back(0);
        entry.push_back(0);
      } else {
        entry.push_back(inline_null_val(agg_info.sql_type, float_argument_input));
      }
    } else {
      entry.push_back(inline_null_val(agg_info.sql_type, float_argument_input));
    }
  }
}

ResultSetPtr build_row_for_empty_input(
    const std::vector<Analyzer::Expr*>& target_exprs_in,
    const QueryMemoryDescriptor& query_mem_desc,
    const ExecutorDeviceType device_type) {
  std::vector<std::shared_ptr<Analyzer::Expr>> target_exprs_owned_copies;
  std::vector<Analyzer::Expr*> target_exprs;
  for (const auto target_expr : target_exprs_in) {
    const auto target_expr_copy =
        std::dynamic_pointer_cast<Analyzer::AggExpr>(target_expr->deep_copy());
    CHECK(target_expr_copy);
    auto ti = target_expr->get_type_info();
    ti.set_notnull(false);
    target_expr_copy->set_type_info(ti);
    if (target_expr_copy->get_arg()) {
      auto arg_ti = target_expr_copy->get_arg()->get_type_info();
      arg_ti.set_notnull(false);
      target_expr_copy->get_arg()->set_type_info(arg_ti);
    }
    target_exprs_owned_copies.push_back(target_expr_copy);
    target_exprs.push_back(target_expr_copy.get());
  }
  std::vector<TargetInfo> target_infos;
  std::vector<int64_t> entry;
  fill_entries_for_empty_input(target_infos, entry, target_exprs, query_mem_desc);
  const auto executor = query_mem_desc.getExecutor();
  CHECK(executor);
  auto row_set_mem_owner = executor->getRowSetMemoryOwner();
  CHECK(row_set_mem_owner);
  auto rs = std::make_shared<ResultSet>(target_infos,
                                        device_type,
                                        query_mem_desc,
                                        row_set_mem_owner,
                                        executor->getCatalog(),
                                        executor->blockSize(),
                                        executor->gridSize());
  rs->allocateStorage();
  rs->fillOneEntry(entry);
  return rs;
}

}  // namespace

ResultSetPtr Executor::collectAllDeviceResults(
    SharedKernelContext& shared_context,
    const RelAlgExecutionUnit& ra_exe_unit,
    const QueryMemoryDescriptor& query_mem_desc,
    const ExecutorDeviceType device_type,
    std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner) {
  auto timer = DEBUG_TIMER(__func__);
  auto& result_per_device = shared_context.getFragmentResults();
  if (result_per_device.empty() && query_mem_desc.getQueryDescriptionType() ==
                                       QueryDescriptionType::NonGroupedAggregate) {
    return build_row_for_empty_input(
        ra_exe_unit.target_exprs, query_mem_desc, device_type);
  }
  if (use_speculative_top_n(ra_exe_unit, query_mem_desc)) {
    try {
      return reduceSpeculativeTopN(
          ra_exe_unit, result_per_device, row_set_mem_owner, query_mem_desc);
    } catch (const std::bad_alloc&) {
      throw SpeculativeTopNFailed("Failed during multi-device reduction.");
    }
  }
  const auto shard_count =
      device_type == ExecutorDeviceType::GPU
          ? GroupByAndAggregate::shard_count_for_top_groups(ra_exe_unit, *catalog_)
          : 0;

  if (shard_count && !result_per_device.empty()) {
    return collectAllDeviceShardedTopResults(shared_context, ra_exe_unit);
  }
  return reduceMultiDeviceResults(
      ra_exe_unit, result_per_device, row_set_mem_owner, query_mem_desc);
}

namespace {
size_t permute_storage_columnar(const ResultSetStorage* input_storage,
                                const QueryMemoryDescriptor& input_query_mem_desc,
                                const ResultSetStorage* output_storage,
                                size_t output_row_index,
                                const QueryMemoryDescriptor& output_query_mem_desc,
                                const std::vector<uint32_t>& top_permutation) {
  const auto output_buffer = output_storage->getUnderlyingBuffer();
  const auto input_buffer = input_storage->getUnderlyingBuffer();
  for (const auto sorted_idx : top_permutation) {
    // permuting all group-columns in this result set into the final buffer:
    for (size_t group_idx = 0; group_idx < input_query_mem_desc.getKeyCount();
         group_idx++) {
      const auto input_column_ptr =
          input_buffer + input_query_mem_desc.getPrependedGroupColOffInBytes(group_idx) +
          sorted_idx * input_query_mem_desc.groupColWidth(group_idx);
      const auto output_column_ptr =
          output_buffer +
          output_query_mem_desc.getPrependedGroupColOffInBytes(group_idx) +
          output_row_index * output_query_mem_desc.groupColWidth(group_idx);
      memcpy(output_column_ptr,
             input_column_ptr,
             output_query_mem_desc.groupColWidth(group_idx));
    }
    // permuting all agg-columns in this result set into the final buffer:
    for (size_t slot_idx = 0; slot_idx < input_query_mem_desc.getSlotCount();
         slot_idx++) {
      const auto input_column_ptr =
          input_buffer + input_query_mem_desc.getColOffInBytes(slot_idx) +
          sorted_idx * input_query_mem_desc.getPaddedSlotWidthBytes(slot_idx);
      const auto output_column_ptr =
          output_buffer + output_query_mem_desc.getColOffInBytes(slot_idx) +
          output_row_index * output_query_mem_desc.getPaddedSlotWidthBytes(slot_idx);
      memcpy(output_column_ptr,
             input_column_ptr,
             output_query_mem_desc.getPaddedSlotWidthBytes(slot_idx));
    }
    ++output_row_index;
  }
  return output_row_index;
}

size_t permute_storage_row_wise(const ResultSetStorage* input_storage,
                                const ResultSetStorage* output_storage,
                                size_t output_row_index,
                                const QueryMemoryDescriptor& output_query_mem_desc,
                                const std::vector<uint32_t>& top_permutation) {
  const auto output_buffer = output_storage->getUnderlyingBuffer();
  const auto input_buffer = input_storage->getUnderlyingBuffer();
  for (const auto sorted_idx : top_permutation) {
    const auto row_ptr = input_buffer + sorted_idx * output_query_mem_desc.getRowSize();
    memcpy(output_buffer + output_row_index * output_query_mem_desc.getRowSize(),
           row_ptr,
           output_query_mem_desc.getRowSize());
    ++output_row_index;
  }
  return output_row_index;
}
}  // namespace

// Collect top results from each device, stitch them together and sort. Partial
// results from each device are guaranteed to be disjunct because we only go on
// this path when one of the columns involved is a shard key.
ResultSetPtr Executor::collectAllDeviceShardedTopResults(
    SharedKernelContext& shared_context,
    const RelAlgExecutionUnit& ra_exe_unit) const {
  auto& result_per_device = shared_context.getFragmentResults();
  const auto first_result_set = result_per_device.front().first;
  CHECK(first_result_set);
  auto top_query_mem_desc = first_result_set->getQueryMemDesc();
  CHECK(!top_query_mem_desc.hasInterleavedBinsOnGpu());
  const auto top_n = ra_exe_unit.sort_info.limit + ra_exe_unit.sort_info.offset;
  top_query_mem_desc.setEntryCount(0);
  for (auto& result : result_per_device) {
    const auto result_set = result.first;
    CHECK(result_set);
    result_set->sort(ra_exe_unit.sort_info.order_entries, top_n, this);
    size_t new_entry_cnt = top_query_mem_desc.getEntryCount() + result_set->rowCount();
    top_query_mem_desc.setEntryCount(new_entry_cnt);
  }
  auto top_result_set = std::make_shared<ResultSet>(first_result_set->getTargetInfos(),
                                                    first_result_set->getDeviceType(),
                                                    top_query_mem_desc,
                                                    first_result_set->getRowSetMemOwner(),
                                                    catalog_,
                                                    blockSize(),
                                                    gridSize());
  auto top_storage = top_result_set->allocateStorage();
  size_t top_output_row_idx{0};
  for (auto& result : result_per_device) {
    const auto result_set = result.first;
    CHECK(result_set);
    const auto& top_permutation = result_set->getPermutationBuffer();
    CHECK_LE(top_permutation.size(), top_n);
    if (top_query_mem_desc.didOutputColumnar()) {
      top_output_row_idx = permute_storage_columnar(result_set->getStorage(),
                                                    result_set->getQueryMemDesc(),
                                                    top_storage,
                                                    top_output_row_idx,
                                                    top_query_mem_desc,
                                                    top_permutation);
    } else {
      top_output_row_idx = permute_storage_row_wise(result_set->getStorage(),
                                                    top_storage,
                                                    top_output_row_idx,
                                                    top_query_mem_desc,
                                                    top_permutation);
    }
  }
  CHECK_EQ(top_output_row_idx, top_query_mem_desc.getEntryCount());
  return top_result_set;
}

std::unordered_map<int, const Analyzer::BinOper*> Executor::getInnerTabIdToJoinCond()
    const {
  std::unordered_map<int, const Analyzer::BinOper*> id_to_cond;
  const auto& join_info = plan_state_->join_info_;
  CHECK_EQ(join_info.equi_join_tautologies_.size(), join_info.join_hash_tables_.size());
  for (size_t i = 0; i < join_info.join_hash_tables_.size(); ++i) {
    int inner_table_id = join_info.join_hash_tables_[i]->getInnerTableId();
    id_to_cond.insert(
        std::make_pair(inner_table_id, join_info.equi_join_tautologies_[i].get()));
  }
  return id_to_cond;
}

namespace {

bool has_lazy_fetched_columns(const std::vector<ColumnLazyFetchInfo>& fetched_cols) {
  for (const auto& col : fetched_cols) {
    if (col.is_lazily_fetched) {
      return true;
    }
  }
  return false;
}

}  // namespace

std::vector<std::unique_ptr<ExecutionKernel>> Executor::createKernels(
    SharedKernelContext& shared_context,
    const RelAlgExecutionUnit& ra_exe_unit,
    ColumnFetcher& column_fetcher,
    const std::vector<InputTableInfo>& table_infos,
    const ExecutionOptions& eo,
    const bool is_agg,
    const bool allow_single_frag_table_opt,
    const size_t context_count,
    const QueryCompilationDescriptor& query_comp_desc,
    const QueryMemoryDescriptor& query_mem_desc,
    RenderInfo* render_info,
    std::unordered_set<int>& available_gpus,
    int& available_cpus) {
  std::vector<std::unique_ptr<ExecutionKernel>> execution_kernels;

  QueryFragmentDescriptor fragment_descriptor(
      ra_exe_unit,
      table_infos,
      query_comp_desc.getDeviceType() == ExecutorDeviceType::GPU
          ? catalog_->getDataMgr().getMemoryInfo(Data_Namespace::MemoryLevel::GPU_LEVEL)
          : std::vector<Data_Namespace::MemoryInfo>{},
      eo.gpu_input_mem_limit_percent,
      eo.outer_fragment_indices);
  CHECK(!ra_exe_unit.input_descs.empty());

  const auto device_type = query_comp_desc.getDeviceType();
  const bool uses_lazy_fetch =
      plan_state_->allow_lazy_fetch_ &&
      has_lazy_fetched_columns(getColLazyFetchInfo(ra_exe_unit.target_exprs));
  const bool use_multifrag_kernel = (device_type == ExecutorDeviceType::GPU) &&
                                    eo.allow_multifrag && (!uses_lazy_fetch || is_agg);
  const auto device_count = deviceCount(device_type);
  CHECK_GT(device_count, 0);

  fragment_descriptor.buildFragmentKernelMap(ra_exe_unit,
                                             shared_context.getFragOffsets(),
                                             device_count,
                                             device_type,
                                             use_multifrag_kernel,
                                             g_inner_join_fragment_skipping,
                                             this);
  if (eo.with_watchdog && fragment_descriptor.shouldCheckWorkUnitWatchdog()) {
    checkWorkUnitWatchdog(ra_exe_unit, table_infos, *catalog_, device_type, device_count);
  }

  if (use_multifrag_kernel) {
    VLOG(1) << "Creating multifrag execution kernels";
    VLOG(1) << query_mem_desc.toString();

    // NB: We should never be on this path when the query is retried because of running
    // out of group by slots; also, for scan only queries on CPU we want the
    // high-granularity, fragment by fragment execution instead. For scan only queries on
    // GPU, we want the multifrag kernel path to save the overhead of allocating an output
    // buffer per fragment.
    auto multifrag_kernel_dispatch = [&ra_exe_unit,
                                      &execution_kernels,
                                      &column_fetcher,
                                      &eo,
                                      &query_comp_desc,
                                      &query_mem_desc,
                                      render_info](const int device_id,
                                                   const FragmentsList& frag_list,
                                                   const int64_t rowid_lookup_key) {
      execution_kernels.emplace_back(
          std::make_unique<ExecutionKernel>(ra_exe_unit,
                                            ExecutorDeviceType::GPU,
                                            device_id,
                                            eo,
                                            column_fetcher,
                                            query_comp_desc,
                                            query_mem_desc,
                                            frag_list,
                                            ExecutorDispatchMode::MultifragmentKernel,
                                            render_info,
                                            rowid_lookup_key));
    };
    fragment_descriptor.assignFragsToMultiDispatch(multifrag_kernel_dispatch);
  } else {
    VLOG(1) << "Creating one execution kernel per fragment";
    VLOG(1) << query_mem_desc.toString();

    if (!ra_exe_unit.use_bump_allocator && allow_single_frag_table_opt &&
        (query_mem_desc.getQueryDescriptionType() == QueryDescriptionType::Projection) &&
        table_infos.size() == 1 && table_infos.front().table_id > 0) {
      const auto max_frag_size =
          table_infos.front().info.getFragmentNumTuplesUpperBound();
      if (max_frag_size < query_mem_desc.getEntryCount()) {
        LOG(INFO) << "Lowering scan limit from " << query_mem_desc.getEntryCount()
                  << " to match max fragment size " << max_frag_size
                  << " for kernel per fragment execution path.";
        throw CompilationRetryNewScanLimit(max_frag_size);
      }
    }

    size_t frag_list_idx{0};
    auto fragment_per_kernel_dispatch = [&ra_exe_unit,
                                         &execution_kernels,
                                         &column_fetcher,
                                         &eo,
                                         &frag_list_idx,
                                         &device_type,
                                         &query_comp_desc,
                                         &query_mem_desc,
                                         render_info](const int device_id,
                                                      const FragmentsList& frag_list,
                                                      const int64_t rowid_lookup_key) {
      if (!frag_list.size()) {
        return;
      }
      CHECK_GE(device_id, 0);

      execution_kernels.emplace_back(
          std::make_unique<ExecutionKernel>(ra_exe_unit,
                                            device_type,
                                            device_id,
                                            eo,
                                            column_fetcher,
                                            query_comp_desc,
                                            query_mem_desc,
                                            frag_list,
                                            ExecutorDispatchMode::KernelPerFragment,
                                            render_info,
                                            rowid_lookup_key));
      ++frag_list_idx;
    };

    fragment_descriptor.assignFragsToKernelDispatch(fragment_per_kernel_dispatch,
                                                    ra_exe_unit);
  }

  return execution_kernels;
}

template <typename THREAD_POOL>
void Executor::launchKernels(SharedKernelContext& shared_context,
                             std::vector<std::unique_ptr<ExecutionKernel>>&& kernels) {
  auto clock_begin = timer_start();
  std::lock_guard<std::mutex> kernel_lock(kernel_mutex_);
  kernel_queue_time_ms_ += timer_stop(clock_begin);

  THREAD_POOL thread_pool;
  VLOG(1) << "Launching " << kernels.size() << " kernels for query.";
  for (auto& kernel : kernels) {
    thread_pool.spawn(
        [this, &shared_context, parent_thread_id = logger::thread_id()](
            ExecutionKernel* kernel) {
          CHECK(kernel);
          DEBUG_TIMER_NEW_THREAD(parent_thread_id);
          kernel->run(this, shared_context);
        },
        kernel.get());
  }
  thread_pool.join();
}

std::vector<size_t> Executor::getTableFragmentIndices(
    const RelAlgExecutionUnit& ra_exe_unit,
    const ExecutorDeviceType device_type,
    const size_t table_idx,
    const size_t outer_frag_idx,
    std::map<int, const TableFragments*>& selected_tables_fragments,
    const std::unordered_map<int, const Analyzer::BinOper*>&
        inner_table_id_to_join_condition) {
  const int table_id = ra_exe_unit.input_descs[table_idx].getTableId();
  auto table_frags_it = selected_tables_fragments.find(table_id);
  CHECK(table_frags_it != selected_tables_fragments.end());
  const auto& outer_input_desc = ra_exe_unit.input_descs[0];
  const auto outer_table_fragments_it =
      selected_tables_fragments.find(outer_input_desc.getTableId());
  const auto outer_table_fragments = outer_table_fragments_it->second;
  CHECK(outer_table_fragments_it != selected_tables_fragments.end());
  CHECK_LT(outer_frag_idx, outer_table_fragments->size());
  if (!table_idx) {
    return {outer_frag_idx};
  }
  const auto& outer_fragment_info = (*outer_table_fragments)[outer_frag_idx];
  auto& inner_frags = table_frags_it->second;
  CHECK_LT(size_t(1), ra_exe_unit.input_descs.size());
  std::vector<size_t> all_frag_ids;
  for (size_t inner_frag_idx = 0; inner_frag_idx < inner_frags->size();
       ++inner_frag_idx) {
    const auto& inner_frag_info = (*inner_frags)[inner_frag_idx];
    if (skipFragmentPair(outer_fragment_info,
                         inner_frag_info,
                         table_idx,
                         inner_table_id_to_join_condition,
                         ra_exe_unit,
                         device_type)) {
      continue;
    }
    all_frag_ids.push_back(inner_frag_idx);
  }
  return all_frag_ids;
}

// Returns true iff the join between two fragments cannot yield any results, per
// shard information. The pair can be skipped to avoid full broadcast.
bool Executor::skipFragmentPair(
    const Fragmenter_Namespace::FragmentInfo& outer_fragment_info,
    const Fragmenter_Namespace::FragmentInfo& inner_fragment_info,
    const int table_idx,
    const std::unordered_map<int, const Analyzer::BinOper*>&
        inner_table_id_to_join_condition,
    const RelAlgExecutionUnit& ra_exe_unit,
    const ExecutorDeviceType device_type) {
  if (device_type != ExecutorDeviceType::GPU) {
    return false;
  }
  CHECK(table_idx >= 0 &&
        static_cast<size_t>(table_idx) < ra_exe_unit.input_descs.size());
  const int inner_table_id = ra_exe_unit.input_descs[table_idx].getTableId();
  // Both tables need to be sharded the same way.
  if (outer_fragment_info.shard == -1 || inner_fragment_info.shard == -1 ||
      outer_fragment_info.shard == inner_fragment_info.shard) {
    return false;
  }
  const Analyzer::BinOper* join_condition{nullptr};
  if (ra_exe_unit.join_quals.empty()) {
    CHECK(!inner_table_id_to_join_condition.empty());
    auto condition_it = inner_table_id_to_join_condition.find(inner_table_id);
    CHECK(condition_it != inner_table_id_to_join_condition.end());
    join_condition = condition_it->second;
    CHECK(join_condition);
  } else {
    CHECK_EQ(plan_state_->join_info_.equi_join_tautologies_.size(),
             plan_state_->join_info_.join_hash_tables_.size());
    for (size_t i = 0; i < plan_state_->join_info_.join_hash_tables_.size(); ++i) {
      if (plan_state_->join_info_.join_hash_tables_[i]->getInnerTableRteIdx() ==
          table_idx) {
        CHECK(!join_condition);
        join_condition = plan_state_->join_info_.equi_join_tautologies_[i].get();
      }
    }
  }
  if (!join_condition) {
    return false;
  }
  // TODO(adb): support fragment skipping based on the overlaps operator
  if (join_condition->is_overlaps_oper()) {
    return false;
  }
  size_t shard_count{0};
  if (dynamic_cast<const Analyzer::ExpressionTuple*>(
          join_condition->get_left_operand())) {
    auto inner_outer_pairs =
        normalize_column_pairs(join_condition, *getCatalog(), getTemporaryTables());
    shard_count = BaselineJoinHashTable::getShardCountForCondition(
        join_condition, this, inner_outer_pairs);
  } else {
    shard_count = get_shard_count(join_condition, this);
  }
  if (shard_count && !ra_exe_unit.join_quals.empty()) {
    plan_state_->join_info_.sharded_range_table_indices_.emplace(table_idx);
  }
  return shard_count;
}

namespace {

const ColumnDescriptor* try_get_column_descriptor(const InputColDescriptor* col_desc,
                                                  const Catalog_Namespace::Catalog& cat) {
  const int table_id = col_desc->getScanDesc().getTableId();
  const int col_id = col_desc->getColId();
  return get_column_descriptor_maybe(col_id, table_id, cat);
}

}  // namespace

std::map<size_t, std::vector<uint64_t>> get_table_id_to_frag_offsets(
    const std::vector<InputDescriptor>& input_descs,
    const std::map<int, const TableFragments*>& all_tables_fragments) {
  std::map<size_t, std::vector<uint64_t>> tab_id_to_frag_offsets;
  for (auto& desc : input_descs) {
    const auto fragments_it = all_tables_fragments.find(desc.getTableId());
    CHECK(fragments_it != all_tables_fragments.end());
    const auto& fragments = *fragments_it->second;
    std::vector<uint64_t> frag_offsets(fragments.size(), 0);
    for (size_t i = 0, off = 0; i < fragments.size(); ++i) {
      frag_offsets[i] = off;
      off += fragments[i].getNumTuples();
    }
    tab_id_to_frag_offsets.insert(std::make_pair(desc.getTableId(), frag_offsets));
  }
  return tab_id_to_frag_offsets;
}

std::pair<std::vector<std::vector<int64_t>>, std::vector<std::vector<uint64_t>>>
Executor::getRowCountAndOffsetForAllFrags(
    const RelAlgExecutionUnit& ra_exe_unit,
    const CartesianProduct<std::vector<std::vector<size_t>>>& frag_ids_crossjoin,
    const std::vector<InputDescriptor>& input_descs,
    const std::map<int, const TableFragments*>& all_tables_fragments) {
  std::vector<std::vector<int64_t>> all_num_rows;
  std::vector<std::vector<uint64_t>> all_frag_offsets;
  const auto tab_id_to_frag_offsets =
      get_table_id_to_frag_offsets(input_descs, all_tables_fragments);
  std::unordered_map<size_t, size_t> outer_id_to_num_row_idx;
  for (const auto& selected_frag_ids : frag_ids_crossjoin) {
    std::vector<int64_t> num_rows;
    std::vector<uint64_t> frag_offsets;
    if (!ra_exe_unit.union_all) {
      CHECK_EQ(selected_frag_ids.size(), input_descs.size());
    }
    for (size_t tab_idx = 0; tab_idx < input_descs.size(); ++tab_idx) {
      const auto frag_id = ra_exe_unit.union_all ? 0 : selected_frag_ids[tab_idx];
      const auto fragments_it =
          all_tables_fragments.find(input_descs[tab_idx].getTableId());
      CHECK(fragments_it != all_tables_fragments.end());
      const auto& fragments = *fragments_it->second;
      if (ra_exe_unit.join_quals.empty() || tab_idx == 0 ||
          plan_state_->join_info_.sharded_range_table_indices_.count(tab_idx)) {
        const auto& fragment = fragments[frag_id];
        num_rows.push_back(fragment.getNumTuples());
      } else {
        size_t total_row_count{0};
        for (const auto& fragment : fragments) {
          total_row_count += fragment.getNumTuples();
        }
        num_rows.push_back(total_row_count);
      }
      const auto frag_offsets_it =
          tab_id_to_frag_offsets.find(input_descs[tab_idx].getTableId());
      CHECK(frag_offsets_it != tab_id_to_frag_offsets.end());
      const auto& offsets = frag_offsets_it->second;
      CHECK_LT(frag_id, offsets.size());
      frag_offsets.push_back(offsets[frag_id]);
    }
    all_num_rows.push_back(num_rows);
    // Fragment offsets of outer table should be ONLY used by rowid for now.
    all_frag_offsets.push_back(frag_offsets);
  }
  return {all_num_rows, all_frag_offsets};
}

// Only fetch columns of hash-joined inner fact table whose fetch are not deferred from
// all the table fragments.
bool Executor::needFetchAllFragments(const InputColDescriptor& inner_col_desc,
                                     const RelAlgExecutionUnit& ra_exe_unit,
                                     const FragmentsList& selected_fragments) const {
  const auto& input_descs = ra_exe_unit.input_descs;
  const int nest_level = inner_col_desc.getScanDesc().getNestLevel();
  if (nest_level < 1 ||
      inner_col_desc.getScanDesc().getSourceType() != InputSourceType::TABLE ||
      ra_exe_unit.join_quals.empty() || input_descs.size() < 2 ||
      (ra_exe_unit.join_quals.empty() &&
       plan_state_->isLazyFetchColumn(inner_col_desc))) {
    return false;
  }
  const int table_id = inner_col_desc.getScanDesc().getTableId();
  CHECK_LT(static_cast<size_t>(nest_level), selected_fragments.size());
  CHECK_EQ(table_id, selected_fragments[nest_level].table_id);
  const auto& fragments = selected_fragments[nest_level].fragment_ids;
  return fragments.size() > 1;
}

std::ostream& operator<<(std::ostream& os, FetchResult const& fetch_result) {
  return os << "col_buffers" << shared::printContainer(fetch_result.col_buffers)
            << " num_rows" << shared::printContainer(fetch_result.num_rows)
            << " frag_offsets" << shared::printContainer(fetch_result.frag_offsets);
}

FetchResult Executor::fetchChunks(
    const ColumnFetcher& column_fetcher,
    const RelAlgExecutionUnit& ra_exe_unit,
    const int device_id,
    const Data_Namespace::MemoryLevel memory_level,
    const std::map<int, const TableFragments*>& all_tables_fragments,
    const FragmentsList& selected_fragments,
    const Catalog_Namespace::Catalog& cat,
    std::list<ChunkIter>& chunk_iterators,
    std::list<std::shared_ptr<Chunk_NS::Chunk>>& chunks,
    DeviceAllocator* device_allocator) {
  auto timer = DEBUG_TIMER(__func__);
  INJECT_TIMER(fetchChunks);
  const auto& col_global_ids = ra_exe_unit.input_col_descs;
  std::vector<std::vector<size_t>> selected_fragments_crossjoin;
  std::vector<size_t> local_col_to_frag_pos;
  buildSelectedFragsMapping(selected_fragments_crossjoin,
                            local_col_to_frag_pos,
                            col_global_ids,
                            selected_fragments,
                            ra_exe_unit);

  CartesianProduct<std::vector<std::vector<size_t>>> frag_ids_crossjoin(
      selected_fragments_crossjoin);

  std::vector<std::vector<const int8_t*>> all_frag_col_buffers;
  std::vector<std::vector<int64_t>> all_num_rows;
  std::vector<std::vector<uint64_t>> all_frag_offsets;

  for (const auto& selected_frag_ids : frag_ids_crossjoin) {
    std::vector<const int8_t*> frag_col_buffers(
        plan_state_->global_to_local_col_ids_.size());
    for (const auto& col_id : col_global_ids) {
      // check whether the interrupt flag turns on (non kernel-time query interrupt)
      if ((g_enable_dynamic_watchdog || g_enable_runtime_query_interrupt) &&
          interrupted_.load()) {
        resetInterrupt();
        throw QueryExecutionError(ERR_INTERRUPTED);
      }
      CHECK(col_id);
      const int table_id = col_id->getScanDesc().getTableId();
      const auto cd = try_get_column_descriptor(col_id.get(), cat);
      if (cd && cd->isVirtualCol) {
        CHECK_EQ("rowid", cd->columnName);
        continue;
      }
      const auto fragments_it = all_tables_fragments.find(table_id);
      CHECK(fragments_it != all_tables_fragments.end());
      const auto fragments = fragments_it->second;
      auto it = plan_state_->global_to_local_col_ids_.find(*col_id);
      CHECK(it != plan_state_->global_to_local_col_ids_.end());
      CHECK_LT(static_cast<size_t>(it->second),
               plan_state_->global_to_local_col_ids_.size());
      const size_t frag_id = selected_frag_ids[local_col_to_frag_pos[it->second]];
      if (!fragments->size()) {
        return {};
      }
      CHECK_LT(frag_id, fragments->size());
      auto memory_level_for_column = memory_level;
      if (plan_state_->columns_to_fetch_.find(
              std::make_pair(col_id->getScanDesc().getTableId(), col_id->getColId())) ==
          plan_state_->columns_to_fetch_.end()) {
        memory_level_for_column = Data_Namespace::CPU_LEVEL;
      }
      if (col_id->getScanDesc().getSourceType() == InputSourceType::RESULT) {
        frag_col_buffers[it->second] = column_fetcher.getResultSetColumn(
            col_id.get(), memory_level_for_column, device_id, device_allocator);
      } else {
        if (needFetchAllFragments(*col_id, ra_exe_unit, selected_fragments)) {
          frag_col_buffers[it->second] =
              column_fetcher.getAllTableColumnFragments(table_id,
                                                        col_id->getColId(),
                                                        all_tables_fragments,
                                                        memory_level_for_column,
                                                        device_id,
                                                        device_allocator);
        } else {
          frag_col_buffers[it->second] =
              column_fetcher.getOneTableColumnFragment(table_id,
                                                       frag_id,
                                                       col_id->getColId(),
                                                       all_tables_fragments,
                                                       chunks,
                                                       chunk_iterators,
                                                       memory_level_for_column,
                                                       device_id,
                                                       device_allocator);
        }
      }
    }
    all_frag_col_buffers.push_back(frag_col_buffers);
  }
  std::tie(all_num_rows, all_frag_offsets) = getRowCountAndOffsetForAllFrags(
      ra_exe_unit, frag_ids_crossjoin, ra_exe_unit.input_descs, all_tables_fragments);
  return {all_frag_col_buffers, all_num_rows, all_frag_offsets};
}

// fetchChunks() is written under the assumption that multiple inputs implies a JOIN.
// This is written under the assumption that multiple inputs implies a UNION ALL.
FetchResult Executor::fetchUnionChunks(
    const ColumnFetcher& column_fetcher,
    const RelAlgExecutionUnit& ra_exe_unit,
    const int device_id,
    const Data_Namespace::MemoryLevel memory_level,
    const std::map<int, const TableFragments*>& all_tables_fragments,
    const FragmentsList& selected_fragments,
    const Catalog_Namespace::Catalog& cat,
    std::list<ChunkIter>& chunk_iterators,
    std::list<std::shared_ptr<Chunk_NS::Chunk>>& chunks,
    DeviceAllocator* device_allocator) {
  auto timer = DEBUG_TIMER(__func__);
  INJECT_TIMER(fetchUnionChunks);

  std::vector<std::vector<const int8_t*>> all_frag_col_buffers;
  std::vector<std::vector<int64_t>> all_num_rows;
  std::vector<std::vector<uint64_t>> all_frag_offsets;

  CHECK(!selected_fragments.empty());
  CHECK_LE(2u, ra_exe_unit.input_descs.size());
  CHECK_LE(2u, ra_exe_unit.input_col_descs.size());
  using TableId = int;
  TableId const selected_table_id = selected_fragments.front().table_id;
  bool const input_descs_index =
      selected_table_id == ra_exe_unit.input_descs[1].getTableId();
  if (!input_descs_index) {
    CHECK_EQ(selected_table_id, ra_exe_unit.input_descs[0].getTableId());
  }
  bool const input_col_descs_index =
      selected_table_id ==
      (*std::next(ra_exe_unit.input_col_descs.begin()))->getScanDesc().getTableId();
  if (!input_col_descs_index) {
    CHECK_EQ(selected_table_id,
             ra_exe_unit.input_col_descs.front()->getScanDesc().getTableId());
  }
  VLOG(2) << "selected_fragments.size()=" << selected_fragments.size()
          << " selected_table_id=" << selected_table_id
          << " input_descs_index=" << int(input_descs_index)
          << " input_col_descs_index=" << int(input_col_descs_index)
          << " ra_exe_unit.input_descs="
          << shared::printContainer(ra_exe_unit.input_descs)
          << " ra_exe_unit.input_col_descs="
          << shared::printContainer(ra_exe_unit.input_col_descs);

  // Partition col_global_ids by table_id
  std::unordered_map<TableId, std::list<std::shared_ptr<const InputColDescriptor>>>
      table_id_to_input_col_descs;
  for (auto const& input_col_desc : ra_exe_unit.input_col_descs) {
    TableId const table_id = input_col_desc->getScanDesc().getTableId();
    table_id_to_input_col_descs[table_id].push_back(input_col_desc);
  }
  for (auto const& pair : table_id_to_input_col_descs) {
    std::vector<std::vector<size_t>> selected_fragments_crossjoin;
    std::vector<size_t> local_col_to_frag_pos;

    buildSelectedFragsMappingForUnion(selected_fragments_crossjoin,
                                      local_col_to_frag_pos,
                                      pair.second,
                                      selected_fragments,
                                      ra_exe_unit);

    CartesianProduct<std::vector<std::vector<size_t>>> frag_ids_crossjoin(
        selected_fragments_crossjoin);

    for (const auto& selected_frag_ids : frag_ids_crossjoin) {
      std::vector<const int8_t*> frag_col_buffers(
          plan_state_->global_to_local_col_ids_.size());
      for (const auto& col_id : pair.second) {
        CHECK(col_id);
        const int table_id = col_id->getScanDesc().getTableId();
        CHECK_EQ(table_id, pair.first);
        const auto cd = try_get_column_descriptor(col_id.get(), cat);
        if (cd && cd->isVirtualCol) {
          CHECK_EQ("rowid", cd->columnName);
          continue;
        }
        const auto fragments_it = all_tables_fragments.find(table_id);
        CHECK(fragments_it != all_tables_fragments.end());
        const auto fragments = fragments_it->second;
        auto it = plan_state_->global_to_local_col_ids_.find(*col_id);
        CHECK(it != plan_state_->global_to_local_col_ids_.end());
        CHECK_LT(static_cast<size_t>(it->second),
                 plan_state_->global_to_local_col_ids_.size());
        const size_t frag_id = ra_exe_unit.union_all
                                   ? 0
                                   : selected_frag_ids[local_col_to_frag_pos[it->second]];
        if (!fragments->size()) {
          return {};
        }
        CHECK_LT(frag_id, fragments->size());
        auto memory_level_for_column = memory_level;
        if (plan_state_->columns_to_fetch_.find(
                std::make_pair(col_id->getScanDesc().getTableId(), col_id->getColId())) ==
            plan_state_->columns_to_fetch_.end()) {
          memory_level_for_column = Data_Namespace::CPU_LEVEL;
        }
        if (col_id->getScanDesc().getSourceType() == InputSourceType::RESULT) {
          frag_col_buffers[it->second] = column_fetcher.getResultSetColumn(
              col_id.get(), memory_level_for_column, device_id, device_allocator);
        } else {
          if (needFetchAllFragments(*col_id, ra_exe_unit, selected_fragments)) {
            frag_col_buffers[it->second] =
                column_fetcher.getAllTableColumnFragments(table_id,
                                                          col_id->getColId(),
                                                          all_tables_fragments,
                                                          memory_level_for_column,
                                                          device_id,
                                                          device_allocator);
          } else {
            frag_col_buffers[it->second] =
                column_fetcher.getOneTableColumnFragment(table_id,
                                                         frag_id,
                                                         col_id->getColId(),
                                                         all_tables_fragments,
                                                         chunks,
                                                         chunk_iterators,
                                                         memory_level_for_column,
                                                         device_id,
                                                         device_allocator);
          }
        }
      }
      all_frag_col_buffers.push_back(frag_col_buffers);
    }
    std::vector<std::vector<int64_t>> num_rows;
    std::vector<std::vector<uint64_t>> frag_offsets;
    std::tie(num_rows, frag_offsets) = getRowCountAndOffsetForAllFrags(
        ra_exe_unit, frag_ids_crossjoin, ra_exe_unit.input_descs, all_tables_fragments);
    all_num_rows.insert(all_num_rows.end(), num_rows.begin(), num_rows.end());
    all_frag_offsets.insert(
        all_frag_offsets.end(), frag_offsets.begin(), frag_offsets.end());
  }
  // UNION ALL hacks.
  VLOG(2) << "all_frag_col_buffers=" << shared::printContainer(all_frag_col_buffers);
  for (size_t i = 0; i < all_frag_col_buffers.front().size(); ++i) {
    all_frag_col_buffers[i & 1][i] = all_frag_col_buffers[i & 1][i ^ 1];
  }
  if (input_descs_index == input_col_descs_index) {
    std::swap(all_frag_col_buffers[0], all_frag_col_buffers[1]);
  }

  VLOG(2) << "all_frag_col_buffers=" << shared::printContainer(all_frag_col_buffers)
          << " all_num_rows=" << shared::printContainer(all_num_rows)
          << " all_frag_offsets=" << shared::printContainer(all_frag_offsets)
          << " input_col_descs_index=" << input_col_descs_index;
  return {{all_frag_col_buffers[input_descs_index]},
          {{all_num_rows[0][input_descs_index]}},
          {{all_frag_offsets[0][input_descs_index]}}};
}

std::vector<size_t> Executor::getFragmentCount(const FragmentsList& selected_fragments,
                                               const size_t scan_idx,
                                               const RelAlgExecutionUnit& ra_exe_unit) {
  if ((ra_exe_unit.input_descs.size() > size_t(2) || !ra_exe_unit.join_quals.empty()) &&
      scan_idx > 0 &&
      !plan_state_->join_info_.sharded_range_table_indices_.count(scan_idx) &&
      !selected_fragments[scan_idx].fragment_ids.empty()) {
    // Fetch all fragments
    return {size_t(0)};
  }

  return selected_fragments[scan_idx].fragment_ids;
}

void Executor::buildSelectedFragsMapping(
    std::vector<std::vector<size_t>>& selected_fragments_crossjoin,
    std::vector<size_t>& local_col_to_frag_pos,
    const std::list<std::shared_ptr<const InputColDescriptor>>& col_global_ids,
    const FragmentsList& selected_fragments,
    const RelAlgExecutionUnit& ra_exe_unit) {
  local_col_to_frag_pos.resize(plan_state_->global_to_local_col_ids_.size());
  size_t frag_pos{0};
  const auto& input_descs = ra_exe_unit.input_descs;
  for (size_t scan_idx = 0; scan_idx < input_descs.size(); ++scan_idx) {
    const int table_id = input_descs[scan_idx].getTableId();
    CHECK_EQ(selected_fragments[scan_idx].table_id, table_id);
    selected_fragments_crossjoin.push_back(
        getFragmentCount(selected_fragments, scan_idx, ra_exe_unit));
    for (const auto& col_id : col_global_ids) {
      CHECK(col_id);
      const auto& input_desc = col_id->getScanDesc();
      if (input_desc.getTableId() != table_id ||
          input_desc.getNestLevel() != static_cast<int>(scan_idx)) {
        continue;
      }
      auto it = plan_state_->global_to_local_col_ids_.find(*col_id);
      CHECK(it != plan_state_->global_to_local_col_ids_.end());
      CHECK_LT(static_cast<size_t>(it->second),
               plan_state_->global_to_local_col_ids_.size());
      local_col_to_frag_pos[it->second] = frag_pos;
    }
    ++frag_pos;
  }
}

void Executor::buildSelectedFragsMappingForUnion(
    std::vector<std::vector<size_t>>& selected_fragments_crossjoin,
    std::vector<size_t>& local_col_to_frag_pos,
    const std::list<std::shared_ptr<const InputColDescriptor>>& col_global_ids,
    const FragmentsList& selected_fragments,
    const RelAlgExecutionUnit& ra_exe_unit) {
  local_col_to_frag_pos.resize(plan_state_->global_to_local_col_ids_.size());
  size_t frag_pos{0};
  const auto& input_descs = ra_exe_unit.input_descs;
  for (size_t scan_idx = 0; scan_idx < input_descs.size(); ++scan_idx) {
    const int table_id = input_descs[scan_idx].getTableId();
    // selected_fragments here is from assignFragsToKernelDispatch
    // execution_kernel.fragments
    if (selected_fragments[0].table_id != table_id) {  // TODO 0
      continue;
    }
    // CHECK_EQ(selected_fragments[scan_idx].table_id, table_id);
    selected_fragments_crossjoin.push_back(
        // getFragmentCount(selected_fragments, scan_idx, ra_exe_unit));
        {size_t(1)});  // TODO
    for (const auto& col_id : col_global_ids) {
      CHECK(col_id);
      const auto& input_desc = col_id->getScanDesc();
      if (input_desc.getTableId() != table_id ||
          input_desc.getNestLevel() != static_cast<int>(scan_idx)) {
        continue;
      }
      auto it = plan_state_->global_to_local_col_ids_.find(*col_id);
      CHECK(it != plan_state_->global_to_local_col_ids_.end());
      CHECK_LT(static_cast<size_t>(it->second),
               plan_state_->global_to_local_col_ids_.size());
      local_col_to_frag_pos[it->second] = frag_pos;
    }
    ++frag_pos;
  }
}

namespace {

class OutVecOwner {
 public:
  OutVecOwner(const std::vector<int64_t*>& out_vec) : out_vec_(out_vec) {}
  ~OutVecOwner() {
    for (auto out : out_vec_) {
      delete[] out;
    }
  }

 private:
  std::vector<int64_t*> out_vec_;
};
}  // namespace

int32_t Executor::executePlanWithoutGroupBy(
    const RelAlgExecutionUnit& ra_exe_unit,
    const CompilationResult& compilation_result,
    const bool hoist_literals,
    ResultSetPtr& results,
    const std::vector<Analyzer::Expr*>& target_exprs,
    const ExecutorDeviceType device_type,
    std::vector<std::vector<const int8_t*>>& col_buffers,
    QueryExecutionContext* query_exe_context,
    const std::vector<std::vector<int64_t>>& num_rows,
    const std::vector<std::vector<uint64_t>>& frag_offsets,
    Data_Namespace::DataMgr* data_mgr,
    const int device_id,
    const uint32_t start_rowid,
    const uint32_t num_tables,
    RenderInfo* render_info) {
  INJECT_TIMER(executePlanWithoutGroupBy);
  auto timer = DEBUG_TIMER(__func__);
  CHECK(!results);
  if (col_buffers.empty()) {
    return 0;
  }

  RenderAllocatorMap* render_allocator_map_ptr = nullptr;
  if (render_info) {
    // TODO(adb): make sure that we either never get here in the CPU case, or if we do get
    // here, we are in non-insitu mode.
    CHECK(render_info->useCudaBuffers() || !render_info->isPotentialInSituRender())
        << "CUDA disabled rendering in the executePlanWithoutGroupBy query path is "
           "currently unsupported.";
    render_allocator_map_ptr = render_info->render_allocator_map_ptr.get();
  }

  int32_t error_code = device_type == ExecutorDeviceType::GPU ? 0 : start_rowid;
  std::vector<int64_t*> out_vec;
  const auto hoist_buf = serializeLiterals(compilation_result.literal_values, device_id);
  const auto join_hash_table_ptrs = getJoinHashTablePtrs(device_type, device_id);
  std::unique_ptr<OutVecOwner> output_memory_scope;
  if ((g_enable_dynamic_watchdog || g_enable_runtime_query_interrupt) &&
      interrupted_.load()) {
    resetInterrupt();
    throw QueryExecutionError(ERR_INTERRUPTED);
  }
  if (device_type == ExecutorDeviceType::CPU) {
    auto cpu_generated_code = std::dynamic_pointer_cast<CpuCompilationContext>(
        compilation_result.generated_code);
    CHECK(cpu_generated_code);
    out_vec = query_exe_context->launchCpuCode(ra_exe_unit,
                                               cpu_generated_code.get(),
                                               hoist_literals,
                                               hoist_buf,
                                               col_buffers,
                                               num_rows,
                                               frag_offsets,
                                               0,
                                               &error_code,
                                               num_tables,
                                               join_hash_table_ptrs);
    output_memory_scope.reset(new OutVecOwner(out_vec));
  } else {
    auto gpu_generated_code = std::dynamic_pointer_cast<GpuCompilationContext>(
        compilation_result.generated_code);
    CHECK(gpu_generated_code);
    try {
      out_vec = query_exe_context->launchGpuCode(
          ra_exe_unit,
          gpu_generated_code.get(),
          hoist_literals,
          hoist_buf,
          col_buffers,
          num_rows,
          frag_offsets,
          0,
          data_mgr,
          blockSize(),
          gridSize(),
          device_id,
          compilation_result.gpu_smem_context.getSharedMemorySize(),
          &error_code,
          num_tables,
          join_hash_table_ptrs,
          render_allocator_map_ptr);
      output_memory_scope.reset(new OutVecOwner(out_vec));
    } catch (const OutOfMemory&) {
      return ERR_OUT_OF_GPU_MEM;
    } catch (const std::exception& e) {
      LOG(FATAL) << "Error launching the GPU kernel: " << e.what();
    }
  }
  if (error_code == Executor::ERR_OVERFLOW_OR_UNDERFLOW ||
      error_code == Executor::ERR_DIV_BY_ZERO ||
      error_code == Executor::ERR_OUT_OF_TIME ||
      error_code == Executor::ERR_INTERRUPTED ||
      error_code == Executor::ERR_SINGLE_VALUE_FOUND_MULTIPLE_VALUES ||
      error_code == Executor::ERR_GEOS) {
    return error_code;
  }
  if (ra_exe_unit.estimator) {
    CHECK(!error_code);
    results =
        std::shared_ptr<ResultSet>(query_exe_context->estimator_result_set_.release());
    return 0;
  }
  std::vector<int64_t> reduced_outs;
  const auto num_frags = col_buffers.size();
  const size_t entry_count =
      device_type == ExecutorDeviceType::GPU
          ? (compilation_result.gpu_smem_context.isSharedMemoryUsed()
                 ? 1
                 : blockSize() * gridSize() * num_frags)
          : num_frags;
  if (size_t(1) == entry_count) {
    for (auto out : out_vec) {
      CHECK(out);
      reduced_outs.push_back(*out);
    }
  } else {
    size_t out_vec_idx = 0;

    for (const auto target_expr : target_exprs) {
      const auto agg_info = get_target_info(target_expr, g_bigint_count);
      CHECK(agg_info.is_agg);

      const int num_iterations = agg_info.sql_type.is_geometry()
                                     ? agg_info.sql_type.get_physical_coord_cols()
                                     : 1;

      for (int i = 0; i < num_iterations; i++) {
        int64_t val1;
        const bool float_argument_input = takes_float_argument(agg_info);
        if (is_distinct_target(agg_info)) {
          CHECK(agg_info.agg_kind == kCOUNT ||
                agg_info.agg_kind == kAPPROX_COUNT_DISTINCT);
          val1 = out_vec[out_vec_idx][0];
          error_code = 0;
        } else {
          const auto chosen_bytes = static_cast<size_t>(
              query_exe_context->query_mem_desc_.getPaddedSlotWidthBytes(out_vec_idx));
          std::tie(val1, error_code) = Executor::reduceResults(
              agg_info.agg_kind,
              agg_info.sql_type,
              query_exe_context->getAggInitValForIndex(out_vec_idx),
              float_argument_input ? sizeof(int32_t) : chosen_bytes,
              out_vec[out_vec_idx],
              entry_count,
              false,
              float_argument_input);
        }
        if (error_code) {
          break;
        }
        reduced_outs.push_back(val1);
        if (agg_info.agg_kind == kAVG ||
            (agg_info.agg_kind == kSAMPLE &&
             (agg_info.sql_type.is_varlen() || agg_info.sql_type.is_geometry()))) {
          const auto chosen_bytes = static_cast<size_t>(
              query_exe_context->query_mem_desc_.getPaddedSlotWidthBytes(out_vec_idx +
                                                                         1));
          int64_t val2;
          std::tie(val2, error_code) = Executor::reduceResults(
              agg_info.agg_kind == kAVG ? kCOUNT : agg_info.agg_kind,
              agg_info.sql_type,
              query_exe_context->getAggInitValForIndex(out_vec_idx + 1),
              float_argument_input ? sizeof(int32_t) : chosen_bytes,
              out_vec[out_vec_idx + 1],
              entry_count,
              false,
              false);
          if (error_code) {
            break;
          }
          reduced_outs.push_back(val2);
          ++out_vec_idx;
        }
        ++out_vec_idx;
      }
    }
  }

  if (error_code) {
    return error_code;
  }

  CHECK_EQ(size_t(1), query_exe_context->query_buffers_->result_sets_.size());
  auto rows_ptr = std::shared_ptr<ResultSet>(
      query_exe_context->query_buffers_->result_sets_[0].release());
  rows_ptr->fillOneEntry(reduced_outs);
  results = std::move(rows_ptr);
  return error_code;
}

namespace {

bool check_rows_less_than_needed(const ResultSetPtr& results, const size_t scan_limit) {
  CHECK(scan_limit);
  return results && results->rowCount() < scan_limit;
}

}  // namespace

int32_t Executor::executePlanWithGroupBy(
    const RelAlgExecutionUnit& ra_exe_unit,
    const CompilationResult& compilation_result,
    const bool hoist_literals,
    ResultSetPtr& results,
    const ExecutorDeviceType device_type,
    std::vector<std::vector<const int8_t*>>& col_buffers,
    const std::vector<size_t> outer_tab_frag_ids,
    QueryExecutionContext* query_exe_context,
    const std::vector<std::vector<int64_t>>& num_rows,
    const std::vector<std::vector<uint64_t>>& frag_offsets,
    Data_Namespace::DataMgr* data_mgr,
    const int device_id,
    const int outer_table_id,
    const int64_t scan_limit,
    const uint32_t start_rowid,
    const uint32_t num_tables,
    RenderInfo* render_info) {
  auto timer = DEBUG_TIMER(__func__);
  INJECT_TIMER(executePlanWithGroupBy);
  CHECK(!results);
  if (col_buffers.empty()) {
    return 0;
  }
  CHECK_NE(ra_exe_unit.groupby_exprs.size(), size_t(0));
  // TODO(alex):
  // 1. Optimize size (make keys more compact).
  // 2. Resize on overflow.
  // 3. Optimize runtime.
  auto hoist_buf = serializeLiterals(compilation_result.literal_values, device_id);
  int32_t error_code = device_type == ExecutorDeviceType::GPU ? 0 : start_rowid;
  const auto join_hash_table_ptrs = getJoinHashTablePtrs(device_type, device_id);
  if ((g_enable_dynamic_watchdog || g_enable_runtime_query_interrupt) &&
      interrupted_.load()) {
    return ERR_INTERRUPTED;
  }

  RenderAllocatorMap* render_allocator_map_ptr = nullptr;
  if (render_info && render_info->useCudaBuffers()) {
    render_allocator_map_ptr = render_info->render_allocator_map_ptr.get();
  }

  VLOG(2) << "bool(ra_exe_unit.union_all)=" << bool(ra_exe_unit.union_all)
          << " ra_exe_unit.input_descs="
          << shared::printContainer(ra_exe_unit.input_descs)
          << " ra_exe_unit.input_col_descs="
          << shared::printContainer(ra_exe_unit.input_col_descs)
          << " ra_exe_unit.scan_limit=" << ra_exe_unit.scan_limit
          << " num_rows=" << shared::printContainer(num_rows)
          << " frag_offsets=" << shared::printContainer(frag_offsets)
          << " query_exe_context->query_buffers_->num_rows_="
          << query_exe_context->query_buffers_->num_rows_
          << " query_exe_context->query_mem_desc_.getEntryCount()="
          << query_exe_context->query_mem_desc_.getEntryCount()
          << " device_id=" << device_id << " outer_table_id=" << outer_table_id
          << " scan_limit=" << scan_limit << " start_rowid=" << start_rowid
          << " num_tables=" << num_tables;

  RelAlgExecutionUnit ra_exe_unit_copy = ra_exe_unit;
  // For UNION ALL, filter out input_descs and input_col_descs that are not associated
  // with outer_table_id.
  if (ra_exe_unit_copy.union_all) {
    // Sort outer_table_id first, then pop the rest off of ra_exe_unit_copy.input_descs.
    std::stable_sort(ra_exe_unit_copy.input_descs.begin(),
                     ra_exe_unit_copy.input_descs.end(),
                     [outer_table_id](auto const& a, auto const& b) {
                       return a.getTableId() == outer_table_id &&
                              b.getTableId() != outer_table_id;
                     });
    while (!ra_exe_unit_copy.input_descs.empty() &&
           ra_exe_unit_copy.input_descs.back().getTableId() != outer_table_id) {
      ra_exe_unit_copy.input_descs.pop_back();
    }
    // Filter ra_exe_unit_copy.input_col_descs.
    ra_exe_unit_copy.input_col_descs.remove_if(
        [outer_table_id](auto const& input_col_desc) {
          return input_col_desc->getScanDesc().getTableId() != outer_table_id;
        });
    query_exe_context->query_mem_desc_.setEntryCount(ra_exe_unit_copy.scan_limit);
  }

  if (device_type == ExecutorDeviceType::CPU) {
    auto cpu_generated_code = std::dynamic_pointer_cast<CpuCompilationContext>(
        compilation_result.generated_code);
    CHECK(cpu_generated_code);
    query_exe_context->launchCpuCode(
        ra_exe_unit_copy,
        cpu_generated_code.get(),
        hoist_literals,
        hoist_buf,
        col_buffers,
        num_rows,
        frag_offsets,
        ra_exe_unit_copy.union_all ? ra_exe_unit_copy.scan_limit : scan_limit,
        &error_code,
        num_tables,
        join_hash_table_ptrs);
  } else {
    try {
      auto gpu_generated_code = std::dynamic_pointer_cast<GpuCompilationContext>(
          compilation_result.generated_code);
      CHECK(gpu_generated_code);
      query_exe_context->launchGpuCode(
          ra_exe_unit_copy,
          gpu_generated_code.get(),
          hoist_literals,
          hoist_buf,
          col_buffers,
          num_rows,
          frag_offsets,
          ra_exe_unit_copy.union_all ? ra_exe_unit_copy.scan_limit : scan_limit,
          data_mgr,
          blockSize(),
          gridSize(),
          device_id,
          compilation_result.gpu_smem_context.getSharedMemorySize(),
          &error_code,
          num_tables,
          join_hash_table_ptrs,
          render_allocator_map_ptr);
    } catch (const OutOfMemory&) {
      return ERR_OUT_OF_GPU_MEM;
    } catch (const OutOfRenderMemory&) {
      return ERR_OUT_OF_RENDER_MEM;
    } catch (const StreamingTopNNotSupportedInRenderQuery&) {
      return ERR_STREAMING_TOP_N_NOT_SUPPORTED_IN_RENDER_QUERY;
    } catch (const std::exception& e) {
      LOG(FATAL) << "Error launching the GPU kernel: " << e.what();
    }
  }

  if (error_code == Executor::ERR_OVERFLOW_OR_UNDERFLOW ||
      error_code == Executor::ERR_DIV_BY_ZERO ||
      error_code == Executor::ERR_OUT_OF_TIME ||
      error_code == Executor::ERR_INTERRUPTED ||
      error_code == Executor::ERR_SINGLE_VALUE_FOUND_MULTIPLE_VALUES ||
      error_code == Executor::ERR_GEOS) {
    return error_code;
  }

  if (error_code != Executor::ERR_OVERFLOW_OR_UNDERFLOW &&
      error_code != Executor::ERR_DIV_BY_ZERO && !render_allocator_map_ptr) {
    results = query_exe_context->getRowSet(ra_exe_unit_copy,
                                           query_exe_context->query_mem_desc_);
    CHECK(results);
    VLOG(2) << "results->rowCount()=" << results->rowCount();
    results->holdLiterals(hoist_buf);
  }
  if (error_code < 0 && render_allocator_map_ptr) {
    auto const adjusted_scan_limit =
        ra_exe_unit_copy.union_all ? ra_exe_unit_copy.scan_limit : scan_limit;
    // More rows passed the filter than available slots. We don't have a count to check,
    // so assume we met the limit if a scan limit is set
    if (adjusted_scan_limit != 0) {
      return 0;
    } else {
      return error_code;
    }
  }
  if (error_code && (!scan_limit || check_rows_less_than_needed(results, scan_limit))) {
    return error_code;  // unlucky, not enough results and we ran out of slots
  }

  return 0;
}

std::vector<int64_t> Executor::getJoinHashTablePtrs(const ExecutorDeviceType device_type,
                                                    const int device_id) {
  std::vector<int64_t> table_ptrs;
  const auto& join_hash_tables = plan_state_->join_info_.join_hash_tables_;
  for (auto hash_table : join_hash_tables) {
    if (!hash_table) {
      CHECK(table_ptrs.empty());
      return {};
    }
    table_ptrs.push_back(hash_table->getJoinHashBuffer(
        device_type, device_type == ExecutorDeviceType::GPU ? device_id : 0));
  }
  return table_ptrs;
}

void Executor::nukeOldState(const bool allow_lazy_fetch,
                            const std::vector<InputTableInfo>& query_infos,
                            const PlanState::DeletedColumnsMap& deleted_cols_map,
                            const RelAlgExecutionUnit* ra_exe_unit) {
  kernel_queue_time_ms_ = 0;
  compilation_queue_time_ms_ = 0;
  const bool contains_left_deep_outer_join =
      ra_exe_unit && std::find_if(ra_exe_unit->join_quals.begin(),
                                  ra_exe_unit->join_quals.end(),
                                  [](const JoinCondition& join_condition) {
                                    return join_condition.type == JoinType::LEFT;
                                  }) != ra_exe_unit->join_quals.end();
  cgen_state_.reset(new CgenState(query_infos.size(), contains_left_deep_outer_join));
  plan_state_.reset(new PlanState(allow_lazy_fetch && !contains_left_deep_outer_join,
                                  query_infos,
                                  deleted_cols_map,
                                  this));
}

void Executor::preloadFragOffsets(const std::vector<InputDescriptor>& input_descs,
                                  const std::vector<InputTableInfo>& query_infos) {
  AUTOMATIC_IR_METADATA(cgen_state_.get());
  const auto ld_count = input_descs.size();
  auto frag_off_ptr = get_arg_by_name(cgen_state_->row_func_, "frag_row_off");
  for (size_t i = 0; i < ld_count; ++i) {
    CHECK_LT(i, query_infos.size());
    const auto frag_count = query_infos[i].info.fragments.size();
    if (i > 0) {
      cgen_state_->frag_offsets_.push_back(nullptr);
    } else {
      if (frag_count > 1) {
        cgen_state_->frag_offsets_.push_back(
            cgen_state_->ir_builder_.CreateLoad(frag_off_ptr));
      } else {
        cgen_state_->frag_offsets_.push_back(nullptr);
      }
    }
  }
}

Executor::JoinHashTableOrError Executor::buildHashTableForQualifier(
    const std::shared_ptr<Analyzer::BinOper>& qual_bin_oper,
    const std::vector<InputTableInfo>& query_infos,
    const MemoryLevel memory_level,
    const HashType preferred_hash_type,
    ColumnCacheMap& column_cache,
    const QueryHint& query_hint) {
  if (!g_enable_overlaps_hashjoin && qual_bin_oper->is_overlaps_oper()) {
    return {nullptr, "Overlaps hash join disabled, attempting to fall back to loop join"};
  }
  // check whether the interrupt flag turns on (non kernel-time query interrupt)
  if ((g_enable_dynamic_watchdog || g_enable_runtime_query_interrupt) &&
      interrupted_.load()) {
    resetInterrupt();
    throw QueryExecutionError(ERR_INTERRUPTED);
  }
  try {
    auto tbl = HashJoin::getInstance(qual_bin_oper,
                                     query_infos,
                                     memory_level,
                                     preferred_hash_type,
                                     deviceCountForMemoryLevel(memory_level),
                                     column_cache,
                                     this,
                                     query_hint);
    return {tbl, ""};
  } catch (const HashJoinFail& e) {
    return {nullptr, e.what()};
  }
}

int8_t Executor::warpSize() const {
  CHECK(catalog_);
  const auto cuda_mgr = catalog_->getDataMgr().getCudaMgr();
  CHECK(cuda_mgr);
  const auto& dev_props = cuda_mgr->getAllDeviceProperties();
  CHECK(!dev_props.empty());
  return dev_props.front().warpSize;
}

unsigned Executor::gridSize() const {
  CHECK(catalog_);
  const auto cuda_mgr = catalog_->getDataMgr().getCudaMgr();
  if (!cuda_mgr) {
    return 0;
  }
  return grid_size_x_ ? grid_size_x_ : 2 * cuda_mgr->getMinNumMPsForAllDevices();
}

unsigned Executor::numBlocksPerMP() const {
  CHECK(catalog_);
  const auto cuda_mgr = catalog_->getDataMgr().getCudaMgr();
  CHECK(cuda_mgr);
  return grid_size_x_ ? std::ceil(grid_size_x_ / cuda_mgr->getMinNumMPsForAllDevices())
                      : 2;
}

unsigned Executor::blockSize() const {
  CHECK(catalog_);
  const auto cuda_mgr = catalog_->getDataMgr().getCudaMgr();
  if (!cuda_mgr) {
    return 0;
  }
  const auto& dev_props = cuda_mgr->getAllDeviceProperties();
  return block_size_x_ ? block_size_x_ : dev_props.front().maxThreadsPerBlock;
}

size_t Executor::maxGpuSlabSize() const {
  return max_gpu_slab_size_;
}

int64_t Executor::deviceCycles(int milliseconds) const {
  CHECK(catalog_);
  const auto cuda_mgr = catalog_->getDataMgr().getCudaMgr();
  CHECK(cuda_mgr);
  const auto& dev_props = cuda_mgr->getAllDeviceProperties();
  return static_cast<int64_t>(dev_props.front().clockKhz) * milliseconds;
}

llvm::Value* Executor::castToFP(llvm::Value* value,
                                SQLTypeInfo const& from_ti,
                                SQLTypeInfo const& to_ti) {
  AUTOMATIC_IR_METADATA(cgen_state_.get());
  if (value->getType()->isIntegerTy() && from_ti.is_number() && to_ti.is_fp() &&
      (!from_ti.is_fp() || from_ti.get_size() != to_ti.get_size())) {
    llvm::Type* fp_type{nullptr};
    switch (to_ti.get_size()) {
      case 4:
        fp_type = llvm::Type::getFloatTy(cgen_state_->context_);
        break;
      case 8:
        fp_type = llvm::Type::getDoubleTy(cgen_state_->context_);
        break;
      default:
        LOG(FATAL) << "Unsupported FP size: " << to_ti.get_size();
    }
    value = cgen_state_->ir_builder_.CreateSIToFP(value, fp_type);
    if (from_ti.get_scale()) {
      value = cgen_state_->ir_builder_.CreateFDiv(
          value,
          llvm::ConstantFP::get(value->getType(), exp_to_scale(from_ti.get_scale())));
    }
  }
  return value;
}

llvm::Value* Executor::castToIntPtrTyIn(llvm::Value* val, const size_t bitWidth) {
  AUTOMATIC_IR_METADATA(cgen_state_.get());
  CHECK(val->getType()->isPointerTy());

  const auto val_ptr_type = static_cast<llvm::PointerType*>(val->getType());
  const auto val_type = val_ptr_type->getElementType();
  size_t val_width = 0;
  if (val_type->isIntegerTy()) {
    val_width = val_type->getIntegerBitWidth();
  } else {
    if (val_type->isFloatTy()) {
      val_width = 32;
    } else {
      CHECK(val_type->isDoubleTy());
      val_width = 64;
    }
  }
  CHECK_LT(size_t(0), val_width);
  if (bitWidth == val_width) {
    return val;
  }
  return cgen_state_->ir_builder_.CreateBitCast(
      val, llvm::PointerType::get(get_int_type(bitWidth, cgen_state_->context_), 0));
}

#define EXECUTE_INCLUDE
#include "ArrayOps.cpp"
#include "DateAdd.cpp"
#include "StringFunctions.cpp"
#undef EXECUTE_INCLUDE

std::tuple<RelAlgExecutionUnit, PlanState::DeletedColumnsMap> Executor::addDeletedColumn(
    const RelAlgExecutionUnit& ra_exe_unit,
    const CompilationOptions& co) {
  if (!co.filter_on_deleted_column) {
    return std::make_tuple(ra_exe_unit, PlanState::DeletedColumnsMap{});
  }
  auto ra_exe_unit_with_deleted = ra_exe_unit;
  PlanState::DeletedColumnsMap deleted_cols_map;
  for (const auto& input_table : ra_exe_unit_with_deleted.input_descs) {
    if (input_table.getSourceType() != InputSourceType::TABLE) {
      continue;
    }
    const auto td = catalog_->getMetadataForTable(input_table.getTableId());
    CHECK(td);
    const auto deleted_cd = catalog_->getDeletedColumnIfRowsDeleted(td);
    if (!deleted_cd) {
      continue;
    }
    CHECK(deleted_cd->columnType.is_boolean());
    // check deleted column is not already present
    bool found = false;
    for (const auto& input_col : ra_exe_unit_with_deleted.input_col_descs) {
      if (input_col.get()->getColId() == deleted_cd->columnId &&
          input_col.get()->getScanDesc().getTableId() == deleted_cd->tableId &&
          input_col.get()->getScanDesc().getNestLevel() == input_table.getNestLevel()) {
        found = true;
      }
    }
    if (!found) {
      // add deleted column
      ra_exe_unit_with_deleted.input_col_descs.emplace_back(new InputColDescriptor(
          deleted_cd->columnId, deleted_cd->tableId, input_table.getNestLevel()));
      auto deleted_cols_it = deleted_cols_map.find(deleted_cd->tableId);
      if (deleted_cols_it == deleted_cols_map.end()) {
        CHECK(deleted_cols_map.insert(std::make_pair(deleted_cd->tableId, deleted_cd))
                  .second);
      } else {
        CHECK_EQ(deleted_cd, deleted_cols_it->second);
      }
    }
  }
  return std::make_tuple(ra_exe_unit_with_deleted, deleted_cols_map);
}

namespace {
// Note(Wamsi): `get_hpt_overflow_underflow_safe_scaled_value` will return `true` for safe
// scaled epoch value and `false` for overflow/underflow values as the first argument of
// return type.
std::tuple<bool, int64_t, int64_t> get_hpt_overflow_underflow_safe_scaled_values(
    const int64_t chunk_min,
    const int64_t chunk_max,
    const SQLTypeInfo& lhs_type,
    const SQLTypeInfo& rhs_type) {
  const int32_t ldim = lhs_type.get_dimension();
  const int32_t rdim = rhs_type.get_dimension();
  CHECK(ldim != rdim);
  const auto scale = DateTimeUtils::get_timestamp_precision_scale(abs(rdim - ldim));
  if (ldim > rdim) {
    // LHS type precision is more than RHS col type. No chance of overflow/underflow.
    return {true, chunk_min / scale, chunk_max / scale};
  }

  using checked_int64_t = boost::multiprecision::number<
      boost::multiprecision::cpp_int_backend<64,
                                             64,
                                             boost::multiprecision::signed_magnitude,
                                             boost::multiprecision::checked,
                                             void>>;

  try {
    auto ret =
        std::make_tuple(true,
                        int64_t(checked_int64_t(chunk_min) * checked_int64_t(scale)),
                        int64_t(checked_int64_t(chunk_max) * checked_int64_t(scale)));
    return ret;
  } catch (const std::overflow_error& e) {
    // noop
  }
  return std::make_tuple(false, chunk_min, chunk_max);
}

}  // namespace

std::pair<bool, int64_t> Executor::skipFragment(
    const InputDescriptor& table_desc,
    const Fragmenter_Namespace::FragmentInfo& fragment,
    const std::list<std::shared_ptr<Analyzer::Expr>>& simple_quals,
    const std::vector<uint64_t>& frag_offsets,
    const size_t frag_idx) {
  const int table_id = table_desc.getTableId();
  for (const auto& simple_qual : simple_quals) {
    const auto comp_expr =
        std::dynamic_pointer_cast<const Analyzer::BinOper>(simple_qual);
    if (!comp_expr) {
      // is this possible?
      return {false, -1};
    }
    const auto lhs = comp_expr->get_left_operand();
    auto lhs_col = dynamic_cast<const Analyzer::ColumnVar*>(lhs);
    if (!lhs_col || !lhs_col->get_table_id() || lhs_col->get_rte_idx()) {
      // See if lhs is a simple cast that was allowed through normalize_simple_predicate
      auto lhs_uexpr = dynamic_cast<const Analyzer::UOper*>(lhs);
      if (lhs_uexpr) {
        CHECK(lhs_uexpr->get_optype() ==
              kCAST);  // We should have only been passed a cast expression
        lhs_col = dynamic_cast<const Analyzer::ColumnVar*>(lhs_uexpr->get_operand());
        if (!lhs_col || !lhs_col->get_table_id() || lhs_col->get_rte_idx()) {
          continue;
        }
      } else {
        continue;
      }
    }
    const auto rhs = comp_expr->get_right_operand();
    const auto rhs_const = dynamic_cast<const Analyzer::Constant*>(rhs);
    if (!rhs_const) {
      // is this possible?
      return {false, -1};
    }
    if (!lhs->get_type_info().is_integer() && !lhs->get_type_info().is_time()) {
      continue;
    }
    const int col_id = lhs_col->get_column_id();
    auto chunk_meta_it = fragment.getChunkMetadataMap().find(col_id);
    int64_t chunk_min{0};
    int64_t chunk_max{0};
    bool is_rowid{false};
    size_t start_rowid{0};
    if (chunk_meta_it == fragment.getChunkMetadataMap().end()) {
      auto cd = get_column_descriptor(col_id, table_id, *catalog_);
      if (cd->isVirtualCol) {
        CHECK(cd->columnName == "rowid");
        const auto& table_generation = getTableGeneration(table_id);
        start_rowid = table_generation.start_rowid;
        chunk_min = frag_offsets[frag_idx] + start_rowid;
        chunk_max = frag_offsets[frag_idx + 1] - 1 + start_rowid;
        is_rowid = true;
      }
    } else {
      const auto& chunk_type = lhs_col->get_type_info();
      chunk_min = extract_min_stat(chunk_meta_it->second->chunkStats, chunk_type);
      chunk_max = extract_max_stat(chunk_meta_it->second->chunkStats, chunk_type);
    }
    if (chunk_min > chunk_max) {
      // invalid metadata range, do not skip fragment
      return {false, -1};
    }
    if (lhs->get_type_info().is_timestamp() &&
        (lhs_col->get_type_info().get_dimension() !=
         rhs_const->get_type_info().get_dimension()) &&
        (lhs_col->get_type_info().is_high_precision_timestamp() ||
         rhs_const->get_type_info().is_high_precision_timestamp())) {
      // If original timestamp lhs col has different precision,
      // column metadata holds value in original precision
      // therefore adjust rhs value to match lhs precision

      // Note(Wamsi): We adjust rhs const value instead of lhs value to not
      // artificially limit the lhs column range. RHS overflow/underflow is already
      // been validated in `TimeGM::get_overflow_underflow_safe_epoch`.
      bool is_valid;
      std::tie(is_valid, chunk_min, chunk_max) =
          get_hpt_overflow_underflow_safe_scaled_values(
              chunk_min, chunk_max, lhs_col->get_type_info(), rhs_const->get_type_info());
      if (!is_valid) {
        VLOG(4) << "Overflow/Underflow detecting in fragments skipping logic.\nChunk min "
                   "value: "
                << std::to_string(chunk_min)
                << "\nChunk max value: " << std::to_string(chunk_max)
                << "\nLHS col precision is: "
                << std::to_string(lhs_col->get_type_info().get_dimension())
                << "\nRHS precision is: "
                << std::to_string(rhs_const->get_type_info().get_dimension()) << ".";
        return {false, -1};
      }
    }
    llvm::LLVMContext local_context;
    CgenState local_cgen_state(local_context);
    CodeGenerator code_generator(&local_cgen_state, nullptr);

    const auto rhs_val =
        CodeGenerator::codegenIntConst(rhs_const, &local_cgen_state)->getSExtValue();

    switch (comp_expr->get_optype()) {
      case kGE:
        if (chunk_max < rhs_val) {
          return {true, -1};
        }
        break;
      case kGT:
        if (chunk_max <= rhs_val) {
          return {true, -1};
        }
        break;
      case kLE:
        if (chunk_min > rhs_val) {
          return {true, -1};
        }
        break;
      case kLT:
        if (chunk_min >= rhs_val) {
          return {true, -1};
        }
        break;
      case kEQ:
        if (chunk_min > rhs_val || chunk_max < rhs_val) {
          return {true, -1};
        } else if (is_rowid) {
          return {false, rhs_val - start_rowid};
        }
        break;
      default:
        break;
    }
  }
  return {false, -1};
}

/*
 *   The skipFragmentInnerJoins process all quals stored in the execution unit's
 * join_quals and gather all the ones that meet the "simple_qual" characteristics
 * (logical expressions with AND operations, etc.). It then uses the skipFragment function
 * to decide whether the fragment should be skipped or not. The fragment will be skipped
 * if at least one of these skipFragment calls return a true statment in its first value.
 *   - The code depends on skipFragment's output to have a meaningful (anything but -1)
 * second value only if its first value is "false".
 *   - It is assumed that {false, n  > -1} has higher priority than {true, -1},
 *     i.e., we only skip if none of the quals trigger the code to update the
 * rowid_lookup_key
 *   - Only AND operations are valid and considered:
 *     - `select * from t1,t2 where A and B and C`: A, B, and C are considered for causing
 * the skip
 *     - `select * from t1,t2 where (A or B) and C`: only C is considered
 *     - `select * from t1,t2 where A or B`: none are considered (no skipping).
 *   - NOTE: (re: intermediate projections) the following two queries are fundamentally
 * implemented differently, which cause the first one to skip correctly, but the second
 * one will not skip.
 *     -  e.g. #1, select * from t1 join t2 on (t1.i=t2.i) where (A and B); -- skips if
 * possible
 *     -  e.g. #2, select * from t1 join t2 on (t1.i=t2.i and A and B); -- intermediate
 * projection, no skipping
 */
std::pair<bool, int64_t> Executor::skipFragmentInnerJoins(
    const InputDescriptor& table_desc,
    const RelAlgExecutionUnit& ra_exe_unit,
    const Fragmenter_Namespace::FragmentInfo& fragment,
    const std::vector<uint64_t>& frag_offsets,
    const size_t frag_idx) {
  std::pair<bool, int64_t> skip_frag{false, -1};
  for (auto& inner_join : ra_exe_unit.join_quals) {
    if (inner_join.type != JoinType::INNER) {
      continue;
    }

    // extracting all the conjunctive simple_quals from the quals stored for the inner
    // join
    std::list<std::shared_ptr<Analyzer::Expr>> inner_join_simple_quals;
    for (auto& qual : inner_join.quals) {
      auto temp_qual = qual_to_conjunctive_form(qual);
      inner_join_simple_quals.insert(inner_join_simple_quals.begin(),
                                     temp_qual.simple_quals.begin(),
                                     temp_qual.simple_quals.end());
    }
    auto temp_skip_frag = skipFragment(
        table_desc, fragment, inner_join_simple_quals, frag_offsets, frag_idx);
    if (temp_skip_frag.second != -1) {
      skip_frag.second = temp_skip_frag.second;
      return skip_frag;
    } else {
      skip_frag.first = skip_frag.first || temp_skip_frag.first;
    }
  }
  return skip_frag;
}

AggregatedColRange Executor::computeColRangesCache(
    const std::unordered_set<PhysicalInput>& phys_inputs) {
  AggregatedColRange agg_col_range_cache;
  CHECK(catalog_);
  std::unordered_set<int> phys_table_ids;
  for (const auto& phys_input : phys_inputs) {
    phys_table_ids.insert(phys_input.table_id);
  }
  std::vector<InputTableInfo> query_infos;
  for (const int table_id : phys_table_ids) {
    query_infos.emplace_back(InputTableInfo{table_id, getTableInfo(table_id)});
  }
  for (const auto& phys_input : phys_inputs) {
    const auto cd =
        catalog_->getMetadataForColumn(phys_input.table_id, phys_input.col_id);
    CHECK(cd);
    if (ExpressionRange::typeSupportsRange(cd->columnType)) {
      const auto col_var = boost::make_unique<Analyzer::ColumnVar>(
          cd->columnType, phys_input.table_id, phys_input.col_id, 0);
      const auto col_range = getLeafColumnRange(col_var.get(), query_infos, this, false);
      agg_col_range_cache.setColRange(phys_input, col_range);
    }
  }
  return agg_col_range_cache;
}

StringDictionaryGenerations Executor::computeStringDictionaryGenerations(
    const std::unordered_set<PhysicalInput>& phys_inputs) {
  StringDictionaryGenerations string_dictionary_generations;
  CHECK(catalog_);
  for (const auto& phys_input : phys_inputs) {
    const auto cd =
        catalog_->getMetadataForColumn(phys_input.table_id, phys_input.col_id);
    CHECK(cd);
    const auto& col_ti =
        cd->columnType.is_array() ? cd->columnType.get_elem_type() : cd->columnType;
    if (col_ti.is_string() && col_ti.get_compression() == kENCODING_DICT) {
      const int dict_id = col_ti.get_comp_param();
      const auto dd = catalog_->getMetadataForDict(dict_id);
      CHECK(dd && dd->stringDict);
      string_dictionary_generations.setGeneration(dict_id,
                                                  dd->stringDict->storageEntryCount());
    }
  }
  return string_dictionary_generations;
}

TableGenerations Executor::computeTableGenerations(
    std::unordered_set<int> phys_table_ids) {
  TableGenerations table_generations;
  for (const int table_id : phys_table_ids) {
    const auto table_info = getTableInfo(table_id);
    table_generations.setGeneration(
        table_id,
        TableGeneration{static_cast<int64_t>(table_info.getPhysicalNumTuples()), 0});
  }
  return table_generations;
}

void Executor::setupCaching(const std::unordered_set<PhysicalInput>& phys_inputs,
                            const std::unordered_set<int>& phys_table_ids) {
  CHECK(catalog_);
  row_set_mem_owner_ = std::make_shared<RowSetMemoryOwner>(Executor::getArenaBlockSize());
  row_set_mem_owner_->setDictionaryGenerations(
      computeStringDictionaryGenerations(phys_inputs));
  agg_col_range_cache_ = computeColRangesCache(phys_inputs);
  table_generations_ = computeTableGenerations(phys_table_ids);
}

mapd_shared_mutex& Executor::getSessionLock() {
  return executor_session_mutex_;
}

std::string& Executor::getCurrentQuerySession(
    mapd_shared_lock<mapd_shared_mutex>& read_lock) {
  return current_query_session_;
}

size_t Executor::getRunningExecutorId(mapd_shared_lock<mapd_shared_mutex>& read_lock) {
  return running_query_executor_id_;
}

bool Executor::checkCurrentQuerySession(const QuerySessionId& candidate_query_session,
                                        mapd_shared_lock<mapd_shared_mutex>& read_lock) {
  // if current_query_session is equal to the candidate_query_session,
  // or it is empty session we consider
  return !candidate_query_session.empty() &&
         (current_query_session_ == candidate_query_session);
}

void Executor::invalidateRunningQuerySession(
    mapd_unique_lock<mapd_shared_mutex>& write_lock) {
  current_query_session_ = "";
  running_query_executor_id_ = Executor::UNITARY_EXECUTOR_ID;
}

CurrentQueryStatus Executor::attachExecutorToQuerySession(
    std::shared_ptr<const query_state::QueryState>& query_state) {
  QuerySessionId query_session_id = "";
  std::string query_str = "N/A";
  {
    mapd_shared_lock<mapd_shared_mutex> read_lock(executor_session_mutex_);
    // gather necessary query's info
    if (query_state != nullptr && query_state->getConstSessionInfo() != nullptr) {
      query_session_id = query_state->getConstSessionInfo()->get_session_id();
      query_str = query_state->getQueryStr();
    } else if (!getCurrentQuerySession(read_lock).empty()) {
      query_session_id = getCurrentQuerySession(read_lock);
    }
  }
  if (!query_session_id.empty()) {
    // if session is valid, do update 1) the exact executor id and 2) query status
    mapd_unique_lock<mapd_shared_mutex> write_lock(executor_session_mutex_);
    updateQuerySessionExecutorAssignment(
        query_session_id, query_state->getQuerySubmittedTime(), executor_id_, write_lock);
    updateQuerySessionStatusWithLock(query_session_id,
                                     query_state->getQuerySubmittedTime(),
                                     QuerySessionStatus::QueryStatus::PENDING_EXECUTOR,
                                     write_lock);
  }
  return {query_session_id, query_str};
}

void Executor::checkPendingQueryStatus(const QuerySessionId& query_session) {
  // check whether we are okay to execute the "pending" query
  // i.e., before running the query check if this query session is "ALREADY" interrupted
  mapd_unique_lock<mapd_shared_mutex> session_write_lock(executor_session_mutex_);
  if (query_session.empty()) {
    return;
  }
  if (queries_interrupt_flag_.find(query_session) == queries_interrupt_flag_.end()) {
    // something goes wrong since we assume this is caller's responsibility
    // (call this function only for enrolled query session)
    if (!queries_session_map_.count(query_session)) {
      VLOG(1) << "Interrupting pending query is not available since the query session is "
                 "not enrolled";
    } else {
      // here the query session is enrolled but the interrupt flag is not registered
      VLOG(1)
          << "Interrupting pending query is not available since its interrupt flag is "
             "not registered";
    }
    return;
  }
  if (queries_interrupt_flag_[query_session]) {
    throw QueryExecutionError(Executor::ERR_INTERRUPTED);
  }
}

void Executor::clearQuerySessionStatus(
    const QuerySessionId& query_session,
    const std::chrono::time_point<std::chrono::system_clock> submitted,
    bool acquire_spin_lock) {
  mapd_unique_lock<mapd_shared_mutex> session_write_lock(executor_session_mutex_);
  // clear the interrupt-related info for a finished query
  if (query_session.empty()) {
    return;
  }
  removeFromQuerySessionList(query_session, submitted, session_write_lock);
  if (query_session.compare(current_query_session_) == 0 &&
      running_query_executor_id_ == executor_id_) {
    invalidateRunningQuerySession(session_write_lock);
    if (acquire_spin_lock) {
      // for single-node execution, and dist mode uses external spin_lock
      // instead of executor's internal spin_lock
      execute_spin_lock_.clear(std::memory_order_release);
      resetInterrupt();
    }
  }
}

void Executor::updateQuerySessionStatus(
    std::shared_ptr<const query_state::QueryState>& query_state,
    const QuerySessionStatus::QueryStatus new_query_status) {
  // update the running query session's the current status
  if (query_state) {
    mapd_unique_lock<mapd_shared_mutex> session_write_lock(executor_session_mutex_);
    auto query_session = query_state->getConstSessionInfo()->get_session_id();
    if (query_session.empty()) {
      return;
    }
    if (new_query_status == QuerySessionStatus::QueryStatus::RUNNING) {
      current_query_session_ = query_session;
      running_query_executor_id_ = executor_id_;
    }
    updateQuerySessionStatusWithLock(query_session,
                                     query_state->getQuerySubmittedTime(),
                                     new_query_status,
                                     session_write_lock);
  }
}

void Executor::updateQuerySessionStatus(
    const QuerySessionId& query_session,
    const std::chrono::time_point<std::chrono::system_clock> submitted,
    const QuerySessionStatus::QueryStatus new_query_status) {
  // update the running query session's the current status
  mapd_unique_lock<mapd_shared_mutex> session_write_lock(executor_session_mutex_);
  if (query_session.empty()) {
    return;
  }
  if (new_query_status == QuerySessionStatus::QueryStatus::RUNNING) {
    current_query_session_ = query_session;
    running_query_executor_id_ = executor_id_;
  }
  updateQuerySessionStatusWithLock(
      query_session, submitted, new_query_status, session_write_lock);
}

void Executor::enrollQuerySession(
    const QuerySessionId& query_session,
    const std::string& query_str,
    const std::chrono::time_point<std::chrono::system_clock> submitted,
    const size_t executor_id,
    const QuerySessionStatus::QueryStatus query_session_status) {
  // enroll the query session into the Executor's session map
  mapd_unique_lock<mapd_shared_mutex> session_write_lock(executor_session_mutex_);
  if (query_session.empty()) {
    return;
  }
  addToQuerySessionList(query_session,
                        query_str,
                        submitted,
                        executor_id,
                        query_session_status,
                        session_write_lock);
}

bool Executor::addToQuerySessionList(
    const QuerySessionId& query_session,
    const std::string& query_str,
    const std::chrono::time_point<std::chrono::system_clock> submitted,
    const size_t executor_id,
    const QuerySessionStatus::QueryStatus query_status,
    mapd_unique_lock<mapd_shared_mutex>& write_lock) {
  // an internal API that enrolls the query session into the Executor's session map
  auto t_str = ::toString(submitted);
  if (queries_session_map_.count(query_session)) {
    if (queries_session_map_.at(query_session).count(t_str)) {
      queries_session_map_.at(query_session).erase(t_str);
      queries_session_map_.at(query_session)
          .emplace(t_str,
                   QuerySessionStatus(
                       query_session, executor_id, query_str, submitted, query_status));
    } else {
      queries_session_map_.at(query_session)
          .emplace(t_str,
                   QuerySessionStatus(
                       query_session, executor_id, query_str, submitted, query_status));
    }
  } else {
    std::map<std::string, QuerySessionStatus> executor_per_query_map;
    executor_per_query_map.emplace(
        t_str,
        QuerySessionStatus(
            query_session, executor_id, query_str, submitted, query_status));
    queries_session_map_.emplace(query_session, executor_per_query_map);
  }
  return queries_interrupt_flag_.emplace(query_session, false).second;
}

bool Executor::updateQuerySessionStatusWithLock(
    const QuerySessionId& query_session,
    const std::chrono::time_point<std::chrono::system_clock> submitted,
    const QuerySessionStatus::QueryStatus updated_query_status,
    mapd_unique_lock<mapd_shared_mutex>& write_lock) {
  // an internal API that updates query session status
  if (query_session.empty()) {
    return false;
  }
  auto t_str = ::toString(submitted);
  if (queries_session_map_.count(query_session)) {
    for (auto& query_status : queries_session_map_.at(query_session)) {
      auto target_submitted_t_str =
          ::toString(query_status.second.getQuerySubmittedTime());
      // no time difference --> found the target query status
      if (t_str.compare(target_submitted_t_str) == 0) {
        auto prev_status = query_status.second.getQueryStatus();
        if (prev_status == updated_query_status) {
          return false;
        }
        query_status.second.setQueryStatus(updated_query_status);
        return true;
      }
    }
  }
  return false;
}

bool Executor::updateQuerySessionExecutorAssignment(
    const QuerySessionId& query_session,
    const std::chrono::time_point<std::chrono::system_clock> submitted,
    const size_t executor_id,
    mapd_unique_lock<mapd_shared_mutex>& write_lock) {
  // update the executor id of the query session
  if (query_session.empty()) {
    return false;
  }
  auto t_str = ::toString(submitted);
  if (queries_session_map_.count(query_session)) {
    auto storage = queries_session_map_.at(query_session);
    for (auto it = storage.begin(); it != storage.end(); it++) {
      auto target_submitted_t_str = ::toString(it->second.getQuerySubmittedTime());
      // no time difference --> found the target query status
      if (t_str.compare(target_submitted_t_str) == 0) {
        queries_session_map_.at(query_session).at(t_str).setExecutorId(executor_id);
        return true;
      }
    }
  }
  return false;
}

bool Executor::removeFromQuerySessionList(
    const QuerySessionId& query_session,
    const std::chrono::time_point<std::chrono::system_clock> submitted,
    mapd_unique_lock<mapd_shared_mutex>& write_lock) {
  if (query_session.empty()) {
    return false;
  }
  auto t_str = ::toString(submitted);
  if (queries_session_map_.count(query_session)) {
    auto& storage = queries_session_map_.at(query_session);
    if (storage.size() > 1) {
      // in this case we only remove query executor info
      for (auto it = storage.begin(); it != storage.end(); it++) {
        auto target_submitted_t_str = ::toString(it->second.getQuerySubmittedTime());
        // no time difference && have the same executor id--> found the target query
        if (it->second.getExecutorId() == executor_id_ &&
            t_str.compare(target_submitted_t_str) == 0) {
          storage.erase(it);
          return true;
        }
      }
    } else if (storage.size() == 1) {
      // here this session only has a single query executor
      // so we clear both executor info and its interrupt flag
      queries_session_map_.erase(query_session);
      queries_interrupt_flag_.erase(query_session);
      if (interrupted_.load()) {
        interrupted_.store(false);
        VLOG(1) << "RESET Executor " << this << " that had previously been interrupted";
      }
      return true;
    }
  }
  return false;
}

void Executor::setQuerySessionAsInterrupted(
    const QuerySessionId& query_session,
    mapd_unique_lock<mapd_shared_mutex>& write_lock) {
  if (query_session.empty()) {
    return;
  }
  if (queries_interrupt_flag_.find(query_session) != queries_interrupt_flag_.end()) {
    queries_interrupt_flag_[query_session] = true;
  }
}

bool Executor::checkIsQuerySessionInterrupted(
    const QuerySessionId& query_session,
    mapd_shared_lock<mapd_shared_mutex>& read_lock) {
  auto flag_it = queries_interrupt_flag_.find(query_session);
  return !query_session.empty() && flag_it != queries_interrupt_flag_.end() &&
         flag_it->second;
}

bool Executor::checkIsQuerySessionEnrolled(
    const QuerySessionId& query_session,
    mapd_shared_lock<mapd_shared_mutex>& read_lock) {
  return !query_session.empty() && queries_session_map_.count(query_session);
}

void Executor::enableRuntimeQueryInterrupt(
    const double runtime_query_check_freq,
    const unsigned pending_query_check_freq) const {
  // The only one scenario that we intentionally call this function is
  // to allow runtime query interrupt in QueryRunner for test cases.
  // Because test machine's default setting does not allow runtime query interrupt,
  // so we have to turn it on within test code if necessary.
  g_enable_runtime_query_interrupt = true;
  g_pending_query_interrupt_freq = pending_query_check_freq;
  g_running_query_interrupt_freq = runtime_query_check_freq;
  if (g_running_query_interrupt_freq) {
    g_running_query_interrupt_freq = 0.5;
  }
}

void Executor::addToCardinalityCache(const std::string& cache_key,
                                     const size_t cache_value) {
  if (g_use_estimator_result_cache) {
    mapd_unique_lock<mapd_shared_mutex> lock(recycler_mutex_);
    cardinality_cache_[cache_key] = cache_value;
    VLOG(1) << "Put estimated cardinality to the cache";
  }
}

Executor::CachedCardinality Executor::getCachedCardinality(const std::string& cache_key) {
  mapd_shared_lock<mapd_shared_mutex> lock(recycler_mutex_);
  if (g_use_estimator_result_cache &&
      cardinality_cache_.find(cache_key) != cardinality_cache_.end()) {
    VLOG(1) << "Reuse cached cardinality";
    return {true, cardinality_cache_[cache_key]};
  }
  return {false, -1};
}

std::vector<QuerySessionStatus> Executor::getQuerySessionInfo(
    const QuerySessionId& query_session,
    mapd_shared_lock<mapd_shared_mutex>& read_lock) {
  if (!queries_session_map_.empty() && queries_session_map_.count(query_session)) {
    auto& query_infos = queries_session_map_.at(query_session);
    std::vector<QuerySessionStatus> ret;
    for (auto& info : query_infos) {
      ret.push_back(QuerySessionStatus(query_session,
                                       info.second.getExecutorId(),
                                       info.second.getQueryStr(),
                                       info.second.getQuerySubmittedTime(),
                                       info.second.getQueryStatus()));
    }
    return ret;
  }
  return {};
}

std::map<int, std::shared_ptr<Executor>> Executor::executors_;

std::atomic_flag Executor::execute_spin_lock_ = ATOMIC_FLAG_INIT;
// current running query's session ID
std::string Executor::current_query_session_{""};
// running executor's id
size_t Executor::running_query_executor_id_{0};
// contain the interrupt flag's status per query session
InterruptFlagMap Executor::queries_interrupt_flag_;
// contain a list of queries per query session
QuerySessionMap Executor::queries_session_map_;
// session lock
mapd_shared_mutex Executor::executor_session_mutex_;

mapd_shared_mutex Executor::execute_mutex_;
mapd_shared_mutex Executor::executors_cache_mutex_;

std::mutex Executor::gpu_active_modules_mutex_;
uint32_t Executor::gpu_active_modules_device_mask_{0x0};
void* Executor::gpu_active_modules_[max_gpu_count];
std::atomic<bool> Executor::interrupted_{false};

std::mutex Executor::compilation_mutex_;
std::mutex Executor::kernel_mutex_;

mapd_shared_mutex Executor::recycler_mutex_;
std::unordered_map<std::string, size_t> Executor::cardinality_cache_;