chore(cubestore): Upgrade DF: finer grained app_metrics in planning

Author: srh

rust/cubestore/cubestore/src/app_metrics.rs

@@ -15,8 +15,11 @@ pub static DATA_QUERIES_CACHE_SIZE: Gauge = metrics::gauge("cs.sql.query.data.ca
 // Approximate total weighted size of entries in this cache.
 pub static DATA_QUERIES_CACHE_WEIGHT: Gauge = metrics::gauge("cs.sql.query.data.cache.weight");
 pub static DATA_QUERY_TIME_MS: Histogram = metrics::histogram("cs.sql.query.data.ms");
+pub static DATA_QUERY_LOGICAL_PLAN_TOTAL_CREATION_TIME_MS: Histogram = metrics::histogram("cs.sql.query.data.planning.logical_plan_total_creation.ms");
 pub static DATA_QUERY_LOGICAL_PLAN_OPTIMIZE_TIME_MS: Histogram = metrics::histogram("cs.sql.query.data.planning.logical_plan_optimize.ms");
+pub static DATA_QUERY_CHOOSE_INDEX_TIME_MS: Histogram = metrics::histogram("cs.sql.query.data.planning.choose_index.ms");
 pub static DATA_QUERY_CHOOSE_INDEX_AND_WORKERS_TIME_MS: Histogram = metrics::histogram("cs.sql.query.data.planning.choose_index_and_workers.ms");
+pub static DATA_QUERY_TO_SERIALIZED_PLAN_TIME_MS: Histogram = metrics::histogram("cs.sql.query.data.planning.to_serialized_plan.ms");
 pub static DATA_QUERY_CREATE_ROUTER_PHYSICAL_PLAN_MS: Histogram = metrics::histogram("cs.sql.query.data.planning.router_plan.ms");
 pub static DATA_QUERY_CREATE_WORKER_PHYSICAL_PLAN_MS: Histogram = metrics::histogram("cs.sql.query.data.planning.worker_plan.ms");
 

rust/cubestore/cubestore/src/queryplanner/mod.rs

@@ -194,11 +194,13 @@ impl QueryPlanner for QueryPlannerImpl {
                 self.config.enable_topk(),
             )
             .await?;
+            let choose_index_ext_end = SystemTime::now();
             let workers = compute_workers(
                 self.config.as_ref(),
                 &logical_plan,
                 &meta.multi_part_subtree,
             )?;
+            app_metrics::DATA_QUERY_CHOOSE_INDEX_TIME_MS.report(choose_index_ext_end.duration_since(choose_index_ext_start)?.as_millis() as i64);
             app_metrics::DATA_QUERY_CHOOSE_INDEX_AND_WORKERS_TIME_MS.report(choose_index_ext_start.elapsed()?.as_millis() as i64);
             QueryPlan::Select(
                 PreSerializedPlan::try_new(logical_plan, meta, trace_obj)?,

rust/cubestore/cubestore/src/sql/mod.rs

@@ -2,7 +2,7 @@ use std::collections::HashMap;
 use std::convert::TryFrom;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, SystemTime};
 
 use async_trait::async_trait;
 use chrono::format::Fixed::Nanosecond3;
@@ -1114,6 +1114,7 @@ impl SqlService for SqlServiceImpl {
                     .await
             }
             CubeStoreStatement::Statement(Statement::Query(q)) => {
+                let logical_plan_time_start = SystemTime::now();
                 let logical_plan = self
                     .query_planner
                     .logical_plan(
@@ -1123,6 +1124,8 @@ impl SqlService for SqlServiceImpl {
                     )
                     .await?;
 
+                app_metrics::DATA_QUERY_LOGICAL_PLAN_TOTAL_CREATION_TIME_MS.report(logical_plan_time_start.elapsed()?.as_millis() as i64);
+
                 // TODO distribute and combine
                 let res = match logical_plan {
                     QueryPlan::Meta(logical_plan) => {
@@ -1137,13 +1140,16 @@ impl SqlService for SqlServiceImpl {
 
                         let cluster = self.cluster.clone();
                         let executor = self.query_executor.clone();
+                        let serialized_plan_time_start = SystemTime::now();
+                        let serialized_plan = serialized.to_serialized_plan()?;
+                        app_metrics::DATA_QUERY_TO_SERIALIZED_PLAN_TIME_MS.report(serialized_plan_time_start.elapsed()?.as_millis() as i64);
                         timeout(
                             self.query_timeout,
                             self.cache
                                 .get(
                                     query,
                                     context,
-                                    serialized.to_serialized_plan()?,
+                                    serialized_plan,
                                     async move |plan| {
                                         let records;
                                         if workers.len() == 0 {