chore: fix unit test and clean code

Author: dqhl76

src/query/service/tests/it/storages/fuse/pruning_pipeline.rs

@@ -62,7 +62,7 @@ use databend_storages_common_table_meta::table::OPT_KEY_DATABASE_ID;
 use databend_storages_common_table_meta::table::OPT_KEY_SNAPSHOT_LOCATION;
 use opendal::Operator;
 
-async fn apply_block_pruning(
+async fn apply_snapshot_pruning(
     table_snapshot: Arc<TableSnapshot>,
     schema: TableSchemaRef,
     push_down: &Option<PushDownInfo>,
@@ -102,12 +102,9 @@ async fn apply_block_pruning(
         GlobalIORuntime::instance().spawn(async move {
             // avoid block global io runtime
             let runtime = Runtime::with_worker_threads(2, None)?;
-
             let join_handler = runtime.spawn(async move {
-                let segment_pruned_result =
-                    fuse_pruner.clone().segment_pruning(segment_locs).await?;
-                for segment in segment_pruned_result {
-                    let _ = segment_tx.send(Ok(segment)).await;
+                for segment in segment_locs {
+                    let _ = segment_tx.send(segment).await;
                 }
                 Ok::<_, ErrorCode>(())
             });
@@ -140,7 +137,7 @@ async fn apply_block_pruning(
 }
 
 #[tokio::test(flavor = "multi_thread")]
-async fn test_block_pruner() -> Result<()> {
+async fn test_snapshot_pruner() -> Result<()> {
     let fixture = TestFixture::setup().await?;
     let ctx = fixture.new_query_ctx().await?;
 
@@ -320,7 +317,7 @@ async fn test_block_pruner() -> Result<()> {
 
     for (id, (extra, expected_blocks, expected_rows)) in extras.into_iter().enumerate() {
         let cache_key = Some(format!("test_block_pruner_{}", id));
-        let parts = apply_block_pruning(
+        let parts = apply_snapshot_pruning(
             snapshot.clone(),
             table.get_table_info().schema(),
             &extra,

src/query/storages/fuse/src/pruning/fuse_pruner.rs

@@ -443,53 +443,6 @@ impl FusePruner {
         }
     }
 
-    // Temporarily using, will remove after finish pruning refactor.
-    pub async fn segment_pruning(
-        &self,
-        mut segment_locs: Vec<SegmentLocation>,
-    ) -> Result<Vec<(SegmentLocation, Arc<CompactSegmentInfo>)>> {
-        // Segment pruner.
-        let segment_pruner =
-            SegmentPruner::create(self.pruning_ctx.clone(), self.table_schema.clone())?;
-
-        let mut remain = segment_locs.len() % self.max_concurrency;
-        let batch_size = segment_locs.len() / self.max_concurrency;
-        let mut works = Vec::with_capacity(self.max_concurrency);
-        while !segment_locs.is_empty() {
-            let gap_size = std::cmp::min(1, remain);
-            let batch_size = batch_size + gap_size;
-            remain -= gap_size;
-
-            let mut batch = segment_locs.drain(0..batch_size).collect::<Vec<_>>();
-            works.push(self.pruning_ctx.pruning_runtime.spawn({
-                let segment_pruner = segment_pruner.clone();
-                let pruning_ctx = self.pruning_ctx.clone();
-                async move {
-                    // Build pruning tasks.
-                    if let Some(internal_column_pruner) = &pruning_ctx.internal_column_pruner {
-                        batch = batch
-                            .into_iter()
-                            .filter(|segment| {
-                                internal_column_pruner
-                                    .should_keep(SEGMENT_NAME_COL_NAME, &segment.location.0)
-                            })
-                            .collect::<Vec<_>>();
-                    }
-                    let pruned_segments = segment_pruner.pruning(batch).await?;
-                    Result::<_>::Ok(pruned_segments)
-                }
-            }));
-        }
-
-        let workers = futures::future::try_join_all(works).await?;
-        let mut pruned_segments = vec![];
-        for worker in workers {
-            let res = worker?;
-            pruned_segments.extend(res);
-        }
-        Ok(pruned_segments)
-    }
-
     fn extract_block_metas(
         segment_path: &str,
         segment: &CompactSegmentInfo,