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

perf: improve data load and code clean up (#9150)

Browse Source
This commit is contained in:
Huaijin
2025-11-17 21:47:59 +08:00
committed by GitHub
parent 6ca371ec99
commit dc2cdcbf0b
33 changed files with 289 additions and 725 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/service/promql/aggregations/avg.rs
+1 -6
View File
@@ -22,15 +22,10 @@ use crate::service::promql::{
value::{EvalContext, Sample, Value},
};
/// Aggregates Matrix input for range queries
pub fn avg(param: &Option<LabelModifier>, data: Value, eval_ctx: &EvalContext) -> Result<Value> {
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,
);
src/service/promql/aggregations/count.rs
+1 -6
View File
@@ -22,15 +22,10 @@ use crate::service::promql::{
value::{EvalContext, Sample, Value},
};
/// Aggregates Matrix input for range queries
pub fn count(param: &Option<LabelModifier>, data: Value, eval_ctx: &EvalContext) -> Result<Value> {
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,
);
src/service/promql/aggregations/group.rs
+1 -6
View File
@@ -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<LabelModifier>, data: Value, eval_ctx: &EvalContext) -> Result<Value> {
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,
);
src/service/promql/aggregations/max.rs
+1 -6
View File
@@ -22,15 +22,10 @@ use crate::service::promql::{
value::{EvalContext, Sample, Value},
};
/// Aggregates Matrix input for range queries
pub fn max(param: &Option<LabelModifier>, data: Value, eval_ctx: &EvalContext) -> Result<Value> {
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,
);
src/service/promql/aggregations/min.rs
+1 -5
View File
@@ -25,12 +25,8 @@ use crate::service::promql::{
/// Aggregates Matrix input for range queries
pub fn min(param: &Option<LabelModifier>, data: Value, eval_ctx: &EvalContext) -> Result<Value> {
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
);
src/service/promql/aggregations/mod.rs
+27 -79
View File
@@ -177,16 +177,19 @@ pub(crate) fn eval_aggregate<F>(
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<i64> = 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<series_idx>
let start2 = std::time::Instant::now();
let mut groups: HashMap<u64, Vec<usize>> = 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<Sample>)> = groups
let start3 = std::time::Instant::now();
let results: Vec<RangeValue> = 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<u64, (Labels, Vec<Sample>)> =
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<RangeValue> = 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)]
src/service/promql/aggregations/quantile.rs
+1 -6
View File
@@ -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<Value> {
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,
);
src/service/promql/aggregations/stddev.rs
+1 -6
View File
@@ -23,15 +23,10 @@ use crate::service::promql::{
value::{EvalContext, Sample, Value},
};
/// Aggregates Matrix input for range queries
pub fn stddev(param: &Option<LabelModifier>, data: Value, eval_ctx: &EvalContext) -> Result<Value> {
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,
);
src/service/promql/aggregations/stdvar.rs
+1 -6
View File
@@ -23,15 +23,10 @@ use crate::service::promql::{
value::{EvalContext, Sample, Value},
};
/// Aggregates Matrix input for range queries
pub fn stdvar(param: &Option<LabelModifier>, data: Value, eval_ctx: &EvalContext) -> Result<Value> {
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,
);
src/service/promql/aggregations/sum.rs
+1 -5
View File
@@ -25,12 +25,8 @@ use crate::service::promql::{
/// Aggregates Matrix input for range queries
pub fn sum(param: &Option<LabelModifier>, data: Value, eval_ctx: &EvalContext) -> Result<Value> {
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
);
src/service/promql/engine.rs
+183 -222
View File
@@ -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<Result<(HashMap<u64, Vec<Sample>>, HashSet<i64>)>>;
type TokioExemplarsResult =
tokio::task::JoinHandle<Result<(HashMap<u64, Vec<Arc<Exemplar>>>, HashSet<i64>)>>;
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::<Vec<_>>();
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<HashSet<String>>,
query_exemplars: bool,
) -> Result<HashMap<u64, RangeValue>> {
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::<Vec<_>>();
// 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<i64>, HashSet<u64>) =
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::<Int64Array>()
.unwrap();
let hash = batch
.column_by_name(HASH_LABEL)
.unwrap()
.as_any()
.downcast_ref::<UInt64Array>()
.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::<Int64Array>()
.unwrap();
let hash = batch
.column_by_name(HASH_LABEL)
.unwrap()
.as_any()
.downcast_ref::<StringArray>()
.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::<Vec<_>>();
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::<Vec<_>>(),
false,
))?
.select(label_cols)?
.collect()
.await?;
let mut metrics: HashMap<u64, RangeValue> = 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<u64> = 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<u64, RangeValue>,
df: DataFrame,
) -> Result<()> {
let start_time = std::time::Instant::now();
) -> Result<(HashMap<u64, RangeValue>, HashSet<i64>)> {
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<u64, Vec<Sample>> =
HashMap::with_capacity(metrics.len() * 2 / target_partitions);
let task: tokio::task::JoinHandle<Result<HashMap<u64, Vec<Sample>>>> =
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<u64, Vec<Sample>> = 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::<Int64Array>()
.unwrap();
let value_values = batch
.column_by_name(VALUE_LABEL)
.unwrap()
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
if hash_field_type == DataType::UInt64 {
let hash_values = batch
.column_by_name(HASH_LABEL)
.unwrap()
.as_any()
.downcast_ref::<Int64Array>()
.downcast_ref::<UInt64Array>()
.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::<Float64Array>()
.downcast_ref::<StringArray>()
.unwrap();
if hash_field_type == DataType::UInt64 {
let hash_values = batch
.column_by_name(HASH_LABEL)
.unwrap()
.as_any()
.downcast_ref::<UInt64Array>()
.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::<StringArray>()
.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<i64> = 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<i64> = HashSet::new();
let mut metrics: HashMap<u64, RangeValue> = 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<u64, RangeValue>,
df: DataFrame,
) -> Result<()> {
let start_time = std::time::Instant::now();
) -> Result<(HashMap<u64, RangeValue>, HashSet<i64>)> {
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<u64, Vec<Arc<Exemplar>>> =
HashMap::with_capacity(metrics.len() * 2 / target_partitions);
let task: tokio::task::JoinHandle<Result<HashMap<u64, Vec<Arc<Exemplar>>>>> =
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<u64, Vec<Arc<Exemplar>>> = 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::<StringArray>()
.unwrap();
if hash_field_type == DataType::UInt64 {
let hash_values = batch
.column_by_name(HASH_LABEL)
.unwrap()
.as_any()
.downcast_ref::<UInt64Array>()
.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::<Vec<json::Value>>(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::<StringArray>()
.unwrap();
if hash_field_type == DataType::UInt64 {
let hash_values = batch
.column_by_name(HASH_LABEL)
.unwrap()
.as_any()
.downcast_ref::<UInt64Array>()
.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::<Vec<json::Value>>(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::<StringArray>()
.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::<Vec<json::Value>>(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::<Vec<json::Value>>(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<i64> = 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<i64> = HashSet::new();
let mut metrics: HashMap<u64, RangeValue> = 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;
src/service/promql/functions/absent_over_time.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/avg_over_time.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/changes.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/count_over_time.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/delta.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/deriv.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/histogram.rs
+1 -25
View File
@@ -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<Value> {
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))
}
src/service/promql/functions/holt_winters.rs
+2 -14
View File
@@ -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<Value> {
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 {
src/service/promql/functions/idelta.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/increase.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/irate.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/last_over_time.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/max_over_time.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/min_over_time.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/mod.rs
+48 -86
View File
@@ -211,24 +211,19 @@ pub trait RangeFunc: Sync {
fn exec(&self, samples: &[Sample], eval_ts: i64, range: &Duration) -> Option<f64>;
}
/// Enhanced version that processes all timestamps at once for range queries
pub(crate) fn eval_range<F>(data: Value, func: F, eval_ctx: &EvalContext) -> Result<Value>
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<Option<RangeValue>> = data
let results: Vec<RangeValue> = 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 &timestamps {
// 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 &timestamps {
// 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::<Vec<_>>()
} 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))
}
src/service/promql/functions/predict_linear.rs
+1 -13
View File
@@ -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<Value> {
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 {
src/service/promql/functions/quantile_over_time.rs
+1 -13
View File
@@ -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<Value> {
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 {
src/service/promql/functions/rate.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/resets.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/stddev_over_time.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/stdvar_over_time.rs
+1 -13
View File
@@ -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<Value> {
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;
src/service/promql/functions/sum_over_time.rs
+1 -13
View File
@@ -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<Value> {
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;