feat(promql): run info() natively (#10005)

* feat(promql): run info() natively

info(v, {selector}) fell back to the existing plan, which has no info
and refuses the query ("Unsupported expr type: info", or "Cannot accept
multiple vector as function input" when a selector is given).

Compile it into the native tree and run it on the info kernel:

- Plan: NativeExpr::Info { base, info, selector }. The info side is an
  instant selection reporting each selected sample's own time in
  milliseconds (InstantValue::InfoTimestamp, which replaces the
  BoundInstant.timestamp flag); validation refuses any other info side.
- Compiler: info_input derives the info selector the way Prometheus
  does: target_info or one __name__="..." matcher, the data matchers,
  and the @ and offset every selector of the first argument shares
  (subquery offsets added until an @ anchors them, none when they
  differ). Regex or negative name matchers, which can select several
  tables, __field__/__schema__/__database__ matchers and multi-field
  info tables are refused. The planner's selector walk uses the same
  helper, so both bind the info table as the last input.
- Runtime: BinaryRuntime runs the kernel as PairKernel::Info, with the
  retention and pool accounting of binary and set nodes; the kernel
  reports its workspace through info_with_reservation.
- Planner: the info table is a second table, so it is read through its
  own RawSeriesSource and the node runs on the frontend. The output adds
  the info table's tags, or only the data labels the selector names.
  Partitioned or non-Mito tables and a missing info table keep the
  existing plan.

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

* test(promql): pin region-local placement of info over its own table

info(rate(target_info[5m])) reads the info table twice. Check that the
planner binds both as direct scans under one FusedSeries node bound to
the table, that the node is placed in the region, and the columns it
reports.

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

* fix(promql): keep sample-less base series out of info's identifying patterns

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

* fix(promql): share identifying label strings in info() instead of copying them

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

---------

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

src/promql/src/extension_plan/aggregate_finalize.rs

@@ -501,6 +501,17 @@ fn split(node: &NativeNode, fields: &[FieldSpec], time_index: &str) -> DfResult<
                 _ => Err(declined("an aggregate under a set operation")),
             }
         }
+        // The suffix above a split aggregate reads the merged state alone, so it has no source for
+        // the info series.
+        NativeExpr::Info { base, info, .. } => {
+            match (
+                split(base, fields, time_index)?,
+                split(info, fields, time_index)?,
+            ) {
+                (None, None) => Ok(None),
+                _ => Err(declined("an aggregate under info")),
+            }
+        }
     }
 }
 
@@ -527,9 +538,13 @@ fn prefix_fields(tree: &NativeTree) -> DfResult<Vec<FieldSpec>> {
             | NativeExpr::Timestamp { input }
             | NativeExpr::Aggregate { input, .. }
             | NativeExpr::Vector { input, .. } => raw_fields(input),
-            NativeExpr::Binary { left, right, .. } | NativeExpr::Set { left, right, .. } => {
-                raw_fields(left).or_else(|| raw_fields(right))
-            }
+            NativeExpr::Binary { left, right, .. }
+            | NativeExpr::Set { left, right, .. }
+            | NativeExpr::Info {
+                base: left,
+                info: right,
+                ..
+            } => raw_fields(left).or_else(|| raw_fields(right)),
         }
     }
 

src/promql/src/extension_plan/aggregate_finalize/tests.rs

@@ -91,7 +91,7 @@ fn instant() -> NativeNode {
                 lookback_ms: 300_000,
                 offset_ms: 0,
                 at_ms: None,
-                timestamp: false,
+                value: crate::native::plan::InstantValue::Sample,
                 output_grid: grid(),
             },
         },
@@ -619,7 +619,13 @@ fn the_prefix_and_the_suffix_each_name_their_own_source_zero() {
             | NativeExpr::Timestamp { input }
             | NativeExpr::Aggregate { input, .. }
             | NativeExpr::Vector { input, .. } => inputs(input, found),
-            NativeExpr::Binary { left, right, .. } | NativeExpr::Set { left, right, .. } => {
+            NativeExpr::Binary { left, right, .. }
+            | NativeExpr::Set { left, right, .. }
+            | NativeExpr::Info {
+                base: left,
+                info: right,
+                ..
+            } => {
                 inputs(left, found);
                 inputs(right, found);
             }

src/promql/src/extension_plan/fused_series.rs

@@ -655,7 +655,13 @@ fn tree_inputs(tree: &NativeTree) -> Vec<&crate::native::plan::BoundInput> {
             | NativeExpr::Timestamp { input }
             | NativeExpr::Aggregate { input, .. }
             | NativeExpr::Vector { input, .. } => collect(input, inputs),
-            NativeExpr::Binary { left, right, .. } | NativeExpr::Set { left, right, .. } => {
+            NativeExpr::Binary { left, right, .. }
+            | NativeExpr::Set { left, right, .. }
+            | NativeExpr::Info {
+                base: left,
+                info: right,
+                ..
+            } => {
                 collect(left, inputs);
                 collect(right, inputs);
             }

src/promql/src/extension_plan/fused_series/tests.rs

@@ -89,7 +89,7 @@ use crate::map_label::LabelRewrite;
 use crate::native::execute::{InputId, NativeExecutionState, NativeExecutor};
 use crate::native::plan::{
     AggregateKind, AggregateStage, BoundAggregate, BoundInput, BoundInstant, BoundMap, BoundScalar,
-    BoundWindow, Grouping, MapOp, NativeExpr, NativeNode, NativeTree, NativeType,
+    BoundWindow, Grouping, InstantValue, MapOp, NativeExpr, NativeNode, NativeTree, NativeType,
 };
 use crate::range_eval::{RangeFunction, RangeWindow};
 use crate::vector_match::{BinaryMatching, BinaryOp, Cardinality, FillValues, MatchingLabels};
@@ -249,7 +249,7 @@ fn binary_tree_over_sources(
                 lookback_ms: 0,
                 offset_ms: 0,
                 at_ms: None,
-                timestamp: false,
+                value: InstantValue::Sample,
                 output_grid: Arc::clone(&grid),
             },
         },
@@ -269,7 +269,7 @@ fn binary_tree_over_sources(
                 lookback_ms: 0,
                 offset_ms: 0,
                 at_ms: None,
-                timestamp: false,
+                value: InstantValue::Sample,
                 output_grid: Arc::clone(&grid),
             },
         },
@@ -321,7 +321,7 @@ fn instant_of(source_index: usize, grid: &Arc<EvalGrid>) -> NativeNode {
                 lookback_ms: 0,
                 offset_ms: 0,
                 at_ms: None,
-                timestamp: false,
+                value: InstantValue::Sample,
                 output_grid: Arc::clone(grid),
             },
         },
@@ -362,7 +362,13 @@ fn declaring_orderings(tree: &NativeTree, orderings: &[SeriesOrdering]) -> Nativ
             | NativeExpr::Vector { input, .. } => {
                 **input = rewrite(input, orderings, seen);
             }
-            NativeExpr::Binary { left, right, .. } | NativeExpr::Set { left, right, .. } => {
+            NativeExpr::Binary { left, right, .. }
+            | NativeExpr::Set { left, right, .. }
+            | NativeExpr::Info {
+                base: left,
+                info: right,
+                ..
+            } => {
                 **left = rewrite(left, orderings, seen);
                 **right = rewrite(right, orderings, seen);
             }

src/promql/src/extension_plan/fused_series/tests/frame_output.rs

@@ -15,7 +15,7 @@
 use super::*;
 use crate::map::MathOp;
 use crate::map_label::LabelRewrite;
-use crate::native::plan::{BoundMap, MapOp};
+use crate::native::plan::{BoundMap, InstantValue, MapOp};
 
 fn instant_tree(schema: Arc<SeriesSchema>, grid: Arc<EvalGrid>) -> NativeTree {
     let output = NativeType::Samples(Arc::new(SeriesSchema {
@@ -36,7 +36,7 @@ fn instant_tree(schema: Arc<SeriesSchema>, grid: Arc<EvalGrid>) -> NativeTree {
                 lookback_ms: 0,
                 offset_ms: 0,
                 at_ms: None,
-                timestamp: false,
+                value: InstantValue::Sample,
                 output_grid: grid,
             },
         },

src/promql/src/info.rs

@@ -29,14 +29,15 @@ use std::collections::{BTreeMap, BTreeSet, HashMap};
 use std::sync::Arc;
 
 use datafusion::error::{DataFusionError, Result};
-use datatypes::series::SeriesLabels;
 use datatypes::series::samples::{SampleRef, Series, SeriesSchema};
+use datatypes::series::{Label, SeriesLabels};
 use regex::Regex;
 
 use crate::grid_series::{
     GridSample, GridSeriesBuilder, LabelSet, external, label_set, owned, render_label_set,
 };
 use crate::select::MatchOp;
+use crate::vector_match::{Budget, HISTOGRAM_ROW_BYTES, owned_histogram_bytes};
 
 /// The labels an info series is joined on, as in Prometheus.
 const IDENTIFYING_LABELS: [&str; 2] = ["instance", "job"];
@@ -70,6 +71,33 @@ pub fn info(
     base: &[Series],
     info: &[Series],
     selector: &InfoSelector,
+) -> Result<Vec<Series>> {
+    info_inner(schema, base, info, selector, None)
+}
+
+/// [`info()`], handing the live workspace it holds to `reserve` before it allocates it.
+///
+/// What the index of the info series and the rows of the results keep is held until the operation
+/// returns; the merge of one label set's rows is a phase, and the builder's columns join the phase
+/// of the output they are finished as. A total `reserve` refuses ends the operation with that error.
+pub(crate) fn info_with_reservation(
+    schema: &Arc<SeriesSchema>,
+    base: &[Series],
+    info: &[Series],
+    selector: &InfoSelector,
+    reserve: &mut dyn FnMut(usize) -> Result<()>,
+) -> Result<Vec<Series>> {
+    let mut budget = Budget::new(reserve);
+    info_inner(schema, base, info, selector, Some(&mut budget))
+}
+
+/// The one info kernel, whether or not a caller accounts for its workspace.
+fn info_inner(
+    schema: &Arc<SeriesSchema>,
+    base: &[Series],
+    info: &[Series],
+    selector: &InfoSelector,
+    mut budget: Option<&mut Budget<'_>>,
 ) -> Result<Vec<Series>> {
     let grid = schema
         .eval_grid
@@ -96,13 +124,21 @@ pub fn info(
     }
     let name_matchers = effective_name_matchers(&selector.name_matchers)?;
     let is_info_metric = |name: &str| name_matchers.iter().all(|matcher| matcher.matches(name));
+    let width = schema.fields.len();
 
+    // The presence of the base at each point and the identifying values of every base series
+    // stay live until the operation returns.
+    if let Some(budget) = budget.as_deref_mut() {
+        budget.hold(points.saturating_add(base.len().saturating_mul(PATTERN_BYTES)))?;
+    }
     let mut base_present = vec![false; points];
-    for series in base {
+    let mut sampled = vec![false; base.len()];
+    for (series, sampled) in base.iter().zip(&mut sampled) {
         for field in series.fields.iter() {
             for ((point, _), sample) in series.time.positions().zip(field.iter()) {
                 if !matches!(sample, SampleRef::Missing) {
                     base_present[point] = true;
+                    *sampled = true;
                 }
             }
         }
@@ -114,8 +150,10 @@ pub fn info(
     // base series it enriches: a present label matches any value the pattern's base series have,
     // an absent one matches only absence. An info series no pattern admits is never fetched, so
     // it neither enriches nor conflicts.
+    // A base series without a sample is not in Prometheus' base matrix at all, so it selects no
+    // info series.
     let mut patterns: HashMap<Vec<bool>, Vec<BTreeSet<Arc<str>>>> = HashMap::new();
-    for series in base {
+    for (series, _) in base.iter().zip(&sampled).filter(|(_, sampled)| **sampled) {
         if is_info_metric(series.metric.name().unwrap_or("")) {
             continue;
         }
@@ -146,6 +184,17 @@ pub fn info(
         if !fetched {
             continue;
         }
+        // The entry of a signature and the newest series it keeps at each point the series
+        // holds a sample, charged before they are allocated.
+        if let Some(budget) = budget.as_deref_mut() {
+            budget.hold(
+                series
+                    .time
+                    .len()
+                    .saturating_mul(KEPT_POINT_BYTES)
+                    .saturating_add(SIGNATURE_BYTES),
+            )?;
+        }
         let name: Arc<str> = Arc::from(series.metric.name().unwrap_or(""));
         info_names.insert(name.clone());
         let kept = by_signature.entry((name, values)).or_default();
@@ -185,7 +234,6 @@ pub fn info(
         }
     }
 
-    let width = schema.fields.len();
     let mut result_index: HashMap<LabelSet, usize> = HashMap::new();
     let mut results: Vec<(SeriesLabels, Vec<Row>)> = Vec::new();
     let mut added: Vec<(Arc<str>, Arc<str>)> = Vec::new();
@@ -205,13 +253,21 @@ pub fn info(
         };
         // The result of the previous point, reused while the same info series are kept.
         let mut previous: Option<Choice> = None;
+        // A row of this series is kept by the result it goes to, so every row it can have is held
+        // before the walk; a histogram is held as it is copied.
+        if let Some(budget) = budget.as_deref_mut() {
+            budget.hold(series.time.len().saturating_mul(row_bytes(width)))?;
+        }
         let mut cursors: Vec<_> = series.fields.iter().map(|field| field.cursor()).collect();
         for (point, row) in series.time.positions() {
             let mut values = Vec::with_capacity(width);
             for cursor in &mut cursors {
-                values.push(owned(
-                    cursor.at(row).expect("the axis and the fields are aligned"),
-                )?);
+                let sample = cursor.at(row).expect("the axis and the fields are aligned");
+                if let (Some(budget), SampleRef::Histogram(view)) = (budget.as_deref_mut(), &sample)
+                {
+                    budget.hold(owned_histogram_bytes(view))?;
+                }
+                values.push(owned(sample)?);
             }
             if values.iter().all(Option::is_none) {
                 continue;
@@ -246,6 +302,9 @@ pub fn info(
                     Some(match result_index.entry(label_set(&metric)) {
                         std::collections::hash_map::Entry::Occupied(entry) => *entry.get(),
                         std::collections::hash_map::Entry::Vacant(entry) => {
+                            if let Some(budget) = budget.as_deref_mut() {
+                                budget.hold(result_set_bytes(&metric))?;
+                            }
                             results.push((metric, Vec::new()));
                             *entry.insert(results.len() - 1)
                         }
@@ -270,6 +329,10 @@ pub fn info(
     // Rows of one label set arrive per base series; sorted by point, a field held twice at one
     // point is a collision.
     for (metric, rows) in &mut results {
+        // The sort's buffer and the merged rows, while the rows they replace are still alive.
+        if let Some(budget) = budget.as_deref_mut() {
+            budget.phase(rows.len().saturating_mul(2 * std::mem::size_of::<Row>()))?;
+        }
         rows.sort_by_key(|(point, _)| *point);
         let mut merged: Vec<Row> = Vec::with_capacity(rows.len());
         for (point, values) in rows.drain(..) {
@@ -301,8 +364,17 @@ pub fn info(
 
     let mut builder = GridSeriesBuilder::try_new(schema.clone())?;
     let mut output = Vec::with_capacity(results.len());
+    if let Some(budget) = budget.as_deref_mut() {
+        budget.phase(0)?;
+    }
     for (metric, rows) in results {
         builder.begin_series();
+        // This series' columns, reserved before its first row. They stay live inside the series
+        // it is finished as, so the phase keeps them.
+        if let Some(budget) = budget.as_deref_mut() {
+            budget.grow(column_bytes(&rows, width))?;
+        }
+        builder.reserve_rows(rows.len())?;
         for (point, row) in rows {
             builder.push_samples(point, &row)?;
         }
@@ -314,6 +386,66 @@ pub fn info(
 /// The samples of one result series at one grid point, one per field.
 type Row = (usize, Vec<Option<GridSample>>);
 
+/// The identifying values one base series adds to its presence pattern, charged twice for the
+/// spare room of the sets they go to.
+const PATTERN_BYTES: usize = 2 * IDENTIFYING_LABELS.len() * std::mem::size_of::<Arc<str>>();
+
+/// The entry of one info series' signature: its key, its map of kept series, and the name the
+/// info metric list keeps, charged twice for the spare room of the tables.
+const SIGNATURE_BYTES: usize = 2
+    * (std::mem::size_of::<(Signature, BTreeMap<usize, (f64, usize)>)>()
+        + IDENTIFYING_LABELS.len() * std::mem::size_of::<Option<Arc<str>>>()
+        + std::mem::size_of::<Arc<str>>());
+
+/// The series one signature keeps at one point, charged twice for the B-tree's spare room.
+const KEPT_POINT_BYTES: usize = 2 * std::mem::size_of::<(usize, (f64, usize))>();
+
+/// One row a result keeps: its entry in the result's row list, charged twice for the list's
+/// growth, and its samples, allocated at their exact count.
+fn row_bytes(width: usize) -> usize {
+    (2 * std::mem::size_of::<Row>())
+        .saturating_add(width.saturating_mul(std::mem::size_of::<Option<GridSample>>()))
+}
+
+/// The bytes a result label set keeps: its metric, rebuilt by each label it is given, the label set
+/// it is filed under with the copy of the name that set makes, and its entries in the result list
+/// and index.
+fn result_set_bytes(metric: &SeriesLabels) -> usize {
+    let labels = metric.tags().len().saturating_add(1);
+    labels
+        .saturating_mul(
+            2 * std::mem::size_of::<Label>() + std::mem::size_of::<(Arc<str>, Arc<str>)>(),
+        )
+        .saturating_add(metric.name().map_or(0, str::len))
+        .saturating_add(
+            metric
+                .tags()
+                .iter()
+                .map(|tag| tag.name.len())
+                .sum::<usize>(),
+        )
+        .saturating_add(2 * std::mem::size_of::<(SeriesLabels, Vec<Row>)>())
+        .saturating_add(2 * std::mem::size_of::<(LabelSet, usize)>())
+}
+
+/// The bytes one result series' columns take: per row, the grid index, the value and validity
+/// column of every field, charged twice for the capacity each grows to, and a histogram index; per
+/// histogram, a row of the builder's histogram table and the buckets copied into it.
+fn column_bytes(rows: &[Row], width: usize) -> usize {
+    let row = std::mem::size_of::<usize>()
+        + 2 * width.saturating_mul(std::mem::size_of::<f64>() + 1)
+        + width.saturating_mul(std::mem::size_of::<Option<u32>>());
+    rows.iter().flat_map(|(_, values)| values).fold(
+        rows.len().saturating_mul(row),
+        |bytes, value| match value {
+            Some(GridSample::Histogram(histogram)) => bytes
+                .saturating_add(HISTOGRAM_ROW_BYTES)
+                .saturating_add(histogram.heap_bytes()),
+            _ => bytes,
+        },
+    )
+}
+
 /// What the info series kept at a point make of one base series.
 struct Choice {
     /// The info series, one per info metric that has one.
@@ -346,14 +478,19 @@ impl FirstError {
 type Signature = (Arc<str>, Vec<Option<Arc<str>>>);
 
 /// The value of each identifying label, `None` where the series does not have it.
+///
+/// The values share the series' own label strings, which the retained operands already hold and
+/// charge, so the patterns and signatures built from them allocate no label text of their own.
 fn identifying_values(metric: &SeriesLabels) -> Vec<Option<Arc<str>>> {
     IDENTIFYING_LABELS
         .iter()
         .map(|label| {
             metric
-                .get(label)
-                .filter(|value| !value.is_empty())
-                .map(Arc::from)
+                .tags()
+                .iter()
+                .find(|tag| tag.name.as_ref() == *label)
+                .filter(|tag| !tag.value.is_empty())
+                .map(|tag| Arc::clone(&tag.value))
         })
         .collect()
 }

src/promql/src/info/tests.rs

@@ -239,6 +239,54 @@ fn results_with_one_label_set_merge_by_field() {
     assert!(error.to_string().contains("same labelset"), "{error}");
 }
 
+/// A base series a filter emptied, such as `b` in `info(up > 0)`, selects no info series: its two
+/// info series would otherwise collide for the series that does have a sample.
+#[test]
+fn a_base_series_without_samples_selects_no_info_series() {
+    let schema = grid_schema(1);
+    let labels = |pairs: &[(&str, &str)]| -> Labels {
+        pairs
+            .iter()
+            .map(|(name, value)| (name.to_string(), value.to_string()))
+            .collect()
+    };
+    let kept = one_point(
+        &schema,
+        &[("__name__", "up"), ("instance", "a"), ("job", "j")],
+        vec![Some(GridSample::Float(1.0))],
+    );
+    let filtered = sparse_series(
+        &schema,
+        &labels(&[("__name__", "up"), ("instance", "b"), ("job", "j")]),
+        &Points::new(),
+    )
+    .unwrap();
+    let info_of = |version| {
+        one_point(
+            &schema,
+            &[
+                ("__name__", "target_info"),
+                ("instance", "b"),
+                ("job", "j"),
+                ("version", version),
+            ],
+            vec![Some(GridSample::Float(1.0))],
+        )
+    };
+    let result = info(
+        &schema,
+        &[kept, filtered],
+        &[info_of("v1"), info_of("v2")],
+        &all_labels(),
+    )
+    .unwrap();
+    assert_eq!(result.len(), 1);
+    assert_eq!(
+        labels_of(&result[0].metric),
+        labels(&[("__name__", "up"), ("instance", "a"), ("job", "j")])
+    );
+}
+
 /// A native histogram base sample is enriched like a float one.
 #[test]
 fn histogram_base_samples_are_enriched() {

src/promql/src/native/display.rs

@@ -27,11 +27,11 @@ use datatypes::series::{SeriesLabels, SeriesOrdering};
 use crate::histogram_eval::HistogramFunction;
 use crate::map::MathOp;
 use crate::native::plan::{
-    AggregateKind, AggregateStage, BoundVectorOp, BoundWindow, Grouping, MapOp, NativeExpr,
-    NativeNode, NativeTree, NativeType,
+    AggregateKind, AggregateStage, BoundVectorOp, BoundWindow, Grouping, InstantValue, MapOp,
+    NativeExpr, NativeNode, NativeTree, NativeType,
 };
 use crate::range_eval::RangeFunction;
-use crate::select::SortOrder;
+use crate::select::{MatchOp, SortOrder};
 use crate::vector_match::{BinaryOp, Cardinality, FillValues, MatchingLabels, SetOp};
 
 /// The header line of every tree.
@@ -98,8 +98,10 @@ fn describe(node: &NativeNode) -> String {
             if let Some(at) = op.at_ms {
                 text.push_str(&format!(" @ {at}ms"));
             }
-            if op.timestamp {
-                text.push_str(" timestamp");
+            match op.value {
+                InstantValue::Sample => {}
+                InstantValue::Timestamp => text.push_str(" timestamp"),
+                InstantValue::InfoTimestamp => text.push_str(" info timestamp"),
             }
             text
         }
@@ -174,6 +176,29 @@ fn describe(node: &NativeNode) -> String {
             },
             matching_labels(&op.labels)
         ),
+        NativeExpr::Info { selector, .. } => format!(
+            "names [{}] data [{}]{}",
+            selector
+                .name_matchers
+                .iter()
+                .map(|(op, value)| format!(
+                    "__name__{}\"{value}\"",
+                    match op {
+                        MatchOp::Equal => "=",
+                        MatchOp::NotEqual => "!=",
+                        MatchOp::Regex => "=~",
+                        MatchOp::NotRegex => "!~",
+                    }
+                ))
+                .collect::<Vec<_>>()
+                .join(", "),
+            label_list(&selector.data_labels),
+            if selector.keep_unmatched {
+                " keep unmatched"
+            } else {
+                ""
+            }
+        ),
     }
 }
 
@@ -365,6 +390,7 @@ fn kind(expr: &NativeExpr) -> &'static str {
         NativeExpr::Vector { .. } => "Vector",
         NativeExpr::Binary { .. } => "Binary",
         NativeExpr::Set { .. } => "Set",
+        NativeExpr::Info { .. } => "Info",
     }
 }
 
@@ -394,6 +420,10 @@ fn display_node(
             display_node(f, left, depth + 1, "left")?;
             display_node(f, right, depth + 1, "right")?;
         }
+        NativeExpr::Info { base, info, .. } => {
+            display_node(f, base, depth + 1, "base")?;
+            display_node(f, info, depth + 1, "info")?;
+        }
     }
     Ok(())
 }

src/promql/src/native/display/tests.rs

@@ -24,7 +24,7 @@ use super::super::plan::{
     NativeType,
 };
 use crate::map::MathOp;
-use crate::native::plan::MapOp;
+use crate::native::plan::{InstantValue, MapOp};
 use crate::range_eval::{RangeFunction, RangeModifier, RangeWindow};
 use crate::select::SortOrder;
 use crate::vector_match::{
@@ -269,7 +269,7 @@ fn a_scalar_and_an_instant_line_show_what_they_hold() {
                 lookback_ms: 300_000,
                 offset_ms: 60_000,
                 at_ms: None,
-                timestamp: false,
+                value: InstantValue::Sample,
                 output_grid: Arc::clone(&out),
             },
         },
@@ -465,7 +465,7 @@ fn instant_at_and_timestamp_are_displayed() {
                 lookback_ms: 300_000,
                 offset_ms: 0,
                 at_ms: Some(1_234),
-                timestamp: true,
+                value: InstantValue::Timestamp,
                 output_grid: Arc::clone(&out),
             },
         },

src/promql/src/native/execute.rs

@@ -41,8 +41,10 @@ use crate::grid_series::{
     series_bytes, table_bytes,
 };
 use crate::histogram_eval::{evaluate_histograms, map_native_histogram};
+use crate::info::{InfoSelector, info_with_reservation};
 use crate::instant_eval::{
-    select_instant_at, select_instants, timestamp_of_evaluated, timestamp_of_selection,
+    info_sample_times, select_instant_at, select_instants, timestamp_of_evaluated,
+    timestamp_of_selection,
 };
 use crate::map::{
     ArithmeticOp, Comparison, ComparisonOp, ScalarOperand, drop_metric_name, map_arithmetic,
@@ -52,8 +54,8 @@ use crate::native::aggregate::AggregateRuntime;
 use crate::native::batch::EvaluatedChunk;
 use crate::native::metrics::NodeMetrics;
 use crate::native::plan::{
-    BoundInstant, BoundScalar, BoundSetOp, BoundVectorOp, Grouping, MapOp, NativeExpr, NativeNode,
-    NativeTree, NativeType,
+    BoundInstant, BoundScalar, BoundSetOp, BoundVectorOp, Grouping, InstantValue, MapOp,
+    NativeExpr, NativeNode, NativeTree, NativeType,
 };
 #[cfg(test)]
 pub(crate) use crate::native::runs::absorb_rebuilds;
@@ -967,7 +969,9 @@ impl InstantRuntime {
         // The grid's own points are what the selection decides, so the range and the step it is
         // asked for are the grid's: `start..=start + (len - 1) * step`, step by step.
         if let Some(at_ms) = self.selection.at_ms {
-            let timestamp_window = if self.selection.timestamp && self.selection.lookback_ms >= 0 {
+            let timestamp_window = if self.selection.value == InstantValue::Timestamp
+                && self.selection.lookback_ms >= 0
+            {
                 let anchor = at_ms as i128;
                 let shifted = anchor - self.selection.offset_ms as i128;
                 let lower = anchor.max(shifted) - self.selection.lookback_ms as i128;
@@ -992,38 +996,23 @@ impl InstantRuntime {
                 &fields,
                 selection_at,
                 selection_lookback,
-                if self.selection.timestamp {
+                if self.selection.value == InstantValue::Timestamp {
                     0
                 } else {
                     self.selection.offset_ms
                 },
             )?
             else {
-                let metric = if self.selection.timestamp {
-                    drop_metric_name(&metric)?
-                } else {
-                    metric
-                };
-                return no_row_chunk(&self.schema, metric);
+                return no_row_chunk(&self.schema, self.output_metric(&metric)?);
             };
             let grid_len = usize::try_from(spec.len).unwrap_or(usize::MAX);
-            if self.selection.timestamp {
+            if self.selection.value != InstantValue::Sample {
                 let indices = vec![index; grid_len];
                 let timestamps = grid
                     .timestamps()
                     .map_err(crate::grid_series::external)?
                     .to_vec();
-                return match timestamp_of_selection(
-                    &self.schema,
-                    &metric,
-                    &time,
-                    &fields,
-                    &indices,
-                    &timestamps,
-                )? {
-                    Some(series) => EvaluatedChunk::try_new(series, true),
-                    None => no_row_chunk(&self.schema, drop_metric_name(&metric)?),
-                };
+                return self.project(&metric, &time, &fields, &indices, &timestamps);
             }
             self.output.begin_series();
             let mut cursors: Vec<RowCursor<'_>> = fields.iter().map(Samples::cursor).collect();
@@ -1051,18 +1040,8 @@ impl InstantRuntime {
             self.selection.lookback_ms,
             self.selection.offset_ms,
         )?;
-        if self.selection.timestamp {
-            return match timestamp_of_selection(
-                &self.schema,
-                &metric,
-                &time,
-                &fields,
-                &indices,
-                &timestamps,
-            )? {
-                Some(series) => EvaluatedChunk::try_new(series, true),
-                None => no_row_chunk(&self.schema, drop_metric_name(&metric)?),
-            };
+        if self.selection.value != InstantValue::Sample {
+            return self.project(&metric, &time, &fields, &indices, &timestamps);
         }
         self.output.begin_series();
         // One cursor per field, so a row is read as the field holds it rather than by walking the
@@ -1095,6 +1074,42 @@ impl InstantRuntime {
             None => no_row_chunk(&self.schema, metric),
         }
     }
+
+    /// The labels this selection reports a series under: `timestamp(x)` drops the metric name.
+    fn output_metric(&self, metric: &SeriesLabels) -> Result<SeriesLabels> {
+        match self.selection.value {
+            InstantValue::Timestamp => drop_metric_name(metric),
+            InstantValue::Sample | InstantValue::InfoTimestamp => Ok(metric.clone()),
+        }
+    }
+
+    /// The sample times of a selection, for a selection that reports them rather than the samples.
+    fn project(
+        &self,
+        metric: &SeriesLabels,
+        time: &RawTimestamps,
+        fields: &SeriesFields,
+        indices: &[u64],
+        timestamps: &[i64],
+    ) -> Result<EvaluatedChunk> {
+        let projected = match self.selection.value {
+            InstantValue::Timestamp => {
+                timestamp_of_selection(&self.schema, metric, time, fields, indices, timestamps)?
+            }
+            InstantValue::InfoTimestamp => {
+                info_sample_times(&self.schema, metric, time, fields, indices, timestamps)?
+            }
+            InstantValue::Sample => {
+                return Err(DataFusionError::Internal(
+                    "an instant selection of samples has no sample times to project".to_string(),
+                ));
+            }
+        };
+        match projected {
+            Some(series) => EvaluatedChunk::try_new(series, true),
+            None => no_row_chunk(&self.schema, self.output_metric(metric)?),
+        }
+    }
 }
 
 /// An input a node reads whole before it computes -- one operand of a binary node, or the input of a
@@ -1430,6 +1445,8 @@ enum PairKernel {
     Binary(Arc<BinaryMatching>),
     /// `and`, `or` or `unless`, which keeps samples of either side as they are.
     Set(BoundSetOp),
+    /// `info`, whose left operand is the base and right operand the info series' sample times.
+    Info(InfoSelector),
 }
 
 impl PairKernel {
@@ -1438,6 +1455,7 @@ impl PairKernel {
         match self {
             Self::Binary(_) => "the binary node",
             Self::Set(_) => "the set node",
+            Self::Info(_) => "the info node",
         }
     }
 }
@@ -1644,6 +1662,11 @@ impl BinaryRuntime {
                     right,
                     &mut reserve,
                 ),
+                // The result keeps the base's fields on the base's grid, so the base's layout is
+                // the one the kernel builds it on.
+                PairKernel::Info(selector) => {
+                    info_with_reservation(left_schema, left, right, selector, &mut reserve)
+                }
             }
         };
         // The operands and the assembled series have been handed to the kernel and the reservation
@@ -2304,13 +2327,13 @@ impl RuntimeNode {
                                 Arc::clone(&metrics),
                                 pool,
                             );
-                            if op.timestamp {
+                            if op.value == InstantValue::Timestamp {
                                 state.emitted = Some(Emitted::default());
                             }
                             state
                         },
                     },
-                    shared || op.timestamp,
+                    shared || op.value == InstantValue::Timestamp,
                 )
             }
             NativeExpr::Merge { input } => {
@@ -2505,6 +2528,39 @@ impl RuntimeNode {
                     false,
                 )
             }
+            NativeExpr::Info {
+                base,
+                info,
+                selector,
+            } => {
+                let (Some(base_schema), Some(info_schema)) =
+                    (base.output.samples(), info.output.samples())
+                else {
+                    return Err(DataFusionError::Execution(
+                        "info consumes two vectors".to_string(),
+                    ));
+                };
+                let runtime = BinaryRuntime::new(
+                    PairKernel::Info(selector.clone()),
+                    [Arc::clone(base_schema), Arc::clone(info_schema)],
+                );
+                let (base_node, _) = Self::build(base, pool, inputs, format!("{location}.base"))?;
+                let (info_node, _) = Self::build(info, pool, inputs, format!("{location}.info"))?;
+                // The kernel merges the results that share a label set, so its output shares none.
+                (
+                    RuntimeOp::Binary {
+                        left: Box::new(base_node),
+                        right: Box::new(info_node),
+                        binary: Some(runtime),
+                        state: NodeState::new(
+                            format!("native.info at {location}"),
+                            Arc::clone(&metrics),
+                            pool,
+                        ),
+                    },
+                    false,
+                )
+            }
             NativeExpr::Aggregate { input, op } => {
                 let schema = node
                     .output
@@ -3987,7 +4043,7 @@ impl NodeState {
         if let Some(output) = &output {
             metrics.record_grid_points(output.series.time.len());
             metrics.record_output(1, valid_values(output));
-            if instant.selection.timestamp {
+            if instant.selection.value == InstantValue::Timestamp {
                 self.track(output)?;
             }
         }

src/promql/src/native/execute/tests.rs

@@ -48,14 +48,15 @@ use super::{
 use crate::evaluated::EvaluatedChunk;
 use crate::grid_series::merge_same_labelset;
 use crate::histogram_eval::{HistogramFunction, NativeHistogramFunction};
+use crate::info::InfoSelector;
 use crate::instant_eval::select_instants;
 use crate::map::MathOp;
 use crate::map_label::LabelRewrite;
 use crate::native::metrics::{NodeMetric, NodeMetrics};
 use crate::native::plan::{
     AggregateKind, AggregateStage, AggregateStateSchema, BoundAggregate, BoundInput, BoundInstant,
-    BoundMap, BoundScalar, BoundSetOp, BoundVectorOp, BoundWindow, Grouping, MapOp, NativeExpr,
-    NativeNode, NativeTree, NativeType,
+    BoundMap, BoundScalar, BoundSetOp, BoundVectorOp, BoundWindow, Grouping, InstantValue, MapOp,
+    NativeExpr, NativeNode, NativeTree, NativeType,
 };
 use crate::range_eval::{RangeEvaluator, RangeFunction, RangeWindow};
 use crate::select::SortOrder;
@@ -262,7 +263,7 @@ fn instant(
                 lookback_ms,
                 offset_ms,
                 at_ms: None,
-                timestamp: false,
+                value: InstantValue::Sample,
                 output_grid: Arc::clone(&output_grid),
             },
         },
@@ -280,7 +281,7 @@ fn timestamp_instant_rejects_negative_lookback() {
     let NativeExpr::Instant { op, .. } = &mut root.expr else {
         unreachable!();
     };
-    op.timestamp = true;
+    op.value = InstantValue::Timestamp;
     op.at_ms = Some(500);
     let tree = NativeTree::try_new(root).unwrap();
     let pool = pool();
@@ -7985,6 +7986,181 @@ fn queued_set_results_hold_their_labels_until_handed_on() {
     assert_eq!(dropped.reserved(), 0);
 }
 
+// ---------------------------------------------------------------------------
+// `info`, driven through `NativeExecutor`.
+// ---------------------------------------------------------------------------
+
+/// `name` with `labels`.
+fn labelled(name: &str, labels: &[(&str, &str)]) -> SeriesLabels {
+    SeriesLabels::try_new(
+        Some(name.into()),
+        labels
+            .iter()
+            .map(|(name, value)| Label {
+                name: (*name).into(),
+                value: (*value).into(),
+            })
+            .collect(),
+    )
+    .unwrap()
+}
+
+/// `info` over evaluated base chunks at source 0 and the raw info series at source 1, selected the
+/// way the compiler selects them: within `lookback_ms`, reporting each selected sample's time.
+fn info_tree(grid: &Arc<EvalGrid>, lookback_ms: i64, selector: InfoSelector) -> NativeTree {
+    let base = evaluated_input_with_source_index(Arc::clone(grid), 0);
+    let mut info = instant(
+        raw_input_with_source_index_and_ordering(1, SeriesOrdering::KeyThenTime),
+        lookback_ms,
+        0,
+        Arc::clone(grid),
+    );
+    let NativeExpr::Instant { op, .. } = &mut info.expr else {
+        unreachable!("`instant` builds an instant selection");
+    };
+    op.value = InstantValue::InfoTimestamp;
+    let output = base.output.clone();
+    NativeTree::try_new(NativeNode {
+        expr: NativeExpr::Info {
+            base: Box::new(base),
+            info: Box::new(info),
+            selector,
+        },
+        output,
+    })
+    .unwrap()
+}
+
+/// A base series takes the data labels of the newest info series sharing its identifying labels at
+/// each point, which the runtime can only tell from the sample times its info side reports, and a
+/// base series no info series matches is dropped when the selector's data matcher needs a value.
+///
+/// `version="v1"` is sampled at 0s and 10s and `version="v2"` at 15s and 30s, read within 15s: the
+/// point at 20s sees both and keeps `v2`, the newer one.
+#[test]
+fn info_labels_each_point_from_the_newest_info_series() {
+    let grid = grid(0, 10_000, 4);
+    let schema = evaluated_schema(Arc::clone(&grid));
+    let tree = info_tree(
+        &grid,
+        15_000,
+        InfoSelector {
+            name_matchers: Vec::new(),
+            data_labels: vec!["version".into()],
+            keep_unmatched: false,
+        },
+    );
+    let mut execution = executor(&tree, pool());
+    let base = execution.state().input_ids(0).next().unwrap();
+    let info = execution.state().input_ids(1).next().unwrap();
+    for (instance, values) in [("a", [1.0, 2.0, 3.0, 4.0]), ("b", [5.0, 6.0, 7.0, 8.0])] {
+        execution
+            .push_evaluated(
+                base,
+                evaluated_chunk_of(
+                    &schema,
+                    labelled("up", &[("instance", instance), ("job", "j")]),
+                    0,
+                    values.into_iter().map(Some).collect(),
+                    true,
+                ),
+            )
+            .unwrap();
+    }
+    for (key, version, timestamps) in [
+        (b"info-v1", "v1", [0, 10_000]),
+        (b"info-v2", "v2", [15_000, 30_000]),
+    ] {
+        execution
+            .push_raw(
+                info,
+                raw_chunk_of(
+                    &raw_schema(),
+                    key,
+                    labelled(
+                        "target_info",
+                        &[("instance", "a"), ("job", "j"), ("version", version)],
+                    ),
+                    &timestamps,
+                    &[1.0, 1.0],
+                    true,
+                ),
+            )
+            .unwrap();
+    }
+    assert_eq!(
+        decided_by_label_set(&run(&mut execution)),
+        vec![
+            (
+                "__name__=up,instance=a,job=j,version=v1".to_string(),
+                vec![(0, Some(1.0)), (1, Some(2.0))]
+            ),
+            (
+                "__name__=up,instance=a,job=j,version=v2".to_string(),
+                vec![(2, Some(3.0)), (3, Some(4.0))]
+            ),
+        ]
+    );
+    assert_eq!(execution.state().reserved_bytes(), 0);
+}
+
+/// The rows `info` keeps for every base point are reserved from the pool before the kernel files
+/// them: a pool that holds the operands but not those rows refuses the operation and gets back what
+/// the node held.
+#[test]
+fn an_oversized_info_workspace_is_refused_by_the_pool() {
+    let points = 10_000;
+    let grid = grid(0, 60_000, points);
+    let schema = evaluated_schema(Arc::clone(&grid));
+    let tree = info_tree(
+        &grid,
+        60_000,
+        InfoSelector {
+            name_matchers: Vec::new(),
+            data_labels: Vec::new(),
+            keep_unmatched: true,
+        },
+    );
+    // About two operand copies of the base, far less than a row per point.
+    let pool = bounded_pool(points as usize * 20);
+    let mut execution = executor(&tree, Arc::clone(&pool));
+    let base = execution.state().input_ids(0).next().unwrap();
+    let info = execution.state().input_ids(1).next().unwrap();
+    execution
+        .push_evaluated(
+            base,
+            evaluated_chunk_of(
+                &schema,
+                labelled("up", &[("job", "j")]),
+                0,
+                vec![Some(1.0); points as usize],
+                true,
+            ),
+        )
+        .unwrap();
+    execution
+        .push_raw(
+            info,
+            raw_chunk_of(
+                &raw_schema(),
+                b"info",
+                labelled("target_info", &[("job", "j"), ("version", "v1")]),
+                &[0],
+                &[1.0],
+                true,
+            ),
+        )
+        .unwrap();
+    let error = execution.finish().unwrap_err();
+    assert!(
+        matches!(error.find_root(), DataFusionError::ResourcesExhausted(_)),
+        "the info workspace was not refused by the pool: {error}"
+    );
+    assert_eq!(pool.reserved(), 0);
+    assert_eq!(execution.state().reserved_bytes(), 0);
+    assert!(!root_holds(&execution, holds_binary));
+}
+
 // ---------------------------------------------------------------------------
 // Scalar broadcasts, driven through `NativeExecutor`.
 // ---------------------------------------------------------------------------

src/promql/src/native/plan.rs

@@ -35,6 +35,7 @@ use datatypes::series::time::EvalGrid;
 use datatypes::series::{SeriesLabels, SeriesOrdering};
 
 use crate::histogram_eval::{HistogramFunction, NativeHistogramFunction};
+use crate::info::InfoSelector;
 use crate::map::MathOp;
 use crate::map_label::LabelRewrite;
 use crate::range_eval::{RangeFunction, RangeWindow};
@@ -160,12 +161,24 @@ pub struct BoundInstant {
     pub offset_ms: i64,
     /// Anchored selection instant; samples are selected at this time for every grid point.
     pub at_ms: Option<i64>,
-    /// Emit the selected sample timestamp in seconds rather than its value.
-    pub timestamp: bool,
+    /// What the selection reports for the sample it selects.
+    pub value: InstantValue,
     /// The grid the selection produces values on.
     pub output_grid: Arc<EvalGrid>,
 }
 
+/// What an instant selection reports at a grid point for the sample it selected there.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum InstantValue {
+    /// The sample itself.
+    Sample,
+    /// `timestamp(x)`: the sample's own time in seconds, without the metric name.
+    Timestamp,
+    /// The info series `info()` reads: the sample's own time in milliseconds, under all of its
+    /// labels, as Prometheus' `evalSeries` records it for `info`.
+    InfoTimestamp,
+}
+
 /// A window function over raw samples or over an evaluated input.
 ///
 /// The input and output grids are bound separately: a subquery evaluates its inner expression on
@@ -369,6 +382,14 @@ pub enum NativeExpr {
         right: Box<NativeNode>,
         op: BoundSetOp,
     },
+    /// `info(base, {selector})`: the series of `base` with the data labels of the info series that
+    /// share their identifying labels. `info` is the instant selection of those info series, whose
+    /// values are the times of the samples it selected ([`InstantValue::InfoTimestamp`]).
+    Info {
+        base: Box<NativeNode>,
+        info: Box<NativeNode>,
+        selector: InfoSelector,
+    },
 }
 
 /// One node of a bound tree.
@@ -383,8 +404,9 @@ pub struct NativeNode {
 /// Whether `node` reads the whole input vector at a point rather than one series at a time.
 ///
 /// An `Aggregate` folds every series of its group, a `Binary` and a `Set` pair two vectors' series
-/// by label, and a `Vector` operation (`sort`, `absent`, `topk`, `limitk`, `histogram_quantile`,
-/// ...) reorders or reads the whole vector at a grid point.
+/// by label, an `Info` looks the info series of each series up by its identifying labels, and a
+/// `Vector` operation (`sort`, `absent`, `topk`, `limitk`, `histogram_quantile`, ...) reorders or
+/// reads the whole vector at a grid point.
 ///
 /// `Merge` answers one series per label set, and the label sets it answers span every input series:
 /// two of them can collapse into the one it emits. It is therefore whole-vector too, which is why it
@@ -416,6 +438,7 @@ pub fn reads_across_series(node: &NativeNode) -> bool {
         NativeExpr::Aggregate { .. }
         | NativeExpr::Vector { .. }
         | NativeExpr::Set { .. }
+        | NativeExpr::Info { .. }
         | NativeExpr::Merge { .. } => true,
     }
 }
@@ -779,6 +802,44 @@ impl NativeNode {
                 ensure_same_fields(&left.output, &self.output, "a set operation")?;
                 ensure_grid(&self.output, left_grid, "a set operation")?;
             }
+            NativeExpr::Info { base, info, .. } => {
+                base.validate()?;
+                info.validate()?;
+                let Some((_, base_grid)) = evaluated_vector(&base.output) else {
+                    return Err(invalid(&format!(
+                        "info consumes an evaluated vector, not {}",
+                        base.output.describe()
+                    )));
+                };
+                // The kernel reads the info side's values as sample times, which only an instant
+                // selection that reports them produces.
+                let NativeExpr::Instant { op, .. } = &info.expr else {
+                    return Err(invalid(
+                        "info reads its info series from an instant selection",
+                    ));
+                };
+                if op.value != InstantValue::InfoTimestamp {
+                    return Err(invalid(
+                        "info reads the sample times of its info series, not their values",
+                    ));
+                }
+                let Some((info_schema, info_grid)) = evaluated_vector(&info.output) else {
+                    return Err(invalid("info's info series are not an evaluated vector"));
+                };
+                if info_schema.fields.len() != 1 {
+                    return Err(invalid(&format!(
+                        "info reads one sample time per info series, and its info series carry {} \
+                         fields",
+                        info_schema.fields.len()
+                    )));
+                }
+                if base_grid.spec() != info_grid.spec() {
+                    return Err(invalid("info aligns two different grids"));
+                }
+                ensure_samples(&self.output, "info")?;
+                ensure_same_fields(&base.output, &self.output, "info")?;
+                ensure_grid(&self.output, base_grid, "info")?;
+            }
         }
         Ok(())
     }

src/promql/src/native/plan/tests.rs

@@ -21,10 +21,11 @@ use datatypes::series::time::{EvalGrid, GridSpec};
 
 use super::{
     AggregateKind, AggregateStage, BoundAggregate, BoundInput, BoundInstant, BoundMap, BoundScalar,
-    BoundVectorOp, BoundWindow, Grouping, MapOp, NativeExpr, NativeNode, NativeTree, NativeType,
-    reads_across_series,
+    BoundVectorOp, BoundWindow, Grouping, InstantValue, MapOp, NativeExpr, NativeNode, NativeTree,
+    NativeType, reads_across_series,
 };
 use crate::extension_plan::{FusedSeries, FusedSeriesBinding};
+use crate::info::InfoSelector;
 use crate::map::MathOp;
 use crate::range_eval::{RangeFunction, RangeWindow};
 use crate::vector_match::{BinaryMatching, BinaryOp, Cardinality, FillValues, MatchingLabels};
@@ -493,7 +494,7 @@ fn an_instant_selection_cannot_change_the_field_count() {
                 lookback_ms: 300_000,
                 offset_ms: 0,
                 at_ms: None,
-                timestamp: false,
+                value: InstantValue::Sample,
                 output_grid: Arc::clone(&out),
             },
         },
@@ -520,7 +521,7 @@ fn an_instant_selection_reads_raw_samples() {
                 lookback_ms: 300_000,
                 offset_ms: 0,
                 at_ms: None,
-                timestamp: false,
+                value: InstantValue::Sample,
                 output_grid: Arc::clone(&out),
             },
         },
@@ -545,7 +546,7 @@ fn an_instant_selection_cannot_consume_state() {
                 lookback_ms: 300_000,
                 offset_ms: 0,
                 at_ms: None,
-                timestamp: false,
+                value: InstantValue::Sample,
                 output_grid: Arc::clone(&out),
             },
         },
@@ -599,7 +600,7 @@ fn an_instant_selection_cannot_read_interleaved_input() {
                 lookback_ms: 300_000,
                 offset_ms: 0,
                 at_ms: None,
-                timestamp: false,
+                value: InstantValue::Sample,
                 output_grid: Arc::clone(&out),
             },
         },
@@ -1435,3 +1436,43 @@ fn a_samples_root_without_an_input_is_refused() {
         "the refusal must name the shape a zero-child tree has to be: {error}"
     );
 }
+
+/// `info` reads its info side's values as sample times, so it accepts only the instant selection
+/// that reports them: one that reports the samples themselves would compare every info series as
+/// equally new.
+#[test]
+fn info_reads_the_sample_times_of_one_field() {
+    let out = grid(0, 60_000, 4);
+    let info_of = |value: InstantValue, fields: usize| NativeNode {
+        expr: NativeExpr::Info {
+            base: Box::new(window_fields(input_with_fields(1), Arc::clone(&out), 1)),
+            info: Box::new(NativeNode {
+                expr: NativeExpr::Instant {
+                    input: Box::new(input_with_fields(fields)),
+                    op: BoundInstant {
+                        lookback_ms: 300_000,
+                        offset_ms: 0,
+                        at_ms: None,
+                        value,
+                        output_grid: Arc::clone(&out),
+                    },
+                },
+                output: evaluated(self::fields(fields), Arc::clone(&out)),
+            }),
+            selector: InfoSelector {
+                name_matchers: Vec::new(),
+                data_labels: Vec::new(),
+                keep_unmatched: true,
+            },
+        },
+        output: evaluated(self::fields(1), Arc::clone(&out)),
+    };
+    info_of(InstantValue::InfoTimestamp, 1).validate().unwrap();
+    for (value, fields, expected) in [
+        (InstantValue::Sample, 1, "not their values"),
+        (InstantValue::InfoTimestamp, 2, "carry 2 fields"),
+    ] {
+        let error = info_of(value, fields).validate().unwrap_err();
+        assert!(error.to_string().contains(expected), "{error}");
+    }
+}

src/promql/src/vector_match.rs

@@ -307,7 +307,7 @@ const MERGE_SLOT_BYTES: usize =
     2 * std::mem::size_of::<(GroupKey, (SeriesLabels, Vec<(usize, usize, GridSample)>))>();
 
 /// One row of the builder's histogram table: the scalar columns and the seven list offsets.
-const HISTOGRAM_ROW_BYTES: usize = 4 * std::mem::size_of::<f64>()
+pub(crate) const HISTOGRAM_ROW_BYTES: usize = 4 * std::mem::size_of::<f64>()
     + 2 * std::mem::size_of::<i32>()
     + std::mem::size_of::<i64>()
     + 7 * std::mem::size_of::<i32>();
@@ -323,7 +323,8 @@ fn output_bytes(fields: usize) -> usize {
         .saturating_add(std::mem::size_of::<SeriesSchema>())
 }
 
-/// The live workspace a binary operation admits as it allocates, at three granularities.
+/// The live workspace a binary, set or `info` operation admits as it allocates, at three
+/// granularities.
 ///
 /// A resource limit belongs at an allocation boundary, not at a value:
 ///
@@ -336,7 +337,7 @@ fn output_bytes(fields: usize) -> usize {
 /// - [`Budget::grow`]: a block a container of that phase took when it outgrew what it held.
 ///
 /// It exists because the caller is handed a *total*, and this turns those figures into that total.
-struct Budget<'a> {
+pub(crate) struct Budget<'a> {
     /// Admits the live workspace, or refuses it.
     reserve: &'a mut dyn FnMut(usize) -> Result<()>,
     /// The bytes the results keep until the operation returns.
@@ -348,7 +349,7 @@ struct Budget<'a> {
 }
 
 impl<'a> Budget<'a> {
-    fn new(reserve: &'a mut dyn FnMut(usize) -> Result<()>) -> Self {
+    pub(crate) fn new(reserve: &'a mut dyn FnMut(usize) -> Result<()>) -> Self {
         Self {
             reserve,
             held: 0,
@@ -358,19 +359,19 @@ impl<'a> Budget<'a> {
     }
 
     /// Retains `bytes` of results, which stay live until the operation returns.
-    fn hold(&mut self, bytes: usize) -> Result<()> {
+    pub(crate) fn hold(&mut self, bytes: usize) -> Result<()> {
         self.held = self.held.saturating_add(bytes);
         self.admit()
     }
 
     /// Admits `bytes` as the whole workspace of the phase being run, replacing the phase before it.
-    fn phase(&mut self, bytes: usize) -> Result<()> {
+    pub(crate) fn phase(&mut self, bytes: usize) -> Result<()> {
         self.phase = bytes;
         self.admit()
     }
 
     /// Adds `bytes` to the phase for a container of it that has outgrown what was reserved.
-    fn grow(&mut self, bytes: usize) -> Result<()> {
+    pub(crate) fn grow(&mut self, bytes: usize) -> Result<()> {
         self.phase = self.phase.saturating_add(bytes);
         self.admit()
     }
@@ -627,7 +628,7 @@ fn label_text_bytes(metric: &SeriesLabels) -> usize {
 
 /// The bytes one owned copy of a histogram view holds: its scalars inline, and on the heap the
 /// buckets of both sides, the span layout of the view and its custom bounds.
-fn owned_histogram_bytes(view: &HistogramView<'_>) -> usize {
+pub(crate) fn owned_histogram_bytes(view: &HistogramView<'_>) -> usize {
     let sides = [view.positive, view.negative];
     let buckets = sides
         .iter()

src/query/src/dist_plan.rs

@@ -25,6 +25,8 @@ mod remote_dyn_filter_receiver;
 mod remote_dyn_filter_registry;
 
 pub use analyzer::{DistPlannerAnalyzer, DistPlannerOptions};
+#[cfg(test)]
+pub(crate) use commutativity::{Categorizer, Commutativity};
 pub use filter_id::{FilterFingerprint, FilterId, ParseFilterIdError, RemoteDynFilterProducerId};
 pub use merge_scan::{MergeScanExec, MergeScanLogicalPlan};
 pub(crate) use merge_sort::{MergeSortExec, MergeSortLogicalPlan};

src/query/src/promql/native_compile.rs

@@ -36,6 +36,7 @@
 //!   <constant> <operator> <expression>             e.g. 1 / x, 0 < x
 //!   -<expression>                                  e.g. -x
 //!   <expression> <set operator> <expression>       e.g. x and on(job) y, x unless y
+//!   info(<expression>[, {<matchers>}])             e.g. info(rate(x[5m]), {version=~".+"})
 //!   <aggregator> [(by|without)(<labels>)] (<expression>)
 //!                                                  e.g. avg by(job)(rate(x[5m]) * 100)
 //!   <aggregator> (<expression>) [(by|without)(<labels>)]
@@ -81,7 +82,8 @@
 //! argument evaluates to) and, over
 //! any of those, `Binary` (an arithmetic or comparison operation between vectors or with a
 //! constant on one side), `Scalar` (that constant), `Set` (`and`, `or` or `unless` between two
-//! vectors), `Aggregate` (one of the nine aggregators that
+//! vectors), `Info` (the expression and an `Instant` of the info metric reporting sample times),
+//! `Aggregate` (one of the nine aggregators that
 //! fold a group's values into one value per grid point, optionally grouped) and `Vector` (an
 //! operation over the whole vector at each point: a sort, `absent`, a classic histogram function, a
 //! selecting aggregator or `count_values`). Everything else is a
@@ -139,20 +141,21 @@ use datatypes::series::samples::SeriesSchema;
 use datatypes::series::time::{EvalGrid, GridSpec};
 use datatypes::series::{SeriesLabels, SeriesOrdering};
 use promql::histogram_eval::{HistogramFunction, NativeHistogramFunction};
+use promql::info::InfoSelector;
 use promql::map::{ArithmeticOp, ComparisonOp, MathOp};
 use promql::map_label::LabelRewrite;
 use promql::map_time::DateFunction;
 use promql::native::plan::{
     AggregateKind, AggregateStage, BoundAggregate, BoundInput, BoundInstant, BoundMap, BoundScalar,
-    BoundSetOp, BoundVectorOp, BoundWindow, Grouping, MapOp, NativeExpr, NativeNode, NativeTree,
-    NativeType,
+    BoundSetOp, BoundVectorOp, BoundWindow, Grouping, InstantValue, MapOp, NativeExpr, NativeNode,
+    NativeTree, NativeType,
 };
 use promql::range_eval::{RangeFunction, RangeWindow, subquery_grid};
 use promql::select::{LabelMatcher, MatchOp, SortOrder, absent_labels};
 use promql::vector_match::{
     BinaryMatching, BinaryOp, Cardinality, FillValues, MatchingLabels, SetOp, binary_output_schema,
 };
-use promql_parser::label::{MatchOp as ParserMatchOp, Matchers};
+use promql_parser::label::{MatchOp as ParserMatchOp, Matcher, Matchers};
 use promql_parser::parser::token::TokenType;
 use promql_parser::parser::value::ValueType;
 use promql_parser::parser::{
@@ -244,7 +247,8 @@ const SUPPORTED: &str = "a selector — `x`, optionally with an `offset` — a r
                          (`label_replace(x, \"a\", \"$1\", \"b\", \"(.*)\")`), a binary \
                          operation between one of them and a constant (`2 / (x + 1)`, \
                          `x > bool 0`), a negation of one of them (`-x`), a set operation between \
-                         two of them (`x and on(job) y`, `x or y`, `x unless y`), or one of the \
+                         two of them (`x and on(job) y`, `x or y`, `x unless y`), `info` over any \
+                         of them with one info metric (`info(x, {version=~\".+\"})`), or one of the \
                          aggregators that folds a group's values into one value per grid point — \
                          `sum`, `avg`, `count`, `min`, `max`, `group`, `stddev`, `stdvar` and \
                          `quantile(<constant>, <expression>)` — over any of them, optionally \
@@ -374,9 +378,13 @@ fn validate_inputs(root: &NativeNode, inputs: &[CompiledInput]) -> Result<()> {
             | NativeExpr::Timestamp { input }
             | NativeExpr::Vector { input, .. }
             | NativeExpr::Aggregate { input, .. } => check(input, inputs, next),
-            NativeExpr::Binary { left, right, .. } | NativeExpr::Set { left, right, .. } => {
-                check(left, inputs, next) && check(right, inputs, next)
-            }
+            NativeExpr::Binary { left, right, .. }
+            | NativeExpr::Set { left, right, .. }
+            | NativeExpr::Info {
+                base: left,
+                info: right,
+                ..
+            } => check(left, inputs, next) && check(right, inputs, next),
         }
     }
     let mut next = 0;
@@ -884,7 +892,7 @@ fn timestamp_call(context: &Context, args: &[Box<PromExpr>]) -> Result<NativeNod
         return Err(not_implemented("timestamp argument is not a vector"));
     }
     if let PromExpr::VectorSelector(selector) = argument {
-        return instant_selection_timestamp(context, selector, true);
+        return instant_selection_of(context, selector, InstantValue::Timestamp);
     }
     let input = compile_expr(context, argument)?;
     let NativeType::Samples(schema) = &input.output else {
@@ -904,6 +912,265 @@ fn timestamp_call(context: &Context, args: &[Box<PromExpr>]) -> Result<NativeNod
     })
 }
 
+/// Compiles `info(v[, {selector}])` as its base `v` and the instant selection of the info series it
+/// reads.
+///
+/// The info series are selected the way Prometheus' `fetchInfoSeries` and `evalSeries` select
+/// them: by the info metric's name and the data matchers, at the reference time the selectors of
+/// `v` share ([`info_input`]), with the statement's lookback. Which of them a base series matches by
+/// its identifying labels is the kernel's.
+fn info_call(context: &Context, args: &[Box<PromExpr>]) -> Result<NativeNode> {
+    let InfoInput { selector, kernel } = info_input(context.stmt, args)?;
+    let base = compile_expr(context, &args[0])?;
+    if base
+        .output
+        .samples()
+        .is_none_or(|schema| schema.eval_grid.is_none())
+    {
+        return Err(DataFusionError::Internal(format!(
+            "info's first argument `{}` compiled to something other than an evaluated vector: the \
+             parser only accepts a vector there",
+            args[0]
+        )));
+    }
+    let info = instant_selection_of(context, &selector, InstantValue::InfoTimestamp)?;
+    let fields = info
+        .output
+        .samples()
+        .map_or(0, |schema| schema.fields.len());
+    // Prometheus reads one sample per info series and point, and the existing planner has no
+    // `info` to follow for several fields.
+    if fields != 1 {
+        return Err(not_implemented(&format!(
+            "info over the info metric `{selector}` with {fields} fields: an info series is read as \
+             one field"
+        )));
+    }
+    let output = base.output.clone();
+    Ok(NativeNode {
+        expr: NativeExpr::Info {
+            base: Box::new(base),
+            info: Box::new(info),
+            selector: kernel,
+        },
+        output,
+    })
+}
+
+/// The info series `info(v[, {selector}])` reads, as a selector, and the selector its kernel
+/// combines by.
+pub(crate) struct InfoInput {
+    /// The info metric with the data matchers, at the reference time the selectors of `v` share.
+    pub selector: VectorSelector,
+    pub kernel: InfoSelector,
+}
+
+/// Splits the arguments of `info` into the selector of its info series and the kernel's selector.
+///
+/// The compiler binds the selector as the call's last input and the planner prepares a scan for it
+/// at the same position, so both read it here. One info metric is one table, so the name has to be
+/// `target_info` (no name matcher) or one `__name__="..."`: a regular expression or a negative
+/// matcher can select several tables. `__field__`, `__schema__` and `__database__` choose what a
+/// selector reads rather than match a label, which Prometheus would read them as, so they are
+/// refused too.
+pub(crate) fn info_input(stmt: &EvalStmt, args: &[Box<PromExpr>]) -> Result<InfoInput> {
+    let (base, labels) = match args {
+        [base] => (base, None),
+        [base, labels] => (base, Some(labels)),
+        _ => {
+            return Err(not_implemented(&format!(
+                "a call to `info` with {} argument(s)",
+                args.len()
+            )));
+        }
+    };
+    let matchers: &[Matcher] = match labels.map(|labels| labels.as_ref()) {
+        None => &[],
+        Some(PromExpr::VectorSelector(VectorSelector {
+            name: None,
+            matchers,
+            offset: None,
+            at: None,
+        })) if matchers.or_matchers.is_empty() => &matchers.matchers,
+        Some(other) => {
+            return Err(not_implemented(&format!(
+                "`info` whose second argument `{other}` is not one label selector without a name \
+                 or modifiers"
+            )));
+        }
+    };
+    let (names, data): (Vec<&Matcher>, Vec<&Matcher>) = matchers
+        .iter()
+        .partition(|matcher| matcher.name == "__name__");
+    let metric = match names.as_slice() {
+        [] => "target_info".to_string(),
+        [name] if name.op == ParserMatchOp::Equal => name.value.clone(),
+        _ => {
+            return Err(not_implemented(
+                "`info` over the info metrics a regular expression, a negative matcher or several \
+                 `__name__` matchers select, which can be several tables",
+            ));
+        }
+    };
+    if data.iter().any(|matcher| {
+        matches!(
+            matcher.name.as_str(),
+            "__field__" | "__schema__" | "__database__"
+        )
+    }) {
+        return Err(not_implemented(
+            "`info` with a `__field__`, `__schema__` or `__database__` matcher, which selects \
+             what is read rather than a data label",
+        ));
+    }
+    let (at, offset_ms) = info_reference(stmt, base)?;
+    let offset = match offset_ms {
+        0 => None,
+        ms if ms > 0 => Some(Offset::Pos(Duration::from_millis(ms.unsigned_abs()))),
+        ms => Some(Offset::Neg(Duration::from_millis(ms.unsigned_abs()))),
+    };
+    let kernel = InfoSelector {
+        name_matchers: names
+            .iter()
+            .map(|matcher| (match_op(&matcher.op), Arc::from(matcher.value.as_str())))
+            .collect(),
+        data_labels: data
+            .iter()
+            .map(|matcher| Arc::from(matcher.name.as_str()))
+            .collect(),
+        keep_unmatched: data.iter().all(|matcher| matcher.is_match("")),
+    };
+    Ok(InfoInput {
+        selector: VectorSelector {
+            name: Some(metric),
+            matchers: Matchers::new(data.into_iter().cloned().collect()),
+            offset,
+            at,
+        },
+        kernel,
+    })
+}
+
+/// The `@` anchor and offset the info series are selected at: those of the first selector of
+/// `expr` when every vector-producing path of it reaches a selector with the same effective
+/// reference time, and none otherwise, as Prometheus `infoSeriesSelectTimestampAndOffset` decides.
+///
+/// A selector inside subqueries adds their offsets until one of them anchors it with `@`. A path
+/// that produces a vector without a selector, such as `vector(1)`, is evaluated at the step itself,
+/// so it leaves no shared reference.
+fn info_reference(stmt: &EvalStmt, expr: &PromExpr) -> Result<(Option<AtModifier>, i64)> {
+    #[derive(Default)]
+    struct Walk {
+        /// The first selector's anchor, its resolved time and its offset.
+        first: Option<(Option<AtModifier>, Option<i64>, i64)>,
+        uniform: bool,
+        reference_free: bool,
+    }
+
+    fn inspect<'a>(
+        stmt: &EvalStmt,
+        expr: &'a PromExpr,
+        subqueries: &mut Vec<&'a SubqueryExpr>,
+        walk: &mut Walk,
+    ) -> Result<bool> {
+        if !matches!(expr.value_type(), ValueType::Vector | ValueType::Matrix) {
+            return Ok(false);
+        }
+        let selector = match expr {
+            PromExpr::VectorSelector(selector) => Some(selector),
+            PromExpr::MatrixSelector(MatrixSelector { vs, .. }) => Some(vs),
+            _ => None,
+        };
+        if let Some(selector) = selector {
+            let mut at = selector.at.clone();
+            let mut offset = selector_offset_ms(&selector.offset)?;
+            for subquery in subqueries.iter().rev() {
+                if at.is_some() {
+                    break;
+                }
+                offset = offset
+                    .checked_add(selector_offset_ms(&subquery.offset)?)
+                    .ok_or_else(|| DataFusionError::Plan("selector offset overflows".into()))?;
+                at = subquery.at.clone();
+            }
+            let at_ms = selector_anchor_ms(&at, stmt)?;
+            match &walk.first {
+                None => walk.first = Some((at, at_ms, offset)),
+                Some((_, first_at, first_offset)) => {
+                    let same = match (first_at, at_ms) {
+                        (None, None) => *first_offset == offset,
+                        (Some(first_at), Some(at)) => {
+                            i128::from(*first_at) - i128::from(*first_offset)
+                                == i128::from(at) - i128::from(offset)
+                        }
+                        _ => false,
+                    };
+                    walk.uniform &= same;
+                }
+            }
+            return Ok(true);
+        }
+        let children: Vec<&PromExpr> = match expr {
+            PromExpr::Paren(paren) => vec![&paren.expr],
+            PromExpr::Unary(unary) => vec![&unary.expr],
+            PromExpr::Binary(binary) => vec![&binary.lhs, &binary.rhs],
+            PromExpr::Aggregate(aggregate) => aggregate
+                .param
+                .iter()
+                .chain([&aggregate.expr])
+                .map(|child| child.as_ref())
+                .collect(),
+            // The second argument of a nested `info` is selector syntax, not a vector it evaluates.
+            PromExpr::Call(call) if call.func.name == "info" => call
+                .args
+                .args
+                .iter()
+                .take(1)
+                .map(|arg| arg.as_ref())
+                .collect(),
+            PromExpr::Call(call) => call.args.args.iter().map(|arg| arg.as_ref()).collect(),
+            PromExpr::Subquery(subquery) => {
+                subqueries.push(subquery);
+                let found = inspect(stmt, &subquery.expr, subqueries, walk);
+                subqueries.pop();
+                let found = found?;
+                walk.reference_free |= !found;
+                return Ok(found);
+            }
+            other => {
+                return Err(not_implemented(&format!(
+                    "`info` over `{other}`, whose reference time is not derived here"
+                )));
+            }
+        };
+        let mut found = false;
+        for child in children {
+            found |= inspect(stmt, child, subqueries, walk)?;
+        }
+        walk.reference_free |= !found;
+        Ok(found)
+    }
+
+    let mut walk = Walk {
+        uniform: true,
+        ..Default::default()
+    };
+    inspect(stmt, expr, &mut Vec::new(), &mut walk)?;
+    match walk.first {
+        Some((at, _, offset)) if walk.uniform && !walk.reference_free => Ok((at, offset)),
+        _ => Ok((None, 0)),
+    }
+}
+
+fn match_op(op: &ParserMatchOp) -> MatchOp {
+    match op {
+        ParserMatchOp::Equal => MatchOp::Equal,
+        ParserMatchOp::NotEqual => MatchOp::NotEqual,
+        ParserMatchOp::Re(_) => MatchOp::Regex,
+        ParserMatchOp::NotRe(_) => MatchOp::NotRegex,
+    }
+}
+
 /// Compiles a range, pointwise, histogram or label function call.
 ///
 /// The name lists are disjoint, and a range function is decided first because it binds its own
@@ -914,6 +1181,9 @@ fn compile_call(context: &Context, call: &Call) -> Result<NativeNode> {
     if call.func.name == "timestamp" {
         return timestamp_call(context, &call.args.args);
     }
+    if call.func.name == "info" {
+        return info_call(context, &call.args.args);
+    }
     let Call { func, args } = call;
     if let Some(signature) = range_signature(func.name) {
         return window_call(context, func.name, signature, &args.args);
@@ -1157,12 +1427,7 @@ fn absent_labels_of(expr: &PromExpr) -> Result<SeriesLabels> {
         })
         .map(|matcher| LabelMatcher {
             name: Arc::from(matcher.name.as_str()),
-            op: match matcher.op {
-                ParserMatchOp::Equal => MatchOp::Equal,
-                ParserMatchOp::NotEqual => MatchOp::NotEqual,
-                ParserMatchOp::Re(_) => MatchOp::Regex,
-                ParserMatchOp::NotRe(_) => MatchOp::NotRegex,
-            },
+            op: match_op(&matcher.op),
             value: Arc::from(matcher.value.as_str()),
             matches_empty: matcher.is_match(""),
         })
@@ -1356,13 +1621,13 @@ fn subquery_window(
 /// The lookback, the offset and the grid are the whole of an instant selection: it reports the last
 /// sample at or before each grid point, within the statement's lookback.
 fn instant_selection(context: &Context, selector: &VectorSelector) -> Result<NativeNode> {
-    instant_selection_timestamp(context, selector, false)
+    instant_selection_of(context, selector, InstantValue::Sample)
 }
 
-fn instant_selection_timestamp(
+fn instant_selection_of(
     context: &Context,
     selector: &VectorSelector,
-    timestamp: bool,
+    value: InstantValue,
 ) -> Result<NativeNode> {
     let lookback_ms = lookback_millis(context.stmt.lookback_delta.as_millis())?;
     let offset_ms = selector_offset_ms(&selector.offset)?;
@@ -1381,7 +1646,7 @@ fn instant_selection_timestamp(
             timing: SelectorTiming::Instant { lookback_ms },
         },
     )?;
-    if timestamp && schema.fields.len() != 1 {
+    if value == InstantValue::Timestamp && schema.fields.len() != 1 {
         return Err(not_implemented(
             "timestamp() over a multi-value input is not supported natively",
         ));
@@ -1405,7 +1670,7 @@ fn instant_selection_timestamp(
                 lookback_ms,
                 offset_ms,
                 at_ms,
-                timestamp,
+                value,
                 output_grid,
             },
         },

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

@@ -28,9 +28,10 @@ use promql::map::MathOp;
 use promql::native::execute::{InputId, NativeExecutionState, NativeExecutor};
 use promql::native::plan::{
     AggregateKind, AggregateStage, BoundAggregate, BoundInput, BoundInstant, BoundMap, BoundSetOp,
-    BoundWindow, Grouping, MapOp, NativeExpr, NativeNode, NativeTree, NativeType,
+    BoundWindow, Grouping, InstantValue, MapOp, NativeExpr, NativeNode, NativeTree, NativeType,
 };
 use promql::range_eval::RangeFunction;
+use promql::select::MatchOp;
 use promql::vector_match::{BinaryMatching, BinaryOp, Cardinality, MatchingLabels, SetOp};
 use promql_parser::parser::{
     AtModifier, BinModifier, EvalStmt, Expr as PromExpr, NumberLiteral, Offset,
@@ -1926,6 +1927,171 @@ fn a_set_operation_refuses_a_fill_value_and_several_fields() {
     }
 }
 
+/// `info` binds the base first and then the instant selection of its info metric, reporting the
+/// times of the samples it selects, with the selector split the way Prometheus uses it: the name
+/// picks the info metric, `target_info` when there is none, and the other matchers both filter the
+/// info series and name the labels that are copied.
+#[test]
+fn info_binds_its_info_metric_after_the_base() {
+    for (query, metric, matchers, names, keep_unmatched) in [
+        ("info(rate(x[5m]))", "target_info", vec![], vec![], true),
+        (
+            "info(rate(x[5m]), {__name__=\"build_info\", version=~\"v.+\", zone=~\".*\"})",
+            "build_info",
+            vec!["version", "zone"],
+            vec!["build_info"],
+            false,
+        ),
+    ] {
+        let compiled = compile_selectors(
+            &eval_stmt(query),
+            &|_| Ok(raw_schema()),
+            SeriesOrdering::KeyThenTime,
+        )
+        .unwrap_or_else(|error| panic!("`{query}` was refused: {error}"));
+        let [base_input, info_input] = &compiled.inputs[..] else {
+            panic!("{query}: {} inputs", compiled.inputs.len());
+        };
+        assert_eq!(
+            base_input.requirement.metric.as_deref(),
+            Some("x"),
+            "{query}"
+        );
+        let requirement = &info_input.requirement;
+        assert_eq!(requirement.metric.as_deref(), Some(metric), "{query}");
+        assert_eq!(
+            requirement
+                .matchers
+                .matchers
+                .iter()
+                .map(|matcher| matcher.name.as_str())
+                .collect::<Vec<_>>(),
+            matchers,
+            "{query}"
+        );
+        assert_eq!(
+            requirement.timing,
+            SelectorTiming::Instant { lookback_ms: 5_000 },
+            "{query}"
+        );
+        let NativeExpr::Info {
+            base,
+            info,
+            selector,
+        } = &compiled.tree.root().expr
+        else {
+            panic!(
+                "{query}: the root is not info: {:?}",
+                compiled.tree.root().expr
+            );
+        };
+        assert!(matches!(base.expr, NativeExpr::Window { .. }), "{query}");
+        let NativeExpr::Instant { input, op } = &info.expr else {
+            panic!("{query}: the info side is not a selection: {:?}", info.expr);
+        };
+        assert_eq!(op.value, InstantValue::InfoTimestamp, "{query}");
+        assert!(
+            matches!(&input.expr, NativeExpr::Input(bound) if bound.source_index == 1),
+            "{query}"
+        );
+        assert_eq!(
+            selector
+                .name_matchers
+                .iter()
+                .map(|(op, value)| (*op, &**value))
+                .collect::<Vec<_>>(),
+            names
+                .into_iter()
+                .map(|name| (MatchOp::Equal, name))
+                .collect::<Vec<_>>(),
+            "{query}"
+        );
+        assert_eq!(
+            selector
+                .data_labels
+                .iter()
+                .map(|label| &**label)
+                .collect::<Vec<_>>(),
+            matchers,
+            "{query}"
+        );
+        assert_eq!(selector.keep_unmatched, keep_unmatched, "{query}");
+    }
+}
+
+/// The info series are selected at the reference time the selectors of the first argument share,
+/// as Prometheus `infoSeriesSelectTimestampAndOffset` decides it: one `@` and offset when every
+/// selector resolves to the same instant, the offsets of the subqueries a selector sits in added
+/// until one anchors it, and the evaluation time itself when the selectors disagree.
+#[test]
+fn info_selects_its_info_series_at_the_shared_reference_time() {
+    for (query, offset_ms, at_ms) in [
+        ("info(x offset 1m)", 60_000, None),
+        ("info(rate(x[1m] @ 100 offset 5s))", 5_000, Some(100_000)),
+        (
+            "info(x @ 50 offset 10s + on(job) y @ 60 offset 20s)",
+            10_000,
+            Some(50_000),
+        ),
+        ("info(x offset 1m + on(job) y)", 0, None),
+        ("info(x @ 50 + on(job) y)", 0, None),
+        ("info(rate(x[1m:10s] offset 1m))", 60_000, None),
+        (
+            "info(rate(x[1m:10s] @ 100 offset 1m))",
+            60_000,
+            Some(100_000),
+        ),
+        (
+            "info(max_over_time(rate(x[1m] @ 30 offset 2s)[1m:10s] offset 1m))",
+            2_000,
+            Some(30_000),
+        ),
+    ] {
+        let compiled = compile_selectors(
+            &eval_stmt(query),
+            &|_| Ok(raw_schema()),
+            SeriesOrdering::KeyThenTime,
+        )
+        .unwrap_or_else(|error| panic!("`{query}` was refused: {error}"));
+        let info = &compiled.inputs.last().unwrap().requirement;
+        assert_eq!(info.metric.as_deref(), Some("target_info"), "{query}");
+        assert_eq!((info.offset_ms, info.at_ms), (offset_ms, at_ms), "{query}");
+    }
+}
+
+/// What the native `info` cannot read is refused rather than read some other way: name matchers
+/// that can select several info tables, a matcher that chooses what is read instead of matching a
+/// label, and an info metric of several fields.
+#[test]
+fn info_refuses_info_metrics_it_cannot_read_as_one_table() {
+    for (query, fields, expected) in [
+        ("info(x, {__name__=~\".+_info\"})", 1, "several tables"),
+        (
+            "info(x, {__name__!=\"build_info\", version=\"v1\"})",
+            1,
+            "several tables",
+        ),
+        ("info(x, {__field__=\"v\"})", 1, "`__field__`"),
+        ("info(x)", 2, "with 2 fields"),
+    ] {
+        let error = compile_selectors(
+            &eval_stmt(query),
+            &|requirement| {
+                Ok(match requirement.metric.as_deref() {
+                    Some("x") => raw_schema(),
+                    _ => raw_schema_with_fields(fields),
+                })
+            },
+            SeriesOrdering::KeyThenTime,
+        )
+        .expect_err(&format!("`{query}` compiled"));
+        assert!(
+            matches!(&error, DataFusionError::NotImplemented(message) if message.contains(expected)),
+            "{query}: {error}"
+        );
+    }
+}
+
 /// A statement with no vector has nothing to read, so it is refused: a tree reads a scan and reports
 /// evaluated samples, and a constant is bound next to a vector rather than on its own. The refusal
 /// names the expression and its type, because the type is why it cannot be bound.
@@ -4883,7 +5049,13 @@ fn validate_inputs_rejects_misbound_occurrences_and_accepts_scalar_wrappers() {
             | NativeExpr::Timestamp { input }
             | NativeExpr::Vector { input, .. }
             | NativeExpr::Aggregate { input, .. } => set_input_source(input, source),
-            NativeExpr::Binary { left, right, .. } | NativeExpr::Set { left, right, .. } => {
+            NativeExpr::Binary { left, right, .. }
+            | NativeExpr::Set { left, right, .. }
+            | NativeExpr::Info {
+                base: left,
+                info: right,
+                ..
+            } => {
                 set_input_source(left, source);
                 set_input_source(right, source);
             }

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

@@ -32,6 +32,7 @@
 //!
 //! A constant statement is planned with no scan: one node, no child, the time index and value.
 
+use std::borrow::Cow;
 use std::sync::Arc;
 use std::time::{Duration, UNIX_EPOCH};
 
@@ -60,7 +61,7 @@ use store_api::metric_engine_consts::{DATA_SCHEMA_TABLE_ID_COLUMN_NAME, METRIC_E
 
 use crate::promql::error::{DataFusionPlanningSnafu, Result};
 use crate::promql::native_compile::{
-    compile_selectors, compile_statement, constant_value, statement_grid,
+    InfoInput, compile_selectors, compile_statement, constant_value, info_input, statement_grid,
 };
 use crate::promql::planner::{DEFAULT_FIELD_COLUMN, DEFAULT_TIME_INDEX_COLUMN, PromPlanner};
 use crate::query_engine::QueryEngineState;
@@ -219,8 +220,11 @@ impl PromPlanner {
             return Ok(None);
         }
 
+        let Some(stmt) = self.native_statement(prom_expr) else {
+            return Ok(None);
+        };
         let mut selectors = Vec::new();
-        if !collect_selectors(prom_expr, &mut selectors) {
+        if !collect_selectors(&stmt, prom_expr, &mut selectors) {
             return Ok(None);
         }
         // Preserve the original single-selector admission allowlist. Broader binary syntax is only
@@ -228,9 +232,6 @@ impl PromPlanner {
         if selectors.len() == 1 && !admitted_single_selector(prom_expr) {
             return Ok(None);
         }
-        let Some(stmt) = self.native_statement(prom_expr) else {
-            return Ok(None);
-        };
         let Ok(grid) = statement_grid(&stmt) else {
             return Ok(None);
         };
@@ -771,7 +772,7 @@ fn scan_time_filter(start: i64, end: i64, inclusive: bool, time_column: &str) ->
     lower.and(column.lt_eq(upper))
 }
 
-/// Whether `root` is a supported one-to-one match or set operation over independent scan
+/// Whether `root` is a supported one-to-one match, set operation or `info` over independent scan
 /// occurrences, under any number of point-wise wrappers.
 ///
 /// Any arithmetic or comparison operator qualifies: they pair series by the same matching rules,
@@ -797,6 +798,9 @@ pub(crate) fn independent_scan_pair(root: &NativeNode) -> bool {
         NativeExpr::Set { left, right, .. } => {
             return scan_operand(left) && scan_operand(right);
         }
+        NativeExpr::Info { base, info, .. } => {
+            return pointwise_operand(base) && pointwise_operand(info);
+        }
         _ => return false,
     };
 
@@ -830,9 +834,10 @@ pub(crate) fn same_table_scans<'a>(plans: impl IntoIterator<Item = &'a LogicalPl
 ///
 /// They follow Prometheus' result labels: a one-to-one match keeps the left side's labels, narrowed
 /// by `on` or `ignoring`; `and` and `unless` keep the left side's series and `or` adds the right
-/// side's; an operation with a constant or a function of each value keeps its operand's. `None` for
-/// a node whose labels this walk does not derive - an aggregate's groups, or a whole-vector
-/// operation - so a tree holding one is declined rather than given its input's labels.
+/// side's; `info` adds the data labels of the info table; an operation with a constant or a function
+/// of each value keeps its operand's. `None` for a node whose labels this walk does not derive - an
+/// aggregate's groups, or a whole-vector operation - so a tree holding one is declined rather than
+/// given its input's labels.
 fn fused_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Option<Vec<String>> {
     match &node.expr {
         NativeExpr::Input(input) => source_tags.get(input.source_index).cloned(),
@@ -890,6 +895,26 @@ fn fused_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Option<V
             }
             Some(tags)
         }
+        // A base series keeps its labels and takes the data labels of its info series: the ones the
+        // selector names, or all of them when it names none.
+        NativeExpr::Info {
+            base,
+            info,
+            selector,
+        } => {
+            let mut tags = fused_output_tags(base, source_tags)?;
+            for tag in fused_output_tags(info, source_tags)? {
+                let copied = selector.data_labels.is_empty()
+                    || selector
+                        .data_labels
+                        .iter()
+                        .any(|label| label.as_ref() == tag.as_str());
+                if copied && !tags.contains(&tag) {
+                    tags.push(tag);
+                }
+            }
+            Some(tags)
+        }
     }
 }
 
@@ -928,7 +953,8 @@ pub(crate) fn pointwise_operand(node: &NativeNode) -> bool {
         | NativeExpr::Scalar(_)
         | NativeExpr::Aggregate { .. }
         | NativeExpr::Vector { .. }
-        | NativeExpr::Set { .. } => false,
+        | NativeExpr::Set { .. }
+        | NativeExpr::Info { .. } => false,
     }
 }
 
@@ -974,42 +1000,54 @@ fn admitted_selector(expr: &PromExpr) -> Option<&VectorSelector> {
     }
 }
 
-fn collect_selectors<'a>(expr: &'a PromExpr, out: &mut Vec<&'a VectorSelector>) -> bool {
+/// The selectors `expr` reads, in the order the compiler binds them.
+///
+/// `info` reads its info series through a selector it does not write: [`info_input`] derives it, for
+/// the compiler as for this walk, and the compiler binds it after the selectors of the base.
+fn collect_selectors<'a>(
+    stmt: &EvalStmt,
+    expr: &'a PromExpr,
+    out: &mut Vec<Cow<'a, VectorSelector>>,
+) -> bool {
     match expr {
-        PromExpr::VectorSelector(selector) => out.push(selector),
-        PromExpr::MatrixSelector(MatrixSelector { vs, .. }) => out.push(vs),
-        PromExpr::Paren(ParenExpr { expr }) => {
-            return collect_selectors(expr, out);
-        }
+        PromExpr::VectorSelector(selector) => out.push(Cow::Borrowed(selector)),
+        PromExpr::MatrixSelector(MatrixSelector { vs, .. }) => out.push(Cow::Borrowed(vs)),
+        PromExpr::Paren(ParenExpr { expr }) => return collect_selectors(stmt, expr, out),
         PromExpr::Call(call)
             if matches!(
                 call.func.name,
                 "rate" | "increase" | "sum_over_time" | "abs" | "timestamp"
             ) && call.args.args.len() == 1 =>
         {
-            return collect_selectors(&call.args.args[0], out);
+            return collect_selectors(stmt, &call.args.args[0], out);
+        }
+        PromExpr::Call(call) if call.func.name == "info" => {
+            let Ok(InfoInput { selector, .. }) = info_input(stmt, &call.args.args) else {
+                return false;
+            };
+            if !collect_selectors(stmt, &call.args.args[0], out) {
+                return false;
+            }
+            out.push(Cow::Owned(selector));
         }
         PromExpr::Binary(binary) => {
-            return collect_selectors(&binary.lhs, out) && collect_selectors(&binary.rhs, out);
+            return collect_selectors(stmt, &binary.lhs, out)
+                && collect_selectors(stmt, &binary.rhs, out);
         }
         // Literals are permitted only as compiler-recognized broadcasts in a bounded binary tree.
         PromExpr::NumberLiteral(_) => return true,
-        PromExpr::Unary(unary) => {
-            return collect_selectors(&unary.expr, out);
-        }
+        PromExpr::Unary(unary) => return collect_selectors(stmt, &unary.expr, out),
         PromExpr::Call(call) => {
             if matches!(call.func.name, "label_replace" | "label_join") {
                 return false;
             }
             for arg in &call.args.args {
-                if !collect_selectors(arg, out) {
+                if !collect_selectors(stmt, arg, out) {
                     return false;
                 }
             }
         }
-        PromExpr::Aggregate(aggregate) => {
-            return collect_selectors(&aggregate.expr, out);
-        }
+        PromExpr::Aggregate(aggregate) => return collect_selectors(stmt, &aggregate.expr, out),
         PromExpr::Subquery(_) => return false,
         _ => {}
     }
@@ -1122,7 +1160,13 @@ mod tests {
                 | NativeExpr::Merge { input }
                 | NativeExpr::Map { input, .. }
                 | NativeExpr::Timestamp { input } => collect(input, sources),
-                NativeExpr::Binary { left, right, .. } | NativeExpr::Set { left, right, .. } => {
+                NativeExpr::Binary { left, right, .. }
+                | NativeExpr::Set { left, right, .. }
+                | NativeExpr::Info {
+                    base: left,
+                    info: right,
+                    ..
+                } => {
                     collect(left, sources);
                     collect(right, sources);
                 }
@@ -1140,8 +1184,15 @@ mod tests {
     /// The selectors `query` names, in the order the compiler binds them.
     fn collected_selectors(query: &str) -> Vec<String> {
         let expr = promql_parser::parser::parse(query).expect("the query parses");
+        let statement = promql_parser::parser::EvalStmt {
+            expr: expr.clone(),
+            start: UNIX_EPOCH,
+            end: UNIX_EPOCH + Duration::from_secs(10),
+            interval: Duration::from_secs(5),
+            lookback_delta: Duration::from_secs(5),
+        };
         let mut selectors = Vec::new();
-        assert!(collect_selectors(&expr, &mut selectors));
+        assert!(collect_selectors(&statement, &expr, &mut selectors));
         selectors
             .iter()
             .map(|selector| selector.name.clone().unwrap_or_default())
@@ -1160,17 +1211,7 @@ mod tests {
                 vec!["x", "y"],
             ),
         ] {
-            let expr = promql_parser::parser::parse(query).unwrap();
-            let mut selectors = Vec::new();
-            assert!(collect_selectors(&expr, &mut selectors));
-            assert_eq!(
-                selectors
-                    .iter()
-                    .map(|s| s.name.as_deref().unwrap())
-                    .collect::<Vec<_>>(),
-                expected,
-                "{query}"
-            );
+            assert_eq!(collected_selectors(query), expected, "{query}");
         }
     }
 

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

@@ -1873,7 +1873,7 @@ async fn native_fused_timestamp_admits_raw_and_evaluated_vectors() {
         if raw_selector {
             assert!(matches!(
                 &fused.tree().root().expr,
-                NativeExpr::Instant { op, .. } if op.timestamp
+                NativeExpr::Instant { op, .. } if op.value == promql::native::plan::InstantValue::Timestamp
             ));
         } else {
             assert!(matches!(
@@ -2539,7 +2539,13 @@ async fn native_fused_binary_keeps_independent_scan_occurrences() {
                 | NativeExpr::Merge { input }
                 | NativeExpr::Map { input, .. }
                 | NativeExpr::Timestamp { input } => sources(input, out),
-                NativeExpr::Binary { left, right, .. } | NativeExpr::Set { left, right, .. } => {
+                NativeExpr::Binary { left, right, .. }
+                | NativeExpr::Set { left, right, .. }
+                | NativeExpr::Info {
+                    base: left,
+                    info: right,
+                    ..
+                } => {
                     sources(left, out);
                     sources(right, out);
                 }
@@ -2713,12 +2719,16 @@ async fn native_fused_cross_table_pair_on_partitioned_tables_falls_back() {
 /// `some_metric` as [`native_two_table_provider`] builds it, and `other_metric` whose tags are
 /// `tag_0` and `field_0`: the second is the name of `some_metric`'s value field.
 async fn native_colliding_tag_provider() -> DfTableSourceProvider {
+    native_tables_provider(&[
+        ("some_metric", &["tag_0"], "timestamp", "field_0"),
+        ("other_metric", &["tag_0", "field_0"], "ts", "value"),
+    ])
+}
+
+/// Single-region tables of `(name, tags, time index, field)`.
+fn native_tables_provider(tables: &[(&str, &[&str], &str, &str)]) -> DfTableSourceProvider {
     let catalog_list = MemoryCatalogManager::with_default_setup();
-    let tables = [
-        ("some_metric", vec!["tag_0"], "timestamp", "field_0"),
-        ("other_metric", vec!["tag_0", "field_0"], "ts", "value"),
-    ];
-    for (table_id, (name, tags, time_index, field)) in (1024..).zip(tables) {
+    for (table_id, &(name, tags, time_index, field)) in (1024..).zip(tables) {
         let mut columns = tags
             .iter()
             .map(|tag| {
@@ -2802,6 +2812,97 @@ async fn native_fused_cross_table_union_with_a_colliding_column_falls_back() {
     assert!(find_fused_series(&plan).is_some(), "{plan}");
 }
 
+/// `up` and the info table `target_info`, whose data labels are `version` and `zone`.
+fn native_info_provider() -> DfTableSourceProvider {
+    native_tables_provider(&[
+        ("up", &["instance", "job"], "ts", "val"),
+        (
+            "target_info",
+            &["instance", "job", "version", "zone"],
+            "greptime_timestamp",
+            "greptime_value",
+        ),
+    ])
+}
+
+/// `info` reads its info table through a raw source of its own, after the base, and reports the
+/// data labels it can copy: every label of the info table, or only the ones the selector names.
+#[tokio::test]
+async fn native_fused_info_reads_the_info_table_after_the_base() {
+    for (query, columns) in [
+        (
+            "info(up)",
+            vec!["instance", "job", "ts", "val", "version", "zone"],
+        ),
+        (
+            "info(rate(up[5m]), {version=~\"v.+\"})",
+            vec!["instance", "job", "ts", "val", "version"],
+        ),
+    ] {
+        let plan = build_native_fused_plan_with_provider(query, true, native_info_provider()).await;
+        let fused = find_fused_series(&plan).unwrap_or_else(|| panic!("{query}: {plan}"));
+        assert!(
+            matches!(fused.tree().root().expr, NativeExpr::Info { .. }),
+            "{query}"
+        );
+        let tables = fused
+            .inputs()
+            .iter()
+            .map(|input| {
+                let LogicalPlan::Extension(Extension { node }) = input else {
+                    panic!("{query}: expected a raw source, got {input}");
+                };
+                let source = node
+                    .as_any()
+                    .downcast_ref::<RawSeriesSource>()
+                    .unwrap_or_else(|| panic!("{query}: expected a raw source, got {input}"));
+                source.metric_name().to_string()
+            })
+            .collect::<Vec<_>>();
+        assert_eq!(vec!["up", "target_info"], tables, "{query}");
+        assert_eq!(columns, sorted_column_names(&plan), "{query}");
+    }
+}
+
+/// `info` over the info table itself reads both scans of that one table under the node, which
+/// stays in the region: `rate` drops the base's name, so its series are enriched rather than
+/// passed through as info series.
+#[tokio::test]
+async fn native_fused_info_over_its_own_table_reads_two_scans_in_the_region() {
+    let query = "info(rate(target_info[5m]))";
+    let plan = build_native_fused_plan_with_provider(query, true, native_info_provider()).await;
+    let fused = find_fused_series(&plan).unwrap_or_else(|| panic!("{plan}"));
+    assert!(matches!(fused.tree().root().expr, NativeExpr::Info { .. }));
+    let tables = fused
+        .inputs()
+        .iter()
+        .map(|input| match input {
+            LogicalPlan::TableScan(scan) => scan.table_name.table().to_string(),
+            other => panic!("expected a direct scan, got {other}"),
+        })
+        .collect::<Vec<_>>();
+    assert_eq!(vec!["target_info", "target_info"], tables);
+    assert_eq!(Some("target_info"), fused.binding().metric_name.as_deref());
+    assert!(matches!(
+        crate::dist_plan::Categorizer::check_extension_plan(
+            fused as &dyn datafusion::logical_expr::UserDefinedLogicalNode,
+            &Default::default(),
+        ),
+        crate::dist_plan::Commutativity::Commutative
+    ));
+    assert_eq!(
+        vec![
+            "greptime_timestamp",
+            "greptime_value",
+            "instance",
+            "job",
+            "version",
+            "zone"
+        ],
+        sorted_column_names(&plan)
+    );
+}
+
 /// A function of each value or an operation with a constant around a pair keeps the pair native,
 /// on one table with both scans under the node.
 #[tokio::test]

tests-integration/src/tests/promql_test.rs

@@ -3731,6 +3731,141 @@ async fn native_promql_cross_table_comparisons() {
     }
 }
 
+/// `info()` over a base table and a `target_info` table whose time index and field are named
+/// differently, answered natively above both regions with Prometheus' answer, which the existing
+/// plan does not give: it has no `info`.
+///
+/// `target_info` holds `version="v1"` for `a` from 0s to 4s and `version="v2"` from 6s, and `v1` for
+/// `b` throughout; `c` has none. With a 2s lookback, `a` reads `v1` up to 5s and `v2` from 6s.
+#[tokio::test(flavor = "multi_thread")]
+async fn native_promql_info_enriches_from_the_info_table() {
+    common_telemetry::init_default_ut_logging();
+    let (standalone, observer) =
+        native_test_instance(GreptimeDbStandaloneBuilder::new("native_promql_info")).await;
+    let instance = standalone.fe_instance().clone();
+    execute_all(
+        &instance,
+        "CREATE TABLE native_info_up (instance STRING, job STRING, ts TIMESTAMP TIME INDEX, \
+         val DOUBLE, PRIMARY KEY(instance, job))",
+        QueryContext::arc(),
+    )
+    .await;
+    execute_all(
+        &instance,
+        "CREATE TABLE target_info (instance STRING, job STRING, \"version\" STRING, \
+         greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE, \
+         PRIMARY KEY(instance, job, \"version\"))",
+        QueryContext::arc(),
+    )
+    .await;
+    let mut up = Vec::new();
+    let mut info = Vec::new();
+    for second in 0..10_i64 {
+        for (target, job, base) in [("a", "j", 0), ("b", "j", 100), ("c", "k", 200)] {
+            up.push(format!(
+                "('{target}', '{job}', {}, {})",
+                second * 1_000,
+                base + second
+            ));
+        }
+        let a_version = match second {
+            0..=4 => Some("v1"),
+            5 => None,
+            _ => Some("v2"),
+        };
+        for (target, version) in [("a", a_version), ("b", Some("v1"))] {
+            if let Some(version) = version {
+                info.push(format!(
+                    "('{target}', 'j', '{version}', {}, 1)",
+                    second * 1_000
+                ));
+            }
+        }
+    }
+    for (table, values) in [("native_info_up", up), ("target_info", info)] {
+        execute_all(
+            &instance,
+            &format!("INSERT INTO {table} VALUES {}", values.join(", ")),
+            QueryContext::arc(),
+        )
+        .await;
+    }
+
+    let row = |second: i64, labels: &[(&str, &str)], value: i64| {
+        (
+            second * 1_000,
+            labels
+                .iter()
+                .map(|(name, value)| (name.to_string(), value.to_string()))
+                .collect::<std::collections::BTreeMap<_, _>>(),
+            value as f64,
+        )
+    };
+    let a = |version| [("instance", "a"), ("job", "j"), ("version", version)];
+    let b = [("instance", "b"), ("job", "j"), ("version", "v1")];
+    let c = [("instance", "c"), ("job", "k")];
+    // `a` takes the version current at each step, `b` keeps `v1`, and `c`, which no info series
+    // matches, passes through unless the selector requires a version.
+    let mut enriched = Vec::new();
+    let mut v1_only = Vec::new();
+    let mut shifted = Vec::new();
+    for second in 3..10_i64 {
+        let version = if second <= 5 { "v1" } else { "v2" };
+        enriched.push(row(second, &a(version), second));
+        enriched.push(row(second, &b, 100 + second));
+        enriched.push(row(second, &c, 200 + second));
+        if second <= 5 {
+            v1_only.push(row(second, &a("v1"), second));
+        }
+        v1_only.push(row(second, &b, 100 + second));
+        // An offset moves the info series along with the base: each step reads both a second back.
+        let version = if second <= 6 { "v1" } else { "v2" };
+        shifted.push(row(second, &a(version), second - 1));
+        shifted.push(row(second, &b, 99 + second));
+        shifted.push(row(second, &c, 199 + second));
+    }
+    for (query, columns, mut expected) in [
+        (
+            "info(native_info_up)",
+            vec!["instance", "job", "ts", "val", "version"],
+            enriched,
+        ),
+        (
+            "info(native_info_up, {version=\"v1\"})",
+            vec!["instance", "job", "ts", "val", "version"],
+            v1_only,
+        ),
+        (
+            "info(native_info_up offset 1s)",
+            vec!["instance", "job", "ts", "val", "version"],
+            shifted,
+        ),
+    ] {
+        expected.sort_by(|left, right| (&left.1, left.0).cmp(&(&right.1, right.0)));
+        let sql = format!("TQL EVAL (3, 9, '1s', '2s') {query}");
+        observer.clear();
+        let (native_columns, native) =
+            native_labelled_rows(native_tql_output(&instance, &sql, true).await).await;
+        let plan = observer.take();
+        assert_eq!(plan.name(), "FusedSeriesExec", "{query}");
+        assert_eq!(plan.children().len(), 2, "{query}");
+        assert_eq!(observer.take_raw_sources().len(), 2, "{query}");
+        assert_eq!(native_columns, columns, "{query}");
+        assert_eq!(native, expected, "{query}");
+
+        // The existing plan refuses `info`: "Unsupported expr type: info" for one argument, and
+        // "Cannot accept multiple vector as function input" for two.
+        let legacy = instance
+            .do_query(&sql, native_query_context(false, 4))
+            .await
+            .remove(0);
+        assert!(
+            legacy.is_err(),
+            "{query}: the existing plan answered `info`"
+        );
+    }
+}
+
 /// A two-selector query over a table split into regions never fuses the binary itself.
 ///
 /// A fused node with more than one input has no placement on such a table — the dist analyzer

tests-integration/tests/http.rs

@@ -1927,7 +1927,7 @@ async fn assert_raw_scan_fused_formats(observer: &NativeFusedObserver, frontend:
                 lookback_ms: 300_000,
                 offset_ms: 0,
                 at_ms: None,
-                timestamp: false,
+                value: promql::native::plan::InstantValue::Sample,
                 output_grid: Arc::clone(&grid),
             },
         },