feat: add per-partition convert, result cache metrics (#7539)

* fix: show convert cost in explain analyze verbose

Signed-off-by: evenyag <realevenyag@gmail.com>

* fix: increase puffin metadata cache metric

Signed-off-by: evenyag <realevenyag@gmail.com>

* feat: add result cache hit/miss to filter metrics

Signed-off-by: evenyag <realevenyag@gmail.com>

* feat: print flat format in debug

Signed-off-by: evenyag <realevenyag@gmail.com>

* test: update sqlness test

Signed-off-by: evenyag <realevenyag@gmail.com>

* feat: make scan cost contains part/reader build cost

Signed-off-by: evenyag <realevenyag@gmail.com>

* feat: collect divider cost

Signed-off-by: evenyag <realevenyag@gmail.com>

* refactor: remove unused field in ScannerMetrics

Signed-off-by: evenyag <realevenyag@gmail.com>

* feat: collect metadata read bytes

Signed-off-by: evenyag <realevenyag@gmail.com>

* chore: collect read metrics in get_parquet_meta_data

Signed-off-by: evenyag <realevenyag@gmail.com>

---------

Signed-off-by: evenyag <realevenyag@gmail.com>

Author: evenyag

src/mito2/src/cache.rs

@@ -350,7 +350,10 @@ impl CacheManager {
         // Try to get metadata from write cache
         let key = IndexKey::new(file_id.region_id(), file_id.file_id(), FileType::Parquet);
         if let Some(write_cache) = &self.write_cache
-            && let Some(metadata) = write_cache.file_cache().get_parquet_meta_data(key).await
+            && let Some(metadata) = write_cache
+                .file_cache()
+                .get_parquet_meta_data(key, metrics)
+                .await
         {
             metrics.file_cache_hit += 1;
             let metadata = Arc::new(metadata);

src/mito2/src/cache/file_cache.rs

@@ -43,6 +43,7 @@ use crate::metrics::{
 use crate::region::opener::RegionLoadCacheTask;
 use crate::sst::parquet::helper::fetch_byte_ranges;
 use crate::sst::parquet::metadata::MetadataLoader;
+use crate::sst::parquet::reader::MetadataCacheMetrics;
 
 /// Subdirectory of cached files for write.
 ///
@@ -566,7 +567,11 @@ impl FileCache {
 
     /// Get the parquet metadata in file cache.
     /// If the file is not in the cache or fail to load metadata, return None.
-    pub(crate) async fn get_parquet_meta_data(&self, key: IndexKey) -> Option<ParquetMetaData> {
+    pub(crate) async fn get_parquet_meta_data(
+        &self,
+        key: IndexKey,
+        cache_metrics: &mut MetadataCacheMetrics,
+    ) -> Option<ParquetMetaData> {
         // Check if file cache contains the key
         if let Some(index_value) = self.inner.parquet_index.get(&key).await {
             // Load metadata from file cache
@@ -575,7 +580,7 @@ impl FileCache {
             let file_size = index_value.file_size as u64;
             let metadata_loader = MetadataLoader::new(local_store, &file_path, file_size);
 
-            match metadata_loader.load().await {
+            match metadata_loader.load(cache_metrics).await {
                 Ok(metadata) => {
                     CACHE_HIT
                         .with_label_values(&[key.file_type.metric_label()])

src/mito2/src/read.rs

@@ -1058,10 +1058,6 @@ impl<T: BatchReader + ?Sized> BatchReader for Box<T> {
 /// Local metrics for scanners.
 #[derive(Debug, Default)]
 pub(crate) struct ScannerMetrics {
-    /// Duration to prepare the scan task.
-    prepare_scan_cost: Duration,
-    /// Duration to build the (merge) reader.
-    build_reader_cost: Duration,
     /// Duration to scan data.
     scan_cost: Duration,
     /// Duration while waiting for `yield`.
@@ -1070,10 +1066,6 @@ pub(crate) struct ScannerMetrics {
     num_batches: usize,
     /// Number of rows returned.
     num_rows: usize,
-    /// Number of mem ranges scanned.
-    num_mem_ranges: usize,
-    /// Number of file ranges scanned.
-    num_file_ranges: usize,
 }
 
 #[cfg(test)]

src/mito2/src/read/scan_region.rs

@@ -1498,6 +1498,7 @@ impl StreamContext {
                         .entries(self.input.files.iter().map(|file| FileWrapper { file }))
                         .finish()?;
                 }
+                write!(f, ", \"flat_format\": {}", self.input.flat_format)?;
 
                 #[cfg(feature = "enterprise")]
                 self.format_extension_ranges(f)?;

src/mito2/src/read/scan_util.rs

@@ -115,6 +115,8 @@ pub(crate) struct ScanMetricsSet {
     scan_cost: Duration,
     /// Duration while waiting for `yield`.
     yield_cost: Duration,
+    /// Duration to convert [`Batch`]es.
+    convert_cost: Option<Time>,
     /// Duration of the scan.
     total_cost: Duration,
     /// Number of rows returned.
@@ -165,6 +167,18 @@ pub(crate) struct ScanMetricsSet {
     rows_vector_filtered: usize,
     /// Number of rows filtered by precise filter.
     rows_precise_filtered: usize,
+    /// Number of index result cache hits for fulltext index.
+    fulltext_index_cache_hit: usize,
+    /// Number of index result cache misses for fulltext index.
+    fulltext_index_cache_miss: usize,
+    /// Number of index result cache hits for inverted index.
+    inverted_index_cache_hit: usize,
+    /// Number of index result cache misses for inverted index.
+    inverted_index_cache_miss: usize,
+    /// Number of index result cache hits for bloom filter index.
+    bloom_filter_cache_hit: usize,
+    /// Number of index result cache misses for bloom filter index.
+    bloom_filter_cache_miss: usize,
     /// Number of record batches read from SST.
     num_sst_record_batches: usize,
     /// Number of batches decoded from SST.
@@ -187,6 +201,8 @@ pub(crate) struct ScanMetricsSet {
     distributor_scan_cost: Duration,
     /// Duration of the series distributor to yield.
     distributor_yield_cost: Duration,
+    /// Duration spent in divider operations.
+    distributor_divider_cost: Duration,
 
     /// Merge metrics.
     merge_metrics: MergeMetrics,
@@ -247,6 +263,7 @@ impl fmt::Debug for ScanMetricsSet {
             build_reader_cost,
             scan_cost,
             yield_cost,
+            convert_cost,
             total_cost,
             num_rows,
             num_batches,
@@ -266,6 +283,12 @@ impl fmt::Debug for ScanMetricsSet {
             rows_bloom_filtered,
             rows_vector_filtered,
             rows_precise_filtered,
+            fulltext_index_cache_hit,
+            fulltext_index_cache_miss,
+            inverted_index_cache_hit,
+            inverted_index_cache_miss,
+            bloom_filter_cache_hit,
+            bloom_filter_cache_miss,
             num_sst_record_batches,
             num_sst_batches,
             num_sst_rows,
@@ -276,6 +299,7 @@ impl fmt::Debug for ScanMetricsSet {
             num_distributor_batches,
             distributor_scan_cost,
             distributor_yield_cost,
+            distributor_divider_cost,
             merge_metrics,
             dedup_metrics,
             stream_eof,
@@ -313,6 +337,12 @@ impl fmt::Debug for ScanMetricsSet {
             \"first_poll\":\"{first_poll:?}\""
         )?;
 
+        // Write convert_cost if present
+        if let Some(time) = convert_cost {
+            let duration = Duration::from_nanos(time.value() as u64);
+            write!(f, ", \"convert_cost\":\"{duration:?}\"")?;
+        }
+
         // Write non-zero filter counters
         if *rg_fulltext_filtered > 0 {
             write!(f, ", \"rg_fulltext_filtered\":{rg_fulltext_filtered}")?;
@@ -344,6 +374,36 @@ impl fmt::Debug for ScanMetricsSet {
         if *rows_precise_filtered > 0 {
             write!(f, ", \"rows_precise_filtered\":{rows_precise_filtered}")?;
         }
+        if *fulltext_index_cache_hit > 0 {
+            write!(
+                f,
+                ", \"fulltext_index_cache_hit\":{fulltext_index_cache_hit}"
+            )?;
+        }
+        if *fulltext_index_cache_miss > 0 {
+            write!(
+                f,
+                ", \"fulltext_index_cache_miss\":{fulltext_index_cache_miss}"
+            )?;
+        }
+        if *inverted_index_cache_hit > 0 {
+            write!(
+                f,
+                ", \"inverted_index_cache_hit\":{inverted_index_cache_hit}"
+            )?;
+        }
+        if *inverted_index_cache_miss > 0 {
+            write!(
+                f,
+                ", \"inverted_index_cache_miss\":{inverted_index_cache_miss}"
+            )?;
+        }
+        if *bloom_filter_cache_hit > 0 {
+            write!(f, ", \"bloom_filter_cache_hit\":{bloom_filter_cache_hit}")?;
+        }
+        if *bloom_filter_cache_miss > 0 {
+            write!(f, ", \"bloom_filter_cache_miss\":{bloom_filter_cache_miss}")?;
+        }
 
         // Write non-zero distributor metrics
         if *num_series_send_timeout > 0 {
@@ -370,6 +430,12 @@ impl fmt::Debug for ScanMetricsSet {
                 ", \"distributor_yield_cost\":\"{distributor_yield_cost:?}\""
             )?;
         }
+        if !distributor_divider_cost.is_zero() {
+            write!(
+                f,
+                ", \"distributor_divider_cost\":\"{distributor_divider_cost:?}\""
+            )?;
+        }
 
         // Write non-zero memtable metrics
         if *mem_rows > 0 {
@@ -478,27 +544,25 @@ impl ScanMetricsSet {
         self
     }
 
+    /// Attaches the `convert_cost` to the metrics set.
+    fn with_convert_cost(mut self, time: Time) -> Self {
+        self.convert_cost = Some(time);
+        self
+    }
+
     /// Merges the local scanner metrics.
     fn merge_scanner_metrics(&mut self, other: &ScannerMetrics) {
         let ScannerMetrics {
-            prepare_scan_cost,
-            build_reader_cost,
             scan_cost,
             yield_cost,
             num_batches,
             num_rows,
-            num_mem_ranges,
-            num_file_ranges,
         } = other;
 
-        self.prepare_scan_cost += *prepare_scan_cost;
-        self.build_reader_cost += *build_reader_cost;
         self.scan_cost += *scan_cost;
         self.yield_cost += *yield_cost;
         self.num_rows += *num_rows;
         self.num_batches += *num_batches;
-        self.num_mem_ranges += *num_mem_ranges;
-        self.num_file_ranges += *num_file_ranges;
     }
 
     /// Merges the local reader metrics.
@@ -519,6 +583,12 @@ impl ScanMetricsSet {
                     rows_bloom_filtered,
                     rows_vector_filtered,
                     rows_precise_filtered,
+                    fulltext_index_cache_hit,
+                    fulltext_index_cache_miss,
+                    inverted_index_cache_hit,
+                    inverted_index_cache_miss,
+                    bloom_filter_cache_hit,
+                    bloom_filter_cache_miss,
                     inverted_index_apply_metrics,
                     bloom_filter_apply_metrics,
                     fulltext_index_apply_metrics,
@@ -548,6 +618,13 @@ impl ScanMetricsSet {
         self.rows_vector_filtered += *rows_vector_filtered;
         self.rows_precise_filtered += *rows_precise_filtered;
 
+        self.fulltext_index_cache_hit += *fulltext_index_cache_hit;
+        self.fulltext_index_cache_miss += *fulltext_index_cache_miss;
+        self.inverted_index_cache_hit += *inverted_index_cache_hit;
+        self.inverted_index_cache_miss += *inverted_index_cache_miss;
+        self.bloom_filter_cache_hit += *bloom_filter_cache_hit;
+        self.bloom_filter_cache_miss += *bloom_filter_cache_miss;
+
         self.num_sst_record_batches += *num_record_batches;
         self.num_sst_batches += *num_batches;
         self.num_sst_rows += *num_rows;
@@ -598,6 +675,7 @@ impl ScanMetricsSet {
             num_batches,
             scan_cost,
             yield_cost,
+            divider_cost,
         } = distributor_metrics;
 
         self.num_series_send_timeout += *num_series_send_timeout;
@@ -606,6 +684,7 @@ impl ScanMetricsSet {
         self.num_distributor_batches += *num_batches;
         self.distributor_scan_cost += *scan_cost;
         self.distributor_yield_cost += *yield_cost;
+        self.distributor_divider_cost += *divider_cost;
     }
 
     /// Observes metrics.
@@ -622,6 +701,11 @@ impl ScanMetricsSet {
         READ_STAGE_ELAPSED
             .with_label_values(&["yield"])
             .observe(self.yield_cost.as_secs_f64());
+        if let Some(time) = &self.convert_cost {
+            READ_STAGE_ELAPSED
+                .with_label_values(&["convert"])
+                .observe(Duration::from_nanos(time.value() as u64).as_secs_f64());
+        }
         READ_STAGE_ELAPSED
             .with_label_values(&["total"])
             .observe(self.total_cost.as_secs_f64());
@@ -746,21 +830,19 @@ impl Drop for PartitionMetricsInner {
 
         if self.explain_verbose {
             common_telemetry::info!(
-                "{} finished, region_id: {}, partition: {}, scan_metrics: {:?}, convert_batch_costs: {}",
+                "{} finished, region_id: {}, partition: {}, scan_metrics: {:?}",
                 self.scanner_type,
                 self.region_id,
                 self.partition,
                 metrics,
-                self.convert_cost,
             );
         } else {
             common_telemetry::debug!(
-                "{} finished, region_id: {}, partition: {}, scan_metrics: {:?}, convert_batch_costs: {}",
+                "{} finished, region_id: {}, partition: {}, scan_metrics: {:?}",
                 self.scanner_type,
                 self.region_id,
                 self.partition,
                 metrics,
-                self.convert_cost,
             );
         }
     }
@@ -807,7 +889,10 @@ impl PartitionMetrics {
         let partition_str = partition.to_string();
         let in_progress_scan = IN_PROGRESS_SCAN.with_label_values(&[scanner_type, &partition_str]);
         in_progress_scan.inc();
-        let metrics = ScanMetricsSet::default().with_prepare_scan_cost(query_start.elapsed());
+        let convert_cost = MetricBuilder::new(metrics_set).subset_time("convert_cost", partition);
+        let metrics = ScanMetricsSet::default()
+            .with_prepare_scan_cost(query_start.elapsed())
+            .with_convert_cost(convert_cost.clone());
         let inner = PartitionMetricsInner {
             region_id,
             partition,
@@ -822,7 +907,7 @@ impl PartitionMetrics {
                 .subset_time("build_reader_cost", partition),
             scan_cost: MetricBuilder::new(metrics_set).subset_time("scan_cost", partition),
             yield_cost: MetricBuilder::new(metrics_set).subset_time("yield_cost", partition),
-            convert_cost: MetricBuilder::new(metrics_set).subset_time("convert_cost", partition),
+            convert_cost,
             elapsed_compute: MetricBuilder::new(metrics_set).elapsed_compute(partition),
         };
         Self(Arc::new(inner))
@@ -874,9 +959,6 @@ impl PartitionMetrics {
 
     /// Merges [ScannerMetrics], `build_reader_cost`, `scan_cost` and `yield_cost`.
     pub(crate) fn merge_metrics(&self, metrics: &ScannerMetrics) {
-        self.0
-            .build_reader_cost
-            .add_duration(metrics.build_reader_cost);
         self.0.scan_cost.add_duration(metrics.scan_cost);
         self.record_elapsed_compute(metrics.scan_cost);
         self.0.yield_cost.add_duration(metrics.yield_cost);
@@ -956,6 +1038,8 @@ pub(crate) struct SeriesDistributorMetrics {
     pub(crate) scan_cost: Duration,
     /// Duration of the series distributor to yield.
     pub(crate) yield_cost: Duration,
+    /// Duration spent in divider operations.
+    pub(crate) divider_cost: Duration,
 }
 
 /// Scans memtable ranges at `index`.