diff --git a/Cargo.lock b/Cargo.lock index 4a864d96cd..fcf281c868 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9582,4 +9582,4 @@ checksum = "9e16efa8a874a0481a574084d34cc26fdb3b99627480f785888deb6386506656" dependencies = [ "cc", "pkg-config", -] +] \ No newline at end of file diff --git a/src/common/meta/search.rs b/src/common/meta/search.rs index 126a2ba14b..c9e5ba07b8 100644 --- a/src/common/meta/search.rs +++ b/src/common/meta/search.rs @@ -21,12 +21,13 @@ use utoipa::ToSchema; pub struct CachedQueryResponse { pub cached_response: Response, pub deltas: Vec, - pub has_pre_cache_delta: bool, pub has_cached_data: bool, pub cache_query_response: bool, pub response_start_time: i64, pub response_end_time: i64, pub ts_column: String, + pub is_descending: bool, + pub limit: i64, } #[derive(Clone, Debug, Serialize, Deserialize, ToSchema, Default)] pub struct QueryDelta { @@ -34,3 +35,13 @@ pub struct QueryDelta { pub delta_end_time: i64, pub delta_removed_hits: bool, } + +#[derive(Clone, Debug, Serialize, Deserialize, ToSchema, Default)] +pub struct CacheQueryRequest { + pub q_start_time: i64, + pub q_end_time: i64, + pub is_aggregate: bool, + pub ts_column: String, + pub discard_interval: i64, + pub is_descending: bool, +} diff --git a/src/common/utils/http.rs b/src/common/utils/http.rs index a6fb373c06..574a480ebf 100644 --- a/src/common/utils/http.rs +++ b/src/common/utils/http.rs @@ -96,7 +96,7 @@ pub(crate) fn get_folder(query: &Query>) -> String { } #[inline(always)] -pub(crate) fn get_or_create_trace_id(headers: &HeaderMap, span: &Option) -> String { +pub(crate) fn get_or_create_trace_id(headers: &HeaderMap, span: &tracing::Span) -> String { let cfg = config::get_config(); if let Some(traceparent) = headers.get("traceparent") { if cfg.common.tracing_enabled || cfg.common.tracing_search_enabled { @@ -105,7 +105,7 @@ pub(crate) fn get_or_create_trace_id(headers: &HeaderMap, span: &Option Result, Status> { let req: QueryCacheRequest = request.into_inner(); match cacher::get_cached_results( - req.start_time, - req.end_time, - req.is_aggregate, &req.file_path, - &req.timestamp_col, &req.trace_id, - req.discard_interval, + CacheQueryRequest { + q_start_time: req.start_time, + q_end_time: req.end_time, + is_aggregate: req.is_aggregate, + ts_column: req.timestamp_col, + discard_interval: req.discard_interval, + is_descending: req.is_descending, + }, ) .await { Some(res) => { - let deltas = res - .deltas - .iter() - .map(|d| QueryDelta { - delta_start_time: d.delta_start_time, - delta_end_time: d.delta_end_time, - delta_removed_hits: d.delta_removed_hits, - }) - .collect(); - let res = QueryCacheRes { cached_response: Some(QueryResponse { data: serde_json::to_vec(&res.cached_response).unwrap(), }), - deltas, - has_pre_cache_delta: res.has_pre_cache_delta, has_cached_data: res.has_cached_data, cache_query_response: res.cache_query_response, cache_start_time: res.response_start_time, cache_end_time: res.response_end_time, + is_descending: res.is_descending, }; Ok(Response::new(QueryCacheResponse { diff --git a/src/handler/http/request/search/mod.rs b/src/handler/http/request/search/mod.rs index 3d92c38c9e..b5e07e8137 100644 --- a/src/handler/http/request/search/mod.rs +++ b/src/handler/http/request/search/mod.rs @@ -13,7 +13,7 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use std::{collections::HashMap, io::Error}; +use std::{cmp, collections::HashMap, io::Error}; use actix_web::{get, http::StatusCode, post, web, HttpRequest, HttpResponse}; use chrono::{Duration, Utc}; @@ -143,12 +143,9 @@ pub async fn search( let mut range_error = String::new(); let cfg = get_config(); let http_span = if cfg.common.tracing_search_enabled { - Some(tracing::info_span!( - "/api/{org_id}/_search", - org_id = org_id.clone() - )) + tracing::info_span!("/api/{org_id}/_search", org_id = org_id.clone()) } else { - None + Span::none() }; let trace_id = get_or_create_trace_id(in_req.headers(), &http_span); @@ -285,6 +282,11 @@ pub async fn search( "[trace_id {trace_id}] query deltas are: {:?}", c_resp.deltas ); + log::info!( + "[trace_id {trace_id}] Query original start time: {}, end time : {}", + req.query.start_time, + req.query.end_time + ); } // Result caching check ends let mut results = Vec::new(); @@ -308,7 +310,6 @@ pub async fn search( .with_label_values(&[&org_id]) .inc(); - let cfg = get_config(); // get a local search queue lock #[cfg(not(feature = "enterprise"))] let locker = SearchService::QUEUE_LOCKER.clone(); @@ -329,17 +330,8 @@ pub async fn search( .with_label_values(&[&org_id]) .dec(); - let search_tracing = !cfg.common.tracing_enabled && cfg.common.tracing_search_enabled; let mut tasks = Vec::new(); - if cfg.common.result_cache_enabled && cfg.common.print_key_sql { - log::info!( - "[trace_id {trace_id}] Query original start time: {}, end time : {}", - req.query.start_time, - req.query.end_time - ); - } - for (i, delta) in c_resp.deltas.into_iter().enumerate() { let http_span_local = http_span.clone(); let mut req = req.clone(); @@ -353,7 +345,10 @@ pub async fn search( req.query.end_time = delta.delta_end_time; let cfg = get_config(); - if cfg.common.result_cache_enabled && cfg.common.print_key_sql { + if cfg.common.result_cache_enabled + && cfg.common.print_key_sql + && c_resp.has_cached_data + { log::info!( "[trace_id {trace_id}] Query new start time: {}, end time : {}", req.query.start_time, @@ -361,19 +356,17 @@ pub async fn search( ); } - let search_fut = SearchService::search( + SearchService::search( &trace_id, &org_id, stream_type, Some(user_id.to_string()), &req, - ); - if search_tracing { - search_fut.instrument(http_span_local.unwrap()).await - } else { - search_fut.await - } - }); + ) + .instrument(http_span_local) + .await + }) + .instrument(http_span.clone()); tasks.push(task); } @@ -423,7 +416,12 @@ pub async fn search( } } if c_resp.has_cached_data { - merge_response(&mut c_resp.cached_response, &results, &c_resp.ts_column); + merge_response( + &mut c_resp.cached_response, + &results, + &c_resp.ts_column, + c_resp.limit, + ); c_resp.cached_response } else { results[0].clone() @@ -494,6 +492,7 @@ pub async fn search( file_path, is_aggregate, trace_id, + c_resp.is_descending, ) .await; } @@ -556,13 +555,13 @@ pub async fn around( let (org_id, stream_name) = path.into_inner(); let cfg = get_config(); let http_span = if cfg.common.tracing_search_enabled { - Some(tracing::info_span!( + tracing::info_span!( "/api/{org_id}/{stream_name}/_around", org_id = org_id.clone(), stream_name = stream_name.clone() - )) + ) } else { - None + Span::none() }; let trace_id = get_or_create_trace_id(in_req.headers(), &http_span); @@ -692,12 +691,10 @@ pub async fn around( .to_str() .ok() .map(|v| v.to_string()); - let search_fut = SearchService::search(&trace_id, &org_id, stream_type, user_id.clone(), &req); - let search_res = if !cfg.common.tracing_enabled && cfg.common.tracing_search_enabled { - search_fut.instrument(http_span.clone().unwrap()).await - } else { - search_fut.await - }; + let search_res = SearchService::search(&trace_id, &org_id, stream_type, user_id.clone(), &req) + .instrument(http_span.clone()) + .await; + let resp_forward = match search_res { Ok(res) => res, Err(err) => { @@ -747,12 +744,10 @@ pub async fn around( timeout, search_type: Some(SearchEventType::UI), }; - let search_fut = SearchService::search(&trace_id, &org_id, stream_type, user_id, &req); - let search_res = if !cfg.common.tracing_enabled && cfg.common.tracing_search_enabled { - search_fut.instrument(http_span.unwrap()).await - } else { - search_fut.await - }; + let search_res = SearchService::search(&trace_id, &org_id, stream_type, user_id, &req) + .instrument(http_span) + .await; + let resp_backward = match search_res { Ok(res) => res, Err(err) => { @@ -899,13 +894,13 @@ pub async fn values( None => "", }; let http_span = if config::get_config().common.tracing_search_enabled { - Some(tracing::info_span!( + tracing::info_span!( "/api/{org_id}/{stream_name}/_values", org_id = org_id.clone(), stream_name = stream_name.clone() - )) + ) } else { - None + Span::none() }; let trace_id = get_or_create_trace_id(in_req.headers(), &http_span); @@ -983,7 +978,7 @@ async fn values_v1( query: &web::Query>, user_id: &str, trace_id: String, - http_span: Option, + http_span: Span, ) -> Result { let start = std::time::Instant::now(); let started_at = Utc::now().timestamp_micros(); @@ -1155,18 +1150,16 @@ async fn values_v1( ), ); } - let search_fut = SearchService::search( + let search_res = SearchService::search( &trace_id, org_id, stream_type, Some(user_id.to_string()), &req, - ); - let search_res = if !cfg.common.tracing_enabled && cfg.common.tracing_search_enabled { - search_fut.instrument(http_span.unwrap()).await - } else { - search_fut.await - }; + ) + .instrument(http_span) + .await; + let resp_search = match search_res { Ok(res) => res, Err(err) => { @@ -1255,7 +1248,7 @@ async fn values_v2( query: &web::Query>, user_id: &str, trace_id: String, - http_span: Option, + http_span: Span, ) -> Result { let start = std::time::Instant::now(); let started_at = Utc::now().timestamp_micros(); @@ -1316,14 +1309,14 @@ async fn values_v2( metrics::QUERY_PENDING_NUMS .with_label_values(&[org_id]) .inc(); - let cfg = get_config(); + // get a local search queue lock #[cfg(not(feature = "enterprise"))] let locker = SearchService::QUEUE_LOCKER.clone(); #[cfg(not(feature = "enterprise"))] let locker = locker.lock().await; #[cfg(not(feature = "enterprise"))] - if !cfg.common.feature_query_queue_enabled { + if !get_config().common.feature_query_queue_enabled { drop(locker); } #[cfg(not(feature = "enterprise"))] @@ -1363,18 +1356,16 @@ async fn values_v2( timeout, search_type: Some(SearchEventType::Values), }; - let search_fut = SearchService::search( + let search_res = SearchService::search( &trace_id, org_id, StreamType::Metadata, Some(user_id.to_string()), &req, - ); - let search_res = if !cfg.common.tracing_enabled && cfg.common.tracing_search_enabled { - search_fut.instrument(http_span.unwrap()).await - } else { - search_fut.await - }; + ) + .instrument(http_span) + .await; + let resp_search = match search_res { Ok(res) => res, Err(err) => { @@ -1491,12 +1482,9 @@ pub async fn search_partition( let start = std::time::Instant::now(); let cfg = get_config(); let http_span = if cfg.common.tracing_search_enabled { - Some(tracing::info_span!( - "/api/{org_id}/_search_partition", - org_id = org_id.clone(), - )) + tracing::info_span!("/api/{org_id}/_search_partition", org_id = org_id.clone(),) } else { - None + Span::none() }; let trace_id = get_or_create_trace_id(in_req.headers(), &http_span); @@ -1515,12 +1503,9 @@ pub async fn search_partition( return Ok(MetaHttpResponse::bad_request(e)); } - let search_fut = SearchService::search_partition(&trace_id, &org_id, stream_type, &req); - let search_res = if !cfg.common.tracing_enabled && cfg.common.tracing_search_enabled { - search_fut.instrument(http_span.unwrap()).await - } else { - search_fut.await - }; + let search_res = SearchService::search_partition(&trace_id, &org_id, stream_type, &req) + .instrument(http_span) + .await; // do search match search_res { @@ -1550,8 +1535,14 @@ fn merge_response( cache_response: &mut config::meta::search::Response, search_response: &Vec, ts_column: &str, + limit: i64, ) { + if cache_response.hits.is_empty() && search_response.is_empty() { + return; + } + if cache_response.hits.is_empty() + && search_response.is_empty() && search_response.first().unwrap().hits.is_empty() && search_response.last().unwrap().hits.is_empty() { @@ -1562,6 +1553,7 @@ fn merge_response( } return; } + let cache_hits_len = cache_response.hits.len(); let mut files_cache_ratio = 0; let mut result_cache_ratio = 0; @@ -1597,12 +1589,15 @@ fn merge_response( .collect(); } } + if cache_response.hits.len() > limit as usize { + cache_response.hits.truncate(limit as usize); + } cache_response.cached_ratio = files_cache_ratio / search_response.len(); - cache_response.result_cache_ratio = (cache_response.total as f64 * 100_f64 - / (result_cache_ratio + cache_response.total) as f64) - as usize; + cache_response.result_cache_ratio = + (cache_hits_len as f64 * 100_f64 / (result_cache_ratio + cache_hits_len) as f64) as usize; } +#[allow(clippy::too_many_arguments)] async fn write_results( ts_column: &str, req_query_start_time: i64, @@ -1611,18 +1606,30 @@ async fn write_results( file_path: String, is_aggregate: bool, trace_id: String, + is_descending: bool, ) { let last_rec_ts = get_ts_value(ts_column, res.hits.last().unwrap()); - let cache_end_time = if last_rec_ts > 0 && last_rec_ts < req_query_end_time { + let first_rec_ts = get_ts_value(ts_column, res.hits.first().unwrap()); + + if (last_rec_ts - first_rec_ts).abs() + < get_config().common.result_cache_discard_duration * 1000 * 1000 + { + return; + } + + let largest_ts = cmp::max(first_rec_ts, last_rec_ts); + + let cache_end_time = if largest_ts > 0 && largest_ts < req_query_end_time { last_rec_ts } else { req_query_end_time }; let file_name = format!( - "{}_{}_{}.json", + "{}_{}_{}_{}.json", req_query_start_time, cache_end_time, - if is_aggregate { 1 } else { 0 } + if is_aggregate { 1 } else { 0 }, + if is_descending { 1 } else { 0 } ); let res_cache = json::to_string(&res).unwrap(); @@ -1646,6 +1653,7 @@ async fn write_results( start_time: req_query_start_time, end_time: cache_end_time, is_aggregate, + is_descending, }); drop(w); } diff --git a/src/handler/http/request/search/multi_streams.rs b/src/handler/http/request/search/multi_streams.rs index 73fbf00bbc..78a399f8dd 100644 --- a/src/handler/http/request/search/multi_streams.rs +++ b/src/handler/http/request/search/multi_streams.rs @@ -28,7 +28,7 @@ use config::{ utils::{base64, json}, }; use infra::errors; -use tracing::Instrument; +use tracing::{Instrument, Span}; use crate::{ common::{ @@ -119,12 +119,9 @@ pub async fn search_multi( let cfg = get_config(); let started_at = Utc::now().timestamp_micros(); let http_span = if cfg.common.tracing_search_enabled { - Some(tracing::info_span!( - "/api/{org_id}/_search_multi", - org_id = org_id.clone() - )) + tracing::info_span!("/api/{org_id}/_search_multi", org_id = org_id.clone()) } else { - None + Span::none() }; let trace_id = get_or_create_trace_id(in_req.headers(), &http_span); @@ -253,19 +250,15 @@ pub async fn search_multi( let trace_id = trace_id.clone(); // do search - let search_fut = SearchService::search( + let search_res = SearchService::search( &trace_id, &org_id, stream_type, Some(user_id.to_string()), &req, - ); - - let search_res = if !cfg.common.tracing_enabled && cfg.common.tracing_search_enabled { - search_fut.instrument(http_span.clone().unwrap()).await - } else { - search_fut.await - }; + ) + .instrument(http_span.clone()) + .await; match search_res { Ok(mut res) => { @@ -426,12 +419,12 @@ pub async fn _search_partition_multi( let start = std::time::Instant::now(); let cfg = get_config(); let http_span = if cfg.common.tracing_search_enabled { - Some(tracing::info_span!( + tracing::info_span!( "/api/{org_id}/_search_partition_multi", org_id = org_id.clone() - )) + ) } else { - None + Span::none() }; let trace_id = get_or_create_trace_id(in_req.headers(), &http_span); @@ -449,7 +442,7 @@ pub async fn _search_partition_multi( let search_fut = SearchService::search_partition_multi(&trace_id, &org_id, stream_type, &req); let search_res = if !cfg.common.tracing_enabled && cfg.common.tracing_search_enabled { - search_fut.instrument(http_span.unwrap()).await + search_fut.instrument(http_span).await } else { search_fut.await }; @@ -567,13 +560,13 @@ pub async fn around_multi( let cfg = get_config(); let http_span = if cfg.common.tracing_search_enabled { - Some(tracing::info_span!( + tracing::info_span!( "/api/{org_id}/{stream_names}/_around_multi", org_id = org_id.clone(), stream_names = stream_names.clone() - )) + ) } else { - None + Span::none() }; let trace_id = get_or_create_trace_id(in_req.headers(), &http_span); @@ -714,13 +707,11 @@ pub async fn around_multi( timeout, search_type: Some(search::SearchEventType::UI), }; - let search_fut = - SearchService::search(&trace_id, &org_id, stream_type, user_id.clone(), &req); - let search_res = if !cfg.common.tracing_enabled && cfg.common.tracing_search_enabled { - search_fut.instrument(http_span.clone().unwrap()).await - } else { - search_fut.await - }; + let search_res = + SearchService::search(&trace_id, &org_id, stream_type, user_id.clone(), &req) + .instrument(http_span.clone()) + .await; + let resp_forward = match search_res { Ok(res) => res, Err(err) => { @@ -791,13 +782,11 @@ pub async fn around_multi( timeout, search_type: Some(search::SearchEventType::UI), }; - let search_fut = - SearchService::search(&trace_id, &org_id, stream_type, user_id.clone(), &req); - let search_res = if !cfg.common.tracing_enabled && cfg.common.tracing_search_enabled { - search_fut.instrument(http_span.clone().unwrap()).await - } else { - search_fut.await - }; + let search_res = + SearchService::search(&trace_id, &org_id, stream_type, user_id.clone(), &req) + .instrument(http_span.clone()) + .await; + let resp_backward = match search_res { Ok(res) => res, Err(err) => { diff --git a/src/handler/http/request/stream/mod.rs b/src/handler/http/request/stream/mod.rs index efa3f8d61e..9ee2cf3e2f 100644 --- a/src/handler/http/request/stream/mod.rs +++ b/src/handler/http/request/stream/mod.rs @@ -318,6 +318,12 @@ async fn delete_stream_cache( path: web::Path<(String, String)>, req: HttpRequest, ) -> Result { + if !config::get_config().common.result_cache_enabled { + return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + "Result Cache is disabled".to_string(), + ))); + } let (org_id, stream_name) = path.into_inner(); let query = web::Query::>::from_query(req.query_string()).unwrap(); let stream_type = match get_stream_type_from_request(&query) { diff --git a/src/handler/http/request/traces/mod.rs b/src/handler/http/request/traces/mod.rs index 534367cd02..bdbbe2b657 100644 --- a/src/handler/http/request/traces/mod.rs +++ b/src/handler/http/request/traces/mod.rs @@ -19,7 +19,7 @@ use actix_web::{get, http, post, web, HttpRequest, HttpResponse}; use config::{get_config, meta::stream::StreamType, metrics, utils::json}; use infra::errors; use serde::Serialize; -use tracing::Instrument; +use tracing::{Instrument, Span}; use crate::{ common::{ @@ -133,13 +133,13 @@ pub async fn get_latest_traces( let start = std::time::Instant::now(); let (org_id, stream_name) = path.into_inner(); let http_span = if cfg.common.tracing_search_enabled { - Some(tracing::info_span!( + tracing::info_span!( "/api/{org_id}/{stream_name}/traces/latest", org_id = org_id.clone(), stream_name = stream_name.clone() - )) + ) } else { - None + Span::none() }; let trace_id = get_or_create_trace_id(in_req.headers(), &http_span); @@ -275,12 +275,9 @@ pub async fn get_latest_traces( .ok() .map(|v| v.to_string()); - let search_fut = SearchService::search(&trace_id, &org_id, stream_type, user_id.clone(), &req); - let search_res = if !cfg.common.tracing_enabled && cfg.common.tracing_search_enabled { - search_fut.instrument(http_span.clone().unwrap()).await - } else { - search_fut.await - }; + let search_res = SearchService::search(&trace_id, &org_id, stream_type, user_id.clone(), &req) + .instrument(http_span.clone()) + .await; let resp_search = match search_res { Ok(res) => res, @@ -367,13 +364,11 @@ pub async fn get_latest_traces( let mut traces_service_name: HashMap> = HashMap::new(); loop { - let search_fut = - SearchService::search(&trace_id, &org_id, stream_type, user_id.clone(), &req); - let search_res = if !cfg.common.tracing_enabled && cfg.common.tracing_search_enabled { - search_fut.instrument(http_span.clone().unwrap()).await - } else { - search_fut.await - }; + let search_res = + SearchService::search(&trace_id, &org_id, stream_type, user_id.clone(), &req) + .instrument(http_span.clone()) + .await; + let resp_search = match search_res { Ok(res) => res, Err(err) => { diff --git a/src/infra/src/cache/file_data/disk.rs b/src/infra/src/cache/file_data/disk.rs index f5b4236164..5d90629bc9 100644 --- a/src/infra/src/cache/file_data/disk.rs +++ b/src/infra/src/cache/file_data/disk.rs @@ -412,11 +412,13 @@ async fn load(root_dir: &PathBuf, scan_dir: &PathBuf) -> Result<(), anyhow::Erro ); let meta = columns[5].split('_').collect::>(); let is_aggregate = meta[2] == "1"; + let is_descending = meta[3] == "1"; result_cache.entry(query_key).or_insert_with(Vec::new).push( ResultCacheMeta { start_time: meta[0].parse().unwrap(), end_time: meta[1].parse().unwrap(), is_aggregate, + is_descending, }, ); }; diff --git a/src/infra/src/cache/meta.rs b/src/infra/src/cache/meta.rs index c62735604c..1c20b9d5d7 100644 --- a/src/infra/src/cache/meta.rs +++ b/src/infra/src/cache/meta.rs @@ -20,4 +20,5 @@ pub struct ResultCacheMeta { pub start_time: i64, pub end_time: i64, pub is_aggregate: bool, + pub is_descending: bool, } diff --git a/src/proto/proto/cluster/querycache.proto b/src/proto/proto/cluster/querycache.proto index d67e68a06b..8d24c8515a 100644 --- a/src/proto/proto/cluster/querycache.proto +++ b/src/proto/proto/cluster/querycache.proto @@ -27,16 +27,16 @@ message QueryCacheRequest { string timestamp_col = 6; string trace_id = 7; int64 discard_interval = 8; + bool is_descending = 9; } message QueryCacheRes { QueryResponse cached_response = 1; - repeated QueryDelta deltas = 2; - bool has_pre_cache_delta = 3; - bool has_cached_data = 4; - bool cache_query_response = 5; - int64 cache_start_time = 6; - int64 cache_end_time = 7; + bool has_cached_data = 2; + bool cache_query_response = 3; + int64 cache_start_time = 4; + int64 cache_end_time = 5; + bool is_descending = 6; } message QueryCacheResponse { diff --git a/src/service/search/cache/cacher.rs b/src/service/search/cache/cacher.rs index f112c85cb1..32238de58f 100644 --- a/src/service/search/cache/cacher.rs +++ b/src/service/search/cache/cacher.rs @@ -26,7 +26,7 @@ use infra::cache::{ }; use crate::{ - common::meta::search::{CachedQueryResponse, QueryDelta}, + common::meta::search::{CacheQueryRequest, CachedQueryResponse, QueryDelta}, service::search::{ cache::result_utils::{get_ts_value, round_down_to_nearest_minute}, sql::{generate_histogram_interval, SqlMode, RE_HISTOGRAM, RE_SELECT_FROM}, @@ -57,8 +57,10 @@ pub async fn check_cache( }; let sql_mode: SqlMode = meta.sql_mode; - // skip the count queries - if sql_mode.eq(&SqlMode::Full) && req.query.track_total_hits { + let order_by = meta.meta.order_by; + + // skip the count queries & queries with multiple order by fields + if (sql_mode.eq(&SqlMode::Full) && req.query.track_total_hits) || order_by.len() > 1 { return CachedQueryResponse::default(); } let mut result_ts_col = get_ts_col(parsed_sql, &cfg.common.column_timestamp, is_aggregate); @@ -110,21 +112,46 @@ pub async fn check_cache( } let query_key = file_path.replace('/', "_"); + let mut is_descending = true; + + if !order_by.is_empty() { + for (field, order) in &order_by { + if field.eq(&result_ts_col) { + is_descending = *order; + } + } + } + let mut c_resp = match crate::service::search::cluster::cacher::get_cached_results( - req.query.start_time, - req.query.end_time, - is_aggregate, query_key.to_owned(), file_path.to_string(), trace_id.to_owned(), - result_ts_col.clone(), - discard_interval, + CacheQueryRequest { + q_start_time: req.query.start_time, + q_end_time: req.query.end_time, + is_aggregate, + ts_column: result_ts_col.clone(), + discard_interval, + is_descending, + }, ) .await { Some(mut cached_resp) => { - let search_delta: Vec = cached_resp - .deltas + let mut deltas = vec![]; + calculate_deltas_v1( + &ResultCacheMeta { + start_time: cached_resp.response_start_time, + end_time: cached_resp.response_end_time, + is_aggregate, + is_descending: true, + }, + req.query.start_time, + req.query.end_time, + &mut deltas, + ); + + let search_delta: Vec = deltas .iter() .filter(|d| !d.delta_removed_hits) .cloned() @@ -133,7 +160,10 @@ pub async fn check_cache( log::debug!("cached response found"); *should_exec_query = false; }; - if cached_resp.cached_response.total == (meta.meta.limit as usize) { + + if cached_resp.cached_response.total == (meta.meta.limit as usize) + && cached_resp.response_end_time == req.query.end_time + { *should_exec_query = false; cached_resp.deltas = vec![]; } else { @@ -144,24 +174,25 @@ pub async fn check_cache( cached_resp } None => { + // since there is no cache & will be cached in the end we should return the response log::debug!("cached response not found"); - CachedQueryResponse::default() + CachedQueryResponse { + is_descending, + ..Default::default() + } } }; c_resp.cache_query_response = true; + + c_resp.limit = meta.meta.limit as i64; c_resp.ts_column = result_ts_col; c_resp } -#[allow(clippy::too_many_arguments)] pub async fn get_cached_results( - start_time: i64, - end_time: i64, - is_aggregate: bool, file_path: &str, - result_ts_column: &str, trace_id: &str, - discard_interval: i64, + cache_req: CacheQueryRequest, ) -> Option { let r = QUERY_RESULT_CACHE.read().await; let query_key = file_path.replace('/', "_"); @@ -174,16 +205,20 @@ pub async fn get_cached_results( .filter(|cache_meta| { // to make sure there is overlap between cache time range and query time range - cache_meta.start_time <= end_time && cache_meta.end_time >= start_time + cache_meta.start_time <= cache_req.q_end_time + && cache_meta.end_time >= cache_req.q_start_time }) - .max_by_key(|result| result.end_time.min(end_time) - result.start_time.max(start_time)) - { + .max_by_key(|result| { + result.end_time.min(cache_req.q_end_time) + - result.start_time.max(cache_req.q_start_time) + }) { Some(matching_meta) => { let file_name = format!( - "{}_{}_{}.json", + "{}_{}_{}_{}.json", matching_meta.start_time, matching_meta.end_time, - if is_aggregate { 1 } else { 0 } + if cache_req.is_aggregate { 1 } else { 0 }, + if cache_req.is_descending { 1 } else { 0 } ); let mut matching_cache_meta = matching_meta.clone(); // calculate delta time range to fetch the delta data using search query @@ -196,80 +231,55 @@ pub async fn get_cached_results( return None; } - let mut deltas = vec![]; - - calculate_deltas_v1(&matching_cache_meta, start_time, end_time, &mut deltas); - - let remove_hits: Vec<&QueryDelta> = - deltas.iter().filter(|d| d.delta_removed_hits).collect(); - match get_results(file_path, &file_name).await { Ok(v) => { let mut cached_response: Response = json::from_str::(&v).unwrap(); - // remove hits if time range is lesser than cached time range - let mut to_retain = Vec::new(); + let first_ts = get_ts_value( + &cache_req.ts_column, + cached_response.hits.first().unwrap(), + ); - let first_ts = - get_ts_value(result_ts_column, cached_response.hits.first().unwrap()); + let last_ts = get_ts_value( + &cache_req.ts_column, + cached_response.hits.last().unwrap(), + ); - let last_ts = - get_ts_value(result_ts_column, cached_response.hits.last().unwrap()); - - let discard_ts = if discard_interval > 0 { - // for histogram - if first_ts < last_ts { - last_ts - } else { + let discard_ts = if cache_req.is_descending { + if cache_req.discard_interval > 0 { first_ts + } else { + // non-aggregate query + let m_first_ts = round_down_to_nearest_minute(first_ts); + if Utc::now().timestamp_micros() - discard_duration < m_first_ts { + m_first_ts - discard_duration + } else { + m_first_ts + } } - } else if first_ts < last_ts { - // non-aggregate quer + } else if cache_req.discard_interval > 0 { + last_ts + } else { + // non-aggregate query let m_last_ts = round_down_to_nearest_minute(last_ts); if Utc::now().timestamp_micros() - discard_duration < last_ts { m_last_ts - discard_duration } else { - matching_cache_meta.end_time - } - } else { - let m_first_ts = round_down_to_nearest_minute(first_ts); - if Utc::now().timestamp_micros() - discard_duration < m_first_ts { - m_first_ts - discard_duration - } else { - matching_cache_meta.start_time + m_last_ts } }; - if !remove_hits.is_empty() { - for delta in remove_hits { - for hit in &cached_response.hits { - let hit_ts = get_ts_value(result_ts_column, hit); + cached_response.hits.retain(|hit| { + let hit_ts = get_ts_value(&cache_req.ts_column, hit); + hit_ts <= cache_req.q_end_time + && hit_ts >= cache_req.q_start_time + && hit_ts < discard_ts + }); - if !(hit_ts >= delta.delta_start_time - && hit_ts < delta.delta_end_time) - && (hit_ts <= end_time && hit_ts >= start_time) - && hit_ts < discard_ts - { - to_retain.push(hit.clone()); - } - } - } - cached_response.hits = to_retain; - cached_response.total = cached_response.hits.len(); - if discard_interval < 0 { - matching_cache_meta.end_time = discard_ts; - }; + cached_response.total = cached_response.hits.len(); + if cache_req.discard_interval < 0 { + matching_cache_meta.end_time = discard_ts; }; - // recalculate deltas - let mut deltas = vec![]; - - let has_pre_cache_delta = calculate_deltas_v1( - &matching_cache_meta, - start_time, - end_time, - &mut deltas, - ); - log::info!( "[trace_id {trace_id}] Get results from disk success for query key: {} with start time {} - end time {} ", query_key, @@ -278,13 +288,14 @@ pub async fn get_cached_results( ); Some(CachedQueryResponse { cached_response, - deltas, - has_pre_cache_delta, + deltas: vec![], has_cached_data: true, cache_query_response: true, response_start_time: matching_cache_meta.start_time, response_end_time: matching_cache_meta.end_time, - ts_column: result_ts_column.to_string(), + ts_column: cache_req.ts_column.to_string(), + is_descending: cache_req.is_descending, + limit: -1, }) } Err(e) => { diff --git a/src/service/search/cluster/cacher.rs b/src/service/search/cluster/cacher.rs index 60fefabe63..96dea24494 100644 --- a/src/service/search/cluster/cacher.rs +++ b/src/service/search/cluster/cacher.rs @@ -19,18 +19,16 @@ use proto::cluster_rpc::{self, DeleteResultCacheRequest, QueryCacheRequest}; use tonic::{codec::CompressionEncoding, metadata::MetadataValue, transport::Channel, Request}; use tracing::{info_span, Instrument}; -use crate::{common::meta::search::CachedQueryResponse, service::search::infra_cluster}; +use crate::{ + common::meta::search::{CacheQueryRequest, CachedQueryResponse}, + service::search::infra_cluster, +}; -#[allow(clippy::too_many_arguments)] pub async fn get_cached_results( - start_time: i64, - end_time: i64, - is_aggregate: bool, query_key: String, file_path: String, trace_id: String, - result_ts_column: String, - discard_interval: i64, + cache_req: CacheQueryRequest, ) -> Option { let start = std::time::Instant::now(); // get nodes from cluster @@ -51,6 +49,7 @@ pub async fn get_cached_results( return None; }; + let ts_column = cache_req.ts_column; let mut tasks = Vec::new(); for node in nodes { let cfg = config::get_config(); @@ -64,18 +63,19 @@ pub async fn get_cached_results( let query_key = query_key.clone(); let file_path = file_path.clone(); let trace_id = trace_id.clone(); - let result_ts_column = result_ts_column.to_string(); + let ts_column = ts_column.to_string(); let task = tokio::task::spawn( async move { let req = QueryCacheRequest { - start_time, - end_time, - is_aggregate, + start_time: cache_req.q_start_time, + end_time: cache_req.q_end_time, + is_aggregate :cache_req.is_aggregate, query_key, file_path, - timestamp_col: result_ts_column, + timestamp_col: ts_column.to_string(), trace_id:trace_id.clone(), - discard_interval, + discard_interval:cache_req.discard_interval, + is_descending:cache_req.is_descending, }; let request = tonic::Request::new(req); @@ -145,15 +145,6 @@ pub async fn get_cached_results( Ok(res) => match res { Ok((node, node_res)) => match node_res.response { Some(res) => { - let deltas = res - .deltas - .iter() - .map(|d| crate::common::meta::search::QueryDelta { - delta_start_time: d.delta_start_time, - delta_end_time: d.delta_end_time, - delta_removed_hits: d.delta_removed_hits, - }) - .collect(); let cached_res: config::meta::search::Response = match res.cached_response { Some(cached_response) => { match serde_json::from_slice(&cached_response.data) { @@ -181,13 +172,14 @@ pub async fn get_cached_results( node, CachedQueryResponse { cached_response: cached_res, - deltas, - has_pre_cache_delta: res.has_pre_cache_delta, + deltas: vec![], has_cached_data: res.has_cached_data, cache_query_response: res.cache_query_response, response_start_time: res.cache_start_time, response_end_time: res.cache_end_time, - ts_column: result_ts_column.to_string(), + ts_column: ts_column.clone(), + is_descending: res.is_descending, + limit: -1, }, )); } @@ -213,13 +205,16 @@ pub async fn get_cached_results( } if let Some(local_resp) = crate::service::search::cache::cacher::get_cached_results( - start_time, - end_time, - is_aggregate, &file_path, - result_ts_column.as_str(), &trace_id, - discard_interval, + CacheQueryRequest { + q_start_time: cache_req.q_start_time, + q_end_time: cache_req.q_end_time, + is_aggregate: cache_req.is_aggregate, + ts_column, + discard_interval: cache_req.discard_interval, + is_descending: cache_req.is_descending, + }, ) .await { @@ -231,10 +226,12 @@ pub async fn get_cached_results( .filter(|(_, cache_meta)| { // to make sure there is overlap between cache time range and query time range & - cache_meta.response_start_time <= end_time && cache_meta.response_end_time >= start_time + cache_meta.response_start_time <= cache_req.q_end_time + && cache_meta.response_end_time >= cache_req.q_start_time }) .max_by_key(|(_, result)| { - result.response_end_time.min(end_time) - result.response_start_time.max(start_time) + result.response_end_time.min(cache_req.q_end_time) + - result.response_start_time.max(cache_req.q_start_time) }) { Some((node, result)) => { log::info!( diff --git a/src/service/search/mod.rs b/src/service/search/mod.rs index dc0ca818d1..d9081f812d 100644 --- a/src/service/search/mod.rs +++ b/src/service/search/mod.rs @@ -67,7 +67,6 @@ pub(crate) static QUEUE_LOCKER: Lazy>> = static RE_SELECT_WILDCARD: Lazy = Lazy::new(|| Regex::new(r"(?i)select\s+\*\s+from").unwrap()); -#[tracing::instrument(name = "service:search:enter", skip(in_req))] pub async fn search( trace_id: &str, org_id: &str,