diff --git a/src/mito2/src/access_layer.rs b/src/mito2/src/access_layer.rs index 05595bbae3b..5587fe574b1 100644 --- a/src/mito2/src/access_layer.rs +++ b/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 => { diff --git a/src/mito2/src/flush.rs b/src/mito2/src/flush.rs index d9494ed79d8..0da04eeb7d4 100644 --- a/src/mito2/src/flush.rs +++ b/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 { + 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>, + prepare_source: Duration, } struct DoFlushMemtablesResult { diff --git a/src/mito2/src/sst/parquet/writer.rs b/src/mito2/src/sst/parquet/writer.rs index 1525c3bcee9..06bda44a62b 100644 --- a/src/mito2/src/sst/parquet/writer.rs +++ b/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)