reduce copy in multifrag query mem init

Author: akroviakov

.gitignore

@@ -11,7 +11,7 @@ version.py
 __pycache__
 
 # Logs
-hdk_log/*
+*/hdk_log/*
 omniscidb/Calcite/java/calcite/hdk-log-java-test/*
 omniscidb/Calcite/java/calcite/dependency-reduced-pom.xml
 hdk-log-java-test

omniscidb/QueryEngine/QueryExecutionContext.cpp

@@ -124,9 +124,9 @@ ResultSetPtr QueryExecutionContext::groupBufferToDeinterleavedResults(
   auto deinterleaved_result_set =
       std::make_shared<ResultSet>(result_set->getTargetInfos(),
                                   std::vector<ColumnLazyFetchInfo>{},
-                                  std::vector<std::vector<const int8_t*>>{},
-                                  std::vector<std::vector<int64_t>>{},
-                                  std::vector<int64_t>{},
+                                  nullptr,
+                                  nullptr,
+                                  nullptr,
                                   ExecutorDeviceType::CPU,
                                   -1,
                                   deinterleaved_query_mem_desc,

omniscidb/QueryEngine/QueryMemoryInitializer.cpp

@@ -383,6 +383,12 @@ QueryMemoryInitializer::QueryMemoryInitializer(
     group_by_buffers_.push_back(varlen_output_buffer);
   }
 
+  ColBuffersPtr res_set_col_buf = std::make_shared<const ColBuffers>(col_buffers);
+  FragOffsetsPtr res_set_col_frag_offsets = std::make_shared<const FragOffsets>(
+      get_col_frag_offsets(ra_exe_unit.target_exprs, frag_offsets));
+  FragSizesPtr res_set_frag_sizes = std::make_shared<const FragSizes>(
+      get_consistent_frags_sizes(ra_exe_unit.target_exprs, consistent_frag_sizes));
+
   for (size_t i = 0; i < group_buffers_count; i += step) {
     auto group_by_buffer = alloc_group_by_buffer(
         actual_group_buffer_size, thread_idx_, row_set_mem_owner_.get());
@@ -419,19 +425,14 @@ QueryMemoryInitializer::QueryMemoryInitializer(
       group_by_buffers_.push_back(nullptr);
     }
 
-    const auto column_frag_offsets =
-        get_col_frag_offsets(ra_exe_unit.target_exprs, frag_offsets);
-    const auto column_frag_sizes =
-        get_consistent_frags_sizes(ra_exe_unit.target_exprs, consistent_frag_sizes);
-
     result_sets_.emplace_back(new ResultSet(
         target_exprs_to_infos(ra_exe_unit.target_exprs,
                               query_mem_desc,
                               executor->getConfig().exec.group_by.bigint_count),
         executor->getColLazyFetchInfo(ra_exe_unit.target_exprs),
-        col_buffers,
-        column_frag_offsets,
-        column_frag_sizes,
+        res_set_col_buf,
+        res_set_col_frag_offsets,
+        res_set_frag_sizes,
         device_type,
         device_id,
         ResultSet::fixupQueryMemoryDescriptor(query_mem_desc),

omniscidb/ResultSet/ResultSet.cpp

@@ -84,9 +84,9 @@ ResultSet::ResultSet(const std::vector<TargetInfo>& targets,
 
 ResultSet::ResultSet(const std::vector<TargetInfo>& targets,
                      const std::vector<ColumnLazyFetchInfo>& lazy_fetch_info,
-                     const std::vector<std::vector<const int8_t*>>& col_buffers,
-                     const std::vector<std::vector<int64_t>>& frag_offsets,
-                     const std::vector<int64_t>& consistent_frag_sizes,
+                     const ColBuffersPtr& col_buffers,
+                     const FragOffsetsPtr& frag_offsets,
+                     const FragSizesPtr& consistent_frag_sizes,
                      const ExecutorDeviceType device_type,
                      const int device_id,
                      const QueryMemoryDescriptor& query_mem_desc,

omniscidb/ResultSet/ResultSet.h

@@ -161,6 +161,13 @@ using Permutation = std::vector<PermutationIdx>;
 using PermutationView = VectorView<PermutationIdx>;
 using ResultSetPtr = std::shared_ptr<ResultSet>;
 
+using FragSizes = std::vector<int64_t>;
+using FragSizesPtr = std::shared_ptr<const FragSizes>;
+using ColBuffers = std::vector<std::vector<const int8_t*>>;
+using ColBuffersPtr = std::shared_ptr<const ColBuffers>;
+using FragOffsets = std::vector<std::vector<int64_t>>;
+using FragOffsetsPtr = std::shared_ptr<const FragOffsets>;
+
 class ResultSet {
  public:
   friend ResultSetBuilder;
@@ -176,9 +183,9 @@ class ResultSet {
 
   ResultSet(const std::vector<TargetInfo>& targets,
             const std::vector<ColumnLazyFetchInfo>& lazy_fetch_info,
-            const std::vector<std::vector<const int8_t*>>& col_buffers,
-            const std::vector<std::vector<int64_t>>& frag_offsets,
-            const std::vector<int64_t>& consistent_frag_sizes,
+            const ColBuffersPtr& col_buffers,
+            const FragOffsetsPtr& frag_offsets,
+            const FragSizesPtr& consistent_frag_sizes,
             const ExecutorDeviceType device_type,
             const int device_id,
             const QueryMemoryDescriptor& query_mem_desc,
@@ -802,9 +809,9 @@ class ResultSet {
   //   setting offset instead of ptr in group by buffer.
   std::vector<std::vector<int8_t>> literal_buffers_;
   const std::vector<ColumnLazyFetchInfo> lazy_fetch_info_;
-  std::vector<std::vector<std::vector<const int8_t*>>> col_buffers_;
-  std::vector<std::vector<std::vector<int64_t>>> frag_offsets_;
-  std::vector<std::vector<int64_t>> consistent_frag_sizes_;
+  std::vector<ColBuffersPtr> col_buffers_;
+  std::vector<FragOffsetsPtr> frag_offsets_;
+  std::vector<FragSizesPtr> consistent_frag_sizes_;
 
   const std::shared_ptr<const hdk::ir::Estimator> estimator_;
   Data_Namespace::AbstractBuffer* device_estimator_buffer_{nullptr};

omniscidb/ResultSet/ResultSetIteration.cpp

@@ -930,24 +930,25 @@ const std::vector<const int8_t*>& ResultSet::getColumnFrag(const size_t storage_
                                                            const size_t col_logical_idx,
                                                            int64_t& global_idx) const {
   CHECK_LT(static_cast<size_t>(storage_idx), col_buffers_.size());
-  if (col_buffers_[storage_idx].size() > 1) {
+  if (col_buffers_[storage_idx]->size() > 1) {
     int64_t frag_id = 0;
     int64_t local_idx = global_idx;
-    if (consistent_frag_sizes_[storage_idx][col_logical_idx] != -1) {
-      frag_id = global_idx / consistent_frag_sizes_[storage_idx][col_logical_idx];
-      local_idx = global_idx % consistent_frag_sizes_[storage_idx][col_logical_idx];
+    const auto frag_size = (*(consistent_frag_sizes_[storage_idx]))[col_logical_idx];
+    if (frag_size != -1) {
+      frag_id = global_idx / frag_size;
+      local_idx = global_idx % frag_size;
     } else {
       std::tie(frag_id, local_idx) = get_frag_id_and_local_idx(
-          frag_offsets_[storage_idx], col_logical_idx, global_idx);
+          *(frag_offsets_[storage_idx]), col_logical_idx, global_idx);
       CHECK_LE(local_idx, global_idx);
     }
     CHECK_GE(frag_id, int64_t(0));
-    CHECK_LT(static_cast<size_t>(frag_id), col_buffers_[storage_idx].size());
+    CHECK_LT(static_cast<size_t>(frag_id), col_buffers_[storage_idx]->size());
     global_idx = local_idx;
-    return col_buffers_[storage_idx][frag_id];
+    return (*(col_buffers_[storage_idx]))[frag_id];
   } else {
-    CHECK_EQ(size_t(1), col_buffers_[storage_idx].size());
-    return col_buffers_[storage_idx][0];
+    CHECK_EQ(size_t(1), col_buffers_[storage_idx]->size());
+    return (*(col_buffers_[storage_idx]))[0];
   }
 }