diff --git a/src/promql/src/native/plan.rs b/src/promql/src/native/plan.rs index fd308c6d9fb..2d6ee4090cc 100644 --- a/src/promql/src/native/plan.rs +++ b/src/promql/src/native/plan.rs @@ -347,6 +347,28 @@ pub struct NativeNode { pub output: NativeType, } +/// Whether `node` reads the whole input vector at a point rather than one series at a time. +/// +/// Three kinds do: an `Aggregate` folds every series of its group, a `Binary` pairs two vectors' +/// series by label, and a `Vector` operation (`sort`, `absent`, `topk`, `limitk`, +/// `histogram_quantile`, ...) reorders or reads the whole vector at a grid point. Everything else - +/// an input, a scalar, a selection, a window, a map - answers one series from one series' rows. +/// +/// Both a region-side placement and the native route's admission ask this, and they must agree: a +/// tree admitted here but refused there becomes a node the physical layer cannot build, and one +/// refused here but admitted there would be pushed to a region that holds part of the metric. +pub fn reads_across_series(node: &NativeNode) -> bool { + match &node.expr { + NativeExpr::Input(_) | NativeExpr::Scalar(_) => false, + NativeExpr::Instant { input, .. } + | NativeExpr::Window { input, .. } + | NativeExpr::Map { input, .. } => reads_across_series(input), + NativeExpr::Aggregate { .. } | NativeExpr::Binary { .. } | NativeExpr::Vector { .. } => { + true + } + } +} + /// A bound evaluation tree. /// /// The tree itself carries no run-time state: every execution builds its own, so one compiled plan diff --git a/src/query/src/dist_plan/commutativity.rs b/src/query/src/dist_plan/commutativity.rs index ce8466cd68f..17114e5dc55 100644 --- a/src/query/src/dist_plan/commutativity.rs +++ b/src/query/src/dist_plan/commutativity.rs @@ -23,7 +23,7 @@ use datafusion_expr::{Expr, LogicalPlan, UserDefinedLogicalNode}; use promql::extension_plan::{ EmptyMetric, FusedSeries, InstantManipulate, RangeManipulate, SeriesDivide, SeriesNormalize, }; -use promql::native::plan::{NativeExpr, NativeNode}; +use promql::native::plan::reads_across_series; use store_api::metric_engine_consts::DATA_SCHEMA_TSID_COLUMN_NAME; use crate::dist_plan::MergeScanLogicalPlan; @@ -393,18 +393,6 @@ impl Categorizer { /// several series at once. Neither has a compiler producer yet, which is why refusing them costs /// nothing today: a tree built some other way is exactly what this check is for, because evaluating /// it one region at a time would fold a partial input and answer a wrong query silently. -fn reads_across_series(node: &NativeNode) -> bool { - match &node.expr { - NativeExpr::Input(_) | NativeExpr::Scalar(_) => false, - NativeExpr::Instant { input, .. } - | NativeExpr::Window { input, .. } - | NativeExpr::Map { input, .. } => reads_across_series(input), - NativeExpr::Aggregate { .. } | NativeExpr::Binary { .. } => true, - // A vector operation reads the whole vector at each grid point (`topk`, `absent`, - // `histogram_quantile`, ...), so it is not a per-series node either. - NativeExpr::Vector { .. } => true, - } -} #[cfg(test)] mod tests { @@ -431,7 +419,7 @@ mod tests { }; use promql::native::plan::{ AggregateKind, AggregateStage, AggregateStateSchema, BoundAggregate, BoundInput, Grouping, - NativeType, + NativeExpr, NativeNode, NativeType, }; use promql_parser::parser; use store_api::storage::RegionId; diff --git a/src/query/src/promql/planner/native.rs b/src/query/src/promql/planner/native.rs index 88e06e33f7a..6dabf979649 100644 --- a/src/query/src/promql/planner/native.rs +++ b/src/query/src/promql/planner/native.rs @@ -30,7 +30,7 @@ use std::sync::Arc; use std::time::{Duration, UNIX_EPOCH}; -use common_catalog::consts::{METRIC_ENGINE, MITO_ENGINE, MITO2_ENGINE}; +use common_catalog::consts::{MITO_ENGINE, MITO2_ENGINE}; use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, TimeUnit as ArrowTimeUnit}; use datafusion::common::{DFSchema, DFSchemaRef}; use datafusion::error::DataFusionError; @@ -38,6 +38,7 @@ use datafusion::logical_expr::{Extension, LogicalPlan}; use datatypes::series::SeriesOrdering; use datatypes::series::samples::{FieldSpec, SeriesSchema}; use promql::extension_plan::{FusedSeries, FusedSeriesBinding}; +use promql::native::plan::reads_across_series; use promql_parser::parser::value::ValueType; use promql_parser::parser::{ EvalStmt, Expr as PromExpr, MatrixSelector, ParenExpr, VectorSelector, @@ -387,12 +388,15 @@ fn selector_range_millis(expr: &PromExpr) -> Option { /// A whitelist, not a probe: the logical planner cannot ask a region what its scanner supports /// without reading the table, and a false negative only costs the query its native route. /// -/// Mito's series scanner implements the raw path. The metric engine is named too because a logical -/// region reads through its physical region and hands back that same scanner - its own raw-series -/// test is `test_raw_series_scan_uses_logical_metric_route_and_bound_metadata` - and the ordinary -/// metric rewrite puts a filter and an alias between the scan and this node, which the bare-scan -/// check above already refuses. A logical table reaches that check with a bare scan only when the -/// physical table cannot be resolved, and there the logical scan is served by the same scanner. +/// Mito's series scanner implements the raw path. +/// +/// The metric engine is deliberately absent. A logical metric table does read through its physical +/// region, which is a mito scanner, but the planner reaches this node with a bare scan only when it +/// could not resolve that physical table: the ordinary rewrite puts a `__table_id` filter and an +/// alias between the scan and this node, which the bare-scan check above already refuses. A logical +/// table whose physical table is missing is not a table this route can serve - its `TableInfo` +/// carries no `append_mode` either, so the preflight's own check could not be made here - and +/// naming the engine would admit exactly that unverifiable case. /// /// Every other engine in the workspace leaves `raw_series_schema` at its default `None`: a /// file-engine table, for one, is read by a `SinglePartitionScanner` that cannot frame a series. @@ -401,26 +405,7 @@ fn selector_range_millis(expr: &PromExpr) -> Option { /// /// [`supports_pk_dictionary_encoding`]: table::table::scan fn supports_native_raw_series(engine: &str) -> bool { - matches!(engine, MITO_ENGINE | MITO2_ENGINE | METRIC_ENGINE) -} - -/// Whether any node of `tree` reads the whole input vector at a point rather than one series. -/// -/// Such a node folds or pairs across series (`Aggregate`, `Binary`) or reorders/reads a whole -/// vector (`Vector`: `sort`, `absent`, `topk`, `limitk`, `histogram_quantile`, ...). This node runs -/// one executor per scan partition, and a partition holds only part of a metric's series, so those -/// nodes would answer from a fragment. -fn reads_across_series(node: &promql::native::plan::NativeNode) -> bool { - use promql::native::plan::NativeExpr; - match &node.expr { - NativeExpr::Input(_) | NativeExpr::Scalar(_) => false, - NativeExpr::Instant { input, .. } - | NativeExpr::Window { input, .. } - | NativeExpr::Map { input, .. } => reads_across_series(input), - NativeExpr::Aggregate { .. } | NativeExpr::Binary { .. } | NativeExpr::Vector { .. } => { - true - } - } + matches!(engine, MITO_ENGINE | MITO2_ENGINE) } fn single_selector(expr: &PromExpr) -> Option<&VectorSelector> { diff --git a/src/query/src/promql/planner/test.rs b/src/query/src/promql/planner/test.rs index 7eddd3b81db..01af409fca3 100644 --- a/src/query/src/promql/planner/test.rs +++ b/src/query/src/promql/planner/test.rs @@ -1881,14 +1881,15 @@ async fn native_fused_tag_partitioned_table_still_plans_fused_series() { } /// The flag on and a shape the compiler refuses: the existing planner plans it unchanged. -/// A metric-engine logical table is served by its physical region's scanner, so it is admitted. +/// A metric-engine logical table keeps the existing plan. /// -/// The ordinary metric rewrite puts a filter and an alias between the scan and this node, which the -/// bare-scan check refuses - that is why such a query usually falls back. This test pins the one -/// shape that reaches the node bare (the physical table could not be resolved), so that the engine -/// whitelist does not refuse a table whose region really does serve frames. +/// A logical table reads through its physical region, which is a mito scanner, but the planner +/// only reaches this node with a bare scan when that physical table could not be resolved: the +/// ordinary rewrite puts a `__table_id` filter and an alias in between. Such a table's `TableInfo` +/// also carries no `append_mode`, so the preflight's own check could not be made here. Both are +/// reasons the engine is not on the route's whitelist, and this test pins that it falls back. #[tokio::test] -async fn native_fused_metric_engine_bare_scan_is_admitted() { +async fn native_fused_metric_engine_table_falls_back() { let native = build_native_fused_plan_with_provider( "rate(some_metric[5m])", true, @@ -1905,10 +1906,30 @@ async fn native_fused_metric_engine_bare_scan_is_admitted() { .await, ) .await; + let without_flag = build_native_fused_plan_with_provider( + "rate(some_metric[5m])", + false, + build_test_table_provider_with_options( + &[(DEFAULT_SCHEMA_NAME.to_string(), "some_metric".to_string())], + 1, + 1, + vec![], + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ConcreteDataType::float64_datatype(), + METRIC_ENGINE, + ) + .await, + ) + .await; + assert!( + find_fused_series(&native).is_none(), + "a metric-engine table must not reach the native route" + ); assert_eq!( - find_fused_series(&native).map(|fused| fused.promql().to_string()), - Some("rate(some_metric[5m])".to_string()), - "a metric-engine table serves frames, so a bare scan over one keeps the native route" + native.display_indent().to_string(), + without_flag.display_indent().to_string(), + "a metric-engine table must keep the existing plan" ); }