Fix excessive memory usage for large amount of streams in readInOrder

Forcefully reduce the requested amount of num_streams while spreading
mark ranges for inOrder reading to prevent excessive memory usage

Co-authored-by: Azat Khuzhin <[email protected]>

Author: korowa

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/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;