mz_compute/render/
errors.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//! Helpers for handling errors encountered by operators.
11//!
12//! # `DataflowErrorSer`
13//!
14//! [`DataflowErrorSer`] is a serialized byte representation of
15//! [`DataflowError`] used on compute-internal dataflow edges instead of
16//! `DataflowError` directly.
17//!
18//! It is backed by proto-encoded [`ProtoDataflowError`] bytes. Because proto3 +
19//! prost + no map fields = deterministic encoding, byte-equality implies
20//! semantic equality, which lets us use `Ord`, `Hash`, etc. directly on the
21//! bytes.
22//!
23//! **Invariant**: NEVER add `map` fields to `ProtoDataflowError` or any of its
24//! transitive message types, as map fields have non-deterministic encoding
25//! order in protobuf.
26
27use columnar::Columnar;
28use columnation::{Columnation, Region};
29use mz_expr::EvalError;
30use mz_proto::{ProtoType, RustType};
31use mz_repr::Row;
32use mz_storage_types::errors::{DataflowError, ProtoDataflowError};
33use prost::Message;
34use serde::{Deserialize, Serialize};
35use std::fmt;
36
37/// Serialized representation of a [`DataflowError`], backed by proto-encoded bytes.
38///
39/// This type is used on compute-internal dataflow edges to avoid the cost of
40/// carrying a full `DataflowError` enum through the dataflow graph. Because the
41/// proto encoding is canonical (proto3 + prost + no map fields), byte-equality
42/// implies semantic equality.
43#[derive(
44    Clone,
45    Eq,
46    PartialEq,
47    Ord,
48    PartialOrd,
49    Hash,
50    Serialize,
51    Deserialize,
52    Columnar
53)]
54#[columnar(derive(Eq, PartialEq, Ord, PartialOrd))]
55pub struct DataflowErrorSer(Vec<u8>);
56
57impl DataflowErrorSer {
58    /// Decode the serialized bytes back into a [`DataflowError`].
59    ///
60    /// # Panics
61    ///
62    /// Panics if the bytes do not represent a valid `ProtoDataflowError`.
63    pub fn deserialize(&self) -> DataflowError {
64        let proto = ProtoDataflowError::decode(self.0.as_slice())
65            .expect("DataflowErrorSer: invalid proto bytes");
66        proto
67            .into_rust()
68            .expect("DataflowErrorSer: failed to convert proto to DataflowError")
69    }
70}
71
72impl From<DataflowError> for DataflowErrorSer {
73    fn from(err: DataflowError) -> Self {
74        DataflowErrorSer(err.into_proto().encode_to_vec())
75    }
76}
77
78impl From<EvalError> for DataflowErrorSer {
79    fn from(err: EvalError) -> Self {
80        // Note: this allocates a Box via DataflowError::EvalError(Box::new(e)).
81        // Acceptable in v1.
82        DataflowErrorSer::from(DataflowError::from(err))
83    }
84}
85
86impl fmt::Display for DataflowErrorSer {
87    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
88        self.deserialize().fmt(f)
89    }
90}
91
92impl fmt::Debug for DataflowErrorSer {
93    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
94        f.debug_tuple("DataflowErrorSer")
95            .field(&format_args!("{} bytes", self.0.len()))
96            .finish()
97    }
98}
99
100impl Columnation for DataflowErrorSer {
101    type InnerRegion = DataflowErrorSerRegion;
102}
103
104impl crate::sink::correction_v2::DataBytes for DataflowErrorSer {
105    fn data_bytes(&self) -> usize {
106        std::mem::size_of::<Self>() + self.0.len()
107    }
108}
109
110/// A [`Region`] for [`DataflowErrorSer`], delegating to the region for `Vec<u8>`.
111#[derive(Default)]
112pub struct DataflowErrorSerRegion {
113    inner: <Vec<u8> as Columnation>::InnerRegion,
114}
115
116impl Region for DataflowErrorSerRegion {
117    type Item = DataflowErrorSer;
118
119    unsafe fn copy(&mut self, item: &Self::Item) -> Self::Item {
120        // SAFETY: delegating to the inner Vec<u8> region which handles the allocation.
121        DataflowErrorSer(unsafe { self.inner.copy(&item.0) })
122    }
123
124    fn clear(&mut self) {
125        self.inner.clear();
126    }
127
128    fn reserve_items<'a, I>(&mut self, items: I)
129    where
130        I: Iterator<Item = &'a Self::Item> + Clone,
131    {
132        self.inner.reserve_items(items.map(|item| &item.0));
133    }
134
135    fn reserve_regions<'a, I>(&mut self, regions: I)
136    where
137        I: Iterator<Item = &'a Self> + Clone,
138    {
139        self.inner.reserve_regions(regions.map(|r| &r.inner));
140    }
141
142    fn heap_size(&self, callback: impl FnMut(usize, usize)) {
143        self.inner.heap_size(callback);
144    }
145}
146
147/// Used to make possibly-validating code generic: think of this as a kind of `MaybeResult`,
148/// specialized for use in compute.  Validation code will only run when the error constructor is
149/// Some.
150pub(super) trait MaybeValidatingRow<T, E> {
151    fn ok(t: T) -> Self;
152    fn into_error() -> Option<fn(E) -> Self>;
153}
154
155impl<E> MaybeValidatingRow<Row, E> for Row {
156    fn ok(t: Row) -> Self {
157        t
158    }
159
160    fn into_error() -> Option<fn(E) -> Self> {
161        None
162    }
163}
164
165impl<E> MaybeValidatingRow<(), E> for () {
166    fn ok(t: ()) -> Self {
167        t
168    }
169
170    fn into_error() -> Option<fn(E) -> Self> {
171        None
172    }
173}
174
175impl<E, R> MaybeValidatingRow<Vec<R>, E> for Vec<R> {
176    fn ok(t: Vec<R>) -> Self {
177        t
178    }
179
180    fn into_error() -> Option<fn(E) -> Self> {
181        None
182    }
183}
184
185impl<T, E> MaybeValidatingRow<T, E> for Result<T, E> {
186    fn ok(row: T) -> Self {
187        Ok(row)
188    }
189
190    fn into_error() -> Option<fn(E) -> Self> {
191        Some(Err)
192    }
193}
194
195/// Error logger to be used by rendering code.
196// TODO: Consider removing this struct.
197#[derive(Clone)]
198pub(super) struct ErrorLogger {
199    dataflow_name: String,
200}
201
202impl ErrorLogger {
203    pub fn new(dataflow_name: String) -> Self {
204        Self { dataflow_name }
205    }
206
207    /// Log the given error.
208    ///
209    /// The logging format is optimized for surfacing errors with Sentry:
210    ///  * `error` is logged at ERROR level and will appear as the error title in Sentry.
211    ///    We require it to be a static string, to ensure that Sentry always merges instances of
212    ///    the same error together.
213    ///  * `details` is logged at WARN level and will appear in the breadcrumbs.
214    ///    Put relevant dynamic information here.
215    ///
216    /// The message that's logged at WARN level has the format
217    ///   "[customer-data] {message} ({details})"
218    /// We include the [customer-data] tag out of the expectation that `details` will always
219    /// contain some sensitive customer data. We include the `message` to make it possible to match
220    /// the breadcrumbs to their associated error in Sentry.
221    ///
222    // TODO(database-issues#5362): Rethink or justify our error logging strategy.
223    pub fn log(&self, message: &'static str, details: &str) {
224        tracing::warn!(
225            dataflow = self.dataflow_name,
226            "[customer-data] {message} ({details})"
227        );
228        tracing::error!(message);
229    }
230
231    /// Like [`Self::log`], but panics in debug mode.
232    ///
233    /// Use this method to notify about errors that are certainly caused by bugs in Materialize.
234    pub fn soft_panic_or_log(&self, message: &'static str, details: &str) {
235        tracing::warn!(
236            dataflow = self.dataflow_name,
237            "[customer-data] {message} ({details})"
238        );
239        mz_ore::soft_panic_or_log!("{}", message);
240    }
241}
242
243#[cfg(test)]
244mod tests {
245    use super::*;
246    use mz_storage_types::errors::DataflowError;
247    use proptest::prelude::*;
248
249    #[mz_ore::test]
250    #[cfg_attr(miri, ignore)]
251    fn proptest_roundtrip_canonical() {
252        proptest!(|(err in any::<DataflowError>())| {
253            let ser = DataflowErrorSer::from(err.clone());
254
255            // Round-trip: ser -> deser -> ser must produce identical bytes.
256            let deserialized = ser.deserialize();
257            let re_serialized = DataflowErrorSer::from(deserialized);
258            prop_assert_eq!(&ser, &re_serialized,
259                "Canonicality violation: round-trip produced different bytes");
260
261            // Equality: equal errors must produce equal bytes.
262            let ser2 = DataflowErrorSer::from(err);
263            prop_assert_eq!(&ser, &ser2,
264                "Canonicality violation: same error produced different bytes");
265        });
266    }
267
268    #[mz_ore::test]
269    fn display_roundtrip() {
270        let eval_err = EvalError::DivisionByZero;
271        let dfe = DataflowError::from(eval_err.clone());
272        let ser = DataflowErrorSer::from(eval_err);
273
274        assert_eq!(dfe.to_string(), ser.to_string());
275    }
276}
mz_compute/render/errors.rs

mz_compute/render/
errors.rs
