diff --git a/src/frontend/src/error.rs b/src/frontend/src/error.rs index c0f9e2fe6b..9fd2312e73 100644 --- a/src/frontend/src/error.rs +++ b/src/frontend/src/error.rs @@ -174,6 +174,15 @@ pub enum Error { location: Location, }, + #[snafu(display("Unexpected type {data_type} for column '{column}' of table '{table_name}'"))] + UnexpectedColumnType { + table_name: String, + column: String, + data_type: String, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Failed to collect recordbatch"))] CollectRecordbatch { #[snafu(implicit)] @@ -414,7 +423,9 @@ impl ErrorExt for Error { Error::RequestQuery { source, .. } => source.status_code(), - Error::CacheRequired { .. } => StatusCode::Internal, + Error::CacheRequired { .. } | Error::UnexpectedColumnType { .. } => { + StatusCode::Internal + } Error::TableNotFound { .. } => StatusCode::TableNotFound, diff --git a/src/frontend/src/instance.rs b/src/frontend/src/instance.rs index 44949f932f..e16955f287 100644 --- a/src/frontend/src/instance.rs +++ b/src/frontend/src/instance.rs @@ -1510,6 +1510,20 @@ impl PrometheusHandler for Instance { .context(ExecuteQuerySnafu) } + async fn query_metric_names_by_labels( + &self, + matchers: Vec, + schema: &str, + start: SystemTime, + end: SystemTime, + ctx: &QueryContextRef, + ) -> server_error::Result> { + self.handle_query_metric_names_by_labels(matchers, schema, start, end, ctx) + .await + .map_err(BoxedError::new) + .context(ExecuteQuerySnafu) + } + async fn query_label_values( &self, metric: String, diff --git a/src/frontend/src/instance/promql.rs b/src/frontend/src/instance/promql.rs index 3df7022470..3ad9ae217a 100644 --- a/src/frontend/src/instance/promql.rs +++ b/src/frontend/src/instance/promql.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::collections::HashSet; use std::sync::Arc; use std::time::SystemTime; @@ -20,20 +21,25 @@ use catalog::information_schema::TABLES; use client::OutputData; use common_catalog::consts::INFORMATION_SCHEMA_NAME; use common_catalog::format_full_table_name; -use common_recordbatch::util; +use common_recordbatch::{RecordBatch, util}; use common_telemetry::tracing; use datafusion_expr::LogicalPlan; +use datatypes::arrow::array::{Array, UInt32Array}; +use futures::StreamExt; use promql_parser::label::{Matcher, Matchers}; use query::promql; use query::promql::planner::PromPlanner; use servers::prometheus; use session::context::QueryContextRef; use snafu::{OptionExt, ResultExt}; +use store_api::metric_engine_consts::DATA_SCHEMA_TABLE_ID_COLUMN_NAME; +use store_api::storage::TableId; +use table::TableRef; use crate::error::{ CatalogSnafu, CollectRecordbatchSnafu, ExecLogicalPlanSnafu, PrometheusLabelValuesQueryPlanSnafu, PrometheusMetricNamesQueryPlanSnafu, ReadTableSnafu, - Result, TableNotFoundSnafu, TableSnafu, + Result, TableNotFoundSnafu, TableSnafu, UnexpectedColumnTypeSnafu, }; use crate::instance::Instance; @@ -117,6 +123,173 @@ impl Instance { Ok(results) } + /// Handles a metric names query constrained by matchers on ordinary labels. + /// + /// [`Instance::handle_query_metric_names`] answers from table metadata, which + /// cannot resolve a label matcher: whether a metric carries `pod="abc"` is a + /// property of its data. The metric engine multiplexes the logical tables of + /// one physical table into a region holding the union of their label columns + /// alongside `__table_id`, so a single distinct scan per physical table + /// resolves the matchers for all of its logical tables at once. + /// + /// Only metric engine tables are covered. Tables on other engines share no + /// column space to scan, and a scan per table does not scale to the table + /// counts this API is expected to answer over. + #[tracing::instrument(skip_all)] + pub(crate) async fn handle_query_metric_names_by_labels( + &self, + matchers: Vec, + schema: &str, + start: SystemTime, + end: SystemTime, + ctx: &QueryContextRef, + ) -> Result> { + let _timer = crate::metrics::PROMQL_QUERY_METRICS_ELAPSED + .with_label_values(&[ctx.get_db_string().as_str()]) + .start_timer(); + + let catalog = ctx.current_catalog(); + let mut table_ids = HashSet::new(); + for physical in self.physical_metric_tables(catalog, schema, ctx).await? { + table_ids.extend( + self.scan_matching_table_ids(physical, &matchers, start, end, ctx) + .await?, + ); + } + + // Batch-resolve only the ids the scan produced. An id dropped between the + // scan and here simply has no entry. + let table_ids = table_ids.into_iter().collect::>(); + let mut names = self + .catalog_manager + .tables_by_ids(catalog, schema, &table_ids) + .await + .context(CatalogSnafu)? + .into_iter() + .map(|table| table.table_info().name.clone()) + .collect::>(); + names.sort_unstable(); + Ok(names) + } + + /// Returns the metric engine physical tables of a schema. + /// + /// Their data regions carry the union of their logical tables' label columns, + /// so scanning these covers every metric of the schema. + async fn physical_metric_tables( + &self, + catalog: &str, + schema: &str, + ctx: &QueryContextRef, + ) -> Result> { + let mut tables = self.catalog_manager.tables(catalog, schema, Some(ctx)); + let mut physical_tables = Vec::new(); + + while let Some(table) = tables.next().await { + let table = table.context(CatalogSnafu)?; + if table.table_info().is_physical_table() { + physical_tables.push(table); + } + } + + Ok(physical_tables) + } + + /// Scans `table` for the distinct values of `column` that match `matchers` + /// within the time range. + /// + /// Callers look the table up and decode the batches; the plan between is the + /// same whether the projected column is a label or `__table_id`. + async fn scan_distinct_column( + &self, + table: TableRef, + matchers: Vec, + column: String, + start: SystemTime, + end: SystemTime, + ctx: &QueryContextRef, + ) -> Result> { + let dataframe = self + .query_engine + .read_table(table.clone()) + .with_context(|_| ReadTableSnafu { + table_name: table.table_info().full_table_name(), + })?; + + let scan_plan = dataframe.into_unoptimized_plan(); + let conditions = PromPlanner::matchers_to_expr(Matchers::new(matchers), scan_plan.schema()) + .context(PrometheusLabelValuesQueryPlanSnafu)?; + let logical_plan = promql::label_values::rewrite_label_values_query( + table, scan_plan, conditions, column, start, end, + ) + .context(PrometheusLabelValuesQueryPlanSnafu)?; + + let results = self + .query_engine + .execute(logical_plan, ctx.clone()) + .await + .context(ExecLogicalPlanSnafu)?; + + match results.data { + OutputData::Stream(stream) => { + util::collect(stream).await.context(CollectRecordbatchSnafu) + } + OutputData::RecordBatches(rbs) => Ok(rbs.take()), + _ => unreachable!("should not happen"), + } + } + + /// Returns the `__table_id`s of `physical` carrying a row that matches every + /// matcher within the time range. + async fn scan_matching_table_ids( + &self, + physical: TableRef, + matchers: &[Matcher], + start: SystemTime, + end: SystemTime, + ctx: &QueryContextRef, + ) -> Result> { + // `__table_id` attributes a row to its logical table, and a physical + // table that never took a column from one does not expose it. Skipping + // such a table can only miss a metric carrying no labels at all. + if physical + .schema() + .column_schema_by_name(DATA_SCHEMA_TABLE_ID_COLUMN_NAME) + .is_none() + { + return Ok(Vec::new()); + } + + let table_name = physical.table_info().full_table_name(); + let batches = self + .scan_distinct_column( + physical, + matchers.to_vec(), + DATA_SCHEMA_TABLE_ID_COLUMN_NAME.to_string(), + start, + end, + ctx, + ) + .await?; + + let mut table_ids = Vec::new(); + for batch in batches { + // Only one column in results, ensured by `rewrite_label_values_query`. + let column = batch.column(0); + let ids = column + .as_any() + .downcast_ref::() + .with_context(|| UnexpectedColumnTypeSnafu { + table_name: table_name.clone(), + column: DATA_SCHEMA_TABLE_ID_COLUMN_NAME, + data_type: column.data_type().to_string(), + })?; + table_ids.extend(ids.iter().flatten()); + } + + Ok(table_ids) + } + /// Handles label values query request, returns the values. #[tracing::instrument(skip_all)] pub(crate) async fn handle_query_label_values( @@ -150,40 +323,9 @@ impl Instance { .context(TableSnafu); } - let dataframe = self - .query_engine - .read_table(table.clone()) - .with_context(|_| ReadTableSnafu { - table_name: full_table_name, - })?; - - let scan_plan = dataframe.into_unoptimized_plan(); - let filter_conditions = - PromPlanner::matchers_to_expr(Matchers::new(matchers), scan_plan.schema()) - .context(PrometheusLabelValuesQueryPlanSnafu)?; - let logical_plan = promql::label_values::rewrite_label_values_query( - table, - scan_plan, - filter_conditions, - label_name, - start, - end, - ) - .context(PrometheusLabelValuesQueryPlanSnafu)?; - - let results = self - .query_engine - .execute(logical_plan, ctx.clone()) - .await - .context(ExecLogicalPlanSnafu)?; - - let batches = match results.data { - OutputData::Stream(stream) => util::collect(stream) - .await - .context(CollectRecordbatchSnafu)?, - OutputData::RecordBatches(rbs) => rbs.take(), - _ => unreachable!("should not happen"), - }; + let batches = self + .scan_distinct_column(table, matchers, label_name, start, end, ctx) + .await?; let mut results = Vec::with_capacity(batches.iter().map(|b| b.num_rows()).sum()); for batch in batches { diff --git a/src/query/src/promql/error.rs b/src/query/src/promql/error.rs index dba55c8df8..45e0834962 100644 --- a/src/query/src/promql/error.rs +++ b/src/query/src/promql/error.rs @@ -13,6 +13,7 @@ // limitations under the License. use std::any::Any; +use std::time::SystemTime; use common_error::ext::ErrorExt; use common_error::status_code::StatusCode; @@ -202,6 +203,13 @@ pub enum Error { location: Location, }, + #[snafu(display("Time out of the representable millisecond range: {:?}", time))] + SystemTimeOutOfRange { + time: SystemTime, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("vector cannot contain metrics with the same labelset"))] SameLabelSet { #[snafu(implicit)] @@ -243,6 +251,7 @@ impl ErrorExt for Error { | UnsupportedMatcherOp { .. } | SameLabelSet { .. } | TimestampOutOfRange { .. } + | SystemTimeOutOfRange { .. } | InvalidRegularExpression { .. } | InvalidDestinationLabelName { .. } => StatusCode::InvalidArguments, diff --git a/src/query/src/promql/label_values.rs b/src/query/src/promql/label_values.rs index 4f6ea42a60..44c173477b 100644 --- a/src/query/src/promql/label_values.rs +++ b/src/query/src/promql/label_values.rs @@ -23,9 +23,31 @@ use snafu::{OptionExt, ResultExt}; use table::TableRef; use crate::promql::error::{ - DataFusionPlanningSnafu, Result, TimeIndexNotFoundSnafu, TimestampOutOfRangeSnafu, + DataFusionPlanningSnafu, Result, SystemTimeOutOfRangeSnafu, TimeIndexNotFoundSnafu, + TimestampOutOfRangeSnafu, }; +/// Converts a [`SystemTime`] to a millisecond [`Timestamp`]. +/// +/// `duration_since` reports an instant before the epoch as an error rather than +/// a negative duration, and an RFC3339 query parameter can name one, so the sign +/// is recovered here instead of unwrapping. +fn millis_since_epoch(time: SystemTime) -> Result { + let (millis, before_epoch) = match time.duration_since(UNIX_EPOCH) { + Ok(duration) => (duration.as_millis(), false), + Err(earlier) => (earlier.duration().as_millis(), true), + }; + let millis = i64::try_from(millis) + .ok() + .with_context(|| SystemTimeOutOfRangeSnafu { time })?; + + Ok(Timestamp::new_millisecond(if before_epoch { + -millis + } else { + millis + })) +} + fn build_time_filter(time_index_expr: Expr, start: Timestamp, end: Timestamp) -> Expr { time_index_expr .clone() @@ -67,14 +89,12 @@ pub fn rewrite_label_values_query( })?; // We only support millisecond precision at most. - let start = - Timestamp::new_millisecond(start.duration_since(UNIX_EPOCH).unwrap().as_millis() as i64); + let start = millis_since_epoch(start)?; let start = start.convert_to(unit).context(TimestampOutOfRangeSnafu { timestamp: start.value(), unit, })?; - let end = - Timestamp::new_millisecond(end.duration_since(UNIX_EPOCH).unwrap().as_millis() as i64); + let end = millis_since_epoch(end)?; let end = end.convert_to(unit).context(TimestampOutOfRangeSnafu { timestamp: end.value(), unit, @@ -98,3 +118,28 @@ pub fn rewrite_label_values_query( Ok(logical_plan) } + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::*; + + #[test] + fn millis_before_the_epoch_are_negative() { + // `SystemTime::duration_since` reports these as an error; unwrapping it + // panicked on any request whose RFC3339 start named a pre-epoch instant. + let time = UNIX_EPOCH - Duration::from_millis(1); + assert_eq!(millis_since_epoch(time).unwrap().value(), -1); + assert_eq!(millis_since_epoch(UNIX_EPOCH).unwrap().value(), 0); + + let time = UNIX_EPOCH + Duration::from_millis(1); + assert_eq!(millis_since_epoch(time).unwrap().value(), 1); + } + + #[test] + fn millis_beyond_i64_are_rejected() { + let time = UNIX_EPOCH + Duration::from_secs(1 << 60); + assert!(millis_since_epoch(time).is_err()); + } +} diff --git a/src/query/src/promql/planner.rs b/src/query/src/promql/planner.rs index 17cc36a41d..bea2cae586 100644 --- a/src/query/src/promql/planner.rs +++ b/src/query/src/promql/planner.rs @@ -2687,13 +2687,17 @@ impl PromPlanner { let accepts_empty = matcher.is_match(""); let column_name = Self::find_case_sensitive_column(table_schema, matcher.name.as_str()); + // Prometheus reads a label a series does not carry as the empty + // string. A row can miss a label two ways: the table has no column + // for it, or the column exists but is NULL on that row — the latter + // is the norm for logical metrics sharing a physical table, which + // holds the union of their label columns. let col = if let Some(column_name) = column_name { let column = DfExpr::Column(Column::from_name(&column_name)); let field = table_schema .index_of_column_by_name(None, &column_name) .map(|index| table_schema.field(index)); if accepts_empty - && column_name == OTLP_AGGREGATION_TEMPORALITY_LABEL && let Some(data_type) = field .filter(|field| { field.is_nullable() @@ -2712,7 +2716,6 @@ impl PromPlanner { } } else { DfExpr::Literal(ScalarValue::Utf8(Some(String::new())), None) - .alias(matcher.name.clone()) }; let lit = DfExpr::Literal(ScalarValue::Utf8(Some(matcher.value)), None); let expr = match matcher.op { diff --git a/src/query/src/promql/planner/test/delta.rs b/src/query/src/promql/planner/test/delta.rs index f6c37bb24a..3727ea9a30 100644 --- a/src/query/src/promql/planner/test/delta.rs +++ b/src/query/src/promql/planner/test/delta.rs @@ -524,7 +524,7 @@ async fn delta_mixed_ranges_drop_and_float_ranges_sum() { } #[tokio::test] -async fn temporality_matchers_treat_null_as_absent() { +async fn matchers_read_absent_labels_as_empty() { let marker = OTLP_AGGREGATION_TEMPORALITY_LABEL; let schema = Arc::new(ArrowSchema::new(vec![Field::new( marker, @@ -596,31 +596,82 @@ async fn temporality_matchers_treat_null_as_absent() { ); } - let ordinary_schema = Arc::new(ArrowSchema::new(vec![Field::new( + // The rule is about NULL, not about the marker: any nullable label column + // can be NULL where the series does not carry the label. A non-nullable one + // has nothing to normalize. + for (nullable, wants_coalesce) in [(true, true), (false, false)] { + let ordinary_schema = Arc::new(ArrowSchema::new(vec![Field::new( + "label", + ArrowDataType::Utf8, + nullable, + )])); + let ordinary_scan = LogicalPlanBuilder::scan( + "ordinary_labels", + provider_as_source(Arc::new( + MemTable::try_new(ordinary_schema, vec![vec![]]).unwrap(), + )), + None, + ) + .unwrap() + .build() + .unwrap(); + let PromExpr::VectorSelector(selector) = + parser::parse(r#"metric{label!="delta"}"#).unwrap() + else { + unreachable!() + }; + let expressions = PromPlanner::matchers_to_expr(selector.matchers, ordinary_scan.schema()) + .unwrap() + .iter() + .map(ToString::to_string) + .join(" AND "); + assert_eq!( + wants_coalesce, + expressions.contains("coalesce"), + "nullable={nullable}: {expressions}" + ); + } + + // The other way a label goes absent is having no column for it at all. The + // literal standing in for the label has to be usable as a predicate. + let schema = Arc::new(ArrowSchema::new(vec![Field::new( "label", ArrowDataType::Utf8, true, )])); - let ordinary_scan = LogicalPlanBuilder::scan( + let batch = RecordBatch::try_new( + schema.clone(), + vec![Arc::new(StringArray::from(vec![Some("value")]))], + ) + .unwrap(); + let scan = LogicalPlanBuilder::scan( "ordinary_labels", provider_as_source(Arc::new( - MemTable::try_new(ordinary_schema, vec![vec![]]).unwrap(), + MemTable::try_new(schema, vec![vec![batch]]).unwrap(), )), None, ) .unwrap() .build() .unwrap(); - let PromExpr::VectorSelector(selector) = parser::parse(r#"metric{label!="delta"}"#).unwrap() + let PromExpr::VectorSelector(selector) = parser::parse(r#"metric{absent!="delta"}"#).unwrap() else { unreachable!() }; - let expressions = PromPlanner::matchers_to_expr(selector.matchers, ordinary_scan.schema()) + let expressions = PromPlanner::matchers_to_expr(selector.matchers, scan.schema()).unwrap(); + assert_eq!( + r#"Utf8("") != Utf8("delta")"#, + expressions.iter().map(ToString::to_string).join(" AND ") + ); + + let plan = LogicalPlanBuilder::from(scan) + .filter(conjunction(expressions).unwrap()) .unwrap() - .iter() - .map(ToString::to_string) - .join(" AND "); - assert!(!expressions.contains("coalesce"), "{expressions}"); + .build() + .unwrap(); + let (_, batches) = execute(plan, &build_query_engine_state()).await; + let rows = batches.iter().map(|batch| batch.num_rows()).sum::(); + assert_eq!(1, rows); } #[tokio::test] diff --git a/src/servers/src/http/prometheus.rs b/src/servers/src/http/prometheus.rs index 1f9988ede4..420a0c8f4b 100644 --- a/src/servers/src/http/prometheus.rs +++ b/src/servers/src/http/prometheus.rs @@ -55,7 +55,7 @@ use promql_parser::parser::{ AggregateExpr, BinaryExpr, Call, Expr as PromqlExpr, LabelModifier, MatrixSelector, ParenExpr, SubqueryExpr, UnaryExpr, VectorSelector, }; -use query::parser::{DEFAULT_LOOKBACK_STRING, PromQuery, QueryStatement}; +use query::parser::{DEFAULT_LOOKBACK_STRING, PromQuery, QueryLanguageParser, QueryStatement}; use serde::de::{self, MapAccess, Visitor}; use serde::{Deserialize, Serialize}; use serde_json::Value; @@ -74,7 +74,7 @@ use table::requests::{ pub use super::result::prometheus_resp::{PromSampleValue, PrometheusJsonResponse}; use crate::error::{ CollectRecordbatchSnafu, ConvertScalarValueSnafu, DataFusionSnafu, Error, InvalidQuerySnafu, - NotSupportedSnafu, Result, TableNotFoundSnafu, UnexpectedResultSnafu, + NotSupportedSnafu, ParseTimestampSnafu, Result, TableNotFoundSnafu, UnexpectedResultSnafu, }; use crate::http::header::collect_plan_metrics; use crate::otlp::metrics::ucum_to_openmetrics_unit; @@ -1711,9 +1711,35 @@ pub async fn label_values_query( ); let catalog_manager = handler.catalog_manager(); - let mut table_names = try_call_return_response!( - retrieve_table_names(&query_ctx, catalog_manager, matches).await - ); + // An empty `match[]` enumerates every metric; otherwise only the + // selectors answerable from metadata go down that path. + let enumerate_all = matches.is_empty(); + let (metadata_selectors, label_selectors) = + try_call_return_response!(split_selectors_by_label_use(&matches)); + + let mut table_names = if enumerate_all || !metadata_selectors.is_empty() { + try_call_return_response!( + retrieve_table_names(&query_ctx, catalog_manager, metadata_selectors).await + ) + } else { + Vec::new() + }; + + if !label_selectors.is_empty() { + table_names.extend(try_call_return_response!( + retrieve_table_names_by_labels( + &handler, + label_selectors, + params.start.as_deref(), + params.end.as_deref(), + &query_ctx, + ) + .await + )); + table_names.sort_unstable(); + table_names.dedup(); + } + table_names = try_call_return_response!( handler .filter_metadata_metric_names( @@ -1897,6 +1923,131 @@ fn take_metric_name(selector: &mut VectorSelector) -> Option { Some(name) } +/// Removes every `__name__` matcher from the selector and returns them, so the +/// rest can be planned as column predicates. A name given as `VectorSelector::name` +/// comes back as an equality matcher, making both spellings filter alike. +fn take_metric_name_matchers(selector: &mut VectorSelector) -> Vec { + let mut taken = Vec::new(); + if let Some(name) = selector.name.take() { + taken.push(Matcher::new(MatchOp::Equal, METRIC_NAME_LABEL, &name)); + } + + let (name_matchers, rest) = std::mem::take(&mut selector.matchers.matchers) + .into_iter() + .partition(|matcher| matcher.name == METRIC_NAME_LABEL); + selector.matchers.matchers = rest; + taken.extend(name_matchers); + + taken +} + +/// Whether a metric name satisfies every `__name__` matcher of one selector. +/// +/// Negated matchers are honoured here, unlike in [`retrieve_table_names`] where +/// they keep every table so the caller authorizes the full candidate set: the +/// names reaching this point are already narrowed by the data, so filtering them +/// can only remove names, never widen what a caller gets to see. +fn metric_name_matches(table_name: &str, matchers: &[Matcher]) -> bool { + matchers.iter().all(|matcher| match &matcher.op { + MatchOp::Equal => table_name == matcher.value, + MatchOp::NotEqual => table_name != matcher.value, + MatchOp::Re(re) => re.is_match(table_name), + MatchOp::NotRe(re) => !re.is_match(table_name), + }) +} + +/// Whether a matcher constrains an ordinary label. The others name the metric, +/// the database or the field, none of which is a column to scan. +fn is_ordinary_label_matcher(matcher: &Matcher) -> bool { + matcher.name != METRIC_NAME_LABEL + && matcher.name != FIELD_NAME_LABEL + && !is_database_selection_label(&matcher.name) +} + +/// Splits `match[]` selectors by whether they constrain an ordinary label. The +/// first group is answerable from table metadata; the second needs the data read +/// and is returned as parsed selectors. +/// +/// `or` matchers stay in the metadata group, which ignores them, rather than +/// being silently dropped from a data scan that cannot express them. +fn split_selectors_by_label_use(matches: &[String]) -> Result<(Vec, Vec)> { + let mut metadata_only = Vec::new(); + let mut with_labels = Vec::new(); + + for selector in matches { + let expr = promql_parser::parser::parse(selector) + .map_err(|reason| InvalidQuerySnafu { reason }.build())?; + let PromqlExpr::VectorSelector(vector_selector) = expr else { + return InvalidQuerySnafu { + reason: "expected vector selector".to_string(), + } + .fail(); + }; + + let constrains_labels = vector_selector.matchers.or_matchers.is_empty() + && vector_selector + .matchers + .matchers + .iter() + .any(is_ordinary_label_matcher); + if constrains_labels { + with_labels.push(vector_selector); + } else { + metadata_only.push(selector.clone()); + } + } + + Ok((metadata_only, with_labels)) +} + +/// Resolves selectors constraining ordinary labels into metric names: the data +/// answers the label matchers, then each selector's `__name__` matchers narrow +/// the names it found. +async fn retrieve_table_names_by_labels( + handler: &PrometheusHandlerRef, + selectors: Vec, + start: Option<&str>, + end: Option<&str>, + query_ctx: &QueryContextRef, +) -> Result> { + let start_arg = start.map(str::to_string).unwrap_or_else(yesterday_rfc3339); + let end_arg = end.map(str::to_string).unwrap_or_else(current_time_rfc3339); + let start = QueryLanguageParser::parse_promql_timestamp(&start_arg).with_context(|_| { + ParseTimestampSnafu { + timestamp: start_arg.clone(), + } + })?; + let end = QueryLanguageParser::parse_promql_timestamp(&end_arg).with_context(|_| { + ParseTimestampSnafu { + timestamp: end_arg.clone(), + } + })?; + + let schema = query_ctx.current_schema(); + let mut table_names = Vec::new(); + for mut selector in selectors { + let name_matchers = take_metric_name_matchers(&mut selector); + // The database and field matchers name no column, and the metadata path + // ignores them too. + let label_matchers = selector + .matchers + .matchers + .into_iter() + .filter(is_ordinary_label_matcher) + .collect(); + let matched = handler + .query_metric_names_by_labels(label_matchers, &schema, start, end, query_ctx) + .await?; + table_names.extend( + matched + .into_iter() + .filter(|name| metric_name_matches(name, &name_matchers)), + ); + } + + Ok(table_names) +} + async fn retrieve_table_names( query_ctx: &QueryContext, catalog_manager: CatalogManagerRef, @@ -2407,6 +2558,10 @@ mod tests { deny_operation: bool, denied_table: Option<&'static str>, metric_names: Vec, + /// Names the label-matcher path resolves, kept apart from `metric_names` + /// so a test can tell which path answered. + label_metric_names: Vec, + label_lookups: Mutex>>, queries: Mutex>, ordered_outputs: Mutex>, } @@ -2482,6 +2637,18 @@ mod tests { Ok(self.metric_names.clone()) } + async fn query_metric_names_by_labels( + &self, + matchers: Vec, + _: &str, + _: std::time::SystemTime, + _: std::time::SystemTime, + _: &QueryContextRef, + ) -> Result> { + self.label_lookups.lock().unwrap().push(matchers); + Ok(self.label_metric_names.clone()) + } + async fn query_label_values( &self, _: String, @@ -2547,6 +2714,8 @@ mod tests { deny_operation: false, denied_table: None, metric_names: Vec::new(), + label_metric_names: Vec::new(), + label_lookups: Mutex::new(Vec::new()), queries: Mutex::new(Vec::new()), ordered_outputs: Mutex::new(Vec::new()), }); @@ -2600,6 +2769,8 @@ mod tests { deny_operation: false, denied_table: None, metric_names: Vec::new(), + label_metric_names: Vec::new(), + label_lookups: Mutex::new(Vec::new()), queries: Mutex::new(Vec::new()), ordered_outputs: Mutex::new(Vec::new()), }); @@ -2684,6 +2855,8 @@ mod tests { deny_operation: false, denied_table: Some("denied"), metric_names: Vec::new(), + label_metric_names: Vec::new(), + label_lookups: Mutex::new(Vec::new()), queries: Mutex::new(Vec::new()), ordered_outputs: Mutex::new(Vec::new()), })), @@ -2700,6 +2873,153 @@ mod tests { ); } + /// A handler over the logical metric tables `cpu_user` and `cpu_system`, + /// which is what the metadata path enumerates, with the label-matcher path + /// answering `label_metric_names`. + fn label_values_handler(label_metric_names: Vec<&str>) -> Arc { + let mut cpu_user = test_table_info( + 1024, + "cpu_user", + DEFAULT_SCHEMA_NAME, + DEFAULT_CATALOG_NAME, + Arc::new(Schema::new(vec![])), + ); + cpu_user.meta.options.extra_options.insert( + LOGICAL_TABLE_METADATA_KEY.to_string(), + "physical_metrics".to_string(), + ); + let manager = MemoryCatalogManager::new_with_table(EmptyTable::from_table_info(&cpu_user)); + let mut cpu_system = cpu_user.clone(); + cpu_system.ident.table_id = 1025; + cpu_system.name = "cpu_system".to_string(); + manager + .register_table_sync(RegisterTableRequest { + catalog: DEFAULT_CATALOG_NAME.to_string(), + schema: DEFAULT_SCHEMA_NAME.to_string(), + table_name: cpu_system.name.clone(), + table_id: cpu_system.table_id(), + table: EmptyTable::from_table_info(&cpu_system), + }) + .unwrap(); + + Arc::new(TestPrometheusHandler { + catalog_manager: manager, + deny_operation: false, + denied_table: None, + metric_names: Vec::new(), + label_metric_names: label_metric_names.into_iter().map(String::from).collect(), + label_lookups: Mutex::new(Vec::new()), + queries: Mutex::new(Vec::new()), + ordered_outputs: Mutex::new(Vec::new()), + }) + } + + async fn query_metric_name_values( + handler: Arc, + matches: Vec<&str>, + ) -> Vec { + let state: PrometheusHandlerRef = handler; + let response = label_values_query( + State(state), + Path(METRIC_NAME_LABEL.to_string()), + Extension(QueryContext::with( + DEFAULT_CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + )), + Query(LabelValueQuery { + matches: Matches(matches.into_iter().map(String::from).collect()), + ..Default::default() + }), + ) + .await; + + assert!( + response.status_code.is_none(), + "status={:?}, error={:?}", + response.status_code, + response.error + ); + match response.data { + PrometheusResponse::LabelValues(values) => values, + other => panic!("expected label values, got {other:?}"), + } + } + + #[tokio::test] + async fn label_matchers_resolve_metric_names_from_data() { + let handler = label_values_handler(vec!["cpu_user"]); + let values = query_metric_name_values(handler.clone(), vec![r#"{pod="abc"}"#]).await; + + // The metadata path would have enumerated both metrics. + assert_eq!(vec!["cpu_user".to_string()], values); + + let lookups = handler.label_lookups.lock().unwrap(); + assert_eq!(1, lookups.len()); + assert_eq!( + vec!["pod".to_string()], + lookups[0] + .iter() + .map(|matcher| matcher.name.clone()) + .collect::>() + ); + } + + #[tokio::test] + async fn metric_name_matchers_narrow_data_resolved_names() { + let handler = label_values_handler(vec!["cpu_user", "cpu_system"]); + let values = + query_metric_name_values(handler, vec![r#"{__name__=~"cpu_u.*", pod="abc"}"#]).await; + + assert_eq!(vec!["cpu_user".to_string()], values); + } + + #[tokio::test] + async fn special_matchers_are_stripped_before_the_data_lookup() { + let handler = label_values_handler(vec!["cpu_user"]); + let values = query_metric_name_values( + handler.clone(), + vec![r#"{pod="abc", __field__="value", __database__="public"}"#], + ) + .await; + + assert_eq!(vec!["cpu_user".to_string()], values); + let lookups = handler.label_lookups.lock().unwrap(); + assert_eq!( + vec!["pod".to_string()], + lookups[0] + .iter() + .map(|matcher| matcher.name.clone()) + .collect::>() + ); + } + + #[tokio::test] + async fn database_and_field_matchers_stay_on_the_metadata_path() { + let handler = label_values_handler(vec!["never_returned"]); + let values = query_metric_name_values( + handler.clone(), + vec![r#"{__name__=~"cpu_.*", __field__="value", __database__="other"}"#], + ) + .await; + + assert_eq!( + vec!["cpu_system".to_string(), "cpu_user".to_string()], + values + ); + assert!(handler.label_lookups.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn empty_match_still_enumerates_every_metric() { + let handler = label_values_handler(Vec::new()); + let values = query_metric_name_values(handler, Vec::new()).await; + + assert_eq!( + vec!["cpu_system".to_string(), "cpu_user".to_string()], + values + ); + } + #[tokio::test] async fn test_series_query_expands_metric_name_regex() { let cpu_user = test_table_info( @@ -2728,6 +3048,8 @@ mod tests { deny_operation: false, denied_table: None, metric_names: vec!["cpu_user".to_string(), "cpu_system".to_string()], + label_metric_names: Vec::new(), + label_lookups: Mutex::new(Vec::new()), queries: Mutex::new(Vec::new()), ordered_outputs: Mutex::new(Vec::new()), }); @@ -3635,6 +3957,8 @@ mod tests { deny_operation: true, denied_table: None, metric_names: Vec::new(), + label_metric_names: Vec::new(), + label_lookups: Mutex::new(Vec::new()), queries: Mutex::new(Vec::new()), ordered_outputs: Mutex::new(Vec::new()), })), @@ -3649,6 +3973,8 @@ mod tests { deny_operation: false, denied_table: Some("denied"), metric_names: Vec::new(), + label_metric_names: Vec::new(), + label_lookups: Mutex::new(Vec::new()), queries: Mutex::new(Vec::new()), ordered_outputs: Mutex::new(Vec::new()), }); diff --git a/src/servers/src/prometheus_handler.rs b/src/servers/src/prometheus_handler.rs index 4f4b7c5d19..1e211b501a 100644 --- a/src/servers/src/prometheus_handler.rs +++ b/src/servers/src/prometheus_handler.rs @@ -184,6 +184,21 @@ pub trait PrometheusHandler { ctx: &QueryContextRef, ) -> Result>; + /// Query metric table names that carry data matching `matchers` in the time + /// range. `matchers` must hold only ordinary label matchers: `__name__` + /// names a table and the database and field matchers name no column, so the + /// caller resolves all three before calling. + /// + /// Only metric engine tables are covered. + async fn query_metric_names_by_labels( + &self, + matchers: Vec, + schema: &str, + start: SystemTime, + end: SystemTime, + ctx: &QueryContextRef, + ) -> Result>; + async fn query_label_values( &self, metric: String, diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 12b9ac883d..bd761f39a3 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -1524,6 +1524,117 @@ pub async fn test_prom_http_api(store_type: StorageType) { .unwrap() ); + // query `__name__` by a matcher on an ordinary label: the metric engine + // physical tables are scanned, so only metrics carrying the label value are + // returned. `demo_metrics` shares `phy` with `demo` but has no `host` value. + let res = client + .get("/v1/prometheus/api/v1/label/__name__/values?match[]={host=\"host1\"}&start=0&end=600") + .send() + .await; + let status = res.status(); + let text = res.text().await; + assert_eq!(status, StatusCode::OK, "{text}"); + let prom_resp = serde_json::from_str::(&text).unwrap(); + assert_eq!(prom_resp.status, "success"); + assert!(prom_resp.error.is_none()); + assert_eq!( + prom_resp.data, + serde_json::from_value::(json!(["demo", "multi_labels"])).unwrap() + ); + + // `__name__` matchers narrow the names the data resolved: `multi_labels` + // also carries `idc="idc1"` but its name does not match. + let res = client + .get("/v1/prometheus/api/v1/label/__name__/values?match[]={__name__=~\"demo.*\", idc=\"idc1\"}&start=0&end=600") + .send() + .await; + assert_eq!(res.status(), StatusCode::OK); + let prom_resp = res.json::().await; + assert_eq!(prom_resp.status, "success"); + assert!(prom_resp.error.is_none()); + assert_eq!( + prom_resp.data, + serde_json::from_value::(json!([ + "demo_metrics", + "demo_metrics_with_nanos", + ])) + .unwrap() + ); + + // The time range selects the series: `demo` carries `host="host2"` only at + // t=600, so narrowing the range drops it while `multi_labels` at t=0 stays. + let res = client + .get("/v1/prometheus/api/v1/label/__name__/values?match[]={host=\"host2\"}&start=0&end=600") + .send() + .await; + assert_eq!(res.status(), StatusCode::OK); + let prom_resp = res.json::().await; + assert_eq!( + prom_resp.data, + serde_json::from_value::(json!(["demo", "multi_labels"])).unwrap() + ); + let res = client + .get("/v1/prometheus/api/v1/label/__name__/values?match[]={host=\"host2\"}&start=0&end=300") + .send() + .await; + assert_eq!(res.status(), StatusCode::OK); + let prom_resp = res.json::().await; + assert_eq!( + prom_resp.data, + serde_json::from_value::(json!(["multi_labels"])).unwrap() + ); + + // Logical metrics sharing a physical table have NULL in the label columns + // they don't use. Prometheus reads a label a series doesn't carry as the + // empty string, so `demo_metrics` and `demo_metrics_with_nanos` — neither of + // which has a `host` label — match both of these. + // + // `.%2B` is `.+`; a bare `+` decodes to a space in a query string. Grafana + // encodes it the same way. + let res = client + .get("/v1/prometheus/api/v1/label/__name__/values?match[]={__name__=~\".%2B\", host=\"\"}&start=0&end=600") + .send() + .await; + assert_eq!(res.status(), StatusCode::OK); + let prom_resp = res.json::().await; + assert_eq!( + prom_resp.data, + serde_json::from_value::(json!([ + "demo_metrics", + "demo_metrics_with_nanos", + ])) + .unwrap() + ); + + let res = client + .get("/v1/prometheus/api/v1/label/__name__/values?match[]={__name__=~\".%2B\", host!=\"host1\"}&start=0&end=600") + .send() + .await; + assert_eq!(res.status(), StatusCode::OK); + let prom_resp = res.json::().await; + assert_eq!( + prom_resp.data, + serde_json::from_value::(json!([ + "demo", + "demo_metrics", + "demo_metrics_with_nanos", + "multi_labels", + ])) + .unwrap() + ); + + // A pre-epoch RFC3339 bound is a valid range, not a panic. + let res = client + .get("/v1/prometheus/api/v1/label/__name__/values?match[]={host=\"host1\"}&start=1969-12-31T23:59:59Z&end=600") + .send() + .await; + assert_eq!(res.status(), StatusCode::OK); + let prom_resp = res.json::().await; + assert_eq!( + prom_resp.data, + serde_json::from_value::(json!(["demo", "multi_labels"])).unwrap() + ); + // buildinfo let res = client .get("/v1/prometheus/api/v1/status/buildinfo") diff --git a/tests/cases/standalone/common/promql/regex.result b/tests/cases/standalone/common/promql/regex.result index 7e0132c2f4..c76b743eaf 100644 --- a/tests/cases/standalone/common/promql/regex.result +++ b/tests/cases/standalone/common/promql/regex.result @@ -146,8 +146,9 @@ TQL ANALYZE VERBOSE (0, 0, '1s') test{host!~".+"}; |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..0], lookback=[300000], interval=[1000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["host"] REDACTED +|_|_|_FilterExec: CASE WHEN host@1 IS NOT NULL THEN host@1 ELSE_END = REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, {"partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "selector":"LastRow { after_merge: true }", "distribution":"PerSeries", "projection": ["ts", "host", "val"], "filters": ["host = Dictionary(UInt32, Utf8(\"\"))", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(0, None)"], "REDACTED +|_|_|_SeriesScan: region=REDACTED, {"partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "distribution":"PerSeries", "projection": ["ts", "host", "val"], "filters": ["CASE WHEN host IS NOT NULL THEN host ELSE Dictionary(UInt32, Utf8(\"\")) END = Dictionary(UInt32, Utf8(\"\"))", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(0, None)"], "REDACTED |_|_|_| |_|_| Total rows: 0_| +-+-+-+