diff --git a/src/common/meta/search.rs b/src/common/meta/search.rs index cab8d2614d..fd667f3133 100644 --- a/src/common/meta/search.rs +++ b/src/common/meta/search.rs @@ -35,7 +35,6 @@ pub struct CachedQueryResponse { pub struct QueryDelta { pub delta_start_time: i64, pub delta_end_time: i64, - pub delta_removed_hits: bool, } #[derive(Clone, Debug, Serialize, Deserialize, ToSchema, Default)] @@ -128,7 +127,6 @@ mod tests { deltas: vec![QueryDelta { delta_start_time: 1000, delta_end_time: 2000, - delta_removed_hits: false, }], has_cached_data: true, cache_query_response: true, @@ -154,12 +152,10 @@ mod tests { let delta = QueryDelta { delta_start_time: 1000, delta_end_time: 2000, - delta_removed_hits: true, }; assert_eq!(delta.delta_start_time, 1000); assert_eq!(delta.delta_end_time, 2000); - assert!(delta.delta_removed_hits); } #[test] @@ -167,13 +163,11 @@ mod tests { let delta1 = QueryDelta { delta_start_time: 1000, delta_end_time: 2000, - delta_removed_hits: false, }; let delta2 = QueryDelta { delta_start_time: 2000, delta_end_time: 3000, - delta_removed_hits: false, }; assert!(delta1 < delta2); diff --git a/src/service/runtime_metrics.rs b/src/service/runtime_metrics.rs index 679979881a..2d2e4ca057 100644 --- a/src/service/runtime_metrics.rs +++ b/src/service/runtime_metrics.rs @@ -234,7 +234,7 @@ pub async fn start_metrics_collector() { #[cfg(not(tokio_unstable))] log::info!( - "Starting runtime metrics collector (basic mode - compile with --cfg tokio_unstable for detailed metrics)" + "Starting runtime metrics collector (basic mode - compile with --cfg tokio_unstable for detail)" ); tokio::spawn(async { diff --git a/src/service/search/cache/cacher.rs b/src/service/search/cache/cacher.rs index 57edd31dd9..c66107bfc1 100644 --- a/src/service/search/cache/cacher.rs +++ b/src/service/search/cache/cacher.rs @@ -153,7 +153,7 @@ pub async fn check_cache( }; let mut histogram_interval = -1; - if let Some(interval) = sql.histogram_interval { + if is_aggregate && let Some(interval) = sql.histogram_interval { *file_path = format!("{file_path}_{interval}_{result_ts_col}"); let mut req_time_range = (req.query.start_time, req.query.end_time); @@ -186,12 +186,17 @@ pub async fn check_cache( if is_aggregate && order_by.is_empty() && result_ts_col.is_empty() { return MultiCachedQueryResponse::default(); } - let mut multi_resp = MultiCachedQueryResponse::default(); + let mut multi_resp = MultiCachedQueryResponse { + trace_id: trace_id.to_string(), + is_aggregate, + is_descending, + ..Default::default() + }; if histogram_interval > -1 { multi_resp.histogram_interval = histogram_interval / 1000 / 1000; } log::info!( - "[CACHE CANDIDATES {trace_id}] request result_ts_col: {}, histogram_interval: {}, time range: {} - {}", + "[trace_id {trace_id}] check_cache: result_ts_col: {}, histogram_interval: {}, time range: {} - {}", result_ts_col, histogram_interval, req.query.start_time, @@ -256,11 +261,7 @@ pub async fn check_cache( } if !deltas.is_empty() { - let search_delta: Vec = deltas - .iter() - .filter(|d| !d.delta_removed_hits) - .cloned() - .collect(); + let search_delta: Vec = deltas.to_vec(); if search_delta.is_empty() { log::debug!("cached response found"); *should_exec_query = false; @@ -269,7 +270,6 @@ pub async fn check_cache( } } multi_resp.cache_query_response = true; - multi_resp.is_descending = is_descending; multi_resp.limit = sql.limit as i64; multi_resp.ts_column = result_ts_col; multi_resp.took = start.elapsed().as_millis() as usize; @@ -328,11 +328,7 @@ pub async fn check_cache( &mut deltas, ); - let search_delta: Vec = deltas - .iter() - .filter(|d| !d.delta_removed_hits) - .cloned() - .collect(); + let search_delta: Vec = deltas.to_vec(); if search_delta.is_empty() { log::debug!("cached response found"); *should_exec_query = false; @@ -361,7 +357,6 @@ pub async fn check_cache( }; multi_resp.deltas = c_resp.deltas.clone(); multi_resp.has_cached_data = c_resp.has_cached_data; - multi_resp.is_descending = is_descending; multi_resp.cached_response.push(c_resp); multi_resp.took = start.elapsed().as_millis() as usize; multi_resp.cache_query_response = true; @@ -398,27 +393,28 @@ pub async fn get_cached_results( .filter(|m| { // to make sure there is overlap between cache time range and query time range log::info!( - "[CACHE CANDIDATES {trace_id}] get_cached_results: cache_meta response time_range: {} - {}", + "[CACHE CANDIDATES {trace_id}] get_cached_results: cache_meta time_range: {} - {}", m.start_time, m.end_time ); // check if the data is matching for histogram - if cache_req.histogram_interval > 0 - && ( - m.start_time % cache_req.histogram_interval != 0 - || m.end_time % cache_req.histogram_interval != 0 - ) { + if cache_req.is_aggregate + && cache_req.histogram_interval > 0 + && (m.start_time % cache_req.histogram_interval != 0 + || m.end_time % cache_req.histogram_interval != 0) + { return false; } - m.start_time <= cache_req.q_end_time && - m.end_time >= cache_req.q_start_time + m.start_time <= cache_req.q_end_time && m.end_time >= cache_req.q_start_time }) - .max_by_key(|result| select_cache_meta(result, &cache_req, &selection_strategy)) { - Some(v) => v.clone(), - None => { + .max_by_key(|result| select_cache_meta(result, &cache_req, &selection_strategy)) + { + Some(v) => v.clone(), + None => { log::debug!("No matching cache found for query key: {query_key}"); return None; - }}; + } + }; // get the cache data from disk let file_name = format!( @@ -431,7 +427,9 @@ pub async fn get_cached_results( let data = match get_results(file_path, &file_name).await { Ok(v) => v, Err(e) => { - log::error!("[trace_id {trace_id}] Get results from disk failed: {e}"); + log::error!( + "[trace_id {trace_id}] Get results from disk failed: file: {file_path}/{file_name}, error: {e}" + ); return None; } }; @@ -444,15 +442,8 @@ pub async fn get_cached_results( }; // filter data based on request time range - let (hits_allowed_start_time, hits_allowed_end_time) = if cache_req.histogram_interval > 0 { - // calculation in line with date bin of datafusion - ( - cache_req.q_start_time - (cache_req.q_start_time % cache_req.histogram_interval), - cache_req.q_end_time - (cache_req.q_end_time % cache_req.histogram_interval), - ) - } else { - (cache_req.q_start_time, cache_req.q_end_time) - }; + let (hits_allowed_start_time, hits_allowed_end_time) = + (cache_req.q_start_time, cache_req.q_end_time); let first_ts = get_ts_value(&cache_req.ts_column, cached_response.hits.first().unwrap()); let last_ts = get_ts_value(&cache_req.ts_column, cached_response.hits.last().unwrap()); let data_start_time = std::cmp::min(first_ts, last_ts); @@ -461,7 +452,7 @@ pub async fn get_cached_results( if data_start_time < cache_req.q_start_time || data_end_time > cache_req.q_end_time { cached_response.hits.retain(|hit| { let hit_ts = get_ts_value(&cache_req.ts_column, hit); - hit_ts <= hits_allowed_end_time && hit_ts >= hits_allowed_start_time + hit_ts < hits_allowed_end_time && hit_ts >= hits_allowed_start_time }); // if the data is empty after filtering, return None if cached_response.hits.is_empty() { @@ -476,8 +467,7 @@ pub async fn get_cached_results( cached_response.total = cached_response.hits.len(); log::info!( - "[CACHE RESULT {trace_id}] Get results from disk success for query key: {} with time range {} - {} ", - query_key, + "[CACHE RESULT {trace_id}] Get results from disk success for query key: {query_key} with time range {} - {} ", matching_meta.start_time, matching_meta.end_time ); @@ -497,70 +487,51 @@ pub async fn get_cached_results( pub fn calculate_deltas( result_meta: &ResultCacheMeta, - start_time: i64, - end_time: i64, + query_start_time: i64, + query_end_time: i64, deltas: &mut Vec, -) -> bool { - let mut has_pre_cache_delta = false; - if start_time == result_meta.start_time && end_time == result_meta.end_time { +) { + if query_start_time == result_meta.start_time && query_end_time == result_meta.end_time { // If query start time and end time are the same as cache times, return results from cache - return has_pre_cache_delta; + return; } - // Query Start time < ResultCacheMeta start time & Query End time > ResultCacheMeta End time - if end_time != result_meta.end_time { - if end_time > result_meta.end_time { - // q end time : 11:00, r end time : 10:45 - deltas.push(QueryDelta { - delta_start_time: result_meta.end_time, - delta_end_time: end_time, - delta_removed_hits: false, - }); - } else { - deltas.push(QueryDelta { - delta_start_time: end_time, - delta_end_time: result_meta.end_time, - delta_removed_hits: true, - }); - } + if query_end_time > result_meta.end_time { + // q end time : 11:00, r end time : 10:45 + // for delta start time we need to add 1 microsecond to the end time + // because we will include the start_time, if we don't add 1 microsecond + // the start_time will be include in the next search, we will get duplicate data + deltas.push(QueryDelta { + delta_start_time: result_meta.end_time + 1, + delta_end_time: query_end_time, + }); } - // Query Start time > ResultCacheMeta start time & Query End time > ResultCacheMeta End time -> - // typical last x min/hours/days of data - if start_time != result_meta.start_time { - if start_time > result_meta.start_time { - // q start time : 10:00, r start time : 09:00 - // Fetch data between ResultCacheMeta Start time & Query start time - deltas.push(QueryDelta { - delta_start_time: result_meta.start_time, - delta_end_time: start_time, - delta_removed_hits: true, - }); - } else { - deltas.push(QueryDelta { - delta_start_time: start_time, - delta_end_time: result_meta.start_time, - delta_removed_hits: false, - }); - has_pre_cache_delta = true; - } + + if query_start_time < result_meta.start_time { + // q start time : 10:00, r start time : 10:15 + deltas.push(QueryDelta { + delta_start_time: query_start_time, + delta_end_time: result_meta.start_time, + }); } - has_pre_cache_delta } pub async fn cache_results_to_disk( file_path: &str, file_name: &str, data: String, -) -> std::io::Result<()> { +) -> std::io::Result { let file = format!("results/{file_path}/{file_name}"); + if disk::exist(&file).await { + return Ok(false); + } match disk::set(&file, Bytes::from(data)).await { - Ok(_) => (), + Ok(_) => Ok(true), Err(e) => { log::error!("Error caching results to disk: {e}"); - return Err(std::io::Error::other("Error caching results to disk")); + Err(std::io::Error::other("Error caching results to disk")) } } - Ok(()) } pub async fn get_results(file_path: &str, file_name: &str) -> std::io::Result { @@ -805,12 +776,6 @@ fn calculate_deltas_multi( let mut current_end_time = start_time; for meta in results { - log::info!( - "meta time {} - {} - records {}", - meta.response_start_time, - meta.response_end_time, - meta.cached_response.hits.len() - ); cache_duration += meta.response_end_time - meta.response_start_time; let delta_end_time = if histogram_interval > 0 && !meta.cached_response.hits.is_empty() { // If histogram interval > 0, we need to adjust the end time to the nearest interval @@ -830,7 +795,6 @@ fn calculate_deltas_multi( deltas.push(QueryDelta { delta_start_time: current_end_time, delta_end_time, - delta_removed_hits: false, }); } // Update the current end time to the end of the current meta @@ -852,7 +816,6 @@ fn calculate_deltas_multi( deltas.push(QueryDelta { delta_start_time: expected_delta_start_time, delta_end_time: end_time, - delta_removed_hits: false, }); } @@ -952,7 +915,6 @@ mod tests { let expected_deltas = vec![QueryDelta { delta_start_time: 1747659555000000, delta_end_time: 1747659600000000, - delta_removed_hits: false, }]; assert_eq!(deltas, expected_deltas); } diff --git a/src/service/search/cache/mod.rs b/src/service/search/cache/mod.rs index a06c164c9d..941fb50300 100644 --- a/src/service/search/cache/mod.rs +++ b/src/service/search/cache/mod.rs @@ -94,18 +94,9 @@ pub async fn search( false }; - // Result caching check start - let mut origin_sql = in_req.query.sql.clone(); - let is_aggregate = is_aggregate_query(&origin_sql).unwrap_or(true); - let (stream_name, all_streams) = match resolve_stream_names(&origin_sql) { - // result cache doesn't support multiple stream names - Ok(v) => (v[0].clone(), v.join(",")), - Err(e) => { - return Err(Error::Message(e.to_string())); - } - }; - let mut req = in_req.clone(); + + // check the original query function first let mut query_fn = req .query .query_fn @@ -123,67 +114,19 @@ pub async fn search( query_fn = None; } - let action = req - .query - .action_id - .as_ref() - .and_then(|v| svix_ksuid::Ksuid::from_str(v).ok()); + // Result caching check start + let (mut c_resp, should_exec_query) = + prepare_cache_response(trace_id, org_id, stream_type, &mut req, use_cache).await?; + let file_path = c_resp.file_path.clone(); - // calculate hash for the query with version - let mut hash_body = vec![CACHE_VERSION.to_string(), origin_sql.to_string()]; - if let Some(vrl_function) = &query_fn { - hash_body.push(vrl_function.to_string()); - } - if let Some(action_id) = action { - hash_body.push(action_id.to_string()); - } - if !req.regions.is_empty() { - hash_body.extend(req.regions.clone()); - } - if !req.clusters.is_empty() { - hash_body.extend(req.clusters.clone()); - } - let mut h = config::utils::hash::gxhash::new(); - let hashed_query = h.sum64(&hash_body.join(",")); - - let mut should_exec_query = true; - - let mut file_path = format!("{org_id}/{stream_type}/{stream_name}/{hashed_query}"); - let mut c_resp: MultiCachedQueryResponse = if use_cache { - // if cache is used, we need to check the cache - check_cache( - trace_id, - org_id, - stream_type, - &mut req, - &mut origin_sql, - &mut file_path, - is_aggregate, - &mut should_exec_query, - ) - .await - } else { - // if cache is not used, we need to parse the sql to get the ts column and is descending - let query: SearchQuery = req.query.clone().into(); - match crate::service::search::Sql::new(&query, org_id, stream_type, req.search_type).await { - Ok(v) => { - let (ts_column, is_descending) = - cacher::get_ts_col_order_by(&v, TIMESTAMP_COL_NAME, is_aggregate) - .unwrap_or_default(); - - let order_by = v.order_by; - - MultiCachedQueryResponse { - ts_column, - is_descending, - order_by, - ..Default::default() - } - } - Err(e) => { - log::error!("Error parsing sql: {e}"); - MultiCachedQueryResponse::default() - } + // get the modified original sql from req + let origin_sql = in_req.query.sql.clone(); + let is_aggregate = is_aggregate_query(&origin_sql).unwrap_or(true); + let (stream_name, all_streams) = match resolve_stream_names(&origin_sql) { + // result cache doesn't support multiple stream names + Ok(v) => (v[0].clone(), v.join(",")), + Err(e) => { + return Err(Error::Message(e.to_string())); } }; @@ -192,7 +135,6 @@ pub async fn search( c_resp.deltas.push(QueryDelta { delta_start_time: req.query.start_time, delta_end_time: req.query.end_time, - delta_removed_hits: false, }); } else if use_cache { log::info!( @@ -252,7 +194,10 @@ pub async fn search( log::info!( "{}", search_inspector_fields( - format!("[trace_id {trace_id}] deltas are : {:?}", c_resp.deltas), + format!( + "[trace_id {trace_id}] Qeury deltas are: {:?}", + c_resp.deltas + ), SearchInspectorFieldsBuilder::new() .node_name(LOCAL_NODE.name.clone()) .component("cacher:search deltas".to_string()) @@ -266,8 +211,9 @@ pub async fn search( .build() ) ); + log::info!( - "[trace_id {trace_id}] Query original start time: {}, end time : {}", + "[trace_id {trace_id}] Query original start time: {}, end time: {}", req.query.start_time, req.query.end_time ); @@ -298,7 +244,7 @@ pub async fn search( && c_resp.has_cached_data { log::info!( - "[trace_id {trace_id}] Query new start time: {}, end time : {}", + "[trace_id {trace_id}] Query new start time: {}, end time: {}", req.query.start_time, req.query.end_time ); @@ -511,6 +457,113 @@ pub async fn search( Ok(res) } +#[tracing::instrument(name = "service:search:cacher:prepare_cache_response", skip_all)] +pub async fn prepare_cache_response( + trace_id: &str, + org_id: &str, + stream_type: StreamType, + req: &mut search::Request, + use_cache: bool, +) -> Result<(MultiCachedQueryResponse, bool), Error> { + let mut origin_sql = req.query.sql.clone(); + let is_aggregate = is_aggregate_query(&origin_sql).unwrap_or(true); + let stream_name = match resolve_stream_names(&origin_sql) { + // result cache doesn't support multiple stream names + Ok(v) => { + if v.is_empty() { + return Err(Error::Message("Stream name is empty".to_string())); + } else { + v[0].clone() + } + } + Err(e) => { + return Err(Error::Message(e.to_string())); + } + }; + + let mut query_fn = req + .query + .query_fn + .as_ref() + .map(|v| match base64::decode_url(v) { + Ok(v) => v, + Err(_) => v.to_string(), + }); + let is_result_array_skip_vrl = query_fn + .as_ref() + .map(|v| is_result_array_skip_vrl(v)) + .unwrap_or(false); + if is_result_array_skip_vrl { + query_fn = None; + } + + let action = req + .query + .action_id + .as_ref() + .and_then(|v| svix_ksuid::Ksuid::from_str(v).ok()); + + // calculate hash for the query with version + let mut hash_body = vec![CACHE_VERSION.to_string(), origin_sql.to_string()]; + if let Some(vrl_function) = &query_fn { + hash_body.push(vrl_function.to_string()); + } + if let Some(action_id) = action { + hash_body.push(action_id.to_string()); + } + if !req.regions.is_empty() { + hash_body.extend(req.regions.clone()); + } + if !req.clusters.is_empty() { + hash_body.extend(req.clusters.clone()); + } + let mut h = config::utils::hash::gxhash::new(); + let hashed_query = h.sum64(&hash_body.join(",")); + + let mut should_exec_query = true; + + let mut file_path = format!("{org_id}/{stream_type}/{stream_name}/{hashed_query}"); + let resp = if use_cache { + // if cache is used, we need to check the cache + check_cache( + trace_id, + org_id, + stream_type, + req, + &mut origin_sql, + &mut file_path, + is_aggregate, + &mut should_exec_query, + ) + .await + } else { + // if cache is not used, we need to parse the sql to get the ts column and is descending + let query: SearchQuery = req.query.clone().into(); + match crate::service::search::Sql::new(&query, org_id, stream_type, req.search_type).await { + Ok(v) => { + let (ts_column, is_descending) = + cacher::get_ts_col_order_by(&v, TIMESTAMP_COL_NAME, is_aggregate) + .unwrap_or_default(); + + let order_by = v.order_by; + + MultiCachedQueryResponse { + ts_column, + is_aggregate, + is_descending, + order_by, + ..Default::default() + } + } + Err(e) => { + log::error!("Error parsing sql: {e}"); + MultiCachedQueryResponse::default() + } + } + }; + Ok((resp, should_exec_query)) +} + // based on _timestamp of first record in config::meta::search::Response either add it in start // or end to cache response #[tracing::instrument(name = "service:search:cache:merge_response", skip_all)] @@ -751,35 +804,32 @@ pub async fn write_results( return; } - // 1. remove incomplete records for histogram - if let Some(interval) = res.histogram_interval { + // 1. alignment time range for incomplete records for histogram + let mut accept_start_time = req_query_start_time; + let mut accept_end_time = req_query_end_time; + if is_aggregate && let Some(interval) = res.histogram_interval { let interval = interval * 1000 * 1000; // convert to microseconds - if req_query_start_time % interval != 0 || req_query_end_time % interval != 0 { - res.hits.remove(0); // remove first record - res.hits.pop(); // remove last record - res.total = res.hits.len(); - res.size = res.hits.len() as i64; - } + // next interval of start_time + accept_start_time = accept_start_time - (accept_start_time % interval) + interval; + // previous interval of end_time + accept_end_time = accept_end_time - (accept_end_time % interval) - interval; } - // 2. get the latest record, used to check if need to remove records with discard_duration + // 2. get the data time range, check if need to remove records with discard_duration let delay_ts = second_micros(get_config().limit.cache_delay_secs); - let max_ts = Utc::now().timestamp_micros() - delay_ts; - let remove_item = if is_descending { - res.hits.first() - } else { - res.hits.last() - }; - if let Some(item) = remove_item - && let Some(val) = item.get(ts_column) - && let Some(ts) = convert_ts_value_to_datetime(val) - && ts.timestamp_micros() > max_ts - { + let accept_end_time = std::cmp::min(Utc::now().timestamp_micros() - delay_ts, accept_end_time); + let last_rec_ts = get_ts_value(ts_column, res.hits.last().unwrap()); + let first_rec_ts = get_ts_value(ts_column, res.hits.first().unwrap()); + let data_start_time = std::cmp::min(first_rec_ts, last_rec_ts); + let data_end_time = std::cmp::max(first_rec_ts, last_rec_ts); + if data_start_time < accept_start_time || data_end_time > accept_end_time { res.hits.retain(|hit| { if let Some(hit_ts) = hit.get(ts_column) && let Some(hit_ts_datetime) = convert_ts_value_to_datetime(hit_ts) { - hit_ts_datetime.timestamp_micros() <= max_ts + let item_ts = hit_ts_datetime.timestamp_micros(); + // only keep the records within the accept time range + item_ts >= accept_start_time && item_ts <= accept_end_time } else { true } @@ -800,11 +850,12 @@ pub async fn write_results( let cache_start_time = std::cmp::min(first_rec_ts, last_rec_ts); let mut cache_end_time = std::cmp::max(first_rec_ts, last_rec_ts); if (cache_end_time - cache_start_time) < delay_ts { + log::info!("[trace_id {trace_id}] Time range is too short for caching, skipping caching"); return; } // 5. adjust the cache time range - if let Some(interval) = res.histogram_interval { + if is_aggregate && let Some(interval) = res.histogram_interval { cache_end_time += interval * 1000 * 1000; } @@ -822,18 +873,22 @@ pub async fn write_results( match SearchService::cache::cacher::cache_results_to_disk(&file_path, &file_name, res_cache) .await { - Ok(_) => { - QUERY_RESULT_CACHE - .write() - .await - .entry(query_key) - .or_insert_with(Vec::new) - .push(ResultCacheMeta { - start_time: cache_start_time, - end_time: cache_end_time, - is_aggregate, - is_descending, - }); + Ok(success) => { + if success { + // success: true, cache to disk success + // success: false, cache to disk already exists, skipping caching + QUERY_RESULT_CACHE + .write() + .await + .entry(query_key) + .or_insert_with(Vec::new) + .push(ResultCacheMeta { + start_time: cache_start_time, + end_time: cache_end_time, + is_aggregate, + is_descending, + }); + } } Err(e) => { log::error!("Cache results to disk failed: {e}"); @@ -843,13 +898,12 @@ pub async fn write_results( } pub fn apply_vrl_to_response( - backup_query_fn: Option, + query_fn: Option, res: &mut config::meta::search::Response, org_id: &str, stream_name: &str, trace_id: &str, ) -> Vec { - let query_fn = backup_query_fn.clone(); let mut local_res = res.clone(); local_res.hits = if let Some(query_fn) = query_fn @@ -931,89 +985,6 @@ pub fn apply_vrl_to_response( local_res.hits } -#[tracing::instrument(name = "service:search:cacher:check_cache_v2", skip_all)] -pub async fn check_cache_v2( - trace_id: &str, - org_id: &str, - stream_type: StreamType, - in_req: &search::Request, - use_cache: bool, -) -> Result { - // Result caching check start - let mut origin_sql = in_req.query.sql.clone(); - origin_sql = origin_sql.replace('\n', " "); - let is_aggregate = is_aggregate_query(&origin_sql).unwrap_or_default(); - let stream_name = match resolve_stream_names(&origin_sql) { - // TODO: cache don't not support multiple stream names - Ok(v) => v[0].clone(), - Err(e) => { - return Err(Error::Message(e.to_string())); - } - }; - - let mut req = in_req.clone(); - let query_fn = req - .query - .query_fn - .as_ref() - .and_then(|v| base64::decode_url(v).ok()); - - // calculate hash for the query - let mut hash_body = vec![origin_sql.to_string()]; - if let Some(vrl_function) = &query_fn { - hash_body.push(vrl_function.to_string()); - } - if !req.regions.is_empty() { - hash_body.extend(req.regions.clone()); - } - if !req.clusters.is_empty() { - hash_body.extend(req.clusters.clone()); - } - let mut h = config::utils::hash::gxhash::new(); - let hashed_query = h.sum64(&hash_body.join(",")); - - let mut should_exec_query = true; - - let mut file_path = format!("{org_id}/{stream_type}/{stream_name}/{hashed_query}"); - Ok(if use_cache { - let mut resp = check_cache( - trace_id, - org_id, - stream_type, - &mut req, - &mut origin_sql, - &mut file_path, - is_aggregate, - &mut should_exec_query, - ) - .await; - resp.is_aggregate = is_aggregate; - resp.trace_id = trace_id.to_string(); - resp.file_path = file_path; - resp - } else { - let query = req.query.into(); - match crate::service::search::Sql::new(&query, org_id, stream_type, req.search_type).await { - Ok(v) => { - let (ts_column, is_descending) = - cacher::get_ts_col_order_by(&v, TIMESTAMP_COL_NAME, is_aggregate) - .unwrap_or_default(); - - MultiCachedQueryResponse { - ts_column, - is_descending, - file_path, - ..Default::default() - } - } - Err(e) => { - log::error!("[trace_id {trace_id}]: Error parsing sql: {e:?}"); - MultiCachedQueryResponse::default() - } - } - }) -} - fn convert_ts_value_to_datetime(ts_value: &serde_json::Value) -> Option> { match ts_value { // Handle the case where ts_value is a number (microseconds) diff --git a/src/service/search/cache/multi.rs b/src/service/search/cache/multi.rs index 0023f1d101..e62cfff006 100644 --- a/src/service/search/cache/multi.rs +++ b/src/service/search/cache/multi.rs @@ -38,6 +38,7 @@ pub async fn get_cached_results( ); return res; }; + cache_metas.sort_by_key(|meta| meta.start_time); let selection_strategy = ResultCacheSelectionStrategy::from( config::get_config() @@ -51,17 +52,23 @@ pub async fn get_cached_results( }) else { break; }; - let cache_meta = cache_metas.remove(idx); + let best_cache_meta = cache_metas.remove(idx); if let Some(result) = super::cacher::get_cached_results( trace_id, file_path, cache_req.clone(), - Some(vec![cache_meta]), + Some(vec![best_cache_meta.clone()]), ) .await { res.push(result); } + // Remove all overlapping metas with the selected largest meta and call recursively + cache_metas.retain(|meta| { + !best_cache_meta.eq(meta) + && (meta.end_time <= best_cache_meta.start_time + || meta.start_time >= best_cache_meta.end_time) + }); } res diff --git a/src/service/search/mod.rs b/src/service/search/mod.rs index 05d9193012..6b8e234d13 100644 --- a/src/service/search/mod.rs +++ b/src/service/search/mod.rs @@ -937,7 +937,7 @@ pub async fn search_partition( .unwrap() .num_microseconds() .unwrap(); - if is_aggregate && ts_column.is_some() { + if (is_aggregate && ts_column.is_some()) || enable_align_histogram { let hist_int = if let Some(hist_int) = sql.histogram_interval { hist_int } else { @@ -957,27 +957,6 @@ pub async fn search_partition( min_step *= hist_int; } } - // Only for UI search or query param is `true`, we need to generate histogram interval - else if enable_align_histogram { - if let Some(hist_int) = sql.histogram_interval { - // convert seconds to microseconds - min_step = hist_int * 1_000_000; - } else { - let time_range = (req.start_time, req.end_time); - let interval = generate_histogram_interval(Some(time_range)); - match convert_histogram_interval_to_seconds(interval) { - Ok(v) => { - // convert seconds to microseconds - min_step = v * 1_000_000 - } - Err(e) => { - log::error!( - "[trace_id {trace_id}] search_partition: convert_histogram_interval_to_seconds error: {e:?}", - ); - } - } - } - } // Calculate original step with all factors considered let mut total_secs = resp.original_size / cfg.limit.query_group_base_speed / cpu_cores; @@ -1056,13 +1035,13 @@ pub async fn search_partition( part_num, step, min_step, - is_histogram + is_histogram || enable_align_histogram ); // Create a partition generator let generator = partition::PartitionGenerator::new( min_step, cfg.limit.search_mini_partition_duration_secs, - is_histogram, + is_histogram || enable_align_histogram, ); if cfg.common.align_partitions_for_index && use_inverted_index(&sql) { diff --git a/src/service/search/partition.rs b/src/service/search/partition.rs index 389449eb8b..3c7ec86048 100644 --- a/src/service/search/partition.rs +++ b/src/service/search/partition.rs @@ -211,30 +211,29 @@ impl PartitionGenerator { let mut partitions = Vec::new(); let duration = end_time - start_time; - // Generate partitions by DESC order first, then reverse if needed - let mut end = end_time; - let mut last_partition_step = end % self.min_step; + // Handle the alignment for the first and last partition + let mut new_end = end_time; + let last_partition_step = end_time % self.min_step; + if last_partition_step > 0 && duration > self.min_step * 3 { + new_end = end_time - last_partition_step; + partitions.push([new_end, end_time]); + } + let mut new_start = start_time; + let last_partition_step = start_time % self.min_step; + if last_partition_step > 0 && duration > self.min_step * 3 { + new_start = start_time + self.min_step - last_partition_step; + partitions.push([start_time, new_start]); + } - while end > start_time { - let mut start = max(end - step, start_time); - // If the step is greater than the duration, handle the alignment for the boundary - // partition - if last_partition_step > 0 && duration > self.min_step { - // Handle alignment for the first partition - partitions.push([end - last_partition_step, end]); - start -= last_partition_step; - end -= last_partition_step; - } else { - start = max(start - last_partition_step, start_time); - } - // Ensure the start time is not less than the start time of the query - start = max(start, start_time); + let mut end = new_end; + while end > new_start { + let start = max(end - step, new_start); partitions.push([start, end]); end = start; - last_partition_step = 0; } // We need to reverse partitions if query is ASC order + partitions.sort_by(|a, b| b[0].cmp(&a[0])); if order_by == OrderBy::Asc { partitions.reverse(); } diff --git a/src/service/search/sql/mod.rs b/src/service/search/sql/mod.rs index 9d93139815..f41893f292 100644 --- a/src/service/search/sql/mod.rs +++ b/src/service/search/sql/mod.rs @@ -224,7 +224,7 @@ impl Sql { let mut histogram_interval_visitor = HistogramIntervalVisitor::new(Some((query.start_time, query.end_time))); let _ = statement.visit(&mut histogram_interval_visitor); - let histogram_interval = if query.histogram_interval > 0 { + let mut histogram_interval = if query.histogram_interval > 0 { Some(validate_and_adjust_histogram_interval( query.histogram_interval, Some((query.start_time, query.end_time)), @@ -232,6 +232,9 @@ impl Sql { } else { histogram_interval_visitor.interval }; + if !histogram_interval_visitor.is_histogram { + histogram_interval = None; + } //********************Change the sql start*********************************// // 11. replace approx_percentile_cont to new format diff --git a/src/service/search/sql/visitor/histogram_interval.rs b/src/service/search/sql/visitor/histogram_interval.rs index 920e2f988d..f6e9cf0410 100644 --- a/src/service/search/sql/visitor/histogram_interval.rs +++ b/src/service/search/sql/visitor/histogram_interval.rs @@ -20,6 +20,7 @@ use sqlparser::ast::{Expr, FunctionArguments, VisitorMut}; #[derive(Debug)] pub struct HistogramIntervalVisitor { + pub is_histogram: bool, pub interval: Option, time_range: Option<(i64, i64)>, } @@ -27,6 +28,7 @@ pub struct HistogramIntervalVisitor { impl HistogramIntervalVisitor { pub fn new(time_range: Option<(i64, i64)>) -> Self { Self { + is_histogram: false, interval: None, time_range, } @@ -40,6 +42,7 @@ impl VisitorMut for HistogramIntervalVisitor { if let Expr::Function(func) = expr && func.name.to_string().eq_ignore_ascii_case("histogram") { + self.is_histogram = true; if let FunctionArguments::List(list) = &func.args { let mut args = list.args.iter(); // first is field diff --git a/src/service/search/streaming/mod.rs b/src/service/search/streaming/mod.rs index 85644d462e..41b4732575 100644 --- a/src/service/search/streaming/mod.rs +++ b/src/service/search/streaming/mod.rs @@ -20,10 +20,13 @@ use std::time::Instant; -use config::meta::{ - search::{StreamResponses, ValuesEventContext}, - sql::OrderBy, - stream::StreamType, +use config::{ + meta::{ + search::{StreamResponses, ValuesEventContext}, + sql::OrderBy, + stream::StreamType, + }, + utils::time::hour_micros, }; use log; #[cfg(feature = "enterprise")] @@ -135,11 +138,11 @@ pub async fn process_search_stream_request( if req.query.from == 0 && !req.query.track_total_hits && req.query.streaming_id.is_none() { // check cache for the first page - let c_resp = match search_cache::check_cache_v2( + let (c_resp, _should_exec_query) = match search_cache::prepare_cache_response( &trace_id, &org_id, stream_type, - &req, + &mut req, use_cache, ) .instrument(search_span.clone()) @@ -187,6 +190,7 @@ pub async fn process_search_stream_request( return; } }; + let local_c_resp = c_resp.clone(); let cached_resp = local_c_resp.cached_response; let mut deltas = local_c_resp.deltas; @@ -208,17 +212,20 @@ pub async fn process_search_stream_request( .unwrap_or_default(); log::info!( - "[HTTP2_STREAM trace_id {trace_id}] found cache responses len:{}, cache_start_time: {c_start_time}, cache_end_time: {c_end_time}", + "[HTTP2_STREAM trace_id {trace_id}] found cached files: {}, records: {cached_hits}, cache_time: {c_start_time} - {c_end_time}", cached_resp.len(), ); // handle cache responses and deltas if !cached_resp.is_empty() && cached_hits > 0 { // `max_query_range` is used initialize `remaining_query_range` - // set max_query_range to i64::MAX if it is 0, to ensure unlimited query range - // for cache only search + // set max_query_range to `end_time - start_time` as hour if it is 0, to ensure + // unlimited query range for cache only search let remaining_query_range = if max_query_range == 0 { - i64::MAX + std::cmp::max( + 1, + (req.query.end_time - req.query.start_time) / hour_micros(1), + ) } else { max_query_range }; // hours