From 168c7dbb5c28d89e9f6cefdf8d22c4665b466f44 Mon Sep 17 00:00:00 2001 From: Jannik Steinmann Date: Wed, 29 Jul 2026 00:14:18 +0200 Subject: [PATCH 1/6] Add RestSessionCatalog --- crates/catalog/rest/public-api.txt | 36 +- crates/catalog/rest/src/catalog.rs | 627 ++++++++++++++++++++++++----- 2 files changed, 557 insertions(+), 106 deletions(-) diff --git a/crates/catalog/rest/public-api.txt b/crates/catalog/rest/public-api.txt index 6f3fa58a47..446821606f 100644 --- a/crates/catalog/rest/public-api.txt +++ b/crates/catalog/rest/public-api.txt @@ -273,14 +273,15 @@ pub fn iceberg_catalog_rest::RestCatalog::purge_table<'life0, 'life1, 'async_tra pub fn iceberg_catalog_rest::RestCatalog::register_table<'life0, 'life1, 'async_trait>(&'life0 self, table_ident: &'life1 iceberg::catalog::TableIdent, metadata_location: alloc::string::String) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg_catalog_rest::RestCatalog::rename_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, src: &'life1 iceberg::catalog::TableIdent, dest: &'life2 iceberg::catalog::TableIdent) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait pub fn iceberg_catalog_rest::RestCatalog::table_exists<'life0, 'life1, 'async_trait>(&'life0 self, table: &'life1 iceberg::catalog::TableIdent) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait -pub fn iceberg_catalog_rest::RestCatalog::update_namespace<'life0, 'life1, 'async_trait>(&'life0 self, _namespace: &'life1 iceberg::catalog::NamespaceIdent, _properties: std::collections::hash::map::HashMap) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg_catalog_rest::RestCatalog::update_namespace<'life0, 'life1, 'async_trait>(&'life0 self, namespace: &'life1 iceberg::catalog::NamespaceIdent, properties: std::collections::hash::map::HashMap) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg_catalog_rest::RestCatalog::update_table<'life0, 'async_trait>(&'life0 self, commit: iceberg::catalog::TableCommit) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub struct iceberg_catalog_rest::RestCatalogBuilder impl iceberg_catalog_rest::RestCatalogBuilder pub fn iceberg_catalog_rest::RestCatalogBuilder::with_auth_manager(self, auth_manager: alloc::sync::Arc) -> Self pub fn iceberg_catalog_rest::RestCatalogBuilder::with_client(self, client: reqwest::async_impl::client::Client) -> Self +pub fn iceberg_catalog_rest::RestCatalogBuilder::with_session_context(self, context: iceberg::catalog::session::SessionContext) -> Self impl core::default::Default for iceberg_catalog_rest::RestCatalogBuilder -pub fn iceberg_catalog_rest::RestCatalogBuilder::default() -> Self +pub fn iceberg_catalog_rest::RestCatalogBuilder::default() -> iceberg_catalog_rest::RestCatalogBuilder impl core::fmt::Debug for iceberg_catalog_rest::RestCatalogBuilder pub fn iceberg_catalog_rest::RestCatalogBuilder::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl iceberg::catalog::CatalogBuilder for iceberg_catalog_rest::RestCatalogBuilder @@ -289,6 +290,37 @@ pub fn iceberg_catalog_rest::RestCatalogBuilder::load(self, name: impl core::con pub fn iceberg_catalog_rest::RestCatalogBuilder::with_kms_client_factory(self, kms_client_factory: alloc::sync::Arc) -> Self pub fn iceberg_catalog_rest::RestCatalogBuilder::with_runtime(self, runtime: iceberg::runtime::Runtime) -> Self pub fn iceberg_catalog_rest::RestCatalogBuilder::with_storage_factory(self, storage_factory: alloc::sync::Arc) -> Self +pub struct iceberg_catalog_rest::RestSessionCatalog +impl core::fmt::Debug for iceberg_catalog_rest::RestSessionCatalog +pub fn iceberg_catalog_rest::RestSessionCatalog::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl iceberg::catalog::session::SessionCatalog for iceberg_catalog_rest::RestSessionCatalog +pub fn iceberg_catalog_rest::RestSessionCatalog::create_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, namespace: &'life2 iceberg::catalog::NamespaceIdent, properties: std::collections::hash::map::HashMap) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::create_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, namespace: &'life2 iceberg::catalog::NamespaceIdent, creation: iceberg::catalog::TableCreation) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::drop_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, namespace: &'life2 iceberg::catalog::NamespaceIdent) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::drop_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, table: &'life2 iceberg::catalog::TableIdent) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::get_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, namespace: &'life2 iceberg::catalog::NamespaceIdent) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::list_namespaces<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, parent: core::option::Option<&'life2 iceberg::catalog::NamespaceIdent>) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::list_tables<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, namespace: &'life2 iceberg::catalog::NamespaceIdent) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::load_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, table_ident: &'life2 iceberg::catalog::TableIdent) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::namespace_exists<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, ns: &'life2 iceberg::catalog::NamespaceIdent) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::purge_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, table: &'life2 iceberg::catalog::TableIdent) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::register_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, table_ident: &'life2 iceberg::catalog::TableIdent, metadata_location: alloc::string::String) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::rename_table<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, src: &'life2 iceberg::catalog::TableIdent, dest: &'life3 iceberg::catalog::TableIdent) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::table_exists<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, table: &'life2 iceberg::catalog::TableIdent) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::update_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, _namespace: &'life2 iceberg::catalog::NamespaceIdent, _properties: std::collections::hash::map::HashMap) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::update_table<'life0, 'life1, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, commit: iceberg::catalog::TableCommit) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub struct iceberg_catalog_rest::RestSessionCatalogBuilder +impl iceberg_catalog_rest::RestSessionCatalogBuilder +pub fn iceberg_catalog_rest::RestSessionCatalogBuilder::load(self, name: impl core::convert::Into, props: std::collections::hash::map::HashMap) -> impl core::future::future::Future> + core::marker::Send +pub fn iceberg_catalog_rest::RestSessionCatalogBuilder::with_auth_manager(self, auth_manager: alloc::sync::Arc) -> Self +pub fn iceberg_catalog_rest::RestSessionCatalogBuilder::with_client(self, client: reqwest::async_impl::client::Client) -> Self +pub fn iceberg_catalog_rest::RestSessionCatalogBuilder::with_kms_client_factory(self, kms_client_factory: alloc::sync::Arc) -> Self +pub fn iceberg_catalog_rest::RestSessionCatalogBuilder::with_runtime(self, runtime: iceberg::runtime::Runtime) -> Self +pub fn iceberg_catalog_rest::RestSessionCatalogBuilder::with_storage_factory(self, storage_factory: alloc::sync::Arc) -> Self +impl core::default::Default for iceberg_catalog_rest::RestSessionCatalogBuilder +pub fn iceberg_catalog_rest::RestSessionCatalogBuilder::default() -> Self +impl core::fmt::Debug for iceberg_catalog_rest::RestSessionCatalogBuilder +pub fn iceberg_catalog_rest::RestSessionCatalogBuilder::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result pub struct iceberg_catalog_rest::StorageCredential pub iceberg_catalog_rest::StorageCredential::config: std::collections::hash::map::HashMap pub iceberg_catalog_rest::StorageCredential::prefix: alloc::string::String diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs index 0c15c50663..003e4329ee 100644 --- a/crates/catalog/rest/src/catalog.rs +++ b/crates/catalog/rest/src/catalog.rs @@ -29,7 +29,7 @@ use iceberg::io::{FileIO, FileIOBuilder, StorageFactory}; use iceberg::table::Table; use iceberg::{ Catalog, CatalogBuilder, Error, ErrorKind, Namespace, NamespaceIdent, Result, Runtime, - TableCommit, TableCreation, TableIdent, + SessionCatalog, SessionContext, TableCommit, TableCreation, TableIdent, }; use itertools::Itertools; use reqwest::header::{ @@ -68,101 +68,41 @@ const CARGO_PKG_VERSION: &str = env!("CARGO_PKG_VERSION"); const PATH_V1: &str = "v1"; /// Builder for [`RestCatalog`]. -#[derive(Debug)] +#[derive(Debug, Default)] pub struct RestCatalogBuilder { - config: RestCatalogConfig, - auth_manager: Option>, - storage_factory: Option>, - kms_client_factory: Option>, - runtime: Option, -} - -impl Default for RestCatalogBuilder { - fn default() -> Self { - Self { - config: RestCatalogConfig { - name: None, - uri: "".to_string(), - warehouse: None, - props: HashMap::new(), - client: None, - default_client: Arc::new(OnceLock::new()), - }, - auth_manager: None, - storage_factory: None, - kms_client_factory: None, - runtime: None, - } - } + session_context: Option, + inner: RestSessionCatalogBuilder, } impl CatalogBuilder for RestCatalogBuilder { type C = RestCatalog; fn with_storage_factory(mut self, storage_factory: Arc) -> Self { - self.storage_factory = Some(storage_factory); + self.inner = self.inner.with_storage_factory(storage_factory); self } fn with_kms_client_factory(mut self, kms_client_factory: Arc) -> Self { - self.kms_client_factory = Some(kms_client_factory); + self.inner = self.inner.with_kms_client_factory(kms_client_factory); self } fn with_runtime(mut self, runtime: Runtime) -> Self { - self.runtime = Some(runtime); + self.inner = self.inner.with_runtime(runtime); self } fn load( - mut self, + self, name: impl Into, props: HashMap, ) -> impl Future> + Send { - self.config.name = Some(name.into()); - - if props.contains_key(REST_CATALOG_PROP_URI) { - self.config.uri = props - .get(REST_CATALOG_PROP_URI) - .cloned() - .unwrap_or_default(); - } - - if props.contains_key(REST_CATALOG_PROP_WAREHOUSE) { - self.config.warehouse = props.get(REST_CATALOG_PROP_WAREHOUSE).cloned() - } - - // Collect other remaining properties - self.config.props = props - .into_iter() - .filter(|(k, _)| k != REST_CATALOG_PROP_URI && k != REST_CATALOG_PROP_WAREHOUSE) - .collect(); - + let name = name.into(); async move { - if self.config.name.is_none() { - Err(Error::new( - ErrorKind::DataInvalid, - "Catalog name is required", - )) - } else if self.config.uri.is_empty() { - Err(Error::new( - ErrorKind::DataInvalid, - "Catalog uri is required", - )) - } else { - let runtime = self.runtime.unwrap_or_else(Runtime::current); - let kms_client = match self.kms_client_factory { - Some(factory) => Some(factory.create_kms_client(&self.config.props).await?), - None => None, - }; - Ok(RestCatalog::new( - self.config, - self.auth_manager, - self.storage_factory, - runtime, - kms_client, - )) - } + let context = self.session_context.unwrap_or_else(SessionContext::empty); + let session_catalog = Arc::new(self.inner.load(name, props).await?); + + Ok(RestCatalog::from_session_catalog(context, session_catalog)) } } } @@ -170,13 +110,20 @@ impl CatalogBuilder for RestCatalogBuilder { impl RestCatalogBuilder { /// Configures the catalog with a custom HTTP client. pub fn with_client(mut self, client: Client) -> Self { - self.config.client = Some(client); + self.inner = self.inner.with_client(client); + self + } + + /// Configures the session context that will be used with this catalog. + /// Overwrites the default empty context from SessionContext::empty(). + pub fn with_session_context(mut self, context: SessionContext) -> Self { + self.session_context = Some(context); self } /// Injects a custom auth manager, overriding the `rest.auth.type` configuration. pub fn with_auth_manager(mut self, auth_manager: Arc) -> Self { - self.auth_manager = Some(auth_manager); + self.inner = self.inner.with_auth_manager(auth_manager); self } } @@ -487,8 +434,170 @@ impl RestClient { /// Rest catalog implementation. #[derive(Debug)] pub struct RestCatalog { - /// Injected through [`RestCatalogBuilder::with_auth_manager`]; otherwise - /// one is resolved from `rest.auth.type` when the context is built. + session_context: SessionContext, + inner: Arc, +} + +impl RestCatalog { + /// Creates a `RestCatalog` from a [`RestCatalogConfig`]. + #[cfg(test)] + fn new( + context: SessionContext, + config: RestCatalogConfig, + auth_manager: Option>, + storage_factory: Option>, + runtime: Runtime, + kms_client: Option>, + ) -> Self { + let session_catalog = Arc::new(RestSessionCatalog::new( + config, + auth_manager, + storage_factory, + runtime, + kms_client, + )); + + Self::from_session_catalog(context, session_catalog) + } + + fn from_session_catalog(context: SessionContext, inner: Arc) -> Self { + Self { + session_context: context, + inner, + } + } + + #[cfg(test)] + async fn client(&self) -> Result<&RestClient> { + self.inner.client().await + } +} + +/// All requests and expected responses are derived from the REST catalog API spec: +/// https://github.com/apache/iceberg/blob/main/open-api/rest-catalog-open-api.yaml +#[async_trait] +impl Catalog for RestCatalog { + async fn list_namespaces( + &self, + parent: Option<&NamespaceIdent>, + ) -> Result> { + self.inner + .list_namespaces(&self.session_context, parent) + .await + } + + async fn create_namespace( + &self, + namespace: &NamespaceIdent, + properties: HashMap, + ) -> Result { + self.inner + .create_namespace(&self.session_context, namespace, properties) + .await + } + + async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result { + self.inner + .get_namespace(&self.session_context, namespace) + .await + } + + async fn namespace_exists(&self, ns: &NamespaceIdent) -> Result { + self.inner.namespace_exists(&self.session_context, ns).await + } + + async fn update_namespace( + &self, + namespace: &NamespaceIdent, + properties: HashMap, + ) -> Result<()> { + self.inner + .update_namespace(&self.session_context, namespace, properties) + .await + } + + async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()> { + self.inner + .drop_namespace(&self.session_context, namespace) + .await + } + + async fn list_tables(&self, namespace: &NamespaceIdent) -> Result> { + self.inner + .list_tables(&self.session_context, namespace) + .await + } + + /// Create a new table inside the namespace. + /// + /// In the resulting table, if there are any config properties that + /// are present in both the response from the REST server and the + /// config provided when creating this `RestCatalog` instance then + /// the value provided locally to the `RestCatalog` will take precedence. + async fn create_table( + &self, + namespace: &NamespaceIdent, + creation: TableCreation, + ) -> Result { + self.inner + .create_table(&self.session_context, namespace, creation) + .await + } + + /// Load table from the catalog. + /// + /// If there are any config properties that are present in both the response from the REST + /// server and the config provided when creating this `RestCatalog` instance, then the value + /// provided locally to the `RestCatalog` will take precedence. + async fn load_table(&self, table_ident: &TableIdent) -> Result
{ + self.inner + .load_table(&self.session_context, table_ident) + .await + } + + /// Drop a table from the catalog. + async fn drop_table(&self, table: &TableIdent) -> Result<()> { + self.inner.drop_table(&self.session_context, table).await + } + + /// Drop a table from the catalog and purge its data by sending + /// `purgeRequested=true` to the REST server. + async fn purge_table(&self, table: &TableIdent) -> Result<()> { + self.inner.purge_table(&self.session_context, table).await + } + + /// Check if a table exists in the catalog. + async fn table_exists(&self, table: &TableIdent) -> Result { + self.inner.table_exists(&self.session_context, table).await + } + + /// Rename a table in the catalog. + async fn rename_table(&self, src: &TableIdent, dest: &TableIdent) -> Result<()> { + self.inner + .rename_table(&self.session_context, src, dest) + .await + } + + async fn register_table( + &self, + table_ident: &TableIdent, + metadata_location: String, + ) -> Result
{ + self.inner + .register_table(&self.session_context, table_ident, metadata_location) + .await + } + + async fn update_table(&self, commit: TableCommit) -> Result
{ + self.inner.update_table(&self.session_context, commit).await + } +} + +/// Rest catalog implementation. +#[derive(Debug)] +pub struct RestSessionCatalog { + /// Injected through [`RestSessionCatalogBuilder::with_auth_manager`]; otherwise + /// one is resolved from `rest.auth.type` when the client is built. auth_manager: Option>, /// User config is stored as-is and never be changed. /// @@ -502,8 +611,8 @@ pub struct RestCatalog { kms_client: Option>, } -impl RestCatalog { - /// Creates a `RestCatalog` from a [`RestCatalogConfig`]. +impl RestSessionCatalog { + /// Creates a `RestSessionCatalog` from a [`RestCatalogConfig`]. fn new( config: RestCatalogConfig, auth_manager: Option>, @@ -522,7 +631,12 @@ impl RestCatalog { } /// Sends a DELETE request for the given table, optionally requesting purge. - async fn delete_table(&self, table: &TableIdent, purge: bool) -> Result<()> { + async fn delete_table( + &self, + _context: &SessionContext, + table: &TableIdent, + purge: bool, + ) -> Result<()> { let client = self.client().await?; let mut request_builder = client @@ -638,7 +752,7 @@ impl RestCatalog { &Self::auth_props(&self.user_config), ) .await?; - RestCatalog::load_config( + RestSessionCatalog::load_config( &http_client.with_auth_session(Arc::from(init_session)), &self.user_config, ) @@ -765,12 +879,11 @@ impl RestCatalog { } } -/// All requests and expected responses are derived from the REST catalog API spec: -/// https://github.com/apache/iceberg/blob/main/open-api/rest-catalog-open-api.yaml #[async_trait] -impl Catalog for RestCatalog { +impl SessionCatalog for RestSessionCatalog { async fn list_namespaces( &self, + _context: &SessionContext, parent: Option<&NamespaceIdent>, ) -> Result> { let client = self.client().await?; @@ -826,6 +939,7 @@ impl Catalog for RestCatalog { async fn create_namespace( &self, + _context: &SessionContext, namespace: &NamespaceIdent, properties: HashMap, ) -> Result { @@ -861,7 +975,11 @@ impl Catalog for RestCatalog { } } - async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result { + async fn get_namespace( + &self, + _context: &SessionContext, + namespace: &NamespaceIdent, + ) -> Result { let client = self.client().await?; let request = HttpRequest::build( @@ -890,13 +1008,17 @@ impl Catalog for RestCatalog { } } - async fn namespace_exists(&self, ns: &NamespaceIdent) -> Result { + async fn namespace_exists( + &self, + context: &SessionContext, + ns: &NamespaceIdent, + ) -> Result { // Prefer a cheap HEAD when the server advertises it; otherwise fall back // to loading the namespace (GET) and treating a missing namespace as // `false`, so this still works against servers that don't advertise the // HEAD route. if !self.supports_endpoint(&V1_NAMESPACE_EXISTS).await? { - return match self.get_namespace(ns).await { + return match self.get_namespace(context, ns).await { Ok(_) => Ok(true), Err(e) if e.kind() == ErrorKind::NamespaceNotFound => Ok(false), Err(e) => Err(e), @@ -910,6 +1032,7 @@ impl Catalog for RestCatalog { async fn update_namespace( &self, + _context: &SessionContext, _namespace: &NamespaceIdent, _properties: HashMap, ) -> Result<()> { @@ -919,7 +1042,11 @@ impl Catalog for RestCatalog { )) } - async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()> { + async fn drop_namespace( + &self, + _context: &SessionContext, + namespace: &NamespaceIdent, + ) -> Result<()> { let client = self.client().await?; let request = HttpRequest::build( @@ -944,7 +1071,11 @@ impl Catalog for RestCatalog { } } - async fn list_tables(&self, namespace: &NamespaceIdent) -> Result> { + async fn list_tables( + &self, + _context: &SessionContext, + namespace: &NamespaceIdent, + ) -> Result> { let client = self.client().await?; let endpoint = client.config.tables_endpoint(namespace); let mut identifiers = Vec::new(); @@ -998,6 +1129,7 @@ impl Catalog for RestCatalog { /// the value provided locally to the `RestCatalog` will take precedence. async fn create_table( &self, + _context: &SessionContext, namespace: &NamespaceIdent, creation: TableCreation, ) -> Result
{ @@ -1083,7 +1215,11 @@ impl Catalog for RestCatalog { /// If there are any config properties that are present in both the response from the REST /// server and the config provided when creating this `RestCatalog` instance, then the value /// provided locally to the `RestCatalog` will take precedence. - async fn load_table(&self, table_ident: &TableIdent) -> Result
{ + async fn load_table( + &self, + _context: &SessionContext, + table_ident: &TableIdent, + ) -> Result
{ let client = self.client().await?; let request = HttpRequest::build( @@ -1140,23 +1276,23 @@ impl Catalog for RestCatalog { } /// Drop a table from the catalog. - async fn drop_table(&self, table: &TableIdent) -> Result<()> { - self.delete_table(table, false).await + async fn drop_table(&self, context: &SessionContext, table: &TableIdent) -> Result<()> { + self.delete_table(context, table, false).await } /// Drop a table from the catalog and purge its data by sending /// `purgeRequested=true` to the REST server. - async fn purge_table(&self, table: &TableIdent) -> Result<()> { - self.delete_table(table, true).await + async fn purge_table(&self, context: &SessionContext, table: &TableIdent) -> Result<()> { + self.delete_table(context, table, true).await } /// Check if a table exists in the catalog. - async fn table_exists(&self, table: &TableIdent) -> Result { + async fn table_exists(&self, context: &SessionContext, table: &TableIdent) -> Result { // Prefer a cheap HEAD when the server advertises it; otherwise fall back // to loading the table (GET) and treating a missing table as `false`, so // this still works against servers that don't advertise the HEAD route. if !self.supports_endpoint(&V1_TABLE_EXISTS).await? { - return match self.load_table(table).await { + return match self.load_table(context, table).await { Ok(_) => Ok(true), Err(e) if e.kind() == ErrorKind::TableNotFound => Ok(false), Err(e) => Err(e), @@ -1169,7 +1305,12 @@ impl Catalog for RestCatalog { } /// Rename a table in the catalog. - async fn rename_table(&self, src: &TableIdent, dest: &TableIdent) -> Result<()> { + async fn rename_table( + &self, + _context: &SessionContext, + src: &TableIdent, + dest: &TableIdent, + ) -> Result<()> { let client = self.client().await?; let request = HttpRequest::build( @@ -1204,6 +1345,7 @@ impl Catalog for RestCatalog { async fn register_table( &self, + _context: &SessionContext, table_ident: &TableIdent, metadata_location: String, ) -> Result
{ @@ -1271,7 +1413,11 @@ impl Catalog for RestCatalog { table_builder.build() } - async fn update_table(&self, mut commit: TableCommit) -> Result
{ + async fn update_table( + &self, + _context: &SessionContext, + mut commit: TableCommit, + ) -> Result
{ let client = self.client().await?; let request = HttpRequest::build( @@ -1349,6 +1495,170 @@ impl Catalog for RestCatalog { } } +/// Builder for [`RestSessionCatalog`]. +#[derive(Debug)] +pub struct RestSessionCatalogBuilder { + config: RestCatalogConfig, + auth_manager: Option>, + storage_factory: Option>, + kms_client_factory: Option>, + runtime: Option, +} + +impl Default for RestSessionCatalogBuilder { + fn default() -> Self { + Self { + config: RestCatalogConfig { + name: None, + uri: "".to_string(), + warehouse: None, + props: HashMap::new(), + client: None, + default_client: Arc::new(OnceLock::new()), + }, + auth_manager: None, + storage_factory: None, + kms_client_factory: None, + runtime: None, + } + } +} + +impl RestSessionCatalogBuilder { + /// Configures the catalog with a custom HTTP client. + pub fn with_client(mut self, client: Client) -> Self { + self.config.client = Some(client); + self + } + + /// Injects a custom auth manager, overriding the `rest.auth.type` configuration. + pub fn with_auth_manager(mut self, auth_manager: Arc) -> Self { + self.auth_manager = Some(auth_manager); + self + } + + /// Set a custom StorageFactory to use for storage operations. + /// + /// When a StorageFactory is provided, the catalog will use it to build FileIO + /// instances for all storage operations instead of using the default factory. + /// + /// # Arguments + /// + /// * `storage_factory` - The StorageFactory to use for creating storage instances + /// + /// # Example + /// + /// ```rust,ignore + /// use iceberg::CatalogBuilder; + /// use iceberg::io::StorageFactory; + /// use iceberg_storage_opendal::OpenDalStorageFactory; + /// use std::sync::Arc; + /// + /// let catalog = MyCatalogBuilder::default() + /// .with_storage_factory(Arc::new(OpenDalStorageFactory::S3 { + /// customized_credential_load: None, + /// })) + /// .load("my_catalog", props) + /// .await?; + /// ``` + pub fn with_storage_factory(mut self, storage_factory: Arc) -> Self { + self.storage_factory = Some(storage_factory); + self + } + + /// Set a [`KmsClientFactory`] to enable table encryption. + /// + /// When provided, the catalog calls the factory once during + /// [`load`](Self::load) with the catalog properties to create a shared + /// [`KeyManagementClient`](crate::encryption::KeyManagementClient). + /// That client is then passed to each table's `TableBuilder` so tables + /// with `encryption.key-id` set can construct an `EncryptionManager`. + /// + /// # Example + /// + /// ```rust,ignore + /// use iceberg::CatalogBuilder; + /// use iceberg::encryption::kms::KmsClientFactory; + /// use std::sync::Arc; + /// + /// let catalog = MyCatalogBuilder::default() + /// .with_kms_client_factory(Arc::new(MyKmsClientFactory)) + /// .load("my_catalog", props) + /// .await?; + /// ``` + pub fn with_kms_client_factory( + mut self, + kms_client_factory: Arc, + ) -> Self { + self.kms_client_factory = Some(kms_client_factory); + self + } + + /// Set a custom tokio Runtime to use for spawning async tasks. + /// + /// When a Runtime is provided, the catalog will propagate it to all tables + /// it creates. Tasks such as scan planning and delete file processing + /// will be spawned on this runtime. + pub fn with_runtime(mut self, runtime: Runtime) -> Self { + self.runtime = Some(runtime); + self + } + + /// Create a new catalog instance. + pub fn load( + mut self, + name: impl Into, + props: HashMap, + ) -> impl Future> + Send { + self.config.name = Some(name.into()); + + if props.contains_key(REST_CATALOG_PROP_URI) { + self.config.uri = props + .get(REST_CATALOG_PROP_URI) + .cloned() + .unwrap_or_default(); + } + + if props.contains_key(REST_CATALOG_PROP_WAREHOUSE) { + self.config.warehouse = props.get(REST_CATALOG_PROP_WAREHOUSE).cloned() + } + + // Collect other remaining properties + self.config.props = props + .into_iter() + .filter(|(k, _)| k != REST_CATALOG_PROP_URI && k != REST_CATALOG_PROP_WAREHOUSE) + .collect(); + + async move { + if self.config.name.is_none() { + Err(Error::new( + ErrorKind::DataInvalid, + "Catalog name is required", + )) + } else if self.config.uri.is_empty() { + Err(Error::new( + ErrorKind::DataInvalid, + "Catalog uri is required", + )) + } else { + let runtime = self.runtime.unwrap_or_else(Runtime::current); + let kms_client = match self.kms_client_factory { + Some(factory) => Some(factory.create_kms_client(&self.config.props).await?), + None => None, + }; + + Ok(RestSessionCatalog::new( + self.config, + self.auth_manager, + self.storage_factory, + runtime, + kms_client, + )) + } + } + } +} + #[cfg(test)] mod tests { use std::fs::File; @@ -1411,6 +1721,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -1489,6 +1800,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -1499,12 +1811,12 @@ mod tests { let plan = "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan" .parse::() .unwrap(); - assert!(catalog.supports_endpoint(&plan).await.unwrap()); + assert!(catalog.inner.supports_endpoint(&plan).await.unwrap()); // Advertised list is present but does not include this route. let delete_ns = "DELETE /v1/{prefix}/namespaces/{namespace}" .parse::() .unwrap(); - assert!(!catalog.supports_endpoint(&delete_ns).await.unwrap()); + assert!(!catalog.inner.supports_endpoint(&delete_ns).await.unwrap()); config_mock.assert_async().await; } @@ -1521,6 +1833,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -1533,12 +1846,12 @@ mod tests { let load_table = "GET /v1/{prefix}/namespaces/{namespace}/tables/{table}" .parse::() .unwrap(); - assert!(catalog.supports_endpoint(&load_table).await.unwrap()); + assert!(catalog.inner.supports_endpoint(&load_table).await.unwrap()); // But not an optional endpoint that must be advertised. let plan = "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan" .parse::() .unwrap(); - assert!(!catalog.supports_endpoint(&plan).await.unwrap()); + assert!(!catalog.inner.supports_endpoint(&plan).await.unwrap()); config_mock.assert_async().await; } @@ -1557,6 +1870,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -1567,7 +1881,7 @@ mod tests { let load_table = "GET /v1/{prefix}/namespaces/{namespace}/tables/{table}" .parse::() .unwrap(); - assert!(catalog.supports_endpoint(&load_table).await.unwrap()); + assert!(catalog.inner.supports_endpoint(&load_table).await.unwrap()); config_mock.assert_async().await; } @@ -1609,6 +1923,7 @@ mod tests { props.insert("credential".to_string(), "client1:secret1".to_string()); let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder() .uri(server.url()) .props(props) @@ -1659,6 +1974,7 @@ mod tests { let config_mock = create_config_mock(&mut server).await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder() .uri(server.url()) .props(props) @@ -1765,6 +2081,7 @@ mod tests { ); let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder() .uri(server.url()) .props(props) @@ -1818,6 +2135,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -1851,6 +2169,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -2649,6 +2968,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -2751,6 +3071,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -2807,6 +3128,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -2853,6 +3175,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -2889,6 +3212,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -2927,6 +3251,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -2958,6 +3283,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -3001,6 +3327,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -3072,6 +3399,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -3206,6 +3534,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -3253,6 +3582,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -3285,6 +3615,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -3325,6 +3656,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -3359,6 +3691,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -3396,6 +3729,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -3516,6 +3850,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -3555,6 +3890,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -3707,6 +4043,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -3779,6 +4116,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -3926,6 +4264,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -3994,6 +4333,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -4048,6 +4388,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), @@ -4090,7 +4431,8 @@ mod tests { assert!(catalog.is_ok()); - let catalog_config = catalog.unwrap().user_config; + let catalog = catalog.unwrap(); + let catalog_config = &catalog.inner.user_config; assert_eq!(catalog_config.name.as_deref(), Some("test")); assert_eq!(catalog_config.uri, "http://localhost:8080"); assert_eq!(catalog_config.warehouse, None); @@ -4120,4 +4462,81 @@ mod tests { assert_eq!(err.message(), "Catalog uri is required"); } } + + #[tokio::test] + async fn test_create_session_catalog() { + let builder = RestSessionCatalogBuilder::default(); + + let result = builder + .load( + "test", + HashMap::from([ + ( + REST_CATALOG_PROP_URI.to_string(), + "http://localhost:8080".to_string(), + ), + ("a".to_string(), "b".to_string()), + ]), + ) + .await; + + assert!(result.is_ok()); + + let catalog = result.unwrap(); + + let catalog_config = catalog.user_config; + assert_eq!(catalog_config.name.as_deref(), Some("test")); + assert_eq!(catalog_config.uri, "http://localhost:8080"); + assert_eq!(catalog_config.warehouse, None); + // The default builder sets no client (only `with_client` does). + assert!(catalog_config.client.is_none()); + + // `uri` is consumed into its own field; other props are retained. + assert_eq!(catalog_config.props.get("a"), Some(&"b".to_string())); + assert!(!catalog_config.props.contains_key(REST_CATALOG_PROP_URI)); + } + + #[tokio::test] + async fn test_create_rest_catalog_with_session() { + let context = SessionContext::builder() + .session_id("test-id".to_string()) + .build(); + + let result = RestCatalogBuilder::default() + .with_session_context(context) + .load( + "test", + HashMap::from([( + REST_CATALOG_PROP_URI.to_string(), + "http://localhost:8080".to_string(), + )]), + ) + .await; + + assert!(result.is_ok()); + + // The context passed to `with_session_context` is the one the catalog is bound to. + let catalog = result.unwrap(); + assert_eq!(catalog.session_context.session_id(), "test-id"); + } + + #[tokio::test] + async fn test_create_rest_catalog_default_session() { + let result = RestCatalogBuilder::default() + .load( + "test", + HashMap::from([( + REST_CATALOG_PROP_URI.to_string(), + "http://localhost:8080".to_string(), + )]), + ) + .await; + + assert!(result.is_ok()); + + // Without `with_session`, the catalog falls back to `SessionContext::empty()`, + // which assigns a fresh v4 UUID. + let catalog = result.unwrap(); + assert!(uuid::Uuid::parse_str(catalog.session_context.session_id()).is_ok()); + } } From 0a981462e3424541e49ecb234ab3a3e5b14a1796 Mon Sep 17 00:00:00 2001 From: Jannik Steinmann Date: Thu, 13 Aug 2026 22:50:51 +0200 Subject: [PATCH 2/6] Extract RestClient::init() --- crates/catalog/rest/src/catalog.rs | 88 ++++++++++++++++-------------- 1 file changed, 46 insertions(+), 42 deletions(-) diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs index 003e4329ee..58ea0477ef 100644 --- a/crates/catalog/rest/src/catalog.rs +++ b/crates/catalog/rest/src/catalog.rs @@ -419,6 +419,51 @@ struct RestClient { } impl RestClient { + async fn init( + user_config: &RestCatalogConfig, + auth_manager: Arc, + ) -> Result { + let http_client = HttpClient::new(user_config)?; + // The init session lives only for the config handshake, so a + // manager whose session guards a one-shot resource can release + // it before deriving the catalog session. + let catalog_config = { + let init_session = auth_manager + .init_session( + &http_client.without_auth_session(), + &RestSessionCatalog::auth_props(user_config), + ) + .await?; + RestSessionCatalog::load_config( + &http_client.with_auth_session(Arc::from(init_session)), + user_config, + ) + .await? + }; + // Use the advertised endpoints as-is, falling back to + // `DEFAULT_ENDPOINTS` when absent or empty. + let endpoints = match &catalog_config.endpoints { + Some(advertised) if !advertised.is_empty() => advertised.iter().cloned().collect(), + _ => crate::endpoint::DEFAULT_ENDPOINTS.clone(), + }; + let config = user_config.clone().merge_with_config(catalog_config); + let http_client = http_client.update_with(&config)?; + // The manager is handed an unauthenticated client: its own + // requests must not be signed by the session it is deriving. + let session = auth_manager + .catalog_session( + &http_client.without_auth_session(), + &RestSessionCatalog::auth_props(&config), + ) + .await?; + + Ok(Self { + config, + http_client: http_client.with_auth_session(session), + endpoints, + }) + } + /// Testing only: the bearer token the catalog session would attach. #[cfg(test)] async fn token(&self) -> Option { @@ -740,48 +785,7 @@ impl RestSessionCatalog { async fn client(&self) -> Result<&RestClient> { self.client .get_or_try_init(|| async { - let http_client = HttpClient::new(&self.user_config)?; - let auth_manager = self.resolve_auth_manager()?; - // The init session lives only for the config handshake, so a - // manager whose session guards a one-shot resource can release - // it before deriving the catalog session. - let catalog_config = { - let init_session = auth_manager - .init_session( - &http_client.without_auth_session(), - &Self::auth_props(&self.user_config), - ) - .await?; - RestSessionCatalog::load_config( - &http_client.with_auth_session(Arc::from(init_session)), - &self.user_config, - ) - .await? - }; - // Use the advertised endpoints as-is, falling back to - // `DEFAULT_ENDPOINTS` when absent or empty. - let endpoints = match &catalog_config.endpoints { - Some(advertised) if !advertised.is_empty() => { - advertised.iter().cloned().collect() - } - _ => crate::endpoint::DEFAULT_ENDPOINTS.clone(), - }; - let config = self.user_config.clone().merge_with_config(catalog_config); - let http_client = http_client.update_with(&config)?; - // The manager is handed an unauthenticated client: its own - // requests must not be signed by the session it is deriving. - let session = auth_manager - .catalog_session( - &http_client.without_auth_session(), - &Self::auth_props(&config), - ) - .await?; - - Ok(RestClient { - config, - http_client: http_client.with_auth_session(session), - endpoints, - }) + RestClient::init(&self.user_config, self.resolve_auth_manager()?).await }) .await } From 49cb24e949a03c90e531b2e7707d64e41dda3071 Mon Sep 17 00:00:00 2001 From: Jannik Steinmann Date: Fri, 17 Jul 2026 01:10:29 +0200 Subject: [PATCH 3/6] Refactor test suite to target RestSessionCatalog --- crates/catalog/rest/src/catalog.rs | 369 ++++++++++++----------------- 1 file changed, 150 insertions(+), 219 deletions(-) diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs index 58ea0477ef..fd2362e5a5 100644 --- a/crates/catalog/rest/src/catalog.rs +++ b/crates/catalog/rest/src/catalog.rs @@ -1686,15 +1686,15 @@ mod tests { use crate::auth::AuthSession; use crate::request::HttpRequest; - fn test_catalog(config: RestCatalogConfig) -> RestCatalog { + fn test_catalog(config: RestCatalogConfig) -> RestSessionCatalog { test_catalog_with(config, None) } fn test_catalog_with( config: RestCatalogConfig, auth_manager: Option>, - ) -> RestCatalog { - RestCatalog::new(config, auth_manager, None, Runtime::current(), None) + ) -> RestSessionCatalog { + RestSessionCatalog::new(config, auth_manager, None, Runtime::current(), None) } fn test_client() -> HttpClient { @@ -1706,6 +1706,17 @@ mod tests { .unwrap() } + /// Builds a [`RestSessionCatalog`] with the default test storage factory and runtime. + fn session_catalog(config: RestCatalogConfig) -> RestSessionCatalog { + RestSessionCatalog::new( + config, + None, + Some(Arc::new(LocalFsStorageFactory)), + Runtime::current(), + None, + ) + } + #[tokio::test] async fn test_update_config() { let mut server = Server::new_async().await; @@ -1724,14 +1735,7 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); assert_eq!( catalog @@ -1926,16 +1930,11 @@ mod tests { let mut props = HashMap::new(); props.insert("credential".to_string(), "client1:secret1".to_string()); - let catalog = RestCatalog::new( - SessionContext::empty(), + let catalog = session_catalog( RestCatalogConfig::builder() .uri(server.url()) .props(props) .build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, ); let token = catalog.client().await.unwrap().token().await; @@ -1977,16 +1976,11 @@ mod tests { let config_mock = create_config_mock(&mut server).await; - let catalog = RestCatalog::new( - SessionContext::empty(), + let catalog = session_catalog( RestCatalogConfig::builder() .uri(server.url()) .props(props) .build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, ); let token = catalog.client().await.unwrap().token().await; @@ -2084,16 +2078,11 @@ mod tests { format!("{}{}", auth_server.url(), auth_server_path).to_string(), ); - let catalog = RestCatalog::new( - SessionContext::empty(), + let catalog = session_catalog( RestCatalogConfig::builder() .uri(server.url()) .props(props) .build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, ); let token = catalog.client().await.unwrap().token().await; @@ -2138,16 +2127,12 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); - let _namespaces = catalog.list_namespaces(None).await.unwrap(); + let _namespaces = catalog + .list_namespaces(&SessionContext::empty(), None) + .await + .unwrap(); config_mock.assert_async().await; list_ns_mock.assert_async().await; @@ -2172,16 +2157,12 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); - let namespaces = catalog.list_namespaces(None).await.unwrap(); + let namespaces = catalog + .list_namespaces(&SessionContext::empty(), None) + .await + .unwrap(); let expected_ns = vec![ NamespaceIdent::from_vec(vec!["ns1".to_string(), "ns11".to_string()]).unwrap(), @@ -2241,6 +2222,7 @@ mod tests { ("header.authorization".to_string(), "Basic xyz".to_string()), ]); let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder() .uri(server.url()) .props(props) @@ -2279,6 +2261,7 @@ mod tests { let props = HashMap::from([("credential".to_string(), "client1:secret1".to_string())]); let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder() .uri(bootstrap.url()) .props(props) @@ -2436,6 +2419,7 @@ mod tests { "aud-1".to_string(), )])); let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), Some(Arc::new(manager)), Some(Arc::new(LocalFsStorageFactory)), @@ -2548,6 +2532,7 @@ mod tests { .await; let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder() .uri(server.url()) .props(HashMap::from([( @@ -2605,6 +2590,7 @@ mod tests { let config_mock = create_config_mock(&mut server).await; let captured = Arc::new(AsyncMutex::new(None)); let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder() .uri(server.url()) .props(HashMap::from([( @@ -2679,6 +2665,7 @@ mod tests { .await; let captured = Arc::new(AsyncMutex::new(None)); let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder() .uri(server.url()) .warehouse("client-wh".to_string()) @@ -2710,6 +2697,7 @@ mod tests { .await; let captured = Arc::new(AsyncMutex::new(None)); let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder() .uri(server.url()) .warehouse("client-wh".to_string()) @@ -2788,6 +2776,7 @@ mod tests { let dropped = Arc::new(AtomicBool::new(false)); let catalog = RestCatalog::new( + SessionContext::empty(), RestCatalogConfig::builder().uri(server.url()).build(), Some(Arc::new(GuardManager(dropped.clone()))), Some(Arc::new(LocalFsStorageFactory)), @@ -2971,16 +2960,12 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); - let namespaces = catalog.list_namespaces(None).await.unwrap(); + let namespaces = catalog + .list_namespaces(&SessionContext::empty(), None) + .await + .unwrap(); let expected_ns = vec![ NamespaceIdent::from_vec(vec!["ns1".to_string(), "ns11".to_string()]).unwrap(), @@ -3074,16 +3059,12 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); - let namespaces = catalog.list_namespaces(None).await.unwrap(); + let namespaces = catalog + .list_namespaces(&SessionContext::empty(), None) + .await + .unwrap(); let expected_ns = vec![ NamespaceIdent::from_vec(vec!["ns1".to_string(), "ns11".to_string()]).unwrap(), @@ -3131,17 +3112,11 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); let namespaces = catalog .create_namespace( + &SessionContext::empty(), &NamespaceIdent::from_vec(vec!["ns1".to_string(), "ns11".to_string()]).unwrap(), HashMap::from([("key1".to_string(), "value1".to_string())]), ) @@ -3178,17 +3153,13 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); let namespaces = catalog - .get_namespace(&NamespaceIdent::new("ns1".to_string())) + .get_namespace( + &SessionContext::empty(), + &NamespaceIdent::new("ns1".to_string()), + ) .await .unwrap(); @@ -3215,18 +3186,14 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); assert!( catalog - .namespace_exists(&NamespaceIdent::new("ns1".to_string())) + .namespace_exists( + &SessionContext::empty(), + &NamespaceIdent::new("ns1".to_string()) + ) .await .unwrap() ); @@ -3286,17 +3253,13 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); catalog - .drop_namespace(&NamespaceIdent::new("ns1".to_string())) + .drop_namespace( + &SessionContext::empty(), + &NamespaceIdent::new("ns1".to_string()), + ) .await .unwrap(); @@ -3330,17 +3293,13 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); let tables = catalog - .list_tables(&NamespaceIdent::new("ns1".to_string())) + .list_tables( + &SessionContext::empty(), + &NamespaceIdent::new("ns1".to_string()), + ) .await .unwrap(); @@ -3402,17 +3361,13 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); let tables = catalog - .list_tables(&NamespaceIdent::new("ns1".to_string())) + .list_tables( + &SessionContext::empty(), + &NamespaceIdent::new("ns1".to_string()), + ) .await .unwrap(); @@ -3537,17 +3492,13 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); let tables = catalog - .list_tables(&NamespaceIdent::new("ns1".to_string())) + .list_tables( + &SessionContext::empty(), + &NamespaceIdent::new("ns1".to_string()), + ) .await .unwrap(); @@ -3585,20 +3536,13 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); catalog - .drop_table(&TableIdent::new( - NamespaceIdent::new("ns1".to_string()), - "table1".to_string(), - )) + .drop_table( + &SessionContext::empty(), + &TableIdent::new(NamespaceIdent::new("ns1".to_string()), "table1".to_string()), + ) .await .unwrap(); @@ -3618,21 +3562,14 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); assert!( catalog - .table_exists(&TableIdent::new( - NamespaceIdent::new("ns1".to_string()), - "table1".to_string(), - )) + .table_exists( + &SessionContext::empty(), + &TableIdent::new(NamespaceIdent::new("ns1".to_string()), "table1".to_string(),), + ) .await .unwrap() ); @@ -3694,17 +3631,11 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); catalog .rename_table( + &SessionContext::empty(), &TableIdent::new(NamespaceIdent::new("ns1".to_string()), "table1".to_string()), &TableIdent::new(NamespaceIdent::new("ns1".to_string()), "table2".to_string()), ) @@ -3732,20 +3663,13 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); let table = catalog - .load_table(&TableIdent::new( - NamespaceIdent::new("ns1".to_string()), - "test1".to_string(), - )) + .load_table( + &SessionContext::empty(), + &TableIdent::new(NamespaceIdent::new("ns1".to_string()), "test1".to_string()), + ) .await .unwrap(); @@ -3853,20 +3777,13 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); let table = catalog - .load_table(&TableIdent::new( - NamespaceIdent::new("ns1".to_string()), - "test1".to_string(), - )) + .load_table( + &SessionContext::empty(), + &TableIdent::new(NamespaceIdent::new("ns1".to_string()), "test1".to_string()), + ) .await; assert!(table.is_err()); @@ -3893,14 +3810,7 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); let table_creation = TableCreation::builder() .name("test1".to_string()) @@ -3947,7 +3857,11 @@ mod tests { .build(); let table = catalog - .create_table(&NamespaceIdent::from_strs(["ns1"]).unwrap(), table_creation) + .create_table( + &SessionContext::empty(), + &NamespaceIdent::from_strs(["ns1"]).unwrap(), + table_creation, + ) .await .unwrap(); @@ -4046,14 +3960,7 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); let table_creation = TableCreation::builder() .name("test1".to_string()) @@ -4075,7 +3982,11 @@ mod tests { .build(); let table_result = catalog - .create_table(&NamespaceIdent::from_strs(["ns1"]).unwrap(), table_creation) + .create_table( + &SessionContext::empty(), + &NamespaceIdent::from_strs(["ns1"]).unwrap(), + table_creation, + ) .await; assert!(table_result.is_err()); @@ -4336,14 +4247,7 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); let table_ident = TableIdent::new(NamespaceIdent::new("ns1".to_string()), "test1".to_string()); let metadata_location = String::from( @@ -4351,7 +4255,7 @@ mod tests { ); let table = catalog - .register_table(&table_ident, metadata_location) + .register_table(&SessionContext::empty(), &table_ident, metadata_location) .await .unwrap(); @@ -4391,14 +4295,7 @@ mod tests { .create_async() .await; - let catalog = RestCatalog::new( - SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), - None, - Some(Arc::new(LocalFsStorageFactory)), - Runtime::current(), - None, - ); + let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); let table_ident = TableIdent::new(NamespaceIdent::new("ns1".to_string()), "test1".to_string()); @@ -4406,7 +4303,7 @@ mod tests { "s3://warehouse/database/table/metadata/00001-5f2f8166-244c-4eae-ac36-384ecdec81fc.gz.metadata.json", ); let table = catalog - .register_table(&table_ident, metadata_location) + .register_table(&SessionContext::empty(), &table_ident, metadata_location) .await; assert!(table.is_err()); @@ -4543,4 +4440,38 @@ mod tests { let catalog = result.unwrap(); assert!(uuid::Uuid::parse_str(catalog.session_context.session_id()).is_ok()); } + + /// Smoke test: a [`Catalog`] trait method delegates through the façade to `RestSessionCatalog`. + #[tokio::test] + async fn test_rest_catalog_delegates() { + let mut server = Server::new_async().await; + let config_mock = create_config_mock(&mut server).await; + let list_ns_mock = server + .mock("GET", "/v1/namespaces") + .with_body( + r#"{ + "namespaces": [["ns1"]] + }"#, + ) + .create_async() + .await; + + let catalog = RestCatalog::new( + SessionContext::empty(), + RestCatalogConfig::builder().uri(server.url()).build(), + None, + Some(Arc::new(LocalFsStorageFactory)), + Runtime::current(), + None, + ); + + let namespaces = catalog.list_namespaces(None).await.unwrap(); + + assert_eq!(namespaces, vec![ + NamespaceIdent::from_vec(vec!["ns1".to_string()]).unwrap() + ]); + + config_mock.assert_async().await; + list_ns_mock.assert_async().await; + } } From 908c82058a8459ba5efcc9ebc23da8c205f9eb34 Mon Sep 17 00:00:00 2001 From: Jannik Steinmann Date: Fri, 14 Aug 2026 09:04:10 +0200 Subject: [PATCH 4/6] Doc + comment updates Add RestSessionCatalog example Revised comments --- crates/catalog/rest/src/auth/mod.rs | 6 +- crates/catalog/rest/src/catalog.rs | 94 ++++++++++--------- crates/catalog/rest/src/client.rs | 2 +- crates/catalog/rest/src/endpoint.rs | 5 +- crates/catalog/rest/src/lib.rs | 38 +++++++- crates/examples/Cargo.toml | 4 + .../src/rest_session_catalog_namespace.rs | 94 +++++++++++++++++++ website/src/api.md | 36 ++++++- 8 files changed, 227 insertions(+), 52 deletions(-) create mode 100644 crates/examples/src/rest_session_catalog_namespace.rs diff --git a/crates/catalog/rest/src/auth/mod.rs b/crates/catalog/rest/src/auth/mod.rs index 8f7a3f425a..e8bda50b91 100644 --- a/crates/catalog/rest/src/auth/mod.rs +++ b/crates/catalog/rest/src/auth/mod.rs @@ -39,8 +39,10 @@ pub const AUTH_TYPE_OAUTH2: &str = "oauth2"; /// Creates the [`AuthSession`]s used to authenticate REST catalog requests. /// /// A manager is created once per catalog, either from the `rest.auth.type` -/// property or injected through `RestCatalogBuilder::with_auth_manager`. It -/// builds the sessions the catalog then keeps. +/// property or injected through +/// [`RestCatalogBuilder::with_auth_manager`](crate::RestCatalogBuilder::with_auth_manager) or +/// [`RestSessionCatalogBuilder::with_auth_manager`](crate::RestSessionCatalogBuilder::with_auth_manager). +/// It builds the sessions the catalog then keeps. /// /// Both methods are handed the catalog's [`HttpClient`], which an /// implementation may reuse for its own requests (e.g. a token exchange) so diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs index fd2362e5a5..3c091c7d15 100644 --- a/crates/catalog/rest/src/catalog.rs +++ b/crates/catalog/rest/src/catalog.rs @@ -67,7 +67,11 @@ const ICEBERG_REST_SPEC_VERSION: &str = "0.14.1"; const CARGO_PKG_VERSION: &str = env!("CARGO_PKG_VERSION"); const PATH_V1: &str = "v1"; -/// Builder for [`RestCatalog`]. +/// Builder for [`RestCatalog`], the [`Catalog`]-compatible façade over a +/// [`RestSessionCatalog`]. +/// +/// The resulting catalog binds one [`SessionContext`] to every operation. Use +/// [`RestSessionCatalogBuilder`] when the caller supplies a context per operation. #[derive(Debug, Default)] pub struct RestCatalogBuilder { session_context: Option, @@ -114,8 +118,10 @@ impl RestCatalogBuilder { self } - /// Configures the session context that will be used with this catalog. - /// Overwrites the default empty context from SessionContext::empty(). + /// Binds the session context forwarded with every catalog operation. + /// + /// If this is not called, [`load`](CatalogBuilder::load) creates a fresh + /// [`SessionContext::empty`] context. pub fn with_session_context(mut self, context: SessionContext) -> Self { self.session_context = Some(context); self @@ -164,8 +170,8 @@ fn is_sensitive_prop(key: &str) -> bool { || key.starts_with("header.") } -/// Redacts secret property values: this config is printed by -/// [`RestCatalog`]'s derived `Debug`. +/// Redacts secret property values: this config is printed by the derived +/// [`Debug`] implementations of [`RestSessionCatalog`] and [`RestCatalog`]. impl Debug for RestCatalogConfig { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { let props: HashMap<&str, &str> = self @@ -412,13 +418,15 @@ struct RestClient { http_client: HttpClient, /// Runtime config is fetched from rest server and stored here. /// - /// It's could be different from the user config. + /// It could be different from the user config. config: RestCatalogConfig, - /// Capabilities the server advertises (see [`RestCatalog::supports_endpoint`]). + /// Capabilities the server advertises (see [`RestSessionCatalog::supports_endpoint`]). endpoints: HashSet, } impl RestClient { + /// Initializes the runtime config, advertised endpoints, and authentication + /// sessions shared by one REST catalog instance. async fn init( user_config: &RestCatalogConfig, auth_manager: Arc, @@ -476,7 +484,11 @@ impl RestClient { } } -/// Rest catalog implementation. +/// A [`Catalog`]-compatible façade over [`RestSessionCatalog`]. +/// +/// Every operation is forwarded with the single [`SessionContext`] selected by +/// [`RestCatalogBuilder`]. Use [`RestSessionCatalog`] when the caller needs to +/// provide a context per operation. #[derive(Debug)] pub struct RestCatalog { session_context: SessionContext, @@ -518,8 +530,9 @@ impl RestCatalog { } } -/// All requests and expected responses are derived from the REST catalog API spec: -/// https://github.com/apache/iceberg/blob/main/open-api/rest-catalog-open-api.yaml +/// Every operation forwards to its [`RestSessionCatalog`] equivalent with the +/// bound [`SessionContext`]; see that implementation for the REST-specific +/// behavior. #[async_trait] impl Catalog for RestCatalog { async fn list_namespaces( @@ -573,12 +586,6 @@ impl Catalog for RestCatalog { .await } - /// Create a new table inside the namespace. - /// - /// In the resulting table, if there are any config properties that - /// are present in both the response from the REST server and the - /// config provided when creating this `RestCatalog` instance then - /// the value provided locally to the `RestCatalog` will take precedence. async fn create_table( &self, namespace: &NamespaceIdent, @@ -589,34 +596,24 @@ impl Catalog for RestCatalog { .await } - /// Load table from the catalog. - /// - /// If there are any config properties that are present in both the response from the REST - /// server and the config provided when creating this `RestCatalog` instance, then the value - /// provided locally to the `RestCatalog` will take precedence. async fn load_table(&self, table_ident: &TableIdent) -> Result
{ self.inner .load_table(&self.session_context, table_ident) .await } - /// Drop a table from the catalog. async fn drop_table(&self, table: &TableIdent) -> Result<()> { self.inner.drop_table(&self.session_context, table).await } - /// Drop a table from the catalog and purge its data by sending - /// `purgeRequested=true` to the REST server. async fn purge_table(&self, table: &TableIdent) -> Result<()> { self.inner.purge_table(&self.session_context, table).await } - /// Check if a table exists in the catalog. async fn table_exists(&self, table: &TableIdent) -> Result { self.inner.table_exists(&self.session_context, table).await } - /// Rename a table in the catalog. async fn rename_table(&self, src: &TableIdent, dest: &TableIdent) -> Result<()> { self.inner .rename_table(&self.session_context, src, dest) @@ -638,13 +635,16 @@ impl Catalog for RestCatalog { } } -/// Rest catalog implementation. +/// REST catalog implementation of [`SessionCatalog`]. +/// +/// Each operation accepts a [`SessionContext`]. REST configuration, authentication sessions, +/// and the HTTP client are initialized lazily once per catalog and shared across all operations. #[derive(Debug)] pub struct RestSessionCatalog { /// Injected through [`RestSessionCatalogBuilder::with_auth_manager`]; otherwise /// one is resolved from `rest.auth.type` when the client is built. auth_manager: Option>, - /// User config is stored as-is and never be changed. + /// User config is stored as-is and never changed. /// /// It could be different from the config fetched from the server and used at runtime. user_config: RestCatalogConfig, @@ -774,8 +774,8 @@ impl RestSessionCatalog { ErrorKind::DataInvalid, format!( "unknown '{REST_CATALOG_PROP_AUTH_TYPE}': {other}; use \ - `RestCatalogBuilder::with_auth_manager` to inject a \ - custom auth manager" + `RestSessionCatalogBuilder::with_auth_manager` or \ + `RestCatalogBuilder::with_auth_manager` to inject a custom auth manager" ), )), } @@ -873,7 +873,7 @@ impl RestSessionCatalog { .ok_or_else(|| { Error::new( ErrorKind::Unexpected, - "StorageFactory must be provided for RestCatalog. Use `with_storage_factory` to configure it.", + "StorageFactory must be provided for REST catalog table operations. Use `with_storage_factory` to configure it.", ) })?; @@ -883,6 +883,8 @@ impl RestSessionCatalog { } } +/// All requests and expected responses are derived from the REST catalog API spec: +/// #[async_trait] impl SessionCatalog for RestSessionCatalog { async fn list_namespaces( @@ -1129,8 +1131,8 @@ impl SessionCatalog for RestSessionCatalog { /// /// In the resulting table, if there are any config properties that /// are present in both the response from the REST server and the - /// config provided when creating this `RestCatalog` instance then - /// the value provided locally to the `RestCatalog` will take precedence. + /// config provided when creating this `RestSessionCatalog` instance, then + /// the value provided locally to the `RestSessionCatalog` will take precedence. async fn create_table( &self, _context: &SessionContext, @@ -1217,8 +1219,8 @@ impl SessionCatalog for RestSessionCatalog { /// Load table from the catalog. /// /// If there are any config properties that are present in both the response from the REST - /// server and the config provided when creating this `RestCatalog` instance, then the value - /// provided locally to the `RestCatalog` will take precedence. + /// server and the config provided when creating this `RestSessionCatalog` instance, then the + /// value provided locally to the `RestSessionCatalog` will take precedence. async fn load_table( &self, _context: &SessionContext, @@ -1499,7 +1501,10 @@ impl SessionCatalog for RestSessionCatalog { } } -/// Builder for [`RestSessionCatalog`]. +/// Builder for an unbound [`RestSessionCatalog`]. +/// +/// Unlike [`RestCatalogBuilder`], the resulting catalog accepts a +/// [`SessionContext`] with each [`SessionCatalog`] operation. #[derive(Debug)] pub struct RestSessionCatalogBuilder { config: RestCatalogConfig, @@ -1553,12 +1558,12 @@ impl RestSessionCatalogBuilder { /// # Example /// /// ```rust,ignore - /// use iceberg::CatalogBuilder; /// use iceberg::io::StorageFactory; + /// use iceberg_catalog_rest::RestSessionCatalogBuilder; /// use iceberg_storage_opendal::OpenDalStorageFactory; /// use std::sync::Arc; /// - /// let catalog = MyCatalogBuilder::default() + /// let catalog = RestSessionCatalogBuilder::default() /// .with_storage_factory(Arc::new(OpenDalStorageFactory::S3 { /// customized_credential_load: None, /// })) @@ -1574,18 +1579,18 @@ impl RestSessionCatalogBuilder { /// /// When provided, the catalog calls the factory once during /// [`load`](Self::load) with the catalog properties to create a shared - /// [`KeyManagementClient`](crate::encryption::KeyManagementClient). + /// [`KeyManagementClient`]. /// That client is then passed to each table's `TableBuilder` so tables /// with `encryption.key-id` set can construct an `EncryptionManager`. /// /// # Example /// /// ```rust,ignore - /// use iceberg::CatalogBuilder; /// use iceberg::encryption::kms::KmsClientFactory; + /// use iceberg_catalog_rest::RestSessionCatalogBuilder; /// use std::sync::Arc; /// - /// let catalog = MyCatalogBuilder::default() + /// let catalog = RestSessionCatalogBuilder::default() /// .with_kms_client_factory(Arc::new(MyKmsClientFactory)) /// .load("my_catalog", props) /// .await?; @@ -1608,7 +1613,10 @@ impl RestSessionCatalogBuilder { self } - /// Create a new catalog instance. + /// Creates a new session catalog instance. + /// + /// The server configuration handshake, endpoint negotiation, and + /// authentication sessions are initialized lazily on the first operation. pub fn load( mut self, name: impl Into, @@ -4435,7 +4443,7 @@ mod tests { assert!(result.is_ok()); - // Without `with_session`, the catalog falls back to `SessionContext::empty()`, + // Without `with_session_context`, the catalog falls back to `SessionContext::empty()`, // which assigns a fresh v4 UUID. let catalog = result.unwrap(); assert!(uuid::Uuid::parse_str(catalog.session_context.session_id()).is_ok()); diff --git a/crates/catalog/rest/src/client.rs b/crates/catalog/rest/src/client.rs index 4ed3d6d023..add697444c 100644 --- a/crates/catalog/rest/src/client.rs +++ b/crates/catalog/rest/src/client.rs @@ -86,7 +86,7 @@ impl HttpClient { /// not merged in. /// /// Like every request, it carries this client's session; call - /// [`Self::without_session`] first to send it unauthenticated. + /// [`Self::without_auth_session`] first to send it unauthenticated. /// /// [`AuthManager`]: crate::auth::AuthManager pub async fn post_form( diff --git a/crates/catalog/rest/src/endpoint.rs b/crates/catalog/rest/src/endpoint.rs index 1f4e7f354a..3d82071f21 100644 --- a/crates/catalog/rest/src/endpoint.rs +++ b/crates/catalog/rest/src/endpoint.rs @@ -23,9 +23,8 @@ //! Each entry is a `"{method} {path}"` string, for example //! `"POST /v1/{prefix}/namespaces/{namespace}/tables"`; parse one through //! [`Endpoint`]'s [`FromStr`] implementation. -//! -//! Use [`RestCatalog::supports_endpoint`](crate::RestCatalog::supports_endpoint) -//! to check whether the connected server advertised a given [`Endpoint`]. +//! The REST catalog uses the advertised set internally to select supported +//! operations and compatibility fallbacks. use std::collections::HashSet; use std::fmt::{self, Display, Formatter}; diff --git a/crates/catalog/rest/src/lib.rs b/crates/catalog/rest/src/lib.rs index 5670ef3c9c..d9ad38dfab 100644 --- a/crates/catalog/rest/src/lib.rs +++ b/crates/catalog/rest/src/lib.rs @@ -17,8 +17,14 @@ //! Iceberg REST API implementation. //! -//! To build a rest catalog with configurations -//! # Example +//! The crate provides two REST catalog APIs: +//! +//! - [`RestCatalog`] implements [`iceberg::Catalog`] by binding one +//! [`iceberg::SessionContext`] to every operation. +//! - [`RestSessionCatalog`] implements [`iceberg::SessionCatalog`] and accepts a +//! session context with each operation. +//! +//! # Catalog compatibility API //! //! ```rust, no_run //! use std::collections::HashMap; @@ -48,6 +54,34 @@ //! .unwrap(); //! } //! ``` +//! +//! # Session catalog API +//! +//! ```rust, no_run +//! use std::collections::HashMap; +//! +//! use iceberg::{SessionCatalog, SessionContext}; +//! use iceberg_catalog_rest::{REST_CATALOG_PROP_URI, RestSessionCatalogBuilder}; +//! +//! #[tokio::main] +//! async fn main() { +//! let catalog = RestSessionCatalogBuilder::default() +//! .load( +//! "rest", +//! HashMap::from([( +//! REST_CATALOG_PROP_URI.to_string(), +//! "http://localhost:8181".to_string(), +//! )]), +//! ) +//! .await +//! .unwrap(); +//! let context = SessionContext::builder() +//! .identity("user123".to_string()) +//! .build(); +//! +//! let namespaces = catalog.list_namespaces(&context, None).await.unwrap(); +//! } +//! ``` #![deny(missing_docs)] diff --git a/crates/examples/Cargo.toml b/crates/examples/Cargo.toml index 0492fed343..27e7c9969d 100644 --- a/crates/examples/Cargo.toml +++ b/crates/examples/Cargo.toml @@ -40,6 +40,10 @@ path = "src/rest_catalog_namespace.rs" name = "rest-catalog-table" path = "src/rest_catalog_table.rs" +[[example]] +name = "rest-session-catalog-namespace" +path = "src/rest_session_catalog_namespace.rs" + [[example]] name = "oss-backend" path = "src/oss_backend.rs" diff --git a/crates/examples/src/rest_session_catalog_namespace.rs b/crates/examples/src/rest_session_catalog_namespace.rs new file mode 100644 index 0000000000..60f5d33780 --- /dev/null +++ b/crates/examples/src/rest_session_catalog_namespace.rs @@ -0,0 +1,94 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::HashMap; + +use iceberg::{NamespaceIdent, SessionCatalog, SessionContext}; +use iceberg_catalog_rest::{REST_CATALOG_PROP_URI, RestSessionCatalogBuilder}; + +static REST_URI: &str = "http://localhost:8181"; + +/// This is a simple example that demonstrates how to use [`RestSessionCatalog`] to create +/// namespaces. +/// +/// Unlike [`RestCatalog`](iceberg_catalog_rest::RestCatalog), every operation +/// takes the [`SessionContext`] it should run under, so a single catalog instance +/// can serve requests on behalf of different sessions. +/// +/// A running instance of the iceberg-rest catalog on port 8181 is required. +/// You can find how to run the iceberg-rest catalog with `docker compose` in +/// the official [quickstart documentation](https://iceberg.apache.org/spark-quickstart/). +#[tokio::main] +async fn main() { + // ANCHOR: create_catalog + // Create the REST iceberg session catalog. + let catalog = RestSessionCatalogBuilder::default() + .load( + "rest", + HashMap::from([(REST_CATALOG_PROP_URI.to_string(), REST_URI.to_string())]), + ) + .await + .unwrap(); + + // Build the session context passed to each operation. + let context = SessionContext::builder() + .identity("user123".to_string()) + .build(); + // ANCHOR_END: create_catalog + + // ANCHOR: list_all_namespace + // List all namespaces already in the catalog. + let existing_namespaces = catalog.list_namespaces(&context, None).await.unwrap(); + println!("Namespaces alreading in the existing catalog: {existing_namespaces:?}"); + // ANCHOR_END: list_all_namespace + + // ANCHOR: create_namespace + // Create a new namespace identifier. + let namespace_ident = + NamespaceIdent::from_vec(vec!["ns1".to_string(), "ns11".to_string()]).unwrap(); + + // Drop the namespace if it already exists. + if catalog + .namespace_exists(&context, &namespace_ident) + .await + .unwrap() + { + println!("Namespace already exists, dropping now.",); + catalog + .drop_namespace(&context, &namespace_ident) + .await + .unwrap(); + } + + // Create the new namespace in the catalog. + let _created_namespace = catalog + .create_namespace( + &context, + &namespace_ident, + HashMap::from([("key1".to_string(), "value1".to_string())]), + ) + .await + .unwrap(); + println!("Namespace {namespace_ident:?} created!"); + + let loaded_namespace = catalog + .get_namespace(&context, &namespace_ident) + .await + .unwrap(); + println!("Namespace loaded!\n\nNamespace: {loaded_namespace:#?}",); + // ANCHOR_END: create_namespace +} diff --git a/website/src/api.md b/website/src/api.md index 27a68cbdcf..acf0270413 100644 --- a/website/src/api.md +++ b/website/src/api.md @@ -24,6 +24,9 @@ * Create and list namespaces. * Create, load, and drop tables +`SessionCatalog` offers the same operations, but takes a `SessionContext` with each call instead of +binding one for the lifetime of the catalog. It is currently implemented by the REST catalog only. + There is support for the following catalogs: | Catalog | Description | @@ -37,7 +40,16 @@ There is support for the following catalogs: Not all catalog implementations are complete. -## `RestCatalog` +## REST catalog APIs + +The REST catalog provides two interfaces: + +* `RestCatalog` implements `Catalog` and binds one `SessionContext` to every operation. +* `RestSessionCatalog` implements `SessionCatalog` and accepts a context with each operation. + +REST configuration, authentication, and the HTTP client are shared by the catalog instance. + +### `RestCatalog` Here is an example of how to create a `RestCatalog`: @@ -52,10 +64,32 @@ You can run following code to list all root namespaces: ``` Then you can run following code to create namespace: + ```rust,no_run,noplayground {{#rustdoc_include ../../crates/examples/src/rest_catalog_namespace.rs:create_namespace}} ``` +### `RestSessionCatalog` + +Use `RestSessionCatalog` when the caller supplies a session context for each operation. Build the +catalog and the context it should run under: + +```rust,no_run,noplayground +{{#rustdoc_include ../../crates/examples/src/rest_session_catalog_namespace.rs:create_catalog}} +``` + +Every operation then takes that context. You can run following code to list all root namespaces: + +```rust,no_run,noplayground +{{#rustdoc_include ../../crates/examples/src/rest_session_catalog_namespace.rs:list_all_namespace}} +``` + +Then you can run following code to create namespace: + +```rust,no_run,noplayground +{{#rustdoc_include ../../crates/examples/src/rest_session_catalog_namespace.rs:create_namespace}} +``` + # Table After creating `Catalog`, we can manipulate tables through `Catalog`. From 0ea8a83b9a676238ce1a75f91353ab7956d83158 Mon Sep 17 00:00:00 2001 From: Jannik Steinmann Date: Fri, 14 Aug 2026 10:20:15 +0200 Subject: [PATCH 5/6] Move helper functions into RestClient --- crates/catalog/rest/src/catalog.rs | 98 +++++++++++++++--------------- 1 file changed, 49 insertions(+), 49 deletions(-) diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs index 3c091c7d15..e4fd204ba5 100644 --- a/crates/catalog/rest/src/catalog.rs +++ b/crates/catalog/rest/src/catalog.rs @@ -439,10 +439,10 @@ impl RestClient { let init_session = auth_manager .init_session( &http_client.without_auth_session(), - &RestSessionCatalog::auth_props(user_config), + &Self::auth_props(user_config), ) .await?; - RestSessionCatalog::load_config( + Self::load_config( &http_client.with_auth_session(Arc::from(init_session)), user_config, ) @@ -461,7 +461,7 @@ impl RestClient { let session = auth_manager .catalog_session( &http_client.without_auth_session(), - &RestSessionCatalog::auth_props(&config), + &Self::auth_props(&config), ) .await?; @@ -482,6 +482,52 @@ impl RestClient { async fn query_catalog(&self, request: HttpRequest) -> Result { self.http_client.query_catalog(request).await } + + /// The properties handed to the [`AuthManager`], with the catalog `uri` + /// and `warehouse` made explicit. + fn auth_props(config: &RestCatalogConfig) -> HashMap { + // `oauth2-server-uri` stays absent unless explicitly configured, so an + // injected manager keeps its own endpoint. The resolved `uri` and + // `warehouse` ARE passed: the builder moved them off the props, and + // the built-in manager recomputes its token endpoint from the URI. + let mut props = config.props.clone(); + props.insert(REST_CATALOG_PROP_URI.to_string(), config.uri.clone()); + if let Some(warehouse) = &config.warehouse { + // A fallback only: after the handshake the merged props hold + // the resolved warehouse, server override included. + props + .entry(REST_CATALOG_PROP_WAREHOUSE.to_string()) + .or_insert_with(|| warehouse.clone()); + } + props + } + + /// Loads the runtime config from the server using `user_config`. + /// + /// It's required for a REST catalog to update its config after creation. + async fn load_config( + http_client: &HttpClient, + user_config: &RestCatalogConfig, + ) -> Result { + let mut request_builder = http_client.request(Method::GET, user_config.config_endpoint()); + + if let Some(warehouse_location) = &user_config.warehouse { + request_builder = request_builder.query(&[("warehouse", warehouse_location)]); + } + + let request = HttpRequest::build(request_builder)?; + + let http_response = http_client.query_catalog(request).await?; + + match http_response.status() { + StatusCode::OK => deserialize_catalog_response(http_response).await, + _ => Err(deserialize_unexpected_catalog_error( + http_response, + http_client.disable_header_redaction(), + ) + .await), + } + } } /// A [`Catalog`]-compatible façade over [`RestSessionCatalog`]. @@ -709,25 +755,6 @@ impl RestSessionCatalog { } } - /// The properties handed to the [`AuthManager`], with the catalog `uri` - /// and `warehouse` made explicit. - fn auth_props(config: &RestCatalogConfig) -> HashMap { - // `oauth2-server-uri` stays absent unless explicitly configured, so an - // injected manager keeps its own endpoint. The resolved `uri` and - // `warehouse` ARE passed: the builder moved them off the props, and - // the built-in manager recomputes its token endpoint from the URI. - let mut props = config.props.clone(); - props.insert(REST_CATALOG_PROP_URI.to_string(), config.uri.clone()); - if let Some(warehouse) = &config.warehouse { - // A fallback only: after the handshake the merged props hold - // the resolved warehouse, server override included. - props - .entry(REST_CATALOG_PROP_WAREHOUSE.to_string()) - .or_insert_with(|| warehouse.clone()); - } - props - } - /// The configured auth scheme: explicit `rest.auth.type` (matched /// case-insensitively) when set; otherwise `oauth2` when a `token`, /// `credential` or `oauth2-server-uri` is configured (preserving @@ -814,33 +841,6 @@ impl RestSessionCatalog { } } - /// Load the runtime config from the server by `user_config`. - /// - /// It's required for a REST catalog to update its config after creation. - async fn load_config( - http_client: &HttpClient, - user_config: &RestCatalogConfig, - ) -> Result { - let mut request_builder = http_client.request(Method::GET, user_config.config_endpoint()); - - if let Some(warehouse_location) = &user_config.warehouse { - request_builder = request_builder.query(&[("warehouse", warehouse_location)]); - } - - let request = HttpRequest::build(request_builder)?; - - let http_response = http_client.query_catalog(request).await?; - - match http_response.status() { - StatusCode::OK => deserialize_catalog_response(http_response).await, - _ => Err(deserialize_unexpected_catalog_error( - http_response, - http_client.disable_header_redaction(), - ) - .await), - } - } - async fn load_file_io( &self, metadata_location: Option<&str>, From 7c398434d7fe993e7ba5a63482317b2fed13f117 Mon Sep 17 00:00:00 2001 From: Jannik Steinmann Date: Sun, 26 Jul 2026 20:20:38 +0200 Subject: [PATCH 6/6] SessionCatalog conversion to Catalog --- crates/iceberg/src/catalog/session.rs | 117 +++++++++++++++++++++++++- 1 file changed, 116 insertions(+), 1 deletion(-) diff --git a/crates/iceberg/src/catalog/session.rs b/crates/iceberg/src/catalog/session.rs index 1790f4978a..f5abefef1b 100644 --- a/crates/iceberg/src/catalog/session.rs +++ b/crates/iceberg/src/catalog/session.rs @@ -19,6 +19,7 @@ use std::collections::HashMap; use std::fmt::Debug; +use std::sync::Arc; use async_trait::async_trait; #[cfg(test)] @@ -28,7 +29,7 @@ use uuid::Uuid; use zeroize::Zeroizing; use crate::table::Table; -use crate::{Namespace, NamespaceIdent, Result, TableCommit, TableCreation, TableIdent}; +use crate::{Catalog, Namespace, NamespaceIdent, Result, TableCommit, TableCreation, TableIdent}; /// Context for a session. /// @@ -233,6 +234,120 @@ pub trait SessionCatalog: Debug + Send + Sync { async fn update_table(&self, context: &SessionContext, commit: TableCommit) -> Result
; } +impl dyn SessionCatalog { + /// Bind this catalog to a session, exposing the ordinary Catalog API. + /// + /// # Example + /// ``` + /// # fn into_catalog(session_catalog: Arc, id: String) { + /// let session = SessionContext::builder().session_id(id).build(); + /// + /// // Use the plain catalog API for the duration of this session. + /// let catalog = session_catalog.into_catalog(session); + /// # let _ = catalog; + /// # } + /// ``` + pub fn into_catalog(self: Arc, session: SessionContext) -> Arc { + Arc::new(SessionBoundCatalog { + inner: self, + session, + }) + } +} + +/// Allows any [`SessionCatalog`] to implement the [`Catalog`] trait. +#[derive(Debug)] +struct SessionBoundCatalog { + inner: Arc, + session: SessionContext, +} + +#[async_trait] +impl Catalog for SessionBoundCatalog { + async fn list_namespaces( + &self, + parent: Option<&NamespaceIdent>, + ) -> Result> { + self.inner.list_namespaces(&self.session, parent).await + } + + async fn create_namespace( + &self, + namespace: &NamespaceIdent, + properties: HashMap, + ) -> Result { + self.inner + .create_namespace(&self.session, namespace, properties) + .await + } + + async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result { + self.inner.get_namespace(&self.session, namespace).await + } + + async fn namespace_exists(&self, namespace: &NamespaceIdent) -> Result { + self.inner.namespace_exists(&self.session, namespace).await + } + + async fn update_namespace( + &self, + namespace: &NamespaceIdent, + properties: HashMap, + ) -> Result<()> { + self.inner + .update_namespace(&self.session, namespace, properties) + .await + } + + async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()> { + self.inner.drop_namespace(&self.session, namespace).await + } + + async fn list_tables(&self, namespace: &NamespaceIdent) -> Result> { + self.inner.list_tables(&self.session, namespace).await + } + + async fn create_table( + &self, + namespace: &NamespaceIdent, + creation: TableCreation, + ) -> Result
{ + self.inner + .create_table(&self.session, namespace, creation) + .await + } + + async fn load_table(&self, table: &TableIdent) -> Result
{ + self.inner.load_table(&self.session, table).await + } + + async fn drop_table(&self, table: &TableIdent) -> Result<()> { + self.inner.drop_table(&self.session, table).await + } + + async fn purge_table(&self, table: &TableIdent) -> Result<()> { + self.inner.purge_table(&self.session, table).await + } + + async fn table_exists(&self, table: &TableIdent) -> Result { + self.inner.table_exists(&self.session, table).await + } + + async fn rename_table(&self, src: &TableIdent, dest: &TableIdent) -> Result<()> { + self.inner.rename_table(&self.session, src, dest).await + } + + async fn register_table(&self, table: &TableIdent, metadata_location: String) -> Result
{ + self.inner + .register_table(&self.session, table, metadata_location) + .await + } + + async fn update_table(&self, commit: TableCommit) -> Result
{ + self.inner.update_table(&self.session, commit).await + } +} + #[cfg(test)] mod tests { use std::collections::HashMap;