Add non-lazy data import.

Signed-off-by: ienkovich <[email protected]>

Author: ienkovich

omniscidb/ArrowStorage/ArrowStorage.cpp

@@ -715,7 +715,8 @@ void ArrowStorage::appendArrowTable(std::shared_ptr<arrow::Table> at, int table_
         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 ||
+            if (config_->storage.enable_non_lazy_data_import ||
+                !config_->storage.enable_lazy_dict_materialization ||
                 dict_data->is_materialized) {
               col_arr = createDictionaryEncodedColumn(
                   dict_data->dict()->stringDict.get(), col_arr, col_type);
@@ -730,8 +731,11 @@ void ArrowStorage::appendArrowTable(std::shared_ptr<arrow::Table> at, int table_
         }
       } else if (col_type->isString()) {
       } else {
-        col_arr = replaceNullValues(
-            col_arr, col_type, dict_data ? dict_data->dict()->stringDict.get() : nullptr);
+        col_arr =
+            replaceNullValues(col_arr,
+                              col_type,
+                              dict_data ? dict_data->dict()->stringDict.get() : nullptr,
+                              config_->storage.enable_non_lazy_data_import);
       }
 
       col_data[col_idx] = col_arr;

omniscidb/ArrowStorage/ArrowStorageUtils.cpp

@@ -174,8 +174,9 @@ void copyArrayDataReplacingNulls(T* dst, std::shared_ptr<arrow::Array> arr) {
 
 template <typename T>
 std::shared_ptr<arrow::ChunkedArray> replaceNullValuesImpl(
-    std::shared_ptr<arrow::ChunkedArray> arr) {
-  if (!std::is_same_v<T, bool> && arr->null_count() == 0) {
+    std::shared_ptr<arrow::ChunkedArray> arr,
+    bool force_copy) {
+  if (!force_copy && !std::is_same_v<T, bool> && arr->null_count() == 0) {
     // for boolean columns we still need to convert bitmaps to array
     return arr;
   }
@@ -884,7 +885,9 @@ std::shared_ptr<arrow::ChunkedArray> convertDecimalToInteger(
 std::shared_ptr<arrow::ChunkedArray> replaceNullValues(
     std::shared_ptr<arrow::ChunkedArray> arr,
     const hdk::ir::Type* type,
-    StringDictionary* dict) {
+    StringDictionary* dict,
+    bool force_single_chunk) {
+  bool force_copy = force_single_chunk && (arr->chunks().size() > 1);
   if (type->isTime()) {
     if (type->size() != 8) {
       throw std::runtime_error("Unsupported time type for Arrow import: "s +
@@ -902,7 +905,7 @@ std::shared_ptr<arrow::ChunkedArray> replaceNullValues(
       case 2:
         return convertDateReplacingNulls<int32_t, int16_t>(arr);
       case 4:
-        return replaceNullValuesImpl<int32_t>(arr);
+        return replaceNullValuesImpl<int32_t>(arr, force_copy);
       case 8:
         return convertDateReplacingNulls<int32_t, int64_t>(arr);
       default:
@@ -912,26 +915,26 @@ std::shared_ptr<arrow::ChunkedArray> replaceNullValues(
   } else if (type->isInteger() || type->isTimestamp()) {
     switch (type->size()) {
       case 1:
-        return replaceNullValuesImpl<int8_t>(arr);
+        return replaceNullValuesImpl<int8_t>(arr, force_copy);
       case 2:
-        return replaceNullValuesImpl<int16_t>(arr);
+        return replaceNullValuesImpl<int16_t>(arr, force_copy);
       case 4:
-        return replaceNullValuesImpl<int32_t>(arr);
+        return replaceNullValuesImpl<int32_t>(arr, force_copy);
       case 8:
-        return replaceNullValuesImpl<int64_t>(arr);
+        return replaceNullValuesImpl<int64_t>(arr, force_copy);
       default:
         throw std::runtime_error("Unsupported integer/datetime type for Arrow import: "s +
                                  type->toString());
     }
   } else if (type->isFloatingPoint()) {
     switch (type->as<hdk::ir::FloatingPointType>()->precision()) {
       case hdk::ir::FloatingPointType::kFloat:
-        return replaceNullValuesImpl<float>(arr);
+        return replaceNullValuesImpl<float>(arr, force_copy);
       case hdk::ir::FloatingPointType::kDouble:
-        return replaceNullValuesImpl<double>(arr);
+        return replaceNullValuesImpl<double>(arr, force_copy);
     }
   } else if (type->isBoolean()) {
-    return replaceNullValuesImpl<bool>(arr);
+    return replaceNullValuesImpl<bool>(arr, force_copy);
   } else if (type->isFixedLenArray()) {
     return replaceNullValuesFixedSizeArray(arr, type, dict);
   } else if (type->isVarLenArray()) {

omniscidb/ArrowStorage/ArrowStorageUtils.h

@@ -28,7 +28,8 @@ std::shared_ptr<arrow::DataType> getArrowImportType(hdk::ir::Context& ctx,
 std::shared_ptr<arrow::ChunkedArray> replaceNullValues(
     std::shared_ptr<arrow::ChunkedArray> arr,
     const hdk::ir::Type* type,
-    StringDictionary* dict = nullptr);
+    StringDictionary* dict = nullptr,
+    bool force_single_chunk = false);
 
 std::shared_ptr<arrow::ChunkedArray> convertDecimalToInteger(
     std::shared_ptr<arrow::ChunkedArray> arr,

omniscidb/ConfigBuilder/ConfigBuilder.cpp

@@ -588,6 +588,15 @@ bool ConfigBuilder::parseCommandLineArgs(int argc,
           ->default_value(config_->storage.enable_lazy_dict_materialization)
           ->implicit_value(true),
       "Enable lazy materialization of string dictionary columns from Arrow Storage.");
+  opt_desc.add_options()(
+      "enable-non-lazy-data-import",
+      po::value<bool>(&config_->storage.enable_non_lazy_data_import)
+          ->default_value(config_->storage.enable_non_lazy_data_import)
+          ->implicit_value(true),
+      "Enable non-lazy data import in Arrow Storage. When enabled, we do as much data "
+      "processing on import as we might require. This might increase overall execution "
+      "time. This option can be used to split data import and execution for performance "
+      "measurements.");
 
   if (allow_gtest_flags) {
     opt_desc.add_options()("gtest_list_tests", "list all test");

omniscidb/Shared/Config.h

@@ -177,6 +177,7 @@ struct DebugConfig {
 
 struct StorageConfig {
   bool enable_lazy_dict_materialization = false;
+  bool enable_non_lazy_data_import = false;
 };
 
 struct Config {