mz_compute_types/
optimizer_metrics.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Metrics collected by the optimizer.
11
12use std::sync::Arc;
13use std::sync::atomic::{AtomicU64, Ordering};
14use std::time::Duration;
15
16use mz_ore::metric;
17use mz_ore::metrics::MetricsRegistry;
18use mz_ore::metrics::raw::{HistogramVec, IntCounterVec};
19use mz_ore::stats::histogram_seconds_buckets;
20
21use crate::plan::LoweringMetrics;
22
23/// Optimizer metrics.
24#[derive(Debug, Clone)]
25pub struct OptimizerMetrics {
26    e2e_optimization_time_seconds: HistogramVec,
27    /// Threshold in nanoseconds above which optimization emits a "slow
28    /// optimization" `warn!`. Zero disables the warning. Shared via `Arc` so a
29    /// runtime threshold change reaches every clone, including the long-lived
30    /// copies held by the coordinator and `PeekClient`.
31    e2e_optimization_time_seconds_log_threshold: Arc<AtomicU64>,
32    outer_join_lowering_cases: IntCounterVec,
33    transform_hits: IntCounterVec,
34    transform_total: IntCounterVec,
35    /// Local storage of transform times; these are emitted as part of the
36    /// log-line when end-to-end optimization times exceed the configured threshold.
37    transform_time_seconds: std::collections::BTreeMap<String, Vec<Duration>>,
38    /// Metrics recorded during MIR to LIR lowering.
39    lowering: LoweringMetrics,
40}
41
42impl OptimizerMetrics {
43    /// Registers the optimizer metrics into `registry`.
44    pub fn register_into(
45        registry: &MetricsRegistry,
46        e2e_optimization_time_seconds_log_threshold: Duration,
47    ) -> Self {
48        Self {
49            e2e_optimization_time_seconds: registry.register(metric!(
50                 name: "mz_optimizer_e2e_optimization_time_seconds",
51                 help: "A histogram of end-to-end optimization times since restart.",
52                 var_labels: ["object_type"],
53                 buckets: histogram_seconds_buckets(0.000_128, 8.0),
54            )),
55            e2e_optimization_time_seconds_log_threshold: Arc::new(AtomicU64::new(
56                duration_to_nanos(e2e_optimization_time_seconds_log_threshold),
57            )),
58            outer_join_lowering_cases: registry.register(metric!(
59                name: "outer_join_lowering_cases",
60                help: "How many times the different outer join lowering cases happened.",
61                var_labels: ["case"],
62            )),
63            transform_hits: registry.register(metric!(
64                name: "transform_hits",
65                help: "How many times a given transform changed the plan.",
66                var_labels: ["transform"],
67            )),
68            transform_total: registry.register(metric!(
69                name: "transform_total",
70                help: "How many times a given transform was applied.",
71                var_labels: ["transform"],
72            )),
73            transform_time_seconds: std::collections::BTreeMap::new(),
74            lowering: LoweringMetrics::register_into(registry),
75        }
76    }
77
78    /// Updates the "slow optimization" warning threshold. Shared via `Arc`, so
79    /// the change reaches every existing clone.
80    pub fn set_e2e_optimization_time_log_threshold(&self, threshold: Duration) {
81        self.e2e_optimization_time_seconds_log_threshold
82            .store(duration_to_nanos(threshold), Ordering::Relaxed);
83    }
84
85    /// The metrics recorded during MIR to LIR lowering.
86    pub fn lowering(&self) -> &LoweringMetrics {
87        &self.lowering
88    }
89
90    /// Records an end-to-end optimization time for `object_type`, and emits a
91    /// "slow optimization" `warn!` when it exceeds the configured threshold.
92    pub fn observe_e2e_optimization_time(&self, object_type: &str, duration: Duration) {
93        self.e2e_optimization_time_seconds
94            .with_label_values(&[object_type])
95            .observe(duration.as_secs_f64());
96        // Log it when it's big. Zero disables the warning, matching the
97        // `optimizer_e2e_latency_warning_threshold` var contract.
98        let configured = Duration::from_nanos(
99            self.e2e_optimization_time_seconds_log_threshold
100                .load(Ordering::Relaxed),
101        );
102        if configured.is_zero() {
103            return;
104        }
105        let debug_threshold = cfg!(debug_assertions);
106        let threshold = if debug_threshold {
107            // Debug builds are much slower to optimize (despite mz-transform being built
108            // with `opt-level = 3` even in debug builds), so we have a larger threshold.
109            // (A big part of the slowness comes from not optimizing mz-expr, but turning on
110            // optimizations for that in debug builds would slow down the build
111            // considerably.)
112            configured * 6
113        } else {
114            configured
115        };
116        if duration > threshold {
117            let transform_times = self
118                .transform_time_seconds
119                .iter()
120                .map(|(k, v)| {
121                    (
122                        k,
123                        v.into_iter()
124                            .map(|duration| duration.as_micros())
125                            .collect::<Vec<_>>(),
126                    )
127                })
128                .collect::<Vec<_>>();
129            let threshold_string = if debug_threshold {
130                format!("{}ms (debug)", threshold.as_millis())
131            } else {
132                format!("{}ms", threshold.as_millis())
133            };
134            tracing::warn!(
135                duration = format!("{}ms", duration.as_millis()),
136                threshold = threshold_string,
137                object_type = object_type,
138                transform_times_μs = serde_json::to_string(&transform_times)
139                    .unwrap_or_else(|_| format!("{:?}", transform_times)),
140                "slow optimization",
141            );
142        }
143    }
144
145    /// Records that outer join lowering took the `case` path.
146    pub fn inc_outer_join_lowering(&self, case: &str) {
147        self.outer_join_lowering_cases
148            .with_label_values(&[case])
149            .inc()
150    }
151
152    /// Records an application of `transform`, and whether it changed the plan.
153    pub fn inc_transform(&self, hit: bool, transform: &str) {
154        if hit {
155            self.transform_hits.with_label_values(&[transform]).inc();
156        }
157        self.transform_total.with_label_values(&[transform]).inc();
158    }
159
160    /// Records a run time of `transform` for the "slow optimization" log line.
161    pub fn observe_transform_time(&mut self, transform: &str, duration: Duration) {
162        let transform_time_seconds = &mut self.transform_time_seconds;
163        if let Some(times) = transform_time_seconds.get_mut(transform) {
164            times.push(duration);
165        } else {
166            transform_time_seconds.insert(transform.to_string(), vec![duration]);
167        }
168    }
169}
170
171/// Saturates at `u64::MAX` nanoseconds (~584 years), beyond any real threshold.
172fn duration_to_nanos(duration: Duration) -> u64 {
173    u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX)
174}
mz_compute_types/optimizer_metrics.rs

mz_compute_types/
optimizer_metrics.rs
