From 54ef9762d69842c01aee90048d60b7593eeb3a58 Mon Sep 17 00:00:00 2001 From: Jan Kaul Date: Wed, 15 Jul 2026 09:23:25 +0200 Subject: [PATCH 1/2] refactor: replace FromResponse trait with fetch_inner + fetch_conditional - Extract HTTP mechanics into private fetch_inner returning reqwest::Response - fetch becomes a thin wrapper (T: DeserializeOwned), absorbing fetch_empty - fetch_conditional returns Conditional for 304-aware endpoints - Conditional::Modified now carries the full HeaderMap instead of a single extracted etag string, letting callers access any response header - load_table is non-generic, always returns Conditional - No FromResponse trait needed; no fetch_empty function needed --- .../src/apis/catalog_api_api.rs | 29 ++-- .../iceberg-rest-catalog/src/apis/fetch.rs | 142 +++++++----------- catalogs/iceberg-rest-catalog/src/apis/mod.rs | 8 +- catalogs/iceberg-rest-catalog/src/catalog.rs | 4 + 4 files changed, 80 insertions(+), 103 deletions(-) diff --git a/catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs b/catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs index 5752aa20..3a1e1c73 100644 --- a/catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs +++ b/catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs @@ -444,7 +444,7 @@ pub async fn commit_transaction( let uri_str = "transactions/commit".to_string(); let method = reqwest::Method::POST; - fetch::fetch_empty( + fetch::fetch( configuration, method, prefix, @@ -548,7 +548,7 @@ pub async fn drop_namespace( let method = reqwest::Method::DELETE; - fetch::fetch_empty(configuration, method, prefix, &uri_str, &(), None, None).await + fetch::fetch(configuration, method, prefix, &uri_str, &(), None, None).await } /// Remove a table from the catalog @@ -571,7 +571,7 @@ pub async fn drop_table( ); let method = reqwest::Method::DELETE; - fetch::fetch_empty( + fetch::fetch( configuration, method, prefix, @@ -597,7 +597,7 @@ pub async fn drop_view( ); let method = reqwest::Method::DELETE; - fetch::fetch_empty(configuration, method, prefix, &uri_str, &(), None, None).await + fetch::fetch(configuration, method, prefix, &uri_str, &(), None, None).await } /// List all namespaces at a certain level, optionally starting from a given parent namespace. If table accounting.tax.paid.info exists, using 'SELECT NAMESPACE IN accounting' would translate into `GET /namespaces?parent=accounting` and must return a namespace, [\"accounting\", \"tax\"] only. Using 'SELECT NAMESPACE IN accounting.tax' would translate into `GET /namespaces?parent=accounting%1Ftax` and must return a namespace, [\"accounting\", \"tax\", \"paid\"]. If `parent` is not provided, all top-level namespaces should be listed. @@ -785,17 +785,14 @@ pub async fn load_namespace_metadata( /// /// The headers argument can control how the Iceberg catalog handles credentials (via /// `X-Iceberg-Access-Delegation`) and whether to use HTTP cache semantics (via `If-None-Match`). -pub async fn load_table( +pub async fn load_table( configuration: &configuration::Configuration, prefix: Option<&str>, namespace: &str, table: &str, headers: HashMap, snapshots: Option<&str>, -) -> Result> -where - T: super::FromResponse, -{ +) -> Result, Error> { let mut query_params = HashMap::new(); if let Some(snapshots) = snapshots { query_params.insert("snapshots".to_owned(), snapshots.to_string()); @@ -807,7 +804,7 @@ where table = crate::apis::urlencode(table) ); - fetch::fetch( + fetch::fetch_conditional( configuration, reqwest::Method::GET, prefix, @@ -849,7 +846,7 @@ pub async fn namespace_exists( let method = reqwest::Method::HEAD; - fetch::fetch_empty(configuration, method, prefix, &uri_str, &(), None, None).await + fetch::fetch(configuration, method, prefix, &uri_str, &(), None, None).await } /// Register a table using given metadata file location. @@ -887,7 +884,7 @@ pub async fn rename_table( let uri_str = "tables/rename".to_string(); let method = reqwest::Method::POST; - fetch::fetch_empty( + fetch::fetch( configuration, method, prefix, @@ -908,7 +905,7 @@ pub async fn rename_view( let uri_str = "views/rename".to_string(); let method = reqwest::Method::POST; - fetch::fetch_empty( + fetch::fetch( configuration, method, prefix, @@ -961,7 +958,7 @@ pub async fn report_metrics( ); let method = reqwest::Method::POST; - fetch::fetch_empty( + fetch::fetch( configuration, method, prefix, @@ -988,7 +985,7 @@ pub async fn table_exists( let method = reqwest::Method::HEAD; - fetch::fetch_empty(configuration, method, prefix, &uri_str, &(), None, None).await + fetch::fetch(configuration, method, prefix, &uri_str, &(), None, None).await } /// Set and/or remove properties on a namespace. The request body specifies a list of properties to remove and a map of key value pairs to update. Properties that are not in the request are not modified or removed by this call. Server implementations are not required to support namespace properties. @@ -1058,5 +1055,5 @@ pub async fn view_exists( let method = reqwest::Method::HEAD; - fetch::fetch_empty(configuration, method, prefix, &uri_str, &(), None, None).await + fetch::fetch(configuration, method, prefix, &uri_str, &(), None, None).await } diff --git a/catalogs/iceberg-rest-catalog/src/apis/fetch.rs b/catalogs/iceberg-rest-catalog/src/apis/fetch.rs index 5b576690..c2d97979 100644 --- a/catalogs/iceberg-rest-catalog/src/apis/fetch.rs +++ b/catalogs/iceberg-rest-catalog/src/apis/fetch.rs @@ -2,76 +2,9 @@ use crate::apis::{configuration, ResponseContent}; use super::{Conditional, Error}; -use async_trait::async_trait; use std::collections::HashMap; -/// Turns a raw [`reqwest::Response`] into the caller's chosen output type. The output type -/// selects how the response is interpreted, so a single [`fetch`] serves every case: -/// - a `Deserialize` type parses the JSON body (an empty body is treated as `null`, so `()` -/// works for endpoints with no content); -/// - [`Conditional`] additionally honours `304 Not Modified` and captures the response `ETag`. -#[async_trait] -pub trait FromResponse: Sized { - async fn from_response(resp: reqwest::Response) -> Result>; -} - -#[async_trait] -impl FromResponse for T -where - T: serde::de::DeserializeOwned, - E: serde::de::DeserializeOwned, -{ - async fn from_response(resp: reqwest::Response) -> Result> { - let status = resp.status(); - let content = resp.text().await?; - if !status.is_client_error() && !status.is_server_error() { - let body = if content.is_empty() { "null" } else { &content }; - serde_json::from_str(body).map_err(Error::from) - } else { - let entity: Option = serde_json::from_str(&content).ok(); - Err(Error::ResponseError(ResponseContent { - status, - content, - entity, - })) - } - } -} - -#[async_trait] -impl FromResponse for Conditional -where - T: serde::de::DeserializeOwned, - E: serde::de::DeserializeOwned, -{ - async fn from_response(resp: reqwest::Response) -> Result> { - if resp.status() == reqwest::StatusCode::NOT_MODIFIED { - return Ok(Conditional::NotModified); - } - // Capture the ETag before consuming the response body. - let etag = resp - .headers() - .get(reqwest::header::ETAG) - .and_then(|v| v.to_str().ok()) - .map(str::to_owned); - let status = resp.status(); - let content = resp.text().await?; - if !status.is_client_error() && !status.is_server_error() { - let body = if content.is_empty() { "null" } else { &content }; - let value = serde_json::from_str(body).map_err(Error::from)?; - Ok(Conditional::Modified { value, etag }) - } else { - let entity: Option = serde_json::from_str(&content).ok(); - Err(Error::ResponseError(ResponseContent { - status, - content, - entity, - })) - } - } -} - -pub(crate) async fn fetch( +pub(crate) async fn fetch_inner( configuration: &configuration::Configuration, method: reqwest::Method, prefix: Option<&str>, @@ -79,10 +12,9 @@ pub(crate) async fn fetch( request: &R, headers: Option>, query_params: Option>, -) -> Result> +) -> Result> where R: serde::Serialize + ?Sized, - T: FromResponse, { let uri_base = build_uri_base(configuration, prefix); let client = &configuration.client; @@ -135,11 +67,40 @@ where } let req = req_builder.build()?; - let resp = client.execute(req).await?; - T::from_response(resp).await + client.execute(req).await.map_err(Error::from) } -pub(crate) async fn fetch_empty( +pub(crate) async fn fetch( + configuration: &configuration::Configuration, + method: reqwest::Method, + prefix: Option<&str>, + uri_str: &str, + request: &R, + headers: Option>, + query_params: Option>, +) -> Result> +where + R: serde::Serialize + ?Sized, + T: serde::de::DeserializeOwned, + E: serde::de::DeserializeOwned, +{ + let resp = fetch_inner(configuration, method, prefix, uri_str, request, headers, query_params).await?; + let status = resp.status(); + let content = resp.text().await?; + if !status.is_client_error() && !status.is_server_error() { + let body = if content.is_empty() { "null" } else { &content }; + serde_json::from_str(body).map_err(Error::from) + } else { + let entity: Option = serde_json::from_str(&content).ok(); + Err(Error::ResponseError(ResponseContent { + status, + content, + entity, + })) + } +} + +pub(crate) async fn fetch_conditional( configuration: &configuration::Configuration, method: reqwest::Method, prefix: Option<&str>, @@ -147,21 +108,34 @@ pub(crate) async fn fetch_empty( request: &R, headers: Option>, query_params: Option>, -) -> Result<(), Error> +) -> Result, Error> where R: serde::Serialize + ?Sized, + T: serde::de::DeserializeOwned, E: serde::de::DeserializeOwned, { - fetch( - configuration, - method, - prefix, - uri_str, - request, - headers, - query_params, - ) - .await + let resp = fetch_inner(configuration, method, prefix, uri_str, request, headers, query_params).await?; + if resp.status() == reqwest::StatusCode::NOT_MODIFIED { + return Ok(Conditional::NotModified); + } + let resp_headers = resp.headers().clone(); + let status = resp.status(); + let content = resp.text().await?; + if !status.is_client_error() && !status.is_server_error() { + let body = if content.is_empty() { "null" } else { &content }; + let value = serde_json::from_str(body).map_err(Error::from)?; + Ok(Conditional::Modified { + value, + headers: resp_headers, + }) + } else { + let entity: Option = serde_json::from_str(&content).ok(); + Err(Error::ResponseError(ResponseContent { + status, + content, + entity, + })) + } } /// Build the base URI for REST API calls with proper prefix encoding diff --git a/catalogs/iceberg-rest-catalog/src/apis/mod.rs b/catalogs/iceberg-rest-catalog/src/apis/mod.rs index 18abeecf..56be115d 100644 --- a/catalogs/iceberg-rest-catalog/src/apis/mod.rs +++ b/catalogs/iceberg-rest-catalog/src/apis/mod.rs @@ -15,8 +15,11 @@ pub struct ResponseContent { pub enum Conditional { /// Server returned `304 Not Modified`; the caller's cached value is still current. NotModified, - /// Server returned a fresh body, along with its `ETag` if one was present. - Modified { value: T, etag: Option }, + /// Server returned a fresh body along with all HTTP response headers. + Modified { + value: T, + headers: reqwest::header::HeaderMap, + }, } #[derive(Debug)] @@ -114,6 +117,5 @@ pub mod configuration_api_api; pub(crate) mod fetch; pub mod o_auth2_api_api; -pub use fetch::FromResponse; pub mod configuration; diff --git a/catalogs/iceberg-rest-catalog/src/catalog.rs b/catalogs/iceberg-rest-catalog/src/catalog.rs index c3a4817e..a32785c1 100644 --- a/catalogs/iceberg-rest-catalog/src/catalog.rs +++ b/catalogs/iceberg-rest-catalog/src/catalog.rs @@ -359,6 +359,10 @@ impl Catalog for RestCatalog { .await .map_err(|_| Error::CatalogNotFound)?; + let apis::Conditional::Modified { value: response, .. } = response else { + return Err(Error::CatalogNotFound); + }; + let object_store = self.get_object_store(&response)?; self.cache From 94be523636f0d808eedb4617050a2c89e31f3a17 Mon Sep 17 00:00:00 2001 From: Jan Kaul Date: Wed, 15 Jul 2026 11:14:23 +0200 Subject: [PATCH 2/2] refactor: split load_table into load_table + load_table_conditional MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit load_table returns LoadTableResult directly — no wrapping, ergonomic for callers that don't need cache semantics. load_table_conditional returns Conditional and is the right call when passing If-None-Match; it can return NotModified. This avoids surprising callers of load_table with a Conditional return type when no If-None-Match header was sent. --- .../src/apis/catalog_api_api.rs | 39 ++++++++++++++++++- catalogs/iceberg-rest-catalog/src/catalog.rs | 4 -- 2 files changed, 37 insertions(+), 6 deletions(-) diff --git a/catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs b/catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs index 3a1e1c73..db9d6dd7 100644 --- a/catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs +++ b/catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs @@ -783,8 +783,8 @@ pub async fn load_namespace_metadata( /// Load a table from the catalog. The response contains the table's full metadata plus any extra /// configuration to apply to it, which may include credentials to use for subsequent requests. /// -/// The headers argument can control how the Iceberg catalog handles credentials (via -/// `X-Iceberg-Access-Delegation`) and whether to use HTTP cache semantics (via `If-None-Match`). +/// The headers argument can control how the Iceberg catalog handles credentials via +/// `X-Iceberg-Access-Delegation`. For HTTP cache semantics use [`load_table_conditional`]. pub async fn load_table( configuration: &configuration::Configuration, prefix: Option<&str>, @@ -792,6 +792,41 @@ pub async fn load_table( table: &str, headers: HashMap, snapshots: Option<&str>, +) -> Result> { + let mut query_params = HashMap::new(); + if let Some(snapshots) = snapshots { + query_params.insert("snapshots".to_owned(), snapshots.to_string()); + } + + let uri_str = format!( + "namespaces/{namespace}/tables/{table}", + namespace = crate::apis::urlencode(namespace), + table = crate::apis::urlencode(table) + ); + + fetch::fetch( + configuration, + reqwest::Method::GET, + prefix, + &uri_str, + &(), + Some(headers), + Some(query_params), + ) + .await +} + +/// Load a table from the catalog with HTTP cache semantics. Pass an `If-None-Match` header with +/// a previously received ETag to skip re-fetching unchanged metadata; the server returns +/// `304 Not Modified` in that case and this function returns [`super::Conditional::NotModified`]. +/// Use [`load_table`] when caching is not needed. +pub async fn load_table_conditional( + configuration: &configuration::Configuration, + prefix: Option<&str>, + namespace: &str, + table: &str, + headers: HashMap, + snapshots: Option<&str>, ) -> Result, Error> { let mut query_params = HashMap::new(); if let Some(snapshots) = snapshots { diff --git a/catalogs/iceberg-rest-catalog/src/catalog.rs b/catalogs/iceberg-rest-catalog/src/catalog.rs index a32785c1..c3a4817e 100644 --- a/catalogs/iceberg-rest-catalog/src/catalog.rs +++ b/catalogs/iceberg-rest-catalog/src/catalog.rs @@ -359,10 +359,6 @@ impl Catalog for RestCatalog { .await .map_err(|_| Error::CatalogNotFound)?; - let apis::Conditional::Modified { value: response, .. } = response else { - return Err(Error::CatalogNotFound); - }; - let object_store = self.get_object_store(&response)?; self.cache