diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 610675c83d..0404f35597 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -3510,6 +3510,16 @@ impl serde_core::ser::Serialize for iceberg::ViewUpdate pub fn iceberg::ViewUpdate::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg::ViewUpdate pub fn iceberg::ViewUpdate::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg::Credential(_) +impl iceberg::Credential +pub fn iceberg::Credential::expose(&self) -> &str +pub fn iceberg::Credential::new(value: alloc::string::String) -> Self +impl core::clone::Clone for iceberg::Credential +pub fn iceberg::Credential::clone(&self) -> iceberg::Credential +impl core::convert::From for iceberg::Credential +pub fn iceberg::Credential::from(value: alloc::string::String) -> Self +impl core::fmt::Debug for iceberg::Credential +pub fn iceberg::Credential::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result pub struct iceberg::Error impl iceberg::Error pub fn iceberg::Error::backtrace(&self) -> &std::backtrace::Backtrace @@ -3658,6 +3668,19 @@ impl core::clone::Clone for iceberg::RuntimeHandle pub fn iceberg::RuntimeHandle::clone(&self) -> iceberg::RuntimeHandle impl core::fmt::Debug for iceberg::RuntimeHandle pub fn iceberg::RuntimeHandle::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +pub struct iceberg::SessionContext +impl iceberg::SessionContext +pub fn iceberg::SessionContext::credentials(&self) -> &std::collections::hash::map::HashMap +pub fn iceberg::SessionContext::empty() -> Self +pub fn iceberg::SessionContext::identity(&self) -> core::option::Option<&str> +pub fn iceberg::SessionContext::properties(&self) -> &std::collections::hash::map::HashMap +pub fn iceberg::SessionContext::session_id(&self) -> &str +impl core::clone::Clone for iceberg::SessionContext +pub fn iceberg::SessionContext::clone(&self) -> iceberg::SessionContext +impl core::fmt::Debug for iceberg::SessionContext +pub fn iceberg::SessionContext::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl iceberg::SessionContext +pub fn iceberg::SessionContext::builder() -> SessionContextBuilder<((), (), (), ())> pub struct iceberg::TableCommit impl iceberg::TableCommit pub fn iceberg::TableCommit::apply(self, table: iceberg::table::Table) -> iceberg::Result @@ -3765,5 +3788,26 @@ pub fn iceberg::memory::MemoryCatalogBuilder::load(self, name: impl core::conver pub fn iceberg::memory::MemoryCatalogBuilder::with_kms_client_factory(self, kms_client_factory: alloc::sync::Arc) -> Self pub fn iceberg::memory::MemoryCatalogBuilder::with_runtime(self, runtime: iceberg::Runtime) -> Self pub fn iceberg::memory::MemoryCatalogBuilder::with_storage_factory(self, storage_factory: alloc::sync::Arc) -> Self +pub trait iceberg::SessionCatalog: core::fmt::Debug + core::marker::Send + core::marker::Sync +pub fn iceberg::SessionCatalog::create_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, namespace: &'life2 iceberg::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::SessionCatalog::create_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, namespace: &'life2 iceberg::NamespaceIdent, creation: iceberg::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::SessionCatalog::drop_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, namespace: &'life2 iceberg::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::SessionCatalog::drop_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, table: &'life2 iceberg::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::SessionCatalog::get_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, namespace: &'life2 iceberg::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::SessionCatalog::list_namespaces<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, parent: core::option::Option<&'life2 iceberg::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::SessionCatalog::list_tables<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, namespace: &'life2 iceberg::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::SessionCatalog::load_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, table: &'life2 iceberg::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::SessionCatalog::namespace_exists<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, namespace: &'life2 iceberg::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::SessionCatalog::purge_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, table: &'life2 iceberg::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::SessionCatalog::register_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, table: &'life2 iceberg::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::SessionCatalog::rename_table<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, src: &'life2 iceberg::TableIdent, dest: &'life3 iceberg::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::SessionCatalog::table_exists<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, table: &'life2 iceberg::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::SessionCatalog::update_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, namespace: &'life2 iceberg::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::SessionCatalog::update_table<'life0, 'life1, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, commit: iceberg::TableCommit) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub trait iceberg::SessionCatalogBuilder: core::default::Default + core::fmt::Debug + core::marker::Send + core::marker::Sync +pub type iceberg::SessionCatalogBuilder::C: iceberg::SessionCatalog +pub fn iceberg::SessionCatalogBuilder::load(self, name: impl core::convert::Into, props: std::collections::hash::map::HashMap) -> impl core::future::future::Future> + core::marker::Send +pub fn iceberg::SessionCatalogBuilder::with_runtime(self, runtime: iceberg::Runtime) -> Self +pub fn iceberg::SessionCatalogBuilder::with_storage_factory(self, storage_factory: alloc::sync::Arc) -> Self pub async fn iceberg::drop_table_data(table_info: &iceberg::table::Table) -> iceberg::Result<()> pub type iceberg::Result = core::result::Result diff --git a/crates/iceberg/src/catalog/mod.rs b/crates/iceberg/src/catalog/mod.rs index 64cf2617e5..d2ecc7d51e 100644 --- a/crates/iceberg/src/catalog/mod.rs +++ b/crates/iceberg/src/catalog/mod.rs @@ -19,6 +19,7 @@ pub mod memory; mod metadata_location; +mod session; pub(crate) mod utils; use std::collections::HashMap; @@ -36,6 +37,7 @@ pub use metadata_location::*; #[cfg(test)] use mockall::automock; use serde_derive::{Deserialize, Serialize}; +pub use session::*; use typed_builder::TypedBuilder; use uuid::Uuid; diff --git a/crates/iceberg/src/catalog/session.rs b/crates/iceberg/src/catalog/session.rs new file mode 100644 index 0000000000..8a733d6c48 --- /dev/null +++ b/crates/iceberg/src/catalog/session.rs @@ -0,0 +1,270 @@ +// 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. + +//! Session catalog API for Apache Iceberg. + +use std::collections::HashMap; +use std::fmt::Debug; +use std::future::Future; +use std::sync::Arc; + +use async_trait::async_trait; +use typed_builder::TypedBuilder; +use uuid::Uuid; +use zeroize::Zeroizing; + +use crate::io::StorageFactory; +use crate::runtime::Runtime; +use crate::table::Table; +use crate::{Namespace, NamespaceIdent, Result, TableCommit, TableCreation, TableIdent}; + +/// Context for a session. +#[derive(Debug, Clone, TypedBuilder)] +pub struct SessionContext { + /// The unique identifier for this session. + /// + /// Note that the session_id may be used for caching session-scoped state + /// and re-use of a session_id with different session context may result in + /// unexpected behavior. + #[builder(default=Uuid::new_v4().to_string())] + session_id: String, + + /// An optional user or principal associated with the session. + #[builder(default, setter(strip_option))] + identity: Option, + + #[builder(default)] + properties: HashMap, + + #[builder(default)] + credentials: HashMap, +} + +impl SessionContext { + /// Creates a new unique but empty session. + pub fn empty() -> Self { + Self::builder().build() + } + + /// Returns the identifier for this session. + /// + /// The identifier may be used for caching state within a session. + pub fn session_id(&self) -> &str { + &self.session_id + } + + /// Returns a string that identifies the current user or principal. + pub fn identity(&self) -> Option<&str> { + self.identity.as_deref() + } + + /// Returns a map of properties currently set for the session. + pub fn properties(&self) -> &HashMap { + &self.properties + } + + /// Returns the session's credential map. + pub fn credentials(&self) -> &HashMap { + &self.credentials + } +} + +/// A string-like type containing sensitive information such as passwords or tokens. +/// +/// It is redacted from logs and automatically zeroized. +#[derive(Clone)] +pub struct Credential(Zeroizing); + +impl Credential { + /// Creates a new [`Credential`] from a string. + pub fn new(value: String) -> Self { + Self(Zeroizing::new(value)) + } + + /// Returns the raw value of the credential. + pub fn expose(&self) -> &str { + &self.0 + } +} + +impl Debug for Credential { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("Credential([REDACTED])") + } +} + +impl From for Credential { + fn from(value: String) -> Self { + Self::new(value) + } +} + +/// The catalog API for Iceberg Rust that includes session handling. +#[async_trait] +pub trait SessionCatalog: Debug + Send + Sync { + /// List namespaces inside the catalog. + async fn list_namespaces( + &self, + context: &SessionContext, + parent: Option<&NamespaceIdent>, + ) -> Result>; + + /// Create a new namespace inside the catalog. + async fn create_namespace( + &self, + context: &SessionContext, + namespace: &NamespaceIdent, + properties: HashMap, + ) -> Result; + + /// Get a namespace information from the catalog. + async fn get_namespace( + &self, + context: &SessionContext, + namespace: &NamespaceIdent, + ) -> Result; + + /// Check if namespace exists in catalog. + async fn namespace_exists( + &self, + context: &SessionContext, + namespace: &NamespaceIdent, + ) -> Result; + + /// Update a namespace inside the catalog. + /// + /// # Behavior + /// + /// The properties must be the full set of namespace. + async fn update_namespace( + &self, + context: &SessionContext, + namespace: &NamespaceIdent, + properties: HashMap, + ) -> Result<()>; + + /// Drop a namespace from the catalog, or returns error if it doesn't exist. + async fn drop_namespace( + &self, + context: &SessionContext, + namespace: &NamespaceIdent, + ) -> Result<()>; + + /// List tables from namespace. + async fn list_tables( + &self, + context: &SessionContext, + namespace: &NamespaceIdent, + ) -> Result>; + + /// Create a new table inside the namespace. + async fn create_table( + &self, + context: &SessionContext, + namespace: &NamespaceIdent, + creation: TableCreation, + ) -> Result; + + /// Load table from the catalog. + async fn load_table(&self, context: &SessionContext, table: &TableIdent) -> Result
; + + /// Drop a table from the catalog, or returns error if it doesn't exist. + async fn drop_table(&self, context: &SessionContext, table: &TableIdent) -> Result<()>; + + /// Drop a table from the catalog and delete the underlying table data. + /// + /// Implementations should load the table metadata, drop the table + /// from the catalog, then delete all associated data and metadata files. + /// The [`drop_table_data`](super::utils::drop_table_data) utility function can + /// be used for the file cleanup step. + async fn purge_table(&self, context: &SessionContext, table: &TableIdent) -> Result<()>; + + /// Check if a table exists in the catalog. + async fn table_exists(&self, context: &SessionContext, table: &TableIdent) -> Result; + + /// Rename a table in the catalog. + async fn rename_table( + &self, + context: &SessionContext, + src: &TableIdent, + dest: &TableIdent, + ) -> Result<()>; + + /// Register an existing table to the catalog. + async fn register_table( + &self, + context: &SessionContext, + table: &TableIdent, + metadata_location: String, + ) -> Result
; + + /// Update a table to the catalog. + async fn update_table(&self, context: &SessionContext, commit: TableCommit) -> Result
; +} + +/// Common interface for all session catalog builders. +pub trait SessionCatalogBuilder: Default + Debug + Send + Sync { + /// The session catalog type that this builder creates. + type C: SessionCatalog; + + /// 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 + fn with_storage_factory(self, storage_factory: Arc) -> 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. + fn with_runtime(self, runtime: Runtime) -> Self; + + /// Create a new session catalog instance. + fn load( + self, + name: impl Into, + props: HashMap, + ) -> impl Future> + Send; +} + +#[cfg(test)] +mod tests { + use uuid::Uuid; + + use crate::{Credential, SessionContext}; + + #[test] + fn test_empty_session_context_has_uuid_session_id() { + let session = SessionContext::empty(); + let session_id = session.session_id(); + + assert!(Uuid::parse_str(session_id).is_ok()); + } + + #[test] + fn test_credential_redacts_value() { + let sensitive_value = "my-pw-12346"; + + let logged = format!("{:?}", Credential::new(sensitive_value.to_string())); + assert!(!logged.contains(sensitive_value)); + } +}