diff --git a/src/query/src/promql/planner.rs b/src/query/src/promql/planner.rs index 6cd5138d14b..a23c5004616 100644 --- a/src/query/src/promql/planner.rs +++ b/src/query/src/promql/planner.rs @@ -22,8 +22,6 @@ pub use native::{NATIVE_PROMQL_EXTENSION_KEY, discovers_metric_name}; mod set_operator; use std::collections::{BTreeSet, HashSet, VecDeque}; -use std::future::Future; -use std::pin::Pin; use std::sync::Arc; use arrow::datatypes::IntervalDayTime; @@ -310,11 +308,6 @@ pub struct PromPlanner { type BinaryFieldPair<'a> = (&'a String, &'a String); -/// [`PromPlanner::try_plan_native_operand`]'s future, boxed; see -/// [`PromPlanner::native_operand_future`] for why it is not kept inline. -type NativeOperandFuture<'a> = - Pin>> + Send + 'a>>; - impl PromPlanner { pub async fn stmt_to_plan( table_provider: DfTableSourceProvider, @@ -1177,63 +1170,6 @@ impl PromPlanner { }) } - /// Boxes [`Self::try_plan_native_operand`]'s future. - /// - /// The planner re-enters [`Self::prom_expr_to_plan_inner`] and - /// [`Self::prom_binary_expr_to_plan`] once per nesting level of the statement, and both frames - /// are large in a debug build. The native route's future is large as well, so a caller that - /// also plans the same expression with the existing path would carry a copy of it on the stack - /// at every level. Built here, the future moves into the box before that recursion starts, and - /// only the attempt itself pays for it. - fn native_operand_future<'a>( - &'a mut self, - expr: &'a PromExpr, - query_engine_state: &'a QueryEngineState, - ) -> NativeOperandFuture<'a> { - Box::pin(self.try_plan_native_operand(expr, query_engine_state)) - } - - /// Re-plans one operand that [`matching_filters::propagate`] rewrote, publishing the plan and - /// its context through `context`. - /// - /// A copied matcher belongs to the scan, not to the operand's identity: `absent()` turns - /// `selector_matcher` into the labels it reports, so the re-planned context keeps the matchers - /// the operand was written with. - /// - /// A natively planned operand is re-planned natively: the matcher has to reach the fused node's - /// own scan, or the join keeps the non-joining rows that node reported. The route decides per - /// expression, so a decline falls back to the existing plan exactly as it would have on its own. - async fn replan_rewritten_operand( - &mut self, - rewritten: &PromExpr, - used_native: bool, - query_engine_state: &QueryEngineState, - context: &mut PromPlannerContext, - ) -> Result { - let selectors = std::mem::take(&mut context.selector_matcher); - let native = if used_native { - self.native_operand_future(rewritten, query_engine_state) - .await? - } else { - None - }; - let plan = match native { - Some((plan, replanned)) => { - *context = replanned; - plan - } - None => { - let plan = self - .prom_expr_to_plan(rewritten, query_engine_state) - .await?; - *context = self.ctx.clone(); - plan - } - }; - context.selector_matcher = selectors; - Ok(plan) - } - async fn prom_binary_expr_to_plan( &mut self, query_engine_state: &QueryEngineState, @@ -1483,32 +1419,39 @@ impl PromPlanner { // Both operands are planned first because only the planned contexts tell a tag // from a value field. The rewrite merely adds matchers to a selector, so the // table reference, time index and field columns captured above stay valid. - if let Some(rewritten) = matching_filters::propagate( - binary_expr, - &left_context.tag_columns, - &right_context.tag_columns, - &left_context.aggregation_field_labels, - &right_context.aggregation_field_labels, - ) { + // Matcher propagation belongs to the legacy route: copying a matcher into the + // other operand's scan drops the groups that have no partner, which is exactly + // what lets the legacy join answer where Prometheus would report a duplicated + // match group. A native operand must not be rewritten this way - the native + // matcher checks the complete one-side groups itself, and filtering them first + // would hide that error. So the rewrite only runs when neither operand is native. + if !used_native + && let Some(rewritten) = matching_filters::propagate( + binary_expr, + &left_context.tag_columns, + &right_context.tag_columns, + &left_context.aggregation_field_labels, + &right_context.aggregation_field_labels, + ) + { + // A copied matcher belongs to the scan, not to the operand's identity: + // `absent()` turns `selector_matcher` into the labels it reports, so the + // re-planned context keeps the matchers the operand was written with. if rewritten.lhs.as_ref() != lhs.as_ref() { + let selectors = std::mem::take(&mut left_context.selector_matcher); left_input = self - .replan_rewritten_operand( - &rewritten.lhs, - used_native, - query_engine_state, - &mut left_context, - ) + .prom_expr_to_plan(&rewritten.lhs, query_engine_state) .await?; + left_context = self.ctx.clone(); + left_context.selector_matcher = selectors; } if rewritten.rhs.as_ref() != rhs.as_ref() { + let selectors = std::mem::take(&mut right_context.selector_matcher); right_input = self - .replan_rewritten_operand( - &rewritten.rhs, - used_native, - query_engine_state, - &mut right_context, - ) + .prom_expr_to_plan(&rewritten.rhs, query_engine_state) .await?; + right_context = self.ctx.clone(); + right_context.selector_matcher = selectors; } // The code below reads `self.ctx` as the right operand's context. self.ctx = right_context.clone(); diff --git a/src/query/src/promql/planner/native.rs b/src/query/src/promql/planner/native.rs index 68382fff99c..5791e39d607 100644 --- a/src/query/src/promql/planner/native.rs +++ b/src/query/src/promql/planner/native.rs @@ -864,17 +864,11 @@ impl PromPlanner { .map(|(scan, ..)| scan) .filter(|scan| !is_empty_discovered_input(scan)), ) || logical_ids_differ; - // A multi-selector tree over a partitioned table has no in-process placement whose matching - // and uniqueness semantics equal the frontend route. One node over several scans of one - // partitioned table reads neither the other scans nor the fragments another region holds, - // and one node over several tables' sources matches every series those sources delivered, - // where the frontend route copies the operands' matchers into their scans first and checks - // uniqueness above the join. Keep the existing plan in process; the hybrid route lifts such - // a tree one operand at a time. Over the wire every occurrence is already its own raw source - // reconciled above the regions, so the route keeps those trees. - if selectors.len() > 1 && !self_contained && prepared.iter().any(|input| input.5) { - return Ok(None); - } + // A multi-selector tree over a partitioned table has no region-local placement: one node + // over several scans reads neither the other scans nor the fragments another region holds. + // It does have a frontend placement - each occurrence reads the table through its own raw + // source and the node matches above every region's contribution - so the tree is admitted + // on both routes and nothing is refused here. // The first input names the output's time index and, on one table, binds every scan. let Some((_, _, metric, time, tags, _, _)) = prepared.first() else { return Ok(None); @@ -1054,6 +1048,43 @@ impl PromPlanner { // region-local producer cannot see the pieces another region holds, so this fact decides // whether that boundary may be claimed at all. let any_partitioned = prepared.iter().any(|input| input.5); + // Admission and placement are separate: `independent_scan_pair` admits nested native + // matches, while the physical locality classification decides whether their inputs need + // gathering or segmentation. + // A label rewrite can give two series one label set, which every consumer that answers a + // whole label set has to see as one series. The pass puts a [`NativeExpr::Merge`] where that + // matters and nowhere else, so a tree with no collapsing rewrite keeps every partition. + let tree = match promql::native::canonicalize::merge_label_rewrites( + &compiled.tree, + &source_tags, + ) { + Ok(tree) => tree, + Err(_) => return Ok(None), + }; + let root = tree.root(); + // The placement facts the split and the segmentation below are chosen from, and the same + // facts that say whether one node can run over the scans at all. + let per_partition = + !reads_across_series(root) || pointwise_operand(root) || independent_scan_pair(root); + let locality = input_locality(root); + let computed_key_case = + matches!(locality, InputLocality::SinglePartition) && matching_operands_drop_key(root); + let required_placement = !per_partition || computed_key_case; + // The node reads more than one input when the tree has more than one selector occurrence: + // the compiler binds one input per occurrence. What it needs from them is that each is + // bound to its own source - a raw source, in process or over the wire - or, where a region + // can run the node itself, one bare scan per occurrence. Either way the consumer above them + // reads every occurrence's contribution, so the metric-id fact this used to turn on is not + // what admission turns on. + let multi_input = selectors.len() > 1; + // A region runs one node over its own scans only when the table it reads is not split into + // regions and the tree's root is per-partition work: a partitioned table's region reads + // neither the other scans nor the fragments another region holds, and a whole-vector + // consumer has no region-local placement. Such a tree runs above the regions, and then each + // occurrence reads through its own raw source. + let per_source_inputs = cross_table + || self_contained + || (multi_input && (any_partitioned || required_placement)); // A segment carries one source's identity. Keeping each occurrence's scan and metadata // aligned here lets the cutting step name that source independently of selector count. // Read the split-series fact before `prepared` is consumed by the child mapping below. @@ -1090,10 +1121,12 @@ impl PromPlanner { // A scan stays bare only where a region runs the node over it in process. Over the // wire every input is a raw source, whichever table it reads: the source is what the // transport carries, and it is also what reconciles the fragments of a series that - // several regions hold. - // A partition that splits one series across regions has to be read through a raw - // source even in process: a bare scan would hand a region only its own fragment. - if (!cross_table && !self_contained && !split_series) || already_sourced { + // several regions hold. A partition that splits one series across regions has to + // be read through a raw source even in process: a bare scan would hand a region only + // its own fragment. A tree the regions cannot run as one node reads its occurrences + // through their own sources too, so the consumer above matches every region's + // contribution instead of the rows of one. + if (!per_source_inputs && !split_series) || already_sourced { return Ok(scan); } let Some(metric) = metric else { @@ -1175,49 +1208,27 @@ impl PromPlanner { } Some(FieldKind::Float) | None => FieldKind::Float, }; - // 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 a node whose locality is `SinglePartition` normalizes every input to one - // partition, 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 tree over more than one input needs no split: each input already answers from its own + // source, and the one node above them folds every partition in one execution. The split is + // an optimization for the one-input shapes it can carry, and so is the segmentation below; + // the node over the inputs is what admits the rest. // - // 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; - // The aggregate split folds float state, and the multi-input node reads several scans, so - // a histogram keeps the existing plan for both rather than running a tree whose output the - // runtime could not finish. The kind entering each aggregate is what decides, not the + // The aggregate split folds float state, and a node reading several sources runs over + // frames it reconciled rather than over its own scans, so a histogram keeps the existing + // plan for both rather than running a tree whose output the runtime could not finish. A + // node over bare scans is not that shape: a region runs it over each scan itself, exactly + // as it does for one selector. The kind entering each aggregate is what decides, not the // shape of the root: `sum(h) * 2` has a binary root and would otherwise hide the aggregate // under it. if let Some(kind @ FieldKind::Histogram) = statement_kind - && (multi_input || aggregate_reads_histogram(root, kind)) + && ((multi_input && per_source_inputs) || aggregate_reads_histogram(root, kind)) { return Ok(None); } - // 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 || (self_contained && children.len() > 1)) && depends_on_input_order(root) { - return Ok(None); - } - // Two sources that report groups by metric name need a column for that name: without one, - // groups that differ only by metric would be reported as the same label set. A discovered - // selector carries that column itself; several explicit selectors do not. - if (multi_input || (self_contained && children.len() > 1)) && groups_by_metric_name(root) { + // The split keeps its prefix per-series, which is why it declines an operation whose + // answer depends on the order its input arrives in. A node over more than one input reads + // its inputs' partitions in the order they became ready, so that refusal holds there too. + if multi_input && depends_on_input_order(root) { return Ok(None); } // The native output schema can name only compiled value fields, the time index and the tag @@ -1246,8 +1257,7 @@ impl PromPlanner { // series frames, and a canonical router collects the pieces of one label set from every // partition. Declining here would hand the statement back to a path that answers it per // partition. - if !multi_input - && selectors.len() == 1 + if selectors.len() == 1 && matches!(&tree.root().expr, NativeExpr::Merge { input } if pointwise_operand(input) && canonical_local_prefix(input) && !contains_global_consumer(input)) && let Some(plan) = canonical_boundary_plan( prom_expr, @@ -1263,13 +1273,6 @@ impl PromPlanner { { return Ok(Some((plan, single_input_context))); } - let root = tree.root(); - let per_partition = - !reads_across_series(root) || pointwise_operand(root) || independent_scan_pair(root); - let locality = input_locality(root); - let computed_key_case = - matches!(locality, InputLocality::SinglePartition) && matching_operands_drop_key(root); - let required_placement = !per_partition || computed_key_case; if required_placement { // A tree whose fold can be split keeps that plan: it exports aggregate state rather // than whole samples, so it moves less between the two halves than a sample boundary @@ -1279,7 +1282,7 @@ impl PromPlanner { // without this route. Its producer reads whatever input the statement prepared: over // the wire that is the raw series source, and the partial/final halves stay on the // frontend above the regions, so per-series work keeps the scan width. - let split = if !multi_input && selectors.len() == 1 { + let split = if selectors.len() == 1 { FusedSeries::split_aggregate(&tree, &time).context(DataFusionPlanningSnafu)? } else { None @@ -1319,11 +1322,6 @@ impl PromPlanner { }); return Ok(Some((plan, single_input_context))); } - // Preserve the established refusal for repeated same-table operands unless - // segmentation is needed to preserve a computed matching key. - if selectors.len() != 1 && !cross_table && !computed_key_case { - return Ok(None); - } } // A segment over the wire keeps the raw source as its input rather than a bare scan: the // segment's prefixes run on the frontend above the regions, so the raw series source is @@ -1344,22 +1342,27 @@ impl PromPlanner { FusedOutputFormat::PlainRows, contains_canonical_merge(root), root_kind, - // A split-series table must keep the raw source under the cut: a bare-scan segment - // would evaluate a window over one region's fragment on the frontend. - self_contained || any_split_series, + // A cut must keep the source whenever the table behind it can split a series + // across regions: a per-region prefix would evaluate a window over one region's + // fragment, so `rate` never sees the samples the other region holds. That is a + // partitioned table whose occurrences are source-backed, and always a partition on + // a column that is not a tag. An unpartitioned table's scans are whole in one + // region, so its cuts keep the bare-scan form. + self_contained || any_split_series || (per_source_inputs && any_partitioned), )) .await?; if let Some(plan) = plan { return Ok(Some((plan, single_input_context))); } - // A raw scan alone does not prove a region contains all contributions required by a - // cross-series consumer. Without a split or segmentation, keep that consumer out of - // the region rather than treating its local gather as a global answer. + // One input with no split or segment above it leaves a raw scan doing the work of a + // cross-series consumer, and a single scan does not prove its region holds every + // contribution the consumer needs. Keep that consumer on the existing plan. A tree over + // more than one input is not this shape: every occurrence reaches it through its own + // source, so the node itself is the collection. if required_placement && !multi_input { return Ok(None); } } - // Multi-input fused execution already reads every input in this execution. // 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. If it cannot derive them, this route declines. @@ -1387,14 +1390,12 @@ impl PromPlanner { return Ok(None); } let output_schema = fused_output_schema(&root_fields, &output_tags, &time, root_kind)?; - // 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. The same holds over the wire, - // where every input is a raw source. - let (tags, metric) = if cross_table || self_contained { + // The children a node reads bind it: a `RawSeriesSource` child carries the frames and names + // the metric it reads, so a node over them binds neither tags nor one metric name of its + // own. A node whose children stay direct scans of one table reads that table through the + // one metric and tag set they share, and keeps this binding, which is the same identity + // those scans carry. The same holds over the wire, where every input is a raw source. + let (tags, metric) = if per_source_inputs { (Vec::new(), None) } else { (tags, metric) diff --git a/src/query/src/promql/planner/test.rs b/src/query/src/promql/planner/test.rs index cb0cce7d198..435bd957bb5 100644 --- a/src/query/src/promql/planner/test.rs +++ b/src/query/src/promql/planner/test.rs @@ -3787,37 +3787,18 @@ async fn native_fused_same_table_nested_match_keeps_prior_admission() { ); } -/// On the wire, partitioned cross-table operands are independently transported by raw sources: -/// the pair is one fused node whose two inputs each name the metric they read. In process the same -/// pair keeps the frontend hybrid route instead, because only that route copies the matcher's -/// labels into each operand's scan and checks uniqueness above the join. Repeated scans of one -/// partitioned table retain the region-local multi-scan refusal on both routes. +/// Partitioned operands are independently transported by raw sources: the pair is one native +/// binary whose two inputs each name the metric they read. Repeated selectors of one partitioned +/// table read that table once per occurrence too, so no region is handed a node over bare scans. #[tokio::test] async fn native_fused_partitioned_pair_requires_per_selector_raw_sources() { - let state = QueryEngineState::try_new_with_region_transport( - new_memory_catalog_manager().unwrap(), - None, - None, - None, - None, - None, + let plan = build_native_fused_plan_with_provider( + "some_metric / other_metric", true, - false, - Plugins::default(), - QueryOptions::default(), - ) - .unwrap(); - let eval_stmt = build_eval_stmt("some_metric / other_metric"); - let plan = PromPlanner::stmt_to_plan_with_annotations( native_two_table_provider(vec![0]).await, - &eval_stmt, - &state, - None, - true, ) - .await - .unwrap(); - let fused = find_fused_series(&plan).expect("cross-table pair is native on the wire"); + .await; + let fused = find_fused_series(&plan).expect("cross-table pair is native"); assert_eq!(fused.inputs().len(), 2, "{plan}"); for input in fused.inputs() { assert!( @@ -3826,26 +3807,25 @@ async fn native_fused_partitioned_pair_requires_per_selector_raw_sources() { "{plan}" ); } - // The same pair in process must not become one two-input node: the frontend hybrid route is - // what the matcher propagation and the uniqueness check live in, and that is the plan the - // partitioned-RHS hybrid cases assert. - let in_process = build_native_fused_plan_with_provider( - "some_metric / other_metric", - true, - native_two_table_provider(vec![0]).await, - ) - .await; - assert!( - !has_pair_node(&in_process), - "an in-process partitioned pair keeps the hybrid route: {in_process}" - ); + // Repeated selectors of one partitioned table are admitted the same way: each occurrence reads + // that table through its own raw source, so the match runs above the regions rather than + // inside one of them. let repeated = build_native_fused_plan_with_provider( "some_metric / some_metric", true, native_two_table_provider(vec![0]).await, ) .await; - assert!(!has_pair_node(&repeated), "{repeated}"); + let repeated_fused = + find_fused_series(&repeated).unwrap_or_else(|| panic!("repeated selectors: {repeated}")); + assert_eq!(2, repeated_fused.inputs().len(), "{repeated}"); + for input in repeated_fused.inputs() { + assert!( + matches!(input, LogicalPlan::Extension(extension) + if extension.node.as_any().is::()), + "each occurrence reads its own source: {repeated}" + ); + } } /// `some_metric` as [`native_two_table_provider`] builds it, and `other_metric` whose tags are @@ -4193,17 +4173,28 @@ async fn native_fused_three_selector_trees_bind_every_occurrence() { } } -/// More than two same-table selectors on a partitioned table must not be placed as a -/// multi-child region-local node. Cross-table occurrences may each use a raw source. +/// More than two same-table selectors on a partitioned table each read that table through their own +/// source, so the match runs above the regions. No region is handed a node over the bare scans, which +/// is the shape it refuses. #[tokio::test] -async fn native_fused_same_table_three_selectors_on_partitioned_table_fall_back() { +async fn native_fused_same_table_three_selectors_on_partitioned_table_bind_their_sources() { let plan = build_native_fused_plan_with_provider( "some_metric + some_metric + some_metric", true, native_two_table_provider(vec![0]).await, ) .await; - assert!(!has_pair_node(&plan), "{plan}"); + let fused = find_fused_series(&plan).unwrap_or_else(|| panic!("{plan}")); + assert!(fused.inputs().len() > 1, "{plan}"); + assert_eq!( + vec![ + "some_metric".to_string(), + "some_metric".to_string(), + "some_metric".to_string() + ], + fused_input_metrics(&plan), + "each selector occurrence reads its own source: {plan}" + ); } /// Find the final half of a split aggregate. @@ -4897,9 +4888,10 @@ async fn native_classic_histogram_provider() -> DfTableSourceProvider { ) } -/// Two classic-histogram functions over raw series, each admitted as one native operand: a -/// `histogram_quantile` reads each histogram off the bucket series that share its other labels and -/// drops `le`, so neither operand keeps `le` in its binding nor in the statement's output. +/// Two classic-histogram functions over raw series: a `histogram_quantile` reads each histogram off +/// the bucket series that share its other labels and drops `le`, so neither operand keeps `le` in +/// its binding nor in the statement's output. Each occurrence reads its own source, so the division +/// matches above both of them instead of joining two separately fused operands. #[tokio::test] async fn native_fused_histogram_quantile_binary_drops_le_from_operand_matching() { let provider = native_classic_histogram_provider().await; @@ -4910,9 +4902,15 @@ async fn native_fused_histogram_quantile_binary_drops_le_from_operand_matching() ) .await; let display = plan.display_indent().to_string(); - // The whole statement has two same-table selectors and declines at that guard. Each - // histogram_quantile is nevertheless a one-selector operand and is admitted by the hybrid path. - assert_eq!(4, display.matches("FusedSeries").count(), "{display}"); + // One node over one segment per selector occurrence: the two reads are independent, so the + // division matches above them rather than below a join of two fused operands. + let fused = find_fused_series(&plan).unwrap_or_else(|| panic!("{display}")); + assert_eq!(2, fused.inputs().len(), "{display}"); + assert_eq!( + vec!["m".to_string(), "m".to_string()], + fused_input_metrics(&plan), + "each selector occurrence reads its own source: {display}" + ); assert!( !display.contains(".le ="), "join key still references dropped `le`: {display}" @@ -4952,11 +4950,13 @@ async fn native_fused_histogram_quantile_binary_drops_le_from_operand_matching() } } -/// A shape the split cannot keep correct keeps the existing plan: 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 by the state rows. A literal scalar -/// sibling is not one of them: the operation is computed per series below the fold, or rebuilt over -/// the merged values above it. +/// A shape neither the split nor the node can keep correct keeps the existing plan: a sibling a +/// call computes has no native node at all, a group by a value field cannot be named by the state +/// rows, and a kernel the compiler does not fold keeps it too. +/// +/// `sum(m) / m` and `m + sum(m)` are not on this list. The fold and the plain selector are two +/// occurrences, so each reads its own source and the operation above them answers over both; the +/// split is only the optimization the one-input shapes keep. /// /// `timestamp(sum(m))` and `sum(timestamp(abs(m)))` are not on this list. `timestamp` over a /// computed vector reports the evaluation time, which is what Prometheus reports too - an aggregate @@ -4968,10 +4968,6 @@ async fn native_fused_unsplittable_aggregate_falls_back() { for query in [ "quantile(0.5, some_metric)", "sum(quantile(0.5, some_metric))", - // The sibling is data rather than a literal scalar, so the operation would pair each series - // with the aggregate's series before the merge. - "sum(some_metric) / some_metric", - "some_metric + sum(some_metric)", // A scalar a call computes is not a literal the split could keep: `scalar(...)` has no // native node, so the statement keeps the existing plan. "sum(some_metric) / scalar(some_metric)", @@ -4997,6 +4993,38 @@ async fn native_fused_unsplittable_aggregate_falls_back() { } } +/// A fold beside a plain selector is two occurrences, so each reads its own source and the fold and +/// the match run above both. No partition needs to hold the whole of the input for that: the inputs +/// are reconciled before the node evaluates them. +#[tokio::test] +async fn native_fused_aggregate_above_an_independent_selector_is_admitted() { + for query in [ + "sum(some_metric) / some_metric", + "some_metric + sum(some_metric)", + ] { + let native = build_native_fused_plan(query, true).await; + let fused = find_fused_series(&native).unwrap_or_else(|| panic!("`{query}`: {native}")); + assert_eq!(2, fused.inputs().len(), "`{query}`: {native}"); + assert_eq!( + vec!["some_metric".to_string(), "some_metric".to_string()], + fused_input_metrics(&native), + "`{query}`: each selector occurrence reads its own source: {native}" + ); + assert!( + find_final_split(&native).is_none(), + "`{query}`: a tree over two inputs is not a one-input split: {native}" + ); + let without_flag = build_native_fused_plan(query, false).await; + assert!( + !without_flag + .display_indent() + .to_string() + .contains("FusedSeries"), + "`{query}` takes the native plan only when it is enabled: {without_flag}" + ); + } +} + /// Suffixes whose answers need the whole vector cannot fuse into a partition-local `Final`, so /// the split declines them: it neither exports aggregate state nor runs a `Final` aggregate. /// Ranking (`topk`) and a second aggregate cannot be answered from a partition's groups alone. @@ -5216,22 +5244,6 @@ async fn native_fused_preparation_decline_preserves_legacy_plan() { ); } -/// An aggregate is answered natively now (`sum(rate(..))`, `avg by(..)(..)` are split), so what -/// still falls back here is the shape the split cannot carry: a fold over two actual sources, which -/// no partition holds the whole of. -#[tokio::test] -async fn native_fused_aggregate_falls_back_to_legacy_plan() { - for query in ["some_metric + sum(some_metric)"] { - let plan = build_native_fused_plan(query, true).await; - let display = plan.display_indent().to_string(); - assert!(!display.contains("FusedSeries"), "{query}: {display}"); - assert!(find_final_split(&plan).is_none(), "{query}: {display}"); - assert!(display.contains("Aggregate:"), "{query}: {display}"); - let legacy = build_native_fused_plan(query, false).await; - assert_eq!(display, legacy.display_indent().to_string(), "{query}"); - } -} - #[tokio::test] async fn native_fused_subquery_falls_back_to_legacy_plan() { for query in [ diff --git a/tests-integration/src/tests/native_aggregate_test.rs b/tests-integration/src/tests/native_aggregate_test.rs index ed87d009d48..519a703fe9f 100644 --- a/tests-integration/src/tests/native_aggregate_test.rs +++ b/tests-integration/src/tests/native_aggregate_test.rs @@ -97,7 +97,6 @@ use crate::standalone::GreptimeDbStandaloneBuilder; /// crate-private, so it is repeated here. const NATIVE_PROMQL_EXTENSION_KEY: &str = "query.experimental_native_promql"; /// The physical name of the fused producer. -const FUSED_EXEC_NAME: &str = "FusedSeriesExec"; /// The metric a fused producer registers for the values its own root node produced, per partition. const ROOT_OUTPUT_VALUES_METRIC: &str = "native.root.output_values"; /// The number of sample positions consumed by the root node, per partition. @@ -567,38 +566,38 @@ async fn native_hybrid_binary_05_06_08() { .map(|row| row.value.unwrap()) .collect::>() ); + // The native route matches the whole right operand and checks the one-side groups itself, so + // it reports the duplicated `other` group that Prometheus reports. The legacy value above is + // the one that differs: copying the left matcher into the right scan drops every group without + // a partner, and a dropped group cannot be reported as duplicated. A partition column on the + // right table does not change that - the pair reads independent raw sources either way. let mark = observer.mark(); - let (hybrid, capture) = run_query(&instance, &propagated, true, &observer).await; - assert_eq!( - vec![5.0], - hybrid - .rows - .iter() - .map(|row| row.value.unwrap()) - .collect::>() - ); - assert_eq!( - 2, - capture.fused.len(), - "both binary operands use native execution" - ); - let plans = observer.since(mark); - let frontend = plans - .iter() - .find(|plan| find_binary_join(plan).is_some()) - .expect("frontend Join plan"); - let join = find_binary_join(frontend).unwrap(); - let uniqueness = join - .children() - .iter() - .map(|branch| uniqueness_operators_above_merge(branch)) - .fold((false, false), |(window, filter), found| { - (window || found.0, filter || found.1) - }); + let mut query_ctx = QueryContext::with_db_name(None); + query_ctx.set_extension(QUERY_PARALLELISM_HINT, SCAN_PARTITIONS.to_string()); + query_ctx.set_extension(NATIVE_PROMQL_EXTENSION_KEY, "true".to_string()); + let output = instance + .do_query(&tql_eval(&propagated), Arc::new(query_ctx)) + .await + .remove(0) + .expect("the query plans"); + let error = match output.data { + OutputData::Stream(stream) => collect_batches(stream) + .await + .expect_err("a duplicate match group fails"), + other => panic!("`{propagated}` was expected to stream: {other:?}"), + }; assert!( - uniqueness.0 && uniqueness.1, - "on(job) FE uniqueness operators: {}", - displayable(frontend.as_ref()).indent(false) + error.to_string().contains( + "found duplicate series for the match group {job=\"other\"} on the right hand-side" + ), + "{propagated}: {error}" + ); + // The statement is admitted natively: a fallback to the legacy plan would have answered the + // value instead of reporting the duplicate. + let capture = NativeCapture::from_plans(&observer.since(mark)); + assert!( + capture.fused.iter().any(|node| node.children().len() == 2), + "the partitioned pair runs as one native binary" ); // The left selector is a nonempty single `api` series, while the unpartitioned RHS also @@ -771,7 +770,6 @@ async fn check_reduced_label_matching(instance: &Arc, observer: &Nativ ); let (legacy, legacy_capture) = run_query(instance, &unique_query, false, observer).await; assert!(legacy_capture.fused.is_empty()); - let mark = observer.mark(); let (native, capture) = run_query(instance, &unique_query, true, observer).await; assert_eq!( legacy.rows.len(), @@ -796,43 +794,15 @@ async fn check_reduced_label_matching(instance: &Arc, observer: &Nativ native.rows.len(), "the matching instance filters leave one matching job" ); - assert_eq!(2, capture.fused.len(), "{unique_query}"); - let plans = observer.since(mark); - let frontend = plans - .iter() - .find(|plan| find_binary_join(plan).is_some()) - .expect("frontend Join plan"); - let join = find_binary_join(frontend).unwrap(); - let uniqueness = join - .children() - .iter() - .map(|branch| uniqueness_operators_above_merge(branch)) - .fold((false, false), |(window, assertion_filter), found| { - (window || found.0, assertion_filter || found.1) - }); + // The whole binary is native: one fused node matches both operands above their own sources, + // and the duplicate-group check is the native matcher's, not a frontend WindowAgg/Filter pair. assert!( - uniqueness.0, - "on(job) requires a frontend uniqueness WindowAggExec above MergeScan: {}", - displayable(frontend.as_ref()).indent(false) + capture.fused.iter().any(|node| node.children().len() > 1), + "`{unique_query}` must run as one native binary: {:?}", + capture.fused ); - assert!( - uniqueness.1, - "on(job) requires a frontend prom_assert_unique_match_group FilterExec above MergeScan: {}", - displayable(frontend.as_ref()).indent(false) - ); - for branch in join.children() { - assert!( - contains_execution_node(branch, "MergeScanExec"), - "on(job) FE branch must retain remote boundary: {}", - displayable(frontend.as_ref()).indent(false) - ); - assert!( - !contains_execution_node(branch, FUSED_EXEC_NAME), - "on(job) fused node must stay in region" - ); - } for fused in &capture.fused { - assert_fused_reads_scan_directly(fused, &unique_query); + assert_fused_reads_sources(fused, &unique_query); } let duplicate_query = format!("{GAUGE_TABLE} / on(job) {COUNTER_RHS_TABLE}"); @@ -941,7 +911,10 @@ async fn check_hybrid_binary_cases(instance: &Arc, observer: &NativePl "one-to-one match must drop unmatched series" ); } - assert_eq!(2, capture.fused.len(), "{promql}: plans since {mark}"); + assert!( + !capture.fused.is_empty(), + "{promql}: the pair must stay native: plans since {mark}" + ); let region_plans = observer .since(mark) .into_iter() @@ -970,75 +943,37 @@ async fn check_hybrid_binary_cases(instance: &Arc, observer: &NativePl scan_tables.contains(&COUNTER_RHS_TABLE), "{promql}: independent RHS scan missing: {scan_tables:?}" ); - for fused in &capture.fused { - assert_fused_reads_scan_directly(fused, &format!("hybrid `{promql}`")); - } - for fused in &capture.fused { - let per_fused = NativeCapture { - fused: vec![Arc::clone(fused)], - finalize: Vec::new(), - }; - let counts = fused_root_metric_per_partition( - &per_fused, - ROOT_OUTPUT_VALUES_METRIC, - &format!("hybrid `{promql}`"), - ); - assert!( - counts.iter().all(|count| count.is_some_and(|n| n > 0)), - "{promql}: each scan partition must emit values, got {counts:?}" - ); - } - let plans = observer.since(mark); - let frontend = plans + // The binary routes by label set, so its output partitions are keyed by the match group and + // an input partition may contribute nothing to a given one. What must still hold is that + // each operand registered its values on every partition it reads - the same check the + // unpartitioned pair makes with the operand metrics. + let binary = capture + .fused .iter() - .find(|plan| find_binary_join(plan).is_some()) - .unwrap_or_else(|| panic!("{promql}: FE JoinExec missing; plans: {plans:#?}")); - let join = find_binary_join(frontend).unwrap(); - assert_eq!( - 2, - join.children().len(), - "{promql}: FE join branches: {}", - displayable(frontend.as_ref()).indent(false) - ); - for branch in join.children() { - assert!( - contains_execution_node(branch, "MergeScanExec"), - "{promql}: FE Join branch lacks MergeScanExec: {}", - displayable(frontend.as_ref()).indent(false) - ); - assert!( - !contains_execution_node(branch, FUSED_EXEC_NAME), - "{promql}: FusedSeriesExec leaked onto frontend: {}", - displayable(frontend.as_ref()).indent(false) - ); - } + .find(|node| node.children().len() > 1) + .unwrap_or_else(|| panic!("{promql}: no native binary among {:?}", capture.fused)); + let left = + fused_node_count_per_partition(binary, "native.root.left.output_values", &promql); + let right = + fused_node_count_per_partition(binary, "native.root.right.output_values", &promql); assert!( - !is_nested_under_merge(frontend, "HashJoinExec"), - "{promql}: FE Join is under MergeScan: {}", - displayable(frontend.as_ref()).indent(false) + left.iter().chain(&right).all(Option::is_some), + "{promql}: operand metrics must be registered for every partition: \ + left={left:?}, right={right:?}" ); - if index == 1 { - assert!( - has_projection_parent_of_join(frontend), - "06 outer scalar arithmetic must project above JoinExec: {}", - displayable(frontend.as_ref()).indent(false) - ); + // The pair is matched by the native binary above its own sources; the frontend join and + // its uniqueness operators only existed while the binary could not be planned natively. + assert!( + capture.fused.iter().any(|node| node.children().len() > 1), + "{promql}: the pair must be one native binary: {:?}", + capture.fused + ); + for fused in &capture.fused { + assert_fused_reads_sources(fused, &format!("hybrid `{promql}`")); } } } -fn find_binary_join(plan: &Arc) -> Option<&Arc> { - if matches!( - plan.name(), - "HashJoinExec" | "NestedLoopJoinExec" | "CrossJoinExec" - ) { - return Some(plan); - } - plan.children() - .iter() - .find_map(|child| find_binary_join(child)) -} - fn find_execution_node<'a>( plan: &'a Arc, name: &str, @@ -1055,48 +990,6 @@ fn contains_execution_node(plan: &Arc, name: &str) -> bool { find_execution_node(plan, name).is_some() } -/// Finds the PromQL uniqueness validation operators on the frontend side of a remote scan. -/// Traversal stops at each MergeScan so it cannot accidentally match region-side execution. -fn uniqueness_operators_above_merge(plan: &Arc) -> (bool, bool) { - if plan.name() == "MergeScanExec" { - return (false, false); - } - - let mut has_window = matches!(plan.name(), "WindowAggExec" | "BoundedWindowAggExec"); - let mut has_assertion_filter = plan.name() == "FilterExec" - && format!("{plan:?}").contains("prom_assert_unique_match_group"); - for child in plan.children() { - let (child_window, child_filter) = uniqueness_operators_above_merge(child); - has_window |= child_window; - has_assertion_filter |= child_filter; - } - (has_window, has_assertion_filter) -} - -fn is_nested_under_merge(plan: &Arc, name: &str) -> bool { - fn walk(plan: &Arc, under_merge: bool, name: &str) -> bool { - if under_merge && plan.name() == name { - return true; - } - plan.children() - .iter() - .any(|child| walk(child, under_merge || plan.name() == "MergeScanExec", name)) - } - walk(plan, false, name) -} - -fn has_projection_parent_of_join(plan: &Arc) -> bool { - plan.name() == "ProjectionExec" - && plan - .children() - .iter() - .any(|child| find_binary_join(child).is_some()) - || plan - .children() - .iter() - .any(|child| has_projection_parent_of_join(child)) -} - /// Executes one statement and drops its result. async fn execute_sql(instance: &Arc, sql: &str) { for result in instance.do_query(sql, QueryContext::arc()).await { @@ -1514,6 +1407,31 @@ fn assert_fused_reads_scan_directly(fused: &Arc, context: &st ); } +/// The fused node keeps its scan width and reads every operand through its own source: a single +/// operand reads one scan, and a native binary takes one input per operand, each a raw source or a +/// merge scan over the regions. +fn assert_fused_reads_sources(fused: &Arc, context: &str) { + let plan = displayable(fused.as_ref()).indent(false).to_string(); + assert_eq!( + SCAN_PARTITIONS, + fused.properties().output_partitioning().partition_count(), + "{context}: the fused producer must keep the {SCAN_PARTITIONS} scan partitions:\n{plan}" + ); + let children = fused.children(); + assert!( + !children.is_empty(), + "{context}: the fused producer must read at least one input:\n{plan}" + ); + for child in children { + assert!( + contains_execution_node(child, "SeriesScanExec") + || contains_execution_node(child, "MergeScanExec") + || contains_execution_node(child, "RawSeriesSourceExec"), + "{context}: every operand must read its own source:\n{plan}" + ); + } +} + /// The node a transparent cooperative-scheduling wrapper stands in for, or `node` itself. /// /// The decorator clones its input's properties, changes only the scheduling type and forwards diff --git a/tests-integration/src/tests/promql_test.rs b/tests-integration/src/tests/promql_test.rs index fd5bb1e116d..aaeac805e8b 100644 --- a/tests-integration/src/tests/promql_test.rs +++ b/tests-integration/src/tests/promql_test.rs @@ -1868,6 +1868,21 @@ fn observed_physical_nodes( nodes } +/// The region reads a native node answers from, descending through cooperative wrappers and the +/// segments a statement cut out of itself. +fn native_region_inputs( + plan: &Arc, + out: &mut Vec>, +) { + for child in plan.children() { + if child.is::() { + out.push(Arc::clone(child)); + } else { + native_region_inputs(child, out); + } + } +} + fn observed_series_scans(observer: &NativePlanObserver) -> Vec> { observed_physical_nodes(observer) .into_iter() @@ -2880,11 +2895,11 @@ fn repeated_selector_fixture_inserts(prefix: &str) -> String { } /// Real two-datanode TCP acceptance for a repeated selector over ONE partitioned metric table. -/// Over the wire every prepared scan becomes its own raw source with its own time bounds, so two -/// occurrences of the same fragmented table - a ratio of two windows, or a binary expression over -/// two identical ones - stay independent instead of falling back. In process the same expression -/// would place one node region-locally over both scans of the one table, which no region can serve, -/// so this admission is remote-only. +/// Every prepared scan becomes its own raw source with its own time bounds, so two occurrences of +/// the same fragmented table - a ratio of two windows, or a binary expression over two identical +/// ones - stay independent instead of falling back. Over the wire the transport carries those +/// sources; in process they keep the node out of the region, which cannot serve one node over both +/// scans of the one table. #[tokio::test(flavor = "multi_thread")] async fn distributed_native_promql_repeated_selector_over_partitioned_table() { common_telemetry::init_default_ut_logging(); @@ -6587,57 +6602,45 @@ async fn native_promql_info_enriches_from_the_info_table() { } } -/// A two-selector query over a table split into regions never fuses the binary itself. +/// A window over a table partitioned on a column that is not a tag sees the whole series. /// -/// A fused node with more than one input has no placement on such a table — the dist analyzer -/// refuses it with "only accept plan with at most one child" — so admitting one used to make the -/// whole query fail instead of falling back. Each operand may still be fused on its own: that is a -/// single-input node over a region-local scan, which places fine, and it is what the hybrid -/// operand route does. What must never appear is a fused node with two children, so that is what -/// is asserted rather than "no fused node at all". -/// A window over a table partitioned on its time index sees the whole series. -/// -/// One selector is enough: such a partition splits a series across regions, so a per-region window -/// would evaluate over a fragment and drop every range that straddles the boundary. The selector -/// therefore reads the table through a raw source and the window runs above the reconciliation. -/// The fixture puts one series' two samples on either side of the boundary. +/// Such a table splits one series across regions, so a per-region prefix evaluates `rate` over a +/// fragment and loses every window that straddles the boundary. The cut therefore has to keep each +/// occurrence's raw source and run the prefix above the reconciliation, exactly as the remote route +/// does. The fixture puts the two samples of one series on either side of the boundary and compares +/// the native answer with the existing plan's. #[tokio::test(flavor = "multi_thread")] -async fn native_promql_time_partitioned_selector_reads_whole_series() { +async fn native_promql_window_over_a_time_partitioned_table_keeps_whole_series() { let (standalone, observer) = native_test_instance(GreptimeDbStandaloneBuilder::new( - "native_promql_time_partitioned_selector", + "native_promql_time_partitioned_window", )) .await; let instance = standalone.fe_instance().clone(); execute_all( &instance, - "CREATE TABLE native_time_split (host STRING, ts TIMESTAMP TIME INDEX, val DOUBLE, \ + "CREATE TABLE native_time_partitioned (host STRING, ts TIMESTAMP TIME INDEX, val DOUBLE, \ PRIMARY KEY(host)) PARTITION ON COLUMNS (ts) (ts < '1970-01-01 00:02:30', \ ts >= '1970-01-01 00:02:30')", QueryContext::arc(), ) .await; + // One series, one sample in each region: the second is 240s after the first, and the boundary + // sits between them, so neither region holds a complete window on its own. execute_all( &instance, - "INSERT INTO native_time_split VALUES ('h0', 60000, 60), ('h0', 240000, 240)", + "INSERT INTO native_time_partitioned VALUES ('h0', 60000, 60), ('h0', 240000, 240)", QueryContext::arc(), ) .await; for query in [ - "rate(native_time_split[5m])", - "sum(rate(native_time_split[5m]))", + "rate(native_time_partitioned[5m]) + rate(native_time_partitioned[5m])", + "(rate(native_time_partitioned[5m]) + rate(native_time_partitioned[5m])) \ + + rate(native_time_partitioned[5m])", ] { let sql = format!("TQL EVAL (300, 300, '1m') {query}"); observer.clear(); - let (_, native) = - native_labelled_rows(native_tql_output(&instance, &sql, true).await).await; - let plan = observer.root_plan(); - assert!( - find_plan_node(&plan, "FusedSeriesExec").is_some(), - "{query}: the frontend tree must run natively, not on the existing plan: {:?}", - display_plan_tree(&plan) - ); - let (_, legacy) = - native_labelled_rows(native_tql_output(&instance, &sql, false).await).await; + let native = native_result_rows(native_tql_output(&instance, &sql, true).await).await; + let legacy = native_result_rows(native_tql_output(&instance, &sql, false).await).await; assert!( !native.is_empty(), "{query}: the native plan answered nothing" @@ -6646,11 +6649,33 @@ async fn native_promql_time_partitioned_selector_reads_whole_series() { native, legacy, "{query}: the native plan must match the existing one" ); + let root = observer.root_plan(); + let mut inputs = Vec::new(); + native_region_inputs(&root, &mut inputs); + assert!( + !inputs.is_empty(), + "{query}: the window runs above the region reads: {:?}", + display_plan_tree(&root) + ); + // A fallback plan also produces the existing plan's rows, so the admission has to be + // asserted separately: the frontend tree must carry a fused node. + assert!( + find_plan_node(&root, "FusedSeriesExec").is_some(), + "{query}: the tree must run natively, not on the existing plan: {:?}", + display_plan_tree(&root) + ); } } +/// A query over two selectors of one partitioned table matches above the regions. +/// +/// A fused node over the bare scans of such a table has no placement in a region — the dist +/// analyzer refuses it with "only accept plan with at most one child" — so each selector +/// occurrence reads that table through its own raw region input instead. Every input is one read of +/// the whole table, and the matching the query asks for runs once above them, over every region's +/// contribution: two inputs for a two-selector query, three for a three-selector one. #[tokio::test(flavor = "multi_thread")] -async fn native_promql_two_selectors_on_a_partitioned_table_keep_the_binary_unfused() { +async fn native_promql_two_selectors_on_a_partitioned_table_match_above_independent_regions() { let (standalone, observer) = native_test_instance(GreptimeDbStandaloneBuilder::new( "native_promql_partitioned_pair", )) @@ -6670,10 +6695,11 @@ async fn native_promql_two_selectors_on_a_partitioned_table_keep_the_binary_unfu QueryContext::arc(), ) .await; - for (query, expected) in [ + for (query, expected, sources) in [ ( "native_partitioned_pair{kind=\"a\"} and on(host) native_partitioned_pair{kind=\"b\",host=\"h0\"}", vec![(5_000, "h0".to_string(), 1.0)], + 2, ), ( "native_partitioned_pair{kind=\"a\"} - on(host) native_partitioned_pair{kind=\"b\"}", @@ -6681,8 +6707,10 @@ async fn native_promql_two_selectors_on_a_partitioned_table_keep_the_binary_unfu (5_000, "h0".to_string(), -1.0), (5_000, "h2".to_string(), -1.0), ], + 2, ), - // Three selectors: the inner pair may take the hybrid route under the outer join. + // Three selectors: the inner pair matches above its own two region inputs, under the outer + // match. ( "native_partitioned_pair{kind=\"a\"} - on(host) native_partitioned_pair{kind=\"b\"} \ + on(host) native_partitioned_pair{kind=\"a\"}", @@ -6690,19 +6718,30 @@ async fn native_promql_two_selectors_on_a_partitioned_table_keep_the_binary_unfu (5_000, "h0".to_string(), 0.0), (5_000, "h2".to_string(), 2.0), ], + 3, ), ] { let sql = format!("TQL EVAL (5, 5, '1s') {query}"); observer.clear(); let native = native_result_rows(native_tql_output(&instance, &sql, true).await).await; - if let Some(fused) = observer.fused() { - assert_eq!( - fused.children().len(), - 1, - "`{query}` fused the two-selector binary, which has no placement on a partitioned \ - table: {fused:?}" - ); - } + // The frontend's own tree: the fused node above the regions, not the region-side plan the + // observer also wraps. + let root = observer.root_plan(); + let mut inputs = Vec::new(); + native_region_inputs(&root, &mut inputs); + assert_eq!( + sources, + inputs.len(), + "`{query}` reads one region input per selector occurrence: {:?}", + display_plan_tree(&root) + ); + // The values alone cannot tell native matching from a legacy frontend join that reads the + // same regions, so require the native binary in the frontend tree. + assert!( + find_plan_node(&root, "FusedSeriesExec").is_some(), + "`{query}` must match natively above its region inputs: {:?}", + display_plan_tree(&root) + ); assert_eq!(native, expected, "{query}"); let legacy = native_result_rows(native_tql_output(&instance, &sql, false).await).await; assert_eq!(native, legacy, "{query}");