unitycatalog_client/
delta_v1.rs

1//! Hand-written client for the UC Delta REST API (`/delta/v1/...`).
2//!
3//! The Delta API is a standalone REST protocol (not a generated resource API),
4//! so — like [`crate::temporary_credentials`] — it is hand-maintained. The wire
5//! DTOs are shared with the server via
6//! [`unitycatalog_delta_api::models`]. This client covers the full
7//! `delta.yaml` surface: catalog config negotiation, the catalog-managed table
8//! lifecycle (`createStagingTable`, `createTable`, `loadTable`, `updateTable`,
9//! `deleteTable`, `tableExists`, `renameTable`), credential vending
10//! (`getTableCredentials`, `getStagingTableCredentials`,
11//! `getTemporaryPathCredentials`), and commit-metrics reporting (`reportMetrics`).
12
13use percent_encoding::{NON_ALPHANUMERIC, utf8_percent_encode};
14use unitycatalog_delta_api::models::{
15    DeltaCatalogConfig, DeltaCreateStagingTableRequest, DeltaCreateTableRequest,
16    DeltaCredentialOperation, DeltaCredentialsResponse, DeltaLoadTableResponse,
17    DeltaRenameTableRequest, DeltaReportMetricsRequest, DeltaStagingTableResponse,
18    DeltaUpdateTableRequest,
19};
20use url::Url;
21
22use crate::{Result, Transport};
23
24/// Percent-encode a single path segment so names containing `/ ? # space ..`
25/// route to the intended resource instead of being interpreted as path
26/// structure. We encode every non-alphanumeric byte (a superset of the RFC 3986
27/// path-segment reserved set), which is safe because each call encodes exactly
28/// one segment that is then joined with literal `/` separators.
29fn encode_segment(segment: &str) -> String {
30    utf8_percent_encode(segment, NON_ALPHANUMERIC).to_string()
31}
32
33/// The wire string for a credential operation as used in `operation` query params.
34fn operation_param(operation: DeltaCredentialOperation) -> &'static str {
35    match operation {
36        DeltaCredentialOperation::Read => "READ",
37        DeltaCredentialOperation::ReadWrite => "READ_WRITE",
38    }
39}
40
41/// Client for the `/delta/v1/` Delta REST API.
42///
43/// Constructed from the per-target [`Transport`] (carrying auth) and the Unity
44/// Catalog base URL. Prefer
45/// [`UnityCatalogClient::delta_v1`](crate::UnityCatalogClient::delta_v1).
46#[derive(Clone)]
47pub struct DeltaV1Client {
48    client: Transport,
49    base_url: Url,
50}
51
52impl DeltaV1Client {
53    /// Create a new Delta v1 client. `base_url` is normalized to end in `/` so it
54    /// joins cleanly with the relative endpoint paths.
55    pub fn new(client: Transport, mut base_url: Url) -> Self {
56        if !base_url.path().ends_with('/') {
57            base_url.set_path(&format!("{}/", base_url.path()));
58        }
59        Self { client, base_url }
60    }
61
62    /// Build a `/delta/v1/...` URL from already-encoded path components.
63    fn url(&self, rest: &str) -> Result<Url> {
64        Ok(self.base_url.join(&format!("delta/v1/{rest}"))?)
65    }
66
67    /// Get catalog configuration and supported endpoints —
68    /// `GET /delta/v1/config?catalog={catalog}&protocol-versions={versions}`.
69    ///
70    /// This is the protocol-negotiation entry point: the client advertises the
71    /// highest protocol versions it supports per major version (`protocol_versions`,
72    /// e.g. `"1.1,2.3"`) and the server returns the endpoints it supports for the
73    /// newest mutually supported version. Used to discover whether the
74    /// `/delta/v1` surface is available before falling back to the legacy API.
75    pub async fn get_config(
76        &self,
77        catalog: &str,
78        protocol_versions: &str,
79    ) -> Result<DeltaCatalogConfig> {
80        let url = self.url("config")?;
81        let response = self
82            .client
83            .get(url)
84            .query(&[
85                ("catalog", catalog),
86                ("protocol-versions", protocol_versions),
87            ])
88            .send_raw()
89            .await
90            .map_err(crate::Error::from_delta_send)?;
91        let response = crate::error::check_delta_response(response).await?;
92        let result = response.bytes().await?;
93        Ok(serde_json::from_slice(&result)?)
94    }
95
96    /// Create a staging table — `POST /delta/v1/catalogs/{catalog}/schemas/{schema}/staging-tables`.
97    ///
98    /// Allocates an immutable table id and a managed storage `location`, and
99    /// advertises the catalog-managed contract (required/suggested protocol +
100    /// properties) plus `READ_WRITE` credentials for writing the initial
101    /// `_delta_log/0.json`. The catalog/schema are taken from the path; the
102    /// request body carries only the table name.
103    pub async fn create_staging_table(
104        &self,
105        catalog: &str,
106        schema: &str,
107        request: &DeltaCreateStagingTableRequest,
108    ) -> Result<DeltaStagingTableResponse> {
109        let url = self.url(&format!(
110            "catalogs/{}/schemas/{}/staging-tables",
111            encode_segment(catalog),
112            encode_segment(schema),
113        ))?;
114        let response = self
115            .client
116            .post(url)
117            .json(request)
118            .send_raw()
119            .await
120            .map_err(crate::Error::from_delta_send)?;
121        let response = crate::error::check_delta_response(response).await?;
122        let result = response.bytes().await?;
123        Ok(serde_json::from_slice(&result)?)
124    }
125
126    /// Finalize a managed table from its staging reservation —
127    /// `POST /delta/v1/catalogs/{catalog}/schemas/{schema}/tables`.
128    ///
129    /// Called after the client has written `_delta_log/0.json` (carrying the
130    /// required features/properties, including `io.unitycatalog.tableId`) to the
131    /// staging `location`. The server validates the contract and registers the
132    /// table at version 0.
133    pub async fn create_table(
134        &self,
135        catalog: &str,
136        schema: &str,
137        request: &DeltaCreateTableRequest,
138    ) -> Result<DeltaLoadTableResponse> {
139        let url = self.url(&format!(
140            "catalogs/{}/schemas/{}/tables",
141            encode_segment(catalog),
142            encode_segment(schema),
143        ))?;
144        let response = self
145            .client
146            .post(url)
147            .json(request)
148            .send_raw()
149            .await
150            .map_err(crate::Error::from_delta_send)?;
151        let response = crate::error::check_delta_response(response).await?;
152        let result = response.bytes().await?;
153        Ok(serde_json::from_slice(&result)?)
154    }
155
156    /// Load a Delta table's metadata, unbackfilled CCv2 commits, and latest
157    /// ratified version — `GET /delta/v1/catalogs/{catalog}/schemas/{schema}/tables/{table}`.
158    ///
159    /// For a catalog-managed (MANAGED) table the response carries the `commits`
160    /// log tail and `latest_table_version` a reader needs to materialize the
161    /// catalog's ratified snapshot.
162    pub async fn load_table(
163        &self,
164        catalog: &str,
165        schema: &str,
166        table: &str,
167    ) -> Result<DeltaLoadTableResponse> {
168        let url = self.table_url(catalog, schema, table, "")?;
169        let response = self
170            .client
171            .get(url)
172            .send_raw()
173            .await
174            .map_err(crate::Error::from_delta_send)?;
175        let response = crate::error::check_delta_response(response).await?;
176        let result = response.bytes().await?;
177        Ok(serde_json::from_slice(&result)?)
178    }
179
180    /// Update a table — `POST /delta/v1/catalogs/{catalog}/schemas/{schema}/tables/{table}`.
181    ///
182    /// This is the catalog-managed commit surface: the `add-commit` action
183    /// proposes a staged commit (the server ratifies iff `version == last + 1`,
184    /// returning `409` on conflict), and `set-latest-backfilled-version` notifies
185    /// the catalog that commits up to a version have been published into
186    /// `_delta_log/`. Metadata/property/schema changes ride the same call.
187    pub async fn update_table(
188        &self,
189        catalog: &str,
190        schema: &str,
191        table: &str,
192        request: &DeltaUpdateTableRequest,
193    ) -> Result<DeltaLoadTableResponse> {
194        let url = self.table_url(catalog, schema, table, "")?;
195        let response = self
196            .client
197            .post(url)
198            .json(request)
199            .send_raw()
200            .await
201            .map_err(crate::Error::from_delta_send)?;
202        let response = crate::error::check_delta_response(response).await?;
203        let result = response.bytes().await?;
204        Ok(serde_json::from_slice(&result)?)
205    }
206
207    /// Delete a table — `DELETE /delta/v1/catalogs/{catalog}/schemas/{schema}/tables/{table}`.
208    ///
209    /// The server responds `204 No Content` on success.
210    pub async fn delete_table(&self, catalog: &str, schema: &str, table: &str) -> Result<()> {
211        let url = self.table_url(catalog, schema, table, "")?;
212        let response = self
213            .client
214            .delete(url)
215            .send_raw()
216            .await
217            .map_err(crate::Error::from_delta_send)?;
218        crate::error::check_delta_response(response).await?;
219        Ok(())
220    }
221
222    /// Check whether a table exists —
223    /// `HEAD /delta/v1/catalogs/{catalog}/schemas/{schema}/tables/{table}`.
224    ///
225    /// Returns `Ok(true)` on `204`, `Ok(false)` on `404`, and an error for any
226    /// other non-success status. A HEAD response carries no body, so a `404` here
227    /// is mapped directly rather than parsed as a Delta error envelope.
228    pub async fn table_exists(&self, catalog: &str, schema: &str, table: &str) -> Result<bool> {
229        let url = self.table_url(catalog, schema, table, "")?;
230        // The two transports classify a 404 differently: CloudClient's retry layer
231        // surfaces it as `Err(SendRawError::Retry)` with the status attached,
232        // whereas WasmClient's `send_raw` is a plain send with no `error_for_status`
233        // so a 404 comes back as `Ok(Response)` with a 404 status.
234        #[cfg(not(target_arch = "wasm32"))]
235        {
236            match self.client.head(url).send_raw().await {
237                // Any 2xx (the server responds 204) means the table exists.
238                Ok(_) => Ok(true),
239                // A 404 is the documented "does not exist" signal — a HEAD carries
240                // no body, so map it directly rather than parsing an error envelope.
241                Err(olai_http::SendRawError::Retry(ref e))
242                    if e.status() == Some(reqwest::StatusCode::NOT_FOUND) =>
243                {
244                    Ok(false)
245                }
246                Err(e) => Err(crate::Error::from_delta_send(e)),
247            }
248        }
249        #[cfg(target_arch = "wasm32")]
250        {
251            let response = self
252                .client
253                .head(url)
254                .send_raw()
255                .await
256                .map_err(crate::Error::from_delta_send)?;
257            let status = response.status();
258            if status == reqwest::StatusCode::NOT_FOUND {
259                Ok(false)
260            } else if status.is_success() {
261                Ok(true)
262            } else {
263                Err(crate::Error::delta_status(status))
264            }
265        }
266    }
267
268    /// Rename a table within the same catalog and schema —
269    /// `POST /delta/v1/catalogs/{catalog}/schemas/{schema}/tables/{table}/rename`.
270    ///
271    /// Cross-schema and cross-catalog moves are not supported. The server responds
272    /// `204 No Content` on success.
273    pub async fn rename_table(
274        &self,
275        catalog: &str,
276        schema: &str,
277        table: &str,
278        request: &DeltaRenameTableRequest,
279    ) -> Result<()> {
280        let url = self.table_url(catalog, schema, table, "/rename")?;
281        let response = self
282            .client
283            .post(url)
284            .json(request)
285            .send_raw()
286            .await
287            .map_err(crate::Error::from_delta_send)?;
288        crate::error::check_delta_response(response).await?;
289        Ok(())
290    }
291
292    /// Vend temporary credentials for a table's data —
293    /// `GET /delta/v1/catalogs/{catalog}/schemas/{schema}/tables/{table}/credentials?operation={op}`.
294    ///
295    /// `operation` scopes the credential to `READ` or `READ_WRITE` access.
296    pub async fn get_table_credentials(
297        &self,
298        catalog: &str,
299        schema: &str,
300        table: &str,
301        operation: DeltaCredentialOperation,
302    ) -> Result<DeltaCredentialsResponse> {
303        let url = self.table_url(catalog, schema, table, "/credentials")?;
304        let response = self
305            .client
306            .get(url)
307            .query(&[("operation", operation_param(operation))])
308            .send_raw()
309            .await
310            .map_err(crate::Error::from_delta_send)?;
311        let response = crate::error::check_delta_response(response).await?;
312        let result = response.bytes().await?;
313        Ok(serde_json::from_slice(&result)?)
314    }
315
316    /// Vend `READ_WRITE` credentials for a staging table identified by its UUID —
317    /// `GET /delta/v1/staging-tables/{table_id}/credentials`.
318    ///
319    /// A staging table is globally identified by its UUID (no catalog/schema). This
320    /// is how staging-phase credentials are refreshed while the client writes the
321    /// initial commit.
322    pub async fn get_staging_table_credentials(
323        &self,
324        table_id: &str,
325    ) -> Result<DeltaCredentialsResponse> {
326        let url = self.url(&format!(
327            "staging-tables/{}/credentials",
328            encode_segment(table_id),
329        ))?;
330        let response = self
331            .client
332            .get(url)
333            .send_raw()
334            .await
335            .map_err(crate::Error::from_delta_send)?;
336        let response = crate::error::check_delta_response(response).await?;
337        let result = response.bytes().await?;
338        Ok(serde_json::from_slice(&result)?)
339    }
340
341    /// Vend temporary credentials for a storage path (for creating a new external
342    /// table) — `GET /delta/v1/temporary-path-credentials?location={location}&operation={op}`.
343    ///
344    /// The path is not yet part of any catalog/namespace, so it is passed as a
345    /// query parameter rather than a path segment.
346    pub async fn get_temporary_path_credentials(
347        &self,
348        location: &str,
349        operation: DeltaCredentialOperation,
350    ) -> Result<DeltaCredentialsResponse> {
351        let url = self.url("temporary-path-credentials")?;
352        let response = self
353            .client
354            .get(url)
355            .query(&[
356                ("location", location),
357                ("operation", operation_param(operation)),
358            ])
359            .send_raw()
360            .await
361            .map_err(crate::Error::from_delta_send)?;
362        let response = crate::error::check_delta_response(response).await?;
363        let result = response.bytes().await?;
364        Ok(serde_json::from_slice(&result)?)
365    }
366
367    /// Report commit metrics (telemetry) for a table —
368    /// `POST /delta/v1/catalogs/{catalog}/schemas/{schema}/tables/{table}/metrics`.
369    ///
370    /// The path `{table}` and the body `table-id` must identify the same table.
371    /// The server responds `204 No Content` on success.
372    pub async fn report_metrics(
373        &self,
374        catalog: &str,
375        schema: &str,
376        table: &str,
377        request: &DeltaReportMetricsRequest,
378    ) -> Result<()> {
379        let url = self.table_url(catalog, schema, table, "/metrics")?;
380        let response = self
381            .client
382            .post(url)
383            .json(request)
384            .send_raw()
385            .await
386            .map_err(crate::Error::from_delta_send)?;
387        crate::error::check_delta_response(response).await?;
388        Ok(())
389    }
390
391    /// Build a `catalogs/{c}/schemas/{s}/tables/{t}{suffix}` URL with each name
392    /// percent-encoded. `suffix` is a literal sub-path (e.g. `"/rename"`) or empty.
393    fn table_url(&self, catalog: &str, schema: &str, table: &str, suffix: &str) -> Result<Url> {
394        self.url(&format!(
395            "catalogs/{}/schemas/{}/tables/{}{suffix}",
396            encode_segment(catalog),
397            encode_segment(schema),
398            encode_segment(table),
399        ))
400    }
401}
402
403// Native-only: these tests use mockito + `CloudClient::new_unauthenticated` and a
404// tokio runtime, none of which exist on wasm32.
405#[cfg(all(test, not(target_arch = "wasm32")))]
406mod tests {
407    use super::*;
408    use mockito::Server;
409    use unitycatalog_delta_api::models::DeltaErrorType;
410
411    fn test_client(server: &Server) -> DeltaV1Client {
412        let base = Url::parse(&server.url()).unwrap();
413        DeltaV1Client::new(Transport::new_unauthenticated(), base)
414    }
415
416    /// A minimal `DeltaLoadTableResponse` JSON body for happy-path assertions.
417    fn load_table_body() -> &'static str {
418        r#"{"metadata":{"etag":"e","table-type":"MANAGED","table-uuid":"u",
419            "location":"s3://b/t","created-time":0,"updated-time":0,
420            "columns":{"type":"struct","fields":[]},"properties":{}}}"#
421    }
422
423    #[tokio::test]
424    async fn load_table_maps_delta_not_found() {
425        let mut server = Server::new_async().await;
426        let m = server
427            .mock("GET", "/delta/v1/catalogs/c/schemas/s/tables/t")
428            .with_status(404)
429            .with_header("content-type", "application/json")
430            .with_body(
431                r#"{"error":{"message":"no table","type":"NoSuchTableException","code":404}}"#,
432            )
433            .create_async()
434            .await;
435
436        let err = test_client(&server)
437            .load_table("c", "s", "t")
438            .await
439            .unwrap_err();
440        m.assert_async().await;
441        assert!(err.is_not_found());
442        assert!(matches!(
443            err,
444            crate::Error::Delta(ref model)
445                if model.error_type == DeltaErrorType::NoSuchTableException
446        ));
447    }
448
449    #[tokio::test]
450    async fn update_table_maps_commit_conflict() {
451        let mut server = Server::new_async().await;
452        let m = server
453            .mock("POST", "/delta/v1/catalogs/c/schemas/s/tables/t")
454            .with_status(409)
455            .with_body(
456                r#"{"error":{"message":"conflict","type":"CommitVersionConflictException","code":409}}"#,
457            )
458            .create_async()
459            .await;
460
461        let req = DeltaUpdateTableRequest {
462            requirements: vec![],
463            updates: vec![],
464        };
465        let err = test_client(&server)
466            .update_table("c", "s", "t", &req)
467            .await
468            .unwrap_err();
469        m.assert_async().await;
470        assert!(err.is_commit_conflict());
471    }
472
473    #[tokio::test]
474    async fn url_segments_are_percent_encoded() {
475        let mut server = Server::new_async().await;
476        // A schema with a space and a table with a slash must be encoded so they
477        // route to a single resource rather than extra path structure.
478        let m = server
479            .mock(
480                "GET",
481                "/delta/v1/catalogs/c/schemas/my%20schema/tables/weird%2Fname",
482            )
483            .with_status(200)
484            .with_header("content-type", "application/json")
485            .with_body(load_table_body())
486            .create_async()
487            .await;
488
489        let resp = test_client(&server)
490            .load_table("c", "my schema", "weird/name")
491            .await
492            .unwrap();
493        m.assert_async().await;
494        assert_eq!(resp.metadata.location, "s3://b/t");
495    }
496
497    #[tokio::test]
498    async fn table_exists_true_false() {
499        let mut server = Server::new_async().await;
500        let exists = server
501            .mock("HEAD", "/delta/v1/catalogs/c/schemas/s/tables/yes")
502            .with_status(204)
503            .create_async()
504            .await;
505        let missing = server
506            .mock("HEAD", "/delta/v1/catalogs/c/schemas/s/tables/no")
507            .with_status(404)
508            .create_async()
509            .await;
510
511        let client = test_client(&server);
512        assert!(client.table_exists("c", "s", "yes").await.unwrap());
513        assert!(!client.table_exists("c", "s", "no").await.unwrap());
514        exists.assert_async().await;
515        missing.assert_async().await;
516    }
517
518    #[tokio::test]
519    async fn get_config_sends_query_params() {
520        let mut server = Server::new_async().await;
521        let m = server
522            .mock("GET", "/delta/v1/config")
523            .match_query(mockito::Matcher::AllOf(vec![
524                mockito::Matcher::UrlEncoded("catalog".into(), "main".into()),
525                mockito::Matcher::UrlEncoded("protocol-versions".into(), "1.1,2.3".into()),
526            ]))
527            .with_status(200)
528            .with_header("content-type", "application/json")
529            .with_body(r#"{"endpoints":["GET /v1/config"],"protocol-version":"1.0"}"#)
530            .create_async()
531            .await;
532
533        let cfg = test_client(&server)
534            .get_config("main", "1.1,2.3")
535            .await
536            .unwrap();
537        m.assert_async().await;
538        assert_eq!(cfg.protocol_version, "1.0");
539    }
540
541    #[tokio::test]
542    async fn get_table_credentials_sends_operation() {
543        let mut server = Server::new_async().await;
544        let m = server
545            .mock("GET", "/delta/v1/catalogs/c/schemas/s/tables/t/credentials")
546            .match_query(mockito::Matcher::UrlEncoded(
547                "operation".into(),
548                "READ_WRITE".into(),
549            ))
550            .with_status(200)
551            .with_header("content-type", "application/json")
552            .with_body(r#"{"storage-credentials":[]}"#)
553            .create_async()
554            .await;
555
556        let resp = test_client(&server)
557            .get_table_credentials("c", "s", "t", DeltaCredentialOperation::ReadWrite)
558            .await
559            .unwrap();
560        m.assert_async().await;
561        assert!(resp.storage_credentials.is_empty());
562    }
563
564    #[tokio::test]
565    async fn rename_and_delete_and_metrics_no_content() {
566        let mut server = Server::new_async().await;
567        let rename = server
568            .mock("POST", "/delta/v1/catalogs/c/schemas/s/tables/t/rename")
569            .with_status(204)
570            .create_async()
571            .await;
572        let delete = server
573            .mock("DELETE", "/delta/v1/catalogs/c/schemas/s/tables/t")
574            .with_status(204)
575            .create_async()
576            .await;
577        let metrics = server
578            .mock("POST", "/delta/v1/catalogs/c/schemas/s/tables/t/metrics")
579            .with_status(204)
580            .create_async()
581            .await;
582
583        let client = test_client(&server);
584        client
585            .rename_table(
586                "c",
587                "s",
588                "t",
589                &DeltaRenameTableRequest {
590                    new_name: "t2".into(),
591                },
592            )
593            .await
594            .unwrap();
595        client.delete_table("c", "s", "t").await.unwrap();
596        client
597            .report_metrics(
598                "c",
599                "s",
600                "t",
601                &DeltaReportMetricsRequest {
602                    table_id: "u".into(),
603                    report: None,
604                },
605            )
606            .await
607            .unwrap();
608        rename.assert_async().await;
609        delete.assert_async().await;
610        metrics.assert_async().await;
611    }
612
613    #[tokio::test]
614    async fn temporary_path_credentials_query() {
615        let mut server = Server::new_async().await;
616        let m = server
617            .mock("GET", "/delta/v1/temporary-path-credentials")
618            .match_query(mockito::Matcher::AllOf(vec![
619                mockito::Matcher::UrlEncoded("location".into(), "s3://bucket/path".into()),
620                mockito::Matcher::UrlEncoded("operation".into(), "READ".into()),
621            ]))
622            .with_status(200)
623            .with_header("content-type", "application/json")
624            .with_body(r#"{"storage-credentials":[]}"#)
625            .create_async()
626            .await;
627
628        test_client(&server)
629            .get_temporary_path_credentials("s3://bucket/path", DeltaCredentialOperation::Read)
630            .await
631            .unwrap();
632        m.assert_async().await;
633    }
634}
unitycatalog_client/delta_v1.rs

unitycatalog_client/
delta_v1.rs
