[fix](auth)unified workload and resource permission logic (#32907)

- `Grant resource` can no longer grant global `usage_priv`
-  `grant resource %` instead of `grant resource *`

before change:
```
grant usage_priv on resource * to f;
show grants for f\G
*************************** 1. row ***************************
      UserIdentity: 'f'@'%'
           Comment: 
          Password: No
             Roles: 
       GlobalPrivs: Usage_priv 
      CatalogPrivs: NULL
     DatabasePrivs: internal.information_schema: Select_priv ; internal.mysql: Select_priv 
        TablePrivs: NULL
          ColPrivs: NULL
     ResourcePrivs: NULL
 CloudClusterPrivs: NULL
WorkloadGroupPrivs: normal: Usage_priv 
```
after change
```
grant usage_priv on resource '%' to f;
show grants for f\G
*************************** 1. row ***************************
      UserIdentity: 'f'@'%'
           Comment: 
          Password: No
             Roles: 
       GlobalPrivs: NULL
      CatalogPrivs: NULL
     DatabasePrivs: internal.information_schema: Select_priv ; internal.mysql: Select_priv 
        TablePrivs: NULL
          ColPrivs: NULL
     ResourcePrivs: %: Usage_priv 
 CloudClusterPrivs: NULL
WorkloadGroupPrivs: normal: Usage_priv 

```

fe/fe-core/src/main/java/org/apache/doris/analysis/ResourcePattern.java

@@ -22,6 +22,7 @@
 import org.apache.doris.common.io.Text;
 import org.apache.doris.common.io.Writable;
 import org.apache.doris.mysql.privilege.Auth.PrivLevel;
+import org.apache.doris.persist.gson.GsonPostProcessable;
 import org.apache.doris.persist.gson.GsonUtils;
 
 import com.google.common.base.Strings;
@@ -34,7 +35,7 @@
 // only the following 2 formats are allowed
 // *
 // resource
-public class ResourcePattern implements Writable {
+public class ResourcePattern implements Writable, GsonPostProcessable {
     @SerializedName(value = "resourceName")
     private String resourceName;
 
@@ -48,9 +49,9 @@ public class ResourcePattern implements Writable {
     public static ResourcePattern ALL_STAGE;
 
     static {
-        ALL_GENERAL = new ResourcePattern("*", ResourceTypeEnum.GENERAL);
-        ALL_CLUSTER = new ResourcePattern("*", ResourceTypeEnum.CLUSTER);
-        ALL_STAGE = new ResourcePattern("*", ResourceTypeEnum.STAGE);
+        ALL_GENERAL = new ResourcePattern("%", ResourceTypeEnum.GENERAL);
+        ALL_CLUSTER = new ResourcePattern("%", ResourceTypeEnum.CLUSTER);
+        ALL_STAGE = new ResourcePattern("%", ResourceTypeEnum.STAGE);
 
         try {
             ALL_GENERAL.analyze();
@@ -65,7 +66,11 @@ private ResourcePattern() {
     }
 
     public ResourcePattern(String resourceName, ResourceTypeEnum type) {
-        this.resourceName = Strings.isNullOrEmpty(resourceName) ? "*" : resourceName;
+        // To be compatible with previous syntax
+        if ("*".equals(resourceName)) {
+            resourceName = "%";
+        }
+        this.resourceName = Strings.isNullOrEmpty(resourceName) ? "%" : resourceName;
         resourceType = type;
     }
 
@@ -86,15 +91,11 @@ public String getResourceName() {
     }
 
     public PrivLevel getPrivLevel() {
-        if (resourceName.equals("*")) {
-            return PrivLevel.GLOBAL;
-        } else {
-            return PrivLevel.RESOURCE;
-        }
+        return PrivLevel.RESOURCE;
     }
 
     public void analyze() throws AnalysisException {
-        if (!resourceName.equals("*")) {
+        if (!resourceName.equals("%")) {
             FeNameFormat.checkResourceName(resourceName, resourceType);
         }
     }
@@ -130,4 +131,12 @@ public static ResourcePattern read(DataInput in) throws IOException {
         String json = Text.readString(in);
         return GsonUtils.GSON.fromJson(json, ResourcePattern.class);
     }
+
+    @Override
+    public void gsonPostProcess() throws IOException {
+        // // To be compatible with previous syntax
+        if ("*".equals(resourceName)) {
+            resourceName = "%";
+        }
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/ResourcePrivEntry.java

@@ -27,11 +27,8 @@
 import java.io.IOException;
 
 public class ResourcePrivEntry extends PrivEntry {
-    protected static final String ANY_RESOURCE = "*";
-
     protected PatternMatcher resourcePattern;
     protected String origResource;
-    protected boolean isAnyResource;
 
     protected ResourcePrivEntry() {
     }
@@ -41,15 +38,12 @@ protected ResourcePrivEntry(PatternMatcher resourcePattern,
         super(privSet);
         this.resourcePattern = resourcePattern;
         this.origResource = origResource;
-        if (origResource.equals(ANY_RESOURCE)) {
-            isAnyResource = true;
-        }
     }
 
     public static ResourcePrivEntry create(String resourceName, PrivBitSet privs)
             throws AnalysisException, PatternMatcherException {
         PatternMatcher resourcePattern = PatternMatcher.createMysqlPattern(
-                resourceName.equals(ANY_RESOURCE) ? "%" : resourceName,
+                resourceName,
                 CaseSensibility.RESOURCE.getCaseSensibility());
         if (privs.containsNodePriv() || privs.containsDbTablePriv()) {
             throw new AnalysisException("Resource privilege can not contains node or db table privileges: " + privs);
@@ -112,6 +106,5 @@ public void readFields(DataInput in) throws IOException {
         } catch (PatternMatcherException e) {
             throw new IOException(e);
         }
-        isAnyResource = origResource.equals(ANY_RESOURCE);
     }
 }

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/ResourcePrivTable.java

@@ -26,26 +26,15 @@
 public class ResourcePrivTable extends PrivTable {
     private static final Logger LOG = LogManager.getLogger(ResourcePrivTable.class);
 
-    /*
-     * Return first priv which match the user@host on resourceName The returned priv will be
-     * saved in 'savedPrivs'.
-     */
     public void getPrivs(String resourceName, PrivBitSet savedPrivs) {
-        ResourcePrivEntry matchedEntry = null;
+        // need check all entries, because may have 2 entries match resourceName,
+        // For example, if the resourceName is g1, there are two entry `%` and `g1` compound requirements
         for (PrivEntry entry : entries) {
             ResourcePrivEntry resourcePrivEntry = (ResourcePrivEntry) entry;
             // check resource
-            if (!resourcePrivEntry.getResourcePattern().match(resourceName)) {
-                continue;
+            if (resourcePrivEntry.getResourcePattern().match(resourceName)) {
+                savedPrivs.or(resourcePrivEntry.getPrivSet());
             }
-
-            matchedEntry = resourcePrivEntry;
-            break;
-        }
-        if (matchedEntry == null) {
-            return;
         }
-
-        savedPrivs.or(matchedEntry.getPrivSet());
     }
 }

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/Role.java

@@ -471,8 +471,8 @@ public boolean checkWorkloadGroupPriv(String workloadGroupName, PrivPredicate wa
             return true;
         }
         PrivBitSet savedPrivs = PrivBitSet.of();
-        // Workload groups do not support global usage_priv, so only global admin_priv and usage_priv are checked.
-        if (checkGlobalInternal(PrivPredicate.ADMIN, savedPrivs)
+        // usage priv not in global, but grant_priv may in global
+        if (checkGlobalInternal(wanted, savedPrivs)
                 || checkWorkloadGroupInternal(workloadGroupName, wanted, savedPrivs)) {
             return true;
         }
@@ -573,22 +573,11 @@ private void grantPrivs(ResourcePattern resourcePattern, PrivBitSet privs) throw
         if (privs.isEmpty()) {
             return;
         }
-        // grant privs to user
-        switch (resourcePattern.getPrivLevel()) {
-            case GLOBAL:
-                grantGlobalPrivs(privs);
-                break;
-            case RESOURCE:
-                if (resourcePattern.isClusterResource()) {
-                    grantCloudClusterPrivs(resourcePattern.getResourceName(), false, false, privs);
-                } else {
-                    grantResourcePrivs(resourcePattern.getResourceName(), privs);
-                }
-                break;
-            default:
-                Preconditions.checkNotNull(null, resourcePattern.getPrivLevel());
+        if (resourcePattern.isClusterResource()) {
+            grantCloudClusterPrivs(resourcePattern.getResourceName(), false, false, privs);
+        } else {
+            grantResourcePrivs(resourcePattern.getResourceName(), privs);
         }
-
     }
 
     private void grantPrivs(WorkloadGroupPattern workloadGroupPattern, PrivBitSet privs) throws DdlException {

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/WorkloadGroupPrivTable.java

@@ -20,20 +20,13 @@
 public class WorkloadGroupPrivTable extends PrivTable {
 
     public void getPrivs(String workloadGroupName, PrivBitSet savedPrivs) {
-        WorkloadGroupPrivEntry matchedEntry = null;
+        // need check all entries, because may have 2 entries match workloadGroupName,
+        // For example, if the workloadGroupName is g1, there are two entry `%` and `g1` compound requirements
         for (PrivEntry entry : entries) {
             WorkloadGroupPrivEntry workloadGroupPrivEntry = (WorkloadGroupPrivEntry) entry;
-            if (!workloadGroupPrivEntry.getWorkloadGroupPattern().match(workloadGroupName)) {
-                continue;
+            if (workloadGroupPrivEntry.getWorkloadGroupPattern().match(workloadGroupName)) {
+                savedPrivs.or(workloadGroupPrivEntry.getPrivSet());
             }
-
-            matchedEntry = workloadGroupPrivEntry;
-            break;
-        }
-        if (matchedEntry == null) {
-            return;
         }
-
-        savedPrivs.or(matchedEntry.getPrivSet());
     }
 }

fe/fe-core/src/test/java/org/apache/doris/analysis/GrantStmtTest.java

@@ -108,8 +108,8 @@ public void testResourceNormal() throws UserException {
         stmt = new GrantStmt(new UserIdentity("testUser", "%"), null,
                 new ResourcePattern("*", ResourceTypeEnum.GENERAL), privileges, ResourceTypeEnum.GENERAL);
         stmt.analyze(analyzer);
-        Assert.assertEquals(Auth.PrivLevel.GLOBAL, stmt.getResourcePattern().getPrivLevel());
-        Assert.assertEquals("GRANT Usage_priv ON RESOURCE '*' TO 'testUser'@'%'", stmt.toSql());
+        Assert.assertEquals(Auth.PrivLevel.RESOURCE, stmt.getResourcePattern().getPrivLevel());
+        Assert.assertEquals("GRANT Usage_priv ON RESOURCE '%' TO 'testUser'@'%'", stmt.toSql());
     }
 
     @Test(expected = AnalysisException.class)

fe/fe-core/src/test/java/org/apache/doris/mysql/privilege/AuthTest.java

@@ -18,6 +18,7 @@
 package org.apache.doris.mysql.privilege;
 
 import org.apache.doris.analysis.Analyzer;
+import org.apache.doris.analysis.CompoundPredicate.Operator;
 import org.apache.doris.analysis.CreateRoleStmt;
 import org.apache.doris.analysis.CreateUserStmt;
 import org.apache.doris.analysis.DropRoleStmt;
@@ -1677,15 +1678,17 @@ public void testGrantRole() {
     }
 
     @Test
-    public void testResource() {
+    public void testResource() throws UserException {
         UserIdentity userIdentity = new UserIdentity("testUser", "%");
         String role = "role0";
         String resourceName = "spark0";
         ResourcePattern resourcePattern = new ResourcePattern(resourceName, ResourceTypeEnum.GENERAL);
-        String anyResource = "*";
+        String anyResource = "%";
         ResourcePattern anyResourcePattern = new ResourcePattern(anyResource, ResourceTypeEnum.GENERAL);
         List<AccessPrivilegeWithCols> usagePrivileges = Lists
                 .newArrayList(new AccessPrivilegeWithCols(AccessPrivilege.USAGE_PRIV));
+        List<AccessPrivilegeWithCols> grantPrivileges = Lists
+                .newArrayList(new AccessPrivilegeWithCols(AccessPrivilege.GRANT_PRIV));
         UserDesc userDesc = new UserDesc(userIdentity, "12345", true);
 
         // ------ grant|revoke resource to|from user ------
@@ -1833,8 +1836,9 @@ public void testResource() {
             Assert.fail();
         }
         Assert.assertTrue(accessManager.checkResourcePriv(userIdentity, resourceName, PrivPredicate.USAGE));
-        Assert.assertTrue(accessManager.checkGlobalPriv(userIdentity, PrivPredicate.USAGE));
-        Assert.assertTrue(accessManager.checkGlobalPriv(userIdentity, PrivPredicate.SHOW_RESOURCES));
+        // anyResource not belong to global auth
+        Assert.assertFalse(accessManager.checkGlobalPriv(userIdentity, PrivPredicate.USAGE));
+        Assert.assertFalse(accessManager.checkGlobalPriv(userIdentity, PrivPredicate.SHOW_RESOURCES));
         Assert.assertFalse(accessManager.checkGlobalPriv(userIdentity, PrivPredicate.SHOW));
 
         // 3. revoke usage_priv on resource '*' from 'testUser'@'%'
@@ -1891,7 +1895,7 @@ public void testResource() {
             Assert.fail();
         }
         Assert.assertTrue(accessManager.checkResourcePriv(userIdentity, resourceName, PrivPredicate.USAGE));
-        Assert.assertTrue(accessManager.checkGlobalPriv(userIdentity, PrivPredicate.USAGE));
+        Assert.assertFalse(accessManager.checkGlobalPriv(userIdentity, PrivPredicate.USAGE));
 
         // 3. revoke usage_priv on resource '*' from role 'role0'
         revokeStmt = new RevokeStmt(null, role, anyResourcePattern, usagePrivileges, ResourceTypeEnum.GENERAL);
@@ -1960,10 +1964,26 @@ public void testResource() {
         GrantStmt grantStmt3 = new GrantStmt(userIdentity, "test_role", tablePattern, usagePrivileges);
         ExceptionChecker.expectThrowsWithMsg(AnalysisException.class,
                 "Can not grant/revoke USAGE_PRIV to/from database or table", () -> grantStmt3.analyze(analyzer));
+
+        // 4.drop user
+        dropUser(userIdentity);
+
+        // -------- test anyPattern has usage_priv and g1 has grant_priv  ----------
+        // 1. create user with no role
+        createUser(userIdentity);
+        // 2. grant usage_priv on workload group '%' to user
+        grant(new GrantStmt(userIdentity, null, anyResourcePattern, usagePrivileges, ResourceTypeEnum.GENERAL));
+        // 3.grant grant_priv on workload group g1
+        grant(new GrantStmt(userIdentity, null, resourcePattern, grantPrivileges, ResourceTypeEnum.GENERAL));
+        Assert.assertTrue(accessManager.checkResourcePriv(userIdentity, resourceName,
+                PrivPredicate.of(PrivBitSet.of(Privilege.USAGE_PRIV, Privilege.GRANT_PRIV),
+                        Operator.AND)));
+        // 4. drop user
+        dropUser(userIdentity);
     }
 
     @Test
-    public void testWorkloadGroupPriv() {
+    public void testWorkloadGroupPriv() throws UserException {
         UserIdentity userIdentity = new UserIdentity("testUser", "%");
         String role = "role0";
         String workloadGroupName = "g1";
@@ -1972,6 +1992,8 @@ public void testWorkloadGroupPriv() {
         WorkloadGroupPattern anyWorkloadGroupPattern = new WorkloadGroupPattern(anyWorkloadGroup);
         List<AccessPrivilegeWithCols> usagePrivileges = Lists
                 .newArrayList(new AccessPrivilegeWithCols(AccessPrivilege.USAGE_PRIV));
+        List<AccessPrivilegeWithCols> grantPrivileges = Lists
+                .newArrayList(new AccessPrivilegeWithCols(AccessPrivilege.GRANT_PRIV));
         UserDesc userDesc = new UserDesc(userIdentity, "12345", true);
 
         // ------ grant|revoke workload group to|from user ------
@@ -2244,6 +2266,27 @@ public void testWorkloadGroupPriv() {
         GrantStmt grantStmt3 = new GrantStmt(userIdentity, "test_role", tablePattern, usagePrivileges);
         ExceptionChecker.expectThrowsWithMsg(AnalysisException.class,
                 "Can not grant/revoke USAGE_PRIV to/from database or table", () -> grantStmt3.analyze(analyzer));
+        // 4.drop user
+        dropUser(userIdentity);
+
+        // -------- test anyPattern has usage_priv and g1 has grant_priv  ----------
+        // 1. create user with no role
+        createUser(userIdentity);
+        // 2. grant usage_priv on workload group '%' to user
+        grant(new GrantStmt(userIdentity, null, anyWorkloadGroupPattern, usagePrivileges));
+        // 3.grant grant_priv on workload group g1
+        grant(new GrantStmt(userIdentity, null, workloadGroupPattern, grantPrivileges));
+        Assert.assertTrue(accessManager.checkWorkloadGroupPriv(userIdentity, workloadGroupName,
+                PrivPredicate.of(PrivBitSet.of(Privilege.USAGE_PRIV, Privilege.GRANT_PRIV),
+                        Operator.AND)));
+        // 4. drop user
+        dropUser(userIdentity);
+    }
+
+    private void dropUser(UserIdentity userIdentity) throws UserException {
+        DropUserStmt dropUserStmt = new DropUserStmt(userIdentity);
+        dropUserStmt.analyze(analyzer);
+        auth.dropUser(dropUserStmt);
     }
 
     private void createUser(UserIdentity userIdentity) throws UserException {