rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-06 19:06:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(promql): print each fused node's own compiled tree

Browse Source 
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 <discord9@163.com>
This commit is contained in:
discord9
2026-11-06 16:45:28 +08:00
parent 3cebe452f3
commit fe18484261
18 changed files with 326 additions and 353 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/promql/src/extension_plan/fused_series.rs
+32 -65
View File
@@ -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<SchemaR
/// The logical plan of one node that runs a compiled native tree.
///
/// The node adds nothing to the tree: it records the query the tree was compiled from, so
/// `EXPLAIN` shows the reader their own query rather than the shape it compiled to, and it carries
/// the binding the tree's input was compiled against so the execution plan can be built from it.
/// The node carries the binding the tree's input was compiled against so the execution plan can
/// be built from it. `EXPLAIN` shows the compiled tree.
#[derive(Debug)]
pub struct FusedSeries {
/// The PromQL text this tree was compiled from.
promql: Arc<str>,
/// 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<String>,
format: FusedOutputFormat,
promql: Arc<str>,
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<str>,
tree: NativeTree,
binding: FusedSeriesBinding,
children: Vec<LogicalPlan>,
output_schema: DFSchemaRef,
) -> DataFusionResult<Self> {
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<str>,
tree: NativeTree,
binding: FusedSeriesBinding,
children: Vec<LogicalPlan>,
@@ -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<str>,
tree: NativeTree,
binding: FusedSeriesBinding,
children: Vec<LogicalPlan>,
@@ -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<str>,
tree: NativeTree,
binding: FusedSeriesBinding,
children: Vec<LogicalPlan>,
@@ -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::<DataFusionResult<Vec<_>>>()?;
let exec: Arc<dyn ExecutionPlan> = 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<Self> {
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<H: Hasher>(&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<Ordering> {
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<str>,
/// The tree this node runs, shared by every partition's execution.
tree: Arc<NativeTree>,
/// 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<str>,
inputs: Vec<(Arc<dyn ExecutionPlan>, RawLabelSource, SeriesOrdering)>,
tree: Arc<NativeTree>,
output_schema: SchemaRef,
@@ -2267,7 +2238,6 @@ impl FusedSeriesExec {
tags: Vec<String>,
) -> DataFusionResult<Self> {
Self::build(
promql,
inputs,
tree,
output_schema,
@@ -2281,7 +2251,6 @@ impl FusedSeriesExec {
#[allow(clippy::too_many_arguments)]
fn build(
promql: Arc<str>,
inputs: Vec<(Arc<dyn ExecutionPlan>, RawLabelSource, SeriesOrdering)>,
tree: Arc<NativeTree>,
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(),
src/promql/src/extension_plan/fused_series/tests.rs
+70 -107
View File
@@ -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<dyn ExecutionPlan>) -> 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<String>) -> 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<String>, 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<SeriesSchema>| {
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<Arc<dyn ExecutionPlan>> {
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<RecordBatch>) -> Vec<StateSeries> {
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::<Vec<_>>();
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<dyn ExecutionPlan>;
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(
src/promql/src/extension_plan/fused_series/tests/frame_output.rs
-3
View File
@@ -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())],
src/promql/src/extension_plan/series_repartition/tests.rs
-1
View File
@@ -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),
src/promql/src/native/display.rs
+1 -1
View File
@@ -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 {
src/promql/src/native/plan/tests.rs
+4 -16
View File
@@ -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}");
}
src/query/src/analyze.rs
+116 -41
View File
@@ -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 {
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]
src/query/src/dist_plan/analyzer/test.rs
+6 -2
View File
@@ -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::<FusedSeries>()
.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:?}");
src/query/src/dist_plan/commutativity.rs
+4 -24
View File
@@ -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 {
src/query/src/optimizer/scan_hint.rs
+1 -8
View File
@@ -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(),
),
})
}
src/query/src/optimizer/series_frame_output.rs
-1
View File
@@ -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,
src/query/src/promql/native_compile/tests.rs
+5 -5
View File
@@ -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}`");
}
}
src/query/src/promql/planner/native.rs
+6 -32
View File
@@ -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<String>],
@@ -1873,7 +1867,6 @@ fn canonical_boundary_plan(
root_kind: FieldKind,
) -> Option<LogicalPlan> {
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<String>],
@@ -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<String>],
@@ -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<String>],
@@ -2268,7 +2253,6 @@ async fn segment_tree(
return Ok(None);
};
let plan = Box::pin(segment_tree(
prom_expr,
&cut_tree,
&current_children,
&current_tags,
@@ -2303,7 +2287,6 @@ async fn segment_tree(
}
let built = built.or_else(|| {
build_evaluated_segment(
prom_expr,
cut,
&current_children,
&current_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(&current_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<str>,
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 {
src/query/src/promql/planner/test.rs
+50 -13
View File
@@ -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<String> {
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::<FusedSeries>()
.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.
src/query/src/query_engine/state.rs
-1
View File
@@ -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),
tests-integration/src/tests/promql_test.rs
+31 -29
View File
@@ -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::<usize>();
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<dyn datafusion::physical_plan::ExecutionPlan>) -
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::<Vec<_>>();
@@ -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!(
tests-integration/tests/http.rs
-1
View File
@@ -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()],
tests-integration/tests/native_prefix.rs
-3
View File
@@ -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],