change all to runtime::execute_futures_in_parallel

Author: BohuTANG

src/query/pipeline/sources/src/input_formats/impls/input_format_parquet.rs

@@ -148,6 +148,7 @@ impl InputFormat for InputFormatParquet {
                     .map(|location| Self::get_split_batch(location.to_vec(), op.clone()))
             });
 
+            // TODO: Get from ctx.
             let thread_nums = 16;
             let permit_nums = 64;
             let result = execute_futures_in_parallel(

src/query/storages/fuse/src/io/files.rs

@@ -14,12 +14,11 @@
 
 use std::sync::Arc;
 
+use common_base::runtime::execute_futures_in_parallel;
 use common_catalog::table_context::TableContext;
 use common_exception::Result;
 use opendal::Operator;
 
-use crate::io::execute_futures_in_parallel;
-
 // File related operations.
 pub struct Files {
     ctx: Arc<dyn TableContext>,
@@ -52,9 +51,12 @@ impl Files {
                     .map(|location| Self::delete_files(self.operator.clone(), location.to_vec()))
             });
 
+            let threads_nums = self.ctx.get_settings().get_max_threads()? as usize;
+            let permit_nums = self.ctx.get_settings().get_max_storage_io_requests()? as usize;
             execute_futures_in_parallel(
-                self.ctx.clone(),
                 tasks,
+                threads_nums,
+                permit_nums,
                 "batch-remove-files-worker".to_owned(),
             )
             .await?;

src/query/storages/fuse/src/io/mod.rs

@@ -14,15 +14,13 @@
 
 mod files;
 mod locations;
-mod parallel;
 mod read;
 mod segments;
 mod snapshots;
 mod write;
 
 pub use files::Files;
 pub use locations::TableMetaLocationGenerator;
-pub use parallel::execute_futures_in_parallel;
 pub use read::BlockReader;
 pub use read::BloomBlockFilterReader;
 pub use read::MergeIOReadResult;

src/query/storages/fuse/src/io/parallel.rs

@@ -14,6 +14,7 @@
 
 use std::sync::Arc;
 
+use common_base::runtime::execute_futures_in_parallel;
 use common_catalog::table_context::TableContext;
 use common_exception::Result;
 use common_expression::TableSchemaRef;
@@ -23,7 +24,6 @@ use storages_common_table_meta::meta::Location;
 use storages_common_table_meta::meta::SegmentInfo;
 use tracing::Instrument;
 
-use crate::io::execute_futures_in_parallel;
 use crate::io::MetaReaders;
 
 // Read segment related operations.
@@ -88,9 +88,12 @@ impl SegmentsIO {
             }
         });
 
+        let threads_nums = self.ctx.get_settings().get_max_threads()? as usize;
+        let permit_nums = self.ctx.get_settings().get_max_storage_io_requests()? as usize;
         execute_futures_in_parallel(
-            self.ctx.clone(),
             tasks,
+            threads_nums,
+            permit_nums,
             "fuse-req-segments-worker".to_owned(),
         )
         .await
@@ -143,9 +146,12 @@ impl SegmentsIO {
             })
         });
 
+        let threads_nums = self.ctx.get_settings().get_max_threads()? as usize;
+        let permit_nums = self.ctx.get_settings().get_max_storage_io_requests()? as usize;
         execute_futures_in_parallel(
-            self.ctx.clone(),
             tasks,
+            threads_nums,
+            permit_nums,
             "fuse-req-segments-worker".to_owned(),
         )
         .await

src/query/storages/fuse/src/io/segments.rs

@@ -20,6 +20,7 @@ use std::time::Instant;
 
 use chrono::DateTime;
 use chrono::Utc;
+use common_base::runtime::execute_futures_in_parallel;
 use common_catalog::table_context::TableContext;
 use common_exception::ErrorCode;
 use common_exception::Result;
@@ -37,7 +38,6 @@ use tracing::info;
 use tracing::warn;
 use tracing::Instrument;
 
-use crate::io::execute_futures_in_parallel;
 use crate::io::MetaReaders;
 use crate::io::SnapshotHistoryReader;
 use crate::io::TableMetaLocationGenerator;
@@ -168,9 +168,12 @@ impl SnapshotsIO {
             })
         });
 
+        let threads_nums = self.ctx.get_settings().get_max_threads()? as usize;
+        let permit_nums = self.ctx.get_settings().get_max_storage_io_requests()? as usize;
         execute_futures_in_parallel(
-            self.ctx.clone(),
             tasks,
+            threads_nums,
+            permit_nums,
             "fuse-req-snapshots-worker".to_owned(),
         )
         .await

src/query/storages/fuse/src/io/snapshots.rs

@@ -15,6 +15,7 @@
 
 use std::sync::Arc;
 
+use common_base::runtime::execute_futures_in_parallel;
 use common_catalog::table_context::TableContext;
 use common_exception::ErrorCode;
 use common_exception::Result;
@@ -30,7 +31,6 @@ use storages_common_table_meta::meta::Location;
 use storages_common_table_meta::meta::SegmentInfo;
 use tracing::debug;
 
-use crate::io::execute_futures_in_parallel;
 use crate::io::SegmentsIO;
 use crate::io::TableMetaLocationGenerator;
 use crate::operations::merge_into::mutation_meta::mutation_log::CommitMeta;
@@ -148,10 +148,10 @@ impl TableMutationAggregator {
 
     // TODO use batch_meta_writer
     async fn write_segments(&self, segments: Vec<SerializedSegment>) -> Result<()> {
-        let mut handles = Vec::with_capacity(segments.len());
+        let mut tasks = Vec::with_capacity(segments.len());
         for segment in segments {
             let op = self.dal.clone();
-            handles.push(async move {
+            tasks.push(async move {
                 op.write(&segment.path, segment.raw_data).await?;
                 if let Some(segment_cache) = CacheManager::instance().get_table_segment_cache() {
                     segment_cache.put(segment.path.clone(), segment.segment.clone());
@@ -160,9 +160,12 @@ impl TableMutationAggregator {
             });
         }
 
+        let threads_nums = self.ctx.get_settings().get_max_threads()? as usize;
+        let permit_nums = self.ctx.get_settings().get_max_storage_io_requests()? as usize;
         execute_futures_in_parallel(
-            self.ctx.clone(),
-            handles,
+            tasks,
+            threads_nums,
+            permit_nums,
             "mutation-write-segments-worker".to_owned(),
         )
         .await?

src/query/storages/fuse/src/operations/merge_into/processors/transform_mutation_aggregator.rs

@@ -17,6 +17,7 @@ use std::collections::BTreeMap;
 use std::collections::HashMap;
 use std::sync::Arc;
 
+use common_base::runtime::execute_futures_in_parallel;
 use common_catalog::table_context::TableContext;
 use common_exception::ErrorCode;
 use common_exception::Result;
@@ -31,7 +32,6 @@ use storages_common_table_meta::meta::Location;
 use storages_common_table_meta::meta::SegmentInfo;
 use storages_common_table_meta::meta::Statistics;
 
-use crate::io::execute_futures_in_parallel;
 use crate::io::SegmentsIO;
 use crate::io::TableMetaLocationGenerator;
 use crate::operations::mutation::AbortOperation;
@@ -149,10 +149,10 @@ impl MutationTransform {
     }
 
     async fn write_segments(&self, segments: Vec<SerializedData>) -> Result<()> {
-        let mut handles = Vec::with_capacity(segments.len());
+        let mut tasks = Vec::with_capacity(segments.len());
         for segment in segments {
             let op = self.dal.clone();
-            handles.push(async move {
+            tasks.push(async move {
                 op.write(&segment.location, segment.data).await?;
                 if let Some(segment_cache) = CacheManager::instance().get_table_segment_cache() {
                     segment_cache.put(segment.location.clone(), segment.segment.clone());
@@ -161,9 +161,12 @@ impl MutationTransform {
             });
         }
 
+        let threads_nums = self.ctx.get_settings().get_max_threads()? as usize;
+        let permit_nums = self.ctx.get_settings().get_max_storage_io_requests()? as usize;
         execute_futures_in_parallel(
-            self.ctx.clone(),
-            handles,
+            tasks,
+            threads_nums,
+            permit_nums,
             "mutation-write-segments-worker".to_owned(),
         )
         .await?