Remove all allocations in metrics system

Author: mwylde

datafusion/physical-plan/src/metrics/builder.rs

@@ -17,10 +17,10 @@
 
 //! Builder for creating arbitrary metrics
 
-use std::{borrow::Cow, sync::Arc};
+use std::{borrow::Cow};
 
 use super::{
-    Count, ExecutionPlanMetricsSet, Gauge, Label, Metric, MetricValue, Time, Timestamp,
+    Count, ExecutionPlanMetricsSet, Gauge, Label, MetricValue, Time, Timestamp,
 };
 
 /// Structure for constructing metrics, counters, timers, etc.
@@ -45,7 +45,7 @@ use super::{
 /// ```
 pub struct MetricBuilder<'a> {
     /// Location that the metric created by this builder will be added do
-    metrics: &'a ExecutionPlanMetricsSet,
+    _metrics: &'a ExecutionPlanMetricsSet,
 
     /// optional partition number
     partition: Option<usize>,
@@ -58,7 +58,7 @@ impl<'a> MetricBuilder<'a> {
     /// Create a new `MetricBuilder` that will register the result of `build()` with the `metrics`
     pub fn new(metrics: &'a ExecutionPlanMetricsSet) -> Self {
         Self {
-            metrics,
+            _metrics: metrics,
             partition: None,
             labels: vec![],
         }
@@ -87,14 +87,7 @@ impl<'a> MetricBuilder<'a> {
 
     /// Consume self and create a metric of the specified value
     /// registered with the MetricsSet
-    pub fn build(self, value: MetricValue) {
-        let Self {
-            labels,
-            partition,
-            metrics,
-        } = self;
-        let metric = Arc::new(Metric::new_with_labels(value, partition, labels));
-        metrics.register(metric);
+    pub fn build(self, _value: MetricValue) {
     }
 
     /// Consume self and create a new counter for recording output rows

datafusion/physical-plan/src/metrics/mod.rs

@@ -369,8 +369,8 @@ impl ExecutionPlanMetricsSet {
     }
 
     /// Add the specified metric to the underlying metric set
-    pub fn register(&self, metric: Arc<Metric>) {
-        self.inner.lock().push(metric)
+    pub fn register(&self, _metric: Arc<Metric>) {
+        //self.inner.lock().push(metric)
     }
 
     /// Return a clone of the inner [`MetricsSet`]

datafusion/physical-plan/src/metrics/value.rs

@@ -20,25 +20,18 @@
 use std::{
     borrow::{Borrow, Cow},
     fmt::Display,
-    sync::{
-        atomic::{AtomicUsize, Ordering},
-        Arc,
-    },
     time::Duration,
 };
 
 use chrono::{DateTime, Utc};
 use datafusion_common::instant::Instant;
 use datafusion_execution::memory_pool::human_readable_size;
-use parking_lot::Mutex;
 
 /// A counter to record things such as number of input or output rows
 ///
 /// Note `clone`ing counters update the same underlying metrics
 #[derive(Debug, Clone)]
 pub struct Count {
-    /// value of the metric counter
-    value: Arc<AtomicUsize>,
 }
 
 impl PartialEq for Count {
@@ -63,20 +56,16 @@ impl Count {
     /// create a new counter
     pub fn new() -> Self {
         Self {
-            value: Arc::new(AtomicUsize::new(0)),
         }
     }
 
     /// Add `n` to the metric's value
-    pub fn add(&self, n: usize) {
-        // relaxed ordering for operations on `value` poses no issues
-        // we're purely using atomic ops with no associated memory ops
-        self.value.fetch_add(n, Ordering::Relaxed);
+    pub fn add(&self, _n: usize) {
     }
 
     /// Get the current value
     pub fn value(&self) -> usize {
-        self.value.load(Ordering::Relaxed)
+        0
     }
 }
 
@@ -86,8 +75,6 @@ impl Count {
 /// Note `clone`ing gauge update the same underlying metrics
 #[derive(Debug, Clone)]
 pub struct Gauge {
-    /// value of the metric gauge
-    value: Arc<AtomicUsize>,
 }
 
 impl PartialEq for Gauge {
@@ -112,47 +99,35 @@ impl Gauge {
     /// create a new gauge
     pub fn new() -> Self {
         Self {
-            value: Arc::new(AtomicUsize::new(0)),
         }
     }
 
     /// Add `n` to the metric's value
-    pub fn add(&self, n: usize) {
-        // relaxed ordering for operations on `value` poses no issues
-        // we're purely using atomic ops with no associated memory ops
-        self.value.fetch_add(n, Ordering::Relaxed);
+    pub fn add(&self, _n: usize) {
     }
 
     /// Sub `n` from the metric's value
-    pub fn sub(&self, n: usize) {
-        // relaxed ordering for operations on `value` poses no issues
-        // we're purely using atomic ops with no associated memory ops
-        self.value.fetch_sub(n, Ordering::Relaxed);
+    pub fn sub(&self, _n: usize) {
     }
 
     /// Set metric's value to maximum of `n` and current value
-    pub fn set_max(&self, n: usize) {
-        self.value.fetch_max(n, Ordering::Relaxed);
+    pub fn set_max(&self, _n: usize) {
     }
 
     /// Set the metric's value to `n` and return the previous value
-    pub fn set(&self, n: usize) -> usize {
-        // relaxed ordering for operations on `value` poses no issues
-        // we're purely using atomic ops with no associated memory ops
-        self.value.swap(n, Ordering::Relaxed)
+    pub fn set(&self, _n: usize) -> usize {
+        0
     }
 
     /// Get the current value
     pub fn value(&self) -> usize {
-        self.value.load(Ordering::Relaxed)
+        0
     }
 }
 
 /// Measure a potentially non contiguous duration of time
 #[derive(Debug, Clone)]
 pub struct Time {
-    /// elapsed time, in nanoseconds
-    nanos: Arc<AtomicUsize>,
 }
 
 impl Default for Time {
@@ -179,13 +154,11 @@ impl Time {
     /// times for operations.
     pub fn new() -> Self {
         Self {
-            nanos: Arc::new(AtomicUsize::new(0)),
         }
     }
 
     /// Add elapsed nanoseconds since `start`to self
-    pub fn add_elapsed(&self, start: Instant) {
-        self.add_duration(start.elapsed());
+    pub fn add_elapsed(&self, _start: Instant) {
     }
 
     /// Add duration of time to self
@@ -198,14 +171,11 @@ impl Time {
     /// This is based on the assumption that the timing logic in most cases is likely
     /// to take at least a nanosecond, and so this is reasonable mechanism to avoid
     /// ambiguity, especially on systems with low-resolution monotonic clocks
-    pub fn add_duration(&self, duration: Duration) {
-        let more_nanos = duration.as_nanos() as usize;
-        self.nanos.fetch_add(more_nanos.max(1), Ordering::Relaxed);
+    pub fn add_duration(&self, _duration: Duration) {
     }
 
     /// Add the number of nanoseconds of other `Time` to self
-    pub fn add(&self, other: &Time) {
-        self.add_duration(Duration::from_nanos(other.value() as u64))
+    pub fn add(&self, _other: &Time) {
     }
 
     /// return a scoped guard that adds the amount of time elapsed
@@ -220,16 +190,14 @@ impl Time {
 
     /// Get the number of nanoseconds record by this Time metric
     pub fn value(&self) -> usize {
-        self.nanos.load(Ordering::Relaxed)
+        0
     }
 }
 
 /// Stores a single timestamp, stored as the number of nanoseconds
 /// elapsed from Jan 1, 1970 UTC
 #[derive(Debug, Clone)]
 pub struct Timestamp {
-    /// Time thing started
-    timestamp: Arc<Mutex<Option<DateTime<Utc>>>>,
 }
 
 impl Default for Timestamp {
@@ -242,7 +210,6 @@ impl Timestamp {
     /// Create a new timestamp and sets its value to 0
     pub fn new() -> Self {
         Self {
-            timestamp: Arc::new(Mutex::new(None)),
         }
     }
 
@@ -252,40 +219,23 @@ impl Timestamp {
     }
 
     /// Sets the timestamps value to a specified time
-    pub fn set(&self, now: DateTime<Utc>) {
-        *self.timestamp.lock() = Some(now);
+    pub fn set(&self, _now: DateTime<Utc>) {
     }
 
     /// return the timestamps value at the last time `record()` was
     /// called.
     ///
     /// Returns `None` if `record()` has not been called
     pub fn value(&self) -> Option<DateTime<Utc>> {
-        *self.timestamp.lock()
+        None
     }
 
     /// sets the value of this timestamp to the minimum of this and other
-    pub fn update_to_min(&self, other: &Timestamp) {
-        let min = match (self.value(), other.value()) {
-            (None, None) => None,
-            (Some(v), None) => Some(v),
-            (None, Some(v)) => Some(v),
-            (Some(v1), Some(v2)) => Some(if v1 < v2 { v1 } else { v2 }),
-        };
-
-        *self.timestamp.lock() = min;
+    pub fn update_to_min(&self, _other: &Timestamp) {
     }
 
     /// sets the value of this timestamp to the maximum of this and other
-    pub fn update_to_max(&self, other: &Timestamp) {
-        let max = match (self.value(), other.value()) {
-            (None, None) => None,
-            (Some(v), None) => Some(v),
-            (None, Some(v)) => Some(v),
-            (Some(v1), Some(v2)) => Some(if v1 < v2 { v2 } else { v1 }),
-        };
-
-        *self.timestamp.lock() = max;
+    pub fn update_to_max(&self, _other: &Timestamp) {
     }
 }