diff --git a/src/service/promql/aggregations/avg.rs b/src/service/promql/aggregations/avg.rs index 783b1befc5..94a0322616 100644 --- a/src/service/promql/aggregations/avg.rs +++ b/src/service/promql/aggregations/avg.rs @@ -22,15 +22,10 @@ use crate::service::promql::{ value::{EvalContext, Sample, Value}, }; -/// Aggregates Matrix input for range queries pub fn avg(param: &Option, data: Value, eval_ctx: &EvalContext) -> Result { let start = std::time::Instant::now(); - let (input_size, timestamps_count) = match &data { - Value::Matrix(m) => (m.len(), eval_ctx.timestamps().len()), - _ => (0, 0), - }; log::info!( - "[trace_id: {}] [PromQL Timing] avg() started with {input_size} series and {timestamps_count} timestamps", + "[trace_id: {}] [PromQL Timing] avg() started", eval_ctx.trace_id, ); diff --git a/src/service/promql/aggregations/count.rs b/src/service/promql/aggregations/count.rs index c02bc2ec13..d364a81ff7 100644 --- a/src/service/promql/aggregations/count.rs +++ b/src/service/promql/aggregations/count.rs @@ -22,15 +22,10 @@ use crate::service::promql::{ value::{EvalContext, Sample, Value}, }; -/// Aggregates Matrix input for range queries pub fn count(param: &Option, data: Value, eval_ctx: &EvalContext) -> Result { let start = std::time::Instant::now(); - let (input_size, timestamps_count) = match &data { - Value::Matrix(m) => (m.len(), eval_ctx.timestamps().len()), - _ => (0, 0), - }; log::info!( - "[trace_id: {}] [PromQL Timing] count() started with {input_size} series and {timestamps_count} timestamps", + "[trace_id: {}] [PromQL Timing] count() started", eval_ctx.trace_id, ); diff --git a/src/service/promql/aggregations/group.rs b/src/service/promql/aggregations/group.rs index d3dc79c2e5..421bd398b1 100644 --- a/src/service/promql/aggregations/group.rs +++ b/src/service/promql/aggregations/group.rs @@ -22,16 +22,11 @@ use crate::service::promql::{ value::{EvalContext, Sample, Value}, }; -/// Aggregates Matrix input for range queries /// https://prometheus.io/docs/prometheus/latest/querying/operators/#aggregation-operators pub fn group(param: &Option, data: Value, eval_ctx: &EvalContext) -> Result { let start = std::time::Instant::now(); - let (input_size, timestamps_count) = match &data { - Value::Matrix(m) => (m.len(), eval_ctx.timestamps().len()), - _ => (0, 0), - }; log::info!( - "[trace_id: {}] [PromQL Timing] group() started with {input_size} series and {timestamps_count} timestamps", + "[trace_id: {}] [PromQL Timing] group() started", eval_ctx.trace_id, ); diff --git a/src/service/promql/aggregations/max.rs b/src/service/promql/aggregations/max.rs index 5e17957d8e..2dd4057332 100644 --- a/src/service/promql/aggregations/max.rs +++ b/src/service/promql/aggregations/max.rs @@ -22,15 +22,10 @@ use crate::service::promql::{ value::{EvalContext, Sample, Value}, }; -/// Aggregates Matrix input for range queries pub fn max(param: &Option, data: Value, eval_ctx: &EvalContext) -> Result { let start = std::time::Instant::now(); - let (input_size, timestamps_count) = match &data { - Value::Matrix(m) => (m.len(), eval_ctx.timestamps().len()), - _ => (0, 0), - }; log::info!( - "[trace_id: {}] [PromQL Timing] max() started with {input_size} series and {timestamps_count} timestamps", + "[trace_id: {}] [PromQL Timing] max() started", eval_ctx.trace_id, ); diff --git a/src/service/promql/aggregations/min.rs b/src/service/promql/aggregations/min.rs index be79bf205b..7def22f5c0 100644 --- a/src/service/promql/aggregations/min.rs +++ b/src/service/promql/aggregations/min.rs @@ -25,12 +25,8 @@ use crate::service::promql::{ /// Aggregates Matrix input for range queries pub fn min(param: &Option, data: Value, eval_ctx: &EvalContext) -> Result { let start = std::time::Instant::now(); - let (input_size, timestamps_count) = match &data { - Value::Matrix(m) => (m.len(), eval_ctx.timestamps().len()), - _ => (0, 0), - }; log::info!( - "[trace_id: {}] [PromQL Timing] min() started with {input_size} series and {timestamps_count} timestamps", + "[trace_id: {}] [PromQL Timing] min() started", eval_ctx.trace_id ); diff --git a/src/service/promql/aggregations/mod.rs b/src/service/promql/aggregations/mod.rs index 48b2e59281..3487a8c4c2 100644 --- a/src/service/promql/aggregations/mod.rs +++ b/src/service/promql/aggregations/mod.rs @@ -177,16 +177,19 @@ pub(crate) fn eval_aggregate( where F: AggFunc, { - let start = std::time::Instant::now(); let func_name = func.name(); - log::info!( - "[trace_id: {}] [PromQL Timing] eval_aggregate({func_name}) started", - eval_ctx.trace_id, - ); + let trace_id = &eval_ctx.trace_id; + log::info!("[trace_id: {trace_id}] [PromQL Timing] eval_aggregate({func_name}) started"); // Handle Matrix input for range queries let matrix = match data { - Value::Matrix(m) => m, + Value::Matrix(m) => { + log::info!( + "[trace_id: {trace_id}] [PromQL Timing] eval_aggregate({func_name}) started with {} series", + m.len() + ); + m + } Value::None => return Ok(Value::None), _ => { return Err(DataFusionError::Plan(format!( @@ -202,33 +205,14 @@ where // Use the eval timestamps from the context to ensure consistent alignment let eval_timestamps = eval_ctx.timestamps(); let eval_timestamps: hashbrown::HashSet = eval_timestamps.iter().cloned().collect(); - - if eval_timestamps.is_empty() { - return Ok(Value::None); - } - log::info!( - "[trace_id: {}] [PromQL Timing] eval_aggregate({func_name}) processing {} time points", - eval_ctx.trace_id, + "[trace_id: {trace_id}] [PromQL Timing] eval_aggregate({func_name}) processing {} time points", eval_timestamps.len() ); - // For each eval timestamp, aggregate across all series - // Parallelize processing using query_thread_num - let cfg = config::get_config(); - let thread_num = cfg.limit.query_thread_num; - let chunk_size = (eval_timestamps.len() / thread_num).max(1); - log::info!( - "[trace_id: {}] [PromQL Timing] eval_aggregate({func_name}) using {} threads with chunk_size {}", - eval_ctx.trace_id, - thread_num, - chunk_size - ); - - let start1 = std::time::Instant::now(); - // Step 1: Compute label hash for each series once based on param // This avoids recomputing the hash for every timestamp + let start1 = std::time::Instant::now(); let series_label_hashes: Vec<(u64, Labels)> = matrix .iter() .map(|rv| { @@ -247,34 +231,30 @@ where .collect(); log::info!( - "[trace_id: {}] [PromQL Timing] eval_aggregate({func_name}) computed label hashes in {:?}", - eval_ctx.trace_id, + "[trace_id: {trace_id}] [PromQL Timing] eval_aggregate({func_name}) computed label hashes in {:?}", start1.elapsed() ); - let start2 = std::time::Instant::now(); - // Step 2: Group series indices by their label hash // Build index: label_hash -> Vec + let start2 = std::time::Instant::now(); let mut groups: HashMap> = HashMap::new(); for (series_idx, (hash, _)) in series_label_hashes.iter().enumerate() { groups.entry(*hash).or_default().push(series_idx); } log::info!( - "[trace_id: {}] [PromQL Timing] eval_aggregate({func_name}) built {} groups in {:?}", - eval_ctx.trace_id, + "[trace_id: {trace_id}] [PromQL Timing] eval_aggregate({func_name}) built {} groups in {:?}", groups.len(), start2.elapsed() ); - let start3 = std::time::Instant::now(); - // Step 3: Process each group in parallel // For each group, aggregate all samples across timestamps - let results: Vec<(u64, Labels, Vec)> = groups + let start3 = std::time::Instant::now(); + let results: Vec = groups .par_iter() - .map(|(hash, series_indices)| { + .map(|(_, series_indices)| { // Get the labels for this group (from the first series in the group) let labels = series_label_hashes[series_indices[0]].1.clone(); @@ -295,54 +275,22 @@ where // Sort by timestamp to maintain order samples.sort_by_key(|s| s.timestamp); - (*hash, labels, samples) + RangeValue { + labels, + samples, + exemplars: None, + time_window: None, + } }) .collect(); log::info!( - "[trace_id: {}] [PromQL Timing] eval_aggregate({func_name}) parallel aggregation took: {:?}", - eval_ctx.trace_id, - start3.elapsed() + "[trace_id: {trace_id}] [PromQL Timing] eval_aggregate({func_name}) parallel aggregation took: {:?}, produced {} series", + start3.elapsed(), + results.len() ); - let start4 = std::time::Instant::now(); - - // Step 4: Convert results to final format - let mut result_by_labels: HashMap)> = - HashMap::with_capacity(results.len()); - for (hash, labels, samples) in results { - if !samples.is_empty() { - result_by_labels.insert(hash, (labels, samples)); - } - } - - log::info!( - "[trace_id: {}] [PromQL Timing] eval_aggregate({func_name}) final conversion took: {:?}", - eval_ctx.trace_id, - start4.elapsed() - ); - - if result_by_labels.is_empty() { - return Ok(Value::None); - } - - let result_matrix: Vec = result_by_labels - .into_values() - .map(|(labels, samples)| RangeValue { - labels, - samples, - exemplars: None, - time_window: None, - }) - .collect(); - - log::info!( - "[trace_id: {}] [PromQL Timing] eval_aggregate({func_name}) completed in {:?}, produced {} series", - eval_ctx.trace_id, - start.elapsed(), - result_matrix.len() - ); - Ok(Value::Matrix(result_matrix)) + Ok(Value::Matrix(results)) } #[cfg(test)] diff --git a/src/service/promql/aggregations/quantile.rs b/src/service/promql/aggregations/quantile.rs index 5fef701512..b7eb77105d 100644 --- a/src/service/promql/aggregations/quantile.rs +++ b/src/service/promql/aggregations/quantile.rs @@ -22,16 +22,11 @@ use crate::service::promql::{ value::{EvalContext, RangeValue, Sample, Value}, }; -/// Aggregates Matrix input for range queries /// Note: quantile aggregates all series into a single result (no label grouping) pub fn quantile(qtile: f64, data: Value, eval_ctx: &EvalContext) -> Result { let start = std::time::Instant::now(); - let (input_size, timestamps_count) = match &data { - Value::Matrix(m) => (m.len(), eval_ctx.timestamps().len()), - _ => (0, 0), - }; log::info!( - "[trace_id: {}] [PromQL Timing] quantile_range({qtile}) started with {input_size} series and {timestamps_count} timestamps", + "[trace_id: {}] [PromQL Timing] quantile_range({qtile}) started", eval_ctx.trace_id, ); diff --git a/src/service/promql/aggregations/stddev.rs b/src/service/promql/aggregations/stddev.rs index 792df88e41..ff10413a2c 100644 --- a/src/service/promql/aggregations/stddev.rs +++ b/src/service/promql/aggregations/stddev.rs @@ -23,15 +23,10 @@ use crate::service::promql::{ value::{EvalContext, Sample, Value}, }; -/// Aggregates Matrix input for range queries pub fn stddev(param: &Option, data: Value, eval_ctx: &EvalContext) -> Result { let start = std::time::Instant::now(); - let (input_size, timestamps_count) = match &data { - Value::Matrix(m) => (m.len(), eval_ctx.timestamps().len()), - _ => (0, 0), - }; log::info!( - "[trace_id: {}] [PromQL Timing] stddev() started with {input_size} series and {timestamps_count} timestamps", + "[trace_id: {}] [PromQL Timing] stddev() started", eval_ctx.trace_id, ); diff --git a/src/service/promql/aggregations/stdvar.rs b/src/service/promql/aggregations/stdvar.rs index 7c7e5417a0..d6232a88df 100644 --- a/src/service/promql/aggregations/stdvar.rs +++ b/src/service/promql/aggregations/stdvar.rs @@ -23,15 +23,10 @@ use crate::service::promql::{ value::{EvalContext, Sample, Value}, }; -/// Aggregates Matrix input for range queries pub fn stdvar(param: &Option, data: Value, eval_ctx: &EvalContext) -> Result { let start = std::time::Instant::now(); - let (input_size, timestamps_count) = match &data { - Value::Matrix(m) => (m.len(), eval_ctx.timestamps().len()), - _ => (0, 0), - }; log::info!( - "[trace_id: {}] [PromQL Timing] stdvar() started with {input_size} series and {timestamps_count} timestamps", + "[trace_id: {}] [PromQL Timing] stdvar() started", eval_ctx.trace_id, ); diff --git a/src/service/promql/aggregations/sum.rs b/src/service/promql/aggregations/sum.rs index 83fdbe31c6..67cece0704 100644 --- a/src/service/promql/aggregations/sum.rs +++ b/src/service/promql/aggregations/sum.rs @@ -25,12 +25,8 @@ use crate::service::promql::{ /// Aggregates Matrix input for range queries pub fn sum(param: &Option, data: Value, eval_ctx: &EvalContext) -> Result { let start = std::time::Instant::now(); - let (input_size, timestamps_count) = match &data { - Value::Matrix(m) => (m.len(), eval_ctx.timestamps().len()), - _ => (0, 0), - }; log::info!( - "[trace_id: {}] [PromQL Timing] sum() started with {input_size} series and {timestamps_count} timestamps", + "[trace_id: {}] [PromQL Timing] sum() started", eval_ctx.trace_id ); diff --git a/src/service/promql/engine.rs b/src/service/promql/engine.rs index 13e6701760..a2554bba6f 100644 --- a/src/service/promql/engine.rs +++ b/src/service/promql/engine.rs @@ -31,7 +31,6 @@ use datafusion::{ datatypes::{DataType, Schema}, }, error::{DataFusionError, Result}, - functions_aggregate::min_max::max, physical_plan::{ Partitioning, execute_stream_partitioned, expressions::Column, repartition::RepartitionExec, }, @@ -56,6 +55,9 @@ use super::{ use crate::service::promql::{ aggregations, binaries, functions, micros, rewrite::remove_filter_all, value::*, }; +type TokioResult = tokio::task::JoinHandle>, HashSet)>>; +type TokioExemplarsResult = + tokio::task::JoinHandle>>, HashSet)>>; pub struct Engine { trace_id: String, @@ -537,10 +539,12 @@ impl Engine { return Ok(Value::None); } - // cache data + let start = std::time::Instant::now(); let mut metric_values = metrics.into_values().collect::>(); metric_values.par_iter_mut().for_each(|metric| { - metric.samples.sort_by(|a, b| a.timestamp.cmp(&b.timestamp)); + metric + .samples + .sort_unstable_by(|a, b| a.timestamp.cmp(&b.timestamp)); if self.ctx.query_exemplars && metric.exemplars.is_some() { metric .exemplars @@ -554,6 +558,11 @@ impl Engine { } else { Value::Matrix(metric_values) }; + log::info!( + "[trace_id: {}] sort samples by timestamps took: {:?}", + self.trace_id, + start.elapsed() + ); Ok(values) } @@ -1117,6 +1126,7 @@ async fn selector_load_data_from_datafusion( label_selector: &Option>, query_exemplars: bool, ) -> Result> { + let start_time = std::time::Instant::now(); let table_name = selector.name.as_ref().unwrap(); let mut df_group = match ctx.table(table_name).await { Ok(v) => v.filter( @@ -1163,78 +1173,40 @@ async fn selector_load_data_from_datafusion( let label_cols = label_cols.into_iter().map(col).collect::>(); // get hash & timestamp - let start_time = std::time::Instant::now(); - let sub_batch = df_group - .clone() - .aggregate( - vec![col(HASH_LABEL)], - vec![max(col(TIMESTAMP_COL_NAME)).alias(TIMESTAMP_COL_NAME)], - )? - .collect() - .await?; - + let start1 = std::time::Instant::now(); let hash_field_type = schema.field_with_name(HASH_LABEL).unwrap().data_type(); - let (timestamp_values, hash_value_set): (HashSet, HashSet) = - if hash_field_type == &DataType::UInt64 { - sub_batch - .iter() - .flat_map(|batch| { - let ts = batch - .column_by_name(TIMESTAMP_COL_NAME) - .unwrap() - .as_any() - .downcast_ref::() - .unwrap(); - let hash = batch - .column_by_name(HASH_LABEL) - .unwrap() - .as_any() - .downcast_ref::() - .unwrap(); - ts.iter() - .zip(hash.iter()) - .map(|(t, h)| (t.unwrap_or_default(), h.unwrap_or_default())) - }) - .unzip() - } else { - sub_batch - .iter() - .flat_map(|batch| { - let ts = batch - .column_by_name(TIMESTAMP_COL_NAME) - .unwrap() - .as_any() - .downcast_ref::() - .unwrap(); - let hash = batch - .column_by_name(HASH_LABEL) - .unwrap() - .as_any() - .downcast_ref::() - .unwrap(); - ts.iter() - .zip(hash.iter()) - .map(|(t, h)| (t.unwrap_or_default(), gxhash::new().sum64(h.unwrap_or("")))) - }) - .unzip() - }; - let timestamp_values = timestamp_values.into_iter().map(lit).collect::>(); + let (mut metrics, timestamp_set) = if query_exemplars { + load_exemplars_from_datafusion(hash_field_type, df_group.clone()).await? + } else { + load_samples_from_datafusion(hash_field_type, df_group.clone()).await? + }; log::info!( - "[trace_id: {trace_id}] load hashing took: {:?}", - start_time.elapsed() + "[trace_id: {trace_id}] load hashing and sample took: {:?}, metrics count: {}, timestamp count: {}", + start1.elapsed(), + metrics.len(), + timestamp_set.len() ); // get series + let start2 = std::time::Instant::now(); let series = df_group .clone() - .filter(col(TIMESTAMP_COL_NAME).in_list(timestamp_values, false))? + .filter(col(TIMESTAMP_COL_NAME).in_list( + timestamp_set.iter().map(|&v| lit(v)).collect::>(), + false, + ))? .select(label_cols)? .collect() .await?; - let mut metrics: HashMap = HashMap::with_capacity(hash_value_set.len()); + log::info!( + "[trace_id: {trace_id}] load all labels took: {:?}", + start2.elapsed() + ); + let mut labels = Vec::new(); + let mut hash_label_set: HashSet = HashSet::with_capacity(metrics.len()); for batch in series { let columns = batch.columns(); let schema = batch.schema(); @@ -1261,10 +1233,7 @@ async fn selector_load_data_from_datafusion( .unwrap(); for i in 0..batch.num_rows() { let hash = hash_values.value(i); - if !hash_value_set.contains(&hash) { - continue; - } - if metrics.contains_key(&hash) { + if hash_label_set.contains(&hash) { continue; } labels.clear(); // reset and reuse the same vector @@ -1277,7 +1246,10 @@ async fn selector_load_data_from_datafusion( value: value.value(i).to_string(), })); } - metrics.insert(hash, RangeValue::new(labels.clone(), Vec::new())); + hash_label_set.insert(hash); + if let Some(range_val) = metrics.get_mut(&hash) { + range_val.labels = labels.clone(); + } } } else { let hash_values = batch @@ -1288,10 +1260,7 @@ async fn selector_load_data_from_datafusion( .unwrap(); for i in 0..batch.num_rows() { let hash: u64 = gxhash::new().sum64(hash_values.value(i)); - if !hash_value_set.contains(&hash) { - continue; - } - if metrics.contains_key(&hash) { + if hash_label_set.contains(&hash) { continue; } labels.clear(); // reset and reuse the same vector @@ -1304,25 +1273,16 @@ async fn selector_load_data_from_datafusion( value: value.value(i).to_string(), })); } - metrics.insert(hash, RangeValue::new(labels.clone(), Vec::new())); + hash_label_set.insert(hash); + if let Some(range_val) = metrics.get_mut(&hash) { + range_val.labels = labels.clone(); + } } } } log::info!( - "[trace_id: {trace_id}] load hashing and series took: {:?}", - start_time.elapsed() - ); - - // get values - if query_exemplars { - load_exemplars_from_datafusion(trace_id, hash_field_type, &mut metrics, df_group).await?; - } else { - load_samples_from_datafusion(trace_id, hash_field_type, &mut metrics, df_group).await?; - } - - log::info!( - "[trace_id: {trace_id}] load data took: {:?}", + "[trace_id: {trace_id}] load data from datafusion took: {:?}", start_time.elapsed() ); @@ -1330,12 +1290,9 @@ async fn selector_load_data_from_datafusion( } async fn load_samples_from_datafusion( - trace_id: &str, hash_field_type: &DataType, - metrics: &mut HashMap, df: DataFrame, -) -> Result<()> { - let start_time = std::time::Instant::now(); +) -> Result<(HashMap, HashSet)> { let ctx = Arc::new(df.task_ctx()); let target_partitions = ctx.session_config().target_partitions(); let plan = df @@ -1355,96 +1312,98 @@ async fn load_samples_from_datafusion( let mut tasks = Vec::new(); for mut stream in streams { let hash_field_type = hash_field_type.clone(); - let mut series: HashMap> = - HashMap::with_capacity(metrics.len() * 2 / target_partitions); - let task: tokio::task::JoinHandle>>> = - tokio::task::spawn(async move { - loop { - match stream.try_next().await { - Ok(Some(batch)) => { - let time_values = batch - .column_by_name(TIMESTAMP_COL_NAME) + let mut series: HashMap> = HashMap::new(); + let task: TokioResult = tokio::task::spawn(async move { + loop { + match stream.try_next().await { + Ok(Some(batch)) => { + let time_values = batch + .column_by_name(TIMESTAMP_COL_NAME) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let value_values = batch + .column_by_name(VALUE_LABEL) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + + if hash_field_type == DataType::UInt64 { + let hash_values = batch + .column_by_name(HASH_LABEL) .unwrap() .as_any() - .downcast_ref::() + .downcast_ref::() .unwrap(); - let value_values = batch - .column_by_name(VALUE_LABEL) + for i in 0..batch.num_rows() { + let timestamp = time_values.value(i); + let hash: u64 = hash_values.value(i); + let entry = series.entry(hash).or_default(); + entry.push(Sample::new(timestamp, value_values.value(i))); + } + } else { + let hash_values = batch + .column_by_name(HASH_LABEL) .unwrap() .as_any() - .downcast_ref::() + .downcast_ref::() .unwrap(); - if hash_field_type == DataType::UInt64 { - let hash_values = batch - .column_by_name(HASH_LABEL) - .unwrap() - .as_any() - .downcast_ref::() - .unwrap(); - for i in 0..batch.num_rows() { - let hash: u64 = hash_values.value(i); - let entry = series.entry(hash).or_default(); - entry.push(Sample::new( - time_values.value(i), - value_values.value(i), - )); - } - } else { - let hash_values = batch - .column_by_name(HASH_LABEL) - .unwrap() - .as_any() - .downcast_ref::() - .unwrap(); - for i in 0..batch.num_rows() { - let hash: u64 = gxhash::new().sum64(hash_values.value(i)); - let entry = series.entry(hash).or_default(); - entry.push(Sample::new( - time_values.value(i), - value_values.value(i), - )); - } + for i in 0..batch.num_rows() { + let timestamp = time_values.value(i); + let hash: u64 = gxhash::new().sum64(hash_values.value(i)); + let entry = series.entry(hash).or_default(); + entry.push(Sample::new(timestamp, value_values.value(i))); } } - Ok(None) => break, - Err(e) => { - log::error!("load samples from datafusion execute stream Error: {e}"); - return Err(e); - } + } + Ok(None) => break, + Err(e) => { + log::error!("load samples from datafusion execute stream Error: {e}"); + return Err(e); } } - Ok(series) - }); + } + let mut unique_timestamps: HashSet = HashSet::new(); + for (_, samples) in &series { + if let Some(min_timestamp) = samples.iter().map(|s| s.timestamp).max() { + unique_timestamps.insert(min_timestamp); + } + } + Ok((series, unique_timestamps)) + }); tasks.push(task); } // collect results + let mut all_unique_timestamps: HashSet = HashSet::new(); + let mut metrics: HashMap = HashMap::new(); for task in tasks { - let m = task + let (m, timestamps) = task .await .map_err(|e| DataFusionError::Execution(e.to_string()))??; + all_unique_timestamps.extend(timestamps); for (hash, samples) in m { - if let Some(range_val) = metrics.get_mut(&hash) { - range_val.samples = samples; - } + metrics.insert( + hash, + RangeValue { + labels: vec![], + samples, + exemplars: None, + time_window: None, + }, + ); } } - log::info!( - "[trace_id: {trace_id}] load samples from datafusion took: {:?}", - start_time.elapsed() - ); - - Ok(()) + Ok((metrics, all_unique_timestamps)) } async fn load_exemplars_from_datafusion( - trace_id: &str, hash_field_type: &DataType, - metrics: &mut HashMap, df: DataFrame, -) -> Result<()> { - let start_time = std::time::Instant::now(); +) -> Result<(HashMap, HashSet)> { let ctx = Arc::new(df.task_ctx()); let target_partitions = ctx.session_config().target_partitions(); let plan = df @@ -1465,98 +1424,100 @@ async fn load_exemplars_from_datafusion( let mut tasks = Vec::new(); for mut stream in streams { let hash_field_type = hash_field_type.clone(); - let mut series: HashMap>> = - HashMap::with_capacity(metrics.len() * 2 / target_partitions); - let task: tokio::task::JoinHandle>>>> = - tokio::task::spawn(async move { - loop { - match stream.try_next().await { - Ok(Some(batch)) => { - let exemplars_values = batch - .column_by_name(EXEMPLARS_LABEL) + let mut series: HashMap>> = HashMap::new(); + let task: TokioExemplarsResult = tokio::task::spawn(async move { + loop { + match stream.try_next().await { + Ok(Some(batch)) => { + let exemplars_values = batch + .column_by_name(EXEMPLARS_LABEL) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + if hash_field_type == DataType::UInt64 { + let hash_values = batch + .column_by_name(HASH_LABEL) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + for i in 0..batch.num_rows() { + let hash: u64 = hash_values.value(i); + let exemplar = exemplars_values.value(i); + if let Ok(exemplars) = json::from_str::>(exemplar) + { + let entry = series.entry(hash).or_default(); + for exemplar in exemplars { + if let Some(exemplar) = exemplar.as_object() { + entry.push(Arc::new(Exemplar::from(exemplar))); + } + } + } + } + } else { + let hash_values = batch + .column_by_name(HASH_LABEL) .unwrap() .as_any() .downcast_ref::() .unwrap(); - if hash_field_type == DataType::UInt64 { - let hash_values = batch - .column_by_name(HASH_LABEL) - .unwrap() - .as_any() - .downcast_ref::() - .unwrap(); - for i in 0..batch.num_rows() { - let hash: u64 = hash_values.value(i); - let exemplar = exemplars_values.value(i); - if let Ok(exemplars) = - json::from_str::>(exemplar) - { - let entry = series.entry(hash).or_default(); - for exemplar in exemplars { - if let Some(exemplar) = exemplar.as_object() { - entry.push(Arc::new(Exemplar::from(exemplar))); - } - } - } - } - } else { - let hash_values = batch - .column_by_name(HASH_LABEL) - .unwrap() - .as_any() - .downcast_ref::() - .unwrap(); - for i in 0..batch.num_rows() { - let hash: u64 = gxhash::new().sum64(hash_values.value(i)); - let exemplar = exemplars_values.value(i); - if let Ok(exemplars) = - json::from_str::>(exemplar) - { - let entry = series.entry(hash).or_default(); - for exemplar in exemplars { - if let Some(exemplar) = exemplar.as_object() { - entry.push(Arc::new(Exemplar::from(exemplar))); - } + for i in 0..batch.num_rows() { + let hash: u64 = gxhash::new().sum64(hash_values.value(i)); + let exemplar = exemplars_values.value(i); + if let Ok(exemplars) = json::from_str::>(exemplar) + { + let entry = series.entry(hash).or_default(); + for exemplar in exemplars { + if let Some(exemplar) = exemplar.as_object() { + entry.push(Arc::new(Exemplar::from(exemplar))); } } } } } - Ok(None) => break, - Err(e) => { - log::error!("load exemplars from datafusion execute stream Error: {e}"); - return Err(e); - } + } + Ok(None) => break, + Err(e) => { + log::error!("load exemplars from datafusion execute stream Error: {e}"); + return Err(e); } } - Ok(series) - }); + } + let mut unique_timestamps: HashSet = HashSet::new(); + for (_, samples) in &series { + if let Some(min_timestamp) = samples.iter().map(|s| s.timestamp).max() { + unique_timestamps.insert(min_timestamp); + } + } + Ok((series, unique_timestamps)) + }); tasks.push(task); } // collect results + let mut all_unique_timestamps: HashSet = HashSet::new(); + let mut metrics: HashMap = HashMap::new(); for task in tasks { - let m = task + let (m, timestamps) = task .await .map_err(|e| DataFusionError::Execution(e.to_string()))??; + all_unique_timestamps.extend(timestamps); for (hash, exemplars) in m { - if let Some(range_val) = metrics.get_mut(&hash) { - if range_val.exemplars.is_none() { - range_val.exemplars = Some(vec![]); - } - range_val.exemplars = Some(exemplars); - } + metrics.insert( + hash, + RangeValue { + labels: vec![], + samples: vec![], + exemplars: Some(exemplars), + time_window: None, + }, + ); } } - log::info!( - "[trace_id: {trace_id}] load exemplars from datafusion took: {:?}", - start_time.elapsed() - ); - - Ok(()) + Ok((metrics, all_unique_timestamps)) } - #[cfg(test)] mod tests { use std::sync::Arc; diff --git a/src/service/promql/functions/absent_over_time.rs b/src/service/promql/functions/absent_over_time.rs index 1fd24c3a02..5179530039 100644 --- a/src/service/promql/functions/absent_over_time.rs +++ b/src/service/promql/functions/absent_over_time.rs @@ -23,20 +23,8 @@ use crate::service::promql::{ }; /// https://prometheus.io/docs/prometheus/latest/querying/functions/#absent_over_time -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn absent_over_time(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] absent_over_time() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, AbsentOverTimeFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] absent_over_time() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, AbsentOverTimeFunc::new(), eval_ctx) } pub struct AbsentOverTimeFunc; diff --git a/src/service/promql/functions/avg_over_time.rs b/src/service/promql/functions/avg_over_time.rs index f174ca7c36..699782f203 100644 --- a/src/service/promql/functions/avg_over_time.rs +++ b/src/service/promql/functions/avg_over_time.rs @@ -22,20 +22,8 @@ use crate::service::promql::{ value::{EvalContext, Sample, Value}, }; -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn avg_over_time(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] avg_over_time() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, AvgOverTimeFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] avg_over_time() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, AvgOverTimeFunc::new(), eval_ctx) } pub struct AvgOverTimeFunc; diff --git a/src/service/promql/functions/changes.rs b/src/service/promql/functions/changes.rs index 2c2317fbb2..ebff1c48f4 100644 --- a/src/service/promql/functions/changes.rs +++ b/src/service/promql/functions/changes.rs @@ -23,20 +23,8 @@ use crate::service::promql::{ }; /// https://prometheus.io/docs/prometheus/latest/querying/functions/#changes -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn changes(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] changes() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, ChangesFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] changes() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, ChangesFunc::new(), eval_ctx) } pub struct ChangesFunc; diff --git a/src/service/promql/functions/count_over_time.rs b/src/service/promql/functions/count_over_time.rs index d5d5e99d3d..492c5f4598 100644 --- a/src/service/promql/functions/count_over_time.rs +++ b/src/service/promql/functions/count_over_time.rs @@ -22,20 +22,8 @@ use crate::service::promql::{ value::{EvalContext, Sample, Value}, }; -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn count_over_time(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] count_over_time() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, CountOverTimeFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] count_over_time() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, CountOverTimeFunc::new(), eval_ctx) } pub struct CountOverTimeFunc; diff --git a/src/service/promql/functions/delta.rs b/src/service/promql/functions/delta.rs index 162ea5000f..0feaf1c4c1 100644 --- a/src/service/promql/functions/delta.rs +++ b/src/service/promql/functions/delta.rs @@ -22,20 +22,8 @@ use crate::service::promql::{ value::{EvalContext, ExtrapolationKind, Sample, Value, extrapolated_rate}, }; -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn delta(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] delta() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, DeltaFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] delta() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, DeltaFunc::new(), eval_ctx) } pub struct DeltaFunc; diff --git a/src/service/promql/functions/deriv.rs b/src/service/promql/functions/deriv.rs index a02f66ae8e..8f3bbf605c 100644 --- a/src/service/promql/functions/deriv.rs +++ b/src/service/promql/functions/deriv.rs @@ -24,20 +24,8 @@ use crate::service::promql::{ }; /// https://prometheus.io/docs/prometheus/latest/querying/functions/#deriv -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn deriv(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] deriv() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, DerivFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] deriv() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, DerivFunc::new(), eval_ctx) } pub struct DerivFunc; diff --git a/src/service/promql/functions/histogram.rs b/src/service/promql/functions/histogram.rs index 86875c2ec3..d9350b499e 100644 --- a/src/service/promql/functions/histogram.rs +++ b/src/service/promql/functions/histogram.rs @@ -39,28 +39,10 @@ impl Bucket { /// Enhanced version that processes all timestamps at once for range queries pub(crate) fn histogram_quantile(phi: f64, data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] histogram_quantile() started with phi={phi}, {} time points", - eval_ctx.trace_id, - eval_ctx.timestamps().len() - ); - // Handle input data - convert to matrix format if needed let in_matrix = match data { - Value::Matrix(m) => { - log::info!( - "[trace_id: {}] [PromQL Timing] histogram_quantile() processing {} series", - eval_ctx.trace_id, - m.len() - ); - m - } + Value::Matrix(m) => m, Value::None => { - log::info!( - "[trace_id: {}] [PromQL Timing] histogram_quantile() received None input", - eval_ctx.trace_id - ); return Ok(Value::None); } _ => { @@ -134,12 +116,6 @@ pub(crate) fn histogram_quantile(phi: f64, data: Value, eval_ctx: &EvalContext) } } - log::info!( - "[trace_id: {}] [PromQL Timing] histogram_quantile() total execution took: {:?}, produced {} series", - eval_ctx.trace_id, - start.elapsed(), - range_values.len() - ); Ok(Value::Matrix(range_values)) } diff --git a/src/service/promql/functions/holt_winters.rs b/src/service/promql/functions/holt_winters.rs index 101fd6047f..666684ccc7 100644 --- a/src/service/promql/functions/holt_winters.rs +++ b/src/service/promql/functions/holt_winters.rs @@ -24,29 +24,17 @@ use crate::service::promql::{ }; /// https://prometheus.io/docs/prometheus/latest/querying/functions/#holt_winters -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn holt_winters( data: Value, scaling_factor: f64, trend_factor: f64, eval_ctx: &EvalContext, ) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] holt_winters() started", - eval_ctx.trace_id - ); - let result = super::eval_range( + super::eval_range( data, HoltWintersFunc::new(scaling_factor, trend_factor), eval_ctx, - ); - log::info!( - "[trace_id: {}] [PromQL Timing] holt_winters() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + ) } pub struct HoltWintersFunc { diff --git a/src/service/promql/functions/idelta.rs b/src/service/promql/functions/idelta.rs index f95c37a53b..0df8acb725 100644 --- a/src/service/promql/functions/idelta.rs +++ b/src/service/promql/functions/idelta.rs @@ -22,20 +22,8 @@ use crate::service::promql::{ value::{EvalContext, Sample, Value}, }; -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn idelta(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] idelta() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, IdeltaFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] idelta() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, IdeltaFunc::new(), eval_ctx) } pub struct IdeltaFunc; diff --git a/src/service/promql/functions/increase.rs b/src/service/promql/functions/increase.rs index 261e6457a1..e2c83bb537 100644 --- a/src/service/promql/functions/increase.rs +++ b/src/service/promql/functions/increase.rs @@ -22,20 +22,8 @@ use crate::service::promql::{ value::{EvalContext, ExtrapolationKind, Sample, Value, extrapolated_rate}, }; -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn increase(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] increase() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, IncreaseFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] increase() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, IncreaseFunc::new(), eval_ctx) } pub struct IncreaseFunc; diff --git a/src/service/promql/functions/irate.rs b/src/service/promql/functions/irate.rs index 0d7d6b3641..3807ac8ccf 100644 --- a/src/service/promql/functions/irate.rs +++ b/src/service/promql/functions/irate.rs @@ -22,20 +22,8 @@ use crate::service::promql::{ value::{EvalContext, Sample, Value}, }; -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn irate(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] irate() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, IrateFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] irate() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, IrateFunc::new(), eval_ctx) } pub struct IrateFunc; diff --git a/src/service/promql/functions/last_over_time.rs b/src/service/promql/functions/last_over_time.rs index cfe4ac94e0..53fbec7e7e 100644 --- a/src/service/promql/functions/last_over_time.rs +++ b/src/service/promql/functions/last_over_time.rs @@ -22,20 +22,8 @@ use crate::service::promql::{ value::{EvalContext, Sample, Value}, }; -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn last_over_time(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] last_over_time() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, LastOverTimeFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] last_over_time() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, LastOverTimeFunc::new(), eval_ctx) } pub struct LastOverTimeFunc; diff --git a/src/service/promql/functions/max_over_time.rs b/src/service/promql/functions/max_over_time.rs index ea1f8b0e7e..550f6adbce 100644 --- a/src/service/promql/functions/max_over_time.rs +++ b/src/service/promql/functions/max_over_time.rs @@ -23,20 +23,8 @@ use crate::service::promql::{ value::{EvalContext, Sample, Value}, }; -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn max_over_time(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] max_over_time() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, MaxOverTimeFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] max_over_time() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, MaxOverTimeFunc::new(), eval_ctx) } pub struct MaxOverTimeFunc; diff --git a/src/service/promql/functions/min_over_time.rs b/src/service/promql/functions/min_over_time.rs index 36c27fa6af..52624e6af1 100644 --- a/src/service/promql/functions/min_over_time.rs +++ b/src/service/promql/functions/min_over_time.rs @@ -23,20 +23,8 @@ use crate::service::promql::{ value::{EvalContext, Sample, Value}, }; -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn min_over_time(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] min_over_time() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, MinOverTimeFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] min_over_time() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, MinOverTimeFunc::new(), eval_ctx) } pub struct MinOverTimeFunc; diff --git a/src/service/promql/functions/mod.rs b/src/service/promql/functions/mod.rs index f81f797745..857db4a145 100644 --- a/src/service/promql/functions/mod.rs +++ b/src/service/promql/functions/mod.rs @@ -211,24 +211,19 @@ pub trait RangeFunc: Sync { fn exec(&self, samples: &[Sample], eval_ts: i64, range: &Duration) -> Option; } -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn eval_range(data: Value, func: F, eval_ctx: &EvalContext) -> Result where F: RangeFunc, { let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] eval_range({}) started", - eval_ctx.trace_id, - func.name() - ); + let trace_id = &eval_ctx.trace_id; + let func_name = func.name(); + log::info!("[trace_id: {trace_id}] [PromQL Timing] eval_range({func_name}) started"); let data = match data { Value::Matrix(v) => { log::info!( - "[trace_id: {}] [PromQL Timing] eval_range({}) processing {} series", - eval_ctx.trace_id, - func.name(), + "[trace_id: {trace_id}] [PromQL Timing] eval_range({func_name}) processing {} series", v.len() ); v @@ -236,8 +231,7 @@ where Value::None => return Ok(Value::None), v => { return Err(DataFusionError::Plan(format!( - "{}: matrix argument expected but got {}", - func.name(), + "{func_name}: matrix argument expected but got {}", v.get_type() ))); } @@ -245,96 +239,64 @@ where // Always use range query path - compute all timestamps at once let timestamps = eval_ctx.timestamps(); - let mut range_values = Vec::with_capacity(data.len()); - log::info!( - "[trace_id: {}] [PromQL Timing] eval_range({}) processing {} time points in range query mode", - eval_ctx.trace_id, - func.name(), + "[trace_id: {trace_id}] [PromQL Timing] eval_range({func_name}) processing {} time points", timestamps.len() ); - let cfg = config::get_config(); - let thread_num = cfg.limit.query_thread_num; - let chunk_size = (data.len() / thread_num).max(1); - log::info!( - "[trace_id: {}] [PromQL Timing] eval_range({}) using {} threads with chunk_size {}", - eval_ctx.trace_id, - func.name(), - thread_num, - chunk_size - ); - - let parallel_start = std::time::Instant::now(); - let results: Vec> = data + let results: Vec = data .into_par_iter() - .chunks(chunk_size) - .flat_map(|chunk| { - chunk - .into_iter() - .map(|mut metric| { - let mut labels = std::mem::take(&mut metric.labels); - if !KEEP_METRIC_NAME_FUNC.contains(func.name()) { - labels = labels.without_metric_name(); - } - let time_window = metric.time_window.as_ref().unwrap(); - let range = time_window.range; - let range_micros = micros(range); - let mut result_samples = Vec::with_capacity(timestamps.len()); + .flat_map(|mut metric| { + let mut labels = std::mem::take(&mut metric.labels); + if !KEEP_METRIC_NAME_FUNC.contains(func.name()) { + labels = labels.without_metric_name(); + } + let time_window = metric.time_window.as_ref().unwrap(); + let range = time_window.range; + let range_micros = micros(range); + let mut result_samples = Vec::with_capacity(timestamps.len()); - // For each eval timestamp, compute the function value - for &eval_ts in ×tamps { - // Find samples in the window [eval_ts - range, eval_ts] - let window_start = eval_ts - range_micros; - let window_end = eval_ts; + // For each eval timestamp, compute the function value + for &eval_ts in ×tamps { + // Find samples in the window [eval_ts - range, eval_ts] + let window_start = eval_ts - range_micros; + let window_end = eval_ts; - // Extract samples within this window using binary search - let start_index = metric - .samples - .partition_point(|s| s.timestamp < window_start); - let end_index = metric - .samples - .partition_point(|s| s.timestamp <= window_end); - let window_samples = &metric.samples[start_index..end_index]; + // Extract samples within this window using binary search + let start_index = metric + .samples + .partition_point(|s| s.timestamp < window_start); + let end_index = metric + .samples + .partition_point(|s| s.timestamp <= window_end); + let window_samples = &metric.samples[start_index..end_index]; - if window_samples.is_empty() { - continue; - } + if window_samples.is_empty() { + continue; + } - if let Some(value) = func.exec(window_samples, eval_ts, &range) { - result_samples.push(Sample::new(eval_ts, value)); - } - } + if let Some(value) = func.exec(window_samples, eval_ts, &range) { + result_samples.push(Sample::new(eval_ts, value)); + } + } - if !result_samples.is_empty() { - Some(RangeValue { - labels, - samples: result_samples, - exemplars: None, - time_window: metric.time_window, - }) - } else { - None - } + if !result_samples.is_empty() { + Some(RangeValue { + labels, + samples: result_samples, + exemplars: None, + time_window: metric.time_window, }) - .collect::>() + } else { + None + } }) .collect(); - log::info!( - "[trace_id: {}] [PromQL Timing] eval_range({}) parallel processing took: {:?}", - eval_ctx.trace_id, - func.name(), - parallel_start.elapsed() - ); - - range_values.extend(results.into_iter().flatten()); log::info!( - "[trace_id: {}] [PromQL Timing] eval_range({}) completed in {:?}, produced {} series", - eval_ctx.trace_id, - func.name(), + "[trace_id: {trace_id}] [PromQL Timing] eval_range({func_name}) completed in {:?}, produced {} series", start.elapsed(), - range_values.len() + results.len() ); - Ok(Value::Matrix(range_values)) + Ok(Value::Matrix(results)) } diff --git a/src/service/promql/functions/predict_linear.rs b/src/service/promql/functions/predict_linear.rs index 62d58c0f05..d66006a51a 100644 --- a/src/service/promql/functions/predict_linear.rs +++ b/src/service/promql/functions/predict_linear.rs @@ -24,20 +24,8 @@ use crate::service::promql::{ }; /// https://prometheus.io/docs/prometheus/latest/querying/functions/#predict_linear -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn predict_linear(data: Value, duration: f64, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] predict_linear() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, PredictLinearFunc::new(duration), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] predict_linear() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, PredictLinearFunc::new(duration), eval_ctx) } pub struct PredictLinearFunc { diff --git a/src/service/promql/functions/quantile_over_time.rs b/src/service/promql/functions/quantile_over_time.rs index 202458d3be..748aac8c6d 100644 --- a/src/service/promql/functions/quantile_over_time.rs +++ b/src/service/promql/functions/quantile_over_time.rs @@ -24,24 +24,12 @@ use crate::service::promql::{ }; /// https://prometheus.io/docs/prometheus/latest/querying/functions/#quantile_over_time -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn quantile_over_time( phi_quantile: f64, data: Value, eval_ctx: &EvalContext, ) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] quantile_over_time() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, QuantileOverTimeFunc::new(phi_quantile), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] quantile_over_time() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, QuantileOverTimeFunc::new(phi_quantile), eval_ctx) } pub struct QuantileOverTimeFunc { diff --git a/src/service/promql/functions/rate.rs b/src/service/promql/functions/rate.rs index dcad5f8f53..f9259c2ac6 100644 --- a/src/service/promql/functions/rate.rs +++ b/src/service/promql/functions/rate.rs @@ -22,20 +22,8 @@ use crate::service::promql::{ value::{EvalContext, ExtrapolationKind, Sample, Value, extrapolated_rate}, }; -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn rate(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] rate() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, RateFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] rate() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, RateFunc::new(), eval_ctx) } pub struct RateFunc; diff --git a/src/service/promql/functions/resets.rs b/src/service/promql/functions/resets.rs index c53ec598ce..e384da54d2 100644 --- a/src/service/promql/functions/resets.rs +++ b/src/service/promql/functions/resets.rs @@ -23,20 +23,8 @@ use crate::service::promql::{ }; /// https://prometheus.io/docs/prometheus/latest/querying/functions/#resets -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn resets(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] resets() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, ResetsFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] resets() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, ResetsFunc::new(), eval_ctx) } pub struct ResetsFunc; diff --git a/src/service/promql/functions/stddev_over_time.rs b/src/service/promql/functions/stddev_over_time.rs index a59edb8881..f03685c542 100644 --- a/src/service/promql/functions/stddev_over_time.rs +++ b/src/service/promql/functions/stddev_over_time.rs @@ -24,20 +24,8 @@ use crate::service::promql::{ }; /// https://prometheus.io/docs/prometheus/latest/querying/functions/#stddev_over_time -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn stddev_over_time(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] stddev_over_time() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, StddevOverTimeFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] stddev_over_time() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, StddevOverTimeFunc::new(), eval_ctx) } pub struct StddevOverTimeFunc; diff --git a/src/service/promql/functions/stdvar_over_time.rs b/src/service/promql/functions/stdvar_over_time.rs index 1793be5108..acd19cccd3 100644 --- a/src/service/promql/functions/stdvar_over_time.rs +++ b/src/service/promql/functions/stdvar_over_time.rs @@ -24,20 +24,8 @@ use crate::service::promql::{ }; /// https://prometheus.io/docs/prometheus/latest/querying/functions/#stdvar_over_time -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn stdvar_over_time(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] stdvar_over_time() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, StdvarOverTimeFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] stdvar_over_time() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, StdvarOverTimeFunc::new(), eval_ctx) } pub struct StdvarOverTimeFunc; diff --git a/src/service/promql/functions/sum_over_time.rs b/src/service/promql/functions/sum_over_time.rs index a6c915f108..6d23d4cc48 100644 --- a/src/service/promql/functions/sum_over_time.rs +++ b/src/service/promql/functions/sum_over_time.rs @@ -22,20 +22,8 @@ use crate::service::promql::{ value::{EvalContext, Sample, Value}, }; -/// Enhanced version that processes all timestamps at once for range queries pub(crate) fn sum_over_time(data: Value, eval_ctx: &EvalContext) -> Result { - let start = std::time::Instant::now(); - log::info!( - "[trace_id: {}] [PromQL Timing] sum_over_time() started", - eval_ctx.trace_id - ); - let result = super::eval_range(data, SumOverTimeFunc::new(), eval_ctx); - log::info!( - "[trace_id: {}] [PromQL Timing] sum_over_time() execution took: {:?}", - eval_ctx.trace_id, - start.elapsed() - ); - result + super::eval_range(data, SumOverTimeFunc::new(), eval_ctx) } pub struct SumOverTimeFunc;