blackmwk commented on code in PR #3106:
URL: https://github.com/apache/iceberg-rust/pull/3106#discussion_r4410659186
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -256,37 +365,11 @@ impl RestCatalogConfig {
}
pub(crate) fn get_token_endpoint(&self) -> String {
Review Comment:
These values only need to compute once, figure out a pattern to avoid
repeated computing.
##########
crates/catalog/rest/src/catalog.rs:
##########
Review Comment:
See https://github.com/apache/iceberg-rust/pull/3106/changes#r4410687751
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -181,72 +198,164 @@ 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)]
+#[cfg_attr(test, derive(PartialEq, Eq))]
Review Comment:
```suggestion
#[derive(PartialEq)]
```
Just dervie PartialEq, and add comment to explain why we need it.
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -316,204 +399,140 @@ 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(REST_CATALOG_PROP_SCOPE.to_string())
+ .or_insert_with(|| "catalog".to_string());
+ params
+ }
+
+ pub(crate) fn explicit_oauth_params(&self) -> HashMap<String, String> {
+ [
+ (REST_CATALOG_PROP_SCOPE, self.scope()),
+ (REST_CATALOG_PROP_AUDIENCE, self.audience()),
+ (REST_CATALOG_PROP_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
+ })
}
-}
-/// 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 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
+ }
+ })
}
}
-/// 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_header_redaction(value: &str) -> Result<bool> {
+ Ok(value.eq_ignore_ascii_case("true"))
}
-/// The default OAuth2 token endpoint for a catalog `uri`.
-pub(crate) fn default_token_endpoint(uri: &str) -> String {
- [uri, PATH_V1, "oauth", "tokens"].join("/")
-}
-
-/// Only the headers explicitly configured via `header.xxx` props (no
defaults).
-pub(crate) fn explicit_headers_from_props(props: &HashMap<String, String>) ->
Result<HeaderMap> {
- let mut headers = HeaderMap::new();
- for (key, value) in props
- .iter()
- .filter_map(|(k, v)| k.strip_prefix("header.").map(|k| (k, v)))
- {
- headers.insert(
- HeaderName::from_str(key).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Invalid header name: {key}"),
- )
- .with_source(e)
- })?,
- HeaderValue::from_str(value).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- // The value itself is omitted: it may be a secret.
- format!("Invalid value for header: {key}"),
- )
- .with_source(e)
- })?,
- );
+fn parse_page_size(value: &str) -> Result<NonZeroU32> {
+ let invalid = || {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Invalid {REST_CATALOG_PROP_PAGE_SIZE}: {value}"),
+ )
+ };
+ let page_size = value
+ .parse::<NonZeroU32>()
+ .map_err(|error| invalid().with_source(error))?;
+ if page_size.get() > i32::MAX as u32 {
+ return Err(invalid());
}
-
- Ok(headers)
+ Ok(page_size)
}
-/// The optional OAuth parameters added to each authentication request.
-pub(crate) fn oauth_params_from_props(props: &HashMap<String, String>) ->
HashMap<String, String> {
- let mut params = HashMap::new();
-
- if let Some(scope) = props.get("scope") {
- params.insert("scope".to_string(), scope.to_string());
- } else {
- params.insert("scope".to_string(), "catalog".to_string());
+fn parse_cache_expiry(value: &str) -> Result<Duration> {
Review Comment:
Aboove
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -316,204 +399,140 @@ 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(REST_CATALOG_PROP_SCOPE.to_string())
+ .or_insert_with(|| "catalog".to_string());
+ params
+ }
+
+ pub(crate) fn explicit_oauth_params(&self) -> HashMap<String, String> {
+ [
+ (REST_CATALOG_PROP_SCOPE, self.scope()),
+ (REST_CATALOG_PROP_AUDIENCE, self.audience()),
+ (REST_CATALOG_PROP_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
+ })
}
-}
-/// 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 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
+ }
+ })
}
}
-/// 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_header_redaction(value: &str) -> Result<bool> {
+ Ok(value.eq_ignore_ascii_case("true"))
}
-/// The default OAuth2 token endpoint for a catalog `uri`.
-pub(crate) fn default_token_endpoint(uri: &str) -> String {
- [uri, PATH_V1, "oauth", "tokens"].join("/")
-}
-
-/// Only the headers explicitly configured via `header.xxx` props (no
defaults).
-pub(crate) fn explicit_headers_from_props(props: &HashMap<String, String>) ->
Result<HeaderMap> {
- let mut headers = HeaderMap::new();
- for (key, value) in props
- .iter()
- .filter_map(|(k, v)| k.strip_prefix("header.").map(|k| (k, v)))
- {
- headers.insert(
- HeaderName::from_str(key).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Invalid header name: {key}"),
- )
- .with_source(e)
- })?,
- HeaderValue::from_str(value).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- // The value itself is omitted: it may be a secret.
- format!("Invalid value for header: {key}"),
- )
- .with_source(e)
- })?,
- );
+fn parse_page_size(value: &str) -> Result<NonZeroU32> {
+ let invalid = || {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Invalid {REST_CATALOG_PROP_PAGE_SIZE}: {value}"),
+ )
+ };
+ let page_size = value
+ .parse::<NonZeroU32>()
+ .map_err(|error| invalid().with_source(error))?;
+ if page_size.get() > i32::MAX as u32 {
+ return Err(invalid());
}
-
- Ok(headers)
+ Ok(page_size)
}
-/// The optional OAuth parameters added to each authentication request.
-pub(crate) fn oauth_params_from_props(props: &HashMap<String, String>) ->
HashMap<String, String> {
- let mut params = HashMap::new();
-
- if let Some(scope) = props.get("scope") {
- params.insert("scope".to_string(), scope.to_string());
- } else {
- params.insert("scope".to_string(), "catalog".to_string());
+fn parse_cache_expiry(value: &str) -> Result<Duration> {
+ let millis = value.parse::<u64>().map_err(|error| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Invalid
{REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS}: {value}"),
+ )
+ .with_source(error)
+ })?;
+ let duration = Duration::from_millis(millis);
+ if duration.is_zero() || duration > TABLE_CACHE_EXPIRE_AFTER_WRITE_MAX {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "Invalid
{REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS}: {millis} (must be
between 1 ms and 1000 years)"
+ ),
+ ));
}
+ Ok(duration)
+}
- let optional_params = ["audience", "resource"];
- for param_name in optional_params {
- if let Some(value) = props.get(param_name) {
- params.insert(param_name.to_string(), value.to_string());
+#[derive(Clone)]
+#[cfg_attr(test, derive(PartialEq, Eq))]
Review Comment:
Just derive Partialeq, add comments to explain .
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -316,204 +399,140 @@ 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(REST_CATALOG_PROP_SCOPE.to_string())
+ .or_insert_with(|| "catalog".to_string());
+ params
+ }
+
+ pub(crate) fn explicit_oauth_params(&self) -> HashMap<String, String> {
+ [
+ (REST_CATALOG_PROP_SCOPE, self.scope()),
+ (REST_CATALOG_PROP_AUDIENCE, self.audience()),
+ (REST_CATALOG_PROP_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
+ })
}
-}
-/// 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 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
+ }
+ })
}
}
-/// 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_header_redaction(value: &str) -> Result<bool> {
Review Comment:
THis is a commonly used pattern, create a mod in iceberg crate so that they
could be reused. For example, add sth like
```rust
pub struct BoolProp(bool);
impl FromStr for BoolProp {
}
```
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -316,204 +399,140 @@ 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(REST_CATALOG_PROP_SCOPE.to_string())
+ .or_insert_with(|| "catalog".to_string());
+ params
+ }
+
+ pub(crate) fn explicit_oauth_params(&self) -> HashMap<String, String> {
+ [
+ (REST_CATALOG_PROP_SCOPE, self.scope()),
+ (REST_CATALOG_PROP_AUDIENCE, self.audience()),
+ (REST_CATALOG_PROP_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
+ })
}
-}
-/// 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 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
+ }
+ })
}
}
-/// 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_header_redaction(value: &str) -> Result<bool> {
+ Ok(value.eq_ignore_ascii_case("true"))
}
-/// The default OAuth2 token endpoint for a catalog `uri`.
-pub(crate) fn default_token_endpoint(uri: &str) -> String {
- [uri, PATH_V1, "oauth", "tokens"].join("/")
-}
-
-/// Only the headers explicitly configured via `header.xxx` props (no
defaults).
-pub(crate) fn explicit_headers_from_props(props: &HashMap<String, String>) ->
Result<HeaderMap> {
- let mut headers = HeaderMap::new();
- for (key, value) in props
- .iter()
- .filter_map(|(k, v)| k.strip_prefix("header.").map(|k| (k, v)))
- {
- headers.insert(
- HeaderName::from_str(key).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Invalid header name: {key}"),
- )
- .with_source(e)
- })?,
- HeaderValue::from_str(value).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- // The value itself is omitted: it may be a secret.
- format!("Invalid value for header: {key}"),
- )
- .with_source(e)
- })?,
- );
+fn parse_page_size(value: &str) -> Result<NonZeroU32> {
Review Comment:
Above
--
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]
