From 3b5d3a2a284e79f5908e26a7d3c04a75f78c995a Mon Sep 17 00:00:00 2001 From: Dmitriy Rusov Date: Fri, 17 Jul 2026 15:51:31 +0200 Subject: [PATCH] dev --- packages/cubejs-api-gateway/src/gateway.ts | 10 ++++ .../cubejs-backend-native/src/node_export.rs | 1 + .../cubesql/src/compile/engine/df/scan.rs | 14 ++++- rust/cubesql/cubesql/src/sql/postgres/shim.rs | 4 +- rust/cubesql/cubesql/src/transport/service.rs | 53 +++++++++++++++++++ 5 files changed, 79 insertions(+), 3 deletions(-) diff --git a/packages/cubejs-api-gateway/src/gateway.ts b/packages/cubejs-api-gateway/src/gateway.ts index cf1526fdf2754..cee55593de93f 100644 --- a/packages/cubejs-api-gateway/src/gateway.ts +++ b/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, ); diff --git a/packages/cubejs-backend-native/src/node_export.rs b/packages/cubejs-backend-native/src/node_export.rs index c5bdc1cc687d4..d45c386ce3aac 100644 --- a/packages/cubejs-backend-native/src/node_export.rs +++ b/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?; diff --git a/rust/cubesql/cubesql/src/compile/engine/df/scan.rs b/rust/cubesql/cubesql/src/compile/engine/df/scan.rs index 6d0790b00d640..929ab9409929c 100644 --- a/rust/cubesql/cubesql/src/compile/engine/df/scan.rs +++ b/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( diff --git a/rust/cubesql/cubesql/src/sql/postgres/shim.rs b/rust/cubesql/cubesql/src/sql/postgres/shim.rs index d487ff93b55df..1af7ba9a86950 100644 --- a/rust/cubesql/cubesql/src/sql/postgres/shim.rs +++ b/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?; diff --git a/rust/cubesql/cubesql/src/transport/service.rs b/rust/cubesql/cubesql/src/transport/service.rs index 509290d6223f5..1de989e4b2237 100644 --- a/rust/cubesql/cubesql/src/transport/service.rs +++ b/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, + last_refresh_time: RWLockAsync>>, } 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) { + 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 { + 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()) + ); + } +}