Execute.h

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.
 */

#ifndef QUERYENGINE_EXECUTE_H
#define QUERYENGINE_EXECUTE_H

#include "AggregatedColRange.h"
#include "BufferCompaction.h"
#include "CartesianProduct.h"
#include "CgenState.h"
#include "CodeCache.h"
#include "DateTimeUtils.h"
#include "Descriptors/QueryFragmentDescriptor.h"
#include "GpuSharedMemoryContext.h"
#include "GroupByAndAggregate.h"
#include "JoinHashTable.h"
#include "LoopControlFlow/JoinLoop.h"
#include "NvidiaKernel.h"
#include "PlanState.h"
#include "RelAlgExecutionUnit.h"
#include "RelAlgTranslator.h"
#include "StringDictionaryGenerations.h"
#include "TableGenerations.h"
#include "TargetMetaInfo.h"
#include "WindowContext.h"

#include "../Shared/Logger.h"
#include "../Shared/SystemParameters.h"
#include "../Shared/mapd_shared_mutex.h"
#include "../Shared/measure.h"
#include "../Shared/thread_count.h"
#include "../StringDictionary/LruCache.hpp"
#include "../StringDictionary/StringDictionary.h"
#include "../StringDictionary/StringDictionaryProxy.h"
#include "DataMgr/Chunk/Chunk.h"
#include "ThriftHandler/CommandLineOptions.h"

#include <llvm/IR/Function.h>
#include <llvm/IR/Value.h>
#include <llvm/Linker/Linker.h>
#include <llvm/Transforms/Utils/ValueMapper.h>
#include <rapidjson/document.h>

#include <algorithm>
#include <atomic>
#include <condition_variable>
#include <cstddef>
#include <cstdlib>
#include <deque>
#include <functional>
#include <limits>
#include <map>
#include <mutex>
#include <stack>
#include <unordered_map>
#include <unordered_set>

class QueryCompilationDescriptor;
using QueryCompilationDescriptorOwned = std::unique_ptr<QueryCompilationDescriptor>;
class QueryMemoryDescriptor;
using QueryMemoryDescriptorOwned = std::unique_ptr<QueryMemoryDescriptor>;
using InterruptFlagMap = std::map<std::string, bool>;

extern void read_udf_gpu_module(const std::string& udf_ir_filename);
extern void read_udf_cpu_module(const std::string& udf_ir_filename);
extern bool is_udf_module_present(bool cpu_only = false);
extern void read_rt_udf_gpu_module(const std::string& udf_ir);
extern void read_rt_udf_cpu_module(const std::string& udf_ir);
extern bool is_rt_udf_module_present(bool cpu_only = false);

class ColumnFetcher;
class ExecutionResult;

class WatchdogException : public std::runtime_error {
 public:
  WatchdogException(const std::string& cause) : std::runtime_error(cause) {}
};

class Executor;

inline llvm::Value* get_arg_by_name(llvm::Function* func, const std::string& name) {
  for (auto& arg : func->args()) {
    if (arg.getName() == name) {
      return &arg;
    }
  }
  CHECK(false);
  return nullptr;
}

inline uint32_t log2_bytes(const uint32_t bytes) {
  switch (bytes) {
    case 1:
      return 0;
    case 2:
      return 1;
    case 4:
      return 2;
    case 8:
      return 3;
    default:
      abort();
  }
}

inline const ColumnDescriptor* get_column_descriptor(
    const int col_id,
    const int table_id,
    const Catalog_Namespace::Catalog& cat) {
  CHECK_GT(table_id, 0);
  const auto col_desc = cat.getMetadataForColumn(table_id, col_id);
  CHECK(col_desc);
  return col_desc;
}

inline const Analyzer::Expr* extract_cast_arg(const Analyzer::Expr* expr) {
  const auto cast_expr = dynamic_cast<const Analyzer::UOper*>(expr);
  if (!cast_expr || cast_expr->get_optype() != kCAST) {
    return expr;
  }
  return cast_expr->get_operand();
}

inline std::string numeric_type_name(const SQLTypeInfo& ti) {
  CHECK(ti.is_integer() || ti.is_decimal() || ti.is_boolean() || ti.is_time() ||
        ti.is_fp() || (ti.is_string() && ti.get_compression() == kENCODING_DICT) ||
        ti.is_timeinterval());
  if (ti.is_integer() || ti.is_decimal() || ti.is_boolean() || ti.is_time() ||
      ti.is_string() || ti.is_timeinterval()) {
    return "int" + std::to_string(ti.get_logical_size() * 8) + "_t";
  }
  return ti.get_type() == kDOUBLE ? "double" : "float";
}

inline const ColumnDescriptor* get_column_descriptor_maybe(
    const int col_id,
    const int table_id,
    const Catalog_Namespace::Catalog& cat) {
  CHECK(table_id);
  return table_id > 0 ? get_column_descriptor(col_id, table_id, cat) : nullptr;
}

inline const ResultSetPtr& get_temporary_table(const TemporaryTables* temporary_tables,
                                               const int table_id) {
  CHECK_LT(table_id, 0);
  const auto it = temporary_tables->find(table_id);
  CHECK(it != temporary_tables->end());
  return it->second;
}

inline const SQLTypeInfo get_column_type(const int col_id,
                                         const int table_id,
                                         const ColumnDescriptor* cd,
                                         const TemporaryTables* temporary_tables) {
  CHECK(cd || temporary_tables);
  if (cd) {
    CHECK_EQ(col_id, cd->columnId);
    CHECK_EQ(table_id, cd->tableId);
    return cd->columnType;
  }
  const auto& temp = get_temporary_table(temporary_tables, table_id);
  return temp->getColType(col_id);
}

template <typename PtrTy>
inline const ColumnarResults* rows_to_columnar_results(
    std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner,
    const PtrTy& result,
    const int number) {
  std::vector<SQLTypeInfo> col_types;
  for (size_t i = 0; i < result->colCount(); ++i) {
    col_types.push_back(get_logical_type_info(result->getColType(i)));
  }
  return new ColumnarResults(row_set_mem_owner, *result, number, col_types);
}

// TODO(alex): Adjust interfaces downstream and make this not needed.
inline std::vector<Analyzer::Expr*> get_exprs_not_owned(
    const std::vector<std::shared_ptr<Analyzer::Expr>>& exprs) {
  std::vector<Analyzer::Expr*> exprs_not_owned;
  for (const auto& expr : exprs) {
    exprs_not_owned.push_back(expr.get());
  }
  return exprs_not_owned;
}

inline const ColumnarResults* columnarize_result(
    std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner,
    const ResultSetPtr& result,
    const int frag_id) {
  INJECT_TIMER(columnarize_result);
  CHECK_EQ(0, frag_id);
  return rows_to_columnar_results(row_set_mem_owner, result, result->colCount());
}

class CompilationRetryNoLazyFetch : public std::runtime_error {
 public:
  CompilationRetryNoLazyFetch()
      : std::runtime_error("Retry query compilation with no GPU lazy fetch.") {}
};

class CompilationRetryNewScanLimit : public std::runtime_error {
 public:
  CompilationRetryNewScanLimit(const size_t new_scan_limit)
      : std::runtime_error("Retry query compilation with new scan limit.")
      , new_scan_limit_(new_scan_limit) {}

  size_t new_scan_limit_;
};

class TooManyLiterals : public std::runtime_error {
 public:
  TooManyLiterals() : std::runtime_error("Too many literals in the query") {}
};

class CompilationRetryNoCompaction : public std::runtime_error {
 public:
  CompilationRetryNoCompaction()
      : std::runtime_error("Retry query compilation with no compaction.") {}
};

class QueryMustRunOnCpu : public std::runtime_error {
 public:
  QueryMustRunOnCpu() : std::runtime_error("Query must run in cpu mode.") {}
};

class SringConstInResultSet : public std::runtime_error {
 public:
  SringConstInResultSet()
      : std::runtime_error(
            "NONE ENCODED String types are not supported as input result set.") {}
};

class ExtensionFunction;

namespace std {
template <>
struct hash<std::vector<int>> {
  size_t operator()(const std::vector<int>& vec) const {
    return vec.size() ^ boost::hash_range(vec.begin(), vec.end());
  }
};

template <>
struct hash<std::pair<int, int>> {
  size_t operator()(const std::pair<int, int>& p) const {
    return boost::hash<std::pair<int, int>>()(p);
  }
};

}  // namespace std

using RowDataProvider = Fragmenter_Namespace::RowDataProvider;

class UpdateLogForFragment : public RowDataProvider {
 public:
  using FragmentInfoType = Fragmenter_Namespace::FragmentInfo;

  UpdateLogForFragment(FragmentInfoType const& fragment_info,
                       size_t const,
                       const std::shared_ptr<ResultSet>& rs);

  std::vector<TargetValue> getEntryAt(const size_t index) const override;
  std::vector<TargetValue> getTranslatedEntryAt(const size_t index) const override;

  size_t const getRowCount() const override;
  StringDictionaryProxy* getLiteralDictionary() const override {
    return rs_->getRowSetMemOwner()->getLiteralStringDictProxy();
  }
  size_t const getEntryCount() const override;
  size_t const getFragmentIndex() const;
  FragmentInfoType const& getFragmentInfo() const;
  decltype(FragmentInfoType::physicalTableId) const getPhysicalTableId() const {
    return fragment_info_.physicalTableId;
  }
  decltype(FragmentInfoType::fragmentId) const getFragmentId() const {
    return fragment_info_.fragmentId;
  }

  SQLTypeInfo getColumnType(const size_t col_idx) const;

  using Callback = std::function<void(const UpdateLogForFragment&)>;

  auto getResultSet() const { return rs_; }

 private:
  FragmentInfoType const& fragment_info_;
  size_t fragment_index_;
  std::shared_ptr<ResultSet> rs_;
};

using LLVMValueVector = std::vector<llvm::Value*>;

class QueryCompilationDescriptor;

struct FetchResult {
  std::vector<std::vector<const int8_t*>> col_buffers;
  std::vector<std::vector<int64_t>> num_rows;
  std::vector<std::vector<uint64_t>> frag_offsets;
};

std::ostream& operator<<(std::ostream&, FetchResult const&);

class Executor {
  static_assert(sizeof(float) == 4 && sizeof(double) == 8,
                "Host hardware not supported, unexpected size of float / double.");
  static_assert(sizeof(time_t) == 8,
                "Host hardware not supported, 64-bit time support is required.");

 public:
  using ExecutorId = size_t;
  static const ExecutorId UNITARY_EXECUTOR_ID = 0;

  Executor(const ExecutorId 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);

  static std::shared_ptr<Executor> getExecutor(
      const ExecutorId id,
      const std::string& debug_dir = "",
      const std::string& debug_file = "",
      const SystemParameters system_parameters = SystemParameters());

  static void nukeCacheOfExecutors() {
    mapd_unique_lock<mapd_shared_mutex> flush_lock(
        execute_mutex_);  // don't want native code to vanish while executing
    mapd_unique_lock<mapd_shared_mutex> lock(executors_cache_mutex_);
    (decltype(executors_){}).swap(executors_);
  }

  static void clearMemory(const Data_Namespace::MemoryLevel memory_level);

  static size_t getArenaBlockSize();

  StringDictionaryProxy* getStringDictionaryProxy(
      const int dictId,
      const std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner,
      const bool with_generation) const;

  bool isCPUOnly() const;

  bool isArchMaxwell(const ExecutorDeviceType dt) const;

  bool containsLeftDeepOuterJoin() const {
    return cgen_state_->contains_left_deep_outer_join_;
  }

  const ColumnDescriptor* getColumnDescriptor(const Analyzer::ColumnVar*) const;

  const ColumnDescriptor* getPhysicalColumnDescriptor(const Analyzer::ColumnVar*,
                                                      int) const;

  const Catalog_Namespace::Catalog* getCatalog() const;
  void setCatalog(const Catalog_Namespace::Catalog* catalog);

  const std::shared_ptr<RowSetMemoryOwner> getRowSetMemoryOwner() const;

  const TemporaryTables* getTemporaryTables() const;

  Fragmenter_Namespace::TableInfo getTableInfo(const int table_id) const;

  const TableGeneration& getTableGeneration(const int table_id) const;

  ExpressionRange getColRange(const PhysicalInput&) const;

  size_t getNumBytesForFetchedRow(const std::set<int>& table_ids_to_fetch) const;

  std::vector<ColumnLazyFetchInfo> getColLazyFetchInfo(
      const std::vector<Analyzer::Expr*>& target_exprs) const;

  void registerActiveModule(void* module, const int device_id) const;
  void unregisterActiveModule(void* module, const int device_id) const;
  void interrupt(const std::string& query_session = "",
                 const std::string& interrupt_session = "");
  void resetInterrupt();

  // only for testing usage
  void enableRuntimeQueryInterrupt(const unsigned interrupt_freq) const;

  static const size_t high_scan_limit{32000000};

  int8_t warpSize() const;
  unsigned gridSize() const;
  unsigned numBlocksPerMP() const;
  unsigned blockSize() const;
  size_t maxGpuSlabSize() const;

 private:
  void clearMetaInfoCache();

  int deviceCount(const ExecutorDeviceType) const;
  int deviceCountForMemoryLevel(const Data_Namespace::MemoryLevel memory_level) const;

  // Generate code for a window function target.
  llvm::Value* codegenWindowFunction(const size_t target_index,
                                     const CompilationOptions& co);

  // Generate code for an aggregate window function target.
  llvm::Value* codegenWindowFunctionAggregate(const CompilationOptions& co);

  // The aggregate state requires a state reset when starting a new partition. Generate
  // the new partition check and return the continuation basic block.
  llvm::BasicBlock* codegenWindowResetStateControlFlow();

  // Generate code for initializing the state of a window aggregate.
  void codegenWindowFunctionStateInit(llvm::Value* aggregate_state);

  // Generates the required calls for an aggregate window function and returns the final
  // result.
  llvm::Value* codegenWindowFunctionAggregateCalls(llvm::Value* aggregate_state,
                                                   const CompilationOptions& co);

  // The AVG window function requires some post-processing: the sum is divided by count
  // and the result is stored back for the current row.
  void codegenWindowAvgEpilogue(llvm::Value* crt_val,
                                llvm::Value* window_func_null_val,
                                llvm::Value* multiplicity_lv);

  // Generates code which loads the current aggregate value for the window context.
  llvm::Value* codegenAggregateWindowState();

  llvm::Value* aggregateWindowStatePtr();

  struct CompilationResult {
    std::shared_ptr<CompilationContext> generated_code;
    std::unordered_map<int, CgenState::LiteralValues> literal_values;
    bool output_columnar;
    std::string llvm_ir;
    GpuSharedMemoryContext gpu_smem_context;
  };

  bool isArchPascalOrLater(const ExecutorDeviceType dt) const {
    if (dt == ExecutorDeviceType::GPU) {
      const auto cuda_mgr = catalog_->getDataMgr().getCudaMgr();
      LOG_IF(FATAL, cuda_mgr == nullptr)
          << "No CudaMgr instantiated, unable to check device architecture";
      return cuda_mgr->isArchPascalOrLater();
    }
    return false;
  }

  bool needFetchAllFragments(const InputColDescriptor& col_desc,
                             const RelAlgExecutionUnit& ra_exe_unit,
                             const FragmentsList& selected_fragments) const;

  class ExecutionDispatch {
   private:
    Executor* executor_;
    const RelAlgExecutionUnit& ra_exe_unit_;
    const std::vector<InputTableInfo>& query_infos_;
    const Catalog_Namespace::Catalog& cat_;
    mutable std::vector<uint64_t> all_frag_row_offsets_;
    mutable std::mutex all_frag_row_offsets_mutex_;
    const std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner_;
    RenderInfo* render_info_;
    std::vector<std::pair<ResultSetPtr, std::vector<size_t>>> all_fragment_results_;
    std::atomic_flag dynamic_watchdog_set_ = ATOMIC_FLAG_INIT;
    static std::mutex reduce_mutex_;

    void runImpl(const ExecutorDeviceType chosen_device_type,
                 int chosen_device_id,
                 const ExecutionOptions& eo,
                 const ColumnFetcher& column_fetcher,
                 const QueryCompilationDescriptor& query_comp_desc,
                 const QueryMemoryDescriptor& query_mem_desc,
                 const FragmentsList& frag_list,
                 const ExecutorDispatchMode kernel_dispatch_mode,
                 const int64_t rowid_lookup_key);

   public:
    ExecutionDispatch(Executor* executor,
                      const RelAlgExecutionUnit& ra_exe_unit,
                      const std::vector<InputTableInfo>& query_infos,
                      const Catalog_Namespace::Catalog& cat,
                      const std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner,
                      RenderInfo* render_info);

    ExecutionDispatch(const ExecutionDispatch&) = delete;

    ExecutionDispatch& operator=(const ExecutionDispatch&) = delete;

    ExecutionDispatch(ExecutionDispatch&&) = delete;

    ExecutionDispatch& operator=(ExecutionDispatch&&) = delete;

    std::tuple<QueryCompilationDescriptorOwned, QueryMemoryDescriptorOwned> compile(
        const size_t max_groups_buffer_entry_guess,
        const int8_t crt_min_byte_width,
        const CompilationOptions& co,
        const ExecutionOptions& eo,
        const ColumnFetcher& column_fetcher,
        const bool has_cardinality_estimation);

    void run(const ExecutorDeviceType chosen_device_type,
             int chosen_device_id,
             const ExecutionOptions& eo,
             const ColumnFetcher& column_fetcher,
             const QueryCompilationDescriptor& query_comp_desc,
             const QueryMemoryDescriptor& query_mem_desc,
             const FragmentsList& frag_ids,
             const ExecutorDispatchMode kernel_dispatch_mode,
             const int64_t rowid_lookup_key);

    const RelAlgExecutionUnit& getExecutionUnit() const;

    const std::vector<uint64_t>& getFragOffsets() const;

    std::vector<std::pair<ResultSetPtr, std::vector<size_t>>>& getFragmentResults();

    friend class QueryCompilationDescriptor;
  };

  ResultSetPtr executeWorkUnit(size_t& max_groups_buffer_entry_guess,
                               const bool is_agg,
                               const std::vector<InputTableInfo>&,
                               const RelAlgExecutionUnit&,
                               const CompilationOptions&,
                               const ExecutionOptions& options,
                               const Catalog_Namespace::Catalog&,
                               std::shared_ptr<RowSetMemoryOwner>,
                               RenderInfo* render_info,
                               const bool has_cardinality_estimation,
                               ColumnCacheMap& column_cache);

  void executeUpdate(const RelAlgExecutionUnit& ra_exe_unit,
                     const std::vector<InputTableInfo>& table_infos,
                     const CompilationOptions& co,
                     const ExecutionOptions& eo,
                     const Catalog_Namespace::Catalog& cat,
                     std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner,
                     const UpdateLogForFragment::Callback& cb,
                     const bool is_agg);

  using PerFragmentCallBack =
      std::function<void(ResultSetPtr, const Fragmenter_Namespace::FragmentInfo&)>;

  void executeWorkUnitPerFragment(const RelAlgExecutionUnit& ra_exe_unit,
                                  const InputTableInfo& table_info,
                                  const CompilationOptions& co,
                                  const ExecutionOptions& eo,
                                  const Catalog_Namespace::Catalog& cat,
                                  PerFragmentCallBack& cb);

  ResultSetPtr executeExplain(const QueryCompilationDescriptor&);

  ResultSetPtr executeTableFunction(const TableFunctionExecutionUnit exe_unit,
                                    const std::vector<InputTableInfo>& table_infos,
                                    const CompilationOptions& co,
                                    const ExecutionOptions& eo,
                                    const Catalog_Namespace::Catalog& cat);

  // TODO(alex): remove
  ExecutorDeviceType getDeviceTypeForTargets(
      const RelAlgExecutionUnit& ra_exe_unit,
      const ExecutorDeviceType requested_device_type);

  ResultSetPtr collectAllDeviceResults(
      ExecutionDispatch& execution_dispatch,
      const std::vector<Analyzer::Expr*>& target_exprs,
      const QueryMemoryDescriptor& query_mem_desc,
      const ExecutorDeviceType device_type,
      std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner);

  ResultSetPtr collectAllDeviceShardedTopResults(
      ExecutionDispatch& execution_dispatch) const;

  std::unordered_map<int, const Analyzer::BinOper*> getInnerTabIdToJoinCond() const;

  template <typename THREAD_POOL>
  void dispatchFragments(
      const std::function<void(const ExecutorDeviceType chosen_device_type,
                               int chosen_device_id,
                               const QueryCompilationDescriptor& query_comp_desc,
                               const QueryMemoryDescriptor& query_mem_desc,
                               const FragmentsList& frag_list,
                               const ExecutorDispatchMode kernel_dispatch_mode,
                               const int64_t rowid_lookup_key)> dispatch,
      const ExecutionDispatch& execution_dispatch,
      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,
      std::unordered_set<int>& available_gpus,
      int& available_cpus);

  std::vector<size_t> 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);

  bool skipFragmentPair(const Fragmenter_Namespace::FragmentInfo& outer_fragment_info,
                        const Fragmenter_Namespace::FragmentInfo& inner_fragment_info,
                        const int inner_table_id,
                        const std::unordered_map<int, const Analyzer::BinOper*>&
                            inner_table_id_to_join_condition,
                        const RelAlgExecutionUnit& ra_exe_unit,
                        const ExecutorDeviceType device_type);

  FetchResult fetchChunks(const ColumnFetcher&,
                          const RelAlgExecutionUnit& ra_exe_unit,
                          const int device_id,
                          const Data_Namespace::MemoryLevel,
                          const std::map<int, const TableFragments*>&,
                          const FragmentsList& selected_fragments,
                          const Catalog_Namespace::Catalog&,
                          std::list<ChunkIter>&,
                          std::list<std::shared_ptr<Chunk_NS::Chunk>>&,
                          DeviceAllocator* device_allocator);

  FetchResult fetchUnionChunks(const ColumnFetcher&,
                               const RelAlgExecutionUnit& ra_exe_unit,
                               const int device_id,
                               const Data_Namespace::MemoryLevel,
                               const std::map<int, const TableFragments*>&,
                               const FragmentsList& selected_fragments,
                               const Catalog_Namespace::Catalog&,
                               std::list<ChunkIter>&,
                               std::list<std::shared_ptr<Chunk_NS::Chunk>>&,
                               DeviceAllocator* device_allocator);

  std::pair<std::vector<std::vector<int64_t>>, std::vector<std::vector<uint64_t>>>
  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);

  void 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);

  void 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);

  std::vector<size_t> getFragmentCount(const FragmentsList& selected_fragments,
                                       const size_t scan_idx,
                                       const RelAlgExecutionUnit& ra_exe_unit);

  int32_t executePlanWithGroupBy(const RelAlgExecutionUnit& ra_exe_unit,
                                 const CompilationResult&,
                                 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*,
                                 const std::vector<std::vector<int64_t>>& num_rows,
                                 const std::vector<std::vector<uint64_t>>& frag_offsets,
                                 Data_Namespace::DataMgr*,
                                 const int device_id,
                                 const int outer_table_id,
                                 const int64_t limit,
                                 const uint32_t start_rowid,
                                 const uint32_t num_tables,
                                 RenderInfo* render_info);
  int32_t executePlanWithoutGroupBy(
      const RelAlgExecutionUnit& ra_exe_unit,
      const CompilationResult&,
      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);

 public:  // Temporary, ask saman about this
  static std::pair<int64_t, int32_t> 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);

  static void addCodeToCache(const CodeCacheKey&,
                             std::shared_ptr<CompilationContext>,
                             llvm::Module*,
                             CodeCache&);

 private:
  ResultSetPtr resultsUnion(ExecutionDispatch& execution_dispatch);
  std::vector<int64_t> getJoinHashTablePtrs(const ExecutorDeviceType device_type,
                                            const int device_id);
  ResultSetPtr reduceMultiDeviceResults(
      const RelAlgExecutionUnit&,
      std::vector<std::pair<ResultSetPtr, std::vector<size_t>>>& all_fragment_results,
      std::shared_ptr<RowSetMemoryOwner>,
      const QueryMemoryDescriptor&) const;
  ResultSetPtr reduceMultiDeviceResultSets(
      std::vector<std::pair<ResultSetPtr, std::vector<size_t>>>& all_fragment_results,
      std::shared_ptr<RowSetMemoryOwner>,
      const QueryMemoryDescriptor&) const;
  ResultSetPtr reduceSpeculativeTopN(
      const RelAlgExecutionUnit&,
      std::vector<std::pair<ResultSetPtr, std::vector<size_t>>>& all_fragment_results,
      std::shared_ptr<RowSetMemoryOwner>,
      const QueryMemoryDescriptor&) const;

  ResultSetPtr executeWorkUnitImpl(size_t& max_groups_buffer_entry_guess,
                                   const bool is_agg,
                                   const bool allow_single_frag_table_opt,
                                   const std::vector<InputTableInfo>&,
                                   const RelAlgExecutionUnit&,
                                   const CompilationOptions&,
                                   const ExecutionOptions& options,
                                   const Catalog_Namespace::Catalog&,
                                   std::shared_ptr<RowSetMemoryOwner>,
                                   RenderInfo* render_info,
                                   const bool has_cardinality_estimation,
                                   ColumnCacheMap& column_cache);

  std::vector<llvm::Value*> inlineHoistedLiterals();

  std::tuple<Executor::CompilationResult, std::unique_ptr<QueryMemoryDescriptor>>
  compileWorkUnit(const std::vector<InputTableInfo>& query_infos,
                  const RelAlgExecutionUnit& ra_exe_unit,
                  const CompilationOptions& co,
                  const ExecutionOptions& eo,
                  const CudaMgr_Namespace::CudaMgr* cuda_mgr,
                  const bool allow_lazy_fetch,
                  std::shared_ptr<RowSetMemoryOwner>,
                  const size_t max_groups_buffer_entry_count,
                  const int8_t crt_min_byte_width,
                  const bool has_cardinality_estimation,
                  ColumnCacheMap& column_cache,
                  RenderInfo* render_info = nullptr);
  // Generate code to skip the deleted rows in the outermost table.
  llvm::BasicBlock* codegenSkipDeletedOuterTableRow(
      const RelAlgExecutionUnit& ra_exe_unit,
      const CompilationOptions& co);
  std::vector<JoinLoop> buildJoinLoops(RelAlgExecutionUnit& ra_exe_unit,
                                       const CompilationOptions& co,
                                       const ExecutionOptions& eo,
                                       const std::vector<InputTableInfo>& query_infos,
                                       ColumnCacheMap& column_cache);
  // Create a callback which generates code which returns true iff the row on the given
  // level is deleted.
  std::function<llvm::Value*(const std::vector<llvm::Value*>&, llvm::Value*)>
  buildIsDeletedCb(const RelAlgExecutionUnit& ra_exe_unit,
                   const size_t level_idx,
                   const CompilationOptions& co);
  // Builds a join hash table for the provided conditions on the current level.
  // Returns null iff on failure and provides the reasons in `fail_reasons`.
  std::shared_ptr<JoinHashTableInterface> buildCurrentLevelHashTable(
      const JoinCondition& current_level_join_conditions,
      RelAlgExecutionUnit& ra_exe_unit,
      const CompilationOptions& co,
      const std::vector<InputTableInfo>& query_infos,
      ColumnCacheMap& column_cache,
      std::vector<std::string>& fail_reasons);
  llvm::Value* addJoinLoopIterator(const std::vector<llvm::Value*>& prev_iters,
                                   const size_t level_idx);
  void codegenJoinLoops(const std::vector<JoinLoop>& join_loops,
                        const RelAlgExecutionUnit& ra_exe_unit,
                        GroupByAndAggregate& group_by_and_aggregate,
                        llvm::Function* query_func,
                        llvm::BasicBlock* entry_bb,
                        const QueryMemoryDescriptor& query_mem_desc,
                        const CompilationOptions& co,
                        const ExecutionOptions& eo);
  bool compileBody(const RelAlgExecutionUnit& ra_exe_unit,
                   GroupByAndAggregate& group_by_and_aggregate,
                   const QueryMemoryDescriptor& query_mem_desc,
                   const CompilationOptions& co,
                   const GpuSharedMemoryContext& gpu_smem_context = {});

  void createErrorCheckControlFlow(llvm::Function* query_func,
                                   bool run_with_dynamic_watchdog,
                                   bool run_with_allowing_runtime_interrupt,
                                   ExecutorDeviceType device_type);

  void preloadFragOffsets(const std::vector<InputDescriptor>& input_descs,
                          const std::vector<InputTableInfo>& query_infos);

  struct JoinHashTableOrError {
    std::shared_ptr<JoinHashTableInterface> hash_table;
    std::string fail_reason;
  };

  JoinHashTableOrError buildHashTableForQualifier(
      const std::shared_ptr<Analyzer::BinOper>& qual_bin_oper,
      const std::vector<InputTableInfo>& query_infos,
      const MemoryLevel memory_level,
      const JoinHashTableInterface::HashType preferred_hash_type,
      ColumnCacheMap& column_cache);
  void nukeOldState(const bool allow_lazy_fetch,
                    const std::vector<InputTableInfo>& query_infos,
                    const RelAlgExecutionUnit* ra_exe_unit);

  std::shared_ptr<CompilationContext> optimizeAndCodegenCPU(
      llvm::Function*,
      llvm::Function*,
      const std::unordered_set<llvm::Function*>&,
      const CompilationOptions&);
  std::shared_ptr<CompilationContext> optimizeAndCodegenGPU(
      llvm::Function*,
      llvm::Function*,
      std::unordered_set<llvm::Function*>&,
      const bool no_inline,
      const CudaMgr_Namespace::CudaMgr* cuda_mgr,
      const CompilationOptions&);
  std::string generatePTX(const std::string&) const;
  void initializeNVPTXBackend() const;

  int64_t deviceCycles(int milliseconds) const;

  struct GroupColLLVMValue {
    llvm::Value* translated_value;
    llvm::Value* original_value;
  };

  GroupColLLVMValue groupByColumnCodegen(Analyzer::Expr* group_by_col,
                                         const size_t col_width,
                                         const CompilationOptions&,
                                         const bool translate_null_val,
                                         const int64_t translated_null_val,
                                         GroupByAndAggregate::DiamondCodegen&,
                                         std::stack<llvm::BasicBlock*>&,
                                         const bool thread_mem_shared);

  llvm::Value* castToFP(llvm::Value* val);
  llvm::Value* castToIntPtrTyIn(llvm::Value* val, const size_t bit_width);

  RelAlgExecutionUnit addDeletedColumn(const RelAlgExecutionUnit& ra_exe_unit,
                                       const CompilationOptions& co);

  std::pair<bool, int64_t> skipFragment(
      const InputDescriptor& table_desc,
      const Fragmenter_Namespace::FragmentInfo& frag_info,
      const std::list<std::shared_ptr<Analyzer::Expr>>& simple_quals,
      const std::vector<uint64_t>& frag_offsets,
      const size_t frag_idx);

  std::pair<bool, int64_t> 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);

  AggregatedColRange computeColRangesCache(
      const std::unordered_set<PhysicalInput>& phys_inputs);
  StringDictionaryGenerations computeStringDictionaryGenerations(
      const std::unordered_set<PhysicalInput>& phys_inputs);
  TableGenerations computeTableGenerations(std::unordered_set<int> phys_table_ids);

 public:
  void setupCaching(const std::unordered_set<PhysicalInput>& phys_inputs,
                    const std::unordered_set<int>& phys_table_ids);

  template <typename SESSION_MAP_LOCK>
  void setCurrentQuerySession(const std::string& query_session,
                              SESSION_MAP_LOCK& write_lock);
  template <typename SESSION_MAP_LOCK>
  std::string& getCurrentQuerySession(SESSION_MAP_LOCK& read_lock);
  template <typename SESSION_MAP_LOCK>
  bool checkCurrentQuerySession(const std::string& candidate_query_session,
                                SESSION_MAP_LOCK& read_lock);
  template <typename SESSION_MAP_LOCK>
  void invalidateQuerySession(SESSION_MAP_LOCK& write_lock);
  template <typename SESSION_MAP_LOCK>
  bool addToQuerySessionList(const std::string& query_session,
                             SESSION_MAP_LOCK& write_lock);
  template <typename SESSION_MAP_LOCK>
  bool removeFromQuerySessionList(const std::string& query_session,
                                  SESSION_MAP_LOCK& write_lock);
  template <typename SESSION_MAP_LOCK>
  void setQuerySessionAsInterrupted(const std::string& query_session,
                                    SESSION_MAP_LOCK& write_lock);
  template <typename SESSION_MAP_LOCK>
  bool checkIsQuerySessionInterrupted(const std::string& query_session,
                                      SESSION_MAP_LOCK& read_lock);
  mapd_shared_mutex& getSessionLock();

  // true when we have matched cardinality, and false otherwise
  using CachedCardinality = std::pair<bool, size_t>;
  void addToCardinalityCache(const std::string& cache_key, const size_t cache_value);
  CachedCardinality getCachedCardinality(const std::string& cache_key);

 private:
  std::shared_ptr<CompilationContext> getCodeFromCache(const CodeCacheKey&,
                                                       const CodeCache&);

  std::vector<int8_t> serializeLiterals(
      const std::unordered_map<int, CgenState::LiteralValues>& literals,
      const int device_id);

  static size_t align(const size_t off_in, const size_t alignment) {
    size_t off = off_in;
    if (off % alignment != 0) {
      off += (alignment - off % alignment);
    }
    return off;
  }

  std::unique_ptr<CgenState> cgen_state_;

  class FetchCacheAnchor {
   public:
    FetchCacheAnchor(CgenState* cgen_state)
        : cgen_state_(cgen_state), saved_fetch_cache(cgen_state_->fetch_cache_) {}
    ~FetchCacheAnchor() { cgen_state_->fetch_cache_.swap(saved_fetch_cache); }

   private:
    CgenState* cgen_state_;
    std::unordered_map<int, std::vector<llvm::Value*>> saved_fetch_cache;
  };

  llvm::Value* spillDoubleElement(llvm::Value* elem_val, llvm::Type* elem_ty);

  std::unique_ptr<PlanState> plan_state_;
  std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner_;

  static const int max_gpu_count{16};
  std::mutex gpu_exec_mutex_[max_gpu_count];

  mutable std::mutex gpu_active_modules_mutex_;
  mutable uint32_t gpu_active_modules_device_mask_;
  mutable void* gpu_active_modules_[max_gpu_count];
  std::atomic<bool> interrupted_;

  mutable std::shared_ptr<StringDictionaryProxy> lit_str_dict_proxy_;
  mutable std::mutex str_dict_mutex_;

  mutable std::unique_ptr<llvm::TargetMachine> nvptx_target_machine_;

  CodeCache cpu_code_cache_;
  CodeCache gpu_code_cache_;

  static const size_t baseline_threshold{
      1000000};  // if a perfect hash needs more entries, use baseline
  static const size_t code_cache_size{1000};

  const unsigned block_size_x_;
  const unsigned grid_size_x_;
  const size_t max_gpu_slab_size_;
  const std::string debug_dir_;
  const std::string debug_file_;

  const ExecutorId executor_id_;
  const Catalog_Namespace::Catalog* catalog_;
  const TemporaryTables* temporary_tables_;

  int64_t kernel_queue_time_ms_ = 0;
  int64_t compilation_queue_time_ms_ = 0;

  // Singleton instance used for an execution unit which is a project with window
  // functions.
  std::unique_ptr<WindowProjectNodeContext> window_project_node_context_owned_;
  // The active window function.
  WindowFunctionContext* active_window_function_{nullptr};

  mutable InputTableInfoCache input_table_info_cache_;
  AggregatedColRange agg_col_range_cache_;
  StringDictionaryGenerations string_dictionary_generations_;
  TableGenerations table_generations_;
  static mapd_shared_mutex executor_session_mutex_;
  static std::string current_query_session_;
  // a pair of <query_session, interrupted_flag>
  static InterruptFlagMap queries_interrupt_flag_;

  static std::map<int, std::shared_ptr<Executor>> executors_;
  static std::atomic_flag execute_spin_lock_;

  // SQL queries take a shared lock, exclusive options (cache clear, memory clear) take a
  // write lock
  static mapd_shared_mutex execute_mutex_;
  static mapd_shared_mutex executors_cache_mutex_;

  // for now we use recycler_mutex only for cardinality_cache_
  // and will expand its coverage for more interesting caches for query excution
  static mapd_shared_mutex recycler_mutex_;
  static std::unordered_map<std::string, size_t> cardinality_cache_;

 public:
  static const int32_t ERR_DIV_BY_ZERO{1};
  static const int32_t ERR_OUT_OF_GPU_MEM{2};
  static const int32_t ERR_OUT_OF_SLOTS{3};
  static const int32_t ERR_UNSUPPORTED_SELF_JOIN{4};
  static const int32_t ERR_OUT_OF_RENDER_MEM{5};
  static const int32_t ERR_OUT_OF_CPU_MEM{6};
  static const int32_t ERR_OVERFLOW_OR_UNDERFLOW{7};
  static const int32_t ERR_OUT_OF_TIME{9};
  static const int32_t ERR_INTERRUPTED{10};
  static const int32_t ERR_COLUMNAR_CONVERSION_NOT_SUPPORTED{11};
  static const int32_t ERR_TOO_MANY_LITERALS{12};
  static const int32_t ERR_STRING_CONST_IN_RESULTSET{13};
  static const int32_t ERR_STREAMING_TOP_N_NOT_SUPPORTED_IN_RENDER_QUERY{14};
  static const int32_t ERR_SINGLE_VALUE_FOUND_MULTIPLE_VALUES{15};
  static const int32_t ERR_GEOS{16};

  static std::mutex compilation_mutex_;
  static std::mutex kernel_mutex_;

  friend class BaselineJoinHashTable;
  friend class CodeGenerator;
  friend class ColumnFetcher;
  friend class OverlapsJoinHashTable;
  friend class GroupByAndAggregate;
  friend class QueryCompilationDescriptor;
  friend class QueryMemoryDescriptor;
  friend class QueryMemoryInitializer;
  friend class QueryFragmentDescriptor;
  friend class QueryExecutionContext;
  friend class ResultSet;
  friend class InValuesBitmap;
  friend class JoinHashTable;
  friend class LeafAggregator;
  friend class QueryRewriter;
  friend class PendingExecutionClosure;
  friend class RelAlgExecutor;
  friend class TableOptimizer;
  friend class TableFunctionCompilationContext;
  friend class TableFunctionExecutionContext;
  friend struct TargetExprCodegenBuilder;
  friend struct TargetExprCodegen;
  friend class WindowProjectNodeContext;
};

inline std::string get_null_check_suffix(const SQLTypeInfo& lhs_ti,
                                         const SQLTypeInfo& rhs_ti) {
  if (lhs_ti.get_notnull() && rhs_ti.get_notnull()) {
    return "";
  }
  std::string null_check_suffix{"_nullable"};
  if (lhs_ti.get_notnull()) {
    CHECK(!rhs_ti.get_notnull());
    null_check_suffix += "_rhs";
  } else if (rhs_ti.get_notnull()) {
    CHECK(!lhs_ti.get_notnull());
    null_check_suffix += "_lhs";
  }
  return null_check_suffix;
}

inline bool is_unnest(const Analyzer::Expr* expr) {
  return dynamic_cast<const Analyzer::UOper*>(expr) &&
         static_cast<const Analyzer::UOper*>(expr)->get_optype() == kUNNEST;
}

bool is_trivial_loop_join(const std::vector<InputTableInfo>& query_infos,
                          const RelAlgExecutionUnit& ra_exe_unit);

std::unordered_set<int> get_available_gpus(const Catalog_Namespace::Catalog& cat);

size_t get_context_count(const ExecutorDeviceType device_type,
                         const size_t cpu_count,
                         const size_t gpu_count);

extern "C" void register_buffer_with_executor_rsm(int64_t exec, int8_t* buffer);

const Analyzer::Expr* remove_cast_to_int(const Analyzer::Expr* expr);

#endif  // QUERYENGINE_EXECUTE_H