Setting parallelize_output_from_storages

src/Core/Settings.h

@@ -709,6 +709,8 @@ class IColumn;
     \
     M(String, workload, "default", "Name of workload to be used to access resources", 0) \
     \
+    M(Bool, parallelize_output_from_storages, false, "Parallelize output for reading step from storage. It allows parallelizing query processing right after reading from storage if possible", 0) \
+    \
     /** Experimental functions */ \
     M(Bool, allow_experimental_funnel_functions, false, "Enable experimental functions for funnel analysis.", 0) \
     M(Bool, allow_experimental_nlp_functions, false, "Enable experimental functions for natural language processing.", 0) \

src/Storages/IStorage.cpp

@@ -136,8 +136,8 @@ void IStorage::read(
 
     /// parallelize processing if not yet
     const size_t output_ports = pipe.numOutputPorts();
-    const auto storage_name = getName();
-    if (parallelizeOutputAfterReading() && output_ports > 0 && output_ports < num_streams)
+    const bool parallelize_output = context->getSettingsRef().parallelize_output_from_storages;
+    if (parallelize_output && parallelizeOutputAfterReading() && output_ports > 0 && output_ports < num_streams)
         pipe.resize(num_streams);
 
     readFromPipe(query_plan, std::move(pipe), column_names, storage_snapshot, query_info, context, getName());

tests/queries/0_stateless/02723_parallelize_output_setting.reference

@@ -0,0 +1,7 @@
+-- { echoOn }
+set parallelize_output_from_storages=1;
+select startsWith(trimLeft(explain),'Resize') as resize from (explain pipeline select * from file(data_02723.csv)) where resize;
+1
+-- no Resize in pipeline
+set parallelize_output_from_storages=0;
+select startsWith(trimLeft(explain),'Resize') as resize from (explain pipeline select * from file(data_02723.csv)) where resize;

tests/queries/0_stateless/02723_parallelize_output_setting.sql

@@ -0,0 +1,12 @@
+-- Tags: no-parallel
+
+insert into function file(data_02723.csv) select number from numbers(5) settings engine_file_truncate_on_insert=1;
+
+set max_threads=2;
+-- { echoOn }
+set parallelize_output_from_storages=1;
+select startsWith(trimLeft(explain),'Resize') as resize from (explain pipeline select * from file(data_02723.csv)) where resize;
+-- no Resize in pipeline
+set parallelize_output_from_storages=0;
+select startsWith(trimLeft(explain),'Resize') as resize from (explain pipeline select * from file(data_02723.csv)) where resize;
+