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<SeriesSchema>,
-    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<Vec<Series>> {
+) -> 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<SeriesSchema>,
+    input: &[Series],
+    label: &str,
+    grouping: Grouping<'_>,
+) -> Result<Vec<Series>> {
+    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<SeriesSchema>) -> Self {
-        Self {
+    /// Validates schema-dependent requirements before the native route is selected.
+    fn try_new(op: BoundVectorOp, schema: Arc<SeriesSchema>) -> Result<Self> {
+        // 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::<str>::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::<str>::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::<str>::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::<str>::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<f64>)>;
 
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<PreparedInput> = 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<u32> = 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::<std::result::Result<Vec<_>, 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::<Vec<_>>();
-                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::<Vec<_>>(),
             // 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<String>, Vec<String>)> {
+        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<Item = &'a LogicalPl
 /// by `on` or `ignoring`; `and` and `unless` keep the left side's series and `or` adds the right
 /// side's; `info` adds the data labels of the info table; an operation with a scalar or a function
 /// of each value keeps its operand's; a scalar and `vector(s)` have none. `None` for a node whose
-/// labels this walk does not derive - an aggregate's groups, or a whole-vector operation - so a tree
-/// holding one is declined rather than given its input's labels.
+/// labels this walk does not derive - an aggregate's groups, a whole-vector operation, or a label
+/// rewrite, which renames the labels below it - so a tree holding one reports no label set here.
+/// `None` is not the query's answer: the caller asks the existing planner for the columns the node is
+/// named after instead of declining the query.
 fn fused_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Option<Vec<String>> {
     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<String>]) -> Option<V
         }
         NativeExpr::Instant { input, .. }
         | NativeExpr::Window { input, .. }
-        | NativeExpr::Map { input, .. }
-        | NativeExpr::ParameterizedMap { input, .. }
         | NativeExpr::Timestamp { input } => 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 = <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<String> {
+    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::<RawSeriesSource>()
+                .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<Instance>) {
         }
     }
 
+    // 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::<Vec<_>>();
+        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<dyn MockInst
     }
 }
 
-/// The fused node of the last wrapped plan that held one, and every region-side raw source.
+/// The fused node of the last wrapped plan that held one, every region-side raw source, and the
+/// root of the last wrapped plan.
 #[derive(Default)]
 struct NativePlanObserver(
     Mutex<Option<Arc<dyn datafusion::physical_plan::ExecutionPlan>>>,
     Mutex<Vec<Arc<dyn datafusion::physical_plan::ExecutionPlan>>>,
+    Mutex<Option<Arc<dyn datafusion::physical_plan::ExecutionPlan>>>,
 );
 
 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<dyn datafusion::physical_plan::ExecutionPlan> {
+        self.2
+            .lock()
+            .unwrap()
+            .clone()
+            .expect("a wrapped plan was captured")
     }
 
     fn take_raw_sources(&self) -> Vec<Arc<dyn datafusion::physical_plan::ExecutionPlan>> {
@@ -1807,6 +1820,12 @@ impl PhysicalPlanWrapper for NativePlanObserver {
         origin: Arc<dyn datafusion::physical_plan::ExecutionPlan>,
         _ctx: session::context::QueryContextRef,
     ) -> Arc<dyn datafusion::physical_plan::ExecutionPlan> {
+        {
+            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<dyn datafusion::physical_plan::ExecutionPlan>) -> 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::<Vec<_>>())
+            .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(<value field>)` 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