Require TBB for CPU multithreading

CMakeLists.txt

@@ -165,8 +165,6 @@ include_directories(${Boost_INCLUDE_DIR})
 
 # TBB
 find_package(TBB REQUIRED)
-add_definitions("-DENABLE_TBB")
-add_definitions("-DHAVE_TBB")
 add_definitions("-DTBB_PREVIEW_TASK_GROUP_EXTENSIONS=1")
 
 # Cost Model

omniscidb/ArrowStorage/ArrowStorage.cpp

@@ -18,7 +18,6 @@
 #include "IR/Type.h"
 #include "Shared/ArrowUtil.h"
 #include "Shared/measure.h"
-#include "Shared/threading.h"
 
 #ifdef __GNUC__
 #pragma GCC diagnostic push
@@ -32,6 +31,8 @@
 #include <arrow/util/value_parsing.h>
 #include <parquet/api/reader.h>
 #include <parquet/arrow/reader.h>
+#include <tbb/blocked_range.h>
+#include <tbb/parallel_for.h>
 
 #ifdef __GNUC__
 #pragma GCC diagnostic pop
@@ -682,131 +683,126 @@ void ArrowStorage::appendArrowTable(std::shared_ptr<arrow::Table> at, int table_
     }
   }
 
-  threading::parallel_for(
-      threading::blocked_range(0, (int)at->columns().size()), [&](auto range) {
-        for (auto col_idx = range.begin(); col_idx != range.end(); col_idx++) {
-          auto col_info = getColumnInfo(db_id_, table_id, columnId(col_idx));
-          auto col_type = col_info->type;
-          auto col_arr = at->column(col_idx);
-
-          // Conversion of empty string to Nulls and further processing handled
-          // separately.
-          if (!col_type->nullable() && col_arr->null_count() != 0 &&
-              col_arr->type()->id() != arrow::Type::STRING) {
-            throw std::runtime_error("Null values used in non-nullable type: "s +
-                                     col_type->toString());
-          }
+  tbb::parallel_for(tbb::blocked_range(0, (int)at->columns().size()), [&](auto range) {
+    for (auto col_idx = range.begin(); col_idx != range.end(); col_idx++) {
+      auto col_info = getColumnInfo(db_id_, table_id, columnId(col_idx));
+      auto col_type = col_info->type;
+      auto col_arr = at->column(col_idx);
 
-          DictionaryData* dict_data = nullptr;
-          auto elem_type =
-              col_type->isArray()
-                  ? dynamic_cast<const hdk::ir::ArrayBaseType*>(col_type)->elemType()
-                  : col_type;
-          if (elem_type->isExtDictionary()) {
-            dict_data = dicts_
-                            .at(dynamic_cast<const hdk::ir::ExtDictionaryType*>(elem_type)
-                                    ->dictId())
-                            .get();
-          }
+      // Conversion of empty string to Nulls and further processing handled
+      // separately.
+      if (!col_type->nullable() && col_arr->null_count() != 0 &&
+          col_arr->type()->id() != arrow::Type::STRING) {
+        throw std::runtime_error("Null values used in non-nullable type: "s +
+                                 col_type->toString());
+      }
 
-          if (col_type->isDecimal()) {
-            col_arr = convertDecimalToInteger(col_arr, col_type);
-          } else if (col_type->isExtDictionary()) {
-            switch (col_arr->type()->id()) {
-              case arrow::Type::STRING:
-                // if the dictionary has already been materialized, append indices
-                if (!config_->storage.enable_lazy_dict_materialization ||
-                    dict_data->is_materialized) {
-                  col_arr = createDictionaryEncodedColumn(
-                      dict_data->dict()->stringDict.get(), col_arr, col_type);
-                }
-                break;
-              case arrow::Type::DICTIONARY:
-                col_arr = convertArrowDictionary(
-                    dict_data->dict()->stringDict.get(), col_arr, col_type);
-                break;
-              default:
-                CHECK(false);
+      DictionaryData* dict_data = nullptr;
+      auto elem_type =
+          col_type->isArray()
+              ? dynamic_cast<const hdk::ir::ArrayBaseType*>(col_type)->elemType()
+              : col_type;
+      if (elem_type->isExtDictionary()) {
+        dict_data =
+            dicts_
+                .at(dynamic_cast<const hdk::ir::ExtDictionaryType*>(elem_type)->dictId())
+                .get();
+      }
+
+      if (col_type->isDecimal()) {
+        col_arr = convertDecimalToInteger(col_arr, col_type);
+      } else if (col_type->isExtDictionary()) {
+        switch (col_arr->type()->id()) {
+          case arrow::Type::STRING:
+            // if the dictionary has already been materialized, append indices
+            if (!config_->storage.enable_lazy_dict_materialization ||
+                dict_data->is_materialized) {
+              col_arr = createDictionaryEncodedColumn(
+                  dict_data->dict()->stringDict.get(), col_arr, col_type);
             }
-          } else if (col_type->isString()) {
-          } else {
-            col_arr = replaceNullValues(
-                col_arr,
-                col_type,
-                dict_data ? dict_data->dict()->stringDict.get() : nullptr);
-          }
+            break;
+          case arrow::Type::DICTIONARY:
+            col_arr = convertArrowDictionary(
+                dict_data->dict()->stringDict.get(), col_arr, col_type);
+            break;
+          default:
+            CHECK(false);
+        }
+      } else if (col_type->isString()) {
+      } else {
+        col_arr = replaceNullValues(
+            col_arr, col_type, dict_data ? dict_data->dict()->stringDict.get() : nullptr);
+      }
 
-          col_data[col_idx] = col_arr;
+      col_data[col_idx] = col_arr;
 
-          bool compute_stats = !col_type->isString();
-          if (compute_stats) {
-            size_t elems_count = 1;
-            if (col_type->isFixedLenArray()) {
-              elems_count = col_type->size() / elem_type->size();
-            }
-            // Compute stats for each fragment.
-            threading::parallel_for(
-                threading::blocked_range(size_t(0), frag_count), [&](auto frag_range) {
-                  for (size_t frag_idx = frag_range.begin(); frag_idx != frag_range.end();
-                       ++frag_idx) {
-                    auto& frag = fragments[frag_idx];
-
-                    frag.offset =
-                        frag_idx
-                            ? ((frag_idx - 1) * table.fragment_size + first_frag_size)
-                            : 0;
-                    frag.row_count =
-                        frag_idx
-                            ? std::min(table.fragment_size,
-                                       static_cast<size_t>(at->num_rows()) - frag.offset)
-                            : first_frag_size;
-
-                    size_t num_bytes;
-                    if (col_type->isFixedLenArray()) {
-                      num_bytes = frag.row_count * col_type->size();
-                    } else if (col_type->isVarLenArray()) {
-                      num_bytes =
-                          computeTotalStringsLength(col_arr, frag.offset, frag.row_count);
-                    } else {
-                      num_bytes = frag.row_count * col_type->size();
-                    }
-                    auto meta = std::make_shared<ChunkMetadata>(
-                        col_info->type, num_bytes, frag.row_count);
-
-                    if (!lazy_fetch_cols[col_idx]) {
-                      meta->fillChunkStats(computeStats(
-                          col_arr->Slice(frag.offset, frag.row_count * elems_count),
-                          col_type));
-                    } else {
-                      int32_t min = 0;
-                      int32_t max = -1;
-                      meta->fillChunkStats(min, max, /*has_nulls=*/true);
-                    }
-                    frag.metadata[col_idx] = meta;
-                  }
-                });  // each fragment
-          } else {
-            for (size_t frag_idx = 0; frag_idx < frag_count; ++frag_idx) {
-              auto& frag = fragments[frag_idx];
-              frag.offset =
-                  frag_idx ? ((frag_idx - 1) * table.fragment_size + first_frag_size) : 0;
-              frag.row_count =
-                  frag_idx ? std::min(table.fragment_size,
-                                      static_cast<size_t>(at->num_rows()) - frag.offset)
-                           : first_frag_size;
-              CHECK(col_type->isText());
-              auto meta = std::make_shared<ChunkMetadata>(
-                  col_info->type,
-                  computeTotalStringsLength(col_arr, frag.offset, frag.row_count),
-                  frag.row_count);
-              meta->fillStringChunkStats(
-                  col_arr->Slice(frag.offset, frag.row_count)->null_count());
-
-              frag.metadata[col_idx] = meta;
-            }
-          }
+      bool compute_stats = !col_type->isString();
+      if (compute_stats) {
+        size_t elems_count = 1;
+        if (col_type->isFixedLenArray()) {
+          elems_count = col_type->size() / elem_type->size();
         }
-      });  // each column
+        // Compute stats for each fragment.
+        tbb::parallel_for(
+            tbb::blocked_range(size_t(0), frag_count), [&](auto frag_range) {
+              for (size_t frag_idx = frag_range.begin(); frag_idx != frag_range.end();
+                   ++frag_idx) {
+                auto& frag = fragments[frag_idx];
+
+                frag.offset =
+                    frag_idx ? ((frag_idx - 1) * table.fragment_size + first_frag_size)
+                             : 0;
+                frag.row_count =
+                    frag_idx ? std::min(table.fragment_size,
+                                        static_cast<size_t>(at->num_rows()) - frag.offset)
+                             : first_frag_size;
+
+                size_t num_bytes;
+                if (col_type->isFixedLenArray()) {
+                  num_bytes = frag.row_count * col_type->size();
+                } else if (col_type->isVarLenArray()) {
+                  num_bytes =
+                      computeTotalStringsLength(col_arr, frag.offset, frag.row_count);
+                } else {
+                  num_bytes = frag.row_count * col_type->size();
+                }
+                auto meta = std::make_shared<ChunkMetadata>(
+                    col_info->type, num_bytes, frag.row_count);
+
+                if (!lazy_fetch_cols[col_idx]) {
+                  meta->fillChunkStats(computeStats(
+                      col_arr->Slice(frag.offset, frag.row_count * elems_count),
+                      col_type));
+                } else {
+                  int32_t min = 0;
+                  int32_t max = -1;
+                  meta->fillChunkStats(min, max, /*has_nulls=*/true);
+                }
+                frag.metadata[col_idx] = meta;
+              }
+            });  // each fragment
+      } else {
+        for (size_t frag_idx = 0; frag_idx < frag_count; ++frag_idx) {
+          auto& frag = fragments[frag_idx];
+          frag.offset =
+              frag_idx ? ((frag_idx - 1) * table.fragment_size + first_frag_size) : 0;
+          frag.row_count =
+              frag_idx ? std::min(table.fragment_size,
+                                  static_cast<size_t>(at->num_rows()) - frag.offset)
+                       : first_frag_size;
+          CHECK(col_type->isText());
+          auto meta = std::make_shared<ChunkMetadata>(
+              col_info->type,
+              computeTotalStringsLength(col_arr, frag.offset, frag.row_count),
+              frag.row_count);
+          meta->fillStringChunkStats(
+              col_arr->Slice(frag.offset, frag.row_count)->null_count());
+
+          frag.metadata[col_idx] = meta;
+        }
+      }
+    }
+  });  // each column
   dict_lock.unlock();
 
   if (table.row_count) {

omniscidb/ArrowStorage/ArrowStorageUtils.cpp

@@ -17,7 +17,6 @@
 #include "IR/Context.h"
 #include "Shared/InlineNullValues.h"
 
-// TODO: use <Shared/threading.h>
 #include <tbb/parallel_for.h>
 #include <tbb/task_group.h>
 

omniscidb/CMakeLists.txt

@@ -590,26 +590,11 @@ endif()
 
 # TBB
 
-option(ENABLE_TBB "Enable OneTBB for threading (if found)" ON)
-set(TBB_LIBS "")
-find_package(TBB)
+find_package(TBB REQUIRED)
 if(TBB_FOUND)
   message(STATUS "TBB library is found with ${TBB_DIR}")
-  add_definitions("-DHAVE_TBB")
   add_definitions("-DTBB_PREVIEW_TASK_GROUP_EXTENSIONS=1")
   list(APPEND TBB_LIBS ${TBB_LIBRARIES})
-  if(ENABLE_TBB)
-    add_definitions("-DENABLE_TBB")
-  else()
-    message(STATUS "Using TBB for threading is DISABLED")
-  endif()
-else()
-  set(ENABLE_TBB OFF)
-endif()
-
-option(DISABLE_CONCURRENCY "Disable parallellism at the threading layer" OFF)
-if(DISABLE_CONCURRENCY)
-  add_definitions("-DDISABLE_CONCURRENCY")
 endif()
 
 list(APPEND ADDITIONAL_MAKE_CLEAN_FILES ${CMAKE_BINARY_DIR}/gen-cpp/)

omniscidb/QueryEngine/ArrowResultSetConverter.cpp

@@ -20,7 +20,6 @@
 #include "Execute.h"
 #include "Shared/ArrowUtil.h"
 #include "Shared/DateConverters.h"
-#include "Shared/threading.h"
 #include "Shared/toString.h"
 
 //  arrow headers
@@ -341,7 +340,7 @@ int64_t create_bitmap_parallel_for_avx512(uint8_t* bitmap_data,
         bitmap_data_ptr, const_cast<TYPE*>(values_data_ptr), processing_count);
   };
 
-  threading::parallel_for(
+  tbb::parallel_for(
       tbb::blocked_range<size_t>(
           0, (avx512_processing_count + min_block_size - 1) / min_block_size),
       br_par_processor);
@@ -394,7 +393,7 @@ void convert_column(ResultSetPtr result,
 
   std::vector<std::shared_ptr<arrow::Array>> fragments(values.size(), nullptr);
 
-  threading::parallel_for(static_cast<size_t>(0), values.size(), [&](size_t idx) {
+  tbb::parallel_for(static_cast<size_t>(0), values.size(), [&](size_t idx) {
     size_t chunk_rows_count = chunks[idx].second;
 
     auto res = arrow::AllocateBuffer((chunk_rows_count + 7) / 8);
@@ -421,7 +420,7 @@ void convert_column(ResultSetPtr result,
                          ? std::make_shared<NumArray>(
                                chunk_rows_count, values[idx], is_valid, null_count)
                          : std::make_shared<NumArray>(chunk_rows_count, values[idx]);
-  });  // threading::parallel_for
+  });  // tbb::parallel_for
 
   out = std::make_shared<arrow::ChunkedArray>(std::move(fragments));
 }
@@ -1492,7 +1491,7 @@ std::shared_ptr<arrow::Table> ArrowResultSetConverter::getArrowTable(
                                    results_->isTruncated());
     } else {
       auto timer = DEBUG_TIMER("fetch data in parallel_for");
-      threading::parallel_for(
+      tbb::parallel_for(
           static_cast<size_t>(0), entry_count, stride, [&](size_t start_entry) {
             const size_t i = start_entry / stride;
             const size_t end_entry = std::min(entry_count, start_entry + stride);
@@ -1513,7 +1512,7 @@ std::shared_ptr<arrow::Table> ArrowResultSetConverter::getArrowTable(
 
     {
       auto timer = DEBUG_TIMER("append rows to arrow, finish builders");
-      threading::parallel_for(static_cast<size_t>(0), col_count, [&](size_t i) {
+      tbb::parallel_for(static_cast<size_t>(0), col_count, [&](size_t i) {
         if (!columnar_conversion_flags[i]) {
           for (size_t j = 0; j < segments_count; ++j) {
             if (column_value_segs[j][i]) {