functional correctness

Author: akroviakov

omniscidb/QueryEngine/Execute.cpp

@@ -2992,12 +2992,16 @@ FetchResult Executor::fetchChunks(
   if(memory_level == Data_Namespace::MemoryLevel::GPU_LEVEL){
     std::mutex all_frag;
     std::atomic<bool> empty_frags{false};
-    tbb::task_arena limitedArena(16);
+    tbb::task_arena limitedArena(1);
+    for (const auto& selected_frag_ids : frag_ids_crossjoin) {
+      selected_frag_ids_vec.push_back(selected_frag_ids);
+    }
+    all_frag_col_buffers.resize(selected_frag_ids_vec.size());
+
     limitedArena.execute([&]() {
-      tbb::parallel_for_each(
-          frag_ids_crossjoin.begin(),
-          frag_ids_crossjoin.end(),
-          [&](const std::vector<size_t>& selected_frag_ids) {
+      tbb::parallel_for(
+          0ul, selected_frag_ids_vec.size(), [&](const size_t idx) {
+            const auto& selected_frag_ids = selected_frag_ids_vec[idx];
             std::vector<const int8_t*> frag_col_buffers(
                 plan_state_->global_to_local_col_ids_.size());
             for (const auto& col_id : col_global_ids) {
@@ -3029,6 +3033,7 @@ FetchResult Executor::fetchChunks(
                 // determine if we need special treatment to linearlize multi-frag table
                 // i.e., a column that is classified as varlen type, i.e., array
                 // for now, we can support more types in this way
+                all_frag.lock();
                 if (needLinearizeAllFragments(
                         *col_id, ra_exe_unit, selected_fragments, memory_level)) {
                   bool for_lazy_fetch = false;
@@ -3055,7 +3060,9 @@ FetchResult Executor::fetchChunks(
                                                                 device_allocator,
                                                                 /*thread_idx=*/0);
                 }
+                all_frag.unlock();
               } else {
+                LOG(INFO) << "Pushing to idx " << idx;
                 frag_col_buffers[it->second] =
                     column_fetcher.getOneTableColumnFragment(col_id->getColInfo(),
                                                             frag_id,
@@ -3067,10 +3074,8 @@ FetchResult Executor::fetchChunks(
                                                             device_allocator);
               }
             }
-            all_frag.lock();
-            selected_frag_ids_vec.push_back(selected_frag_ids);
-            all_frag_col_buffers.push_back(frag_col_buffers);
-            all_frag.unlock();
+            LOG(INFO) << "frag_col_buffers size to push: " << frag_col_buffers.size();
+            all_frag_col_buffers[idx] = frag_col_buffers;
           });
     });
     if (empty_frags) {
@@ -3152,6 +3157,29 @@ FetchResult Executor::fetchChunks(
   }
   std::tie(all_num_rows, all_frag_offsets) = getRowCountAndOffsetForAllFrags(
       ra_exe_unit, selected_frag_ids_vec, ra_exe_unit.input_descs, all_tables_fragments);
+  // if(memory_level == Data_Namespace::MemoryLevel::GPU_LEVEL){
+  //   LOG(INFO) << "selected_frag_ids_vec - all_frag_col_buffers - all_num_rows -
+  //   all_frag_offsets"; LOG(INFO) << "Sizes: " << selected_frag_ids_vec.size() << " - "
+  //   <<  all_frag_col_buffers.size() << " - "  << all_num_rows.size() << " - " <<
+  //   all_frag_offsets.size(); for(size_t idx = 0; idx < selected_frag_ids_vec.size();
+  //   idx++){
+  //     LOG(INFO) << "Sizes[" << idx << "]: " << selected_frag_ids_vec[idx].size() << " -
+  //     " <<  all_frag_col_buffers[idx].size() << " - "  << all_num_rows[idx].size() << "
+  //     - " << all_frag_offsets[idx].size(); for(size_t iidx = 0; iidx <
+  //     selected_frag_ids_vec[idx].size(); iidx++){
+  //       LOG(INFO) << " -- "<< selected_frag_ids_vec[idx][iidx] << " - " <<
+  //       (all_frag_col_buffers[idx].size()? ((void*)(all_frag_col_buffers[idx][iidx])) :
+  //       "__") << " - " << all_num_rows[idx][iidx] << " - " <<
+  //       all_frag_offsets[idx][iidx];
+  //     }
+  //     for(size_t iidx = selected_frag_ids_vec[idx].size(); iidx <
+  //     all_frag_col_buffers[idx].size(); iidx++){
+  //       LOG(INFO) << " -- "<< " __ "<< " - " <<
+  //       (void*)(all_frag_col_buffers[idx][iidx]) << " - " <<  " __ " << " - " <<  " __
+  //       ";
+  //     }
+  //   }
+  // }
   return {all_frag_col_buffers, all_num_rows, all_frag_offsets};
 }