Skip to main content

datafusion_physical_expr_common/metrics/
value.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! Value representation of metrics
19
20use super::CustomMetricValue;
21use chrono::{DateTime, Utc};
22use datafusion_common::{
23    human_readable_count, human_readable_duration, human_readable_size, instant::Instant,
24};
25use parking_lot::Mutex;
26use std::{
27    borrow::{Borrow, Cow},
28    fmt::{Debug, Display},
29    sync::{
30        Arc,
31        atomic::{AtomicUsize, Ordering},
32    },
33    time::Duration,
34};
35
36/// A counter to record things such as number of input or output rows
37///
38/// Note `clone`ing counters update the same underlying metrics
39#[derive(Debug, Clone)]
40pub struct Count {
41    /// value of the metric counter
42    value: Arc<AtomicUsize>,
43}
44
45impl PartialEq for Count {
46    fn eq(&self, other: &Self) -> bool {
47        self.value().eq(&other.value())
48    }
49}
50
51impl Display for Count {
52    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
53        write!(f, "{}", human_readable_count(self.value()))
54    }
55}
56
57impl Default for Count {
58    fn default() -> Self {
59        Self::new()
60    }
61}
62
63impl Count {
64    /// create a new counter
65    pub fn new() -> Self {
66        Self {
67            value: Arc::new(AtomicUsize::new(0)),
68        }
69    }
70
71    /// Add `n` to the metric's value
72    pub fn add(&self, n: usize) {
73        // relaxed ordering for operations on `value` poses no issues
74        // we're purely using atomic ops with no associated memory ops
75        self.value.fetch_add(n, Ordering::Relaxed);
76    }
77
78    /// Get the current value
79    pub fn value(&self) -> usize {
80        self.value.load(Ordering::Relaxed)
81    }
82}
83
84/// A gauge is the simplest metrics type. It just returns a value.
85/// For example, you can easily expose current memory consumption with a gauge.
86///
87/// Note `clone`ing gauge update the same underlying metrics
88#[derive(Debug, Clone)]
89pub struct Gauge {
90    /// value of the metric gauge
91    value: Arc<AtomicUsize>,
92}
93
94impl PartialEq for Gauge {
95    fn eq(&self, other: &Self) -> bool {
96        self.value().eq(&other.value())
97    }
98}
99
100impl Display for Gauge {
101    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
102        write!(f, "{}", self.value())
103    }
104}
105
106impl Default for Gauge {
107    fn default() -> Self {
108        Self::new()
109    }
110}
111
112impl Gauge {
113    /// create a new gauge
114    pub fn new() -> Self {
115        Self {
116            value: Arc::new(AtomicUsize::new(0)),
117        }
118    }
119
120    /// Add `n` to the metric's value
121    pub fn add(&self, n: usize) {
122        // relaxed ordering for operations on `value` poses no issues
123        // we're purely using atomic ops with no associated memory ops
124        self.value.fetch_add(n, Ordering::Relaxed);
125    }
126
127    /// Sub `n` from the metric's value
128    pub fn sub(&self, n: usize) {
129        // relaxed ordering for operations on `value` poses no issues
130        // we're purely using atomic ops with no associated memory ops
131        self.value.fetch_sub(n, Ordering::Relaxed);
132    }
133
134    /// Set metric's value to maximum of `n` and current value
135    pub fn set_max(&self, n: usize) {
136        self.value.fetch_max(n, Ordering::Relaxed);
137    }
138
139    /// Set the metric's value to `n` and return the previous value
140    pub fn set(&self, n: usize) -> usize {
141        // relaxed ordering for operations on `value` poses no issues
142        // we're purely using atomic ops with no associated memory ops
143        self.value.swap(n, Ordering::Relaxed)
144    }
145
146    /// Get the current value
147    pub fn value(&self) -> usize {
148        self.value.load(Ordering::Relaxed)
149    }
150}
151
152/// Measure a potentially non contiguous duration of time
153#[derive(Debug, Clone)]
154pub struct Time {
155    /// elapsed time, in nanoseconds
156    nanos: Arc<AtomicUsize>,
157}
158
159impl Default for Time {
160    fn default() -> Self {
161        Self::new()
162    }
163}
164
165impl PartialEq for Time {
166    fn eq(&self, other: &Self) -> bool {
167        self.value().eq(&other.value())
168    }
169}
170
171impl Display for Time {
172    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
173        write!(f, "{}", human_readable_duration(self.value() as u64))
174    }
175}
176
177impl Time {
178    /// Create a new [`Time`] wrapper suitable for recording elapsed
179    /// times for operations.
180    pub fn new() -> Self {
181        Self {
182            nanos: Arc::new(AtomicUsize::new(0)),
183        }
184    }
185
186    /// Add elapsed nanoseconds since `start`to self
187    pub fn add_elapsed(&self, start: Instant) {
188        self.add_duration(start.elapsed());
189    }
190
191    /// Add duration of time to self
192    ///
193    /// Note: this will always increment the recorded time by at least 1 nanosecond
194    /// to distinguish between the scenario of no values recorded, in which
195    /// case the value will be 0, and no measurable amount of time having passed,
196    /// in which case the value will be small but not 0.
197    ///
198    /// This is based on the assumption that the timing logic in most cases is likely
199    /// to take at least a nanosecond, and so this is reasonable mechanism to avoid
200    /// ambiguity, especially on systems with low-resolution monotonic clocks
201    pub fn add_duration(&self, duration: Duration) {
202        let more_nanos = duration.as_nanos() as usize;
203        self.nanos.fetch_add(more_nanos.max(1), Ordering::Relaxed);
204    }
205
206    /// Add the number of nanoseconds of other `Time` to self
207    pub fn add(&self, other: &Time) {
208        self.add_duration(Duration::from_nanos(other.value() as u64))
209    }
210
211    /// return a scoped guard that adds the amount of time elapsed
212    /// between its creation and its drop or call to `stop` to the
213    /// underlying metric.
214    pub fn timer(&self) -> ScopedTimerGuard<'_> {
215        ScopedTimerGuard {
216            inner: self,
217            start: Some(Instant::now()),
218        }
219    }
220
221    /// Get the number of nanoseconds record by this Time metric
222    pub fn value(&self) -> usize {
223        self.nanos.load(Ordering::Relaxed)
224    }
225
226    /// Return a scoped guard that adds the amount of time elapsed between the
227    /// given instant and its drop (or the call to `stop`) to the underlying metric
228    pub fn timer_with(&self, now: Instant) -> ScopedTimerGuard<'_> {
229        ScopedTimerGuard {
230            inner: self,
231            start: Some(now),
232        }
233    }
234}
235
236/// Stores a single timestamp, stored as the number of nanoseconds
237/// elapsed from Jan 1, 1970 UTC
238#[derive(Debug, Clone)]
239pub struct Timestamp {
240    /// Time thing started
241    timestamp: Arc<Mutex<Option<DateTime<Utc>>>>,
242}
243
244impl Default for Timestamp {
245    fn default() -> Self {
246        Self::new()
247    }
248}
249
250impl Timestamp {
251    /// Create a new timestamp and sets its value to 0
252    pub fn new() -> Self {
253        Self {
254            timestamp: Arc::new(Mutex::new(None)),
255        }
256    }
257
258    /// Sets the timestamps value to the current time
259    pub fn record(&self) {
260        self.set(Utc::now())
261    }
262
263    /// Sets the timestamps value to a specified time
264    pub fn set(&self, now: DateTime<Utc>) {
265        *self.timestamp.lock() = Some(now);
266    }
267
268    /// return the timestamps value at the last time `record()` was
269    /// called.
270    ///
271    /// Returns `None` if `record()` has not been called
272    pub fn value(&self) -> Option<DateTime<Utc>> {
273        *self.timestamp.lock()
274    }
275
276    /// sets the value of this timestamp to the minimum of this and other
277    pub fn update_to_min(&self, other: &Timestamp) {
278        let min = match (self.value(), other.value()) {
279            (None, None) => None,
280            (Some(v), None) => Some(v),
281            (None, Some(v)) => Some(v),
282            (Some(v1), Some(v2)) => Some(if v1 < v2 { v1 } else { v2 }),
283        };
284
285        *self.timestamp.lock() = min;
286    }
287
288    /// sets the value of this timestamp to the maximum of this and other
289    pub fn update_to_max(&self, other: &Timestamp) {
290        let max = match (self.value(), other.value()) {
291            (None, None) => None,
292            (Some(v), None) => Some(v),
293            (None, Some(v)) => Some(v),
294            (Some(v1), Some(v2)) => Some(if v1 < v2 { v2 } else { v1 }),
295        };
296
297        *self.timestamp.lock() = max;
298    }
299}
300
301impl PartialEq for Timestamp {
302    fn eq(&self, other: &Self) -> bool {
303        self.value().eq(&other.value())
304    }
305}
306
307impl Display for Timestamp {
308    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
309        match self.value() {
310            None => write!(f, "NONE"),
311            Some(v) => {
312                write!(f, "{v}")
313            }
314        }
315    }
316}
317
318/// RAAI structure that adds all time between its construction and
319/// destruction to the CPU time or the first call to `stop` whichever
320/// comes first
321pub struct ScopedTimerGuard<'a> {
322    inner: &'a Time,
323    start: Option<Instant>,
324}
325
326impl ScopedTimerGuard<'_> {
327    /// Stop the timer timing and record the time taken
328    pub fn stop(&mut self) {
329        if let Some(start) = self.start.take() {
330            self.inner.add_elapsed(start)
331        }
332    }
333
334    /// Restarts the timer recording from the current time
335    pub fn restart(&mut self) {
336        self.start = Some(Instant::now())
337    }
338
339    /// Stop the timer, record the time taken and consume self
340    pub fn done(mut self) {
341        self.stop()
342    }
343
344    /// Stop the timer timing and record the time taken since the given endpoint.
345    pub fn stop_with(&mut self, end_time: Instant) {
346        if let Some(start) = self.start.take() {
347            let elapsed = end_time - start;
348            self.inner.add_duration(elapsed)
349        }
350    }
351
352    /// Stop the timer, record the time taken since `end_time` endpoint, and
353    /// consume self.
354    pub fn done_with(mut self, end_time: Instant) {
355        self.stop_with(end_time)
356    }
357}
358
359impl Drop for ScopedTimerGuard<'_> {
360    fn drop(&mut self) {
361        self.stop()
362    }
363}
364
365/// Counters tracking pruning metrics
366///
367/// For example, a file scanner initially is planned to scan 10 files, but skipped
368/// 8 of them using statistics, the pruning metrics would look like: 10 total -> 2 matched
369///
370/// Note `clone`ing update the same underlying metrics
371#[derive(Debug, Clone)]
372pub struct PruningMetrics {
373    pruned: Arc<AtomicUsize>,
374    matched: Arc<AtomicUsize>,
375    fully_matched: Arc<AtomicUsize>,
376}
377
378impl Display for PruningMetrics {
379    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
380        let matched = self.matched.load(Ordering::Relaxed);
381        let total = self.pruned.load(Ordering::Relaxed) + matched;
382        let fully_matched = self.fully_matched.load(Ordering::Relaxed);
383
384        if fully_matched != 0 {
385            write!(
386                f,
387                "{} total → {} matched -> {} fully matched",
388                human_readable_count(total),
389                human_readable_count(matched),
390                human_readable_count(fully_matched)
391            )
392        } else {
393            write!(
394                f,
395                "{} total → {} matched",
396                human_readable_count(total),
397                human_readable_count(matched)
398            )
399        }
400    }
401}
402
403impl Default for PruningMetrics {
404    fn default() -> Self {
405        Self::new()
406    }
407}
408
409impl PruningMetrics {
410    /// create a new PruningMetrics
411    pub fn new() -> Self {
412        Self {
413            pruned: Arc::new(AtomicUsize::new(0)),
414            matched: Arc::new(AtomicUsize::new(0)),
415            fully_matched: Arc::new(AtomicUsize::new(0)),
416        }
417    }
418
419    /// Add `n` to the metric's pruned value
420    pub fn add_pruned(&self, n: usize) {
421        // relaxed ordering for operations on `value` poses no issues
422        // we're purely using atomic ops with no associated memory ops
423        self.pruned.fetch_add(n, Ordering::Relaxed);
424    }
425
426    /// Add `n` to the metric's matched value
427    pub fn add_matched(&self, n: usize) {
428        // relaxed ordering for operations on `value` poses no issues
429        // we're purely using atomic ops with no associated memory ops
430        self.matched.fetch_add(n, Ordering::Relaxed);
431    }
432
433    /// Add `n` to the metric's fully matched value
434    pub fn add_fully_matched(&self, n: usize) {
435        // relaxed ordering for operations on `value` poses no issues
436        // we're purely using atomic ops with no associated memory ops
437        self.fully_matched.fetch_add(n, Ordering::Relaxed);
438    }
439
440    /// Subtract `n` to the metric's matched value.
441    pub fn subtract_matched(&self, n: usize) {
442        // relaxed ordering for operations on `value` poses no issues
443        // we're purely using atomic ops with no associated memory ops
444        self.matched.fetch_sub(n, Ordering::Relaxed);
445    }
446
447    /// Number of items pruned
448    pub fn pruned(&self) -> usize {
449        self.pruned.load(Ordering::Relaxed)
450    }
451
452    /// Number of items matched (not pruned)
453    pub fn matched(&self) -> usize {
454        self.matched.load(Ordering::Relaxed)
455    }
456
457    /// Number of items fully matched
458    pub fn fully_matched(&self) -> usize {
459        self.fully_matched.load(Ordering::Relaxed)
460    }
461}
462
463/// Counters tracking ratio metrics (e.g. matched vs total)
464///
465/// The counters are thread-safe and shared across clones.
466#[derive(Debug, Clone, Default)]
467pub struct RatioMetrics {
468    part: Arc<AtomicUsize>,
469    total: Arc<AtomicUsize>,
470    merge_strategy: RatioMergeStrategy,
471    /// Ratios are displayed as `1% (1/100)`; this controls the latter part.
472    display_raw_values: bool,
473}
474
475#[derive(Debug, Clone, Default)]
476pub enum RatioMergeStrategy {
477    #[default]
478    AddPartAddTotal,
479    AddPartSetTotal,
480    SetPartAddTotal,
481}
482
483impl RatioMetrics {
484    /// Create a new [`RatioMetrics`]
485    pub fn new() -> Self {
486        Self {
487            part: Arc::new(AtomicUsize::new(0)),
488            total: Arc::new(AtomicUsize::new(0)),
489            merge_strategy: RatioMergeStrategy::AddPartAddTotal,
490            display_raw_values: true,
491        }
492    }
493
494    pub fn with_merge_strategy(mut self, merge_strategy: RatioMergeStrategy) -> Self {
495        self.merge_strategy = merge_strategy;
496        self
497    }
498
499    pub fn with_display_raw_values(mut self, display_raw_values: bool) -> Self {
500        self.display_raw_values = display_raw_values;
501        self
502    }
503
504    /// Add `n` to the numerator (`part`) value
505    pub fn add_part(&self, n: usize) {
506        self.part.fetch_add(n, Ordering::Relaxed);
507    }
508
509    /// Add `n` to the denominator (`total`) value
510    pub fn add_total(&self, n: usize) {
511        self.total.fetch_add(n, Ordering::Relaxed);
512    }
513
514    /// Set the numerator (`part`) value to `n`, overwriting any existing value
515    pub fn set_part(&self, n: usize) {
516        self.part.store(n, Ordering::Relaxed);
517    }
518
519    /// Set the denominator (`total`) value to `n`, overwriting any existing value
520    pub fn set_total(&self, n: usize) {
521        self.total.store(n, Ordering::Relaxed);
522    }
523
524    /// Merge the value from `other` into `self`
525    pub fn merge(&self, other: &Self) {
526        match self.merge_strategy {
527            RatioMergeStrategy::AddPartAddTotal => {
528                self.add_part(other.part());
529                self.add_total(other.total());
530            }
531            RatioMergeStrategy::AddPartSetTotal => {
532                self.add_part(other.part());
533                self.set_total(other.total());
534            }
535            RatioMergeStrategy::SetPartAddTotal => {
536                self.set_part(other.part());
537                self.add_total(other.total());
538            }
539        }
540    }
541
542    /// Return the numerator (`part`) value
543    pub fn part(&self) -> usize {
544        self.part.load(Ordering::Relaxed)
545    }
546
547    /// Return the denominator (`total`) value
548    pub fn total(&self) -> usize {
549        self.total.load(Ordering::Relaxed)
550    }
551
552    /// Return the strategy used to merge two [`RatioMetrics`] values
553    pub fn merge_strategy(&self) -> &RatioMergeStrategy {
554        &self.merge_strategy
555    }
556
557    /// Whether `Display` for this metric appends the raw `(part/total)` numbers
558    /// alongside the percentage
559    pub fn display_raw_values(&self) -> bool {
560        self.display_raw_values
561    }
562}
563
564impl PartialEq for RatioMetrics {
565    fn eq(&self, other: &Self) -> bool {
566        self.part() == other.part()
567            && self.total() == other.total()
568            && self.display_raw_values == other.display_raw_values
569    }
570}
571
572impl Display for RatioMetrics {
573    /// Format the ratio to a format like '18.26% (220/1150)'
574    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
575        let part = self.part();
576        let total = self.total();
577
578        // Format the ratio first (for example, `6667/10000` -> `66.67%`),
579        // then optionally append the raw values as ` (6.67 K/10.00 K)`.
580        if total == 0 {
581            write!(f, "N/A")?;
582        } else {
583            // Use basis points so we can round with integer math:
584            // e.g. 18.26% has basis point 1826
585            let basis_points = (((part as u128 * 10_000) + (total as u128 / 2))
586                / total as u128) as usize;
587            let whole = basis_points / 100;
588            let fractional = basis_points % 100;
589
590            if fractional == 0 {
591                write!(f, "{whole}%")?;
592            } else if fractional.is_multiple_of(10) {
593                write!(f, "{whole}.{}%", fractional / 10)?;
594            } else {
595                write!(f, "{whole}.{fractional:02}%")?;
596            }
597        }
598
599        if !self.display_raw_values {
600            return Ok(());
601        }
602
603        if total == 0 {
604            if part == 0 {
605                write!(f, " (0/0)")
606            } else {
607                write!(f, " ({}/0)", human_readable_count(part))
608            }
609        } else {
610            write!(
611                f,
612                " ({}/{})",
613                human_readable_count(part),
614                human_readable_count(total)
615            )
616        }
617    }
618}
619
620/// Possible values for a [super::Metric].
621///
622/// Among other differences, the metric types have different ways to
623/// logically interpret their underlying values and some metrics are
624/// so common they are given special treatment.
625#[derive(Debug, Clone)]
626pub enum MetricValue {
627    /// Number of output rows produced: "output_rows" metric
628    OutputRows(Count),
629    /// Elapsed Compute Time: the wall clock time spent in "cpu
630    /// intensive" work.
631    ///
632    /// This measurement represents, roughly:
633    /// ```
634    /// use std::time::Instant;
635    /// let start = Instant::now();
636    /// // ...CPU intensive work here...
637    /// let elapsed_compute = (Instant::now() - start).as_nanos();
638    /// ```
639    ///
640    /// Note 1: Does *not* include time other operators spend
641    /// computing input.
642    ///
643    /// Note 2: *Does* includes time when the thread could have made
644    /// progress but the OS did not schedule it (e.g. due to CPU
645    /// contention), thus making this value different than the
646    /// classical definition of "cpu_time", which is the time reported
647    /// from `clock_gettime(CLOCK_THREAD_CPUTIME_ID, ..)`.
648    ElapsedCompute(Time),
649    /// Number of spills produced: "spill_count" metric
650    SpillCount(Count),
651    /// Total size of spilled bytes produced: "spilled_bytes" metric
652    SpilledBytes(Count),
653    /// Total size of output bytes produced: "output_bytes" metric
654    OutputBytes(Count),
655    /// Total number of output batches produced: "output_batches" metric
656    OutputBatches(Count),
657    /// Total size of spilled rows produced: "spilled_rows" metric
658    SpilledRows(Count),
659    /// Current memory used
660    CurrentMemoryUsage(Gauge),
661    /// Operator defined count.
662    Count {
663        /// The provided name of this metric
664        name: Cow<'static, str>,
665        /// The value of the metric
666        count: Count,
667    },
668    /// Operator defined gauge.
669    Gauge {
670        /// The provided name of this metric
671        name: Cow<'static, str>,
672        /// The value of the metric
673        gauge: Gauge,
674    },
675    /// Operator defined peak memory usage in bytes.
676    PeakMemoryUsage {
677        /// The provided name of this metric
678        name: Cow<'static, str>,
679        /// The value of the metric
680        gauge: Gauge,
681    },
682    /// Operator defined time
683    Time {
684        /// The provided name of this metric
685        name: Cow<'static, str>,
686        /// The value of the metric
687        time: Time,
688    },
689    /// The time at which execution started
690    StartTimestamp(Timestamp),
691    /// The time at which execution ended
692    EndTimestamp(Timestamp),
693    /// Metrics related to scan pruning
694    PruningMetrics {
695        name: Cow<'static, str>,
696        pruning_metrics: PruningMetrics,
697    },
698    /// Metrics that should be displayed as ratio like (42%)
699    Ratio {
700        name: Cow<'static, str>,
701        ratio_metrics: RatioMetrics,
702    },
703    Custom {
704        /// The provided name of this metric
705        name: Cow<'static, str>,
706        /// A custom implementation of the metric value.
707        value: Arc<dyn CustomMetricValue>,
708    },
709}
710
711// Manually implement PartialEq for `MetricValue` because it contains CustomMetricValue in its
712// definition which is a dyn trait. This wouldn't allow us to just derive PartialEq.
713impl PartialEq for MetricValue {
714    fn eq(&self, other: &Self) -> bool {
715        match (self, other) {
716            (MetricValue::OutputRows(count), MetricValue::OutputRows(other)) => {
717                count == other
718            }
719            (MetricValue::ElapsedCompute(time), MetricValue::ElapsedCompute(other)) => {
720                time == other
721            }
722            (MetricValue::SpillCount(count), MetricValue::SpillCount(other)) => {
723                count == other
724            }
725            (MetricValue::SpilledBytes(count), MetricValue::SpilledBytes(other)) => {
726                count == other
727            }
728            (MetricValue::OutputBytes(count), MetricValue::OutputBytes(other)) => {
729                count == other
730            }
731            (MetricValue::OutputBatches(count), MetricValue::OutputBatches(other)) => {
732                count == other
733            }
734            (MetricValue::SpilledRows(count), MetricValue::SpilledRows(other)) => {
735                count == other
736            }
737            (
738                MetricValue::CurrentMemoryUsage(gauge),
739                MetricValue::CurrentMemoryUsage(other),
740            ) => gauge == other,
741            (
742                MetricValue::Count { name, count },
743                MetricValue::Count {
744                    name: other_name,
745                    count: other_count,
746                },
747            ) => name == other_name && count == other_count,
748            (
749                MetricValue::Gauge { name, gauge },
750                MetricValue::Gauge {
751                    name: other_name,
752                    gauge: other_gauge,
753                },
754            )
755            | (
756                MetricValue::PeakMemoryUsage { name, gauge },
757                MetricValue::PeakMemoryUsage {
758                    name: other_name,
759                    gauge: other_gauge,
760                },
761            ) => name == other_name && gauge == other_gauge,
762            (
763                MetricValue::Time { name, time },
764                MetricValue::Time {
765                    name: other_name,
766                    time: other_time,
767                },
768            ) => name == other_name && time == other_time,
769
770            (
771                MetricValue::StartTimestamp(timestamp),
772                MetricValue::StartTimestamp(other),
773            ) => timestamp == other,
774            (MetricValue::EndTimestamp(timestamp), MetricValue::EndTimestamp(other)) => {
775                timestamp == other
776            }
777            (
778                MetricValue::PruningMetrics {
779                    name,
780                    pruning_metrics,
781                },
782                MetricValue::PruningMetrics {
783                    name: other_name,
784                    pruning_metrics: other_pruning_metrics,
785                },
786            ) => {
787                name == other_name
788                    && pruning_metrics.pruned() == other_pruning_metrics.pruned()
789                    && pruning_metrics.matched() == other_pruning_metrics.matched()
790            }
791            (
792                MetricValue::Ratio {
793                    name,
794                    ratio_metrics,
795                },
796                MetricValue::Ratio {
797                    name: other_name,
798                    ratio_metrics: other_ratio_metrics,
799                },
800            ) => name == other_name && ratio_metrics == other_ratio_metrics,
801            (
802                MetricValue::Custom { name, value },
803                MetricValue::Custom {
804                    name: other_name,
805                    value: other_value,
806                },
807            ) => name == other_name && value.is_eq(other_value),
808            // Default case when the two sides do not have the same type.
809            _ => false,
810        }
811    }
812}
813
814impl MetricValue {
815    /// Return the name of this SQL metric
816    pub fn name(&self) -> &str {
817        match self {
818            Self::OutputRows(_) => "output_rows",
819            Self::SpillCount(_) => "spill_count",
820            Self::SpilledBytes(_) => "spilled_bytes",
821            Self::OutputBytes(_) => "output_bytes",
822            Self::OutputBatches(_) => "output_batches",
823            Self::SpilledRows(_) => "spilled_rows",
824            Self::CurrentMemoryUsage(_) => "mem_used",
825            Self::ElapsedCompute(_) => "elapsed_compute",
826            Self::Count { name, .. } => name.borrow(),
827            Self::Gauge { name, .. } | Self::PeakMemoryUsage { name, .. } => {
828                name.borrow()
829            }
830            Self::Time { name, .. } => name.borrow(),
831            Self::StartTimestamp(_) => "start_timestamp",
832            Self::EndTimestamp(_) => "end_timestamp",
833            Self::PruningMetrics { name, .. } => name.borrow(),
834            Self::Ratio { name, .. } => name.borrow(),
835            Self::Custom { name, .. } => name.borrow(),
836        }
837    }
838
839    /// Return the value of the metric as a usize value, used to aggregate metric
840    /// value across partitions.
841    pub fn as_usize(&self) -> usize {
842        match self {
843            Self::OutputRows(count) => count.value(),
844            Self::SpillCount(count) => count.value(),
845            Self::SpilledBytes(bytes) => bytes.value(),
846            Self::OutputBytes(bytes) => bytes.value(),
847            Self::OutputBatches(count) => count.value(),
848            Self::SpilledRows(count) => count.value(),
849            Self::CurrentMemoryUsage(used) => used.value(),
850            Self::ElapsedCompute(time) => time.value(),
851            Self::Count { count, .. } => count.value(),
852            Self::Gauge { gauge, .. } | Self::PeakMemoryUsage { gauge, .. } => {
853                gauge.value()
854            }
855            Self::Time { time, .. } => time.value(),
856            Self::StartTimestamp(timestamp) => timestamp
857                .value()
858                .and_then(|ts| ts.timestamp_nanos_opt())
859                .map(|nanos| nanos as usize)
860                .unwrap_or(0),
861            Self::EndTimestamp(timestamp) => timestamp
862                .value()
863                .and_then(|ts| ts.timestamp_nanos_opt())
864                .map(|nanos| nanos as usize)
865                .unwrap_or(0),
866            // This function is a utility for aggregating metrics, for complex metric
867            // like `PruningMetrics`, this function is not supposed to get called.
868            // Metrics aggregation for them are implemented inside `MetricsSet` directly.
869            Self::PruningMetrics { .. } => 0,
870            // Should not be used. See comments in `PruningMetrics` for details.
871            Self::Ratio { .. } => 0,
872            Self::Custom { value, .. } => value.as_usize(),
873        }
874    }
875
876    /// create a new MetricValue with the same type as `self` suitable
877    /// for accumulating
878    pub fn new_empty(&self) -> Self {
879        match self {
880            Self::OutputRows(_) => Self::OutputRows(Count::new()),
881            Self::SpillCount(_) => Self::SpillCount(Count::new()),
882            Self::SpilledBytes(_) => Self::SpilledBytes(Count::new()),
883            Self::OutputBytes(_) => Self::OutputBytes(Count::new()),
884            Self::OutputBatches(_) => Self::OutputBatches(Count::new()),
885            Self::SpilledRows(_) => Self::SpilledRows(Count::new()),
886            Self::CurrentMemoryUsage(_) => Self::CurrentMemoryUsage(Gauge::new()),
887            Self::ElapsedCompute(_) => Self::ElapsedCompute(Time::new()),
888            Self::Count { name, .. } => Self::Count {
889                name: name.clone(),
890                count: Count::new(),
891            },
892            Self::Gauge { name, .. } => Self::Gauge {
893                name: name.clone(),
894                gauge: Gauge::new(),
895            },
896            Self::PeakMemoryUsage { name, .. } => Self::PeakMemoryUsage {
897                name: name.clone(),
898                gauge: Gauge::new(),
899            },
900            Self::Time { name, .. } => Self::Time {
901                name: name.clone(),
902                time: Time::new(),
903            },
904            Self::StartTimestamp(_) => Self::StartTimestamp(Timestamp::new()),
905            Self::EndTimestamp(_) => Self::EndTimestamp(Timestamp::new()),
906            Self::PruningMetrics { name, .. } => Self::PruningMetrics {
907                name: name.clone(),
908                pruning_metrics: PruningMetrics::new(),
909            },
910            Self::Ratio {
911                name,
912                ratio_metrics,
913            } => {
914                let merge_strategy = ratio_metrics.merge_strategy.clone();
915                Self::Ratio {
916                    name: name.clone(),
917                    ratio_metrics: RatioMetrics::new()
918                        .with_merge_strategy(merge_strategy)
919                        .with_display_raw_values(ratio_metrics.display_raw_values),
920                }
921            }
922            Self::Custom { name, value } => Self::Custom {
923                name: name.clone(),
924                value: value.new_empty(),
925            },
926        }
927    }
928
929    /// Aggregates the value of other to `self`. panic's if the types
930    /// are mismatched or aggregating does not make sense for this
931    /// value
932    ///
933    /// Note this is purposely marked `mut` (even though atomics are
934    /// used) so Rust's type system can be used to ensure the
935    /// appropriate API access. `MetricValues` should be modified
936    /// using the original [`Count`] or [`Time`] they were created
937    /// from.
938    pub fn aggregate(&mut self, other: &Self) {
939        match (self, other) {
940            (Self::OutputRows(count), Self::OutputRows(other_count))
941            | (Self::SpillCount(count), Self::SpillCount(other_count))
942            | (Self::SpilledBytes(count), Self::SpilledBytes(other_count))
943            | (Self::OutputBytes(count), Self::OutputBytes(other_count))
944            | (Self::OutputBatches(count), Self::OutputBatches(other_count))
945            | (Self::SpilledRows(count), Self::SpilledRows(other_count))
946            | (
947                Self::Count { count, .. },
948                Self::Count {
949                    count: other_count, ..
950                },
951            ) => count.add(other_count.value()),
952            (Self::CurrentMemoryUsage(gauge), Self::CurrentMemoryUsage(other_gauge))
953            | (
954                Self::Gauge { gauge, .. },
955                Self::Gauge {
956                    gauge: other_gauge, ..
957                },
958            )
959            | (
960                Self::PeakMemoryUsage { gauge, .. },
961                Self::PeakMemoryUsage {
962                    gauge: other_gauge, ..
963                },
964            ) => gauge.add(other_gauge.value()),
965            (Self::ElapsedCompute(time), Self::ElapsedCompute(other_time))
966            | (
967                Self::Time { time, .. },
968                Self::Time {
969                    time: other_time, ..
970                },
971            ) => time.add(other_time),
972            // timestamps are aggregated by min/max
973            (Self::StartTimestamp(timestamp), Self::StartTimestamp(other_timestamp)) => {
974                timestamp.update_to_min(other_timestamp);
975            }
976            // timestamps are aggregated by min/max
977            (Self::EndTimestamp(timestamp), Self::EndTimestamp(other_timestamp)) => {
978                timestamp.update_to_max(other_timestamp);
979            }
980            (
981                Self::PruningMetrics {
982                    pruning_metrics, ..
983                },
984                Self::PruningMetrics {
985                    pruning_metrics: other_pruning_metrics,
986                    ..
987                },
988            ) => {
989                let pruned = other_pruning_metrics.pruned.load(Ordering::Relaxed);
990                let matched = other_pruning_metrics.matched.load(Ordering::Relaxed);
991                let fully_matched =
992                    other_pruning_metrics.fully_matched.load(Ordering::Relaxed);
993                pruning_metrics.add_pruned(pruned);
994                pruning_metrics.add_matched(matched);
995                pruning_metrics.add_fully_matched(fully_matched);
996            }
997            (
998                Self::Ratio { ratio_metrics, .. },
999                Self::Ratio {
1000                    ratio_metrics: other_ratio_metrics,
1001                    ..
1002                },
1003            ) => {
1004                ratio_metrics.merge(other_ratio_metrics);
1005            }
1006            (
1007                Self::Custom { value, .. },
1008                Self::Custom {
1009                    value: other_value, ..
1010                },
1011            ) => {
1012                value.aggregate(Arc::clone(other_value));
1013            }
1014            m @ (_, _) => {
1015                panic!(
1016                    "Mismatched metric types. Can not aggregate {:?} with value {:?}",
1017                    m.0, m.1
1018                )
1019            }
1020        }
1021    }
1022
1023    /// Returns a number by which to sort metrics by display. Lower
1024    /// numbers are "more useful" (and displayed first)
1025    pub fn display_sort_key(&self) -> u8 {
1026        match self {
1027            // `BaselineMetrics` that is common for most operators
1028            Self::OutputRows(_) => 0,
1029            Self::ElapsedCompute(_) => 1,
1030            Self::OutputBytes(_) => 2,
1031            Self::OutputBatches(_) => 3,
1032            // Other metrics
1033            Self::PruningMetrics { name, .. } => match name.as_ref() {
1034                // The following metrics belong to `DataSourceExec` with a Parquet data source.
1035                // They are displayed in a specific order that reflects the actual pruning process,
1036                // from coarse-grained to fine-grained pruning levels.
1037                //
1038                // You may update these metrics as long as their relative order remains unchanged.
1039                //
1040                // Reference PR: <https://github.com/apache/datafusion/pull/18379>
1041                "files_ranges_pruned_statistics" => 4,
1042                "row_groups_pruned_statistics" => 5,
1043                "row_groups_pruned_bloom_filter" => 6,
1044                "page_index_pages_pruned" => 7,
1045                "page_index_rows_pruned" => 8,
1046                _ => 9,
1047            },
1048            Self::SpillCount(_) => 10,
1049            Self::SpilledBytes(_) => 11,
1050            Self::SpilledRows(_) => 12,
1051            Self::CurrentMemoryUsage(_) => 13,
1052            Self::Count { name, .. } => match name.as_ref() {
1053                // This Parquet page-index metric is a plain Count because it
1054                // records pages that skipped page-index evaluation, not a
1055                // pruned/matched pair. Keep it grouped with the other
1056                // page-index pruning metrics in EXPLAIN output.
1057                "page_index_pages_skipped_by_fully_matched" => 8,
1058                _ => 14,
1059            },
1060            Self::PeakMemoryUsage { .. } => 13,
1061            Self::Gauge { .. } => 15,
1062            Self::Time { .. } => 16,
1063            Self::Ratio { .. } => 17,
1064            Self::StartTimestamp(_) => 18, // show timestamps last
1065            Self::EndTimestamp(_) => 19,
1066            Self::Custom { .. } => 20,
1067        }
1068    }
1069
1070    /// returns true if this metric has a timestamp value
1071    pub fn is_timestamp(&self) -> bool {
1072        matches!(self, Self::StartTimestamp(_) | Self::EndTimestamp(_))
1073    }
1074}
1075
1076impl Display for MetricValue {
1077    /// Prints the value of this metric
1078    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
1079        match self {
1080            Self::OutputRows(count)
1081            | Self::OutputBatches(count)
1082            | Self::SpillCount(count)
1083            | Self::SpilledRows(count)
1084            | Self::Count { count, .. } => {
1085                write!(f, "{count}")
1086            }
1087            Self::SpilledBytes(count) | Self::OutputBytes(count) => {
1088                let readable_count = human_readable_size(count.value());
1089                write!(f, "{readable_count}")
1090            }
1091            Self::CurrentMemoryUsage(gauge) => {
1092                // CurrentMemoryUsage is in bytes, format like SpilledBytes
1093                let readable_size = human_readable_size(gauge.value());
1094                write!(f, "{readable_size}")
1095            }
1096            Self::PeakMemoryUsage { gauge, .. } => {
1097                let readable_size = human_readable_size(gauge.value());
1098                write!(f, "{readable_size}")
1099            }
1100            Self::Gauge { gauge, .. } => {
1101                // Generic gauge metrics - format with human-readable count
1102                write!(f, "{}", human_readable_count(gauge.value()))
1103            }
1104            Self::ElapsedCompute(time) | Self::Time { time, .. } => {
1105                // distinguish between no time recorded and very small
1106                // amount of time recorded
1107                if time.value() > 0 {
1108                    write!(f, "{time}")
1109                } else {
1110                    write!(f, "NOT RECORDED")
1111                }
1112            }
1113            Self::StartTimestamp(timestamp) | Self::EndTimestamp(timestamp) => {
1114                write!(f, "{timestamp}")
1115            }
1116            Self::PruningMetrics {
1117                pruning_metrics, ..
1118            } => {
1119                write!(f, "{pruning_metrics}")
1120            }
1121            Self::Ratio { ratio_metrics, .. } => write!(f, "{ratio_metrics}"),
1122            Self::Custom { value, .. } => {
1123                write!(f, "{value}")
1124            }
1125        }
1126    }
1127}
1128
1129#[cfg(test)]
1130mod tests {
1131    use std::any::Any;
1132
1133    use chrono::TimeZone;
1134    use datafusion_common::units::MB;
1135
1136    use super::*;
1137
1138    #[derive(Debug, Default)]
1139    pub struct CustomCounter {
1140        count: AtomicUsize,
1141    }
1142
1143    impl Display for CustomCounter {
1144        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1145            write!(f, "count: {}", self.count.load(Ordering::Relaxed))
1146        }
1147    }
1148
1149    impl CustomMetricValue for CustomCounter {
1150        fn new_empty(&self) -> Arc<dyn CustomMetricValue> {
1151            Arc::new(CustomCounter::default())
1152        }
1153
1154        fn aggregate(&self, other: Arc<dyn CustomMetricValue + 'static>) {
1155            let other = other.as_any().downcast_ref::<Self>().unwrap();
1156            self.count
1157                .fetch_add(other.count.load(Ordering::Relaxed), Ordering::Relaxed);
1158        }
1159
1160        fn as_any(&self) -> &dyn Any {
1161            self
1162        }
1163
1164        fn is_eq(&self, other: &Arc<dyn CustomMetricValue>) -> bool {
1165            let Some(other) = other.as_any().downcast_ref::<Self>() else {
1166                return false;
1167            };
1168
1169            self.count.load(Ordering::Relaxed) == other.count.load(Ordering::Relaxed)
1170        }
1171    }
1172
1173    fn new_custom_counter(name: &'static str, value: usize) -> MetricValue {
1174        let custom_counter = CustomCounter::default();
1175        custom_counter.count.fetch_add(value, Ordering::Relaxed);
1176
1177        MetricValue::Custom {
1178            name: Cow::Borrowed(name),
1179            value: Arc::new(custom_counter),
1180        }
1181    }
1182
1183    #[test]
1184    fn test_custom_metric_with_mismatching_names() {
1185        let mut custom_val = new_custom_counter("Hi", 1);
1186        let other_custom_val = new_custom_counter("Hello", 1);
1187
1188        // Not equal since the name differs.
1189        assert!(other_custom_val != custom_val);
1190
1191        // Should work even though the name differs
1192        custom_val.aggregate(&other_custom_val);
1193
1194        let expected_val = new_custom_counter("Hi", 2);
1195        assert!(expected_val == custom_val);
1196    }
1197
1198    #[test]
1199    fn test_custom_metric() {
1200        let mut custom_val = new_custom_counter("hi", 11);
1201        let other_custom_val = new_custom_counter("hi", 20);
1202
1203        custom_val.aggregate(&other_custom_val);
1204
1205        assert!(custom_val != other_custom_val);
1206
1207        if let MetricValue::Custom { value, .. } = custom_val {
1208            let counter = value
1209                .as_any()
1210                .downcast_ref::<CustomCounter>()
1211                .expect("Expected CustomCounter");
1212            assert_eq!(counter.count.load(Ordering::Relaxed), 31);
1213        } else {
1214            panic!("Unexpected value");
1215        }
1216    }
1217
1218    #[test]
1219    fn test_display_custom_metric() {
1220        let custom_val = new_custom_counter("hi", 11);
1221        assert_eq!(custom_val.to_string(), "count: 11");
1222    }
1223
1224    #[test]
1225    fn test_display_output_rows() {
1226        let count = Count::new();
1227        let values = vec![
1228            MetricValue::OutputRows(count.clone()),
1229            MetricValue::Count {
1230                name: "my_counter".into(),
1231                count: count.clone(),
1232            },
1233        ];
1234
1235        for value in &values {
1236            assert_eq!("0", value.to_string(), "value {value:?}");
1237        }
1238
1239        count.add(42);
1240        for value in &values {
1241            assert_eq!("42", value.to_string(), "value {value:?}");
1242        }
1243    }
1244
1245    #[test]
1246    fn test_display_spilled_bytes() {
1247        let count = Count::new();
1248        let spilled_byte = MetricValue::SpilledBytes(count.clone());
1249
1250        assert_eq!("0.0 B", spilled_byte.to_string());
1251
1252        count.add((100 * MB) as usize);
1253        assert_eq!("100.0 MB", spilled_byte.to_string());
1254
1255        count.add((0.5 * MB as f64) as usize);
1256        assert_eq!("100.5 MB", spilled_byte.to_string());
1257    }
1258
1259    #[test]
1260    fn test_display_time() {
1261        let time = Time::new();
1262        let values = vec![
1263            MetricValue::ElapsedCompute(time.clone()),
1264            MetricValue::Time {
1265                name: "my_time".into(),
1266                time: time.clone(),
1267            },
1268        ];
1269
1270        // if time is not set, it should not be reported as zero
1271        for value in &values {
1272            assert_eq!("NOT RECORDED", value.to_string(), "value {value:?}");
1273        }
1274
1275        time.add_duration(Duration::from_nanos(1042));
1276        for value in &values {
1277            assert_eq!("1.04µs", value.to_string(), "value {value:?}");
1278        }
1279    }
1280
1281    #[test]
1282    fn test_display_ratio() {
1283        let ratio_metrics = RatioMetrics::new();
1284        let ratio = MetricValue::Ratio {
1285            name: Cow::Borrowed("ratio_metric"),
1286            ratio_metrics: ratio_metrics.clone(),
1287        };
1288
1289        assert_eq!("N/A (0/0)", ratio.to_string());
1290
1291        ratio_metrics.add_part(10);
1292        assert_eq!("N/A (10/0)", ratio.to_string());
1293
1294        ratio_metrics.add_total(40);
1295        assert_eq!("25% (10/40)", ratio.to_string());
1296
1297        let tiny_ratio_metrics = RatioMetrics::new();
1298        let tiny_ratio = MetricValue::Ratio {
1299            name: Cow::Borrowed("tiny_ratio_metric"),
1300            ratio_metrics: tiny_ratio_metrics.clone(),
1301        };
1302        tiny_ratio_metrics.add_part(1);
1303        tiny_ratio_metrics.add_total(3000);
1304        assert_eq!("0.03% (1/3.00 K)", tiny_ratio.to_string());
1305
1306        ratio_metrics.set_part(6667);
1307        ratio_metrics.set_total(10_000);
1308        assert_eq!("66.67% (6.67 K/10.00 K)", ratio.to_string());
1309
1310        let percentage_only = RatioMetrics::new().with_display_raw_values(false);
1311        let ratio = MetricValue::Ratio {
1312            name: Cow::Borrowed("percentage_only"),
1313            ratio_metrics: percentage_only.clone(),
1314        };
1315        assert_eq!("N/A", ratio.to_string());
1316        percentage_only.set_part(6667);
1317        percentage_only.set_total(10_000);
1318        assert_eq!("66.67%", ratio.to_string());
1319    }
1320
1321    #[test]
1322    fn test_ratio_set_methods() {
1323        let ratio_metrics = RatioMetrics::new();
1324
1325        // Ensure set methods don't increment
1326        ratio_metrics.set_part(10);
1327        ratio_metrics.set_part(10);
1328        ratio_metrics.set_total(40);
1329        ratio_metrics.set_total(40);
1330        assert_eq!("25% (10/40)", ratio_metrics.to_string());
1331
1332        let ratio_metrics = RatioMetrics::new();
1333
1334        // Calling set should change the value
1335        ratio_metrics.set_part(10);
1336        ratio_metrics.set_part(30);
1337        ratio_metrics.set_total(40);
1338        ratio_metrics.set_total(50);
1339        assert_eq!("60% (30/50)", ratio_metrics.to_string());
1340    }
1341
1342    #[test]
1343    fn test_ratio_merge_strategy() {
1344        // Test AddPartSetTotal strategy
1345        let ratio_metrics1 =
1346            RatioMetrics::new().with_merge_strategy(RatioMergeStrategy::AddPartSetTotal);
1347
1348        ratio_metrics1.set_part(10);
1349        ratio_metrics1.set_total(40);
1350        assert_eq!("25% (10/40)", ratio_metrics1.to_string());
1351        let ratio_metrics2 =
1352            RatioMetrics::new().with_merge_strategy(RatioMergeStrategy::AddPartSetTotal);
1353        ratio_metrics2.set_part(20);
1354        ratio_metrics2.set_total(40);
1355        assert_eq!("50% (20/40)", ratio_metrics2.to_string());
1356
1357        ratio_metrics1.merge(&ratio_metrics2);
1358        assert_eq!("75% (30/40)", ratio_metrics1.to_string());
1359
1360        // Test SetPartAddTotal strategy
1361        let ratio_metrics1 =
1362            RatioMetrics::new().with_merge_strategy(RatioMergeStrategy::SetPartAddTotal);
1363        ratio_metrics1.set_part(20);
1364        ratio_metrics1.set_total(50);
1365        let ratio_metrics2 = RatioMetrics::new();
1366        ratio_metrics2.set_part(20);
1367        ratio_metrics2.set_total(50);
1368        ratio_metrics1.merge(&ratio_metrics2);
1369        assert_eq!("20% (20/100)", ratio_metrics1.to_string());
1370
1371        // Test AddPartAddTotal strategy (default)
1372        let ratio_metrics1 = RatioMetrics::new();
1373        ratio_metrics1.set_part(20);
1374        ratio_metrics1.set_total(50);
1375        let ratio_metrics2 = RatioMetrics::new();
1376        ratio_metrics2.set_part(20);
1377        ratio_metrics2.set_total(50);
1378        ratio_metrics1.merge(&ratio_metrics2);
1379        assert_eq!("40% (40/100)", ratio_metrics1.to_string());
1380    }
1381
1382    #[test]
1383    fn test_display_timestamp() {
1384        let timestamp = Timestamp::new();
1385        let values = vec![
1386            MetricValue::StartTimestamp(timestamp.clone()),
1387            MetricValue::EndTimestamp(timestamp.clone()),
1388        ];
1389
1390        // if time is not set, it should not be reported as zero
1391        for value in &values {
1392            assert_eq!("NONE", value.to_string(), "value {value:?}");
1393        }
1394
1395        timestamp.set(Utc.timestamp_nanos(1431648000000000));
1396        for value in &values {
1397            assert_eq!(
1398                "1970-01-17 13:40:48 UTC",
1399                value.to_string(),
1400                "value {value:?}"
1401            );
1402        }
1403    }
1404
1405    #[test]
1406    fn test_timer_with_custom_instant() {
1407        let time = Time::new();
1408        let start_time = Instant::now();
1409
1410        // Sleep a bit to ensure some time passes
1411        std::thread::sleep(Duration::from_millis(1));
1412
1413        // Create timer with the earlier start time
1414        let mut timer = time.timer_with(start_time);
1415
1416        // Sleep a bit more
1417        std::thread::sleep(Duration::from_millis(1));
1418
1419        // Stop the timer
1420        timer.stop();
1421
1422        // The recorded time should be at least 20ms (both sleeps)
1423        assert!(
1424            time.value() >= 2_000_000,
1425            "Expected at least 2ms, got {} ns",
1426            time.value()
1427        );
1428    }
1429
1430    #[test]
1431    fn test_stop_with_custom_endpoint() {
1432        let time = Time::new();
1433        let start = Instant::now();
1434        let mut timer = time.timer_with(start);
1435
1436        // Simulate exactly 10ms passing
1437        let end = start + Duration::from_millis(10);
1438
1439        // Stop with custom endpoint
1440        timer.stop_with(end);
1441
1442        // Should record exactly 10ms (10_000_000 nanoseconds)
1443        // Allow for small variations due to timer resolution
1444        let recorded = time.value();
1445        assert!(
1446            (10_000_000..=10_100_000).contains(&recorded),
1447            "Expected ~10ms, got {recorded} ns"
1448        );
1449
1450        // Calling stop_with again should not add more time
1451        timer.stop_with(end);
1452        assert_eq!(
1453            recorded,
1454            time.value(),
1455            "Time should not change after second stop"
1456        );
1457    }
1458
1459    #[test]
1460    fn test_done_with_custom_endpoint() {
1461        let time = Time::new();
1462        let start = Instant::now();
1463
1464        // Create a new scope for the timer
1465        {
1466            let timer = time.timer_with(start);
1467
1468            // Simulate 50ms passing
1469            let end = start + Duration::from_millis(5);
1470
1471            // Call done_with to stop and consume the timer
1472            timer.done_with(end);
1473
1474            // Timer is consumed, can't use it anymore
1475        }
1476
1477        // Should record exactly 5ms
1478        let recorded = time.value();
1479        assert!(
1480            (5_000_000..=5_100_000).contains(&recorded),
1481            "Expected ~5ms, got {recorded} ns",
1482        );
1483
1484        // Test that done_with prevents drop from recording time again
1485        {
1486            let timer2 = time.timer_with(start);
1487            let end2 = start + Duration::from_millis(5);
1488            timer2.done_with(end2);
1489            // drop happens here but should not record additional time
1490        }
1491
1492        // Should have added only 5ms more
1493        let new_recorded = time.value();
1494        assert!(
1495            (10_000_000..=10_100_000).contains(&new_recorded),
1496            "Expected ~10ms total, got {new_recorded} ns",
1497        );
1498    }
1499
1500    #[test]
1501    fn test_human_readable_metric_formatting() {
1502        // Test Count formatting with various sizes
1503        let small_count = Count::new();
1504        small_count.add(42);
1505        assert_eq!(
1506            MetricValue::OutputRows(small_count.clone()).to_string(),
1507            "42"
1508        );
1509
1510        let thousand_count = Count::new();
1511        thousand_count.add(10_100);
1512        assert_eq!(
1513            MetricValue::OutputRows(thousand_count.clone()).to_string(),
1514            "10.10 K"
1515        );
1516
1517        let million_count = Count::new();
1518        million_count.add(1_532_000);
1519        assert_eq!(
1520            MetricValue::SpilledRows(million_count.clone()).to_string(),
1521            "1.53 M"
1522        );
1523
1524        let billion_count = Count::new();
1525        billion_count.add(2_500_000_000);
1526        assert_eq!(
1527            MetricValue::OutputBatches(billion_count.clone()).to_string(),
1528            "2.50 B"
1529        );
1530
1531        // Test Time formatting with various durations
1532        let micros_time = Time::new();
1533        micros_time.add_duration(Duration::from_nanos(1_234));
1534        assert_eq!(
1535            MetricValue::ElapsedCompute(micros_time.clone()).to_string(),
1536            "1.23µs"
1537        );
1538
1539        let millis_time = Time::new();
1540        millis_time.add_duration(Duration::from_nanos(11_295_377));
1541        assert_eq!(
1542            MetricValue::ElapsedCompute(millis_time.clone()).to_string(),
1543            "11.30ms"
1544        );
1545
1546        let seconds_time = Time::new();
1547        seconds_time.add_duration(Duration::from_nanos(1_234_567_890));
1548        assert_eq!(
1549            MetricValue::ElapsedCompute(seconds_time.clone()).to_string(),
1550            "1.23s"
1551        );
1552
1553        // Test CurrentMemoryUsage formatting (should use size, not count)
1554        let mem_gauge = Gauge::new();
1555        mem_gauge.add(100 * MB as usize);
1556        assert_eq!(
1557            MetricValue::CurrentMemoryUsage(mem_gauge.clone()).to_string(),
1558            "100.0 MB"
1559        );
1560
1561        // Test PeakMemoryUsage formatting (should use size, not count)
1562        let peak_mem_gauge = Gauge::new();
1563        peak_mem_gauge.add(100 * MB as usize);
1564        assert_eq!(
1565            MetricValue::PeakMemoryUsage {
1566                name: "peak_mem_used".into(),
1567                gauge: peak_mem_gauge.clone()
1568            }
1569            .to_string(),
1570            "100.0 MB"
1571        );
1572
1573        // Test custom Gauge formatting (should use count)
1574        let custom_gauge = Gauge::new();
1575        custom_gauge.add(50_000);
1576        assert_eq!(
1577            MetricValue::Gauge {
1578                name: "custom".into(),
1579                gauge: custom_gauge.clone()
1580            }
1581            .to_string(),
1582            "50.00 K"
1583        );
1584
1585        // Test PruningMetrics formatting
1586        let pruning = PruningMetrics::new();
1587        pruning.add_matched(500_000);
1588        pruning.add_pruned(500_000);
1589        assert_eq!(
1590            MetricValue::PruningMetrics {
1591                name: "test_pruning".into(),
1592                pruning_metrics: pruning.clone()
1593            }
1594            .to_string(),
1595            "1.00 M total → 500.0 K matched"
1596        );
1597
1598        // Test RatioMetrics formatting
1599        let ratio = RatioMetrics::new();
1600        ratio.add_part(250_000);
1601        ratio.add_total(1_000_000);
1602        assert_eq!(
1603            MetricValue::Ratio {
1604                name: "test_ratio".into(),
1605                ratio_metrics: ratio.clone()
1606            }
1607            .to_string(),
1608            "25% (250.0 K/1.00 M)"
1609        );
1610    }
1611}