sunchao commented on code in PR #6059:
URL: https://github.com/apache/datafusion-comet/pull/6059#discussion_r4178199074
##########
native/core/src/parquet/objectstore/azure.rs:
##########
@@ -118,12 +432,487 @@ pub fn create_store(
.map(|(k, _)| k.as_ref())
.collect::<Vec<_>>()
);
+
+ let env: Vec<(String, String)> = env.collect();
+ validate_translated(
+ configs,
+ &translated,
+ account.as_deref(),
+ container.as_deref(),
+ oauth_keys,
+ env_token_file(&env).is_some(),
+ )?;
+ let store = build_builder(
+ url,
+ configs,
+ account.as_deref(),
+ container.as_deref(),
+ oauth_keys,
+ &translated,
+ env.into_iter(),
+ )
+ .build()?;
+ Ok((Box::new(store), path))
+}
+
+fn config_error(message: String) -> object_store::Error {
+ object_store::Error::Generic {
+ store: "MicrosoftAzure",
+ source: message.into(),
+ }
+}
+
+/// Reject a Hadoop configuration that `object_store` would silently build a
different
+/// identity from: a blank credential, an auth type or mechanism the native
scan cannot
+/// build, a named principal with no token file in Hadoop or the environment,
or a client
+/// secret or token file without the client id and tenant that complete it.
+/// `has_env_token_file` says whether `AZURE_FEDERATED_TOKEN_FILE` is set.
+fn validate_translated(
+ configs: &HashMap<String, String>,
+ translated: &[(AzureConfigKey, String)],
+ account: Option<&str>,
+ container: Option<&str>,
+ oauth_keys: OAuthKeys,
+ has_env_token_file: bool,
+) -> Result<(), object_store::Error> {
+ let account_name = account.unwrap_or("<unknown>");
+ let fail = |reason: String| {
+ Err(config_error(format!(
+ "Hadoop configuration for account {account_name}: {reason}"
+ )))
+ };
+ if let Some(reason) = hadoop_problem(configs, account, container,
oauth_keys, translated) {
+ return fail(reason);
+ }
+ let has = |wanted: AzureConfigKey| translated.iter().any(|(key, _)| *key
== wanted);
+ let borrows_env_token_file = env_policy(configs, account, container,
oauth_keys, translated)
+ == EnvPolicy::TokenFileOnly
+ && !has(AzureConfigKey::FederatedTokenFile);
+ if borrows_env_token_file && !has_env_token_file {
+ return fail(format!(
+ "the principal named by the Hadoop keys needs a token file from \
+ `{HADOOP_WI_TOKEN_FILE}` or `{ENV_FEDERATED_TOKEN_FILE}`"
+ ));
+ }
+ let mechanism = if has(AzureConfigKey::ClientSecret) {
+ HADOOP_OAUTH_CLIENT_SECRET
+ } else if has(AzureConfigKey::FederatedTokenFile) {
+ HADOOP_WI_TOKEN_FILE
+ } else if borrows_env_token_file {
+ ENV_FEDERATED_TOKEN_FILE
+ } else {
+ return Ok(());
+ };
+ let mut missing = Vec::new();
+ if !has(AzureConfigKey::ClientId) {
+ missing.push(format!("`{HADOOP_OAUTH_CLIENT_ID}`"));
+ }
+ if !has(AzureConfigKey::AuthorityId) {
+ let tenant_keys: Vec<String> = [HADOOP_MSI_TENANT,
HADOOP_OAUTH_CLIENT_ENDPOINT]
+ .into_iter()
+ .filter(|base| key_is_read(configs, account, oauth_keys, base,
AuthMechanism::OAuth))
+ .map(|base| format!("`{base}`"))
+ .collect();
+ missing.push(tenant_keys.join(" or "));
+ }
+ if missing.is_empty() {
+ return Ok(());
+ }
+ fail(format!(
+ "`{mechanism}` also needs {}",
+ missing.join(" and ")
+ ))
+}
+
+/// Why the Hadoop keys cannot be built natively as configured, or `None` when
they can:
+/// a blank value first, then an explicit auth type the translated keys do not
satisfy,
+/// then a provider class they do not satisfy, then a key that selects a
mechanism with no
+/// native counterpart.
+fn hadoop_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ container: Option<&str>,
+ oauth_keys: OAuthKeys,
+ translated: &[(AzureConfigKey, String)],
+) -> Option<String> {
+ blank_value_problem(configs, account, container, oauth_keys)
+ .or_else(|| auth_type_problem(configs, account, translated))
+ .or_else(|| provider_class_problem(configs, account, translated))
+ .or_else(|| key_provider_problem(configs, account, translated))
+ .or_else(|| unsupported_key_problem(configs, account, oauth_keys))
+}
+
+/// Whether an explicit `SimpleKeyProvider` has the account key it reads.
Hadoop fails
+/// when that key is missing, so the scan must not proceed with another
credential.
+fn key_provider_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ translated: &[(AzureConfigKey, String)],
+) -> Option<String> {
+ if !mechanism_is_read(configs, account, AuthMechanism::SharedKey) {
+ return None;
+ }
+ let (key, class) = account_scoped_entry(configs, HADOOP_KEY_PROVIDER,
account)?;
+ let has_key = translated
+ .iter()
+ .any(|(key, _)| *key == AzureConfigKey::AccessKey);
+ (is_simple_key_provider(&class) && !has_key)
+ .then(|| format!("`{key}={class}` needs `{HADOOP_KEY}`"))
+}
+
+/// A blank SAS token or credential value, named by the exact key that holds
it.
+///
+/// Blank values are errors rather than absent, so a templated configuration
that
+/// substitutes an empty string fails loudly instead of silently using another
credential.
+/// The exceptions are the keys the active provider class accepts blank
+/// (`optional_provider_keys`): Hadoop reads them with a default or as plain
strings, so
+/// they are absent and the builder proceeds as it does when they are unset.
+fn blank_value_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ container: Option<&str>,
+ oauth_keys: OAuthKeys,
+) -> Option<String> {
+ if let Some((key, value)) = active_sas_token(configs, account, container) {
+ if value.trim().is_empty() {
+ return Some(format!("`{key}` is blank"));
+ }
+ }
+ let optional = optional_provider_keys(configs, account);
+ HADOOP_CREDENTIAL_MAPPINGS
+ .iter()
+ .filter(|(base, _, mechanism)| key_is_read(configs, account,
oauth_keys, base, *mechanism))
+ .filter(|(base, _, _)| !optional.contains(base))
+ .find_map(|(base, _, _)| {
+ account_scoped_entry(configs, base, account)
+ .filter(|(_, value)| value.trim().is_empty())
+ .map(|(key, _)| format!("`{key}` is blank"))
+ })
+}
+
+/// The keys the active provider class accepts blank, or none when there is no
provider
+/// class or one the scan rejects.
+fn optional_provider_keys(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+) -> &'static [&'static str] {
+ let Some((_, class)) = active_provider_class(configs, account) else {
+ return &[];
+ };
+ if is_provider_class(&class, HADOOP_MSI_PROVIDER_CLASS) {
+ HADOOP_MSI_OPTIONAL_KEYS
+ } else if is_provider_class(&class, HADOOP_WI_PROVIDER_CLASS) {
+ HADOOP_WI_OPTIONAL_KEYS
+ } else {
+ &[]
+ }
+}
+
+/// Whether an explicit `fs.azure.account.auth.type` is one the translated
keys satisfy.
+///
+/// Setting it is Hadoop choosing a mechanism, so it is validated rather than
ignored:
+/// `SharedKey` needs the account key, `OAuth` a provider the scan can build,
`SAS` a SAS
+/// token, `Custom` has no native counterpart and any other value is a typo.
+fn auth_type_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ translated: &[(AzureConfigKey, String)],
+) -> Option<String> {
+ let (key, value) = account_scoped_entry(configs, HADOOP_AUTH_TYPE,
account)?;
+ let auth_type = value.trim();
+ if auth_type.is_empty() {
+ return Some(format!("`{key}` is blank"));
+ }
+ let has = |wanted: AzureConfigKey| translated.iter().any(|(key, _)| *key
== wanted);
+ let setting = format!("`{key}={auth_type}`");
+ if auth_type.eq_ignore_ascii_case("SharedKey") {
+ return (!has(AzureConfigKey::AccessKey))
+ .then(|| format!("{setting} needs `{HADOOP_KEY}`"));
+ }
+ if auth_type.eq_ignore_ascii_case("OAuth") {
+ return oauth_problem(configs, account, translated, &setting);
+ }
+ if auth_type.eq_ignore_ascii_case("SAS") {
+ return (!has(AzureConfigKey::SasKey)).then(|| {
+ format!(
+ "{setting} needs `{HADOOP_SAS_FIXED_TOKEN}`; a SAS token
provider class is \
+ not supported by the native scan"
+ )
+ });
+ }
+ if auth_type.eq_ignore_ascii_case("Custom") {
+ return Some(format!(
+ "{setting} loads a custom token provider class, which the native
scan does not \
+ support"
+ ));
+ }
+ Some(format!(
+ "{setting} is not supported; the native scan supports
`{HADOOP_AUTH_TYPE}` values \
+ SharedKey, OAuth and SAS"
+ ))
+}
+
+/// Whether `fs.azure.account.auth.type=OAuth` can be satisfied: through the
provider class
+/// when one is set, otherwise through a translated secret or token file.
+fn oauth_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ translated: &[(AzureConfigKey, String)],
+ setting: &str,
+) -> Option<String> {
+ if active_provider_class(configs, account).is_some() {
+ return provider_class_problem(configs, account, translated);
+ }
+ let has = |wanted: AzureConfigKey| translated.iter().any(|(key, _)| *key
== wanted);
+ if has(AzureConfigKey::ClientSecret) ||
has(AzureConfigKey::FederatedTokenFile) {
+ return None;
+ }
+ Some(format!(
+ "{setting} needs `{HADOOP_OAUTH_CLIENT_SECRET}` or
`{HADOOP_WI_TOKEN_FILE}`"
+ ))
+}
+
+/// Whether a `fs.azure.account.oauth.provider.type` class, validated whenever
it is set
+/// and OAuth is in use, names a token provider the native scan can
+/// satisfy: MSI stands alone, Workload Identity needs the client id and
tenant (the token
+/// file may still come from `AZURE_FEDERATED_TOKEN_FILE`), client credentials
need the
+/// secret, and any other class has no native counterpart.
+fn provider_class_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ translated: &[(AzureConfigKey, String)],
+) -> Option<String> {
+ let (provider_key, class) = active_provider_class(configs, account)?;
+ let class = class.trim();
+ if class.is_empty() {
+ return Some(format!("`{provider_key}` is blank"));
+ }
+ let has = |wanted: AzureConfigKey| translated.iter().any(|(key, _)| *key
== wanted);
+ let provider = format!("`{provider_key}={class}`");
+ if is_provider_class(class, HADOOP_MSI_PROVIDER_CLASS) {
+ return None;
+ }
+ if is_provider_class(class, HADOOP_WI_PROVIDER_CLASS) {
+ return (!workload_identity_named(configs, account,
translated)).then(|| {
+ format!("{provider} needs `{HADOOP_OAUTH_CLIENT_ID}` and
`{HADOOP_MSI_TENANT}`")
+ });
+ }
+ if is_provider_class(class, HADOOP_CLIENT_CREDS_PROVIDER_CLASS) {
+ return (!has(AzureConfigKey::ClientSecret))
+ .then(|| format!("{provider} needs
`{HADOOP_OAUTH_CLIENT_SECRET}`"));
+ }
+ Some(format!(
+ "{provider} is not a token provider the native scan supports"
+ ))
+}
+
+/// The first Hadoop key present that selects a mechanism with no native
counterpart,
+/// named exactly as the user set it. An explicit `SimpleKeyProvider` is the
default key
+/// provider, not a separate mechanism.
+fn unsupported_key_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ oauth_keys: OAuthKeys,
+) -> Option<String> {
+ HADOOP_UNSUPPORTED_MECHANISM_KEYS
+ .iter()
+ .filter(|(base, mechanism)| key_is_read(configs, account, oauth_keys,
base, *mechanism))
+ .find_map(|(base, _)| {
+ account_scoped_entry(configs, base, account)
Review Comment:
[P2] Honor Hadoop's auth-type guard before rejecting a global SAS provider.
When global `fs.azure.account.auth.type=OAuth` is overridden by account-scoped
`SAS`, Hadoop's `getTokenProviderClass` ignores a global
`fs.azure.sas.token.provider.type` because the auth types differ. With a fixed
SAS token, Hadoop therefore initializes successfully. This unconditional lookup
instead treats the inactive global provider as selected and rejects the scan. A
configuration also supplying the matching legacy container token works on the
base revision, so this introduces a scan failure for previously working
configurations. Could provider resolution mirror Hadoop's guarded global
fallback while retaining rejection of providers actually selected for the
account?
Evidence: Reproduced with exact base/head `create_store` implementations,
actual `object_store 0.13.2`, and an environment without Azure settings. URL:
`abfss://[email protected]/path/file.parquet`. Configuration:
`fs.azure.account.auth.type=OAuth`;
`fs.azure.account.auth.type.myacct.dfs.core.windows.net=SAS`;
`fs.azure.sas.token.provider.type=example.UnusedSasProvider`; both
`fs.azure.sas.fixed.token.myacct.dfs.core.windows.net` and
`fs.azure.sas.data.myacct.dfs.core.windows.net` set to
`sv=2020-08-04&sig=synthetic-container`. Base returns Ok. Head returns `Hadoop
configuration for account myacct: fs.azure.sas.token.provider.type selects an
authentication mechanism the native scan does not support`. Removing only the
inactive provider makes both revisions succeed. Upstream Hadoop 3.4.1
`AbfsConfiguration.getTokenProviderClass`, lines 550–558, guards global
fallback by auth-type equality. Hadoop 3.4.2 and 3.5.0 retain that guard. The
reproduction made no storage o
r credential requests.
--
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]
