add setting table_function_max_readers

src/Core/Settings.h

@@ -74,6 +74,7 @@ class IColumn;
     M(UInt64, idle_connection_timeout, 3600, "Close idle TCP connections after specified number of seconds.", 0) \
     M(UInt64, distributed_connections_pool_size, 1024, "Maximum number of connections with one remote server in the pool.", 0) \
     M(UInt64, connections_with_failover_max_tries, DBMS_CONNECTION_POOL_WITH_FAILOVER_DEFAULT_MAX_TRIES, "The maximum number of attempts to connect to replicas.", 0) \
+    M(MaxThreads, table_function_max_readers, 0, "The maximum number of reader threads for s3/url/hdfsCluster to execute the request. By default, it is determined automatically.", 0) \
     M(UInt64, s3_strict_upload_part_size, 0, "The exact size of part to upload during multipart upload to S3 (some implementations does not supports variable size parts).", 0) \
     M(UInt64, s3_min_upload_part_size, 16*1024*1024, "The minimum size of part to upload during multipart upload to S3.", 0) \
     M(UInt64, s3_max_upload_part_size, 5ull*1024*1024*1024, "The maximum size of part to upload during multipart upload to S3.", 0) \

src/Storages/HDFS/StorageHDFS.cpp

@@ -782,6 +782,8 @@ Pipe StorageHDFS::read(
 
     Pipes pipes;
     auto this_ptr = std::static_pointer_cast<StorageHDFS>(shared_from_this());
+    const size_t table_function_max_readers = context_->getSettingsRef().table_function_max_readers;
+    num_streams = std::min(num_streams, table_function_max_readers);
     for (size_t i = 0; i < num_streams; ++i)
     {
         pipes.emplace_back(std::make_shared<HDFSSource>(

src/Storages/StorageS3.cpp

@@ -1100,6 +1100,8 @@ Pipe StorageS3::read(
     }
 
     const size_t max_download_threads = local_context->getSettingsRef().max_download_threads;
+    const size_t table_function_max_readers = local_context->getSettingsRef().table_function_max_readers;
+    num_streams = std::min(num_streams, table_function_max_readers);
     for (size_t i = 0; i < num_streams; ++i)
     {
         pipes.emplace_back(std::make_shared<StorageS3Source>(

src/Storages/StorageURL.cpp

@@ -781,6 +781,8 @@ Pipe IStorageURLBase::read(
     pipes.reserve(num_streams);
 
     size_t download_threads = num_streams >= max_download_threads ? 1 : (max_download_threads / num_streams);
+    const size_t table_function_max_readers = local_context->getSettingsRef().table_function_max_readers;
+    num_streams = std::min(num_streams, table_function_max_readers);
     for (size_t i = 0; i < num_streams; ++i)
     {
         pipes.emplace_back(std::make_shared<StorageURLSource>(