fix: allow credential chain for system_history tables

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

@@ -34,6 +34,7 @@ use databend_common_ast::ast::TableAlias;
 use databend_common_ast::ast::TypeName;
 use databend_common_ast::parser::parse_values;
 use databend_common_ast::parser::tokenize_sql;
+use databend_common_base::runtime::ThreadTracker;
 use databend_common_catalog::plan::StageTableInfo;
 use databend_common_catalog::plan::list_stage_files;
 use databend_common_catalog::table_context::StageAttachment;
@@ -61,6 +62,7 @@ use databend_common_users::UserApiProvider;
 use databend_storages_common_table_meta::table::OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH;
 use databend_storages_common_table_meta::table::OPT_KEY_ENABLE_SCHEMA_EVOLUTION;
 use derive_visitor::Drive;
+use log::LevelFilter;
 use log::debug;
 use log::warn;
 use parking_lot::RwLock;
@@ -661,13 +663,20 @@ pub async fn resolve_stage_location(
         ));
     }
 
-    let stage = if names[0] == "~" {
+    let mut stage = if names[0] == "~" {
         StageInfo::new_user_stage(&ctx.get_current_user()?.name)
     } else {
         UserApiProvider::instance()
             .get_stage(&ctx.get_tenant(), names[0])
             .await?
     };
+    if ThreadTracker::capture_log_settings()
+        .is_some_and(|settings| settings.level == LevelFilter::Off)
+    {
+        // History log transform queries use the internal history stage.
+        // Enable credential chain at runtime since the flag is not persisted in meta.
+        stage.allow_credential_chain = true;
+    }
 
     let path = names.get(1).unwrap_or(&"").trim_start_matches('/');
     let path = if path.is_empty() { "/" } else { path };

src/query/storages/fuse/src/fuse_table.rs

@@ -192,6 +192,10 @@ impl FuseTable {
                     // External or attached table.
                     Some(sp) => {
                         let sp = apply_storage_class(&table_info, sp, storage_class_specs);
+                        // Special handling for history tables.
+                        // Since history tables storage params are fully generated from config,
+                        // we can safely allow credential chain.
+                        let sp = allow_system_history_credential_chain(&table_info, sp);
                         let operator = init_operator(&sp)?;
 
                         let table_meta_options = &table_info.meta.options;
@@ -1367,3 +1371,22 @@ pub enum RetentionPolicy {
     ByTimePeriod(TimeDelta),
     ByNumOfSnapshotsToKeep(usize),
 }
+
+fn allow_system_history_credential_chain(
+    table_info: &TableInfo,
+    storage_params: StorageParams,
+) -> StorageParams {
+    let mut sp = storage_params;
+    let Ok(db_name) = table_info.database_name() else {
+        return sp;
+    };
+    if !db_name.eq_ignore_ascii_case("system_history") {
+        return sp;
+    }
+    if let StorageParams::S3(cfg) = &mut sp {
+        if cfg.allow_credential_chain.is_none() {
+            cfg.allow_credential_chain = Some(true);
+        }
+    }
+    sp
+}