blackmwk commented on code in PR #3106:
URL: https://github.com/apache/iceberg-rust/pull/3106#discussion_r4409707756
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -181,72 +184,151 @@ impl RestCatalogBuilder {
}
}
-/// Rest catalog configuration.
-#[derive(Clone, TypedBuilder)]
-pub(crate) struct RestCatalogConfig {
- #[builder(default, setter(strip_option))]
- name: Option<String>,
-
+/// Typed REST catalog properties.
+#[derive(Clone, Properties, PartialEq, Eq)]
+pub(crate) struct RestCatalogProperties {
+ /// Catalog URI.
+ #[property(key = REST_CATALOG_PROP_URI, default = "", getter)]
uri: String,
-
- #[builder(default, setter(strip_option(fallback = warehouse_opt)))]
+ /// Warehouse location or identifier.
+ #[property(key = REST_CATALOG_PROP_WAREHOUSE, default = None, getter)]
warehouse: Option<String>,
+ /// Catalog endpoint prefix.
+ #[property(key = "prefix", default = None, getter)]
Review Comment:
We also need a constant for the "prefix" key.
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -181,72 +184,151 @@ impl RestCatalogBuilder {
}
}
-/// Rest catalog configuration.
-#[derive(Clone, TypedBuilder)]
-pub(crate) struct RestCatalogConfig {
- #[builder(default, setter(strip_option))]
- name: Option<String>,
-
+/// Typed REST catalog properties.
+#[derive(Clone, Properties, PartialEq, Eq)]
Review Comment:
Why you need to derive PartialEq and Eq? Remove them if not reqquired.
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -181,72 +184,151 @@ impl RestCatalogBuilder {
}
}
-/// Rest catalog configuration.
-#[derive(Clone, TypedBuilder)]
-pub(crate) struct RestCatalogConfig {
- #[builder(default, setter(strip_option))]
- name: Option<String>,
-
+/// Typed REST catalog properties.
+#[derive(Clone, Properties, PartialEq, Eq)]
+pub(crate) struct RestCatalogProperties {
+ /// Catalog URI.
+ #[property(key = REST_CATALOG_PROP_URI, default = "", getter)]
uri: String,
-
- #[builder(default, setter(strip_option(fallback = warehouse_opt)))]
+ /// Warehouse location or identifier.
+ #[property(key = REST_CATALOG_PROP_WAREHOUSE, default = None, getter)]
warehouse: Option<String>,
+ /// Catalog endpoint prefix.
+ #[property(key = "prefix", default = None, getter)]
+ prefix: Option<String>,
+ /// Explicit authentication scheme.
+ #[property(key = REST_CATALOG_PROP_AUTH_TYPE, default = None, getter)]
+ auth_type: Option<AuthType>,
+ /// Configured bearer token.
+ #[property(key = "token", default = None, parse_with =
parse_sensitive_string, getter)]
Review Comment:
Ditto.
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -30,17 +30,20 @@ use iceberg::encryption::kms::{KeyManagementClient,
KmsClientFactory};
use iceberg::io::{
ADLS_SAS_TOKEN, FileIO, FileIOBuilder, GCS_TOKEN, S3_SESSION_TOKEN,
StorageFactory,
};
+use iceberg::sensitive::SensitiveString;
use iceberg::table::Table;
use iceberg::{
Catalog, CatalogBuilder, Error, ErrorKind, Namespace, NamespaceIdent,
Result, Runtime,
SessionCatalog, SessionContext, TableCommit, TableCreation, TableIdent,
};
+use iceberg_property_macro::Properties;
use itertools::Itertools;
use reqwest::header::{
HeaderMap, HeaderName, HeaderValue, {self},
};
use reqwest::{Client, Method, StatusCode, Url};
use tokio::sync::OnceCell;
+#[cfg(test)]
Review Comment:
If this is not used in prod code, move it to test mod.
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -181,72 +184,151 @@ impl RestCatalogBuilder {
}
}
-/// Rest catalog configuration.
-#[derive(Clone, TypedBuilder)]
-pub(crate) struct RestCatalogConfig {
- #[builder(default, setter(strip_option))]
- name: Option<String>,
-
+/// Typed REST catalog properties.
+#[derive(Clone, Properties, PartialEq, Eq)]
+pub(crate) struct RestCatalogProperties {
+ /// Catalog URI.
+ #[property(key = REST_CATALOG_PROP_URI, default = "", getter)]
uri: String,
-
- #[builder(default, setter(strip_option(fallback = warehouse_opt)))]
+ /// Warehouse location or identifier.
+ #[property(key = REST_CATALOG_PROP_WAREHOUSE, default = None, getter)]
warehouse: Option<String>,
+ /// Catalog endpoint prefix.
+ #[property(key = "prefix", default = None, getter)]
+ prefix: Option<String>,
+ /// Explicit authentication scheme.
+ #[property(key = REST_CATALOG_PROP_AUTH_TYPE, default = None, getter)]
+ auth_type: Option<AuthType>,
+ /// Configured bearer token.
+ #[property(key = "token", default = None, parse_with =
parse_sensitive_string, getter)]
+ token: Option<SensitiveString>,
+ /// Configured OAuth2 client ID and secret.
+ #[property(key = "credential", default = None, parse_with =
parse_credential, getter)]
Review Comment:
All keys should have a constant.
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -316,204 +372,152 @@ impl RestCatalogConfig {
])
}
- /// The HTTP client: the configured one, or a lazily-created default that
- /// is shared across every user of this config (and its clones), so token
- /// and catalog requests keep sharing one connection pool.
- pub(crate) fn client(&self) -> Client {
- self.client
- .clone()
- .unwrap_or_else(||
self.default_client.get_or_init(Client::default).clone())
- }
-
- /// Get the token from the config.
- ///
- /// The client can use this token to send requests.
- pub(crate) fn token(&self) -> Option<String> {
- self.props.get("token").cloned()
- }
-
- /// Get the credentials from the config. The client can use these
credentials to fetch a new
- /// token.
- pub(crate) fn credential(&self) -> Option<(Option<String>, String)> {
- credential_from_props(&self.props)
- }
-
- /// Get the extra headers from config, see [`extra_headers_from_props`].
- pub(crate) fn extra_headers(&self) -> Result<HeaderMap> {
- extra_headers_from_props(&self.props)
+ pub(crate) fn extra_headers(&self) -> HeaderMap {
+ let mut headers = HeaderMap::from_iter([
+ (
+ header::CONTENT_TYPE,
+ HeaderValue::from_static("application/json"),
+ ),
+ (
+ HeaderName::from_static("x-client-version"),
+ HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION),
+ ),
+ (
+ header::USER_AGENT,
+
HeaderValue::from_str(&format!("iceberg-rs/{CARGO_PKG_VERSION}")).unwrap(),
+ ),
+ ]);
+ headers.extend(self.headers.0.clone());
+ headers
}
- /// Get the optional OAuth headers from the config.
pub(crate) fn extra_oauth_params(&self) -> HashMap<String, String> {
- oauth_params_from_props(&self.props)
- }
-
- /// Check if header redaction is disabled in error logs.
- ///
- /// Returns true if the `disable-header-redaction` property is set to
"true".
- /// Defaults to false for security (headers are redacted by default).
- pub(crate) fn disable_header_redaction(&self) -> bool {
- self.props
- .get(REST_CATALOG_PROP_DISABLE_HEADER_REDACTION)
- .map(|v| v.eq_ignore_ascii_case("true"))
- .unwrap_or(false)
- }
-
- /// Builds the table cache for freshness-aware loading, or `None` when
- /// `rest-table-cache.max-entries` is 0.
- fn table_cache(&self) -> Result<Option<TableCache>> {
- let parse = |key: &str, default: u64| -> Result<u64> {
- self.props.get(key).map_or(Ok(default), |v| {
- v.parse().map_err(|e| {
- Error::new(ErrorKind::DataInvalid, format!("Invalid {key}:
{v}")).with_source(e)
- })
- })
- };
- let expire_after_write_ms = parse(
- REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS,
- TABLE_CACHE_EXPIRE_AFTER_WRITE_MS_DEFAULT,
- )?;
- let expire_after_write = Duration::from_millis(expire_after_write_ms);
- // Zero is ambiguous: it could mean "disable" or "never expire".
- if expire_after_write.is_zero() || expire_after_write >
TABLE_CACHE_EXPIRE_AFTER_WRITE_MAX {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid
{REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS}: {expire_after_write_ms}
(must be between 1 ms and 1000 years)"
- ),
- ));
- }
- let max_entries = parse(
- REST_CATALOG_PROP_TABLE_CACHE_MAX_ENTRIES,
- TABLE_CACHE_MAX_ENTRIES_DEFAULT,
- )?;
- Ok((max_entries > 0).then(|| {
+ let mut params = self.explicit_oauth_params();
+ params
+ .entry("scope".to_string())
+ .or_insert_with(|| "catalog".to_string());
+ params
+ }
+
+ pub(crate) fn explicit_oauth_params(&self) -> HashMap<String, String> {
+ [
+ ("scope", self.scope()),
+ ("audience", self.audience()),
+ ("resource", self.resource()),
+ ]
+ .into_iter()
+ .filter_map(|(key, value)| value.as_ref().map(|value|
(key.to_string(), value.clone())))
+ .collect()
+ }
+
+ fn table_cache(&self) -> Option<TableCache> {
+ (self.table_cache_max_entries() > 0).then(|| {
TableCache::builder()
- .max_capacity(max_entries)
- .time_to_live(expire_after_write)
+ .max_capacity(self.table_cache_max_entries())
+ .time_to_live(*self.table_cache_expire_after_write())
.build()
- }))
+ })
}
- /// Merge the `RestCatalogConfig` with the a [`CatalogConfig`] (fetched
from the REST server).
- pub(crate) fn merge_with_config(mut self, mut config: CatalogConfig) ->
Self {
- if let Some(uri) = config.overrides.remove(REST_CATALOG_PROP_URI) {
- self.uri = uri;
- }
-
- let mut props = config.defaults;
- props.extend(self.props);
- // The builder moved the client warehouse off the props; restore it
- // between defaults and overrides (default < client < override).
- if let Some(warehouse) = &self.warehouse {
- props.insert(REST_CATALOG_PROP_WAREHOUSE.to_string(),
warehouse.clone());
- }
- props.extend(config.overrides);
-
- self.props = props;
- self
+ fn resolved_auth_type(&self) -> &AuthType {
+ self.auth_type().as_ref().unwrap_or_else(|| {
+ if self.token.is_some() || self.credential.is_some() ||
self.oauth2_server_uri.is_some()
+ {
+ &AuthType::OAuth2
+ } else {
+ &AuthType::None
+ }
+ })
}
}
-/// Parses the `credential` property.
-///
-/// ## Output
-///
-/// - `None`: No credential is set.
-/// - `Some(None, client_secret)`: No client_id is set, use client_secret
directly.
-/// - `Some(Some(client_id), client_secret)`: Both client_id and client_secret
are set.
-pub(crate) fn credential_from_props(
- props: &HashMap<String, String>,
-) -> Option<(Option<String>, String)> {
- let cred = props.get("credential")?;
-
- match cred.split_once(':') {
- Some((client_id, client_secret)) => {
- Some((Some(client_id.to_string()), client_secret.to_string()))
- }
- None => Some((None, cred.to_string())),
- }
+fn parse_sensitive_string(value: &str) -> Result<SensitiveString> {
+ Ok(value.to_string().into())
}
-/// The extra headers added to each request, which include:
-///
-/// - `content-type`
-/// - `x-client-version`
-/// - `user-agent`
-/// - All headers specified by `header.xxx` in props.
-pub(crate) fn extra_headers_from_props(props: &HashMap<String, String>) ->
Result<HeaderMap> {
- let mut headers = HeaderMap::from_iter([
- (
- header::CONTENT_TYPE,
- HeaderValue::from_static("application/json"),
- ),
- (
- HeaderName::from_static("x-client-version"),
- HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION),
- ),
- (
- header::USER_AGENT,
-
HeaderValue::from_str(&format!("iceberg-rs/{CARGO_PKG_VERSION}")).unwrap(),
- ),
- ]);
-
- headers.extend(explicit_headers_from_props(props)?);
-
- Ok(headers)
+fn parse_credential(value: &str) -> Result<OAuth2Credential> {
Review Comment:
Can we replace this with a FromStr implementaion?
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -316,204 +372,152 @@ impl RestCatalogConfig {
])
}
- /// The HTTP client: the configured one, or a lazily-created default that
- /// is shared across every user of this config (and its clones), so token
- /// and catalog requests keep sharing one connection pool.
- pub(crate) fn client(&self) -> Client {
- self.client
- .clone()
- .unwrap_or_else(||
self.default_client.get_or_init(Client::default).clone())
- }
-
- /// Get the token from the config.
- ///
- /// The client can use this token to send requests.
- pub(crate) fn token(&self) -> Option<String> {
- self.props.get("token").cloned()
- }
-
- /// Get the credentials from the config. The client can use these
credentials to fetch a new
- /// token.
- pub(crate) fn credential(&self) -> Option<(Option<String>, String)> {
- credential_from_props(&self.props)
- }
-
- /// Get the extra headers from config, see [`extra_headers_from_props`].
- pub(crate) fn extra_headers(&self) -> Result<HeaderMap> {
- extra_headers_from_props(&self.props)
+ pub(crate) fn extra_headers(&self) -> HeaderMap {
+ let mut headers = HeaderMap::from_iter([
+ (
+ header::CONTENT_TYPE,
+ HeaderValue::from_static("application/json"),
+ ),
+ (
+ HeaderName::from_static("x-client-version"),
+ HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION),
+ ),
+ (
+ header::USER_AGENT,
+
HeaderValue::from_str(&format!("iceberg-rs/{CARGO_PKG_VERSION}")).unwrap(),
+ ),
+ ]);
+ headers.extend(self.headers.0.clone());
+ headers
}
- /// Get the optional OAuth headers from the config.
pub(crate) fn extra_oauth_params(&self) -> HashMap<String, String> {
- oauth_params_from_props(&self.props)
- }
-
- /// Check if header redaction is disabled in error logs.
- ///
- /// Returns true if the `disable-header-redaction` property is set to
"true".
- /// Defaults to false for security (headers are redacted by default).
- pub(crate) fn disable_header_redaction(&self) -> bool {
- self.props
- .get(REST_CATALOG_PROP_DISABLE_HEADER_REDACTION)
- .map(|v| v.eq_ignore_ascii_case("true"))
- .unwrap_or(false)
- }
-
- /// Builds the table cache for freshness-aware loading, or `None` when
- /// `rest-table-cache.max-entries` is 0.
- fn table_cache(&self) -> Result<Option<TableCache>> {
- let parse = |key: &str, default: u64| -> Result<u64> {
- self.props.get(key).map_or(Ok(default), |v| {
- v.parse().map_err(|e| {
- Error::new(ErrorKind::DataInvalid, format!("Invalid {key}:
{v}")).with_source(e)
- })
- })
- };
- let expire_after_write_ms = parse(
- REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS,
- TABLE_CACHE_EXPIRE_AFTER_WRITE_MS_DEFAULT,
- )?;
- let expire_after_write = Duration::from_millis(expire_after_write_ms);
- // Zero is ambiguous: it could mean "disable" or "never expire".
- if expire_after_write.is_zero() || expire_after_write >
TABLE_CACHE_EXPIRE_AFTER_WRITE_MAX {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid
{REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS}: {expire_after_write_ms}
(must be between 1 ms and 1000 years)"
- ),
- ));
- }
- let max_entries = parse(
- REST_CATALOG_PROP_TABLE_CACHE_MAX_ENTRIES,
- TABLE_CACHE_MAX_ENTRIES_DEFAULT,
- )?;
- Ok((max_entries > 0).then(|| {
+ let mut params = self.explicit_oauth_params();
+ params
+ .entry("scope".to_string())
+ .or_insert_with(|| "catalog".to_string());
+ params
+ }
+
+ pub(crate) fn explicit_oauth_params(&self) -> HashMap<String, String> {
+ [
+ ("scope", self.scope()),
+ ("audience", self.audience()),
+ ("resource", self.resource()),
+ ]
+ .into_iter()
+ .filter_map(|(key, value)| value.as_ref().map(|value|
(key.to_string(), value.clone())))
+ .collect()
+ }
+
+ fn table_cache(&self) -> Option<TableCache> {
+ (self.table_cache_max_entries() > 0).then(|| {
TableCache::builder()
- .max_capacity(max_entries)
- .time_to_live(expire_after_write)
+ .max_capacity(self.table_cache_max_entries())
+ .time_to_live(*self.table_cache_expire_after_write())
.build()
- }))
+ })
}
- /// Merge the `RestCatalogConfig` with the a [`CatalogConfig`] (fetched
from the REST server).
- pub(crate) fn merge_with_config(mut self, mut config: CatalogConfig) ->
Self {
- if let Some(uri) = config.overrides.remove(REST_CATALOG_PROP_URI) {
- self.uri = uri;
- }
-
- let mut props = config.defaults;
- props.extend(self.props);
- // The builder moved the client warehouse off the props; restore it
- // between defaults and overrides (default < client < override).
- if let Some(warehouse) = &self.warehouse {
- props.insert(REST_CATALOG_PROP_WAREHOUSE.to_string(),
warehouse.clone());
- }
- props.extend(config.overrides);
-
- self.props = props;
- self
+ fn resolved_auth_type(&self) -> &AuthType {
+ self.auth_type().as_ref().unwrap_or_else(|| {
+ if self.token.is_some() || self.credential.is_some() ||
self.oauth2_server_uri.is_some()
+ {
+ &AuthType::OAuth2
+ } else {
+ &AuthType::None
+ }
+ })
}
}
-/// Parses the `credential` property.
-///
-/// ## Output
-///
-/// - `None`: No credential is set.
-/// - `Some(None, client_secret)`: No client_id is set, use client_secret
directly.
-/// - `Some(Some(client_id), client_secret)`: Both client_id and client_secret
are set.
-pub(crate) fn credential_from_props(
- props: &HashMap<String, String>,
-) -> Option<(Option<String>, String)> {
- let cred = props.get("credential")?;
-
- match cred.split_once(':') {
- Some((client_id, client_secret)) => {
- Some((Some(client_id.to_string()), client_secret.to_string()))
- }
- None => Some((None, cred.to_string())),
- }
+fn parse_sensitive_string(value: &str) -> Result<SensitiveString> {
+ Ok(value.to_string().into())
}
-/// The extra headers added to each request, which include:
-///
-/// - `content-type`
-/// - `x-client-version`
-/// - `user-agent`
-/// - All headers specified by `header.xxx` in props.
-pub(crate) fn extra_headers_from_props(props: &HashMap<String, String>) ->
Result<HeaderMap> {
- let mut headers = HeaderMap::from_iter([
- (
- header::CONTENT_TYPE,
- HeaderValue::from_static("application/json"),
- ),
- (
- HeaderName::from_static("x-client-version"),
- HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION),
- ),
- (
- header::USER_AGENT,
-
HeaderValue::from_str(&format!("iceberg-rs/{CARGO_PKG_VERSION}")).unwrap(),
- ),
- ]);
-
- headers.extend(explicit_headers_from_props(props)?);
-
- Ok(headers)
+fn parse_credential(value: &str) -> Result<OAuth2Credential> {
Review Comment:
If possible, replace all parse function with FromStr implementation.
--
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]
