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

unitycatalog_client/
delta_v1.rs
