This is an automated email from the ASF dual-hosted git repository.
CTTY pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 10c45ed5a feat(catalog): Add SessionCatalog trait (#2836)
10c45ed5a is described below
commit 10c45ed5a3e2b22a13f0368da3d403d622ef44c3
Author: Jannik Steinmann <[email protected]>
AuthorDate: Wed Aug 5 02:35:04 2026 +0200
feat(catalog): Add SessionCatalog trait (#2836)
* Add SessionCatalog trait and SessionContext
* Add remaining SessionContext fields
* Use safe Credential type in SessionContext
* Default to UUID session_id
* Dropped the redundant Credential::new and added an example
* More SessionContext-level tests + example
* Automock the SessionCatalog trait
* Compile-assert that types are Send+Sync and dyn-compatible
* Remove SessionCatalogBuilder
Signed-off-by: Jannik Steinmann <[email protected]>
Re-generate public API
Signed-off-by: Jannik Steinmann <[email protected]>
---------
Signed-off-by: Jannik Steinmann <[email protected]>
---
crates/iceberg/public-api.txt | 38 +++++
crates/iceberg/src/catalog/mod.rs | 2 +
crates/iceberg/src/catalog/session.rs | 300 ++++++++++++++++++++++++++++++++++
3 files changed, 340 insertions(+)
diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt
index 1f2829878..530aa4cbf 100644
--- a/crates/iceberg/public-api.txt
+++ b/crates/iceberg/public-api.txt
@@ -3539,6 +3539,15 @@ 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<Self, <__D as serde_core::de::Deserializer>::Error> where
__D: serde_core::de::Deserializer<'de>
+pub struct iceberg::Credential(_)
+impl iceberg::Credential
+pub fn iceberg::Credential::expose(&self) -> &str
+impl core::clone::Clone for iceberg::Credential
+pub fn iceberg::Credential::clone(&self) -> iceberg::Credential
+impl core::convert::From<alloc::string::String> 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
@@ -3687,6 +3696,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<alloc::string::String,
iceberg::Credential>
+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<alloc::string::String,
alloc::string::String>
+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<iceberg::table::Table>
@@ -3794,5 +3816,21 @@ 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<dyn
iceberg::encryption::kms::KmsClientFactory>) -> 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<dyn iceberg::io::StorageFactory>) -> 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<alloc::string::String,
alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn
core::future::future::Future<Output = iceberg::Result<iceberg::Namespace>> +
core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0:
'async_trait, 'life1 [...]
+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<alloc::boxed::Box<(dyn core::future::future::Future<Output =
iceberg::Result<iceberg::table::Table>> + 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<alloc::boxed::Box<(dyn core::future::future::Future<Output =
iceberg::Result<()>> + 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<alloc::boxed::Box<(dyn
core::future::future::Future<Output = iceberg::Result<()>> + 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<alloc::boxed::Box<(dyn core::future::future::Future<Output =
iceberg::Result<iceberg::Namespace>> + 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<alloc::boxed::Box<(dyn core::future::future::Future<Output =
iceberg::Result<alloc::vec::Vec<iceberg::NamespaceIdent>>> + 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<alloc::boxed::Box<(dyn core::future::future::Future<Output =
iceberg::Result<alloc::vec::Vec<iceberg::TableIdent>>> + 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<alloc::boxed::Box<(dyn
core::future::future::Future<Output = iceberg::Result<iceberg::table::Table>> +
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<alloc::boxed::Box<(dyn core::future::future::Future<Output =
iceberg::Result<bool>> + 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<alloc::boxed::Box<(dyn
core::future::future::Future<Output = iceberg::Result<()>> + 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<alloc::boxed::Box<(dyn core::future::future::Future<Output =
iceberg::Result<iceberg::table::Table>> + 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<alloc::boxed::Box<(dyn core::future::future::Future<Output =
iceberg::Result<()>> + 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<alloc::boxed::Box<(dyn
core::future::future::Future<Output = iceberg::Result<bool>> +
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<alloc::string::String,
alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn
core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send
+ 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1:
'async_trait, [...]
+pub fn iceberg::SessionCatalog::update_table<'life0, 'life1,
'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, commit:
iceberg::TableCommit) -> core::pin::Pin<alloc::boxed::Box<(dyn
core::future::future::Future<Output = iceberg::Result<iceberg::table::Table>> +
core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0:
'async_trait, 'life1: 'async_trait
pub async fn iceberg::drop_table_data(table_info: &iceberg::table::Table) ->
iceberg::Result<()>
pub type iceberg::Result<T> = core::result::Result<T, iceberg::Error>
diff --git a/crates/iceberg/src/catalog/mod.rs
b/crates/iceberg/src/catalog/mod.rs
index 64cf2617e..d2ecc7d51 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 000000000..1790f4978
--- /dev/null
+++ b/crates/iceberg/src/catalog/session.rs
@@ -0,0 +1,300 @@
+// 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 async_trait::async_trait;
+#[cfg(test)]
+use mockall::automock;
+use typed_builder::TypedBuilder;
+use uuid::Uuid;
+use zeroize::Zeroizing;
+
+use crate::table::Table;
+use crate::{Namespace, NamespaceIdent, Result, TableCommit, TableCreation,
TableIdent};
+
+/// Context for a session.
+///
+/// # Example
+/// ```rust
+/// use iceberg::SessionContext;
+///
+/// let session = SessionContext::builder()
+/// .identity("user123".to_string())
+/// .build();
+///
+/// assert_eq!(session.identity(), Some("user123"));
+/// assert!(!session.session_id().is_empty());
+/// ```
+#[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<String>,
+
+ #[builder(default)]
+ properties: HashMap<String, String>,
+
+ #[builder(default)]
+ credentials: HashMap<String, Credential>,
+}
+
+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<String, String> {
+ &self.properties
+ }
+
+ /// Returns the session's credential map.
+ pub fn credentials(&self) -> &HashMap<String, Credential> {
+ &self.credentials
+ }
+}
+
+/// A string-like type containing sensitive information such as passwords or
tokens.
+///
+/// It is redacted from logs and automatically zeroized.
+///
+/// # Example
+/// ```rust
+/// use iceberg::Credential;
+///
+/// let sensitive_value = "my-pw-12345";
+/// let credential = Credential::from(sensitive_value.to_string());
+///
+/// // Not contained in debug logs.
+/// assert!(!format!("{:?}", credential).contains(sensitive_value));
+/// ```
+#[derive(Clone)]
+pub struct Credential(Zeroizing<String>);
+
+impl Credential {
+ /// 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<String> for Credential {
+ fn from(value: String) -> Self {
+ Self(Zeroizing::new(value))
+ }
+}
+
+/// The catalog API for Iceberg Rust that includes session handling.
+#[async_trait]
+#[cfg_attr(test, automock)]
+pub trait SessionCatalog: Debug + Send + Sync {
+ /// List namespaces inside the catalog.
+ async fn list_namespaces(
+ &self,
+ context: &SessionContext,
+ parent: Option<&NamespaceIdent>,
+ ) -> Result<Vec<NamespaceIdent>>;
+
+ /// Create a new namespace inside the catalog.
+ async fn create_namespace(
+ &self,
+ context: &SessionContext,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
+ ) -> Result<Namespace>;
+
+ /// Get a namespace information from the catalog.
+ async fn get_namespace(
+ &self,
+ context: &SessionContext,
+ namespace: &NamespaceIdent,
+ ) -> Result<Namespace>;
+
+ /// Check if namespace exists in catalog.
+ async fn namespace_exists(
+ &self,
+ context: &SessionContext,
+ namespace: &NamespaceIdent,
+ ) -> Result<bool>;
+
+ /// 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<String, String>,
+ ) -> 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<Vec<TableIdent>>;
+
+ /// Create a new table inside the namespace.
+ async fn create_table(
+ &self,
+ context: &SessionContext,
+ namespace: &NamespaceIdent,
+ creation: TableCreation,
+ ) -> Result<Table>;
+
+ /// Load table from the catalog.
+ async fn load_table(&self, context: &SessionContext, table: &TableIdent)
-> Result<Table>;
+
+ /// 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<bool>;
+
+ /// 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<Table>;
+
+ /// Update a table to the catalog.
+ async fn update_table(&self, context: &SessionContext, commit:
TableCommit) -> Result<Table>;
+}
+
+#[cfg(test)]
+mod tests {
+ use std::collections::HashMap;
+
+ use uuid::Uuid;
+
+ use crate::{Credential, SessionCatalog, 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_empty_sessions_get_unique_id() {
+ let session1 = SessionContext::empty();
+ let session2 = SessionContext::empty();
+
+ assert_ne!(session1.session_id(), session2.session_id())
+ }
+
+ #[test]
+ fn test_empty_sessions_get_unique_id_via_builder() {
+ let session1 = SessionContext::builder().build();
+ let session2 = SessionContext::builder().build();
+
+ assert_ne!(session1.session_id(), session2.session_id());
+ }
+
+ #[test]
+ fn test_session_with_credentials_does_not_display_them() {
+ let sensitive_value = "my-pw-123456";
+ let session = SessionContext::builder()
+ .credentials(HashMap::from([(
+ "key".to_string(),
+ Credential::from(sensitive_value.to_string()),
+ )]))
+ .build();
+
+ let logged = format!("{:?}", session);
+ assert!(!logged.contains(sensitive_value))
+ }
+
+ #[test]
+ fn test_credential_redacts_value() {
+ let sensitive_value = "my-pw-12346";
+
+ let logged = format!("{:?}",
Credential::from(sensitive_value.to_string()));
+ assert!(!logged.contains(sensitive_value));
+ }
+
+ #[test]
+ fn test_types_are_send_sync() {
+ assert_send_sync::<Credential>();
+ assert_send_sync::<SessionContext>();
+ assert_send_sync::<dyn SessionCatalog>();
+
+ fn _dyn_compatible(_: &dyn SessionCatalog) {}
+ }
+
+ fn assert_send_sync<T: Send + Sync + ?Sized>() {}
+}
