chore(query): allow_anonymous when key token is empty

src/common/storage/src/operator.rs

@@ -421,6 +421,17 @@ fn init_s3_operator(cfg: &StorageS3Config) -> Result<impl Builder> {
         builder = builder.disable_config_load().disable_ec2_metadata();
     }
 
+    // If credential loading is disabled and no credentials are provided, use unsigned requests.
+    // This allows accessing public buckets reliably in environments where signing could be rejected.
+    if cfg.disable_credential_loader
+        && cfg.access_key_id.is_empty()
+        && cfg.secret_access_key.is_empty()
+        && cfg.security_token.is_empty()
+        && cfg.role_arn.is_empty()
+    {
+        builder = builder.allow_anonymous();
+    }
+
     // Enable virtual host style
     if cfg.enable_virtual_host_style {
         builder = builder.enable_virtual_host_style();

src/common/storage/src/stage.rs

@@ -24,6 +24,7 @@ use databend_common_exception::Result;
 use databend_common_meta_app::principal::StageInfo;
 use databend_common_meta_app::principal::StageType;
 use databend_common_meta_app::principal::UserIdentity;
+use databend_common_meta_app::storage::StorageParams;
 use futures::Stream;
 use futures::StreamExt;
 use futures::TryStreamExt;
@@ -89,7 +90,18 @@ impl StageFileInfo {
 
 pub fn init_stage_operator(stage_info: &StageInfo) -> Result<Operator> {
     if stage_info.stage_type == StageType::External {
-        Ok(init_operator(&stage_info.stage_params.storage)?)
+        // External S3 stages don't load credentials by default; `role_arn` opts into assume-role.
+        let storage = match stage_info.stage_params.storage.clone() {
+            StorageParams::S3(mut cfg) => {
+                if cfg.role_arn.is_empty() {
+                    cfg.disable_credential_loader = true;
+                }
+                StorageParams::S3(cfg)
+            }
+            v => v,
+        };
+
+        Ok(init_operator(&storage)?)
     } else {
         let stage_prefix = stage_info.stage_prefix();
         let param = DataOperator::instance()

src/query/sql/src/planner/binder/ddl/stage.rs

@@ -20,6 +20,7 @@ use databend_common_exception::Result;
 use databend_common_meta_app::principal::FileFormatOptionsReader;
 use databend_common_meta_app::principal::FileFormatParams;
 use databend_common_meta_app::principal::StageInfo;
+use databend_common_meta_app::storage::StorageParams;
 use databend_common_storage::init_operator;
 
 use super::super::copy_into_table::resolve_stage_location;
@@ -89,6 +90,17 @@ impl Binder {
                 )
                 .await?;
 
+                // External S3 stages don't load credentials by default; `role_arn` opts into assume-role.
+                let stage_storage = match stage_storage {
+                    StorageParams::S3(mut cfg) => {
+                        if cfg.role_arn.is_empty() {
+                            cfg.disable_credential_loader = true;
+                        }
+                        StorageParams::S3(cfg)
+                    }
+                    v => v,
+                };
+
                 // Check the storage params via init operator.
                 let _ = init_operator(&stage_storage).map_err(|err| {
                     ErrorCode::InvalidConfig(format!(

tests/sqllogictests/suites/base/05_ddl/05_0016_ddl_stage_public_s3_list.test

@@ -0,0 +1,11 @@
+statement ok
+DROP STAGE IF EXISTS wizardbend_tpch
+
+statement ok
+CREATE OR REPLACE STAGE wizardbend_tpch URL='s3://wizardbend/TPC-H/1TB/customer/'
+
+statement ok
+LIST @wizardbend_tpch
+
+statement ok
+DROP STAGE IF EXISTS wizardbend_tpch