[1/6] feat: add refreshable storage credential providers

crates/iceberg/public-api.txt

@@ -693,6 +693,13 @@ pub fn iceberg::inspect::SnapshotsTable<'a>::new(table: &'a iceberg::table::Tabl
 pub async fn iceberg::inspect::SnapshotsTable<'a>::scan(&self) -> iceberg::Result<iceberg::scan::ArrowRecordBatchStream>
 pub fn iceberg::inspect::SnapshotsTable<'a>::schema(&self) -> iceberg::spec::Schema
 pub mod iceberg::io
+pub enum iceberg::io::StorageCredentialKind
+pub iceberg::io::StorageCredentialKind::Gcs(iceberg::io::GcsCredential)
+pub iceberg::io::StorageCredentialKind::S3(iceberg::io::S3Credential)
+impl core::clone::Clone for iceberg::io::StorageCredentialKind
+pub fn iceberg::io::StorageCredentialKind::clone(&self) -> iceberg::io::StorageCredentialKind
+impl core::fmt::Debug for iceberg::io::StorageCredentialKind
+pub fn iceberg::io::StorageCredentialKind::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 pub struct iceberg::io::AzdlsConfig
 pub iceberg::io::AzdlsConfig::account_key: core::option::Option<alloc::string::String>
 pub iceberg::io::AzdlsConfig::account_name: core::option::Option<alloc::string::String>
@@ -743,6 +750,7 @@ impl iceberg::io::FileIOBuilder
 pub fn iceberg::io::FileIOBuilder::build(self) -> iceberg::io::FileIO
 pub fn iceberg::io::FileIOBuilder::config(&self) -> &iceberg::io::StorageConfig
 pub fn iceberg::io::FileIOBuilder::new(factory: alloc::sync::Arc<dyn iceberg::io::StorageFactory>) -> Self
+pub fn iceberg::io::FileIOBuilder::with_credential_provider(self, provider: alloc::sync::Arc<dyn iceberg::io::StorageCredentialProvider>) -> Self
 pub fn iceberg::io::FileIOBuilder::with_prop(self, key: impl alloc::string::ToString, value: impl alloc::string::ToString) -> Self
 pub fn iceberg::io::FileIOBuilder::with_props(self, args: impl core::iter::traits::collect::IntoIterator<Item = (impl alloc::string::ToString, impl alloc::string::ToString)>) -> Self
 impl core::clone::Clone for iceberg::io::FileIOBuilder
@@ -779,6 +787,15 @@ impl serde_core::ser::Serialize for iceberg::io::GcsConfig
 pub fn iceberg::io::GcsConfig::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::io::GcsConfig
 pub fn iceberg::io::GcsConfig::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::io::GcsCredential
+impl iceberg::io::GcsCredential
+pub fn iceberg::io::GcsCredential::into_token(self) -> alloc::string::String
+pub fn iceberg::io::GcsCredential::new(token: impl core::convert::Into<alloc::string::String>) -> Self
+pub fn iceberg::io::GcsCredential::token(&self) -> &str
+impl core::clone::Clone for iceberg::io::GcsCredential
+pub fn iceberg::io::GcsCredential::clone(&self) -> iceberg::io::GcsCredential
+impl core::fmt::Debug for iceberg::io::GcsCredential
+pub fn iceberg::io::GcsCredential::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 pub struct iceberg::io::HfConfig
 pub iceberg::io::HfConfig::endpoint: core::option::Option<alloc::string::String>
 pub iceberg::io::HfConfig::revision: core::option::Option<alloc::string::String>
@@ -846,6 +863,7 @@ impl core::fmt::Debug for iceberg::io::LocalFsStorageFactory
 pub fn iceberg::io::LocalFsStorageFactory::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 impl iceberg::io::StorageFactory for iceberg::io::LocalFsStorageFactory
 pub fn iceberg::io::LocalFsStorageFactory::build(&self, _config: &iceberg::io::StorageConfig) -> iceberg::Result<alloc::sync::Arc<dyn iceberg::io::Storage>>
+pub fn iceberg::io::LocalFsStorageFactory::build_with_credentials(&self, config: &iceberg::io::StorageConfig, credential_provider: core::option::Option<alloc::sync::Arc<dyn iceberg::io::StorageCredentialProvider>>) -> iceberg::Result<alloc::sync::Arc<dyn iceberg::io::Storage>>
 impl serde_core::ser::Serialize for iceberg::io::LocalFsStorageFactory
 pub fn iceberg::io::LocalFsStorageFactory::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::io::LocalFsStorageFactory
@@ -884,6 +902,7 @@ impl core::fmt::Debug for iceberg::io::MemoryStorageFactory
 pub fn iceberg::io::MemoryStorageFactory::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 impl iceberg::io::StorageFactory for iceberg::io::MemoryStorageFactory
 pub fn iceberg::io::MemoryStorageFactory::build(&self, _config: &iceberg::io::StorageConfig) -> iceberg::Result<alloc::sync::Arc<dyn iceberg::io::Storage>>
+pub fn iceberg::io::MemoryStorageFactory::build_with_credentials(&self, config: &iceberg::io::StorageConfig, credential_provider: core::option::Option<alloc::sync::Arc<dyn iceberg::io::StorageCredentialProvider>>) -> iceberg::Result<alloc::sync::Arc<dyn iceberg::io::Storage>>
 impl serde_core::ser::Serialize for iceberg::io::MemoryStorageFactory
 pub fn iceberg::io::MemoryStorageFactory::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::io::MemoryStorageFactory
@@ -959,6 +978,17 @@ impl serde_core::ser::Serialize for iceberg::io::S3Config
 pub fn iceberg::io::S3Config::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::io::S3Config
 pub fn iceberg::io::S3Config::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::io::S3Credential
+impl iceberg::io::S3Credential
+pub fn iceberg::io::S3Credential::access_key_id(&self) -> &str
+pub fn iceberg::io::S3Credential::into_parts(self) -> (alloc::string::String, alloc::string::String, core::option::Option<alloc::string::String>)
+pub fn iceberg::io::S3Credential::new(access_key_id: impl core::convert::Into<alloc::string::String>, secret_access_key: impl core::convert::Into<alloc::string::String>, session_token: core::option::Option<alloc::string::String>) -> Self
+pub fn iceberg::io::S3Credential::secret_access_key(&self) -> &str
+pub fn iceberg::io::S3Credential::session_token(&self) -> core::option::Option<&str>
+impl core::clone::Clone for iceberg::io::S3Credential
+pub fn iceberg::io::S3Credential::clone(&self) -> iceberg::io::S3Credential
+impl core::fmt::Debug for iceberg::io::S3Credential
+pub fn iceberg::io::S3Credential::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 pub struct iceberg::io::StorageConfig
 impl iceberg::io::StorageConfig
 pub fn iceberg::io::StorageConfig::from_props(props: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> Self
@@ -996,6 +1026,19 @@ impl serde_core::ser::Serialize for iceberg::io::StorageConfig
 pub fn iceberg::io::StorageConfig::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::io::StorageConfig
 pub fn iceberg::io::StorageConfig::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::io::StorageCredential
+impl iceberg::io::StorageCredential
+pub fn iceberg::io::StorageCredential::expires_at(&self) -> core::option::Option<std::time::SystemTime>
+pub fn iceberg::io::StorageCredential::into_kind(self) -> iceberg::io::StorageCredentialKind
+pub fn iceberg::io::StorageCredential::kind(&self) -> &iceberg::io::StorageCredentialKind
+pub fn iceberg::io::StorageCredential::new(kind: iceberg::io::StorageCredentialKind) -> Self
+pub fn iceberg::io::StorageCredential::prefix(&self) -> core::option::Option<&str>
+pub fn iceberg::io::StorageCredential::with_expiration(self, expires_at: std::time::SystemTime) -> Self
+pub fn iceberg::io::StorageCredential::with_prefix(self, prefix: impl core::convert::Into<alloc::string::String>) -> Self
+impl core::clone::Clone for iceberg::io::StorageCredential
+pub fn iceberg::io::StorageCredential::clone(&self) -> iceberg::io::StorageCredential
+impl core::fmt::Debug for iceberg::io::StorageCredential
+pub fn iceberg::io::StorageCredential::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 pub const iceberg::io::ADLS_ACCOUNT_KEY: &str
 pub const iceberg::io::ADLS_ACCOUNT_NAME: &str
 pub const iceberg::io::ADLS_AUTHORITY_HOST: &str
@@ -1004,15 +1047,20 @@ pub const iceberg::io::ADLS_CLIENT_SECRET: &str
 pub const iceberg::io::ADLS_CONNECTION_STRING: &str
 pub const iceberg::io::ADLS_SAS_TOKEN: &str
 pub const iceberg::io::ADLS_TENANT_ID: &str
+pub const iceberg::io::AWS_REFRESH_CREDENTIALS_ENABLED: &str
+pub const iceberg::io::AWS_REFRESH_CREDENTIALS_ENDPOINT: &str
 pub const iceberg::io::CLIENT_REGION: &str
 pub const iceberg::io::GCS_ALLOW_ANONYMOUS: &str
 pub const iceberg::io::GCS_CREDENTIALS_JSON: &str
 pub const iceberg::io::GCS_DISABLE_CONFIG_LOAD: &str
 pub const iceberg::io::GCS_DISABLE_VM_METADATA: &str
 pub const iceberg::io::GCS_NO_AUTH: &str
 pub const iceberg::io::GCS_PROJECT_ID: &str
+pub const iceberg::io::GCS_REFRESH_CREDENTIALS_ENABLED: &str
+pub const iceberg::io::GCS_REFRESH_CREDENTIALS_ENDPOINT: &str
 pub const iceberg::io::GCS_SERVICE_HOST: &str
 pub const iceberg::io::GCS_TOKEN: &str
+pub const iceberg::io::GCS_TOKEN_EXPIRES_AT: &str
 pub const iceberg::io::GCS_USER_PROJECT: &str
 pub const iceberg::io::HF_ENDPOINT: &str
 pub const iceberg::io::HF_REVISION: &str
@@ -1032,6 +1080,7 @@ pub const iceberg::io::S3_PATH_STYLE_ACCESS: &str
 pub const iceberg::io::S3_REGION: &str
 pub const iceberg::io::S3_SECRET_ACCESS_KEY: &str
 pub const iceberg::io::S3_SESSION_TOKEN: &str
+pub const iceberg::io::S3_SESSION_TOKEN_EXPIRES_AT_MS: &str
 pub const iceberg::io::S3_SSE_KEY: &str
 pub const iceberg::io::S3_SSE_MD5: &str
 pub const iceberg::io::S3_SSE_TYPE: &str
@@ -1083,12 +1132,18 @@ pub fn iceberg::io::MemoryStorage::read<'life0, 'life1, 'async_trait>(&'life0 se
 pub fn iceberg::io::MemoryStorage::reader<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<alloc::boxed::Box<dyn iceberg::io::FileRead>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait
 pub fn iceberg::io::MemoryStorage::write<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str, bs: bytes::bytes::Bytes) -> 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::io::MemoryStorage::writer<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<alloc::boxed::Box<dyn iceberg::io::FileWrite>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait
+pub trait iceberg::io::StorageCredentialProvider: core::fmt::Debug + core::marker::Send + core::marker::Sync
+pub fn iceberg::io::StorageCredentialProvider::load_credential<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<iceberg::io::StorageCredential>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait
+pub fn iceberg::io::StorageCredentialProvider::supports_path(&self, _path: &str) -> bool
 pub trait iceberg::io::StorageFactory: core::fmt::Debug + core::marker::Send + core::marker::Sync + typetag::Serialize + typetag::Deserialize
 pub fn iceberg::io::StorageFactory::build(&self, config: &iceberg::io::StorageConfig) -> iceberg::Result<alloc::sync::Arc<dyn iceberg::io::Storage>>
+pub fn iceberg::io::StorageFactory::build_with_credentials(&self, config: &iceberg::io::StorageConfig, credential_provider: core::option::Option<alloc::sync::Arc<dyn iceberg::io::StorageCredentialProvider>>) -> iceberg::Result<alloc::sync::Arc<dyn iceberg::io::Storage>>
 impl iceberg::io::StorageFactory for iceberg::io::LocalFsStorageFactory
 pub fn iceberg::io::LocalFsStorageFactory::build(&self, _config: &iceberg::io::StorageConfig) -> iceberg::Result<alloc::sync::Arc<dyn iceberg::io::Storage>>
+pub fn iceberg::io::LocalFsStorageFactory::build_with_credentials(&self, config: &iceberg::io::StorageConfig, credential_provider: core::option::Option<alloc::sync::Arc<dyn iceberg::io::StorageCredentialProvider>>) -> iceberg::Result<alloc::sync::Arc<dyn iceberg::io::Storage>>
 impl iceberg::io::StorageFactory for iceberg::io::MemoryStorageFactory
 pub fn iceberg::io::MemoryStorageFactory::build(&self, _config: &iceberg::io::StorageConfig) -> iceberg::Result<alloc::sync::Arc<dyn iceberg::io::Storage>>
+pub fn iceberg::io::MemoryStorageFactory::build_with_credentials(&self, config: &iceberg::io::StorageConfig, credential_provider: core::option::Option<alloc::sync::Arc<dyn iceberg::io::StorageCredentialProvider>>) -> iceberg::Result<alloc::sync::Arc<dyn iceberg::io::Storage>>
 pub mod iceberg::memory
 pub struct iceberg::memory::MemoryCatalog
 impl core::fmt::Debug for iceberg::memory::MemoryCatalog

crates/iceberg/src/io/file_io.rs

@@ -22,7 +22,8 @@ use bytes::Bytes;
 use futures::{Stream, StreamExt};
 
 use super::storage::{
-    LocalFsStorageFactory, MemoryStorageFactory, Storage, StorageConfig, StorageFactory,
+    LocalFsStorageFactory, MemoryStorageFactory, Storage, StorageConfig, StorageCredentialProvider,
+    StorageFactory,
 };
 use crate::Result;
 
@@ -65,6 +66,8 @@ pub struct FileIO {
     config: StorageConfig,
     /// Factory for creating storage instances
     factory: Arc<dyn StorageFactory>,
+    /// Optional provider of refreshable, backend-specific credentials
+    credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
     /// Cached storage instance (lazily initialized)
     storage: Arc<OnceLock<Arc<dyn Storage>>>,
 }
@@ -77,6 +80,7 @@ impl FileIO {
         Self {
             config: StorageConfig::new(),
             factory: Arc::new(MemoryStorageFactory),
+            credential_provider: None,
             storage: Arc::new(OnceLock::new()),
         }
     }
@@ -88,6 +92,7 @@ impl FileIO {
         Self {
             config: StorageConfig::new(),
             factory: Arc::new(LocalFsStorageFactory),
+            credential_provider: None,
             storage: Arc::new(OnceLock::new()),
         }
     }
@@ -107,8 +112,11 @@ impl FileIO {
             return Ok(storage.clone());
         }
 
-        // Build the storage
-        let storage = self.factory.build(&self.config)?;
+        // Build the storage, passing any credential provider so backends that
+        // support refreshable credentials can wire it into their operators.
+        let storage = self
+            .factory
+            .build_with_credentials(&self.config, self.credential_provider.clone())?;
 
         // Try to set it (another thread might have set it first)
         let _ = self.storage.set(storage.clone());
@@ -191,6 +199,8 @@ pub struct FileIOBuilder {
     factory: Arc<dyn StorageFactory>,
     /// Storage configuration
     config: StorageConfig,
+    /// Optional provider of refreshable, backend-specific credentials
+    credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
 }
 
 impl FileIOBuilder {
@@ -199,6 +209,7 @@ impl FileIOBuilder {
         Self {
             factory,
             config: StorageConfig::new(),
+            credential_provider: None,
         }
     }
 
@@ -224,11 +235,21 @@ impl FileIOBuilder {
         &self.config
     }
 
+    /// Attach a provider of refreshable, backend-specific credentials.
+    pub fn with_credential_provider(
+        mut self,
+        provider: Arc<dyn StorageCredentialProvider>,
+    ) -> Self {
+        self.credential_provider = Some(provider);
+        self
+    }
+
     /// Builds [`FileIO`].
     pub fn build(self) -> FileIO {
         FileIO {
             config: self.config,
             factory: self.factory,
+            credential_provider: self.credential_provider,
             storage: Arc::new(OnceLock::new()),
         }
     }
@@ -397,7 +418,20 @@ mod tests {
     use tempfile::TempDir;
 
     use super::{FileIO, FileIOBuilder};
-    use crate::io::{LocalFsStorageFactory, MemoryStorageFactory};
+    use crate::io::{
+        LocalFsStorageFactory, MemoryStorageFactory, StorageCredential, StorageCredentialProvider,
+    };
+    use crate::{ErrorKind, Result};
+
+    #[derive(Debug)]
+    struct TestCredentialProvider;
+
+    #[async_trait::async_trait]
+    impl StorageCredentialProvider for TestCredentialProvider {
+        async fn load_credential(&self, _path: &str) -> Result<StorageCredential> {
+            unreachable!("unsupported factories must reject the provider before loading from it")
+        }
+    }
 
     fn create_local_file_io() -> FileIO {
         FileIO::new_with_fs()
@@ -544,4 +578,14 @@ mod tests {
         assert_eq!(file_io.config().get("key1"), Some(&"value1".to_string()));
         assert_eq!(file_io.config().get("key2"), Some(&"value2".to_string()));
     }
+
+    #[tokio::test]
+    async fn test_file_io_rejects_credentials_for_unsupported_factory() {
+        let file_io = FileIOBuilder::new(Arc::new(MemoryStorageFactory))
+            .with_credential_provider(Arc::new(TestCredentialProvider))
+            .build();
+
+        let err = file_io.exists("memory://file").await.unwrap_err();
+        assert_eq!(err.kind(), ErrorKind::FeatureUnsupported, "{err}");
+    }
 }

crates/iceberg/src/io/storage/config/gcs.rs

@@ -41,6 +41,12 @@ pub const GCS_NO_AUTH: &str = "gcs.no-auth";
 pub const GCS_CREDENTIALS_JSON: &str = "gcs.credentials-json";
 /// Google Cloud Storage token.
 pub const GCS_TOKEN: &str = "gcs.oauth2.token";
+/// Epoch-millisecond timestamp at which the vended GCS OAuth2 token expires.
+pub const GCS_TOKEN_EXPIRES_AT: &str = "gcs.oauth2.token-expires-at";
+/// Endpoint used to fetch and refresh vended GCS OAuth2 credentials.
+pub const GCS_REFRESH_CREDENTIALS_ENDPOINT: &str = "gcs.oauth2.refresh-credentials-endpoint";
+/// Whether vended GCS OAuth2 credentials should be refreshed. Defaults to `true`.
+pub const GCS_REFRESH_CREDENTIALS_ENABLED: &str = "gcs.oauth2.refresh-credentials-enabled";
 /// Option to skip signing requests (e.g. for public buckets/folders).
 pub const GCS_ALLOW_ANONYMOUS: &str = "gcs.allow-anonymous";
 /// Option to skip loading the credential from GCE metadata server.

crates/iceberg/src/io/storage/config/s3.rs

@@ -35,10 +35,16 @@ pub const S3_ACCESS_KEY_ID: &str = "s3.access-key-id";
 pub const S3_SECRET_ACCESS_KEY: &str = "s3.secret-access-key";
 /// S3 session token (required when using temporary credentials).
 pub const S3_SESSION_TOKEN: &str = "s3.session-token";
+/// Epoch-millisecond timestamp at which the vended S3 session token expires.
+pub const S3_SESSION_TOKEN_EXPIRES_AT_MS: &str = "s3.session-token-expires-at-ms";
 /// S3 region.
 pub const S3_REGION: &str = "s3.region";
 /// Region to use for the S3 client (takes precedence over [`S3_REGION`]).
 pub const CLIENT_REGION: &str = "client.region";
+/// Endpoint used to fetch and refresh vended AWS credentials.
+pub const AWS_REFRESH_CREDENTIALS_ENDPOINT: &str = "client.refresh-credentials-endpoint";
+/// Whether vended AWS credentials should be refreshed. Defaults to `true`.
+pub const AWS_REFRESH_CREDENTIALS_ENABLED: &str = "client.refresh-credentials-enabled";
 /// S3 Path Style Access.
 pub const S3_PATH_STYLE_ACCESS: &str = "s3.path-style-access";
 /// S3 Server Side Encryption Type.