chore: tweak deletion batch size & log messages (#15005)

* adjust purge batch size

* add explicit timing logs for list and purge

Author: dantengsky

src/query/service/src/pipelines/builders/builder_on_finished.rs

@@ -24,6 +24,7 @@ use databend_common_pipeline_core::Pipeline;
 use databend_common_storages_fuse::io::Files;
 use databend_common_storages_stage::StageTable;
 use log::error;
+use log::info;
 
 use crate::pipelines::PipelineBuilder;
 use crate::sessions::QueryContext;
@@ -66,16 +67,7 @@ impl PipelineBuilder {
                         // 2. Try to purge copied files if purge option is true, if error will skip.
                         // If a file is already copied(status with AlreadyCopied) we will try to purge them.
                         if !is_active && copy_purge_option {
-                            let start = Instant::now();
                             Self::try_purge_files(ctx.clone(), &stage_info, &files).await;
-
-                            // Perf.
-                            {
-                                metrics_inc_copy_purge_files_counter(files.len() as u32);
-                                metrics_inc_copy_purge_files_cost_milliseconds(
-                                    start.elapsed().as_millis() as u32,
-                                );
-                            }
                         }
 
                         Ok(())
@@ -107,21 +99,17 @@ impl PipelineBuilder {
             }
         }
 
-        let start = Instant::now();
         Self::try_purge_files(ctx.clone(), &stage_info, &files).await;
 
-        // Perf.
-        {
-            metrics_inc_copy_purge_files_counter(files.len() as u32);
-            metrics_inc_copy_purge_files_cost_milliseconds(start.elapsed().as_millis() as u32);
-        }
         Ok(())
     }
 
     #[async_backtrace::framed]
     pub async fn try_purge_files(ctx: Arc<QueryContext>, stage_info: &StageInfo, files: &[String]) {
+        let start = Instant::now();
         let table_ctx: Arc<dyn TableContext> = ctx.clone();
         let op = StageTable::get_op(stage_info);
+
         match op {
             Ok(op) => {
                 let file_op = Files::create(table_ctx, op);
@@ -133,5 +121,18 @@ impl PipelineBuilder {
                 error!("Failed to get stage table op, error: {}", e);
             }
         }
+
+        let elapsed = start.elapsed();
+        info!(
+            "purged files: number {}, time used {:?} ",
+            files.len(),
+            elapsed
+        );
+
+        // Perf.
+        {
+            metrics_inc_copy_purge_files_counter(files.len() as u32);
+            metrics_inc_copy_purge_files_cost_milliseconds(elapsed.as_millis() as u32);
+        }
     }
 }

src/query/sql/src/planner/binder/binder.rs

@@ -15,6 +15,7 @@
 use std::collections::HashMap;
 use std::str::FromStr;
 use std::sync::Arc;
+use std::time::Instant;
 
 use chrono_tz::Tz;
 use databend_common_ast::ast::format_statement;
@@ -34,6 +35,7 @@ use databend_common_expression::Expr;
 use databend_common_functions::BUILTIN_FUNCTIONS;
 use databend_common_meta_app::principal::StageFileFormatType;
 use indexmap::IndexMap;
+use log::info;
 use log::warn;
 
 use super::Finder;
@@ -129,10 +131,12 @@ impl<'a> Binder {
     #[async_backtrace::framed]
     #[minitrace::trace]
     pub async fn bind(mut self, stmt: &Statement) -> Result<Plan> {
+        let start = Instant::now();
         self.ctx.set_status_info("binding");
         let mut init_bind_context = BindContext::new();
         let plan = self.bind_statement(&mut init_bind_context, stmt).await?;
         self.bind_query_index(&mut init_bind_context, &plan).await?;
+        info!("bind stmt to plan, time used: {:?}", start.elapsed());
         Ok(plan)
     }
 

src/query/sql/src/planner/planner.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::sync::Arc;
+use std::time::Instant;
 
 use databend_common_ast::ast::Expr;
 use databend_common_ast::ast::Literal;
@@ -68,6 +69,7 @@ impl Planner {
     #[async_backtrace::framed]
     #[minitrace::trace]
     pub async fn plan_sql(&mut self, sql: &str) -> Result<(Plan, PlanExtras)> {
+        let start = Instant::now();
         let settings = self.ctx.get_settings();
         let sql_dialect = settings.get_sql_dialect()?;
         // compile prql to sql for prql dialect
@@ -205,6 +207,7 @@ impl Planner {
                     tokens.extend(iter);
                 };
             } else {
+                info!("logical plan built, time used: {:?}", start.elapsed());
                 return res;
             }
         }

src/query/sql/src/planner/plans/copy_into_table.rs

@@ -150,7 +150,11 @@ impl CopyIntoTablePlan {
         let cost_get_all_files = end_get_all_source.duration_since(start).as_millis();
         metrics_inc_copy_collect_files_get_all_source_files_milliseconds(cost_get_all_files as u64);
 
-        ctx.set_status_info(&format!("end list files: got {} files", num_all_files));
+        ctx.set_status_info(&format!(
+            "end list files: got {} files, time used {:?}",
+            num_all_files,
+            start.elapsed()
+        ));
 
         let (need_copy_file_infos, duplicated) = if self.force {
             if !self.stage_table_info.stage_info.copy_options.purge
@@ -167,6 +171,7 @@ impl CopyIntoTablePlan {
             // Status.
             ctx.set_status_info("begin filtering out copied files");
 
+            let filter_start = Instant::now();
             let FilteredCopyFiles {
                 files_to_copy,
                 duplicated_files,
@@ -180,8 +185,9 @@ impl CopyIntoTablePlan {
                 )
                 .await?;
             ctx.set_status_info(&format!(
-                "end filtering out copied files: {}",
-                num_all_files
+                "end filtering out copied files: {}, time used {:?}",
+                num_all_files,
+                filter_start.elapsed()
             ));
 
             let end_filter_out = Instant::now();
@@ -194,11 +200,11 @@ impl CopyIntoTablePlan {
         };
 
         info!(
-            "copy: read files with max_files={:?} finished, all:{}, need copy:{}, elapsed:{}",
+            "copy: read files with max_files={:?} finished, all:{}, need copy:{}, elapsed:{:?}",
             max_files,
             num_all_files,
             need_copy_file_infos.len(),
-            start.elapsed().as_secs()
+            start.elapsed()
         );
 
         if need_copy_file_infos.is_empty() {

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

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::sync::Arc;
+use std::time::Instant;
 
 use databend_common_base::runtime::execute_futures_in_parallel;
 use databend_common_catalog::table_context::TableContext;
@@ -39,9 +40,23 @@ impl Files {
         &self,
         file_locations: impl IntoIterator<Item = impl AsRef<str>>,
     ) -> Result<()> {
-        let batch_size = 1000;
         let locations = Vec::from_iter(file_locations.into_iter().map(|v| v.as_ref().to_string()));
 
+        if locations.is_empty() {
+            return Ok(());
+        }
+
+        // adjusts batch_size according to the `max_threads` settings,
+        // limits its min/max value to 1 and 1000.
+        let threads_nums = self.ctx.get_settings().get_max_threads()? as usize;
+        let batch_size = (locations.len() / threads_nums).min(1000).max(1);
+
+        info!(
+            "remove file in batch, batch_size: {}, number of chunks {}",
+            batch_size,
+            (locations.len() / batch_size).max(1)
+        );
+
         if locations.len() <= batch_size {
             Self::delete_files(self.operator.clone(), locations).await?;
         } else {
@@ -53,8 +68,6 @@ 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;
-
             execute_futures_in_parallel(
                 tasks,
                 threads_nums,
@@ -69,14 +82,23 @@ impl Files {
 
     #[async_backtrace::framed]
     async fn delete_files(op: Operator, locations: Vec<String>) -> Result<()> {
+        let start = Instant::now();
         // temporary fix for https://github.com/datafuselabs/databend/issues/13804
         let locations = locations
             .into_iter()
             .map(|loc| loc.trim_start_matches('/').to_owned())
             .filter(|loc| !loc.is_empty())
             .collect::<Vec<_>>();
-        info!("deleting files: {:?}", &locations);
+        info!("deleting files {:?}", &locations);
+        let num_of_files = locations.len();
+
         op.remove(locations).await?;
+
+        info!(
+            "deleted files, number of files {}, time used {:?}",
+            num_of_files,
+            start.elapsed(),
+        );
         Ok(())
     }
 }

src/query/storages/fuse/src/operations/gc.rs

@@ -588,8 +588,7 @@ impl FuseTable {
         locations_to_be_purged: HashSet<String>,
     ) -> Result<()> {
         let fuse_file = Files::create(ctx.clone(), self.operator.clone());
-        let locations = Vec::from_iter(locations_to_be_purged);
-        fuse_file.remove_file_in_batch(&locations).await
+        fuse_file.remove_file_in_batch(locations_to_be_purged).await
     }
 
     // Purge file by location chunks.