refactor(promql): keep one cross-series predicate for both callers

Signed-off-by: discord9 <discord9@163.com>

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

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;

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<i64> {
 /// 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<i64> {
 ///
 /// [`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> {

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"
     );
 }