34 std::vector<std::unique_ptr<
char[]>>& literals_owner,
37 switch (device_type) {
39 literals_owner.emplace_back(std::make_unique<
char[]>(
sizeof(int64_t)));
40 std::memcpy(literals_owner.back().get(), &literal,
sizeof(
T));
41 return reinterpret_cast<const int8_t*
>(literals_owner.back().get());
45 const auto gpu_literal_buf_ptr = gpu_allocator->
alloc(
sizeof(int64_t));
47 gpu_literal_buf_ptr, reinterpret_cast<const int8_t*>(&literal),
sizeof(
T));
48 return gpu_literal_buf_ptr;
61 std::vector<std::unique_ptr<
char[]>>& literals_owner,
63 const int64_t string_size = literal->size();
64 const int64_t padded_string_size =
65 (string_size + 7) / 8 * 8;
66 switch (device_type) {
68 literals_owner.emplace_back(
69 std::make_unique<
char[]>(
sizeof(int64_t) + padded_string_size));
70 std::memcpy(literals_owner.back().get(), &string_size,
sizeof(int64_t));
72 literals_owner.back().get() +
sizeof(int64_t), literal->data(), string_size);
73 return reinterpret_cast<const int8_t*
>(literals_owner.back().get());
77 const auto gpu_literal_buf_ptr =
78 gpu_allocator->
alloc(
sizeof(int64_t) + padded_string_size);
80 reinterpret_cast<const int8_t*>(&string_size),
82 gpu_allocator->
copyToDevice(gpu_literal_buf_ptr +
sizeof(int64_t),
83 reinterpret_cast<const int8_t*>(literal->data()),
85 return gpu_literal_buf_ptr;
93 size_t input_element_count) {
94 size_t allocated_output_row_count = 0;
103 allocated_output_row_count =
108 allocated_output_row_count = input_element_count;
115 return allocated_output_row_count;
122 const std::vector<InputTableInfo>& table_infos,
123 const std::shared_ptr<CompilationContext>& compilation_context,
127 bool is_pre_launch_udtf) {
129 CHECK(compilation_context);
130 std::vector<std::shared_ptr<Chunk_NS::Chunk>> chunks_owner;
131 std::vector<std::unique_ptr<char[]>> literals_owner;
133 const int device_id = 0;
134 std::unique_ptr<CudaAllocator> device_allocator;
136 auto data_mgr = executor->getDataMgr();
140 std::vector<const int8_t*> col_buf_ptrs;
141 std::vector<int64_t> col_sizes;
142 std::vector<const int8_t*> input_str_dict_proxy_ptrs;
143 std::optional<size_t> input_num_rows;
148 std::vector<std::vector<const int8_t*>> col_list_bufs;
149 std::vector<std::vector<const int8_t*>> input_col_list_str_dict_proxy_ptrs;
150 for (
const auto& input_expr : exe_unit.
input_exprs) {
151 auto ti = input_expr->get_type_info();
152 if (!ti.is_column_list()) {
155 if (
auto col_var = dynamic_cast<Analyzer::ColumnVar*>(input_expr)) {
156 auto table_id = col_var->get_table_id();
157 auto table_info_it = std::find_if(
158 table_infos.begin(), table_infos.end(), [&table_id](
const auto& table_info) {
159 return table_info.table_id == table_id;
161 CHECK(table_info_it != table_infos.end());
165 table_info_it->info.fragments.front(),
169 device_allocator.get(),
175 if (!input_num_rows) {
176 input_num_rows = (buf_elem_count > 0 ? buf_elem_count : 1);
179 int8_t* input_str_dict_proxy_ptr =
nullptr;
180 if (ti.is_subtype_dict_encoded_string()) {
181 const auto input_string_dictionary_proxy = executor->getStringDictionaryProxy(
182 ti.get_comp_param(), executor->getRowSetMemoryOwner(),
true);
183 input_str_dict_proxy_ptr =
184 reinterpret_cast<int8_t*
>(input_string_dictionary_proxy);
186 if (ti.is_column_list()) {
187 if (col_index == -1) {
188 col_list_bufs.push_back({});
189 input_col_list_str_dict_proxy_ptrs.push_back({});
190 col_list_bufs.back().reserve(ti.get_dimension());
191 input_col_list_str_dict_proxy_ptrs.back().reserve(ti.get_dimension());
193 CHECK_EQ(col_sizes.back(), buf_elem_count);
196 col_list_bufs.back().push_back(col_buf);
197 input_col_list_str_dict_proxy_ptrs.back().push_back(input_str_dict_proxy_ptr);
199 if (col_index + 1 == ti.get_dimension()) {
203 col_buf_ptrs.push_back((
const int8_t*)col_list_bufs.back().data());
204 input_str_dict_proxy_ptrs.push_back(
205 (
const int8_t*)input_col_list_str_dict_proxy_ptrs.back().data());
207 col_buf_ptrs.push_back(col_buf);
208 input_str_dict_proxy_ptrs.push_back(input_str_dict_proxy_ptr);
210 col_sizes.push_back(buf_elem_count);
211 }
else if (
const auto& constant_val = dynamic_cast<Analyzer::Constant*>(input_expr)) {
214 col_sizes.push_back(0);
215 input_str_dict_proxy_ptrs.push_back(
nullptr);
216 const auto const_val_datum = constant_val->get_constval();
217 const auto& ti = constant_val->get_type_info();
224 device_allocator.get()));
230 device_allocator.get()));
235 }
else if (ti.is_integer() || ti.is_timestamp()) {
241 device_allocator.get()));
247 device_allocator.get()));
253 device_allocator.get()));
259 device_allocator.get()));
264 }
else if (ti.is_boolean()) {
268 device_allocator.get()));
269 }
else if (ti.is_bytes()) {
273 device_allocator.get()));
276 " is not yet supported.");
290 CHECK(input_num_rows);
292 std::vector<int8_t*> output_str_dict_proxy_ptrs;
294 int8_t* output_str_dict_proxy_ptr =
nullptr;
295 auto ti = output_expr->get_type_info();
296 if (ti.is_dict_encoded_string()) {
297 const auto output_string_dictionary_proxy = executor->getStringDictionaryProxy(
298 ti.get_comp_param(), executor->getRowSetMemoryOwner(),
true);
299 output_str_dict_proxy_ptr =
300 reinterpret_cast<int8_t*
>(output_string_dictionary_proxy);
302 output_str_dict_proxy_ptrs.emplace_back(output_str_dict_proxy_ptr);
305 if (is_pre_launch_udtf) {
309 std::dynamic_pointer_cast<CpuCompilationContext>(compilation_context),
316 switch (device_type) {
320 std::dynamic_pointer_cast<CpuCompilationContext>(compilation_context),
323 input_str_dict_proxy_ptrs,
325 output_str_dict_proxy_ptrs,
330 std::dynamic_pointer_cast<GpuCompilationContext>(compilation_context),
333 input_str_dict_proxy_ptrs,
335 output_str_dict_proxy_ptrs,
348 const std::shared_ptr<CpuCompilationContext>& compilation_context,
349 std::vector<const int8_t*>& col_buf_ptrs,
350 std::vector<int64_t>& col_sizes,
351 const size_t elem_count,
352 Executor* executor) {
354 int64_t output_row_count = 0;
360 auto mgr = std::make_unique<TableFunctionManager>(
365 !exe_unit.table_func.usesManager());
369 const auto byte_stream_ptr = !col_buf_ptrs.empty()
370 ?
reinterpret_cast<const int8_t**
>(col_buf_ptrs.data())
372 if (!col_buf_ptrs.empty()) {
373 CHECK(byte_stream_ptr);
375 const auto col_sizes_ptr = !col_sizes.empty() ? col_sizes.data() :
nullptr;
376 if (!col_sizes.empty()) {
377 CHECK(col_sizes_ptr);
381 const auto err = compilation_context->table_function_entry_point()(
382 reinterpret_cast<const int8_t*
>(mgr.get()),
390 if (exe_unit.table_func.hasPreFlightOutputSizer()) {
391 exe_unit.output_buffer_size_param = output_row_count;
396 std::string(mgr->get_error_message()));
405 const std::shared_ptr<CpuCompilationContext>& compilation_context,
406 std::vector<const int8_t*>& col_buf_ptrs,
407 std::vector<int64_t>& col_sizes,
408 std::vector<const int8_t*>& input_str_dict_proxy_ptrs,
409 const size_t elem_count,
410 std::vector<int8_t*>& output_str_dict_proxy_ptrs,
411 Executor* executor) {
413 int64_t output_row_count = 0;
419 auto mgr = std::make_unique<TableFunctionManager>(
424 !exe_unit.table_func.usesManager());
426 if (exe_unit.table_func.hasOutputSizeKnownPreLaunch()) {
431 }
else if (exe_unit.table_func.hasPreFlightOutputSizer()) {
432 output_row_count = exe_unit.output_buffer_size_param;
437 const auto byte_stream_ptr = !col_buf_ptrs.empty()
438 ?
reinterpret_cast<const int8_t**
>(col_buf_ptrs.data())
440 if (!col_buf_ptrs.empty()) {
441 CHECK(byte_stream_ptr);
443 const auto col_sizes_ptr = !col_sizes.empty() ? col_sizes.data() :
nullptr;
444 if (!col_sizes.empty()) {
445 CHECK(col_sizes_ptr);
447 const auto input_str_dict_proxy_byte_stream_ptr =
448 !input_str_dict_proxy_ptrs.empty()
449 ?
reinterpret_cast<const int8_t**
>(input_str_dict_proxy_ptrs.data())
452 const auto output_str_dict_proxy_byte_stream_ptr =
453 !output_str_dict_proxy_ptrs.empty()
454 ?
reinterpret_cast<int8_t**
>(output_str_dict_proxy_ptrs.data())
458 const auto err = compilation_context->table_function_entry_point()(
459 reinterpret_cast<const int8_t*
>(mgr.get()),
462 input_str_dict_proxy_byte_stream_ptr,
464 output_str_dict_proxy_byte_stream_ptr,
469 std::string(mgr->get_error_message()));
477 if (exe_unit.table_func.hasCompileTimeOutputSizeConstant()) {
478 if (static_cast<size_t>(output_row_count) != mgr->get_nrows()) {
480 "Table function with constant sizing parameter must return " +
485 if (output_row_count < 0 || (
size_t)output_row_count > mgr->get_nrows()) {
486 output_row_count = mgr->get_nrows();
490 if (exe_unit.table_func.hasTableFunctionSpecifiedParameter() && !mgr->query_buffers) {
492 if (output_row_count == 0) {
494 mgr->allocate_output_buffers(0);
500 mgr->query_buffers->getResultSet(0)->updateStorageEntryCount(output_row_count);
502 auto group_by_buffers_ptr = mgr->query_buffers->getGroupByBuffersPtr();
503 CHECK(group_by_buffers_ptr);
504 auto output_buffers_ptr =
reinterpret_cast<int64_t*
>(group_by_buffers_ptr[0]);
506 auto num_out_columns = exe_unit.target_exprs.size();
507 int8_t* src =
reinterpret_cast<int8_t*
>(output_buffers_ptr);
508 int8_t* dst =
reinterpret_cast<int8_t*
>(output_buffers_ptr);
511 for (
size_t col_idx = 0; col_idx < num_out_columns; col_idx++) {
512 auto ti = exe_unit.target_exprs[col_idx]->get_type_info();
518 CHECK_EQ(mgr->get_nrows(), output_row_count);
521 const size_t actual_column_size = allocated_column_size;
524 if (ti.is_text_encoding_dict_array()) {
525 CHECK_EQ(m.getDTypeMetadataDictId(),
526 ti.get_comp_param());
529 const size_t target_width = ti.get_size();
530 const size_t allocated_column_size = target_width * mgr->get_nrows();
531 const size_t actual_column_size = target_width * output_row_count;
533 auto t = memmove(dst, src, actual_column_size);
540 return mgr->query_buffers->getResultSetOwned(0);
559 const std::shared_ptr<GpuCompilationContext>& compilation_context,
560 std::vector<const int8_t*>& col_buf_ptrs,
561 std::vector<int64_t>& col_sizes,
562 std::vector<const int8_t*>& input_str_dict_proxy_ptrs,
563 const size_t elem_count,
564 std::vector<int8_t*>& output_str_dict_proxy_ptrs,
566 Executor* executor) {
574 auto data_mgr = executor->getDataMgr();
575 auto gpu_allocator = std::make_unique<CudaAllocator>(
577 CHECK(gpu_allocator);
585 reinterpret_cast<CUdeviceptr>(gpu_allocator->alloc(
sizeof(int8_t*)));
588 auto byte_stream_ptr = !(col_buf_ptrs.empty())
589 ? gpu_allocator->alloc(col_buf_ptrs.size() *
sizeof(int64_t))
591 if (byte_stream_ptr) {
592 gpu_allocator->copyToDevice(byte_stream_ptr,
593 reinterpret_cast<int8_t*>(col_buf_ptrs.data()),
594 col_buf_ptrs.size() *
sizeof(int64_t));
598 auto col_sizes_ptr = !(col_sizes.empty())
599 ? gpu_allocator->alloc(col_sizes.size() *
sizeof(int64_t))
602 gpu_allocator->copyToDevice(col_sizes_ptr,
603 reinterpret_cast<int8_t*>(col_sizes.data()),
604 col_sizes.size() *
sizeof(int64_t));
609 reinterpret_cast<CUdeviceptr>(gpu_allocator->alloc(
sizeof(int32_t)));
617 for (
size_t i = 0; i < num_out_columns; i++) {
618 const size_t col_width = exe_unit.
target_exprs[i]->get_type_info().get_size();
619 query_mem_desc.
addColSlotInfo({std::make_tuple(col_width, col_width)});
622 auto query_buffers = std::make_unique<QueryMemoryInitializer>(
627 (allocated_output_row_count == 0 ? 1 : allocated_output_row_count),
628 std::vector<std::vector<const int8_t*>>{col_buf_ptrs},
629 std::vector<std::vector<uint64_t>>{{0}},
635 int64_t output_row_count = allocated_output_row_count;
638 reinterpret_cast<CUdeviceptr>(gpu_allocator->alloc(
sizeof(int64_t*)));
639 gpu_allocator->copyToDevice(reinterpret_cast<int8_t*>(kernel_params[
OUTPUT_ROW_COUNT]),
640 reinterpret_cast<int8_t*>(&output_row_count),
641 sizeof(output_row_count));
650 const unsigned block_size_x =
651 (exe_unit.table_func.isRuntime() ? 1 : executor->blockSize());
652 const unsigned block_size_y = 1;
653 const unsigned block_size_z = 1;
654 const unsigned grid_size_x =
655 (exe_unit.table_func.isRuntime() ? 1 : executor->gridSize());
656 const unsigned grid_size_y = 1;
657 const unsigned grid_size_z = 1;
659 auto gpu_output_buffers =
660 query_buffers->setupTableFunctionGpuBuffers(query_mem_desc,
671 std::vector<void*> param_ptrs;
672 for (
auto& param : kernel_params) {
673 param_ptrs.push_back(¶m);
678 CHECK(compilation_context);
679 const auto native_code = compilation_context->getNativeCode(device_id);
680 auto cu_func =
static_cast<CUfunction>(native_code.first);
696 gpu_allocator->copyFromDevice(
697 reinterpret_cast<int8_t*>(&output_row_count),
698 reinterpret_cast<int8_t*>(kernel_params[OUTPUT_ROW_COUNT]),
700 if (exe_unit.table_func.hasNonUserSpecifiedOutputSize()) {
701 if (static_cast<size_t>(output_row_count) != allocated_output_row_count) {
703 "Table function with constant sizing parameter must return " +
708 if (output_row_count < 0 || (
size_t)output_row_count > allocated_output_row_count) {
709 output_row_count = allocated_output_row_count;
714 query_buffers->getResultSet(0)->updateStorageEntryCount(output_row_count);
717 query_buffers->copyFromTableFunctionGpuBuffers(data_mgr,
725 return query_buffers->getResultSetOwned(0);
Defines data structures for the semantic analysis phase of query processing.
size_t get_output_row_count(const TableFunctionExecutionUnit &exe_unit, size_t input_element_count)
void launchPreCodeOnCpu(const TableFunctionExecutionUnit &exe_unit, const std::shared_ptr< CpuCompilationContext > &compilation_context, std::vector< const int8_t * > &col_buf_ptrs, std::vector< int64_t > &col_sizes, const size_t elem_count, Executor *executor)
std::vector< Analyzer::Expr * > input_exprs
const table_functions::TableFunction table_func
void checkCudaErrors(CUresult err)
unsigned long long CUdeviceptr
void setOutputColumnar(const bool val)
ColumnCacheMap columnarized_table_cache_
std::shared_ptr< ResultSet > ResultSetPtr
size_t output_buffer_size_param
bool containsPreFlightFn() const
size_t get_bit_width(const SQLTypeInfo &ti)
std::shared_ptr< RowSetMemoryOwner > row_set_mem_owner_
std::mutex TableFunctionManager_singleton_mutex
void copyToDevice(void *device_dst, const void *host_src, const size_t num_bytes) const override
ResultSetPtr launchCpuCode(const TableFunctionExecutionUnit &exe_unit, const std::shared_ptr< CpuCompilationContext > &compilation_context, std::vector< const int8_t * > &col_buf_ptrs, std::vector< int64_t > &col_sizes, std::vector< const int8_t * > &input_str_dict_proxy_ptrs, const size_t elem_count, std::vector< int8_t * > &output_str_dict_proxy_ptrs, Executor *executor)
int8_t * alloc(const size_t num_bytes) override
ResultSetPtr execute(const TableFunctionExecutionUnit &exe_unit, const std::vector< InputTableInfo > &table_infos, const std::shared_ptr< CompilationContext > &compilation_context, const ColumnFetcher &column_fetcher, const ExecutorDeviceType device_type, Executor *executor, bool is_pre_launch_udtf)
static std::pair< const int8_t *, size_t > getOneColumnFragment(Executor *executor, const Analyzer::ColumnVar &hash_col, const Fragmenter_Namespace::FragmentInfo &fragment, const Data_Namespace::MemoryLevel effective_mem_lvl, const int device_id, DeviceAllocator *device_allocator, const size_t thread_idx, std::vector< std::shared_ptr< Chunk_NS::Chunk >> &chunks_owner, ColumnCacheMap &column_cache)
Gets one chunk's pointer and element count on either CPU or GPU.
CUstream getQueryEngineCudaStreamForDevice(int device_num)
int64_t flatbufferSize() const
bool hasTableFunctionSpecifiedParameter() const
const int8_t * create_literal_buffer(const T literal, const ExecutorDeviceType device_type, std::vector< std::unique_ptr< char[]>> &literals_owner, CudaAllocator *gpu_allocator)
#define DEBUG_TIMER(name)
std::vector< Analyzer::Expr * > target_exprs
void addColSlotInfo(const std::vector< std::tuple< int8_t, int8_t >> &slots_for_col)
bool hasOutputSizeIndependentOfInputSize() const
OutputBufferSizeType getOutputRowSizeType() const
ResultSetPtr launchGpuCode(const TableFunctionExecutionUnit &exe_unit, const std::shared_ptr< GpuCompilationContext > &compilation_context, std::vector< const int8_t * > &col_buf_ptrs, std::vector< int64_t > &col_sizes, std::vector< const int8_t * > &input_str_dict_proxy_ptrs, const size_t elem_count, std::vector< int8_t * > &output_str_dict_proxy_ptrs, const int device_id, Executor *executor)
FORCE_INLINE HOST DEVICE T align_to_int64(T addr)