add more metrics

Signed-off-by: luofucong <luofc@foxmail.com>

src/mito2/src/access_layer.rs

@@ -63,14 +63,44 @@ pub enum WriteType {
     Compaction,
 }
 
+/// Timings for SST write stages.
 #[derive(Debug)]
 pub struct Metrics {
+    /// SST write operation type.
     pub(crate) write_type: WriteType,
+    /// Time spent waiting for a flush concurrency permit.
+    pub(crate) wait_flush_permit: Duration,
+    /// Time spent building memtable ranges.
+    pub(crate) build_ranges: Duration,
+    /// Time spent preparing flat sources.
+    pub(crate) prepare_source: Duration,
+    /// Time spent initializing the Parquet writer.
+    pub(crate) init_writer: Duration,
+    /// Time spent fetching source batches.
     pub(crate) iter_source: Duration,
+    /// Time spent converting source batches to the write schema.
+    pub(crate) convert_batch: Duration,
+    /// Time spent writing batches to the Parquet writer.
     pub(crate) write_batch: Duration,
+    /// Time spent updating SST statistics.
+    pub(crate) update_stats: Duration,
+    /// Time spent updating indexes with source batches.
     pub(crate) update_index: Duration,
+    /// Time spent finishing indexes.
+    pub(crate) finish_index: Duration,
+    /// Time spent flushing the final Parquet row group.
+    pub(crate) flush_writer: Duration,
+    /// Time spent closing the Parquet writer and completing the object write.
+    pub(crate) close_writer: Duration,
+    /// Time spent building SST information from Parquet metadata.
+    pub(crate) build_sst_info: Duration,
+    /// Time spent converting SST information to file metadata.
+    pub(crate) build_file_meta: Duration,
+    /// Time spent uploading Parquet files when using the write cache.
     pub(crate) upload_parquet: Duration,
+    /// Time spent uploading Puffin files.
     pub(crate) upload_puffin: Duration,
+    /// Time spent compacting memtables before flush.
     pub(crate) compact_memtable: Duration,
 }
 
@@ -78,9 +108,20 @@ impl Metrics {
     pub fn new(write_type: WriteType) -> Self {
         Self {
             write_type,
+            wait_flush_permit: Default::default(),
+            build_ranges: Default::default(),
+            prepare_source: Default::default(),
+            init_writer: Default::default(),
             iter_source: Default::default(),
+            convert_batch: Default::default(),
             write_batch: Default::default(),
+            update_stats: Default::default(),
             update_index: Default::default(),
+            finish_index: Default::default(),
+            flush_writer: Default::default(),
+            close_writer: Default::default(),
+            build_sst_info: Default::default(),
+            build_file_meta: Default::default(),
             upload_parquet: Default::default(),
             upload_puffin: Default::default(),
             compact_memtable: Default::default(),
@@ -89,9 +130,20 @@ impl Metrics {
 
     pub(crate) fn merge(mut self, other: Self) -> Self {
         assert_eq!(self.write_type, other.write_type);
+        self.wait_flush_permit += other.wait_flush_permit;
+        self.build_ranges += other.build_ranges;
+        self.prepare_source += other.prepare_source;
+        self.init_writer += other.init_writer;
         self.iter_source += other.iter_source;
+        self.convert_batch += other.convert_batch;
         self.write_batch += other.write_batch;
+        self.update_stats += other.update_stats;
         self.update_index += other.update_index;
+        self.finish_index += other.finish_index;
+        self.flush_writer += other.flush_writer;
+        self.close_writer += other.close_writer;
+        self.build_sst_info += other.build_sst_info;
+        self.build_file_meta += other.build_file_meta;
         self.upload_parquet += other.upload_parquet;
         self.upload_puffin += other.upload_puffin;
         self.compact_memtable += other.compact_memtable;
@@ -119,7 +171,7 @@ impl Metrics {
                 if !self.compact_memtable.is_zero() {
                     FLUSH_ELAPSED
                         .with_label_values(&["compact_memtable"])
-                        .observe(self.upload_puffin.as_secs_f64());
+                        .observe(self.compact_memtable.as_secs_f64());
                 }
             }
             WriteType::Compaction => {

src/mito2/src/flush.rs

@@ -18,7 +18,7 @@ use std::collections::HashMap;
 use std::num::NonZeroU64;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
-use std::time::Instant;
+use std::time::{Duration, Instant};
 
 use bytes::Bytes;
 use common_telemetry::{debug, error, info};
@@ -559,8 +559,10 @@ impl RegionFlushTask {
                 mem_stats.bytes_allocated() as u64,
             )]);
             // Sets `for_flush` flag and propagates the reader batch size.
+            let start = Instant::now();
             let mem_ranges =
                 mem.ranges(None, RangesOptions::for_flush().with_batch_size(batch_size))?;
+            flush_metrics.build_ranges += start.elapsed();
             let num_mem_ranges = mem_ranges.ranges.len();
 
             // Aggregate stats from all ranges
@@ -576,9 +578,11 @@ impl RegionFlushTask {
                 num_encoded,
                 num_sources,
                 results,
+                prepare_source,
             } = self
                 .flush_flat_mem_ranges(version, &write_opts, mem_ranges)
                 .await?;
+            flush_metrics.prepare_source += prepare_source;
             encoded_part_count += num_encoded;
             for (source_idx, result) in results.into_iter().enumerate() {
                 let (max_sequence, ssts_written, metrics) = result?;
@@ -598,6 +602,7 @@ impl RegionFlushTask {
 
                 flush_metrics = flush_metrics.merge(metrics);
 
+                let start = Instant::now();
                 for sst_info in &ssts_written {
                     flushed_bytes += sst_info.file_size;
                     let pk_range = sst_info
@@ -615,6 +620,7 @@ impl RegionFlushTask {
                 if hook.is_some() {
                     all_sst_infos.extend(ssts_written);
                 }
+                flush_metrics.build_file_meta += start.elapsed();
             }
 
             common_telemetry::debug!(
@@ -646,6 +652,7 @@ impl RegionFlushTask {
         write_opts: &WriteOptions,
         mem_ranges: MemtableRanges,
     ) -> Result<FlushFlatMemResult> {
+        let start = Instant::now();
         let batch_schema = to_flat_sst_arrow_schema(
             &version.metadata,
             &FlatSchemaOptions::from_encoding(version.metadata.primary_key_encoding),
@@ -658,6 +665,7 @@ impl RegionFlushTask {
             &version.options,
             field_column_start,
         )?;
+        let prepare_source = start.elapsed();
         let mut tasks = Vec::with_capacity(flat_sources.encoded.len() + flat_sources.sources.len());
         let num_encoded = flat_sources.encoded.len();
         for (source, max_sequence) in flat_sources.sources {
@@ -666,8 +674,10 @@ impl RegionFlushTask {
             let write_opts = write_opts.clone();
             let semaphore = self.flush_semaphore.clone();
             let task = common_runtime::spawn_global(async move {
+                let start = Instant::now();
                 let _permit = semaphore.acquire().await.unwrap();
                 let mut metrics = Metrics::new(WriteType::Flush);
+                metrics.wait_flush_permit = start.elapsed();
                 let ssts = access_layer
                     .write_sst(write_request, &write_opts, &mut metrics)
                     .await?;
@@ -682,10 +692,13 @@ impl RegionFlushTask {
             let region_id = version.metadata.region_id;
             let semaphore = self.flush_semaphore.clone();
             let task = common_runtime::spawn_global(async move {
+                let start = Instant::now();
                 let _permit = semaphore.acquire().await.unwrap();
-                let metrics = access_layer
+                let wait_flush_permit = start.elapsed();
+                let mut metrics = access_layer
                     .put_sst(&encoded.data, region_id, &encoded.sst_info, &cache_manager)
                     .await?;
+                metrics.wait_flush_permit = wait_flush_permit;
                 FLUSH_FILE_TOTAL.inc();
                 Ok((max_sequence, smallvec![encoded.sst_info], metrics))
             });
@@ -699,6 +712,7 @@ impl RegionFlushTask {
             num_encoded,
             num_sources,
             results,
+            prepare_source,
         })
     }
 
@@ -806,6 +820,7 @@ struct FlushFlatMemResult {
     num_encoded: usize,
     num_sources: usize,
     results: Vec<Result<(SequenceNumber, SstInfoArray, Metrics)>>,
+    prepare_source: Duration,
 }
 
 struct DoFlushMemtablesResult {

src/mito2/src/sst/parquet/writer.rs

@@ -177,7 +177,9 @@ where
         metrics: &'a mut Metrics,
     ) -> ParquetWriter<'a, F, I, P> {
         let init_file = FileId::random();
+        let start = Instant::now();
         let indexer = indexer_builder.build(init_file, 0, None).await;
+        metrics.init_writer += start.elapsed();
 
         ParquetWriter {
             path_provider,
@@ -218,7 +220,9 @@ where
             let mut index_output = IndexOutput::default();
             match self.index_config.build_mode {
                 IndexBuildMode::Sync => {
+                    let start = Instant::now();
                     index_output = self.current_indexer.as_mut().unwrap().finish().await;
+                    self.metrics.finish_index += start.elapsed();
                 }
                 IndexBuildMode::Async => {
                     debug!(
@@ -227,11 +231,16 @@ where
                     );
                 }
             }
+            let start = Instant::now();
             current_writer.flush().await.context(WriteParquetSnafu)?;
+            self.metrics.flush_writer += start.elapsed();
 
+            let start = Instant::now();
             let parquet_metadata = current_writer.close().await.context(WriteParquetSnafu)?;
+            self.metrics.close_writer += start.elapsed();
             let file_size = self.bytes_written.load(Ordering::Relaxed) as u64;
 
+            let start = Instant::now();
             // Safety: num rows > 0 so we must have min/max.
             let time_range = stats.time_range.unwrap();
 
@@ -258,6 +267,7 @@ where
                 index_metadata: index_output,
                 num_series,
             });
+            self.metrics.build_sst_info += start.elapsed();
             self.current_file = FileId::random();
             self.bytes_written.store(0, Ordering::Relaxed)
         };
@@ -346,7 +356,9 @@ where
         {
             match record_batch {
                 Ok(batch) => {
+                    let start = Instant::now();
                     stats.update_flat(&batch)?;
+                    self.metrics.update_stats += start.elapsed();
                     if matches!(self.index_config.build_mode, IndexBuildMode::Sync) {
                         let start = Instant::now();
                         // safety: self.current_indexer must be set when first batch has been written.
@@ -413,7 +425,9 @@ where
         };
         self.metrics.iter_source += start.elapsed();
 
+        let start = Instant::now();
         let arrow_batch = converter.convert_batch(&record_batch)?;
+        self.metrics.convert_batch += start.elapsed();
 
         let start = Instant::now();
         self.maybe_init_writer(arrow_batch.schema_ref(), opts)