unitycatalog_client/
error.rs

1use unitycatalog_delta_api::models::DeltaErrorModel;
2
3pub type Result<T, E = Error> = std::result::Result<T, E>;
4
5#[derive(Debug, thiserror::Error)]
6pub enum Error {
7    #[error("Common Error: {source}")]
8    Common {
9        #[from]
10        source: unitycatalog_common::Error,
11    },
12
13    #[error("Delta API error {}: [{:?}] {}", .0.code, .0.error_type, .0.message)]
14    Delta(DeltaErrorModel),
15
16    #[error("Client Error: {source}")]
17    ClientError {
18        #[from]
19        source: olai_http::Error,
20    },
21
22    #[error("Malformed response: {source}")]
23    MalformedResponse {
24        #[from]
25        source: serde_json::Error,
26    },
27
28    #[error("Malformed url: {source}")]
29    MalformedUrl {
30        #[from]
31        source: url::ParseError,
32    },
33
34    #[error("Reqwuest error: {0}")]
35    RequestError(#[from] reqwest::Error),
36
37    #[error("API error: {0}")]
38    Api(#[from] UcApiError),
39
40    #[error("Generic error: {0}")]
41    Generic(String),
42}
43
44impl Error {
45    pub fn generic(message: impl ToString) -> Self {
46        Error::Generic(message.to_string())
47    }
48
49    /// Map a failed `/delta/v1` request into a typed [`Error`].
50    ///
51    /// A request sent via [`olai_http::CloudRequestBuilder::send_raw`] fails with
52    /// a [`SendRawError`]: a signing failure carries no HTTP response and maps to
53    /// [`Error::ClientError`]; a request failure carrying an HTTP status + body
54    /// (a [`olai_http::RetryError`]) is parsed as a Delta error envelope via
55    /// [`parse_delta_error`]. A transport failure with no status also maps to
56    /// [`Error::ClientError`].
57    pub(crate) fn from_delta_send(err: olai_http::SendRawError) -> Self {
58        match err {
59            olai_http::SendRawError::Sign(e) => Error::ClientError { source: e },
60            olai_http::SendRawError::Retry(e) => match e.status() {
61                Some(status) => parse_delta_error(status.as_u16(), e.body().map(str::as_bytes)),
62                // No status: a transport/timeout failure, not an HTTP error body.
63                None => Error::ClientError { source: e.error() },
64            },
65        }
66    }
67
68    /// Map a failed UC API request into a typed [`Error`], mirroring
69    /// [`from_delta_send`](Self::from_delta_send) but parsing the UC API error
70    /// body (`{ error_code, message }`) via [`parse_error`].
71    pub(crate) fn from_api_send(err: olai_http::SendRawError) -> Self {
72        match err {
73            olai_http::SendRawError::Sign(e) => Error::ClientError { source: e },
74            olai_http::SendRawError::Retry(e) => match e.status() {
75                Some(status) => parse_error(status.as_u16(), e.body().map(str::as_bytes)),
76                None => Error::ClientError { source: e.error() },
77            },
78        }
79    }
80
81    pub fn is_not_found(&self) -> bool {
82        match self {
83            Error::Api(UcApiError::NotFound { .. }) => true,
84            Error::Delta(model) => model.error_type.is_not_found(),
85            _ => false,
86        }
87    }
88
89    pub fn is_already_exists(&self) -> bool {
90        match self {
91            Error::Api(UcApiError::AlreadyExists { .. }) => true,
92            Error::Delta(model) => model.error_type.is_already_exists(),
93            _ => false,
94        }
95    }
96
97    pub fn is_permission_denied(&self) -> bool {
98        match self {
99            Error::Api(UcApiError::PermissionDenied { .. }) => true,
100            Error::Delta(model) => {
101                matches!(
102                    model.error_type,
103                    unitycatalog_delta_api::models::DeltaErrorType::PermissionDeniedException
104                )
105            }
106            _ => false,
107        }
108    }
109
110    pub fn is_unauthenticated(&self) -> bool {
111        match self {
112            Error::Api(UcApiError::Unauthenticated { .. }) => true,
113            Error::Delta(model) => {
114                matches!(
115                    model.error_type,
116                    unitycatalog_delta_api::models::DeltaErrorType::NotAuthorizedException
117                )
118            }
119            _ => false,
120        }
121    }
122
123    /// Whether this is a Delta `CommitVersionConflictException` (409): a
124    /// concurrent commit ratified the proposed version first. The caller should
125    /// rebuild its snapshot and retry the commit.
126    pub fn is_commit_conflict(&self) -> bool {
127        matches!(self, Error::Delta(model) if model.error_type.is_commit_conflict())
128    }
129
130    /// Whether this is a Delta `UpdateRequirementConflictException` (409): an
131    /// `assert-etag`/`assert-table-uuid` requirement was not met. The caller
132    /// should reload the table and retry.
133    pub fn is_update_requirement_conflict(&self) -> bool {
134        matches!(self, Error::Delta(model) if model.error_type.is_update_requirement_conflict())
135    }
136
137    /// Whether this is a Delta `ResourceExhaustedException` / `TooManyRequestsException`
138    /// (429): the request was throttled or hit the unbackfilled-commit limit. The
139    /// caller should back off (and backfill pending commits) before retrying.
140    pub fn is_resource_exhausted(&self) -> bool {
141        matches!(self, Error::Delta(model) if model.error_type.is_resource_exhausted())
142    }
143
144    /// Whether this is a Delta `CommitStateUnknownException` (500): the commit
145    /// outcome is unknown. The caller must check table state before retrying to
146    /// avoid duplicate commits.
147    pub fn is_commit_state_unknown(&self) -> bool {
148        matches!(self, Error::Delta(model) if model.error_type.is_commit_state_unknown())
149    }
150
151    /// Whether this is a Delta `UnsupportedTableFormatException` (400): the table
152    /// is not Delta, or is a Delta table this `/delta/v1` endpoint does not
153    /// support. The caller should fall back to the legacy UC table API.
154    pub fn is_unsupported_table_format(&self) -> bool {
155        matches!(self, Error::Delta(model) if model.error_type.is_unsupported_table_format())
156    }
157
158    /// Whether this is a Delta `NotImplementedException` (501): the server does
159    /// not implement this `/delta/v1` functionality. The caller should fall back
160    /// to the legacy UC table API.
161    pub fn is_not_implemented(&self) -> bool {
162        matches!(self, Error::Delta(model) if model.error_type.is_not_implemented())
163    }
164
165    /// Whether this is a `404` that is **not** a recognizable Delta error envelope
166    /// — i.e. the `/delta/v1` route itself is absent (no `UnsupportedTableFormat`
167    /// / `NoSuchTable` envelope was returned), as on a UC deployment that does not
168    /// serve `/delta/v1` at all.
169    ///
170    /// This is deliberately distinct from [`Error::is_not_found`]: an *enveloped*
171    /// `NoSuchTableException` (a genuinely missing table) is an [`Error::Delta`]
172    /// and returns `false` here, so callers can fall back on a missing route
173    /// without masking a missing table.
174    pub fn is_route_missing(&self) -> bool {
175        matches!(self, Error::Api(UcApiError::Other { status: 404, .. }))
176    }
177
178    /// Whether the caller should react to this `/delta/v1` loadTable error by
179    /// falling back to the legacy UC table API (filesystem snapshot): an
180    /// unsupported table format, an unimplemented endpoint, or an entirely missing
181    /// route. A genuine `NoSuchTable`/auth/other error returns `false` and must be
182    /// propagated.
183    pub fn should_fall_back_to_legacy(&self) -> bool {
184        self.is_unsupported_table_format() || self.is_not_implemented() || self.is_route_missing()
185    }
186}
187
188/// Typed error variants mapped to the Databricks Unity Catalog API error code spec.
189#[derive(Debug, thiserror::Error, PartialEq, Eq, Clone)]
190#[non_exhaustive]
191pub enum UcApiError {
192    #[error("Invalid parameter: {message}")]
193    InvalidParameter { message: String },
194
195    #[error("Unauthenticated: {message}")]
196    Unauthenticated { message: String },
197
198    #[error("Permission denied: {message}")]
199    PermissionDenied { message: String },
200
201    #[error("Resource not found: {message}")]
202    NotFound { message: String },
203
204    #[error("Resource already exists: {message}")]
205    AlreadyExists { message: String },
206
207    #[error("Request limit exceeded: {message}")]
208    RequestLimitExceeded { message: String },
209
210    #[error("Internal server error: {message}")]
211    InternalError { message: String },
212
213    #[error("Temporarily unavailable: {message}")]
214    TemporarilyUnavailable { message: String },
215
216    #[error("API error {status}: [{error_code}] {message}")]
217    Other {
218        status: u16,
219        error_code: String,
220        message: String,
221    },
222}
223
224impl UcApiError {
225    /// Returns the UC API error code string.
226    pub fn error_code(&self) -> &str {
227        match self {
228            UcApiError::InvalidParameter { .. } => "INVALID_PARAMETER_VALUE",
229            UcApiError::Unauthenticated { .. } => "UNAUTHENTICATED",
230            UcApiError::PermissionDenied { .. } => "PERMISSION_DENIED",
231            UcApiError::NotFound { .. } => "RESOURCE_NOT_FOUND",
232            UcApiError::AlreadyExists { .. } => "RESOURCE_ALREADY_EXISTS",
233            UcApiError::RequestLimitExceeded { .. } => "REQUEST_LIMIT_EXCEEDED",
234            UcApiError::InternalError { .. } => "INTERNAL_ERROR",
235            UcApiError::TemporarilyUnavailable { .. } => "TEMPORARILY_UNAVAILABLE",
236            UcApiError::Other { error_code, .. } => error_code,
237        }
238    }
239
240    /// Returns the HTTP status code associated with this error.
241    pub fn http_status(&self) -> u16 {
242        match self {
243            UcApiError::InvalidParameter { .. } => 400,
244            UcApiError::Unauthenticated { .. } => 401,
245            UcApiError::PermissionDenied { .. } => 403,
246            UcApiError::NotFound { .. } => 404,
247            UcApiError::AlreadyExists { .. } => 409,
248            UcApiError::RequestLimitExceeded { .. } => 429,
249            UcApiError::InternalError { .. } => 500,
250            UcApiError::TemporarilyUnavailable { .. } => 503,
251            UcApiError::Other { status, .. } => *status,
252        }
253    }
254
255    /// Construct from an API response with status code, error code string, and message.
256    pub fn from_api_response(status: u16, error_code: &str, message: String) -> Self {
257        match error_code {
258            "INVALID_PARAMETER_VALUE" => UcApiError::InvalidParameter { message },
259            "UNAUTHENTICATED" => UcApiError::Unauthenticated { message },
260            "PERMISSION_DENIED" => UcApiError::PermissionDenied { message },
261            "RESOURCE_NOT_FOUND" => UcApiError::NotFound { message },
262            "RESOURCE_ALREADY_EXISTS" => UcApiError::AlreadyExists { message },
263            "REQUEST_LIMIT_EXCEEDED" => UcApiError::RequestLimitExceeded { message },
264            "INTERNAL_ERROR" => UcApiError::InternalError { message },
265            "TEMPORARILY_UNAVAILABLE" => UcApiError::TemporarilyUnavailable { message },
266            other => UcApiError::Other {
267                status,
268                error_code: other.to_string(),
269                message,
270            },
271        }
272    }
273}
274
275/// Serde helper for parsing the UC API error body.
276#[derive(serde::Deserialize)]
277struct ApiErrorBody {
278    #[serde(alias = "errorCode")]
279    error_code: String,
280    message: String,
281}
282
283/// Parse a UC API error from its HTTP `status` and response `body`, returning a
284/// typed [`Error`]. Falls back to [`UcApiError::Other`] with the raw bytes when
285/// the body is not the expected `{ error_code, message }` JSON.
286pub(crate) fn parse_error(status: u16, body: Option<&[u8]>) -> Error {
287    let body = body.unwrap_or_default();
288    match serde_json::from_slice::<ApiErrorBody>(body) {
289        Ok(api_err) => {
290            UcApiError::from_api_response(status, &api_err.error_code, api_err.message).into()
291        }
292        Err(_) => UcApiError::Other {
293            status,
294            error_code: String::new(),
295            message: String::from_utf8_lossy(body).into_owned(),
296        }
297        .into(),
298    }
299}
300
301/// Read an error HTTP response, parse the UC API JSON body, and return a typed
302/// [`Error`]. Thin wrapper over [`parse_error`] retained for the generated
303/// resource clients (`codegen/**`), which hold a [`reqwest::Response`] because
304/// they still call `.send()` (a 2xx/non-2xx split, not `send_raw`).
305pub(crate) async fn parse_error_response(response: reqwest::Response) -> Error {
306    let status = response.status().as_u16();
307    match response.bytes().await {
308        Ok(body) => parse_error(status, Some(&body)),
309        Err(e) => Error::RequestError(e),
310    }
311}
312
313/// Parse a `/delta/v1` error from its HTTP `status` and response `body`, matching
314/// the Delta API error envelope (`{ "error": { message, type, code, stack? } }`)
315/// and returning a typed [`Error::Delta`].
316///
317/// Falls back to [`UcApiError::Other`] when the body is not a recognizable Delta
318/// error envelope (e.g. a bare proxy 502 or a truncated body), preserving the raw
319/// bytes for diagnostics.
320///
321/// The caller passes `status` and `body` directly (rather than a
322/// [`reqwest::Response`]) because a failed `/delta/v1` request surfaces through
323/// [`olai_http::CloudRequestBuilder::send_raw`] as an
324/// [`olai_http::RetryError`], which has already consumed the response body and
325/// exposes it via [`olai_http::RetryError::status`]/[`body`](olai_http::RetryError::body).
326pub(crate) fn parse_delta_error(status: u16, body: Option<&[u8]>) -> Error {
327    use unitycatalog_delta_api::models::DeltaErrorResponse;
328
329    let body = body.unwrap_or_default();
330    match serde_json::from_slice::<DeltaErrorResponse>(body) {
331        Ok(envelope) => Error::Delta(envelope.error),
332        Err(_) => UcApiError::Other {
333            status,
334            error_code: String::new(),
335            message: String::from_utf8_lossy(body).into_owned(),
336        }
337        .into(),
338    }
339}
340
341#[cfg(test)]
342mod tests {
343    use super::*;
344
345    fn make_response(status: u16, body: &'static str) -> reqwest::Response {
346        http::Response::builder()
347            .status(status)
348            .header("content-type", "application/json")
349            .body(bytes::Bytes::from_static(body.as_bytes()))
350            .map(reqwest::Response::from)
351            .unwrap()
352    }
353
354    #[tokio::test]
355    async fn test_parse_error_resource_not_found() {
356        let resp = make_response(
357            404,
358            r#"{"error_code":"RESOURCE_NOT_FOUND","message":"catalog 'foo' not found"}"#,
359        );
360        let err = parse_error_response(resp).await;
361        assert!(err.is_not_found());
362        assert!(matches!(
363            err,
364            Error::Api(UcApiError::NotFound { ref message }) if message == "catalog 'foo' not found"
365        ));
366    }
367
368    #[tokio::test]
369    async fn test_parse_error_camel_case_alias() {
370        let resp = make_response(
371            404,
372            r#"{"errorCode":"RESOURCE_NOT_FOUND","message":"not found"}"#,
373        );
374        let err = parse_error_response(resp).await;
375        assert!(err.is_not_found());
376    }
377
378    #[tokio::test]
379    async fn test_parse_error_non_json_body() {
380        let resp = make_response(500, "Internal Server Error");
381        let err = parse_error_response(resp).await;
382        assert!(matches!(
383            err,
384            Error::Api(UcApiError::Other {
385                status: 500,
386                ref message,
387                ..
388            }) if message == "Internal Server Error"
389        ));
390    }
391
392    #[tokio::test]
393    async fn test_parse_error_already_exists() {
394        let resp = make_response(
395            409,
396            r#"{"error_code":"RESOURCE_ALREADY_EXISTS","message":"already exists"}"#,
397        );
398        let err = parse_error_response(resp).await;
399        assert!(err.is_already_exists());
400    }
401
402    /// Parse a Delta error envelope directly from `(status, body)`, matching the
403    /// shape [`Error::from_delta_send`] feeds in after `send_raw` classifies a
404    /// non-2xx response.
405    fn delta_err(status: u16, body: &str) -> Error {
406        parse_delta_error(status, Some(body.as_bytes()))
407    }
408
409    #[test]
410    fn test_parse_delta_error_not_found() {
411        let err = delta_err(
412            404,
413            r#"{"error":{"message":"table 'x' not found","type":"NoSuchTableException","code":404}}"#,
414        );
415        assert!(err.is_not_found());
416        assert!(!err.is_already_exists());
417        assert!(matches!(
418            err,
419            Error::Delta(ref m) if m.message == "table 'x' not found" && m.code == 404
420        ));
421    }
422
423    #[test]
424    fn test_parse_delta_error_already_exists() {
425        let err = delta_err(
426            409,
427            r#"{"error":{"message":"exists","type":"AlreadyExistsException","code":409}}"#,
428        );
429        assert!(err.is_already_exists());
430        assert!(!err.is_not_found());
431    }
432
433    #[test]
434    fn test_parse_delta_error_commit_conflict() {
435        let err = delta_err(
436            409,
437            r#"{"error":{"message":"conflict","type":"CommitVersionConflictException","code":409}}"#,
438        );
439        assert!(err.is_commit_conflict());
440        assert!(!err.is_update_requirement_conflict());
441        assert!(!err.is_already_exists());
442    }
443
444    #[test]
445    fn test_parse_delta_error_update_requirement_conflict() {
446        let err = delta_err(
447            409,
448            r#"{"error":{"message":"etag mismatch","type":"UpdateRequirementConflictException","code":409}}"#,
449        );
450        assert!(err.is_update_requirement_conflict());
451        assert!(!err.is_commit_conflict());
452    }
453
454    #[test]
455    fn test_parse_delta_error_resource_exhausted() {
456        for ty in ["ResourceExhaustedException", "TooManyRequestsException"] {
457            let body = format!(r#"{{"error":{{"message":"slow down","type":"{ty}","code":429}}}}"#);
458            let err = delta_err(429, &body);
459            assert!(err.is_resource_exhausted(), "type {ty} should be exhausted");
460        }
461    }
462
463    #[test]
464    fn test_parse_delta_error_commit_state_unknown() {
465        let err = delta_err(
466            500,
467            r#"{"error":{"message":"unknown","type":"CommitStateUnknownException","code":500}}"#,
468        );
469        assert!(err.is_commit_state_unknown());
470    }
471
472    #[test]
473    fn test_parse_delta_error_with_stack() {
474        let err = delta_err(
475            500,
476            r#"{"error":{"message":"boom","type":"InternalServerErrorException","code":500,"stack":["a","b"]}}"#,
477        );
478        assert!(matches!(
479            err,
480            Error::Delta(ref m) if m.stack.as_deref() == Some(&["a".to_string(), "b".to_string()][..])
481        ));
482    }
483
484    #[test]
485    fn test_parse_delta_error_non_envelope_body() {
486        let err = delta_err(502, "Bad Gateway");
487        assert!(matches!(
488            err,
489            Error::Api(UcApiError::Other { status: 502, ref message, .. }) if message == "Bad Gateway"
490        ));
491    }
492
493    #[test]
494    fn test_parse_delta_error_unsupported_table_format() {
495        // A 400 with the typed envelope: the table is not Delta / not supported by
496        // /delta/v1. Should trigger the legacy-API fallback but is not a not-found.
497        let err = delta_err(
498            400,
499            r#"{"error":{"message":"not a delta table","type":"UnsupportedTableFormatException","code":400}}"#,
500        );
501        assert!(err.is_unsupported_table_format());
502        assert!(err.should_fall_back_to_legacy());
503        assert!(!err.is_not_found());
504        assert!(!err.is_route_missing());
505    }
506
507    #[test]
508    fn test_parse_delta_error_not_implemented() {
509        let err = delta_err(
510            501,
511            r#"{"error":{"message":"nope","type":"NotImplementedException","code":501}}"#,
512        );
513        assert!(err.is_not_implemented());
514        assert!(err.should_fall_back_to_legacy());
515    }
516
517    #[test]
518    fn test_route_missing_is_non_envelope_404() {
519        // A 404 with no Delta error envelope = the /delta/v1 route is absent. This
520        // must fall back to the legacy API, distinct from an enveloped
521        // NoSuchTableException (a genuinely missing table), which must propagate.
522        let err = delta_err(404, "Not Found");
523        assert!(err.is_route_missing());
524        assert!(err.should_fall_back_to_legacy());
525        // is_not_found also reports true for the missing route via the Api arm, but
526        // the enveloped not-found below must NOT be treated as a missing route.
527        let enveloped = delta_err(
528            404,
529            r#"{"error":{"message":"no table","type":"NoSuchTableException","code":404}}"#,
530        );
531        assert!(enveloped.is_not_found());
532        assert!(!enveloped.is_route_missing());
533        assert!(!enveloped.should_fall_back_to_legacy());
534    }
535}
unitycatalog_client/error.rs

unitycatalog_client/
error.rs
