From 72bee4ae9cd4d8eebae2e8cb951046b5ab804e73 Mon Sep 17 00:00:00 2001 From: discord9 Date: Thu, 29 Oct 2026 20:16:44 +0800 Subject: [PATCH] feat(promql): answer whole-vector and multi-metric statements natively (#10025) * feat(promql): name a fused tree's tags from the existing plan A fused node declares exactly the tags the planner gives it, and reads each one back off the series, so a label set this route cannot derive is a label set the answer would lose. The walk over the tree derives labels for a match, a set operation and the point-wise wrappers over them, and names none for an aggregate's groups or a whole-vector operation - which today declines every such statement. Take those columns from the existing planner instead, the way the aggregate split already does: its tag columns are what it projects for the same statement, and they are now shared through one helper rather than written out twice. Only the tag columns are copied, because the node names its field columns after its own tree while the existing planner can name a computed column after its expression. A label rewrite is a shape the walk cannot derive at all: label_replace adds its destination label and an empty replacement removes it, so passing the input's labels through is an inference the rewrite contradicts. Report no label set there. The rewrite itself keeps the existing plan, and this does not change that: the native runtime refuses two series a rewrite gives one label set, where that plan answers them, so admitting the shape would turn an answer into an error. The test that holds this now covers the two-occurrence shape as well. An aggregate or whole-vector root does not reach this branch today: both read across series, so they are planned by the aggregate split or fall back. It is the prerequisite for the placement that runs such a tree in one execution, and reaches the branch when that lands. Signed-off-by: discord9 * feat(promql): run a classic histogram function above the aggregate cut The split accepts a suffix that runs once over the fully merged series, which is why TopK/BottomK were already admitted: the merge has folded every partition's contribution to each group, so the series the suffix reads are the ones the statement's input would have produced in one execution. A classic histogram function is that kind of operation. It reads each histogram off the bucket series that share its other labels and drops le, and the merge has already folded each (..., le) group into one complete series on the evaluated grid, so the buckets it reads are exactly the ones the input would have had. Reading le off those merged series is the same computation, not an approximation. LimitK still declines: it depends on the storage input order, which the merge does not preserve. This is what the o11ybench query `histogram_quantile(0.99, sum(rate(...[5m])) by (instance, le))` needs: it was planned by the existing path only because the split refused a whole-vector operation above the cut. The acceptance test runs the shape over real data and compares the rows two executions report: the fixture spreads six bucket series over several files and a memtable range, so the partial sums genuinely have partitions to merge, and the native answer must equal the complete input's for quantiles at both ends of the interpolation and for histogram_fraction. It also asserts the native run really merged state, so a silent fall back cannot pass it. Signed-off-by: discord9 * feat(promql): read two metric logical tables as two inputs A two-selector statement whose two sides are aggregates kept the existing plan, even where both sides read a metric-engine logical table. The o11ybench query `sum(max by (...) (a)) by (namespace) / sum(b) by (namespace)` is that shape, and it fell back for three separate reasons. First, the engine gate refused every multi-selector metric-engine statement. The clause after it already requires each such selector to be a recognized logical rewrite, so the removed clause refused only the case this change handles. Second, a metric logical table is rewritten to a scan of the physical table restricted by `__table_id = `, so two logical tables over one physical table prepare two scans of the same table. The check deciding whether the node binds one metric or one input per occurrence compared table names only, so the two metrics read as one and both inputs would carry the first one's name - a wrong answer, not a fall back. The id the rewrite already requires is now part of that identity. Third, a tree over two aggregates reads across series, so it is not answered per partition, and the branch for such trees was the aggregate split, which takes one child. A node with more than one input needs no split: it declares one partition and reads all of every input's partitions in one execution, so every group is folded in one place. Such a tree now falls through to the node. The one-selector path is unchanged: it still requires `can_split` and still declines a tree the split cannot cut. So is the multi-selector case whose inputs stay direct scans of one table - such a node is not region-local, as `Categorizer::check_extension_plan` refuses a cross-series plan over more than one scan - which is why the condition is the metric-logical-table one. The acceptance test runs both shapes over two logical tables of one physical table, with values that make a swapped input visibly wrong, and compares the rows the two routes report. Signed-off-by: discord9 * test(promql): log the metric pair's rows through telemetry, not stdout The acceptance test printed its evidence with println!, which the crate lints as an error under -D warnings, and echoed a Display value through to_string inside the format arguments. Every other test in the file logs through common_telemetry::info!, so this one does too. Signed-off-by: discord9 * fix(promql): keep the existing plan for a tree the runtime cannot execute The multi-input route skipped the aggregate split, and the split was what refused a tree the runtime cannot execute: `check_cut` declines a `quantile` and a grouping by a value field. Skipping it turned queries the existing plan answers into errors, because the node is chosen while planning the logical plan and a refusal the physical layer raises later has nowhere to fall back to. `quantile(0.5, a) / sum(b)` over two metric logical tables reported `NotImplemented: the native runtime does not execute the Quantile { .. } aggregate yet` where the existing plan answers it, and `count(count by (val) (a)) / sum(b)` failed on the grouping by a value field, which the outer `count` had hidden from the column check. Ask the runtime before taking the route, with the same construction the physical layer validates by, and keep the existing plan for whatever it refuses. Asking it rather than listing the unsupported kinds is what makes the check cover refusals this planner does not know about. Signed-off-by: discord9 * fix(promql): keep the existing plan for a grouping and a selection the node cannot answer A multi-input node skips the aggregate split, so the refusals the split makes for reasons that are not about what the runtime computes were skipped with it. Two of them matter here. A grouping by the metric name keeps a label no column of the existing plan can name. The native aggregate folds its groups by `__name__` as by any other label, but the label set the node declares comes from the existing planner's columns, and `__name__` is the name a raw frame carries rather than a column of any scan - so the existing planner's `by(__name__)` finds nothing and reports no label. The schema would then report one label set for two groups the tree keeps apart: `sum by (__name__) (a or b)` over two metrics would answer two rows a reader cannot tell apart, which is a series identity lost rather than an answer given. A selection that depends on the order its input arrives in is the other. `limitk` keeps the first `k` series of each group in input order, while a node with more than one input reads its inputs' partitions through a gather that hands them over in the order they became ready: the schedule would decide which series a query keeps. Prometheus defines the selection over a series set. The split declines `LimitK` for the same reason. Every other vector operation orders its input itself - `topk` by label, `sort` by value, `limit_ratio` by a label hash - which is why only this one is asked about. Both shapes keep the existing plan. `limitk` is not answerable by either route today, since the existing planner has no kernel for the operator; the point is that the native route is not the first to answer a selection the schedule would decide. Each guard is checked with the guard removed, so both tests are known to fail without it. Signed-off-by: discord9 * fix(promql): decline unsupported count_values groupings before execution Signed-off-by: discord9 * docs(promql): scope the native capability probe guarantee Signed-off-by: discord9 * test(promql): pin the same-table pair the multi-selector exclusion used to keep out State that a Mito input joins a multi-input tree only when a second metric logical id makes the tree multi-input, and that only LimitK's selection depends on arrival order. Signed-off-by: discord9 * fix(promql): decline ranking grouped by a value field before native execution topk, bottomk and limitk key their groups by series labels, while the existing plan also accepts a value field there and forms one group per sample value. Admitted multi-input trees could therefore keep one row where the existing plan keeps one per group. Validate the grouping when the vector runtime is built so the statement keeps the existing plan. Signed-off-by: discord9 --------- Signed-off-by: discord9 --- src/promql/src/aggregate.rs | 46 +- .../src/extension_plan/aggregate_finalize.rs | 27 +- .../aggregate_finalize/tests.rs | 69 ++ src/promql/src/native/execute.rs | 35 +- src/promql/src/native/execute/tests.rs | 95 +++ src/query/src/promql/planner/native.rs | 347 ++++++++-- src/query/src/promql/planner/test.rs | 134 +++- .../src/tests/native_aggregate_test.rs | 157 ++++- tests-integration/src/tests/promql_test.rs | 600 +++++++++++++++++- 9 files changed, 1418 insertions(+), 92 deletions(-) diff --git a/src/promql/src/aggregate.rs b/src/promql/src/aggregate.rs index 6ea4d0bcb3a..7d99ff1567a 100644 --- a/src/promql/src/aggregate.rs +++ b/src/promql/src/aggregate.rs @@ -582,16 +582,12 @@ pub fn aggregate_quantile( Ok(output) } -/// `count_values(label, v)`: at each grid point, the number of samples in each group of series -/// that share a value, with the value written to `label` in Prometheus' `FormatFloat(v, 'f', -1)` -/// form, or for a native histogram as its `FloatHistogram.String`. With `by`, `label` is part of -/// the grouping; with `without`, it is dropped if listed. -pub fn count_values( - schema: &Arc, - input: &[Series], +/// Validates the schema-only requirements shared by the kernel and native runtime construction. +pub(crate) fn validate_count_values( + schema: &SeriesSchema, label: &str, grouping: Grouping<'_>, -) -> Result> { +) -> Result<()> { // `by` may list the value label, which this operator writes itself, even when an input field // has that name; grouping by any other field name stays unsupported. let other_labels: Vec<&str>; @@ -606,16 +602,30 @@ pub fn count_values( } without => without, }; - // Unlike the other aggregations, this one reads histogram samples too. validate_grouping(schema, checked)?; - for series in input { - validate_grid_series(schema, series)?; - } if label.is_empty() { return Err(DataFusionError::Execution(format!( "invalid label name {label:?}" ))); } + Ok(()) +} + +/// `count_values(label, v)`: at each grid point, the number of samples in each group of series +/// that share a value, with the value written to `label` in Prometheus' `FormatFloat(v, 'f', -1)` +/// form, or for a native histogram as its `FloatHistogram.String`. With `by`, `label` is part of +/// the grouping; with `without`, it is dropped if listed. +pub fn count_values( + schema: &Arc, + input: &[Series], + label: &str, + grouping: Grouping<'_>, +) -> Result> { + validate_count_values(schema, label, grouping)?; + // Unlike the other aggregations, this one reads histogram samples too. + for series in input { + validate_grid_series(schema, series)?; + } let by_labels: Vec<&str>; let grouping = match grouping { Grouping::By(labels) => { @@ -704,6 +714,18 @@ fn validate_input(schema: &SeriesSchema, input: &[Series], grouping: Grouping<'_ Ok(()) } +/// Validates the grouping a whole-vector operation addresses its series by. +/// +/// The native route keys `topk`, `bottomk` and `limitk` groups by series labels only, while the +/// existing planner also accepts a value field there, where it varies between samples. Refusing +/// the field here keeps such a statement on the existing plan before native execution is chosen. +pub(crate) fn validate_vector_grouping( + schema: &SeriesSchema, + grouping: Grouping<'_>, +) -> Result<()> { + validate_grouping(schema, grouping) +} + fn validate_grouping(schema: &SeriesSchema, grouping: Grouping<'_>) -> Result<()> { if schema.eval_grid.is_none() { return Err(DataFusionError::Execution( diff --git a/src/promql/src/extension_plan/aggregate_finalize.rs b/src/promql/src/extension_plan/aggregate_finalize.rs index c29df4f4d45..0c83434bac1 100644 --- a/src/promql/src/extension_plan/aggregate_finalize.rs +++ b/src/promql/src/extension_plan/aggregate_finalize.rs @@ -68,8 +68,8 @@ use crate::grid_series::series_bytes; use crate::map::MathOp; use crate::native::execute::{InputId, NativeExecutionState, NativeExecutor}; use crate::native::plan::{ - AggregateKind, AggregateStage, BoundAggregate, BoundInput, Grouping, MapOp, NativeExpr, - NativeNode, NativeTree, NativeType, + AggregateKind, AggregateStage, BoundAggregate, BoundInput, BoundVectorOp, Grouping, MapOp, + NativeExpr, NativeNode, NativeTree, NativeType, }; /// One node that merges the partition-local state of a split aggregate. @@ -348,12 +348,21 @@ struct Split { /// Splits the innermost aggregate out of `node`. /// /// `Ok(None)` is a subtree holding no aggregate. The currently verified suffix operations are maps, -/// aggregates, TopK/BottomK and binary operations with a constant on one side. TopK/BottomK run once -/// in the suffix over the fully merged aggregate series, per evaluation timestamp, and never in the -/// partition-local producer. A map is point-wise and an aggregate runs over the complete merge. -/// LimitK is declined because it selects the first `k` series in input order -/// ([`crate::select::limitk`]), while this merge does not preserve the storage input order. Other -/// vector operations have not yet been verified/supported by this split. +/// aggregates, TopK/BottomK, classic histogram functions (`histogram_quantile`, +/// `histogram_fraction`) and binary operations with a constant on one side. TopK/BottomK and the +/// histogram functions run once in the suffix over the fully merged aggregate series, per evaluation +/// timestamp, and never in the partition-local producer. +/// +/// A histogram function reads each classic histogram from the bucket series that share its other +/// labels, and drops `le`. The merge has already folded every partition's contribution to each +/// `(..., le)` group into one complete series on the evaluated grid, so the bucket series it reads +/// are exactly the ones this statement's input would have produced in one execution: reading `le` +/// off those merged series is the same computation, not an approximation. TopK is already accepted +/// for that reason: it runs once over the fully merged series. This admits a second operation of +/// that kind, not a new mechanism. LimitK stays declined because it depends on the storage input +/// order, which the merge does not preserve ([`crate::select::limitk`]). A map is point-wise and an +/// aggregate runs over the complete merge. Other vector operations have not yet been +/// verified/supported by this split. /// /// An instant selection or window above the cut cannot be applied to the merged values under the /// tree's current binding: the cut output is evaluated, whereas these nodes consume raw samples. @@ -421,7 +430,7 @@ fn split(node: &NativeNode, fields: &[FieldSpec], time_index: &str) -> DfResult< } NativeExpr::Vector { input, - op: op @ crate::native::plan::BoundVectorOp::TopK { .. }, + op: op @ (BoundVectorOp::TopK { .. } | BoundVectorOp::Histogram(_)), } => { let Some(mut inner) = split(input, fields, time_index)? else { return Ok(None); diff --git a/src/promql/src/extension_plan/aggregate_finalize/tests.rs b/src/promql/src/extension_plan/aggregate_finalize/tests.rs index 097eb7ffcf3..4215358db43 100644 --- a/src/promql/src/extension_plan/aggregate_finalize/tests.rs +++ b/src/promql/src/extension_plan/aggregate_finalize/tests.rs @@ -24,7 +24,9 @@ use datafusion::common::DFSchema; use datatypes::series::samples::FieldSpec; use datatypes::series::time::{EvalGrid, GridSpec}; +use crate::histogram_eval::HistogramFunction; use crate::native::plan::BoundVectorOp; +use crate::select::SortOrder; use crate::vector_match::{BinaryMatching, BinaryOp, Cardinality, FillValues, MatchingLabels}; /// The statement's grid, which every evaluated chunk is addressed on. @@ -384,6 +386,73 @@ fn topk_and_bottomk_run_after_the_aggregate_cut() { } } +/// A classic histogram function above the cut reads each histogram off the bucket series that share +/// its other labels, and the merge has already folded every partition's `(..., le)` group into one +/// complete series: the set it reads is the one the statement's input would have produced in one +/// execution, so this is the same computation and not an approximation. This is the o11ybench Q10 +/// shape. +#[test] +fn histogram_quantile_over_an_aggregate_splits() { + let tree = NativeTree::try_new(vector_over( + complete(AggregateKind::Sum, by(&["instance", "le"]))(window()), + BoundVectorOp::Histogram(HistogramFunction::Quantile(0.99)), + )) + .unwrap(); + assert!(AggregateFinalize::can_split(&tree, &raw_fields(), "ts")); + + // The function stays above the cut, reading the merged series at its own source `0`. + let node = finalize( + tree.root().clone(), + output_schema(&["sum(v0)"], &["instance", "le"]), + ) + .unwrap(); + let NativeExpr::Vector { + input, + op: BoundVectorOp::Histogram(HistogramFunction::Quantile(quantile)), + } = &node.suffix.root().expr + else { + panic!( + "the histogram function stays above the cut: {}", + node.suffix + ); + }; + assert_eq!(0.99, *quantile); + assert!( + matches!(&input.expr, NativeExpr::Input(bound) if bound.source_index == 0), + "the merged values enter where the aggregate was: {}", + node.suffix + ); + + // The prefix is the per-series tree: the histogram function is not one partition's to compute. + let prefix = producer(&node).tree().to_string(); + assert!(!prefix.contains("histogram_quantile"), "{prefix}"); + assert!(!prefix.contains("Aggregate"), "{prefix}"); + assert!(prefix.contains("Window"), "{prefix}"); +} + +/// Admitting the histogram functions widens one arm, not the whole `Vector` family: every other +/// vector operation is still declined, `sort` here because it reorders the input rather than reading +/// each series of it. +#[test] +fn a_vector_operation_that_is_neither_topk_nor_a_histogram_is_declined() { + let tree = NativeTree::try_new(vector_over( + complete(AggregateKind::Sum, by(&["instance"]))(window()), + BoundVectorOp::Sort(SortOrder::Asc), + )) + .unwrap(); + assert!(!AggregateFinalize::can_split(&tree, &raw_fields(), "ts")); + + let error = finalize( + tree.root().clone(), + output_schema(&["sum(v0)"], &["instance"]), + ) + .unwrap_err(); + assert!( + matches!(error, DataFusionError::NotImplemented(_)), + "{error}" + ); +} + #[test] fn limitk_above_the_cut_is_declined_because_merge_does_not_preserve_input_order() { let error = finalize( diff --git a/src/promql/src/native/execute.rs b/src/promql/src/native/execute.rs index bf0f4e40de7..86d58d8377c 100644 --- a/src/promql/src/native/execute.rs +++ b/src/promql/src/native/execute.rs @@ -1707,14 +1707,41 @@ struct VectorRuntime { } impl VectorRuntime { - fn new(op: BoundVectorOp, schema: Arc) -> Self { - Self { + /// Validates schema-dependent requirements before the native route is selected. + fn try_new(op: BoundVectorOp, schema: Arc) -> Result { + // Every grouping these operations address their series by is checked here: `topk`, + // `bottomk` and `limitk` key groups by series labels only, so a `by` naming a value field + // would fold series the existing plan keeps apart into one group. + let bound_grouping = match &op { + BoundVectorOp::TopK { group, .. } + | BoundVectorOp::LimitK { group, .. } + | BoundVectorOp::CountValues { group, .. } => Some(group), + _ => None, + }; + if let Some(group) = bound_grouping { + let names: Vec<&str> = match group { + Grouping::By(names) | Grouping::Without(names) => { + names.iter().map(AsRef::as_ref).collect() + } + }; + let grouping = match group { + Grouping::By(_) => aggregate::Grouping::By(&names), + Grouping::Without(_) => aggregate::Grouping::Without(&names), + }; + match &op { + BoundVectorOp::CountValues { label, .. } => { + aggregate::validate_count_values(&schema, label, grouping)? + } + _ => aggregate::validate_vector_grouping(&schema, grouping)?, + } + } + Ok(Self { op, schema, input: RetainedInput::default(), merge_input: false, computed: false, - } + }) } /// Runs the kernel over the whole input and returns the series it kept, or `None` when it @@ -3055,7 +3082,7 @@ impl RuntimeNode { .to_string(), )); } - let mut vector = VectorRuntime::new(op.clone(), schema); + let mut vector = VectorRuntime::try_new(op.clone(), schema)?; let (child, child_shared) = Self::build(input, pool, inputs, format!("{location}.input"))?; let limitk = matches!(op, BoundVectorOp::LimitK { .. }); diff --git a/src/promql/src/native/execute/tests.rs b/src/promql/src/native/execute/tests.rs index dc13671be34..1f34b8a362a 100644 --- a/src/promql/src/native/execute/tests.rs +++ b/src/promql/src/native/execute/tests.rs @@ -10159,6 +10159,101 @@ fn vector_op(inner: NativeNode, op: BoundVectorOp) -> NativeNode { } } +#[test] +fn count_values_runtime_rejects_grouping_by_another_value_field() { + let grid = grid(0, 60_000, 1); + let input = instant(raw_input_of(2), 60_000, 0, Arc::clone(&grid)); + let tree = NativeTree::try_new(vector_op( + input, + BoundVectorOp::CountValues { + label: "sample".into(), + group: Grouping::By(Arc::from([Arc::::from("v1")])), + }, + )) + .unwrap(); + let error = NativeExecutionState::try_new(&tree, pool()) + .err() + .expect("grouping by another value field is unsupported"); + assert!( + matches!(error, DataFusionError::NotImplemented(_)), + "{error}" + ); + assert!( + error + .to_string() + .contains("aggregate by value field requires a per-point label/schema adapter"), + "{error}" + ); +} + +#[test] +fn count_values_runtime_accepts_grouping_by_its_target_label() { + let grid = grid(0, 60_000, 1); + let tree = NativeTree::try_new(vector_op( + instant(input(), 60_000, 0, Arc::clone(&grid)), + BoundVectorOp::CountValues { + label: "v0".into(), + group: Grouping::By(Arc::from([Arc::::from("v0")])), + }, + )) + .unwrap(); + let mut execution = executor(&tree, pool()); + execution + .push_raw( + InputId::new(0), + raw_chunk(&raw_schema(), &[0], &[3.0], true), + ) + .unwrap(); + let chunks = run(&mut execution); + assert_eq!(chunks.len(), 1); + assert_eq!(chunks[0].series.metric.get("v0"), Some("3")); + assert_eq!(decided(&chunks), vec![(0, Some(1.0))]); +} + +#[test] +fn topk_runtime_rejects_grouping_by_a_value_field() { + let grid = grid(0, 60_000, 1); + let input = instant(input(), 60_000, 0, Arc::clone(&grid)); + let tree = NativeTree::try_new(vector_op( + input, + BoundVectorOp::TopK { + k: 1.0, + top: true, + group: Grouping::By(Arc::from([Arc::::from("v0")])), + }, + )) + .unwrap(); + let error = NativeExecutionState::try_new(&tree, pool()) + .err() + .expect("grouping by a value field is unsupported"); + assert!( + matches!(error, DataFusionError::NotImplemented(_)), + "{error}" + ); + assert!( + error + .to_string() + .contains("aggregate by value field requires a per-point label/schema adapter"), + "{error}" + ); +} + +#[test] +fn topk_runtime_accepts_grouping_by_a_label() { + let grid = grid(0, 60_000, 1); + let input = instant(input(), 60_000, 0, Arc::clone(&grid)); + let tree = NativeTree::try_new(vector_op( + input, + BoundVectorOp::TopK { + k: 1.0, + top: true, + group: Grouping::By(Arc::from([Arc::::from("host")])), + }, + )) + .unwrap(); + NativeExecutionState::try_new(&tree, pool()).expect("a label grouping stays supported"); +} + /// The positions a series decided, with their values. type Decided = Vec<(usize, Option)>; diff --git a/src/query/src/promql/planner/native.rs b/src/query/src/promql/planner/native.rs index bd14e9d4f2f..b24a14002a3 100644 --- a/src/query/src/promql/planner/native.rs +++ b/src/query/src/promql/planner/native.rs @@ -42,15 +42,20 @@ use common_query::prelude::OTLP_AGGREGATION_TEMPORALITY_LABEL; use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, TimeUnit as ArrowTimeUnit}; use datafusion::common::{DFSchema, DFSchemaRef}; use datafusion::error::DataFusionError; +use datafusion::execution::memory_pool::UnboundedMemoryPool; use datafusion::logical_expr::{Extension, LogicalPlan, LogicalPlanBuilder, TableScanBuilder}; use datafusion::prelude::Expr as DfExpr; use datafusion_expr::Expr; use datatypes::series::SeriesOrdering; use datatypes::series::samples::{FieldSpec, SeriesSchema}; use promql::extension_plan::{AggregateFinalize, FusedSeries, FusedSeriesBinding, RawSeriesSource}; -use promql::native::plan::{NativeExpr, NativeNode, NativeType, reads_across_series}; +use promql::native::execute::NativeExecutionState; +use promql::native::plan::{ + BoundAggregate, BoundVectorOp, Grouping, MapOp, NativeExpr, NativeNode, NativeType, + reads_across_series, +}; use promql::vector_match::{Cardinality, MatchingLabels, SetOp}; -use promql_parser::label::Matchers; +use promql_parser::label::{METRIC_NAME, Matchers}; use promql_parser::parser::value::ValueType; use promql_parser::parser::{ BinModifier, EvalStmt, Expr as PromExpr, LabelModifier, MatrixSelector, ParenExpr, @@ -178,8 +183,9 @@ impl PromPlanner { /// Plans `prom_expr` as one native node, or returns `None` to keep the existing plan. /// - /// `query_engine_state` is only read to build the *existing* plan of an aggregate query, whose - /// column names and grouped labels are what the rows of a split aggregate are named after. That + /// `query_engine_state` is only read to build the *existing* plan of the same statement, for the + /// column names the existing path gives it: they are what the rows of a split aggregate are named + /// after, and what names the tags of a tree whose labels the walk below does not derive. That /// plan is built for its schema, its names are copied, and it is dropped: it is never executed /// and never stored, so the batch that answers the query is the native route's own. /// @@ -283,6 +289,10 @@ impl PromPlanner { let preparation = async { let mut prepared: Vec = Vec::new(); let mut single_input_context = None; + // The first metric logical-table id a selector's rewrite carried, and whether a later one + // carried a different id. See the `cross_table` use below. + let mut logical_scan_id: Option = None; + let mut logical_ids_differ = false; for selector in &selectors { self.ctx = original_context.clone(); // These matchers are alternatives the one scan below the node cannot represent, so the @@ -337,8 +347,20 @@ impl PromPlanner { let mut input = self.create_table_scan_plan(table_ref).await?; let logical_scan = metric_scan_and_table_id(&input); let has_verified_metric_rewrite = logical_scan.is_some(); + // A metric logical table's id is part of the identity its own scan reads: two + // logical tables served through one physical table prepare two scans of that one + // table, and the ids are what says they are two metrics. A selector whose scan is not + // a logical rewrite contributes no id, and a missing id on either side is not a + // difference. + if let Some((_, _, table_id)) = &logical_scan { + match logical_scan_id { + None => logical_scan_id = Some(*table_id), + Some(first) if first != *table_id => logical_ids_differ = true, + Some(_) => {} + } + } let scan = match logical_scan { - Some((mut scan, predicate)) => { + Some((mut scan, predicate, _)) => { // Preserve logical field order and the table restriction on the direct // physical scan. let source_schema = scan.source.schema(); @@ -381,11 +403,12 @@ impl PromPlanner { // there is a planning error, not a fallback. So the engine is checked here, where the // existing plan is still the answer. A file-engine table is the concrete case: its // scanner is a `SinglePartitionScanner`, which leaves `raw_series_schema` at the - // default `None`. A metric-engine table is served only through the logical rewrite - // recognised above, which is what makes its series the table's own. + // default `None`. A metric-engine selector is admitted through its verified logical + // rewrite, with no separate multi-selector exclusion: each input is a metric logical + // table served through that rewrite, and the ids those rewrites carry are what tells + // two different ones apart below. Two occurrences of one logical table stay two + // scans of the one physical table, answered where that table's region runs them. if !supports_native_raw_series(&table_meta.engine) - || (selectors.len() > 1 - && !matches!(table_meta.engine.as_str(), MITO_ENGINE | MITO2_ENGINE)) || (table_meta.engine == METRIC_ENGINE_NAME && !has_verified_metric_rewrite) { return Ok(None); @@ -451,17 +474,20 @@ impl PromPlanner { } prepared.push((input, series_schema, Some(Arc::from(metric)), time, tags)); } - Ok(Some((prepared, single_input_context))) + Ok(Some((prepared, single_input_context, logical_ids_differ))) } .await; self.ctx = original_context; - let Some((mut prepared, single_input_context)) = preparation? else { + let Some((mut prepared, single_input_context, logical_ids_differ)) = preparation? else { return Ok(None); }; // Selectors over one table share one region, so the node runs there over every scan. When // they read more than one table, each scan answers from its own region and the node runs - // above them; see `RawSeriesSource`. - let cross_table = !same_table_scans(prepared.iter().map(|(scan, ..)| scan)); + // above them; see `RawSeriesSource`. A metric logical table's id is part of the identity it + // reads: two ids are two metrics even when one physical table holds both, so two selectors + // that differ only by that id still read different series and are bound as two inputs. + let cross_table = + !same_table_scans(prepared.iter().map(|(scan, ..)| scan)) || logical_ids_differ; // The first input names the output's time index and, on one table, binds every scan. let Some((_, series_schema, metric, time, tags)) = prepared.first() else { return Ok(None); @@ -588,25 +614,70 @@ impl PromPlanner { .collect::, DataFusionError>>() .context(DataFusionPlanningSnafu)?; // Ordinary pointwise trees and trees of one-to-one matches, on one table or across - // several, are answered per partition. Aggregate splits are restricted to one raw scan - // child. + // several, are answered per partition. A tree that folds across series - an aggregate - is + // not, and one selector's fold is carried by the aggregate split below. let root = compiled.tree.root(); let per_partition = !reads_across_series(root) || pointwise_operand(root) || independent_scan_pair(root); - if !per_partition { + // The split is only ever a one-child plan: it exports each partition's leaf state and merges + // them above the region. A tree over two metric logical tables of ONE physical table is not + // split, because that node already runs one execution over every input partition: a + // `FusedSeries` with more than one input declares one partition, and its `execute` reads all + // of every input's partitions (`if self.inputs.len() > 1 { 0..count }` in `fused_series.rs`), + // so every group is folded in one place. Both of those inputs are `RawSeriesSource` children + // (the two ids made `cross_table` true), each answering from its own region with frames that + // carry their own labels, so the node is plannable above both. That plan needs no split to be + // correct, so it falls through to the node below. + // + // A multi-selector tree whose inputs stay direct scans of one table is not admitted here: its + // scans are not raw sources, so the categorizer refuses to place the node region-locally + // (`Categorizer::check_extension_plan` is `NonCommutative` for a cross-series node over more + // than one scan) and the physical planner cannot bind a merged plain scan under it either. So + // such a tree keeps its one-selector-or-decline behaviour, unchanged. + // + // A selector whose scan carries no metric logical id - a Mito table's, for one - adds no + // id, so one Mito aggregate beside one metric aggregate is not a multi-input tree: + // `sum(m) + sum(mito)` keeps the existing plan. A second metric logical id does make the + // tree multi-input, and the one execution above then reads the Mito input through its own + // raw source as well. + let multi_input = selectors.len() > 1 && logical_ids_differ; + // A multi-input tree skips the split below, so the refusals the split makes for reasons that + // are not about what the runtime computes have to be made here. + // + // The aggregate split keeps its prefix per-series, which is why it declines an operation + // whose answer depends on the order its input arrives in. A multi-input node reads its + // inputs' partitions in the order they became ready, so that refusal is repeated here. + if multi_input && depends_on_input_order(root) { + return Ok(None); + } + // The split names the columns its groups report from the existing planner. A grouping by the + // metric name keeps a label no column of the existing plan can name, so the schema the node + // declares would report one label set for groups the tree keeps apart. + if multi_input && groups_by_metric_name(root) { + return Ok(None); + } + // Catch construction-time refusals before choosing native execution: errors raised during + // physical planning cannot fall back. This includes unsupported aggregates and `topk`, + // `limitk` or CountValues groupings; it does not cover data-dependent errors raised while + // executing the tree. + if multi_input + && NativeExecutionState::try_new( + &compiled.tree, + Arc::new(UnboundedMemoryPool::default()), + ) + .is_err() + { + return Ok(None); + } + if !per_partition && !multi_input { if selectors.len() != 1 || !AggregateFinalize::can_split(&compiled.tree, &series_schema.fields, &time) { return Ok(None); } - let saved = self.ctx.clone(); - let planned = self - .prom_root_expr_to_plan(prom_expr, query_engine_state) - .await; - let legacy_fields = self.ctx.field_columns.clone(); - let legacy_tags = self.ctx.tag_columns.clone(); - self.ctx = saved; - drop(planned?); + let (legacy_fields, legacy_tags) = self + .legacy_output_columns(prom_expr, query_engine_state) + .await?; let output_schema = fused_output_schema(&legacy_fields, &legacy_tags, &time)?; let Some(rawinput) = children.into_iter().next() else { return Ok(None); @@ -635,22 +706,40 @@ impl PromPlanner { ))); } let tree = compiled.tree; - let (root_fields, output_tags) = match &tree.root().output { + // The tags are asked for before the fields, because the walk answers for a tree the + // compiler produced: a scalar root has no labels at all, and a sample root's are the ones + // the walk derives or - where it derives none - the ones the existing planner names. + let output_tags = match &tree.root().output { + // A scalar has no labels, and its value is named as a bare literal's is. + NativeType::Scalar(_) => Vec::new(), + NativeType::Samples(_) => match fused_output_tags(tree.root(), &source_tags) { + Some(output_tags) => output_tags, + None => { + // The walk cannot name this tree's labels, so the columns the existing planner + // names for the same statement are the answer rather than a decline. + let (_, legacy_tags) = self + .legacy_output_columns(prom_expr, query_engine_state) + .await?; + // Only the tag columns are copied: the node names its field columns after the + // tree's own root schema, while the existing planner can name a computed column + // after its expression, so its field names are not this node's. A tag that + // collides with a field or the time index is still rejected by the uniqueness + // check below. + legacy_tags + } + }, + NativeType::AggregateState(_) => return Ok(None), + }; + let root_fields = match &tree.root().output { // The fields are the compiled root's own: a raw field's name, or the one `vector(s)` // gives its result. - NativeType::Samples(schema) => { - let fields = schema - .fields - .iter() - .map(|f| f.name.to_string()) - .collect::>(); - let Some(output_tags) = fused_output_tags(tree.root(), &source_tags) else { - return Ok(None); - }; - (fields, output_tags) - } + NativeType::Samples(schema) => schema + .fields + .iter() + .map(|f| f.name.to_string()) + .collect::>(), // A scalar has no labels, and its value is named as a bare literal's is. - NativeType::Scalar(_) => (vec![DEFAULT_FIELD_COLUMN.to_string()], Vec::new()), + NativeType::Scalar(_) => vec![DEFAULT_FIELD_COLUMN.to_string()], NativeType::AggregateState(_) => return Ok(None), }; // A root field can share a name with a tag or the time index, and under `or` the right @@ -665,7 +754,12 @@ impl PromPlanner { return Ok(None); } let output_schema = fused_output_schema(&root_fields, &output_tags, &time)?; - // Across tables the region scans are bound by their `RawSeriesSource` children. + // Across tables the region scans are bound by their `RawSeriesSource` children, and this is + // also the binding a node over more than one input needs: such a node reads frames that + // already carry their own labels, so it binds neither tags nor one metric name of its own - + // each child names the metric it reads. A multi-input plan whose children stay direct + // scans reads one table through the one metric and tag set they share, and keeps this + // binding, which is the same identity those scans carry. let (tags, metric) = if cross_table { (Vec::new(), None) } else { @@ -692,6 +786,33 @@ impl PromPlanner { ))) } + /// The field and tag columns the *existing* planner names for `prom_expr`. + /// + /// The names are read off the plan the existing path builds for the same statement, and that + /// plan is then dropped: it is never executed and never stored, so the batch that answers the + /// query is the native route's own. A statement the existing planner cannot plan fails here the + /// same way it fails there, because the existing plan is the answer for this statement: an error + /// planning it is the error the statement has, whichever route asked. + /// + /// Only the planner context is saved and restored: the table provider's resolved-table cache + /// keeps the entries this speculative planning resolved, exactly as the aggregate-split branch, + /// which planned the same way before this helper existed, always did. + async fn legacy_output_columns( + &mut self, + prom_expr: &PromExpr, + query_engine_state: &QueryEngineState, + ) -> Result<(Vec, Vec)> { + let saved = self.ctx.clone(); + let planned = self + .prom_root_expr_to_plan(prom_expr, query_engine_state) + .await; + let legacy_fields = self.ctx.field_columns.clone(); + let legacy_tags = self.ctx.tag_columns.clone(); + self.ctx = saved; + drop(planned?); + Ok((legacy_fields, legacy_tags)) + } + /// The statement being planned, rebuilt from the evaluation numbers the planner holds. /// /// The compiler reads the statement's grid, lookback and `@` anchors rather than the planner's @@ -841,8 +962,10 @@ pub(crate) fn same_table_scans<'a>(plans: impl IntoIterator]) -> Option> { match &node.expr { NativeExpr::Input(input) => source_tags.get(input.source_index).cloned(), @@ -851,9 +974,19 @@ fn fused_output_tags(node: &NativeNode, source_tags: &[Vec]) -> Option fused_output_tags(input, source_tags), + // A label rewrite changes the label set itself: `label_replace` adds its destination label + // and an empty replacement removes it, so the input's labels are not the output's. A map of + // math alone keeps them. + NativeExpr::Map { input, op } => { + if op.ops.iter().any(|op| matches!(op, MapOp::Label(_))) { + return None; + } + fused_output_tags(input, source_tags) + } + // Its parameters are scalars, which carry no labels, so what the node reports are its + // input's labels. + NativeExpr::ParameterizedMap { input, .. } => fused_output_tags(input, source_tags), NativeExpr::Aggregate { .. } | NativeExpr::Vector { .. } | NativeExpr::Merge { .. } => None, NativeExpr::Binary { left, @@ -937,6 +1070,92 @@ fn data_free(node: &NativeNode) -> bool { } } +/// Whether `node` holds an operation whose answer depends on the order its input arrives in. +/// +/// `limitk` keeps the first `k` series of each group *in input order* ([`promql::select::limitk`]). +/// A node with more than one input reads its inputs' partitions through a gather that hands over +/// whichever partition became ready first, so the schedule would decide which series a query keeps +/// and the same data could answer differently between two runs. Prometheus defines the selection +/// over a series set, not over an arrival order. +/// +/// Only this operation's *selection* depends on that order. `topk` breaks ties by label and +/// `limit_ratio` selects by a label hash, so a reordered gather changes neither. The instant +/// `sort`s keep tied series in the order they arrived, and a floating-point fold such as `stdvar` +/// can differ in its last bits, but neither changes which series a statement keeps. The aggregate +/// split declines `LimitK` for the same reason, and only a multi-input node repeats that refusal. +fn depends_on_input_order(node: &NativeNode) -> bool { + match &node.expr { + NativeExpr::Input(_) + | NativeExpr::Scalar(_) + | NativeExpr::EvalTime { .. } + | NativeExpr::VectorOf { .. } => false, + NativeExpr::Instant { input, .. } + | NativeExpr::Merge { input } + | NativeExpr::Window { input, .. } + | NativeExpr::Map { input, .. } + | NativeExpr::Timestamp { input } + | NativeExpr::Aggregate { input, .. } => depends_on_input_order(input), + NativeExpr::ParameterizedMap { + input, parameters, .. + } => depends_on_input_order(input) || parameters.iter().any(depends_on_input_order), + NativeExpr::Vector { input, op } => { + matches!(op, BoundVectorOp::LimitK { .. }) || depends_on_input_order(input) + } + NativeExpr::Binary { left, right, .. } | NativeExpr::Set { left, right, .. } => { + depends_on_input_order(left) || depends_on_input_order(right) + } + NativeExpr::Info { base, info, .. } => { + depends_on_input_order(base) || depends_on_input_order(info) + } + } +} + +/// Whether a grouping in `node` keeps a label the existing planner's column list cannot name. +/// +/// An aggregate's `by(...)` keeps the labels it lists that its input carries, and `__name__` is one +/// of them: the aggregate folds its groups by the metric name as by any other label. It is not a +/// column of any scan, though - it is the name the raw frame carries - so the existing planner's +/// `by(__name__)` finds no such column and reports no label at all. A schema taken from that list +/// would drop the metric name the native aggregate kept, and two groups differing only in their +/// metric would come out as two rows under one label set, which is a series identity lost rather +/// than an answer given. +/// +/// `without(...)` needs no such check: both routes drop `__name__` there. Nor does a tree with one +/// selector: it reads one metric, so every group it folds carries the same name and naming that name +/// or dropping it cannot merge two groups. What the check is for is the tree that reads more than one +/// metric, which is the one a multi-input node is bound for. +fn groups_by_metric_name(node: &NativeNode) -> bool { + let grouping_names_metric = |op: &BoundAggregate| matches!(&op.group, Grouping::By(names) if names.iter().any(|name| name.as_ref() == METRIC_NAME)); + match &node.expr { + NativeExpr::Input(_) + | NativeExpr::Scalar(_) + | NativeExpr::EvalTime { .. } + | NativeExpr::VectorOf { .. } => false, + NativeExpr::Instant { input, .. } + | NativeExpr::Merge { input } + | NativeExpr::Window { input, .. } + | NativeExpr::Map { input, .. } + | NativeExpr::Timestamp { input } => groups_by_metric_name(input), + NativeExpr::Aggregate { input, op } => { + grouping_names_metric(op) || groups_by_metric_name(input) + } + NativeExpr::ParameterizedMap { + input, parameters, .. + } => groups_by_metric_name(input) || parameters.iter().any(groups_by_metric_name), + NativeExpr::Vector { input, op } => { + matches!(op, BoundVectorOp::CountValues { group: Grouping::By(names), .. } + if names.iter().any(|name| name.as_ref() == METRIC_NAME)) + || groups_by_metric_name(input) + } + NativeExpr::Binary { left, right, .. } | NativeExpr::Set { left, right, .. } => { + groups_by_metric_name(left) || groups_by_metric_name(right) + } + NativeExpr::Info { base, info, .. } => { + groups_by_metric_name(base) || groups_by_metric_name(info) + } + } +} + /// Whether `node` reads samples point-wise, so that one partitioned execution is one answer. /// /// A node's output type is not checked here: [`NativeNode::validate`] already refuses a selection, @@ -1091,6 +1310,11 @@ fn collect_selectors<'a>( PromExpr::NumberLiteral(_) => return true, PromExpr::Unary(unary) => return collect_selectors(stmt, &unary.expr, out), PromExpr::Call(call) => { + // A label rewrite is a point-wise map, so its selectors would be its argument's - but + // the runtime refuses two series that a rewrite gives one label set + // (`a_label_rewrite_relabels_series_and_refuses_a_shared_point`), where the existing + // plan answers them. Admitting the shape would turn an answer into an error, so the + // statement keeps the existing plan until that refusal has a fallback. if matches!(call.func.name, "label_replace" | "label_join") { return false; } @@ -1114,10 +1338,12 @@ fn supports_native_raw_series(engine: &str) -> bool { matches!(engine, MITO_ENGINE | MITO2_ENGINE | METRIC_ENGINE) } -/// Recognizes only the metric logical-table scan emitted by `create_table_scan_plan`. +/// Recognizes only the metric logical-table scan emitted by `create_table_scan_plan`, and reports +/// the logical table's id together with the scan: the rewrite's `__table_id = ` restriction is +/// the logical table the rewrite stands for, so it is the identity a selector over it reads. fn metric_scan_and_table_id( plan: &LogicalPlan, -) -> Option<(datafusion::logical_expr::TableScan, Expr)> { +) -> Option<(datafusion::logical_expr::TableScan, Expr, u32)> { let plan = match plan { LogicalPlan::Projection(projection) => projection.input.as_ref(), plan => plan, @@ -1137,16 +1363,15 @@ fn metric_scan_and_table_id( { return None; } - if !matches!( - binary.right.as_ref(), - Expr::Literal(datafusion::scalar::ScalarValue::UInt32(Some(_)), _) - ) { + let Expr::Literal(datafusion::scalar::ScalarValue::UInt32(Some(table_id)), _) = + binary.right.as_ref() + else { return None; - } + }; let LogicalPlan::TableScan(scan) = filter.input.as_ref() else { return None; }; - Some((scan.clone(), filter.predicate.clone())) + Some((scan.clone(), filter.predicate.clone(), *table_id)) } /// Whether `expr` calls `rate`/`increase` anywhere in its admitted expression tree. @@ -1417,9 +1642,29 @@ mod tests { "{query}" ); } - // An aggregate's labels are its groups, which this walk does not derive. + // An aggregate's labels are its groups, which this walk does not derive. The route still + // answers the query: it asks the existing planner for the columns instead of declining. let aggregate = compiled_test_tree("sum by (host) (x)"); assert_eq!(None, fused_output_tags(aggregate.root(), &sources)); + // A label rewrite renames the label set below it, which this walk does not derive either. + for query in [ + r#"label_replace(x, "dst", "$1", "src", "(.*)")"#, + r#"label_join(x, "dst", "-", "src", "src")"#, + ] { + let rewrite = compiled_test_tree(query); + assert_eq!(None, fused_output_tags(rewrite.root(), &sources), "{query}"); + } + } + + /// A map that only computes values passes its input's labels through unchanged. + #[test] + fn match_output_tags_pass_through_pointwise_math() { + let sources = [vec!["host".to_string(), "kind".to_string()]]; + let tree = compiled_test_tree("abs(x) * 2"); + assert_eq!( + Some(vec!["host".to_string(), "kind".to_string()]), + fused_output_tags(tree.root(), &sources) + ); } #[test] diff --git a/src/query/src/promql/planner/test.rs b/src/query/src/promql/planner/test.rs index e00d0ca13f2..a54e4d4bf44 100644 --- a/src/query/src/promql/planner/test.rs +++ b/src/query/src/promql/planner/test.rs @@ -2220,10 +2220,33 @@ async fn native_fused_metric_fields_follow_logical_order() { })); } -/// More than one selector over metric-engine logical tables is not one multi-input node, though -/// one selector over the same tables is native. +/// Every selector `plan`'s fused node reads answers with a raw source naming its own metric. +fn fused_input_metrics(plan: &LogicalPlan) -> Vec { + let fused = find_fused_series(plan).unwrap_or_else(|| panic!("no fused node in {plan}")); + fused + .inputs() + .iter() + .map(|input| { + let LogicalPlan::Extension(Extension { node }) = input else { + panic!("expected a raw source, got {input}"); + }; + node.as_any() + .downcast_ref::() + .unwrap_or_else(|| panic!("expected a raw source, got {input}")) + .metric_name() + .to_string() + }) + .collect() +} + +/// Two metric logical tables over one physical table are two metrics, one input each, though one +/// selector over either of them is one input and one binding. +/// +/// Both logical tables are rewritten to scans of the same physical table, so their scans name the +/// same table and only the id the rewrite restricts to tells them apart: a plan that read the first +/// binding for both inputs would report the first metric's answer twice. #[tokio::test] -async fn native_fused_multiple_metric_engine_selectors_fall_back() { +async fn native_fused_metric_engine_selectors_keep_their_own_logical_table() { let tables = [ (DEFAULT_SCHEMA_NAME.to_string(), "some_metric".to_string()), (DEFAULT_SCHEMA_NAME.to_string(), "other_metric".to_string()), @@ -2231,14 +2254,72 @@ async fn native_fused_multiple_metric_engine_selectors_fall_back() { let provider = || build_test_table_provider_with_tsid(&tables, 1, 1); let single = build_native_fused_plan_with_provider("other_metric", true, provider().await).await; - assert!(find_fused_series(&single).is_some(), "{single}"); + let fused = find_fused_series(&single).unwrap_or_else(|| panic!("{single}")); + assert_eq!(Some("other_metric"), fused.binding().metric_name.as_deref()); for query in [ "some_metric / other_metric", - "some_metric + some_metric", "some_metric + other_metric + some_metric", ] { let native = build_native_fused_plan_with_provider(query, true, provider().await).await; - assert!(!has_pair_node(&native), "{query}: {native}"); + assert!(has_pair_node(&native), "{query}: {native}"); + let fused = find_fused_series(&native).unwrap_or_else(|| panic!("{query}: {native}")); + assert_eq!(None, fused.binding().metric_name, "{query}"); + assert!( + fused_input_metrics(&native) + .iter() + .any(|metric| metric == "other_metric"), + "{query}: {native}" + ); + } + let pair = + build_native_fused_plan_with_provider("some_metric / other_metric", true, provider().await) + .await; + assert_eq!( + vec!["some_metric".to_string(), "other_metric".to_string()], + fused_input_metrics(&pair), + "{pair}" + ); + // Two selectors of one logical table are one metric read twice, so they keep the one binding + // and the direct scans that name it. + let repeated = + build_native_fused_plan_with_provider("some_metric + some_metric", true, provider().await) + .await; + let fused = find_fused_series(&repeated).unwrap_or_else(|| panic!("{repeated}")); + assert_eq!(2, fused.inputs().len(), "{repeated}"); + assert_eq!( + Some("some_metric"), + fused.binding().metric_name.as_deref(), + "{repeated}" + ); +} + +/// Two aggregates over two metric logical tables of one physical table are one multi-input +/// evaluation, not an aggregate split: the node folds every input partition in one execution. +#[tokio::test] +async fn native_fused_metric_engine_two_aggregate_pair_is_one_execution() { + let tables = [ + (DEFAULT_SCHEMA_NAME.to_string(), "some_metric".to_string()), + (DEFAULT_SCHEMA_NAME.to_string(), "other_metric".to_string()), + ]; + let provider = || build_test_table_provider_with_tsid(&tables, 1, 1); + for query in [ + "sum(some_metric) / sum(other_metric)", + "sum(sum(some_metric)) / sum(other_metric)", + "sum by (tag_0) (some_metric) / sum by (tag_0) (other_metric)", + ] { + let native = build_native_fused_plan_with_provider(query, true, provider().await).await; + let fused = find_fused_series(&native) + .unwrap_or_else(|| panic!("{query}: no fused node: {native}")); + assert_eq!(2, fused.inputs().len(), "{query}: {native}"); + assert!( + fused.partial_fold().is_none(), + "{query}: a multi-input aggregate is not a split: {native}" + ); + assert_eq!( + vec!["some_metric".to_string(), "other_metric".to_string()], + fused_input_metrics(&native), + "{query}: {native}" + ); } } @@ -3195,6 +3276,27 @@ async fn native_fused_nested_aggregate_keeps_the_outer_names() { ); } +/// A classic histogram function above the cut is the split's: `histogram_quantile` reads each +/// histogram off the bucket series the merge produced, which are the ones the statement's input +/// would have produced in one execution, so the statement is answered by the merge over the partial +/// producer instead of falling back to the existing plan. +#[tokio::test] +async fn native_fused_histogram_quantile_over_aggregate_splits() { + let query = "histogram_quantile(0.99, sum by (tag_0) (rate(some_metric[5m])))"; + let plan = build_native_fused_plan(query, true).await; + let display = plan.display_indent().to_string(); + assert!( + find_aggregate_finalize(&plan).is_some(), + "`{query}` must be answered by a split aggregate: {display}" + ); + assert!( + find_fused_series(&plan).is_some(), + "`{query}` must keep the producer that exports the state: {display}" + ); + assert!(display.contains("AggregateFinalize: "), "{display}"); + assert!(display.contains("histogram_quantile"), "{display}"); +} + /// A shape the split cannot keep correct keeps the existing plan: a quantile's state is the group's /// samples, a binary operation whose sibling is data would match one side's series before the merge, /// a sibling a call computes has no native node at all, and a group by a value field cannot be named @@ -3275,12 +3377,15 @@ async fn native_fused_clamp_aggregate_keeps_the_existing_error() { } } -/// A label rewrite stays on the existing plan, which is the one that projects its new label. +/// A label rewrite keeps the existing plan, on every shape, because the native runtime refuses a +/// label set its rewrite creates twice where the existing plan answers it. /// -/// The native node's output schema is the table's own time index, fields and tags, so a tree that -/// rewrote labels would report a label set the existing plan's projection emits and the node cannot. -/// Each query below writes `newlabel`, which the fixture table does not have (its only tag is -/// `tag_0`), so a fused plan would visibly lose it. +/// The node's tags are copied from the existing plan, so a rewritten label is declarable - but the +/// runtime is why this shape is not admitted: two series a rewrite gives one label set collide, and +/// `a_label_rewrite_relabels_series_and_refuses_a_shared_point` (`promql/src/native/execute/tests.rs`) +/// pins that refusal. The existing plan has no such check, so admitting the shape would turn an +/// answer into an error. Every shape below therefore keeps the existing plan, the two-occurrence one +/// included: the refusal sits in `collect_selectors`, ahead of any placement decision. #[tokio::test] async fn native_fused_label_rewrites_keep_legacy_output_labels() { for query in [ @@ -3290,12 +3395,17 @@ async fn native_fused_label_rewrites_keep_legacy_output_labels() { // ... and below another call, which is the same shape one level deeper. r#"abs(label_replace(some_metric, "newlabel", "$1", "tag_0", "(.*)"))"#, r#"abs(label_join(some_metric, "newlabel", "-", "tag_0", "tag_0"))"#, + // A second occurrence does not admit it either: the collision is the runtime's, not the + // placement's, so the pair keeps the existing plan as well. + r#"label_replace(some_metric, "newlabel", "$1", "tag_0", "(.*)") + on(newlabel) some_metric"#, + r#"label_join(some_metric, "newlabel", "-", "tag_0", "tag_0") + some_metric"#, ] { let native = build_native_fused_plan(query, true).await; let display = native.display_indent().to_string(); assert!( find_fused_series(&native).is_none(), - "`{query}` rewrites labels the node's schema cannot declare: {display}" + "`{query}` writes a label the runtime would refuse twice, so it keeps the existing \ + plan: {display}" ); assert!( display.contains("newlabel"), diff --git a/tests-integration/src/tests/native_aggregate_test.rs b/tests-integration/src/tests/native_aggregate_test.rs index b99ec5c8a72..979726cadd4 100644 --- a/tests-integration/src/tests/native_aggregate_test.rs +++ b/tests-integration/src/tests/native_aggregate_test.rs @@ -127,6 +127,9 @@ const COUNTER_TABLE: &str = "native_agg_counter"; /// The right operand of the hybrid binary cases. It has one region but a partition column, so a /// pair reading it is declined by the two-scan native route and planned by the hybrid one. const COUNTER_RHS_TABLE: &str = "native_agg_counter_rhs"; +/// The table the histogram queries read: one bucket series per `(job, le)`, with `le` in the primary +/// key so the raw-series scan hands the bound over as a label. +const HIST_TABLE: &str = "native_agg_hist_bucket"; /// The table the range-window smoke test reads. const RANGE_TABLE: &str = "native_agg_range"; const HYBRID_MATCH_LEFT: &str = "native_hybrid_match_left"; @@ -191,6 +194,33 @@ const COUNTER_WEST_REGION: &str = "us-west-2"; /// `instance = h5`, whose rate is 4. const COUNTER_EAST_REGION: &str = "us-east-1"; +/// The count both samples of every histogram bucket series start from: the same value as +/// [`COUNTER_FIRST_VALUE`], and for the same reason - `rate`'s counter-zero clamp must not bind. +/// Here it is asked to cover increases of at most 6, so `interval * first / increase` is +/// `60 * 1500 / 6 = 15000s`, far above the 30s the boundary extrapolation adds. +const HIST_FIRST_VALUE: f64 = COUNTER_FIRST_VALUE; + +/// The bucket series of the histogram fixture in insert order, as `(job, le, increase)`: three +/// bounds per job, each holding a cumulative count, so a bound's count is never below the count of +/// the bound beneath it - the monotonicity `histogram_quantile` and `histogram_fraction` require +/// before they interpolate. The increases are 1, 2, 3 for `job = a` and 4, 5, 6 for `job = b`, all +/// distinct, so a route that read a bucket twice or lost one answers a different number. +/// +/// `rate` over the two samples at the counter fixture's instants divides by 150 exactly as +/// [`COUNTER_SERIES`] does, so `sum by (le)` over both jobs leaves 5/150, 7/150 and 9/150 in the +/// buckets `0.1`, `1` and `+Inf`. Those three counts put `histogram_quantile(0.99, ...)`'s rank +/// inside the `+Inf` bucket, where the answer is the bound below it - 1 - and +/// `histogram_quantile(0.5, ...)`'s nine tenths into the first bucket - 0.09 - while +/// `histogram_fraction(0, 1, ...)` closes under the `+Inf` count at 7/9. +const HIST_SERIES: [(&str, &str, f64); 6] = [ + ("a", "0.1", 1.0), + ("a", "1", 2.0), + ("a", "+Inf", 3.0), + ("b", "0.1", 4.0), + ("b", "1", 5.0), + ("b", "+Inf", 6.0), +]; + /// 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 @@ -1023,6 +1053,22 @@ fn counter_table_sql() -> String { ) } +/// The DDL of [`HIST_TABLE`]: [`create_table_sql`]'s shape with the instance replaced by the bucket +/// bound. +/// +/// `le` is part of the primary key so that it is a label of the series the raw-series scan reads, +/// not a field of one: the histogram functions find a bucket by a `le` label whose value parses as a +/// bound, and a bound that arrived as a field would leave them no bucket to read. TWCS is set as on +/// every other fixture table, so one flush per bucket series leaves several files for the scan to +/// split. +fn hist_table_sql() -> String { + format!( + "CREATE TABLE {HIST_TABLE} (ts TIMESTAMP(3) TIME INDEX, job STRING, le STRING, v DOUBLE, \ + PRIMARY KEY (job, le)) 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 @@ -1032,7 +1078,10 @@ fn counter_table_sql() -> String { /// one memtable range - but its samples are the two of [`COUNTER_SERIES`], [`COUNTER_STEP_SECS`] /// 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 +/// [`COUNTER_EAST_REGION`]. The histogram table holds the six bucket series of [`HIST_SERIES`] at +/// the counter samples' instants, flushed after each series but the last, so its buckets reach the +/// scan the same way - several files, one memtable range - and a bucket's partial sum is one +/// partition's to fold. 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. @@ -1090,6 +1139,26 @@ async fn setup_fixture(instance: &Arc) { } } + // The bucket series sit at the counter samples' instants, so `rate` over them extrapolates + // exactly as derived for [`COUNTER_SERIES`]; only the counts differ, and those are cumulative + // per bound. + execute_sql(instance, &hist_table_sql()).await; + for (index, (job, bound, increase)) in HIST_SERIES.iter().enumerate() { + let second_value = HIST_FIRST_VALUE + increase; + execute_sql( + instance, + &format!( + "INSERT INTO {HIST_TABLE} (ts, job, le, v) VALUES \ + ({counter_first}, '{job}', '{bound}', {HIST_FIRST_VALUE:?}), \ + ({counter_second}, '{job}', '{bound}', {second_value:?})" + ), + ) + .await; + if index + 1 < HIST_SERIES.len() { + execute_sql(instance, &format!("ADMIN FLUSH_TABLE('{HIST_TABLE}')")).await; + } + } + execute_sql(instance, &create_table_sql(RANGE_TABLE)).await; execute_sql( instance, @@ -1981,3 +2050,89 @@ async fn native_aggregate_complete_over_four_nonempty_mito_partitions() { check_fallback_case(&instance, &observer, case).await; } } + +#[tokio::test(flavor = "multi_thread")] +async fn native_histogram_quantile_over_the_cut_answers_what_the_complete_input_answers() { + common_telemetry::init_default_ut_logging(); + + let observer = Arc::new(NativePlanObserver::default()); + let plugins = Plugins::new(); + plugins.insert(Arc::clone(&observer) as PhysicalPlanWrapperRef); + let standalone = GreptimeDbStandaloneBuilder::new("native_histogram_over_cut") + .with_query_options(QueryOptions { + experimental_native_promql: true, + ..Default::default() + }) + .with_plugin(plugins) + .build() + .await; + let instance = Arc::clone(standalone.fe_instance()); + setup_fixture(&instance).await; + + // A histogram function above the cut is admitted because the merge has already folded every + // partition's `(..., le)` group into one complete series, so the buckets it reads are the ones + // the statement's input would have produced in one execution. That claim is checked here on the + // rows two real executions report, not on a plan: the bucket series of [`HIST_SERIES`] are + // spread over several files and a memtable range, and both routes compute real rates over them. + // The quantiles cover the two ends of the classic interpolation - 0.99's rank is above the last + // finite bucket's, where the answer is that bucket's upper bound rather than an interpolation, + // and 0.5 interpolates inside the first bucket - and the fraction reads the same buckets a third + // way. + let queries = [ + format!("histogram_quantile(0.99, sum by (le) (rate({HIST_TABLE}[5m])))"), + format!("histogram_quantile(0.5, sum by (le) (rate({HIST_TABLE}[5m])))"), + format!("histogram_fraction(0.0, 1.0, sum by (le) (rate({HIST_TABLE}[5m])))"), + ]; + for promql in &queries { + let query = tql_eval(promql); + let (legacy, legacy_capture) = run_query(&instance, promql, false, &observer).await; + assert!( + legacy_capture.fused.is_empty() && legacy_capture.finalize.is_empty(), + "`{query}`: the legacy run must not plan a native node, but it planned {} fused \ + producers and {} frontend merges", + legacy_capture.fused.len(), + legacy_capture.finalize.len() + ); + // Without this the comparison below would also pass over two empty answers, which says + // nothing about either route. + assert!( + !legacy.rows.is_empty(), + "`{query}`: the fixture must produce a value on the complete input, otherwise the two \ + routes would agree on nothing" + ); + + let mark = observer.mark(); + let (native, capture) = run_query(&instance, promql, true, &observer).await; + assert!( + !capture.finalize.is_empty(), + "`{query}`: the split must answer a histogram function above the cut, so the native run \ + must merge the state the scan partitions export; instead its plans held {} fused \ + producers and {} frontend merges - a fall back - and it answered {} rows against the \ + legacy run's {}", + capture.fused.len(), + capture.finalize.len(), + native.rows.len(), + legacy.rows.len() + ); + assert_same_columns(&native, &legacy, &query); + // The equivalence the admission rests on: the rows the same statement reports over the + // complete input, labels included. + assert_rows( + &native.rows, + &legacy.rows, + &format!("native `{query}` vs legacy"), + ); + + let plans = observer + .since(mark) + .iter() + .map(|plan| displayable(plan.as_ref()).indent(true).to_string()) + .collect::>(); + common_telemetry::info!( + "`{query}`\n native rows: {:#?}\n legacy rows: {:#?}\n plans:\n{}", + native.rows, + legacy.rows, + plans.join("\n") + ); + } +} diff --git a/tests-integration/src/tests/promql_test.rs b/tests-integration/src/tests/promql_test.rs index 5b9d8068a60..bc811782589 100644 --- a/tests-integration/src/tests/promql_test.rs +++ b/tests-integration/src/tests/promql_test.rs @@ -1770,17 +1770,30 @@ async fn f06_scalar_metric_table_lag_order_by_ts_desc(instance: Arc>>, Mutex>>, + Mutex>>, ); impl NativePlanObserver { fn clear(&self) { *self.0.lock().unwrap() = None; self.1.lock().unwrap().clear(); + *self.2.lock().unwrap() = None; + } + + /// The root of the first plan wrapped after the last `clear`, which is the frontend's own tree + /// for the statement: the region-side sub-plans are wrapped later, while it executes. + fn root_plan(&self) -> Arc { + self.2 + .lock() + .unwrap() + .clone() + .expect("a wrapped plan was captured") } fn take_raw_sources(&self) -> Vec> { @@ -1807,6 +1820,12 @@ impl PhysicalPlanWrapper for NativePlanObserver { origin: Arc, _ctx: session::context::QueryContextRef, ) -> Arc { + { + let mut root = self.2.lock().unwrap(); + if root.is_none() { + *root = Some(origin.clone()); + } + } let mut stack = vec![origin.clone()]; while let Some(node) = stack.pop() { if node.name() == "FusedSeriesExec" { @@ -3344,9 +3363,9 @@ async fn native_labelled_rows( DataType::Timestamp(..) => { time = Some(cast(column.as_ref(), &DataType::Int64).unwrap()); } - DataType::Float64 => { + DataType::Float64 | DataType::Int64 => { assert!(value.is_none(), "one value column: {schema:?}"); - value = Some(column.clone()); + value = Some(cast(column.as_ref(), &DataType::Float64).unwrap()); } _ => labels.push(( field.name().clone(), @@ -4568,6 +4587,581 @@ async fn native_promql_metric_logical_tables_are_isolated() { } } +/// Two occurrences of one metric logical table bind two scans of the same physical table, each +/// with its own offset. The pair is answered where that table's region runs, with no raw source +/// per input, and matches the existing plan row for row. +#[tokio::test(flavor = "multi_thread")] +async fn native_promql_metric_same_logical_table_offset_pair_matches_legacy() { + common_telemetry::init_default_ut_logging(); + let (standalone, observer) = native_test_instance(GreptimeDbStandaloneBuilder::new( + "native_promql_metric_offset_pair_e2e", + )) + .await; + let instance = standalone.fe_instance().clone(); + execute_all( + &instance, + r#"CREATE TABLE native_offset_phy ( + ts TIMESTAMP TIME INDEX, + val DOUBLE + ) ENGINE=metric WITH ("physical_metric_table" = "true"); + CREATE TABLE native_offset_a ( + host STRING, + ts TIMESTAMP TIME INDEX, + val DOUBLE, + PRIMARY KEY (host) + ) ENGINE=metric WITH ("on_physical_table" = 'native_offset_phy'); + INSERT INTO native_offset_a (host, ts, val) VALUES ('a', 0, 10), ('a', 60000, 12), ('a', 120000, 15);"#, + QueryContext::arc(), + ) + .await; + + // The left occurrence reads each step's sample and the offset one reads a minute earlier: + // 12 - 10 = 2, then 15 - 12 = 3. + let sql = "TQL EVAL (60, 120, '1m') native_offset_a - native_offset_a offset 1m"; + let expected = vec![ + (60_000, "a".to_string(), 2.0), + (120_000, "a".to_string(), 3.0), + ]; + + observer.clear(); + let legacy_rows = native_result_rows(native_tql_output(&instance, sql, false).await).await; + assert!( + observer.fused().is_none(), + "the existing plan must not hold a native node" + ); + assert_eq!(legacy_rows, expected, "legacy"); + + observer.clear(); + let native_rows = native_result_rows(native_tql_output(&instance, sql, true).await).await; + let plan = observer.take(); + assert_eq!(plan.name(), "FusedSeriesExec"); + assert_eq!(plan.output_partitioning().partition_count(), 1); + assert_eq!(plan.children().len(), 2); + assert!( + observer.take_raw_sources().is_empty(), + "one logical table keeps both scans region-local instead of one raw source per input" + ); + assert_eq!(native_rows, expected, "native"); + assert_eq!(native_rows, legacy_rows); +} + +/// What every name of the frontend-side merge a split aggregate runs in front of contains. +const FINALIZE_EXEC_NAME: &str = "AggregateFinalize"; + +/// Whether any node of `plan` is the merge a split aggregate runs in front of. +fn holds_split_merge(plan: &Arc) -> bool { + let mut stack = vec![plan.clone()]; + while let Some(node) = stack.pop() { + if node.name().contains(FINALIZE_EXEC_NAME) { + return true; + } + stack.extend(node.children().into_iter().cloned()); + } + false +} + +/// Two aggregates, one per metric logical table of one physical table, answered by one multi-input +/// native node and by the existing plan alike. +/// +/// The two logical tables are two metrics over one physical table, so both rewrites prepare a scan +/// of that same table and only the id each one restricts to tells them apart: one binding read for +/// both inputs would report one metric's sum where the query asks for a ratio. The pair is not a +/// split either - the node has more than one input, so it declares one partition and folds every +/// input's partitions in that one execution - and the plans the engines ran say so. +#[tokio::test(flavor = "multi_thread")] +async fn native_promql_metric_two_aggregate_pair_matches_legacy() { + common_telemetry::init_default_ut_logging(); + let (standalone, observer) = native_test_instance(GreptimeDbStandaloneBuilder::new( + "native_promql_metric_pair_e2e", + )) + .await; + let instance = standalone.fe_instance().clone(); + execute_all( + &instance, + r#"CREATE TABLE native_pair_phy ( + ts TIMESTAMP TIME INDEX, + val DOUBLE + ) ENGINE=metric WITH ("physical_metric_table" = "true"); + CREATE TABLE native_pair_demo ( + host STRING, + ts TIMESTAMP TIME INDEX, + val DOUBLE, + PRIMARY KEY (host) + ) ENGINE=metric WITH ("on_physical_table" = 'native_pair_phy'); + CREATE TABLE native_pair_other ( + host STRING, + ts TIMESTAMP TIME INDEX, + val DOUBLE, + PRIMARY KEY (host) + ) ENGINE=metric WITH ("on_physical_table" = 'native_pair_phy'); + INSERT INTO native_pair_demo (host, ts, val) VALUES ('a', 0, 2), ('b', 0, 4); + INSERT INTO native_pair_other (host, ts, val) VALUES ('a', 0, 8), ('b', 0, 8);"#, + QueryContext::arc(), + ) + .await; + + // `sum` drops the labels, so both sides are one label-less series and the division matches: + // (2 + 4) / (8 + 8) = 0.375. Grouped by host each source keeps its own host's ratio, so the + // two groups differ and a swapped input would not answer the same: a = 2 / 8, b = 4 / 8. + let grouped = |host: &str, value: f64| { + ( + 0, + std::collections::BTreeMap::from([("host".to_string(), host.to_string())]), + value, + ) + }; + for (query, expected) in [ + ( + "sum(native_pair_demo) / sum(native_pair_other)", + vec![(0, std::collections::BTreeMap::new(), 0.375)], + ), + ( + "sum by (host) (native_pair_demo) / sum by (host) (native_pair_other)", + vec![grouped("a", 0.25), grouped("b", 0.5)], + ), + ] { + let sql = format!("TQL EVAL (0, 0, '1s') {query}"); + + // The existing plan: the route is off, so no native node ran and the answer is the one the + // legacy planner derives from the fixture by hand. + observer.clear(); + let (legacy_columns, legacy_rows) = + native_labelled_rows(native_tql_output(&instance, &sql, false).await).await; + assert!( + observer.fused().is_none(), + "{query}: the existing plan must not hold a native node" + ); + assert!( + !legacy_rows.is_empty(), + "{query}: the comparison cannot pass vacuously" + ); + assert_eq!(legacy_rows, expected, "legacy: {query}"); + + // The native plan: one two-input node over both logical tables' scans, and no split. + observer.clear(); + let (native_columns, native_rows) = + native_labelled_rows(native_tql_output(&instance, &sql, true).await).await; + let plan = observer.take(); + assert_eq!(plan.name(), "FusedSeriesExec", "{query}"); + assert_eq!(plan.output_partitioning().partition_count(), 1, "{query}"); + assert_eq!(plan.children().len(), 2, "{query}"); + assert_eq!( + observer.take_raw_sources().len(), + 2, + "{query}: each logical table answers through its own raw source" + ); + assert!( + !holds_split_merge(&observer.root_plan()), + "{query}: a multi-input aggregate is one execution, not a split" + ); + + assert_eq!(native_rows, legacy_rows, "{query}: rows"); + assert_eq!(native_rows, expected, "{query}"); + + // The columns both routes report: the time index, one value column, and one column per + // label. The labels themselves are compared row by row above, and the time index is `ts` on + // both. The value column's *name* is not equal, and that is this route's existing naming for + // a computed value column rather than anything this shape introduces: the node names its + // fields after the tree's own root schema, while the existing planner names a computed column + // after its expression (see the comment above `legacy_output_columns` in + // `src/query/src/promql/planner/native.rs`; the same difference is pinned for a cross-table + // pair by `native_fused_cross_table_pair_reads_each_table_through_its_own_source`). So the + // shape is asserted, and the value column's name is pinned as the raw field it comes from. + assert_eq!( + native_columns.len(), + legacy_columns.len(), + "{query}: columns" + ); + assert!( + legacy_columns.iter().any(|column| column == "ts"), + "{query}" + ); + assert!( + native_columns.iter().any(|column| column == "ts"), + "{query}" + ); + assert!( + native_columns.iter().any(|column| column == "val"), + "{query}: {native_columns:?}" + ); + let labels = native_rows + .first() + .map(|(_, labels, _)| labels.keys().cloned().collect::>()) + .unwrap_or_default(); + for label in labels { + assert!( + native_columns.contains(&label) && legacy_columns.contains(&label), + "{query}: label {label} is a column on both routes" + ); + } + common_telemetry::info!( + "{query}: native columns {native_columns:?}, legacy columns {legacy_columns:?}, rows \ + {native_rows:?}, native plan:\n{}", + datafusion::physical_plan::displayable(observer.root_plan().as_ref()).indent(true) + ); + } +} + +/// A grouping by the metric name keeps series the schema would have no column for. +/// +/// `sum by (__name__) (a or b)` folds two metrics into two groups that differ only in their metric +/// name. The native aggregate keeps that name and folds by it, but the label set the node declares is +/// taken from the existing planner's columns, and `__name__` is no column of any scan - so the two +/// groups would be reported under one label set, as two rows a reader cannot tell apart. The number +/// of rows and their full label sets are what this asserts: the values alone would still look +/// plausible, and only the labels say the identity was lost. +#[tokio::test(flavor = "multi_thread")] +async fn native_promql_declines_a_grouping_by_the_metric_name() { + common_telemetry::init_default_ut_logging(); + let (standalone, observer) = native_test_instance(GreptimeDbStandaloneBuilder::new( + "native_promql_group_by_name", + )) + .await; + let instance = standalone.fe_instance().clone(); + execute_all( + &instance, + r#"CREATE TABLE native_name_phy ( + ts TIMESTAMP TIME INDEX, + val DOUBLE + ) ENGINE=metric WITH ("physical_metric_table" = "true"); + CREATE TABLE native_name_demo ( + host STRING, + ts TIMESTAMP TIME INDEX, + val DOUBLE, + PRIMARY KEY (host) + ) ENGINE=metric WITH ("on_physical_table" = 'native_name_phy'); + CREATE TABLE native_name_other ( + host STRING, + ts TIMESTAMP TIME INDEX, + val DOUBLE, + PRIMARY KEY (host) + ) ENGINE=metric WITH ("on_physical_table" = 'native_name_phy'); + INSERT INTO native_name_demo (host, ts, val) VALUES ('a', 0, 2), ('b', 0, 4); + INSERT INTO native_name_other (host, ts, val) VALUES ('a', 0, 8), ('b', 0, 8);"#, + QueryContext::arc(), + ) + .await; + + // The two sides carry different hosts, so `or` keeps both series and the grouping has two groups + // whose only distinguishing label is the metric name. + let query = r#"sum by (__name__) (native_name_demo{host="a"} or native_name_other{host="b"})"#; + let sql = format!("TQL EVAL (0, 0, '1s') {query}"); + let (_, legacy_rows) = + native_labelled_rows(native_tql_output(&instance, &sql, false).await).await; + assert!( + !legacy_rows.is_empty(), + "{query}: the existing plan must answer this, otherwise there is nothing to keep" + ); + + observer.clear(); + let (_, native_rows) = + native_labelled_rows(native_tql_output(&instance, &sql, true).await).await; + assert!( + observer.fused().is_none(), + "{query}: the route must decline a grouping by the metric name, whose groups its declared \ + schema cannot name" + ); + assert_eq!( + native_rows, legacy_rows, + "{query}: the full label sets of every row, not only the values" + ); +} + +/// Unsupported `count_values` groupings keep the legacy plan before labels are lost or execution +/// reaches a kernel refusal. The outer `sum` prevents a final-schema check from masking the latter. +#[tokio::test(flavor = "multi_thread")] +async fn native_promql_declines_count_values_groupings_the_node_cannot_honor() { + common_telemetry::init_default_ut_logging(); + let (standalone, observer) = native_test_instance(GreptimeDbStandaloneBuilder::new( + "native_promql_count_values_groups", + )) + .await; + let instance = standalone.fe_instance().clone(); + execute_all( + &instance, + r#"CREATE TABLE native_cv_phy ( + ts TIMESTAMP TIME INDEX, + val DOUBLE + ) ENGINE=metric WITH ("physical_metric_table" = "true"); + CREATE TABLE native_cv_demo ( + host STRING, + ts TIMESTAMP TIME INDEX, + val DOUBLE, + PRIMARY KEY (host) + ) ENGINE=metric WITH ("on_physical_table" = 'native_cv_phy'); + CREATE TABLE native_cv_other ( + host STRING, + ts TIMESTAMP TIME INDEX, + val DOUBLE, + PRIMARY KEY (host) + ) ENGINE=metric WITH ("on_physical_table" = 'native_cv_phy'); + INSERT INTO native_cv_demo (host, ts, val) VALUES ('a', 0, 1), ('b', 0, 1); + INSERT INTO native_cv_other (host, ts, val) VALUES ('a', 0, 1), ('b', 0, 1);"#, + QueryContext::arc(), + ) + .await; + + // `by (__name__)` names a label absent from the declared schema; the outer sum hides the + // unsupported `by (val)` grouping from the final schema. + let by_name = r#"count_values by (__name__) ("sample", native_cv_demo{host="a"} or native_cv_other{host="b"})"#; + let by_field = r#"sum(count_values by (val) ("sample", native_cv_demo{host="a"} or native_cv_other{host="b"}))"#; + for query in [by_name, by_field] { + let sql = format!("TQL EVAL (0, 0, '1s') {query}"); + let (_, legacy_rows) = + native_labelled_rows(native_tql_output(&instance, &sql, false).await).await; + assert!( + !legacy_rows.is_empty(), + "{query}: the existing plan must answer this, otherwise there is nothing to keep" + ); + + observer.clear(); + let (_, native_rows) = + native_labelled_rows(native_tql_output(&instance, &sql, true).await).await; + assert!( + observer.fused().is_none(), + "{query}: the route must decline a grouping its declared schema or kernel cannot honor" + ); + assert_eq!( + native_rows, legacy_rows, + "{query}: the full label sets of every row, not only the values" + ); + if query == by_field { + assert_eq!( + native_rows, + vec![(0, std::collections::BTreeMap::new(), 2.0)], + "the outer sum counts both input series" + ); + } + } +} + +/// `topk`/`bottomk` key their groups by series labels, while the existing plan also accepts a value +/// field there and forms one group per sample value. A `by()` therefore keeps every +/// group's winner on the existing plan and must not reach the native node, which would fold both +/// series into one group and return fewer rows. +#[tokio::test(flavor = "multi_thread")] +async fn native_promql_declines_ranking_grouped_by_a_value_field() { + common_telemetry::init_default_ut_logging(); + let (standalone, observer) = native_test_instance(GreptimeDbStandaloneBuilder::new( + "native_promql_ranking_value_field", + )) + .await; + let instance = standalone.fe_instance().clone(); + execute_all( + &instance, + r#"CREATE TABLE native_rank_phy ( + ts TIMESTAMP TIME INDEX, + val DOUBLE + ) ENGINE=metric WITH ("physical_metric_table" = "true"); + CREATE TABLE native_rank_demo ( + host STRING, + ts TIMESTAMP TIME INDEX, + val DOUBLE, + PRIMARY KEY (host) + ) ENGINE=metric WITH ("on_physical_table" = 'native_rank_phy'); + CREATE TABLE native_rank_other ( + host STRING, + ts TIMESTAMP TIME INDEX, + val DOUBLE, + PRIMARY KEY (host) + ) ENGINE=metric WITH ("on_physical_table" = 'native_rank_phy'); + INSERT INTO native_rank_demo (host, ts, val) VALUES ('x', 0, 2); + INSERT INTO native_rank_other (host, ts, val) VALUES ('y', 0, 8);"#, + QueryContext::arc(), + ) + .await; + + let pair = r#"native_rank_demo{host="x"} or native_rank_other{host="y"}"#; + // `by (val)` forms two groups of one sample each, so both series survive: the existing plan + // answers two rows where the native node would keep the single winner of one group. + for query in [ + format!("topk by (val) (1, {pair})"), + format!("bottomk by (val) (1, {pair})"), + ] { + let sql = format!("TQL EVAL (0, 0, '1s') {query}"); + observer.clear(); + let legacy_rows = native_result_rows(native_tql_output(&instance, &sql, false).await).await; + assert_eq!( + legacy_rows.len(), + 2, + "{query}: the existing plan keeps one row per value group" + ); + assert!( + legacy_rows + .iter() + .map(|(_, _, value)| *value) + .eq([2.0, 8.0]), + "{query}: {legacy_rows:?}" + ); + + observer.clear(); + let native_rows = native_result_rows(native_tql_output(&instance, &sql, true).await).await; + assert!( + observer.fused().is_none(), + "{query}: the route must decline a grouping its groups cannot honor" + ); + assert_eq!(native_rows, legacy_rows, "{query}: full rows"); + } + + // A label grouping is what the native node does honor, so it still runs natively and keeps every + // group's winner. + let sql = format!("TQL EVAL (0, 0, '1s') topk by (host) (1, {pair})"); + observer.clear(); + let native_rows = native_result_rows(native_tql_output(&instance, &sql, true).await).await; + assert_eq!(observer.take().name(), "FusedSeriesExec"); + let legacy_rows = native_result_rows(native_tql_output(&instance, &sql, false).await).await; + assert_eq!(native_rows, legacy_rows); + assert_eq!(native_rows.len(), 2); +} + +/// `limitk` keeps the series that arrive first, so a multi-input node must not answer it. +/// +/// A node with more than one input hands its inputs' partitions over in the order they became ready, +/// while `limitk` keeps the first `k` series of each group in input order: the schedule would decide +/// which series a query keeps, where Prometheus defines the selection over a series set. The +/// aggregate split declines it for the same reason, and that refusal is repeated for a multi-input +/// node. +/// +/// The existing planner cannot plan `limitk` at all - it has no kernel for the operator - so the +/// statement fails on both routes today. What this pins is that the native route does not change +/// that outcome by answering it, and in particular does not answer it differently between runs. +#[tokio::test(flavor = "multi_thread")] +async fn native_promql_declines_a_selection_that_depends_on_arrival_order() { + common_telemetry::init_default_ut_logging(); + let (standalone, observer) = native_test_instance(GreptimeDbStandaloneBuilder::new( + "native_promql_limitk_order", + )) + .await; + let instance = standalone.fe_instance().clone(); + execute_all( + &instance, + r#"CREATE TABLE native_order_phy ( + ts TIMESTAMP TIME INDEX, + val DOUBLE + ) ENGINE=metric WITH ("physical_metric_table" = "true"); + CREATE TABLE native_order_left ( + host STRING, + ts TIMESTAMP TIME INDEX, + val DOUBLE, + PRIMARY KEY (host) + ) ENGINE=metric WITH ("on_physical_table" = 'native_order_phy'); + CREATE TABLE native_order_right ( + host STRING, + ts TIMESTAMP TIME INDEX, + val DOUBLE, + PRIMARY KEY (host) + ) ENGINE=metric WITH ("on_physical_table" = 'native_order_phy'); + INSERT INTO native_order_left (host, ts, val) VALUES ('a', 0, 2), ('b', 0, 4); + INSERT INTO native_order_right (host, ts, val) VALUES ('a', 0, 8), ('b', 0, 8);"#, + QueryContext::arc(), + ) + .await; + + let query = "native_order_left and on(host) limitk(1, native_order_right)"; + let sql = format!("TQL EVAL (0, 0, '1s') {query}"); + + // The route declines, so the statement is planned by the one that was always its answer. + observer.clear(); + let native = instance + .do_query(&sql, native_query_context(true, 4)) + .await + .remove(0); + assert!( + observer.fused().is_none(), + "{query}: the route must decline a selection whose answer depends on arrival order" + ); + let legacy = instance + .do_query(&sql, native_query_context(false, 4)) + .await + .remove(0); + // Both routes report the same outcome: the decline leads to the plan this statement already had, + // not to an answer the schedule could change. + assert_eq!( + native.is_err(), + legacy.is_err(), + "{query}: the route must not change whether the statement is answerable" + ); + if let (Err(native_error), Err(legacy_error)) = (native, legacy) { + assert_eq!( + native_error.to_string(), + legacy_error.to_string(), + "{query}: the decline must lead to the plan this statement already had" + ); + } +} + +/// A tree the native runtime cannot execute keeps the existing plan, rather than failing the +/// statement. +/// +/// The multi-input route skips the aggregate split, and the split is what refuses a tree the runtime +/// cannot run. The native node is chosen while planning the logical plan, so a refusal the physical +/// layer raises later is an error the statement reports, not a fall back - which would turn a query +/// the existing plan answers into a failure. `quantile` is outside the eight complete kinds the +/// runtime computes, and a grouping by a value field needs a per-point adapter it does not have, so +/// both are checked here on the shape that would otherwise reach the node: two aggregates, one per +/// logical table of one physical table, which is the multi-input case. +#[tokio::test(flavor = "multi_thread")] +async fn native_promql_declines_a_tree_the_runtime_cannot_execute() { + common_telemetry::init_default_ut_logging(); + let (standalone, observer) = native_test_instance(GreptimeDbStandaloneBuilder::new( + "native_promql_unexecutable_tree", + )) + .await; + let instance = standalone.fe_instance().clone(); + execute_all( + &instance, + r#"CREATE TABLE native_unexec_phy ( + ts TIMESTAMP TIME INDEX, + val DOUBLE + ) ENGINE=metric WITH ("physical_metric_table" = "true"); + CREATE TABLE native_unexec_demo ( + host STRING, + ts TIMESTAMP TIME INDEX, + val DOUBLE, + PRIMARY KEY (host) + ) ENGINE=metric WITH ("on_physical_table" = 'native_unexec_phy'); + CREATE TABLE native_unexec_other ( + host STRING, + ts TIMESTAMP TIME INDEX, + val DOUBLE, + PRIMARY KEY (host) + ) ENGINE=metric WITH ("on_physical_table" = 'native_unexec_phy'); + INSERT INTO native_unexec_demo (host, ts, val) VALUES ('a', 0, 2), ('b', 0, 4); + INSERT INTO native_unexec_other (host, ts, val) VALUES ('a', 0, 8), ('b', 0, 8);"#, + QueryContext::arc(), + ) + .await; + + for query in [ + // `quantile` computes a value the eight complete kinds do not, so the runtime refuses it. + "quantile(0.5, native_unexec_demo) / sum(native_unexec_other)", + // The outer `count` clears the labels the inner grouping writes, so no column check stops + // this one: the refusal is the grouping by a value field, which the runtime has no adapter + // for. + "count(count by (val) (native_unexec_demo)) / sum(native_unexec_other)", + ] { + let sql = format!("TQL EVAL (0, 0, '1s') {query}"); + let (_, legacy_rows) = + native_labelled_rows(native_tql_output(&instance, &sql, false).await).await; + assert!( + !legacy_rows.is_empty(), + "{query}: the existing plan must answer this, otherwise there is nothing to keep" + ); + + observer.clear(); + let (_, native_rows) = + native_labelled_rows(native_tql_output(&instance, &sql, true).await).await; + assert_eq!( + native_rows, legacy_rows, + "{query}: a tree the runtime cannot execute must keep the existing plan" + ); + assert!( + observer.fused().is_none(), + "{query}: the route must decline it instead of planning a node the runtime refuses" + ); + } +} + /// OTLP delta temporality on the public TQL route: a `delta`-marked series answers the raw sample /// sum (10 + 10 = 20) while a cumulative series with the same samples answers 0. The fused node /// computes one cumulative window for every series, so it must decline a marker-bearing table