fix(prometheus): honor label matchers in __name__ values query (#9134)

* fix(prometheus): honor label matchers in __name__ values query

`/api/v1/label/__name__/values?match[]={pod="abc"}` dropped every matcher
other than `__name__` and returned all metrics in the schema. No error,
just the wrong list. Grafana's metrics browser sends this request, so
picking a label value there did nothing.

Selectors that only constrain `__name__` keep answering from table
metadata. A selector constraining an ordinary label now goes to the data:
scan each metric engine physical table for distinct `__table_id` in the
time range, map the ids back to metric names, then apply the selector's
own `__name__` matchers.

Only metric engine tables are covered; other engines share no column space
to scan.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* refactor(prometheus): batch-resolve metric names by table id

Building a full table-id-to-name map meant walking every table in the
schema and holding all of them in memory, just to name the handful the
scan returned. Use `tables_by_ids` instead — one batch KV read over the
ids the scan actually produced.

The catalog walk stays, but only to find the physical tables to scan.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix(promql): read an absent label as the empty string

A matcher on a label the series does not carry only worked when the table
had no column for it at all. Where the column exists but is NULL on that
row -- the norm for logical metrics sharing a metric engine physical
table, which holds the union of their label columns -- three-valued logic
dropped the row, so `host!="host1"` and `host=""` missed every metric
without a host label.

Coalesce nullable string label columns to "" for matchers that accept the
empty string, rather than only for the OTLP temporality marker. Equality
matchers are untouched; they cannot match NULL either way.

This is the Prometheus compatibility fix #8970 deliberately kept out of
its own scope. The cost is visible in the regex sqlness plan: the
predicate becomes a CASE, so the scan loses its LastRow selector and
grows a FilterExec. Only negative and empty-accepting matchers pay it.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix(promql): don't panic on pre-epoch label value bounds

`rewrite_label_values_query` unwrapped `duration_since(UNIX_EPOCH)`, which
returns an error for an instant before the epoch. `start=1969-12-31T23:59:59Z`
parses as valid RFC3339, so the request panicked instead of answering.

Recover the sign from the error branch, and report a value beyond i64
milliseconds as an error rather than wrapping the cast.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* refactor(prometheus): drop applicable_matchers, share the distinct scan

With the planner reading an absent label as empty, the frontend no longer
needs to pre-filter matchers per physical table. Removing that exposed a
second problem: a physical table that never took a column from a logical
table exposes no `__table_id`, and projecting it failed the whole request.
Skip those tables; the only thing that can miss is a metric with no labels.

Also pulls out the plan-build-execute-collect sequence the two label value
scans had in common.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

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,
 

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<Matcher>,
+        schema: &str,
+        start: SystemTime,
+        end: SystemTime,
+        ctx: &QueryContextRef,
+    ) -> server_error::Result<Vec<String>> {
+        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,

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<Matcher>,
+        schema: &str,
+        start: SystemTime,
+        end: SystemTime,
+        ctx: &QueryContextRef,
+    ) -> Result<Vec<String>> {
+        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::<Vec<_>>();
+        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::<Vec<_>>();
+        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<Vec<TableRef>> {
+        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<Matcher>,
+        column: String,
+        start: SystemTime,
+        end: SystemTime,
+        ctx: &QueryContextRef,
+    ) -> Result<Vec<RecordBatch>> {
+        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<Vec<TableId>> {
+        // `__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::<UInt32Array>()
+                .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 {

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,
 

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<Timestamp> {
+    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());
+    }
+}

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 {

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::<usize>();
+    assert_eq!(1, rows);
 }
 
 #[tokio::test]

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<String> {
     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<Matcher> {
+    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<String>, Vec<VectorSelector>)> {
+    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<VectorSelector>,
+    start: Option<&str>,
+    end: Option<&str>,
+    query_ctx: &QueryContextRef,
+) -> Result<Vec<String>> {
+    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<String>,
+        /// Names the label-matcher path resolves, kept apart from `metric_names`
+        /// so a test can tell which path answered.
+        label_metric_names: Vec<String>,
+        label_lookups: Mutex<Vec<Vec<Matcher>>>,
         queries: Mutex<Vec<String>>,
         ordered_outputs: Mutex<Vec<bool>>,
     }
@@ -2482,6 +2637,18 @@ mod tests {
             Ok(self.metric_names.clone())
         }
 
+        async fn query_metric_names_by_labels(
+            &self,
+            matchers: Vec<Matcher>,
+            _: &str,
+            _: std::time::SystemTime,
+            _: std::time::SystemTime,
+            _: &QueryContextRef,
+        ) -> Result<Vec<String>> {
+            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<TestPrometheusHandler> {
+        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<TestPrometheusHandler>,
+        matches: Vec<&str>,
+    ) -> Vec<String> {
+        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::<Vec<_>>()
+        );
+    }
+
+    #[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::<Vec<_>>()
+        );
+    }
+
+    #[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()),
         });

src/servers/src/prometheus_handler.rs

@@ -184,6 +184,21 @@ pub trait PrometheusHandler {
         ctx: &QueryContextRef,
     ) -> Result<Vec<String>>;
 
+    /// 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<Matcher>,
+        schema: &str,
+        start: SystemTime,
+        end: SystemTime,
+        ctx: &QueryContextRef,
+    ) -> Result<Vec<String>>;
+
     async fn query_label_values(
         &self,
         metric: String,

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::<PrometheusJsonResponse>(&text).unwrap();
+    assert_eq!(prom_resp.status, "success");
+    assert!(prom_resp.error.is_none());
+    assert_eq!(
+        prom_resp.data,
+        serde_json::from_value::<PrometheusResponse>(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::<PrometheusJsonResponse>().await;
+    assert_eq!(prom_resp.status, "success");
+    assert!(prom_resp.error.is_none());
+    assert_eq!(
+        prom_resp.data,
+        serde_json::from_value::<PrometheusResponse>(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::<PrometheusJsonResponse>().await;
+    assert_eq!(
+        prom_resp.data,
+        serde_json::from_value::<PrometheusResponse>(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::<PrometheusJsonResponse>().await;
+    assert_eq!(
+        prom_resp.data,
+        serde_json::from_value::<PrometheusResponse>(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::<PrometheusJsonResponse>().await;
+    assert_eq!(
+        prom_resp.data,
+        serde_json::from_value::<PrometheusResponse>(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::<PrometheusJsonResponse>().await;
+    assert_eq!(
+        prom_resp.data,
+        serde_json::from_value::<PrometheusResponse>(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::<PrometheusJsonResponse>().await;
+    assert_eq!(
+        prom_resp.data,
+        serde_json::from_value::<PrometheusResponse>(json!(["demo", "multi_labels"])).unwrap()
+    );
+
     // buildinfo
     let res = client
         .get("/v1/prometheus/api/v1/status/buildinfo")

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_|
 +-+-+-+