feat(promql): compile absent_over_time on the native route (#10198)

`absent_over_time(<selector>[<range>])` had no native kernel, so the
statement declined to the existing planner; there a nameless selector has
no metric name to look up, and the query failed instead of answering the
label set Prometheus reports.

Bind it to the whole-vector operation `absent` already uses, over a window
read of the selector's range. The fold keeps the metric name: a presence
fold that drops it would fold two metrics which differ only by name into
one series and refuse the second value, while `absent` only asks whether
the window held anything at all. Labels come from the same derivation
`absent` uses, now also accepting a matrix selector.

A statement holding a nameless selector that discovery expands into more
than one metric table keeps declining: keeping the name would declare
`__name__` without a bound metric name, and dropping it would merge the
metrics. The typed empty input of a selector that matched no metric table
is excluded from that decline, so a zero-match query plans natively and
answers from it.

The legacy planner is untouched.

Signed-off-by: discord9 <discord9@163.com>

src/query/src/promql/native_compile.rs

@@ -31,6 +31,7 @@
 //!   <histogram function>(<expression>)             e.g. histogram_count(x)
 //!   <sort>(<expression>[, <string>]*)              e.g. sort_desc(x), sort_by_label(x, "job")
 //!   absent(<expression>)                           e.g. absent(x{job="a"})
+//!   absent_over_time(<matrix>)                     e.g. absent_over_time(x{job="a"}[5m])
 //!   <histogram>(<constant>+, <expression>)         e.g. histogram_quantile(0.9, rate(x[5m]))
 //!   <label function>(<expression>, <string>*)       e.g. label_replace(x, "a", "$1", "b", "(.*)")
 //!   <expression> <operator> <scalar>               e.g. x * 100, x > bool time()
@@ -265,8 +266,9 @@ const SUPPORTED: &str = "a selector — `x`, optionally with an `offset` — a r
                          grouped by `by` or `without` as in `avg by(job)(rate(x[5m]) * 100)`; \
                          `topk`, `bottomk`, `limitk`, `limit_ratio` or `count_values` over any of \
                          them, as in `topk(3, x) by(job)` and `count_values(\"v\", x)`; a \
-                         sort, `absent` or `histogram_quantile` / `histogram_fraction` over any of \
-                         them (`sort_desc(x)`, `absent(x{job=\"a\"})`, \
+                         sort, `absent`, `absent_over_time` or `histogram_quantile` / \
+                         `histogram_fraction` over any of them (`sort_desc(x)`, \
+                         `absent(x{job=\"a\"})`, `absent_over_time(x{job=\"a\"}[5m])`, \
                          `histogram_quantile(0.9, rate(x[5m]))`); a range function's own \
                          constants sit where that function takes them, before \
                          the matrix for `quantile_over_time(0.9, x[5m])` and after it for \
@@ -1394,6 +1396,11 @@ fn compile_call(context: &Context, call: &Call) -> Result<NativeNode> {
     if call.func.name == "info" {
         return info_call(context, &call.args.args);
     }
+    // `absent_over_time` reads a matrix where `absent` reads a vector, so it cannot ride the vector
+    // path below: the operation it binds is `absent`'s own, over a presence read of the window.
+    if call.func.name == ABSENT_OVER_TIME {
+        return absent_over_time_call(context, &call.args.args);
+    }
     let Call { func, args } = call;
     if let Some(node) = scalar_call(context, func.name, &args.args)? {
         return Ok(node);
@@ -1716,6 +1723,71 @@ fn parameterized_histogram_call(
     }))
 }
 
+/// The PromQL name of `absent_over_time`, which is not a range function here: it reads a matrix but
+/// answers with `absent`'s vector operation rather than with one value per series.
+const ABSENT_OVER_TIME: &str = "absent_over_time";
+
+/// The range function `absent_over_time` folds a window with, to learn whether a series has a sample
+/// in it.
+///
+/// `absent` answers at every point its input holds no series at, so the matrix has to become one
+/// series per series holding a sample: a range function at all, rather than a value no series
+/// states. Which range function is therefore a question of which points a window reports, and
+/// `present_over_time` and `last_over_time` report the same ones — a window with at least one
+/// sample. The name is what separates them: `present_over_time` drops the metric name, and a
+/// selector written without one is discovered as one source per metric, so two metrics carrying the
+/// same ordinary labels would arrive at the fold as one label set. The window refuses two values of
+/// one label set at one point, because everything above it reads them as one series: the query
+/// would fail where Prometheus answers with the empty vector. Keeping the name keeps those two
+/// series apart, and `absent` reads presence alone, so the value this fold computes is not part of
+/// the answer.
+const PRESENCE_FOLD: RangeFunction = RangeFunction::LastOverTime;
+
+/// Binds `absent_over_time(m[d])` into `absent`'s own vector operation, over a presence read of the
+/// window.
+///
+/// Both halves are `absent`'s: the labels come from the matrix selector's matchers exactly as
+/// `absent` derives them from a vector selector's, and the operation tests whether its input holds
+/// a series at each grid point. What differs is only where that input comes from — a matrix folds a
+/// window rather than selecting an instant — which is why the call is bound here and not by
+/// [`vector_call`]: `absent_over_time` reads a matrix, and nothing in it is a vector expression the
+/// recursion could compile.
+///
+/// The existing plan answers this call per series with a UDF, which needs a metric name to read a
+/// table from; a selector written without one names its metrics only by the labels that pick them,
+/// which is exactly the shape `absent_over_time` is asked for when nothing matches.
+fn absent_over_time_call(context: &Context, arguments: &[Box<PromExpr>]) -> Result<NativeNode> {
+    let [matrix] = arguments else {
+        return Err(not_implemented(&format!(
+            "a call to `{ABSENT_OVER_TIME}` with {} argument(s), where it takes one matrix",
+            arguments.len()
+        )));
+    };
+    let labels = absent_labels_of(matrix)?;
+    let input = match unparen(matrix) {
+        PromExpr::MatrixSelector(MatrixSelector { vs, range }) => {
+            window_selection(context, PRESENCE_FOLD, vs, range)?
+        }
+        PromExpr::Subquery(subquery) => {
+            subquery_window(context, ABSENT_OVER_TIME, PRESENCE_FOLD, subquery)?
+        }
+        other => {
+            return Err(not_implemented(&format!(
+                "a call to `{ABSENT_OVER_TIME}` whose matrix is `{other}`, where it has to be a \
+                 range selector like `x[5m]` or a subquery like `x[5m:1m]`"
+            )));
+        }
+    };
+    let output = input.output.clone();
+    Ok(NativeNode {
+        expr: NativeExpr::Vector {
+            input: Box::new(input),
+            op: BoundVectorOp::Absent { labels },
+        },
+        output,
+    })
+}
+
 /// The vector operation a sort, `absent` or a classic histogram function binds, and the vector it reads,
 /// or `None` when `name` is none of them.
 fn vector_op<'a>(
@@ -1794,22 +1866,30 @@ fn vector_op<'a>(
     Ok(Some((BoundVectorOp::Sort(order), vector)))
 }
 
-/// The labels `absent(expr)` answers with, as Prometheus `createLabelsForAbsentFunction` derives
-/// them: from the matchers of a vector selector, and none for any other expression.
+/// The labels `absent(expr)` and `absent_over_time(m[d])` answer with, as Prometheus
+/// `createLabelsForAbsentFunction` derives them: from the matchers of a vector or range selector,
+/// and none for any other expression.
+///
+/// Both selectors carry the same matchers, and Prometheus derives the labels from them the same way
+/// in either position, so the range selector of `absent_over_time` reads exactly as the vector
+/// selector of `absent` does.
 ///
 /// `__field__`, `__schema__` and `__database__` choose what is read rather than match a label, so
 /// they are not labels of the answer, as in the existing planner.
 fn absent_labels_of(expr: &PromExpr) -> Result<SeriesLabels> {
-    let PromExpr::VectorSelector(selector) = unparen(expr) else {
-        return absent_labels(&[]);
+    let (selector, alternatives) = match unparen(expr) {
+        PromExpr::VectorSelector(selector) => (&selector.matchers, &selector.matchers.or_matchers),
+        PromExpr::MatrixSelector(MatrixSelector { vs, .. }) => {
+            (&vs.matchers, &vs.matchers.or_matchers)
+        }
+        _ => return absent_labels(&[]),
     };
-    if !selector.matchers.or_matchers.is_empty() {
+    if !alternatives.is_empty() {
         return Err(not_implemented(&format!(
-            "`absent` over `{selector}`, whose matchers are alternatives"
+            "`absent` or `absent_over_time` over `{expr}`, whose matchers are alternatives"
         )));
     }
     let matchers = selector
-        .matchers
         .matchers
         .iter()
         .filter(|matcher| {

src/query/src/promql/native_compile/tests.rs

@@ -4330,6 +4330,265 @@ fn whole_vector_functions_bind_a_vector_operation() {
     }
 }
 
+/// `absent_over_time(m[d])` binds `absent`'s own operation over a read of the window, so the labels
+/// it answers with are derived from the range selector's matchers exactly as `absent` derives them
+/// from a vector selector's.
+#[test]
+fn absent_over_time_binds_absent_over_a_presence_read() {
+    let cases = [
+        // A selector that names no metric answers with every equality matcher: this is the shape
+        // the existing plan cannot answer, because its per-series UDF needs a metric name to read a
+        // table from and a nameless selector writes none.
+        (
+            r#"absent_over_time({no_such_label="x", job="api"}[5m])"#,
+            r#"absent {job="api", no_such_label="x"}"#,
+        ),
+        // The table-choosing matchers are not labels of the answer, as they are not for `absent`.
+        (
+            r#"absent_over_time(x{job="a", env=~"p.*", instance!="", __field__="v"}[5m])"#,
+            r#"absent {job="a"}"#,
+        ),
+        // `job!="b"` accepts a missing `job`, so it reads after `job="a"` and removes it.
+        (r#"absent_over_time(x{job!="b", job="a"}[5m])"#, "absent {}"),
+        // A subquery is a matrix too, and it is not a selector, so it fixes no label.
+        (r#"absent_over_time(rate(x[1m])[5m:1m])"#, "absent {}"),
+    ];
+    for (query, op) in cases {
+        let tree = compile(&eval_stmt(query), raw_schema(), SeriesOrdering::KeyThenTime)
+            .unwrap_or_else(|error| panic!("`{query}` was refused: {error}"));
+        let display = tree.to_string();
+        let root = display.lines().nth(1).expect("a compiled tree has a root");
+        assert_eq!(root, format!("  root Vector {op}"), "`{query}`");
+        // The fold that reports presence keeps the metric name, so two discovered metrics that
+        // share their ordinary labels stay two series on the way up: `present_over_time` drops it.
+        assert!(
+            display.contains("input Window last_over_time [300000ms]"),
+            "`{query}` does not read its window through the presence fold: {display}"
+        );
+    }
+}
+
+/// `absent_over_time(x)` with `argument` in place of its matrix.
+///
+/// The parser types and counts a call's arguments, so a call whose matrix is not a matrix can only
+/// reach this compiler on a statement built here.
+fn with_absent_over_time_argument(argument: &str) -> EvalStmt {
+    let mut stmt = eval_stmt("absent_over_time(x[5m])");
+    let PromExpr::Call(call) = &mut stmt.expr else {
+        unreachable!("`absent_over_time(x[5m])` parses to a call");
+    };
+    *call.args.args[0] = promql_parser::parser::parse(argument).unwrap();
+    stmt
+}
+
+/// A call that is not `absent_over_time(<matrix>)` is refused as that, rather than read as another
+/// statement's shape.
+#[test]
+fn absent_over_time_refuses_what_is_not_a_matrix() {
+    // A call's argument count and types are the parser's, so neither of these reaches a plan.
+    for query in ["absent_over_time(x)", "absent_over_time(x[5m], 1)"] {
+        assert!(
+            promql_parser::parser::parse(query).is_err(),
+            "`{query}` is a call this compiler would have had to refuse"
+        );
+    }
+    // A matrix position holding something else, which only a built statement can express.
+    let error = compile(
+        &with_absent_over_time_argument("x"),
+        raw_schema(),
+        SeriesOrdering::KeyThenTime,
+    )
+    .expect_err("a call over a vector was accepted");
+    assert!(
+        error
+            .to_string()
+            .contains("whose matrix is `x`, where it has to be a range selector"),
+        "{error}"
+    );
+    // Alternative (`or`) matchers are alternatives no one scan can filter, so the labels they would
+    // fix are not this compiler's to derive.
+    let error = compile(
+        &eval_stmt(r#"absent_over_time(x{a="b" or c="d"}[5m])"#),
+        raw_schema(),
+        SeriesOrdering::KeyThenTime,
+    )
+    .expect_err("alternative matchers were accepted");
+    assert!(
+        error
+            .to_string()
+            .contains("whose matchers are alternatives"),
+        "{error}"
+    );
+}
+
+/// `value` at each of [`eval_stmt`]'s 21 grid points.
+fn grid_values(value: f64) -> Vec<(usize, Option<f64>)> {
+    (0..21).map(|position| (position, Some(value))).collect()
+}
+
+/// The points `absent_over_time` answers at are the ones no series has a sample at, and the labels
+/// are the ones its selector's matchers fix.
+///
+/// Prometheus v0.315 answers the same statements the same way: `absent_over_time(a[5m])` reports
+/// nothing while the window holds a sample and one unlabelled series once it does not, and
+/// `absent_over_time({no_such_label="x", job="api"}[5m])` reports `{job="api", no_such_label="x"}
+/// = 1`, because a series that carries no `no_such_label` reads as the empty string and the matcher
+/// wants `x`.
+#[test]
+fn absent_over_time_answers_only_the_points_no_series_has_a_sample_at() {
+    let schema = raw_schema();
+    let timestamps = agg_timestamps_0_to_100();
+    let values: Vec<f64> = (1..=21).map(f64::from).collect();
+    let with_samples = || {
+        vec![agg_raw_chunk(
+            &schema,
+            b"a",
+            agg_metric("a", &[("job", "api")]),
+            &timestamps,
+            &values,
+        )]
+    };
+
+    // Every window holds a sample of `a`, so there is nothing to report.
+    assert!(
+        aggregate_results(&execute_compiled("absent_over_time(a[5m])", with_samples())).is_empty(),
+        "a series holding a sample in every window is not absent"
+    );
+
+    // The metric exists but holds nothing in the window: the answer has no `__name__` to report,
+    // because the matcher that named the metric is not a label of an `absent` answer.
+    assert_eq!(
+        aggregate_results(&execute_compiled("absent_over_time(a[5m])", Vec::new())),
+        vec![(None, Vec::new(), grid_values(1.0))],
+        "a metric with no sample in any window answers the empty label set"
+    );
+
+    // A matcher that no row satisfies is the shape the existing plan declines: the labels still
+    // have to reach the answer, at every point of the grid.
+    assert_eq!(
+        aggregate_results(&execute_compiled(
+            r#"absent_over_time({no_such_label="x", job="api"}[5m])"#,
+            Vec::new()
+        )),
+        vec![(
+            None,
+            vec![
+                ("job".to_string(), "api".to_string()),
+                ("no_such_label".to_string(), "x".to_string()),
+            ],
+            grid_values(1.0),
+        )]
+    );
+
+    // An equality matcher on a label a present series carries fixes that label too.
+    assert_eq!(
+        aggregate_results(&execute_compiled(
+            r#"absent_over_time(a{job="api"}[5m])"#,
+            Vec::new()
+        )),
+        vec![(
+            None,
+            vec![("job".to_string(), "api".to_string())],
+            grid_values(1.0)
+        )]
+    );
+}
+
+/// The fold reads one window per grid point, so a sample outside the window does not make the point
+/// present, and an instant statement answers at its one point alone.
+#[test]
+fn absent_over_time_reads_the_window_and_not_the_series() {
+    let schema = raw_schema();
+    let chunk_at = |timestamps: &[i64]| {
+        vec![agg_raw_chunk(
+            &schema,
+            b"a",
+            agg_metric("a", &[("job", "api")]),
+            timestamps,
+            &[1.0; 2],
+        )]
+    };
+    let labels = vec![("job".to_string(), "api".to_string())];
+
+    // Samples a minute and a half before the statement's start are inside every five-minute window
+    // it evaluates, so every point reads the series and `absent` has nothing to say.
+    assert!(
+        aggregate_results(&execute_compiled(
+            "absent_over_time(a[5m])",
+            chunk_at(&[-90_000, -60_000])
+        ))
+        .is_empty(),
+        "a window reaching back past the samples holds them"
+    );
+
+    // Samples four minutes before it are outside every one of those windows: the metric exists and
+    // holds nothing the windows read, so the equality matcher's label is the whole answer.
+    assert_eq!(
+        aggregate_results(&execute_compiled(
+            r#"absent_over_time(a{job="api"}[5m])"#,
+            chunk_at(&[-400_000, -350_000])
+        )),
+        vec![(None, labels.clone(), grid_values(1.0))]
+    );
+
+    // The same statement as an instant query answers at its one point alone.
+    let instant = eval_stmt_between(
+        r#"absent_over_time(a{job="api"}[5m])"#,
+        Duration::from_secs(100),
+        Duration::from_secs(100),
+    );
+    let tree = compile(&instant, raw_schema(), SeriesOrdering::KeyThenTime).unwrap();
+    let pool: Arc<dyn MemoryPool> = Arc::new(GreedyMemoryPool::new(1 << 20));
+    let mut execution =
+        NativeExecutor::new(NativeExecutionState::try_new(&tree, pool, None).unwrap());
+    for chunk in chunk_at(&[-400_000, -350_000]) {
+        execution.push_raw(InputId::new(0), chunk).unwrap();
+    }
+    let output = execution.finish().unwrap();
+    assert_eq!(
+        aggregate_results(&output),
+        vec![(None, labels, vec![(0, Some(1.0))])],
+        "an instant statement answers at its one grid point"
+    );
+}
+
+/// Two metrics that share their ordinary labels are two series, not one: the presence fold keeps
+/// the name, so `absent` still sees a vector and answers nothing rather than reading the pair as one
+/// series and failing.
+///
+/// This is the shape a selector that names no metric is discovered into: one source per metric,
+/// each reporting its own name.
+#[test]
+fn absent_over_time_keeps_two_discovered_metrics_apart() {
+    let schema = raw_schema();
+    let timestamps = agg_timestamps_0_to_100();
+    let values: Vec<f64> = (1..=21).map(f64::from).collect();
+    let chunks = vec![
+        agg_raw_chunk(
+            &schema,
+            b"a",
+            agg_metric("a", &[("job", "api")]),
+            &timestamps,
+            &values,
+        ),
+        agg_raw_chunk(
+            &schema,
+            b"b",
+            agg_metric("b", &[("job", "api")]),
+            &timestamps,
+            &values,
+        ),
+    ];
+    assert!(
+        aggregate_results(&execute_compiled(
+            r#"absent_over_time({job="api"}[5m])"#,
+            chunks
+        ))
+        .is_empty(),
+        "two metrics holding samples are present, whatever labels they share"
+    );
+}
+
 /// One raw chunk of `metric`, keyed by `key`, closing the series.
 ///
 /// `timestamps` are milliseconds and `values` one per timestamp, which is what a scan of the series

src/query/src/promql/planner/native.rs

@@ -1072,6 +1072,23 @@ impl PromPlanner {
         let Some((mut prepared, logical_ids_differ, statement_kind)) = preparation? else {
             return Ok(None);
         };
+        // `absent_over_time` reads its window through a fold that keeps the metric name, so the
+        // segment that runs it reports names across a boundary. A nameless selector that discovery
+        // expands into more than one metric binds no single name for that boundary, and the fused
+        // node refuses a tree declaring `__name__` without one at physical planning. A fold that
+        // dropped the name would plan, but it would read two metrics sharing their ordinary labels
+        // as one series, so the route declines instead of planning either tree.
+        //
+        // A selector that matched no table at all is not that shape: its typed empty input holds no
+        // series, so a presence read over it answers at every point and needs no name to hand one
+        // over.
+        if reads_presence(prom_expr)
+            && prepared
+                .iter()
+                .any(|(scan, _, metric, ..)| !is_empty_discovered_input(scan) && metric.is_none())
+        {
+            return Ok(None);
+        }
         // Selectors over one table share one region, so the node runs there over every scan. When
         // they read more than one table, each scan answers from its own region and the node runs
         // above them; see `RawSeriesSource`. A metric logical table's id is part of the identity it
@@ -1869,6 +1886,28 @@ pub(crate) fn is_empty_discovered_input(plan: &LogicalPlan) -> bool {
     relation.schema.as_arrow() == frame.as_ref()
 }
 
+/// Whether `expr` asks for `absent_over_time`, whose window is read through a fold that keeps the
+/// metric name.
+fn reads_presence(expr: &PromExpr) -> bool {
+    #[derive(Default)]
+    struct Presence(bool);
+
+    impl promql_parser::util::ExprVisitor for Presence {
+        type Error = std::convert::Infallible;
+
+        fn pre_visit(&mut self, expr: &PromExpr) -> std::result::Result<bool, Self::Error> {
+            if matches!(expr, PromExpr::Call(call) if call.func.name == "absent_over_time") {
+                self.0 = true;
+            }
+            Ok(true)
+        }
+    }
+
+    let mut presence = Presence::default();
+    let _ = promql_parser::util::walk_expr(&mut presence, expr);
+    presence.0
+}
+
 /// The subtrees a [`NativeNode`] reads, in the order the node evaluates them.
 fn node_children(expr: &NativeExpr) -> Vec<&NativeNode> {
     match expr {

src/query/src/promql/planner/test.rs

@@ -2470,6 +2470,140 @@ async fn native_zero_match_takes_the_layout_of_its_set_operand() {
 /// The table is the one [`build_test_table_provider`] builds — one tag `tag_0`, the millisecond
 /// time index `timestamp`, one field `field_0` — which is the shape the selector context resolves
 /// to a raw series schema of one field.
+/// `absent_over_time` is answered natively, by `absent`'s own operation over a presence read of the
+/// window, and its labels come from the selector's matchers.
+///
+/// The existing plan answers the call with a per-series UDF, which needs a metric name to read a
+/// table from: a selector that names none - the shape `absent_over_time` is asked for when nothing
+/// matches - has no name to write, so this is the plan that has to answer it.
+#[tokio::test]
+async fn native_absent_over_time_plans_over_discovered_metrics() {
+    // Only one of the two metrics carries `no_such_label`, so the matcher that wants a value of it
+    // narrows the discovery to that one metric and the route plans. A selector that narrowed to
+    // both would decline: its presence read keeps the names, and one segment could not bind them.
+    let specs = ["alpha_total", "beta_total"]
+        .into_iter()
+        .map(|name| ((DEFAULT_SCHEMA_NAME.to_string(), name.to_string()), 2, 1))
+        .collect::<Vec<_>>();
+    let tags = vec![
+        vec!["tag_0".to_string(), "no_such_label".to_string()],
+        vec!["tag_0".to_string()],
+    ];
+    let provider = build_test_table_provider_with_tsid_tag_names(&specs, &tags, None).await;
+    let plan = build_native_fused_plan_with_provider(
+        r#"absent_over_time({no_such_label="x", tag_0="api"}[5m])"#,
+        true,
+        provider,
+    )
+    .await;
+    let display = plan.display_indent().to_string();
+    assert!(display.contains("FusedSeries"), "{display}");
+    assert!(
+        !display.contains("prom_absent_over_time"),
+        "the answer must not be the existing per-series UDF: {display}"
+    );
+    let fused = find_fused_series(&plan).expect("fused plan");
+    let tree = fused.tree().to_string();
+    assert!(
+        tree.contains(r#"root Vector absent {no_such_label="x", tag_0="api"}"#),
+        "the labels are the selector's matchers: {tree}"
+    );
+    // A statement whose window runs where its samples are is planned as one tree per stage, so the
+    // presence read is asserted across the plan rather than under the root node the frontend holds.
+    let trees = fused_trees(&plan).join("\n");
+    assert!(
+        trees.contains("Window last_over_time [300000ms]"),
+        "the presence read folds the window: {trees}"
+    );
+    // The metric name is not a label of an `absent` answer, so the output has none.
+    let names: Vec<&str> = plan
+        .schema()
+        .fields()
+        .iter()
+        .map(|field| field.name().as_str())
+        .collect();
+    for expected in ["tag_0", "no_such_label", "timestamp"] {
+        assert!(
+            names.contains(&expected),
+            "{expected} missing from {names:?}"
+        );
+    }
+    assert!(!names.contains(&METRIC_NAME), "{names:?}");
+
+    // A selector that matches no metric table at all reads the typed empty input, and the route
+    // still answers: no candidate table carries `no_such_label`, so the presence read finds nothing
+    // and `absent` reports its matchers' labels at every point.
+    let plan = build_native_fused_plan_with_provider(
+        r#"absent_over_time({no_such_label="x", tag_0="api"}[5m])"#,
+        true,
+        metric_discovery_provider(&["alpha_total", "beta_total"]).await,
+    )
+    .await;
+    let display = plan.display_indent().to_string();
+    assert!(display.contains("FusedSeries"), "{display}");
+    let fused = find_fused_series(&plan).expect("fused plan");
+    let tree = fused.tree().to_string();
+    assert!(
+        tree.contains(r#"root Vector absent {no_such_label="x", tag_0="api"}"#),
+        "a selector that matched no table still answers its matchers: {tree}"
+    );
+
+    // The same selector over metrics that all carry the label narrows to several: the presence
+    // read keeps their names and one segment cannot bind them, so the route declines rather than
+    // planning a tree that cannot be built or one that would read two metrics as one series.
+    let specs = ["alpha_total", "beta_total"]
+        .into_iter()
+        .map(|name| ((DEFAULT_SCHEMA_NAME.to_string(), name.to_string()), 2, 1))
+        .collect::<Vec<_>>();
+    let tags = vec![
+        vec!["tag_0".to_string(), "no_such_label".to_string()],
+        vec!["tag_0".to_string(), "no_such_label".to_string()],
+    ];
+    let provider = build_test_table_provider_with_tsid_tag_names(&specs, &tags, None).await;
+    let error = PromPlanner::stmt_to_plan_with_annotations(
+        provider,
+        &build_eval_stmt(r#"absent_over_time({no_such_label="x", tag_0="api"}[5m])"#),
+        &build_query_engine_state(),
+        None,
+        true,
+    )
+    .await
+    .expect_err("a nameless selector over several metrics must decline");
+    assert!(
+        error.to_string().contains("metric matcher"),
+        "the decline keeps the existing plan's answer: {error}"
+    );
+
+    // A selector that names its metric answers with that metric's other equality matchers alone.
+    let plan =
+        build_native_fused_plan(r#"absent_over_time(some_metric{tag_0="x"}[5m])"#, true).await;
+    let display = plan.display_indent().to_string();
+    assert!(display.contains("FusedSeries"), "{display}");
+    let fused = find_fused_series(&plan).expect("fused plan");
+    let tree = fused.tree().to_string();
+    assert!(
+        tree.contains(r#"root Vector absent {tag_0="x"}"#),
+        "a named metric is not a label of the answer: {tree}"
+    );
+}
+
+/// Every native tree the plan carries, outermost first: one per stage the plan splits into.
+fn fused_trees(plan: &LogicalPlan) -> Vec<String> {
+    fn visit(plan: &LogicalPlan, trees: &mut Vec<String>) {
+        if let LogicalPlan::Extension(Extension { node }) = plan
+            && let Some(fused) = node.as_any().downcast_ref::<FusedSeries>()
+        {
+            trees.push(fused.tree().to_string());
+        }
+        for input in plan.inputs() {
+            visit(input, trees);
+        }
+    }
+    let mut trees = Vec::new();
+    visit(plan, &mut trees);
+    trees
+}
+
 async fn build_native_fused_plan(query: &str, native_fused: bool) -> LogicalPlan {
     let table_provider = build_test_table_provider_with_options(
         &[(DEFAULT_SCHEMA_NAME.to_string(), "some_metric".to_string())],

tests-integration/src/tests/promql_test.rs

@@ -5650,6 +5650,183 @@ async fn native_promql_executes_fused_node_and_matches_legacy() {
     }
 }
 
+/// `promql` over `start..=end` at `step`, through the Prometheus handler.
+///
+/// The handler is the path that authorizes and plans a selector naming no metric: a `TQL EVAL`
+/// statement reaches the planner through the statement executor instead, and that path prepares no
+/// discovery scope, so it keeps declining such a selector. Its times are Prometheus timestamps, so
+/// a bare number is a count of seconds.
+async fn promql_output(
+    instance: &Arc<Instance>,
+    promql: &str,
+    start: &str,
+    end: &str,
+    step: &str,
+) -> Output {
+    let query = PromQuery {
+        query: promql.to_string(),
+        start: start.to_string(),
+        end: end.to_string(),
+        step: step.to_string(),
+        ..PromQuery::default()
+    };
+    servers::prometheus_handler::PrometheusHandler::do_query(
+        instance.as_ref(),
+        &query,
+        native_query_context(true, 4),
+    )
+    .await
+    .unwrap_or_else(|error| panic!("{promql} over {start}..={end} at {step}: {error}"))
+}
+
+/// The Prometheus v0.315 answers `absent_over_time` gives, and that the native route reports.
+///
+/// `absent_over_time` answers with the labels its selector's matchers fix at every point no series
+/// of the selector has a sample at, and with nothing at the points one does. Three metric tables
+/// make the shapes reachable: `absent_probe` carries `probe_label`, which no row's value matches, so
+/// a nameless selector narrows to it; `absent_a` and `absent_b` carry the same ordinary labels, so a
+/// nameless selector over them is one discovery expands into several metrics; and no table carries
+/// `no_such_label`, which is the reported query - the one whose selector matches no metric table and
+/// whose labels therefore have to come from the matchers alone.
+///
+/// The existing plan answers the call per series with a UDF, which needs a metric name to read a
+/// table from: such a selector writes none, and the labels still have to reach the answer at every
+/// step.
+///
+/// The statements are evaluated through the Prometheus handler, which is the path that authorizes a
+/// selector naming no metric; a bare timestamp there is a count of seconds, so the samples below sit
+/// at 0s and 60s and the evaluation times are written as the seconds they are.
+#[tokio::test(flavor = "multi_thread")]
+async fn native_promql_absent_over_time_answers_like_prometheus() {
+    common_telemetry::init_default_ut_logging();
+    let (standalone, observer) = native_test_instance(GreptimeDbStandaloneBuilder::new(
+        "native_promql_absent_over_time_e2e",
+    ))
+    .await;
+    let instance = standalone.fe_instance().clone();
+    execute_all(
+        &instance,
+        "CREATE TABLE absent_phy (ts TIMESTAMP TIME INDEX, val DOUBLE) ENGINE=metric WITH ('physical_metric_table'='true')\
+         ; CREATE TABLE absent_a (job STRING, ts TIMESTAMP TIME INDEX, val DOUBLE, PRIMARY KEY(job)) ENGINE=metric WITH ('on_physical_table'='absent_phy')\
+         ; CREATE TABLE absent_b (job STRING, ts TIMESTAMP TIME INDEX, val DOUBLE, PRIMARY KEY(job)) ENGINE=metric WITH ('on_physical_table'='absent_phy')\
+         ; CREATE TABLE absent_probe (job STRING, probe_label STRING, ts TIMESTAMP TIME INDEX, val DOUBLE, PRIMARY KEY(job, probe_label)) ENGINE=metric WITH ('on_physical_table'='absent_phy')\
+         ; INSERT INTO absent_a (job, ts, val) VALUES ('api', 0, 1), ('api', 60000, 2)\
+         ; INSERT INTO absent_b (job, ts, val) VALUES ('api', 0, 3), ('api', 60000, 4)\
+         ; INSERT INTO absent_probe (job, probe_label, ts, val) VALUES ('api', 'y', 0, 5)",
+        QueryContext::arc(),
+    )
+    .await;
+
+    let labels = |pairs: &[(&str, &str)]| {
+        pairs
+            .iter()
+            .map(|(name, value)| ((*name).to_string(), (*value).to_string()))
+            .collect::<std::collections::BTreeMap<_, _>>()
+    };
+    // Every case is the native route's answer: a shape that fell back to the existing plan would
+    // error, because that plan needs a metric name to read a table from.
+    for (promql, start, end, step, expected) in [
+        // Samples inside the window: nothing to report.
+        (
+            "absent_over_time(absent_a[5m])",
+            "120",
+            "120",
+            "1s",
+            Vec::new(),
+        ),
+        // The metric exists and holds no sample in the window: one series, and no `__name__`.
+        (
+            "absent_over_time(absent_a[5m])",
+            "400",
+            "400",
+            "1s",
+            vec![(400000, labels(&[]), 1.0)],
+        ),
+        // A named metric's other equality matcher is a label of the answer.
+        (
+            r#"absent_over_time(absent_a{job="api"}[5m])"#,
+            "400",
+            "400",
+            "1s",
+            vec![(400000, labels(&[("job", "api")]), 1.0)],
+        ),
+        // The reported query: no metric table carries `no_such_label`, so the selector matches
+        // none, and both equality matchers are the answer's labels.
+        (
+            r#"absent_over_time({no_such_label="x", job="api"}[5m])"#,
+            "120",
+            "120",
+            "1s",
+            vec![(
+                120000,
+                labels(&[("job", "api"), ("no_such_label", "x")]),
+                1.0,
+            )],
+        ),
+        // The same answer at every step of a range.
+        (
+            r#"absent_over_time({no_such_label="x", job="api"}[5m])"#,
+            "120",
+            "400",
+            "70s",
+            [120000i64, 190000, 260000, 330000, 400000]
+                .into_iter()
+                .map(|time| (time, labels(&[("job", "api"), ("no_such_label", "x")]), 1.0))
+                .collect::<Vec<_>>(),
+        ),
+        // One metric table carries `probe_label`, and no row's value matches, so the discovery
+        // narrows to that table and the presence read still finds nothing.
+        (
+            r#"absent_over_time({probe_label="x", job="api"}[5m])"#,
+            "120",
+            "120",
+            "1s",
+            vec![(120000, labels(&[("job", "api"), ("probe_label", "x")]), 1.0)],
+        ),
+    ] {
+        observer.clear();
+        let (columns, native) =
+            native_labelled_rows(promql_output(&instance, promql, start, end, step).await).await;
+        assert!(
+            observer.fused().is_some(),
+            "{promql} did not run on the native route: {columns:?} {native:?}"
+        );
+        assert_eq!(
+            native, expected,
+            "{promql} over {start}..={end} at {step}: {columns:?}"
+        );
+    }
+
+    // A nameless selector that discovery expands into more than one metric declines. The presence
+    // read keeps the metric name, and no single name is bound for the segment that hands it over; a
+    // fold that dropped the name would plan, and would read two metrics sharing their ordinary
+    // labels as one series. Neither tree is planned, so the query keeps the existing plan's answer
+    // for a selector that names no metric - an error - rather than answering wrongly.
+    observer.clear();
+    let declined = servers::prometheus_handler::PrometheusHandler::do_query(
+        instance.as_ref(),
+        &PromQuery {
+            query: r#"absent_over_time({job="api"}[5m])"#.to_string(),
+            start: "120".to_string(),
+            end: "120".to_string(),
+            step: "1s".to_string(),
+            ..PromQuery::default()
+        },
+        native_query_context(true, 4),
+    )
+    .await;
+    assert!(
+        declined.is_err(),
+        "a nameless selector over several metrics must not be answered natively yet"
+    );
+    assert!(
+        !observed_physical_nodes(&observer)
+            .iter()
+            .any(|node| node.name() == "FusedSeriesExec"),
+        "the declining shape must not run a fused node"
+    );
+}
+
 /// Compare value and response ordering for native routes added for common PromQL shapes.
 #[tokio::test(flavor = "multi_thread")]
 async fn native_promql_sort_absent_and_histogram_quantile_match_legacy() {