fix(query): revoke need consider GrantObject::Database/Table (#14567)

* fix(query): revoke need consider GrantObject::Database/Table

* add some comment about revoke

old version, all contains ownership privilege. now will not revoke ownership privilege.

so the user will contains the object's ownership

Then in bind_revoke, if revoke all on <object> from user, need contains
ownership.

.github/actions/fuse_compat/action.yml

@@ -20,15 +20,14 @@ runs:
     - name: Test compatibility
       shell: bash
       run: |
-          bash ./tests/fuse-compat/test-fuse-compat.sh 0.7.150 base
-          bash ./tests/fuse-compat/test-fuse-compat.sh 0.7.151 base
           bash ./tests/fuse-compat/test-fuse-compat.sh 1.0.56 base
           bash ./tests/fuse-compat/test-fuse-compat.sh 1.0.56 01_meta_compression 01_flashback
           bash ./tests/fuse-compat/test-fuse-compat.sh 1.1.30 base
           bash ./tests/fuse-compat/test-fuse-compat.sh 1.1.30 02_meta_compression_v3_to_v4 02_flashback_v3_to_v4
           bash ./tests/fuse-compat/test-fuse-compat.sh 1.1.38 base
           bash ./tests/fuse-compat/test-fuse-compat.sh 1.1.39 base
           bash ./tests/fuse-compat/test-fuse-compat.sh 1.1.46 base
+          bash ./tests/fuse-compat/test-fuse-compat.sh 1.2.241 revoke
           bash ./tests/fuse-compat/test-fuse-compat.sh 1.2.306 rbac
           bash ./tests/fuse-compat/test-fuse-forward-compat.sh 1.2.307 rbac
           bash ./tests/fuse-compat/test-fuse-forward-compat.sh 1.2.318 rbac

src/query/service/src/interpreters/interpreter_privilege_revoke.rs

@@ -53,7 +53,9 @@ impl Interpreter for RevokePrivilegeInterpreter {
 
         let plan = self.plan.clone();
 
-        validate_grant_object_exists(&self.ctx, &plan.on).await?;
+        for object in &plan.on {
+            validate_grant_object_exists(&self.ctx, object).await?;
+        }
 
         // TODO: check user existence
         // TODO: check privilege on granting on the grant object
@@ -63,19 +65,23 @@ impl Interpreter for RevokePrivilegeInterpreter {
 
         match plan.principal {
             PrincipalIdentity::User(user) => {
-                user_mgr
-                    .revoke_privileges_from_user(&tenant, user, plan.on, plan.priv_types)
-                    .await?;
+                for object in plan.on {
+                    user_mgr
+                        .revoke_privileges_from_user(&tenant, user.clone(), object, plan.priv_types)
+                        .await?;
+                }
             }
             PrincipalIdentity::Role(role) => {
                 if role == BUILTIN_ROLE_ACCOUNT_ADMIN {
                     return Err(ErrorCode::IllegalGrant(
                         "Illegal REVOKE command. Can not revoke built-in role [ account_admin ]",
                     ));
                 }
-                user_mgr
-                    .revoke_privileges_from_role(&tenant, &role, plan.on, plan.priv_types)
-                    .await?;
+                for object in plan.on {
+                    user_mgr
+                        .revoke_privileges_from_role(&tenant, &role, object, plan.priv_types)
+                        .await?;
+                }
             }
         }
 

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

@@ -22,6 +22,7 @@ use databend_common_ast::ast::RevokeStmt;
 use databend_common_exception::Result;
 use databend_common_meta_app::principal::AuthInfo;
 use databend_common_meta_app::principal::GrantObject;
+use databend_common_meta_app::principal::PrincipalIdentity;
 use databend_common_meta_app::principal::UserOption;
 use databend_common_meta_app::principal::UserPrivilegeSet;
 use databend_common_users::UserApiProvider;
@@ -97,8 +98,13 @@ impl Binder {
             AccountMgrSource::ALL { level } => {
                 // ALL PRIVILEGES have different available privileges set on different grant objects
                 // Now in this case all is always true.
-                let grant_object = self.convert_to_grant_object(level).await?;
-                let priv_types = grant_object.available_privileges(false);
+                let grant_object = self.convert_to_revoke_grant_object(level).await?;
+                // Note if old version `grant all on db.*/db.t to user`, the user will contains ownership privilege.
+                // revoke all need to revoke it.
+                let priv_types = match principal {
+                    PrincipalIdentity::User(_) => grant_object[0].available_privileges(true),
+                    PrincipalIdentity::Role(_) => grant_object[0].available_privileges(false),
+                };
                 let plan = RevokePrivilegePlan {
                     principal: principal.clone(),
                     on: grant_object,
@@ -107,7 +113,7 @@ impl Binder {
                 Ok(Plan::RevokePriv(Box::new(plan)))
             }
             AccountMgrSource::Privs { privileges, level } => {
-                let grant_object = self.convert_to_grant_object(level).await?;
+                let grant_object = self.convert_to_revoke_grant_object(level).await?;
                 let mut priv_types = UserPrivilegeSet::empty();
                 for x in privileges {
                     priv_types.set_privilege(*x);
@@ -165,6 +171,57 @@ impl Binder {
         }
     }
 
+    // Some old query version use GrantObject::Table store table name.
+    // So revoke need compat the old version.
+    pub(in crate::planner::binder) async fn convert_to_revoke_grant_object(
+        &self,
+        source: &AccountMgrLevel,
+    ) -> Result<Vec<GrantObject>> {
+        // TODO fetch real catalog
+        let catalog_name = self.ctx.get_current_catalog();
+        let tenant = self.ctx.get_tenant();
+        let catalog = self.ctx.get_catalog(&catalog_name).await?;
+        match source {
+            AccountMgrLevel::Global => Ok(vec![GrantObject::Global]),
+            AccountMgrLevel::Table(database_name, table_name) => {
+                let database_name = database_name
+                    .clone()
+                    .unwrap_or_else(|| self.ctx.get_current_database());
+                let db_id = catalog
+                    .get_database(&tenant, &database_name)
+                    .await?
+                    .get_db_info()
+                    .ident
+                    .db_id;
+                let table_id = catalog
+                    .get_table(&tenant, &database_name, table_name)
+                    .await?
+                    .get_id();
+                Ok(vec![
+                    GrantObject::TableById(catalog_name.clone(), db_id, table_id),
+                    GrantObject::Table(catalog_name.clone(), database_name, table_name.clone()),
+                ])
+            }
+            AccountMgrLevel::Database(database_name) => {
+                let database_name = database_name
+                    .clone()
+                    .unwrap_or_else(|| self.ctx.get_current_database());
+                let db_id = catalog
+                    .get_database(&tenant, &database_name)
+                    .await?
+                    .get_db_info()
+                    .ident
+                    .db_id;
+                Ok(vec![
+                    GrantObject::DatabaseById(catalog_name.clone(), db_id),
+                    GrantObject::Database(catalog_name.clone(), database_name),
+                ])
+            }
+            AccountMgrLevel::UDF(udf) => Ok(vec![GrantObject::UDF(udf.clone())]),
+            AccountMgrLevel::Stage(stage) => Ok(vec![GrantObject::Stage(stage.clone())]),
+        }
+    }
+
     #[async_backtrace::framed]
     pub(in crate::planner::binder) async fn bind_create_user(
         &mut self,

src/query/sql/src/planner/plans/ddl/account.rs

@@ -123,7 +123,7 @@ pub struct GrantPrivilegePlan {
 pub struct RevokePrivilegePlan {
     pub principal: PrincipalIdentity,
     pub priv_types: UserPrivilegeSet,
-    pub on: GrantObject,
+    pub on: Vec<GrantObject>,
 }
 
 #[derive(Clone, Debug, PartialEq)]

tests/fuse-compat/compat-logictest/revoke/fuse_compat_read

@@ -0,0 +1,20 @@
+query T
+show grants for a;
+----
+GRANT INSERT ON 'default'.'default'.'t' TO 'a'@'%'
+GRANT SELECT ON 'default'.'default'.* TO 'a'@'%'
+GRANT ALL ON 'default'.'db'.* TO 'a'@'%'
+
+statement ok
+revoke insert on default.t from a;
+
+statement ok
+revoke select on default.* from a;
+
+statement ok
+revoke all on db.* from a;
+
+query T
+show grants for a;
+----
+

tests/fuse-compat/compat-logictest/revoke/fuse_compat_write

@@ -0,0 +1,20 @@
+statement ok
+drop user if exists a;
+
+statement ok
+create table t(id int);
+
+statement ok
+create user a identified by '123';
+
+statement ok
+grant insert on default.t to a;
+
+statement ok
+grant select on default.* to a;
+
+statement ok
+create database if not exists db;
+
+statement ok
+grant all on db.* to a;