Merge pull request ClickHouse#78505 from korowa/fix-zero-streams-reading

Fix zero streams after max_streams_to_max_threads_ratio applied

Author: azat

src/Interpreters/InterpreterSelectQuery.cpp

@@ -2652,8 +2652,6 @@ void InterpreterSelectQuery::executeFetchColumns(QueryProcessingStage::Enum proc
     else if (storage)
     {
         /// Table.
-        if (max_streams == 0)
-            max_streams = 1;
 
         /// If necessary, we request more sources than the number of threads - to distribute the work evenly over the threads.
         if (max_streams > 1 && !is_sync_remote)
@@ -2668,6 +2666,9 @@ void InterpreterSelectQuery::executeFetchColumns(QueryProcessingStage::Enum proc
                     streams_with_ratio);
         }
 
+        if (max_streams == 0)
+            max_streams = 1;
+
         auto & prewhere_info = analysis_result.prewhere_info;
 
         if (prewhere_info)

src/Planner/PlannerJoinTree.cpp

@@ -771,9 +771,6 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres
                     "Setting 'max_block_size' cannot be zero");
         }
 
-        if (max_streams == 0)
-            max_streams = 1;
-
         /// If necessary, we request more sources than the number of threads - to distribute the work evenly over the threads
         if (max_streams > 1 && !is_sync_remote)
         {
@@ -787,6 +784,9 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres
                     streams_with_ratio);
         }
 
+        if (max_streams == 0)
+            max_streams = 1;
+
         if (table_node)
             table_expression_query_info.table_expression_modifiers = table_node->getTableExpressionModifiers();
         else

src/Processors/QueryPlan/ReadFromMergeTree.cpp

@@ -1126,6 +1126,20 @@ Pipe ReadFromMergeTree::spreadMarkRangesAmongStreamsWithOrder(
         return new_ranges;
     };
 
+    if (num_streams > 1)
+    {
+        /// Reduce num_streams if requested value is unnecessarily large.
+        ///
+        /// Additional increase of streams number in case of skewed parts, like it's
+        /// done in `spreadMarkRangesAmongStreams` won't affect overall performance
+        /// due to the single downstream `MergingSortedTransform`.
+        if (info.sum_marks < num_streams * info.min_marks_for_concurrent_read && parts_with_ranges.size() < num_streams)
+        {
+            num_streams = std::max(
+                (info.sum_marks + info.min_marks_for_concurrent_read - 1) / info.min_marks_for_concurrent_read, parts_with_ranges.size());
+        }
+    }
+
     const size_t min_marks_per_stream = (info.sum_marks - 1) / num_streams + 1;
     bool need_preliminary_merge = (parts_with_ranges.size() > settings[Setting::read_in_order_two_level_merge_threshold]);
 

tests/queries/0_stateless/03402_zero_streams_after_max_streams_to_max_threads_ratio.reference

@@ -0,0 +1,7 @@
+49.5
+49.5
+49.5
+
+0
+0
+0

tests/queries/0_stateless/03402_zero_streams_after_max_streams_to_max_threads_ratio.sql

@@ -0,0 +1,16 @@
+DROP TABLE IF EXISTS 03402_data;
+
+CREATE TABLE 03402_data (id UInt32) ENGINE = MergeTree ORDER BY id;
+INSERT INTO 03402_data SELECT * FROM numbers(100);
+
+SELECT avg(id) FROM 03402_data SETTINGS max_threads = 4, max_streams_to_max_threads_ratio = 0;
+SELECT avg(id) FROM 03402_data SETTINGS max_threads = 0, max_streams_to_max_threads_ratio = 0;
+SELECT avg(id) FROM 03402_data SETTINGS max_threads = 2, max_streams_to_max_threads_ratio = 0.2;
+
+SELECT '';
+
+SELECT id FROM 03402_data ORDER BY id LIMIT 1 SETTINGS max_threads = 4, max_streams_to_max_threads_ratio = 0;
+SELECT id FROM 03402_data ORDER BY id LIMIT 1 SETTINGS max_threads = 0, max_streams_to_max_threads_ratio = 0;
+SELECT id FROM 03402_data ORDER BY id LIMIT 1 SETTINGS max_threads = 2, max_streams_to_max_threads_ratio = 0.2;
+
+DROP TABLE 03402_data;

tests/queries/0_stateless/03403_read_in_order_streams_memory_usage.reference

@@ -0,0 +1 @@
+1

tests/queries/0_stateless/03403_read_in_order_streams_memory_usage.sql

@@ -0,0 +1,20 @@
+DROP TABLE IF EXISTS 03403_data;
+CREATE TABLE 03403_data(id UInt32, val String) ENGINE = MergeTree ORDER BY id AS SELECT 1, 'test';
+
+SELECT *
+FROM 03403_data
+ORDER BY id
+FORMAT Null
+SETTINGS max_threads = 1024,
+         max_streams_to_max_threads_ratio = 10000000;
+
+SYSTEM FLUSH LOGS query_log;
+
+SELECT memory_usage < 10_000_000
+FROM system.query_log
+WHERE Settings['max_streams_to_max_threads_ratio'] = '10000000'
+  AND query like '%FROM 03403_data%'
+  AND type = 'QueryFinish'
+  AND current_database = currentDatabase();
+
+DROP TABLE 03403_data;