laskoviymishka commented on code in PR #2999: URL: https://github.com/apache/iceberg-rust/pull/2999#discussion_r4008131055
########## 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:?}"); Review Comment: "alreading" → "already". ########## crates/iceberg/src/catalog/session.rs: ########## @@ -233,6 +234,120 @@ pub trait SessionCatalog: Debug + Send + Sync { async fn update_table(&self, context: &SessionContext, commit: TableCommit) -> Result<Table>; } +impl dyn SessionCatalog { + /// Bind this catalog to a session, exposing the ordinary Catalog API. + /// + /// # Example + /// ``` + /// # fn into_catalog(session_catalog: Arc<dyn SessionCatalog>, 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<Self>, session: SessionContext) -> Arc<dyn Catalog> { Review Comment: `into_` reads as "consumes the receiver," but the `Arc<Self>` here is refcounted — other clones stay usable, which is rather the point. `bind_session` would describe what this does more honestly. Related, and smaller: since it lives on `impl dyn SessionCatalog`, a caller holding a concrete `Arc<RestSessionCatalog>` has to erase to `Arc<dyn SessionCatalog>` first before calling it. A blanket method on the trait (`where Self: 'static`) or a free `bind_session` would avoid that. wdyt? ########## crates/iceberg/src/catalog/session.rs: ########## @@ -233,6 +234,120 @@ pub trait SessionCatalog: Debug + Send + Sync { async fn update_table(&self, context: &SessionContext, commit: TableCommit) -> Result<Table>; } +impl dyn SessionCatalog { + /// Bind this catalog to a session, exposing the ordinary Catalog API. + /// + /// # Example + /// ``` + /// # fn into_catalog(session_catalog: Arc<dyn SessionCatalog>, id: String) { Review Comment: This doctest is compiled (no `ignore`/`no_run`), and the hidden `# fn` line plus `SessionContext::builder()` reference `Arc`, `SessionCatalog`, and `SessionContext` without importing them, so `cargo test --doc` won't compile it — the `#` only hides the lines, it doesn't skip them. A couple of hidden `use` lines before the fn fixes it: ```rust /// # use std::sync::Arc; /// # use iceberg::{SessionCatalog, SessionContext}; ``` ########## crates/catalog/rest/src/catalog.rs: ########## @@ -1349,6 +1501,176 @@ impl Catalog for RestCatalog { } } +/// Builder for an unbound [`RestSessionCatalog`]. +/// +/// Unlike [`RestCatalogBuilder`], the resulting catalog accepts a +/// [`SessionContext`] with each [`SessionCatalog`] operation. +#[derive(Debug)] +pub struct RestSessionCatalogBuilder { + config: RestCatalogConfig, + auth_manager: Option<Arc<dyn AuthManager>>, + storage_factory: Option<Arc<dyn StorageFactory>>, + kms_client_factory: Option<Arc<dyn KmsClientFactory>>, + runtime: Option<Runtime>, +} + +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<dyn AuthManager>) -> 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::io::StorageFactory; + /// use iceberg_catalog_rest::RestSessionCatalogBuilder; + /// use iceberg_storage_opendal::OpenDalStorageFactory; + /// use std::sync::Arc; + /// + /// let catalog = RestSessionCatalogBuilder::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<dyn StorageFactory>) -> 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`]. + /// 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::encryption::kms::KmsClientFactory; + /// use iceberg_catalog_rest::RestSessionCatalogBuilder; + /// use std::sync::Arc; + /// + /// let catalog = RestSessionCatalogBuilder::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<dyn KmsClientFactory>, + ) -> 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 + } + + /// 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<String>, + props: HashMap<String, String>, + ) -> impl Future<Output = Result<RestSessionCatalog>> + 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() { Review Comment: `self.config.name` is set to `Some(name.into())` at the top of `load`, so by the time this `async move` block runs the `is_none()` branch can never fire — an empty name slips through. The `uri` check below it is the one doing real work. Not sure this survives the rebase (it may be pre-existing from #2920), but if this builder is still yours afterward, either drop the dead check or guard on empty with `name.as_deref().map_or(true, str::is_empty)`. ########## 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()) Review Comment: This builds a context with `.identity("user123")` and passes it to `list_namespaces`, but as noted on `into_catalog` the REST impl drops it — so the example demonstrates a distinction that doesn't exist yet, and this is the version that lands in the website docs. Same `.identity("user123")` in the `lib.rs` crate doc. I'd either use `SessionContext::empty()` here or add a one-line comment that identity isn't forwarded yet. ########## crates/iceberg/src/catalog/session.rs: ########## @@ -233,6 +234,120 @@ pub trait SessionCatalog: Debug + Send + Sync { async fn update_table(&self, context: &SessionContext, commit: TableCommit) -> Result<Table>; } +impl dyn SessionCatalog { + /// Bind this catalog to a session, exposing the ordinary Catalog API. + /// + /// # Example + /// ``` + /// # fn into_catalog(session_catalog: Arc<dyn SessionCatalog>, 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<Self>, session: SessionContext) -> Arc<dyn Catalog> { Review Comment: Binding a session here reads like it should change what the catalog does per session, but for the REST implementation today the session is dropped on the floor — every leaf op takes `_context` and never forwards `identity`, `credentials`, or `session_id` into a request. So `into_catalog(admin_session)` and `into_catalog(readonly_session)` produce catalogs that send byte-identical HTTP. That's a fine starting point, but the `credentials` field on `SessionContext` strongly implies per-op auth override, and nothing warns the caller it's a no-op — someone routing per-user tokens through this gets catalog-level auth silently. I'd either document the limitation right here on `into_catalog` (and on `RestSessionCatalog`), noting the REST impl doesn't forward the session yet and citing the follow-up (#2774?), or wire a minimal credential passthrough now. Either is fine, but I'd like it explicit before the follow-ups build on this entry point. wdyt? ########## crates/catalog/rest/src/catalog.rs: ########## @@ -482,15 +482,215 @@ impl RestClient { async fn query_catalog(&self, request: HttpRequest) -> Result<Response> { 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<String, String> { + // `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<CatalogConfig> { + 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), + } + } } -/// 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 { - /// Injected through [`RestCatalogBuilder::with_auth_manager`]; otherwise - /// one is resolved from `rest.auth.type` when the context is built. + session_context: SessionContext, + inner: Arc<RestSessionCatalog>, +} + +impl RestCatalog { + /// Creates a `RestCatalog` from a [`RestCatalogConfig`]. + #[cfg(test)] + fn new( + context: SessionContext, + config: RestCatalogConfig, + auth_manager: Option<Arc<dyn AuthManager>>, + storage_factory: Option<Arc<dyn StorageFactory>>, + runtime: Runtime, + kms_client: Option<Arc<dyn KeyManagementClient>>, + ) -> 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<RestSessionCatalog>) -> Self { + Self { + session_context: context, + inner, + } + } + + #[cfg(test)] + async fn client(&self) -> Result<&RestClient> { + self.inner.client().await + } +} + +/// 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 { Review Comment: `SessionBoundCatalog` and `RestCatalog` are now structurally the same thing — both hold an `Arc<impl SessionCatalog>` + a `SessionContext` and hand-forward all 15 `Catalog` methods with the bound session. That's ~30 near-identical bodies to keep in sync, and a third copy for the next `SessionCatalog` impl. Not blocking — the compiler catches drift — but `RestCatalog` could wrap a `SessionBoundCatalog` (keeping the typed `Arc<RestSessionCatalog>` alongside for the `#[cfg(test)]` `client()`/`supports_endpoint` helpers), or `SessionBoundCatalog` could go `pub(crate)` and generic over `C: SessionCatalog`. wdyt? ########## crates/iceberg/src/catalog/session.rs: ########## @@ -233,6 +234,120 @@ pub trait SessionCatalog: Debug + Send + Sync { async fn update_table(&self, context: &SessionContext, commit: TableCommit) -> Result<Table>; } +impl dyn SessionCatalog { + /// Bind this catalog to a session, exposing the ordinary Catalog API. + /// + /// # Example + /// ``` + /// # fn into_catalog(session_catalog: Arc<dyn SessionCatalog>, 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<Self>, session: SessionContext) -> Arc<dyn Catalog> { + Arc::new(SessionBoundCatalog { + inner: self, + session, + }) + } +} + +/// Allows any [`SessionCatalog`] to implement the [`Catalog`] trait. +#[derive(Debug)] +struct SessionBoundCatalog { Review Comment: The new `into_catalog` path has no tests — nothing checks that the bound session is actually the one forwarded. A refactor that swapped `&self.session` for `SessionContext::empty()` in every method would still typecheck and still pass, so the delegation is effectively unverified. `SessionCatalog` already gets `automock`, so a short test that builds a `MockSessionCatalog`, binds a context with a known `session_id`, calls one read and one mutating method through the returned `Catalog`, and asserts the forwarded `session_id` with `predicate::eq` would lock it down. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
