Executor Class Reference

#include <Execute.h>

Collaboration diagram for Executor:

Classes
class	FetchCacheAnchor
struct	GroupColLLVMValue
struct	JoinHashTableOrError

Public Types
using	ExecutorId = size_t
using	CachedCardinality = std::pair< bool, size_t >

Public Member Functions
	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)
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
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 ()
void	enableRuntimeQueryInterrupt (const unsigned interrupt_freq) const
int8_t	warpSize () const
unsigned	gridSize () const
unsigned	numBlocksPerMP () const
unsigned	blockSize () const
size_t	maxGpuSlabSize () const
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 ()
void	addToCardinalityCache (const std::string &cache_key, const size_t cache_value)
CachedCardinality	getCachedCardinality (const std::string &cache_key)
template<typename THREAD_POOL >
void	launchKernels (SharedKernelContext &shared_context, std::vector< std::unique_ptr< ExecutionKernel >> &&kernels)
template<>
void	setCurrentQuerySession (const std::string &query_session, mapd_unique_lock< mapd_shared_mutex > &write_lock)
template<>
std::string &	getCurrentQuerySession (mapd_shared_lock< mapd_shared_mutex > &read_lock)
template<>
bool	checkCurrentQuerySession (const std::string &candidate_query_session, mapd_shared_lock< mapd_shared_mutex > &read_lock)
template<>
void	invalidateQuerySession (mapd_unique_lock< mapd_shared_mutex > &write_lock)
template<>
bool	addToQuerySessionList (const std::string &query_session, mapd_unique_lock< mapd_shared_mutex > &write_lock)
template<>
bool	removeFromQuerySessionList (const std::string &query_session, mapd_unique_lock< mapd_shared_mutex > &write_lock)
template<>
void	setQuerySessionAsInterrupted (const std::string &query_session, mapd_unique_lock< mapd_shared_mutex > &write_lock)
template<>
bool	checkIsQuerySessionInterrupted (const std::string &query_session, mapd_shared_lock< mapd_shared_mutex > &read_lock)

Static Public Member Functions
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 ()
static void	clearMemory (const Data_Namespace::MemoryLevel memory_level)
static size_t	getArenaBlockSize ()
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 &)

Static Public Attributes
static const ExecutorId	UNITARY_EXECUTOR_ID = 0
static const size_t	high_scan_limit
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_

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

Private Member Functions
void	clearMetaInfoCache ()
int	deviceCount (const ExecutorDeviceType) const
int	deviceCountForMemoryLevel (const Data_Namespace::MemoryLevel memory_level) const
llvm::Value *	codegenWindowFunction (const size_t target_index, const CompilationOptions &co)
llvm::Value *	codegenWindowFunctionAggregate (const CompilationOptions &co)
llvm::BasicBlock *	codegenWindowResetStateControlFlow ()
void	codegenWindowFunctionStateInit (llvm::Value *aggregate_state)
llvm::Value *	codegenWindowFunctionAggregateCalls (llvm::Value *aggregate_state, const CompilationOptions &co)
void	codegenWindowAvgEpilogue (llvm::Value *crt_val, llvm::Value *window_func_null_val, llvm::Value *multiplicity_lv)
llvm::Value *	codegenAggregateWindowState ()
llvm::Value *	aggregateWindowStatePtr ()
bool	isArchPascalOrLater (const ExecutorDeviceType dt) const
bool	needFetchAllFragments (const InputColDescriptor &col_desc, const RelAlgExecutionUnit &ra_exe_unit, const FragmentsList &selected_fragments) const
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)
void	executeWorkUnitPerFragment (const RelAlgExecutionUnit &ra_exe_unit, const InputTableInfo &table_info, const CompilationOptions &co, const ExecutionOptions &eo, const Catalog_Namespace::Catalog &cat, PerFragmentCallBack &cb)
	Compiles and dispatches a work unit per fragment processing results with the per fragment callback. Currently used for computing metrics over fragments (metadata). More...
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)
	Compiles and dispatches a table function; that is, a function that takes as input one or more columns and returns a ResultSet, which can be parsed by subsequent execution steps. More...
ExecutorDeviceType	getDeviceTypeForTargets (const RelAlgExecutionUnit &ra_exe_unit, const ExecutorDeviceType requested_device_type)
ResultSetPtr	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)
ResultSetPtr	collectAllDeviceShardedTopResults (SharedKernelContext &shared_context, const RelAlgExecutionUnit &ra_exe_unit) const
std::unordered_map< int, const Analyzer::BinOper * >	getInnerTabIdToJoinCond () const
std::vector< std::unique_ptr < ExecutionKernel > >	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)
template<typename THREAD_POOL >
void	launchKernels (SharedKernelContext &shared_context, std::vector< std::unique_ptr< ExecutionKernel >> &&kernels)
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)
ResultSetPtr	resultsUnion (SharedKernelContext &shared_context, const RelAlgExecutionUnit &ra_exe_unit)
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< 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)
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)
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)
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)
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
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)
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)
llvm::Value *	spillDoubleElement (llvm::Value *elem_val, llvm::Type *elem_ty)

Static Private Member Functions
static size_t	align (const size_t off_in, const size_t alignment)

Private Attributes
std::unique_ptr< CgenState >	cgen_state_
std::unique_ptr< PlanState >	plan_state_
std::shared_ptr < RowSetMemoryOwner >	row_set_mem_owner_
std::mutex	gpu_exec_mutex_ [max_gpu_count]
std::shared_ptr < StringDictionaryProxy >	lit_str_dict_proxy_
std::mutex	str_dict_mutex_
std::unique_ptr < llvm::TargetMachine >	nvptx_target_machine_
CodeCache	cpu_code_cache_
CodeCache	gpu_code_cache_
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
std::unique_ptr < WindowProjectNodeContext >	window_project_node_context_owned_
WindowFunctionContext *	active_window_function_ {nullptr}
InputTableInfoCache	input_table_info_cache_
AggregatedColRange	agg_col_range_cache_
StringDictionaryGenerations	string_dictionary_generations_
TableGenerations	table_generations_

Static Private Attributes
static const int	max_gpu_count {16}
static std::mutex	gpu_active_modules_mutex_
static uint32_t	gpu_active_modules_device_mask_ {0x0}
static void *	gpu_active_modules_ [max_gpu_count]
static std::atomic< bool >	interrupted_ {false}
static const size_t	baseline_threshold
static const size_t	code_cache_size {1000}
static mapd_shared_mutex	executor_session_mutex_
static std::string	current_query_session_ {""}
static InterruptFlagMap	queries_interrupt_flag_
static std::map< int, std::shared_ptr< Executor > >	executors_
static std::atomic_flag	execute_spin_lock_ = ATOMIC_FLAG_INIT
static mapd_shared_mutex	execute_mutex_
static mapd_shared_mutex	executors_cache_mutex_
static mapd_shared_mutex	recycler_mutex_
static std::unordered_map < std::string, size_t >	cardinality_cache_

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

Detailed Description

Definition at line 295 of file Execute.h.

Member Typedef Documentation

using Executor::CachedCardinality = std::pair<bool, size_t>

Definition at line 835 of file Execute.h.

using Executor::ExecutorId = size_t

Definition at line 302 of file Execute.h.

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

private

Definition at line 454 of file Execute.h.

Constructor & Destructor Documentation

Executor::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
	)

Definition at line 129 of file Execute.cpp.

    : cgen_state_(new CgenState({}, false))

Member Function Documentation

void Executor::addCodeToCache ( const CodeCacheKey &  key, std::shared_ptr< CompilationContext >  compilation_context, llvm::Module *  module, CodeCache &  cache  )

static

Definition at line 219 of file NativeCodegen.cpp.

References LruCache< key_t, value_t, hash_t >::put().

Referenced by StubGenerator::generateStub().

                                                {
  cache.put(key,
            std::make_pair<std::shared_ptr<CompilationContext>, decltype(module)>(
                std::move(compilation_context), std::move(module)));
}

Here is the call graph for this function:

Here is the caller graph for this function:

RelAlgExecutionUnit Executor::addDeletedColumn ( const RelAlgExecutionUnit &  ra_exe_unit, const CompilationOptions &  co  )

private

Definition at line 3198 of file Execute.cpp.

References catalog_(), CHECK(), CompilationOptions::filter_on_deleted_column, and TABLE.

                                                                             {
  if (!co.filter_on_deleted_column) {
    return ra_exe_unit;
  }
  auto ra_exe_unit_with_deleted = ra_exe_unit;
  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()));
    }
  }
  return ra_exe_unit_with_deleted;
}

Here is the call graph for this function:

llvm::Value * Executor::addJoinLoopIterator ( const std::vector< llvm::Value * > &  prev_iters, const size_t  level_idx  )

private

Definition at line 530 of file IRCodegen.cpp.

References CodeGenerator::cgen_state_, CHECK(), and CgenState::scan_idx_to_hash_pos_.

                                                                   {
  // Iterators are added for loop-outer joins when the head of the loop is generated,
  // then once again when the body if generated. Allow this instead of special handling
  // of call sites.
  const auto it = cgen_state_->scan_idx_to_hash_pos_.find(level_idx);
  if (it != cgen_state_->scan_idx_to_hash_pos_.end()) {
    return it->second;
  }
  CHECK(!prev_iters.empty());
  llvm::Value* matching_row_index = prev_iters.back();
  const auto it_ok =
      cgen_state_->scan_idx_to_hash_pos_.emplace(level_idx, matching_row_index);
  CHECK(it_ok.second);
  return matching_row_index;
}

Here is the call graph for this function:

void Executor::addToCardinalityCache	(	const std::string &	cache_key,
		const size_t	cache_value
	)

Definition at line 3583 of file Execute.cpp.

References g_use_estimator_result_cache, and VLOG.

                                                               {
  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";
  }
}

template<typename SESSION_MAP_LOCK >

bool Executor::addToQuerySessionList	(	const std::string &	query_session,
		SESSION_MAP_LOCK &	write_lock
	)

template<>

bool Executor::addToQuerySessionList	(	const std::string &	query_session,
		mapd_unique_lock< mapd_shared_mutex > &	write_lock
	)

Definition at line 3547 of file Execute.cpp.

                                                                                      {
  return queries_interrupt_flag_.emplace(query_session, false).second;
}

llvm::Value * Executor::aggregateWindowStatePtr ( )

private

Definition at line 123 of file WindowFunctionIR.cpp.

References anonymous_namespace{WindowFunctionIR.cpp}::get_adjusted_window_type_info(), get_int_type(), WindowProjectNodeContext::getActiveWindowFunctionContext(), and kFLOAT.

                                             {
  const auto window_func_context =
      WindowProjectNodeContext::getActiveWindowFunctionContext(this);
  const auto window_func = window_func_context->getWindowFunction();
  const auto arg_ti = get_adjusted_window_type_info(window_func);
  llvm::Type* aggregate_state_type =
      arg_ti.get_type() == kFLOAT
          ? llvm::PointerType::get(get_int_type(32, cgen_state_->context_), 0)
          : llvm::PointerType::get(get_int_type(64, cgen_state_->context_), 0);
  const auto aggregate_state_i64 = cgen_state_->llInt(
      reinterpret_cast<const int64_t>(window_func_context->aggregateState()));
  return cgen_state_->ir_builder_.CreateIntToPtr(aggregate_state_i64,
                                                 aggregate_state_type);
}

Here is the call graph for this function:

static size_t Executor::align ( const size_t  off_in, const size_t  alignment  )

inlinestaticprivate

Definition at line 847 of file Execute.h.

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

unsigned Executor::blockSize

(

)

const

Definition at line 3128 of file Execute.cpp.

References block_size_x_(), catalog_(), and CHECK().

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

Here is the call graph for this function:

std::shared_ptr< JoinHashTableInterface > Executor::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  )

private

Definition at line 484 of file IRCodegen.cpp.

References anonymous_namespace{IRCodegen.cpp}::add_qualifier_to_execution_unit(), Data_Namespace::CPU_LEVEL, CompilationOptions::device_type, Executor::JoinHashTableOrError::fail_reason, GPU, Data_Namespace::GPU_LEVEL, Executor::JoinHashTableOrError::hash_table, INNER, IS_EQUIVALENCE, PlanState::join_info_, JoinHashTableInterface::OneToOne, CodeGenerator::plan_state_, JoinCondition::quals, and JoinCondition::type.

                                        {
  if (current_level_join_conditions.type != JoinType::INNER &&
      current_level_join_conditions.quals.size() > 1) {
    fail_reasons.emplace_back("No equijoin expression found for outer join");
    return nullptr;
  }
  std::shared_ptr<JoinHashTableInterface> current_level_hash_table;
  for (const auto& join_qual : current_level_join_conditions.quals) {
    auto qual_bin_oper = std::dynamic_pointer_cast<Analyzer::BinOper>(join_qual);
    if (!qual_bin_oper || !IS_EQUIVALENCE(qual_bin_oper->get_optype())) {
      fail_reasons.emplace_back("No equijoin expression found");
      if (current_level_join_conditions.type == JoinType::INNER) {
        add_qualifier_to_execution_unit(ra_exe_unit, join_qual);
      }
      continue;
    }
    JoinHashTableOrError hash_table_or_error;
    if (!current_level_hash_table) {
      hash_table_or_error = buildHashTableForQualifier(
          qual_bin_oper,
          query_infos,
          co.device_type == ExecutorDeviceType::GPU ? MemoryLevel::GPU_LEVEL
                                                    : MemoryLevel::CPU_LEVEL,
          JoinHashTableInterface::HashType::OneToOne,
          column_cache);
      current_level_hash_table = hash_table_or_error.hash_table;
    }
    if (hash_table_or_error.hash_table) {
      plan_state_->join_info_.join_hash_tables_.push_back(hash_table_or_error.hash_table);
      plan_state_->join_info_.equi_join_tautologies_.push_back(qual_bin_oper);
    } else {
      fail_reasons.push_back(hash_table_or_error.fail_reason);
      if (current_level_join_conditions.type == JoinType::INNER) {
        add_qualifier_to_execution_unit(ra_exe_unit, qual_bin_oper);
      }
    }
  }
  return current_level_hash_table;
}

Here is the call graph for this function:

Executor::JoinHashTableOrError Executor::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  )

private

Definition at line 3074 of file Execute.cpp.

References g_enable_dynamic_watchdog, g_enable_overlaps_hashjoin, g_enable_runtime_query_interrupt, and JoinHashTableInterface::getInstance().

                                  {
  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 =
        JoinHashTableInterface::getInstance(qual_bin_oper,
                                            query_infos,
                                            memory_level,
                                            preferred_hash_type,
                                            deviceCountForMemoryLevel(memory_level),
                                            column_cache,
                                            this);
    return {tbl, ""};
  } catch (const HashJoinFail& e) {
    return {nullptr, e.what()};
  }
}

Here is the call graph for this function:

std::function< llvm::Value *(const std::vector< llvm::Value * > &, llvm::Value *)> Executor::buildIsDeletedCb ( const RelAlgExecutionUnit &  ra_exe_unit, const size_t  level_idx, const CompilationOptions &  co  )

private

Definition at line 425 of file IRCodegen.cpp.

References catalog_(), CodeGenerator::cgen_state_, CHECK(), CHECK_LT, CodeGenerator::codegen(), CgenState::context_, CompilationOptions::filter_on_deleted_column, RelAlgExecutionUnit::input_descs, CgenState::ir_builder_, CgenState::llBool(), CgenState::llInt(), CgenState::row_func_, TABLE, and CodeGenerator::toBool().

                                                         {
  if (!co.filter_on_deleted_column) {
    return nullptr;
  }
  CHECK_LT(level_idx + 1, ra_exe_unit.input_descs.size());
  const auto input_desc = ra_exe_unit.input_descs[level_idx + 1];
  if (input_desc.getSourceType() != InputSourceType::TABLE) {
    return nullptr;
  }
  const auto td = catalog_->getMetadataForTable(input_desc.getTableId());
  CHECK(td);
  const auto deleted_cd = catalog_->getDeletedColumnIfRowsDeleted(td);
  if (!deleted_cd) {
    return nullptr;
  }
  CHECK(deleted_cd->columnType.is_boolean());
  const auto deleted_expr = makeExpr<Analyzer::ColumnVar>(deleted_cd->columnType,
                                                          input_desc.getTableId(),
                                                          deleted_cd->columnId,
                                                          input_desc.getNestLevel());
  return [this, deleted_expr, level_idx, &co](const std::vector<llvm::Value*>& prev_iters,
                                              llvm::Value* have_more_inner_rows) {
    const auto matching_row_index = addJoinLoopIterator(prev_iters, level_idx + 1);
    // Avoid fetching the deleted column from a position which is not valid.
    // An invalid position can be returned by a one to one hash lookup (negative)
    // or at the end of iteration over a set of matching values.
    llvm::Value* is_valid_it{nullptr};
    if (have_more_inner_rows) {
      is_valid_it = have_more_inner_rows;
    } else {
      is_valid_it = cgen_state_->ir_builder_.CreateICmp(
          llvm::ICmpInst::ICMP_SGE, matching_row_index, cgen_state_->llInt<int64_t>(0));
    }
    const auto it_valid_bb = llvm::BasicBlock::Create(
        cgen_state_->context_, "it_valid", cgen_state_->row_func_);
    const auto it_not_valid_bb = llvm::BasicBlock::Create(
        cgen_state_->context_, "it_not_valid", cgen_state_->row_func_);
    cgen_state_->ir_builder_.CreateCondBr(is_valid_it, it_valid_bb, it_not_valid_bb);
    const auto row_is_deleted_bb = llvm::BasicBlock::Create(
        cgen_state_->context_, "row_is_deleted", cgen_state_->row_func_);
    cgen_state_->ir_builder_.SetInsertPoint(it_valid_bb);
    CodeGenerator code_generator(this);
    const auto row_is_deleted = code_generator.toBool(
        code_generator.codegen(deleted_expr.get(), true, co).front());
    cgen_state_->ir_builder_.CreateBr(row_is_deleted_bb);
    cgen_state_->ir_builder_.SetInsertPoint(it_not_valid_bb);
    const auto row_is_deleted_default = cgen_state_->llBool(false);
    cgen_state_->ir_builder_.CreateBr(row_is_deleted_bb);
    cgen_state_->ir_builder_.SetInsertPoint(row_is_deleted_bb);
    auto row_is_deleted_or_default =
        cgen_state_->ir_builder_.CreatePHI(row_is_deleted->getType(), 2);
    row_is_deleted_or_default->addIncoming(row_is_deleted, it_valid_bb);
    row_is_deleted_or_default->addIncoming(row_is_deleted_default, it_not_valid_bb);
    return row_is_deleted_or_default;
  };
}

Here is the call graph for this function:

std::vector< JoinLoop > Executor::buildJoinLoops ( RelAlgExecutionUnit &  ra_exe_unit, const CompilationOptions &  co, const ExecutionOptions &  eo, const std::vector< InputTableInfo > &  query_infos, ColumnCacheMap &  column_cache  )

private

Definition at line 256 of file IRCodegen.cpp.

References CodeGenerator::cgen_state_, CHECK(), CHECK_LT, CodeGenerator::codegen(), INJECT_TIMER, CgenState::ir_builder_, RelAlgExecutionUnit::join_quals, LEFT, CgenState::llBool(), JoinHashTableInterface::OneToOne, CgenState::outer_join_match_found_per_level_, Set, Singleton, JoinLoopDomain::slot_lookup_result, CodeGenerator::toBool(), JoinCondition::type, and JoinLoopDomain::values_buffer.

                                  {
  INJECT_TIMER(buildJoinLoops);
  std::vector<JoinLoop> join_loops;
  for (size_t level_idx = 0, current_hash_table_idx = 0;
       level_idx < ra_exe_unit.join_quals.size();
       ++level_idx) {
    const auto& current_level_join_conditions = ra_exe_unit.join_quals[level_idx];
    std::vector<std::string> fail_reasons;
    const auto build_cur_level_hash_table = [&]() {
      if (current_level_join_conditions.quals.size() > 1) {
        const auto first_qual = *current_level_join_conditions.quals.begin();
        auto qual_bin_oper =
            std::dynamic_pointer_cast<const Analyzer::BinOper>(first_qual);
        if (qual_bin_oper && qual_bin_oper->is_overlaps_oper() &&
            current_level_join_conditions.type == JoinType::LEFT) {
          JoinCondition join_condition{{first_qual}, current_level_join_conditions.type};

          return buildCurrentLevelHashTable(
              join_condition, ra_exe_unit, co, query_infos, column_cache, fail_reasons);
        }
      }
      return buildCurrentLevelHashTable(current_level_join_conditions,
                                        ra_exe_unit,
                                        co,
                                        query_infos,
                                        column_cache,
                                        fail_reasons);
    };
    const auto current_level_hash_table = build_cur_level_hash_table();
    const auto found_outer_join_matches_cb =
        [this, level_idx](llvm::Value* found_outer_join_matches) {
          CHECK_LT(level_idx, cgen_state_->outer_join_match_found_per_level_.size());
          CHECK(!cgen_state_->outer_join_match_found_per_level_[level_idx]);
          cgen_state_->outer_join_match_found_per_level_[level_idx] =
              found_outer_join_matches;
        };
    const auto is_deleted_cb = buildIsDeletedCb(ra_exe_unit, level_idx, co);
    const auto outer_join_condition_multi_quals_cb =
        [this, level_idx, &co, &current_level_join_conditions](
            const std::vector<llvm::Value*>& prev_iters) {
          // The values generated for the match path don't dominate all uses
          // since on the non-match path nulls are generated. Reset the cache
          // once the condition is generated to avoid incorrect reuse.
          FetchCacheAnchor anchor(cgen_state_.get());
          addJoinLoopIterator(prev_iters, level_idx + 1);
          llvm::Value* left_join_cond = cgen_state_->llBool(true);
          CodeGenerator code_generator(this);
          // Do not want to look at all quals! only 1..N quals (ignore first qual)
          // Note(jclay): this may need to support cases larger than 2
          // are there any?
          if (current_level_join_conditions.quals.size() >= 2) {
            auto qual_it = std::next(current_level_join_conditions.quals.begin(), 1);
            for (; qual_it != current_level_join_conditions.quals.end();
                 std::advance(qual_it, 1)) {
              left_join_cond = cgen_state_->ir_builder_.CreateAnd(
                  left_join_cond,
                  code_generator.toBool(
                      code_generator.codegen(qual_it->get(), true, co).front()));
            }
          }
          return left_join_cond;
        };
    if (current_level_hash_table) {
      if (current_level_hash_table->getHashType() == JoinHashTable::HashType::OneToOne) {
        join_loops.emplace_back(
            JoinLoopKind::Singleton,
            current_level_join_conditions.type,
            [this, current_hash_table_idx, level_idx, current_level_hash_table, &co](
                const std::vector<llvm::Value*>& prev_iters) {
              addJoinLoopIterator(prev_iters, level_idx);
              JoinLoopDomain domain{{0}};
              domain.slot_lookup_result =
                  current_level_hash_table->codegenSlot(co, current_hash_table_idx);
              return domain;
            },
            nullptr,
            current_level_join_conditions.type == JoinType::LEFT
                ? std::function<void(llvm::Value*)>(found_outer_join_matches_cb)
                : nullptr,
            is_deleted_cb);
      } else {
        join_loops.emplace_back(
            /*kind=*/JoinLoopKind::Set,
            /*type=*/current_level_join_conditions.type,
            /*iteration_domain_codegen=*/
            [this, current_hash_table_idx, level_idx, current_level_hash_table, &co](
                const std::vector<llvm::Value*>& prev_iters) {
              addJoinLoopIterator(prev_iters, level_idx);
              JoinLoopDomain domain{{0}};
              const auto matching_set = current_level_hash_table->codegenMatchingSet(
                  co, current_hash_table_idx);
              domain.values_buffer = matching_set.elements;
              domain.element_count = matching_set.count;
              return domain;
            },
            /*outer_condition_match=*/
            current_level_join_conditions.type == JoinType::LEFT
                ? std::function<llvm::Value*(const std::vector<llvm::Value*>&)>(
                      outer_join_condition_multi_quals_cb)
                : nullptr,
            /*found_outer_matches=*/current_level_join_conditions.type == JoinType::LEFT
                ? std::function<void(llvm::Value*)>(found_outer_join_matches_cb)
                : nullptr,
            /*is_deleted=*/is_deleted_cb);
      }
      ++current_hash_table_idx;
    } else {
      const auto fail_reasons_str = current_level_join_conditions.quals.empty()
                                        ? "No equijoin expression found"
                                        : boost::algorithm::join(fail_reasons, " | ");
      check_if_loop_join_is_allowed(
          ra_exe_unit, eo, query_infos, level_idx, fail_reasons_str);
      // Callback provided to the `JoinLoop` framework to evaluate the (outer) join
      // condition.
      VLOG(1) << "Unable to build hash table, falling back to loop join: "
              << fail_reasons_str;
      const auto outer_join_condition_cb =
          [this, level_idx, &co, &current_level_join_conditions](
              const std::vector<llvm::Value*>& prev_iters) {
            // The values generated for the match path don't dominate all uses
            // since on the non-match path nulls are generated. Reset the cache
            // once the condition is generated to avoid incorrect reuse.
            FetchCacheAnchor anchor(cgen_state_.get());
            addJoinLoopIterator(prev_iters, level_idx + 1);
            llvm::Value* left_join_cond = cgen_state_->llBool(true);
            CodeGenerator code_generator(this);
            for (auto expr : current_level_join_conditions.quals) {
              left_join_cond = cgen_state_->ir_builder_.CreateAnd(
                  left_join_cond,
                  code_generator.toBool(
                      code_generator.codegen(expr.get(), true, co).front()));
            }
            return left_join_cond;
          };
      join_loops.emplace_back(
          /*kind=*/JoinLoopKind::UpperBound,
          /*type=*/current_level_join_conditions.type,
          /*iteration_domain_codegen=*/
          [this, level_idx](const std::vector<llvm::Value*>& prev_iters) {
            addJoinLoopIterator(prev_iters, level_idx);
            JoinLoopDomain domain{{0}};
            const auto rows_per_scan_ptr = cgen_state_->ir_builder_.CreateGEP(
                get_arg_by_name(cgen_state_->row_func_, "num_rows_per_scan"),
                cgen_state_->llInt(int32_t(level_idx + 1)));
            domain.upper_bound = cgen_state_->ir_builder_.CreateLoad(rows_per_scan_ptr,
                                                                     "num_rows_per_scan");
            return domain;
          },
          /*outer_condition_match=*/
          current_level_join_conditions.type == JoinType::LEFT
              ? std::function<llvm::Value*(const std::vector<llvm::Value*>&)>(
                    outer_join_condition_cb)
              : nullptr,
          /*found_outer_matches=*/
          current_level_join_conditions.type == JoinType::LEFT
              ? std::function<void(llvm::Value*)>(found_outer_join_matches_cb)
              : nullptr,
          /*is_deleted=*/is_deleted_cb);
    }
  }
  return join_loops;
}

Here is the call graph for this function:

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  )

private

Definition at line 2573 of file Execute.cpp.

References CHECK(), CHECK_EQ, CHECK_LT, and RelAlgExecutionUnit::input_descs.

                                            {
  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;
  }
}

Here is the call graph for this function:

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  )

private

Definition at line 2604 of file Execute.cpp.

References CHECK(), CHECK_LT, and RelAlgExecutionUnit::input_descs.

                                            {
  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;
  }
}

Here is the call graph for this function:

llvm::Value * Executor::castToFP ( llvm::Value *  val )

private

Definition at line 3148 of file Execute.cpp.

References logger::FATAL, LOG, and to_string().

                                            {
  if (!val->getType()->isIntegerTy()) {
    return val;
  }

  auto val_width = static_cast<llvm::IntegerType*>(val->getType())->getBitWidth();
  llvm::Type* dest_ty{nullptr};
  switch (val_width) {
    case 32:
      dest_ty = llvm::Type::getFloatTy(cgen_state_->context_);
      break;
    case 64:
      dest_ty = llvm::Type::getDoubleTy(cgen_state_->context_);
      break;
    default:
      LOG(FATAL) << "Unsupported FP width: " << std::to_string(val_width);
  }
  return cgen_state_->ir_builder_.CreateSIToFP(val, dest_ty);
}

Here is the call graph for this function:

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

private

Definition at line 3168 of file Execute.cpp.

References CHECK(), CHECK_LT, and get_int_type().

                                                                           {
  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));
}

Here is the call graph for this function:

template<typename SESSION_MAP_LOCK >

bool Executor::checkCurrentQuerySession	(	const std::string &	candidate_query_session,
		SESSION_MAP_LOCK &	read_lock
	)

template<>

bool Executor::checkCurrentQuerySession	(	const std::string &	candidate_query_session,
		mapd_shared_lock< mapd_shared_mutex > &	read_lock
	)

Definition at line 3534 of file Execute.cpp.

                                                                                        {
  // if current_query_session is equal to the candidate_query_session,
  // or it is empty session we consider
  return (current_query_session_ == candidate_query_session);
}

template<typename SESSION_MAP_LOCK >

bool Executor::checkIsQuerySessionInterrupted	(	const std::string &	query_session,
		SESSION_MAP_LOCK &	read_lock
	)

template<>

bool Executor::checkIsQuerySessionInterrupted	(	const std::string &	query_session,
		mapd_shared_lock< mapd_shared_mutex > &	read_lock
	)

Definition at line 3567 of file Execute.cpp.

                                                    {
  auto flag_it = queries_interrupt_flag_.find(query_session);
  return flag_it != queries_interrupt_flag_.end() && flag_it->second;
}

void Executor::clearMemory ( const Data_Namespace::MemoryLevel  memory_level )

static

Definition at line 170 of file Execute.cpp.

References Data_Namespace::DataMgr::clearMemory(), Data_Namespace::CPU_LEVEL, Catalog_Namespace::SysCatalog::getDataMgr(), Data_Namespace::GPU_LEVEL, Catalog_Namespace::SysCatalog::instance(), and CacheInvalidator< CACHE_HOLDING_TYPES >::invalidateCaches().

Referenced by DBHandler::clear_cpu_memory(), DBHandler::clear_gpu_memory(), QueryRunner::QueryRunner::clearCpuMemory(), and QueryRunner::QueryRunner::clearGpuMemory().

                                                                       {
  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.");
    }
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

void Executor::clearMetaInfoCache ( )

private

Definition at line 346 of file Execute.cpp.

References input_table_info_cache_().

                                  {
  input_table_info_cache_.clear();
  agg_col_range_cache_.clear();
  string_dictionary_generations_.clear();
  table_generations_.clear();
}

Here is the call graph for this function:

llvm::Value * Executor::codegenAggregateWindowState ( )

private

Definition at line 319 of file WindowFunctionIR.cpp.

References AVG, COUNT, anonymous_namespace{WindowFunctionIR.cpp}::get_adjusted_window_type_info(), get_int_type(), WindowProjectNodeContext::getActiveWindowFunctionContext(), Analyzer::WindowFunction::getKind(), kDECIMAL, kDOUBLE, and kFLOAT.

                                                 {
  const auto pi32_type =
      llvm::PointerType::get(get_int_type(32, cgen_state_->context_), 0);
  const auto pi64_type =
      llvm::PointerType::get(get_int_type(64, cgen_state_->context_), 0);
  const auto window_func_context =
      WindowProjectNodeContext::getActiveWindowFunctionContext(this);
  const Analyzer::WindowFunction* window_func = window_func_context->getWindowFunction();
  const auto window_func_ti = get_adjusted_window_type_info(window_func);
  const auto aggregate_state_type =
      window_func_ti.get_type() == kFLOAT ? pi32_type : pi64_type;
  auto aggregate_state = aggregateWindowStatePtr();
  if (window_func->getKind() == SqlWindowFunctionKind::AVG) {
    const auto aggregate_state_count_i64 = cgen_state_->llInt(
        reinterpret_cast<const int64_t>(window_func_context->aggregateStateCount()));
    auto aggregate_state_count = cgen_state_->ir_builder_.CreateIntToPtr(
        aggregate_state_count_i64, aggregate_state_type);
    const auto double_null_lv = cgen_state_->inlineFpNull(SQLTypeInfo(kDOUBLE));
    switch (window_func_ti.get_type()) {
      case kFLOAT: {
        return cgen_state_->emitCall(
            "load_avg_float", {aggregate_state, aggregate_state_count, double_null_lv});
      }
      case kDOUBLE: {
        return cgen_state_->emitCall(
            "load_avg_double", {aggregate_state, aggregate_state_count, double_null_lv});
      }
      case kDECIMAL: {
        return cgen_state_->emitCall(
            "load_avg_decimal",
            {aggregate_state,
             aggregate_state_count,
             double_null_lv,
             cgen_state_->llInt<int32_t>(window_func_ti.get_scale())});
      }
      default: {
        return cgen_state_->emitCall(
            "load_avg_int", {aggregate_state, aggregate_state_count, double_null_lv});
      }
    }
  }
  if (window_func->getKind() == SqlWindowFunctionKind::COUNT) {
    return cgen_state_->ir_builder_.CreateLoad(aggregate_state);
  }
  switch (window_func_ti.get_type()) {
    case kFLOAT: {
      return cgen_state_->emitCall("load_float", {aggregate_state});
    }
    case kDOUBLE: {
      return cgen_state_->emitCall("load_double", {aggregate_state});
    }
    default: {
      return cgen_state_->ir_builder_.CreateLoad(aggregate_state);
    }
  }
}

Here is the call graph for this function:

void Executor::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  )

private

Definition at line 547 of file IRCodegen.cpp.

References ExecutionOptions::allow_runtime_query_interrupt, CodeGenerator::cgen_state_, JoinLoop::codegen(), CgenState::context_, CompilationOptions::device_type, CgenState::ir_builder_, CgenState::llInt(), CgenState::needs_error_check_, CodeGenerator::posArg(), CgenState::row_func_, and ExecutionOptions::with_dynamic_watchdog.

                                                            {
  const auto exit_bb =
      llvm::BasicBlock::Create(cgen_state_->context_, "exit", cgen_state_->row_func_);
  cgen_state_->ir_builder_.SetInsertPoint(exit_bb);
  cgen_state_->ir_builder_.CreateRet(cgen_state_->llInt<int32_t>(0));
  cgen_state_->ir_builder_.SetInsertPoint(entry_bb);
  CodeGenerator code_generator(this);
  const auto loops_entry_bb = JoinLoop::codegen(
      join_loops,
      [this,
       query_func,
       &query_mem_desc,
       &co,
       &eo,
       &group_by_and_aggregate,
       &join_loops,
       &ra_exe_unit](const std::vector<llvm::Value*>& prev_iters) {
        addJoinLoopIterator(prev_iters, join_loops.size());
        auto& builder = cgen_state_->ir_builder_;
        const auto loop_body_bb = llvm::BasicBlock::Create(
            builder.getContext(), "loop_body", builder.GetInsertBlock()->getParent());
        builder.SetInsertPoint(loop_body_bb);
        const bool can_return_error =
            compileBody(ra_exe_unit, group_by_and_aggregate, query_mem_desc, co);
        if (can_return_error || cgen_state_->needs_error_check_ ||
            eo.with_dynamic_watchdog || eo.allow_runtime_query_interrupt) {
          createErrorCheckControlFlow(query_func,
                                      eo.with_dynamic_watchdog,
                                      eo.allow_runtime_query_interrupt,
                                      co.device_type);
        }
        return loop_body_bb;
      },
      code_generator.posArg(nullptr),
      exit_bb,
      cgen_state_->ir_builder_);
  cgen_state_->ir_builder_.SetInsertPoint(entry_bb);
  cgen_state_->ir_builder_.CreateBr(loops_entry_bb);
}

Here is the call graph for this function:

llvm::BasicBlock * Executor::codegenSkipDeletedOuterTableRow ( const RelAlgExecutionUnit &  ra_exe_unit, const CompilationOptions &  co  )

private

Definition at line 2170 of file NativeCodegen.cpp.

                                  {
  if (!co.filter_on_deleted_column) {
    return nullptr;
  }
  CHECK(!ra_exe_unit.input_descs.empty());
  const auto& outer_input_desc = ra_exe_unit.input_descs[0];
  if (outer_input_desc.getSourceType() != InputSourceType::TABLE) {
    return nullptr;
  }
  const auto td = catalog_->getMetadataForTable(outer_input_desc.getTableId());
  CHECK(td);
  const auto deleted_cd = catalog_->getDeletedColumnIfRowsDeleted(td);
  if (!deleted_cd) {
    return nullptr;
  }
  CHECK(deleted_cd->columnType.is_boolean());
  const auto deleted_expr =
      makeExpr<Analyzer::ColumnVar>(deleted_cd->columnType,
                                    outer_input_desc.getTableId(),
                                    deleted_cd->columnId,
                                    outer_input_desc.getNestLevel());
  CodeGenerator code_generator(this);
  const auto is_deleted =
      code_generator.toBool(code_generator.codegen(deleted_expr.get(), true, co).front());
  const auto is_deleted_bb = llvm::BasicBlock::Create(
      cgen_state_->context_, "is_deleted", cgen_state_->row_func_);
  llvm::BasicBlock* bb = llvm::BasicBlock::Create(
      cgen_state_->context_, "is_not_deleted", cgen_state_->row_func_);
  cgen_state_->ir_builder_.CreateCondBr(is_deleted, is_deleted_bb, bb);
  cgen_state_->ir_builder_.SetInsertPoint(is_deleted_bb);
  cgen_state_->ir_builder_.CreateRet(cgen_state_->llInt<int32_t>(0));
  cgen_state_->ir_builder_.SetInsertPoint(bb);
  return bb;
}

void Executor::codegenWindowAvgEpilogue ( llvm::Value *  crt_val, llvm::Value *  window_func_null_val, llvm::Value *  multiplicity_lv  )

private

Definition at line 283 of file WindowFunctionIR.cpp.

References anonymous_namespace{WindowFunctionIR.cpp}::get_adjusted_window_type_info(), get_int_type(), WindowProjectNodeContext::getActiveWindowFunctionContext(), kDOUBLE, and kFLOAT.

                                                                    {
  const auto window_func_context =
      WindowProjectNodeContext::getActiveWindowFunctionContext(this);
  const auto window_func = window_func_context->getWindowFunction();
  const auto window_func_ti = get_adjusted_window_type_info(window_func);
  const auto pi32_type =
      llvm::PointerType::get(get_int_type(32, cgen_state_->context_), 0);
  const auto pi64_type =
      llvm::PointerType::get(get_int_type(64, cgen_state_->context_), 0);
  const auto aggregate_state_type =
      window_func_ti.get_type() == kFLOAT ? pi32_type : pi64_type;
  const auto aggregate_state_count_i64 = cgen_state_->llInt(
      reinterpret_cast<const int64_t>(window_func_context->aggregateStateCount()));
  auto aggregate_state_count = cgen_state_->ir_builder_.CreateIntToPtr(
      aggregate_state_count_i64, aggregate_state_type);
  std::string agg_count_func_name = "agg_count";
  switch (window_func_ti.get_type()) {
    case kFLOAT: {
      agg_count_func_name += "_float";
      break;
    }
    case kDOUBLE: {
      agg_count_func_name += "_double";
      break;
    }
    default: {
      break;
    }
  }
  agg_count_func_name += "_skip_val";
  cgen_state_->emitCall(agg_count_func_name,
                        {aggregate_state_count, crt_val, window_func_null_val});
}

Here is the call graph for this function:

llvm::Value * Executor::codegenWindowFunction ( const size_t  target_index, const CompilationOptions &  co  )

private

Definition at line 21 of file WindowFunctionIR.cpp.

References WindowProjectNodeContext::activateWindowFunctionContext(), run_benchmark_import::args, AVG, CHECK(), CHECK_EQ, CodeGenerator::codegen(), COUNT, CUME_DIST, DENSE_RANK, logger::FATAL, FIRST_VALUE, WindowProjectNodeContext::get(), WindowFunctionContext::getWindowFunction(), LAG, LAST_VALUE, LEAD, LOG, MAX, MIN, NTILE, PERCENT_RANK, CodeGenerator::posArg(), RANK, ROW_NUMBER, and SUM.

                                                                           {
  CodeGenerator code_generator(this);
  const auto window_func_context =
      WindowProjectNodeContext::get(this)->activateWindowFunctionContext(this,
                                                                         target_index);
  const auto window_func = window_func_context->getWindowFunction();
  switch (window_func->getKind()) {
    case SqlWindowFunctionKind::ROW_NUMBER:
    case SqlWindowFunctionKind::RANK:
    case SqlWindowFunctionKind::DENSE_RANK:
    case SqlWindowFunctionKind::NTILE: {
      return cgen_state_->emitCall("row_number_window_func",
                                   {cgen_state_->llInt(reinterpret_cast<const int64_t>(
                                        window_func_context->output())),
                                    code_generator.posArg(nullptr)});
    }
    case SqlWindowFunctionKind::PERCENT_RANK:
    case SqlWindowFunctionKind::CUME_DIST: {
      return cgen_state_->emitCall("percent_window_func",
                                   {cgen_state_->llInt(reinterpret_cast<const int64_t>(
                                        window_func_context->output())),
                                    code_generator.posArg(nullptr)});
    }
    case SqlWindowFunctionKind::LAG:
    case SqlWindowFunctionKind::LEAD:
    case SqlWindowFunctionKind::FIRST_VALUE:
    case SqlWindowFunctionKind::LAST_VALUE: {
      CHECK(WindowProjectNodeContext::get(this));
      const auto& args = window_func->getArgs();
      CHECK(!args.empty());
      const auto arg_lvs = code_generator.codegen(args.front().get(), true, co);
      CHECK_EQ(arg_lvs.size(), size_t(1));
      return arg_lvs.front();
    }
    case SqlWindowFunctionKind::AVG:
    case SqlWindowFunctionKind::MIN:
    case SqlWindowFunctionKind::MAX:
    case SqlWindowFunctionKind::SUM:
    case SqlWindowFunctionKind::COUNT: {
      return codegenWindowFunctionAggregate(co);
    }
    default: {
      LOG(FATAL) << "Invalid window function kind";
    }
  }
  return nullptr;
}

Here is the call graph for this function:

llvm::Value * Executor::codegenWindowFunctionAggregate ( const CompilationOptions &  co )

private

Definition at line 138 of file WindowFunctionIR.cpp.

References AVG, CHECK(), WindowProjectNodeContext::get(), get_int_type(), and WindowProjectNodeContext::getActiveWindowFunctionContext().

                                                                                {
  const auto reset_state_false_bb = codegenWindowResetStateControlFlow();
  auto aggregate_state = aggregateWindowStatePtr();
  llvm::Value* aggregate_state_count = nullptr;
  const auto window_func_context =
      WindowProjectNodeContext::getActiveWindowFunctionContext(this);
  const auto window_func = window_func_context->getWindowFunction();
  if (window_func->getKind() == SqlWindowFunctionKind::AVG) {
    const auto aggregate_state_count_i64 = cgen_state_->llInt(
        reinterpret_cast<const int64_t>(window_func_context->aggregateStateCount()));
    const auto pi64_type =
        llvm::PointerType::get(get_int_type(64, cgen_state_->context_), 0);
    aggregate_state_count =
        cgen_state_->ir_builder_.CreateIntToPtr(aggregate_state_count_i64, pi64_type);
  }
  codegenWindowFunctionStateInit(aggregate_state);
  if (window_func->getKind() == SqlWindowFunctionKind::AVG) {
    const auto count_zero = cgen_state_->llInt(int64_t(0));
    cgen_state_->emitCall("agg_id", {aggregate_state_count, count_zero});
  }
  cgen_state_->ir_builder_.CreateBr(reset_state_false_bb);
  cgen_state_->ir_builder_.SetInsertPoint(reset_state_false_bb);
  CHECK(WindowProjectNodeContext::get(this));
  return codegenWindowFunctionAggregateCalls(aggregate_state, co);
}

Here is the call graph for this function:

llvm::Value * Executor::codegenWindowFunctionAggregateCalls ( llvm::Value *  aggregate_state, const CompilationOptions &  co  )

private

Definition at line 241 of file WindowFunctionIR.cpp.

References run_benchmark_import::args, AVG, CHECK(), CHECK_EQ, CodeGenerator::codegen(), CodeGenerator::codegenCastBetweenIntTypes(), COUNT, anonymous_namespace{WindowFunctionIR.cpp}::get_adjusted_window_type_info(), anonymous_namespace{WindowFunctionIR.cpp}::get_window_agg_name(), WindowProjectNodeContext::getActiveWindowFunctionContext(), kFLOAT, and SUM.

                                                                                         {
  const auto window_func_context =
      WindowProjectNodeContext::getActiveWindowFunctionContext(this);
  const auto window_func = window_func_context->getWindowFunction();
  const auto window_func_ti = get_adjusted_window_type_info(window_func);
  const auto window_func_null_val =
      window_func_ti.is_fp()
          ? cgen_state_->inlineFpNull(window_func_ti)
          : cgen_state_->castToTypeIn(cgen_state_->inlineIntNull(window_func_ti), 64);
  const auto& args = window_func->getArgs();
  llvm::Value* crt_val;
  if (args.empty()) {
    CHECK(window_func->getKind() == SqlWindowFunctionKind::COUNT);
    crt_val = cgen_state_->llInt(int64_t(1));
  } else {
    CodeGenerator code_generator(this);
    const auto arg_lvs = code_generator.codegen(args.front().get(), true, co);
    CHECK_EQ(arg_lvs.size(), size_t(1));
    if (window_func->getKind() == SqlWindowFunctionKind::SUM && !window_func_ti.is_fp()) {
      crt_val = code_generator.codegenCastBetweenIntTypes(
          arg_lvs.front(), args.front()->get_type_info(), window_func_ti, false);
    } else {
      crt_val = window_func_ti.get_type() == kFLOAT
                    ? arg_lvs.front()
                    : cgen_state_->castToTypeIn(arg_lvs.front(), 64);
    }
  }
  const auto agg_name = get_window_agg_name(window_func->getKind(), window_func_ti);
  llvm::Value* multiplicity_lv = nullptr;
  if (args.empty()) {
    cgen_state_->emitCall(agg_name, {aggregate_state, crt_val});
  } else {
    cgen_state_->emitCall(agg_name + "_skip_val",
                          {aggregate_state, crt_val, window_func_null_val});
  }
  if (window_func->getKind() == SqlWindowFunctionKind::AVG) {
    codegenWindowAvgEpilogue(crt_val, window_func_null_val, multiplicity_lv);
  }
  return codegenAggregateWindowState();
}

Here is the call graph for this function:

void Executor::codegenWindowFunctionStateInit ( llvm::Value *  aggregate_state )

private

Definition at line 192 of file WindowFunctionIR.cpp.

References COUNT, anonymous_namespace{WindowFunctionIR.cpp}::get_adjusted_window_type_info(), get_int_type(), WindowProjectNodeContext::getActiveWindowFunctionContext(), kDOUBLE, and kFLOAT.

                                                                        {
  const auto window_func_context =
      WindowProjectNodeContext::getActiveWindowFunctionContext(this);
  const auto window_func = window_func_context->getWindowFunction();
  const auto window_func_ti = get_adjusted_window_type_info(window_func);
  const auto window_func_null_val =
      window_func_ti.is_fp()
          ? cgen_state_->inlineFpNull(window_func_ti)
          : cgen_state_->castToTypeIn(cgen_state_->inlineIntNull(window_func_ti), 64);
  llvm::Value* window_func_init_val;
  if (window_func_context->getWindowFunction()->getKind() ==
      SqlWindowFunctionKind::COUNT) {
    switch (window_func_ti.get_type()) {
      case kFLOAT: {
        window_func_init_val = cgen_state_->llFp(float(0));
        break;
      }
      case kDOUBLE: {
        window_func_init_val = cgen_state_->llFp(double(0));
        break;
      }
      default: {
        window_func_init_val = cgen_state_->llInt(int64_t(0));
        break;
      }
    }
  } else {
    window_func_init_val = window_func_null_val;
  }
  const auto pi32_type =
      llvm::PointerType::get(get_int_type(32, cgen_state_->context_), 0);
  switch (window_func_ti.get_type()) {
    case kDOUBLE: {
      cgen_state_->emitCall("agg_id_double", {aggregate_state, window_func_init_val});
      break;
    }
    case kFLOAT: {
      aggregate_state =
          cgen_state_->ir_builder_.CreateBitCast(aggregate_state, pi32_type);
      cgen_state_->emitCall("agg_id_float", {aggregate_state, window_func_init_val});
      break;
    }
    default: {
      cgen_state_->emitCall("agg_id", {aggregate_state, window_func_init_val});
      break;
    }
  }
}

Here is the call graph for this function:

llvm::BasicBlock * Executor::codegenWindowResetStateControlFlow ( )

private

Definition at line 164 of file WindowFunctionIR.cpp.

References WindowProjectNodeContext::getActiveWindowFunctionContext(), CodeGenerator::posArg(), and CodeGenerator::toBool().

                                                             {
  const auto window_func_context =
      WindowProjectNodeContext::getActiveWindowFunctionContext(this);
  const auto bitset = cgen_state_->llInt(
      reinterpret_cast<const int64_t>(window_func_context->partitionStart()));
  const auto min_val = cgen_state_->llInt(int64_t(0));
  const auto max_val = cgen_state_->llInt(window_func_context->elementCount() - 1);
  const auto null_val = cgen_state_->llInt(inline_int_null_value<int64_t>());
  const auto null_bool_val = cgen_state_->llInt<int8_t>(inline_int_null_value<int8_t>());
  CodeGenerator code_generator(this);
  const auto reset_state =
      code_generator.toBool(cgen_state_->emitCall("bit_is_set",
                                                  {bitset,
                                                   code_generator.posArg(nullptr),
                                                   min_val,
                                                   max_val,
                                                   null_val,
                                                   null_bool_val}));
  const auto reset_state_true_bb = llvm::BasicBlock::Create(
      cgen_state_->context_, "reset_state.true", cgen_state_->row_func_);
  const auto reset_state_false_bb = llvm::BasicBlock::Create(
      cgen_state_->context_, "reset_state.false", cgen_state_->row_func_);
  cgen_state_->ir_builder_.CreateCondBr(
      reset_state, reset_state_true_bb, reset_state_false_bb);
  cgen_state_->ir_builder_.SetInsertPoint(reset_state_true_bb);
  return reset_state_false_bb;
}

Here is the call graph for this function:

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  )

private

Definition at line 1755 of file Execute.cpp.

References anonymous_namespace{Execute.cpp}::build_row_for_empty_input(), catalog_(), DEBUG_TIMER, SharedKernelContext::getFragmentResults(), QueryMemoryDescriptor::getQueryDescriptionType(), GPU, NonGroupedAggregate, GroupByAndAggregate::shard_count_for_top_groups(), RelAlgExecutionUnit::target_exprs, and use_speculative_top_n().

                                                        {
  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);
}

Here is the call graph for this function:

ResultSetPtr Executor::collectAllDeviceShardedTopResults ( SharedKernelContext &  shared_context, const RelAlgExecutionUnit &  ra_exe_unit  ) const

private

Definition at line 1870 of file Execute.cpp.

References CHECK(), CHECK_EQ, CHECK_LE, SharedKernelContext::getFragmentResults(), SortInfo::limit, SortInfo::offset, SortInfo::order_entries, anonymous_namespace{Execute.cpp}::permute_storage_columnar(), anonymous_namespace{Execute.cpp}::permute_storage_row_wise(), run_benchmark_import::result, and RelAlgExecutionUnit::sort_info.

                                                  {
  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);
    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(),
                                                    this);
  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;
}

Here is the call graph for this function:

bool Executor::compileBody ( const RelAlgExecutionUnit &  ra_exe_unit, GroupByAndAggregate &  group_by_and_aggregate, const QueryMemoryDescriptor &  query_mem_desc, const CompilationOptions &  co, const GpuSharedMemoryContext &  gpu_smem_context = {}  )

private

Definition at line 2207 of file NativeCodegen.cpp.

                                                                           {
  // generate the code for the filter
  std::vector<Analyzer::Expr*> primary_quals;
  std::vector<Analyzer::Expr*> deferred_quals;
  bool short_circuited =
      CodeGenerator::prioritizeQuals(ra_exe_unit, primary_quals, deferred_quals);
  if (short_circuited) {
    VLOG(1) << "Prioritized " << std::to_string(primary_quals.size()) << " quals, "
            << "short-circuited and deferred " << std::to_string(deferred_quals.size())
            << " quals";
  }
  llvm::Value* filter_lv = cgen_state_->llBool(true);
  CodeGenerator code_generator(this);
  for (auto expr : primary_quals) {
    // Generate the filter for primary quals
    auto cond = code_generator.toBool(code_generator.codegen(expr, true, co).front());
    filter_lv = cgen_state_->ir_builder_.CreateAnd(filter_lv, cond);
  }
  CHECK(filter_lv->getType()->isIntegerTy(1));
  llvm::BasicBlock* sc_false{nullptr};
  if (!deferred_quals.empty()) {
    auto sc_true = llvm::BasicBlock::Create(
        cgen_state_->context_, "sc_true", cgen_state_->row_func_);
    sc_false = llvm::BasicBlock::Create(
        cgen_state_->context_, "sc_false", cgen_state_->row_func_);
    cgen_state_->ir_builder_.CreateCondBr(filter_lv, sc_true, sc_false);
    cgen_state_->ir_builder_.SetInsertPoint(sc_false);
    if (ra_exe_unit.join_quals.empty()) {
      cgen_state_->ir_builder_.CreateRet(cgen_state_->llInt(int32_t(0)));
    }
    cgen_state_->ir_builder_.SetInsertPoint(sc_true);
    filter_lv = cgen_state_->llBool(true);
  }
  for (auto expr : deferred_quals) {
    filter_lv = cgen_state_->ir_builder_.CreateAnd(
        filter_lv, code_generator.toBool(code_generator.codegen(expr, true, co).front()));
  }

  CHECK(filter_lv->getType()->isIntegerTy(1));
  return group_by_and_aggregate.codegen(
      filter_lv, sc_false, query_mem_desc, co, gpu_smem_context);
}

std::tuple< CompilationResult, std::unique_ptr< QueryMemoryDescriptor > > Executor::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 >  row_set_mem_owner, 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  )

private

Definition at line 1868 of file NativeCodegen.cpp.

                                                   {
  auto timer = DEBUG_TIMER(__func__);
  nukeOldState(allow_lazy_fetch, query_infos, &ra_exe_unit);

  GroupByAndAggregate group_by_and_aggregate(
      this, co.device_type, ra_exe_unit, query_infos, row_set_mem_owner);
  auto query_mem_desc =
      group_by_and_aggregate.initQueryMemoryDescriptor(eo.allow_multifrag,
                                                       max_groups_buffer_entry_guess,
                                                       crt_min_byte_width,
                                                       render_info,
                                                       eo.output_columnar_hint);

  if (query_mem_desc->getQueryDescriptionType() ==
          QueryDescriptionType::GroupByBaselineHash &&
      !has_cardinality_estimation &&
      (!render_info || !render_info->isPotentialInSituRender()) && !eo.just_explain) {
    throw CardinalityEstimationRequired();
  }

  const bool output_columnar = query_mem_desc->didOutputColumnar();
  const bool gpu_shared_mem_optimization =
      is_gpu_shared_mem_supported(query_mem_desc.get(),
                                  ra_exe_unit,
                                  cuda_mgr,
                                  co.device_type,
                                  cuda_mgr ? this->blockSize() : 1,
                                  cuda_mgr ? this->numBlocksPerMP() : 1);
  if (gpu_shared_mem_optimization) {
    // disable interleaved bins optimization on the GPU
    query_mem_desc->setHasInterleavedBinsOnGpu(false);
    LOG(DEBUG1) << "GPU shared memory is used for the " +
                       query_mem_desc->queryDescTypeToString() + " query(" +
                       std::to_string(get_shared_memory_size(gpu_shared_mem_optimization,
                                                             query_mem_desc.get())) +
                       " out of " + std::to_string(g_gpu_smem_threshold) + " bytes).";
  }

  const GpuSharedMemoryContext gpu_smem_context(
      get_shared_memory_size(gpu_shared_mem_optimization, query_mem_desc.get()));

  if (co.device_type == ExecutorDeviceType::GPU) {
    const size_t num_count_distinct_descs =
        query_mem_desc->getCountDistinctDescriptorsSize();
    for (size_t i = 0; i < num_count_distinct_descs; i++) {
      const auto& count_distinct_descriptor =
          query_mem_desc->getCountDistinctDescriptor(i);
      if (count_distinct_descriptor.impl_type_ == CountDistinctImplType::StdSet ||
          (count_distinct_descriptor.impl_type_ != CountDistinctImplType::Invalid &&
           !co.hoist_literals)) {
        throw QueryMustRunOnCpu();
      }
    }
  }

  // Read the module template and target either CPU or GPU
  // by binding the stream position functions to the right implementation:
  // stride access for GPU, contiguous for CPU
  auto rt_module_copy = llvm::CloneModule(
#if LLVM_VERSION_MAJOR >= 7
      *g_rt_module.get(),
#else
      g_rt_module.get(),
#endif
      cgen_state_->vmap_,
      [](const llvm::GlobalValue* gv) {
        auto func = llvm::dyn_cast<llvm::Function>(gv);
        if (!func) {
          return true;
        }
        return (func->getLinkage() == llvm::GlobalValue::LinkageTypes::PrivateLinkage ||
                func->getLinkage() == llvm::GlobalValue::LinkageTypes::InternalLinkage ||
                CodeGenerator::alwaysCloneRuntimeFunction(func));
      });

  if (co.device_type == ExecutorDeviceType::CPU) {
    if (is_udf_module_present(true)) {
      CodeGenerator::link_udf_module(udf_cpu_module, *rt_module_copy, cgen_state_.get());
    }
    if (is_rt_udf_module_present(true)) {
      CodeGenerator::link_udf_module(
          rt_udf_cpu_module, *rt_module_copy, cgen_state_.get());
    }
  } else {
    rt_module_copy->setDataLayout(get_gpu_data_layout());
    rt_module_copy->setTargetTriple(get_gpu_target_triple_string());

    if (is_udf_module_present()) {
      llvm::Triple gpu_triple(udf_gpu_module->getTargetTriple());

      if (!gpu_triple.isNVPTX()) {
        throw QueryMustRunOnCpu();
      }

      CodeGenerator::link_udf_module(udf_gpu_module, *rt_module_copy, cgen_state_.get());
    }
    if (is_rt_udf_module_present()) {
      CodeGenerator::link_udf_module(
          rt_udf_gpu_module, *rt_module_copy, cgen_state_.get());
    }
  }

  cgen_state_->module_ = rt_module_copy.release();

  auto agg_fnames =
      get_agg_fnames(ra_exe_unit.target_exprs, !ra_exe_unit.groupby_exprs.empty());

  const auto agg_slot_count = ra_exe_unit.estimator ? size_t(1) : agg_fnames.size();

  const bool is_group_by{query_mem_desc->isGroupBy()};
  auto query_func = is_group_by ? query_group_by_template(cgen_state_->module_,
                                                          co.hoist_literals,
                                                          *query_mem_desc,
                                                          co.device_type,
                                                          ra_exe_unit.scan_limit,
                                                          gpu_smem_context)
                                : query_template(cgen_state_->module_,
                                                 agg_slot_count,
                                                 co.hoist_literals,
                                                 !!ra_exe_unit.estimator,
                                                 gpu_smem_context);
  bind_pos_placeholders("pos_start", true, query_func, cgen_state_->module_);
  bind_pos_placeholders("group_buff_idx", false, query_func, cgen_state_->module_);
  bind_pos_placeholders("pos_step", false, query_func, cgen_state_->module_);

  cgen_state_->query_func_ = query_func;
  cgen_state_->query_func_entry_ir_builder_.SetInsertPoint(
      &query_func->getEntryBlock().front());

  std::vector<llvm::Value*> col_heads;
  std::tie(cgen_state_->row_func_, col_heads) =
      create_row_function(ra_exe_unit.input_col_descs.size(),
                          is_group_by ? 0 : agg_slot_count,
                          co.hoist_literals,
                          query_func,
                          cgen_state_->module_,
                          cgen_state_->context_);
  CHECK(cgen_state_->row_func_);
  // make sure it's in-lined, we don't want register spills in the inner loop
  mark_function_always_inline(cgen_state_->row_func_);
  auto bb =
      llvm::BasicBlock::Create(cgen_state_->context_, "entry", cgen_state_->row_func_);
  cgen_state_->ir_builder_.SetInsertPoint(bb);
  preloadFragOffsets(ra_exe_unit.input_descs, query_infos);
  RelAlgExecutionUnit body_execution_unit = ra_exe_unit;
  const auto join_loops =
      buildJoinLoops(body_execution_unit, co, eo, query_infos, column_cache);
  plan_state_->allocateLocalColumnIds(ra_exe_unit.input_col_descs);
  const auto is_not_deleted_bb = codegenSkipDeletedOuterTableRow(ra_exe_unit, co);
  if (is_not_deleted_bb) {
    bb = is_not_deleted_bb;
  }
  if (!join_loops.empty()) {
    codegenJoinLoops(join_loops,
                     body_execution_unit,
                     group_by_and_aggregate,
                     query_func,
                     bb,
                     *(query_mem_desc.get()),
                     co,
                     eo);
  } else {
    const bool can_return_error = compileBody(
        ra_exe_unit, group_by_and_aggregate, *query_mem_desc, co, gpu_smem_context);
    if (can_return_error || cgen_state_->needs_error_check_ || eo.with_dynamic_watchdog ||
        eo.allow_runtime_query_interrupt) {
      createErrorCheckControlFlow(query_func,
                                  eo.with_dynamic_watchdog,
                                  eo.allow_runtime_query_interrupt,
                                  co.device_type);
    }
  }
  std::vector<llvm::Value*> hoisted_literals;

  if (co.hoist_literals) {
    VLOG(1) << "number of hoisted literals: "
            << cgen_state_->query_func_literal_loads_.size()
            << " / literal buffer usage: " << cgen_state_->getLiteralBufferUsage(0)
            << " bytes";
  }

  if (co.hoist_literals && !cgen_state_->query_func_literal_loads_.empty()) {
    // we have some hoisted literals...
    hoisted_literals = inlineHoistedLiterals();
  }
  // iterate through all the instruction in the query template function and
  // replace the call to the filter placeholder with the call to the actual filter
  for (auto it = llvm::inst_begin(query_func), e = llvm::inst_end(query_func); it != e;
       ++it) {
    if (!llvm::isa<llvm::CallInst>(*it)) {
      continue;
    }
    auto& filter_call = llvm::cast<llvm::CallInst>(*it);
    if (std::string(filter_call.getCalledFunction()->getName()) == "row_process") {
      std::vector<llvm::Value*> args;
      for (size_t i = 0; i < filter_call.getNumArgOperands(); ++i) {
        args.push_back(filter_call.getArgOperand(i));
      }
      args.insert(args.end(), col_heads.begin(), col_heads.end());
      args.push_back(get_arg_by_name(query_func, "join_hash_tables"));
      // push hoisted literals arguments, if any
      args.insert(args.end(), hoisted_literals.begin(), hoisted_literals.end());

      llvm::ReplaceInstWithInst(&filter_call,
                                llvm::CallInst::Create(cgen_state_->row_func_, args, ""));
      break;
    }
  }

  plan_state_->init_agg_vals_ =
      init_agg_val_vec(ra_exe_unit.target_exprs, ra_exe_unit.quals, *query_mem_desc);

  /*
   * If we have decided to use GPU shared memory (decision is not made here), then
   * we generate proper code for extra components that it needs (buffer initialization and
   * gpu reduction from shared memory to global memory). We then replace these functions
   * into the already compiled query_func (replacing two placeholders, write_back_nop and
   * init_smem_nop). The rest of the code should be as before (row_func, etc.).
   */
  if (gpu_smem_context.isSharedMemoryUsed()) {
    if (query_mem_desc->getQueryDescriptionType() ==
        QueryDescriptionType::GroupByPerfectHash) {
      GpuSharedMemCodeBuilder gpu_smem_code(
          cgen_state_->module_,
          cgen_state_->context_,
          *query_mem_desc,
          target_exprs_to_infos(ra_exe_unit.target_exprs, *query_mem_desc),
          plan_state_->init_agg_vals_);
      gpu_smem_code.codegen();
      gpu_smem_code.injectFunctionsInto(query_func);

      // helper functions are used for caching purposes later
      cgen_state_->helper_functions_.push_back(gpu_smem_code.getReductionFunction());
      cgen_state_->helper_functions_.push_back(gpu_smem_code.getInitFunction());
      LOG(IR) << gpu_smem_code.toString();
    }
  }

  auto multifrag_query_func = cgen_state_->module_->getFunction(
      "multifrag_query" + std::string(co.hoist_literals ? "_hoisted_literals" : ""));
  CHECK(multifrag_query_func);

  bind_query(query_func,
             "query_stub" + std::string(co.hoist_literals ? "_hoisted_literals" : ""),
             multifrag_query_func,
             cgen_state_->module_);

  auto live_funcs =
      CodeGenerator::markDeadRuntimeFuncs(*cgen_state_->module_,
                                          {query_func, cgen_state_->row_func_},
                                          {multifrag_query_func});

  std::string llvm_ir;
  if (eo.just_explain) {
    if (co.explain_type == ExecutorExplainType::Optimized) {
#ifdef WITH_JIT_DEBUG
      throw std::runtime_error(
          "Explain optimized not available when JIT runtime debug symbols are enabled");
#else
      // Note that we don't run the NVVM reflect pass here. Use LOG(IR) to get the
      // optimized IR after NVVM reflect
      llvm::legacy::PassManager pass_manager;
      optimize_ir(query_func, cgen_state_->module_, pass_manager, live_funcs, co);
#endif  // WITH_JIT_DEBUG
    }
    llvm_ir =
        serialize_llvm_object(query_func) + serialize_llvm_object(cgen_state_->row_func_);
  }
  verify_function_ir(cgen_state_->row_func_);

  LOG(IR) << query_mem_desc->toString() << "\nGenerated IR\n"
          << serialize_llvm_object(query_func)
          << serialize_llvm_object(cgen_state_->row_func_) << "\nEnd of IR";

  return std::make_tuple(
      CompilationResult{
          co.device_type == ExecutorDeviceType::CPU
              ? optimizeAndCodegenCPU(query_func, multifrag_query_func, live_funcs, co)
              : optimizeAndCodegenGPU(query_func,
                                      multifrag_query_func,
                                      live_funcs,
                                      is_group_by || ra_exe_unit.estimator,
                                      cuda_mgr,
                                      co),
          cgen_state_->getLiterals(),
          output_columnar,
          llvm_ir,
          std::move(gpu_smem_context)},
      std::move(query_mem_desc));
}

AggregatedColRange Executor::computeColRangesCache ( const std::unordered_set< PhysicalInput > &  phys_inputs )

private

Definition at line 3447 of file Execute.cpp.

References catalog_(), CHECK(), getLeafColumnRange(), AggregatedColRange::setColRange(), and ExpressionRange::typeSupportsRange().

                                                        {
  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;
}

Here is the call graph for this function:

StringDictionaryGenerations Executor::computeStringDictionaryGenerations ( const std::unordered_set< PhysicalInput > &  phys_inputs )

private

Definition at line 3473 of file Execute.cpp.

References catalog_(), CHECK(), kENCODING_DICT, and StringDictionaryGenerations::setGeneration().

                                                        {
  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;
}

Here is the call graph for this function:

TableGenerations Executor::computeTableGenerations ( std::unordered_set< int >  phys_table_ids )

private

Definition at line 3494 of file Execute.cpp.

References TableGenerations::setGeneration().

                                          {
  TableGenerations table_generations;
  for (const int table_id : phys_table_ids) {
    const auto table_info = getTableInfo(table_id);
    table_generations.setGeneration(
        table_id, TableGeneration{table_info.getPhysicalNumTuples(), 0});
  }
  return table_generations;
}

Here is the call graph for this function:

bool Executor::containsLeftDeepOuterJoin ( ) const

inline

Definition at line 338 of file Execute.h.

References cgen_state_.

                                         {
    return cgen_state_->contains_left_deep_outer_join_;
  }

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

private

Definition at line 1449 of file NativeCodegen.cpp.

                                                                           {
  // check whether the row processing was successful; currently, it can
  // fail by running out of group by buffer slots

  if (run_with_dynamic_watchdog && run_with_allowing_runtime_interrupt) {
    // when both dynamic watchdog and runtime interrupt turns on
    // we use dynamic watchdog
    run_with_allowing_runtime_interrupt = false;
  }

  llvm::Value* row_count = nullptr;
  if ((run_with_dynamic_watchdog || run_with_allowing_runtime_interrupt) &&
      device_type == ExecutorDeviceType::GPU) {
    row_count =
        find_variable_in_basic_block<llvm::LoadInst>(query_func, ".entry", "row_count");
  }

  bool done_splitting = false;
  for (auto bb_it = query_func->begin(); bb_it != query_func->end() && !done_splitting;
       ++bb_it) {
    llvm::Value* pos = nullptr;
    for (auto inst_it = bb_it->begin(); inst_it != bb_it->end(); ++inst_it) {
      if ((run_with_dynamic_watchdog || run_with_allowing_runtime_interrupt) &&
          llvm::isa<llvm::PHINode>(*inst_it)) {
        if (inst_it->getName() == "pos") {
          pos = &*inst_it;
        }
        continue;
      }
      if (!llvm::isa<llvm::CallInst>(*inst_it)) {
        continue;
      }
      auto& filter_call = llvm::cast<llvm::CallInst>(*inst_it);
      if (std::string(filter_call.getCalledFunction()->getName()) == "row_process") {
        auto next_inst_it = inst_it;
        ++next_inst_it;
        auto new_bb = bb_it->splitBasicBlock(next_inst_it);
        auto& br_instr = bb_it->back();
        llvm::IRBuilder<> ir_builder(&br_instr);
        llvm::Value* err_lv = &*inst_it;
        if (run_with_dynamic_watchdog) {
          CHECK(pos);
          llvm::Value* call_watchdog_lv = nullptr;
          if (device_type == ExecutorDeviceType::GPU) {
            // In order to make sure all threads within a block see the same barrier,
            // only those blocks whose none of their threads have experienced the critical
            // edge will go through the dynamic watchdog computation
            CHECK(row_count);
            auto crit_edge_rem =
                (blockSize() & (blockSize() - 1))
                    ? ir_builder.CreateSRem(
                          row_count,
                          cgen_state_->llInt(static_cast<int64_t>(blockSize())))
                    : ir_builder.CreateAnd(
                          row_count,
                          cgen_state_->llInt(static_cast<int64_t>(blockSize() - 1)));
            auto crit_edge_threshold = ir_builder.CreateSub(row_count, crit_edge_rem);
            crit_edge_threshold->setName("crit_edge_threshold");

            // only those threads where pos < crit_edge_threshold go through dynamic
            // watchdog call
            call_watchdog_lv =
                ir_builder.CreateICmp(llvm::ICmpInst::ICMP_SLT, pos, crit_edge_threshold);
          } else {
            // CPU path: run watchdog for every 64th row
            auto dw_predicate = ir_builder.CreateAnd(pos, uint64_t(0x3f));
            call_watchdog_lv = ir_builder.CreateICmp(
                llvm::ICmpInst::ICMP_EQ, dw_predicate, cgen_state_->llInt(int64_t(0LL)));
          }
          CHECK(call_watchdog_lv);
          auto error_check_bb = bb_it->splitBasicBlock(
              llvm::BasicBlock::iterator(br_instr), ".error_check");
          auto& watchdog_br_instr = bb_it->back();

          auto watchdog_check_bb = llvm::BasicBlock::Create(
              cgen_state_->context_, ".watchdog_check", query_func, error_check_bb);
          llvm::IRBuilder<> watchdog_ir_builder(watchdog_check_bb);
          auto detected_timeout = watchdog_ir_builder.CreateCall(
              cgen_state_->module_->getFunction("dynamic_watchdog"), {});
          auto timeout_err_lv = watchdog_ir_builder.CreateSelect(
              detected_timeout, cgen_state_->llInt(Executor::ERR_OUT_OF_TIME), err_lv);
          watchdog_ir_builder.CreateBr(error_check_bb);

          llvm::ReplaceInstWithInst(
              &watchdog_br_instr,
              llvm::BranchInst::Create(
                  watchdog_check_bb, error_check_bb, call_watchdog_lv));
          ir_builder.SetInsertPoint(&br_instr);
          auto unified_err_lv = ir_builder.CreatePHI(err_lv->getType(), 2);

          unified_err_lv->addIncoming(timeout_err_lv, watchdog_check_bb);
          unified_err_lv->addIncoming(err_lv, &*bb_it);
          err_lv = unified_err_lv;
        } else if (run_with_allowing_runtime_interrupt) {
          CHECK(pos);
          llvm::Value* call_check_interrupt_lv = nullptr;
          if (device_type == ExecutorDeviceType::GPU) {
            // approximate how many times the %pos variable
            // is increased --> the number of iteration
            int32_t num_shift_by_gridDim = getExpOfTwo(gridSize());
            int32_t num_shift_by_blockDim = getExpOfTwo(blockSize());
            if (!isPowOfTwo(gridSize())) {
              num_shift_by_gridDim++;
            }
            if (!isPowOfTwo(blockSize())) {
              num_shift_by_blockDim++;
            }
            int total_num_shift = num_shift_by_gridDim + num_shift_by_blockDim;
            // check the interrupt flag for every 64th iteration
            llvm::Value* pos_shifted_per_iteration =
                ir_builder.CreateLShr(pos, cgen_state_->llInt(total_num_shift));
            auto interrupt_predicate =
                ir_builder.CreateAnd(pos_shifted_per_iteration, uint64_t(0x3f));
            call_check_interrupt_lv =
                ir_builder.CreateICmp(llvm::ICmpInst::ICMP_EQ,
                                      interrupt_predicate,
                                      cgen_state_->llInt(int64_t(0LL)));
          } else {
            // CPU path: run interrupt checker for every 64th row
            auto interrupt_predicate = ir_builder.CreateAnd(pos, uint64_t(0x3f));
            call_check_interrupt_lv =
                ir_builder.CreateICmp(llvm::ICmpInst::ICMP_EQ,
                                      interrupt_predicate,
                                      cgen_state_->llInt(int64_t(0LL)));
          }
          CHECK(call_check_interrupt_lv);
          auto error_check_bb = bb_it->splitBasicBlock(
              llvm::BasicBlock::iterator(br_instr), ".error_check");
          auto& check_interrupt_br_instr = bb_it->back();

          auto interrupt_check_bb = llvm::BasicBlock::Create(
              cgen_state_->context_, ".interrupt_check", query_func, error_check_bb);
          llvm::IRBuilder<> interrupt_checker_ir_builder(interrupt_check_bb);
          auto detected_interrupt = interrupt_checker_ir_builder.CreateCall(
              cgen_state_->module_->getFunction("check_interrupt"), {});
          auto interrupt_err_lv = interrupt_checker_ir_builder.CreateSelect(
              detected_interrupt, cgen_state_->llInt(Executor::ERR_INTERRUPTED), err_lv);
          interrupt_checker_ir_builder.CreateBr(error_check_bb);

          llvm::ReplaceInstWithInst(
              &check_interrupt_br_instr,
              llvm::BranchInst::Create(
                  interrupt_check_bb, error_check_bb, call_check_interrupt_lv));
          ir_builder.SetInsertPoint(&br_instr);
          auto unified_err_lv = ir_builder.CreatePHI(err_lv->getType(), 2);

          unified_err_lv->addIncoming(interrupt_err_lv, interrupt_check_bb);
          unified_err_lv->addIncoming(err_lv, &*bb_it);
          err_lv = unified_err_lv;
        }
        const auto error_code_arg = get_arg_by_name(query_func, "error_code");
        err_lv =
            ir_builder.CreateCall(cgen_state_->module_->getFunction("record_error_code"),
                                  std::vector<llvm::Value*>{err_lv, error_code_arg});
        if (device_type == ExecutorDeviceType::GPU) {
          // let kernel execution finish as expected, regardless of the observed error,
          // unless it is from the dynamic watchdog where all threads within that block
          // return together.
          if (run_with_allowing_runtime_interrupt) {
            err_lv = ir_builder.CreateICmp(llvm::ICmpInst::ICMP_EQ,
                                           err_lv,
                                           cgen_state_->llInt(Executor::ERR_INTERRUPTED));
          } else {
            err_lv = ir_builder.CreateICmp(llvm::ICmpInst::ICMP_EQ,
                                           err_lv,
                                           cgen_state_->llInt(Executor::ERR_OUT_OF_TIME));
          }

        } else {
          err_lv = ir_builder.CreateICmp(llvm::ICmpInst::ICMP_NE,
                                         err_lv,
                                         cgen_state_->llInt(static_cast<int32_t>(0)));
        }
        auto error_bb = llvm::BasicBlock::Create(
            cgen_state_->context_, ".error_exit", query_func, new_bb);
        llvm::ReturnInst::Create(cgen_state_->context_, error_bb);
        llvm::ReplaceInstWithInst(&br_instr,
                                  llvm::BranchInst::Create(error_bb, new_bb, err_lv));
        done_splitting = true;
        break;
      }
    }
  }
  CHECK(done_splitting);
}

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  )

private

Determines execution dispatch mode and required fragments for a given query step, then creates kernels to execute the query and returns them for launch.

Definition at line 1944 of file Execute.cpp.

References ExecutionOptions::allow_multifrag, catalog_(), CHECK(), CHECK_GE, CHECK_GT, anonymous_namespace{Execute.cpp}::checkWorkUnitWatchdog(), g_inner_join_fragment_skipping, QueryCompilationDescriptor::getDeviceType(), QueryMemoryDescriptor::getEntryCount(), SharedKernelContext::getFragOffsets(), QueryMemoryDescriptor::getQueryDescriptionType(), GPU, ExecutionOptions::gpu_input_mem_limit_percent, Data_Namespace::GPU_LEVEL, anonymous_namespace{Execute.cpp}::has_lazy_fetched_columns(), logger::INFO, RelAlgExecutionUnit::input_descs, KernelPerFragment, LOG, MultifragmentKernel, ExecutionOptions::outer_fragment_indices, Projection, query_mem_desc, RelAlgExecutionUnit::target_exprs, logger::thread_id(), QueryMemoryDescriptor::toString(), RelAlgExecutionUnit::use_bump_allocator, VLOG, and ExecutionOptions::with_watchdog.

                         {
  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,
                                      parent_thread_id = logger::thread_id()](
                                         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,
                                            parent_thread_id));
    };
    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,
                                         parent_thread_id = logger::thread_id()](
                                            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,
                                            parent_thread_id));
      ++frag_list_idx;
    };

    fragment_descriptor.assignFragsToKernelDispatch(fragment_per_kernel_dispatch,
                                                    ra_exe_unit);
  }

  return execution_kernels;
}

Here is the call graph for this function:

int Executor::deviceCount ( const ExecutorDeviceType  device_type ) const

private

Definition at line 615 of file Execute.cpp.

References catalog_(), CHECK(), and GPU.

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

Here is the call graph for this function:

int Executor::deviceCountForMemoryLevel ( const Data_Namespace::MemoryLevel  memory_level ) const

private

Definition at line 625 of file Execute.cpp.

References CPU, GPU, and Data_Namespace::GPU_LEVEL.

                                                        {
  return memory_level == GPU_LEVEL ? deviceCount(ExecutorDeviceType::GPU)
                                   : deviceCount(ExecutorDeviceType::CPU);
}

int64_t Executor::deviceCycles ( int  milliseconds ) const

private

Definition at line 3140 of file Execute.cpp.

References catalog_(), and CHECK().

                                                     {
  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;
}

Here is the call graph for this function:

void Executor::enableRuntimeQueryInterrupt

(

const unsigned

interrupt_freq

)

const

Definition at line 3574 of file Execute.cpp.

References g_enable_runtime_query_interrupt, and g_runtime_query_interrupt_frequency.

                                                                              {
  // 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_runtime_query_interrupt_frequency = interrupt_freq;
}

ResultSetPtr Executor::executeExplain ( const QueryCompilationDescriptor &  query_comp_desc )

private

Definition at line 1624 of file Execute.cpp.

References QueryCompilationDescriptor::getIR().

                                                                                       {
  return std::make_shared<ResultSet>(query_comp_desc.getIR());
}

Here is the call graph for this function:

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  limit, const uint32_t  start_rowid, const uint32_t  num_tables, RenderInfo *  render_info  )

private

Definition at line 2857 of file Execute.cpp.

References CHECK(), CHECK_NE, anonymous_namespace{Execute.cpp}::check_rows_less_than_needed(), CPU, DEBUG_TIMER, ERR_DIV_BY_ZERO, ERR_GEOS, ERR_INTERRUPTED, ERR_OUT_OF_TIME, ERR_OVERFLOW_OR_UNDERFLOW, ERR_SINGLE_VALUE_FOUND_MULTIPLE_VALUES, error_code, logger::FATAL, g_enable_dynamic_watchdog, g_enable_runtime_query_interrupt, CompilationResult::generated_code, QueryMemoryDescriptor::getEntryCount(), QueryExecutionContext::getRowSet(), GpuSharedMemoryContext::getSharedMemorySize(), GPU, CompilationResult::gpu_smem_context, RelAlgExecutionUnit::groupby_exprs, INJECT_TIMER, RelAlgExecutionUnit::input_col_descs, RelAlgExecutionUnit::input_descs, QueryExecutionContext::launchCpuCode(), QueryExecutionContext::launchGpuCode(), CompilationResult::literal_values, LOG, num_rows, shared::printContainer(), QueryExecutionContext::query_buffers_, QueryExecutionContext::query_mem_desc_, RenderInfo::render_allocator_map_ptr, RelAlgExecutionUnit::scan_limit, QueryMemoryDescriptor::setEntryCount(), RelAlgExecutionUnit::union_all, RenderInfo::useCudaBuffers(), and VLOG.

                             {
  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;
}

Here is the call graph for this function:

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  )

private

Definition at line 2657 of file Execute.cpp.

References CHECK(), CHECK_EQ, CPU, DEBUG_TIMER, ERR_DIV_BY_ZERO, ERR_GEOS, ERR_INTERRUPTED, ERR_OUT_OF_TIME, ERR_OVERFLOW_OR_UNDERFLOW, ERR_SINGLE_VALUE_FOUND_MULTIPLE_VALUES, error_code, RelAlgExecutionUnit::estimator, QueryExecutionContext::estimator_result_set_, logger::FATAL, g_bigint_count, g_enable_dynamic_watchdog, g_enable_runtime_query_interrupt, CompilationResult::generated_code, get_target_info(), QueryExecutionContext::getAggInitValForIndex(), QueryMemoryDescriptor::getPaddedSlotWidthBytes(), GpuSharedMemoryContext::getSharedMemorySize(), GPU, CompilationResult::gpu_smem_context, INJECT_TIMER, is_distinct_target(), RenderInfo::isPotentialInSituRender(), GpuSharedMemoryContext::isSharedMemoryUsed(), kAPPROX_COUNT_DISTINCT, kAVG, kCOUNT, kSAMPLE, QueryExecutionContext::launchCpuCode(), QueryExecutionContext::launchGpuCode(), CompilationResult::literal_values, LOG, num_rows, out, QueryExecutionContext::query_buffers_, QueryExecutionContext::query_mem_desc_, reduceResults(), RenderInfo::render_allocator_map_ptr, takes_float_argument(), and RenderInfo::useCudaBuffers().

                             {
  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;
}

Here is the call graph for this function:

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

private

Compiles and dispatches a table function; that is, a function that takes as input one or more columns and returns a ResultSet, which can be parsed by subsequent execution steps.

Definition at line 1598 of file Execute.cpp.

References CHECK_EQ, TableFunctionCompilationContext::compile(), CompilationOptions::device_type, TableFunctionExecutionContext::execute(), and INJECT_TIMER.

                                         {
  INJECT_TIMER(Exec_executeTableFunction);
  nukeOldState(false, table_infos, 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());
  CHECK_EQ(table_infos.size(), size_t(1));
  return exe_context.execute(exe_unit,
                             table_infos.front(),
                             &compilation_context,
                             column_fetcher,
                             co.device_type,
                             this);
}

Here is the call graph for this function:

void Executor::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  )

private

Definition at line 62 of file ExecuteUpdate.cpp.

References CHECK(), CHECK_EQ, CHECK_GT, CPU, FragmentsPerTable::fragment_ids, SharedKernelContext::getFragmentResults(), KernelPerFragment, query_mem_desc, ExecutionKernel::run(), logger::thread_id(), timer_start(), timer_stop(), and VLOG.

                                                {
  CHECK(cb);
  VLOG(1) << "Executing update/delete work unit:" << ra_exe_unit_in;

  const auto ra_exe_unit = addDeletedColumn(ra_exe_unit_in, co);
  ColumnCacheMap column_cache;

  ColumnFetcher column_fetcher(this, column_cache);
  CHECK_GT(ra_exe_unit.input_descs.size(), size_t(0));
  const auto table_id = ra_exe_unit.input_descs[0].getTableId();
  const auto& outer_fragments = table_infos.front().info.fragments;

  std::vector<FragmentsPerTable> fragments = {{0, {0}}};
  for (size_t tab_idx = 1; tab_idx < ra_exe_unit.input_descs.size(); tab_idx++) {
    int table_id = ra_exe_unit.input_descs[tab_idx].getTableId();
    CHECK_EQ(table_infos[tab_idx].table_id, table_id);
    const auto& fragmentsPerTable = table_infos[tab_idx].info.fragments;
    FragmentsPerTable entry = {table_id, {}};
    for (size_t innerFragId = 0; innerFragId < fragmentsPerTable.size(); innerFragId++) {
      entry.fragment_ids.push_back(innerFragId);
    }
    fragments.push_back(entry);
  }

  if (outer_fragments.empty()) {
    return;
  }

  const auto max_tuple_count_fragment_it = std::max_element(
      outer_fragments.begin(), outer_fragments.end(), [](const auto& a, const auto& b) {
        return a.getNumTuples() < b.getNumTuples();
      });
  CHECK(max_tuple_count_fragment_it != outer_fragments.end());
  int64_t global_max_groups_buffer_entry_guess =
      max_tuple_count_fragment_it->getNumTuples();
  if (is_agg) {
    global_max_groups_buffer_entry_guess = std::min(
        2 * global_max_groups_buffer_entry_guess, static_cast<int64_t>(100'000'000));
  }

  auto query_comp_desc = std::make_unique<QueryCompilationDescriptor>();
  std::unique_ptr<QueryMemoryDescriptor> query_mem_desc;
  {
    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 = query_comp_desc->compile(global_max_groups_buffer_entry_guess,
                                              8,
                                              /*has_cardinality_estimation=*/true,
                                              ra_exe_unit,
                                              table_infos,
                                              column_fetcher,
                                              co,
                                              eo,
                                              nullptr,
                                              this);
  }
  CHECK(query_mem_desc);

  for (size_t fragment_index = 0; fragment_index < outer_fragments.size();
       ++fragment_index) {
    const int64_t crt_fragment_tuple_count =
        outer_fragments[fragment_index].getNumTuples();
    if (crt_fragment_tuple_count == 0) {
      // nothing to update
      continue;
    }

    SharedKernelContext shared_context(table_infos);
    fragments[0] = {table_id, {fragment_index}};
    {
      ExecutionKernel current_fragment_kernel(ra_exe_unit,
                                              ExecutorDeviceType::CPU,
                                              0,
                                              eo,
                                              column_fetcher,
                                              *query_comp_desc,
                                              *query_mem_desc,
                                              fragments,
                                              ExecutorDispatchMode::KernelPerFragment,
                                              /*render_info=*/nullptr,
                                              /*rowid_lookup_key=*/-1,
                                              logger::thread_id());

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

      current_fragment_kernel.run(this, shared_context);
    }
    const auto& proj_fragment_results = shared_context.getFragmentResults();
    if (proj_fragment_results.empty()) {
      continue;
    }
    const auto& proj_fragment_result = proj_fragment_results[0];
    const auto proj_result_set = proj_fragment_result.first;
    CHECK(proj_result_set);
    cb({outer_fragments[fragment_index], fragment_index, proj_result_set});
  }
}

Here is the call graph for this function:

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 &  options, const Catalog_Namespace::Catalog &  cat, std::shared_ptr< RowSetMemoryOwner >  row_set_mem_owner, RenderInfo *  render_info, const bool  has_cardinality_estimation, ColumnCacheMap &  column_cache  )

private

Definition at line 1310 of file Execute.cpp.

References executor_id_(), CompilationRetryNewScanLimit::new_scan_limit_, anonymous_namespace{Execute.cpp}::replace_scan_limit(), run_benchmark_import::result, and VLOG.

                                  {
  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_);
    }
    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_);
    }
    return result;
  }
}

Here is the call graph for this function:

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 &  options, const Catalog_Namespace::Catalog &  cat, std::shared_ptr< RowSetMemoryOwner >  row_set_mem_owner, RenderInfo *  render_info, const bool  has_cardinality_estimation, ColumnCacheMap &  column_cache  )

private

Definition at line 1373 of file Execute.cpp.

References CompilationOptions::allow_lazy_fetch, ExecutionOptions::allow_runtime_query_interrupt, CHECK(), anonymous_namespace{Execute.cpp}::compute_buffer_entry_guess(), CPU, cpu_threads(), CompilationOptions::device_type, ExecutionOptions::executor_type, CompilationOptions::explain_type, CompilationOptions::filter_on_deleted_column, g_use_tbb_pool, get_available_gpus(), get_context_count(), get_min_byte_width(), QueryExecutionError::getErrorCode(), CompilationOptions::hoist_literals, INJECT_TIMER, ExecutionOptions::just_explain, ExecutionOptions::just_validate, MAX_BYTE_WIDTH_SUPPORTED, Native, CompilationOptions::opt_level, Projection, CompilationOptions::register_intel_jit_listener, timer_start(), timer_stop(), VLOG, CompilationOptions::with_dynamic_watchdog, and ExecutionOptions::with_dynamic_watchdog.

                                  {
  INJECT_TIMER(Exec_executeWorkUnit);
  const auto ra_exe_unit = 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,
                                           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, 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,
                                     this);
}

Here is the call graph for this function:

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

private

Compiles and dispatches a work unit per fragment processing results with the per fragment callback. Currently used for computing metrics over fragments (metadata).

Definition at line 1527 of file Execute.cpp.

References CHECK(), CHECK_EQ, CompilationOptions::device_type, Fragmenter_Namespace::TableInfo::fragments, SharedKernelContext::getFragmentResults(), InputTableInfo::info, KernelPerFragment, ExecutionKernel::run(), logger::thread_id(), timer_start(), and timer_stop().

                                                                   {
  const auto ra_exe_unit = 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,
                                       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,
                             logger::thread_id());
      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]);
  }
}

Here is the call graph for this function:

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  )

private

Definition at line 2310 of file Execute.cpp.

References cat(), CHECK(), CHECK_EQ, CHECK_LT, Data_Namespace::CPU_LEVEL, DEBUG_TIMER, g_enable_dynamic_watchdog, g_enable_runtime_query_interrupt, ColumnFetcher::getAllTableColumnFragments(), ColumnFetcher::getOneTableColumnFragment(), ColumnFetcher::getResultSetColumn(), INJECT_TIMER, RelAlgExecutionUnit::input_col_descs, RelAlgExecutionUnit::input_descs, RESULT, and anonymous_namespace{Execute.cpp}::try_get_column_descriptor().

                                       {
  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};
}

Here is the call graph for this function:

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  )

private

Definition at line 2409 of file Execute.cpp.

References cat(), CHECK(), CHECK_EQ, CHECK_LE, CHECK_LT, Data_Namespace::CPU_LEVEL, DEBUG_TIMER, ColumnFetcher::getAllTableColumnFragments(), ColumnFetcher::getOneTableColumnFragment(), ColumnFetcher::getResultSetColumn(), INJECT_TIMER, RelAlgExecutionUnit::input_col_descs, RelAlgExecutionUnit::input_descs, num_rows, shared::printContainer(), RESULT, anonymous_namespace{Execute.cpp}::try_get_column_descriptor(), RelAlgExecutionUnit::union_all, and VLOG.

                                       {
  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