fix: pipeline max_threads should use max width of Pipes. (#18837)

Author: youngsofun

src/query/pipeline/core/src/pipeline.rs

@@ -227,7 +227,7 @@ impl Pipeline {
                 new_sinks.push_back((index, idx));
             }
         }
-
+        self.max_threads = self.max_threads.max(new_sinks.len());
         self.sinks = new_sinks;
     }
 
@@ -246,13 +246,11 @@ impl Pipeline {
     }
 
     pub fn set_max_threads(&mut self, max_threads: usize) {
-        let mut max_pipe_size = 0;
         let sinks = self.graph.externals(Direction::Outgoing).count();
         let sources = self.graph.externals(Direction::Incoming).count();
 
-        max_pipe_size = std::cmp::max(max_pipe_size, sinks);
-        max_pipe_size = std::cmp::max(max_pipe_size, sources);
-        self.max_threads = std::cmp::min(max_pipe_size, max_threads);
+        self.max_threads = std::cmp::max(self.max_threads, std::cmp::max(sinks, sources));
+        self.max_threads = std::cmp::min(self.max_threads, max_threads);
     }
 
     pub fn get_max_threads(&self) -> usize {

src/query/service/src/pipelines/executor/queries_pipeline_executor.rs

@@ -27,6 +27,7 @@ use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use fastrace::func_path;
 use fastrace::prelude::*;
+use log::info;
 use log::warn;
 use parking_lot::Mutex;
 
@@ -105,6 +106,7 @@ impl QueriesPipelineExecutor {
 
     fn execute_threads(self: &Arc<Self>, threads: usize) -> Vec<ThreadJoinHandle<Result<()>>> {
         let mut thread_join_handles = Vec::with_capacity(threads);
+        info!(num = threads; "starting worker threads");
 
         for thread_num in 0..threads {
             let this = self.clone();

src/query/service/src/pipelines/executor/query_pipeline_executor.rs

@@ -343,6 +343,7 @@ impl QueryPipelineExecutor {
 
     fn execute_threads(self: &Arc<Self>, threads: usize) -> Vec<ThreadJoinHandle<Result<()>>> {
         let mut thread_join_handles = Vec::with_capacity(threads);
+        info!(num = threads; "starting worker threads");
 
         for thread_num in 0..threads {
             let this = self.clone();

src/query/sql/src/planner/plans/copy_into_table.rs

@@ -139,6 +139,8 @@ pub struct CopyIntoTablePlan {
     // query may be Some even if is_transform=false
     pub is_transform: bool,
 
+    // control by setting `enable_distributed_copy`
+    // set in optimizer
     pub enable_distributed: bool,
 
     pub files_collected: bool,