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

mz_sql/
optimizer_metrics.rs
