feat: complete database-level default connection configuration

This commit completes the database-level default connection configuration feature
allowing databases to specify default storage connection settings for FUSE tables.

Key changes:
- Fix table inheritance logic: FUSE tables inherit database defaults
- Add ALTER DATABASE SET OPTIONS with paired validation
- Implement privilege checking for ALTER DATABASE operations
- Add comprehensive SQL logic tests for feature validation

The feature enables:
- CREATE DATABASE OPTIONS (DEFAULT_STORAGE_CONNECTION = '...', DEFAULT_STORAGE_PATH = '...')
- ALTER DATABASE SET OPTIONS with paired validation (both options required)
- FUSE tables automatically inherit database connection defaults for external storage

Corrected inheritance logic where FUSE tables (external storage capable) inherit
database defaults, while other engines use their own connection mechanisms.

Author: dantengsky

src/query/service/src/interpreters/access/privilege_access.rs

@@ -1693,7 +1693,9 @@ impl AccessChecker for PrivilegeAccess {
             Plan::RenameWorkloadGroup(_) => {}
             Plan::SetWorkloadGroupQuotas(_) => {}
             Plan::UnsetWorkloadGroupQuotas(_) => {}
-            Plan::AlterDatabase(_) => {todo!()}
+            Plan::AlterDatabase(plan) => {
+                self.validate_db_access(&plan.catalog, &plan.database, UserPrivilegeType::Alter, false).await?;
+            }
         }
 
         Ok(())

src/query/service/src/interpreters/interpreter_alter_database.rs

@@ -49,6 +49,17 @@ impl Interpreter for AlterDatabaseInterpreter {
     async fn execute2(&self) -> Result<PipelineBuildResult> {
         debug!("ctx.id" = self.ctx.get_id().as_str(); "alter_database_execute");
 
+        // Validate options before proceeding
+        // Validate that DEFAULT_STORAGE_CONNECTION and DEFAULT_STORAGE_PATH are used together
+        let has_connection = self.plan.options.contains_key("DEFAULT_STORAGE_CONNECTION");
+        let has_path = self.plan.options.contains_key("DEFAULT_STORAGE_PATH");
+
+        if has_connection != has_path {
+            return Err(databend_common_exception::ErrorCode::BadArguments(
+                "DEFAULT_STORAGE_CONNECTION and DEFAULT_STORAGE_PATH options must be used together"
+            ));
+        }
+
         // Get the catalog and database
         let _catalog = self.ctx.get_catalog(&self.plan.catalog).await?;
 

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

@@ -336,13 +336,13 @@ impl Binder {
         let has_path = options.iter().any(|p| p.name == "DEFAULT_STORAGE_PATH");
 
         if has_connection && !has_path {
-            return Err(ErrorCode::InvalidArgument(
+            return Err(ErrorCode::BadArguments(
                 "DEFAULT_STORAGE_CONNECTION requires DEFAULT_STORAGE_PATH to be specified".to_string()
             ));
         }
 
         if has_path && !has_connection {
-            return Err(ErrorCode::InvalidArgument(
+            return Err(ErrorCode::BadArguments(
                 "DEFAULT_STORAGE_PATH requires DEFAULT_STORAGE_CONNECTION to be specified".to_string()
             ));
         }

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

@@ -549,17 +549,21 @@ impl Binder {
 
         // Get database defaults for connection options
         let mut options: BTreeMap<String, String> = BTreeMap::new();
-        if let Ok(database_info) = catalog.get_database(&self.ctx.get_tenant(), &database).await {
-            // Extract database-level default connection options
-            if let Some(default_connection) = database_info.options().get("DEFAULT_STORAGE_CONNECTION") {
-                options.insert("connection".to_string(), default_connection.clone());
-            }
-            if let Some(default_path) = database_info.options().get("DEFAULT_STORAGE_PATH") {
-                options.insert("location".to_string(), default_path.clone());
+
+        // FUSE tables can inherit database connection defaults for external storage
+        let engine = engine.unwrap_or(catalog.default_table_engine());
+        if matches!(engine, Engine::Fuse) {
+            if let Ok(database_info) = catalog.get_database(&self.ctx.get_tenant(), &database).await {
+                // Extract database-level default connection options
+                if let Some(default_connection) = database_info.options().get("DEFAULT_STORAGE_CONNECTION") {
+                    options.insert("connection".to_string(), default_connection.clone());
+                }
+                if let Some(default_path) = database_info.options().get("DEFAULT_STORAGE_PATH") {
+                    options.insert("location".to_string(), default_path.clone());
+                }
             }
         }
 
-        let engine = engine.unwrap_or(catalog.default_table_engine());
         if catalog.support_partition() != (engine == Engine::Iceberg) {
             return Err(ErrorCode::TableEngineNotSupported(format!(
                 "Catalog '{}' engine type is {:?} but table {} engine type is {}",