1use 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#[derive(Debug, Clone)]
40pub struct Count {
41 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 pub fn new() -> Self {
66 Self {
67 value: Arc::new(AtomicUsize::new(0)),
68 }
69 }
70
71 pub fn add(&self, n: usize) {
73 self.value.fetch_add(n, Ordering::Relaxed);
76 }
77
78 pub fn value(&self) -> usize {
80 self.value.load(Ordering::Relaxed)
81 }
82}
83
84#[derive(Debug, Clone)]
89pub struct Gauge {
90 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 pub fn new() -> Self {
115 Self {
116 value: Arc::new(AtomicUsize::new(0)),
117 }
118 }
119
120 pub fn add(&self, n: usize) {
122 self.value.fetch_add(n, Ordering::Relaxed);
125 }
126
127 pub fn sub(&self, n: usize) {
129 self.value.fetch_sub(n, Ordering::Relaxed);
132 }
133
134 pub fn set_max(&self, n: usize) {
136 self.value.fetch_max(n, Ordering::Relaxed);
137 }
138
139 pub fn set(&self, n: usize) -> usize {
141 self.value.swap(n, Ordering::Relaxed)
144 }
145
146 pub fn value(&self) -> usize {
148 self.value.load(Ordering::Relaxed)
149 }
150}
151
152#[derive(Debug, Clone)]
154pub struct Time {
155 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 pub fn new() -> Self {
181 Self {
182 nanos: Arc::new(AtomicUsize::new(0)),
183 }
184 }
185
186 pub fn add_elapsed(&self, start: Instant) {
188 self.add_duration(start.elapsed());
189 }
190
191 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 pub fn add(&self, other: &Time) {
208 self.add_duration(Duration::from_nanos(other.value() as u64))
209 }
210
211 pub fn timer(&self) -> ScopedTimerGuard<'_> {
215 ScopedTimerGuard {
216 inner: self,
217 start: Some(Instant::now()),
218 }
219 }
220
221 pub fn value(&self) -> usize {
223 self.nanos.load(Ordering::Relaxed)
224 }
225
226 pub fn timer_with(&self, now: Instant) -> ScopedTimerGuard<'_> {
229 ScopedTimerGuard {
230 inner: self,
231 start: Some(now),
232 }
233 }
234}
235
236#[derive(Debug, Clone)]
239pub struct Timestamp {
240 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 pub fn new() -> Self {
253 Self {
254 timestamp: Arc::new(Mutex::new(None)),
255 }
256 }
257
258 pub fn record(&self) {
260 self.set(Utc::now())
261 }
262
263 pub fn set(&self, now: DateTime<Utc>) {
265 *self.timestamp.lock() = Some(now);
266 }
267
268 pub fn value(&self) -> Option<DateTime<Utc>> {
273 *self.timestamp.lock()
274 }
275
276 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 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
318pub struct ScopedTimerGuard<'a> {
322 inner: &'a Time,
323 start: Option<Instant>,
324}
325
326impl ScopedTimerGuard<'_> {
327 pub fn stop(&mut self) {
329 if let Some(start) = self.start.take() {
330 self.inner.add_elapsed(start)
331 }
332 }
333
334 pub fn restart(&mut self) {
336 self.start = Some(Instant::now())
337 }
338
339 pub fn done(mut self) {
341 self.stop()
342 }
343
344 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 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#[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 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 pub fn add_pruned(&self, n: usize) {
421 self.pruned.fetch_add(n, Ordering::Relaxed);
424 }
425
426 pub fn add_matched(&self, n: usize) {
428 self.matched.fetch_add(n, Ordering::Relaxed);
431 }
432
433 pub fn add_fully_matched(&self, n: usize) {
435 self.fully_matched.fetch_add(n, Ordering::Relaxed);
438 }
439
440 pub fn subtract_matched(&self, n: usize) {
442 self.matched.fetch_sub(n, Ordering::Relaxed);
445 }
446
447 pub fn pruned(&self) -> usize {
449 self.pruned.load(Ordering::Relaxed)
450 }
451
452 pub fn matched(&self) -> usize {
454 self.matched.load(Ordering::Relaxed)
455 }
456
457 pub fn fully_matched(&self) -> usize {
459 self.fully_matched.load(Ordering::Relaxed)
460 }
461}
462
463#[derive(Debug, Clone, Default)]
467pub struct RatioMetrics {
468 part: Arc<AtomicUsize>,
469 total: Arc<AtomicUsize>,
470 merge_strategy: RatioMergeStrategy,
471 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 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 pub fn add_part(&self, n: usize) {
506 self.part.fetch_add(n, Ordering::Relaxed);
507 }
508
509 pub fn add_total(&self, n: usize) {
511 self.total.fetch_add(n, Ordering::Relaxed);
512 }
513
514 pub fn set_part(&self, n: usize) {
516 self.part.store(n, Ordering::Relaxed);
517 }
518
519 pub fn set_total(&self, n: usize) {
521 self.total.store(n, Ordering::Relaxed);
522 }
523
524 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 pub fn part(&self) -> usize {
544 self.part.load(Ordering::Relaxed)
545 }
546
547 pub fn total(&self) -> usize {
549 self.total.load(Ordering::Relaxed)
550 }
551
552 pub fn merge_strategy(&self) -> &RatioMergeStrategy {
554 &self.merge_strategy
555 }
556
557 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 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
575 let part = self.part();
576 let total = self.total();
577
578 if total == 0 {
581 write!(f, "N/A")?;
582 } else {
583 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#[derive(Debug, Clone)]
626pub enum MetricValue {
627 OutputRows(Count),
629 ElapsedCompute(Time),
649 SpillCount(Count),
651 SpilledBytes(Count),
653 OutputBytes(Count),
655 OutputBatches(Count),
657 SpilledRows(Count),
659 CurrentMemoryUsage(Gauge),
661 Count {
663 name: Cow<'static, str>,
665 count: Count,
667 },
668 Gauge {
670 name: Cow<'static, str>,
672 gauge: Gauge,
674 },
675 PeakMemoryUsage {
677 name: Cow<'static, str>,
679 gauge: Gauge,
681 },
682 Time {
684 name: Cow<'static, str>,
686 time: Time,
688 },
689 StartTimestamp(Timestamp),
691 EndTimestamp(Timestamp),
693 PruningMetrics {
695 name: Cow<'static, str>,
696 pruning_metrics: PruningMetrics,
697 },
698 Ratio {
700 name: Cow<'static, str>,
701 ratio_metrics: RatioMetrics,
702 },
703 Custom {
704 name: Cow<'static, str>,
706 value: Arc<dyn CustomMetricValue>,
708 },
709}
710
711impl 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 _ => false,
810 }
811 }
812}
813
814impl MetricValue {
815 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 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 Self::PruningMetrics { .. } => 0,
870 Self::Ratio { .. } => 0,
872 Self::Custom { value, .. } => value.as_usize(),
873 }
874 }
875
876 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 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 (Self::StartTimestamp(timestamp), Self::StartTimestamp(other_timestamp)) => {
974 timestamp.update_to_min(other_timestamp);
975 }
976 (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 pub fn display_sort_key(&self) -> u8 {
1026 match self {
1027 Self::OutputRows(_) => 0,
1029 Self::ElapsedCompute(_) => 1,
1030 Self::OutputBytes(_) => 2,
1031 Self::OutputBatches(_) => 3,
1032 Self::PruningMetrics { name, .. } => match name.as_ref() {
1034 "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 "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, Self::EndTimestamp(_) => 19,
1066 Self::Custom { .. } => 20,
1067 }
1068 }
1069
1070 pub fn is_timestamp(&self) -> bool {
1072 matches!(self, Self::StartTimestamp(_) | Self::EndTimestamp(_))
1073 }
1074}
1075
1076impl Display for MetricValue {
1077 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 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 write!(f, "{}", human_readable_count(gauge.value()))
1103 }
1104 Self::ElapsedCompute(time) | Self::Time { time, .. } => {
1105 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 assert!(other_custom_val != custom_val);
1190
1191 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 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 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 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 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 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 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 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 std::thread::sleep(Duration::from_millis(1));
1412
1413 let mut timer = time.timer_with(start_time);
1415
1416 std::thread::sleep(Duration::from_millis(1));
1418
1419 timer.stop();
1421
1422 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 let end = start + Duration::from_millis(10);
1438
1439 timer.stop_with(end);
1441
1442 let recorded = time.value();
1445 assert!(
1446 (10_000_000..=10_100_000).contains(&recorded),
1447 "Expected ~10ms, got {recorded} ns"
1448 );
1449
1450 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 {
1466 let timer = time.timer_with(start);
1467
1468 let end = start + Duration::from_millis(5);
1470
1471 timer.done_with(end);
1473
1474 }
1476
1477 let recorded = time.value();
1479 assert!(
1480 (5_000_000..=5_100_000).contains(&recorded),
1481 "Expected ~5ms, got {recorded} ns",
1482 );
1483
1484 {
1486 let timer2 = time.timer_with(start);
1487 let end2 = start + Duration::from_millis(5);
1488 timer2.done_with(end2);
1489 }
1491
1492 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 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 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 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 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 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 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 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}