clickhouse_dfe/
error.rs

1// Project:   clickhouse-dfe
2// File:      src/error.rs
3// Purpose:   Crate error enum, upstream mapping and retry classification
4// Language:  Rust
5//
6// License:   Apache-2.0
7// Copyright: (c) 2026 HYPERI PTY LIMITED
8
9//! Contains [`Error`] and corresponding [`Result`].
10
11use serde::{de, ser};
12use std::{error::Error as StdError, fmt, io, result, str::Utf8Error};
13
14/// A result with a specified [`Error`] type.
15pub type Result<T, E = Error> = result::Result<T, E>;
16
17type BoxedError = Box<dyn StdError + Send + Sync>;
18
19/// Represents all possible errors.
20#[derive(Debug, thiserror::Error)]
21#[non_exhaustive]
22#[allow(missing_docs)]
23pub enum Error {
24    #[error("invalid params: {0}")]
25    InvalidParams(#[source] BoxedError),
26    #[error("network error: {0}")]
27    Network(#[source] BoxedError),
28    #[error("compression error: {0}")]
29    Compression(#[source] BoxedError),
30    #[error("decompression error: {0}")]
31    Decompression(#[source] BoxedError),
32    #[error("no rows returned by a query that expected to return at least one row")]
33    RowNotFound,
34    #[error("sequences must have a known size ahead of time")]
35    SequenceMustHaveLength,
36    #[error("`deserialize_any` is not supported")]
37    DeserializeAnyNotSupported,
38    #[error("not enough data, probably a row type mismatches a database schema")]
39    NotEnoughData,
40    #[error("string is not valid utf8")]
41    InvalidUtf8Encoding(#[from] Utf8Error),
42    #[error("tag for enum is not valid")]
43    InvalidTagEncoding(usize),
44    #[error("max number of types in the Variant data type is 255, got {0}")]
45    VariantDiscriminatorIsOutOfBound(usize),
46    #[error("a custom error message from serde: {0}")]
47    Custom(String),
48    /// Background worker task exited; the inserter is no longer
49    /// accepting commands. Terminal: construct a fresh inserter,
50    /// don't retry the same handle.
51    #[error("background worker task has exited; the inserter is no longer accepting commands")]
52    WorkerExited,
53    /// Structured server-side exception parsed from the response.
54    /// Preferred over [`Error::BadResponse`] when the server returned
55    /// a recognisable `Code: NNN. DB::Exception: ...` body. Callers
56    /// can match on `code` for typed handling and use
57    /// [`Error::is_retriable`] for conservative retry classification.
58    #[error("server error code {code}: {message}")]
59    ServerException {
60        /// `X-ClickHouse-Exception-Code` from the response header.
61        /// Signed `int` on the wire (cpp-client + server source);
62        /// codes are small positives in practice.
63        code: i32,
64        /// Exception name extracted from `(UPPERCASE_NAME)` near
65        /// the end of the message. `None` when the parser couldn't
66        /// recognise it (older CH versions, custom error paths).
67        name: Option<String>,
68        /// Cleaned message body, stripped of the `Code: N.`,
69        /// `DB::Exception:` prefix, exception-name parens, and
70        /// `(version ...)` suffix.
71        message: String,
72        /// Server-side stack trace if `Stack trace:` was present in
73        /// the body.
74        stack_trace: Option<String>,
75    },
76    #[error("bad response: {0}")]
77    BadResponse(String),
78    /// A transport-level failure that a fresh attempt may not hit --
79    /// a pool-acquire timeout, a closed pool, a hook failure. Typed so
80    /// retry classification does not have to match on message text.
81    #[error("transient transport failure: {0}")]
82    Transient(String),
83    /// A connection could not be opened: an unresolvable host, a host
84    /// that resolved to no addresses, an empty endpoint list. Typed for
85    /// the same reason as [`Error::Transient`].
86    #[error("connect failed: {0}")]
87    Connect(String),
88    #[error("timeout expired")]
89    TimedOut,
90    #[error("error while parsing columns header from the response: {0}")]
91    InvalidColumnsHeader(#[source] BoxedError),
92    #[error("schema mismatch: {0}")]
93    SchemaMismatch(String),
94    #[error("unsupported: {0}")]
95    Unsupported(String),
96    /// A dynamic-insert failure, kept typed so a caller can tell a stale
97    /// schema (re-fetch and retry) from a permanent rejection.
98    #[cfg(feature = "dynamic")]
99    #[cfg_attr(docsrs, doc(cfg(feature = "dynamic")))]
100    #[error(transparent)]
101    Dynamic(#[from] crate::dynamic::DynamicError),
102    #[error("{0}")]
103    Other(BoxedError),
104}
105
106// Variant-by-variant where the name matches, `Other` otherwise.
107// Upstream's enum is `#[non_exhaustive]`, so a variant added there later
108// lands in `Other` until this mapping is extended.
109impl From<clickhouse::error::Error> for Error {
110    fn from(error: clickhouse::error::Error) -> Self {
111        use clickhouse::error::Error as Up;
112        match error {
113            Up::InvalidParams(e) => Self::InvalidParams(e),
114            Up::Network(e) => Self::Network(e),
115            Up::Compression(e) => Self::Compression(e),
116            Up::Decompression(e) => Self::Decompression(e),
117            Up::RowNotFound => Self::RowNotFound,
118            Up::SequenceMustHaveLength => Self::SequenceMustHaveLength,
119            Up::DeserializeAnyNotSupported => Self::DeserializeAnyNotSupported,
120            Up::NotEnoughData => Self::NotEnoughData,
121            Up::InvalidUtf8Encoding(e) => Self::InvalidUtf8Encoding(e),
122            Up::InvalidTagEncoding(n) => Self::InvalidTagEncoding(n),
123            Up::VariantDiscriminatorIsOutOfBound(n) => Self::VariantDiscriminatorIsOutOfBound(n),
124            Up::Custom(s) => Self::Custom(s),
125            Up::BadResponse(s) => Self::BadResponse(s),
126            Up::TimedOut => Self::TimedOut,
127            Up::InvalidColumnsHeader(e) => Self::InvalidColumnsHeader(e),
128            Up::SchemaMismatch(s) => Self::SchemaMismatch(s),
129            Up::Unsupported(s) => Self::Unsupported(s),
130            other => Self::Other(Box::new(other)),
131        }
132    }
133}
134
135impl ser::Error for Error {
136    fn custom<T: fmt::Display>(msg: T) -> Self {
137        Self::Custom(msg.to_string())
138    }
139}
140
141impl de::Error for Error {
142    fn custom<T: fmt::Display>(msg: T) -> Self {
143        Self::Custom(msg.to_string())
144    }
145}
146
147impl From<Error> for io::Error {
148    fn from(error: Error) -> Self {
149        io::Error::other(error)
150    }
151}
152
153impl From<io::Error> for Error {
154    fn from(error: io::Error) -> Self {
155        // Unwraps an `Error` this crate boxed on the way out, so a round trip
156        // through `io::Error` keeps the variant.
157        match error.downcast::<Self>() {
158            Ok(inner) => inner,
159            Err(other) => Self::Other(other.into()),
160        }
161    }
162}
163
164impl Error {
165    /// Conservative retriability classification.
166    ///
167    /// Returns `true` for errors that *might* succeed on retry --
168    /// transport-level (`Network`, `TimedOut`, `Transient`, `Connect`),
169    /// a known set of transient server-side codes, and the dynamic-insert
170    /// failures that clear once the schema is re-read.
171    ///
172    /// Returns `false` for everything else, including unknown server codes:
173    /// a caller wanting a more aggressive retry should match on the variant.
174    /// The code table is [`is_retriable_code`][Self::is_retriable_code].
175    #[must_use]
176    pub fn is_retriable(&self) -> bool {
177        match self {
178            // `Transient` and `Connect` are only ever built for a
179            // condition a fresh attempt may not hit, so they classify
180            // here rather than on their message text.
181            Self::Network(_) | Self::TimedOut | Self::Transient(_) | Self::Connect(_) => true,
182            Self::ServerException { code, .. } => Self::is_retriable_code(*code),
183            // A stale, missing or unreadable schema is fixed by re-reading it;
184            // an unsupported type or a bad value is not.
185            #[cfg(feature = "dynamic")]
186            Self::Dynamic(e) => matches!(
187                e,
188                crate::dynamic::DynamicError::SchemaFetch { .. }
189                    | crate::dynamic::DynamicError::SchemaMismatch { .. }
190                    | crate::dynamic::DynamicError::EmptySchema { .. }
191            ),
192            _ => false,
193        }
194    }
195
196    /// ClickHouse error codes considered transient for retry. See
197    /// `src/Common/ErrorCodes.cpp` upstream for the canonical list.
198    /// Conservative -- false negatives are expected, false positives
199    /// should be rare. Each entry names its upstream macro.
200    #[must_use]
201    pub fn is_retriable_code(code: i32) -> bool {
202        matches!(
203            code,
204            // TIMEOUT_EXCEEDED
205            159 |
206            // TOO_SLOW (merge queue lag; usually transient)
207            160 |
208            // TOO_MANY_SIMULTANEOUS_QUERIES
209            202 |
210            // SOCKET_TIMEOUT
211            209 |
212            // NETWORK_ERROR
213            210 |
214            // ABORTED (operator KILL or shutdown -- replayable on
215            // another node)
216            236 |
217            // MEMORY_LIMIT_EXCEEDED (often transient under brief
218            // memory pressure)
219            241 |
220            // TOO_MANY_PARTS (transient under heavy ingest)
221            252 |
222            // ALL_CONNECTION_TRIES_FAILED
223            279 |
224            // LIMIT_EXCEEDED (transient quota)
225            290 |
226            // UNKNOWN_STATUS_OF_INSERT (the "did my INSERT land?"
227            // ambiguity code -- documented mitigation is retry with
228            // an insert_deduplication_token, which the
229            // batch_isolation token variant ships)
230            319 |
231            // RECEIVED_ERROR_TOO_MANY_REQUESTS (server proxied a 429
232            // from a downstream resource -- S3, Kafka, HTTP storage)
233            364 |
234            // PART_IS_TEMPORARILY_LOCKED (mutate/merge contention)
235            384 |
236            // CANNOT_SCHEDULE_TASK (task scheduler full)
237            439 |
238            // DEADLOCK_AVOIDED (server-side avoidance; succeeds on retry)
239            473 |
240            // UNKNOWN_STATUS_OF_TRANSACTION (transaction analogue of
241            // UNKNOWN_STATUS_OF_INSERT -- same "did it land?" ambiguity)
242            659 |
243            // TOO_MANY_UNAVAILABLE_SHARDS (transient cluster condition)
244            904 |
245            // KEEPER_EXCEPTION (replicated-metadata hiccup)
246            999
247        )
248    }
249}
250
251#[cfg(test)]
252mod tests {
253    use crate::error::Error;
254    use std::io;
255
256    #[test]
257    fn roundtrip_io_error() {
258        let orig = Error::NotEnoughData;
259
260        // Error -> io::Error
261        let orig_str = orig.to_string();
262        let io = io::Error::from(orig);
263        assert_eq!(io.kind(), io::ErrorKind::Other);
264        assert_eq!(io.to_string(), orig_str);
265
266        // io::Error -> Error
267        let orig = Error::from(io);
268        assert!(matches!(orig, Error::NotEnoughData));
269    }
270
271    #[test]
272    fn error_traits() {
273        fn assert_traits<T: std::error::Error + Send + Sync>() {}
274
275        assert_traits::<Error>();
276    }
277
278    #[test]
279    fn is_retriable_classifies_known_transient_codes() {
280        let make = |code: i32| Error::ServerException {
281            code,
282            name: None,
283            message: String::new(),
284            stack_trace: None,
285        };
286
287        // Sampled from the retriable table -- known transient codes.
288        for &code in &[
289            159i32, 160, 202, 209, 210, 236, 241, 252, 279, 290, 319, 364, 384, 439, 473, 659, 904,
290            999,
291        ] {
292            assert!(make(code).is_retriable(), "code {code} should be retriable");
293        }
294
295        // Sampled known-not-retriable codes (auth, schema, parse).
296        for &code in &[36i32, 60, 117, 192, 195, 469] {
297            assert!(
298                !make(code).is_retriable(),
299                "code {code} should NOT be retriable"
300            );
301        }
302    }
303
304    #[test]
305    fn is_retriable_for_transport_errors() {
306        assert!(Error::TimedOut.is_retriable());
307        assert!(Error::Transient("pool acquire timed out".into()).is_retriable());
308        assert!(Error::Connect("no address for host".into()).is_retriable());
309        // A foreign `io::Error` lands in `Other`, which stays terminal.
310        let foreign: Error = io::Error::new(io::ErrorKind::ConnectionReset, "reset").into();
311        assert!(!foreign.is_retriable());
312    }
313
314    #[test]
315    fn is_retriable_code_defaults_unknown_to_not_retriable() {
316        // A ClickHouse release can ship codes this table has never seen; until
317        // it is extended, a caller sees them as terminal rather than looping.
318        for &code in &[0i32, 1, 12345, 99999, i32::MAX] {
319            assert!(
320                !Error::is_retriable_code(code),
321                "unknown code {code} should default to NOT retriable"
322            );
323        }
324    }
325
326    /// The consumer branches on this to decide between re-reading the schema
327    /// and dead-lettering the batch, so the split has to hold.
328    #[cfg(feature = "dynamic")]
329    #[test]
330    fn dynamic_schema_failures_are_retriable_but_bad_data_is_not() {
331        use crate::dynamic::DynamicError;
332
333        let retriable = [
334            DynamicError::EmptySchema {
335                table: "db.t".into(),
336            },
337            DynamicError::SchemaMismatch {
338                table: "db.t".into(),
339                message: "drift".into(),
340            },
341            DynamicError::SchemaFetch {
342                table: "db.t".into(),
343                source: clickhouse::error::Error::TimedOut,
344            },
345        ];
346        for e in retriable {
347            assert!(Error::from(e).is_retriable());
348        }
349
350        let terminal = [
351            DynamicError::UnsupportedType {
352                column: "c".into(),
353                type_str: "Point".into(),
354            },
355            DynamicError::EncodingError {
356                column: "c".into(),
357                message: "UInt8 value out of range".into(),
358            },
359        ];
360        for e in terminal {
361            assert!(!Error::from(e).is_retriable());
362        }
363    }
364
365    /// `#[error(transparent)]`, so the wrapper adds no prefix and `source()`
366    /// reaches the original query failure.
367    #[cfg(feature = "dynamic")]
368    #[test]
369    fn dynamic_errors_pass_through_display_and_source() {
370        use std::error::Error as _;
371
372        let err = Error::from(crate::dynamic::DynamicError::SchemaFetch {
373            table: "db.t".into(),
374            source: clickhouse::error::Error::TimedOut,
375        });
376        assert_eq!(
377            err.to_string(),
378            "failed to fetch schema for 'db.t': timeout expired"
379        );
380        // `transparent` forwards `source()`, so the query failure is reachable.
381        let source = err.source().expect("the query failure is the source");
382        assert_eq!(source.to_string(), "timeout expired");
383    }
384}
clickhouse_dfe/error.rs

clickhouse_dfe/
error.rs
