starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-05 16:10:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: search partition alignment (#8581)

Browse Source
This commit is contained in:
Hengfei Yang
2025-09-25 22:27:48 +08:00
committed by GitHub
parent a49641e625
commit fafd4444ef
10 changed files with 276 additions and 351 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/meta/search.rs
-6
View File
@@ -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);
src/service/runtime_metrics.rs
+1 -1
View File
@@ -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 {
src/service/search/cache/cacher.rs
Vendored
+57 -95
View File
@@ -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<QueryDelta> = deltas
.iter()
.filter(|d| !d.delta_removed_hits)
.cloned()
.collect();
let search_delta: Vec<QueryDelta> = 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<QueryDelta> = deltas
.iter()
.filter(|d| !d.delta_removed_hits)
.cloned()
.collect();
let search_delta: Vec<QueryDelta> = 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<QueryDelta>,
) -> 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<bool> {
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<String> {
@@ -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);
}
src/service/search/cache/mod.rs
Vendored
+165 -194
View File
@@ -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<String>,
query_fn: Option<String>,
res: &mut config::meta::search::Response,
org_id: &str,
stream_name: &str,
trace_id: &str,
) -> Vec<serde_json::Value> {
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<MultiCachedQueryResponse, Error> {
// 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<chrono::DateTime<Utc>> {
match ts_value {
// Handle the case where ts_value is a number (microseconds)
src/service/search/cache/multi.rs
Vendored
+9 -2
View File
@@ -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
src/service/search/mod.rs
+3 -24
View File
@@ -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) {
src/service/search/partition.rs
+17 -18
View File
@@ -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();
}
src/service/search/sql/mod.rs
+4 -1
View File
@@ -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
src/service/search/sql/visitor/histogram_interval.rs
+3
View File
@@ -20,6 +20,7 @@ use sqlparser::ast::{Expr, FunctionArguments, VisitorMut};
#[derive(Debug)]
pub struct HistogramIntervalVisitor {
pub is_histogram: bool,
pub interval: Option<i64>,
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
src/service/search/streaming/mod.rs
+17 -10
View File
@@ -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