feat(promql): execute native scalar roots and literal broadcasts (#9936)

I hereby agree to the terms of the [GreptimeDB
CLA](https://github.com/GreptimeTeam/.github/blob/main/CLA.md).

## Refer to a related PR or issue link (optional)

Stacked on #9868 (`feat/narthex-fused-wiring`, currently at
`990d932198ab0c5e41b724a54d990e5a8ffc8acb`). This change needs that PR's
unmerged `FusedSeries`/`SeriesScan`/in-process native route, so the base
is its branch; GitHub will retarget to `narthex` once #9868 merges.

## What's changed and what's your intention?

Enable the existing opt-in native PromQL route for literal vector/scalar
arithmetic such as `x * 100`, unary negation such as `-x`, and constant
scalar statements such as `1 + 2`. These public entries return ordinary
nonempty query RecordBatches with the same values and output schema as
the existing planner.

- Traverse unary and binary expression operands when locating the one
selector and its range requirements. Constant-only statements reuse the
existing constant evaluator and statement grid; unsupported shapes still
use the existing planner.
- Execute constant Scalar roots as one unlabelled evaluated chunk on the
full statement grid, with existing memory admission, output metrics and
error/cleanup paths. A no-input tree has genuinely zero logical/physical
children rather than a discarded placeholder scan.
- Keep each raw input partition attached to its own `FusedSeriesExec`
evaluator. This change adds neither an input-side Coalesce nor a
SeriesFrame exchange. Ordinary output collection continues to use
DataFusion.
- Cover scalar NaN/negative-zero/EOF/memory boundaries, child arity and
output schemas, independent lazy input progress, and the actual public
HTTP/query-handler route. The standalone single-region fixture persists
four disjoint input batches: for both broadcast and unary queries, all
four input/native partitions perform work and the complete result is
exactly 24 labelled timestamp/value rows. A four-stream barrier test
collected by DataFusion verifies concurrent progress; this is not a
CPU-overlap or speedup benchmark.
- Add a generated SQLness compatibility case for scalar arithmetic,
broadcast and unary results, rerun successfully without result changes.

Scope: constant scalar roots and literal broadcasts only.
`scalar(vector)` adaptation, dynamic per-grid scalar
arithmetic/broadcast, global Vector/Select and Complete Aggregate wiring
are separate work and are not enabled here. The experimental
configuration/request gates and current unsupported-shape fallback
remain in place. No persisted or remote wire format changes are
introduced; internal Fused constructor callers are migrated together.

Verification on this exact HEAD
(`d9eb9778f113dd9bd6125305a9ea447b4aa6ab1d`, restacked onto the current
dependency
`990d932198ab0c5e41b724a54d990e5a8ffc8acb`), exit 0:

```text
cargo fmt --all -- --check
git diff --check
cargo check --locked -p promql -p query --tests
cargo check --locked -p tests-integration --lib --test main
cargo nextest run --locked -p promql -p query --retries 0
  1870 passed, 2 skipped
cargo nextest run --locked -p tests-integration --test main --retries 0 \
  -E 'test(=integration_http_file_test::test_native_promql_route_plans_the_query) | test(=integration_http_file_test::test_prometheus_native_promql_route)'
  2 passed, 770 filtered out   (four real input partitions, exact 24 values)
cargo nextest run --locked -p tests-integration --lib --retries 0 -E 'test(native_promql_)'
  4 passed, 251 filtered out   (upstream VIEW/metric/ANALYZE compatibility)
cargo clippy --locked -p promql -p query --all-targets -- -D warnings
cargo clippy --locked -p tests-integration --lib --test main -- -D warnings
```

SQLness (generated, then rerun with an unchanged result):

```text
/mnt/nvme_rust/rust-targets-2/d-scalar-entry/debug/sqlness-runner bare \
  --bins-dir .../debug --case-dir <isolated root> --test-filter '^standalone:scalar_entry$' --jobs 1
```

This SQLness case uses the default public compatibility route, not the
native request flag; it does not replace the native execution assertions
above. Full workspace `make test`/`make clippy`/udeps and external CI
have not been run for this change.

## PR Checklist
Please convert it to a draft if some of the following conditions are not
met.

- [x] I have written the necessary rustdoc comments.
- [x] I have added the necessary unit tests and integration tests.
- [ ] This PR requires documentation updates.
- [x] API changes are backward compatible.
- [x] Schema or data changes are backward compatible.
- [ ] This PR needs to be backported to release branches, and I have
added the `backport-<target>` labels (e.g. `backport-v1.3` targets
`release/v1.3`).

src/promql/AGENTS.md

@@ -77,6 +77,10 @@ operators. Statement-to-plan translation lives in `src/query/src/promql/`.
   its doc is written for a caller that does not exist.
 - **A native window requires non-interleaved input.** Only `Contiguous` and
   `KeyThenTime` are accepted; `Interleaved` is refused when compiled.
+- **A fused tree runs ordinary raw children and a literal Scalar root.** Recursive native
+  execution keeps `source_index` routing for one or more raw children; only a validated
+  zero-source Scalar root is materialized on its bound grid without a scanner. It must not
+  displace the current multi-input vector path or the single-input partial aggregate path.
 - **A series closes exactly once**, on the chunk that says so — per input series,
   not per label set. A node that rewrites labels can hand two input series over
   under one label set, and each closes on its own chunk, so a label set may be

src/promql/src/extension_plan/fused_series.rs

@@ -56,6 +56,7 @@ use datafusion::execution::context::TaskContext;
 use datafusion::execution::memory_pool::{MemoryReservation, UnboundedMemoryPool};
 use datafusion::logical_expr::{Expr, LogicalPlan, UserDefinedLogicalNodeCore};
 use datafusion::physical_expr::{EquivalenceProperties, PhysicalExpr};
+use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet};
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
 use datafusion::physical_plan::{
@@ -498,6 +499,14 @@ fn validate_binding(
         )));
     }
     let schemas = collect_source_schemas(tree)?;
+    if schemas.is_empty()
+        && (!matches!(tree.root().expr, NativeExpr::Scalar(_))
+            || !matches!(tree.root().output, NativeType::Scalar(_)))
+    {
+        return Err(DataFusionError::Plan(
+            "FusedSeries without inputs requires a constant scalar root".into(),
+        ));
+    }
     if schemas.len() != child_count {
         return Err(DataFusionError::Plan(
             "FusedSeries source indexes must densely match its children".into(),
@@ -570,17 +579,7 @@ pub(crate) fn tree_source_fields(tree: &NativeTree) -> DataFusionResult<Vec<Vec<
 
 fn validate_physical_tree(tree: &NativeTree) -> DataFusionResult<Vec<Arc<SeriesSchema>>> {
     let schemas = collect_source_schemas(tree)?;
-    if schemas.is_empty() {
-        return Err(DataFusionError::Plan(
-            "FusedSeries physical plan requires at least one raw source".into(),
-        ));
-    }
-    if !tree
-        .root()
-        .output
-        .samples()
-        .is_some_and(|schema| schema.eval_grid.is_some())
-    {
+    if tree.root().output.grid().is_none() {
         return Err(DataFusionError::Plan(
             "FusedSeries output is not evaluated samples".into(),
         ));
@@ -599,6 +598,37 @@ pub(crate) enum OutputColumn {
     Tag(usize),
 }
 
+/// The caller-named value field used by the ordinary row conversion for a label-free scalar.
+fn scalar_root_fields(
+    output_schema: &SchemaRef,
+    tags: &[String],
+    time_column: &str,
+) -> DataFusionResult<Vec<FieldSpec>> {
+    if !tags.is_empty() {
+        return Err(DataFusionError::Plan(format!(
+            "FusedSeriesExec scalar root cannot produce {} tag columns",
+            tags.len()
+        )));
+    }
+    let mut values = output_schema
+        .fields()
+        .iter()
+        .filter(|field| field.name().as_str() != time_column);
+    let Some(value) = values.next() else {
+        return Err(DataFusionError::Plan(
+            "missing value column in scalar root output schema".to_string(),
+        ));
+    };
+    if values.next().is_some() {
+        return Err(DataFusionError::Plan(
+            "scalar root output schema must contain exactly one value column".into(),
+        ));
+    }
+    Ok(vec![FieldSpec {
+        name: Arc::from(value.name().as_str()),
+    }])
+}
+
 /// Classifies every output column, rejecting anything the conversion cannot produce.
 pub(crate) fn classify_output(
     output_schema: &SchemaRef,
@@ -789,7 +819,7 @@ impl FusedSeriesExec {
         partial_fold: Option<BoundAggregate>,
     ) -> DataFusionResult<Self> {
         let raw_schemas = validate_physical_tree(&tree)?;
-        if inputs.len() != raw_schemas.len() || inputs.is_empty() {
+        if inputs.len() != raw_schemas.len() {
             return Err(DataFusionError::Plan(format!(
                 "FusedSeries physical input count {} does not match its {} raw sources",
                 inputs.len(),
@@ -798,11 +828,8 @@ impl FusedSeriesExec {
         }
         // Building a state proves the runtime executes this tree; drop the validation state now.
         let _ = NativeExecutionState::try_new(&tree, Arc::new(UnboundedMemoryPool::default()))?;
-        let output_series = tree
-            .root()
-            .output
-            .samples()
-            .ok_or_else(|| DataFusionError::Plan("FusedSeries output is not samples".into()))?;
+        let output_series = tree.root().output.samples();
+        let scalar_root = raw_schemas.is_empty();
         if let Some(op) = &partial_fold {
             aggregate_op(op)?;
         }
@@ -827,15 +854,24 @@ impl FusedSeriesExec {
         }
         let classification = if partial_fold.is_some() {
             Vec::new()
+        } else if scalar_root {
+            let scalar_fields = scalar_root_fields(&output_schema, &output_tags, &time_column)?;
+            classify_output(&output_schema, &scalar_fields, &output_tags, &time_column)?
         } else {
             classify_output(
                 &output_schema,
-                &output_series.fields,
+                &output_series
+                    .ok_or_else(|| {
+                        DataFusionError::Plan("FusedSeries output is not samples".into())
+                    })?
+                    .fields,
                 &output_tags,
                 &time_column,
             )?
         };
-        let partition_count = inputs[0].0.properties().partitioning.partition_count();
+        let partition_count = inputs.first().map_or(1, |(input, _, _)| {
+            input.properties().partitioning.partition_count()
+        });
         for (index, (input, label_source, ordering)) in inputs.iter().enumerate() {
             let count = input.properties().partitioning.partition_count();
             if count == 0 {
@@ -868,9 +904,19 @@ impl FusedSeriesExec {
         }
         let properties = Arc::new(PlanProperties::new(
             EquivalenceProperties::new(output_schema.clone()),
-            Partitioning::UnknownPartitioning(if inputs.len() > 1 { 1 } else { partition_count }),
-            inputs[0].0.properties().emission_type,
-            inputs[0].0.properties().boundedness,
+            Partitioning::UnknownPartitioning(if inputs.len() != 1 {
+                1
+            } else {
+                partition_count
+            }),
+            inputs.first().map_or(EmissionType::Final, |(input, _, _)| {
+                input.properties().emission_type
+            }),
+            inputs
+                .first()
+                .map_or(Boundedness::Bounded, |(input, _, _)| {
+                    input.properties().boundedness
+                }),
         ));
         Ok(Self {
             promql,
@@ -989,9 +1035,9 @@ impl ExecutionPlan for FusedSeriesExec {
         partition: usize,
         context: Arc<TaskContext>,
     ) -> DataFusionResult<SendableRecordBatchStream> {
-        if self.inputs.len() > 1 && partition != 0 {
+        if partition >= self.properties.partitioning.partition_count() {
             return Err(DataFusionError::Execution(
-                "multi-source FusedSeries only has output partition 0".into(),
+                "FusedSeries output partition is out of range".into(),
             ));
         }
         let state = NativeExecutionState::try_new(
@@ -1047,9 +1093,13 @@ impl ExecutionPlan for FusedSeriesExec {
                 )
             })
             .transpose()?;
+        let mut executor = NativeExecutor::new(state);
+        if streams.is_empty() {
+            executor.end_of_input()?;
+        }
         Ok(Box::pin(FusedSeriesStream {
             inputs: streams,
-            executor: Some(NativeExecutor::new(state)),
+            executor: Some(executor),
             input_ids,
             classification: Arc::clone(&self.classification),
             tags: Arc::clone(&self.tags),

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

@@ -1900,14 +1900,244 @@ async fn later_source_error_drops_all_gathered_inputs() {
     assert_eq!(0, pool.reserved());
 }
 
-/// The physical boundary rejects zero sources and mismatched physical source counts.
+/// A literal scalar runs as one schema-checked, one-partition output without scan children.
+#[tokio::test]
+async fn physical_plan_accepts_scalar_root_without_sources() {
+    let statement_grid = grid();
+    let tree = scalar_tree(Arc::clone(&statement_grid), 4.0);
+    let binding = FusedSeriesBinding {
+        grid: statement_grid,
+        metric_name: None,
+        time_index: TIME.to_string(),
+        tags: vec![],
+    };
+    let schema = Arc::new(Schema::new(vec![
+        Field::new(VALUE, DataType::Float64, true),
+        Field::new(
+            TIME,
+            DataType::Timestamp(TimeUnit::Millisecond, None),
+            false,
+        ),
+    ]));
+    let node = FusedSeries::try_new(
+        Arc::from("4"),
+        tree,
+        binding.clone(),
+        Vec::new(),
+        schema.clone().to_dfschema_ref().unwrap(),
+    )
+    .unwrap();
+    let exec = node.to_execution_plan(Vec::new()).unwrap();
+    assert_eq!(1, exec.properties().output_partitioning().partition_count());
+    let rebuilt = exec
+        .clone()
+        .replace_children(
+            Vec::new(),
+            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+        )
+        .unwrap();
+    assert!(matches!(
+        exec.properties().boundedness,
+        Boundedness::Bounded
+    ));
+    assert!(exec.children().is_empty());
+    assert!(rebuilt.children().is_empty());
+    let zero_pool = Arc::new(GreedyMemoryPool::new(0));
+    let mut rejected = exec.execute(0, task_context(&zero_pool)).unwrap();
+    assert!(rejected.next().await.unwrap().is_err());
+    drop(rejected);
+    assert_eq!(0, zero_pool.reserved());
+    let pool = Arc::new(GreedyMemoryPool::new(1 << 20));
+    let mut stream = exec.execute(0, task_context(&pool)).unwrap();
+    let first = stream.next().await.unwrap().unwrap();
+    assert!(stream.next().await.is_none());
+    assert!(stream.next().await.is_none());
+    drop(stream);
+    let batches = [first];
+    assert_eq!(1, batches.len());
+    assert_eq!(schema.as_ref(), batches[0].schema().as_ref());
+    assert_eq!(GRID_LEN as usize, batches[0].num_rows());
+    let timestamps = batches[0]
+        .column_by_name(TIME)
+        .unwrap()
+        .as_any()
+        .downcast_ref::<TimestampMillisecondArray>()
+        .unwrap();
+    assert_eq!(grid_timestamps(), timestamps.values().as_ref());
+    let values = batches[0]
+        .column_by_name(VALUE)
+        .unwrap()
+        .as_any()
+        .downcast_ref::<Float64Array>()
+        .unwrap();
+    assert!(values.iter().all(|value| value == Some(4.0)));
+    assert_eq!(0, pool.reserved());
+    let rebuilt_pool = Arc::new(GreedyMemoryPool::new(1 << 20));
+    let rebuilt_batches = rebuilt
+        .execute(0, task_context(&rebuilt_pool))
+        .unwrap()
+        .try_collect::<Vec<_>>()
+        .await
+        .unwrap();
+    assert_eq!(1, rebuilt_batches.len());
+    assert_eq!(batches[0].schema(), rebuilt_batches[0].schema());
+    assert_eq!(batches[0].columns(), rebuilt_batches[0].columns());
+    assert_eq!(0, rebuilt_pool.reserved());
+    let drop_pool = Arc::new(GreedyMemoryPool::new(1 << 20));
+    let unpolled = rebuilt.execute(0, task_context(&drop_pool)).unwrap();
+    drop(unpolled);
+    assert_eq!(0, drop_pool.reserved());
+    let mut early = rebuilt.execute(0, task_context(&drop_pool)).unwrap();
+    let early_batch = early.next().await.unwrap().unwrap();
+    drop(early);
+    assert_eq!(0, drop_pool.reserved());
+    drop(early_batch);
+    assert_eq!(0, drop_pool.reserved());
+    assert!(exec.execute(1, task_context(&pool)).is_err());
+    let metrics = exec.metrics().unwrap();
+    assert_eq!(1, count(&metrics, "native.root.output_batches"));
+    assert_eq!(
+        GRID_LEN as usize,
+        count(&metrics, "native.root.output_values")
+    );
+    assert_eq!(
+        GRID_LEN as usize,
+        count(&metrics, "native.root.evaluated_grid_points")
+    );
+    assert_eq!(1, count(&metrics, "native.root.completed_series"));
+    assert_eq!(0, count(&metrics, "native.root.input_batches"));
+    assert_eq!(0, count(&metrics, "native.root.input_samples"));
+    let partial_error = FusedSeries::try_new_partial(
+        Arc::from("4"),
+        scalar_tree(grid(), 4.0),
+        binding,
+        vec![],
+        partial_op(AggregateKind::Sum, &[]),
+    )
+    .expect_err("scalar roots cannot be partial folds");
+    assert!(partial_error.to_string().contains("one raw child"));
+}
+
+#[tokio::test]
+async fn physical_scalar_rejects_malformed_output_schema() {
+    let tree = scalar_tree(grid(), 4.0);
+    let binding = FusedSeriesBinding {
+        grid: grid(),
+        metric_name: None,
+        time_index: TIME.to_string(),
+        tags: vec![],
+    };
+    let schemas = [
+        Schema::new(vec![Field::new(
+            TIME,
+            DataType::Timestamp(TimeUnit::Millisecond, None),
+            false,
+        )]),
+        Schema::new(vec![Field::new(VALUE, DataType::Float64, true)]),
+        Schema::new(vec![
+            Field::new(VALUE, DataType::Float64, true),
+            Field::new(OTHER, DataType::Float64, true),
+            Field::new(
+                TIME,
+                DataType::Timestamp(TimeUnit::Millisecond, None),
+                false,
+            ),
+        ]),
+        Schema::new(vec![
+            Field::new(VALUE, DataType::Utf8, true),
+            Field::new(
+                TIME,
+                DataType::Timestamp(TimeUnit::Millisecond, None),
+                false,
+            ),
+        ]),
+        Schema::new(vec![
+            Field::new(VALUE, DataType::Float64, true),
+            Field::new(TIME, DataType::Float64, false),
+        ]),
+    ];
+    for schema in schemas {
+        let node = FusedSeries::try_new(
+            Arc::from("4"),
+            tree.clone(),
+            binding.clone(),
+            Vec::new(),
+            Arc::new(schema).to_dfschema_ref().unwrap(),
+        )
+        .unwrap();
+        assert!(node.to_execution_plan(Vec::new()).is_err());
+    }
+    let tagged = FusedSeries::try_new(
+        Arc::from("4"),
+        tree,
+        FusedSeriesBinding {
+            tags: vec![HOST.to_string()],
+            ..binding
+        },
+        Vec::new(),
+        output_schema().to_dfschema_ref().unwrap(),
+    )
+    .unwrap();
+    assert!(tagged.to_execution_plan(Vec::new()).is_err());
+}
+
+#[tokio::test]
+async fn physical_scalar_conversion_preserves_special_float_bits() {
+    for value in [f64::NAN, -0.0, f64::INFINITY, f64::NEG_INFINITY] {
+        let statement_grid = grid();
+        let tree = scalar_tree(Arc::clone(&statement_grid), value);
+        let binding = FusedSeriesBinding {
+            grid: statement_grid,
+            metric_name: None,
+            time_index: TIME.to_string(),
+            tags: vec![],
+        };
+        let schema = Arc::new(Schema::new(vec![
+            Field::new(VALUE, DataType::Float64, true),
+            Field::new(
+                TIME,
+                DataType::Timestamp(TimeUnit::Millisecond, None),
+                false,
+            ),
+        ]));
+        let node = FusedSeries::try_new(
+            Arc::from("4"),
+            tree,
+            binding,
+            Vec::new(),
+            schema.to_dfschema_ref().unwrap(),
+        )
+        .unwrap();
+        let exec = node.to_execution_plan(Vec::new()).unwrap();
+        let batches = exec
+            .execute(0, SessionContext::default().task_ctx())
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        assert_eq!(1, batches.len());
+        assert_eq!(GRID_LEN as usize, batches[0].num_rows());
+        let actual = batches[0]
+            .column_by_name(VALUE)
+            .unwrap()
+            .as_any()
+            .downcast_ref::<Float64Array>()
+            .unwrap();
+        assert!(
+            actual
+                .iter()
+                .all(|actual| actual.unwrap().to_bits() == value.to_bits())
+        );
+    }
+}
+
 #[test]
 fn physical_plan_requires_matching_source_count() {
     let node = fused(rate_tree());
     let zero = node
         .to_execution_plan(Vec::new())
-        .expect_err("physical execution cannot run with zero sources");
-    assert!(zero.to_string().contains("source") || zero.to_string().contains("input"));
+        .expect_err("a raw-series tree cannot run with zero sources");
+    assert!(zero.to_string().contains("count") || zero.to_string().contains("source"));
 
     let source = memory_exec(Vec::new());
     let tuple = || {

src/promql/src/extension_plan/planner.rs

@@ -109,11 +109,14 @@ fn plan_fused_series(
     physical_inputs: &[Arc<dyn ExecutionPlan>],
     target_partitions: usize,
 ) -> DfResult<Option<Arc<dyn ExecutionPlan>>> {
-    if physical_inputs.len() != fused.children().len() || physical_inputs.is_empty() {
+    if physical_inputs.len() != fused.children().len() {
         return Err(DataFusionError::Plan(
             "FusedSeries physical child count mismatch".into(),
         ));
     }
+    if physical_inputs.is_empty() {
+        return fused.to_execution_plan(Vec::new()).map(Some);
+    }
     if physical_inputs
         .iter()
         .any(|input| input.downcast_ref::<RegionScanExec>().is_none())

src/promql/src/native/execute.rs

@@ -30,7 +30,7 @@ use datatypes::arrow::buffer::{Buffer, ScalarBuffer};
 use datatypes::series::histogram::HistogramView;
 use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
 use datatypes::series::samples::{
-    DenseSamplesRef, RowCursor, SampleRef, Samples, Series, SeriesFields, SeriesSchema,
+    DenseSamplesRef, FieldSpec, RowCursor, SampleRef, Samples, Series, SeriesFields, SeriesSchema,
 };
 use datatypes::series::time::{EvalGrid, GridRange, GridSpan, GridSpec, TimeAxis};
 use datatypes::series::{SeriesLabels, SeriesOrdering};
@@ -1974,6 +1974,24 @@ fn grid_position(grid: &EvalGrid, timestamp: i64) -> Result<usize> {
     Ok(position as usize)
 }
 
+/// One unlabelled constant value per statement grid point, closed on its only chunk.
+fn scalar_chunk(grid: Arc<EvalGrid>, value: f64) -> Result<EvaluatedChunk> {
+    let schema = SeriesSchema {
+        fields: Arc::from([FieldSpec {
+            name: "value".into(),
+        }]),
+        eval_grid: Some(Arc::clone(&grid)),
+    };
+    let len = grid.spec().len as usize;
+    let time = GridRange::try_new(grid, 0, len).map_err(series_error)?;
+    let values = Float64Array::from(vec![value; len]);
+    let fields = SeriesFields::one(Samples::from_float(values));
+    let metric = SeriesLabels::try_new(None, Vec::new()).map_err(series_error)?;
+    let series =
+        Series::try_new(&schema, metric, TimeAxis::Grid(time), fields).map_err(series_error)?;
+    EvaluatedChunk::try_new(series, true)
+}
+
 /// A closing chunk carrying `metric` and no row at all.
 ///
 /// A series whose selection picked no source row still closes exactly once, on a chunk that holds
@@ -2068,6 +2086,8 @@ struct InputSlot {
 pub struct NativeExecutionState {
     root: RuntimeNode,
     inputs: Vec<InputSlot>,
+    /// A constant root waiting for its only pull; operand scalars remain chunkless.
+    scalar_root: Option<(Arc<EvalGrid>, f64, NodeState)>,
 }
 
 impl NodeState {
@@ -2834,25 +2854,46 @@ impl NativeExecutionState {
     pub fn try_new(tree: &NativeTree, pool: Arc<dyn MemoryPool>) -> Result<Self> {
         let mut inputs = Vec::new();
         let (root, _) = RuntimeNode::build(tree.root(), &pool, &mut inputs)?;
-        match tree.root().output.samples() {
-            Some(schema) if schema.eval_grid.is_some() => {}
-            Some(_) => {
+        let scalar_root = if inputs.is_empty() {
+            match (&tree.root().expr, &tree.root().output) {
+                (NativeExpr::Scalar(BoundScalar { value }), NativeType::Scalar(grid)) => Some((
+                    Arc::clone(grid),
+                    *value,
+                    NodeState::new("native.scalar", Arc::clone(&root.metrics), &pool),
+                )),
+                _ => {
+                    return Err(DataFusionError::Execution(
+                        "native tree without inputs must be a scalar root".into(),
+                    ));
+                }
+            }
+        } else {
+            None
+        };
+        match &tree.root().output {
+            NativeType::Scalar(_) if inputs.is_empty() => {}
+            NativeType::Scalar(_) => {
+                return Err(DataFusionError::Execution(
+                    "native scalar root unexpectedly has input nodes".into(),
+                ));
+            }
+            NativeType::Samples(schema) if schema.eval_grid.is_some() => {}
+            NativeType::Samples(_) => {
                 return Err(DataFusionError::Execution(
                     "native runtime root produces raw samples".to_string(),
                 ));
             }
-            None => {
+            NativeType::AggregateState(_) => {
                 return Err(DataFusionError::Execution(
                     "native runtime root produces no sample output".to_string(),
                 ));
             }
         }
-        if inputs.is_empty() {
-            return Err(DataFusionError::Execution(
-                "native tree has no input node".to_string(),
-            ));
-        }
-        Ok(Self { root, inputs })
+        Ok(Self {
+            root,
+            inputs,
+            scalar_root,
+        })
     }
 
     /// Registers the metric handles of every node of this execution's tree with `set`.
@@ -2879,6 +2920,11 @@ impl NativeExecutionState {
             .map(|slot| slot.state.buffer.reservation.size())
             .fold(0usize, usize::saturating_add)
             .saturating_add(self.root.reserved_bytes())
+            .saturating_add(
+                self.scalar_root
+                    .as_ref()
+                    .map_or(0, |(_, _, state)| state.buffer.reservation.size()),
+            )
     }
 }
 
@@ -2894,6 +2940,27 @@ impl NativeExecutionState {
     /// `inputs_finished` is the caller's word that no more source chunks will arrive; it is passed
     /// down unchanged, because only a node that reads its whole input needs it.
     fn pull(&mut self, inputs_finished: bool) -> Result<Option<NativeChunk>> {
+        if let Some((grid, value, mut state)) = self.scalar_root.take() {
+            let len = usize::try_from(grid.spec().len).map_err(|_| {
+                DataFusionError::Execution("scalar grid length does not fit memory size".into())
+            })?;
+            let bytes = std::mem::size_of::<EvaluatedChunk>()
+                .checked_add(std::mem::size_of::<Float64Array>())
+                .and_then(|bytes| {
+                    len.checked_mul(std::mem::size_of::<f64>())
+                        .and_then(|values| bytes.checked_add(values))
+                })
+                .ok_or_else(|| {
+                    DataFusionError::Execution("scalar allocation size overflow".into())
+                })?;
+            state.buffer.grow(bytes)?;
+            let chunk = scalar_chunk(grid, value)?;
+            state.buffer.metrics.record_completed_series(1);
+            state
+                .queued
+                .push_back((NativeChunk::Evaluated(chunk), bytes));
+            return state.pop_queued(evaluated_values);
+        }
         self.root.pull(&mut self.inputs, inputs_finished)
     }
 
@@ -2929,6 +2996,9 @@ impl NativeExecutionState {
     fn close(&mut self) -> Result<()> {
         let finished = self.root.finish_windows();
         self.root.release();
+        if let Some((_, _, mut state)) = self.scalar_root.take() {
+            state.release();
+        }
         // The slots are taken rather than only released: what a released slot still holds is its
         // schema and the grid that schema is addressed on, and a caller that keeps the stopped
         // executor would otherwise keep the input's schema and materialized timestamps alive past
@@ -2948,6 +3018,9 @@ impl NativeExecutionState {
     /// vector and nothing left to take.
     fn release(&mut self) {
         self.root.release();
+        if let Some((_, _, mut state)) = self.scalar_root.take() {
+            state.release();
+        }
         for mut slot in std::mem::take(&mut self.inputs) {
             slot.release();
         }

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

@@ -703,6 +703,149 @@ fn run(executor: &mut NativeExecutor) -> Vec<EvaluatedChunk> {
     out
 }
 
+/// A scalar root answers its own value, once, on the grid it was bound to.
+///
+/// It reads nothing, so what it reports is not a series it was handed: one unlabelled chunk holding
+/// one value per grid point and closing on that same chunk. Nothing above it exists, so the value's
+/// field is the only field, and the chunk is addressed on the statement's grid rather than on one of
+/// its own.
+#[test]
+fn scalar_root_emits_one_unlabelled_grid_chunk() {
+    let grid = grid(10, 5, 3);
+    let tree = NativeTree::try_new(scalar(2.5, Arc::clone(&grid))).unwrap();
+    let pool = pool();
+    let mut execution = executor(&tree, Arc::clone(&pool));
+    execution.end_of_input().unwrap();
+    let chunk = execution.next_chunk().unwrap().unwrap();
+    assert_eq!(
+        chunk.series.metric,
+        SeriesLabels::try_new(None, Vec::new()).unwrap(),
+        "a scalar carries no metric name and no labels"
+    );
+    assert_eq!(chunk.series.fields.len(), 1, "one value per grid point");
+    assert_eq!(sample_values(&chunk), vec![Some(2.5); 3]);
+    assert_eq!(chunk.series.time.len(), 3);
+    assert_eq!(
+        chunk.series.time.timestamps().unwrap().as_ref(),
+        grid.timestamps().unwrap().as_ref(),
+        "the chunk's positions are the statement's grid"
+    );
+    assert!(chunk.is_last, "the one chunk closes the value it reports");
+    assert!(
+        execution.next_chunk().unwrap().is_none(),
+        "a scalar owes one chunk and no second one"
+    );
+    assert_eq!(execution.state().reserved_bytes(), 0);
+    assert_eq!(pool.reserved(), 0);
+}
+
+/// The scalar's chunk is reserved before it is allocated, and its own node reports it.
+///
+/// The pool is asked first, so a low budget refuses the chunk instead of allocating it: the failure
+/// leaves nothing held, and an execution stopped by it gives back everything it took. What the node
+/// reports is the one batch and the values it produced, and no input at all, because it reads none.
+#[test]
+fn scalar_root_reserves_before_allocating_and_counts_its_output() {
+    let grid = grid(0, 1, 4);
+    let tree = NativeTree::try_new(scalar(2.0, Arc::clone(&grid))).unwrap();
+    let zero_pool = bounded_pool(0);
+    let mut rejected = executor(&tree, Arc::clone(&zero_pool));
+    rejected.end_of_input().unwrap();
+    let error = rejected.next_chunk().unwrap_err();
+    assert!(error.to_string().contains("memory"), "{error}");
+    assert_eq!(rejected.state().reserved_bytes(), 0);
+    assert_eq!(zero_pool.reserved(), 0);
+    // A stopped execution holds nothing either, and says so rather than answering again.
+    assert!(rejected.next_chunk().is_err());
+
+    let pool = pool();
+    let metrics = ExecutionPlanMetricsSet::new();
+    let state = NativeExecutionState::try_new(&tree, Arc::clone(&pool)).unwrap();
+    state.register_metrics(&metrics, 0);
+    let mut execution = NativeExecutor::new(state);
+    execution.end_of_input().unwrap();
+    let chunk = execution.next_chunk().unwrap().unwrap();
+    assert_eq!(sample_values(&chunk), vec![Some(2.0); 4]);
+    assert!(execution.next_chunk().unwrap().is_none());
+    assert_eq!(pool.reserved(), 0);
+    let peak = "native.p0.root.peak_reserved_bytes";
+    assert!(gauge(&metrics, peak) >= 4 * std::mem::size_of::<f64>());
+    assert_eq!(count(&metrics, "root", NodeMetric::OUTPUT_BATCHES), 1);
+    assert_eq!(count(&metrics, "root", NodeMetric::OUTPUT_VALUES), 4);
+    assert_eq!(
+        count(&metrics, "root", NodeMetric::EVALUATED_GRID_POINTS),
+        4
+    );
+    assert_eq!(count(&metrics, "root", NodeMetric::COMPLETED_SERIES), 1);
+    assert_eq!(count(&metrics, "root", NodeMetric::INPUT_BATCHES), 0);
+    assert_eq!(count(&metrics, "root", NodeMetric::INPUT_SAMPLES), 0);
+}
+
+/// A scalar can produce its value before the input ends.
+///
+/// Nothing arrives for it, so waiting would be waiting for a chunk no producer will send. Ending the
+/// input afterwards changes nothing: the chunk was handed over once and the tree owes no second one.
+#[test]
+fn scalar_root_settles_before_the_input_ends() {
+    let tree = NativeTree::try_new(scalar(3.0, grid(0, 1, 3))).unwrap();
+    let mut execution = executor(&tree, pool());
+    assert_eq!(
+        sample_values(&execution.next_chunk().unwrap().unwrap()),
+        vec![Some(3.0); 3]
+    );
+    assert!(execution.next_chunk().unwrap().is_none());
+    execution.end_of_input().unwrap();
+    assert!(execution.next_chunk().unwrap().is_none());
+    assert_eq!(execution.state().reserved_bytes(), 0);
+}
+
+/// Stopped executors do not retain the scalar grid after handing off or rejecting the value.
+#[test]
+fn scalar_root_releases_grid_after_success_and_memory_failure() {
+    let pool = pool();
+    let (mut execution, weak_grid) = {
+        let grid = grid(0, 1, 3);
+        let tree = NativeTree::try_new(scalar(3.0, Arc::clone(&grid))).unwrap();
+        let weak = Arc::downgrade(&grid);
+        (executor(&tree, Arc::clone(&pool)), weak)
+    };
+    let chunk = execution.next_chunk().unwrap().unwrap();
+    assert!(weak_grid.upgrade().is_some());
+    drop(chunk);
+    execution.end_of_input().unwrap();
+    assert!(execution.next_chunk().unwrap().is_none());
+    assert!(weak_grid.upgrade().is_none());
+
+    let bounded = bounded_pool(0);
+    let (mut rejected, weak_grid) = {
+        let grid = grid(0, 1, 3);
+        let tree = NativeTree::try_new(scalar(3.0, Arc::clone(&grid))).unwrap();
+        let weak = Arc::downgrade(&grid);
+        (executor(&tree, Arc::clone(&bounded)), weak)
+    };
+    rejected.end_of_input().unwrap();
+    assert!(rejected.next_chunk().is_err());
+    assert_eq!(bounded.reserved(), 0);
+    assert!(weak_grid.upgrade().is_none());
+}
+
+/// Non-finite values and signed zero retain their bits rather than becoming missing samples.
+#[test]
+fn scalar_root_preserves_nan_negative_zero_and_infinity() {
+    for value in [f64::NAN, -0.0, f64::INFINITY, f64::NEG_INFINITY] {
+        let tree = NativeTree::try_new(scalar(value, grid(0, 1, 1))).unwrap();
+        let mut execution = executor(&tree, pool());
+        execution.end_of_input().unwrap();
+        let chunk = execution.next_chunk().unwrap().unwrap();
+        let actual = sample_values(&chunk)[0].expect("the point holds the constant");
+        assert_eq!(
+            actual.to_bits(),
+            value.to_bits(),
+            "the constant {value} is reported bit for bit"
+        );
+    }
+}
+
 fn sample_values(chunk: &EvaluatedChunk) -> Vec<Option<f64>> {
     field_values(chunk, 0)
 }

src/promql/src/native/plan.rs

@@ -141,7 +141,7 @@ pub struct BoundInput {
     pub ordering: SeriesOrdering,
 }
 
-/// A constant scalar bound for broadcast.
+/// A constant scalar bound for broadcast or a statement root.
 ///
 /// A scalar that varies per grid point is the same [`NativeType::Scalar`] as this constant: what
 /// keeps a binary operation from label-matching it is its type, not its constness. This binding

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

@@ -14,6 +14,7 @@
 
 use std::sync::Arc;
 
+use datafusion::common::DFSchema;
 use datatypes::series::SeriesOrdering;
 use datatypes::series::samples::{FieldSpec, SeriesSchema};
 use datatypes::series::time::{EvalGrid, GridSpec};
@@ -23,6 +24,7 @@ use super::{
     BoundVectorOp, BoundWindow, Grouping, MapOp, NativeExpr, NativeNode, NativeTree, NativeType,
     reads_across_series,
 };
+use crate::extension_plan::{FusedSeries, FusedSeriesBinding};
 use crate::map::MathOp;
 use crate::range_eval::{RangeFunction, RangeWindow};
 use crate::vector_match::{BinaryMatching, BinaryOp, Cardinality, FillValues, MatchingLabels};
@@ -1302,3 +1304,40 @@ fn a_vector_operation_cannot_consume_raw_samples() {
     let error = node.validate().unwrap_err();
     assert!(error.to_string().contains("an evaluated vector"), "{error}");
 }
+
+/// A node that reports samples while holding no input node is refused where the node is built.
+///
+/// The tree is constructed directly here — a child module of `native::plan` may — because the node
+/// is one the public constructors decline: a scalar root is the only zero-input shape a fused node
+/// accepts, and samples without a source would be an empty vector reported as an answer. Every
+/// public constructor therefore stays a checked one.
+#[test]
+fn a_samples_root_without_an_input_is_refused() {
+    let statement_grid = grid(0, 60_000, 4);
+    let tree = NativeTree {
+        root: NativeNode {
+            expr: NativeExpr::Scalar(BoundScalar { value: 1.0 }),
+            output: evaluated(Arc::new([]), Arc::clone(&statement_grid)),
+        },
+    };
+    let binding = FusedSeriesBinding {
+        grid: statement_grid,
+        time_index: "time".to_string(),
+        tags: vec![],
+        metric_name: None,
+    };
+
+    let error = FusedSeries::try_new(
+        Arc::from("1"),
+        tree,
+        binding,
+        Vec::new(),
+        Arc::new(DFSchema::empty()),
+    )
+    .expect_err("a samples root reads no source while reporting samples");
+
+    assert!(
+        error.to_string().contains("constant scalar"),
+        "the refusal must name the shape a zero-child tree has to be: {error}"
+    );
+}

src/query/src/promql/native_compile.rs

@@ -109,8 +109,8 @@
 //!
 //! Vector matching is bound with each selector's own input, including the labels and cardinality
 //! written on a vector–vector operation. A scalar has no input of its own: it is bound next to the
-//! vector that gives it something to broadcast over, while a statement whose value is only a scalar
-//! has nothing to read and is refused.
+//! vector that gives it something to broadcast over, while a statement whose value is only a
+//! constant is bound by [`compile_statement`] as that constant on the statement's grid.
 //!
 //! An aggregate is bound as [`AggregateStage::Complete`], which is a claim about the whole execution
 //! rather than about the node: it says that every contribution to every group is in this execution's
@@ -289,7 +289,21 @@ pub fn compile(
     Ok(compiled.tree)
 }
 
-/// Compiles a statement and returns a binding record for each selector occurrence.
+/// Compiles a constant statement to a scalar on the statement's grid, reading no input.
+///
+/// A value that depends on data or on the evaluation time is refused, which the same fold
+/// [`compile_expr`] uses for a constant operand decides ([`constant_value`]).
+pub fn compile_statement(stmt: &EvalStmt) -> Result<NativeTree> {
+    let value = constant_value(&stmt.expr)
+        .ok_or_else(|| not_implemented(&format!("the expression `{}`", stmt.expr)))?;
+    let grid = statement_grid(stmt)?;
+    NativeTree::try_new(NativeNode {
+        expr: NativeExpr::Scalar(BoundScalar { value }),
+        output: NativeType::Scalar(grid),
+    })
+}
+
+/// Compiles `stmt` into a tree and returns a binding record for each selector occurrence.
 ///
 /// `bindings` is called once per selector with its requirements. The resulting source indices are
 /// occurrence numbers in compiler traversal order; the caller maps them to scan children.
@@ -305,6 +319,8 @@ pub fn compile_selectors(
              series may not interleave",
         ));
     }
+    // A tree with selector inputs reports evaluated vector samples. A constant root has no input
+    // and is bound separately by [`compile_statement`].
     if stmt.expr.value_type() != ValueType::Vector {
         return Err(not_implemented(&format!(
             "the expression `{}`, which is a {} where a tree reports evaluated samples",
@@ -1808,7 +1824,8 @@ fn refuses_modifier(modifier: Option<&BinModifier>) -> bool {
 /// semantics for `%`, `^` and `atan2` live, the last two through `go_math`, so defining them again
 /// here would be a second answer for the two to disagree over. An operator with no
 /// [`ArithmeticOp`] — a comparison, `and`, `or`, `unless` — is not arithmetic and does not fold.
-fn constant_value(expr: &PromExpr) -> Option<f64> {
+/// Visible to the planner, which asks whether a statement's value is a constant before binding it.
+pub(crate) fn constant_value(expr: &PromExpr) -> Option<f64> {
     match expr {
         PromExpr::NumberLiteral(NumberLiteral { val }) => Some(*val),
         PromExpr::Paren(paren) => constant_value(&paren.expr),

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

@@ -38,7 +38,8 @@ use promql_parser::parser::{
 };
 
 use super::{
-    CompiledInput, SelectorRequirement, SelectorTiming, compile, compile_selectors, validate_inputs,
+    CompiledInput, SelectorRequirement, SelectorTiming, compile, compile_selectors,
+    compile_statement, validate_inputs,
 };
 
 /// Whether a matching names its labels with `on` (`true`) or `ignoring` (`false`), and which.
@@ -1865,6 +1866,41 @@ fn a_statement_with_no_vector_is_refused() {
     }
 }
 
+/// A constant statement compiles to the constant on the statement's grid, and it reads nothing.
+///
+/// The vector entry above refuses it — a tree over that entry reports evaluated samples — so the
+/// statement entry is the one that binds it, and the grid it reports on is the statement's own.
+#[test]
+fn a_constant_statement_is_a_scalar_on_the_statement_grid_without_input() {
+    for (query, value) in [("1 + 2", 3.0), ("-1", -1.0)] {
+        let stmt = eval_stmt(query);
+        let tree = compile_statement(&stmt)
+            .unwrap_or_else(|error| panic!("`{query}` was refused: {error}"));
+        // The root is the number the statement means, on the statement's own grid: 100 seconds in
+        // 5-second steps is 21 points, and no point is the constant's to choose.
+        assert_eq!(Some(value), constant(tree.root()), "query: {query}");
+        assert_eq!((0, 5_000, 21), scalar_grid(tree.root()), "query: {query}");
+    }
+}
+
+/// A statement whose value comes from the data or from the evaluation time has no constant to bind,
+/// so the statement entry refuses it rather than compiling a tree with nothing to read.
+#[test]
+fn a_statement_that_is_not_constant_has_no_statement_compilation() {
+    for query in ["time()", "time() + 1", "vector(1)", "x"] {
+        let error = compile_statement(&eval_stmt(query))
+            .expect_err(&format!("`{query}` compiled, and it should not have"));
+        assert!(
+            matches!(error, DataFusionError::NotImplemented(_)),
+            "`{query}` did not report NotImplemented: {error}"
+        );
+        assert!(
+            error.to_string().contains(&format!("`{query}`")),
+            "`{query}` did not name the expression it found: {error}"
+        );
+    }
+}
+
 /// A vector-matching modifier has nothing to match when one operand is a constant, and the tree's
 /// matching has no field for a fill value. Both are refused rather than dropped: dropping one would
 /// compile a query into an operation other than the one it wrote. `bool` is not a matching modifier

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

@@ -29,6 +29,8 @@
 //! built to the shape the existing unary pipeline produces — the time index, then the series
 //! fields, then the tags — so a node planned here can be consumed by the same code that consumes a
 //! `SeriesNormalize` output.
+//!
+//! A constant statement is planned with no scan: one node, no child, the time index and value.
 
 use std::sync::Arc;
 use std::time::{Duration, UNIX_EPOCH};
@@ -56,8 +58,10 @@ use snafu::ResultExt;
 use store_api::metric_engine_consts::{DATA_SCHEMA_TABLE_ID_COLUMN_NAME, METRIC_ENGINE_NAME};
 
 use crate::promql::error::{DataFusionPlanningSnafu, Result};
-use crate::promql::native_compile::{compile_selectors, statement_grid};
-use crate::promql::planner::PromPlanner;
+use crate::promql::native_compile::{
+    compile_selectors, compile_statement, constant_value, statement_grid,
+};
+use crate::promql::planner::{DEFAULT_FIELD_COLUMN, DEFAULT_TIME_INDEX_COLUMN, PromPlanner};
 use crate::query_engine::QueryEngineState;
 
 /// The query-context key that turns the native fused-series route on for one query.
@@ -141,10 +145,7 @@ impl PromPlanner {
         }
 
         let mut selectors = Vec::new();
-        if !collect_selectors(prom_expr, &mut selectors)
-            || selectors.is_empty()
-            || selectors.len() > 2
-        {
+        if !collect_selectors(prom_expr, &mut selectors) || selectors.len() > 2 {
             return Ok(None);
         }
         // Preserve the original single-selector admission allowlist. Broader binary syntax is only
@@ -158,6 +159,39 @@ impl PromPlanner {
         let Ok(grid) = statement_grid(&stmt) else {
             return Ok(None);
         };
+        // No selector is not proof of a constant: `time()` and other data-free functions are
+        // unsupported here, so the existing fold is asked before a zero-child node is bound.
+        if selectors.is_empty() {
+            if constant_value(prom_expr).is_none() {
+                return Ok(None);
+            }
+            let tree = match compile_statement(&stmt) {
+                Ok(tree) => tree,
+                Err(DataFusionError::NotImplemented(_)) => return Ok(None),
+                Err(error) => return Err(error).context(DataFusionPlanningSnafu),
+            };
+            // A constant has no scan: no metric, no tags, and the value under the name the existing
+            // path gives a bare literal.
+            let time_index = DEFAULT_TIME_INDEX_COLUMN.to_string();
+            let value = DEFAULT_FIELD_COLUMN.to_string();
+            let output_schema = fused_output_schema(&[value], &[], &time_index)?;
+            let node = FusedSeries::try_new(
+                Arc::from(prom_expr.to_string()),
+                tree,
+                FusedSeriesBinding {
+                    grid,
+                    time_index,
+                    tags: Vec::new(),
+                    metric_name: None,
+                },
+                Vec::new(),
+                output_schema,
+            )
+            .context(DataFusionPlanningSnafu)?;
+            return Ok(Some(LogicalPlan::Extension(Extension {
+                node: Arc::new(node),
+            })));
+        }
         let original_context = self.ctx.clone();
         let preparation = async {
             let mut prepared: Vec<PreparedInput> = Vec::new();
@@ -742,6 +776,13 @@ fn admitted_selector(expr: &PromExpr) -> Option<&VectorSelector> {
         {
             admitted_selector(&call.args.args[0])
         }
+        PromExpr::Unary(unary) => admitted_selector(&unary.expr),
+        PromExpr::Binary(binary) if constant_value(&binary.lhs).is_some() => {
+            admitted_selector(&binary.rhs)
+        }
+        PromExpr::Binary(binary) if constant_value(&binary.rhs).is_some() => {
+            admitted_selector(&binary.lhs)
+        }
         _ => None,
     }
 }
@@ -958,8 +999,12 @@ mod tests {
     }
 
     #[test]
-    fn single_selector_admission_keeps_scalar_binaries_on_fallback() {
-        for query in ["x * 2", "2 - x", "(x * 2) + 3"] {
+    fn single_selector_admission_accepts_literal_broadcasts_and_unary() {
+        for query in ["x * 2", "2 - x", "(x * 2) + 3", "-x"] {
+            let expression = promql_parser::parser::parse(query).unwrap();
+            assert!(admitted_selector(&expression).is_some(), "{query}");
+        }
+        for query in ["time() + x", "scalar(x) + x", "x + y"] {
             let expression = promql_parser::parser::parse(query).unwrap();
             assert!(admitted_selector(&expression).is_none(), "{query}");
         }

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

@@ -1859,6 +1859,39 @@ fn find_fused_series(plan: &LogicalPlan) -> Option<&FusedSeries> {
     plan.inputs().into_iter().find_map(find_fused_series)
 }
 
+/// Literal roots need no raw scan, while a literal beside a selector broadcasts over that input.
+#[tokio::test]
+async fn native_fused_supports_constant_roots_and_selector_broadcasts() {
+    for (query, value) in [("1 + 2", 3.0), ("-1", -1.0)] {
+        let plan = build_native_fused_plan(query, true).await;
+        let fused = find_fused_series(&plan).expect("constant root uses native route");
+        assert!(
+            fused.children().is_empty(),
+            "{query}: {:?}",
+            fused.children()
+        );
+        let NativeExpr::Scalar(scalar) = &fused.tree().root().expr else {
+            panic!("{query}: {:?}", fused.tree().root().expr);
+        };
+        assert_eq!(value, scalar.value, "{query}");
+    }
+
+    for query in [
+        "some_metric * 100",
+        "100 / some_metric",
+        "-some_metric",
+        "rate(some_metric[5m]) * 100",
+    ] {
+        let plan = build_native_fused_plan(query, true).await;
+        let fused = find_fused_series(&plan).expect("selector expression uses native route");
+        assert_eq!(1, fused.children().len(), "{query}");
+        assert!(
+            matches!(fused.children()[0], LogicalPlan::TableScan(_)),
+            "{query}"
+        );
+    }
+}
+
 /// The flag on and a shape the compiler accepts: the statement plans to one native node.
 #[tokio::test]
 async fn native_fused_rate_plans_to_a_fused_series_node() {

tests-integration/src/tests/native_aggregate_test.rs

@@ -44,6 +44,19 @@
 //! partial states that only a merge over *all* of them turns into `sum`, `avg`, `count` and
 //! `stddev` - while the gauge queries keep covering the shapes with round answers.
 //!
+//! The counter table carries one tag the gauge table does not, `region`: the four flushed series -
+//! the four files - are in one region and the series left in the memtable range is in another, so
+//! one query can name a matcher (`{region="us-west-2"}`) that only four of the five scan ranges
+//! satisfy. No other counter query reads that tag: an aggregate drops the labels it does not group
+//! by, so their answers are the ones the untagged fixture gives.
+//!
+//! A query the native route declines is checked the other way round. [`check_fallback_case`] runs
+//! shapes the compiler has no node for - `time()`, `scalar(v)`, a fold over two sources and a
+//! subquery - with the native request on and with it off, and requires that no native node ran in
+//! either plan and that both runs report rows derived from the fixture by hand. A hint that quietly
+//! took the route for one of them would show a fused producer or a frontend merge in the plans that
+//! executed, which is what is observed.
+//!
 //! A binary operation whose sibling is a literal scalar is carried by the split on either side of
 //! the fold: below the aggregate it is computed in the per-series prefix (`sum(8 * rate(...))`,
 //! `max(10 - rate(...))`), and above it the merge's suffix rebuilds it over the merged value with
@@ -51,7 +64,8 @@
 //! `-sum(rate(...))`). Deliberately not covered yet: the aggregate split carries a literal scalar
 //! as its only binary sibling, so a computed scalar (`scalar(...)`, `time()`) or a whole-vector
 //! combination such as `sort`, `absent` or `and`/`or`/`unless` beside the aggregate keeps the
-//! existing plan.
+//! existing plan. [`check_fallback_case`] pins that the computed scalars do, and that a fold over a
+//! second source and a subquery keep it too.
 
 use std::collections::BTreeMap;
 use std::sync::{Arc, Mutex};
@@ -150,6 +164,16 @@ const COUNTER_SERIES: [(&str, &str, f64); 5] = [
     ("b", "h5", 600.0),
 ];
 
+/// The `region` tag of the four counter series that were flushed, one series per file.
+///
+/// Only [`COUNTER_TABLE`] carries the tag, and only the matcher case reads it: the four flushed
+/// series are in this region, so `{region="us-west-2"}` selects the four files and drops the series
+/// the memtable range holds. Every other counter query aggregates the tag away.
+const COUNTER_WEST_REGION: &str = "us-west-2";
+/// The `region` tag of the counter series that stays in the memtable range: `job = b`,
+/// `instance = h5`, whose rate is 4.
+const COUNTER_EAST_REGION: &str = "us-east-1";
+
 /// Observes the physical plan each query engine is about to execute, and keeps it.
 ///
 /// The frontend and the datanode it reads its regions from install the same plugin instance, so both
@@ -319,9 +343,10 @@ struct AggregateCase {
     expected: Vec<ResultRow>,
     /// Whether every one of the four scan partitions must report a folded value.
     per_partition: bool,
-    /// The number of values all four partitions' producers must report together, when the answer
-    /// fixes that figure: one value per series that folded, which is what the counter fixture's
-    /// queries make checkable. [`None`] leaves only "some value" to the assertion.
+    /// The number of values the partitions' producers must report together, when the answer fixes
+    /// that figure: one value per series that folded, which is what the counter fixture's queries
+    /// make checkable. With `per_partition` those are all four partitions' values and their spread
+    /// is bounded too. `None` leaves only "some value" to the assertion.
     root_values: Option<usize>,
     /// The `partial state:` line the producer must show, when the fixture fixes which aggregate the
     /// split cut out. The producer then exports that aggregate's *state*, not an answer, so a
@@ -428,6 +453,20 @@ fn create_table_sql(table: &str) -> String {
     )
 }
 
+/// The DDL of [`COUNTER_TABLE`]: [`create_table_sql`]'s columns plus the `region` tag.
+///
+/// The tag is part of the primary key, so it is a label the raw-series scan reports and a matcher
+/// can select, and it costs the other counter queries nothing: an aggregate that does not group by
+/// it drops it. Only the counter table carries it, because only one counter query reads it; the
+/// gauge table keeps the shape [`create_table_sql`] builds for every fixture table.
+fn counter_table_sql() -> String {
+    format!(
+        "CREATE TABLE {COUNTER_TABLE} (ts TIMESTAMP(3) TIME INDEX, job STRING, instance STRING, \
+         \"region\" STRING, v DOUBLE, PRIMARY KEY (job, instance, \"region\")) WITH \
+         ('compaction.type' = 'twcs', 'compaction.twcs.trigger_file_num' = '100')"
+    )
+}
+
 /// Fills the fixture tables.
 ///
 /// The gauge table gets one flush per series but the last, so four files and one memtable range
@@ -435,10 +474,12 @@ fn create_table_sql(table: &str) -> String {
 /// series'. The counter table is filled the same way, with the same keys and the same flush after
 /// each of the first four series, so the scan sees the same five series in the same four files and
 /// one memtable range - but its samples are the two of [`COUNTER_SERIES`], [`COUNTER_STEP_SECS`]
-/// apart and 140s and 200s before [`EVAL_SECS`]. The range table gets one series whose samples are
-/// 390s apart: both are inside a 10m range at [`EVAL_SECS`] and the older one is outside the default
-/// 5m lookback, so a query that reads its own range counts two samples where one that used the
-/// lookback would count one.
+/// apart and 140s and 200s before [`EVAL_SECS`], and its rows carry the `region` tag of the file
+/// they were flushed into ([`COUNTER_WEST_REGION`]) or, for the series left in the memtable range,
+/// [`COUNTER_EAST_REGION`]. The range table gets one series whose samples are 390s apart: both are
+/// inside a 10m range at [`EVAL_SECS`] and the older one is outside the default 5m lookback, so a
+/// query that reads its own range counts two samples where one that used the lookback would count
+/// one.
 async fn setup_fixture(instance: &Arc<Instance>) {
     execute_sql(instance, &create_table_sql(GAUGE_TABLE)).await;
     for (index, (job, host, v)) in GAUGE_SERIES.iter().enumerate() {
@@ -457,19 +498,26 @@ async fn setup_fixture(instance: &Arc<Instance>) {
         }
     }
 
-    execute_sql(instance, &create_table_sql(COUNTER_TABLE)).await;
+    execute_sql(instance, &counter_table_sql()).await;
     let (counter_first, counter_second) = (
         COUNTER_FIRST_SECS * 1000,
         (COUNTER_FIRST_SECS + COUNTER_STEP_SECS) * 1000,
     );
     for (index, (job, host, increase)) in COUNTER_SERIES.iter().enumerate() {
         let second_value = COUNTER_FIRST_VALUE + increase;
+        // The four flushed series are the ones the matcher case selects; the series that stays in
+        // the memtable range is the only one in the other region.
+        let region = if index + 1 < COUNTER_SERIES.len() {
+            COUNTER_WEST_REGION
+        } else {
+            COUNTER_EAST_REGION
+        };
         execute_sql(
             instance,
             &format!(
-                "INSERT INTO {COUNTER_TABLE} (ts, job, instance, v) VALUES \
-                 ({counter_first}, '{job}', '{host}', {COUNTER_FIRST_VALUE:?}), \
-                 ({counter_second}, '{job}', '{host}', {second_value:?})"
+                "INSERT INTO {COUNTER_TABLE} (ts, job, instance, \"region\", v) VALUES \
+                 ({counter_first}, '{job}', '{host}', '{region}', {COUNTER_FIRST_VALUE:?}), \
+                 ({counter_second}, '{job}', '{host}', '{region}', {second_value:?})"
             ),
         )
         .await;
@@ -726,6 +774,39 @@ fn assert_finalize_reads_region_state(
         1, boundary_partitions,
         "{context}: one region is read, so the boundary exposes one output partition:\n{plan}"
     );
+    // What the producer exports is named here, column by column, before the merge's input schema is
+    // compared with it: two schemas that agree on finished values would agree with each other just
+    // as well, and a merge over answers is not the bridge this case is about. The state layout is
+    // the metric name, one column per tag, the point and field a value belongs to, the four
+    // accumulators and the incremental-mean flag; the tag count is read off the schema, so the
+    // assertion holds for both fixtures without repeating either tag list.
+    const STATE_ACCUMULATORS: [&str; 5] =
+        ["value", "mean", "compensation", "count", "incremental_mean"];
+    const STATE_POSITION: [&str; 2] = ["point", "field"];
+    let state_columns = producer
+        .schema()
+        .fields()
+        .iter()
+        .map(|field| field.name().clone())
+        .collect::<Vec<_>>();
+    let fixed_columns = 1 + STATE_POSITION.len() + STATE_ACCUMULATORS.len();
+    assert!(
+        state_columns.len() >= fixed_columns,
+        "{context}: the producer must export aggregate state, but its {} columns are \
+         {state_columns:?}:\n{plan}",
+        state_columns.len()
+    );
+    let tags = state_columns.len() - fixed_columns;
+    let mut expected_columns = Vec::with_capacity(state_columns.len());
+    expected_columns.push("metric_name".to_string());
+    expected_columns.extend((0..tags).map(|at| format!("tag_{at}")));
+    expected_columns.extend(STATE_POSITION.map(str::to_string));
+    expected_columns.extend(STATE_ACCUMULATORS.map(str::to_string));
+    assert_eq!(
+        expected_columns, state_columns,
+        "{context}: the producer must export the aggregate state layout, not one value per \
+         group:\n{plan}"
+    );
     assert_eq!(
         producer.schema(),
         merge_input.schema(),
@@ -733,6 +814,41 @@ fn assert_finalize_reads_region_state(
     );
 }
 
+/// Checks one query the native route declines: it must stay declined on both routes, and both must
+/// answer what the fixture derives by hand.
+///
+/// The native request is on for one run and off for the other, so the hint being asked for and the
+/// hint being absent are both exercised: neither plan may hold a fused producer or a frontend merge,
+/// and the rows - labels, evaluation time, values and their multiplicity - are the same hand-derived
+/// ones for both. The observer is asked to have seen a plan at all, because a run that handed nothing
+/// over would otherwise satisfy "no native node" without observing anything.
+async fn check_fallback_case(
+    instance: &Arc<Instance>,
+    observer: &NativePlanObserver,
+    case: &AggregateCase,
+) {
+    let promql = case.promql.as_str();
+    let query = tql_eval(promql);
+    for native in [false, true] {
+        let route = if native { "native" } else { "legacy" };
+        let mark = observer.mark();
+        let (result, capture) = run_query(instance, promql, native, observer).await;
+        assert!(
+            observer.mark() > mark,
+            "`{query}`: the {route} run handed no plan to the observer, so nothing about the plan \
+             that ran was observed"
+        );
+        assert!(
+            capture.fused.is_empty() && capture.finalize.is_empty(),
+            "`{query}`: the {route} run must keep the existing plan, but the plans that ran held {} \
+             fused producers and {} frontend merges",
+            capture.fused.len(),
+            capture.finalize.len()
+        );
+        assert_rows(&result.rows, &case.expected, &format!("{route} `{query}`"));
+    }
+}
+
 /// Checks one case on both routes: the legacy run first, then the native one.
 async fn check_case(instance: &Arc<Instance>, observer: &NativePlanObserver, case: &AggregateCase) {
     let promql = case.promql.as_str();
@@ -801,34 +917,36 @@ async fn check_case(instance: &Arc<Instance>, observer: &NativePlanObserver, cas
         );
     }
     if let Some(root_values) = case.root_values {
-        // One value per series that folded, so the four partitions' counts add up to the fixture's
-        // series count. With every partition nonempty that leaves `root_values` split as evenly as
-        // four positive counts allow, and no partition holding more than that: a group of three
-        // series (`count by (job)` below reports three for `job = a`) therefore cannot sit in one
-        // partition, so the group's states have to be merged across at least two of them.
+        // One value per series that folded, so the partitions' counts add up to the number of series
+        // the scan selected - every counter series, or only the ones a matcher kept.
         let counts = per_partition
             .iter()
             .map(|values| values.unwrap_or(0))
             .collect::<Vec<_>>();
-        assert!(
-            counts.iter().all(|count| *count > 0),
-            "`{query}`: all {SCAN_PARTITIONS} scan partitions must fold a counter series, got \
-             {counts:?}"
-        );
         assert_eq!(
             root_values,
             counts.iter().sum::<usize>(),
-            "`{query}`: the partitions must fold one value per counter series, {root_values} in \
+            "`{query}`: the partitions must fold one value per selected series, {root_values} in \
              all, got {counts:?}"
         );
-        let largest = counts.iter().copied().max().unwrap_or(0);
-        assert_eq!(
-            root_values - (SCAN_PARTITIONS - 1),
-            largest,
-            "`{query}`: {root_values} values over {SCAN_PARTITIONS} nonempty partitions leave no \
-             partition holding more than {}, got {counts:?}",
-            root_values - (SCAN_PARTITIONS - 1)
-        );
+        // A case whose four partitions are all asserted nonempty above splits those values as
+        // evenly as four positive counts allow, so no partition holds more than
+        // `root_values - (SCAN_PARTITIONS - 1)`: a group of three series (`count by (job)` below
+        // reports three for `job = a`) therefore cannot sit in one partition, and the group's
+        // states have to be merged across at least two of them. A case a matcher narrows pins only
+        // the total: which partitions keep a selected series depends on how the scan ranges were
+        // spread, and the answer itself - not a bound on the split - is what shows the unselected
+        // series were dropped.
+        if case.per_partition {
+            let largest = counts.iter().copied().max().unwrap_or(0);
+            assert_eq!(
+                root_values - (SCAN_PARTITIONS - 1),
+                largest,
+                "`{query}`: {root_values} values over {SCAN_PARTITIONS} nonempty partitions leave \
+                 no partition holding more than {}, got {counts:?}",
+                root_values - (SCAN_PARTITIONS - 1)
+            );
+        }
     }
 }
 
@@ -866,11 +984,18 @@ async fn native_aggregate_complete_over_four_nonempty_mito_partitions() {
             &by_job(18.0, 12.0),
             true,
         ),
-        case(
-            format!("avg by (job) ({GAUGE_TABLE})"),
-            &by_job(6.0, 6.0),
-            true,
-        ),
+        // The gauge's groups average to 6 and 6, and the producer's plan line has to name the
+        // `avg by (job)` the split cut out: what a partition exports is that aggregate's state, not
+        // the group's mean. The counter table's cases below are the ones whose answers can tell a
+        // merged state from a finished one.
+        AggregateCase {
+            partial_state: Some("partial state: avg by (job) Complete"),
+            ..case(
+                format!("avg by (job) ({GAUGE_TABLE})"),
+                &by_job(6.0, 6.0),
+                true,
+            )
+        },
         case(
             format!("count by (job) ({GAUGE_TABLE})"),
             &by_job(3.0, 2.0),
@@ -921,15 +1046,24 @@ async fn native_aggregate_complete_over_four_nonempty_mito_partitions() {
         // folds them together, so the answers are only reachable when every partition's state
         // arrives: `count` reports 3 and 2, and `avg`/`stddev` fold states that do not decompose
         // into per-partition values. `job = a`'s three series cannot all be in one of the four
-        // partitions - the per-partition value counts asserted for every counter case here rule
-        // that out - so its mean and its standard deviation are merged across at least two
-        // partitions. A standard deviation finished per partition and then averaged would report
-        // 0.5 or 1.0 for the 2+1 splits the counts leave possible, never sqrt(8/3).
+        // partitions - the per-partition counts asserted for the counter cases that read every
+        // series rule that out - so its mean and its standard deviation are merged across at least
+        // two partitions. A standard deviation finished per partition and then averaged would
+        // report 0.5 or 1.0 for the 2+1 splits the counts leave possible, never sqrt(8/3).
         counter_case(
             format!("sum(rate({COUNTER_TABLE}[5m]))"),
             &[(&[], 15.0)],
             "partial state: sum by () Complete",
         ),
+        // The same sum with a literal scalar above the fold: the merge's suffix doubles the merged
+        // 15 to 30, where a suffix run over one partition's share - at most 10 of the 15 - would
+        // answer at most 20, and a suffix that divided instead would answer 7.5. The cut is still
+        // the ungrouped `sum`, which the producer's plan line names.
+        counter_case(
+            format!("sum(rate({COUNTER_TABLE}[5m])) * 2"),
+            &[(&[], 30.0)],
+            "partial state: sum by () Complete",
+        ),
         counter_case(
             format!("sum by (job) (rate({COUNTER_TABLE}[5m]))"),
             &by_job(9.0, 6.0),
@@ -947,6 +1081,25 @@ async fn native_aggregate_complete_over_four_nonempty_mito_partitions() {
                 false,
             )
         },
+        // A matcher on the tag only the counter table carries. The four flushed series are the four
+        // `region = us-west-2` files and the series left in the memtable range is `us-east-1`, so
+        // this selects four of the five scan ranges: `job = a` keeps its 1, 3 and 5 - nine - while
+        // `job = b` keeps only `h4`'s 2, where the case above answers 6. The pruned range still
+        // runs and may leave its partition with nothing to fold, so `per_partition` is left off and
+        // only the total is pinned: one value for each of the four selected series. The answer's
+        // labels and values say which series survived, and every case here still requires the
+        // producer to read the four-partition scan.
+        AggregateCase {
+            root_values: Some(COUNTER_SERIES.len() - 1),
+            partial_state: Some("partial state: sum by (job) Complete"),
+            ..case(
+                format!(
+                    "sum by (job) (rate({COUNTER_TABLE}{{region=\"{COUNTER_WEST_REGION}\"}}[5m]))"
+                ),
+                &by_job(9.0, 2.0),
+                false,
+            )
+        },
         counter_case(
             format!("avg by (job) (rate({COUNTER_TABLE}[5m]))"),
             &by_job(3.0, 3.0),
@@ -1032,4 +1185,40 @@ async fn native_aggregate_complete_over_four_nonempty_mito_partitions() {
     for case in &cases {
         check_case(&instance, &observer, case).await;
     }
+
+    // The shapes the native compiler declines, each declined for its own reason, and each checked on
+    // the plans that really ran: none may hold a fused producer or a frontend merge, and both routes
+    // must answer what the fixture derives by hand.
+    //
+    // `time()` is the evaluation instant and nothing else, so the single row is at `EVAL_MILLIS` and
+    // carries `EVAL_SECS`. `scalar` reports the one value present at a point only when exactly one
+    // series is present at it, so the five counter series make it NaN; a route that folded one
+    // partition at a time would report the number of whichever partition held a single series, so
+    // the NaN also says the whole vector was read at once. The division folds two *different* tables
+    // - 15 over 30 - and a subquery evaluates its own body on a grid of its own: both read across
+    // series and neither can be answered by a producer per partition, so both keep the existing
+    // plan. The subquery's window spans the two inner points `EVAL_SECS - 1` and `EVAL_SECS`, at each
+    // of which the same extrapolated rate answers 15 - both samples stay further from the window's
+    // ends than the 66s threshold - so the maximum is 15 as well.
+    let fallback_cases = [
+        case("time()".to_string(), &[(&[], EVAL_SECS as f64)], false),
+        case(
+            format!("scalar({COUNTER_TABLE})"),
+            &[(&[], f64::NAN)],
+            false,
+        ),
+        case(
+            format!("sum(rate({COUNTER_TABLE}[5m])) / sum({GAUGE_TABLE})"),
+            &[(&[], 15.0 / 30.0)],
+            false,
+        ),
+        case(
+            format!("max_over_time((sum(rate({COUNTER_TABLE}[5m])))[2s:1s])"),
+            &[(&[], 15.0)],
+            false,
+        ),
+    ];
+    for case in &fallback_cases {
+        check_fallback_case(&instance, &observer, case).await;
+    }
 }

tests-integration/tests/http.rs

@@ -15,7 +15,7 @@
 use std::collections::BTreeMap;
 use std::io::Write;
 use std::str::FromStr;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use api::greptime_proto::io::prometheus::write::v2::histogram::{Count, ZeroCount};
@@ -46,8 +46,15 @@ use common_frontend::slow_query_event::{
 };
 use common_memory_manager::OnExhaustedPolicy;
 use common_options::plugin_options::StandaloneFlag;
+use datafusion::physical_plan::metrics::MetricValue;
+use datafusion::physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use datatypes::arrow::array::{
+    Array as _, Float64Array, StringArray, StringViewArray, TimestampMillisecondArray,
+};
+use datatypes::arrow::datatypes::{DataType, SchemaRef, TimeUnit};
 use flate2::Compression;
 use flate2::write::GzEncoder;
+use frontend::instance::Instance;
 use log_query::{AggFunc, Context, Limit, LogExpr, LogQuery, TimeFilter};
 use loki_proto::logproto::{EntryAdapter, LabelPairAdapter, PushRequest, StreamAdapter};
 use loki_proto::prost_types::Timestamp;
@@ -62,6 +69,9 @@ use pipeline::{
     GREPTIME_INTERNAL_TRACE_PIPELINE_V1_NAME, GREPTIME_INTERNAL_TRACE_PIPELINE_V2_NAME,
 };
 use prost::Message;
+use query::options::QueryOptions;
+use query::parser::PromQuery;
+use query::physical_wrapper::{PhysicalPlanWrapper, PhysicalPlanWrapperRef};
 use rstest_reuse::apply;
 use serde_json::{Value, json};
 use servers::http::handler::HealthResponse;
@@ -79,16 +89,18 @@ use servers::http::result::greptime_result_v1::GreptimedbV1Response;
 use servers::http::result::influxdb_result_v1::{InfluxdbOutput, InfluxdbV1Response};
 use servers::http::test_helpers::{TestClient, TestResponse};
 use servers::http::{GreptimeQueryOutput, HttpOptions, HttpServerBuilder};
+use servers::interceptor::{PromQueryInterceptor, PromQueryInterceptorRef};
 use servers::prom_remote_write::v2::test_util as remote_write_v2;
 use servers::prom_remote_write::validation::PromValidationMode;
 use servers::prom_store::{self, mock_timeseries_new_label};
+use servers::query_handler::sql::SqlQueryHandler;
 use servers::request_memory_limiter::ServerMemoryLimiter;
-use session::context::QueryContextRef;
+use session::context::{QueryContextBuilder, QueryContextRef};
 use standalone::options::StandaloneOptions;
 use table::table_name::TableName;
 use tests_integration::test_util::{
     MockInstanceImpl, StorageType, TestGuard, assert_wal_delta, build_test_prom_server,
-    setup_test_http_app, setup_test_http_app_with_frontend,
+    execute_sql_and_expect, setup_test_http_app, setup_test_http_app_with_frontend,
     setup_test_http_app_with_frontend_and_custom_options,
     setup_test_http_app_with_frontend_and_slow_query_threshold,
     setup_test_http_app_with_frontend_and_user_provider, setup_test_prom_app_with_frontend,
@@ -165,6 +177,8 @@ macro_rules! http_tests {
                 test_http_analyze_stream_tql,
                 test_http_sql_slow_query,
                 test_prometheus_promql_api,
+                test_prometheus_native_promql_route,
+                test_native_promql_route_plans_the_query,
                 test_promql_over_non_millisecond_physical_tables,
                 test_prometheus_label_replace_response,
                 test_prom_http_api,
@@ -1446,6 +1460,922 @@ pub async fn test_prometheus_promql_api(store_type: StorageType) {
     guard.remove_all().await;
 }
 
+/// Keeps the plan of the last PromQL request so a test can prove which route answered it.
+///
+/// The plan is only complete once the response body has been drained: a frontend plan keeps its
+/// region stages behind `MergeScanExec`, whose `children()` is empty, and those stages fill in their
+/// metrics while the body streams out. `take_plan` therefore runs after the body is consumed and
+/// leaves nothing behind, so no request can read the plan of another one.
+#[derive(Default)]
+struct PromRouteObserver {
+    last_plan: Mutex<Option<Arc<dyn ExecutionPlan>>>,
+}
+
+impl PromQueryInterceptor for PromRouteObserver {
+    type Error = servers::error::Error;
+
+    fn post_execute(
+        &self,
+        output: common_query::Output,
+        _query_ctx: QueryContextRef,
+    ) -> servers::error::Result<common_query::Output> {
+        *self.last_plan.lock().unwrap() = output.meta.plan.clone();
+        Ok(output)
+    }
+}
+
+impl PromRouteObserver {
+    /// Takes the plan of the request that just finished.
+    fn take_plan(&self) -> Arc<dyn ExecutionPlan> {
+        self.last_plan
+            .lock()
+            .unwrap()
+            .take()
+            .expect("the request finished without the observer seeing the plan it executed")
+    }
+}
+
+/// Every plan node of every stage of an executed plan, in plan order.
+fn plan_nodes(plan: &Arc<dyn ExecutionPlan>) -> Vec<Value> {
+    fn push(node: &Value, nodes: &mut Vec<Value>) {
+        nodes.push(node.clone());
+        if let Some(children) = node["children"].as_array() {
+            for child in children {
+                push(child, nodes);
+            }
+        }
+    }
+
+    let stages = query::analyze_plan_metrics_to_json_value(plan, true)
+        .expect("an executed plan's metrics serialize to JSON");
+    let mut nodes = Vec::new();
+    for stage in stages
+        .as_array()
+        .expect("plan metrics are a list of stages")
+    {
+        push(&stage["plan"], &mut nodes);
+    }
+    nodes
+}
+
+/// The nodes of `plan` named `name`.
+fn plan_nodes_named(plan: &Arc<dyn ExecutionPlan>, name: &str) -> Vec<Value> {
+    plan_nodes(plan)
+        .into_iter()
+        .filter(|node| node["name"] == name)
+        .collect()
+}
+
+/// Whether the executed plan answered on the native fused-series route.
+///
+/// Every stage is searched: on a vector query the fused node runs inside the region stage behind
+/// `MergeScanExec`, while on a constant it is the frontend plan itself.
+fn executes_fused_series(plan: &Arc<dyn ExecutionPlan>) -> bool {
+    !plan_nodes_named(plan, "FusedSeriesExec").is_empty()
+}
+
+/// One row of a PromQL result: its tag columns in schema order, its timestamp and its value.
+type PromqlRow = (Vec<(String, String)>, i64, f64);
+
+/// Checks that a native result reads no scan for a constant, and one real scan over
+/// `scan_partitions` non-empty partitions for a vector, and returns the fused node it ran.
+fn assert_native_plan(
+    plan: &Arc<dyn ExecutionPlan>,
+    scan_partitions: Option<usize>,
+    context: &str,
+) -> Value {
+    let fused = {
+        let mut nodes = plan_nodes_named(plan, "FusedSeriesExec");
+        assert_eq!(
+            nodes.len(),
+            1,
+            "{context}: expected one fused node: {nodes:?}"
+        );
+        nodes.remove(0)
+    };
+    // The tree the node will run, rendered the way `EXPLAIN` renders it: one line per node.
+    let tree = fused["param"].as_str().unwrap_or_default();
+    let scans = plan_nodes_named(plan, "SeriesScanExec");
+    let Some(partitions) = scan_partitions else {
+        // A constant reads no frame: its one value per grid point comes from the tree alone, so the
+        // node has no child, plans no scan and binds no source.
+        assert!(
+            scans.is_empty(),
+            "{context}: a constant planned a scan: {scans:?}"
+        );
+        assert_eq!(
+            fused["children"].as_array().map(Vec::len),
+            Some(0),
+            "{context}: a constant's fused node has children: {fused}"
+        );
+        assert!(
+            !tree.contains("Input"),
+            "{context}: a constant's tree binds an input: {tree}"
+        );
+        return fused;
+    };
+    assert_eq!(
+        fused["children"].as_array().map(Vec::len),
+        Some(1),
+        "{context}: a vector must read exactly one child: {fused}"
+    );
+    assert_eq!(
+        scans.len(),
+        1,
+        "{context}: a vector must read exactly one scan: {scans:?}"
+    );
+    let mut scan = &fused["children"][0];
+    // An observed optional transparent wrapper that preserves the scan's partition layout; any
+    // other node between the fused node and its scan is rejected.
+    if scan["name"] == "CooperativeExec" {
+        let children = scan["children"]
+            .as_array()
+            .unwrap_or_else(|| panic!("{context}: the cooperative wrapper has no child: {scan}"));
+        assert_eq!(
+            children.len(),
+            1,
+            "{context}: the cooperative wrapper reads {} children: {scan}",
+            children.len()
+        );
+        scan = &children[0];
+    }
+    assert_eq!(
+        scan["name"], "SeriesScanExec",
+        "{context}: the fused node's child is not the scan it reads: {fused}"
+    );
+    let scan_description = scan["param"].as_str().unwrap_or_default();
+    assert!(
+        scan_description.contains(&format!("partitions={partitions}")),
+        "{context}: the scan does not expose {partitions} partitions: {scan_description}"
+    );
+    // Both inputs of one selector name the one source, so a tree that reads two sources or none
+    // would have had to pair or gather them.
+    assert_eq!(
+        tree.matches("source=0").count(),
+        1,
+        "{context}: the tree does not read the one source: {tree}"
+    );
+    assert!(
+        !tree.contains("source=1"),
+        "{context}: the tree reads a second source: {tree}"
+    );
+    // Every partition the scan exposes did its own work: a gather of the raw input would leave the
+    // partitions without the input figure, and a partition that never ran would leave both at zero.
+    // These are input figures: the JSON drops `metric.partition()`, so it cannot say which partition
+    // produced which values.
+    let metrics = fused["metrics"]
+        .as_object()
+        .unwrap_or_else(|| panic!("{context}: the fused node reports no metrics: {fused}"));
+    for partition in 0..partitions {
+        let root_peak = format!("native.p{partition}.root.peak_reserved_bytes");
+        assert!(
+            metrics
+                .get(&root_peak)
+                .and_then(Value::as_u64)
+                .unwrap_or_default()
+                > 0,
+            "partition {partition} of `{context}` did no native work: {fused}"
+        );
+        let prefix = format!("native.p{partition}.root.");
+        assert!(
+            metrics.iter().any(|(name, value)| name.starts_with(&prefix)
+                && name.ends_with(".input.peak_reserved_bytes")
+                && value.as_u64().unwrap_or_default() > 0),
+            "partition {partition} of `{context}` read no input: {fused}"
+        );
+    }
+    fused
+}
+
+/// Checks that the fused node that actually ran answered the whole result, one partition at a time.
+///
+/// The JSON drops `metric.partition()`, so which partition produced which values is only on the
+/// captured node.
+fn assert_native_partition_output(
+    fused: &Arc<dyn ExecutionPlan>,
+    scan_partitions: Option<usize>,
+    expected_rows: usize,
+    context: &str,
+) {
+    // A constant answers on the one partition it has.
+    let partitions = scan_partitions.unwrap_or(1);
+    let output_partitions = fused.output_partitioning().partition_count();
+    assert_eq!(
+        output_partitions, partitions,
+        "{context}: the fused node runs {output_partitions} partitions, not {partitions}"
+    );
+    if scan_partitions.is_some() {
+        let children = fused.children();
+        assert_eq!(
+            children.len(),
+            1,
+            "{context}: the fused node reads {} children, not one",
+            children.len()
+        );
+        let child_partitions = children[0].output_partitioning().partition_count();
+        assert_eq!(
+            child_partitions, partitions,
+            "{context}: the fused node's one child runs {child_partitions} partitions, not \
+             {partitions}"
+        );
+    }
+    let metrics = fused
+        .metrics()
+        .unwrap_or_else(|| panic!("{context}: the executed fused node reports no metrics"));
+    let mut values_by_partition = vec![0_usize; partitions];
+    for metric in metrics.iter() {
+        if metric.value().name() != "native.root.output_values" {
+            continue;
+        }
+        let MetricValue::Count { count, .. } = metric.value() else {
+            continue;
+        };
+        let partition = metric.partition().unwrap_or_else(|| {
+            panic!("{context}: the fused node reports output values for no partition")
+        });
+        let values = values_by_partition.get_mut(partition).unwrap_or_else(|| {
+            panic!("{context}: partition {partition} of the fused node is not one of {partitions}")
+        });
+        *values += count.value();
+    }
+    assert!(
+        values_by_partition.iter().all(|values| *values > 0),
+        "{context}: every native execution partition must report values: {values_by_partition:?}"
+    );
+    assert_eq!(
+        values_by_partition.iter().sum::<usize>(),
+        expected_rows,
+        "{context}: the partitions reported a different number of values than the result holds"
+    );
+}
+
+/// The columns of a result as (name, type, nullable).
+fn result_columns(schema: &SchemaRef) -> Vec<(String, DataType, bool)> {
+    schema
+        .fields()
+        .iter()
+        .map(|field| {
+            (
+                field.name().clone(),
+                field.data_type().clone(),
+                field.is_nullable(),
+            )
+        })
+        .collect()
+}
+
+/// Checks the schema the native route reports: the time index, then one nullable `Float64` per
+/// series field, then one nullable `Utf8` per tag.
+///
+/// A constant reports no tags and no series fields: its one value per grid point is `value` on the
+/// `time` grid.
+///
+/// The tags are stated by the caller rather than read off the result: a vector that matches no
+/// series reports no row to read them from, and still answers with its tag columns.
+fn assert_native_result_schema(schema: &SchemaRef, tags: &[&str], context: &str) {
+    // The time index and the one field the tree reports, then one `Utf8` column per tag.
+    let mut columns = vec![
+        (
+            if tags.is_empty() { "time" } else { "ts" }.to_string(),
+            DataType::Timestamp(TimeUnit::Millisecond, None),
+            false,
+        ),
+        (
+            if tags.is_empty() { "value" } else { "val" }.to_string(),
+            DataType::Float64,
+            true,
+        ),
+    ];
+    for name in tags {
+        columns.push(((*name).to_string(), DataType::Utf8, true));
+    }
+    assert_eq!(
+        result_columns(schema),
+        columns,
+        "{context}: native result schema"
+    );
+}
+
+/// Checks the schema the legacy route reports by role rather than by position or field name, which
+/// it does not share with the native route: one millisecond time column, one `Float64` value column
+/// whose name contains the field it derives from, and one `Utf8` column per tag, named exactly as
+/// that tag.
+///
+/// The value column's name is checked loosely because a computed legacy vector column may be named
+/// after the expression that computes it; only constants are compared field for field.
+///
+/// As on the native route the tags are stated by the caller, so a vector that matches no series is
+/// still expected to answer with its tag columns.
+fn assert_legacy_result_schema(schema: &SchemaRef, tags: &[&str], context: &str) {
+    let time_column = if tags.is_empty() { "time" } else { "ts" };
+    let columns = result_columns(schema);
+    assert_eq!(
+        columns.len(),
+        2 + tags.len(),
+        "{context}: legacy result columns: {columns:?}"
+    );
+    let times = columns
+        .iter()
+        .filter(|column| matches!(column.1, DataType::Timestamp(TimeUnit::Millisecond, None)))
+        .collect::<Vec<_>>();
+    assert_eq!(
+        times.len(),
+        1,
+        "{context}: legacy result reports one millisecond time column: {columns:?}"
+    );
+    assert_eq!(
+        times[0].0, time_column,
+        "{context}: legacy time column name: {columns:?}"
+    );
+    let values = columns
+        .iter()
+        .filter(|column| column.1 == DataType::Float64)
+        .collect::<Vec<_>>();
+    assert_eq!(
+        values.len(),
+        1,
+        "{context}: legacy result reports one Float64 value column: {columns:?}"
+    );
+    assert!(
+        values[0].0.contains("val"),
+        "{context}: legacy value column does not name the field it derives from: {columns:?}"
+    );
+    let tag_names = columns
+        .iter()
+        .filter(|column| column.1 == DataType::Utf8)
+        .map(|column| column.0.as_str())
+        .collect::<Vec<_>>();
+    assert_eq!(
+        tag_names, tags,
+        "{context}: legacy tag columns: {columns:?}"
+    );
+}
+
+/// Compares two complete row lists: the same series at the same timestamps, and values that agree to
+/// a relative `1e-9`. Comparing the keys row by row is what rejects a prefix, a duplicate or a
+/// missing series; a row count alone would accept the first two.
+fn assert_rows_eq(actual: &[PromqlRow], expected: &[PromqlRow], context: &str) {
+    let keys = |rows: &[PromqlRow]| {
+        rows.iter()
+            .map(|(tags, timestamp, _)| (tags.clone(), *timestamp))
+            .collect::<Vec<_>>()
+    };
+    assert_eq!(
+        keys(actual),
+        keys(expected),
+        "{context}: the result does not name the expected series and times"
+    );
+    for (actual, expected) in actual.iter().zip(expected) {
+        let tolerance = 1e-9 * expected.2.abs().max(1.0);
+        assert!(
+            (actual.2 - expected.2).abs() <= tolerance,
+            "{context}: series {:?} at {} must report {}, found {}",
+            expected.0,
+            expected.1,
+            expected.2,
+            actual.2
+        );
+    }
+}
+
+/// The three grid points of every fixture host: its two stored samples, then the last of them carried
+/// forward to the last point. `value` turns a stored sample into the queried value.
+fn fixture_rows(value: impl Fn(f64) -> f64) -> Vec<PromqlRow> {
+    (0..8)
+        .flat_map(|host| {
+            let tags = vec![("host".to_string(), format!("h{host}"))];
+            [
+                (0, value((host + 1) as f64)),
+                (1_000, value((host + 2) as f64)),
+                (2_000, value((host + 2) as f64)),
+            ]
+            .into_iter()
+            .map(move |(timestamp, value)| (tags.clone(), timestamp, value))
+        })
+        .collect()
+}
+
+/// Runs one PromQL query through the public frontend handler and returns the plan it executed, the
+/// schema it answered with and every row it reported.
+async fn promql_output(
+    frontend: &Arc<Instance>,
+    query: &str,
+    native: bool,
+    start: &str,
+    end: &str,
+) -> (Arc<dyn ExecutionPlan>, SchemaRef, Vec<PromqlRow>) {
+    let mut query_ctx = QueryContextBuilder::default();
+    if native {
+        query_ctx = query_ctx.set_extension(
+            "query.experimental_native_promql".to_string(),
+            "true".to_string(),
+        );
+    }
+    let prom_query = PromQuery {
+        query: query.to_string(),
+        start: start.to_string(),
+        end: end.to_string(),
+        step: "1s".to_string(),
+        ..Default::default()
+    };
+    let outputs = SqlQueryHandler::do_promql_query(
+        frontend.as_ref(),
+        &prom_query,
+        Arc::new(query_ctx.build()),
+    )
+    .await;
+    let output = outputs
+        .into_iter()
+        .next()
+        .expect("a PromQL query answers once")
+        .unwrap_or_else(|error| panic!("`{query}` failed: {error}"));
+    let plan = output
+        .meta
+        .plan
+        .clone()
+        .expect("a query reports the plan it executed");
+    let output_data = match output.data {
+        common_query::OutputData::RecordBatches(batches) => batches,
+        common_query::OutputData::Stream(stream) => {
+            common_recordbatch::RecordBatches::try_collect(stream)
+                .await
+                .unwrap_or_else(|error| panic!("`{query}` result collection failed: {error}"))
+        }
+        common_query::OutputData::AffectedRows(rows) => {
+            panic!("`{query}` returned affected rows: {rows}")
+        }
+    };
+    // The collected batches keep the schema the stream reported, so a result that holds no row still
+    // names the columns its route answers with.
+    let schema = output_data.schema().arrow_schema().clone();
+    let mut rows = Vec::new();
+    for batch in output_data.iter() {
+        let batch = batch.df_record_batch();
+        let arrow_schema = batch.schema();
+        let timestamps = batch
+            .columns()
+            .iter()
+            .find_map(|column| column.as_any().downcast_ref::<TimestampMillisecondArray>())
+            .unwrap_or_else(|| panic!("`{query}` reports no millisecond timestamp column"));
+        let values = batch
+            .columns()
+            .iter()
+            .find_map(|column| column.as_any().downcast_ref::<Float64Array>())
+            .unwrap_or_else(|| panic!("`{query}` reports no Float64 value column"));
+        // The tags are the result's string columns, in schema order: a column the schema names but
+        // no row carries would be caught by the row comparison below.
+        let tag_columns = arrow_schema
+            .fields()
+            .iter()
+            .enumerate()
+            .filter(|(_, field)| matches!(field.data_type(), DataType::Utf8 | DataType::Utf8View))
+            .map(|(index, field)| (index, field.name().clone()))
+            .collect::<Vec<_>>();
+        for row in 0..batch.num_rows() {
+            assert!(
+                timestamps.is_valid(row),
+                "`{query}`: null timestamp at row {row}"
+            );
+            assert!(values.is_valid(row), "`{query}`: null value at row {row}");
+            let tags = tag_columns
+                .iter()
+                .map(|(index, name)| {
+                    let column = batch.column(*index);
+                    let value = match column.data_type() {
+                        DataType::Utf8 => {
+                            let array = column.as_any().downcast_ref::<StringArray>().unwrap();
+                            assert!(
+                                array.is_valid(row),
+                                "`{query}`: null tag {name} at row {row}"
+                            );
+                            array.value(row).to_string()
+                        }
+                        DataType::Utf8View => {
+                            let array = column.as_any().downcast_ref::<StringViewArray>().unwrap();
+                            assert!(
+                                array.is_valid(row),
+                                "`{query}`: null tag {name} at row {row}"
+                            );
+                            array.value(row).to_string()
+                        }
+                        other => panic!("`{query}`: tag column {name} is not a string: {other}"),
+                    };
+                    (name.clone(), value)
+                })
+                .collect::<Vec<_>>();
+            rows.push((tags, timestamps.value(row), values.value(row)));
+        }
+    }
+    rows.sort_unstable_by(|left, right| left.0.cmp(&right.0).then(left.1.cmp(&right.1)));
+    (plan, schema, rows)
+}
+
+/// Checks public PromQL results and the executed plan of every request on both routes.
+///
+/// One standalone frontend serves both routes: the engine-wide option is on, so each request's own
+/// hint decides its route, and the test-local observer proves which one answered. Every response is
+/// compared whole - result type, labels, timestamp and value - and never by prefix.
+pub async fn test_prometheus_native_promql_route(store_type: StorageType) {
+    let plugins = Plugins::default();
+    let observer = Arc::new(PromRouteObserver::default());
+    plugins.insert::<PromQueryInterceptorRef<servers::error::Error>>(observer.clone());
+    let mut instance =
+        tests_integration::standalone::GreptimeDbStandaloneBuilder::new("promql_native_route")
+            .with_default_store_type(store_type)
+            .with_query_options(QueryOptions {
+                experimental_native_promql: true,
+                ..Default::default()
+            })
+            .with_plugin(plugins)
+            .build()
+            .await;
+    let server = build_test_prom_server(instance.fe_instance().clone(), false, None)
+        .with_greptime_config_options(instance.opts.datanode_options().to_toml().unwrap())
+        .build();
+    let client = TestClient::new(server.build(server.make_app()).unwrap()).await;
+    let native_hint = ("x-greptime-hints", "query.experimental_native_promql=true");
+
+    for sql in [
+        // One series to compute over, one whose only value is zero, and one table that keeps no row.
+        "CREATE TABLE native_route (ts TIMESTAMP TIME INDEX, val DOUBLE, host STRING, PRIMARY KEY (host))",
+        "CREATE TABLE native_zero (ts TIMESTAMP TIME INDEX, val DOUBLE, host STRING, PRIMARY KEY (host))",
+        "CREATE TABLE native_empty (ts TIMESTAMP TIME INDEX, val DOUBLE, host STRING, PRIMARY KEY (host))",
+        "INSERT INTO native_route(host, val, ts) VALUES ('h1', 1.0, 0), ('h1', 2.0, 1000)",
+        "INSERT INTO native_zero(host, val, ts) VALUES ('z', 0.0, 0), ('z', 0.0, 1000)",
+    ] {
+        let res = client
+            .get(&format!("/v1/sql?db=public&sql={}", encode(sql)))
+            .send()
+            .await;
+        assert_eq!(res.status(), StatusCode::OK, "{sql}");
+    }
+
+    // One expression per shape a statement root can take: a folded constant, a negated constant, a
+    // broadcast on either side of the input, a negation compiled as one, and two divisions by zero.
+    // `100 / 0` is `+Inf` and `0 / 0` is `NaN`: both are reported as the values they are, not as a
+    // missing sample, so each series keeps its row.
+    let instants = [
+        ("1 + 2", json!({"resultType":"scalar","result":[1.0,"3.0"]})),
+        ("-1", json!({"resultType":"scalar","result":[1.0,"-1.0"]})),
+        (
+            "native_route * 100",
+            json!({"resultType":"vector","result":[{"metric":{"host":"h1"},"value":[1.0,"200.0"]}]}),
+        ),
+        (
+            "100 / native_route",
+            json!({"resultType":"vector","result":[{"metric":{"host":"h1"},"value":[1.0,"50.0"]}]}),
+        ),
+        (
+            "-native_route",
+            json!({"resultType":"vector","result":[{"metric":{"host":"h1"},"value":[1.0,"-2.0"]}]}),
+        ),
+        (
+            "100 / native_zero",
+            json!({"resultType":"vector","result":[{"metric":{"host":"z"},"value":[1.0,"inf"]}]}),
+        ),
+        (
+            "native_zero / 0",
+            json!({"resultType":"vector","result":[{"metric":{"host":"z"},"value":[1.0,"NaN"]}]}),
+        ),
+        // A selector that matches nothing broadcasts over no series, and a table that holds no row
+        // reports no series either; a constant answers on the grid it was asked about in both cases.
+        (
+            "native_route{host=\"absent\"} * 100",
+            json!({"resultType":"vector","result":[]}),
+        ),
+        ("native_empty", json!({"resultType":"vector","result":[]})),
+    ];
+    for native in [false, true] {
+        for (query, expected) in &instants {
+            let mut request = client.get(&format!(
+                "/v1/prometheus/api/v1/query?query={}&time=1",
+                encode(query)
+            ));
+            if native {
+                request = request.header(native_hint.0, native_hint.1);
+            }
+            let res = request.send().await;
+            assert_eq!(res.status(), StatusCode::OK, "`{query}`, native={native}");
+            let body = serde_json::from_str::<PrometheusJsonResponse>(&res.text().await).unwrap();
+            assert_eq!(body.status, "success", "`{query}`, native={native}");
+            let actual = serde_json::to_value(&body.data).unwrap();
+            assert_eq!(&actual, expected, "`{query}`, native={native}");
+            assert_eq!(
+                executes_fused_series(&observer.take_plan()),
+                native,
+                "`{query}` asked native={native} but executed the other route"
+            );
+        }
+    }
+
+    // The same expression over a range: a matrix carries every grid point of every series, with the
+    // last sample carried forward, and both routes answer it with the same one.
+    let query = encode("native_route * 100");
+    let expected = json!({"resultType":"matrix","result":[{"metric":{"host":"h1"},"values":[[0.0,"100.0"],[1.0,"200.0"],[2.0,"200.0"]]}]});
+    for native in [false, true] {
+        let mut request = client.get(&format!(
+            "/v1/prometheus/api/v1/query_range?query={query}&start=0&end=2&step=1s"
+        ));
+        if native {
+            request = request.header(native_hint.0, native_hint.1);
+        }
+        let res = request.send().await;
+        assert_eq!(res.status(), StatusCode::OK, "range query, native={native}");
+        let body = serde_json::from_str::<PrometheusJsonResponse>(&res.text().await).unwrap();
+        assert_eq!(body.status, "success", "range query, native={native}");
+        let actual = serde_json::to_value(&body.data).unwrap();
+        assert_eq!(&actual, &expected, "range query, native={native}");
+        assert_eq!(
+            executes_fused_series(&observer.take_plan()),
+            native,
+            "the hinted range query executed the other route"
+        );
+    }
+
+    instance.guard.remove_all().await;
+}
+
+/// Keeps the native fused node the request executed, for the figures only the node records.
+///
+/// Uses the existing [`PhysicalPlanWrapper`] hook; `clear` before a query and `take` after its body
+/// is drained keep each query reading only the node it ran.
+#[derive(Default)]
+struct NativeFusedObserver(Mutex<Vec<Arc<dyn ExecutionPlan>>>);
+
+impl NativeFusedObserver {
+    /// Drops what an earlier request captured.
+    fn clear(&self) {
+        self.0.lock().unwrap().clear();
+    }
+
+    /// The one fused node the request just executed.
+    fn take(&self, context: &str) -> Arc<dyn ExecutionPlan> {
+        let mut captured = std::mem::take(&mut *self.0.lock().unwrap());
+        assert_eq!(
+            captured.len(),
+            1,
+            "`{context}`: the request executed {} fused nodes: {captured:?}",
+            captured.len()
+        );
+        captured.remove(0)
+    }
+}
+
+impl PhysicalPlanWrapper for NativeFusedObserver {
+    fn wrap(
+        &self,
+        origin: Arc<dyn ExecutionPlan>,
+        _ctx: QueryContextRef,
+    ) -> Arc<dyn ExecutionPlan> {
+        let mut stack = vec![origin.clone()];
+        while let Some(node) = stack.pop() {
+            if node.name() == "FusedSeriesExec" {
+                self.0.lock().unwrap().push(node);
+            } else {
+                stack.extend(node.children().into_iter().cloned());
+            }
+        }
+        origin
+    }
+}
+
+/// Verifies the native route's plan and complete result against the legacy one.
+///
+/// The fixture is four pk-disjoint insert batches, each persisted by its own flush, so the scan has
+/// four series batches to distribute over the four partitions the query asks for.
+pub async fn test_native_promql_route_plans_the_query(store_type: StorageType) {
+    let observer = Arc::new(NativeFusedObserver::default());
+    let plugins = Plugins::default();
+    let wrapper: PhysicalPlanWrapperRef = observer.clone();
+    plugins.insert(wrapper);
+    let mut instance =
+        tests_integration::standalone::GreptimeDbStandaloneBuilder::new("promql_native_route_plan")
+            .with_default_store_type(store_type)
+            .with_query_options(QueryOptions {
+                experimental_native_promql: true,
+                parallelism: 4,
+                ..Default::default()
+            })
+            .with_plugin(plugins)
+            .build()
+            .await;
+    let frontend = instance.fe_instance();
+
+    // The relaxed TWCS thresholds keep the four flushed files from being compacted back into one
+    // unit of work, so four file ranges reach the scan.
+    for sql in [
+        "CREATE TABLE native_demo (ts TIMESTAMP TIME INDEX, val DOUBLE, host STRING, PRIMARY KEY (host)) WITH ('compaction.type' = 'twcs', 'compaction.twcs.active_window.trigger_file_num' = '100', 'compaction.twcs.inactive_window.trigger_file_num' = '100')",
+        "CREATE TABLE native_rate (ts TIMESTAMP TIME INDEX, val DOUBLE, host STRING, PRIMARY KEY (host)) WITH ('compaction.type' = 'twcs', 'compaction.twcs.active_window.trigger_file_num' = '100', 'compaction.twcs.inactive_window.trigger_file_num' = '100')",
+    ] {
+        execute_sql_and_expect(frontend, sql, "Affected Rows: 0").await;
+    }
+    for (first, second) in [(0, 1), (2, 3), (4, 5), (6, 7)] {
+        let values = [first, second]
+            .into_iter()
+            .map(|host| {
+                format!(
+                    "('h{host}', {}.0, 0), ('h{host}', {}.0, 1000)",
+                    host + 1,
+                    host + 2
+                )
+            })
+            .collect::<Vec<_>>()
+            .join(", ");
+        execute_sql_and_expect(
+            frontend,
+            &format!("INSERT INTO native_demo(host, val, ts) VALUES {values}"),
+            "Affected Rows: 4",
+        )
+        .await;
+        // One 0 -> 3 increment between the two samples.
+        let values = [first, second]
+            .into_iter()
+            .map(|host| format!("('h{host}', 0.0, 0), ('h{host}', 3.0, 1000)"))
+            .collect::<Vec<_>>()
+            .join(", ");
+        execute_sql_and_expect(
+            frontend,
+            &format!("INSERT INTO native_rate(host, val, ts) VALUES {values}"),
+            "Affected Rows: 4",
+        )
+        .await;
+        // Flushing before the next pair keeps every round a separate persisted batch.
+        tests_integration::test_util::execute_sql(frontend, "ADMIN FLUSH_TABLE('native_demo')")
+            .await;
+        tests_integration::test_util::execute_sql(frontend, "ADMIN FLUSH_TABLE('native_rate')")
+            .await;
+    }
+
+    for (query, start, end, expected, scan_partitions) in [
+        (
+            "1 + 2",
+            "0",
+            "2",
+            vec![
+                (Vec::new(), 0_i64, 3.0),
+                (Vec::new(), 1_000, 3.0),
+                (Vec::new(), 2_000, 3.0),
+            ],
+            None,
+        ),
+        (
+            "-1",
+            "0",
+            "2",
+            vec![
+                (Vec::new(), 0_i64, -1.0),
+                (Vec::new(), 1_000, -1.0),
+                (Vec::new(), 2_000, -1.0),
+            ],
+            None,
+        ),
+        (
+            "native_demo * 100",
+            "0",
+            "2",
+            fixture_rows(|value| value * 100.0),
+            Some(4),
+        ),
+        (
+            "100 / native_demo",
+            "0",
+            "2",
+            fixture_rows(|value| 100.0 / value),
+            Some(4),
+        ),
+        (
+            "-native_demo",
+            "0",
+            "2",
+            fixture_rows(|value| -value),
+            Some(4),
+        ),
+        // The counter rises 0 -> 3 between ts 0 and 1s and is evaluated at 1s: the inferred zero
+        // point and the evaluation instant both sit on a sample, so neither edge is extrapolated and
+        // (3 / 300s) * 100 = 1.
+        (
+            "rate(native_rate[5m]) * 100",
+            "1",
+            "1",
+            (0..8)
+                .map(|host| (vec![("host".to_string(), format!("h{host}"))], 1_000, 1.0))
+                .collect::<Vec<PromqlRow>>(),
+            Some(4),
+        ),
+    ] {
+        // Only the node this query ran is read: the previous request's capture is dropped first.
+        observer.clear();
+        // Every fixture series carries one tag; a constant carries none.
+        let tags: &[&str] = if scan_partitions.is_some() {
+            &["host"]
+        } else {
+            &[]
+        };
+        let (plan, schema, rows) = promql_output(frontend, query, true, start, end).await;
+        assert_rows_eq(&rows, &expected, query);
+        assert_native_result_schema(&schema, tags, query);
+        let fused = assert_native_plan(&plan, scan_partitions, query);
+        assert_eq!(
+            fused["metrics"]["native.root.output_values"].as_u64(),
+            Some(expected.len() as u64),
+            "`{query}`: the tree reported a different number of values than the result holds"
+        );
+        // The body is drained, so every partition has finished.
+        assert_native_partition_output(
+            &observer.take(query),
+            scan_partitions,
+            expected.len(),
+            query,
+        );
+
+        // Without the per-query hint the same statement runs the legacy pipeline, which must answer
+        // with the same rows.
+        let (legacy_plan, legacy_schema, legacy_rows) =
+            promql_output(frontend, query, false, start, end).await;
+        assert!(
+            !executes_fused_series(&legacy_plan),
+            "`{query}` without the hint executed the fused route"
+        );
+        assert_rows_eq(&legacy_rows, &expected, query);
+        assert_legacy_result_schema(&legacy_schema, tags, query);
+        // Both constant cases preserve the legacy time/value schema; computed vector columns are
+        // compared by role.
+        if scan_partitions.is_none() {
+            assert_eq!(
+                result_columns(&legacy_schema),
+                result_columns(&schema),
+                "`{query}`: the two routes must report the same constant schema"
+            );
+        }
+    }
+
+    // A vector that matches no series is a typed empty result, not a constant: both routes still
+    // report the time, value and tag columns, and the native route still plans its one scan. What
+    // such a plan reports per partition is deliberately not read: an empty input may be pruned to a
+    // different partition layout than the four non-empty queries above.
+    let query = "native_demo{host=\"absent\"} * 100";
+    observer.clear();
+    let (plan, schema, rows) = promql_output(frontend, query, true, "0", "2").await;
+    assert!(rows.is_empty(), "`{query}` reported {} rows", rows.len());
+    assert_native_result_schema(&schema, &["host"], query);
+    assert!(
+        executes_fused_series(&plan),
+        "`{query}` did not execute the fused route"
+    );
+    let fused_nodes = plan_nodes_named(&plan, "FusedSeriesExec");
+    assert_eq!(
+        fused_nodes.len(),
+        1,
+        "`{query}`: expected one fused node: {fused_nodes:?}"
+    );
+    let fused = &fused_nodes[0];
+    let tree = fused["param"].as_str().unwrap_or_default();
+    assert_eq!(
+        fused["children"].as_array().map(Vec::len),
+        Some(1),
+        "`{query}`: an empty vector must read exactly one child: {fused}"
+    );
+    let scans = plan_nodes_named(&plan, "SeriesScanExec");
+    assert_eq!(
+        scans.len(),
+        1,
+        "`{query}`: an empty vector must read exactly one scan: {scans:?}"
+    );
+    // The tree binds the one source instead of answering as a constant, whose tree binds no input.
+    assert!(
+        tree.contains("Input"),
+        "`{query}`: the empty vector planned no input: {tree}"
+    );
+    assert_eq!(
+        tree.matches("source=0").count(),
+        1,
+        "`{query}`: the tree does not read the one source: {tree}"
+    );
+    assert!(
+        !tree.contains("source=1"),
+        "`{query}`: the tree reads a second source: {tree}"
+    );
+    // The body is drained, so the one fused node the request ran has finished.
+    observer.take(query);
+
+    // Without the per-query hint the same statement runs the legacy pipeline, which must answer
+    // with the same empty typed result.
+    let (legacy_plan, legacy_schema, legacy_rows) =
+        promql_output(frontend, query, false, "0", "2").await;
+    assert!(
+        !executes_fused_series(&legacy_plan),
+        "`{query}` without the hint executed the fused route"
+    );
+    assert!(
+        legacy_rows.is_empty(),
+        "`{query}` without the hint reported {} rows",
+        legacy_rows.len()
+    );
+    assert_legacy_result_schema(&legacy_schema, &["host"], query);
+
+    instance.guard.remove_all().await;
+}
+
 pub async fn test_prometheus_label_replace_response(store_type: StorageType) {
     let (app, mut guard) =
         setup_test_prom_app_with_frontend(store_type, "prometheus_label_replace_response").await;

tests/cases/standalone/common/promql/scalar_entry.result

@@ -0,0 +1,61 @@
+-- Legacy SQLness regression: these TQL EVAL cases pin the legacy PromQL planner.
+-- The native fused-series route is verified by test_prometheus_native_promql_route and
+-- test_native_promql_route_plans_the_query in tests-integration/tests/http.rs.
+CREATE TABLE native_scalar_entry (
+    ts TIMESTAMP(3) TIME INDEX,
+    host STRING PRIMARY KEY,
+    val DOUBLE,
+);
+
+Affected Rows: 0
+
+INSERT INTO native_scalar_entry VALUES
+    (0, 'host1', 1.0),
+    (0, 'host2', 2.0),
+    (1000, 'host1', 3.0),
+    (1000, 'host2', 4.0);
+
+Affected Rows: 4
+
+TQL EVAL (0, 2, '1s') 1 + 2;
+
++---------------------+-------+
+| time                | value |
++---------------------+-------+
+| 1970-01-01T00:00:00 | 3.0   |
+| 1970-01-01T00:00:01 | 3.0   |
+| 1970-01-01T00:00:02 | 3.0   |
++---------------------+-------+
+
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (0, 2, '1s') native_scalar_entry * 100;
+
++-------+---------------------+--------------------+
+| host  | ts                  | val * Float64(100) |
++-------+---------------------+--------------------+
+| host1 | 1970-01-01T00:00:00 | 100.0              |
+| host1 | 1970-01-01T00:00:01 | 300.0              |
+| host1 | 1970-01-01T00:00:02 | 300.0              |
+| host2 | 1970-01-01T00:00:00 | 200.0              |
+| host2 | 1970-01-01T00:00:01 | 400.0              |
+| host2 | 1970-01-01T00:00:02 | 400.0              |
++-------+---------------------+--------------------+
+
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (0, 2, '1s') -native_scalar_entry;
+
++-------+---------------------+---------+
+| host  | ts                  | (- val) |
++-------+---------------------+---------+
+| host1 | 1970-01-01T00:00:00 | -1.0    |
+| host1 | 1970-01-01T00:00:01 | -3.0    |
+| host1 | 1970-01-01T00:00:02 | -3.0    |
+| host2 | 1970-01-01T00:00:00 | -2.0    |
+| host2 | 1970-01-01T00:00:01 | -4.0    |
+| host2 | 1970-01-01T00:00:02 | -4.0    |
++-------+---------------------+---------+
+
+DROP TABLE native_scalar_entry;
+
+Affected Rows: 0
+

tests/cases/standalone/common/promql/scalar_entry.sql

@@ -0,0 +1,24 @@
+-- Legacy SQLness regression: these TQL EVAL cases pin the legacy PromQL planner.
+-- The native fused-series route is verified by test_prometheus_native_promql_route and
+-- test_native_promql_route_plans_the_query in tests-integration/tests/http.rs.
+CREATE TABLE native_scalar_entry (
+    ts TIMESTAMP(3) TIME INDEX,
+    host STRING PRIMARY KEY,
+    val DOUBLE,
+);
+
+INSERT INTO native_scalar_entry VALUES
+    (0, 'host1', 1.0),
+    (0, 'host2', 2.0),
+    (1000, 'host1', 3.0),
+    (1000, 'host2', 4.0);
+
+TQL EVAL (0, 2, '1s') 1 + 2;
+
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (0, 2, '1s') native_scalar_entry * 100;
+
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (0, 2, '1s') -native_scalar_entry;
+
+DROP TABLE native_scalar_entry;