## Coverage Report

#### Created: 2024-10-13 08:39

    /Users/andrewlamb/Software/datafusion/datafusion/physical-plan/src/metrics/value.rs

    Line
    Count
    Source (jump to first uncovered line)
    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
    20
    use std::{
    21
        borrow::{Borrow, Cow},
    22
        fmt::Display,
    23
        sync::{
    24
            atomic::{AtomicUsize, Ordering},
    25
            Arc,
    26
        },
    27
        time::Duration,
    28
    };
    29
    30
    use chrono::{DateTime, Utc};
    31
    use datafusion_common::instant::Instant;
    32
    use parking_lot::Mutex;
    33
    34
    /// A counter to record things such as number of input or output rows
    35
    ///
    36
    /// Note `clone`ing counters update the same underlying metrics
    37
    #[derive(Debug, Clone)]
    38
    pub struct Count {
    39
        /// value of the metric counter
    40
        value: std::sync::Arc<AtomicUsize>,
    41
    }
    42
    43
    impl PartialEq for Count {
    44
    1
        fn eq(&self, other: &Self) -> bool {
    45
    1
            self.value().eq(&other.value())
    46
    1
        }
    47
    }
    48
    49
    impl Display for Count {
    50
    8
        fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
    51
    8
            write!(f, "{}", self.value())
    52
    8
        }
    53
    }
    54
    55
    impl Default for Count {
    56
    7
        fn default() -> Self {
    57
    7
            Self::new()
    58
    7
        }
    59
    }
    60
    61
    impl Count {
    62
        /// create a new counter
    63
    20.0k
        pub fn new() -> Self {
    64
    20.0k
            Self {
    65
    20.0k
                value: Arc::new(AtomicUsize::new(0)),
    66
    20.0k
            }
    67
    20.0k
        }
    68
    69
        /// Add `n` to the metric's value
    70
    129k
        pub fn add(&self, n: usize) {
    71
    129k
            // relaxed ordering for operations on `value` poses no issues
    72
    129k
            // we're purely using atomic ops with no associated memory ops
    73
    129k
            self.value.fetch_add(n, Ordering::Relaxed);
    74
    129k
        }
    75
    76
        /// Get the current value
    77
    413
        pub fn value(&self) -> usize {
    78
    413
            self.value.load(Ordering::Relaxed)
    79
    413
        }
    80
    }
    81
    82
    /// A gauge is the simplest metrics type. It just returns a value.
    83
    /// For example, you can easily expose current memory consumption with a gauge.
    84
    ///
    85
    /// Note `clone`ing gauge update the same underlying metrics
    86
    #[derive(Debug, Clone)]
    87
    pub struct Gauge {
    88
        /// value of the metric gauge
    89
        value: std::sync::Arc<AtomicUsize>,
    90
    }
    91
    92
    impl PartialEq for Gauge {
    93
    0
        fn eq(&self, other: &Self) -> bool {
    94
    0
            self.value().eq(&other.value())
    95
    0
        }
    96
    }
    97
    98
    impl Display for Gauge {
    99
    0
        fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
    100
    0
            write!(f, "{}", self.value())
    101
    0
        }
    102
    }
    103
    104
    impl Default for Gauge {
    105
    0
        fn default() -> Self {
    106
    0
            Self::new()
    107
    0
        }
    108
    }
    109
    110
    impl Gauge {
    111
        /// create a new gauge
    112
    3.24k
        pub fn new() -> Self {
    113
    3.24k
            Self {
    114
    3.24k
                value: Arc::new(AtomicUsize::new(0)),
    115
    3.24k
            }
    116
    3.24k
        }
    117
    118
        /// Add `n` to the metric's value
    119
    19.0k
        pub fn add(&self, n: usize) {
    120
    19.0k
            // relaxed ordering for operations on `value` poses no issues
    121
    19.0k
            // we're purely using atomic ops with no associated memory ops
    122
    19.0k
            self.value.fetch_add(n, Ordering::Relaxed);
    123
    19.0k
        }
    124
    125
        /// Sub `n` from the metric's value
    126
    0
        pub fn sub(&self, n: usize) {
    127
    0
            // relaxed ordering for operations on `value` poses no issues
    128
    0
            // we're purely using atomic ops with no associated memory ops
    129
    0
            self.value.fetch_sub(n, Ordering::Relaxed);
    130
    0
        }
    131
    132
        /// Set metric's value to maximum of `n` and current value
    133
    82
        pub fn set_max(&self, n: usize) {
    134
    82
            self.value.fetch_max(n, Ordering::Relaxed);
    135
    82
        }
    136
    137
        /// Set the metric's value to `n` and return the previous value
    138
    8.11k
        pub fn set(&self, n: usize) -> usize {
    139
    8.11k
            // relaxed ordering for operations on `value` poses no issues
    140
    8.11k
            // we're purely using atomic ops with no associated memory ops
    141
    8.11k
            self.value.swap(n, Ordering::Relaxed)
    142
    8.11k
        }
    143
    144
        /// Get the current value
    145
    0
        pub fn value(&self) -> usize {
    146
    0
            self.value.load(Ordering::Relaxed)
    147
    0
        }
    148
    }
    149
    150
    /// Measure a potentially non contiguous duration of time
    151
    #[derive(Debug, Clone)]
    152
    pub struct Time {
    153
        /// elapsed time, in nanoseconds
    154
        nanos: Arc<AtomicUsize>,
    155
    }
    156
    157
    impl Default for Time {
    158
    0
        fn default() -> Self {
    159
    0
            Self::new()
    160
    0
        }
    161
    }
    162
    163
    impl PartialEq for Time {
    164
    0
        fn eq(&self, other: &Self) -> bool {
    165
    0
            self.value().eq(&other.value())
    166
    0
        }
    167
    }
    168
    169
    impl Display for Time {
    170
    2
        fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
    171
    2
            let duration = std::time::Duration::from_nanos(self.value() as u64);
    172
    2
            write!(f, "{duration:?}")
    173
    2
        }
    174
    }
    175
    176
    impl Time {
    177
        /// Create a new [`Time`] wrapper suitable for recording elapsed
    178
        /// times for operations.
    179
    12.3k
        pub fn new() -> Self {
    180
    12.3k
            Self {
    181
    12.3k
                nanos: Arc::new(AtomicUsize::new(0)),
    182
    12.3k
            }
    183
    12.3k
        }
    184
    185
        /// Add elapsed nanoseconds since `start`to self
    186
    84.3k
        pub fn add_elapsed(&self, start: Instant) {
    187
    84.3k
            self.add_duration(start.elapsed());
    188
    84.3k
        }
    189
    190
        /// Add duration of time to self
    191
        ///
    192
        /// Note: this will always increment the recorded time by at least 1 nanosecond
    193
        /// to distinguish between the scenario of no values recorded, in which
    194
        /// case the value will be 0, and no measurable amount of time having passed,
    195
        /// in which case the value will be small but not 0.
    196
        ///
    197
        /// This is based on the assumption that the timing logic in most cases is likely
    198
        /// to take at least a nanosecond, and so this is reasonable mechanism to avoid
    199
        /// ambiguity, especially on systems with low-resolution monotonic clocks
    200
    84.3k
        pub fn add_duration(&self, duration: Duration) {
    201
    84.3k
            let more_nanos = duration.as_nanos() as usize;
    202
    84.3k
            self.nanos.fetch_add(more_nanos.max(1), Ordering::Relaxed);
    203
    84.3k
        }
    204
    205
        /// Add the number of nanoseconds of other `Time` to self
    206
    10
        pub fn add(&self, other: &Time) {
    207
    10
            self.add_duration(Duration::from_nanos(other.value() as u64))
    208
    10
        }
    209
    210
        /// return a scoped guard that adds the amount of time elapsed
    211
        /// between its creation and its drop or call to `stop` to the
    212
        /// underlying metric.
    213
    84.3k
        pub fn timer(&self) -> ScopedTimerGuard<'_> {
    214
    84.3k
            ScopedTimerGuard {
    215
    84.3k
                inner: self,
    216
    84.3k
                start: Some(Instant::now()),
    217
    84.3k
            }
    218
    84.3k
        }
    219
    220
        /// Get the number of nanoseconds record by this Time metric
    221
    23
        pub fn value(&self) -> usize {
    222
    23
            self.nanos.load(Ordering::Relaxed)
    223
    23
        }
    224
    }
    225
    226
    /// Stores a single timestamp, stored as the number of nanoseconds
    227
    /// elapsed from Jan 1, 1970 UTC
    228
    #[derive(Debug, Clone)]
    229
    pub struct Timestamp {
    230
        /// Time thing started
    231
        timestamp: Arc<Mutex<Option<DateTime<Utc>>>>,
    232
    }
    233
    234
    impl Default for Timestamp {
    235
    7
        fn default() -> Self {
    236
    7
            Self::new()
    237
    7
        }
    238
    }
    239
    240
    impl Timestamp {
    241
        /// Create a new timestamp and sets its value to 0
    242
    400
        pub fn new() -> Self {
    243
    400
            Self {
    244
    400
                timestamp: Arc::new(Mutex::new(None)),
    245
    400
            }
    246
    400
        }
    247
    248
        /// Sets the timestamps value to the current time
    249
    393
        pub fn record(&self) {
    250
    393
            self.set(Utc::now())
    251
    393
        }
    252
    253
        /// Sets the timestamps value to a specified time
    254
    398
        pub fn set(&self, now: DateTime<Utc>) {
    255
    398
            *self.timestamp.lock() = Some(now);
    256
    398
        }
    257
    258
        /// return the timestamps value at the last time `record()` was
    259
        /// called.
    260
        ///
    261
        /// Returns `None` if `record()` has not been called
    262
    237
        pub fn value(&self) -> Option<DateTime<Utc>> {
    263
    237
            *self.timestamp.lock()
    264
    237
        }
    265
    266
        /// sets the value of this timestamp to the minimum of this and other
    267
    2
        pub fn update_to_min(&self, other: &Timestamp) {
    268
    2
            let min = match (self.value(), other.value()) {
    269
    0
                (None, None) => None,
    270
    0
                (Some(v), None) => Some(v),
    271
    1
                (None, Some(v)) => Some(v),
    272
    1
                (Some(v1), Some(v2)) => Some(if v1 < v2 { v1 } else { v20 }),
    273
            };
    274
    275
    2
            *self.timestamp.lock() = min;
    276
    2
        }
    277
    278
        /// sets the value of this timestamp to the maximum of this and other
    279
    2
        pub fn update_to_max(&self, other: &Timestamp) {
    280
    2
            let max = match (self.value(), other.value()) {
    281
    0
                (None, None) => None,
    282
    0
                (Some(v), None) => Some(v),
    283
    1
                (None, Some(v)) => Some(v),
    284
    1
                (Some(v1), Some(v2)) => Some(if v1 < v2 { v2 } else { v10 }),
    285
            };
    286
    287
    2
            *self.timestamp.lock() = max;
    288
    2
        }
    289
    }
    290
    291
    impl PartialEq for Timestamp {
    292
    0
        fn eq(&self, other: &Self) -> bool {
    293
    0
            self.value().eq(&other.value())
    294
    0
        }
    295
    }
    296
    297
    impl Display for Timestamp {
    298
    4
        fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
    299
    4
            match self.value() {
    300
    2
                None => write!(f, "NONE"),
    301
    2
                Some(v) => {
    302
    2
                    write!(f, "{v}")
    303
                }
    304
            }
    305
    4
        }
    306
    }
    307
    308
    /// RAAI structure that adds all time between its construction and
    309
    /// destruction to the CPU time or the first call to `stop` whichever
    310
    /// comes first
    311
    pub struct ScopedTimerGuard<'a> {
    312
        inner: &'a Time,
    313
        start: Option<Instant>,
    314
    }
    315
    316
    impl<'a> ScopedTimerGuard<'a> {
    317
        /// Stop the timer timing and record the time taken
    318
    148k
        pub fn stop(&mut self) {
    319
    148k
            if let Some(start84.3k) = self.start.take() {
    320
    84.3k
                self.inner.add_elapsed(start)
    321
    64.2k
            }
    322
    148k
        }
    323
    324
        /// Restarts the timer recording from the current time
    325
    0
        pub fn restart(&mut self) {
    326
    0
            self.start = Some(Instant::now())
    327
    0
        }
    328
    329
        /// Stop the timer, record the time taken and consume self
    330
    64.2k
        pub fn done(mut self) {
    331
    64.2k
            self.stop()
    332
    64.2k
        }
    333
    }
    334
    335
    impl<'a> Drop for ScopedTimerGuard<'a> {
    336
    84.3k
        fn drop(&mut self) {
    337
    84.3k
            self.stop()
    338
    84.3k
        }
    339
    }
    340
    341
    /// Possible values for a [super::Metric].
    342
    ///
    343
    /// Among other differences, the metric types have different ways to
    344
    /// logically interpret their underlying values and some metrics are
    345
    /// so common they are given special treatment.
    346
    #[derive(Debug, Clone, PartialEq)]
    347
    pub enum MetricValue {
    348
        /// Number of output rows produced: "output_rows" metric
    349
        OutputRows(Count),
    350
        /// Elapsed Compute Time: the wall clock time spent in "cpu
    351
        /// intensive" work.
    352
        ///
    353
        /// This measurement represents, roughly:
    354
        /// ```
    355
        /// use std::time::Instant;
    356
        /// let start = Instant::now();
    357
        /// // ...CPU intensive work here...
    358
        /// let elapsed_compute = (Instant::now() - start).as_nanos();
    359
        /// ```
    360
        ///
    361
        /// Note 1: Does *not* include time other operators spend
    362
        /// computing input.
    363
        ///
    364
        /// Note 2: *Does* includes time when the thread could have made
    365
        /// progress but the OS did not schedule it (e.g. due to CPU
    366
        /// contention), thus making this value different than the
    367
        /// classical definition of "cpu_time", which is the time reported
    368
        /// from `clock_gettime(CLOCK_THREAD_CPUTIME_ID, ..)`.
    369
        ElapsedCompute(Time),
    370
        /// Number of spills produced: "spill_count" metric
    371
        SpillCount(Count),
    372
        /// Total size of spilled bytes produced: "spilled_bytes" metric
    373
        SpilledBytes(Count),
    374
        /// Total size of spilled rows produced: "spilled_rows" metric
    375
        SpilledRows(Count),
    376
        /// Current memory used
    377
        CurrentMemoryUsage(Gauge),
    378
        /// Operator defined count.
    379
        Count {
    380
            /// The provided name of this metric
    381
            name: Cow<'static, str>,
    382
            /// The value of the metric
    383
            count: Count,
    384
        },
    385
        /// Operator defined gauge.
    386
        Gauge {
    387
            /// The provided name of this metric
    388
            name: Cow<'static, str>,
    389
            /// The value of the metric
    390
            gauge: Gauge,
    391
        },
    392
        /// Operator defined time
    393
        Time {
    394
            /// The provided name of this metric
    395
            name: Cow<'static, str>,
    396
            /// The value of the metric
    397
            time: Time,
    398
        },
    399
        /// The time at which execution started
    400
        StartTimestamp(Timestamp),
    401
        /// The time at which execution ended
    402
        EndTimestamp(Timestamp),
    403
    }
    404
    405
    impl MetricValue {
    406
        /// Return the name of this SQL metric
    407
    76
        pub fn name(&self) -> &str {
    408
    76
            match self {
    409
    14
                Self::OutputRows(_) => "output_rows",
    410
    0
                Self::SpillCount(_) => "spill_count",
    411
    0
                Self::SpilledBytes(_) => "spilled_bytes",
    412
    0
                Self::SpilledRows(_) => "spilled_rows",
    413
    0
                Self::CurrentMemoryUsage(_) => "mem_used",
    414
    10
                Self::ElapsedCompute(_) => "elapsed_compute",
    415
    23
                Self::Count { name, .. } => name.borrow(),
    416
    0
                Self::Gauge { name, .. } => name.borrow(),
    417
    7
                Self::Time { name, .. } => name.borrow(),
    418
    11
                Self::StartTimestamp(_) => "start_timestamp",
    419
    11
                Self::EndTimestamp(_) => "end_timestamp",
    420
            }
    421
    76
        }
    422
    423
        /// Return the value of the metric as a usize value
    424
    206
        pub fn as_usize(&self) -> usize {
    425
    206
            match self {
    426
    15
                Self::OutputRows(count) => count.value(),
    427
    62
                Self::SpillCount(count) => count.value(),
    428
    61
                Self::SpilledBytes(bytes) => bytes.value(),
    429
    61
                Self::SpilledRows(count) => count.value(),
    430
    0
                Self::CurrentMemoryUsage(used) => used.value(),
    431
    7
                Self::ElapsedCompute(time) => time.value(),
    432
    0
                Self::Count { count, .. } => count.value(),
    433
    0
                Self::Gauge { gauge, .. } => gauge.value(),
    434
    0
                Self::Time { time, .. } => time.value(),
    435
    0
                Self::StartTimestamp(timestamp) => timestamp
    436
    0
                    .value()
    437
    0
                    .and_then(|ts| ts.timestamp_nanos_opt())
    438
    0
                    .map(|nanos| nanos as usize)
    439
    0
                    .unwrap_or(0),
    440
    0
                Self::EndTimestamp(timestamp) => timestamp
    441
    0
                    .value()
    442
    0
                    .and_then(|ts| ts.timestamp_nanos_opt())
    443
    0
                    .map(|nanos| nanos as usize)
    444
    0
                    .unwrap_or(0),
    445
            }
    446
    206
        }
    447
    448
        /// create a new MetricValue with the same type as `self` suitable
    449
        /// for accumulating
    450
    211
        pub fn new_empty(&self) -> Self {
    451
    211
            match self {
    452
    15
                Self::OutputRows(_) => Self::OutputRows(Count::new()),
    453
    62
                Self::SpillCount(_) => Self::SpillCount(Count::new()),
    454
    61
                Self::SpilledBytes(_) => Self::SpilledBytes(Count::new()),
    455
    61
                Self::SpilledRows(_) => Self::SpilledRows(Count::new()),
    456
    0
                Self::CurrentMemoryUsage(_) => Self::CurrentMemoryUsage(Gauge::new()),
    457
    7
                Self::ElapsedCompute(_) => Self::ElapsedCompute(Time::new()),
    458
    3
                Self::Count { name, .. } => Self::Count {
    459
    3
                    name: name.clone(),
    460
    3
                    count: Count::new(),
    461
    3
                },
    462
    0
                Self::Gauge { name, .. } => Self::Gauge {
    463
    0
                    name: name.clone(),
    464
    0
                    gauge: Gauge::new(),
    465
    0
                },
    466
    0
                Self::Time { name, .. } => Self::Time {
    467
    0
                    name: name.clone(),
    468
    0
                    time: Time::new(),
    469
    0
                },
    470
    1
                Self::StartTimestamp(_) => Self::StartTimestamp(Timestamp::new()),
    471
    1
                Self::EndTimestamp(_) => Self::EndTimestamp(Timestamp::new()),
    472
            }
    473
    211
        }
    474
    475
        /// Aggregates the value of other to `self`. panic's if the types
    476
        /// are mismatched or aggregating does not make sense for this
    477
        /// value
    478
        ///
    479
        /// Note this is purposely marked `mut` (even though atomics are
    480
        /// used) so Rust's type system can be used to ensure the
    481
        /// appropriate API access. `MetricValues` should be modified
    482
        /// using the original [`Count`] or [`Time`] they were created
    483
        /// from.
    484
    220
        pub fn aggregate(&mut self, other: &Self) {
    485
    220
            match (self, other) {
    486
    16
                (Self::OutputRows(count), Self::OutputRows(other_count))
    487
    62
                | (Self::SpillCount(count), Self::SpillCount(other_count))
    488
    61
                | (Self::SpilledBytes(count), Self::SpilledBytes(other_count))
    489
    61
                | (Self::SpilledRows(count), Self::SpilledRows(other_count))
    490
                | (
    491
    4
                    Self::Count { count, .. },
    492
    4
                    Self::Count {
    493
    4
                        count: other_count, ..
    494
                    },
    495
    204
                ) => count.add(other_count.value()),
    496
    0
                (Self::CurrentMemoryUsage(gauge), Self::CurrentMemoryUsage(other_gauge))
    497
                | (
    498
    0
                    Self::Gauge { gauge, .. },
    499
    0
                    Self::Gauge {
    500
    0
                        gauge: other_gauge, ..
    501
                    },
    502
    0
                ) => gauge.add(other_gauge.value()),
    503
    10
                (Self::ElapsedCompute(time), Self::ElapsedCompute(other_time))
    504
                | (
    505
    0
                    Self::Time { time, .. },
    506
    0
                    Self::Time {
    507
    0
                        time: other_time, ..
    508
                    },
    509
    10
                ) => time.add(other_time),
    510
                // timestamps are aggregated by min/max
    511
    2
                (Self::StartTimestamp(timestamp), Self::StartTimestamp(other_timestamp)) => {
    512
    2
                    timestamp.update_to_min(other_timestamp);
    513
    2
                }
    514
                // timestamps are aggregated by min/max
    515
    2
                (Self::EndTimestamp(timestamp), Self::EndTimestamp(other_timestamp)) => {
    516
    2
                    timestamp.update_to_max(other_timestamp);
    517
    2
                }
    518
    2
                m @ (_, _) => {
    519
    2
                    panic!(
    520
    2
                        "Mismatched metric types. Can not aggregate {:?} with value {:?}",
    521
    2
                        m.0, m.1
    522
    2
                    )
    523
                }
    524
            }
    525
    218
        }
    526
    527
        /// Returns a number by which to sort metrics by display. Lower
    528
        /// numbers are "more useful" (and displayed first)
    529
    46
        pub fn display_sort_key(&self) -> u8 {
    530
    46
            match self {
    531
    7
                Self::OutputRows(_) => 0,     // show first
    532
    5
                Self::ElapsedCompute(_) => 1, // show second
    533
    0
                Self::SpillCount(_) => 2,
    534
    0
                Self::SpilledBytes(_) => 3,
    535
    0
                Self::SpilledRows(_) => 4,
    536
    0
                Self::CurrentMemoryUsage(_) => 5,
    537
    16
                Self::Count { .. } => 6,
    538
    0
                Self::Gauge { .. } => 7,
    539
    4
                Self::Time { .. } => 8,
    540
    7
                Self::StartTimestamp(_) => 9, // show timestamps last
    541
    7
                Self::EndTimestamp(_) => 10,
    542
            }
    543
    46
        }
    544
    545
        /// returns true if this metric has a timestamp value
    546
    0
        pub fn is_timestamp(&self) -> bool {
    547
    0
            matches!(self, Self::StartTimestamp(_) | Self::EndTimestamp(_))
    548
    0
        }
    549
    }
    550
    551
    impl Display for MetricValue {
    552
        /// Prints the value of this metric
    553
    16
        fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
    554
    16
            match self {
    555
    6
                Self::OutputRows(count)
    556
    0
                | Self::SpillCount(count)
    557
    0
                | Self::SpilledBytes(count)
    558
    0
                | Self::SpilledRows(count)
    559
    2
                | Self::Count { count, .. } => {
    560
    8
                    write!(f, "{count}")
    561
                }
    562
    0
                Self::CurrentMemoryUsage(gauge) | Self::Gauge { gauge, .. } => {
    563
    0
                    write!(f, "{gauge}")
    564
                }
    565
    2
                Self::ElapsedCompute(time) | Self::Time { time, .. } => {
    566
                    // distinguish between no time recorded and very small
    567
                    // amount of time recorded
    568
    4
                    if time.value() > 0 {
    569
    2
                        write!(f, "{time}")
    570
                    } else {
    571
    2
                        write!(f, "NOT RECORDED")
    572
                    }
    573
                }
    574
    2
                Self::StartTimestamp(timestamp) | Self::EndTimestamp(timestamp) => {
    575
    4
                    write!(f, "{timestamp}")
    576
                }
    577
            }
    578
    16
        }
    579
    }
    580
    581
    #[cfg(test)]
    582
    mod tests {
    583
        use chrono::TimeZone;
    584
    585
        use super::*;
    586
    587
        #[test]
    588
    1
        fn test_display_output_rows() {
    589
    1
            let count = Count::new();
    590
    1
            let values = vec![
    591
    1
                MetricValue::OutputRows(count.clone()),
    592
    1
                MetricValue::Count {
    593
    1
                    name: "my_counter".into(),
    594
    1
                    count: count.clone(),
    595
    1
                },
    596
    1
            ];
    597
    598
    3
            for value2 in &values {
    599
    2
                assert_eq!("0", value.to_string(), "value {value:?}"0);
    600
            }
    601
    602
    1
            count.add(42);
    603
    3
            for value2 in &values {
    604
    2
                assert_eq!("42", value.to_string(), "value {value:?}"0);
    605
            }
    606
    1
        }
    607
    608
        #[test]
    609
    1
        fn test_display_time() {
    610
    1
            let time = Time::new();
    611
    1
            let values = vec![
    612
    1
                MetricValue::ElapsedCompute(time.clone()),
    613
    1
                MetricValue::Time {
    614
    1
                    name: "my_time".into(),
    615
    1
                    time: time.clone(),
    616
    1
                },
    617
    1
            ];
    618
    619
            // if time is not set, it should not be reported as zero
    620
    3
            for value2 in &values {
    621
    2
                assert_eq!("NOT RECORDED", value.to_string(), "value {value:?}"0);
    622
            }
    623
    624
    1
            time.add_duration(Duration::from_nanos(1042));
    625
    3
            for value2 in &values {
    626
    2
                assert_eq!("1.042µs", value.to_string(), "value {value:?}"0);
    627
            }
    628
    1
        }
    629
    630
        #[test]
    631
    1
        fn test_display_timestamp() {
    632
    1
            let timestamp = Timestamp::new();
    633
    1
            let values = vec![
    634
    1
                MetricValue::StartTimestamp(timestamp.clone()),
    635
    1
                MetricValue::EndTimestamp(timestamp.clone()),
    636
    1
            ];
    637
    638
            // if time is not set, it should not be reported as zero
    639
    3
            for value2 in &values {
    640
    2
                assert_eq!("NONE", value.to_string(), "value {value:?}"0);
    641
            }
    642
    643
    1
            timestamp.set(Utc.timestamp_nanos(1431648000000000));
    644
    3
            for value2 in &values {
    645
    2
                assert_eq!(
    646
    2
                    "1970-01-17 13:40:48 UTC",
    647
    2
                    value.to_string(),
    648
    0
                    "value {value:?}"
    649
                );
    650
            }
    651
    1
        }
    652
    }