feat(backend-native): expose lastRefreshTime via /cubesql API (#10206)

* add lastRefreshTime to load result in openapi spec

* feat(backend-native): expose lastRefreshTime via /cubesql API

* some fixes and improvements

Author: KSDaemon

packages/cubejs-api-gateway/openspec.yml

@@ -371,6 +371,8 @@ components:
           type: "array"
           items:
             type: "object"
+        lastRefreshTime:
+          type: "string"
     V1Error:
       type: "object"
       required:

packages/cubejs-backend-native/src/node_export.rs

@@ -327,6 +327,13 @@ async fn handle_sql_query(
             let mut schema_response = Map::new();
             schema_response.insert("schema".into(), columns_json);
 
+            if let Some(last_refresh_time) = stream_schema.metadata().get("lastRefreshTime") {
+                schema_response.insert(
+                    "lastRefreshTime".into(),
+                    serde_json::Value::String(last_refresh_time.clone()),
+                );
+            }
+
             write_jsonl_message(
                 channel.clone(),
                 stream_methods.write.clone(),

packages/cubejs-backend-native/src/orchestrator.rs

@@ -37,6 +37,7 @@ pub struct ResultWrapper {
     transform_data: TransformDataRequest,
     data: Arc<QueryResult>,
     transformed_data: Option<TransformedData>,
+    pub last_refresh_time: Option<String>,
 }
 
 impl ResultWrapper {
@@ -114,6 +115,7 @@ impl ResultWrapper {
             transform_data: transform_request,
             data: query_result,
             transformed_data: None,
+            last_refresh_time: None,
         })
     }
 

packages/cubejs-backend-native/src/transport.rs

@@ -5,13 +5,16 @@ use std::fmt::Display;
 
 use crate::auth::NativeSQLAuthContext;
 use crate::channel::{call_raw_js_with_channel_as_callback, NodeSqlGenerator, ValueFromJs};
+use crate::node_obj_deserializer::JsValueDeserializer;
 use crate::node_obj_serializer::NodeObjSerializer;
 use crate::orchestrator::ResultWrapper;
 use crate::{
     auth::TransportRequest, channel::call_js_with_channel_as_callback,
     stream::call_js_with_stream_as_callback,
 };
 use async_trait::async_trait;
+use cubeorchestrator::query_result_transform::RequestResultData;
+use cubesql::compile::arrow::datatypes::Schema;
 use cubesql::compile::engine::df::scan::{
     convert_transport_response, transform_response, CacheMode, MemberField, RecordBatch, SchemaRef,
 };
@@ -26,7 +29,7 @@ use cubesql::{
     transport::{CubeStreamReceiver, LoadRequestMeta, MetaContext, TransportService},
     CubeError,
 };
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 use uuid::Uuid;
 
@@ -402,14 +405,37 @@ impl TransportService for NodeBridgeTransport {
                             .to_vec(cx)
                             .map_cube_err("Can't convert JS result to array")?;
 
-                        let native_wrapped_results = js_res_wrapped_vec
-                            .iter()
-                            .map(|r| ResultWrapper::from_js_result_wrapper(cx, *r))
-                            .collect::<Result<Vec<_>, _>>()
+                        let get_root_result_object_method: Handle<JsFunction> =
+                            js_result_wrapped.get(cx, "getRootResultObject").map_cube_err(
+                                "Can't get getRootResultObject method from JS ResultWrapper object",
+                            )?;
+
+                        let result_data_js_array = get_root_result_object_method
+                            .call(cx, js_result_wrapped.upcast::<JsValue>(), [])
                             .map_cube_err(
-                                "Can't construct result wrapper from JS ResultWrapper object",
+                                "Error calling getRootResultObject() method of JS ResultWrapper object",
                             )?;
 
+                        let result_data_js_vec = result_data_js_array
+                            .downcast::<JsArray, _>(cx)
+                            .map_cube_err("Can't downcast getRootResultObject result to array")?
+                            .to_vec(cx)
+                            .map_cube_err("Can't convert getRootResultObject result to array")?;
+
+                        let mut native_wrapped_results = Vec::with_capacity(js_res_wrapped_vec.len());
+                        for (js_wrapper, js_result_data) in js_res_wrapped_vec.iter().zip(result_data_js_vec.iter()) {
+                            let mut wrapper = ResultWrapper::from_js_result_wrapper(cx, *js_wrapper)
+                                .map_cube_err("Can't construct result wrapper from JS ResultWrapper object")?;
+
+                            let deserializer = JsValueDeserializer::new(cx, *js_result_data);
+                            let result_data: RequestResultData = Deserialize::deserialize(deserializer)
+                                .map_cube_err("Can't deserialize RequestResultData from getRootResultObject")?;
+
+                            wrapper.last_refresh_time = result_data.last_refresh_time;
+
+                            native_wrapped_results.push(wrapper);
+                        }
+
                         Ok(ValueFromJs::ResultWrapper(native_wrapped_results))
                     } else if let Ok(str) = v.downcast::<JsString, _>(cx) {
                         Ok(ValueFromJs::String(str.value(cx)))
@@ -482,7 +508,20 @@ impl TransportService for NodeBridgeTransport {
                     break result_wrappers
                         .into_iter()
                         .map(|mut wrapper| {
-                            transform_response(&mut wrapper, schema.clone(), &member_fields)
+                            let updated_schema = if let Some(last_refresh_time) =
+                                wrapper.last_refresh_time.clone()
+                            {
+                                let mut metadata = schema.metadata().clone();
+                                metadata.insert("lastRefreshTime".to_string(), last_refresh_time);
+                                Arc::new(Schema::new_with_metadata(
+                                    schema.fields().to_vec(),
+                                    metadata,
+                                ))
+                            } else {
+                                schema.clone()
+                            };
+
+                            transform_response(&mut wrapper, updated_schema, &member_fields)
                         })
                         .collect::<Result<Vec<_>, _>>();
                 }

rust/cubesql/cubeclient/src/models/v1_load_result.rs

@@ -21,6 +21,8 @@ pub struct V1LoadResult {
     pub data: Vec<serde_json::Value>,
     #[serde(rename = "refreshKeyValues", skip_serializing_if = "Option::is_none")]
     pub refresh_key_values: Option<Vec<serde_json::Value>>,
+    #[serde(rename = "lastRefreshTime", skip_serializing_if = "Option::is_none")]
+    pub last_refresh_time: Option<String>,
 }
 
 impl V1LoadResult {
@@ -33,6 +35,7 @@ impl V1LoadResult {
             annotation: Box::new(annotation),
             data,
             refresh_key_values: None,
+            last_refresh_time: None,
         }
     }
 }

rust/cubesql/cubesql/src/compile/engine/df/scan.rs

@@ -18,7 +18,7 @@ pub use datafusion::{
             ArrayRef, BooleanBuilder, Date32Builder, DecimalBuilder, Float32Builder,
             Float64Builder, Int16Builder, Int32Builder, Int64Builder, NullArray, StringBuilder,
         },
-        datatypes::{DataType, SchemaRef},
+        datatypes::{DataType, Schema, SchemaRef},
         error::{ArrowError, Result as ArrowResult},
         record_batch::RecordBatch,
     },
@@ -508,12 +508,13 @@ impl ExecutionPlan for CubeScanExecutionPlan {
 
         // For now execute method executes only one query at a time, so we
         // take the first result
+        let rb_schema = response.first().unwrap().schema().clone();
         one_shot_stream.data = Some(response.first().unwrap().clone());
 
         Ok(Box::pin(CubeScanStreamRouter::new(
             None,
             one_shot_stream,
-            self.schema.clone(),
+            rb_schema,
         )))
     }
 
@@ -1177,7 +1178,18 @@ pub fn convert_transport_response(
         .into_iter()
         .map(|r| {
             let mut response = JsonValueObject::new(r.data.clone());
-            transform_response(&mut response, schema.clone(), &member_fields)
+            let updated_schema = if let Some(last_refresh_time) = r.last_refresh_time.clone() {
+                let mut metadata = schema.metadata().clone();
+                metadata.insert("lastRefreshTime".to_string(), last_refresh_time);
+                Arc::new(Schema::new_with_metadata(
+                    schema.fields().to_vec(),
+                    metadata,
+                ))
+            } else {
+                schema.clone()
+            };
+
+            transform_response(&mut response, updated_schema, &member_fields)
         })
         .collect::<std::result::Result<Vec<RecordBatch>, CubeError>>()
 }

rust/cubesql/cubesql/src/transport/service.rs

@@ -193,11 +193,11 @@ struct MetaCacheBucket {
     value: Arc<MetaContext>,
 }
 
-/// This transports is used in standalone mode
+/// This transport is used in standalone mode
 #[derive(Debug)]
 pub struct HttpTransport {
     /// We use simple cache to improve DX with standalone mode
-    /// because currently we dont persist DF in the SessionState
+    /// because currently we don't persist DF in the SessionState,
     /// and it causes a lot of HTTP requests which slow down BI connections
     cache: RwLockAsync<Option<MetaCacheBucket>>,
 }