mbutrovich commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3723620788
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -384,6 +415,27 @@ impl OpenDalStorage {
}
}
+ /// Returns whether `path` uses a path-scoped dynamic credential provider.
+ ///
+ /// Such paths cannot share an OpenDAL deleter unless the provider can
expose
+ /// the credential scope that applies to each path. Process them
independently
+ /// so bulk deletion remains bounded without crossing credential
boundaries.
+ fn uses_dynamic_credentials(&self, path: &str) -> bool {
+ match self {
+ #[cfg(feature = "opendal-s3")]
+ OpenDalStorage::S3 {
+ credential_provider: Some(provider),
+ ..
+ } => provider.supports_path(path),
+ #[cfg(feature = "opendal-gcs")]
+ OpenDalStorage::Gcs {
Review Comment:
`crates/storage/opendal/src/lib.rs:423-431` (`uses_dynamic_credentials`) and
`:610-628` (`delete_stream`).
When a path is served by a credential provider, `delete_stream` skips the
shared per-bucket `Deleter` and instead calls `create_operator` + a single
`op.delete(relative_path)` per path, sequentially, inside the stream loop. The
non-dynamic branch batches deletes through OpenDAL's `Deleter` (which can use
bulk delete APIs); the dynamic branch does neither batching nor concurrency,
and rebuilds the operator from scratch for every single file.
For `expire_snapshots`/purge on a table with vended-credential refresh
enabled, this turns what would be a handful of batched multi-object delete
calls into one HTTP round trip per file, plus an operator-construction cost per
file. The code comment explains *why* deletes can't share a `Deleter` across
different credential-prefix scopes (correctness: `batch_key_for_path` only
groups by bucket, not by credential scope), but the fix taken forfeits batching
entirely rather than partially, i.e. grouping deletes by `(bucket, matched
credential prefix)` instead of just `bucket` would preserve batched delete
within each credential-scope group. Was that considered?
##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,1203 @@
+// 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.
+
+//! Refresh of vended storage credentials against a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache, and
+//! the returned [`StorageCredential`] enum lets the storage adapter enforce
the
+//! expected backend-specific type. This preserves Java's per-cloud refresh
+//! policy while supporting mixed-cloud tables through a resolving FileIO.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant.
To
+//! add a backend, first add its credential type to Iceberg's storage API and
+//! teach the storage adapter to consume it. Then write its `parse_*` function,
+//! add a `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+ AWS_REFRESH_CREDENTIALS_ENABLED, AWS_REFRESH_CREDENTIALS_ENDPOINT,
+ GCS_REFRESH_CREDENTIALS_ENABLED, GCS_REFRESH_CREDENTIALS_ENDPOINT,
GCS_TOKEN,
+ GCS_TOKEN_EXPIRES_AT, GcsCredential, S3_ACCESS_KEY_ID,
S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN,
+ S3_SESSION_TOKEN_EXPIRES_AT_MS, S3Credential, StorageCredential,
StorageCredentialKind,
+ StorageCredentialProvider,
+};
+use iceberg::{Error, ErrorKind, Result};
+use rand::Rng;
+use reqwest::{Method, StatusCode, Url};
+use tokio::sync::Mutex;
+
+use crate::REST_CATALOG_PROP_SCAN_PLAN_ID;
+use crate::client::{HttpClient, deserialize_unexpected_catalog_error};
+use crate::types::LoadCredentialsResponse;
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// It contains the location schemes it backs, the property keys it is
configured
+/// with, and how to parse its credential. The generic provider stays free of
+/// any per-cloud knowledge.
+struct CloudRefresh {
+ /// Location URL schemes this backend serves.
+ schemes: &'static [&'static str],
+ /// Table property naming the refresh endpoint (absolute or
catalog-relative).
+ endpoint_key: &'static str,
+ /// Table property to opt out; refresh is enabled unless this is `"false"`.
+ enabled_key: &'static str,
+ /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+ jitter_prefetch: bool,
+ /// Parse a complete credential from catalog-supplied properties.
+ parse_credential:
+ fn(config: &HashMap<String, String>, prefix: Option<String>) ->
Result<StorageCredential>,
+}
+
+impl CloudRefresh {
+ /// S3 / AWS
+ const AWS: Self = Self {
+ schemes: &["s3", "s3a", "s3n"],
+ endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+ enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+ jitter_prefetch: true,
+ parse_credential: parse_s3_credential,
+ };
+ /// Google Cloud Storage
+ const GCP: Self = Self {
+ schemes: &["gs", "gcs"],
+ endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+ enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+ jitter_prefetch: false,
+ parse_credential: parse_gcs_credential,
+ };
+ // TODO: Azure (ADLS) is not yet supported: opendal 0.57's Azdls builder
exposes no
+ // custom credential-provider hook, and reqsign's SAS-token credential has
no
+ // expiry, so reqsign-based refresh isn't possible.
+
+ /// Backends with refresh support
+ const SUPPORTED: &[Self] = &[Self::AWS, Self::GCP];
+
+ /// The backend that serves `location`, by its URL scheme, or `None` if no
+ /// supported backend matches (in which case static credentials are used
+ /// as-is, as before).
+ fn for_location(location: &str) -> Option<&'static Self> {
+ Self::SUPPORTED
+ .iter()
+ .find(|cloud| cloud.matches_location(location))
+ }
+
+ fn matches_location(&self, location: &str) -> bool {
+ scheme_of(location).is_some_and(|scheme|
self.schemes.contains(&scheme.as_str()))
+ }
+}
+
+/// Re-fetch a credential once it is within this window of expiry, so a fresh
+/// token is in hand before the object store would reject the old one.
+const REFRESH_BUFFER: Duration = Duration::from_mins(5);
+
+/// AWS keeps at least one minute between its jittered prefetch time and
expiry.
+const MIN_REFRESH_BUFFER: Duration = Duration::from_mins(1);
+
+/// Initial ceiling for failure backoff. Equal jitter chooses from half this
+/// value through the full value.
+const INITIAL_FAILURE_BACKOFF: Duration = Duration::from_secs(1);
+
+/// Maximum failure backoff while a cached credential remains usable.
+const MAX_FAILURE_BACKOFF: Duration = Duration::from_secs(30);
+
+/// A cached vended credential and its refresh schedule.
+#[derive(Clone)]
+struct CachedEntry {
+ credential: StorageCredential,
+ /// When this entry becomes eligible for prefetch. `None` means it does not
+ /// expire and therefore never needs proactive refresh.
+ refresh_at: Option<SystemTime>,
+}
+
+impl CachedEntry {
+ fn new(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+ let refresh_at = credential
+ .expires_at
+ .map(|expires_at| prefetch_time(expires_at, jitter_prefetch));
+ Self {
+ credential,
+ refresh_at,
+ }
+ }
+
+ /// Seed entries that are already inside the nominal five-minute window are
+ /// immediately due. Otherwise AWS applies the same jitter as it does to a
+ /// freshly fetched value.
+ fn seed(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+ let due = credential.expires_at.is_some_and(|expires_at| {
+ SystemTime::now()
+ .checked_add(REFRESH_BUFFER)
+ .is_none_or(|refresh_boundary| refresh_boundary >= expires_at)
+ });
+ let mut entry = Self::new(credential, jitter_prefetch);
+ if due {
+ entry.refresh_at = Some(UNIX_EPOCH);
+ }
+ entry
+ }
+
+ fn is_fresh(&self, now: SystemTime) -> bool {
+ self.refresh_at.is_none_or(|refresh_at| now < refresh_at)
+ }
+
+ fn is_unexpired(&self, now: SystemTime) -> bool {
+ self.credential
+ .expires_at
+ .is_none_or(|expires_at| now < expires_at)
+ }
+}
+
+/// Cached credentials plus failure-backoff state.
+struct CacheState {
+ entries: Vec<CachedEntry>,
+ consecutive_failures: u32,
+ retry_not_before: Option<Instant>,
+}
+
+struct ConfiguredCloud {
+ cloud: &'static CloudRefresh,
+ endpoint: String,
+ cache: Mutex<CacheState>,
+ /// Only one caller fetches at a time. The cache lock is deliberately
+ /// separate so other callers can keep using an unexpired credential while
+ /// the refresh is in flight.
+ refresh: Mutex<()>,
+}
+
+/// Fetches and refreshes vended credentials from a REST catalog's table
+/// credentials endpoint.
+///
+/// Each cloud cache is seeded with the credential from the initial table
+/// properties (when complete) and re-fetched from its endpoint as it
+/// nears expiry.
+pub(crate) struct RestVendedCredentialProvider {
+ client: Arc<HttpClient>,
+ /// Optional scan-plan identifier.
+ plan_id: Option<String>,
+ /// Independently configured endpoint and cache for each backing cloud.
+ clouds: Vec<ConfiguredCloud>,
+}
+
+impl RestVendedCredentialProvider {
+ fn new(client: Arc<HttpClient>, plan_id: Option<String>, clouds:
Vec<ConfiguredCloud>) -> Self {
+ Self {
+ client,
+ plan_id,
+ clouds,
+ }
+ }
+
+ fn configured_cloud_for_location(&self, location: &str) ->
Option<&ConfiguredCloud> {
+ self.clouds
+ .iter()
+ .find(|configured| configured.cloud.matches_location(location))
+ }
+
+ /// Fetch fresh credentials from the catalog's credentials endpoint.
+ async fn fetch(&self, configured: &ConfiguredCloud) ->
Result<Vec<CachedEntry>> {
+ let mut request = self.client.request(Method::GET,
&configured.endpoint);
+ if let Some(plan_id) = &self.plan_id {
+ request = request.query(&[("planId", plan_id)]);
+ }
+ let request = request.build()?;
+ let response = self.client.query_catalog(request).await?;
+
+ match response.status() {
+ StatusCode::OK => {
+ let parsed: LoadCredentialsResponse = response.json().await?;
+ parsed
+ .storage_credentials
+ .into_iter()
+ .filter(|sc| configured.cloud.matches_location(&sc.prefix))
+ .map(|sc| {
+ (configured.cloud.parse_credential)(&sc.config,
Some(sc.prefix)).map(
+ |credential| {
+ CachedEntry::new(credential,
configured.cloud.jitter_prefetch)
+ },
+ )
+ })
+ .collect()
+ }
+ _ => Err(deserialize_unexpected_catalog_error(
+ response,
+ self.client.disable_header_redaction(),
+ )
+ .await),
+ }
+ }
+
+ async fn refresh_credential(
+ &self,
+ configured: &ConfiguredCloud,
+ path: &str,
+ fallback: Option<CachedEntry>,
+ ) -> Result<StorageCredential> {
+ let refreshed = self.fetch(configured).await.and_then(|entries| {
+ let credential = longest_prefix_match(&entries, path)
+ .filter(|entry| entry.is_unexpired(SystemTime::now()))
+ .map(|entry| entry.credential.clone())
+ .ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ format!("no unexpired vended credential matches
storage location: {path}"),
+ )
+ })?;
+ Ok((entries, credential))
+ });
+
+ match refreshed {
+ Ok((entries, credential)) => {
+ let mut cache = configured.cache.lock().await;
+ cache.entries = entries;
+ cache.consecutive_failures = 0;
+ cache.retry_not_before = None;
+ Ok(credential)
+ }
+ Err(fetch_error) => {
+ let mut cache = configured.cache.lock().await;
+ cache.consecutive_failures =
cache.consecutive_failures.saturating_add(1);
+ cache.retry_not_before =
+
Instant::now().checked_add(failure_backoff(cache.consecutive_failures));
+
+ // Graceful degradation: while the cached credential remains
+ // usable, serve it and retry after jittered backoff. Expired
+ // credentials are never served.
+ fallback
+ .filter(|entry| entry.is_unexpired(SystemTime::now()))
+ .map(|entry| entry.credential)
+ .ok_or(fetch_error)
+ }
+ }
Review Comment:
`crates/catalog/rest/src/credential.rs:265-300` (`refresh_credential`),
contrast with
`S3FileIO.refreshStorageCredentials()`/`GCSFileIO.refreshStorageCredentials()`
in Java.
Java's actual multi-prefix refresh (used by both `S3FileIO` and `GCSFileIO`,
not the single-credential
`VendedCredentialsProvider`/`OAuth2RefreshCredentialsHandler` used as an SDK
credentials provider) is unconditional: on each scheduled refresh it fetches
the credentials endpoint once, keeps *every* entry matching the cloud's root
prefix, and replaces `storageCredentials` wholesale — no per-path filtering
happens at refresh time at all.
The Rust `refresh_credential` instead does per-path filtering inline: it
fetches all entries for a cloud, then immediately narrows to
`longest_prefix_match(&entries, path).filter(unexpired)` for *this specific
call's* `path`. If that narrowing yields nothing (no entry covers this path, or
the covering entry happens to already be expired), the whole outcome is treated
as `Err` and:
- the freshly-fetched `entries` are never written to `cache.entries` (only
the `Ok((entries, credential))` branch at line 279-284 updates the cache) — so
if the response contained entries for other prefixes (as the existing
`longest_prefix_match_ignores_freshness` test exercises), they're thrown away
even though a subsequent call for a different, valid path would have to fetch
them all over again;
- `cache.consecutive_failures` is incremented and `retry_not_before` backoff
is armed (lines 288-290) even though the catalog responded successfully — it
just didn't vend anything for this path.
Given Java's model of "cache everything the endpoint returns,
unconditionally," was per-path filtering at refresh time (rather than only at
cache-read time, where it already happens in
`cache_decision`/`longest_prefix_match`) intentional here?
##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,1203 @@
+// 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.
+
+//! Refresh of vended storage credentials against a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache, and
+//! the returned [`StorageCredential`] enum lets the storage adapter enforce
the
+//! expected backend-specific type. This preserves Java's per-cloud refresh
+//! policy while supporting mixed-cloud tables through a resolving FileIO.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant.
To
+//! add a backend, first add its credential type to Iceberg's storage API and
+//! teach the storage adapter to consume it. Then write its `parse_*` function,
+//! add a `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+ AWS_REFRESH_CREDENTIALS_ENABLED, AWS_REFRESH_CREDENTIALS_ENDPOINT,
+ GCS_REFRESH_CREDENTIALS_ENABLED, GCS_REFRESH_CREDENTIALS_ENDPOINT,
GCS_TOKEN,
+ GCS_TOKEN_EXPIRES_AT, GcsCredential, S3_ACCESS_KEY_ID,
S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN,
+ S3_SESSION_TOKEN_EXPIRES_AT_MS, S3Credential, StorageCredential,
StorageCredentialKind,
+ StorageCredentialProvider,
+};
+use iceberg::{Error, ErrorKind, Result};
+use rand::Rng;
+use reqwest::{Method, StatusCode, Url};
+use tokio::sync::Mutex;
+
+use crate::REST_CATALOG_PROP_SCAN_PLAN_ID;
+use crate::client::{HttpClient, deserialize_unexpected_catalog_error};
+use crate::types::LoadCredentialsResponse;
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// It contains the location schemes it backs, the property keys it is
configured
+/// with, and how to parse its credential. The generic provider stays free of
+/// any per-cloud knowledge.
+struct CloudRefresh {
+ /// Location URL schemes this backend serves.
+ schemes: &'static [&'static str],
+ /// Table property naming the refresh endpoint (absolute or
catalog-relative).
+ endpoint_key: &'static str,
+ /// Table property to opt out; refresh is enabled unless this is `"false"`.
+ enabled_key: &'static str,
+ /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+ jitter_prefetch: bool,
+ /// Parse a complete credential from catalog-supplied properties.
+ parse_credential:
+ fn(config: &HashMap<String, String>, prefix: Option<String>) ->
Result<StorageCredential>,
+}
+
+impl CloudRefresh {
+ /// S3 / AWS
+ const AWS: Self = Self {
+ schemes: &["s3", "s3a", "s3n"],
+ endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+ enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+ jitter_prefetch: true,
+ parse_credential: parse_s3_credential,
+ };
+ /// Google Cloud Storage
+ const GCP: Self = Self {
+ schemes: &["gs", "gcs"],
+ endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+ enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+ jitter_prefetch: false,
+ parse_credential: parse_gcs_credential,
+ };
+ // TODO: Azure (ADLS) is not yet supported: opendal 0.57's Azdls builder
exposes no
+ // custom credential-provider hook, and reqsign's SAS-token credential has
no
+ // expiry, so reqsign-based refresh isn't possible.
+
+ /// Backends with refresh support
+ const SUPPORTED: &[Self] = &[Self::AWS, Self::GCP];
+
+ /// The backend that serves `location`, by its URL scheme, or `None` if no
+ /// supported backend matches (in which case static credentials are used
+ /// as-is, as before).
+ fn for_location(location: &str) -> Option<&'static Self> {
+ Self::SUPPORTED
+ .iter()
+ .find(|cloud| cloud.matches_location(location))
+ }
+
+ fn matches_location(&self, location: &str) -> bool {
+ scheme_of(location).is_some_and(|scheme|
self.schemes.contains(&scheme.as_str()))
+ }
+}
+
+/// Re-fetch a credential once it is within this window of expiry, so a fresh
+/// token is in hand before the object store would reject the old one.
+const REFRESH_BUFFER: Duration = Duration::from_mins(5);
+
+/// AWS keeps at least one minute between its jittered prefetch time and
expiry.
+const MIN_REFRESH_BUFFER: Duration = Duration::from_mins(1);
+
+/// Initial ceiling for failure backoff. Equal jitter chooses from half this
+/// value through the full value.
+const INITIAL_FAILURE_BACKOFF: Duration = Duration::from_secs(1);
+
+/// Maximum failure backoff while a cached credential remains usable.
+const MAX_FAILURE_BACKOFF: Duration = Duration::from_secs(30);
+
+/// A cached vended credential and its refresh schedule.
+#[derive(Clone)]
+struct CachedEntry {
+ credential: StorageCredential,
+ /// When this entry becomes eligible for prefetch. `None` means it does not
+ /// expire and therefore never needs proactive refresh.
+ refresh_at: Option<SystemTime>,
+}
+
+impl CachedEntry {
+ fn new(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+ let refresh_at = credential
+ .expires_at
+ .map(|expires_at| prefetch_time(expires_at, jitter_prefetch));
+ Self {
+ credential,
+ refresh_at,
+ }
+ }
+
+ /// Seed entries that are already inside the nominal five-minute window are
+ /// immediately due. Otherwise AWS applies the same jitter as it does to a
+ /// freshly fetched value.
+ fn seed(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+ let due = credential.expires_at.is_some_and(|expires_at| {
+ SystemTime::now()
+ .checked_add(REFRESH_BUFFER)
+ .is_none_or(|refresh_boundary| refresh_boundary >= expires_at)
+ });
+ let mut entry = Self::new(credential, jitter_prefetch);
+ if due {
+ entry.refresh_at = Some(UNIX_EPOCH);
+ }
+ entry
+ }
+
+ fn is_fresh(&self, now: SystemTime) -> bool {
+ self.refresh_at.is_none_or(|refresh_at| now < refresh_at)
+ }
+
+ fn is_unexpired(&self, now: SystemTime) -> bool {
+ self.credential
+ .expires_at
+ .is_none_or(|expires_at| now < expires_at)
+ }
+}
+
+/// Cached credentials plus failure-backoff state.
+struct CacheState {
+ entries: Vec<CachedEntry>,
+ consecutive_failures: u32,
+ retry_not_before: Option<Instant>,
+}
+
+struct ConfiguredCloud {
+ cloud: &'static CloudRefresh,
+ endpoint: String,
+ cache: Mutex<CacheState>,
+ /// Only one caller fetches at a time. The cache lock is deliberately
+ /// separate so other callers can keep using an unexpired credential while
+ /// the refresh is in flight.
+ refresh: Mutex<()>,
+}
+
+/// Fetches and refreshes vended credentials from a REST catalog's table
+/// credentials endpoint.
+///
+/// Each cloud cache is seeded with the credential from the initial table
+/// properties (when complete) and re-fetched from its endpoint as it
+/// nears expiry.
+pub(crate) struct RestVendedCredentialProvider {
+ client: Arc<HttpClient>,
+ /// Optional scan-plan identifier.
+ plan_id: Option<String>,
+ /// Independently configured endpoint and cache for each backing cloud.
+ clouds: Vec<ConfiguredCloud>,
+}
+
+impl RestVendedCredentialProvider {
+ fn new(client: Arc<HttpClient>, plan_id: Option<String>, clouds:
Vec<ConfiguredCloud>) -> Self {
+ Self {
+ client,
+ plan_id,
+ clouds,
+ }
+ }
+
+ fn configured_cloud_for_location(&self, location: &str) ->
Option<&ConfiguredCloud> {
+ self.clouds
+ .iter()
+ .find(|configured| configured.cloud.matches_location(location))
+ }
+
+ /// Fetch fresh credentials from the catalog's credentials endpoint.
+ async fn fetch(&self, configured: &ConfiguredCloud) ->
Result<Vec<CachedEntry>> {
+ let mut request = self.client.request(Method::GET,
&configured.endpoint);
+ if let Some(plan_id) = &self.plan_id {
+ request = request.query(&[("planId", plan_id)]);
+ }
+ let request = request.build()?;
+ let response = self.client.query_catalog(request).await?;
+
+ match response.status() {
+ StatusCode::OK => {
+ let parsed: LoadCredentialsResponse = response.json().await?;
+ parsed
+ .storage_credentials
+ .into_iter()
+ .filter(|sc| configured.cloud.matches_location(&sc.prefix))
+ .map(|sc| {
+ (configured.cloud.parse_credential)(&sc.config,
Some(sc.prefix)).map(
+ |credential| {
+ CachedEntry::new(credential,
configured.cloud.jitter_prefetch)
+ },
+ )
+ })
+ .collect()
Review Comment:
`collect()` into `Result<Vec<CachedEntry>>` short-circuits on the first
`parse_credential` error. If a server vends N credentials for one cloud and one
entry is missing a required field, all N become unusable (and, per finding 2,
this also counts as a "failure" for backoff purposes) rather than just the one
bad entry. Java's equivalent (`VendedCredentialsProvider.refreshCredential`)
only ever expects a single S3-prefixed entry and asserts on it directly, so
there's no directly analogous "partial batch" behavior to compare against — but
given this PR's own design supports N entries per cloud, is one bad entry meant
to invalidate all the others?
##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,107 @@ pub trait StorageFactory: Debug + Send + Sync {
/// A `Result` containing an `Arc<dyn Storage>` on success, or an error
/// if the storage could not be created.
fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>>;
+
+ /// Build a new Storage instance, optionally supplying a credential
provider
+ /// that the backend can call to obtain and refresh short-lived
credentials.
+ fn build_with_credentials(
+ &self,
+ config: &StorageConfig,
+ credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+ ) -> Result<Arc<dyn Storage>> {
+ if credential_provider.is_some() {
+ return Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "Storage factory does not support refreshable credential
providers",
+ ));
+ }
+
+ self.build(config)
+ }
+}
+
+/// Supplies fresh, backend-specific storage credentials on demand.
+///
+/// A catalog that vends temporary credentials implements this trait so that
+/// storage backends can re-fetch credentials as they approach expiry instead
+/// of failing once the initial token's TTL runs out.
+///
+/// # Caching
+///
+/// [`load_credential`](Self::load_credential) may be called very frequently —
+/// the S3 backend, for example, rebuilds its operator (and therefore its
+/// signer) on every file operation. Implementations must cache internally and
+/// only re-fetch when the current credential is at or near expiry; otherwise
+/// every object-store request would trigger a call back to the catalog.
+#[async_trait]
+pub trait StorageCredentialProvider: Debug + Send + Sync {
+ /// Return whether this provider has refresh configuration for `path`.
+ ///
+ /// Backends use this before replacing their normal credential chain. The
+ /// default is `true` for single-backend providers; multi-backend providers
+ /// should return `false` for schemes they do not configure.
+ fn supports_path(&self, _path: &str) -> bool {
+ true
+ }
+
+ /// Load a fresh credential for the storage location identified by `path`.
+ ///
+ /// `path` is the absolute location being accessed (e.g.
+ /// `s3://bucket/warehouse/db/table/...`). Providers that vend distinct
+ /// credentials per location prefix use it to select the most specific
+ /// match. When the selected credential has a declared
+ /// [`StorageCredential::prefix`], it must cover `path`.
+ async fn load_credential(&self, path: &str) -> Result<StorageCredential>;
+}
+
+/// A vended storage credential together with its scope and expiry.
+#[derive(Clone, Debug)]
+pub struct StorageCredential {
+ /// Storage-location prefix this credential is scoped to. `None`
represents a
+ /// credential without a declared scope, sourced from flat storage
properties.
+ pub prefix: Option<String>,
+ /// The backend-specific credential material.
+ pub kind: StorageCredentialKind,
+ /// When the credential expires, if known. `None` means non-expiring and
+ /// backends treat such a credential as always valid and never refresh it.
+ pub expires_at: Option<SystemTime>,
+}
+
+/// Backend-specific credential material.
+#[derive(Clone, Debug)]
+pub enum StorageCredentialKind {
+ /// Amazon S3 credentials.
+ S3(S3Credential),
+ /// Google Cloud Storage credentials.
+ Gcs(GcsCredential),
+}
+
+/// Temporary Amazon S3 credentials.
+#[derive(Clone)]
+pub struct S3Credential {
+ /// AWS access key ID.
+ pub access_key_id: String,
+ /// AWS secret access key.
+ pub secret_access_key: String,
+ /// AWS session token, set for temporary (STS/vended) credentials.
+ pub session_token: Option<String>,
+}
+
+impl Debug for S3Credential {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("S3Credential").finish_non_exhaustive()
+ }
+}
+
+/// Temporary Google Cloud Storage credentials (an OAuth2 access token).
+#[derive(Clone)]
+pub struct GcsCredential {
+ /// OAuth2 bearer token used to access GCS.
+ pub token: String,
+}
Review Comment:
These are new public types (in `public-api.txt`) that third-party
`StorageCredentialProvider` implementors must construct by hand. Every field is
`pub`, with no constructor. That's inconsistent with `StorageConfig` in the
same module (`crates/iceberg/src/io/storage/config/mod.rs:55-58`), which keeps
`props` private and exposes `with_prop`/`from_props` instead. Was a constructor
considered, or is direct struct-literal construction intentional here?
##########
crates/catalog/rest/src/client.rs:
##########
@@ -278,29 +298,30 @@ pub(crate) async fn deserialize_catalog_response<R:
DeserializeOwned>(
let bytes = response.bytes().await?;
serde_json::from_slice::<R>(&bytes).map_err(|e| {
+ // Successful REST responses can contain OAuth tokens and delegated
+ // storage credentials. Never copy an unparsable response into an
error.
Error::new(
ErrorKind::Unexpected,
"Failed to parse response from rest catalog server",
)
- .with_context("json", String::from_utf8_lossy(&bytes))
.with_source(e)
})
}
-/// Headers that contain sensitive information and should be excluded from
logs.
-const SENSITIVE_HEADERS: &[&str] = &[
- "authorization",
- "proxy-authorization",
- "set-cookie",
- "cookie",
- "x-api-key",
- "x-auth-token",
-];
-
-/// Returns true if the header name is considered sensitive.
+/// Returns true if the header may carry a secret.
fn is_sensitive_header(name: &str) -> bool {
let name_lower = name.to_lowercase();
- SENSITIVE_HEADERS.iter().any(|h| name_lower == *h)
+ [
+ "auth",
+ "token",
+ "secret",
+ "key",
+ "password",
+ "cookie",
+ "credential",
+ ]
Review Comment:
Given your own note that Java redacts every header regardless of
sensitivity, that seems like the simpler and more conservative choice here too.
It avoids maintaining a keyword list that can both over-match (a header like
`x-auth-region` gets redacted for no reason) and under-match (a secret header
whose name does not contain any of the chosen substrings). Suggest switching to
redact-everything-by-default unless there is a concrete case where seeing an
unredacted non-sensitive header in error logs matters.
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -553,6 +611,12 @@ impl Storage for OpenDalStorage {
let mut deleters: HashMap<String, opendal::Deleter> = HashMap::new();
while let Some(path) = paths.next().await {
+ if self.uses_dynamic_credentials(&path) {
+ let (op, relative_path) = self.create_operator(&path)?;
+ op.delete(relative_path).await.map_err(from_opendal_error)?;
+ continue;
+ }
Review Comment:
Agree this should be fixed in this PR rather than deferred. See finding 1
above for the specific fix and the added detail on `create_operator` also being
rebuilt per file.
##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,1193 @@
+// 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.
+
+//! Refresh of vended storage credentials against a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache, and
+//! the returned [`StorageCredential`] enum lets the storage adapter enforce
the
+//! expected backend-specific type. This preserves Java's per-cloud refresh
+//! policy while supporting mixed-cloud tables through a resolving FileIO.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant.
To
+//! add a backend, first add its credential type to Iceberg's storage API and
+//! teach the storage adapter to consume it. Then write its `parse_*` function,
+//! add a `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+ AWS_REFRESH_CREDENTIALS_ENABLED, AWS_REFRESH_CREDENTIALS_ENDPOINT,
+ GCS_REFRESH_CREDENTIALS_ENABLED, GCS_REFRESH_CREDENTIALS_ENDPOINT,
GCS_TOKEN,
+ GCS_TOKEN_EXPIRES_AT, GcsCredential, S3_ACCESS_KEY_ID,
S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN,
+ S3_SESSION_TOKEN_EXPIRES_AT_MS, S3Credential, StorageCredential,
StorageCredentialKind,
+ StorageCredentialProvider,
+};
+use iceberg::{Error, ErrorKind, Result};
+use rand::Rng;
+use reqwest::{Method, StatusCode, Url};
+use tokio::sync::Mutex;
+
+use crate::REST_CATALOG_PROP_SCAN_PLAN_ID;
+use crate::client::{HttpClient, deserialize_unexpected_catalog_error};
+use crate::types::LoadCredentialsResponse;
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// It contains the location schemes it backs, the property keys it is
configured
+/// with, and how to parse its credential. The generic provider stays free of
+/// any per-cloud knowledge.
+struct CloudRefresh {
+ /// Location URL schemes this backend serves.
+ schemes: &'static [&'static str],
+ /// Table property naming the refresh endpoint (absolute or
catalog-relative).
+ endpoint_key: &'static str,
+ /// Table property to opt out; refresh is enabled unless this is `"false"`.
+ enabled_key: &'static str,
+ /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+ jitter_prefetch: bool,
+ /// Parse a complete credential from catalog-supplied properties.
+ parse_credential: fn(&HashMap<String, String>) ->
Result<StorageCredential>,
+}
+
+impl CloudRefresh {
+ /// S3 / AWS
+ const AWS: Self = Self {
+ schemes: &["s3", "s3a", "s3n"],
+ endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+ enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+ jitter_prefetch: true,
+ parse_credential: parse_s3_credential,
+ };
+ /// Google Cloud Storage
+ const GCP: Self = Self {
+ schemes: &["gs", "gcs"],
+ endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+ enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+ jitter_prefetch: false,
Review Comment:
Checked this against Java directly. Java's scheduled refresh
(`S3FileIO`/`GCSFileIO.refreshStorageCredentials()`) never retries after a
failed fetch at all, it logs a warning and background refresh permanently stops
for that `FileIO` instance until something else rebuilds the client map. The
jittered backoff-and-retry here is already strictly better than Java's
behavior, not something that needs to move toward parity. Might be worth saying
so explicitly in the module doc comment, since it currently frames the design
as "preserving Java's per-cloud refresh policy," and a reader comparing against
Java will notice the retry model does not match 1:1.
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -541,9 +556,20 @@ impl RestCatalog {
)
})?;
- let file_io = FileIOBuilder::new(factory).with_props(props).build();
+ // If the catalog vends refreshable credentials for this table's
storage,
+ // attach a provider so the backend re-fetches them before they expire.
+ let credential_provider =
crate::credential::build_vended_credential_provider(
Review Comment:
The file already has a `use crate::client::{...}` / `use
crate::types::{...}` block at the top; this call could join it as `use
crate::credential::build_vended_credential_provider;` instead of qualifying the
path inline at the call site.
##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,103 @@ pub trait StorageFactory: Debug + Send + Sync {
/// A `Result` containing an `Arc<dyn Storage>` on success, or an error
/// if the storage could not be created.
fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>>;
+
+ /// Build a new Storage instance, optionally supplying a credential
provider
+ /// that the backend can call to obtain and refresh short-lived
credentials.
+ fn build_with_credentials(
+ &self,
+ config: &StorageConfig,
+ credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+ ) -> Result<Arc<dyn Storage>> {
+ if credential_provider.is_some() {
+ return Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "Storage factory does not support refreshable credential
providers",
+ ));
+ }
+
+ self.build(config)
+ }
+}
+
+/// Supplies fresh, backend-specific storage credentials on demand.
+///
+/// A catalog that vends temporary credentials implements this trait so that
+/// storage backends can re-fetch credentials as they approach expiry instead
+/// of failing once the initial token's TTL runs out.
+///
+/// # Caching
+///
+/// [`load_credential`](Self::load_credential) may be called very frequently —
+/// the S3 backend, for example, rebuilds its operator (and therefore its
+/// signer) on every file operation. Implementations must cache internally and
+/// only re-fetch when the current credential is at or near expiry; otherwise
+/// every object-store request would trigger a call back to the catalog.
+#[async_trait]
+pub trait StorageCredentialProvider: Debug + Send + Sync {
Review Comment:
The trait-based design seems fine, Java not having an equivalent is not a
reason to avoid one here. See finding 5 above for the separate, concrete
question on the same code (pub fields, no constructor).
##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,1203 @@
+// 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.
+
+//! Refresh of vended storage credentials against a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache, and
+//! the returned [`StorageCredential`] enum lets the storage adapter enforce
the
+//! expected backend-specific type. This preserves Java's per-cloud refresh
+//! policy while supporting mixed-cloud tables through a resolving FileIO.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant.
To
+//! add a backend, first add its credential type to Iceberg's storage API and
+//! teach the storage adapter to consume it. Then write its `parse_*` function,
+//! add a `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+ AWS_REFRESH_CREDENTIALS_ENABLED, AWS_REFRESH_CREDENTIALS_ENDPOINT,
+ GCS_REFRESH_CREDENTIALS_ENABLED, GCS_REFRESH_CREDENTIALS_ENDPOINT,
GCS_TOKEN,
+ GCS_TOKEN_EXPIRES_AT, GcsCredential, S3_ACCESS_KEY_ID,
S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN,
+ S3_SESSION_TOKEN_EXPIRES_AT_MS, S3Credential, StorageCredential,
StorageCredentialKind,
+ StorageCredentialProvider,
+};
+use iceberg::{Error, ErrorKind, Result};
+use rand::Rng;
+use reqwest::{Method, StatusCode, Url};
+use tokio::sync::Mutex;
+
+use crate::REST_CATALOG_PROP_SCAN_PLAN_ID;
+use crate::client::{HttpClient, deserialize_unexpected_catalog_error};
+use crate::types::LoadCredentialsResponse;
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// It contains the location schemes it backs, the property keys it is
configured
+/// with, and how to parse its credential. The generic provider stays free of
+/// any per-cloud knowledge.
+struct CloudRefresh {
+ /// Location URL schemes this backend serves.
+ schemes: &'static [&'static str],
+ /// Table property naming the refresh endpoint (absolute or
catalog-relative).
+ endpoint_key: &'static str,
+ /// Table property to opt out; refresh is enabled unless this is `"false"`.
+ enabled_key: &'static str,
+ /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+ jitter_prefetch: bool,
+ /// Parse a complete credential from catalog-supplied properties.
+ parse_credential:
+ fn(config: &HashMap<String, String>, prefix: Option<String>) ->
Result<StorageCredential>,
+}
+
+impl CloudRefresh {
+ /// S3 / AWS
+ const AWS: Self = Self {
+ schemes: &["s3", "s3a", "s3n"],
+ endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+ enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+ jitter_prefetch: true,
+ parse_credential: parse_s3_credential,
+ };
+ /// Google Cloud Storage
+ const GCP: Self = Self {
+ schemes: &["gs", "gcs"],
+ endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+ enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+ jitter_prefetch: false,
+ parse_credential: parse_gcs_credential,
+ };
+ // TODO: Azure (ADLS) is not yet supported: opendal 0.57's Azdls builder
exposes no
+ // custom credential-provider hook, and reqsign's SAS-token credential has
no
+ // expiry, so reqsign-based refresh isn't possible.
+
+ /// Backends with refresh support
+ const SUPPORTED: &[Self] = &[Self::AWS, Self::GCP];
+
+ /// The backend that serves `location`, by its URL scheme, or `None` if no
+ /// supported backend matches (in which case static credentials are used
+ /// as-is, as before).
+ fn for_location(location: &str) -> Option<&'static Self> {
+ Self::SUPPORTED
+ .iter()
+ .find(|cloud| cloud.matches_location(location))
+ }
+
+ fn matches_location(&self, location: &str) -> bool {
+ scheme_of(location).is_some_and(|scheme|
self.schemes.contains(&scheme.as_str()))
+ }
+}
+
+/// Re-fetch a credential once it is within this window of expiry, so a fresh
+/// token is in hand before the object store would reject the old one.
+const REFRESH_BUFFER: Duration = Duration::from_mins(5);
+
+/// AWS keeps at least one minute between its jittered prefetch time and
expiry.
+const MIN_REFRESH_BUFFER: Duration = Duration::from_mins(1);
+
+/// Initial ceiling for failure backoff. Equal jitter chooses from half this
+/// value through the full value.
+const INITIAL_FAILURE_BACKOFF: Duration = Duration::from_secs(1);
+
+/// Maximum failure backoff while a cached credential remains usable.
+const MAX_FAILURE_BACKOFF: Duration = Duration::from_secs(30);
+
+/// A cached vended credential and its refresh schedule.
+#[derive(Clone)]
+struct CachedEntry {
+ credential: StorageCredential,
+ /// When this entry becomes eligible for prefetch. `None` means it does not
+ /// expire and therefore never needs proactive refresh.
+ refresh_at: Option<SystemTime>,
+}
+
+impl CachedEntry {
+ fn new(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+ let refresh_at = credential
+ .expires_at
+ .map(|expires_at| prefetch_time(expires_at, jitter_prefetch));
+ Self {
+ credential,
+ refresh_at,
+ }
+ }
+
+ /// Seed entries that are already inside the nominal five-minute window are
+ /// immediately due. Otherwise AWS applies the same jitter as it does to a
+ /// freshly fetched value.
+ fn seed(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+ let due = credential.expires_at.is_some_and(|expires_at| {
+ SystemTime::now()
+ .checked_add(REFRESH_BUFFER)
+ .is_none_or(|refresh_boundary| refresh_boundary >= expires_at)
+ });
+ let mut entry = Self::new(credential, jitter_prefetch);
+ if due {
+ entry.refresh_at = Some(UNIX_EPOCH);
+ }
+ entry
+ }
+
+ fn is_fresh(&self, now: SystemTime) -> bool {
+ self.refresh_at.is_none_or(|refresh_at| now < refresh_at)
+ }
+
+ fn is_unexpired(&self, now: SystemTime) -> bool {
+ self.credential
+ .expires_at
+ .is_none_or(|expires_at| now < expires_at)
+ }
+}
+
+/// Cached credentials plus failure-backoff state.
+struct CacheState {
+ entries: Vec<CachedEntry>,
+ consecutive_failures: u32,
+ retry_not_before: Option<Instant>,
+}
+
+struct ConfiguredCloud {
+ cloud: &'static CloudRefresh,
+ endpoint: String,
+ cache: Mutex<CacheState>,
+ /// Only one caller fetches at a time. The cache lock is deliberately
+ /// separate so other callers can keep using an unexpired credential while
+ /// the refresh is in flight.
+ refresh: Mutex<()>,
+}
+
+/// Fetches and refreshes vended credentials from a REST catalog's table
+/// credentials endpoint.
+///
+/// Each cloud cache is seeded with the credential from the initial table
+/// properties (when complete) and re-fetched from its endpoint as it
+/// nears expiry.
+pub(crate) struct RestVendedCredentialProvider {
+ client: Arc<HttpClient>,
+ /// Optional scan-plan identifier.
+ plan_id: Option<String>,
+ /// Independently configured endpoint and cache for each backing cloud.
+ clouds: Vec<ConfiguredCloud>,
+}
+
+impl RestVendedCredentialProvider {
+ fn new(client: Arc<HttpClient>, plan_id: Option<String>, clouds:
Vec<ConfiguredCloud>) -> Self {
+ Self {
+ client,
+ plan_id,
+ clouds,
+ }
+ }
+
+ fn configured_cloud_for_location(&self, location: &str) ->
Option<&ConfiguredCloud> {
+ self.clouds
+ .iter()
+ .find(|configured| configured.cloud.matches_location(location))
+ }
+
+ /// Fetch fresh credentials from the catalog's credentials endpoint.
+ async fn fetch(&self, configured: &ConfiguredCloud) ->
Result<Vec<CachedEntry>> {
+ let mut request = self.client.request(Method::GET,
&configured.endpoint);
+ if let Some(plan_id) = &self.plan_id {
+ request = request.query(&[("planId", plan_id)]);
+ }
+ let request = request.build()?;
+ let response = self.client.query_catalog(request).await?;
+
+ match response.status() {
+ StatusCode::OK => {
+ let parsed: LoadCredentialsResponse = response.json().await?;
+ parsed
+ .storage_credentials
+ .into_iter()
+ .filter(|sc| configured.cloud.matches_location(&sc.prefix))
+ .map(|sc| {
+ (configured.cloud.parse_credential)(&sc.config,
Some(sc.prefix)).map(
+ |credential| {
+ CachedEntry::new(credential,
configured.cloud.jitter_prefetch)
+ },
+ )
+ })
+ .collect()
+ }
+ _ => Err(deserialize_unexpected_catalog_error(
+ response,
+ self.client.disable_header_redaction(),
+ )
+ .await),
+ }
+ }
+
+ async fn refresh_credential(
+ &self,
+ configured: &ConfiguredCloud,
+ path: &str,
+ fallback: Option<CachedEntry>,
+ ) -> Result<StorageCredential> {
+ let refreshed = self.fetch(configured).await.and_then(|entries| {
+ let credential = longest_prefix_match(&entries, path)
+ .filter(|entry| entry.is_unexpired(SystemTime::now()))
+ .map(|entry| entry.credential.clone())
+ .ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ format!("no unexpired vended credential matches
storage location: {path}"),
+ )
+ })?;
+ Ok((entries, credential))
+ });
+
+ match refreshed {
+ Ok((entries, credential)) => {
+ let mut cache = configured.cache.lock().await;
+ cache.entries = entries;
+ cache.consecutive_failures = 0;
+ cache.retry_not_before = None;
+ Ok(credential)
+ }
+ Err(fetch_error) => {
+ let mut cache = configured.cache.lock().await;
+ cache.consecutive_failures =
cache.consecutive_failures.saturating_add(1);
+ cache.retry_not_before =
+
Instant::now().checked_add(failure_backoff(cache.consecutive_failures));
+
+ // Graceful degradation: while the cached credential remains
+ // usable, serve it and retry after jittered backoff. Expired
+ // credentials are never served.
+ fallback
+ .filter(|entry| entry.is_unexpired(SystemTime::now()))
+ .map(|entry| entry.credential)
+ .ok_or(fetch_error)
+ }
+ }
+ }
+}
+
+impl std::fmt::Debug for RestVendedCredentialProvider {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("RestVendedCredentialProvider")
+ .field("configured_clouds", &self.clouds.len())
+ .finish_non_exhaustive()
+ }
+}
+
+enum CacheDecision {
+ Use(StorageCredential),
+ Refresh(Option<CachedEntry>),
+}
+
+async fn cache_decision(configured: &ConfiguredCloud, path: &str) ->
CacheDecision {
+ let cache = configured.cache.lock().await;
+ let current = longest_prefix_match(&cache.entries, path).cloned();
+ let now = SystemTime::now();
+
+ if let Some(entry) = current.as_ref().filter(|entry| entry.is_fresh(now)) {
+ return CacheDecision::Use(entry.credential.clone());
+ }
+
+ if cache
+ .retry_not_before
+ .is_some_and(|retry_at| Instant::now() < retry_at)
+ && let Some(entry) = current.as_ref().filter(|entry|
entry.is_unexpired(now))
+ {
+ return CacheDecision::Use(entry.credential.clone());
+ }
+
+ CacheDecision::Refresh(current)
+}
+
+#[async_trait]
+impl StorageCredentialProvider for RestVendedCredentialProvider {
+ fn supports_path(&self, path: &str) -> bool {
+ self.configured_cloud_for_location(path).is_some()
+ }
+
+ async fn load_credential(&self, path: &str) -> Result<StorageCredential> {
+ if CloudRefresh::for_location(path).is_none() {
+ return Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ format!("no credential refresh implementation for storage
location: {path}"),
+ ));
+ }
+ let configured =
self.configured_cloud_for_location(path).ok_or_else(|| {
+ Error::new(
+ ErrorKind::FeatureUnsupported,
+ format!("credential refresh is not configured for storage
location: {path}"),
+ )
+ })?;
+
+ let current = match cache_decision(configured, path).await {
+ CacheDecision::Use(credential) => return Ok(credential),
+ CacheDecision::Refresh(current) => current,
+ };
+
+ // One caller refreshes, while concurrent callers immediately keep
using the
+ // unexpired cached credential. With no usable credential, callers wait
+ // for the in-flight refresh instead.
+ let usable = current
+ .as_ref()
+ .filter(|entry| entry.is_unexpired(SystemTime::now()));
+ let _refresh_guard = if let Some(entry) = usable {
+ match configured.refresh.try_lock() {
+ Ok(guard) => guard,
+ Err(_) => return Ok(entry.credential.clone()),
+ }
+ } else {
+ configured.refresh.lock().await
+ };
+
+ // Another caller may have completed a refresh between our cache check
+ // and acquiring the single-flight guard.
+ let current = match cache_decision(configured, path).await {
+ CacheDecision::Use(credential) => return Ok(credential),
+ CacheDecision::Refresh(current) => current,
+ };
+
+ self.refresh_credential(configured, path, current).await
+ }
+}
+
+/// Select the credential whose prefix is the longest match for `path`.
+fn longest_prefix_match<'a>(entries: &'a [CachedEntry], path: &str) ->
Option<&'a CachedEntry> {
+ entries
+ .iter()
+ .filter(|entry| {
+ entry
+ .credential
+ .prefix
+ .as_deref()
+ .is_none_or(|prefix| path.starts_with(prefix))
+ })
+ .max_by_key(|entry| entry.credential.prefix.as_deref().map_or(0,
str::len))
+}
+
+/// Compute a successful credential's prefetch time.
+fn prefetch_time(expires_at: SystemTime, jitter: bool) -> SystemTime {
+ let base = expires_at.checked_sub(REFRESH_BUFFER).unwrap_or(UNIX_EPOCH);
+ if !jitter {
+ return base;
+ }
+
+ let jitter_window = REFRESH_BUFFER.saturating_sub(MIN_REFRESH_BUFFER);
+ let jitter_millis = rand::rng().random_range(0..jitter_window.as_millis()
as u64);
+ base.checked_add(Duration::from_millis(jitter_millis))
+ .unwrap_or(base)
+}
+
+/// Equal-jitter exponential backoff. The random lower half avoids both hot
+/// retry loops and synchronized retries across clients.
+fn failure_backoff(consecutive_failures: u32) -> Duration {
+ let exponent = consecutive_failures.saturating_sub(1).min(5);
+ let ceiling = INITIAL_FAILURE_BACKOFF
+ .checked_mul(1 << exponent)
+ .unwrap_or(MAX_FAILURE_BACKOFF)
+ .min(MAX_FAILURE_BACKOFF);
+ let ceiling_millis = ceiling.as_millis() as u64;
+ let floor_millis = ceiling_millis / 2;
+
Duration::from_millis(rand::rng().random_range(floor_millis..=ceiling_millis))
+}
+
+/// Build a credential provider from a table's properties,
+/// or `None` when no supported cloud advertises an enabled refresh endpoint.
+///
+/// `base_uri` is the catalog URI, used to resolve a relative endpoint.
+pub(crate) fn build_vended_credential_provider(
+ client: Arc<HttpClient>,
+ base_uri: &str,
+ props: &HashMap<String, String>,
+) -> Result<Option<Arc<dyn StorageCredentialProvider>>> {
+ let clouds = CloudRefresh::SUPPORTED
+ .iter()
+ .filter_map(|cloud| {
+ // Refresh is enabled by default and invalid booleans disable
refresh.
+ let enabled = props
+ .get(cloud.enabled_key)
+ .is_none_or(|value| value.parse().unwrap_or(false));
Review Comment:
`str::parse::<bool>()` only accepts the exact strings `"true"`/`"false"`.
Java's `PropertyUtil.propertyAsBoolean` uses `Boolean.parseBoolean`, which is
case-insensitive for `"true"` (`"True"`, `"TRUE"` all parse as `true`). A
config value of `client.refresh-credentials-enabled: "True"` would enable
refresh in Java but silently disable it in Rust (parse error →
`unwrap_or(false)`). Low severity (fails closed either way), but worth a
case-insensitive comparison to match Java's actual accepted input space.
##########
crates/catalog/rest/src/client.rs:
##########
@@ -71,6 +75,24 @@ impl HttpClient {
})
}
+ /// Create a client for table-scoped resources while reusing this client's
+ /// underlying connection pool.
+ ///
+ /// A load-table response may supply a table token or `header.*` values
that
+ /// must be used for subsequent table requests such as credential refresh.
+ pub(crate) fn for_table(
+ &self,
+ catalog_uri: &str,
+ props: HashMap<String, String>,
+ ) -> Result<Self> {
+ let cfg = RestCatalogConfig::builder()
+ .uri(catalog_uri.to_string())
+ .props(props)
+ .client(Some(self.client.clone()))
+ .build();
+ Self::new(&cfg)
+ }
Review Comment:
`for_table` builds a brand-new `HttpClient` via `HttpClient::new(&cfg)`,
which starts `token: Mutex::new(cfg.token())`. `cfg.token()` reads the `token`
property directly — it does not inherit the parent client's already-cached
bearer token. When the catalog authenticates via OAuth client-credentials
(`credential` prop, no static `token` prop), the table-scoped client has no
seeded token and calls `exchange_credential_for_token()` itself on first use.
Since `build_vended_credential_provider` (and therefore `for_table`) is
constructed fresh inside `load_file_io`, and `load_file_io` runs on every
`load_table`/`create_table`/`register_table`, every one of those calls does its
own OAuth token exchange, independent of whatever token the parent
`RestContext.client` already holds. For a query engine that reloads tables
frequently, that's an extra round trip to the OAuth endpoint per load. Is
reusing the parent's cached token (when the table doesn't vend its own) worth
doing here?
--
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]
