Optimize metric label cloning

Author: xudong963

datafusion/datasource-parquet/src/metrics.rs

@@ -16,8 +16,8 @@
 // under the License.
 
 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,46 +100,51 @@ 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", filename.to_string());
+        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);
 
         let files_ranges_pruned_statistics = MetricBuilder::new(metrics)
             .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 +155,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);
 

datafusion/physical-expr-common/src/metrics/builder.rs

@@ -31,7 +31,8 @@ use super::{
 /// 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 +48,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,

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,12 +520,14 @@ 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 {
@@ -533,19 +536,19 @@ impl Label {
         name: impl Into<Cow<'static, str>>,
         value: impl Into<Cow<'static, str>>,
     ) -> Self {
-        let name = name.into();
-        let value = value.into();
+        let name = LabelValue::from(name.into());
+        let value = LabelValue::from(value.into());
         Self { name, value }
     }
 
     /// 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 +558,60 @@ impl Display for Label {
     }
 }
 
+/// Internal representation for label names and values.
+///
+/// `Static` preserves the existing allocation-free path for string literals.
+/// `Shared` stores dynamic strings 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, Eq)]
+enum LabelValue {
+    Static(&'static str),
+    Shared(Arc<str>),
+}
+
+impl LabelValue {
+    fn as_str(&self) -> &str {
+        match self {
+            Self::Static(value) => value,
+            Self::Shared(value) => value.as_ref(),
+        }
+    }
+}
+
+impl From<Cow<'static, str>> for LabelValue {
+    fn from(value: Cow<'static, str>) -> Self {
+        match value {
+            Cow::Borrowed(value) => Self::Static(value),
+            Cow::Owned(value) => Self::Shared(Arc::from(value)),
+        }
+    }
+}
+
+impl PartialEq for LabelValue {
+    fn eq(&self, other: &Self) -> bool {
+        self.as_str() == other.as_str()
+    }
+}
+
+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 +666,15 @@ 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());
+
+        assert_eq!(borrowed, owned);
+        assert_eq!(borrowed.to_string(), owned.to_string());
+    }
+
     #[test]
     fn test_output_rows() {
         let metrics = ExecutionPlanMetricsSet::new();