fix(query): after drop role account_admin should support grant owners…

…hip to new role

1. if has ownership but owner role not exists, the owner role will be
   set to account_admin

2. has_ownership should get ownership and role from meta

src/query/service/src/interpreters/interpreter_privilege_grant.rs

@@ -24,6 +24,7 @@ use databend_common_meta_app::principal::UserPrivilegeType::Ownership;
 use databend_common_sql::plans::GrantPrivilegePlan;
 use databend_common_users::RoleCacheManager;
 use databend_common_users::UserApiProvider;
+use databend_common_users::BUILTIN_ROLE_ACCOUNT_ADMIN;
 use log::debug;
 use log::error;
 use log::info;
@@ -136,7 +137,14 @@ impl GrantPrivilegeInterpreter {
         // if the object's owner is None, it's considered as PUBLIC, everyone could access it
         let owner = user_mgr.get_ownership(tenant, owner_object).await?;
         if let Some(owner) = owner {
-            let can_grant_ownership = available_roles.iter().any(|r| r.name == owner.role);
+            // if object has ownership, but the owner role is not exists, set owner role to ACCOUNT_ADMIN,
+            // only account_admin can grant this object.
+            let role = if !user_mgr.exists_role(tenant, owner.role.clone()).await? {
+                BUILTIN_ROLE_ACCOUNT_ADMIN.to_string()
+            } else {
+                owner.role.clone()
+            };
+            let can_grant_ownership = available_roles.iter().any(|r| r.name == role);
             log_msg = format!(
                 "{}: grant ownership on {:?} from role {} to {}",
                 ctx.get_id(),

src/query/service/src/sessions/session_privilege_mgr.rs

@@ -23,6 +23,7 @@ use databend_common_meta_app::principal::UserInfo;
 use databend_common_meta_app::principal::UserPrivilegeType;
 use databend_common_users::GrantObjectVisibilityChecker;
 use databend_common_users::RoleCacheManager;
+use databend_common_users::UserApiProvider;
 use databend_common_users::BUILTIN_ROLE_ACCOUNT_ADMIN;
 use databend_common_users::BUILTIN_ROLE_PUBLIC;
 
@@ -267,14 +268,20 @@ impl SessionPrivilegeManager for SessionPrivilegeManagerImpl {
 
     #[async_backtrace::framed]
     async fn has_ownership(&self, object: &OwnershipObject) -> Result<bool> {
-        let role_mgr = RoleCacheManager::instance();
+        let user_mgr = UserApiProvider::instance();
         let tenant = self.session_ctx.get_current_tenant();
-
-        // if the object is not owned by any role, then considered as ACCOUNT_ADMIN, the normal users
-        // can not access it unless ACCOUNT_ADMIN grant relevant privileges to them.
-        let owner_role_name = match role_mgr.find_object_owner(&tenant, object).await? {
-            Some(owner_role) => owner_role.name,
+        let owner_role_name = match user_mgr.get_ownership(&tenant, object).await? {
             None => BUILTIN_ROLE_ACCOUNT_ADMIN.to_string(),
+            Some(owner) => match user_mgr.get_role(&tenant, owner.role).await {
+                Ok(owner_role) => owner_role.name,
+                Err(e) => {
+                    if e.code() == ErrorCode::UNKNOWN_ROLE {
+                        BUILTIN_ROLE_ACCOUNT_ADMIN.to_string()
+                    } else {
+                        return Err(e);
+                    }
+                }
+            },
         };
 
         let effective_roles = self.get_all_effective_roles().await?;

tests/suites/0_stateless/18_rbac/18_0002_ownership_cover.result

@@ -37,3 +37,8 @@ t1
 Error: APIError: ResponseError with 1063: Permission denied: User 'a'@'%' does not have the required privileges for database 'db_a'
 t
 t1
+=== fix_issue_14572: test drop role; grant ownership ===
+a	drop_role
+a	drop_role
+a	drop_role1
+GRANT OWNERSHIP ON 'default'.'a'.* TO ROLE `drop_role1`

tests/suites/0_stateless/18_rbac/18_0002_ownership_cover.sh

@@ -126,3 +126,26 @@ echo "drop database db_a;" | $BENDSQL_CLIENT_CONNECT
 echo "drop role if exists role1;" | $BENDSQL_CLIENT_CONNECT
 echo "drop user if exists a;" | $BENDSQL_CLIENT_CONNECT
 echo "drop user if exists b;" | $BENDSQL_CLIENT_CONNECT
+
+echo "=== fix_issue_14572: test drop role; grant ownership ==="
+echo "drop role if exists drop_role;" | $BENDSQL_CLIENT_CONNECT
+echo "drop role if exists drop_role1;" | $BENDSQL_CLIENT_CONNECT
+echo "drop user if exists u1;" | $BENDSQL_CLIENT_CONNECT
+echo "drop database if exists a;" | $BENDSQL_CLIENT_CONNECT
+echo "create role drop_role;" | $BENDSQL_CLIENT_CONNECT
+echo "create role drop_role1;" | $BENDSQL_CLIENT_CONNECT
+echo "create user u1 identified by '123' with DEFAULT_ROLE='drop_role'" | $BENDSQL_CLIENT_CONNECT
+echo "grant role drop_role to u1;" | $BENDSQL_CLIENT_CONNECT
+echo "grant create on *.* to u1;" | $BENDSQL_CLIENT_CONNECT
+export USER_U1_CONNECT="bendsql --user=u1 --password=123 --host=${QUERY_MYSQL_HANDLER_HOST} --port ${QUERY_HTTP_HANDLER_PORT}"
+
+echo "create database a" | $USER_U1_CONNECT
+echo "select name, owner from system.databases where name='a'" | $USER_U1_CONNECT
+echo "drop role drop_role;" | $BENDSQL_CLIENT_CONNECT
+echo "select name, owner from system.databases where name='a'" | $BENDSQL_CLIENT_CONNECT
+echo "grant ownership on a.* to role drop_role1;" | $BENDSQL_CLIENT_CONNECT
+echo "select name, owner from system.databases where name='a'" | $BENDSQL_CLIENT_CONNECT
+echo "show grants for role drop_role1" | $BENDSQL_CLIENT_CONNECT
+echo "drop role drop_role1" | $BENDSQL_CLIENT_CONNECT
+echo "drop user u1" | $BENDSQL_CLIENT_CONNECT
+echo "drop database a" | $BENDSQL_CLIENT_CONNECT