From 5bcd96cead612246b88f6cda9eca5347f20db77a Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Sat, 24 Oct 2026 21:06:02 +0800 Subject: [PATCH] feat(promql): execute zero-input scalar roots through fused children Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/promql/AGENTS.md | 4 + src/promql/src/extension_plan/fused_series.rs | 100 ++++++-- .../src/extension_plan/fused_series/tests.rs | 230 +++++++++++++++++- src/promql/src/extension_plan/planner.rs | 5 +- src/promql/src/native/execute.rs | 95 +++++++- src/promql/src/native/execute/tests.rs | 143 +++++++++++ src/promql/src/native/plan.rs | 2 +- src/promql/src/native/plan/tests.rs | 39 +++ src/query/src/promql/native_compile.rs | 25 +- src/query/src/promql/native_compile/tests.rs | 38 ++- src/query/src/promql/planner/native.rs | 53 +++- src/query/src/promql/planner/test.rs | 33 +++ 12 files changed, 715 insertions(+), 52 deletions(-) diff --git a/src/promql/AGENTS.md b/src/promql/AGENTS.md index 53ad3c61169..8ec551c7ac3 100644 --- a/src/promql/AGENTS.md +++ b/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 diff --git a/src/promql/src/extension_plan/fused_series.rs b/src/promql/src/extension_plan/fused_series.rs index 3bdd896c3fa..916532ff2f0 100644 --- a/src/promql/src/extension_plan/fused_series.rs +++ b/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 DataFusionResult>> { 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> { + 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, ) -> DataFusionResult { 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, ) -> DataFusionResult { - 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), diff --git a/src/promql/src/extension_plan/fused_series/tests.rs b/src/promql/src/extension_plan/fused_series/tests.rs index 8fa3f0246df..d059982a62a 100644 --- a/src/promql/src/extension_plan/fused_series/tests.rs +++ b/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::() + .unwrap(); + assert_eq!(grid_timestamps(), timestamps.values().as_ref()); + let values = batches[0] + .column_by_name(VALUE) + .unwrap() + .as_any() + .downcast_ref::() + .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::>() + .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::>() + .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::() + .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 = || { diff --git a/src/promql/src/extension_plan/planner.rs b/src/promql/src/extension_plan/planner.rs index 63acdeb9fd8..7fd764fc08b 100644 --- a/src/promql/src/extension_plan/planner.rs +++ b/src/promql/src/extension_plan/planner.rs @@ -109,11 +109,14 @@ fn plan_fused_series( physical_inputs: &[Arc], target_partitions: usize, ) -> DfResult>> { - 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::().is_none()) diff --git a/src/promql/src/native/execute.rs b/src/promql/src/native/execute.rs index 46cf228c825..077e1e231df 100644 --- a/src/promql/src/native/execute.rs +++ b/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 { Ok(position as usize) } +/// One unlabelled constant value per statement grid point, closed on its only chunk. +fn scalar_chunk(grid: Arc, value: f64) -> Result { + 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, + /// A constant root waiting for its only pull; operand scalars remain chunkless. + scalar_root: Option<(Arc, f64, NodeState)>, } impl NodeState { @@ -2834,25 +2854,46 @@ impl NativeExecutionState { pub fn try_new(tree: &NativeTree, pool: Arc) -> Result { 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> { + 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::() + .checked_add(std::mem::size_of::()) + .and_then(|bytes| { + len.checked_mul(std::mem::size_of::()) + .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(); } diff --git a/src/promql/src/native/execute/tests.rs b/src/promql/src/native/execute/tests.rs index 3fbd164888f..a7b88a7541a 100644 --- a/src/promql/src/native/execute/tests.rs +++ b/src/promql/src/native/execute/tests.rs @@ -703,6 +703,149 @@ fn run(executor: &mut NativeExecutor) -> Vec { 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::()); + 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> { field_values(chunk, 0) } diff --git a/src/promql/src/native/plan.rs b/src/promql/src/native/plan.rs index 5f87c5725c1..934e8b215c8 100644 --- a/src/promql/src/native/plan.rs +++ b/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 diff --git a/src/promql/src/native/plan/tests.rs b/src/promql/src/native/plan/tests.rs index e22a397208a..67a65f280f5 100644 --- a/src/promql/src/native/plan/tests.rs +++ b/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}" + ); +} diff --git a/src/query/src/promql/native_compile.rs b/src/query/src/promql/native_compile.rs index 243658a0a64..36607b66821 100644 --- a/src/query/src/promql/native_compile.rs +++ b/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 { + 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 { +/// 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 { match expr { PromExpr::NumberLiteral(NumberLiteral { val }) => Some(*val), PromExpr::Paren(paren) => constant_value(&paren.expr), diff --git a/src/query/src/promql/native_compile/tests.rs b/src/query/src/promql/native_compile/tests.rs index 554439b9008..2c49fc394e4 100644 --- a/src/query/src/promql/native_compile/tests.rs +++ b/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 diff --git a/src/query/src/promql/planner/native.rs b/src/query/src/promql/planner/native.rs index e59a32ec50e..9faf3a618cc 100644 --- a/src/query/src/promql/planner/native.rs +++ b/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 = 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, } } diff --git a/src/query/src/promql/planner/test.rs b/src/query/src/promql/planner/test.rs index bce952c7659..a5564e440e0 100644 --- a/src/query/src/promql/planner/test.rs +++ b/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() {