You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/05/23 13:00:20 UTC
(datafusion) 03/04: Optimize metric label cloning (#22406)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-22470-6052ba821b934879a8181640993dee5917759f1b
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 6052ba821b934879a8181640993dee5917759f1b
Author: xudong.w <wx...@gmail.com>
AuthorDate: Sat May 23 20:50:02 2026 +0800
Optimize metric label cloning (#22406)
## Which issue does this PR close?
- Part of #22189.
## Rationale for this change
`ParquetFileMetrics::new` registers many per-file metrics with the same
`filename` label. Before this PR, each metric built its own owned
filename label with `filename.to_string()`, which repeatedly copied the
same dynamic
string during parquet scan setup.
This PR keeps parquet metrics eagerly registered, so
`ExecutionPlan::metrics()` visibility during execution is unchanged,
while reducing repeated label string allocation and copying.
## What changes are included in this PR?
- Store owned `Label` name/value strings behind `Arc<str>` internally,
while
keeping borrowed static label strings allocation-free.
- Reuse one cloned `filename` label across the per-file parquet metrics
in
`ParquetFileMetrics::new`.
- Add a metrics test confirming borrowed and owned label values remain
equal
and display the same way.
## Are these changes tested?
Yes.
## Are there any user-facing changes?
No. Metric registration timing and displayed label values are unchanged.
---
datafusion/datasource-parquet/src/metrics.rs | 78 +++++++-------
.../physical-expr-common/src/metrics/builder.rs | 12 ++-
datafusion/physical-expr-common/src/metrics/mod.rs | 115 +++++++++++++++++++--
3 files changed, 156 insertions(+), 49 deletions(-)
diff --git a/datafusion/datasource-parquet/src/metrics.rs b/datafusion/datasource-parquet/src/metrics.rs
index 262dde024a..4bf009afd6 100644
--- a/datafusion/datasource-parquet/src/metrics.rs
+++ b/datafusion/datasource-parquet/src/metrics.rs
@@ -15,9 +15,11 @@
// specific language governing permissions and limitations
// under the License.
+use std::sync::Arc;
+
use datafusion_physical_plan::metrics::{
- Count, ExecutionPlanMetricsSet, Gauge, MetricBuilder, MetricCategory, MetricType,
- PruningMetrics, RatioMergeStrategy, RatioMetrics, Time,
+ Count, ExecutionPlanMetricsSet, Gauge, Label, MetricBuilder, MetricCategory,
+ MetricType, PruningMetrics, RatioMergeStrategy, RatioMetrics, Time,
};
/// Stores metrics about the parquet execution for a particular parquet file.
@@ -100,37 +102,42 @@ impl ParquetFileMetrics {
filename: &str,
metrics: &ExecutionPlanMetricsSet,
) -> Self {
+ // Share the filename label across all per-file metrics to avoid
+ // allocating the same filename string for each metric.
+ let filename_label = Label::new("filename", Arc::<str>::from(filename));
+ let builder = MetricBuilder::new(metrics).with_label(filename_label);
+
// -----------------------
// 'summary' level metrics
// -----------------------
- let row_groups_pruned_bloom_filter = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let row_groups_pruned_bloom_filter = builder
+ .clone()
.with_type(MetricType::Summary)
.pruning_metrics("row_groups_pruned_bloom_filter", partition);
- let limit_pruned_row_groups = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let limit_pruned_row_groups = builder
+ .clone()
.with_type(MetricType::Summary)
.pruning_metrics("limit_pruned_row_groups", partition);
- let row_groups_pruned_statistics = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let row_groups_pruned_statistics = builder
+ .clone()
.with_type(MetricType::Summary)
.pruning_metrics("row_groups_pruned_statistics", partition);
- let page_index_pages_pruned = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let page_index_pages_pruned = builder
+ .clone()
.with_type(MetricType::Summary)
.pruning_metrics("page_index_pages_pruned", partition);
- let bytes_scanned = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let bytes_scanned = builder
+ .clone()
.with_type(MetricType::Summary)
.with_category(MetricCategory::Bytes)
.counter("bytes_scanned", partition);
- let metadata_load_time = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let metadata_load_time = builder
+ .clone()
.with_type(MetricType::Summary)
.subset_time("metadata_load_time", partition);
@@ -138,8 +145,8 @@ impl ParquetFileMetrics {
.with_type(MetricType::Summary)
.pruning_metrics("files_ranges_pruned_statistics", partition);
- let scan_efficiency_ratio = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let scan_efficiency_ratio = builder
+ .clone()
.with_type(MetricType::Summary)
.ratio_metrics_with_strategy(
"scan_efficiency_ratio",
@@ -150,45 +157,44 @@ impl ParquetFileMetrics {
// -----------------------
// 'dev' level metrics
// -----------------------
- let predicate_evaluation_errors = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let predicate_evaluation_errors = builder
+ .clone()
.with_category(MetricCategory::Rows)
.counter("predicate_evaluation_errors", partition);
- let pushdown_rows_pruned = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let pushdown_rows_pruned = builder
+ .clone()
.with_category(MetricCategory::Rows)
.counter("pushdown_rows_pruned", partition);
- let pushdown_rows_matched = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let pushdown_rows_matched = builder
+ .clone()
.with_category(MetricCategory::Rows)
.counter("pushdown_rows_matched", partition);
- let row_pushdown_eval_time = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let row_pushdown_eval_time = builder
+ .clone()
.subset_time("row_pushdown_eval_time", partition);
- let statistics_eval_time = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let statistics_eval_time = builder
+ .clone()
.subset_time("statistics_eval_time", partition);
- let bloom_filter_eval_time = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let bloom_filter_eval_time = builder
+ .clone()
.subset_time("bloom_filter_eval_time", partition);
- let page_index_eval_time = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let page_index_eval_time = builder
+ .clone()
.subset_time("page_index_eval_time", partition);
- let page_index_rows_pruned = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let page_index_rows_pruned = builder
+ .clone()
.pruning_metrics("page_index_rows_pruned", partition);
- let predicate_cache_inner_records = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let predicate_cache_inner_records = builder
+ .clone()
.with_category(MetricCategory::Rows)
.gauge("predicate_cache_inner_records", partition);
- let predicate_cache_records = MetricBuilder::new(metrics)
- .with_new_label("filename", filename.to_string())
+ let predicate_cache_records = builder
.with_category(MetricCategory::Rows)
.gauge("predicate_cache_records", partition);
diff --git a/datafusion/physical-expr-common/src/metrics/builder.rs b/datafusion/physical-expr-common/src/metrics/builder.rs
index e9c0b76af2..de9d1e03d8 100644
--- a/datafusion/physical-expr-common/src/metrics/builder.rs
+++ b/datafusion/physical-expr-common/src/metrics/builder.rs
@@ -25,13 +25,15 @@ use crate::metrics::{
};
use super::{
- Count, ExecutionPlanMetricsSet, Gauge, Label, Metric, MetricValue, Time, Timestamp,
+ Count, ExecutionPlanMetricsSet, Gauge, Label, LabelValue, Metric, MetricValue, Time,
+ Timestamp,
};
/// Structure for constructing metrics, counters, timers, etc.
///
/// Note the use of `Cow<..>` is to avoid allocations in the common
-/// case of constant strings
+/// case of constant strings. Dynamically created label strings are shared when
+/// [`Label`] values are cloned.
///
/// ```rust
/// use datafusion_physical_expr_common::metrics::*;
@@ -47,6 +49,7 @@ use super::{
/// .with_new_label("filename", "my_awesome_file.parquet")
/// .counter("num_bytes", partition);
/// ```
+#[derive(Clone)]
pub struct MetricBuilder<'a> {
/// Location that the metric created by this builder will be added do
metrics: &'a ExecutionPlanMetricsSet,
@@ -108,7 +111,10 @@ impl<'a> MetricBuilder<'a> {
name: impl Into<Cow<'static, str>>,
value: impl Into<Cow<'static, str>>,
) -> Self {
- self.with_label(Label::new(name.into(), value.into()))
+ self.with_label(Label::new(
+ LabelValue::from(name.into()),
+ LabelValue::from(value.into()),
+ ))
}
/// Set the partition of the metric being constructed
diff --git a/datafusion/physical-expr-common/src/metrics/mod.rs b/datafusion/physical-expr-common/src/metrics/mod.rs
index eecd8cfabd..0a03075b91 100644
--- a/datafusion/physical-expr-common/src/metrics/mod.rs
+++ b/datafusion/physical-expr-common/src/metrics/mod.rs
@@ -30,6 +30,7 @@ use parking_lot::Mutex;
use std::{
borrow::Cow,
fmt::{self, Debug, Display},
+ hash::{Hash, Hasher},
sync::Arc,
vec::IntoIter,
};
@@ -519,20 +520,19 @@ impl From<MetricsSet> for ExecutionPlanMetricsSet {
/// telemetry]<https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/data-model.md>,
/// etc.
///
-/// As the name and value are expected to mostly be constant strings,
-/// use a [`Cow`] to avoid copying / allocations in this common case.
+/// As the name and value are expected to often be constant strings, borrowed
+/// static strings avoid allocations in that common case. Dynamic strings are
+/// stored behind [`Arc<str>`] so cloning labels does not copy the underlying
+/// string data.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct Label {
- name: Cow<'static, str>,
- value: Cow<'static, str>,
+ name: LabelValue,
+ value: LabelValue,
}
impl Label {
/// Create a new [`Label`]
- pub fn new(
- name: impl Into<Cow<'static, str>>,
- value: impl Into<Cow<'static, str>>,
- ) -> Self {
+ pub fn new(name: impl Into<LabelValue>, value: impl Into<LabelValue>) -> Self {
let name = name.into();
let value = value.into();
Self { name, value }
@@ -540,12 +540,12 @@ impl Label {
/// Returns the name of this label
pub fn name(&self) -> &str {
- self.name.as_ref()
+ self.name.as_str()
}
/// Returns the value of this label
pub fn value(&self) -> &str {
- self.value.as_ref()
+ self.value.as_str()
}
}
@@ -555,6 +555,89 @@ impl Display for Label {
}
}
+/// A label name or value.
+///
+/// String literals preserve the existing allocation-free path. Dynamic strings
+/// can be stored behind [`Arc<str>`], so cloning a [`Label`] only increments an
+/// atomic reference count and does not allocate or copy the underlying string
+/// data.
+#[derive(Clone)]
+pub struct LabelValue(LabelValueInner);
+
+/// Internal representation for label names and values.
+///
+/// `LabelValue` is public because `Label::new` accepts it, but these storage
+/// variants are implementation details. Keeping them private prevents external
+/// code from constructing or matching on `Static` and `Shared` directly.
+#[derive(Clone)]
+enum LabelValueInner {
+ Static(&'static str),
+ Shared(Arc<str>),
+}
+
+impl LabelValue {
+ /// Return this label value as a string slice.
+ pub fn as_str(&self) -> &str {
+ match &self.0 {
+ LabelValueInner::Static(value) => value,
+ LabelValueInner::Shared(value) => value.as_ref(),
+ }
+ }
+}
+
+impl From<&'static str> for LabelValue {
+ fn from(value: &'static str) -> Self {
+ Self(LabelValueInner::Static(value))
+ }
+}
+
+impl From<String> for LabelValue {
+ fn from(value: String) -> Self {
+ Self(LabelValueInner::Shared(Arc::from(value)))
+ }
+}
+
+impl From<Arc<str>> for LabelValue {
+ fn from(value: Arc<str>) -> Self {
+ Self(LabelValueInner::Shared(value))
+ }
+}
+
+impl From<Cow<'static, str>> for LabelValue {
+ fn from(value: Cow<'static, str>) -> Self {
+ match value {
+ Cow::Borrowed(value) => value.into(),
+ Cow::Owned(value) => value.into(),
+ }
+ }
+}
+
+impl PartialEq for LabelValue {
+ fn eq(&self, other: &Self) -> bool {
+ self.as_str() == other.as_str()
+ }
+}
+
+impl Eq for LabelValue {}
+
+impl Hash for LabelValue {
+ fn hash<H: Hasher>(&self, state: &mut H) {
+ self.as_str().hash(state);
+ }
+}
+
+impl Debug for LabelValue {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ Debug::fmt(self.as_str(), f)
+ }
+}
+
+impl Display for LabelValue {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ Display::fmt(self.as_str(), f)
+ }
+}
+
#[cfg(test)]
mod tests {
use std::time::Duration;
@@ -609,6 +692,18 @@ mod tests {
assert_eq!("output_rows{partition=2, foo=bar}=66", metric.to_string())
}
+ #[test]
+ fn test_label_owned_and_borrowed_values_are_equal() {
+ let borrowed = Label::new("foo", "bar");
+ let owned = Label::new("foo".to_string(), "bar".to_string());
+ let shared = Label::new("foo", Arc::<str>::from("bar"));
+
+ assert_eq!(borrowed, owned);
+ assert_eq!(borrowed, shared);
+ assert_eq!(borrowed.to_string(), owned.to_string());
+ assert_eq!(borrowed.to_string(), shared.to_string());
+ }
+
#[test]
fn test_output_rows() {
let metrics = ExecutionPlanMetricsSet::new();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org