From 4e8bc6df86bbb1aeb6636bec0bebfcf5d5b49cad Mon Sep 17 00:00:00 2001 From: fys <40801205+fengys1996@users.noreply.github.com> Date: Mon, 19 Oct 2026 06:58:40 +0000 Subject: [PATCH] fix(prom): align remote read label matcher semantics (#9633) * fix(prom): align remote read label matcher semantics Signed-off-by: fys * chore: optimize comment of validate_matchers * fix(prom): preserve existing binary label regex handling Signed-off-by: fys * refactor: condition * test(prom): simplify remote read matcher tests and clarify comments * fix(prom): return binary remote read labels as UTF-8 * fix(prom): reject binary labels in remote read * refactor(prom): align remote read null label handling with PromQL Signed-off-by: fys * refactor(prom): simplify null handling in label matcher branches Signed-off-by: fys * fix(prom): restrict remote read labels to string types Signed-off-by: fys * test(prom): use string label in default value column test Signed-off-by: fys * fix: code review --------- Signed-off-by: fys --- src/frontend/src/error.rs | 8 + src/frontend/src/instance.rs | 4 +- src/frontend/src/instance/prom_store.rs | 84 ++++++- src/servers/src/prom_store.rs | 315 ++++++++++++++++++++++-- tests-integration/src/prom_store.rs | 6 + 5 files changed, 389 insertions(+), 28 deletions(-) diff --git a/src/frontend/src/error.rs b/src/frontend/src/error.rs index 451040c884a..923e6879f35 100644 --- a/src/frontend/src/error.rs +++ b/src/frontend/src/error.rs @@ -231,6 +231,13 @@ pub enum Error { source: servers::error::Error, }, + #[snafu(display("Invalid Prometheus remote read schema: {detail}"))] + InvalidPromRemoteReadSchema { + detail: String, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Failed to create logical plan for prometheus query"))] PromStoreRemoteQueryPlan { #[snafu(implicit)] @@ -394,6 +401,7 @@ impl ErrorExt for Error { | Error::SchemaExists { .. } | Error::ColumnNotFound { .. } | Error::AmbiguousValueColumn { .. } + | Error::InvalidPromRemoteReadSchema { .. } | Error::UnsupportedFormat { .. } | Error::IllegalAuthConfig { .. } | Error::ColumnNoneDefaultValue { .. } diff --git a/src/frontend/src/instance.rs b/src/frontend/src/instance.rs index e7d2e828b28..d72d0677552 100644 --- a/src/frontend/src/instance.rs +++ b/src/frontend/src/instance.rs @@ -3336,7 +3336,7 @@ mod tests { false, ) .with_time_index(true), - ColumnSchema::new("extra_field", ConcreteDataType::float64_datatype(), false), + ColumnSchema::new("extra_field", ConcreteDataType::string_datatype(), false), ColumnSchema::new( greptime_value(), ConcreteDataType::float64_datatype(), @@ -3347,7 +3347,7 @@ mod tests { schema, vec![ Arc::new(TimestampMillisecondVector::from_vec(vec![1000, 2000, 3000])) as VectorRef, - Arc::new(Float64Vector::from_vec(vec![99.0, 99.0, 99.0])) as VectorRef, + Arc::new(StringVector::from(vec!["99", "99", "99"])) as VectorRef, Arc::new(Float64Vector::from_vec(vec![1.0, 2.0, 3.0])) as VectorRef, ], ) diff --git a/src/frontend/src/instance/prom_store.rs b/src/frontend/src/instance/prom_store.rs index 875cbaa873d..7238c38f83d 100644 --- a/src/frontend/src/instance/prom_store.rs +++ b/src/frontend/src/instance/prom_store.rs @@ -35,6 +35,8 @@ use common_query::Output; use common_query::prelude::{GREPTIME_PHYSICAL_TABLE, greptime_value}; use common_recordbatch::RecordBatches; use common_telemetry::{debug, tracing}; +use datatypes::data_type::ConcreteDataType; +use datatypes::schema::Schema; use operator::insert::{ AutoCreateTableType, InserterRef, PerTableSemanticIndex, apply_per_table_semantic_options, build_create_table_expr, fill_table_options_for_create, parse_per_table_semantic_index, @@ -52,7 +54,7 @@ use servers::query_handler::{ PromStoreProtocolHandler, PromStoreProtocolHandlerRef, PromStoreResponse, }; use session::context::QueryContextRef; -use snafu::{OptionExt, ResultExt}; +use snafu::{OptionExt, ResultExt, ensure}; use store_api::metric_engine_consts::{METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY}; use store_api::mito_engine_options::SST_FORMAT_KEY; use table::TableRef; @@ -61,7 +63,7 @@ use tracing::instrument; use crate::error::{ AmbiguousValueColumnSnafu, CatalogSnafu, ColumnNotFoundSnafu, ExecLogicalPlanSnafu, - PromStoreRemoteQueryPlanSnafu, ReadTableSnafu, Result, + InvalidPromRemoteReadSchemaSnafu, PromStoreRemoteQueryPlanSnafu, ReadTableSnafu, Result, }; use crate::instance::Instance; @@ -172,7 +174,7 @@ async fn to_query_result( }) } -fn resolve_column_names(table_name: &str, table: &TableRef) -> Result { +fn resolve_value_column_name(table_name: &str, table: &TableRef) -> Result { let columns = table .field_columns() .map(|column| column.name) @@ -198,6 +200,29 @@ fn resolve_column_names(table_name: &str, table: &TableRef) -> Result { } } +fn ensure_string_label_columns( + schema: &Schema, + timestamp_column_name: &str, + value_column_name: &str, +) -> Result<()> { + for column in schema.column_schemas() { + if column.name == timestamp_column_name || column.name == value_column_name { + continue; + } + // All remaining columns are emitted as labels, including non-key columns. + ensure!( + matches!(column.data_type, ConcreteDataType::String(_)), + InvalidPromRemoteReadSchemaSnafu { + detail: format!( + "Prometheus remote read only supports string labels: column '{}' has type {:?}", + column.name, column.data_type + ), + } + ); + } + Ok(()) +} + impl Instance { #[tracing::instrument(skip_all)] async fn handle_remote_query( @@ -232,7 +257,9 @@ impl Instance { .name .clone(); - let value_column_name = resolve_column_names(table_name, &table)?; + let value_column_name = resolve_value_column_name(table_name, &table)?; + + ensure_string_label_columns(&table.schema(), ×tamp_column_name, &value_column_name)?; let dataframe = self .query_engine @@ -817,10 +844,59 @@ mod tests { use std::sync::Arc; use api::prom_store::remote::LabelMatcher; + use common_error::ext::ErrorExt; + use common_error::status_code::StatusCode; use session::context::QueryContext; use super::*; + #[test] + fn test_remote_read_label_types() { + use datatypes::schema::ColumnSchema; + + for (label_type, supported) in [ + (ConcreteDataType::string_datatype(), true), + (ConcreteDataType::large_string_datatype(), true), + (ConcreteDataType::utf8_view_datatype(), true), + (ConcreteDataType::binary_datatype(), false), + (ConcreteDataType::binary_view_datatype(), false), + (ConcreteDataType::json_datatype(), false), + (ConcreteDataType::vector_datatype(3), false), + (ConcreteDataType::int64_datatype(), false), + (ConcreteDataType::boolean_datatype(), false), + ] { + let schema = Schema::new(vec![ + ColumnSchema::new( + "ts", + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ), + ColumnSchema::new("value", label_type.clone(), true), + ColumnSchema::new("label", ConcreteDataType::string_datatype(), true), + ]); + // Value columns are validated separately, regardless of their type. + ensure_string_label_columns(&schema, "ts", "value").unwrap(); + + let result = ensure_string_label_columns(&schema, "ts", "label"); + assert_eq!(supported, result.is_ok(), "{label_type:?}"); + if let Err(err) = result { + assert!(matches!( + err, + crate::error::Error::InvalidPromRemoteReadSchema { .. } + )); + assert_eq!(err.status_code(), StatusCode::InvalidArguments); + assert!( + err.to_string() + .starts_with("Invalid Prometheus remote read schema: ") + ); + assert!( + err.to_string() + .contains("only supports string labels: column 'value'") + ); + } + } + } + #[test] fn test_auto_create_table_type_for_prom_remote_write_metric_engine() { let mut query_ctx = QueryContext::with( diff --git a/src/servers/src/prom_store.rs b/src/servers/src/prom_store.rs index 0568d967010..b991cc9c524 100644 --- a/src/servers/src/prom_store.rs +++ b/src/servers/src/prom_store.rs @@ -148,15 +148,17 @@ pub fn validate_matchers(matchers: &[LabelMatcher]) -> Result<()> { .build() })?; if matches!(matcher_type, MatcherType::Re | MatcherType::Nre) { - // Validate using the same regex engine as query execution. - // Its syntax differs from Prometheus's; e.g. `(?x) a` is accepted here - // but rejected by Prometheus. - regex::Regex::new(&matcher.value).map_err(|e| { - error::InvalidPromRemoteRequestSnafu { - msg: format!("invalid regex for label '{}': {e}", matcher.name), - } - .build() - })?; + // Like Prometheus, validate the raw regex first, then the anchored pattern. + // We use the same regex engine as query execution, whose syntax differs + // from Prometheus's: e.g. `(?x) a` is accepted here but rejected by Prometheus. + regex::Regex::new(&matcher.value) + .and_then(|_| regex::Regex::new(&anchored_label_regex(&matcher.value))) + .map_err(|e| { + error::InvalidPromRemoteRequestSnafu { + msg: format!("invalid regex for label '{}': {e}", matcher.name), + } + .build() + })?; } } Ok(()) @@ -196,22 +198,61 @@ pub fn query_to_plan( .build() })?; - match m_type { + // Prometheus treats an absent label as "". + let label = if dataframe.schema().has_column_with_unqualified_name(name) { + ident(name) + } else { + lit("") + }; + + // Like PromQL, include NULL when the matcher accepts the empty string. + // Keep EQ/NEQ on the original column for index recognition. + let condition = match m_type { MatcherType::Eq => { - conditions.push(ident(name).eq(lit(value))); + if value.is_empty() { + label.clone().eq(lit(value)).or(label.is_null()) + } else { + label.eq(lit(value)) + } } MatcherType::Neq => { - conditions.push(ident(name).not_eq(lit(value))); + if value.is_empty() { + label.not_eq(lit(value)) + } else { + label.clone().not_eq(lit(value)).or(label.is_null()) + } } - // Case sensitive regexp match - MatcherType::Re => { - conditions.push(regexp_match(ident(name), lit(value), None).is_not_null()); + MatcherType::Re | MatcherType::Nre => { + let pattern = anchored_label_regex(value); + let matches_empty = regex::Regex::new(&pattern) + .map_err(|e| { + error::InvalidPromRemoteRequestSnafu { + msg: format!("invalid regex for label '{name}': {e}"), + } + .build() + })? + .is_match(""); + let matched = regexp_match(label.clone(), lit(pattern), None); + if m_type == MatcherType::Re { + let condition = matched.is_not_null(); + if matches_empty { + condition.or(label.is_null()) + } else { + condition + } + } else { + let condition = matched.is_null(); + // A NULL input also produces NULL. Keep it only when the + // negative matcher accepts the empty string. + if matches_empty { + condition.and(label.is_not_null()) + } else { + condition + } + } } - // Case sensitive regexp not match - MatcherType::Nre => { - conditions.push(regexp_match(ident(name), lit(value), None).is_null()); - } - } + }; + conditions.push(condition); } // Safety: conditions MUST not be empty, reduce always return Some(expr). @@ -224,6 +265,12 @@ pub fn query_to_plan( Ok(dataframe.into_parts().1) } +/// Prometheus's NewFastRegexMatcher also wraps patterns in ^(?s:...)$: +/// https://github.com/prometheus/prometheus/blob/main/model/labels/regexp.go +fn anchored_label_regex(value: &str) -> String { + format!("^(?s:{value})$") +} + #[inline] fn new_label(name: String, value: String) -> Label { Label { name, value } @@ -941,6 +988,32 @@ mod tests { const NEQ_TYPE: i32 = MatcherType::Neq as i32; const RE_TYPE: i32 = MatcherType::Re as i32; + #[test] + fn test_validate_matchers_regex_syntax() { + for op in [MatcherType::Re, MatcherType::Nre] { + for (value, valid) in [ + ("api|web", true), + ("(?s)api.web", true), + ("(?x) api", true), + ("(?x) api # comment", false), + ("[", false), + // Wrapping must not make an invalid original pattern valid. + ("a)|(?:b", false), + ] { + let matcher = LabelMatcher { + name: "job".into(), + value: value.into(), + r#type: op as i32, + }; + assert_eq!( + valid, + validate_matchers(&[matcher]).is_ok(), + "{op:?}: {value}" + ); + } + } + } + #[test] fn test_table_name() { let q = Query { @@ -1095,7 +1168,7 @@ mod tests { }, LabelMatcher { name: "job".to_string(), - value: "*prom*".to_string(), + value: ".*prom.*".to_string(), r#type: RE_TYPE, }, LabelMatcher { @@ -1123,12 +1196,210 @@ mod tests { let ts_col = greptime_timestamp(); let expected = format!( - "Filter: ?table?.{} >= TimestampMillisecond(1000, None) AND ?table?.{} <= TimestampMillisecond(2000, None) AND regexp_match(?table?.job, Utf8(\"*prom*\")) IS NOT NULL AND ?table?.instance != Utf8(\"localhost\") AND ?table?.service.name = Utf8(\"api\") AND ?table?.Region = Utf8(\"us\")\n TableScan: ?table?", + "Filter: ?table?.{} >= TimestampMillisecond(1000, None) AND ?table?.{} <= TimestampMillisecond(2000, None) AND regexp_match(?table?.job, Utf8(\"^(?s:.*prom.*)$\")) IS NOT NULL AND (?table?.instance != Utf8(\"localhost\") OR ?table?.instance IS NULL) AND ?table?.service.name = Utf8(\"api\") AND ?table?.Region = Utf8(\"us\")\n TableScan: ?table?", ts_col, ts_col ); assert_eq!(expected, display_string); } + #[tokio::test] + async fn test_remote_read_label_matcher_semantics() { + let ctx = SessionContext::new(); + let batch = DfRecordBatch::try_from_iter(vec![ + ( + greptime_timestamp(), + Arc::new(TimestampMillisecondArray::from(vec![1000; 6])) as ArrayRef, + ), + ( + "job", + Arc::new(StringArray::from(vec![ + Some("api"), + Some("my-api-server"), + Some(""), + None, + Some("web"), + Some("api\nweb"), + ])) as ArrayRef, + ), + ( + "id", + Arc::new(UInt32Array::from(vec![0, 1, 2, 3, 4, 5])) as ArrayRef, + ), + ]) + .unwrap(); + ctx.register_batch("metrics", batch).unwrap(); + for (name, op, value, expected) in [ + ("job", MatcherType::Eq, "api", vec![0]), + ("job", MatcherType::Neq, "api", vec![1, 2, 3, 4, 5]), + ("job", MatcherType::Eq, "", vec![2, 3]), + ("job", MatcherType::Neq, "", vec![0, 1, 4, 5]), + ("job", MatcherType::Re, "api", vec![0]), + ("job", MatcherType::Nre, "api", vec![1, 2, 3, 4, 5]), + ("job", MatcherType::Re, "api|web", vec![0, 4]), + ("job", MatcherType::Nre, "api|web", vec![1, 2, 3, 5]), + ("job", MatcherType::Re, "", vec![2, 3]), + ("job", MatcherType::Nre, "", vec![0, 1, 4, 5]), + ("job", MatcherType::Re, ".*", vec![0, 1, 2, 3, 4, 5]), + ("job", MatcherType::Nre, ".*", vec![]), + ("job", MatcherType::Re, ".+", vec![0, 1, 4, 5]), + ("job", MatcherType::Nre, ".+", vec![2, 3]), + ("job", MatcherType::Re, "(?m)api", vec![0]), + ("job", MatcherType::Re, "api.web", vec![5]), + ("job", MatcherType::Nre, "api.web", vec![0, 1, 2, 3, 4]), + ("job", MatcherType::Re, "(?-s:api.web)", vec![]), + ("job", MatcherType::Re, "(api)?", vec![0, 2, 3]), + ("missing", MatcherType::Eq, "", vec![0, 1, 2, 3, 4, 5]), + ("missing", MatcherType::Eq, "us", vec![]), + ("missing", MatcherType::Neq, "us", vec![0, 1, 2, 3, 4, 5]), + ("missing", MatcherType::Neq, "", vec![]), + ("missing", MatcherType::Re, ".*", vec![0, 1, 2, 3, 4, 5]), + ("missing", MatcherType::Re, ".+", vec![]), + ("missing", MatcherType::Nre, ".*", vec![]), + ("missing", MatcherType::Nre, ".+", vec![0, 1, 2, 3, 4, 5]), + ("missing", MatcherType::Re, "", vec![0, 1, 2, 3, 4, 5]), + ("missing", MatcherType::Nre, "", vec![]), + ] { + let query = Query { + start_timestamp_ms: 1000, + end_timestamp_ms: 1000, + matchers: vec![LabelMatcher { + name: name.into(), + value: value.into(), + r#type: op as i32, + }], + ..Default::default() + }; + let plan = query_to_plan( + ctx.table("metrics").await.unwrap(), + &query, + greptime_timestamp(), + ) + .unwrap_or_else(|err| panic!("{name} {op:?} {value:?}: {err}")); + let batches = ctx + .execute_logical_plan(plan) + .await + .unwrap() + .collect() + .await + .unwrap(); + let actual: Vec = batches + .iter() + .flat_map(|batch| { + batch + .column(2) + .as_primitive::() + .values() + .to_vec() + }) + .collect(); + assert_eq!(expected, actual, "{name} {op:?} {value:?}"); + } + } + + #[tokio::test] + async fn test_remote_read_matcher_combinations_and_label_types() { + let labels = StringArray::from(vec![Some("api"), Some("api"), None, Some("api")]); + for label_type in [ + DataType::Utf8, + DataType::LargeUtf8, + DataType::Utf8View, + DataType::Dictionary(Box::new(DataType::UInt32), Box::new(DataType::Utf8)), + ] { + let ctx = SessionContext::new(); + let batch = DfRecordBatch::try_from_iter(vec![ + ( + greptime_timestamp(), + Arc::new(TimestampMillisecondArray::from(vec![999, 1000, 2000, 2001])) + as ArrayRef, + ), + ("Job.Name", casts::cast(&labels, &label_type).unwrap()), + ]) + .unwrap(); + ctx.register_batch("metrics", batch).unwrap(); + let mut cases = vec![ + (vec![("Job.Name", MatcherType::Re, ".*")], vec![1000, 2000]), + (vec![("Job.Name", MatcherType::Nre, ".+")], vec![2000]), + ]; + // Exercise combinations once; other representations only need regex/NULL coverage. + if label_type == DataType::Utf8 { + cases.extend([ + ( + vec![ + ("Job.Name", MatcherType::Re, "api|web"), + ("Job.Name", MatcherType::Neq, "web"), + ], + vec![1000], + ), + ( + vec![ + ("Job.Name", MatcherType::Eq, ""), + ("Job.Name", MatcherType::Neq, "api"), + ], + vec![2000], + ), + ( + vec![ + ("Job.Name", MatcherType::Eq, "api"), + ("Job.Name", MatcherType::Eq, "web"), + ], + vec![], + ), + ( + vec![ + ("job.name", MatcherType::Eq, ""), + ("Job.Name", MatcherType::Re, ".*"), + ], + vec![1000, 2000], + ), + ]); + } + for (matchers, expected) in cases { + let mut query = Query { + start_timestamp_ms: 1000, + end_timestamp_ms: 2000, + matchers: matchers + .iter() + .map(|(name, op, value)| LabelMatcher { + name: (*name).into(), + value: (*value).into(), + r#type: *op as i32, + }) + .collect(), + ..Default::default() + }; + query.matchers.push(LabelMatcher { + name: METRIC_NAME_LABEL.into(), + value: "metrics".into(), + r#type: EQ_TYPE, + }); + let plan = query_to_plan( + ctx.table("metrics").await.unwrap(), + &query, + greptime_timestamp(), + ) + .unwrap(); + let batches = ctx + .execute_logical_plan(plan) + .await + .unwrap() + .collect() + .await + .unwrap(); + let actual: Vec = batches + .iter() + .flat_map(|batch| { + batch + .column(0) + .as_primitive::() + .values() + .to_vec() + }) + .collect(); + assert_eq!(expected, actual, "{label_type:?}: {matchers:?}"); + } + } + } + fn column_schemas_with( mut kts_iter: Vec<(&str, ColumnDataType, SemanticType)>, ) -> Vec { diff --git a/tests-integration/src/prom_store.rs b/tests-integration/src/prom_store.rs index 80da82c0e89..87798c282c1 100644 --- a/tests-integration/src/prom_store.rs +++ b/tests-integration/src/prom_store.rs @@ -708,6 +708,12 @@ mod tests { (MatcherType::Neq as i32, "[", 200), (MatcherType::Re as i32, "spark", 200), (MatcherType::Nre as i32, "api.*", 200), + (MatcherType::Re as i32, "(?x) spark", 200), + (MatcherType::Nre as i32, "(?x) spark", 200), + (MatcherType::Re as i32, "(?x) spark # comment", 400), + (MatcherType::Nre as i32, "(?x) spark # comment", 400), + (MatcherType::Re as i32, "a)|(?:b", 400), + (MatcherType::Nre as i32, "a)|(?:b", 400), ] { let mut query = remote_query(metric); query.matchers.push(LabelMatcher {