From 23afa9697ebbce7f7fe2de978292503535a39404 Mon Sep 17 00:00:00 2001 From: discord9 Date: Fri, 6 Nov 2026 14:53:33 +0800 Subject: [PATCH] feat(query): place a native label-set merge above the regions (#10191) * feat(query): place the canonical merge above the regions A canonical producer merges the pieces of one label set that its own input partitions hold, and the canonical router FusedSeries::to_physical_plan inserts above it collects them from exactly those partitions. In process the producer reads the region's own scans, so the region holds every piece its router has to see and that placement is the only one it has. Over the wire the producer reads RawSeriesSource children instead - the wrapper that reconciles fragments several regions hold. Classifying that shape commutative shipped producer -> router as the region sub-plan, one producer and one router per region, each collecting only its own region's pieces, so a label set a time partition spread over several regions was answered from one region's share. Classify a canonical producer by its own contract only: commutative when every child is a bare scan, non-commutative otherwise, and never fall through to the per-series rules, which describe what a node computes and say nothing about where the pieces of one label set are. The analyzer then cuts at the raw source, so the frontend holds consumer -> SeriesRepartitionExec(CanonicalLabels) -> producer -> [raw router] -> MergeScan(RawSeriesSource(TableScan)) and the canonical router collects each label set from the producers of every region, while the fragments of each series are still reconciled below the producer by the raw-primary-key router the interleaved raw input asks for. Admission follows the placement: the in-process route still declines a canonical merge over a partitioned table, because a region-local producer is the only placement it has, and the remote route now admits it. Acceptance is a real two-datanode test over a three-partition metric table: the rows equal an unpartitioned control's, the producer and its canonical router are frontend nodes above the merge scans, and a collision two regions hold reports the Prometheus duplicate-labelset error that no single region can see. Signed-off-by: discord9 * test(promql): partition the canonical acceptance table on a tag GreptimeDB has no implicit time partitioning: a table has more than one region only from a declared partition, and metric tables are normally partitioned on a tag. The acceptance fixture partitioned on the time index instead, which made the test read as if time partitioning were the everyday case. Partition on `host` in three ranges - `early`/`first`/`late`/`second` still land in different regions, so the merge pair and the collision pair keep their cross-region geometry - and rename the test accordingly. A partition on a primary-key column that excludes the time index also proves the regions hold disjoint series, so the two assertions that described fragmented series state what this fixture actually proves instead: the merge scan publishes whole series (`contiguous`), and no raw-key repartition sits below the frontend producer because only the label set crosses regions. Signed-off-by: discord9 * refactor(query): ask the transport where a canonical merge runs `canonical_boundary_region_node` inferred "this execution reads the whole table" from the node's children: a canonical producer whose children were all bare scans was admitted inside a region sub-plan. Over the wire every selector input is a `RawSeriesSource`, so the inference happened to agree with the rule that matters, but a wire plan that ever held a bare scan under a producer would have been admitted silently and answered a label set from one region's share. The analyzer already holds the fact. `in_process_region_transport` is what it was built with, and it is exactly the condition under which a region read is the whole table. Pass it down to `Categorizer::check_plan` and `check_extension_plan` and require it instead of inspecting child shapes. Guard the hole the inference hid: a canonical producer over a bare table scan classifies `Commutative` in process and `NonCommutative` on the wire. Signed-off-by: discord9 * test(promql): cover a fragmented canonical merge too A partition on a primary-key column proves the regions hold disjoint series, so the tag-partitioned fixture could not exercise the other geometry the frontend merge has to read: fragments. In a metric table a tag is always a primary-key column and a field must be Float64, so the only declared partition that proves nothing about where a series lives is the time index. Run the same two fixture rows over both geometries in one loop. The tag case still publishes whole series and has no router below the producer; the time case declares interleaved and asserts exactly one RawPrimaryKey reconciling router between the merge scans and the frontend producer. Both keep the unpartitioned control's rows, the frontend placement of the producer and its canonical router, and the cross-region duplicate-labelset error. Signed-off-by: discord9 * Revert "test(promql): cover a fragmented canonical merge too" This reverts commit 8ff9f430313034cddfda6d698e8542ab0481ba9e. Signed-off-by: discord9 --------- Signed-off-by: discord9 --- src/query/src/dist_plan/analyzer.rs | 21 +- src/query/src/dist_plan/commutativity.rs | 161 +++++++++--- src/query/src/promql/planner/native.rs | 32 ++- src/query/src/promql/planner/test.rs | 2 + tests-integration/src/tests/promql_test.rs | 283 +++++++++++++++++++++ 5 files changed, 455 insertions(+), 44 deletions(-) diff --git a/src/query/src/dist_plan/analyzer.rs b/src/query/src/dist_plan/analyzer.rs index e641579e12d..3882a6b78c7 100644 --- a/src/query/src/dist_plan/analyzer.rs +++ b/src/query/src/dist_plan/analyzer.rs @@ -446,6 +446,14 @@ enum RewriterStatus { #[derive(Debug, Default)] struct PlanRewriter { + /// Whether every region this pass reads is reached in process. + /// + /// [`Categorizer`] needs it to place canonical producers: their router collects the pieces of + /// one label set from the partitions the producer runs on, so the producer belongs inside a + /// region only when region reads are served in process. It is a fact about the transport, not + /// about the plan, and every check this pass makes is a placement decision for the same + /// transport. + in_process_region_transport: bool, /// Whether the whole plan encodes, allowing per-node checks to be skipped. whole_plan_encodable: bool, /// Whether an in-process root fused plan was recognized for placement-gate evaluation. @@ -638,6 +646,7 @@ impl PlanRewriter { .encode(plan, DefaultSerializer) .is_ok(); Self { + in_process_region_transport, in_process_fused_root, whole_plan_encodable, ..Default::default() @@ -688,7 +697,11 @@ impl PlanRewriter { } if self.expand_on_next_part_cond_trans_commutative { - let comm = Categorizer::check_plan(plan, self.partition_cols.clone())?; + let comm = Categorizer::check_plan( + plan, + self.partition_cols.clone(), + self.in_process_region_transport, + )?; match comm { Commutativity::PartialCommutative => { // a small difference is that for partial commutative, we still need to @@ -713,7 +726,11 @@ impl PlanRewriter { } } - match Categorizer::check_plan(plan, self.partition_cols.clone())? { + match Categorizer::check_plan( + plan, + self.partition_cols.clone(), + self.in_process_region_transport, + )? { Commutativity::Commutative => { // PATCH: we should reconsider SORT's commutativity instead of doing this trick. // explain: for a fully commutative SeriesDivide, its child Sort plan only serves it. I.e., that diff --git a/src/query/src/dist_plan/commutativity.rs b/src/query/src/dist_plan/commutativity.rs index 4d727bae0e2..4c9db11affd 100644 --- a/src/query/src/dist_plan/commutativity.rs +++ b/src/query/src/dist_plan/commutativity.rs @@ -32,28 +32,29 @@ use crate::dist_plan::analyzer::AliasMapping; use crate::dist_plan::merge_sort::{MergeSortLogicalPlan, merge_sort_transformer}; use crate::promql::planner::native::{independent_scan_pair, pointwise_operand, same_table_scans}; -/// Whether `fused` is a canonical producer whose prefix and sources can run in one region. -/// Its physical canonical router assembles the complete label sets; ordinary consumers above it -/// do not carry the producer marker and retain their normal placement requirements. -fn canonical_boundary_region_node(fused: &FusedSeries) -> bool { +/// Whether `fused` is a canonical producer that may run inside one region. +/// +/// A canonical producer merges the pieces of one label set that its own input partitions hold, and +/// its physical canonical router collects those pieces from exactly those partitions. It may only +/// run where those partitions cover the whole table. In process that is the region's own read: the +/// region holds every piece its router has to see and the node belongs there. Ordinary consumers +/// above it do not carry the producer marker and retain their normal placement requirements. +/// +/// Over the wire every selector input is a [`RawSeriesSource`] answered per region, so no region +/// holds every piece of a label set. A producer placed below the raw sources would run once per +/// region and its router would collect the pieces of the region it runs in and no others, which +/// answers a label set from one region's share whenever the pieces are spread. Declining this +/// classification is what keeps that from happening: the analyzer expands each raw source into a +/// merge scan and the producer, with the router it inserts above itself, stays on the frontend over +/// every region. +fn canonical_boundary_region_node(fused: &FusedSeries, in_process_region_transport: bool) -> bool { if !fused.is_canonical_producer() { return false; } - fused.format() == FusedOutputFormat::SeriesFrame + in_process_region_transport + && fused.format() == FusedOutputFormat::SeriesFrame && fused.canonical_producer_prefix_is_local() && !fused.children().is_empty() - && fused.children().iter().all(|child| match child { - LogicalPlan::TableScan(_) => true, - LogicalPlan::Extension(extension) - if extension.node.as_any().is::() => - { - matches!( - extension.node.inputs().as_slice(), - [LogicalPlan::TableScan(_)] - ) - } - _ => false, - }) } pub struct StepTransformAction { @@ -121,6 +122,7 @@ impl Categorizer { pub fn check_plan( plan: &LogicalPlan, partition_cols: Option, + in_process_region_transport: bool, ) -> DfResult { // Subquery is treated separately in `inspect_plan_with_subquery`. To avoid rewrite the // "maybe rewritten" plan, stop the check here. @@ -214,9 +216,11 @@ impl Categorizer { Commutativity::Unimplemented } } - LogicalPlan::Extension(extension) => { - Self::check_extension_plan(extension.node.as_ref() as _, &partition_cols) - } + LogicalPlan::Extension(extension) => Self::check_extension_plan( + extension.node.as_ref() as _, + &partition_cols, + in_process_region_transport, + ), LogicalPlan::Distinct(_) => { if partition_cols.is_empty() { Commutativity::Commutative @@ -242,6 +246,7 @@ impl Categorizer { pub fn check_extension_plan( plan: &dyn UserDefinedLogicalNode, partition_cols: &AliasMapping, + in_process_region_transport: bool, ) -> Commutativity { match plan.name() { name if name == SeriesDivide::name() => { @@ -269,8 +274,17 @@ impl Categorizer { } name if name == FusedSeries::name() => { let fused = plan.as_any().downcast_ref::().unwrap(); - if canonical_boundary_region_node(fused) { - return Commutativity::Commutative; + if fused.is_canonical_producer() { + // A canonical producer is placed by its own contract and not by the shape below + // it: either the region holds every piece its router collects, or the node stays + // above the raw sources with its router. It must not fall through to the + // per-series rules below, which describe what a node computes and say nothing + // about where the pieces of one label set are. + return if canonical_boundary_region_node(fused, in_process_region_transport) { + Commutativity::Commutative + } else { + Commutativity::NonCommutative + }; } // The raw-series reader supports one scan, or the narrow same-table multi-scan // one-to-one case. This classification is for region-local execution only. @@ -450,7 +464,7 @@ mod tests { }; use promql::native::plan::{ AggregateKind, AggregateStage, AggregateStateSchema, BoundAggregate, BoundInput, Grouping, - NativeExpr, NativeNode, NativeType, + NativeExpr, NativeNode, NativeTree, NativeType, }; use promql_parser::parser; use store_api::storage::RegionId; @@ -476,7 +490,8 @@ mod tests { BTreeSet::from([Column::from_name("some_partition_col")]), )]); - let commutativity = Categorizer::check_extension_plan(&series_divide, &partition_cols); + let commutativity = + Categorizer::check_extension_plan(&series_divide, &partition_cols, false); assert!(matches!(commutativity, Commutativity::Commutative)); } @@ -617,6 +632,52 @@ mod tests { }) } + /// A canonical producer of `some_metric` over `input`. + /// + /// The tree says what the compiler says at a canonical boundary - one evaluated vector merged by + /// canonical label set - and the `Merge` root is what marks the node as the producer of that + /// collection. The children are exactly what the caller supplies, so a test can hand the node + /// the region's own scan and ask where such a node may run. + fn canonical_producer_plan(input: LogicalPlan) -> LogicalPlan { + let stmt = native_statement(); + let grid = statement_grid(&stmt).unwrap(); + let evaluated = NativeType::Samples(Arc::new(SeriesSchema { + fields: Arc::from([FieldSpec { name: "v0".into() }]), + eval_grid: Some(Arc::clone(&grid)), + })); + let merge = NativeNode { + expr: NativeExpr::Merge { + input: Box::new(NativeNode { + expr: NativeExpr::Input(BoundInput { + output: evaluated.clone(), + source_index: 0, + ordering: SeriesOrdering::KeyThenTime, + }), + output: evaluated.clone(), + }), + }, + output: evaluated, + }; + let binding = FusedSeriesBinding { + grid, + time_index: "ts".to_string(), + tags: Vec::new(), + metric_name: Some(Arc::from("some_metric")), + }; + let tree = NativeTree::try_new(merge).unwrap(); + LogicalPlan::Extension(Extension { + node: Arc::new( + FusedSeries::try_new_canonical_producer( + Arc::from("some_metric"), + tree, + binding, + vec![input], + ) + .unwrap(), + ), + }) + } + /// The region's own view of the metric's table: a scan whose provider is the region provider. fn table_scan_plan() -> LogicalPlan { let provider = mock_table_provider(RegionId::new(1, 1)); @@ -786,7 +847,7 @@ mod tests { let plan = fused_series_plan(table_scan_plan()); assert!(matches!( - Categorizer::check_plan(&plan, None).unwrap(), + Categorizer::check_plan(&plan, None, true).unwrap(), Commutativity::Commutative )); } @@ -804,7 +865,7 @@ mod tests { let plan = fused_series_plan(input); assert!(matches!( - Categorizer::check_plan(&plan, None).unwrap(), + Categorizer::check_plan(&plan, None, true).unwrap(), Commutativity::NonCommutative )); } @@ -816,7 +877,35 @@ mod tests { let plan = fused_series_plan_named(table_scan_plan(), None); assert!(matches!( - Categorizer::check_plan(&plan, None).unwrap(), + Categorizer::check_plan(&plan, None, true).unwrap(), + Commutativity::NonCommutative + )); + } + + /// The transport decides where a canonical producer runs, not the shape of its children: its + /// router collects the pieces of one label set from exactly the partitions the producer runs + /// on, so those partitions have to be the region's whole read - which is only true in process. + /// A bare scan underneath must not on its own admit the producer to a wire plan. + #[test] + fn a_canonical_producer_is_not_commutative_on_the_wire_even_over_a_bare_scan() { + let plan = canonical_producer_plan(table_scan_plan()); + let LogicalPlan::Extension(extension) = &plan else { + unreachable!(); + }; + assert!(matches!( + Categorizer::check_extension_plan( + extension.node.as_ref(), + &AliasMapping::default(), + true + ), + Commutativity::Commutative + )); + assert!(matches!( + Categorizer::check_extension_plan( + extension.node.as_ref(), + &AliasMapping::default(), + false + ), Commutativity::NonCommutative )); } @@ -878,7 +967,11 @@ mod tests { unreachable!(); }; assert!(matches!( - Categorizer::check_extension_plan(extension.node.as_ref(), &AliasMapping::default()), + Categorizer::check_extension_plan( + extension.node.as_ref(), + &AliasMapping::default(), + true + ), Commutativity::Commutative )); @@ -897,7 +990,11 @@ mod tests { unreachable!(); }; assert!(matches!( - Categorizer::check_extension_plan(extension.node.as_ref(), &AliasMapping::default()), + Categorizer::check_extension_plan( + extension.node.as_ref(), + &AliasMapping::default(), + true + ), Commutativity::NonCommutative )); } @@ -1167,7 +1264,7 @@ mod tests { fn a_split_aggregate_is_commutative_only_below_the_cut() { let plan = split_aggregate_plan(table_scan_plan()); assert!(matches!( - Categorizer::check_plan(&plan, None).unwrap(), + Categorizer::check_plan(&plan, None, true).unwrap(), Commutativity::NonCommutative )); @@ -1176,7 +1273,7 @@ mod tests { panic!("expected the final node to have one partial child: {plan}"); }; assert!(matches!( - Categorizer::check_plan(partial, None).unwrap(), + Categorizer::check_plan(partial, None, true).unwrap(), Commutativity::Commutative )); } @@ -1363,7 +1460,7 @@ mod test { fetch: None, }); assert!(matches!( - Categorizer::check_plan(&plan, Some(Default::default())).unwrap(), + Categorizer::check_plan(&plan, Some(Default::default()), false).unwrap(), Commutativity::Commutative )); } diff --git a/src/query/src/promql/planner/native.rs b/src/query/src/promql/planner/native.rs index dd6791493f0..67aaa6e9188 100644 --- a/src/query/src/promql/planner/native.rs +++ b/src/query/src/promql/planner/native.rs @@ -180,8 +180,8 @@ type DiscoveredSelectorMetadata = (Arc, String, Vec, Arc, @@ -368,9 +368,10 @@ impl PromPlanner { }; let table = self.table_from_source(&scan.source)?; let table_meta = &table.table_info().meta; - // A table split along a partition column lives in more than one region. The canonical - // boundary's router only collects the pieces its own region's producer holds, so the caller - // needs this fact to tell a whole-query label-set answer from one region's share of it. + // A table split along a partition column lives in more than one region. In process the + // canonical boundary's router only collects the pieces its own region's producer holds, so + // the caller needs this fact to tell a whole-query label-set answer from one region's share + // of it. let partitioned = table_meta.partition_column_names().next().is_some(); // An engine without a raw-series scanner has no frame to give this node, and by the // time the physical planner asks, the logical plan is already this node: the refusal @@ -1124,11 +1125,22 @@ impl PromPlanner { // region-local producer only sees the partitions of the region it runs in. A table split // along a partition column is planned region by region, so each region would answer a // canonical label set from the piece it holds while the consumer above has no collection of - // the other regions' pieces. Without a proof that this query reads one region, decline the - // route instead of claiming a global canonical answer; the existing planner reads every - // region and reports what Prometheus reports. A tree with no canonical merge claims no such - // answer, so it is untouched: a per-series query on a partitioned table keeps fusing. - if any_partitioned && contains_canonical_merge(root) { + // the other regions' pieces. + // + // In process the producer reads the region's own scans, so that placement is the only one it + // has: without a proof that this query reads one region, decline the route instead of + // claiming a global canonical answer, and let the existing planner - which reads every + // region and reports what Prometheus reports - answer. A tree with no canonical merge claims + // no such answer, so it is untouched: a per-series query on a partitioned table keeps fusing. + // + // Over the wire the producer reads `RawSeriesSource` children, and + // `Categorizer::check_extension_plan` keeps such a producer out of a region + // (`canonical_boundary_region_node` admits only a region's own scans). The analyzer expands + // the raw sources into merge scans and the producer runs on the frontend above every region, + // with its canonical router above all of them, so each label set is collected from the + // producers of every region. That placement is what the remote route claims, and the shape it + // builds is the one the acceptance test asserts: no canonical producer below a merge scan. + if !self_contained && any_partitioned && contains_canonical_merge(root) { return Ok(None); } // The kind the tree's root answers with, which every output column follows. It is not the diff --git a/src/query/src/promql/planner/test.rs b/src/query/src/promql/planner/test.rs index 72ce3ba97d3..b8158c28d34 100644 --- a/src/query/src/promql/planner/test.rs +++ b/src/query/src/promql/planner/test.rs @@ -3989,6 +3989,8 @@ async fn native_fused_info_over_its_own_table_reads_two_scans_in_the_region() { crate::dist_plan::Categorizer::check_extension_plan( fused as &dyn datafusion::logical_expr::UserDefinedLogicalNode, &Default::default(), + // The node reads two scans of the region's own table, which is the in-process read. + true, ), crate::dist_plan::Commutativity::Commutative )); diff --git a/tests-integration/src/tests/promql_test.rs b/tests-integration/src/tests/promql_test.rs index 19df9b06139..21a57a9dfc2 100644 --- a/tests-integration/src/tests/promql_test.rs +++ b/tests-integration/src/tests/promql_test.rs @@ -3030,6 +3030,289 @@ async fn distributed_native_promql_repeated_selector_over_partitioned_table() { } } +/// The rows a collapsing rewrite joins into one series without a collision: the two stored series +/// are ten minutes apart, twice the five-minute lookback, so exactly one of them holds a value at +/// each evaluation point and the label set they share carries both points. +fn canonical_merge_fixture_inserts(prefix: &str) -> String { + format!( + "INSERT INTO {prefix}_b (host, val, ts) VALUES ('first', 10, 0), ('second', 20, 600000)" + ) +} + +/// The rows a collapsing rewrite can only reject: the second series is two minutes after the first, +/// inside the lookback, so both hold a value at the same evaluation point while the declared partition +/// column stores the two hosts under values that place them in different regions. +fn canonical_collision_fixture_inserts(prefix: &str) -> String { + format!("INSERT INTO {prefix}_c (host, val, ts) VALUES ('early', 10, 0), ('late', 20, 120000)") +} + +/// Real two-datanode TCP acceptance for a canonical merge over a tag-partitioned metric table. +/// +/// A rewrite that writes a label its input carries can give two stored series one label set. The two +/// series are stored under different values of the declared partition column, so that partition puts +/// the pieces of the one label set in different regions: no region holds the set whole. A +/// region-local producer merges only the pieces its own region holds, so the boundary has to be +/// placed above the regions: the frontend runs the producer over the reconciled raw sources, and the +/// canonical router it inserts above itself collects each label set from every region. +/// +/// The test states that in the three ways it can be observed: the values (against the same rows in +/// one unpartitioned table), the plan (the producer is a frontend node reading a merge scan, with its +/// canonical router above it), and the error (a collision two different regions hold is reported, +/// which no single region could see). +#[tokio::test(flavor = "multi_thread")] +async fn distributed_native_promql_canonical_merge_over_partitioned_table() { + common_telemetry::init_default_ut_logging(); + let cluster = GreptimeDbClusterBuilder::new("distributed_native_promql_canonical") + .await + .with_datanodes(2) + .with_real_datanode_grpc_addr(true) + .with_frontend_real_datanode_grpc(true) + .with_query_options(QueryOptions { + experimental_native_promql: true, + ..Default::default() + }) + .build(false) + .await; + let instance = cluster.fe_instance().clone(); + execute_all( + &instance, + "CREATE TABLE dist_native_canon_phy (host STRING, val DOUBLE, ts TIMESTAMP TIME INDEX, PRIMARY KEY(host)) PARTITION ON COLUMNS (host) (host < 'first', host >= 'first' AND host < 'second', host >= 'second') ENGINE=metric WITH ('physical_metric_table'='true')", + QueryContext::arc(), + ) + .await; + for table in ["dist_native_canon_b", "dist_native_canon_c"] { + execute_all( + &instance, + &format!( + "CREATE TABLE {table} (host STRING, val DOUBLE, ts TIMESTAMP TIME INDEX, PRIMARY KEY(host)) ENGINE=metric WITH ('on_physical_table'='dist_native_canon_phy')" + ), + QueryContext::arc(), + ) + .await; + } + execute_all( + &instance, + &canonical_merge_fixture_inserts("dist_native_canon"), + QueryContext::arc(), + ) + .await; + execute_all( + &instance, + &canonical_collision_fixture_inserts("dist_native_canon"), + QueryContext::arc(), + ) + .await; + + // The semantic reference: the same rows in one unpartitioned physical table, where one region + // holds every piece of a label set. Its evaluation is what the partitioned plan has to reproduce. + execute_all( + &instance, + "CREATE TABLE dist_native_canon_ctl_phy (host STRING, val DOUBLE, ts TIMESTAMP TIME INDEX, PRIMARY KEY(host)) ENGINE=metric WITH ('physical_metric_table'='true')", + QueryContext::arc(), + ) + .await; + for table in ["dist_native_canon_ctl_b", "dist_native_canon_ctl_c"] { + execute_all( + &instance, + &format!( + "CREATE TABLE {table} (host STRING, val DOUBLE, ts TIMESTAMP TIME INDEX, PRIMARY KEY(host)) ENGINE=metric WITH ('on_physical_table'='dist_native_canon_ctl_phy')" + ), + QueryContext::arc(), + ) + .await; + } + execute_all( + &instance, + &canonical_merge_fixture_inserts("dist_native_canon_ctl"), + QueryContext::arc(), + ) + .await; + execute_all( + &instance, + &canonical_collision_fixture_inserts("dist_native_canon_ctl"), + QueryContext::arc(), + ) + .await; + + // Two stored series rewritten to one label set, holding their values at different evaluation + // points, are one series carrying both points - and the hosts they are stored under land in + // different host ranges, so the two pieces are in different regions. + let expression = r#"label_replace(dist_native_canon_b, "host", "same", "host", ".*")"#; + let query = PromQuery { + query: expression.to_string(), + start: "0".to_string(), + end: "600".to_string(), + step: "600s".to_string(), + ..PromQuery::default() + }; + let expected = vec![ + (0, "same".to_string(), 10.0), + (600_000, "same".to_string(), 20.0), + ]; + let control = expression.replace("dist_native_canon_", "dist_native_canon_ctl_"); + for (route, native) in [("native", true), ("existing", false)] { + let control_rows = native_result_rows( + instance + .do_promql_query( + &PromQuery { + query: control.clone(), + ..query.clone() + }, + native_query_context(native, 2), + ) + .await + .remove(0) + .unwrap(), + ) + .await; + assert_eq!( + control_rows, expected, + "the control over one region defines every row: {route}: {control}" + ); + } + + let output = instance + .do_promql_query(&query, native_query_context(true, 2)) + .await + .remove(0) + .unwrap(); + let plan = output.meta.plan.as_ref().expect("frontend physical plan"); + let shape = native_plan_shape(plan); + assert_eq!( + shape.regions_per_scan, + vec![3], + "the selector reads all three host ranges: {expression}: {plan:?}" + ); + // Declaring a partition on a primary-key column that is not the time index proves the regions + // hold disjoint series, so the merge scan publishes them whole and the boundary above the + // regions only ever sees complete stored series. + assert!( + shape + .scan_orderings + .iter() + .all(|ordering| ordering.as_deref() == Some("contiguous")), + "a declared partition on a primary-key column that excludes the time index proves the \ + regions hold disjoint series (physical_regions_disjoint), so the merge scan publishes \ + whole series, not fragments: {expression}: {:?}: {plan:?}", + shape.scan_orderings + ); + // The boundary's two halves are both frontend nodes: the producer merges what each execution + // partition holds and the consumer above it publishes the routed frames. A producer planned + // inside a region would leave this tree with one fused node and no canonical producer in it. + assert_eq!( + shape.fused, 2, + "the canonical producer and the consumer above it run on the frontend: {expression}: \ + {plan:?}" + ); + let producer = native_plan_nodes(plan) + .into_iter() + .find(|node| promql::extension_plan::is_canonical_producer_exec(node.as_ref())) + .unwrap_or_else(|| { + panic!( + "the canonical producer must be a frontend node above the regions, not a region's \ + own producer: {expression}: {plan:?}" + ) + }); + // What the producer reads is every region's own stored series, each already whole: one merge + // scan over all three regions, with no raw-key repartition below it. + let below_producer = native_plan_nodes(&producer); + assert!( + below_producer + .iter() + .any(|node| node.downcast_ref::().is_some()), + "the producer reads the regions through a merge scan: {expression}: {plan:?}" + ); + assert!( + below_producer + .iter() + .all(|node| node.name() != "SeriesRepartitionExec"), + "no raw-key repartition sits below the producer: each input partition already holds whole \ + stored series, since the partition covers the primary key and only the label set crosses \ + regions: {expression}: {plan:?}" + ); + // The router is the boundary's middle stage: it sits above the producer and collects each label + // set from every partition the producer ran on, which is what a region-local boundary cannot do. + let router = native_plan_nodes(plan) + .into_iter() + .find(|node| { + node.name() == "SeriesRepartitionExec" + && node + .children() + .iter() + .any(|child| promql::extension_plan::is_canonical_producer_exec(child.as_ref())) + }) + .unwrap_or_else(|| { + panic!( + "the canonical router must read the producer above the regions: {expression}: \ + {plan:?}" + ) + }); + assert_eq!( + router.children()[0].name(), + "FusedSeriesExec", + "the router reads the canonical producer: {expression}: {plan:?}" + ); + assert_eq!( + router.properties().output_partitioning().partition_count(), + producer + .properties() + .output_partitioning() + .partition_count(), + "the router keeps the producer's width instead of gathering every piece: {expression}: \ + {plan:?}" + ); + let rows = native_result_rows(output).await; + assert_eq!( + rows, expected, + "the partitioned plan answers the control's rows: {expression}" + ); + + // A collision two regions hold is the error Prometheus reports for two metrics with one label + // set. A region-local producer cannot see it: each region would hold one piece and publish it, + // and the answer would be two series under one label set. The control over one region reports it + // as well, which is the answer the partitioned plan owes. + let collision = r#"label_replace(dist_native_canon_c, "host", "same", "host", ".*")"#; + let collision_query = PromQuery { + query: collision.to_string(), + start: "120".to_string(), + end: "120".to_string(), + step: "60s".to_string(), + ..PromQuery::default() + }; + let collision_control = collision.replace("dist_native_canon_", "dist_native_canon_ctl_"); + let control_error = instance + .do_promql_query( + &PromQuery { + query: collision_control.clone(), + ..collision_query.clone() + }, + native_query_context(true, 2), + ) + .await + .remove(0); + let control_error = match control_error { + Ok(output) => collect_error(output).await, + Err(error) => error.to_string(), + }; + assert!( + control_error.contains("cannot contain metrics with the same labelset"), + "the control over one region reports the overlap: {collision_control}: {control_error}" + ); + let partitioned = instance + .do_promql_query(&collision_query, native_query_context(true, 2)) + .await + .remove(0); + let partitioned_error = match partitioned { + Ok(output) => collect_error(output).await, + Err(error) => error.to_string(), + }; + assert!( + partitioned_error.contains("cannot contain metrics with the same labelset"), + "a label set whose pieces two regions hold is collected by the router and rejected, not \ + published twice: {collision}: {partitioned_error}" + ); +} + /// The rows of the disjoint fixture, inserted into the logical tables `{prefix}_a` and /// `{prefix}_b`. Both hosts carry the same shape and land in different partitions. fn disjoint_fixture_inserts(prefix: &str) -> String {