collect metrics during merge

Author: ShiKaiWi

analytic_engine/src/instance/flush_compaction.rs

@@ -855,6 +855,7 @@ impl SpaceStore {
             let sequence = table_data.last_sequence();
             let mut builder = MergeBuilder::new(MergeConfig {
                 request_id,
+                metrics_collector: None,
                 // no need to set deadline for compaction
                 deadline: None,
                 space_id,

analytic_engine/src/instance/read.rs

@@ -20,7 +20,7 @@ use table_engine::{
     stream::{
         self, ErrWithSource, PartitionedStreams, RecordBatchStream, SendableRecordBatchStream,
     },
-    table::ReadRequest,
+    table::{Metric, ReadRequest},
 };
 use tokio::sync::mpsc::{self, Receiver};
 
@@ -66,6 +66,8 @@ pub enum Error {
 define_result!(Error);
 
 const RECORD_BATCH_READ_BUF_SIZE: usize = 1000;
+const READ_METRIC_MERGE_SORT: &str = "do_merge_sort";
+const READ_METRIC_ITER_NUM: &str = "iter_num";
 
 /// Check whether it needs to apply merge sorting when reading the table with
 /// the `table_options` by the `read_request`.
@@ -91,13 +93,18 @@ impl Instance {
 
         let table_data = space_table.table_data();
 
-        // Collect metrics.
-        table_data.metrics.on_read_request_begin();
-
         let iter_options = self.iter_options.clone();
         let table_options = table_data.table_options();
 
-        if need_merge_sort_streams(&table_data.table_options(), &request) {
+        // Collect metrics.
+        table_data.metrics.on_read_request_begin();
+        let need_merge_sort = need_merge_sort_streams(&table_options, &request);
+        request.metrics_collector.collect(Metric::boolean(
+            READ_METRIC_MERGE_SORT.to_string(),
+            need_merge_sort,
+        ));
+
+        if need_merge_sort {
             let merge_iters = self
                 .build_merge_iters(table_data, &request, iter_options, &table_options)
                 .await?;
@@ -123,16 +130,16 @@ impl Instance {
         };
 
         // Split iterators into `read_parallelism` groups.
-        let mut splited_iters: Vec<_> = std::iter::repeat_with(Vec::new)
+        let mut splitted_iters: Vec<_> = std::iter::repeat_with(Vec::new)
             .take(read_parallelism)
             .collect();
 
         for (i, time_aligned_iter) in partitioned_iters.into_iter().enumerate() {
-            splited_iters[i % read_parallelism].push(time_aligned_iter);
+            splitted_iters[i % read_parallelism].push(time_aligned_iter);
         }
 
         let mut streams = Vec::with_capacity(read_parallelism);
-        for iters in splited_iters {
+        for iters in splitted_iters {
             let stream = iters_to_stream(iters, self.read_runtime(), &request.projected_schema);
             streams.push(stream);
         }
@@ -172,6 +179,7 @@ impl Instance {
         for read_view in read_views {
             let merge_config = MergeConfig {
                 request_id: request.request_id,
+                metrics_collector: Some(request.metrics_collector.clone()),
                 deadline: request.opts.deadline,
                 space_id: table_data.space_id,
                 table_id: table_data.id,
@@ -201,6 +209,11 @@ impl Instance {
             iters.push(dedup_iter);
         }
 
+        request.metrics_collector.collect(Metric::counter(
+            READ_METRIC_ITER_NUM.to_string(),
+            iters.len(),
+        ));
+
         Ok(iters)
     }
 

analytic_engine/src/row_iter/merge.rs

@@ -22,7 +22,10 @@ use common_util::{define_result, error::GenericError};
 use futures::{future::try_join_all, StreamExt};
 use log::{debug, info, trace};
 use snafu::{ensure, Backtrace, ResultExt, Snafu};
-use table_engine::{predicate::PredicateRef, table::TableId};
+use table_engine::{
+    predicate::PredicateRef,
+    table::{Metric, ReadMetricsCollector, TableId},
+};
 
 use crate::{
     row_iter::{
@@ -83,6 +86,7 @@ define_result!(Error);
 #[derive(Debug)]
 pub struct MergeConfig<'a> {
     pub request_id: RequestId,
+    pub metrics_collector: Option<ReadMetricsCollector>,
     /// None for background jobs, such as: compaction
     pub deadline: Option<Instant>,
     pub space_id: SpaceId,
@@ -228,6 +232,7 @@ impl<'a> MergeBuilder<'a> {
             self.config.merge_iter_options,
             self.config.reverse,
             Metrics::new(self.memtables.len(), sst_streams_num, sst_ids),
+            self.config.metrics_collector,
         ))
     }
 }
@@ -556,14 +561,15 @@ impl Ord for HeapBufferedStream {
     }
 }
 
+/// Metrics for merge iterator.
 pub struct Metrics {
     num_memtables: usize,
     num_ssts: usize,
     sst_ids: Vec<FileId>,
-    /// Times to fetch rows from one stream.
-    times_fetch_rows_from_one: usize,
     /// Total rows collected using fetch_rows_from_one_stream().
     total_rows_fetch_from_one: usize,
+    /// Times to fetch rows from one stream.
+    times_fetch_rows_from_one: usize,
     /// Times to fetch one row from multiple stream.
     times_fetch_row_from_multiple: usize,
     /// Create time of the metrics.
@@ -591,6 +597,37 @@ impl Metrics {
             scan_count: 0,
         }
     }
+
+    fn collect(&self, collector: &ReadMetricsCollector) {
+        // TODO: maybe we can define a macro to generate the code.
+        collector.collect(Metric::counter(
+            "num_memtables".to_string(),
+            self.num_memtables,
+        ));
+
+        collector.collect(Metric::counter("num_ssts".to_string(), self.num_ssts));
+        collector.collect(Metric::counter(
+            "times_fetch_rows_from_one".to_string(),
+            self.times_fetch_rows_from_one,
+        ));
+        collector.collect(Metric::counter(
+            "times_rows_fetch_from_one".to_string(),
+            self.times_fetch_row_from_multiple,
+        ));
+        collector.collect(Metric::counter(
+            "total_rows_fetch_from_one".to_string(),
+            self.total_rows_fetch_from_one,
+        ));
+        collector.collect(Metric::elapsed(
+            "init_duration".to_string(),
+            self.init_duration,
+        ));
+        collector.collect(Metric::elapsed(
+            "scan_duration".to_string(),
+            self.scan_duration,
+        ));
+        collector.collect(Metric::counter("scan_count".to_string(), self.scan_count));
+    }
 }
 
 impl fmt::Debug for Metrics {
@@ -630,6 +667,7 @@ pub struct MergeIterator {
     iter_options: IterOptions,
     reverse: bool,
     metrics: Metrics,
+    metrics_collector: Option<ReadMetricsCollector>,
 }
 
 impl MergeIterator {
@@ -643,6 +681,7 @@ impl MergeIterator {
         iter_options: IterOptions,
         reverse: bool,
         metrics: Metrics,
+        metrics_collector: Option<ReadMetricsCollector>,
     ) -> Self {
         let heap_cap = streams.len();
         let record_batch_builder =
@@ -660,6 +699,7 @@ impl MergeIterator {
             iter_options,
             reverse,
             metrics,
+            metrics_collector,
         }
     }
 
@@ -855,6 +895,10 @@ impl MergeIterator {
 
 impl Drop for MergeIterator {
     fn drop(&mut self) {
+        if let Some(collector) = &self.metrics_collector {
+            self.metrics.collect(collector);
+        }
+
         info!(
             "Merge iterator dropped, table_id:{:?}, request_id:{}, metrics:{:?}, iter_options:{:?},",
             self.table_id, self.request_id, self.metrics, self.iter_options,
@@ -925,6 +969,7 @@ mod tests {
             IterOptions::default(),
             false,
             Metrics::new(1, 1, vec![]),
+            None,
         );
 
         check_iterator(
@@ -978,6 +1023,7 @@ mod tests {
             IterOptions::default(),
             true,
             Metrics::new(1, 1, vec![]),
+            None,
         );
 
         check_iterator(

benchmarks/src/merge_memtable_bench.rs

@@ -142,6 +142,7 @@ impl MergeMemTableBench {
         let store_picker: ObjectStorePickerRef = Arc::new(self.store.clone());
         let mut builder = MergeBuilder::new(MergeConfig {
             request_id,
+            metrics_collector: None,
             deadline: None,
             space_id,
             table_id,

benchmarks/src/merge_sst_bench.rs

@@ -125,6 +125,7 @@ impl MergeSstBench {
         let store_picker: ObjectStorePickerRef = Arc::new(self.store.clone());
         let mut builder = MergeBuilder::new(MergeConfig {
             request_id,
+            metrics_collector: None,
             deadline: None,
             space_id,
             table_id,

benchmarks/src/sst_tools.rs

@@ -220,6 +220,7 @@ pub async fn merge_sst(config: MergeSstConfig, runtime: Arc<Runtime>) {
 
         let mut builder = MergeBuilder::new(MergeConfig {
             request_id,
+            metrics_collector: None,
             deadline: None,
             space_id,
             table_id,

table_engine/src/table.rs

@@ -382,71 +382,63 @@ impl ReadOrder {
 pub struct MetricValue<T: Clone + fmt::Debug> {
     pub name: String,
     pub val: T,
-    pub partition: Option<usize>,
 }
 
 #[derive(Clone, Debug)]
 pub enum Metric {
+    Boolean(MetricValue<bool>),
     Counter(MetricValue<usize>),
     Elapsed(MetricValue<Duration>),
 }
 
 impl Metric {
     #[inline]
-    pub fn counter(name: String, val: usize, partition: Option<usize>) -> Self {
-        Metric::Counter(MetricValue {
-            name,
-            val,
-            partition,
-        })
+    pub fn counter(name: String, val: usize) -> Self {
+        Metric::Counter(MetricValue { name, val })
     }
 
     #[inline]
-    pub fn elapsed(name: String, val: Duration, partition: Option<usize>) -> Self {
-        Metric::Elapsed(MetricValue {
-            name,
-            val,
-            partition,
-        })
+    pub fn elapsed(name: String, val: Duration) -> Self {
+        Metric::Elapsed(MetricValue { name, val })
+    }
+
+    #[inline]
+    pub fn boolean(name: String, val: bool) -> Self {
+        Metric::Boolean(MetricValue { name, val })
     }
 }
 
 impl From<Metric> for DfMetric {
     fn from(metric: Metric) -> Self {
-        let (df_metric_val, partition) = match metric {
-            Metric::Counter(MetricValue {
-                name,
-                val,
-                partition,
-            }) => {
+        let df_metric_val = match metric {
+            Metric::Counter(MetricValue { name, val }) => {
                 let count = Count::new();
                 count.add(val);
-                (
-                    DfMetricValue::Count {
-                        name: name.into(),
-                        count,
-                    },
-                    partition,
-                )
+                DfMetricValue::Count {
+                    name: name.into(),
+                    count,
+                }
             }
-            Metric::Elapsed(MetricValue {
-                name,
-                val,
-                partition,
-            }) => {
+            Metric::Elapsed(MetricValue { name, val }) => {
                 let time = Time::new();
                 time.add_duration(val);
-                (
-                    DfMetricValue::Time {
-                        name: name.into(),
-                        time,
-                    },
-                    partition,
-                )
+                DfMetricValue::Time {
+                    name: name.into(),
+                    time,
+                }
+            }
+            Metric::Boolean(MetricValue { name, val }) => {
+                let count = Count::new();
+                // Use 0 for false, 1 for true.
+                count.add(val as usize);
+                DfMetricValue::Count {
+                    name: name.into(),
+                    count,
+                }
             }
         };
 
-        DfMetric::new(df_metric_val, partition)
+        DfMetric::new(df_metric_val, None)
     }
 }