[feature](privilege) Support su user and add admin_readonly role

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisLexer.g4

@@ -540,6 +540,7 @@ SUBJECT: 'SUBJECT';
 SUBSTR: 'SUBSTR';
 SUBSTRING: 'SUBSTRING';
 SUM: 'SUM';
+SU: 'SU';
 SUPERUSER: 'SUPERUSER';
 SWITCH: 'SWITCH';
 SYNC: 'SYNC';

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -509,6 +509,7 @@ supportedOtherStatement
     | UNLOCK TABLES                                                                 #unlockTables
     | INSTALL PLUGIN FROM source=identifierOrText properties=propertyClause?        #installPlugin
     | UNINSTALL PLUGIN name=identifierOrText                                        #uninstallPlugin
+    | SU user=suUser (COMMA? roles+=identifierOrText (COMMA roles+=identifierOrText)*)? #su
     | LOCK TABLES (lockTable (COMMA lockTable)*)?                                   #lockTables
     | RESTORE SNAPSHOT label=multipartIdentifier FROM repo=identifier
         ((ON | EXCLUDE) LEFT_PAREN baseTableRef (COMMA baseTableRef)* RIGHT_PAREN)?
@@ -1114,6 +1115,10 @@ userIdentify
         | LEFT_PAREN host=identifierOrText RIGHT_PAREN))?
     ;
 
+suUser
+    : user=identifierOrText (ATSIGN host+=identifierOrText (DOT host+=identifierOrText)*)?
+    ;
+
 grantUserIdentify
     : userIdentify (IDENTIFIED BY PASSWORD? STRING_LITERAL)?
     ;

fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/ranger/RangerAccessController.java

@@ -22,6 +22,7 @@
 import org.apache.doris.common.AuthorizationException;
 import org.apache.doris.mysql.privilege.CatalogAccessController;
 import org.apache.doris.mysql.privilege.DataMaskPolicy;
+import org.apache.doris.mysql.privilege.PrivilegeContext;
 import org.apache.doris.mysql.privilege.RangerDataMaskPolicy;
 import org.apache.doris.mysql.privilege.RangerRowFilterPolicy;
 import org.apache.doris.mysql.privilege.RowFilterPolicy;
@@ -89,8 +90,9 @@ public static void checkRequestResults(Collection<RangerAccessResult> results, S
     }
 
     @Override
-    public List<? extends RowFilterPolicy> evalRowFilterPolicies(UserIdentity currentUser, String ctl, String db,
-            String tbl) {
+    public List<? extends RowFilterPolicy> evalRowFilterPolicies(PrivilegeContext context, String ctl, String db,
+                                                                 String tbl) {
+        UserIdentity currentUser = context.getCurrentUser();
         RangerAccessResourceImpl resource = createResource(ctl, db, tbl);
         RangerAccessRequestImpl request = createRequest(currentUser);
         // If the access type is not set here, it defaults to ANY1 ACCESS.
@@ -121,8 +123,9 @@ public List<? extends RowFilterPolicy> evalRowFilterPolicies(UserIdentity curren
     }
 
     @Override
-    public Optional<DataMaskPolicy> evalDataMaskPolicy(UserIdentity currentUser, String ctl, String db, String tbl,
-            String col) {
+    public Optional<DataMaskPolicy> evalDataMaskPolicy(PrivilegeContext context, String ctl, String db,
+                                                       String tbl, String col) {
+        UserIdentity currentUser = context.getCurrentUser();
         RangerAccessResourceImpl resource = createResource(ctl, db, tbl, col);
         RangerAccessRequestImpl request = createRequest(currentUser);
         request.setAccessType(DorisAccessType.SELECT.name());

fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/ranger/doris/RangerDorisAccessController.java

@@ -25,6 +25,7 @@
 import org.apache.doris.mysql.privilege.PrivBitSet;
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.mysql.privilege.Privilege;
+import org.apache.doris.mysql.privilege.PrivilegeContext;
 import org.apache.doris.resource.workloadgroup.WorkloadGroupMgr;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -125,7 +126,8 @@ private boolean checkPrivilege(UserIdentity currentUser, PrivPredicate wanted,
     }
 
     @Override
-    public boolean checkGlobalPriv(UserIdentity currentUser, PrivPredicate wanted) {
+    public boolean checkGlobalPriv(PrivilegeContext context, PrivPredicate wanted) {
+        UserIdentity currentUser = context.getCurrentUser();
         PrivBitSet checkedPrivs = PrivBitSet.of();
         return checkGlobalPrivInternal(currentUser, wanted, checkedPrivs);
     }
@@ -136,7 +138,8 @@ private boolean checkGlobalPrivInternal(UserIdentity currentUser, PrivPredicate
     }
 
     @Override
-    public boolean checkCtlPriv(UserIdentity currentUser, String ctl, PrivPredicate wanted) {
+    public boolean checkCtlPriv(PrivilegeContext context, String ctl, PrivPredicate wanted) {
+        UserIdentity currentUser = context.getCurrentUser();
         PrivBitSet checkedPrivs = PrivBitSet.of();
         if (checkGlobalPrivInternal(currentUser, wanted, checkedPrivs)
                 || checkCtlPrivInternal(currentUser, ctl, wanted, checkedPrivs)) {
@@ -160,7 +163,8 @@ private boolean checkAnyPrivWithinCtl(UserIdentity currentUser, String ctl) {
     }
 
     @Override
-    public boolean checkDbPriv(UserIdentity currentUser, String ctl, String db, PrivPredicate wanted) {
+    public boolean checkDbPriv(PrivilegeContext context, String ctl, String db, PrivPredicate wanted) {
+        UserIdentity currentUser = context.getCurrentUser();
         PrivBitSet checkedPrivs = PrivBitSet.of();
         if (checkGlobalPrivInternal(currentUser, wanted, checkedPrivs)
                 || checkCtlPrivInternal(currentUser, ctl, wanted, checkedPrivs)
@@ -187,7 +191,8 @@ private boolean checkAnyPrivWithinDb(UserIdentity currentUser, String ctl, Strin
     }
 
     @Override
-    public boolean checkTblPriv(UserIdentity currentUser, String ctl, String db, String tbl, PrivPredicate wanted) {
+    public boolean checkTblPriv(PrivilegeContext context, String ctl, String db, String tbl, PrivPredicate wanted) {
+        UserIdentity currentUser = context.getCurrentUser();
         PrivBitSet checkedPrivs = PrivBitSet.of();
         if (checkGlobalPrivInternal(currentUser, wanted, checkedPrivs)
                 || checkCtlPrivInternal(currentUser, ctl, wanted, checkedPrivs)
@@ -215,8 +220,9 @@ private boolean checkAnyPrivWithinTbl(UserIdentity currentUser, String ctl, Stri
     }
 
     @Override
-    public void checkColsPriv(UserIdentity currentUser, String ctl, String db, String tbl, Set<String> cols,
-            PrivPredicate wanted) throws AuthorizationException {
+    public void checkColsPriv(PrivilegeContext context, String ctl, String db, String tbl, Set<String> cols,
+                              PrivPredicate wanted) throws AuthorizationException {
+        UserIdentity currentUser = context.getCurrentUser();
         PrivBitSet checkedPrivs = PrivBitSet.of();
         boolean hasTablePriv = checkGlobalPrivInternal(currentUser, wanted, checkedPrivs)
                 || checkCtlPrivInternal(currentUser, ctl, wanted, checkedPrivs)
@@ -243,8 +249,9 @@ private boolean checkColPrivInternal(UserIdentity currentUser, String ctl, Strin
     }
 
     @Override
-    public boolean checkCloudPriv(UserIdentity currentUser, String cloudName,
-            PrivPredicate wanted, ResourceTypeEnum type) {
+    public boolean checkCloudPriv(PrivilegeContext context, String cloudName,
+                                  PrivPredicate wanted, ResourceTypeEnum type) {
+        UserIdentity currentUser = context.getCurrentUser();
         // only support CLUSTER,
         // STORAGE_VAULT should call `checkStorageVaultPriv`
         // GENERAL should call `checkResourcePriv`
@@ -266,7 +273,8 @@ private boolean checkComputeGroupPrivInternal(UserIdentity currentUser, String c
     }
 
     @Override
-    public boolean checkStorageVaultPriv(UserIdentity currentUser, String storageVaultName, PrivPredicate wanted) {
+    public boolean checkStorageVaultPriv(PrivilegeContext context, String storageVaultName, PrivPredicate wanted) {
+        UserIdentity currentUser = context.getCurrentUser();
         PrivBitSet checkedPrivs = PrivBitSet.of();
         return checkGlobalPrivInternal(currentUser, wanted, checkedPrivs)
                 || checkStorageVaultPrivInternal(currentUser, storageVaultName, wanted, checkedPrivs);
@@ -280,7 +288,8 @@ private boolean checkStorageVaultPrivInternal(UserIdentity currentUser, String s
     }
 
     @Override
-    public boolean checkResourcePriv(UserIdentity currentUser, String resourceName, PrivPredicate wanted) {
+    public boolean checkResourcePriv(PrivilegeContext context, String resourceName, PrivPredicate wanted) {
+        UserIdentity currentUser = context.getCurrentUser();
         PrivBitSet checkedPrivs = PrivBitSet.of();
         return checkGlobalPrivInternal(currentUser, wanted, checkedPrivs)
                 || checkResourcePrivInternal(currentUser, resourceName, wanted, checkedPrivs);
@@ -293,7 +302,8 @@ private boolean checkResourcePrivInternal(UserIdentity currentUser, String resou
     }
 
     @Override
-    public boolean checkWorkloadGroupPriv(UserIdentity currentUser, String workloadGroupName, PrivPredicate wanted) {
+    public boolean checkWorkloadGroupPriv(PrivilegeContext context, String workloadGroupName, PrivPredicate wanted) {
+        UserIdentity currentUser = context.getCurrentUser();
         // For compatibility with older versions, it is not needed to check the privileges of the default group.
         if (WorkloadGroupMgr.DEFAULT_GROUP_NAME.equals(workloadGroupName)) {
             return true;
@@ -336,11 +346,12 @@ public static void main(String[] args) {
         RangerDorisAccessController ac = new RangerDorisAccessController("doris");
         UserIdentity user = new UserIdentity("user1", "127.0.0.1");
         user.setIsAnalyzed();
-        boolean res = ac.checkDbPriv(user, "internal", "db1", PrivPredicate.SHOW);
+        PrivilegeContext context = PrivilegeContext.of(user);
+        boolean res = ac.checkDbPriv(context, "internal", "db1", PrivPredicate.SHOW);
         System.out.println("res: " + res);
         user = new UserIdentity("user2", "127.0.0.1");
         user.setIsAnalyzed();
-        res = ac.checkTblPriv(user, "internal", "db1", "tbl1", PrivPredicate.SELECT);
+        res = ac.checkTblPriv(PrivilegeContext.of(user), "internal", "db1", "tbl1", PrivPredicate.SELECT);
         System.out.println("res: " + res);
     }
 }

fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAccessController.java

@@ -26,6 +26,7 @@
 import org.apache.doris.common.ThreadPoolManager;
 import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.mysql.privilege.PrivilegeContext;
 
 import com.google.common.collect.Maps;
 import org.apache.logging.log4j.LogManager;
@@ -137,63 +138,65 @@ private HiveAccessType convertToAccessType(PrivPredicate predicate) {
     }
 
     @Override
-    public boolean checkGlobalPriv(UserIdentity currentUser, PrivPredicate wanted) {
+    public boolean checkGlobalPriv(PrivilegeContext context, PrivPredicate wanted) {
         // hive ranger plugin does not support global privilege
         // use internal access controller to check
         return Env.getCurrentEnv().getAccessManager().getAccessControllerOrDefault(
-                InternalCatalog.INTERNAL_CATALOG_NAME).checkGlobalPriv(currentUser, wanted);
+                InternalCatalog.INTERNAL_CATALOG_NAME).checkGlobalPriv(context, wanted);
     }
 
     @Override
-    public boolean checkCtlPriv(UserIdentity currentUser, String ctl, PrivPredicate wanted) {
+    public boolean checkCtlPriv(PrivilegeContext context, String ctl, PrivPredicate wanted) {
         return true;
     }
 
     @Override
-    public boolean checkDbPriv(UserIdentity currentUser, String ctl, String db, PrivPredicate wanted) {
+    public boolean checkDbPriv(PrivilegeContext context, String ctl, String db, PrivPredicate wanted) {
         RangerHiveResource resource = new RangerHiveResource(HiveObjectType.DATABASE,
                 ClusterNamespace.getNameFromFullName(db));
-        return checkPrivilege(currentUser, convertToAccessType(wanted), resource);
+        return checkPrivilege(context.getCurrentUser(), convertToAccessType(wanted), resource);
     }
 
     @Override
-    public boolean checkTblPriv(UserIdentity currentUser, String ctl, String db, String tbl, PrivPredicate wanted) {
+    public boolean checkTblPriv(PrivilegeContext context, String ctl, String db, String tbl,
+                                PrivPredicate wanted) {
         RangerHiveResource resource = new RangerHiveResource(HiveObjectType.TABLE,
                 ClusterNamespace.getNameFromFullName(db), tbl);
-        return checkPrivilege(currentUser, convertToAccessType(wanted), resource);
+        return checkPrivilege(context.getCurrentUser(), convertToAccessType(wanted), resource);
     }
 
     @Override
-    public void checkColsPriv(UserIdentity currentUser, String ctl, String db, String tbl, Set<String> cols,
-            PrivPredicate wanted) throws AuthorizationException {
+    public void checkColsPriv(PrivilegeContext context, String ctl, String db, String tbl, Set<String> cols,
+                              PrivPredicate wanted) throws AuthorizationException {
         List<RangerHiveResource> resources = new ArrayList<>();
         for (String col : cols) {
             RangerHiveResource resource = new RangerHiveResource(HiveObjectType.COLUMN,
                     ClusterNamespace.getNameFromFullName(db), tbl, col);
             resources.add(resource);
         }
 
-        checkPrivileges(currentUser, convertToAccessType(wanted), resources);
+        checkPrivileges(context.getCurrentUser(), convertToAccessType(wanted), resources);
     }
 
     @Override
-    public boolean checkCloudPriv(UserIdentity currentUser, String cloudName,
-            PrivPredicate wanted, ResourceTypeEnum type) {
+    public boolean checkCloudPriv(PrivilegeContext context, String cloudName,
+                                  PrivPredicate wanted, ResourceTypeEnum type) {
         return false;
     }
 
     @Override
-    public boolean checkStorageVaultPriv(UserIdentity currentUser, String storageVaultName, PrivPredicate wanted) {
+    public boolean checkStorageVaultPriv(PrivilegeContext context, String storageVaultName, PrivPredicate wanted) {
         return false;
     }
 
     @Override
-    public boolean checkResourcePriv(UserIdentity currentUser, String resourceName, PrivPredicate wanted) {
+    public boolean checkResourcePriv(PrivilegeContext context, String resourceName, PrivPredicate wanted) {
         return false;
     }
 
     @Override
-    public boolean checkWorkloadGroupPriv(UserIdentity currentUser, String workloadGroupName, PrivPredicate wanted) {
+    public boolean checkWorkloadGroupPriv(PrivilegeContext context, String workloadGroupName,
+                                          PrivPredicate wanted) {
         // Not support workload group privilege in ranger hive plugin.
         // So always return true to pass the check
         return true;
@@ -228,9 +231,10 @@ public static void main(String[] args) {
         RangerHiveAccessController ac = new RangerHiveAccessController(properties);
         UserIdentity user = new UserIdentity("user1", "127.0.0.1");
         user.setIsAnalyzed();
-        boolean res = ac.checkDbPriv(user, "hive", "tpcds_bin_partitioned_orc_1", PrivPredicate.SHOW);
+        PrivilegeContext context = PrivilegeContext.of(user);
+        boolean res = ac.checkDbPriv(context, "hive", "tpcds_bin_partitioned_orc_1", PrivPredicate.SHOW);
         System.out.println("res: " + res);
-        res = ac.checkTblPriv(user, "internal", "tpch1", "customer", PrivPredicate.SELECT);
+        res = ac.checkTblPriv(context, "internal", "tpch1", "customer", PrivPredicate.SELECT);
         System.out.println("res: " + res);
     }
 }

fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java

@@ -294,7 +294,7 @@ public long saveTransactionState(CountingDataOutputStream dos, long checksum) th
 
     public void checkCloudClusterPriv(String clusterName) throws DdlException {
         // check resource usage privilege
-        if (!Env.getCurrentEnv().getAccessManager().checkCloudPriv(ConnectContext.get().getCurrentUserIdentity(),
+        if (!Env.getCurrentEnv().getAccessManager().checkCloudPriv(ConnectContext.get(),
                 clusterName, PrivPredicate.USAGE, ResourceTypeEnum.CLUSTER)) {
             throw new DdlException("USAGE denied to user "
                     + ConnectContext.get().getCurrentUserIdentity().getQualifiedUser() + "'@'" + ConnectContext.get()

fe/fe-core/src/main/java/org/apache/doris/cloud/stage/StageUtil.java

@@ -78,11 +78,11 @@ public static StagePB getStage(String stage, String user, boolean checkAuth)
             return stagePBs.get(0);
         } else {
             // check stage permission
+            ConnectContext ctx = ConnectContext.get();
             if (checkAuth && !Env.getCurrentEnv().getAccessManager()
-                    .checkCloudPriv(ConnectContext.get().getCurrentUserIdentity(), stage, PrivPredicate.USAGE,
-                            ResourceTypeEnum.STAGE)) {
-                throw new AnalysisException("USAGE denied to user '" + ConnectContext.get().getQualifiedUser()
-                        + "'@'" + ConnectContext.get().getRemoteIP() + "' for cloud stage '" + stage + "'");
+                    .checkCloudPriv(ctx, stage, PrivPredicate.USAGE, ResourceTypeEnum.STAGE)) {
+                throw new AnalysisException("USAGE denied to user '" + ctx.getQualifiedUser()
+                        + "'@'" + ctx.getRemoteIP() + "' for cloud stage '" + stage + "'");
             }
             List<StagePB> stagePBs = ((CloudInternalCatalog) Env.getCurrentInternalCatalog())
                                             .getStage(StageType.EXTERNAL, null, stage, null);

fe/fe-core/src/main/java/org/apache/doris/common/FeNameFormat.java

@@ -159,10 +159,12 @@ public static void checkRoleName(String role, boolean canBeAdmin, String errMsg)
 
         boolean res = false;
         if (CaseSensibility.ROLE.getCaseSensibility()) {
-            res = role.equals(Role.OPERATOR_ROLE) || (!canBeAdmin && role.equals(Role.ADMIN_ROLE));
+            res = role.equals(Role.OPERATOR_ROLE)
+                    || (!canBeAdmin && (role.equals(Role.ADMIN_ROLE) || role.equals(Role.ADMIN_READONLY_ROLE)));
         } else {
             res = role.equalsIgnoreCase(Role.OPERATOR_ROLE)
-                    || (!canBeAdmin && role.equalsIgnoreCase(Role.ADMIN_ROLE));
+                    || (!canBeAdmin && (role.equalsIgnoreCase(Role.ADMIN_ROLE)
+                    || role.equalsIgnoreCase(Role.ADMIN_READONLY_ROLE)));
         }
 
         if (res || role.startsWith(RoleManager.DEFAULT_ROLE_PREFIX)) {

fe/fe-core/src/main/java/org/apache/doris/common/cache/NereidsSqlCacheManager.java

@@ -35,6 +35,7 @@
 import org.apache.doris.datasource.CatalogMgr;
 import org.apache.doris.datasource.hive.HMSExternalTable;
 import org.apache.doris.mysql.privilege.DataMaskPolicy;
+import org.apache.doris.mysql.privilege.PrivilegeContext;
 import org.apache.doris.mysql.privilege.RowFilterPolicy;
 import org.apache.doris.nereids.CascadesContext;
 import org.apache.doris.nereids.SqlCacheContext;
@@ -537,7 +538,10 @@ private boolean rowPoliciesChanged(UserIdentity currentUserIdentity, Env env, Sq
             List<? extends RowFilterPolicy> cachedPolicies = kv.getValue();
 
             List<? extends RowFilterPolicy> rowPolicies = env.getAccessManager().evalRowFilterPolicies(
-                    currentUserIdentity, qualifiedTable.catalog, qualifiedTable.db, qualifiedTable.table);
+                    PrivilegeContext.of(currentUserIdentity),
+                    qualifiedTable.catalog,
+                    qualifiedTable.db,
+                    qualifiedTable.table);
             if (!CollectionUtils.isEqualCollection(cachedPolicies, rowPolicies)) {
                 return true;
             }
@@ -552,7 +556,7 @@ private boolean dataMaskPoliciesChanged(
             Optional<DataMaskPolicy> cachedPolicy = kv.getValue();
 
             Optional<DataMaskPolicy> dataMaskPolicy = env.getAccessManager()
-                    .evalDataMaskPolicy(currentUserIdentity, qualifiedColumn.catalog,
+                    .evalDataMaskPolicy(PrivilegeContext.of(currentUserIdentity), qualifiedColumn.catalog,
                             qualifiedColumn.db, qualifiedColumn.table, qualifiedColumn.column);
             if (!Objects.equals(cachedPolicy, dataMaskPolicy)) {
                 return true;