From fe184842614c66188889412d58eebc85bf92d5cb Mon Sep 17 00:00:00 2001 From: discord9 Date: Fri, 6 Nov 2026 16:45:28 +0800 Subject: [PATCH] feat(promql): print each fused node's own compiled tree The fused node carried the PromQL text it was compiled from and printed it on its EXPLAIN line, so a split aggregate's two halves - the frontend's final tree and the region's partial prefix - showed the same statement over different plans. Both halves now print their own tree: the logical node without figures, the physical node with each tree node's metrics. The query text is gone from the node, its payload and its identity. Signed-off-by: discord9 --- src/promql/src/extension_plan/fused_series.rs | 97 ++++------ .../src/extension_plan/fused_series/tests.rs | 177 +++++++----------- .../fused_series/tests/frame_output.rs | 3 - .../series_repartition/tests.rs | 1 - src/promql/src/native/display.rs | 2 +- src/promql/src/native/plan/tests.rs | 20 +- src/query/src/analyze.rs | 157 ++++++++++++---- src/query/src/dist_plan/analyzer/test.rs | 8 +- src/query/src/dist_plan/commutativity.rs | 28 +-- src/query/src/optimizer/scan_hint.rs | 9 +- .../src/optimizer/series_frame_output.rs | 1 - src/query/src/promql/native_compile/tests.rs | 10 +- src/query/src/promql/planner/native.rs | 38 +--- src/query/src/promql/planner/test.rs | 63 +++++-- src/query/src/query_engine/state.rs | 1 - tests-integration/src/tests/promql_test.rs | 60 +++--- tests-integration/tests/http.rs | 1 - tests-integration/tests/native_prefix.rs | 3 - 18 files changed, 326 insertions(+), 353 deletions(-) diff --git a/src/promql/src/extension_plan/fused_series.rs b/src/promql/src/extension_plan/fused_series.rs index a30bedbb5c3..ea2ef3f2de2 100644 --- a/src/promql/src/extension_plan/fused_series.rs +++ b/src/promql/src/extension_plan/fused_series.rs @@ -14,17 +14,14 @@ //! One node that runs a compiled native tree over raw series. //! -//! [`FusedSeries`] is the logical half: it holds the PromQL text the tree was compiled from, the -//! tree itself and the binding its input was compiled against. [`FusedSeriesExec`] is the physical -//! half: it owns one execution of that tree per partition and decodes its input frames into raw -//! series chunks and converts its evaluated chunks back into batches. +//! [`FusedSeries`] is the logical half: it holds the compiled tree and the binding its input was +//! compiled against. [`FusedSeriesExec`] is the physical half: it owns one execution of that tree +//! per partition, decodes its input frames into raw series chunks, and converts evaluated chunks +//! back into batches. //! -//! The two halves show different things in `EXPLAIN`, which is the point of having both. The -//! logical node prints the query a reader wrote ([`FusedSeries`]'s `fmt_for_explain`), and the -//! physical node prints the tree it will run ([`FusedSeriesExec`]'s `fmt_as`, which embeds -//! [`NativeTree`]'s own display rather than a second renderer). A reader comparing the two plans -//! can therefore see whether a query was compiled the way they expected without decoding node -//! names. +//! Both halves show their own compiled tree in `EXPLAIN`, rendered by [`NativeTree`]'s display. +//! The logical node shows its tree without metrics; the physical node includes each tree node's +//! recorded metrics when available. //! //! Scope of this layer: the input is a stream of native raw series frames //! ([`raw_series_frame_schema`](datatypes::series::frame::raw_series_frame_schema)) — one row per @@ -306,13 +303,10 @@ fn series_frame_output_schema(schema: &SeriesSchema) -> DataFusionResult, /// The compiled tree to run. tree: NativeTree, /// What the tree's input was compiled against. @@ -367,7 +361,6 @@ struct FusedSeriesPayload { time_index: String, tags: Vec, format: FusedOutputFormat, - promql: Arc, output_schema: datafusion::arrow::datatypes::Schema, /// Whether this node is the partition-local producer of a canonical merge; part of the node's /// identity, not of where it may be placed. @@ -377,8 +370,7 @@ struct FusedSeriesPayload { impl FusedSeries { /// Serializes the native tree and the binding required to reconstruct this plan node. - /// The PromQL text is included for diagnostics only and never drives execution; children - /// remain ordinary Substrait inputs. + /// Children remain ordinary Substrait inputs. /// /// This is only the node's own encoding. Whether a planner is allowed to place this node on /// another machine is a separate decision, made by the planner and the transport, not here. @@ -393,7 +385,6 @@ impl FusedSeries { time_index: self.binding.time_index.clone(), tags: self.binding.tags.clone(), format: self.format, - promql: Arc::clone(&self.promql), output_schema: self.output_schema.as_arrow().clone(), canonical_producer: self.canonical_producer, }; @@ -450,7 +441,6 @@ impl FusedSeries { time_index, tags, format, - promql, output_schema, canonical_producer: _, } = payload; @@ -478,8 +468,7 @@ impl FusedSeries { } let output_schema = Arc::new(datafusion::common::DFSchema::try_from(output_schema)?); let canonical_producer = payload.canonical_producer; - let mut node = - Self::try_new_with_format(promql, tree, binding, children, output_schema, format)?; + let mut node = Self::try_new_with_format(tree, binding, children, output_schema, format)?; if canonical_producer { if !matches!(node.tree.root().expr, NativeExpr::Merge { .. }) { return Err(DataFusionError::Plan( @@ -507,14 +496,12 @@ impl FusedSeries { /// Creates the node, checking that `tree` is legal against `binding`. pub fn try_new( - promql: Arc, tree: NativeTree, binding: FusedSeriesBinding, children: Vec, output_schema: DFSchemaRef, ) -> DataFusionResult { Self::try_new_with_format( - promql, tree, binding, children, @@ -530,7 +517,6 @@ impl FusedSeries { /// state frame the tree does not produce, which `try_new_with_format` would then only catch for /// some of its columns. pub fn try_new_state_export( - promql: Arc, tree: NativeTree, binding: FusedSeriesBinding, children: Vec, @@ -542,7 +528,6 @@ impl FusedSeries { )) })?; Self::try_new_with_format( - promql, tree, binding, children, @@ -581,7 +566,6 @@ impl FusedSeries { /// Creates a node with an explicitly selected output format. pub fn try_new_with_format( - promql: Arc, tree: NativeTree, binding: FusedSeriesBinding, children: Vec, @@ -616,7 +600,6 @@ impl FusedSeries { } } Ok(Self { - promql, tree, binding, children, @@ -638,7 +621,6 @@ impl FusedSeries { /// `output_schema` is the frame the evaluated root declares; it is derived here rather than /// taken from the caller so the two cannot disagree. pub fn try_new_canonical_producer( - promql: Arc, tree: NativeTree, binding: FusedSeriesBinding, children: Vec, @@ -654,7 +636,6 @@ impl FusedSeries { frame.as_ref().clone(), )?); Ok(Self { - promql, tree, binding, children, @@ -676,11 +657,6 @@ impl FusedSeries { self.canonical_producer } - /// The PromQL text this node was compiled from. - pub fn promql(&self) -> &str { - &self.promql - } - /// The selected output representation. pub fn format(&self) -> FusedOutputFormat { self.format @@ -961,7 +937,6 @@ impl FusedSeries { }) .collect::>>()?; let exec: Arc = Arc::new(FusedSeriesExec::build( - Arc::clone(&self.promql), inputs, Arc::new(self.tree.clone()), output_schema, @@ -1078,23 +1053,29 @@ impl UserDefinedLogicalNodeCore for FusedSeries { } fn fmt_for_explain(&self, f: &mut Formatter) -> fmt::Result { + write!(f, "FusedSeries")?; + let mut has_params = false; match self.format { - FusedOutputFormat::SeriesFrame if self.canonical_producer => write!( - f, - "FusedSeries: {} (output: series_frame, per-partition canonical merge)", - self.promql - )?, - FusedOutputFormat::SeriesFrame => { - write!(f, "FusedSeries: {} (output: series_frame)", self.promql)? + FusedOutputFormat::SeriesFrame if self.canonical_producer => { + write!(f, ": (output: series_frame, per-partition canonical merge)")?; + has_params = true; } - FusedOutputFormat::PlainRows => write!(f, "FusedSeries: {}", self.promql)?, + FusedOutputFormat::SeriesFrame => { + write!(f, ": (output: series_frame)")?; + has_params = true; + } + FusedOutputFormat::PlainRows => {} } if let Some((_, SeriesRouting::CanonicalLabelsExcept(excluded))) = state_routing(self.tree.root()) { - write!(f, " (state routing: canonical labels except {excluded:?})")?; + let separator = if has_params { " " } else { ": " }; + write!( + f, + "{separator}(state routing: canonical labels except {excluded:?})" + )?; } - Ok(()) + write!(f, "\n{}", self.tree) } fn with_exprs_and_inputs( @@ -1104,7 +1085,6 @@ impl UserDefinedLogicalNodeCore for FusedSeries { ) -> DataFusionResult { validate_binding(&self.tree, &self.binding, &inputs)?; Ok(Self { - promql: Arc::clone(&self.promql), tree: self.tree.clone(), binding: self.binding.clone(), children: inputs, @@ -1117,8 +1097,7 @@ impl UserDefinedLogicalNodeCore for FusedSeries { impl PartialEq for FusedSeries { fn eq(&self, other: &Self) -> bool { - self.promql == other.promql - && self.tree == other.tree + self.tree == other.tree && self.binding == other.binding && self.children == other.children && self.output_schema == other.output_schema @@ -1131,7 +1110,6 @@ impl Eq for FusedSeries {} impl Hash for FusedSeries { fn hash(&self, state: &mut H) { - self.promql.hash(state); self.binding.hash(state); self.children.hash(state); self.output_schema.hash(state); @@ -1143,10 +1121,6 @@ impl Hash for FusedSeries { impl PartialOrd for FusedSeries { fn partial_cmp(&self, other: &Self) -> Option { - match self.promql.partial_cmp(&other.promql) { - Some(Ordering::Equal) => {} - ord => return ord, - } match self.binding.partial_cmp(&other.binding) { Some(Ordering::Equal) => {} ord => return ord, @@ -2126,8 +2100,6 @@ fn claim_evidence( /// The physical node that runs a compiled native tree over its input's raw series. pub struct FusedSeriesExec { - /// The PromQL text the tree was compiled from, which is what the plan line shows. - promql: Arc, /// The tree this node runs, shared by every partition's execution. tree: Arc, /// Each source's compiled raw schema, used to decode only that source's frames. @@ -2254,12 +2226,11 @@ impl FusedSeriesExec { /// [`RawLabelSource::Explicit`]. `time_column` and `tags` name the columns of the output, which /// is a row table rather than a frame. /// - /// The parameters stay separate because each is a fact its own caller has: the query, the plan + /// The parameters stay separate because each is a fact its own caller has: the plan /// below, the frame's label binding and ordering, the tree, and the three things the output /// schema is classified by. A struct over them would be the same list under another name. #[allow(clippy::too_many_arguments)] pub fn try_new( - promql: Arc, inputs: Vec<(Arc, RawLabelSource, SeriesOrdering)>, tree: Arc, output_schema: SchemaRef, @@ -2267,7 +2238,6 @@ impl FusedSeriesExec { tags: Vec, ) -> DataFusionResult { Self::build( - promql, inputs, tree, output_schema, @@ -2281,7 +2251,6 @@ impl FusedSeriesExec { #[allow(clippy::too_many_arguments)] fn build( - promql: Arc, inputs: Vec<(Arc, RawLabelSource, SeriesOrdering)>, tree: Arc, output_schema: SchemaRef, @@ -2557,7 +2526,6 @@ impl FusedSeriesExec { let metric = Arc::new(ExecutionPlanMetricsSet::new()); register_tree_metric(&tree, &metric); Ok(Self { - promql, tree, series_schemas: raw_schemas, source_kinds, @@ -2613,17 +2581,17 @@ impl std::fmt::Debug for FusedSeriesExec { } impl DisplayAs for FusedSeriesExec { - /// The query, then the tree it compiles to, one line per node. + /// The compiled tree, one line per node. /// /// Each node's own figures are read from the metrics this node registered, so a statement that /// never ran - every plain `EXPLAIN`, verbose included - draws the tree without them. That is /// the whole difference between `EXPLAIN` and `ANALYZE` here: nothing switches it on. fn fmt_as(&self, _t: DisplayFormatType, f: &mut Formatter) -> fmt::Result { - write!(f, "FusedSeriesExec: {}", self.promql)?; + write!(f, "FusedSeriesExec")?; // On the operator's own line, where the logical node also puts it: a second line would read // as another node beside the tree this one draws. if self.format == FusedOutputFormat::SeriesFrame { - write!(f, " (output: series_frame)")?; + write!(f, ": (output: series_frame)")?; } write!(f, "\n{}", self.tree.display_with(&self.node_metrics())) } @@ -2708,7 +2676,6 @@ impl ExecutionPlan for FusedSeriesExec { // Rebuilt rather than patched: the new child's schema is what the frame is validated // against, so a child that changed shape is refused instead of decoded with a stale layout. Ok(Arc::new(Self::build( - Arc::clone(&self.promql), children, Arc::clone(&self.tree), self.output_schema.clone(), diff --git a/src/promql/src/extension_plan/fused_series/tests.rs b/src/promql/src/extension_plan/fused_series/tests.rs index 279db772c18..28bb4526648 100644 --- a/src/promql/src/extension_plan/fused_series/tests.rs +++ b/src/promql/src/extension_plan/fused_series/tests.rs @@ -101,9 +101,7 @@ use crate::vector_match::{BinaryMatching, BinaryOp, Cardinality, FillValues, Mat use common_query::series_frame::series_frame_ordering; use common_query::series_frame::unpack_aggregate_state; -/// The query every tree here is compiled from. -const PROMQL: &str = "rate(x[5m])"; -/// The range the statement's window selects, which is the `5m` of [`PROMQL`]. +/// The range the statement's window selects, which is the `5m` of `rate(x[5m])`. const RANGE_MS: i64 = 300_000; /// The statement's grid: `rate(x[5m])` evaluated every minute from five minutes in. const GRID_START: i64 = 300_000; @@ -567,7 +565,6 @@ fn timestamp_match_tree() -> NativeTree { fn partition_disjoint(tree: NativeTree, input: Arc) -> bool { let node = FusedSeries::try_new( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema())], @@ -705,7 +702,6 @@ async fn scalar_binary_matching_key_is_not_routed_by_dropped_names() { let left_input = memory_exec_partitioned(vec![vec![frame(&left)], vec![]]); let right_input = memory_exec_partitioned(vec![vec![], vec![frame(&right)]]); let node = FusedSeries::try_new( - Arc::from("(x * 2) + on(__name__) (y * 2)"), tree.clone(), binding(&tree), vec![input_plan(frame_schema()), input_plan(frame_schema())], @@ -1027,7 +1023,6 @@ fn zone_node() -> FusedSeries { vec![HOST.to_string(), ZONE.to_string(), NAME.to_string()], ); FusedSeries::try_new( - Arc::from(PROMQL), tree, binding, vec![input_plan(frame_schema())], @@ -1397,7 +1392,6 @@ fn canonical_producer_node() -> FusedSeries { }) .unwrap(); FusedSeries::try_new_canonical_producer( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema())], @@ -1411,7 +1405,6 @@ fn fused_series_encoding_preserves_state_and_canonical_producer() { // capability, not a decision about where it may run. let tree = partial_tree(rate_tree(), AggregateKind::Sum, &[HOST, JOB]); let producer = FusedSeries::try_new_state_export( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema())], @@ -1440,7 +1433,6 @@ fn fused_series_ordinary_encoding_round_trip_and_rejects_unsupported_payloads() // `canonical_frame_schema` rather than with the plain row `output_schema`. let declared = canonical_frame_schema(&tree); let node = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree, original_binding.clone(), vec![input_plan(frame_schema())], @@ -1467,7 +1459,6 @@ fn fused_series_ordinary_encoding_round_trip_and_rejects_unsupported_payloads() canonical.field_with_name(TIME).unwrap().clone(), ])); let reordered = FusedSeries::try_new( - Arc::from(PROMQL), rows.tree().clone(), rows.binding().clone(), rows.children().to_vec(), @@ -1483,7 +1474,6 @@ fn fused_series_ordinary_encoding_round_trip_and_rejects_unsupported_payloads() // labels after a tree operation such as label rewriting. let empty_binding = binding_of(rows.tree(), raw_schema(), Vec::new()); let empty_tags = FusedSeries::try_new( - Arc::from(PROMQL), rate_tree(), empty_binding, vec![input_plan(frame_schema())], @@ -1525,20 +1515,18 @@ fn fused_series_ordinary_encoding_round_trip_and_rejects_unsupported_payloads() assert!(FusedSeries::deserialize(&trailing).is_err()); } -/// A payload written under another version is refused by its version, before its body is read. +/// A payload marked with an unsupported version is refused before its body is read. /// -/// The body here is the current payload's body with the version field set to a number this reader -/// does not accept: it decodes cleanly under the current reader, which is exactly what makes it a -/// control. A reader that checked the version only after decoding would accept it and hand back a -/// node whose fields were read at the wrong offsets; this asserts the version is what answers, by -/// name. +/// The body here is the current payload's body with the version field set to a future number this +/// reader does not accept: it decodes cleanly under the current reader, which makes it a control. +/// A reader that checked the version only after decoding would accept it and hand back a node whose +/// fields were read at the wrong offsets; this asserts the version is what answers, by name. #[test] -fn fused_series_envelope_refuses_the_older_payload_version_before_reading_it() { +fn fused_series_envelope_refuses_an_unsupported_payload_version() { let tree = rate_tree(); let declared = canonical_frame_schema(&tree); let bound = binding(&tree); let node = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree, bound, vec![input_plan(frame_schema())], @@ -1556,9 +1544,11 @@ fn fused_series_envelope_refuses_the_older_payload_version_before_reading_it() { // The control: the same body under the accepted version decodes. assert!(FusedSeries::deserialize(&bytes).is_ok()); - let mut old = bytes.clone(); - old[..4].copy_from_slice(&2u32.to_le_bytes()); - let message = FusedSeries::deserialize(&old).unwrap_err().to_string(); + let mut unsupported = bytes.clone(); + unsupported[..4].copy_from_slice(&2u32.to_le_bytes()); + let message = FusedSeries::deserialize(&unsupported) + .unwrap_err() + .to_string(); assert!( message.contains("version") && message.contains('2'), "the version must be what is reported, not a field read at the wrong offset: {message}" @@ -1574,7 +1564,6 @@ fn envelope_node_with_tags(tags: Vec) -> FusedSeries { let binding = binding_of(&tree, raw_schema(), tags); let output_schema = canonical_frame_schema(&tree); FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree, binding, vec![input_plan(frame_schema())], @@ -1614,7 +1603,6 @@ fn scalar_node(tree: NativeTree, value_name: &str) -> FusedSeries { Field::new(value_name, DataType::Float64, true), ])); FusedSeries::try_new( - Arc::from(PROMQL), tree, FusedSeriesBinding { grid: grid(), @@ -1702,7 +1690,6 @@ fn plain_rows_decode_rejects_invalid_schemas() { // Binding tags describe the source; the carried schema independently describes outputs. let empty_binding = binding_of(node.tree(), raw_schema(), Vec::new()); let cross_table = FusedSeries::try_new( - Arc::from(PROMQL), node.tree().clone(), empty_binding, node.children().to_vec(), @@ -1733,7 +1720,6 @@ fn a_non_default_column_order_survives_the_round_trip() { [("schema-origin".to_string(), "carried".to_string())].into(), ); let node = FusedSeries::try_new( - Arc::from(PROMQL), tree, binding, vec![input_plan(frame_schema())], @@ -1769,7 +1755,6 @@ fn series_frame_decode_rejects_schema_mismatch() { fn state_export_decode_rejects_schema_and_format_mismatch() { let tree = partial_tree(rate_tree(), AggregateKind::Sum, &[HOST, JOB]); let node = FusedSeries::try_new_state_export( - Arc::from(PROMQL), tree.clone(), binding_of(&tree, raw_schema(), vec![HOST.to_string(), JOB.to_string()]), vec![input_plan(frame_schema())], @@ -1986,7 +1971,6 @@ fn fused(tree: NativeTree) -> FusedSeries { output_schema() }; FusedSeries::try_new( - Arc::from(PROMQL), tree, binding, vec![input_plan(frame_schema())], @@ -2450,13 +2434,16 @@ impl Display for Explain<'_> { } } -/// The logical node prints the query a reader wrote, not the tree it compiled to. +/// The logical node prints the compiled tree without metrics. #[test] -fn fmt_for_explain_prints_the_promql() { - assert_eq!( - "FusedSeries: rate(x[5m])", - Explain(&fused(rate_tree())).to_string() +fn fmt_for_explain_prints_the_native_tree() { + let text = Explain(&fused(rate_tree())).to_string(); + assert!( + text.starts_with("FusedSeries\n Window: rate [300000ms]"), + "{text}" ); + assert!(text.contains(" Input: raw samples source=0"), "{text}"); + assert!(!text.contains("metrics="), "{text}"); } /// The physical node prints the tree it was built from, rendered by the tree's own display. @@ -2465,10 +2452,12 @@ fn fmt_as_prints_the_native_tree() { let exec = plan(rate_tree(), memory_exec(Vec::new())).expect("the runtime executes a rate tree"); let text = displayable(exec.as_ref()).indent(false).to_string(); + println!("=== physical_plan ===\n{text}"); + assert!(!text.contains("metrics="), "{text}"); assert!( - text.starts_with("FusedSeriesExec: rate(x[5m])"), - "the node line names the query first: {text}" + text.starts_with("FusedSeriesExec\n Window: rate"), + "the node line precedes its tree: {text}" ); // A distinctive line of the compiled tree: the window's bound range, offset, anchor and grid. assert!( @@ -2481,6 +2470,24 @@ fn fmt_as_prints_the_native_tree() { ); } +/// Physical ANALYZE renders the figures registered on the tree nodes that recorded them. +#[tokio::test] +async fn fmt_as_reports_node_metrics_after_execution() { + let exec = plan(rate_tree(), memory_exec(two_series_frames())).unwrap(); + collect(Arc::clone(&exec), SessionContext::default().task_ctx()) + .await + .unwrap(); + let text = displayable(exec.as_ref()).indent(false).to_string(); + println!("=== physical_plan after execution ===\n{text}"); + assert!( + text.starts_with("FusedSeriesExec\n Window: rate"), + "{text}" + ); + assert!(text.contains("metrics=["), "{text}"); + assert!(text.contains("output_values:"), "{text}"); + assert!(text.contains("Input: raw samples source=0"), "{text}"); +} + /// `rate(x[5m])` runs end to end over two raw series: values, tags and the dropped metric name. #[tokio::test] async fn runs_rate_over_two_series() { @@ -2573,7 +2580,6 @@ fn a_tree_bound_to_another_grid_is_rejected() { ..binding(&tree) }; let error = FusedSeries::try_new( - Arc::from(PROMQL), tree, binding, vec![input_plan(frame_schema())], @@ -2604,7 +2610,6 @@ fn a_separately_allocated_grid_with_an_equal_spec_is_accepted() { ..binding(&tree) }; FusedSeries::try_new( - Arc::from(PROMQL), tree, binding, vec![input_plan(frame_schema())], @@ -2625,7 +2630,6 @@ fn scalar_logical_node_has_no_children_and_refuses_a_child() { tags: vec![HOST.to_string()], }; let node = FusedSeries::try_new( - Arc::from("vector(7)"), tree.clone(), binding.clone(), Vec::new(), @@ -2636,7 +2640,6 @@ fn scalar_logical_node_has_no_children_and_refuses_a_child() { assert!(node.inputs().is_empty()); assert!( FusedSeries::try_new( - Arc::from("vector(7)"), tree, binding, vec![input_plan(frame_schema())], @@ -2654,7 +2657,6 @@ fn logical_children_and_inputs_roundtrip_in_source_order() { let left = input_plan_with_rows(frame_schema(), false); let right = input_plan_with_rows(frame_schema(), true); let node = FusedSeries::try_new( - Arc::from("left + right"), tree, binding(&rate_tree()), vec![left.clone(), right.clone()], @@ -2722,7 +2724,6 @@ fn logical_source_indexes_must_match_dense_children() { ] { assert!( FusedSeries::try_new( - Arc::from(name), tree.clone(), binding(&tree), children, @@ -2741,7 +2742,6 @@ fn repeated_source_bindings_validate_fields_and_second_ordering() { binary_tree_over_sources(BinaryOp::Add, raw_schema(), two_field_schema(), 0, 0); assert!( FusedSeries::try_new( - Arc::from("x + x"), fields_mismatch.clone(), binding(&fields_mismatch), vec![input_plan(frame_schema())], @@ -2897,7 +2897,6 @@ fn fragmented_raw_multi_input_is_reconciled_before_matching_router() { use common_query::series_frame::frame_schema_with; let tree = binary_tree_over_sources(BinaryOp::Add, raw_schema(), raw_schema(), 0, 1); let node = FusedSeries::try_new( - Arc::from("left + right"), tree.clone(), binding(&tree), vec![input_plan(frame_schema()), input_plan(frame_schema())], @@ -2955,7 +2954,6 @@ fn fragmented_raw_multi_input_is_reconciled_before_matching_router() { async fn logical_two_source_tree_is_accepted_by_physical_adapter() { let tree = binary_tree_over_sources(BinaryOp::Add, raw_schema(), raw_schema(), 0, 1); let node = FusedSeries::try_new( - Arc::from("left + right"), tree.clone(), binding(&tree), vec![input_plan(frame_schema()), input_plan(frame_schema())], @@ -3035,7 +3033,6 @@ fn binary_identity_rows(batches: &[RecordBatch]) -> Vec<(Option, i64, Op async fn matching_parallelism_uses_configured_width_without_changing_rows() { let tree = binary_tree_over_sources(BinaryOp::Add, raw_schema(), raw_schema(), 0, 1); let node = FusedSeries::try_new( - Arc::from("left + right"), tree.clone(), binding(&tree), vec![input_plan(frame_schema()), input_plan(frame_schema())], @@ -3148,7 +3145,6 @@ async fn independent_sources_execute_all_partitions() { calls: Arc::clone(&right_calls), }); let node = FusedSeries::try_new( - Arc::from("left - right"), tree.clone(), binding(&tree), vec![ @@ -3221,7 +3217,6 @@ async fn both_inputs_are_read_concurrently() { } }); let node = FusedSeries::try_new( - Arc::from("left - right"), tree.clone(), binding(&tree), vec![input_plan(frame_schema()), input_plan(frame_schema())], @@ -3258,7 +3253,6 @@ async fn both_inputs_are_read_concurrently() { async fn binary_output_tags_are_projected_but_raw_binding_is_full() { let tree = binary_tree_over_sources(BinaryOp::Sub, raw_schema(), raw_schema(), 0, 1); let node = FusedSeries::try_new( - Arc::from("left - right"), tree.clone(), binding(&tree), vec![input_plan(frame_schema()), input_plan(frame_schema())], @@ -3303,7 +3297,6 @@ async fn binary_output_tags_are_projected_but_raw_binding_is_full() { async fn empty_first_source_does_not_end_gather_early() { let tree = binary_tree_over_sources(BinaryOp::Sub, raw_schema(), raw_schema(), 0, 1); let node = FusedSeries::try_new( - Arc::from("left - right"), tree.clone(), binding(&tree), vec![input_plan(frame_schema()), input_plan(frame_schema())], @@ -3339,7 +3332,6 @@ fn multiparty_gather_rejects_key_then_time_leaf_requirement() { &[SeriesOrdering::KeyThenTime, SeriesOrdering::Contiguous], ); let node = FusedSeries::try_new( - Arc::from("left + right"), tree.clone(), binding(&tree), vec![input_plan(frame_schema()), input_plan(frame_schema())], @@ -3368,7 +3360,6 @@ fn multiparty_gather_rejects_key_then_time_leaf_requirement() { async fn later_source_error_drops_all_gathered_inputs() { let tree = binary_tree_over_sources(BinaryOp::Sub, raw_schema(), raw_schema(), 0, 1); let node = FusedSeries::try_new( - Arc::from("left - right"), tree.clone(), binding(&tree), vec![input_plan(frame_schema()), input_plan(frame_schema())], @@ -3538,7 +3529,6 @@ async fn physical_plan_accepts_scalar_root_without_sources() { ), ])); let node = FusedSeries::try_new( - Arc::from("4"), tree, binding.clone(), Vec::new(), @@ -3667,7 +3657,6 @@ async fn physical_scalar_rejects_malformed_output_schema() { ]; for schema in schemas { let node = FusedSeries::try_new( - Arc::from("4"), tree.clone(), binding.clone(), Vec::new(), @@ -3677,7 +3666,6 @@ async fn physical_scalar_rejects_malformed_output_schema() { assert!(node.to_execution_plan(Vec::new()).is_err()); } let tagged = FusedSeries::try_new( - Arc::from("4"), tree, FusedSeriesBinding { tags: vec![HOST.to_string()], @@ -3709,14 +3697,9 @@ async fn physical_scalar_conversion_preserves_special_float_bits() { false, ), ])); - let node = FusedSeries::try_new( - Arc::from("4"), - tree, - binding, - Vec::new(), - schema.to_dfschema_ref().unwrap(), - ) - .unwrap(); + let node = + FusedSeries::try_new(tree, binding, Vec::new(), schema.to_dfschema_ref().unwrap()) + .unwrap(); let exec = node.to_execution_plan(Vec::new()).unwrap(); let batches = exec .execute(0, SessionContext::default().task_ctx()) @@ -4492,7 +4475,6 @@ fn unequal_binding_and_output_schema_are_not_ordered_equal() { ), ])); let right_node = FusedSeries::try_new( - Arc::from(PROMQL), tree, binding(&rate_tree()), vec![input_plan(frame_schema())], @@ -4627,7 +4609,7 @@ async fn a_later_repeated_source_enqueue_failure_releases_the_input() { assert_eq!(batches_at_failure, pulled.load(Ordering::SeqCst)); } -/// Logical equality distinguishes source field layouts, even when the PromQL and output match. +/// Logical equality distinguishes source field layouts even when the binding and output match. #[test] fn logical_identity_includes_native_source_field_order() { let ordered_schema = two_field_schema(); @@ -4639,7 +4621,6 @@ fn logical_identity_includes_native_source_field_order() { let make_node = |schema: Arc| { let tree = rate_tree_of(schema); FusedSeries::try_new( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema())], @@ -4649,7 +4630,6 @@ fn logical_identity_includes_native_source_field_order() { }; let left = make_node(Arc::clone(&ordered_schema)); let right = make_node(Arc::new(reversed_schema)); - assert_eq!(left.promql(), right.promql()); assert_eq!(left.binding(), right.binding()); assert_ne!(left, right); assert_ne!(Some(std::cmp::Ordering::Equal), left.partial_cmp(&right)); @@ -4954,7 +4934,6 @@ fn partial_node(kind: AggregateKind) -> FusedSeries { let tree = partial_tree(tree, kind, &[HOST]); let schema = canonical_state_frame_schema(&tree); FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding_of(&tree, raw_schema(), vec![HOST.to_string()]), vec![input_plan(frame_schema())], @@ -4975,7 +4954,6 @@ fn partial_plan( ) -> DataFusionResult> { let tree = partial_tree(tree, op.kind, tags); FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding_of( &tree, @@ -5044,7 +5022,6 @@ async fn decode_state_batches(batches: Vec) -> Vec { async fn a_state_export_keys_its_labels_by_the_cut_aggregate_grouping() { let tree = partial_tree(rate_tree(), AggregateKind::Sum, &[JOB]); let node = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), // The scan is bound with `host`; the aggregate groups by `job`. binding_of(&tree, raw_schema(), vec![HOST.to_string()]), @@ -5402,7 +5379,6 @@ async fn mixed_raw_and_evaluated_inputs_execute_with_source_local_decoders() { ) .unwrap(); let exec = FusedSeries::try_new( - Arc::from("mixed raw/evaluated"), tree.clone(), binding_of( &tree, @@ -5516,7 +5492,6 @@ fn final_aggregate_state_input_constructs_with_evaluated_schema() { let tree = state_tree(AggregateStage::Final, NativeType::AggregateState(state)); let schema = canonical_frame_schema(&tree); let node = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema())], @@ -5538,7 +5513,6 @@ fn partial_aggregate_declares_state_frame_schema() { ); let state_schema = canonical_state_frame_schema(&tree); let node = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema())], @@ -5552,7 +5526,6 @@ fn partial_aggregate_declares_state_frame_schema() { ); let error = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema())], @@ -5632,7 +5605,6 @@ fn repeated_source_with_state_and_samples_is_refused() { async fn series_frame_output_uses_evaluated_codec_schema() { let tree = rate_tree(); let frame_node = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema())], @@ -5649,7 +5621,6 @@ async fn series_frame_output_uses_evaluated_codec_schema() { "frame schema comes from the codec rather than the row output" ); let wrong_schema = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema())], @@ -5674,7 +5645,6 @@ async fn series_frame_output_uses_evaluated_codec_schema() { fn series_frame_accepts_multiple_sources_with_equal_partition_counts() { let tree = binary_tree_over_sources(BinaryOp::Add, raw_schema(), raw_schema(), 0, 1); let node = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema()), input_plan(frame_schema())], @@ -5713,7 +5683,6 @@ async fn series_frame_preserves_four_input_partitions() { .collect::>(); let tree = rate_tree(); let node = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema())], @@ -5764,7 +5733,6 @@ async fn series_frame_rate_matches_plain_rows_bitwise() { ); let tree = rate_tree(); let node = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema())], @@ -5823,7 +5791,6 @@ async fn series_frame_rate_matches_plain_rows_bitwise() { async fn partial_empty_input_yields_valid_empty_state_frame_stream() { let tree = partial_tree(rate_tree(), AggregateKind::Sum, &[HOST]); let node = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema())], @@ -5846,7 +5813,6 @@ async fn partial_root_requires_frame_output() { let prefix = rate_tree(); let tree = partial_tree(prefix, AggregateKind::Sum, &[HOST]); let error = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema())], @@ -5859,18 +5825,16 @@ async fn partial_root_requires_frame_output() { /// `EXPLAIN` says nothing extra about the row table and names the frame output. /// -/// The row table is what a query returns, so its node line has to stay exactly what it was; a frame -/// output answers something else, and the line is the only place a reader can see that. +/// The row-table header has no parameters; a frame output names its format in the header. #[test] fn the_explain_text_names_the_frame_format_only() { let rows = fused(rate_tree()); - assert_eq!( - "FusedSeries: rate(x[5m])", - Explain(&rows).to_string(), - "the row-table line is what it always was" + assert!( + Explain(&rows) + .to_string() + .starts_with("FusedSeries\n Window:") ); let framed = FusedSeries::try_new_with_format( - Arc::from(PROMQL), rows.tree().clone(), rows.binding().clone(), rows.inputs().into_iter().cloned().collect(), @@ -5881,16 +5845,30 @@ fn the_explain_text_names_the_frame_format_only() { assert!( Explain(&framed) .to_string() - .contains("(output: series_frame)"), + .starts_with("FusedSeries: (output: series_frame)\n Window:"), "a frame output names its format: {}", Explain(&framed) ); + let framed_exec = plan_of(framed, memory_exec(Vec::new())).unwrap(); + let text = displayable(framed_exec.as_ref()).indent(false).to_string(); + assert!( + text.starts_with("FusedSeriesExec: (output: series_frame)\n Window:"), + "{text}" + ); + let canonical = canonical_producer_node(); + let text = Explain(&canonical).to_string(); + assert!( + text.starts_with( + "FusedSeries: (output: series_frame, per-partition canonical merge)\n Merge:" + ), + "{text}" + ); let plain = plan(rate_tree(), memory_exec(Vec::new())).expect("the runtime executes the tree"); let text = displayable(plain.as_ref()).indent(false).to_string(); assert!( - text.starts_with("FusedSeriesExec: rate(x[5m])\n"), - "the row-table node line is unchanged: {text}" + text.starts_with("FusedSeriesExec\n Window:"), + "the row-table node line has no parameters: {text}" ); } @@ -5905,7 +5883,6 @@ fn the_logical_schema_follows_the_output_format() { assert_eq!(&output_schema(), rows.schema().inner()); let framed = FusedSeries::try_new_with_format( - Arc::from(PROMQL), rows.tree().clone(), rows.binding().clone(), rows.inputs().into_iter().cloned().collect(), @@ -5934,7 +5911,6 @@ fn the_logical_schema_follows_the_output_format() { fn rebuilding_a_frame_node_keeps_its_output_format() { let rows = fused(rate_tree()); let framed = FusedSeries::try_new_with_format( - Arc::from(PROMQL), rows.tree().clone(), rows.binding().clone(), rows.inputs().into_iter().cloned().collect(), @@ -5971,7 +5947,6 @@ fn a_series_frame_refuses_a_root_it_cannot_carry() { tags: vec![HOST.to_string(), NAME.to_string()], }; let error = FusedSeries::try_new_with_format( - Arc::from(PROMQL), scalar, binding, Vec::new(), @@ -6026,7 +6001,6 @@ async fn a_pending_input_does_not_hold_the_rows_a_frame_buffered() { }); let tree = rate_tree(); let node = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(schema)], @@ -6132,7 +6106,6 @@ async fn fused_frame_input_error_terminates_and_releases_input() { }); let tree = rate_tree(); let node = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(schema)], @@ -6182,7 +6155,6 @@ async fn a_frame_carries_the_closing_chunk_a_row_table_drops() { ); let tree = rate_tree(); let node = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema())], @@ -6267,7 +6239,6 @@ async fn two_chains_that_close_on_one_label_set_each_keep_their_frame() { ]; let tree = rate_tree(); let node = FusedSeries::try_new_with_format( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema())], @@ -6416,7 +6387,6 @@ fn duplicate_evidence_partitioned_exec( let tree = NativeTree::try_new(root).expect("differently bound instant selectors form a valid tree"); let node = FusedSeries::try_new( - Arc::from(PROMQL), tree.clone(), binding(&tree), vec![input_plan(frame_schema()), input_plan(frame_schema())], @@ -6637,7 +6607,6 @@ fn histogram_over_sum_splits_and_declares_complete_bucket_routing() { if op.stage == AggregateStage::Final && op.group == group )); let node = FusedSeries::try_new_with_format( - Arc::from("histogram over sum"), final_tree.clone(), binding(&final_tree), vec![input_plan( @@ -6652,8 +6621,8 @@ fn histogram_over_sum_splits_and_declares_complete_bucket_routing() { }); let display = logical.display_indent().to_string(); assert!( - display.lines().next().unwrap().ends_with( - " (state routing: canonical labels except [\"le\", \"__name__\", \"__type__\", \"__unit__\"])" + display.lines().next().unwrap().starts_with( + "FusedSeries: (output: series_frame) (state routing: canonical labels except [\"le\", \"__name__\", \"__type__\", \"__unit__\"])" ), "{display}" ); @@ -6802,7 +6771,6 @@ fn histogram_final_replaces_incompatible_state_router() { let tree = histogram_over_sum_tree(group, HistogramFunction::Quantile(0.5)); let (partial, final_tree) = FusedSeries::split_aggregate(&tree, TIME).unwrap().unwrap(); let partial_node = FusedSeries::try_new_state_export( - Arc::from("partial"), partial.clone(), binding(&partial), vec![input_plan(frame_schema())], @@ -6824,7 +6792,6 @@ fn histogram_final_replaces_incompatible_state_router() { .unwrap(), ) as Arc; let final_node = FusedSeries::try_new_with_format( - Arc::from("final"), final_tree.clone(), binding(&final_tree), vec![input_plan( @@ -6910,7 +6877,6 @@ async fn histogram_split_routes_whole_state_batches_and_preserves_each_bound() { ]), ]; let partial_node = FusedSeries::try_new_state_export( - Arc::from("partial"), partial.clone(), binding(&partial), vec![input_plan(frame_schema())], @@ -6951,7 +6917,6 @@ async fn histogram_split_routes_whole_state_batches_and_preserves_each_bound() { ); } let final_node = FusedSeries::try_new_with_format( - Arc::from("final"), final_tree.clone(), binding(&final_tree), vec![input_plan( @@ -7096,7 +7061,6 @@ async fn histogram_split_colocates_metadata_distinct_results_before_their_merge( } } let partial_node = FusedSeries::try_new_state_export( - Arc::from("partial"), partial.clone(), binding(&partial), vec![input_plan(frame_schema())], @@ -7110,7 +7074,6 @@ async fn histogram_split_colocates_metadata_distinct_results_before_their_merge( )]) .unwrap(); let final_node = FusedSeries::try_new_with_format( - Arc::from("final"), final_tree.clone(), binding(&final_tree), vec![input_plan( diff --git a/src/promql/src/extension_plan/fused_series/tests/frame_output.rs b/src/promql/src/extension_plan/fused_series/tests/frame_output.rs index ee7f59402e2..88faa9580ee 100644 --- a/src/promql/src/extension_plan/fused_series/tests/frame_output.rs +++ b/src/promql/src/extension_plan/fused_series/tests/frame_output.rs @@ -158,7 +158,6 @@ async fn sparse_instant_frame_preserves_positions_nulls_and_nan_across_formats() let binding = binding_of(&tree, Arc::clone(&schema), vec![HOST.into(), NAME.into()]); let make_node = |format| { FusedSeries::try_new_with_format( - Arc::from("instant(x)"), tree.clone(), binding.clone(), vec![input_plan(frame_schema_of(&schema))], @@ -217,7 +216,6 @@ async fn abs_rate_frame_matches_plain_rows_with_exact_values_and_labels() { let binding = binding(&tree); let make_node = |format| { FusedSeries::try_new_with_format( - Arc::from("abs(rate(x[5m]))"), tree.clone(), binding.clone(), vec![input_plan(frame_schema())], @@ -321,7 +319,6 @@ async fn rewritten_identical_labels_keep_disjoint_chains_in_one_encoder() { let binding = binding_of(&tree, Arc::clone(&schema), vec![HOST.into(), NAME.into()]); let make_node = |format| { FusedSeries::try_new_with_format( - Arc::from("rewritten instant"), tree.clone(), binding.clone(), vec![input_plan(frame_schema())], diff --git a/src/promql/src/extension_plan/series_repartition/tests.rs b/src/promql/src/extension_plan/series_repartition/tests.rs index d374594144b..25a8aaf065f 100644 --- a/src/promql/src/extension_plan/series_repartition/tests.rs +++ b/src/promql/src/extension_plan/series_repartition/tests.rs @@ -2418,7 +2418,6 @@ async fn fused_instant_complete_disjoint_publishes_before_other_input_eof() { .unwrap(); let output_frame_schema = FusedSeries::series_frame_schema(&tree).unwrap(); let logical = FusedSeries::try_new_with_format( - Arc::from("m"), tree, FusedSeriesBinding { grid: Arc::clone(&grid), diff --git a/src/promql/src/native/display.rs b/src/promql/src/native/display.rs index 9cca1fe1dab..a3cb7b21681 100644 --- a/src/promql/src/native/display.rs +++ b/src/promql/src/native/display.rs @@ -519,7 +519,7 @@ impl NativeTree { report_node(self.root(), ROOT_PATH, metrics) } - /// This tree rendered under the operator line that names the query, with each node's own + /// This tree rendered under its operator line, with each node's own /// figures when `metrics` holds them. pub fn display_with<'a>(&'a self, metrics: &'a NodeMetricsSnapshot) -> NativeTreeDisplay<'a> { NativeTreeDisplay { diff --git a/src/promql/src/native/plan/tests.rs b/src/promql/src/native/plan/tests.rs index 32c6a9dc54d..f3b14ada135 100644 --- a/src/promql/src/native/plan/tests.rs +++ b/src/promql/src/native/plan/tests.rs @@ -1589,14 +1589,8 @@ fn a_scalar_root_claiming_samples_is_refused() { metric_name: None, }; - let error = FusedSeries::try_new( - Arc::from("1"), - tree, - binding, - Vec::new(), - Arc::new(DFSchema::empty()), - ) - .expect_err("a samples root reads no source while reporting samples"); + let error = FusedSeries::try_new(tree, binding, Vec::new(), Arc::new(DFSchema::empty())) + .expect_err("a samples root reads no source while reporting samples"); assert!( error.to_string().contains("does not produce a scalar"), @@ -1614,14 +1608,8 @@ fn a_samples_root_without_an_input_is_refused() { tags: vec![], metric_name: None, }; - let error = FusedSeries::try_new( - Arc::from("some_metric"), - tree, - binding, - Vec::new(), - Arc::new(DFSchema::empty()), - ) - .expect_err("a samples root without an input is refused"); + let error = FusedSeries::try_new(tree, binding, Vec::new(), Arc::new(DFSchema::empty())) + .expect_err("a samples root without an input is refused"); assert!(error.to_string().contains("source indexes"), "{error}"); } diff --git a/src/query/src/analyze.rs b/src/query/src/analyze.rs index 572b6008a17..e66e3be1c7b 100644 --- a/src/query/src/analyze.rs +++ b/src/query/src/analyze.rs @@ -463,6 +463,15 @@ impl Display for JsonMetrics { #[cfg(test)] mod tests { use datafusion::physical_plan::empty::EmptyExec; + use datatypes::series::SeriesOrdering; + use datatypes::series::frame::{FrameLabelMode, raw_series_frame_schema}; + use datatypes::series::samples::{FieldSpec, SeriesSchema}; + use datatypes::series::time::{EvalGrid, GridSpec}; + use promql::extension_plan::{FusedSeries, FusedSeriesBinding}; + use promql::native::plan::{ + BoundInput, BoundInstant, InstantValue, NativeExpr, NativeNode, NativeTree, NativeType, + }; + use store_api::region_engine::RawLabelSource; use super::*; @@ -571,49 +580,115 @@ mod tests { /// another node, which is the only place a fused operator runs in a standalone server. #[test] fn qbs_analyze_json_nests_the_native_tree_below_its_operator() { - fn node( - line: &str, - metrics: &[(&str, usize)], - children: Vec, - ) -> NativeTreeMetrics { - NativeTreeMetrics { - line: line.to_string(), - metrics: metrics - .iter() - .map(|(name, value)| (name.to_string(), *value)) - .collect(), - children, - } - } - let metrics = JsonMetrics::from_record_batch_metrics(RecordBatchMetrics { - plan_metrics: vec![PlanMetrics { - plan: "FusedSeriesExec: rate(x[5m])\n Window: rate".to_string(), - plan_name: "FusedSeriesExec".to_string(), - level: 0, - metrics: vec![], - native: Some(node( - "Window: rate [300000ms] offset 0 anchor -", - &[("output_values", 3)], - vec![node("Input: raw samples source=0", &[], vec![])], - )), - }], - ..Default::default() + let grid = Arc::new(EvalGrid::try_new(GridSpec::try_new(0, 1_000, 2).unwrap()).unwrap()); + let schema = Arc::new(SeriesSchema { + fields: Arc::from([FieldSpec { + name: Arc::from("value"), + }]), + eval_grid: None, }); - let value = serde_json::to_value(metrics).unwrap(); + let raw = NativeType::Samples(Arc::clone(&schema)); + let evaluated = NativeType::Samples(Arc::new(SeriesSchema { + fields: schema.fields.clone(), + eval_grid: Some(Arc::clone(&grid)), + })); + let tree = NativeTree::try_new(NativeNode { + expr: NativeExpr::Instant { + input: Box::new(NativeNode { + expr: NativeExpr::Input(BoundInput { + source_index: 0, + output: raw.clone(), + ordering: SeriesOrdering::Contiguous, + }), + output: raw, + }), + op: BoundInstant { + lookback_ms: 0, + offset_ms: 0, + at_ms: None, + value: InstantValue::Sample, + output_grid: Arc::clone(&grid), + }, + }, + output: evaluated, + }) + .unwrap(); + let frame_schema = raw_series_frame_schema(&schema, FrameLabelMode::Explicit).unwrap(); + let child = datafusion::logical_expr::LogicalPlan::EmptyRelation( + datafusion::logical_expr::EmptyRelation { + produce_one_row: false, + schema: Arc::new( + datafusion_common::DFSchema::try_from(frame_schema.clone()).unwrap(), + ), + }, + ); + let binding = FusedSeriesBinding { + grid, + metric_name: None, + time_index: "time".to_string(), + tags: vec![], + }; + let rows = Arc::new(Schema::new(vec![ + Field::new("value", DataType::Float64, true), + Field::new( + "time", + DataType::Timestamp(arrow_schema::TimeUnit::Millisecond, None), + false, + ), + ])); + let frame = FusedSeries::series_frame_schema(&tree).unwrap(); + for (output, format, header, param) in [ + ( + rows, + promql::extension_plan::FusedOutputFormat::PlainRows, + "FusedSeriesExec", + "", + ), + ( + frame, + promql::extension_plan::FusedOutputFormat::SeriesFrame, + "FusedSeriesExec: (output: series_frame)", + "(output: series_frame)", + ), + ] { + let fused = FusedSeries::try_new_with_format( + tree.clone(), + binding.clone(), + vec![child.clone()], + Arc::new(datafusion_common::DFSchema::try_from(output).unwrap()), + format, + ) + .unwrap(); + let exec = fused + .to_execution_plan(vec![( + Arc::new(EmptyExec::new(frame_schema.clone())), + RawLabelSource::Explicit, + SeriesOrdering::Contiguous, + )]) + .unwrap(); + let mut collector = MetricCollector::new(false); + accept(exec.as_ref(), &mut collector).unwrap(); + let mut collected = collector.record_batch_metrics; + collected.plan_metrics.truncate(1); + let plan = &collected.plan_metrics[0]; + assert_eq!(plan.plan.lines().next().unwrap(), header, "{}", plan.plan); + assert!(plan.native.is_some(), "the native tree must be collected"); + let metrics = JsonMetrics::from_record_batch_metrics(collected); + let value = serde_json::to_value(metrics).unwrap(); - assert_eq!(value["name"], "FusedSeriesExec"); - assert_eq!(value["param"], "rate(x[5m])"); - assert_eq!(value["children"][0]["name"], "Window"); - assert_eq!( - value["children"][0]["param"], - "rate [300000ms] offset 0 anchor -" - ); - assert_eq!(value["children"][0]["metrics"]["output_values"], 3); - assert_eq!(value["children"][0]["children"][0]["name"], "Input"); - assert_eq!( - value["children"][0]["children"][0]["param"], - "raw samples source=0" - ); + assert_eq!(value["name"], "FusedSeriesExec"); + assert_eq!(value["param"], param); + assert_eq!(value["children"][0]["name"], "Instant"); + assert_eq!( + value["children"][0]["param"], + "lookback 0ms offset 0 on grid 0/1000x2" + ); + assert_eq!(value["children"][0]["children"][0]["name"], "Input"); + assert_eq!( + value["children"][0]["children"][0]["param"], + "raw samples source=0, contiguous" + ); + } } #[test] diff --git a/src/query/src/dist_plan/analyzer/test.rs b/src/query/src/dist_plan/analyzer/test.rs index 033384474dc..01738db9536 100644 --- a/src/query/src/dist_plan/analyzer/test.rs +++ b/src/query/src/dist_plan/analyzer/test.rs @@ -339,7 +339,6 @@ fn hybrid_fused(scan: LogicalPlan, metric: &str) -> LogicalPlan { }); let tree = compile(&stmt, series, SeriesOrdering::KeyThenTime).unwrap(); let fused = FusedSeries::try_new( - Arc::from(metric), tree, FusedSeriesBinding { grid: statement_grid(&stmt).unwrap(), @@ -424,7 +423,12 @@ fn hybrid_binary_join_rewrites_each_native_operand_boundary() { .as_any() .downcast_ref::() .expect("MergeScan remote input must be a FusedSeries root"); - assert_eq!(expected_table, fused.promql()); + assert!( + fused + .tree() + .to_string() + .contains("Input: raw samples source=0") + ); assert_eq!(1, fused.children().len()); let [LogicalPlan::TableScan(table_scan)] = fused.children() else { panic!("FusedSeries operand must directly consume a TableScan: {fused:?}"); diff --git a/src/query/src/dist_plan/commutativity.rs b/src/query/src/dist_plan/commutativity.rs index 4d727bae0e2..3e70952cd45 100644 --- a/src/query/src/dist_plan/commutativity.rs +++ b/src/query/src/dist_plan/commutativity.rs @@ -551,14 +551,7 @@ mod tests { ); LogicalPlan::Extension(Extension { node: Arc::new( - FusedSeries::try_new( - Arc::from("some_metric{host=\"left\"} + on(host) some_metric{host=\"right\"}"), - tree, - binding, - vec![left, right], - output_schema, - ) - .unwrap(), + FusedSeries::try_new(tree, binding, vec![left, right], output_schema).unwrap(), ), }) } @@ -605,14 +598,7 @@ mod tests { LogicalPlan::Extension(Extension { node: Arc::new( - FusedSeries::try_new( - Arc::from("some_metric"), - tree, - binding, - vec![input], - output_schema, - ) - .unwrap(), + FusedSeries::try_new(tree, binding, vec![input], output_schema).unwrap(), ), }) } @@ -1118,15 +1104,9 @@ mod tests { let (partial_tree, final_tree) = FusedSeries::split_aggregate(&tree, "ts").unwrap().unwrap(); - let partial = FusedSeries::try_new_state_export( - Arc::from("sum(some_metric)"), - partial_tree, - binding.clone(), - vec![input], - ) - .unwrap(); + let partial = + FusedSeries::try_new_state_export(partial_tree, binding.clone(), vec![input]).unwrap(); let final_node = FusedSeries::try_new( - Arc::from("sum(some_metric)"), final_tree, binding, vec![LogicalPlan::Extension(Extension { diff --git a/src/query/src/optimizer/scan_hint.rs b/src/query/src/optimizer/scan_hint.rs index 572d5efd31d..1108d44c44c 100644 --- a/src/query/src/optimizer/scan_hint.rs +++ b/src/query/src/optimizer/scan_hint.rs @@ -1611,14 +1611,7 @@ mod test { LogicalPlan::Extension(Extension { node: Arc::new( - FusedSeries::try_new( - Arc::from("some_metric"), - tree, - binding, - vec![input], - output_schema, - ) - .unwrap(), + FusedSeries::try_new(tree, binding, vec![input], output_schema).unwrap(), ), }) } diff --git a/src/query/src/optimizer/series_frame_output.rs b/src/query/src/optimizer/series_frame_output.rs index 2684d8f0b5f..4a30b50cc98 100644 --- a/src/query/src/optimizer/series_frame_output.rs +++ b/src/query/src/optimizer/series_frame_output.rs @@ -373,7 +373,6 @@ mod tests { .expect("the frame is a legal schema"), ); let fused = FusedSeries::try_new_with_format( - Arc::from("cpu"), tree, FusedSeriesBinding { grid, diff --git a/src/query/src/promql/native_compile/tests.rs b/src/query/src/promql/native_compile/tests.rs index 4643ec80b70..92eaaa5a96d 100644 --- a/src/query/src/promql/native_compile/tests.rs +++ b/src/query/src/promql/native_compile/tests.rs @@ -1286,7 +1286,7 @@ fn the_supported_math_functions_map_to_their_kernels() { let tree = compile(&eval_stmt(query), raw_schema(), SeriesOrdering::KeyThenTime) .unwrap_or_else(|error| panic!("`{query}` was refused: {error}")); let display = tree.to_string(); - let root = display.lines().nth(1).expect("a compiled tree has a root"); + let root = display.lines().next().expect("a compiled tree has a root"); assert_eq!( root, format!(" Map: {op}"), @@ -2852,7 +2852,7 @@ fn the_aggregates_that_are_not_a_fold_bind_a_vector_operation() { let tree = compile(&eval_stmt(query), raw_schema(), SeriesOrdering::KeyThenTime) .unwrap_or_else(|error| panic!("`{query}` was refused: {error}")); let display = tree.to_string(); - let root = display.lines().nth(1).expect("a compiled tree has a root"); + let root = display.lines().next().expect("a compiled tree has a root"); assert_eq!(root, format!(" Vector: {op}"), "`{query}`"); } @@ -4286,7 +4286,7 @@ fn whole_vector_functions_bind_a_vector_operation() { let tree = compile(&eval_stmt(query), raw_schema(), SeriesOrdering::KeyThenTime) .unwrap_or_else(|error| panic!("`{query}` was refused: {error}")); let display = tree.to_string(); - let root = display.lines().nth(1).expect("a compiled tree has a root"); + let root = display.lines().next().expect("a compiled tree has a root"); assert_eq!(root, format!(" Vector: {op}"), "`{query}`"); } } @@ -4829,7 +4829,7 @@ fn label_functions_bind_their_change_into_a_map() { let tree = compile(&eval_stmt(query), raw_schema(), SeriesOrdering::KeyThenTime) .unwrap_or_else(|error| panic!("`{query}` was refused: {error}")); let display = tree.to_string(); - let root = display.lines().nth(1).expect("a compiled tree has a root"); + let root = display.lines().next().expect("a compiled tree has a root"); assert_eq!(root, format!(" Map: {op}"), "`{query}`"); } } @@ -4852,7 +4852,7 @@ fn histogram_functions_bind_their_kernel_into_a_map() { ) .unwrap_or_else(|error| panic!("`{query}` was refused: {error}")); let display = tree.to_string(); - let root = display.lines().nth(1).expect("a compiled tree has a root"); + let root = display.lines().next().expect("a compiled tree has a root"); assert_eq!(root, format!(" Map: {name}"), "`{query}`"); } } diff --git a/src/query/src/promql/planner/native.rs b/src/query/src/promql/planner/native.rs index dd6791493f0..93f399ee87d 100644 --- a/src/query/src/promql/planner/native.rs +++ b/src/query/src/promql/planner/native.rs @@ -563,7 +563,6 @@ impl PromPlanner { let output_schema = fused_output_schema(&values, &output_tags, &time_index, FieldKind::Float)?; let node = FusedSeries::try_new( - Arc::from(prom_expr.to_string()), tree, FusedSeriesBinding { grid, @@ -1233,7 +1232,6 @@ impl PromPlanner { && 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, &tree, &children, &source_tags, @@ -1285,7 +1283,6 @@ impl PromPlanner { return Ok(None); }; let node = plan_split_aggregate( - Arc::from(prom_expr.to_string()), partial, final_tree, FusedSeriesBinding { @@ -1314,7 +1311,6 @@ impl PromPlanner { if required_placement || matches!(locality, InputLocality::SinglePartition) { let segment_metric = if cross_table { None } else { metric.clone() }; let plan = Box::pin(segment_tree( - prom_expr, &tree, &children, &source_tags, @@ -1381,7 +1377,6 @@ impl PromPlanner { (tags, metric) }; let node = FusedSeries::try_new( - Arc::from(prom_expr.to_string()), tree, FusedSeriesBinding { grid, @@ -1862,7 +1857,6 @@ fn cut_segments(node: &NativeNode, canonical_segmentation: bool) -> Segment { /// `None` when the boundary cannot be built, so the caller keeps the plan it would have chosen. #[allow(clippy::too_many_arguments)] fn canonical_boundary_plan( - prom_expr: &PromExpr, tree: &NativeTree, children: &[LogicalPlan], source_tags: &[Vec], @@ -1873,7 +1867,6 @@ fn canonical_boundary_plan( root_kind: FieldKind, ) -> Option { let (plan, _) = canonical_boundary_stages( - prom_expr, tree, children, source_tags, @@ -1893,7 +1886,6 @@ fn canonical_boundary_plan( /// anything else reads the assembled frames as an ordinary evaluated input. #[allow(clippy::too_many_arguments)] fn canonical_boundary_stages( - prom_expr: &PromExpr, tree: &NativeTree, children: &[LogicalPlan], source_tags: &[Vec], @@ -1924,7 +1916,6 @@ fn canonical_boundary_stages( // The producer: the statement's own tree, run per partition, exporting a frame. let producer = FusedSeries::try_new_canonical_producer( - Arc::from(prom_expr.to_string()), tree.clone(), FusedSeriesBinding { grid: Arc::clone(grid), @@ -1963,7 +1954,6 @@ fn canonical_boundary_stages( None }; let consumer = FusedSeries::try_new_with_format( - Arc::from(prom_expr.to_string()), consumer_tree, FusedSeriesBinding { grid: Arc::clone(grid), @@ -2002,7 +1992,6 @@ fn find_input_node(node: &NativeNode, source_index: usize) -> Option<&BoundInput #[allow(clippy::too_many_arguments, clippy::type_complexity)] fn build_evaluated_segment( - prom_expr: &PromExpr, node: &NativeNode, children: &[LogicalPlan], source_tags: &[Vec], @@ -2061,7 +2050,6 @@ fn build_evaluated_segment( // boundary and read as a frame. if matches!(tree.root().expr, NativeExpr::Merge { .. }) { let (plan, tags) = canonical_boundary_stages( - prom_expr, &tree, &inputs, &local_tags, @@ -2082,7 +2070,6 @@ fn build_evaluated_segment( let frame = FusedSeries::series_frame_schema(&tree).ok()?; let frame = DFSchema::try_from(frame.as_ref().clone()).ok()?; let plan = FusedSeries::try_new_with_format( - Arc::from("segment"), tree, FusedSeriesBinding { grid: Arc::clone(grid), @@ -2167,7 +2154,6 @@ fn build_inner_segment( let frame = FusedSeries::series_frame_schema(&tree).ok()?; let frame = DFSchema::try_from(frame.as_ref().clone()).ok()?; let plan = FusedSeries::try_new_with_format( - Arc::from("segment"), tree, FusedSeriesBinding { grid: Arc::clone(grid), @@ -2197,7 +2183,6 @@ fn build_inner_segment( /// root publishes rows. #[allow(clippy::too_many_arguments, clippy::type_complexity)] async fn segment_tree( - prom_expr: &PromExpr, tree: &NativeTree, children: &[LogicalPlan], source_tags: &[Vec], @@ -2268,7 +2253,6 @@ async fn segment_tree( return Ok(None); }; let plan = Box::pin(segment_tree( - prom_expr, &cut_tree, ¤t_children, ¤t_tags, @@ -2303,7 +2287,6 @@ async fn segment_tree( } let built = built.or_else(|| { build_evaluated_segment( - prom_expr, cut, ¤t_children, ¤t_tags, @@ -2377,7 +2360,6 @@ async fn segment_tree( } } return Ok(canonical_boundary_stages( - prom_expr, &parent, &inputs, &child_tags, @@ -2435,7 +2417,6 @@ async fn segment_tree( }; let (segment_metric, segment_time_index) = segment_binding(¤t_source_ids); let Ok(plan) = FusedSeries::try_new_with_format( - Arc::from(prom_expr.to_string()), parent, FusedSeriesBinding { grid: Arc::clone(grid), @@ -2538,7 +2519,6 @@ fn fused_output_schema( /// The statement's output value columns must match the partial state's accumulator count; a /// mismatch means the declared output shape does not match the state schema of the split tree. fn plan_split_aggregate( - promql: Arc, partial: NativeTree, final_tree: NativeTree, binding: FusedSeriesBinding, @@ -2548,9 +2528,9 @@ fn plan_split_aggregate( let state_accumulators = match &partial.root().output { NativeType::AggregateState(state) => state.accumulators.len(), _ => { - return Err(DataFusionError::Plan(format!( - "the partial tree for `{promql}` does not produce aggregate state" - ))) + return Err(DataFusionError::Plan( + "the partial tree does not produce aggregate state".into(), + )) .context(DataFusionPlanningSnafu); } }; @@ -2561,20 +2541,14 @@ fn plan_split_aggregate( .count(); if values != state_accumulators { return Err(DataFusionError::Plan(format!( - "the partial state for `{promql}` has {state_accumulators} accumulator(s), but the output schema declares \ + "the partial state has {state_accumulators} accumulator(s), but the output schema declares \ {values} value column(s)" ))) .context(DataFusionPlanningSnafu); } - let partial = FusedSeries::try_new_state_export( - Arc::clone(&promql), - partial, - binding.clone(), - vec![rawinput], - ) - .context(DataFusionPlanningSnafu)?; + let partial = FusedSeries::try_new_state_export(partial, binding.clone(), vec![rawinput]) + .context(DataFusionPlanningSnafu)?; FusedSeries::try_new( - promql, final_tree, binding, vec![LogicalPlan::Extension(Extension { diff --git a/src/query/src/promql/planner/test.rs b/src/query/src/promql/planner/test.rs index 72ce3ba97d3..9afa4390021 100644 --- a/src/query/src/promql/planner/test.rs +++ b/src/query/src/promql/planner/test.rs @@ -2620,15 +2620,11 @@ async fn native_fused_rate_plans_to_a_fused_series_node() { let plan = build_native_fused_plan("rate(some_metric[5m])", true).await; let display = plan.display_indent().to_string(); - assert!( - display.contains("FusedSeries: rate(some_metric[5m])"), - "{display}" - ); + assert!(display.contains(" Window: rate [300000ms]"), "{display}"); // The node reads the selector's own scan; nothing between them packs the frame it reads. assert!(display.contains("TableScan: some_metric"), "{display}"); - // The node carries the query a reader wrote, not the shape it compiled to. + // The node keeps the compiled tree, not the query text. let fused = find_fused_series(&plan).expect("the plan holds a native node"); - assert_eq!("rate(some_metric[5m])", fused.promql()); assert_eq!( vec!["timestamp", "field_0", "tag_0"], plan.schema() @@ -3401,7 +3397,6 @@ async fn native_histogram_count_drops_float_input() { let query = "histogram_count(some_metric)"; let plan = build_native_fused_plan(query, true).await; let fused = find_fused_series(&plan).expect("histogram_count uses the native route"); - assert_eq!(parser::parse(query).unwrap().to_string(), fused.promql()); assert!(format!("{:?}", fused.tree()).contains("Count")); // The in-memory table scan is not a RegionScanExec or an explicit raw-frame source, so this // unit harness cannot execute a region-backed native plan. Native histogram kernels test that @@ -3413,7 +3408,9 @@ async fn native_ceil_maps_raw_float_input() { let query = "ceil(some_metric)"; let plan = build_native_fused_plan(query, true).await; let fused = find_fused_series(&plan).expect("ceil uses the native route"); - assert_eq!(parser::parse(query).unwrap().to_string(), fused.promql()); + let tree = fused.tree().to_string(); + assert!(tree.contains("Map: ceil"), "{tree}"); + assert!(tree.contains("Input: raw samples source=0"), "{tree}"); let mut names = plan .schema() .fields() @@ -3437,7 +3434,6 @@ async fn native_fused_binary_keeps_independent_scan_occurrences() { ] { let plan = build_native_fused_plan(query, true).await; let fused = find_fused_series(&plan).expect("the plan holds a native node"); - assert_eq!(parser::parse(query).unwrap().to_string(), fused.promql()); assert_eq!(2, fused.inputs().len(), "{query}"); let scans = fused .inputs() @@ -4206,6 +4202,49 @@ fn sorted_column_names(plan: &LogicalPlan) -> Vec { names } +/// Render a split exactly as EXPLAIN displays the frontend and remote region plans. +#[tokio::test] +async fn native_split_explain_renders_each_tree_including_remote_input() { + let query = "100 - avg by (tag_0) (irate(some_metric[1m])) * 100"; + let plan = build_native_fused_plan(query, true).await; + let final_node = find_final_split(&plan).expect("split final"); + let [partial] = final_node.children() else { + panic!("split partial"); + }; + let remote = + crate::dist_plan::MergeScanLogicalPlan::new(partial.clone(), false).into_logical_plan(); + let final_node = final_node + .with_exprs_and_inputs(vec![], vec![remote]) + .unwrap(); + let logical = LogicalPlan::Extension(Extension { + node: Arc::new(final_node), + }); + let text = logical.display_indent().to_string(); + let verbose = logical.display_indent_schema().to_string(); + println!("=== logical_plan ===\n{text}\n=== logical_plan VERBOSE ===\n{verbose}"); + let (final_text, remote_text) = text + .split_once("MergeScan [is_placeholder=false, remote_input=[") + .expect("the partial is inside a merge scan"); + assert!(final_text.starts_with("FusedSeries\n Binary:"), "{text}"); + assert!( + final_text.contains("Aggregate: avg by [tag_0] final"), + "{text}" + ); + assert!( + remote_text + .contains("FusedSeries: (output: series_frame)\n Aggregate: avg by [tag_0] partial"), + "{text}" + ); + assert!(remote_text.contains("Window: irate [60000ms]"), "{text}"); + assert!( + remote_text.contains("Input: raw samples source=0"), + "{text}" + ); + assert_ne!(final_text, remote_text); + assert!(!text.contains("metrics="), "{text}"); + assert!(!verbose.contains("metrics="), "{verbose}"); +} + /// An aggregate is split: the merge stays on the frontend and the per-series prefix runs in the /// producer. Native names follow the compiled tree; grouping labels, values and timestamps retain /// their existing schema roles. @@ -4348,7 +4387,6 @@ async fn native_fused_data_free_vector_does_not_block_segmentation() { .as_any() .downcast_ref::() .unwrap_or_else(|| panic!("expected a fused root rather than a legacy root: {native}")); - assert_eq!(parent.promql(), parser::parse(query).unwrap().to_string()); let NativeExpr::Binary { left, right, .. } = &parent.tree().root().expr else { panic!("expected the whole binary expression in the native parent: {native}"); }; @@ -4514,7 +4552,6 @@ async fn native_split_and_complete_aggregate_produce_identical_values() { let final_node = find_final_split(&split).expect("the query must use a split aggregate"); let partial = partial_below(final_node).unwrap(); let complete = FusedSeries::try_new( - Arc::from(query), complete_tree, final_node.binding().clone(), partial.children().to_vec(), @@ -5023,7 +5060,7 @@ async fn native_fused_histogram_above_aggregate_splits_with_bucket_routing() { // Only the Final node over state publishes this suffix, so the plan's own text pins it. assert!( display.contains( - " (state routing: canonical labels except [\"le\", \"__name__\", \"__type__\", \"__unit__\"])" + "FusedSeries: (state routing: canonical labels except [\"le\", \"__name__\", \"__type__\", \"__unit__\"])" ), "`{query}`: {display}" ); @@ -8080,7 +8117,7 @@ async fn native_histogram_selector_plans_a_native_producer() { let table_provider = build_test_native_histogram_table_provider("some_metric").await; let plan = build_native_fused_plan_with_provider("some_metric", true, table_provider).await; let display = plan.display_indent_schema().to_string(); - assert!(display.contains("FusedSeries: "), "{display}"); + assert!(display.contains("Input: raw samples source=0"), "{display}"); assert!(!display.contains("HistogramFold:"), "{display}"); let fused = find_fused_series(&plan).expect("the statement plans to a native producer"); // The output field column is the canonical native histogram struct. diff --git a/src/query/src/query_engine/state.rs b/src/query/src/query_engine/state.rs index d8b6914b692..c14f481b907 100644 --- a/src/query/src/query_engine/state.rs +++ b/src/query/src/query_engine/state.rs @@ -884,7 +884,6 @@ mod tests { }); for (disjoint, needs_router) in [(true, false), (false, true)] { let fused = FusedSeries::try_new( - Arc::from("metric"), tree.clone(), FusedSeriesBinding { grid: Arc::clone(&grid), diff --git a/tests-integration/src/tests/promql_test.rs b/tests-integration/src/tests/promql_test.rs index 19df9b06139..15825e55c18 100644 --- a/tests-integration/src/tests/promql_test.rs +++ b/tests-integration/src/tests/promql_test.rs @@ -2103,7 +2103,7 @@ async fn native_promql_clamp_aggregates_execute_on_native_segments() { .to_string(); if split { assert!( - display.contains("Aggregate sum by [] final"), + display.contains("Aggregate: sum by [] final"), "{query}: {display}" ); let producer = observer.fused().expect("native partial producer"); @@ -2117,11 +2117,11 @@ async fn native_promql_clamp_aggregates_execute_on_native_segments() { } else { assert!(display.contains("FusedSeriesExec"), "{query}: {display}"); assert!( - display.contains("Vector topk 1 by []"), + display.contains("Vector: topk 1 by []"), "{query}: {display}" ); assert!( - display.contains("Aggregate sum by [] complete"), + display.contains("Aggregate: sum by [] complete"), "{query}: {display}" ); } @@ -3853,13 +3853,13 @@ async fn native_promql_label_rewrites_reach_the_native_route_and_keep_prometheus let histogram_producer = histogram_router.children()[0].clone(); // The boundary's own tree is the merge over the rewrite: `histogram_count` is not in it, so the // router that validates the collected label set runs below the map rather than above it. The - // node's explain line names the whole statement, so the tree is read from its own header on. + // physical node renders its own tree directly below its operator line. let producer_display = datafusion::physical_plan::displayable(histogram_producer.as_ref()) - .indent(false) + .one_line() .to_string(); let producer_tree = producer_display - .split_once("NativeTree") - .map(|(_, tree)| format!("NativeTree{tree}")) + .split_once('\n') + .map(|(_, tree)| tree) .unwrap_or_else(|| panic!("the producer renders its tree: {producer_display}")); assert!( producer_tree.contains("Merge"), @@ -6936,10 +6936,11 @@ async fn native_promql_explain_and_analyze_execute_fused_plan() { .sum::(); assert_eq!(output_values, 1, "ANALYZE executes the fused series"); } else if statement.contains("EXPLAIN") && enabled { - assert!(text.contains("FusedSeries: native_explain")); + assert!(text.contains("Input: raw samples source=0"), "{text}"); assert!(text.contains("MergeScanExec")); + assert!(!text.contains("metrics="), "{text}"); } else if !enabled { - assert!(!text.contains("FusedSeries:")); + assert!(!text.contains("FusedSeries")); } } } @@ -7116,7 +7117,7 @@ fn holds_split_merge(plan: &Arc) - let mut stack = vec![plan.clone()]; while let Some(node) = stack.pop() { let display = datafusion::physical_plan::displayable(node.as_ref()) - .indent(false) + .one_line() .to_string(); if node.name() == "FusedSeriesExec" && display.contains(" final") { let mut descendants = node.children().into_iter().cloned().collect::>(); @@ -8770,21 +8771,22 @@ async fn native_promql_sum_topk_uses_global_topk_segment() { "a global topk answers one partition" ); let parent_display = datafusion::physical_plan::displayable(parent.as_ref()) - .indent(true) + .one_line() .to_string(); let parent_tree = parent_display - .split_once("NativeTree") - .map_or(parent_display.as_str(), |(_, tree)| tree); + .split_once('\n') + .expect("the parent renders its own tree") + .1; assert!( - parent_tree.contains("Aggregate sum by [] complete"), + parent_tree.contains("Aggregate: sum by [] complete"), "the parent owns the complete sum: {parent_tree}" ); assert!( - parent_tree.contains("Vector topk 1 by []"), + parent_tree.contains("Vector: topk 1 by []"), "the parent owns the global topk: {parent_tree}" ); assert!( - parent_tree.contains("Input evaluated samples source=0"), + parent_tree.contains("Input: evaluated samples source=0"), "the parent reads its child's evaluated frame, not a scan: {parent_tree}" ); assert!( @@ -8807,13 +8809,13 @@ async fn native_promql_sum_topk_uses_global_topk_segment() { "the prefix keeps the scan's partitions, which is the point of cutting below the topk" ); let child_display = datafusion::physical_plan::displayable(child.as_ref()) - .indent(true) + .one_line() .to_string(); - // The first line names the statement the node was built for, which is the same for both - // segments; what a segment computes is the tree below it. + // Each segment renders its own compiled tree below the operator line. let child_tree = child_display - .split_once("NativeTree") - .map_or(child_display.as_str(), |(_, tree)| tree); + .split_once('\n') + .expect("the child renders its own tree") + .1; assert!( child_tree.contains("Instant"), "the child owns the per-series prefix: {child_tree}" @@ -9407,20 +9409,20 @@ async fn native_promql_cuts_segments_only_where_a_gather_would_swallow_them() { } if query == "sum(topk(1, seg_a)) + vector(1)" { let parent = datafusion::physical_plan::displayable(frontend[0].as_ref()) - .indent(true) + .one_line() .to_string(); - let parent_tree = parent.split_once("NativeTree").unwrap().1; + let parent_tree = parent.split_once('\n').unwrap().1; assert!(parent_tree.contains("VectorOf"), "{parent_tree}"); assert!( - parent_tree.contains("Input evaluated samples source=0"), + parent_tree.contains("Input: evaluated samples source=0"), "the parent must read the data segment: {parent_tree}" ); assert!(!parent_tree.contains("Instant"), "{parent_tree}"); assert_eq!(regions.len(), 1, "{query}: the data prefix must be cut"); let prefix = datafusion::physical_plan::displayable(regions[0].as_ref()) - .indent(true) + .one_line() .to_string(); - let prefix_tree = prefix.split_once("NativeTree").unwrap().1; + let prefix_tree = prefix.split_once('\n').unwrap().1; assert!(prefix_tree.contains("Instant"), "{prefix_tree}"); assert!(!prefix_tree.contains("VectorOf"), "{prefix_tree}"); let mut rows = eval_values(native_tql_output(&instance, &sql, true).await); @@ -10856,7 +10858,7 @@ async fn native_promql_root_canonical_merge_keeps_the_scan_width() { .indent(true) .to_string(); assert!( - pair_sum_display.contains("Aggregate sum by [] complete"), + pair_sum_display.contains("Aggregate: sum by [] complete"), "the fold is a complete aggregate over the assembled series: {pair_sum_display}" ); assert!( @@ -10931,7 +10933,7 @@ async fn native_promql_root_canonical_merge_keeps_the_scan_width() { // state standing in for the collection. if shape == "a fold" { assert!( - display.contains("Aggregate sum by [] complete"), + display.contains("Aggregate: sum by [] complete"), "{shape}: the fold is a complete aggregate over the assembled frames: {display}" ); assert!( @@ -11026,7 +11028,7 @@ async fn native_promql_root_canonical_merge_keeps_the_scan_width() { .indent(true) .to_string(); assert!( - sum_display.contains("Aggregate sum by [] complete"), + sum_display.contains("Aggregate: sum by [] complete"), "the fold runs natively as a complete aggregate over the boundary: {sum_display}" ); assert!( diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index b968f418024..35b9617a9c8 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -1969,7 +1969,6 @@ async fn assert_raw_scan_fused_formats(observer: &NativeFusedObserver, frontend: ] { let scan_plan = scan_plan.clone().reset_state().unwrap(); let exec = promql::extension_plan::FusedSeries::try_new_with_format( - Arc::from("native_demo"), tree.clone(), binding.clone(), vec![logical_child.clone()], diff --git a/tests-integration/tests/native_prefix.rs b/tests-integration/tests/native_prefix.rs index 4004c8069dc..7edeb30817a 100644 --- a/tests-integration/tests/native_prefix.rs +++ b/tests-integration/tests/native_prefix.rs @@ -157,7 +157,6 @@ async fn native_prefix_multi_partition_single_position_chains() { }; let frame_schema = evaluated_frame_schema(&tree); let fused = FusedSeries::try_new_with_format( - Arc::from("rate(native_prefix_partitioned[10s])"), tree, binding, vec![child], @@ -347,7 +346,6 @@ async fn native_prefix_multi_partition_multi_chunk_chains() { }; let frame_schema = evaluated_frame_schema(&tree); let fused = FusedSeries::try_new_with_format( - Arc::from("rate(native_prefix_partitioned[10s])"), tree, binding, vec![child], @@ -532,7 +530,6 @@ async fn native_prefix_executes_over_real_mito_region() { // the scan below it: `try_new_with_format` compares the two and refuses a mismatch. let frame_schema = evaluated_frame_schema(&tree); let fused = FusedSeries::try_new_with_format( - Arc::from("rate(native_prefix[3s])"), tree, binding, vec![child],