feat(promql): execute zero-input scalar roots through fused children

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

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,238 @@ 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.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().with_new_children(Vec::new()).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 = vec![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,
     }
 }

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() {