Merge pull request #9128 from zhyass/feature_update

fix: too many io requests for read blocks during compact

Author: BohuTANG

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

@@ -28,6 +28,7 @@ pub use read::MetaReaders;
 pub use read::SegmentInfoReader;
 pub use read::SnapshotHistoryReader;
 pub use read::TableSnapshotReader;
+pub use segments::try_join_futures;
 pub use segments::SegmentsIO;
 pub use snapshots::SnapshotsIO;
 pub use write::write_block;

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

@@ -12,6 +12,7 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
+use std::future::Future;
 use std::sync::Arc;
 
 use common_base::base::tokio::sync::Semaphore;
@@ -56,11 +57,7 @@ impl SegmentsIO {
             return Ok(vec![]);
         }
 
-        let ctx = self.ctx.clone();
-        let max_runtime_threads = ctx.get_settings().get_max_threads()? as usize;
-        let max_io_requests = ctx.get_settings().get_max_storage_io_requests()? as usize;
-
-        // 1.1 combine all the tasks.
+        // combine all the tasks.
         let mut iter = segment_locations.iter();
         let tasks = std::iter::from_fn(move || {
             if let Some(location) = iter.next() {
@@ -74,22 +71,42 @@ impl SegmentsIO {
             }
         });
 
-        // 1.2 build the runtime.
-        let semaphore = Semaphore::new(max_io_requests);
-        let segments_runtime = Arc::new(Runtime::with_worker_threads(
-            max_runtime_threads,
-            Some("fuse-req-segments-worker".to_owned()),
-        )?);
+        try_join_futures(
+            self.ctx.clone(),
+            tasks,
+            "fuse-req-segments-worker".to_owned(),
+        )
+        .await
+    }
+}
 
-        // 1.3 spawn all the tasks to the runtime.
-        let join_handlers = segments_runtime.try_spawn_batch(semaphore, tasks).await?;
+pub async fn try_join_futures<Fut>(
+    ctx: Arc<dyn TableContext>,
+    futures: impl IntoIterator<Item = Fut>,
+    thread_name: String,
+) -> Result<Vec<Fut::Output>>
+where
+    Fut: Future + Send + 'static,
+    Fut::Output: Send + 'static,
+{
+    let max_runtime_threads = ctx.get_settings().get_max_threads()? as usize;
+    let max_io_requests = ctx.get_settings().get_max_storage_io_requests()? as usize;
 
-        // 1.4 get all the result.
-        let joint = future::try_join_all(join_handlers)
-            .instrument(tracing::debug_span!("read_segments_join_all"))
-            .await
-            .map_err(|e| ErrorCode::StorageOther(format!("read segments failure, {}", e)))?;
+    // 1. build the runtime.
+    let semaphore = Semaphore::new(max_io_requests);
+    let segments_runtime = Arc::new(Runtime::with_worker_threads(
+        max_runtime_threads,
+        Some(thread_name),
+    )?);
 
-        Ok(joint)
-    }
+    // 2. spawn all the tasks to the runtime.
+    let join_handlers = segments_runtime.try_spawn_batch(semaphore, futures).await?;
+
+    // 3. get all the result.
+    let joint = future::try_join_all(join_handlers)
+        .instrument(tracing::debug_span!("try_join_futures_all"))
+        .await
+        .map_err(|e| ErrorCode::StorageOther(format!("try join futures failure, {}", e)))?;
+
+    Ok(joint)
 }

src/query/storages/fuse/fuse/src/operations/mutation/compact/compact_transform.rs

@@ -92,6 +92,7 @@ pub struct CompactTransform {
 
     // Limit the memory size of the block read.
     max_memory: u64,
+    max_io_requests: usize,
     compact_tasks: VecDeque<CompactTask>,
     block_metas: Vec<Arc<BlockMeta>>,
     order: usize,
@@ -112,9 +113,10 @@ impl CompactTransform {
         thresholds: BlockCompactThresholds,
     ) -> Result<ProcessorPtr> {
         let settings = ctx.get_settings();
-        let max_memory_usage = (settings.get_max_memory_usage()? as f64 * 0.95) as u64;
+        let max_memory_usage = (settings.get_max_memory_usage()? as f64 * 0.8) as u64;
         let max_threads = settings.get_max_threads()?;
         let max_memory = max_memory_usage / max_threads;
+        let max_io_requests = settings.get_max_storage_io_requests()? as usize;
         Ok(ProcessorPtr::create(Box::new(CompactTransform {
             state: State::Consume,
             input,
@@ -125,6 +127,7 @@ impl CompactTransform {
             location_gen,
             dal,
             max_memory,
+            max_io_requests,
             compact_tasks: VecDeque::new(),
             block_metas: Vec::new(),
             order: 0,
@@ -326,7 +329,10 @@ impl Processor for CompactTransform {
                         acc + memory
                     });
 
-                    if memory_usage > self.max_memory {
+                    if (memory_usage > self.max_memory
+                        || task_futures.len() + metas.len() > self.max_io_requests)
+                        && !task_futures.is_empty()
+                    {
                         self.compact_tasks.push_front(task);
                         break;
                     }