fix(query): enable role-based S3 credential chain (#19188)

fix: enable role-based S3 credential chain

Author: everpcpc

src/common/storage/src/operator.rs

@@ -416,8 +416,11 @@ fn init_s3_operator(cfg: &StorageS3Config) -> Result<impl Builder> {
         builder = builder.default_storage_class(cfg.storage_class.to_string().as_ref())
     }
 
-    // If allow_credential_chain is not set, default to false for security.
-    let allow_credential_chain = cfg.allow_credential_chain.unwrap_or(false);
+    // When a role is configured we must allow the credential chain so AWS can assume it,
+    // otherwise we should disable the credential chain for security.
+    let allow_credential_chain = cfg
+        .allow_credential_chain
+        .unwrap_or(!cfg.role_arn.is_empty());
 
     // Disallowing the credential chain forces unsigned or fully explicit access.
     if !allow_credential_chain {

src/common/storage/src/stage.rs

@@ -88,12 +88,13 @@ impl StageFileInfo {
 pub fn init_stage_operator(stage_info: &StageInfo) -> Result<Operator> {
     if stage_info.stage_type == StageType::External {
         // External S3 stages disallow the ambient credential chain by default.
-        // `role_arn` opts into using the credential chain as the source credential.
+        // The S3 operator builder will re-enable it automatically when a role_arn is present.
+        // This hook only needs to force-enable the chain for cases like system history tables.
         let storage = match stage_info.stage_params.storage.clone() {
             StorageParams::S3(mut cfg) => {
-                let allow_credential_chain =
-                    stage_info.allow_credential_chain || !cfg.role_arn.is_empty();
-                cfg.allow_credential_chain = Some(allow_credential_chain);
+                if stage_info.allow_credential_chain {
+                    cfg.allow_credential_chain = Some(true);
+                }
                 StorageParams::S3(cfg)
             }
             v => v,

src/query/sql/src/planner/binder/location.rs

@@ -172,16 +172,13 @@ fn parse_s3_params(l: &mut UriLocation, root: String) -> Result<StorageParams> {
     }
     .to_string();
 
-    // If role_arn is empty, we should not allow credential chain.
-    let mut allow_credential_chain = Some(!role_arn.is_empty());
-
-    // If we are in history table scope, we allow credential loader to be enabled
-    let in_history_table_scope = ThreadTracker::capture_log_settings()
-        .is_some_and(|settings| settings.level == LevelFilter::Off);
-    if in_history_table_scope {
-        info!("Allow credential chain for history tables");
-        allow_credential_chain = Some(true);
-    }
+    // Enable credential chain explicitly only in history-table scope.
+    let allow_credential_chain = ThreadTracker::capture_log_settings()
+        .is_some_and(|settings| settings.level == LevelFilter::Off)
+        .then(|| {
+            info!("Allow credential chain for history tables");
+            true
+        });
 
     let sp = StorageParams::S3(StorageS3Config {
         endpoint_url: secure_omission(endpoint),