naive tbb

Author: Devjiu

omniscidb/QueryEngine/JoinHashTable/Runtime/HashJoinRuntime.cpp

@@ -350,7 +350,6 @@ DEVICE int SUFFIX(fill_hash_join_buff_bucketized_cpu)(
     const int32_t* sd_inner_to_outer_translation_map,
     const int32_t min_inner_elem,
     const int64_t bucket_normalization) {
-  int partial_err = 0;
   auto filling_func = for_semi_join ? SUFFIX(fill_hashtable_for_semi_join)
                                     : SUFFIX(fill_one_to_one_hashtable);
   auto hashtable_filling_func = [&](int64_t elem, size_t index) {
@@ -366,13 +365,14 @@ DEVICE int SUFFIX(fill_hash_join_buff_bucketized_cpu)(
       reinterpret_cast<const struct JoinChunk*>(join_column.col_chunks_buff);
   // BTW it's vector with sz:
   // join_column.num_chunks
-  const int8_t* chunk_mem_ptr = join_chunk_array->col_buff;
-  size_t global_elem_index = 0;
+  // const int8_t* chunk_mem_ptr = join_chunk_array->col_buff;
+
   // LOG(ERROR) << "fill_hash_join_buff_cpu chunk buff size: "
   //            << join_column.col_chunks_buff_sz << " num elems: " <<
   //            join_column.num_elems
   //            << " num_chunks: " << join_column.num_chunks;
-
+  std::atomic<int> err{0};
+  std::atomic<size_t> global_elem_index{0};
   tbb::parallel_for(
       tbb::blocked_range<size_t>(0, join_column.num_chunks),
       [&](const tbb::blocked_range<size_t>& join_chunks_range) {
@@ -395,7 +395,8 @@ DEVICE int SUFFIX(fill_hash_join_buff_bucketized_cpu)(
                 for (size_t elem_i = curr_chnunk_elems_range.begin();
                      elem_i != curr_chnunk_elems_range.end();
                      elem_i++) {
-                  chunk_mem_ptr = curr_chunk.col_buff;
+                  int partial_err = 0;
+                  const int8_t* chunk_mem_ptr = curr_chunk.col_buff;
 
                   // char line[1024];
                   // snprintf(line, sizeof(line), " ptr: %p", chunk_mem_ptr);
@@ -477,20 +478,28 @@ DEVICE int SUFFIX(fill_hash_join_buff_bucketized_cpu)(
                   //            chunk_i
                   //            << " el_i: " << elem_i;
 
-                  if (hashtable_filling_func(elem, global_elem_index)) {
+                  if (hashtable_filling_func(elem, global_elem_index.load())) {
                     partial_err = -1;
                   }
 
                   global_elem_index++;
                   if (partial_err != 0) {
                     // LOG(ERROR) << "error here! " << partial_err;
-                    return partial_err;
+                    int zero{0};
+                    err.compare_exchange_strong(zero, partial_err);
+                    break;
                   }
                   partial_err = 0;
                 }
               });
+          if (err) {
+            break;
+          }
         }
       });
+  if (err) {
+    return -1;
+  }
   return 0;
 }
 #endif