#include <ExecutionKernel.h>

Collaboration diagram for ExecutionKernel:

Public Member Functions
	ExecutionKernel (const RelAlgExecutionUnit &ra_exe_unit, const ExecutorDeviceType chosen_device_type, int chosen_device_id, const ExecutionOptions &eo, const ColumnFetcher &column_fetcher, const QueryCompilationDescriptor &query_comp_desc, const QueryMemoryDescriptor &query_mem_desc, const FragmentsList &frag_list, const ExecutorDispatchMode kernel_dispatch_mode, RenderInfo *render_info, const int64_t rowid_lookup_key)

void	run (Executor *executor, SharedKernelContext &shared_context)

Private Member Functions
void	runImpl (Executor *executor, SharedKernelContext &shared_context)

Private Attributes
const RelAlgExecutionUnit &	ra_exe_unit_

const ExecutorDeviceType	chosen_device_type

int	chosen_device_id

const ExecutionOptions &	eo

const ColumnFetcher &	column_fetcher

const QueryCompilationDescriptor &	query_comp_desc

const QueryMemoryDescriptor &	query_mem_desc

const FragmentsList	frag_list

const ExecutorDispatchMode	kernel_dispatch_mode

RenderInfo *	render_info_

const int64_t	rowid_lookup_key

ResultSetPtr	device_results_

Detailed Description

Definition at line 49 of file ExecutionKernel.h.

Constructor & Destructor Documentation

ExecutionKernel::ExecutionKernel	(	const RelAlgExecutionUnit &	ra_exe_unit,
		const ExecutorDeviceType	chosen_device_type,
		int	chosen_device_id,
		const ExecutionOptions &	eo,
		const ColumnFetcher &	column_fetcher,
		const QueryCompilationDescriptor &	query_comp_desc,
		const QueryMemoryDescriptor &	query_mem_desc,
		const FragmentsList &	frag_list,
		const ExecutorDispatchMode	kernel_dispatch_mode,
		RenderInfo *	render_info,
		const int64_t	rowid_lookup_key
	)

inline

Definition at line 51 of file ExecutionKernel.h.

       : ra_exe_unit_(ra_exe_unit)
       , chosen_device_type(chosen_device_type)
       , chosen_device_id(chosen_device_id)
       , eo(eo)
       , column_fetcher(column_fetcher)
       , query_comp_desc(query_comp_desc)
       , query_mem_desc(query_mem_desc)
       , frag_list(frag_list)
       , kernel_dispatch_mode(kernel_dispatch_mode)
       , render_info_(render_info)
       , rowid_lookup_key(rowid_lookup_key) {}

Member Function Documentation

void ExecutionKernel::run	(	Executor *	executor,
		SharedKernelContext &	shared_context
	)

Definition at line 90 of file ExecutionKernel.cpp.

References DEBUG_TIMER, Executor::ERR_COLUMNAR_CONVERSION_NOT_SUPPORTED, Executor::ERR_OUT_OF_CPU_MEM, Executor::ERR_OUT_OF_GPU_MEM, Executor::ERR_OUT_OF_RENDER_MEM, Executor::ERR_STRING_CONST_IN_RESULTSET, Executor::ERR_TOO_MANY_LITERALS, QueryMemoryDescriptor::getQueryDescriptionType(), INJECT_TIMER, kernel_dispatch_mode, MultifragmentKernel, query_mem_desc, runImpl(), and OutOfHostMemory::what().

Referenced by Executor::executeUpdate(), Executor::executeWorkUnitPerFragment(), and Executor::launchKernels().

                                                                                  {
   DEBUG_TIMER("ExecutionKernel::run");
   INJECT_TIMER(kernel_run);
   try {
     runImpl(executor, shared_context);
   } catch (const OutOfHostMemory& e) {
     throw QueryExecutionError(Executor::ERR_OUT_OF_CPU_MEM, e.what());
   } catch (const std::bad_alloc& e) {
     throw QueryExecutionError(Executor::ERR_OUT_OF_CPU_MEM, e.what());
   } catch (const OutOfRenderMemory& e) {
     throw QueryExecutionError(Executor::ERR_OUT_OF_RENDER_MEM, e.what());
   } catch (const OutOfMemory& e) {
     throw QueryExecutionError(
         Executor::ERR_OUT_OF_GPU_MEM,
         e.what(),
         QueryExecutionProperties{
             query_mem_desc.getQueryDescriptionType(),
             kernel_dispatch_mode == ExecutorDispatchMode::MultifragmentKernel});
   } catch (const ColumnarConversionNotSupported& e) {
     throw QueryExecutionError(Executor::ERR_COLUMNAR_CONVERSION_NOT_SUPPORTED, e.what());
   } catch (const TooManyLiterals& e) {
     throw QueryExecutionError(Executor::ERR_TOO_MANY_LITERALS, e.what());
   } catch (const SringConstInResultSet& e) {
     throw QueryExecutionError(Executor::ERR_STRING_CONST_IN_RESULTSET, e.what());
   } catch (const QueryExecutionError& e) {
     throw e;
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void ExecutionKernel::runImpl	(	Executor *	executor,
		SharedKernelContext &	shared_context
	)

private

Definition at line 119 of file ExecutionKernel.cpp.

References gpu_enabled::accumulate(), SharedKernelContext::addDeviceResults(), CHECK, CHECK_EQ, CHECK_GE, CHECK_GT, CHECK_LT, chosen_device_id, chosen_device_type, FetchResult::col_buffers, column_fetcher, QueryFragmentDescriptor::computeAllTablesFragments(), CPU, Data_Namespace::CPU_LEVEL, device_results_, dynamic_watchdog_init(), SharedKernelContext::dynamic_watchdog_set, ExecutionOptions::dynamic_watchdog_time_limit, eo, Executor::ERR_OUT_OF_CPU_MEM, Executor::ERR_OUT_OF_GPU_MEM, ExecutionOptions::executor_type, Extern, frag_list, FetchResult::frag_offsets, QueryCompilationDescriptor::getCompilationResult(), SharedKernelContext::getFragOffsets(), QueryMemoryDescriptor::getQueryDescriptionType(), QueryMemoryDescriptor::getQueryExecutionContext(), SharedKernelContext::getQueryInfos(), GPU, Data_Namespace::GPU_LEVEL, RelAlgExecutionUnit::groupby_exprs, QueryCompilationDescriptor::hoistLiterals(), logger::INFO, RelAlgExecutionUnit::input_descs, RenderInfo::isPotentialInSituRender(), kernel_dispatch_mode, KernelPerFragment, LOG, Executor::max_gpu_count, MultifragmentKernel, Native, anonymous_namespace{ExecutionKernel.cpp}::need_to_hold_chunk(), FetchResult::num_rows, CompilationResult::output_columnar, Projection, query_comp_desc, anonymous_namespace{ExecutionKernel.cpp}::query_has_inner_join(), query_mem_desc, ra_exe_unit_, render_info_, rowid_lookup_key, run_query_external(), RelAlgExecutionUnit::scan_limit, serialize_to_sql(), QueryMemoryDescriptor::sortOnGpu(), RelAlgExecutionUnit::target_exprs, target_exprs_to_infos(), to_string(), RelAlgExecutionUnit::union_all, VLOG, and ExecutionOptions::with_dynamic_watchdog.

Referenced by run().

                                                                                      {
   CHECK(executor);
   const auto memory_level = chosen_device_type == ExecutorDeviceType::GPU
                                 ? Data_Namespace::GPU_LEVEL
                                 : Data_Namespace::CPU_LEVEL;
   CHECK_GE(frag_list.size(), size_t(1));
   // frag_list[0].table_id is how we tell which query we are running for UNION ALL.
   const int outer_table_id = ra_exe_unit_.union_all
                                  ? frag_list[0].table_id
                                  : ra_exe_unit_.input_descs[0].getTableId();
   CHECK_EQ(frag_list[0].table_id, outer_table_id);
   const auto& outer_tab_frag_ids = frag_list[0].fragment_ids;
 
   CHECK_GE(chosen_device_id, 0);
   CHECK_LT(chosen_device_id, Executor::max_gpu_count);
 
   auto catalog = executor->getCatalog();
   CHECK(catalog);
 
   // need to own them while query executes
   auto chunk_iterators_ptr = std::make_shared<std::list<ChunkIter>>();
   std::list<std::shared_ptr<Chunk_NS::Chunk>> chunks;
   std::unique_ptr<std::lock_guard<std::mutex>> gpu_lock;
   std::unique_ptr<CudaAllocator> device_allocator;
   if (chosen_device_type == ExecutorDeviceType::GPU) {
     gpu_lock.reset(
         new std::lock_guard<std::mutex>(executor->gpu_exec_mutex_[chosen_device_id]));
     device_allocator =
         std::make_unique<CudaAllocator>(&catalog->getDataMgr(), chosen_device_id);
   }
   FetchResult fetch_result;
   try {
     std::map<int, const TableFragments*> all_tables_fragments;
     QueryFragmentDescriptor::computeAllTablesFragments(
         all_tables_fragments, ra_exe_unit_, shared_context.getQueryInfos());
 
     fetch_result = ra_exe_unit_.union_all
                        ? executor->fetchUnionChunks(column_fetcher,
                                                     ra_exe_unit_,
                                                     chosen_device_id,
                                                     memory_level,
                                                     all_tables_fragments,
                                                     frag_list,
                                                     *catalog,
                                                     *chunk_iterators_ptr,
                                                     chunks,
                                                     device_allocator.get())
                        : executor->fetchChunks(column_fetcher,
                                                ra_exe_unit_,
                                                chosen_device_id,
                                                memory_level,
                                                all_tables_fragments,
                                                frag_list,
                                                *catalog,
                                                *chunk_iterators_ptr,
                                                chunks,
                                                device_allocator.get());
     if (fetch_result.num_rows.empty()) {
       return;
     }
     if (eo.with_dynamic_watchdog &&
         !shared_context.dynamic_watchdog_set.test_and_set(std::memory_order_acquire)) {
       CHECK_GT(eo.dynamic_watchdog_time_limit, 0u);
       auto cycle_budget = dynamic_watchdog_init(eo.dynamic_watchdog_time_limit);
       LOG(INFO) << "Dynamic Watchdog budget: CPU: "
                 << std::to_string(eo.dynamic_watchdog_time_limit) << "ms, "
                 << std::to_string(cycle_budget) << " cycles";
     }
   } catch (const OutOfMemory&) {
     throw QueryExecutionError(
         memory_level == Data_Namespace::GPU_LEVEL ? Executor::ERR_OUT_OF_GPU_MEM
                                                   : Executor::ERR_OUT_OF_CPU_MEM,
         QueryExecutionProperties{
             query_mem_desc.getQueryDescriptionType(),
             kernel_dispatch_mode == ExecutorDispatchMode::MultifragmentKernel});
     return;
   }
 
   if (eo.executor_type == ExecutorType::Extern) {
     if (ra_exe_unit_.input_descs.size() > 1) {
       throw std::runtime_error("Joins not supported through external execution");
     }
     const auto query = serialize_to_sql(&ra_exe_unit_, catalog);
     GroupByAndAggregate group_by_and_aggregate(executor,
                                                ExecutorDeviceType::CPU,
                                                ra_exe_unit_,
                                                shared_context.getQueryInfos(),
                                                executor->row_set_mem_owner_,
                                                std::nullopt);
     const auto query_mem_desc =
         group_by_and_aggregate.initQueryMemoryDescriptor(false, 0, 8, nullptr, false);
     device_results_ = run_query_external(
         query,
         fetch_result,
         executor->plan_state_.get(),
         ExternalQueryOutputSpec{
             *query_mem_desc,
             target_exprs_to_infos(ra_exe_unit_.target_exprs, *query_mem_desc),
             executor});
     shared_context.addDeviceResults(std::move(device_results_), outer_tab_frag_ids);
     return;
   }
   const CompilationResult& compilation_result = query_comp_desc.getCompilationResult();
   std::unique_ptr<QueryExecutionContext> query_exe_context_owned;
   const bool do_render = render_info_ && render_info_->isPotentialInSituRender();
 
   int64_t total_num_input_rows{-1};
   if (kernel_dispatch_mode == ExecutorDispatchMode::KernelPerFragment &&
       query_mem_desc.getQueryDescriptionType() == QueryDescriptionType::Projection) {
     total_num_input_rows = 0;
     std::for_each(fetch_result.num_rows.begin(),
                   fetch_result.num_rows.end(),
                   [&total_num_input_rows](const std::vector<int64_t>& frag_row_count) {
                     total_num_input_rows = std::accumulate(frag_row_count.begin(),
                                                            frag_row_count.end(),
                                                            total_num_input_rows);
                   });
     VLOG(2) << "total_num_input_rows=" << total_num_input_rows;
     // TODO(adb): we may want to take this early out for all queries, but we are most
     // likely to see this query pattern on the kernel per fragment path (e.g. with HAVING
     // 0=1)
     if (total_num_input_rows == 0) {
       return;
     }
 
     if (query_has_inner_join(ra_exe_unit_)) {
       total_num_input_rows *= ra_exe_unit_.input_descs.size();
     }
   }
 
   if (eo.executor_type == ExecutorType::Native) {
     try {
       query_exe_context_owned =
           query_mem_desc.getQueryExecutionContext(ra_exe_unit_,
                                                   executor,
                                                   chosen_device_type,
                                                   kernel_dispatch_mode,
                                                   chosen_device_id,
                                                   total_num_input_rows,
                                                   fetch_result.col_buffers,
                                                   fetch_result.frag_offsets,
                                                   executor->getRowSetMemoryOwner(),
                                                   compilation_result.output_columnar,
                                                   query_mem_desc.sortOnGpu(),
                                                   do_render ? render_info_ : nullptr);
     } catch (const OutOfHostMemory& e) {
       throw QueryExecutionError(Executor::ERR_OUT_OF_CPU_MEM);
     }
   }
   QueryExecutionContext* query_exe_context{query_exe_context_owned.get()};
   CHECK(query_exe_context);
   int32_t err{0};
   uint32_t start_rowid{0};
   if (rowid_lookup_key >= 0) {
     if (!frag_list.empty()) {
       const auto& all_frag_row_offsets = shared_context.getFragOffsets();
       start_rowid = rowid_lookup_key -
                     all_frag_row_offsets[frag_list.begin()->fragment_ids.front()];
     }
   }
 
   if (ra_exe_unit_.groupby_exprs.empty()) {
     err = executor->executePlanWithoutGroupBy(ra_exe_unit_,
                                               compilation_result,
                                               query_comp_desc.hoistLiterals(),
                                               device_results_,
                                               ra_exe_unit_.target_exprs,
                                               chosen_device_type,
                                               fetch_result.col_buffers,
                                               query_exe_context,
                                               fetch_result.num_rows,
                                               fetch_result.frag_offsets,
                                               &catalog->getDataMgr(),
                                               chosen_device_id,
                                               start_rowid,
                                               ra_exe_unit_.input_descs.size(),
                                               do_render ? render_info_ : nullptr);
   } else {
     if (ra_exe_unit_.union_all) {
       VLOG(1) << "outer_table_id=" << outer_table_id
               << " ra_exe_unit_.scan_limit=" << ra_exe_unit_.scan_limit;
     }
     err = executor->executePlanWithGroupBy(ra_exe_unit_,
                                            compilation_result,
                                            query_comp_desc.hoistLiterals(),
                                            device_results_,
                                            chosen_device_type,
                                            fetch_result.col_buffers,
                                            outer_tab_frag_ids,
                                            query_exe_context,
                                            fetch_result.num_rows,
                                            fetch_result.frag_offsets,
                                            &catalog->getDataMgr(),
                                            chosen_device_id,
                                            outer_table_id,
                                            ra_exe_unit_.scan_limit,
                                            start_rowid,
                                            ra_exe_unit_.input_descs.size(),
                                            do_render ? render_info_ : nullptr);
   }
   if (device_results_) {
     std::list<std::shared_ptr<Chunk_NS::Chunk>> chunks_to_hold;
     for (const auto& chunk : chunks) {
       if (need_to_hold_chunk(chunk.get(), ra_exe_unit_)) {
         chunks_to_hold.push_back(chunk);
       }
     }
     device_results_->holdChunks(chunks_to_hold);
     device_results_->holdChunkIterators(chunk_iterators_ptr);
   } else {
     VLOG(1) << "null device_results.";
   }
   if (err) {
     throw QueryExecutionError(err);
   }
   shared_context.addDeviceResults(std::move(device_results_), outer_tab_frag_ids);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

int ExecutionKernel::chosen_device_id

private

Definition at line 79 of file ExecutionKernel.h.

Referenced by runImpl().

const ExecutorDeviceType ExecutionKernel::chosen_device_type

private

Definition at line 78 of file ExecutionKernel.h.

Referenced by runImpl().

const ColumnFetcher& ExecutionKernel::column_fetcher

private

Definition at line 81 of file ExecutionKernel.h.

Referenced by runImpl().

ResultSetPtr ExecutionKernel::device_results_

private

Definition at line 89 of file ExecutionKernel.h.

Referenced by runImpl().

const ExecutionOptions& ExecutionKernel::eo

private

Definition at line 80 of file ExecutionKernel.h.

Referenced by runImpl().

const FragmentsList ExecutionKernel::frag_list

private

Definition at line 84 of file ExecutionKernel.h.

Referenced by runImpl().

const ExecutorDispatchMode ExecutionKernel::kernel_dispatch_mode

private

Definition at line 85 of file ExecutionKernel.h.

Referenced by run(), and runImpl().

const QueryCompilationDescriptor& ExecutionKernel::query_comp_desc

private

Definition at line 82 of file ExecutionKernel.h.

Referenced by runImpl().

const QueryMemoryDescriptor& ExecutionKernel::query_mem_desc

private

Definition at line 83 of file ExecutionKernel.h.

Referenced by run(), and runImpl().

const RelAlgExecutionUnit& ExecutionKernel::ra_exe_unit_

private

Definition at line 77 of file ExecutionKernel.h.

Referenced by runImpl().

RenderInfo* ExecutionKernel::render_info_

private

Definition at line 86 of file ExecutionKernel.h.

Referenced by runImpl().

const int64_t ExecutionKernel::rowid_lookup_key

private

Definition at line 87 of file ExecutionKernel.h.

Referenced by runImpl().

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

/home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/ExecutionKernel.h
/home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/ExecutionKernel.cpp

Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation