fix: concurrency control of segment loading
use std::sync::Arc;
+use common_base::base::tokio::sync::Semaphore;
use common_base::base::Runtime;
use common_base::base::TrySpawn;
use common_catalog::table_context::TableContext;
@@ -110,13 +111,18 @@ impl BlockPruner {
ctx.get_settings().get_max_threads()? as usize,
Some("pruning-worker".to_owned()),
)?;
+ let semaphore = Arc::new(Semaphore::new(FUTURE_BUFFER_SIZE));
let mut join_handlers = Vec::with_capacity(segment_locs.len());
for (idx, (seg_loc, ver)) in segment_locs.into_iter().enumerate() {
let ctx = ctx.clone();
let range_filter_pruner = range_filter_pruner.clone();
let bloom_filter_pruner = bloom_filter_pruner.clone();
let limiter = limiter.clone();
+ let semaphore = semaphore.clone();
let segment_pruning_fut = async move {
+ let _permit = semaphore.acquire().await.map_err(|e| {
+ ErrorCode::StorageOther(format!("acquire permit failure, {}", e))
+ })?;
let segment_reader = MetaReaders::segment_info_reader(ctx.as_ref());
if limiter.exceeded() {
// before read segment info, check if limit already exceeded
