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 {