OmniSciDB  72180abbfe
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Pages
StreamingTopN.cpp
Go to the documentation of this file.
1 /*
2  * Copyright 2017 MapD Technologies, Inc.
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at
7  *
8  * http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16 #include "StreamingTopN.h"
17 #include "RelAlgExecutor.h"
18 #include "Shared/checked_alloc.h"
19 #include "TopKSort.h"
20 
21 namespace streaming_top_n {
22 
23 size_t get_heap_size(const size_t row_size, const size_t n, const size_t thread_count) {
24  const auto row_size_quad = row_size / sizeof(int64_t);
25  return (1 + n + row_size_quad * n) * thread_count * sizeof(int64_t);
26 }
27 
28 size_t get_rows_offset_of_heaps(const size_t n, const size_t thread_count) {
29  return (1 + n) * thread_count * sizeof(int64_t);
30 }
31 
32 std::vector<int8_t> get_rows_copy_from_heaps(const int64_t* heaps,
33  const size_t heaps_size,
34  const size_t n,
35  const size_t thread_count) {
36  const auto rows_offset = streaming_top_n::get_rows_offset_of_heaps(n, thread_count);
37  const auto row_buff_size = heaps_size - rows_offset;
38  std::vector<int8_t> rows_copy(row_buff_size);
39  const auto rows_ptr = reinterpret_cast<const int8_t*>(heaps) + rows_offset;
40  std::memcpy(&rows_copy[0], rows_ptr, row_buff_size);
41  return rows_copy;
42 }
43 
44 } // namespace streaming_top_n
45 
46 size_t get_heap_key_slot_index(const std::vector<Analyzer::Expr*>& target_exprs,
47  const size_t target_idx) {
48  size_t slot_idx = 0;
49  for (size_t i = 0; i < target_idx; ++i) {
50  auto agg_info = get_target_info(target_exprs[i], g_bigint_count);
51  slot_idx = advance_slot(slot_idx, agg_info, false);
52  }
53  return slot_idx;
54 }
55 
56 #ifdef HAVE_CUDA
57 std::vector<int8_t> pick_top_n_rows_from_dev_heaps(
58  Data_Namespace::DataMgr* data_mgr,
59  const int64_t* dev_heaps_buffer,
60  const RelAlgExecutionUnit& ra_exe_unit,
62  const size_t thread_count,
63  const int device_id) {
64  CHECK(!query_mem_desc.canOutputColumnar());
65  CHECK_EQ(ra_exe_unit.sort_info.order_entries.size(), size_t(1));
66  const auto& only_oe = ra_exe_unit.sort_info.order_entries.back();
67  const auto oe_col_idx = only_oe.tle_no - 1;
68  const auto n = ra_exe_unit.sort_info.offset + ra_exe_unit.sort_info.limit;
69  const auto group_key_bytes = query_mem_desc.getEffectiveKeyWidth();
70  const PodOrderEntry pod_oe{only_oe.tle_no, only_oe.is_desc, only_oe.nulls_first};
71  const auto key_slot_idx = get_heap_key_slot_index(ra_exe_unit.target_exprs, oe_col_idx);
72  GroupByBufferLayoutInfo oe_layout{
73  n * thread_count,
74  query_mem_desc.getColOffInBytes(key_slot_idx),
75  static_cast<size_t>(query_mem_desc.getPaddedSlotWidthBytes(oe_col_idx)),
76  query_mem_desc.getRowSize(),
77  get_target_info(ra_exe_unit.target_exprs[oe_col_idx], g_bigint_count),
78  -1};
80  data_mgr,
81  dev_heaps_buffer,
82  query_mem_desc.getBufferSizeBytes(
83  ra_exe_unit, thread_count, ExecutorDeviceType::GPU),
84  n,
85  pod_oe,
86  oe_layout,
87  group_key_bytes,
88  thread_count,
89  device_id);
90 }
91 #endif // HAVE_CUDA
std::vector< Analyzer::Expr * > target_exprs
#define CHECK_EQ(x, y)
Definition: Logger.h:205
size_t getBufferSizeBytes(const RelAlgExecutionUnit &ra_exe_unit, const unsigned thread_count, const ExecutorDeviceType device_type) const
Streaming Top N algorithm.
TargetInfo get_target_info(const PointerType target_expr, const bool bigint_count)
Definition: TargetInfo.h:66
size_t get_rows_offset_of_heaps(const size_t n, const size_t thread_count)
const std::list< Analyzer::OrderEntry > order_entries
size_t getEffectiveKeyWidth() const
std::vector< int8_t > pop_n_rows_from_merged_heaps_gpu(Data_Namespace::DataMgr *data_mgr, const int64_t *dev_heaps, const size_t heaps_size, const size_t n, const PodOrderEntry &oe, const GroupByBufferLayoutInfo &layout, const size_t group_key_bytes, const size_t thread_count, const int device_id)
Definition: TopKSort.cu:287
size_t get_heap_key_slot_index(const std::vector< Analyzer::Expr * > &target_exprs, const size_t target_idx)
const size_t limit
size_t advance_slot(const size_t j, const TargetInfo &target_info, const bool separate_varlen_storage)
CHECK(cgen_state)
const SortInfo sort_info
bool g_bigint_count
const int8_t getPaddedSlotWidthBytes(const size_t slot_idx) const
int tle_no
std::vector< int8_t > get_rows_copy_from_heaps(const int64_t *heaps, const size_t heaps_size, const size_t n, const size_t thread_count)
size_t get_heap_size(const size_t row_size, const size_t n, const size_t thread_count)
const size_t offset
size_t getColOffInBytes(const size_t col_idx) const