unitycatalog_client/
error.rs

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

unitycatalog_client/
error.rs
