diff --git a/src/config/src/meta/search.rs b/src/config/src/meta/search.rs index 6a44c67558..0bba4e0bce 100644 --- a/src/config/src/meta/search.rs +++ b/src/config/src/meta/search.rs @@ -578,6 +578,8 @@ pub struct SearchPartitionRequest { pub histogram_interval: i64, #[serde(default)] pub sampling_ratio: Option, + #[serde(default)] + pub search_type: Option, } impl SearchPartitionRequest { @@ -612,6 +614,7 @@ impl From<&Request> for SearchPartitionRequest { streaming_output: req.query.streaming_output, histogram_interval: req.query.histogram_interval, sampling_ratio: req.query.sampling_ratio, + search_type: req.search_type, } } } @@ -2206,6 +2209,7 @@ mod tests { streaming_output: false, histogram_interval: 0, sampling_ratio: None, + search_type: None, }; req.decode().unwrap(); diff --git a/src/handler/grpc/request/search/mod.rs b/src/handler/grpc/request/search/mod.rs index b6ab1c588d..4fce3daed8 100644 --- a/src/handler/grpc/request/search/mod.rs +++ b/src/handler/grpc/request/search/mod.rs @@ -242,7 +242,6 @@ impl Search for Searcher { stream_type, &request, req.skip_max_query_range, - true, true, // allow streamings aggs cache for grpc search partition ) .await; diff --git a/src/handler/http/request/search/mod.rs b/src/handler/http/request/search/mod.rs index 5e5e5c5d29..d5d320f98b 100644 --- a/src/handler/http/request/search/mod.rs +++ b/src/handler/http/request/search/mod.rs @@ -1565,7 +1565,6 @@ pub async fn search_partition( stream_type, &req, false, - true, true, // allow streamings aggs cache for http search partition handler ) .instrument(http_span) diff --git a/src/handler/http/request/traces/mod.rs b/src/handler/http/request/traces/mod.rs index 0e77b2e1d8..f1ab2935db 100644 --- a/src/handler/http/request/traces/mod.rs +++ b/src/handler/http/request/traces/mod.rs @@ -1366,6 +1366,7 @@ async fn process_latest_traces_stream( streaming_output: false, histogram_interval: 0, sampling_ratio: None, + search_type: Some(config::meta::search::SearchEventType::UI), }; let partitions = match SearchService::search_partition( @@ -1375,7 +1376,6 @@ async fn process_latest_traces_stream( stream_type, &partition_req, false, - false, use_cache, ) .await diff --git a/src/service/search/mod.rs b/src/service/search/mod.rs index d28a9dae12..000ab1e6a3 100644 --- a/src/service/search/mod.rs +++ b/src/service/search/mod.rs @@ -23,6 +23,7 @@ use config::{ cluster::LOCAL_NODE, get_config, ider, meta::{ + cluster::RoleGroup, function::RESULT_ARRAY, search::{self}, self_reporting::usage::{RequestStats, UsageType}, @@ -600,12 +601,13 @@ pub async fn search_partition( stream_type: StreamType, req: &search::SearchPartitionRequest, skip_max_query_range: bool, - is_http_req: bool, use_cache: bool, ) -> Result { let cfg = get_config(); - let ctx = PartitionSqlContext::new(req, org_id, stream_type, is_http_req).await?; + let role_group = req.search_type.map(RoleGroup::from); + + let ctx = PartitionSqlContext::new(req, org_id, stream_type).await?; let stream_files = collect_stream_files(trace_id, user_id, &ctx).await?; @@ -627,7 +629,7 @@ pub async fn search_partition( non_ts_order_by_cols: vec![], }; - let total_secs = estimated_secs(trace_id, &ctx, is_http_req, resp.original_size).await?; + let total_secs = estimated_secs(trace_id, &ctx, role_group, resp.original_size).await?; if ctx.use_single_partition || (ctx.is_streaming_aggregate && total_secs <= cfg.limit.aggs_min_num_partition_secs) @@ -1022,9 +1024,9 @@ pub async fn search_partition_multi( streaming_output: req.streaming_output, histogram_interval: req.histogram_interval, sampling_ratio: None, + search_type: None, }, false, - true, false, // disable aggs cache ) .await diff --git a/src/service/search/partition/cpu_cores.rs b/src/service/search/partition/cpu_cores.rs index ffb3e1b2f5..daa96ce64a 100644 --- a/src/service/search/partition/cpu_cores.rs +++ b/src/service/search/partition/cpu_cores.rs @@ -28,14 +28,9 @@ use super::sql_context::PartitionSqlContext; pub(crate) async fn estimated_secs( trace_id: &str, _ctx: &PartitionSqlContext, - is_http_req: bool, + role_group: Option, original_size: usize, ) -> Result { - let role_group = if is_http_req { - Some(RoleGroup::Interactive) - } else { - Some(RoleGroup::Background) - }; let nodes = get_cached_online_querier_nodes(role_group) .await .unwrap_or_default(); diff --git a/src/service/search/partition/sql_context.rs b/src/service/search/partition/sql_context.rs index fb0231e297..7e96a86890 100644 --- a/src/service/search/partition/sql_context.rs +++ b/src/service/search/partition/sql_context.rs @@ -20,9 +20,7 @@ use config::{ }, utils::{ base64, - sql::{ - is_complex_query, is_eligible_for_histogram, is_explain_query, is_simple_distinct_query, - }, + sql::{is_complex_query, is_eligible_for_histogram, is_explain_query}, }, }; use infra::errors::Error; @@ -51,7 +49,6 @@ impl PartitionSqlContext { req: &SearchPartitionRequest, org_id: &str, stream_type: StreamType, - is_http_req: bool, ) -> Result { let query = cluster_rpc::SearchQuery { start_time: req.start_time, @@ -64,15 +61,13 @@ impl PartitionSqlContext { let is_explain = is_explain_query(&req.sql); let is_complex = is_complex_query(&req.sql).unwrap_or(false); - let is_http_distinct = is_simple_distinct_query(&req.sql).unwrap_or(false) && is_http_req; let ts_column = get_ts_col_order_by(&sql, TIMESTAMP_COL_NAME, is_complex).map(|(v, _)| v); let is_streaming_agg = is_streaming_aggregate(&req.sql, ts_column.as_deref()); let apply_over_hits = req.query_fn.as_ref().is_some_and(|v| { !v.is_empty() && RESULT_ARRAY.is_match(&base64::decode_url(v).unwrap_or(v.to_string())) }); - let use_single_partition = is_http_distinct - || is_explain + let use_single_partition = is_explain || ((ts_column.is_none() || apply_over_hits) && !(req.streaming_output && is_streaming_agg)); diff --git a/src/service/search/streaming/execution.rs b/src/service/search/streaming/execution.rs index 5ef8cb9e60..7ab7743fe6 100644 --- a/src/service/search/streaming/execution.rs +++ b/src/service/search/streaming/execution.rs @@ -463,6 +463,7 @@ pub async fn get_partitions( streaming_output: true, histogram_interval: req.query.histogram_interval, sampling_ratio: req.query.sampling_ratio, + search_type: req.search_type, }; let res = SearchService::search_partition( @@ -472,7 +473,6 @@ pub async fn get_partitions( stream_type, &search_partition_req, false, - false, req.use_cache, ) .instrument(tracing::info_span!(