add metrics

src/query/storages/fuse/src/metrics/fuse_metrics.rs

@@ -63,6 +63,14 @@ lazy_static! {
         register_histogram_in_milliseconds(key!("compact_block_read_milliseconds"));
     static ref COMPACT_BLOCK_BUILD_TASK_MILLISECONDS: Histogram =
         register_histogram_in_milliseconds(key!("compact_block_build_task_milliseconds"));
+    static ref RECLUSTER_BLOCK_NUMS_TO_READ: Counter =
+        register_counter(key!("recluster_block_nums_to_read"));
+    static ref RECLUSTER_BLOCK_BYTES_TO_READ: Counter =
+        register_counter(key!("recluster_block_bytes_to_read"));
+    static ref RECLUSTER_ROW_NUMS_TO_READ: Counter =
+        register_counter(key!("recluster_row_nums_to_read"));
+    static ref RECLUSTER_WRITE_BLOCK_NUMS: Counter =
+        register_counter(key!("recluster_write_block_nums"));
     static ref SEGMENTS_RANGE_PRUNING_BEFORE: Counter =
         register_counter(key!("segments_range_pruning_before"));
     static ref SEGMENTS_RANGE_PRUNING_AFTER: Counter =
@@ -413,3 +421,19 @@ pub fn metrics_inc_replace_deleted_blocks_rows(c: u64) {
 pub fn metrics_inc_replace_append_blocks_rows(c: u64) {
     REPLACE_INTO_APPEND_BLOCKS_ROWS.inc_by(c);
 }
+
+pub fn metrics_inc_recluster_block_nums_to_read(c: u64) {
+    RECLUSTER_BLOCK_NUMS_TO_READ.inc_by(c);
+}
+
+pub fn metrics_inc_recluster_block_bytes_to_read(c: u64) {
+    RECLUSTER_BLOCK_BYTES_TO_READ.inc_by(c);
+}
+
+pub fn metrics_inc_recluster_row_nums_to_read(c: u64) {
+    RECLUSTER_ROW_NUMS_TO_READ.inc_by(c);
+}
+
+pub fn metrics_inc_recluster_write_block_nums() {
+    RECLUSTER_WRITE_BLOCK_NUMS.inc();
+}

src/query/storages/fuse/src/operations/mutation/recluster_aggregator.rs

@@ -36,6 +36,7 @@ use storages_common_table_meta::meta::Versioned;
 use crate::io::SegmentsIO;
 use crate::io::SerializedSegment;
 use crate::io::TableMetaLocationGenerator;
+use crate::metrics::metrics_inc_recluster_write_block_nums;
 use crate::operations::common::AbortOperation;
 use crate::operations::common::CommitMeta;
 use crate::operations::common::ConflictResolveContext;
@@ -76,6 +77,7 @@ impl AsyncAccumulatingTransform for ReclusterAggregator {
             self.merged_blocks.push(Arc::new(meta));
             // Refresh status
             {
+                metrics_inc_recluster_write_block_nums();
                 let status = format!(
                     "recluster: generate new blocks:{}, cost:{} sec",
                     self.abort_operation.blocks.len(),

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

@@ -38,6 +38,9 @@ use log::warn;
 use opendal::Operator;
 use storages_common_table_meta::meta::CompactSegmentInfo;
 
+use crate::metrics::metrics_inc_recluster_block_bytes_to_read;
+use crate::metrics::metrics_inc_recluster_block_nums_to_read;
+use crate::metrics::metrics_inc_recluster_row_nums_to_read;
 use crate::operations::common::CommitSink;
 use crate::operations::common::MutationGenerator;
 use crate::operations::common::TransformSerializeBlock;
@@ -166,6 +169,7 @@ impl FuseTable {
         catalog_info: CatalogInfo,
         pipeline: &mut Pipeline,
     ) -> Result<()> {
+        let recluster_block_nums = task.parts.len();
         let block_thresholds = self.get_block_thresholds();
         let table_info = self.get_table_info();
         let description = task.stats.get_description(&table_info.desc);
@@ -187,6 +191,12 @@ impl FuseTable {
         // ReadDataKind to avoid OOM.
         self.do_read_data(ctx.clone(), &plan, pipeline)?;
 
+        {
+            metrics_inc_recluster_block_nums_to_read(recluster_block_nums as u64);
+            metrics_inc_recluster_block_bytes_to_read(task.total_bytes as u64);
+            metrics_inc_recluster_row_nums_to_read(task.total_rows as u64);
+        }
+
         let cluster_stats_gen =
             self.get_cluster_stats_gen(ctx.clone(), task.level + 1, block_thresholds, None)?;
         let operators = cluster_stats_gen.operators.clone();