feat: add lastRefreshTime to Load Request Success telemetry events

packages/cubejs-api-gateway/src/gateway.ts

@@ -2090,6 +2090,13 @@ class ApiGateway {
         })
       );
 
+      // A request may consist of several queries; the oldest refresh time
+      // defines the freshness of the whole result
+      const lastRefreshTimestamps = results
+        .map((r: any) => r.getRootResultObject()[0].lastRefreshTime)
+        .filter(Boolean)
+        .map((t: string) => new Date(t).getTime());
+
       this.log(
         {
           type: 'Load Request Success',
@@ -2109,6 +2116,9 @@ class ApiGateway {
           // queriesWithData:
           //   results.filter((r: any) => r.data?.length).length,
           dbType: results.map(r => r.getRootResultObject()[0].dbType),
+          lastRefreshTime: lastRefreshTimestamps.length
+            ? new Date(Math.min(...lastRefreshTimestamps)).toISOString()
+            : undefined,
         },
         context,
       );

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

@@ -463,6 +463,7 @@ async fn handle_sql_query(
                             "apiType": "sql",
                             "duration": span_id.as_ref().unwrap().duration(),
                             "isDataQuery": span_id.as_ref().unwrap().is_data_query().await,
+                            "lastRefreshTime": span_id.as_ref().unwrap().last_refresh_time().await,
                         }),
                     )
                     .await?;

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

@@ -10,7 +10,7 @@ use crate::{
     CubeError, CubeErrorCauseType,
 };
 use async_trait::async_trait;
-use chrono::{Datelike, NaiveDate};
+use chrono::{DateTime, Datelike, NaiveDate, Utc};
 use cubeclient::models::{
     V1LoadRequestQuery, V1LoadResponse, V1LoadResult, V1LoadResultDataColumnar,
 };
@@ -793,7 +793,7 @@ async fn load_data(
     } else {
         let result = transport
             .load(
-                span_id,
+                span_id.clone(),
                 request,
                 sql_query,
                 auth_context,
@@ -822,6 +822,16 @@ async fn load_data(
             })?;
 
         if let Some(data) = result.into_iter().next() {
+            if let Some(span_id) = &span_id {
+                if let Some(last_refresh_time) = data.schema().metadata().get("lastRefreshTime") {
+                    if let Ok(last_refresh_time) = DateTime::parse_from_rfc3339(last_refresh_time) {
+                        span_id
+                            .set_last_refresh_time(last_refresh_time.with_timezone(&Utc))
+                            .await;
+                    }
+                }
+            }
+
             match (options.max_records, data.num_rows()) {
                 (Some(max_records), len) if len >= max_records => {
                     return Err(ArrowError::ExternalError(Box::new(CubeError::user(

rust/cubesql/cubesql/src/sql/postgres/shim.rs

@@ -382,7 +382,8 @@ impl AsyncPostgresShim {
                                             "query": span_id.query_key.clone(),
                                             "apiType": "sql",
                                             "duration": span_id.duration(),
-                                            "isDataQuery": span_id.is_data_query().await
+                                            "isDataQuery": span_id.is_data_query().await,
+                                            "lastRefreshTime": span_id.last_refresh_time().await
                                         }),
                                     )
                                     .await?;
@@ -1869,6 +1870,7 @@ impl AsyncPostgresShim {
                                 "apiType": "sql",
                                 "duration": start_time.elapsed().unwrap().as_millis() as u64,
                                 "isDataQuery": span_id.is_data_query().await,
+                                "lastRefreshTime": span_id.last_refresh_time().await,
                             }),
                         )
                         .await?;

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

@@ -1,4 +1,5 @@
 use async_trait::async_trait;
+use chrono::{DateTime, SecondsFormat, Utc};
 use cubeclient::apis::{
     configuration::Configuration as ClientConfiguration, default_api as cube_api,
 };
@@ -84,6 +85,7 @@ pub struct SpanId {
     pub query_key: serde_json::Value,
     span_start: SystemTime,
     is_data_query: RWLockAsync<bool>,
+    last_refresh_time: RWLockAsync<Option<DateTime<Utc>>>,
 }
 
 impl SpanId {
@@ -93,6 +95,7 @@ impl SpanId {
             query_key,
             span_start: SystemTime::now(),
             is_data_query: tokio::sync::RwLock::new(false),
+            last_refresh_time: tokio::sync::RwLock::new(None),
         }
     }
 
@@ -106,6 +109,22 @@ impl SpanId {
         *read
     }
 
+    /// Records the refresh time of a load result contributing to this span.
+    /// A span may cover several load requests (e.g. a join of cube scans);
+    /// the oldest value wins, matching `lastRefreshTime` semantics elsewhere.
+    pub async fn set_last_refresh_time(&self, last_refresh_time: DateTime<Utc>) {
+        let mut write = self.last_refresh_time.write().await;
+        *write = Some(match *write {
+            Some(current) => current.min(last_refresh_time),
+            None => last_refresh_time,
+        });
+    }
+
+    pub async fn last_refresh_time(&self) -> Option<String> {
+        let read = self.last_refresh_time.read().await;
+        read.map(|t| t.to_rfc3339_opts(SecondsFormat::Millis, true))
+    }
+
     pub fn duration(&self) -> u64 {
         self.span_start
             .elapsed()
@@ -1041,3 +1060,37 @@ impl SqlTemplates {
         )
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::TimeZone;
+
+    #[tokio::test]
+    async fn span_id_last_refresh_time_keeps_oldest() {
+        let span_id = SpanId::new("test".to_string(), serde_json::json!({}));
+        assert_eq!(span_id.last_refresh_time().await, None);
+
+        let newer = Utc.with_ymd_and_hms(2024, 6, 1, 12, 0, 0).unwrap();
+        let older = Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap();
+
+        span_id.set_last_refresh_time(newer).await;
+        assert_eq!(
+            span_id.last_refresh_time().await,
+            Some("2024-06-01T12:00:00.000Z".to_string())
+        );
+
+        span_id.set_last_refresh_time(older).await;
+        assert_eq!(
+            span_id.last_refresh_time().await,
+            Some("2024-01-01T00:00:00.000Z".to_string())
+        );
+
+        // A newer value must not override the recorded oldest one
+        span_id.set_last_refresh_time(newer).await;
+        assert_eq!(
+            span_id.last_refresh_time().await,
+            Some("2024-01-01T00:00:00.000Z".to_string())
+        );
+    }
+}