This is an automated email from the ASF dual-hosted git repository. gabotechs pushed a commit to branch gabotechs/filter-metrics-by-name in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit f743c5aea68f10f39798a5f62ce05b1c6c181fcd Author: Gabriel <[email protected]> AuthorDate: Sun Jul 19 09:53:51 2026 +0200 Add name filter to metrics --- datafusion/physical-expr-common/src/metrics/mod.rs | 31 ++++++++++++++ datafusion/physical-plan/src/display.rs | 50 ++++++++++++++++++++++ 2 files changed, 81 insertions(+) diff --git a/datafusion/physical-expr-common/src/metrics/mod.rs b/datafusion/physical-expr-common/src/metrics/mod.rs index d6048a0fcd..f510714dae 100644 --- a/datafusion/physical-expr-common/src/metrics/mod.rs +++ b/datafusion/physical-expr-common/src/metrics/mod.rs @@ -418,6 +418,21 @@ impl MetricsSet { .collect::<Vec<_>>(); Self { metrics } } + + /// Returns a new `MetricsSet` filtered by metric name. + /// Only metrics with the names appearing the list will be kept. + pub fn filter_by_names(self, names: &[String]) -> Self { + if names.is_empty() { + return Self { metrics: vec![] }; + } + + let metrics = self + .metrics + .into_iter() + .filter(|metric| names.iter().any(|name| name == metric.value().name())) + .collect::<Vec<_>>(); + Self { metrics } + } } impl Display for MetricsSet { @@ -966,4 +981,20 @@ mod tests { metric_names(&metrics) ); } + + #[test] + fn test_filter_by_names() { + let metrics = ExecutionPlanMetricsSet::new(); + MetricBuilder::new(&metrics).output_rows(0); + MetricBuilder::new(&metrics).counter("custom_counter", 0); + + let names = vec!["output_rows".to_string()]; + let filtered = metrics.clone_inner().filter_by_names(&names); + + assert_eq!(filtered.iter().count(), 1); + assert_eq!( + filtered.iter().next().unwrap().value().name(), + "output_rows" + ); + } } diff --git a/datafusion/physical-plan/src/display.rs b/datafusion/physical-plan/src/display.rs index 6a4d09057b..d14be3575a 100644 --- a/datafusion/physical-plan/src/display.rs +++ b/datafusion/physical-plan/src/display.rs @@ -129,6 +129,9 @@ pub struct DisplayableExecutionPlan<'a> { /// Optional filter by semantic category (rows / bytes / timing). /// `None` means show all categories; `Some(vec![])` means plan-only. metric_categories: Option<Vec<MetricCategory>>, + /// Optional filter by metric names. Only metric names in this list + /// will be rendered. + metric_names: Option<Vec<String>>, // (TreeRender) Maximum total width of the rendered tree tree_maximum_render_width: usize, /// Optional summary totals (currently only used by `pgjson`) — the total @@ -159,6 +162,7 @@ impl<'a> DisplayableExecutionPlan<'a> { show_schema: false, metric_types: Self::default_metric_types(), metric_categories: None, + metric_names: None, tree_maximum_render_width: 240, summary: None, } @@ -175,6 +179,7 @@ impl<'a> DisplayableExecutionPlan<'a> { show_schema: false, metric_types: Self::default_metric_types(), metric_categories: None, + metric_names: None, tree_maximum_render_width: 240, summary: None, } @@ -191,6 +196,7 @@ impl<'a> DisplayableExecutionPlan<'a> { show_schema: false, metric_types: Self::default_metric_types(), metric_categories: None, + metric_names: None, tree_maximum_render_width: 240, summary: None, } @@ -234,6 +240,15 @@ impl<'a> DisplayableExecutionPlan<'a> { self } + /// Specify which metric names to include. + /// + /// - An empty vector means plan-only — suppress all metrics. + /// - `vec!["metric_1"]` means show only the metric named `metric_1`. + pub fn set_metric_names(mut self, metric_names: Vec<String>) -> Self { + self.metric_names = Some(metric_names); + self + } + /// Set the maximum render width for the tree format pub fn set_tree_maximum_render_width(mut self, width: usize) -> Self { self.tree_maximum_render_width = width; @@ -279,6 +294,7 @@ impl<'a> DisplayableExecutionPlan<'a> { show_schema: bool, metric_types: Vec<MetricType>, metric_categories: Option<Vec<MetricCategory>>, + metric_names: Option<Vec<String>>, } impl fmt::Display for Wrapper<'_> { fn fmt(&self, f: &mut Formatter) -> fmt::Result { @@ -291,6 +307,7 @@ impl<'a> DisplayableExecutionPlan<'a> { show_schema: self.show_schema, metric_types: &self.metric_types, metric_categories: self.metric_categories.as_deref(), + metric_names: self.metric_names.as_deref(), }; accept(self.plan, &mut visitor) } @@ -303,6 +320,7 @@ impl<'a> DisplayableExecutionPlan<'a> { show_schema: self.show_schema, metric_types: self.metric_types.clone(), metric_categories: self.metric_categories.clone(), + metric_names: self.metric_names.clone(), } } @@ -324,6 +342,7 @@ impl<'a> DisplayableExecutionPlan<'a> { show_statistics: bool, metric_types: Vec<MetricType>, metric_categories: Option<Vec<MetricCategory>>, + metric_names: Option<Vec<String>>, } impl fmt::Display for Wrapper<'_> { fn fmt(&self, f: &mut Formatter) -> fmt::Result { @@ -336,6 +355,7 @@ impl<'a> DisplayableExecutionPlan<'a> { show_statistics: self.show_statistics, metric_types: &self.metric_types, metric_categories: self.metric_categories.as_deref(), + metric_names: self.metric_names.as_deref(), graphviz_builder: GraphvizBuilder::default(), parents: Vec::new(), }; @@ -355,6 +375,7 @@ impl<'a> DisplayableExecutionPlan<'a> { show_statistics: self.show_statistics, metric_types: self.metric_types.clone(), metric_categories: self.metric_categories.clone(), + metric_names: self.metric_names.clone(), } } @@ -403,6 +424,7 @@ impl<'a> DisplayableExecutionPlan<'a> { show_schema: bool, metric_types: Vec<MetricType>, metric_categories: Option<Vec<MetricCategory>>, + metric_names: Option<Vec<String>>, summary: Option<AnalyzeSummary>, } impl fmt::Display for Wrapper<'_> { @@ -413,6 +435,7 @@ impl<'a> DisplayableExecutionPlan<'a> { show_schema: self.show_schema, metric_types: &self.metric_types, metric_categories: self.metric_categories.as_deref(), + metric_names: self.metric_names.as_deref(), objects: HashMap::new(), parent_ids: Vec::new(), next_id: 0, @@ -446,6 +469,7 @@ impl<'a> DisplayableExecutionPlan<'a> { show_schema: self.show_schema, metric_types: self.metric_types.clone(), metric_categories: self.metric_categories.clone(), + metric_names: self.metric_names.clone(), summary: self.summary, } } @@ -460,6 +484,7 @@ impl<'a> DisplayableExecutionPlan<'a> { show_schema: bool, metric_types: Vec<MetricType>, metric_categories: Option<Vec<MetricCategory>>, + metric_names: Option<Vec<String>>, } impl fmt::Display for Wrapper<'_> { @@ -473,6 +498,7 @@ impl<'a> DisplayableExecutionPlan<'a> { show_schema: self.show_schema, metric_types: &self.metric_types, metric_categories: self.metric_categories.as_deref(), + metric_names: self.metric_names.as_deref(), }; visitor.pre_visit(self.plan)?; Ok(()) @@ -486,6 +512,7 @@ impl<'a> DisplayableExecutionPlan<'a> { show_schema: self.show_schema, metric_types: self.metric_types.clone(), metric_categories: self.metric_categories.clone(), + metric_names: self.metric_names.clone(), } } @@ -544,6 +571,8 @@ struct IndentVisitor<'a, 'b> { metric_types: &'a [MetricType], /// Optional filter by semantic category (rows / bytes / timing). metric_categories: Option<&'a [MetricCategory]>, + /// Optional filter by metric name. + metric_names: Option<&'a [String]>, } impl ExecutionPlanVisitor for IndentVisitor<'_, '_> { @@ -563,6 +592,9 @@ impl ExecutionPlanVisitor for IndentVisitor<'_, '_> { if let Some(cats) = self.metric_categories { metrics = metrics.filter_by_categories(cats); } + if let Some(names) = self.metric_names { + metrics = metrics.filter_by_names(names); + } write!(self.f, ", metrics=[{metrics}]")?; } else { write!(self.f, ", metrics=[]")?; @@ -574,6 +606,9 @@ impl ExecutionPlanVisitor for IndentVisitor<'_, '_> { if let Some(cats) = self.metric_categories { metrics = metrics.filter_by_categories(cats); } + if let Some(names) = self.metric_names { + metrics = metrics.filter_by_names(names); + } write!(self.f, ", metrics=[{metrics}]")?; } else { write!(self.f, ", metrics=[]")?; @@ -616,6 +651,8 @@ struct GraphvizVisitor<'a, 'b> { metric_types: &'a [MetricType], /// Optional filter by semantic category metric_categories: Option<&'a [MetricCategory]>, + /// Optional filter by metric name. + metric_names: Option<&'a [String]>, graphviz_builder: GraphvizBuilder, /// Used to record parent node ids when visiting a plan. @@ -660,6 +697,9 @@ impl ExecutionPlanVisitor for GraphvizVisitor<'_, '_> { if let Some(cats) = self.metric_categories { metrics = metrics.filter_by_categories(cats); } + if let Some(names) = self.metric_names { + metrics = metrics.filter_by_names(names); + } format!("metrics=[{metrics}]") } else { "metrics=[]".to_string() @@ -671,6 +711,9 @@ impl ExecutionPlanVisitor for GraphvizVisitor<'_, '_> { if let Some(cats) = self.metric_categories { metrics = metrics.filter_by_categories(cats); } + if let Some(names) = self.metric_names { + metrics = metrics.filter_by_names(names); + } format!("metrics=[{metrics}]") } else { "metrics=[]".to_string() @@ -729,6 +772,7 @@ struct PgJsonExecutionPlanVisitor<'a> { show_schema: bool, metric_types: &'a [MetricType], metric_categories: Option<&'a [MetricCategory]>, + metric_names: Option<&'a [String]>, objects: HashMap<u32, serde_json::Value>, parent_ids: Vec<u32>, next_id: u32, @@ -813,6 +857,12 @@ impl PgJsonExecutionPlanVisitor<'_> { metrics }; + let metrics = if let Some(names) = self.metric_names { + metrics.filter_by_names(names) + } else { + metrics + }; + // Build the Extras bucket, while extracting PG-canonical keys to the // top level. let mut extras = serde_json::Map::new(); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
