chore: add metrics for copy, test slow reason

src/common/storage/src/common_metrics/copy.rs

@@ -40,3 +40,24 @@ pub fn metrics_inc_copy_read_size_bytes(c: u64) {
 pub fn metrics_inc_copy_read_part_cost_milliseconds(c: u64) {
     increment_gauge!(key!("copy_read_part_cost_milliseconds"), c as f64);
 }
+
+pub fn metrics_inc_filter_out_copied_files_request_milliseconds(c: u64) {
+    increment_gauge!(
+        key!("filter_out_copied_files_request_milliseconds"),
+        c as f64
+    );
+}
+
+pub fn metrics_inc_filter_out_copied_files_entire_milliseconds(c: u64) {
+    increment_gauge!(
+        key!("filter_out_copied_files_entire_milliseconds"),
+        c as f64
+    );
+}
+
+pub fn metrics_inc_collect_files_get_all_source_files_milliseconds(c: u64) {
+    increment_gauge!(
+        key!("collect_files_get_all_source_files_milliseconds"),
+        c as f64
+    );
+}

src/query/service/src/sessions/query_ctx.rs

@@ -23,6 +23,7 @@ use std::str::FromStr;
 use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
+use std::time::Instant;
 use std::time::SystemTime;
 
 use chrono_tz::Tz;
@@ -58,6 +59,7 @@ use common_pipeline_core::InputError;
 use common_settings::ChangeValue;
 use common_settings::Settings;
 use common_sql::IndexType;
+use common_storage::common_metrics::copy::metrics_inc_filter_out_copied_files_request_milliseconds;
 use common_storage::DataOperator;
 use common_storage::StageFileInfo;
 use common_storage::StorageMetrics;
@@ -662,10 +664,15 @@ impl TableContext for QueryContext {
         for chunk in files.chunks(batch_size) {
             let files = chunk.iter().map(|v| v.path.clone()).collect::<Vec<_>>();
             let req = GetTableCopiedFileReq { table_id, files };
+            let start_request = Instant::now();
             let copied_files = catalog
                 .get_table_copied_file_info(&tenant, database_name, req)
                 .await?
                 .file_info;
+
+            metrics_inc_filter_out_copied_files_request_milliseconds(
+                Instant::now().duration_since(start_request).as_millis() as u64,
+            );
             // Colored
             for file in chunk {
                 if let Some(copied_file) = copied_files.get(&file.path) {

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

@@ -24,6 +24,8 @@ use common_expression::DataSchemaRef;
 use common_expression::Scalar;
 use common_meta_app::principal::StageInfo;
 use common_meta_app::schema::CatalogInfo;
+use common_storage::common_metrics::copy::metrics_inc_collect_files_get_all_source_files_milliseconds;
+use common_storage::common_metrics::copy::metrics_inc_filter_out_copied_files_entire_milliseconds;
 use common_storage::init_stage_operator;
 use common_storage::StageFileInfo;
 use log::info;
@@ -134,6 +136,10 @@ impl CopyIntoTablePlan {
 
         let num_all_files = all_source_file_infos.len();
 
+        let end_get_all_source = Instant::now();
+        let cost_get_all_files = end_get_all_source.duration_since(start).as_millis();
+        metrics_inc_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));
 
         let need_copy_file_infos = if self.force {
@@ -145,6 +151,7 @@ impl CopyIntoTablePlan {
         } else {
             // Status.
             ctx.set_status_info("begin filtering out copied files");
+
             let files = ctx
                 .filter_out_copied_files(
                     self.catalog_info.catalog_name(),
@@ -158,6 +165,13 @@ impl CopyIntoTablePlan {
                 "end filtering out copied files: {}",
                 num_all_files
             ));
+
+            let end_filter_out = Instant::now();
+            let cost_filter_out = end_filter_out
+                .duration_since(end_get_all_source)
+                .as_millis();
+            metrics_inc_filter_out_copied_files_entire_milliseconds(cost_filter_out as u64);
+
             files
         };