single threaded simple version

Devjiu · Devjiu · commit 037e88598ed2 · 2023-08-21T22:21:47.000Z
diff --git a/omniscidb/QueryEngine/ColumnFetcher.cpp b/omniscidb/QueryEngine/ColumnFetcher.cpp
@@ -284,15 +284,17 @@ const int8_t* ColumnFetcher::getAllTableColumnFragments(
     const auto& type_width = col_info->type->size();
     auto write_ptr =
         executor_->row_set_mem_owner_->allocate(type_width * total_row_count);
-    tbb::concurrent_vector<std::pair<int8_t*, size_t>> write_ptrs;
+    std::vector<std::pair<int8_t*, size_t>> write_ptrs;
+    std::vector<size_t> valid_fragments;
     for (size_t frag_id = 0; frag_id < frag_count; ++frag_id) {
       const auto& fragment = (*fragments)[frag_id];
-      if (!fragment.getNumTuples()) {
+      if (fragment.isEmptyPhysicalFragment()) {
         continue;
       }
       CHECK_EQ(type_width, fragment.getChunkMetadataMap().at(col_id)->type()->size());
       write_ptrs.push_back({write_ptr, fragment.getNumTuples() * type_width});
       write_ptr += fragment.getNumTuples() * type_width;
+      valid_fragments.push_back(frag_id);
     }
 
     if (write_ptrs.empty()) {
@@ -306,17 +308,16 @@ const int8_t* ColumnFetcher::getAllTableColumnFragments(
           table_column, 0, memory_level, device_id, device_allocator);
     }
 
-    CHECK_EQ(frag_count, write_ptrs.size());
+    size_t valid_frag_count = valid_fragments.size();
     tbb::parallel_for(
-        tbb::blocked_range<size_t>(0, frag_count),
+        tbb::blocked_range<size_t>(0, valid_frag_count),
         [&](const tbb::blocked_range<size_t>& frag_ids) {
-          for (size_t frag_id = frag_ids.begin(); frag_id < frag_ids.end(); ++frag_id) {
+          for (size_t v_frag_id = frag_ids.begin(); v_frag_id < frag_ids.end();
+               ++v_frag_id) {
             std::list<std::shared_ptr<Chunk_NS::Chunk>> chunk_holder;
             std::list<ChunkIter> chunk_iter_holder;
+            size_t frag_id = valid_fragments[v_frag_id];
             const auto& fragment = (*fragments)[frag_id];
-            if (fragment.isEmptyPhysicalFragment()) {
-              continue;
-            }
             auto chunk_meta_it = fragment.getChunkMetadataMap().find(col_id);
             CHECK(chunk_meta_it != fragment.getChunkMetadataMap().end());
             std::shared_ptr<Chunk_NS::Chunk> chunk;
diff --git a/omniscidb/QueryEngine/JoinHashTable/Builders/PerfectHashTableBuilder.h b/omniscidb/QueryEngine/JoinHashTable/Builders/PerfectHashTableBuilder.h
@@ -166,8 +166,8 @@ class PerfectJoinHashTableBuilder {
                                            0);
 
     auto cpu_hash_table_buff = reinterpret_cast<int32_t*>(hash_table_->getCpuBuffer());
-    const int thread_count = cpu_threads();
-    std::vector<std::thread> init_cpu_buff_threads;
+    // const int thread_count = cpu_threads();
+    // std::vector<std::thread> init_cpu_buff_threads;
 
     {
       auto timer_init = DEBUG_TIMER("CPU One-To-One Perfect-Hash: init_hash_join_buff");
@@ -176,54 +176,39 @@ class PerfectJoinHashTableBuilder {
                           hash_join_invalid_val);
     }
     const bool for_semi_join = for_semi_anti_join(join_type);
-    std::atomic<int> err{0};
+    // std::atomic<int> err{0};
     {
       auto timer_fill =
           DEBUG_TIMER("CPU One-To-One Perfect-Hash: fill_hash_join_buff_bucketized");
-      for (int thread_idx = 0; thread_idx < thread_count; ++thread_idx) {
-        init_cpu_buff_threads.emplace_back([hash_join_invalid_val,
-                                            &join_column,
-                                            str_proxy_translation_map,
-                                            thread_idx,
-                                            thread_count,
-                                            type,
-                                            &err,
-                                            &col_range,
-                                            &is_bitwise_eq,
-                                            &for_semi_join,
-                                            cpu_hash_table_buff,
-                                            hash_entry_info] {
-          int partial_err = fill_hash_join_buff_bucketized(
-              cpu_hash_table_buff,
-              hash_join_invalid_val,
-              for_semi_join,
-              join_column,
-              {static_cast<size_t>(type->size()),
-               col_range.getIntMin(),
-               col_range.getIntMax(),
-               inline_fixed_encoding_null_value(type),
-               is_bitwise_eq,
-               col_range.getIntMax() + 1,
-               get_join_column_type_kind(type)},
-              str_proxy_translation_map ? str_proxy_translation_map->data() : nullptr,
-              str_proxy_translation_map ? str_proxy_translation_map->domainStart()
-                                        : 0,  // 0 is dummy value
-              thread_idx,
-              thread_count,
-              hash_entry_info.bucket_normalization);
-          int zero{0};
-          err.compare_exchange_strong(zero, partial_err);
-        });
-      }
-      for (auto& t : init_cpu_buff_threads) {
-        t.join();
+
+      {
+        // int partial_err = 0;
+        JoinColumnTypeInfo type_info{static_cast<size_t>(type->size()),
+                                     col_range.getIntMin(),
+                                     col_range.getIntMax(),
+                                     inline_fixed_encoding_null_value(type),
+                                     is_bitwise_eq,
+                                     col_range.getIntMax() + 1,
+                                     get_join_column_type_kind(type)};
+
+        int error = fill_hash_join_buff_bucketized_cpu(
+            cpu_hash_table_buff,
+            hash_join_invalid_val,
+            for_semi_join,
+            join_column,
+            type_info,
+            str_proxy_translation_map ? str_proxy_translation_map->data() : nullptr,
+            str_proxy_translation_map ? str_proxy_translation_map->domainStart()
+                                      : 0,  // 0 is dummy value
+            hash_entry_info.bucket_normalization);
+        if (error) {
+          // Too many hash entries, need to retry with a 1:many table
+          hash_table_ = nullptr;  // clear the hash table buffer
+          // LOG(ERROR) << "cpu hash_tabel error: " << error;
+          throw NeedsOneToManyHash();
+        }
       }
     }
-    if (err) {
-      // Too many hash entries, need to retry with a 1:many table
-      hash_table_ = nullptr;  // clear the hash table buffer
-      throw NeedsOneToManyHash();
-    }
   }
 
   void initOneToManyHashTableOnCpu(
diff --git a/omniscidb/QueryEngine/JoinHashTable/Runtime/HashJoinRuntime.cpp b/omniscidb/QueryEngine/JoinHashTable/Runtime/HashJoinRuntime.cpp
@@ -261,14 +261,25 @@ DEVICE auto fill_hash_join_buff_impl(int32_t* buff,
   int32_t start = cpu_thread_idx;
   int32_t step = cpu_thread_count;
 #endif
+  // LOG(ERROR) << "fill_hash_join_buff_impl func cur th idx: " << cpu_thread_idx
+  //            << " count: " << cpu_thread_count
+  //            << " get th index: " << std::this_thread::get_id()
+  //            << " chunk buff size: " << join_column.col_chunks_buff_sz
+  //            << " num elems: " << join_column.num_elems
+  //            << " num_chunks: " << join_column.num_chunks;
+  // INJECT_TIMER(fill_hash_join_buff_impl);
   JoinColumnTyped col{&join_column, &type_info};
   for (auto item : col.slice(start, step)) {
+    // LOG(ERROR) << "items: " << item.index;
     const size_t index = item.index;
     int64_t elem = item.element;
     if (elem == type_info.null_val) {
+      // LOG(ERROR) << "null val";
       if (type_info.uses_bw_eq) {
         elem = type_info.translated_null_val;
       } else {
+        // LOG(ERROR) << "initOneToOneHashTableOnCpu(Threaded) cont elem: " << elem
+        //            << " index: " << index;
         continue;
       }
     }
@@ -286,7 +297,13 @@ DEVICE auto fill_hash_join_buff_impl(int32_t* buff,
       elem = outer_id;
     }
 #endif
+    // char line[1024];
+    // snprintf(line, sizeof(line), " entry ptr: %p", buff);
+    // LOG(ERROR) << "initOneToOneHashTableOnCpu(Threaded) elem: " << elem
+    //            << " index: " << index << " invalid_slot_val: " << invalid_slot_val
+    //            << line;
     if (filling_func(elem, index)) {
+      // LOG(ERROR) << "errr";
       return -1;
     }
   }
@@ -323,6 +340,138 @@ DEVICE int SUFFIX(fill_hash_join_buff_bucketized)(
                                   hashtable_filling_func);
 }
 
+#ifndef __CUDACC__
+DEVICE int SUFFIX(fill_hash_join_buff_bucketized_cpu)(
+    int32_t* cpu_hash_table_buff,
+    const int32_t hash_join_invalid_val,
+    const bool for_semi_join,
+    const JoinColumn& join_column,
+    const JoinColumnTypeInfo& type_info,
+    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) {
+    auto entry_ptr = SUFFIX(get_bucketized_hash_slot)(
+        cpu_hash_table_buff, elem, type_info.min_val, bucket_normalization);
+    // LOG(ERROR) << "filling index: " << index << " elem: " << elem
+    //            << " entry_ptr: " << entry_ptr;
+    return filling_func(index, entry_ptr, hash_join_invalid_val);
+  };
+
+  // for some stupid reason int8* ptr is actually JoinChunk* Why?
+  auto join_chunk_array =
+      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;
+  // 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;
+  for (size_t chunk_i = 0; chunk_i < join_column.num_chunks; chunk_i++) {
+    // wtf 1 chunk, but 0 elements.
+    if (join_column.num_elems == 0) {
+      break;
+    }
+    auto curr_chunk = join_chunk_array[chunk_i];
+    for (size_t elem_i = 0; elem_i < curr_chunk.num_elems; elem_i++) {
+      chunk_mem_ptr = curr_chunk.col_buff;
+
+      // char line[1024];
+      // snprintf(line, sizeof(line), " ptr: %p", chunk_mem_ptr);
+      // LOG(ERROR) << "initOneToOneHashTableOnCpu " << line
+      //            << " type: " << type_info.column_type << " index in chunk: " << elem_i
+      //            << " elem_sz: " << type_info.elem_sz
+      //            << " invalid_slot_val: " << hash_join_invalid_val;
+
+      int64_t elem = 0;
+      switch (type_info.column_type) {
+        case SmallDate: {
+          // LOG(ERROR) << "smallDate";
+          elem = fixed_width_small_date_decode_noinline(
+              chunk_mem_ptr,
+              type_info.elem_sz,
+              type_info.elem_sz == 4 ? NULL_INT : NULL_SMALLINT,
+              type_info.elem_sz == 4 ? NULL_INT : NULL_SMALLINT,
+              elem_i);
+          break;
+        }
+        case Signed: {
+          // char line[1024];
+          // snprintf(line, sizeof(line), " ptr: %p", chunk_mem_ptr);
+          // LOG(ERROR) << "Should call fixed_width_int_decode_noinline: " << line
+          //            << " elem_i: " << elem_i << " without func: ";
+
+          // LOG(ERROR) << "int32_cast: "
+          //            << *(reinterpret_cast<const int32_t*>(
+          //                   &chunk_mem_ptr[elem_i * type_info.elem_sz]));
+          elem =
+              fixed_width_int_decode_noinline(chunk_mem_ptr, type_info.elem_sz, elem_i);
+          break;
+        }
+        case Unsigned: {
+          // LOG(ERROR) << "unsigned";
+          elem = fixed_width_unsigned_decode_noinline(
+              chunk_mem_ptr, type_info.elem_sz, elem_i);
+          break;
+        }
+        case Double: {
+          // LOG(ERROR) << "double";
+          elem = fixed_width_double_decode_noinline(chunk_mem_ptr, elem_i);
+          break;
+        }
+        default: {
+          // LOG(ERROR) << "default";
+          assert(0);
+        }
+      }
+
+      if (elem == type_info.null_val) {
+        // LOG(ERROR) << "null elem";
+        if (type_info.uses_bw_eq) {
+          elem = type_info.translated_null_val;
+        } else {
+          // LOG(ERROR) << "cont: elem_i - " << elem_i << " chunk_i - " << chunk_i;
+          break;
+        }
+      }
+      if (sd_inner_to_outer_translation_map &&
+          (!type_info.uses_bw_eq || elem != type_info.translated_null_val)) {
+        const auto outer_id = map_str_id_to_outer_dict(elem,
+                                                       min_inner_elem,
+                                                       type_info.min_val,
+                                                       type_info.max_val,
+                                                       sd_inner_to_outer_translation_map);
+        if (outer_id == StringDictionary::INVALID_STR_ID) {
+          break;
+        }
+        elem = outer_id;
+      }
+
+      // LOG(ERROR) << "initOneToOneHashTableOnCpu elem: " << elem
+      //            << " index: " << global_elem_index << " chunk idx: " << chunk_i
+      //            << " el_i: " << elem_i;
+
+      if (hashtable_filling_func(elem, global_elem_index)) {
+        partial_err = -1;
+      }
+
+      global_elem_index++;
+      if (partial_err != 0) {
+        // LOG(ERROR) << "error here! " << partial_err;
+        return partial_err;
+      }
+      partial_err = 0;
+    }
+  }
+  return 0;
+}
+#endif
+
 DEVICE int SUFFIX(fill_hash_join_buff)(int32_t* buff,
                                        const int32_t invalid_slot_val,
                                        const bool for_semi_join,
diff --git a/omniscidb/QueryEngine/JoinHashTable/Runtime/HashJoinRuntime.h b/omniscidb/QueryEngine/JoinHashTable/Runtime/HashJoinRuntime.h
@@ -146,6 +146,17 @@ int fill_hash_join_buff_bucketized(int32_t* buff,
                                    const int32_t cpu_thread_count,
                                    const int64_t bucket_normalization);
 
+#ifndef __CUDACC_
+int fill_hash_join_buff_bucketized_cpu(int32_t* cpu_hash_table_buff,
+                                       const int32_t hash_join_invalid_val,
+                                       const bool for_semi_join,
+                                       const JoinColumn& join_column,
+                                       const JoinColumnTypeInfo& type_info,
+                                       const int32_t* sd_inner_to_outer_translation_map,
+                                       const int32_t min_inner_elem,
+                                       const int64_t bucket_normalization);
+#endif
+
 int fill_hash_join_buff(int32_t* buff,
                         const int32_t invalid_slot_val,
                         const bool for_semi_join,
diff --git a/omniscidb/QueryEngine/JoinHashTable/Runtime/JoinColumnIterator.h b/omniscidb/QueryEngine/JoinHashTable/Runtime/JoinColumnIterator.h
@@ -73,6 +73,7 @@ struct JoinColumnIterator {
   DEVICE FORCE_INLINE JoinColumnIterator& operator++() {
     index += step;
     index_inside_chunk += step;
+    // this stuff is made to find index_of_chunk by total index of element
     while (chunk_data &&
            index_inside_chunk >= join_chunk_array[index_of_chunk].num_elems) {
       index_inside_chunk -= join_chunk_array[index_of_chunk].num_elems;
diff --git a/omniscidb/Shared/thread_count.h b/omniscidb/Shared/thread_count.h
@@ -16,10 +16,10 @@
 
 #pragma once
 
-#include "Shared/funcannotations.h"
-
 #include <algorithm>
 #include <thread>
+#include "Logger/Logger.h"
+#include "Shared/funcannotations.h"
 
 #ifndef SHARED_EXPORT
 #define SHARED_EXPORT EXTERN
