s

Author: Devjiu

omniscidb/QueryEngine/ColumnFetcher.cpp

@@ -286,15 +286,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()) {
@@ -308,17 +310,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;

omniscidb/QueryEngine/JoinHashTable/Builders/PerfectHashTableBuilder.h

@@ -180,6 +180,7 @@ class PerfectJoinHashTableBuilder {
     {
       auto timer_fill =
           DEBUG_TIMER("CPU One-To-One Perfect-Hash: fill_hash_join_buff_bucketized");
+      LOG(INFO) << "num threads: " << thread_count;
       for (int thread_idx = 0; thread_idx < thread_count; ++thread_idx) {
         init_cpu_buff_threads.emplace_back([hash_join_invalid_val,
                                             &join_column,

omniscidb/QueryEngine/JoinHashTable/Runtime/HashJoinRuntime.cpp

@@ -261,6 +261,8 @@ DEVICE auto fill_hash_join_buff_impl(int32_t* buff,
   int32_t start = cpu_thread_idx;
   int32_t step = cpu_thread_count;
 #endif
+  auto timer = DEBUG_TIMER(__func__);
+  INJECT_TIMER(fill_hash_join_buff_impl);
   JoinColumnTyped col{&join_column, &type_info};
   for (auto item : col.slice(start, step)) {
     const size_t index = item.index;

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
@@ -29,5 +29,7 @@ SHARED_EXPORT extern unsigned g_cpu_threads_override;
 
 inline int cpu_threads() {
   auto ov = g_cpu_threads_override;
-  return (ov <= 0) ? std::max(2 * std::thread::hardware_concurrency(), 1U) : ov;
+  // LOG(INFO) << "g cpu threads: " << g_cpu_threads_override << " hw conur: " <<
+  // std::thread::hardware_concurrency();
+  return (ov <= 0) ? std::max(std::thread::hardware_concurrency(), 1U) : ov;
 }