Enhance database default storage validation

dantengsky · dantengsky · commit 215abcedad7e · 2025-10-28T20:59:27.000+08:00
diff --git a/src/query/service/src/interpreters/access/privilege_access.rs b/src/query/service/src/interpreters/access/privilege_access.rs
@@ -246,6 +246,10 @@ impl PrivilegeAccess {
     ) -> Result<()> {
         self.access_system_history(Some(catalog_name), Some(db_name), None, privileges)?;
         let tenant = self.ctx.get_tenant();
+        let catalog = self.ctx.get_catalog(catalog_name).await?;
+        if if_exists && !catalog.exists_database(&tenant, db_name).await? {
+            return Ok(());
+        }
         let check_current_role_only = match privileges {
             // create table/stream need check db's Create Privilege
             UserPrivilegeType::Create => true,
@@ -264,7 +268,6 @@ impl PrivilegeAccess {
                 return Ok(());
             }
             Err(_err) => {
-                let catalog = self.ctx.get_catalog(catalog_name).await?;
                 match self
                     .convert_to_id(&tenant, &catalog, db_name, None, false)
                     .await
@@ -1694,7 +1697,14 @@ impl AccessChecker for PrivilegeAccess {
             Plan::SetWorkloadGroupQuotas(_) => {}
             Plan::UnsetWorkloadGroupQuotas(_) => {}
             Plan::AlterDatabase(plan) => {
-                self.validate_db_access(&plan.catalog, &plan.database, UserPrivilegeType::Alter, false).await?;
+                self
+                    .validate_db_access(
+                        &plan.catalog,
+                        &plan.database,
+                        UserPrivilegeType::Alter,
+                        plan.if_exists,
+                    )
+                    .await?;
             }
         }
 
diff --git a/src/query/service/src/interpreters/interpreter_alter_database.rs b/src/query/service/src/interpreters/interpreter_alter_database.rs
@@ -16,6 +16,8 @@ use std::sync::Arc;
 
 use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::Result;
+use databend_common_sql::planner::binder::ddl::database::DEFAULT_STORAGE_CONNECTION;
+use databend_common_sql::planner::binder::ddl::database::DEFAULT_STORAGE_PATH;
 use databend_common_sql::plans::AlterDatabasePlan;
 use log::debug;
 
@@ -49,20 +51,30 @@ impl Interpreter for AlterDatabaseInterpreter {
     #[async_backtrace::framed]
     async fn execute2(&self) -> Result<PipelineBuildResult> {
         debug!("ctx.id" = self.ctx.get_id().as_str(); "alter_database_execute");
-
         let catalog = self.ctx.get_catalog(&self.plan.catalog).await?;
-        let database = catalog
+        let database = match catalog
             .get_database(&self.plan.tenant, &self.plan.database)
-            .await?;
+            .await
+        {
+            Ok(db) => db,
+            Err(err) => {
+                if self.plan.if_exists
+                    && err.code() == databend_common_exception::ErrorCode::UNKNOWN_DATABASE
+                {
+                    return Ok(PipelineBuildResult::create());
+                }
+                return Err(err);
+            }
+        };
 
         // Merge provided options with the existing database options
         let mut merged_options = database.options().clone();
         for (key, value) in &self.plan.options {
             merged_options.insert(key.clone(), value.clone());
         }
 
-        let connection_value = merged_options.get("DEFAULT_STORAGE_CONNECTION").cloned();
-        let path_value = merged_options.get("DEFAULT_STORAGE_PATH").cloned();
+        let connection_value = merged_options.get(DEFAULT_STORAGE_CONNECTION).cloned();
+        let path_value = merged_options.get(DEFAULT_STORAGE_PATH).cloned();
 
         // Check if both options are present together in the final merged state
         // This ensures that after ALTER, the database still has both options configured
diff --git a/src/query/service/src/interpreters/interpreter_database_show_create.rs b/src/query/service/src/interpreters/interpreter_database_show_create.rs
@@ -75,7 +75,7 @@ impl Interpreter for ShowCreateDatabaseInterpreter {
                 .iter()
                 .map(|(k, v)| format!("{}='{}'", k, v))
                 .collect::<Vec<_>>()
-                .join(" ");
+                .join(", ");
             if !db.options().is_empty() {
                 write!(info, " OPTIONS ({})", options).expect("failed to format database options");
             }
diff --git a/src/query/sql/src/planner/binder/ddl/database.rs b/src/query/sql/src/planner/binder/ddl/database.rs
@@ -49,6 +49,9 @@ use crate::plans::UndropDatabasePlan;
 use crate::BindContext;
 use crate::SelectBuilder;
 
+pub const DEFAULT_STORAGE_CONNECTION: &str = "DEFAULT_STORAGE_CONNECTION";
+pub const DEFAULT_STORAGE_PATH: &str = "DEFAULT_STORAGE_PATH";
+
 impl Binder {
     #[async_backtrace::framed]
     pub(in crate::planner::binder) async fn bind_show_databases(
@@ -217,20 +220,33 @@ impl Binder {
             ))),
 
             AlterDatabaseAction::SetOptions { options } => {
-                // Validate database options before processing
-                // For ALTER DATABASE, allow modifying single option (the other already exists)
-                self.validate_database_options(options, false).await?;
+                let catalog_arc = self.ctx.get_catalog(&catalog).await?;
+                let db_exists = catalog_arc.exists_database(&tenant, &database).await?;
+                if !db_exists && !*if_exists {
+                    return Err(ErrorCode::UnknownDatabase(format!(
+                        "Unknown database '{}'",
+                        database
+                    )));
+                }
 
-                // Convert SQLProperty to BTreeMap for database options
-                let db_options = options
-                    .iter()
-                    .map(|property| (property.name.clone(), property.value.clone()))
-                    .collect::<BTreeMap<String, String>>();
+                // Validate database options only when the database exists.
+                let db_options = if db_exists {
+                    // For ALTER DATABASE, allow modifying single option (the other already exists)
+                    self.validate_database_options(options, false).await?;
+
+                    options
+                        .iter()
+                        .map(|property| (property.name.clone(), property.value.clone()))
+                        .collect::<BTreeMap<String, String>>()
+                } else {
+                    BTreeMap::new()
+                };
 
                 Ok(Plan::AlterDatabase(Box::new(AlterDatabasePlan {
                     tenant,
                     catalog,
                     database,
+                    if_exists: *if_exists,
                     options: db_options,
                 })))
             }
@@ -331,8 +347,7 @@ impl Binder {
         require_both: bool,
     ) -> Result<()> {
         // Validate database options - only allow specific connection-related options
-        const VALID_DATABASE_OPTIONS: &[&str] =
-            &["DEFAULT_STORAGE_CONNECTION", "DEFAULT_STORAGE_PATH"];
+        const VALID_DATABASE_OPTIONS: &[&str] = &[DEFAULT_STORAGE_CONNECTION, DEFAULT_STORAGE_PATH];
 
         // Check for duplicate options
         let mut seen_options = std::collections::HashSet::new();
@@ -354,33 +369,31 @@ impl Binder {
         }
 
         // Validate pairing requirement based on operation type
-        let has_connection = options
-            .iter()
-            .any(|p| p.name == "DEFAULT_STORAGE_CONNECTION");
-        let has_path = options.iter().any(|p| p.name == "DEFAULT_STORAGE_PATH");
+        let has_connection = options.iter().any(|p| p.name == DEFAULT_STORAGE_CONNECTION);
+        let has_path = options.iter().any(|p| p.name == DEFAULT_STORAGE_PATH);
 
         if require_both {
             // For CREATE DATABASE: both options must be specified together
             if has_connection && !has_path {
-                return Err(ErrorCode::BadArguments(
-                    "DEFAULT_STORAGE_CONNECTION requires DEFAULT_STORAGE_PATH to be specified"
-                        .to_string(),
-                ));
+                return Err(ErrorCode::BadArguments(format!(
+                    "{} requires {} to be specified",
+                    DEFAULT_STORAGE_CONNECTION, DEFAULT_STORAGE_PATH
+                )));
             }
 
             if has_path && !has_connection {
-                return Err(ErrorCode::BadArguments(
-                    "DEFAULT_STORAGE_PATH requires DEFAULT_STORAGE_CONNECTION to be specified"
-                        .to_string(),
-                ));
+                return Err(ErrorCode::BadArguments(format!(
+                    "{} requires {} to be specified",
+                    DEFAULT_STORAGE_PATH, DEFAULT_STORAGE_CONNECTION
+                )));
             }
         }
         // For ALTER DATABASE: allow modifying single option (the other one already exists in database)
 
         // Validate that the specified connection exists
         if let Some(connection_property) = options
             .iter()
-            .find(|p| p.name == "DEFAULT_STORAGE_CONNECTION")
+            .find(|p| p.name == DEFAULT_STORAGE_CONNECTION)
         {
             let connection_name = &connection_property.value;
 
@@ -402,15 +415,46 @@ impl Binder {
         if let (Some(connection_prop), Some(path_prop)) = (
             options
                 .iter()
-                .find(|p| p.name == "DEFAULT_STORAGE_CONNECTION"),
-            options.iter().find(|p| p.name == "DEFAULT_STORAGE_PATH"),
+                .find(|p| p.name == DEFAULT_STORAGE_CONNECTION),
+            options.iter().find(|p| p.name == DEFAULT_STORAGE_PATH),
         ) {
-            // Validate the storage path is accessible
+            // Validate the storage path is accessible and matches the connection protocol
             let connection = self.ctx.get_connection(&connection_prop.value).await?;
+
+            let uri_for_scheme = databend_common_ast::ast::UriLocation::from_uri(
+                path_prop.value.clone(),
+                BTreeMap::new(),
+            )
+            .map_err(|e| {
+                ErrorCode::BadArguments(format!(
+                    "Invalid storage path '{}': {}",
+                    path_prop.value, e
+                ))
+            })?;
+
+            let path_protocol = normalize_storage_protocol(&uri_for_scheme.protocol);
+            let connection_protocol = normalize_storage_protocol(&connection.storage_type);
+
+            if path_protocol != connection_protocol {
+                return Err(ErrorCode::BadArguments(format!(
+                    "{} protocol '{}' does not match connection '{}' protocol '{}'",
+                    DEFAULT_STORAGE_PATH,
+                    uri_for_scheme.protocol,
+                    connection_prop.value,
+                    connection.storage_type
+                )));
+            }
+
             let mut uri_location = databend_common_ast::ast::UriLocation::from_uri(
                 path_prop.value.clone(),
-                connection.storage_params,
-            )?;
+                connection.storage_params.clone(),
+            )
+            .map_err(|e| {
+                ErrorCode::BadArguments(format!(
+                    "Invalid storage path '{}': {}",
+                    path_prop.value, e
+                ))
+            })?;
 
             // Parse and validate the URI location using parse_storage_params_from_uri
             // This enforces that the path must end with '/' (directory requirement)
@@ -490,3 +534,11 @@ impl Binder {
         })
     }
 }
+
+fn normalize_storage_protocol(protocol: &str) -> String {
+    let mut lower = protocol.to_ascii_lowercase();
+    if lower == "file" {
+        lower = "fs".to_string();
+    }
+    lower
+}
diff --git a/src/query/sql/src/planner/binder/ddl/mod.rs b/src/query/sql/src/planner/binder/ddl/mod.rs
@@ -17,7 +17,7 @@ mod catalog;
 mod column;
 mod connection;
 mod data_mask;
-mod database;
+pub mod database;
 mod dictionary;
 mod dynamic_table;
 mod index;
diff --git a/src/query/sql/src/planner/binder/ddl/table.rs b/src/query/sql/src/planner/binder/ddl/table.rs
@@ -117,6 +117,8 @@ use crate::binder::ConstraintExprBinder;
 use crate::binder::Visibility;
 use crate::optimizer::ir::SExpr;
 use crate::parse_computed_expr_to_string;
+use crate::planner::binder::ddl::database::DEFAULT_STORAGE_CONNECTION;
+use crate::planner::binder::ddl::database::DEFAULT_STORAGE_PATH;
 use crate::planner::semantic::normalize_identifier;
 use crate::planner::semantic::resolve_type_name;
 use crate::planner::semantic::IdentifierNormalizer;
@@ -562,8 +564,8 @@ impl Binder {
             {
                 // Extract database-level default connection options
                 let default_connection_name =
-                    database_info.options().get("DEFAULT_STORAGE_CONNECTION");
-                let default_path = database_info.options().get("DEFAULT_STORAGE_PATH");
+                    database_info.options().get(DEFAULT_STORAGE_CONNECTION);
+                let default_path = database_info.options().get(DEFAULT_STORAGE_PATH);
 
                 // If both database defaults exist, construct UriLocation
                 if let (Some(connection_name), Some(path)) = (default_connection_name, default_path)
diff --git a/src/query/sql/src/planner/binder/mod.rs b/src/query/sql/src/planner/binder/mod.rs
@@ -27,7 +27,7 @@ mod column_binding;
 mod constraint_expr;
 mod copy_into_location;
 mod copy_into_table;
-mod ddl;
+pub mod ddl;
 mod default_expr;
 mod distinct;
 mod explain;
@@ -78,6 +78,8 @@ pub use constraint_expr::ConstraintExprBinder;
 pub use copy_into_table::resolve_file_location;
 pub use copy_into_table::resolve_stage_location;
 pub use copy_into_table::resolve_stage_locations;
+pub use ddl::database::DEFAULT_STORAGE_CONNECTION;
+pub use ddl::database::DEFAULT_STORAGE_PATH;
 pub use ddl::table::verify_external_location_privileges;
 pub use default_expr::DefaultExprBinder;
 pub use explain::ExplainConfig;
diff --git a/src/query/sql/src/planner/plans/ddl/database.rs b/src/query/sql/src/planner/plans/ddl/database.rs
@@ -161,6 +161,7 @@ pub struct AlterDatabasePlan {
     pub tenant: Tenant,
     pub catalog: String,
     pub database: String,
+    pub if_exists: bool,
     pub options: BTreeMap<String, String>,
 }
 
diff --git a/tests/sqllogictests/suites/base/05_ddl/05_0001_ddl_create_database_options.test b/tests/sqllogictests/suites/base/05_ddl/05_0001_ddl_create_database_options.test
diff --git a/tests/sqllogictests/suites/base/05_ddl/05_0002_ddl_table_inherit_database_defaults.test b/tests/sqllogictests/suites/base/05_ddl/05_0002_ddl_table_inherit_database_defaults.test
diff --git a/tests/sqllogictests/suites/base/05_ddl/05_0003_ddl_database_connection_validation.test b/tests/sqllogictests/suites/base/05_ddl/05_0003_ddl_database_connection_validation.test

Original file line number	Diff line number	Diff line change
`@@ -75,7 +75,7 @@ impl Interpreter for ShowCreateDatabaseInterpreter {`
`75`	`75`	`.iter()`
`76`	`76`	`.map(\|(k, v)\| format!("{}='{}'", k, v))`
`77`	`77`	`.collect::<Vec<_>>()`
`78`		`- .join(" ");`
	`78`	`+ .join(", ");`
`79`	`79`	`if !db.options().is_empty() {`
`80`	`80`	`write!(info, " OPTIONS ({})", options).expect("failed to format database options");`
`81`	`81`	`}`
Original file line number	Diff line number	Diff line change
`@@ -161,6 +161,7 @@ pub struct AlterDatabasePlan {`
`161`	`161`	`pub tenant: Tenant,`
`162`	`162`	`pub catalog: String,`
`163`	`163`	`pub database: String,`
	`164`	`+ pub if_exists: bool,`
`164`	`165`	`pub options: BTreeMap<String, String>,`
`165`	`166`	`}`
`166`	`167`