chore(cubestore): Upgrade DF: Add planning metrics

Author: srh

rust/cubestore/cubestore/src/app_metrics.rs

@@ -15,6 +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_OPTIMIZE_TIME_MS: Histogram = metrics::histogram("cs.sql.query.data.planning.logical_plan_optimize.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_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");
+
 /// Incoming SQL queries that only read metadata or do trivial computations.
 pub static META_QUERIES: Counter = metrics::counter("cs.sql.query.meta");
 pub static META_QUERY_TIME_MS: Histogram = metrics::histogram("cs.sql.query.meta.ms");

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

@@ -168,7 +168,9 @@ impl QueryPlanner for QueryPlannerImpl {
             )
         );
 
+        let logical_plan_optimize_time = SystemTime::now();
         logical_plan = state.optimize(&logical_plan)?;
+        app_metrics::DATA_QUERY_LOGICAL_PLAN_OPTIMIZE_TIME_MS.report(logical_plan_optimize_time.elapsed()?.as_millis() as i64);
         trace!(
             "Logical Plan: {}",
             pp_plan_ext(
@@ -185,6 +187,7 @@ impl QueryPlanner for QueryPlannerImpl {
         );
 
         let plan = if SerializedPlan::is_data_select_query(&logical_plan) {
+            let choose_index_ext_start = SystemTime::now();
             let (logical_plan, meta) = choose_index_ext(
                 logical_plan,
                 &self.meta_store.as_ref(),
@@ -196,6 +199,7 @@ impl QueryPlanner for QueryPlannerImpl {
                 &logical_plan,
                 &meta.multi_part_subtree,
             )?;
+            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)?,
                 workers,

rust/cubestore/cubestore/src/queryplanner/query_executor.rs

@@ -171,12 +171,14 @@ impl QueryExecutor for QueryExecutorImpl {
     ) -> Result<(SchemaRef, Vec<RecordBatch>), CubeError> {
         let collect_span = tracing::span!(tracing::Level::TRACE, "collect_physical_plan");
         let trace_obj = plan.trace_obj();
+        let create_router_physical_plan_time = SystemTime::now();
         let (physical_plan, logical_plan) = self.router_plan(plan, cluster).await?;
+        app_metrics::DATA_QUERY_CREATE_ROUTER_PHYSICAL_PLAN_MS.report(create_router_physical_plan_time.elapsed()?.as_millis() as i64);
         let split_plan = physical_plan;
 
         trace!(
             "Router Query Physical Plan: {}",
-            pp_phys_plan(split_plan.as_ref())
+            pp_phys_plan_ext(split_plan.as_ref(), &PPOptions{ traverse_past_clustersend: true, ..PPOptions::show_most() })
         );
 
         let flags = PhysicalPlanFlags::with_execution_plan(split_plan.as_ref());
@@ -237,6 +239,7 @@ impl QueryExecutor for QueryExecutorImpl {
         chunk_id_to_record_batches: HashMap<u64, Vec<RecordBatch>>,
     ) -> Result<(SchemaRef, Vec<RecordBatch>, usize), CubeError> {
         let data_loaded_size = DataLoadedSize::new();
+        let create_worker_physical_plan_time = SystemTime::now();
         let (physical_plan, logical_plan) = self
             .worker_plan(
                 plan,
@@ -246,6 +249,8 @@ impl QueryExecutor for QueryExecutorImpl {
                 Some(data_loaded_size.clone()),
             )
             .await?;
+        app_metrics::DATA_QUERY_CREATE_WORKER_PHYSICAL_PLAN_MS.report(create_worker_physical_plan_time.elapsed()?.as_millis() as i64);
+
         let worker_plan;
         let max_batch_rows;
         if let Some((p, s)) = get_worker_plan(&physical_plan) {
@@ -260,7 +265,7 @@ impl QueryExecutor for QueryExecutorImpl {
 
         trace!(
             "Partition Query Physical Plan: {}",
-            pp_phys_plan(worker_plan.as_ref())
+            pp_phys_plan_ext(worker_plan.as_ref(), &PPOptions::show_most())
         );
 
         let execution_time = SystemTime::now();