173 const auto& outer_tab_frag_ids =
frag_list[0].fragment_ids;
178 auto data_mgr = executor->getDataMgr();
181 auto chunk_iterators_ptr = std::make_shared<std::list<ChunkIter>>();
182 std::list<std::shared_ptr<Chunk_NS::Chunk>> chunks;
183 std::unique_ptr<std::lock_guard<std::mutex>> gpu_lock;
184 std::unique_ptr<CudaAllocator> device_allocator;
187 new std::lock_guard<std::mutex>(executor->gpu_exec_mutex_[
chosen_device_id]));
188 device_allocator = std::make_unique<CudaAllocator>(
191 std::shared_ptr<FetchResult> fetch_result(
new FetchResult);
193 std::map<shared::TableKey, const TableFragments*> all_tables_fragments;
202 all_tables_fragments,
204 *chunk_iterators_ptr,
206 device_allocator.get(),
213 all_tables_fragments,
215 *chunk_iterators_ptr,
217 device_allocator.get(),
220 if (fetch_result->num_rows.empty()) {
227 LOG(
INFO) <<
"Dynamic Watchdog budget: CPU: "
243 throw std::runtime_error(
"Joins not supported through external execution");
250 executor->row_set_mem_owner_,
253 group_by_and_aggregate.initQueryMemoryDescriptor(
false, 0, 8,
nullptr,
false);
257 executor->plan_state_.get(),
266 std::unique_ptr<QueryExecutionContext> query_exe_context_owned;
269 int64_t total_num_input_rows{-1};
272 total_num_input_rows = 0;
273 std::for_each(fetch_result->num_rows.begin(),
274 fetch_result->num_rows.end(),
275 [&total_num_input_rows](
const std::vector<int64_t>& frag_row_count) {
277 frag_row_count.end(),
278 total_num_input_rows);
280 VLOG(2) <<
"total_num_input_rows=" << total_num_input_rows;
284 if (total_num_input_rows == 0) {
293 uint32_t start_rowid{0};
296 const auto& all_frag_row_offsets = shared_context.
getFragOffsets();
298 all_frag_row_offsets[
frag_list.begin()->fragment_ids.front()];
303 bool can_run_subkernels = shared_context.getThreadPool() !=
nullptr;
320 can_run_subkernels &&
326 if (can_run_subkernels) {
327 size_t total_rows = fetch_result->num_rows[0][0];
330 for (
size_t sub_start = start_rowid; sub_start < total_rows; sub_start += sub_size) {
331 sub_size = (sub_start + sub_size > total_rows) ? total_rows - sub_start : sub_size;
332 auto subtask = std::make_shared<KernelSubtask>(*
this,
336 total_num_input_rows,
340 shared_context.getThreadPool()->run(
341 [subtask, executor] { subtask->run(executor); });
355 query_exe_context_owned =
362 total_num_input_rows,
363 fetch_result->col_buffers,
364 fetch_result->frag_offsets,
365 executor->getRowSetMemoryOwner(),
367 query_mem_desc.sortOnGpu(),
375 CHECK(query_exe_context);
377 bool optimize_cuda_block_and_grid_sizes =
388 fetch_result->col_buffers,
390 fetch_result->num_rows,
391 fetch_result->frag_offsets,
398 optimize_cuda_block_and_grid_sizes);
401 VLOG(1) <<
"outer_table_key=" << outer_table_key
409 fetch_result->col_buffers,
412 fetch_result->num_rows,
413 fetch_result->frag_offsets,
422 optimize_cuda_block_and_grid_sizes);
425 std::list<std::shared_ptr<Chunk_NS::Chunk>> chunks_to_hold;
426 for (
const auto& chunk : chunks) {
431 chunks_to_hold.push_back(chunk);
437 VLOG(1) <<
"null device_results.";
442 shared_context.addDeviceResults(std::move(
device_results_), outer_tab_frag_ids);
bool need_to_hold_chunk(const Chunk_NS::Chunk *chunk, const RelAlgExecutionUnit &ra_exe_unit, const std::vector< ColumnLazyFetchInfo > &lazy_fetch_info, const ExecutorDeviceType device_type)
std::vector< Analyzer::Expr * > target_exprs
std::atomic_flag dynamic_watchdog_set
const ExecutionOptions & eo
size_t g_cpu_sub_task_size
const std::vector< uint64_t > & getFragOffsets()
static const int max_gpu_count
bool with_dynamic_watchdog
const std::optional< bool > union_all
const ExecutorDispatchMode kernel_dispatch_mode
const RelAlgExecutionUnit & ra_exe_unit_
const int64_t rowid_lookup_key
void addDeviceResults(ResultSetPtr &&device_results, std::vector< size_t > outer_table_fragment_ids)
std::vector< InputDescriptor > input_descs
const ExecutorDeviceType chosen_device_type
bool hoistLiterals() const
const std::list< std::shared_ptr< Analyzer::Expr > > groupby_exprs
std::unique_ptr< ResultSet > run_query_external(const ExecutionUnitSql &sql, const FetchResult &fetch_result, const PlanState *plan_state, const ExternalQueryOutputSpec &output_spec)
RenderInfo * render_info_
ExecutorType executor_type
const QueryMemoryDescriptor & query_mem_desc
DEVICE auto accumulate(ARGS &&...args)
const QueryCompilationDescriptor & query_comp_desc
static void computeAllTablesFragments(std::map< shared::TableKey, const TableFragments * > &all_tables_fragments, const RelAlgExecutionUnit &ra_exe_unit, const std::vector< InputTableInfo > &query_infos)
const std::shared_ptr< Analyzer::Estimator > estimator
static const int32_t ERR_OUT_OF_GPU_MEM
QueryDescriptionType getQueryDescriptionType() const
RUNTIME_EXPORT uint64_t dynamic_watchdog_init(unsigned ms_budget)
const FragmentsList frag_list
ExecutionUnitSql serialize_to_sql(const RelAlgExecutionUnit *ra_exe_unit)
CUstream getQueryEngineCudaStreamForDevice(int device_num)
bool optimize_cuda_block_and_grid_sizes
bool query_has_inner_join(const RelAlgExecutionUnit &ra_exe_unit)
const std::vector< InputTableInfo > & getQueryInfos() const
ResultSetPtr device_results_
std::vector< TargetInfo > target_exprs_to_infos(const std::vector< Analyzer::Expr * > &targets, const QueryMemoryDescriptor &query_mem_desc)
unsigned dynamic_watchdog_time_limit
bool allow_runtime_query_interrupt
auto getCompilationResult() const
static const int32_t ERR_OUT_OF_CPU_MEM
const ColumnFetcher & column_fetcher