rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-14 14:56:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(promql): resolve info signatures per series and cover the review gaps

Browse Source 
Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
This commit is contained in:
Dennis Zhuang
2026-10-21 17:46:56 +08:00
parent d9a1b14314
commit f2673ce10b
3 changed files with 308 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/promql/src/info.rs
+134 -71
View File
@@ -56,8 +56,9 @@ pub struct InfoSelector {
///
/// `info` holds the info series the selector chooses, one float field each, whose value at a
/// point is the original timestamp of the sample selected there. For each info metric a base
/// series takes the info series with the same identifying labels; when several are present at a
/// point the one with the newest timestamp counts, and two equally new ones are an error. Labels
/// series takes the info series with the same identifying labels. Several of them at one point are
/// compared in input order, as Prometheus compares them in the order its storage returns them: a
/// newer one replaces the one kept so far, and one exactly as new is an error. Labels
/// the base series already has are kept, and two info metrics giving one label different values
/// are an error. A base series without identifying labels matches no info series, since
/// Prometheus fetches none for it, and a base series that is itself an info metric is passed
@@ -105,9 +106,7 @@ pub fn info(
}
}
// Errors as `(point, rank, message)`: Prometheus fails at the first failing step, and within
// a step checks the info series, then each base series, then the result label sets.
let mut errors: Vec<(usize, u8, String)> = Vec::new();
let mut error = FirstError::default();
// Prometheus fetches info series per presence pattern of the identifying labels among the
// base series it enriches: a present label matches any value the pattern's base series have,
@@ -131,7 +130,7 @@ pub fn info(
}
}
// The newest info series of each signature at each point.
// The info series kept for each signature at each point.
let mut by_signature: HashMap<Signature, BTreeMap<usize, (f64, usize)>> = HashMap::new();
let mut info_names: BTreeSet<Arc<str>> = BTreeSet::new();
for (index, series) in info.iter().enumerate() {
@@ -147,21 +146,19 @@ pub fn info(
}
let name: Arc<str> = Arc::from(series.metric.name().unwrap_or(""));
info_names.insert(name.clone());
let newest = by_signature
.entry(signature(name, &series.metric))
.or_default();
let kept = by_signature.entry((name, values)).or_default();
let mut cursor = series.fields.as_slice()[0].cursor();
for (point, row) in series.time.positions() {
let timestamp = match cursor.at(row).expect("the axis and the fields are aligned") {
SampleRef::Missing => continue,
SampleRef::Float(timestamp) => timestamp,
SampleRef::Histogram(_) => {
return Err(DataFusionError::Execution(
"info sample should be float".into(),
return Err(DataFusionError::Internal(
"info series must hold the original sample timestamps as floats".into(),
));
}
};
match newest.entry(point) {
match kept.entry(point) {
Entry::Vacant(entry) => {
entry.insert((timestamp, index));
}
@@ -170,18 +167,16 @@ pub fn info(
if timestamp > existing {
entry.insert((timestamp, index));
} else if timestamp == existing && base_present[point] {
errors.push((
point,
0,
error.record(point, 0, || {
format!(
"found duplicate series for info metric: existing {:?} @ {}, \
new {:?} @ {}",
label_set(&info[existing_index].metric),
"found duplicate series for info metric: existing {} @ {}, new \
{} @ {}",
render(&label_set(&info[existing_index].metric)),
existing as i64,
label_set(&series.metric),
render(&label_set(&series.metric)),
timestamp as i64
),
));
)
});
}
}
}
@@ -189,10 +184,25 @@ pub fn info(
}
let width = schema.fields.len();
let mut results: HashMap<LabelSet, (MetricName, Rows)> = HashMap::new();
let mut result_index: HashMap<LabelSet, usize> = HashMap::new();
let mut results: Vec<(MetricName, Vec<Row>)> = Vec::new();
let mut added: Vec<(Arc<str>, Arc<str>)> = Vec::new();
let mut chosen: Vec<usize> = Vec::new();
for series in base {
let ignored = is_info_metric(series.metric.name().unwrap_or(""));
// The signature of each info metric is fixed for a series, so its kept series are looked
// up once; a point then only reads which of them is kept there.
let candidates: Vec<&BTreeMap<usize, (f64, usize)>> = if ignored {
Vec::new()
} else {
let values = identifying_values(&series.metric);
info_names
.iter()
.filter_map(|name| by_signature.get(&(name.clone(), values.clone())))
.collect()
};
// The result of the previous point, reused while the same info series are kept.
let mut previous: Option<Choice> = None;
let mut cursors: Vec<_> = series.fields.iter().map(|field| field.cursor()).collect();
for (point, row) in series.time.positions() {
let mut values = Vec::with_capacity(width);
@@ -204,72 +214,92 @@ pub fn info(
if values.iter().all(Option::is_none) {
continue;
}
let metric = if ignored {
series.metric.clone()
} else {
chosen.clear();
chosen.extend(
candidates
.iter()
.filter_map(|kept| kept.get(&point).map(|&(_, index)| index)),
);
let reuse = previous
.as_ref()
.is_some_and(|previous| previous.info == chosen);
if !reuse {
added.clear();
let mut seen = 0;
for name in &info_names {
let Some(&(_, index)) = by_signature
.get(&signature(name.clone(), &series.metric))
.and_then(|newest| newest.get(&point))
else {
continue;
};
seen += 1;
let mut conflict = None;
for &index in &chosen {
if let Err(label) =
add_data_labels(&info[index].metric, &series.metric, selector, &mut added)
{
errors.push((point, 1, format!("conflicting label: {label}")));
conflict = Some(label);
break;
}
}
if seen == 0 && !selector.keep_unmatched {
continue;
}
let mut metric = series.metric.clone();
for (name, value) in &added {
metric.set(name, value.clone()).map_err(external)?;
}
metric
};
let (_, rows) = results
.entry(label_set(&metric))
.or_insert_with(|| (metric.clone(), BTreeMap::new()));
match rows.entry(point) {
Entry::Vacant(entry) => {
entry.insert(values);
}
Entry::Occupied(mut entry) => {
for (slot, value) in entry.get_mut().iter_mut().zip(values) {
let result = if chosen.is_empty() && !selector.keep_unmatched && !ignored {
None
} else {
let mut metric = series.metric.clone();
for (name, value) in &added {
metric.set(name, value.clone()).map_err(external)?;
}
Some(match result_index.entry(label_set(&metric)) {
std::collections::hash_map::Entry::Occupied(entry) => *entry.get(),
std::collections::hash_map::Entry::Vacant(entry) => {
results.push((metric, Vec::new()));
*entry.insert(results.len() - 1)
}
})
};
previous = Some(Choice {
info: chosen.clone(),
result,
conflict,
});
}
let choice = previous.as_ref().expect("set above");
if let Some(label) = &choice.conflict {
error.record(point, 1, || format!("conflicting label: {label}"));
}
if let Some(result) = choice.result {
results[result].1.push((point, values));
}
}
}
// Rows of one label set arrive per base series; sorted by point, a field held twice at one
// point is a collision.
for (metric, rows) in &mut results {
rows.sort_by_key(|(point, _)| *point);
let mut merged: Vec<Row> = Vec::with_capacity(rows.len());
for (point, values) in rows.drain(..) {
match merged.last_mut() {
Some((last, slots)) if *last == point => {
for (slot, value) in slots.iter_mut().zip(values) {
if value.is_none() {
continue;
}
if slot.is_some() {
errors.push((
point,
2,
error.record(point, 2, || {
format!(
"vector cannot contain metrics with the same labelset {:?}",
label_set(&metric)
),
));
"vector cannot contain metrics with the same labelset {}",
render(&label_set(metric))
)
});
}
*slot = value;
}
}
_ => merged.push((point, values)),
}
}
*rows = merged;
}
if let Some((_, _, message)) = errors
.into_iter()
.min_by_key(|(point, rank, _)| (*point, *rank))
{
if let Some((_, _, message)) = error.0 {
return Err(DataFusionError::Execution(message));
}
let mut builder = GridSeriesBuilder::try_new(schema.clone())?;
let mut output = Vec::with_capacity(results.len());
for (metric, rows) in results.into_values() {
for (metric, rows) in results {
builder.begin_series();
for (point, row) in rows {
builder.push_samples(point, &row)?;
@@ -279,16 +309,49 @@ pub fn info(
Ok(output)
}
/// The samples of one result series by grid point, one per field.
type Rows = BTreeMap<usize, Vec<Option<GridSample>>>;
/// The samples of one result series at one grid point, one per field.
type Row = (usize, Vec<Option<GridSample>>);
/// What the info series kept at a point make of one base series.
struct Choice {
/// The info series, one per info metric that has one.
info: Vec<usize>,
/// The result series the point goes to, `None` when the point is dropped.
result: Option<usize>,
/// A label two of the info series set to different values.
conflict: Option<Arc<str>>,
}
/// The error Prometheus reports: the one of the first failing step, and within a step the first
/// of its checks (the info series, then each base series, then the result label sets).
#[derive(Default)]
struct FirstError(Option<(usize, u8, String)>);
impl FirstError {
/// Records an error of check `rank` at `point`, formatting it only if it comes first.
fn record(&mut self, point: usize, rank: u8, message: impl FnOnce() -> String) {
if self
.0
.as_ref()
.is_none_or(|(first, first_rank, _)| (point, rank) < (*first, *first_rank))
{
self.0 = Some((point, rank, message()));
}
}
}
/// A label set as Prometheus prints one.
fn render(labels: &LabelSet) -> String {
let labels: Vec<String> = labels
.iter()
.map(|(name, value)| format!("{name}=\"{value}\""))
.collect();
format!("{{{}}}", labels.join(", "))
}
/// An info metric name and the identifying label values present on a series.
type Signature = (Arc<str>, Vec<Option<Arc<str>>>);
fn signature(name: Arc<str>, metric: &MetricName) -> Signature {
(name, identifying_values(metric))
}
/// The value of each identifying label, `None` where the series does not have it.
fn identifying_values(metric: &MetricName) -> Vec<Option<Arc<str>>> {
IDENTIFYING_LABELS
src/promql/src/info/prometheus_cases.txt
+47 -2
View File
@@ -1,7 +1,8 @@
# Generated by promql-compat cmd/narthex-golden-info built for amd64 (Prometheus v3.15.0 engine,
# seed 1, lookback 25000ms). `case <data labels> <keep unmatched> <name matchers>`, `base` the
# evaluated base selector, `info` the original sample timestamps of the info series Prometheus
# fetches, then the result series or the error. Points are `grid index:hex value bits`.
# evaluated base expression, `info` the original sample timestamps of the info series matching the
# name and data matchers (a superset of what Prometheus fetches: the identifying-label filter is
# the kernel's), then the result series or the error. Points are `grid index:hex value bits`.
case - true -
expr info({__name__="a"})
grid 100000 10000 2
@@ -4835,3 +4836,47 @@ info __name__=target_info,job=x,k8s_cluster=c1 ; 3:40ffbd0000000000
info __name__=target_info,job=x,k8s_cluster=c2 ; 3:40ffbd0000000000
result __name__=a,job=x ; 0:3ff0000000000000 1:3ff0000000000000 2:3ff0000000000000
end
case - true -
expr info({__name__="a"})
grid 100000 10000 2
base __name__=a,job=x ; 1:4000000000000000
info __name__=target_info,job=x,k8s_cluster=a ; 0:40f86a0000000000 1:40f86a0000000000
info __name__=target_info,job=x,k8s_cluster=b ; 0:40f86a0000000000 1:40f86a0000000000
info __name__=target_info,job=x,k8s_cluster=c ; 1:40fadb0000000000
error found duplicate series for info metric: existing {__name__="target_info", job="x", k8s_cluster="a"} @ 100000, new {__name__="target_info", job="x", k8s_cluster="b"} @ 100000
end
case - true -
expr info({__name__="a"})
grid 100000 10000 2
base __name__=a,job=x ; 1:4000000000000000
info __name__=target_info,job=x,k8s_cluster=c ; 1:40fadb0000000000
info __name__=target_info,job=x,k8s_cluster=a ; 0:40f86a0000000000 1:40f86a0000000000
info __name__=target_info,job=x,k8s_cluster=b ; 0:40f86a0000000000 1:40f86a0000000000
result __name__=a,job=x,k8s_cluster=c ; 1:4000000000000000
end
case - true __name__=~"target_info|build_info"
expr info({__name__="a"}, {__name__=~"target_info|build_info"})
grid 100000 10000 4
base __name__=a,job=x ; 1:4000000000000000 2:4000000000000000 3:4010000000000000
base __name__=a,job=x,k8s_cluster=c1 ; 1:4000000000000000 2:4000000000000000 3:4000000000000000
info __name__=target_info,job=x,k8s_cluster=c1 ; 1:40fadb0000000000 2:40fadb0000000000 3:40fadb0000000000
info __name__=build_info,job=x,k8s_cluster=c2 ; 3:40ffbd0000000000
error vector cannot contain metrics with the same labelset
end
case - true __name__!~"build_.*"
expr info(a * 2, {__name__!~"build_.*"})
grid 100000 10000 3
base instance=i1,job=x ; 0:4000000000000000 1:4010000000000000 2:4018000000000000
base job=y ; 0:4000000000000000 1:4000000000000000 2:4018000000000000
info __name__=target_info,instance=i1,job=x,k8s_cluster=c1 ; 0:40f86a0000000000 1:40f86a0000000000 2:40f86a0000000000
result instance=i1,job=x,k8s_cluster=c1 ; 0:4000000000000000 1:4010000000000000 2:4018000000000000
result job=y ; 0:4000000000000000 1:4000000000000000 2:4018000000000000
end
case - true __name__=~"build_.*"
expr info(a * 2, {__name__=~"build_.*"})
grid 100000 10000 3
base job=y ; 0:4000000000000000 1:4000000000000000 2:4018000000000000
info __name__=build_info,job=y,version=v1 ; 2:40fd4c0000000000
result job=y ; 0:4000000000000000 1:4000000000000000
result job=y,version=v1 ; 2:4018000000000000
end
src/promql/src/info/tests.rs
+127 -2
View File
@@ -15,10 +15,14 @@
use std::collections::BTreeMap;
use std::sync::Arc;
use datatypes::series::samples::{FieldSpec, SeriesSchema};
use datatypes::series::samples::{FieldSpec, SampleRef, Series, SeriesSchema};
use datatypes::series::time::{EvalGrid, GridSpec};
use crate::grid_series::reference::{Labels, Points, labels_of, points_of, sparse_series};
use crate::grid_series::reference::{
Labels, OwnedHistogram, Points, labels_of, metric_of, points_of, sparse_series,
};
use crate::grid_series::{GridSample, GridSeriesBuilder};
use crate::histogram_eval::float_histogram::FloatHistogram;
use crate::info::{InfoSelector, info};
use crate::select::MatchOp;
@@ -140,3 +144,124 @@ fn info_matches_prometheus() {
"{matched} matched, {failed} failed"
);
}
fn grid_schema(fields: usize) -> Arc<SeriesSchema> {
Arc::new(SeriesSchema {
fields: (0..fields)
.map(|field| FieldSpec {
name: Arc::from(format!("f{field}")),
})
.collect::<Vec<_>>()
.into(),
eval_grid: Some(Arc::new(
EvalGrid::try_new(GridSpec::try_new(0, 10_000, 1).unwrap()).unwrap(),
)),
})
}
fn one_point(
schema: &Arc<SeriesSchema>,
labels: &[(&str, &str)],
row: Vec<Option<GridSample>>,
) -> Series {
let labels: Labels = labels
.iter()
.map(|(name, value)| (name.to_string(), value.to_string()))
.collect();
let mut builder = GridSeriesBuilder::try_new(schema.clone()).unwrap();
builder.push_samples(0, &row).unwrap();
builder.finish(metric_of(&labels)).unwrap().unwrap()
}
fn target_info() -> Series {
one_point(
&grid_schema(1),
&[
("__name__", "target_info"),
("job", "x"),
("k8s_cluster", "c1"),
],
vec![Some(GridSample::Float(0.0))],
)
}
fn all_labels() -> InfoSelector {
InfoSelector {
name_matchers: Vec::new(),
data_labels: Vec::new(),
keep_unmatched: true,
}
}
/// Each field is a series of its own: two base series that end up with one label set merge where
/// they hold different fields, and collide only where both hold the same field.
#[test]
fn results_with_one_label_set_merge_by_field() {
let schema = grid_schema(2);
let float = |value| Some(GridSample::Float(value));
let enriched = one_point(
&schema,
&[("__name__", "a"), ("job", "x")],
vec![float(1.0), None],
);
let labelled = |row| {
one_point(
&schema,
&[("__name__", "a"), ("job", "x"), ("k8s_cluster", "c1")],
row,
)
};
let merged = info(
&schema,
&[enriched.clone(), labelled(vec![None, float(2.0)])],
&[target_info()],
&all_labels(),
)
.unwrap();
assert_eq!(merged.len(), 1);
let fields: Vec<_> = merged[0]
.fields
.iter()
.map(|field| match field.sample(0) {
SampleRef::Float(value) => Some(value),
_ => None,
})
.collect();
assert_eq!(fields, [Some(1.0), Some(2.0)]);
let error = info(
&schema,
&[enriched, labelled(vec![float(3.0), None])],
&[target_info()],
&all_labels(),
)
.unwrap_err();
assert!(error.to_string().contains("same labelset"), "{error}");
}
/// A native histogram base sample is enriched like a float one.
#[test]
fn histogram_base_samples_are_enriched() {
let schema = grid_schema(1);
let histogram =
OwnedHistogram::parse("0/0/0/3ff0000000000000/3ff0000000000000/0/1_1/3ff0000000000000///");
let base = one_point(
&schema,
&[("__name__", "a"), ("job", "x")],
vec![Some(GridSample::Histogram(Box::new(
FloatHistogram::try_from_view(&histogram.view()).unwrap(),
)))],
);
let output = info(&schema, &[base], &[target_info()], &all_labels()).unwrap();
assert_eq!(output.len(), 1);
assert_eq!(
labels_of(&output[0].metric)
.get("k8s_cluster")
.map(String::as_str),
Some("c1")
);
assert!(matches!(
output[0].fields.as_slice()[0].sample(0),
SampleRef::Histogram(_)
));
}