chore(cubestore): Upgrade DF: Add data frame serialization time metrics

Author: srh

rust/cubestore/cubestore/src/app_metrics.rs

@@ -53,6 +53,11 @@ pub static DATA_QUERY_CREATE_ROUTER_PHYSICAL_PLAN_US: Histogram =
 pub static DATA_QUERY_CREATE_WORKER_PHYSICAL_PLAN_US: Histogram =
     metrics::histogram("cs.sql.query.data.planning.worker_plan.us");
 
+pub static SQL_DATA_FRAME_SERIALIZATION_TIME_US: Histogram =
+    metrics::histogram("cs.sql.data_frame_serialization.us");
+pub static HTTP_MESSAGE_DATA_FRAME_SERIALIZATION_TIME_US: Histogram =
+    metrics::histogram("cs.http.data_frame_serialization.us");
+
 /// 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/http/mod.rs

@@ -10,7 +10,7 @@ use crate::sql::{InlineTable, InlineTables, SqlQueryContext, SqlService};
 use crate::store::DataFrame;
 use crate::table::{Row, TableValue};
 use crate::util::WorkerLoop;
-use crate::CubeError;
+use crate::{app_metrics, CubeError};
 use async_std::fs::File;
 use cubeshared::codegen::{
     root_as_http_message, HttpColumnValue, HttpColumnValueArgs, HttpError, HttpErrorArgs,
@@ -600,6 +600,7 @@ pub enum HttpCommand {
 impl HttpMessage {
     pub fn bytes(&self) -> Vec<u8> {
         let mut builder = FlatBufferBuilder::with_capacity(1024);
+        let mut data_frame_serialization_start = None::<SystemTime>;
         let args = HttpMessageArgs {
             message_id: self.message_id,
             command_type: match self.command {
@@ -645,6 +646,7 @@ impl HttpMessage {
                     )
                 }
                 HttpCommand::ResultSet { data_frame } => {
+                    data_frame_serialization_start = Some(SystemTime::now());
                     let columns_vec =
                         HttpMessage::build_columns(&mut builder, data_frame.get_columns());
                     let rows = HttpMessage::build_rows(&mut builder, data_frame.clone());
@@ -668,7 +670,16 @@ impl HttpMessage {
         };
         let message = cubeshared::codegen::HttpMessage::create(&mut builder, &args);
         builder.finish(message, None);
-        builder.finished_data().to_vec() // TODO copy
+        let result = builder.finished_data().to_vec(); // TODO copy
+        if let Some(data_frame_serialization_start) = data_frame_serialization_start {
+            app_metrics::HTTP_MESSAGE_DATA_FRAME_SERIALIZATION_TIME_US.report(
+                data_frame_serialization_start
+                    .elapsed()
+                    .unwrap_or_else(|_| Duration::ZERO)
+                    .as_micros() as i64,
+            );
+        }
+        result
     }
 
     pub fn should_close_connection(&self) -> bool {

rust/cubestore/cubestore/src/mysql/mod.rs

@@ -2,7 +2,7 @@ use crate::config::processing_loop::ProcessingLoop;
 use crate::sql::{InlineTables, SqlQueryContext, SqlService};
 use crate::table::TableValue;
 use crate::util::time_span::warn_long;
-use crate::{metastore, CubeError};
+use crate::{app_metrics, metastore, CubeError};
 use async_trait::async_trait;
 use datafusion::cube_ext;
 use hex::ToHex;
@@ -78,6 +78,9 @@ impl<W: io::Write + Send> AsyncMysqlShim<W> for Backend {
         }
         let _s = warn_long("sending query results", Duration::from_millis(100));
         let data_frame = res.unwrap();
+
+        let data_frame_serialization_start_time = SystemTime::now();
+
         let columns = data_frame
             .get_columns()
             .iter()
@@ -133,7 +136,20 @@ impl<W: io::Write + Send> AsyncMysqlShim<W> for Backend {
             rw.end_row()?;
         }
         rw.finish()?;
-        if start.elapsed().unwrap().as_millis() > 200 && query.to_lowercase().starts_with("select")
+
+        let end_time = SystemTime::now();
+        app_metrics::SQL_DATA_FRAME_SERIALIZATION_TIME_US.report(
+            end_time
+                .duration_since(data_frame_serialization_start_time)
+                .unwrap_or_default()
+                .as_micros() as i64,
+        );
+        if end_time
+            .duration_since(start)
+            .unwrap_or_default()
+            .as_millis()
+            > 200
+            && query.to_lowercase().starts_with("select")
         {
             warn!(
                 "Slow Query SQL ({:?}):\n{}",