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..03aae68b069 100644 --- a/src/promql/src/extension_plan/fused_series/tests.rs +++ b/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::() + .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..69accb7b8b7 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, } } @@ -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}"); } 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() { diff --git a/tests-integration/src/tests/native_aggregate_test.rs b/tests-integration/src/tests/native_aggregate_test.rs index 44a261bae67..ab8617dde98 100644 --- a/tests-integration/src/tests/native_aggregate_test.rs +++ b/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, /// 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, /// 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) { 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) { } } - 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::>(); + 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, + 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, observer: &NativePlanObserver, case: &AggregateCase) { let promql = case.promql.as_str(); @@ -801,34 +917,36 @@ async fn check_case(instance: &Arc, 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::>(); - 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::(), - "`{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; + } } diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 493fb9530d3..972ad254257 100644 --- a/tests-integration/tests/http.rs +++ b/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>>, +} + +impl PromQueryInterceptor for PromRouteObserver { + type Error = servers::error::Error; + + fn post_execute( + &self, + output: common_query::Output, + _query_ctx: QueryContextRef, + ) -> servers::error::Result { + *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 { + 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) -> Vec { + fn push(node: &Value, nodes: &mut Vec) { + 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, name: &str) -> Vec { + 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) -> 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, + scan_partitions: Option, + 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, + scan_partitions: Option, + 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::(), + 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::>(); + 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::>(); + 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::>(); + 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::>() + }; + 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 { + (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, + query: &str, + native: bool, + start: &str, + end: &str, +) -> (Arc, SchemaRef, Vec) { + 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::()) + .unwrap_or_else(|| panic!("`{query}` reports no millisecond timestamp column")); + let values = batch + .columns() + .iter() + .find_map(|column| column.as_any().downcast_ref::()) + .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::>(); + 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::().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::().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::>(); + 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::>(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::(&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::(&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>>); + +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 { + 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, + _ctx: QueryContextRef, + ) -> Arc { + 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::>() + .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::>() + .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::>(), + 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; diff --git a/tests/cases/standalone/common/promql/scalar_entry.result b/tests/cases/standalone/common/promql/scalar_entry.result new file mode 100644 index 00000000000..8bf815ba92b --- /dev/null +++ b/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 + diff --git a/tests/cases/standalone/common/promql/scalar_entry.sql b/tests/cases/standalone/common/promql/scalar_entry.sql new file mode 100644 index 00000000000..db29b4d3499 --- /dev/null +++ b/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;