33 int8_t* literal_buffer,
38 std::memcpy(literal_buffer + offset, &d.
floatval,
sizeof(
float));
41 std::memcpy(literal_buffer + offset, &d.
doubleval,
sizeof(
double));
49 std::memcpy(literal_buffer + offset, &d.
tinyintval,
sizeof(int8_t));
52 std::memcpy(literal_buffer + offset, &d.
smallintval,
sizeof(int16_t));
55 std::memcpy(literal_buffer + offset, &d.
intval,
sizeof(int32_t));
58 std::memcpy(literal_buffer + offset, &d.
bigintval,
sizeof(int64_t));
64 std::memcpy(literal_buffer + offset, &d.
boolval,
sizeof(int8_t));
67 std::memcpy(literal_buffer + offset, &string_size,
sizeof(int64_t));
69 literal_buffer + offset +
sizeof(int64_t), d.
stringval->data(), string_size);
72 " is not yet supported.");
77 size_t input_element_count) {
78 size_t allocated_output_row_count = 0;
87 allocated_output_row_count =
92 allocated_output_row_count = input_element_count;
99 return allocated_output_row_count;
106 const std::vector<InputTableInfo>& table_infos,
107 const std::shared_ptr<CompilationContext>& compilation_context,
111 bool is_pre_launch_udtf) {
113 CHECK(compilation_context);
114 std::vector<std::shared_ptr<Chunk_NS::Chunk>> chunks_owner;
115 std::vector<std::unique_ptr<char[]>> literals_owner;
117 const int device_id = 0;
118 std::unique_ptr<CudaAllocator> device_allocator;
120 auto data_mgr = executor->getDataMgr();
124 std::vector<const int8_t*> col_buf_ptrs;
125 std::vector<int64_t> col_sizes;
126 std::vector<const int8_t*> input_str_dict_proxy_ptrs;
127 std::optional<size_t> input_num_rows;
132 std::vector<std::vector<const int8_t*>> col_list_bufs;
133 std::vector<std::vector<const int8_t*>> input_col_list_str_dict_proxy_ptrs;
134 for (
const auto& input_expr : exe_unit.
input_exprs) {
135 auto ti = input_expr->get_type_info();
136 if (!ti.is_column_list()) {
139 if (
auto col_var = dynamic_cast<Analyzer::ColumnVar*>(input_expr)) {
140 CHECK(ti.is_column_list() || ti.is_column()) <<
"ti=" << ti;
141 const auto& table_key = col_var->getTableKey();
142 auto table_info_it = std::find_if(
143 table_infos.begin(), table_infos.end(), [&table_key](
const auto& table_info) {
144 return table_info.table_key == table_key;
146 CHECK(table_info_it != table_infos.end());
150 table_info_it->info.fragments.front(),
154 device_allocator.get(),
160 if (!input_num_rows) {
161 input_num_rows = (buf_elem_count > 0 ? buf_elem_count : 1);
164 int8_t* input_str_dict_proxy_ptr =
nullptr;
165 if (ti.is_subtype_dict_encoded_string()) {
166 const auto input_string_dictionary_proxy = executor->getStringDictionaryProxy(
167 ti.getStringDictKey(), executor->getRowSetMemoryOwner(),
true);
168 input_str_dict_proxy_ptr =
169 reinterpret_cast<int8_t*
>(input_string_dictionary_proxy);
171 if (ti.is_column_list()) {
172 if (col_index == -1) {
173 col_list_bufs.emplace_back();
174 input_col_list_str_dict_proxy_ptrs.emplace_back();
175 col_list_bufs.back().reserve(ti.get_dimension());
176 input_col_list_str_dict_proxy_ptrs.back().reserve(ti.get_dimension());
178 CHECK_EQ(col_sizes.back(), buf_elem_count);
181 col_list_bufs.back().push_back(col_buf);
182 input_col_list_str_dict_proxy_ptrs.back().push_back(input_str_dict_proxy_ptr);
184 if (col_index + 1 == ti.get_dimension()) {
188 col_buf_ptrs.push_back((
const int8_t*)col_list_bufs.back().data());
189 input_str_dict_proxy_ptrs.push_back(
190 (
const int8_t*)input_col_list_str_dict_proxy_ptrs.back().data());
192 col_buf_ptrs.push_back(col_buf);
193 input_str_dict_proxy_ptrs.push_back(input_str_dict_proxy_ptr);
195 col_sizes.push_back(buf_elem_count);
198 col_sizes.push_back(0);
199 input_str_dict_proxy_ptrs.push_back(
nullptr);
200 size_t literal_buffer_size = 0;
201 int8_t* cpu_literal_buf_ptr =
nullptr;
203 if (
const auto& constant_val = dynamic_cast<Analyzer::Constant*>(input_expr)) {
206 const auto const_val_datum = constant_val->get_constval();
207 const auto& ti = constant_val->get_type_info();
208 if (ti.is_text_encoding_none()) {
218 literal_buffer_size =
219 sizeof(int64_t) + ((const_val_datum.stringval->size() + 7) / 8) * 8;
224 literals_owner.emplace_back(std::make_unique<
char[]>(literal_buffer_size));
225 cpu_literal_buf_ptr =
reinterpret_cast<int8_t*
>(literals_owner.back().get());
227 }
else if (
const auto& array_expr =
228 dynamic_cast<Analyzer::ArrayExpr*>(input_expr)) {
229 const auto& ti = input_expr->get_type_info().get_elem_type();
239 int64_t size = array_expr->getElementCount();
240 int64_t
is_null = (array_expr->isNull() ? 0xffffffffffffffff : 0);
243 literal_buffer_size = 2 *
sizeof(int64_t) + (((size + 7) / 8) * 8) * elem_size;
244 literals_owner.emplace_back(std::make_unique<
char[]>(literal_buffer_size));
245 cpu_literal_buf_ptr =
reinterpret_cast<int8_t*
>(literals_owner.back().get());
246 std::memcpy(cpu_literal_buf_ptr, &size,
sizeof(int64_t));
247 std::memcpy(cpu_literal_buf_ptr +
sizeof(int64_t), &is_null,
sizeof(int64_t));
248 for (int64_t i = 0; i < size; i++) {
249 if (
const auto& constant_val =
250 dynamic_cast<const Analyzer::Constant*>(array_expr->getElement(i))) {
251 const auto const_val_datum = constant_val->get_constval();
255 sizeof(int64_t) * 2 + i * elem_size);
262 input_expr->toString() +
263 "\n Only literal constants and columns are supported!");
266 auto* gpu_allocator = device_allocator.get();
267 const auto gpu_literal_buf_ptr = gpu_allocator->alloc(literal_buffer_size);
268 gpu_allocator->copyToDevice(
269 gpu_literal_buf_ptr, cpu_literal_buf_ptr, literal_buffer_size);
270 col_buf_ptrs.push_back(gpu_literal_buf_ptr);
273 col_buf_ptrs.push_back(cpu_literal_buf_ptr);
287 CHECK(input_num_rows);
289 std::vector<int8_t*> output_str_dict_proxy_ptrs;
291 int8_t* output_str_dict_proxy_ptr =
nullptr;
292 auto ti = output_expr->get_type_info();
293 if (ti.is_dict_encoded_string()) {
294 const auto output_string_dictionary_proxy = executor->getStringDictionaryProxy(
295 ti.getStringDictKey(), executor->getRowSetMemoryOwner(),
true);
296 output_str_dict_proxy_ptr =
297 reinterpret_cast<int8_t*
>(output_string_dictionary_proxy);
299 output_str_dict_proxy_ptrs.emplace_back(output_str_dict_proxy_ptr);
302 if (is_pre_launch_udtf) {
306 std::dynamic_pointer_cast<CpuCompilationContext>(compilation_context),
309 input_str_dict_proxy_ptrs,
314 switch (device_type) {
318 std::dynamic_pointer_cast<CpuCompilationContext>(compilation_context),
321 input_str_dict_proxy_ptrs,
323 output_str_dict_proxy_ptrs,
328 std::dynamic_pointer_cast<GpuCompilationContext>(compilation_context),
331 input_str_dict_proxy_ptrs,
333 output_str_dict_proxy_ptrs,
346 const std::shared_ptr<CpuCompilationContext>& compilation_context,
347 std::vector<const int8_t*>& col_buf_ptrs,
348 std::vector<int64_t>& col_sizes,
349 std::vector<const int8_t*>& input_str_dict_proxy_ptrs,
350 const size_t elem_count,
351 Executor* executor) {
353 int64_t output_row_count = 0;
359 auto mgr = std::make_unique<TableFunctionManager>(
364 !exe_unit.table_func.usesManager());
368 const auto byte_stream_ptr = !col_buf_ptrs.empty()
369 ?
reinterpret_cast<const int8_t**
>(col_buf_ptrs.data())
371 if (!col_buf_ptrs.empty()) {
372 CHECK(byte_stream_ptr);
374 const auto col_sizes_ptr = !col_sizes.empty() ? col_sizes.data() :
nullptr;
375 if (!col_sizes.empty()) {
376 CHECK(col_sizes_ptr);
378 const auto input_str_dict_proxy_byte_stream_ptr =
379 !input_str_dict_proxy_ptrs.empty()
380 ?
reinterpret_cast<const int8_t**
>(input_str_dict_proxy_ptrs.data())
384 const auto err = compilation_context->table_function_entry_point()(
385 reinterpret_cast<const int8_t*
>(mgr.get()),
388 input_str_dict_proxy_byte_stream_ptr,
395 output_row_count = mgr->get_nrows();
397 if (exe_unit.table_func.hasPreFlightOutputSizer()) {
398 exe_unit.output_buffer_size_param = output_row_count;
404 std::string(mgr->get_error_message()));
496 const std::shared_ptr<CpuCompilationContext>& compilation_context,
497 std::vector<const int8_t*>& col_buf_ptrs,
498 std::vector<int64_t>& col_sizes,
499 std::vector<const int8_t*>& input_str_dict_proxy_ptrs,
500 const size_t elem_count,
501 std::vector<int8_t*>& output_str_dict_proxy_ptrs,
502 Executor* executor) {
504 int64_t output_row_count = 0;
510 auto mgr = std::make_unique<TableFunctionManager>(
515 !exe_unit.table_func.usesManager());
517 if (exe_unit.table_func.hasOutputSizeKnownPreLaunch()) {
522 }
else if (exe_unit.table_func.hasPreFlightOutputSizer()) {
523 output_row_count = exe_unit.output_buffer_size_param;
528 const auto byte_stream_ptr = !col_buf_ptrs.empty()
529 ?
reinterpret_cast<const int8_t**
>(col_buf_ptrs.data())
531 if (!col_buf_ptrs.empty()) {
532 CHECK(byte_stream_ptr);
534 const auto col_sizes_ptr = !col_sizes.empty() ? col_sizes.data() :
nullptr;
535 if (!col_sizes.empty()) {
536 CHECK(col_sizes_ptr);
538 const auto input_str_dict_proxy_byte_stream_ptr =
539 !input_str_dict_proxy_ptrs.empty()
540 ?
reinterpret_cast<const int8_t**
>(input_str_dict_proxy_ptrs.data())
543 const auto output_str_dict_proxy_byte_stream_ptr =
544 !output_str_dict_proxy_ptrs.empty()
545 ?
reinterpret_cast<int8_t**
>(output_str_dict_proxy_ptrs.data())
551 err = compilation_context->table_function_entry_point()(
552 reinterpret_cast<const int8_t*
>(mgr.get()),
555 input_str_dict_proxy_byte_stream_ptr,
557 output_str_dict_proxy_byte_stream_ptr,
559 }
catch (std::exception
const& e) {
561 std::string(e.what()));
567 output_row_count = mgr->get_nrows();
570 std::string(mgr->get_error_message()));
578 if (exe_unit.table_func.hasCompileTimeOutputSizeConstant()) {
579 if (static_cast<size_t>(output_row_count) != mgr->get_nrows()) {
581 "Table function with constant sizing parameter must return " +
586 if (output_row_count < 0 || (
size_t)output_row_count > mgr->get_nrows()) {
587 output_row_count = mgr->get_nrows();
591 if (exe_unit.table_func.hasTableFunctionSpecifiedParameter() && !mgr->query_buffers) {
593 if (output_row_count == 0) {
595 mgr->allocate_output_buffers(0);
601 mgr->query_buffers->getResultSet(0)->updateStorageEntryCount(output_row_count);
603 auto group_by_buffers_ptr = mgr->query_buffers->getGroupByBuffersPtr();
604 CHECK(group_by_buffers_ptr);
605 auto output_buffers_ptr =
reinterpret_cast<int64_t*
>(group_by_buffers_ptr[0]);
607 auto num_out_columns = exe_unit.target_exprs.size();
608 int8_t* src =
reinterpret_cast<int8_t*
>(output_buffers_ptr);
609 int8_t* dst =
reinterpret_cast<int8_t*
>(output_buffers_ptr);
613 for (
size_t col_idx = 0; col_idx < num_out_columns; col_idx++) {
614 auto ti = exe_unit.target_exprs[col_idx]->get_type_info();
615 if (ti.usesFlatBuffer()) {
620 CHECK_EQ(mgr->get_nrows(), output_row_count);
623 const size_t actual_column_size = allocated_column_size;
626 if (ti.is_text_encoding_dict_array()) {
627 const auto* ti_lite =
633 const size_t target_width = ti.get_size();
634 const size_t allocated_column_size = target_width * mgr->get_nrows();
635 const size_t actual_column_size = target_width * output_row_count;
637 auto t = memmove(dst, src, actual_column_size);
644 return mgr->query_buffers->getResultSetOwned(0);
663 const std::shared_ptr<GpuCompilationContext>& compilation_context,
664 std::vector<const int8_t*>& col_buf_ptrs,
665 std::vector<int64_t>& col_sizes,
666 std::vector<const int8_t*>& input_str_dict_proxy_ptrs,
667 const size_t elem_count,
668 std::vector<int8_t*>& output_str_dict_proxy_ptrs,
670 Executor* executor) {
678 auto data_mgr = executor->getDataMgr();
679 auto gpu_allocator = std::make_unique<CudaAllocator>(
681 CHECK(gpu_allocator);
689 reinterpret_cast<CUdeviceptr>(gpu_allocator->alloc(
sizeof(int8_t*)));
692 auto byte_stream_ptr = !(col_buf_ptrs.empty())
693 ? gpu_allocator->alloc(col_buf_ptrs.size() *
sizeof(int64_t))
695 if (byte_stream_ptr) {
696 gpu_allocator->copyToDevice(byte_stream_ptr,
697 reinterpret_cast<int8_t*>(col_buf_ptrs.data()),
698 col_buf_ptrs.size() *
sizeof(int64_t));
702 auto col_sizes_ptr = !(col_sizes.empty())
703 ? gpu_allocator->alloc(col_sizes.size() *
sizeof(int64_t))
706 gpu_allocator->copyToDevice(col_sizes_ptr,
707 reinterpret_cast<int8_t*>(col_sizes.data()),
708 col_sizes.size() *
sizeof(int64_t));
713 reinterpret_cast<CUdeviceptr>(gpu_allocator->alloc(
sizeof(int32_t)));
718 for (
size_t i = 0; i < num_out_columns; i++) {
719 const size_t col_width = exe_unit.
target_exprs[i]->get_type_info().get_size();
720 query_mem_desc.
addColSlotInfo({std::make_tuple(col_width, col_width)});
723 auto query_buffers = std::make_unique<QueryMemoryInitializer>(
728 (allocated_output_row_count == 0 ? 1 : allocated_output_row_count),
729 std::vector<std::vector<const int8_t*>>{col_buf_ptrs},
730 std::vector<std::vector<uint64_t>>{{0}},
736 int64_t output_row_count = allocated_output_row_count;
739 reinterpret_cast<CUdeviceptr>(gpu_allocator->alloc(
sizeof(int64_t*)));
740 gpu_allocator->copyToDevice(reinterpret_cast<int8_t*>(kernel_params[
OUTPUT_ROW_COUNT]),
741 reinterpret_cast<int8_t*>(&output_row_count),
742 sizeof(output_row_count));
751 const unsigned block_size_x =
752 (exe_unit.table_func.isRuntime() ? 1 : executor->blockSize());
753 const unsigned block_size_y = 1;
754 const unsigned block_size_z = 1;
755 const unsigned grid_size_x =
756 (exe_unit.table_func.isRuntime() ? 1 : executor->gridSize());
757 const unsigned grid_size_y = 1;
758 const unsigned grid_size_z = 1;
760 auto gpu_output_buffers =
761 query_buffers->setupTableFunctionGpuBuffers(query_mem_desc,
772 std::vector<void*> param_ptrs;
773 for (
auto& param : kernel_params) {
774 param_ptrs.push_back(¶m);
779 CHECK(compilation_context);
780 const auto native_code = compilation_context->getNativeCode(device_id);
781 auto cu_func =
static_cast<CUfunction>(native_code.first);
783 VLOG(1) <<
"Launch GPU table function kernel compiled with the following block and "
785 << block_size_x <<
" and " << grid_size_x;
800 gpu_allocator->copyFromDevice(
801 reinterpret_cast<int8_t*>(&output_row_count),
802 reinterpret_cast<int8_t*>(kernel_params[OUTPUT_ROW_COUNT]),
804 if (exe_unit.table_func.hasNonUserSpecifiedOutputSize()) {
805 if (static_cast<size_t>(output_row_count) != allocated_output_row_count) {
807 "Table function with constant sizing parameter must return " +
812 if (output_row_count < 0 || (
size_t)output_row_count > allocated_output_row_count) {
813 output_row_count = allocated_output_row_count;
818 query_buffers->getResultSet(0)->updateStorageEntryCount(output_row_count);
821 query_buffers->copyFromTableFunctionGpuBuffers(data_mgr,
829 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)
std::string DatumToString(Datum d, const SQLTypeInfo &ti)
bool is_timestamp() const
std::vector< Analyzer::Expr * > input_exprs
const table_functions::TableFunction table_func
void checkCudaErrors(CUresult err)
unsigned long long CUdeviceptr
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)
CONSTEXPR DEVICE bool is_null(const T &value)
bool is_timeinterval() const
std::shared_ptr< RowSetMemoryOwner > row_set_mem_owner_
std::mutex TableFunctionManager_singleton_mutex
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)
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.
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, std::vector< const int8_t * > &input_str_dict_proxy_ptrs, const size_t elem_count, Executor *executor)
CUstream getQueryEngineCudaStreamForDevice(int device_num)
bool hasTableFunctionSpecifiedParameter() const
#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
bool is_text_encoding_none() const
OutputBufferSizeType getOutputRowSizeType() const
void append_literal_buffer(const Datum &d, const SQLTypeInfo &ti, int8_t *literal_buffer, int64_t offset)
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)
static int64_t getBufferSize(const void *buffer)