diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AddConstraintCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AddConstraintCommand.java
index 7f9d2c468f2fb1..5eeb3f80f4edc8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AddConstraintCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AddConstraintCommand.java
@@ -24,9 +24,12 @@
 import org.apache.doris.catalog.constraint.PrimaryKeyConstraint;
 import org.apache.doris.catalog.constraint.UniqueConstraint;
 import org.apache.doris.catalog.info.TableNameInfo;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.Pair;
 import org.apache.doris.info.TableNameInfoUtils;
 import org.apache.doris.mtmv.MTMVUtil;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.NereidsPlanner;
 import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.properties.PhysicalProperties;
@@ -71,6 +74,7 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         TableNameInfo tableNameInfo = TableNameInfoUtils.fromCatalogDb(
                 table.getDatabase().getCatalog(), table.getDatabase(), table);
         ImmutableList<String> columns = columnsAndTable.first;
+        checkAlterPriv(ctx, tableNameInfo);
 
         Pair<ImmutableList<String>, TableNameInfo> referencedColumnsAndTable = null;
         if (constraint.isForeignKey()) {
@@ -79,6 +83,8 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
             TableIf refTable = refColumnsAndTable.second;
             TableNameInfo refTableInfo = TableNameInfoUtils.fromCatalogDb(
                     refTable.getDatabase().getCatalog(), refTable.getDatabase(), refTable);
+            // a foreign key also registers a reverse reference on the referenced table
+            checkAlterPriv(ctx, refTableInfo);
             referencedColumnsAndTable = Pair.of(refColumnsAndTable.first, refTableInfo);
         }
         if (constraint.isForeignKey()) {
@@ -97,6 +103,16 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         }
     }
 
+    private void checkAlterPriv(ConnectContext ctx, TableNameInfo tableNameInfo)
+            throws org.apache.doris.common.AnalysisException {
+        if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, tableNameInfo.getCtl(),
+                tableNameInfo.getDb(), tableNameInfo.getTbl(), PrivPredicate.ALTER)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR, "ALTER",
+                    ctx.getQualifiedUser(), ctx.getRemoteIP(),
+                    tableNameInfo.getDb() + ": " + tableNameInfo.getTbl());
+        }
+    }
+
     private void addConstraintAndInvalidate(
             TableNameInfo tableNameInfo, org.apache.doris.catalog.constraint.Constraint constraint)
             throws Exception {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminRotateTdeRootKeyCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminRotateTdeRootKeyCommand.java
index 4952c6c586a7d4..469c871f3a96ed 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminRotateTdeRootKeyCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminRotateTdeRootKeyCommand.java
@@ -17,8 +17,12 @@
 
 package org.apache.doris.nereids.trees.plans.commands;
 
+import org.apache.doris.catalog.Env;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
 import org.apache.doris.encryption.KeyManagerInterface;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
 import org.apache.doris.qe.ConnectContext;
@@ -52,6 +56,7 @@ public AdminRotateTdeRootKeyCommand(Map<String, String> properties) {
 
     @Override
     public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
+        validate();
         KeyManagerInterface keyManager = ctx.getEnv().getKeyManager();
         if (keyManager != null) {
             keyManager.rotateRootKey(properties);
@@ -60,6 +65,16 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         }
     }
 
+    /**
+     * validate
+     */
+    public void validate() throws AnalysisException {
+        // check auth
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
+        }
+    }
+
     @Override
     public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
         return visitor.visitAdminRotateTdeRootKeyCommand(this, context);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminSetEncryptionRootKeyCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminSetEncryptionRootKeyCommand.java
index 5f98cd83b4c572..0e969799c2d666 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminSetEncryptionRootKeyCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminSetEncryptionRootKeyCommand.java
@@ -17,9 +17,13 @@
 
 package org.apache.doris.nereids.trees.plans.commands;
 
+import org.apache.doris.catalog.Env;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
 import org.apache.doris.encryption.EncryptionKey;
 import org.apache.doris.encryption.RootKeyInfo;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
 import org.apache.doris.qe.ConnectContext;
@@ -63,6 +67,11 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
      * validate
      */
     public void validate() throws AnalysisException {
+        // check auth
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
+        }
+
         if (properties == null || properties.isEmpty()) {
             throw new AnalysisException("The properties must not be empty");
         }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelWarmUpJobCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelWarmUpJobCommand.java
index cfb20910f11ccf..05d27a792a2877 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelWarmUpJobCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelWarmUpJobCommand.java
@@ -17,9 +17,13 @@
 
 package org.apache.doris.nereids.trees.plans.commands;
 
+import org.apache.doris.catalog.Env;
 import org.apache.doris.cloud.catalog.CloudEnv;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Config;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.analyzer.UnboundSlot;
 import org.apache.doris.nereids.trees.expressions.EqualTo;
 import org.apache.doris.nereids.trees.expressions.Expression;
@@ -59,6 +63,11 @@ public long getJobId() {
      * @throws AnalysisException check whether this sql is legal
      */
     public void validate(ConnectContext ctx) throws AnalysisException {
+        // check auth
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
+        }
+
         if (!Config.isCloudMode()) {
             throw new AnalysisException("The sql is illegal in disk mode ");
         }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateDictionaryCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateDictionaryCommand.java
index 6b8287200e7bc2..e09d3cf6da7032 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateDictionaryCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateDictionaryCommand.java
@@ -18,7 +18,12 @@
 package org.apache.doris.nereids.trees.plans.commands;
 
 import org.apache.doris.analysis.StmtType;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.dictionary.LayoutType;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.commands.info.CreateDictionaryInfo;
@@ -60,12 +65,20 @@ public StmtType stmtType() {
     }
 
     @Override
-    public void run(ConnectContext ctx, StmtExecutor executor) {
+    public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         try {
             // 1. Validate the dictionary info. names and existence.
             createDictionaryInfo.validateAndSet(ctx);
+        } catch (Exception e) {
+            LOG.warn("Failed to create dictionary: {}", e.getMessage());
+            throw new AnalysisException("Failed to create dictionary: " + e.getMessage());
+        }
+
+        // 2. Check auth. Must run after validateAndSet(), which fills in the default catalog/db names.
+        checkAuth(ctx);
 
-            // 2. Create dictionary and save it in manager. it will schedule data load.
+        try {
+            // 3. Create dictionary and save it in manager. it will schedule data load.
             ctx.getEnv().getDictionaryManager().createDictionary(ctx, createDictionaryInfo);
 
             LOG.info("Created dictionary {} in {} from {}", createDictionaryInfo.getDictName(),
@@ -75,4 +88,25 @@ public void run(ConnectContext ctx, StmtExecutor executor) {
             throw new AnalysisException("Failed to create dictionary: " + e.getMessage());
         }
     }
+
+    /**
+     * A dictionary is created in the internal catalog and its data is loaded by an internal task,
+     * so require CREATE on the dictionary itself and SELECT on the source table. Without the latter
+     * the dictionary would expose data the creator can not read.
+     */
+    private void checkAuth(ConnectContext ctx) throws org.apache.doris.common.AnalysisException {
+        if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, InternalCatalog.INTERNAL_CATALOG_NAME,
+                createDictionaryInfo.getDbName(), createDictionaryInfo.getDictName(), PrivPredicate.CREATE)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "CREATE");
+        }
+
+        String srcCtl = createDictionaryInfo.getSourceCtlName();
+        String srcDb = createDictionaryInfo.getSourceDbName();
+        String srcTbl = createDictionaryInfo.getSourceTableName();
+        if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, srcCtl, srcDb, srcTbl,
+                PrivPredicate.SELECT)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR, "SELECT",
+                    ctx.getQualifiedUser(), ctx.getRemoteIP(), srcDb + ": " + srcTbl);
+        }
+    }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropCatalogRecycleBinCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropCatalogRecycleBinCommand.java
index eaad053343ee56..48edc6bc306311 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropCatalogRecycleBinCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropCatalogRecycleBinCommand.java
@@ -18,6 +18,9 @@
 package org.apache.doris.nereids.trees.plans.commands;
 
 import org.apache.doris.catalog.Env;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
@@ -71,9 +74,21 @@ public DropCatalogRecycleBinCommand(IdType idType, long id) {
 
     @Override
     public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
+        validate();
         Env.getCurrentEnv().dropCatalogRecycleBin(idType, id);
     }
 
+    /**
+     * validate
+     */
+    public void validate() throws org.apache.doris.common.AnalysisException {
+        // Erasing from the recycle bin is irreversible and takes a raw object id, so it can not be
+        // authorized at db/table level. Restrict it to ADMIN, same as the other catalog-wide admin ops.
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
+        }
+    }
+
     @Override
     public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
         return visitor.visitDropCatalogRecycleBinCommand(this, context);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropConstraintCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropConstraintCommand.java
index fb8d506ef97c75..b1b5232af414fa 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropConstraintCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropConstraintCommand.java
@@ -22,8 +22,11 @@
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.catalog.constraint.Constraint;
 import org.apache.doris.catalog.info.TableNameInfo;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
 import org.apache.doris.info.TableNameInfoUtils;
 import org.apache.doris.mtmv.MTMVUtil;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.NereidsPlanner;
 import org.apache.doris.nereids.analyzer.UnboundRelation;
 import org.apache.doris.nereids.exceptions.AnalysisException;
@@ -75,6 +78,14 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
                     + "falling back to name-based lookup: {}", name, e.getMessage());
             tableNameInfo = extractTableNameFromPlan(ctx);
         }
+        // must be checked on both paths above: table resolution failing (which includes an
+        // authorization failure) falls back to a name-only lookup that binds nothing.
+        if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, tableNameInfo.getCtl(),
+                tableNameInfo.getDb(), tableNameInfo.getTbl(), PrivPredicate.ALTER)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR, "ALTER",
+                    ctx.getQualifiedUser(), ctx.getRemoteIP(),
+                    tableNameInfo.getDb() + ": " + tableNameInfo.getTbl());
+        }
         Constraint constraint = Env.getCurrentEnv().getConstraintManager().getConstraint(tableNameInfo, name);
         if (constraint == null) {
             throw new AnalysisException(
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropDictionaryCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropDictionaryCommand.java
index 23f58cf96642c9..3e7a564e58b6ec 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropDictionaryCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropDictionaryCommand.java
@@ -18,6 +18,11 @@
 package org.apache.doris.nereids.trees.plans.commands;
 
 import org.apache.doris.analysis.StmtType;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
@@ -57,10 +62,15 @@ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
     }
 
     @Override
-    public void run(ConnectContext ctx, StmtExecutor executor) {
+    public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         if (dbName == null) { // use current database
             dbName = ctx.getDatabase();
         }
+        // check auth. dictionaries always live in the internal catalog.
+        if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, InternalCatalog.INTERNAL_CATALOG_NAME,
+                dbName, dictName, PrivPredicate.DROP)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "DROP");
+        }
         try {
             ctx.getEnv().getDictionaryManager().dropDictionary(ctx, dbName, dictName, ifExists);
         } catch (Exception e) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropStageCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropStageCommand.java
index bd54e875456950..e38b40ffadb7d2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropStageCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropStageCommand.java
@@ -20,6 +20,10 @@
 import org.apache.doris.analysis.StmtType;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.cloud.catalog.CloudEnv;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
 import org.apache.doris.qe.ConnectContext;
@@ -45,9 +49,20 @@ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
 
     @Override
     public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
+        validate();
         ((CloudEnv) Env.getCurrentEnv()).dropStage(this);
     }
 
+    /**
+     * validate
+     */
+    public void validate() throws UserException {
+        // check auth. keep it aligned with CreateStageCommand.
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
+        }
+    }
+
     @Override
     public StmtType stmtType() {
         return StmtType.DROP;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/WarmUpClusterCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/WarmUpClusterCommand.java
index fd77fb41779de4..acd1f6f4d17497 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/WarmUpClusterCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/WarmUpClusterCommand.java
@@ -17,6 +17,7 @@
 
 package org.apache.doris.nereids.trees.plans.commands;
 
+import org.apache.doris.analysis.ResourceTypeEnum;
 import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.Database;
 import org.apache.doris.catalog.Env;
@@ -33,6 +34,7 @@
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.Triple;
 import org.apache.doris.common.UserException;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.commands.info.WarmUpItem;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
@@ -139,6 +141,14 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         handleWarmUp(ctx, executor);
     }
 
+    private void checkComputeGroupUsage(ConnectContext ctx, String computeGroup) throws AnalysisException {
+        if (!Env.getCurrentEnv().getAccessManager().checkCloudPriv(ctx.getCurrentUserIdentity(),
+                computeGroup, PrivPredicate.USAGE, ResourceTypeEnum.CLUSTER)) {
+            throw new AnalysisException("USAGE denied to user '" + ctx.getQualifiedUser() + "'@'"
+                    + ctx.getRemoteIP() + "' for compute group '" + computeGroup + "'");
+        }
+    }
+
     private void checkWarmupCgs(CloudSystemInfoService cloudSys) throws AnalysisException {
         if (!Strings.isNullOrEmpty(srcCluster)) {
             CloudComputeGroupMeta srcCg = cloudSys.getComputeGroupByName(srcCluster);
@@ -180,6 +190,13 @@ public void validate(ConnectContext connectContext) throws UserException {
             throw new UserException("The sql is just support in cloud mode");
         }
 
+        // check auth. warming up moves data between compute groups, so require USAGE on both ends
+        // instead of global ADMIN. Keep it aligned with UseCloudClusterCommand.
+        checkComputeGroupUsage(connectContext, dstCluster);
+        if (!Strings.isNullOrEmpty(srcCluster)) {
+            checkComputeGroupUsage(connectContext, srcCluster);
+        }
+
         CloudSystemInfoService cloudSys = ((CloudSystemInfoService) Env.getCurrentSystemInfo());
         if (!cloudSys.containClusterName(dstCluster)) {
             throw new AnalysisException("The dstClusterName " + dstCluster + " doesn't exist");
@@ -226,6 +243,12 @@ public void validate(ConnectContext connectContext) throws UserException {
                 if (table == null) {
                     ErrorReport.reportAnalysisException(ErrorCode.ERR_BAD_TABLE_ERROR, tableNameInfo.getTbl());
                 }
+                if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(connectContext, tableNameInfo.getCtl(),
+                        dbName, tableNameInfo.getTbl(), PrivPredicate.SELECT)) {
+                    ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR, "SELECT",
+                            connectContext.getQualifiedUser(), connectContext.getRemoteIP(),
+                            dbName + ": " + tableNameInfo.getTbl());
+                }
                 if (partitionName.length() != 0 && !table.containsPartition(partitionName)) {
                     throw new AnalysisException("The partition " + partitionName + " doesn't exist");
                 }
diff --git a/fe/fe-core/src/test/java/org/apache/doris/cloud/WarmUpClusterOnTablesParseTest.java b/fe/fe-core/src/test/java/org/apache/doris/cloud/WarmUpClusterOnTablesParseTest.java
index ceee9533386d4a..8e678302d3481d 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/cloud/WarmUpClusterOnTablesParseTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/cloud/WarmUpClusterOnTablesParseTest.java
@@ -17,6 +17,7 @@
 
 package org.apache.doris.cloud;
 
+import org.apache.doris.analysis.UserIdentity;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.cloud.OnTablesFilter.TableFilterRule;
 import org.apache.doris.cloud.OnTablesFilter.TableFilterRule.RuleType;
@@ -66,6 +67,11 @@ public static void init() throws Exception {
         originalSystemInfo = getField(env, Env.class, "systemInfo");
         connectContext = new ConnectContext();
         connectContext.setEnv(env);
+        // this test covers ON TABLES parsing and validation, not authorization, so give the
+        // context an identity and let it bypass the privilege checks in validate()
+        connectContext.setCurrentUserIdentity(UserIdentity.ROOT);
+        connectContext.setNoAuth(true);
+        connectContext.setSkipAuth(true);
         connectContext.setThreadLocalInfo();
     }
 
diff --git a/regression-test/suites/auth_call/test_ddl_constraint_auth.groovy b/regression-test/suites/auth_call/test_ddl_constraint_auth.groovy
new file mode 100644
index 00000000000000..6c522eab6d206f
--- /dev/null
+++ b/regression-test/suites/auth_call/test_ddl_constraint_auth.groovy
@@ -0,0 +1,83 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_ddl_constraint_auth", "p0,auth_call") {
+    String user = 'test_ddl_constraint_auth_user'
+    String pwd = 'C123_567p'
+    String dbName = 'test_ddl_constraint_auth_db'
+    String tableName = 'test_ddl_constraint_auth_tb'
+    String constraintName = 'test_ddl_constraint_auth_uk'
+
+    try_sql("DROP USER ${user}")
+    try_sql """drop database if exists ${dbName}"""
+    sql """CREATE USER '${user}' IDENTIFIED BY '${pwd}'"""
+    sql """grant select_priv on regression_test to ${user}"""
+    //cloud-mode
+    if (isCloudMode()) {
+        def clusters = sql " SHOW CLUSTERS; "
+        assertTrue(!clusters.isEmpty())
+        def validCluster = clusters[0][0]
+        sql """GRANT USAGE_PRIV ON CLUSTER `${validCluster}` TO ${user}""";
+    }
+
+    sql """create database ${dbName}"""
+    sql """
+        CREATE TABLE IF NOT EXISTS ${dbName}.${tableName} (
+            id BIGINT,
+            username VARCHAR(30)
+        )
+        DISTRIBUTED BY HASH(id) BUCKETS 2
+        PROPERTIES ("replication_num" = "1");
+        """
+
+    // SELECT alone must not be enough to change constraints
+    sql """grant SELECT_PRIV on ${dbName}.${tableName} to ${user}"""
+    connect(user, "${pwd}", context.config.jdbcUrl) {
+        sql """use ${dbName}"""
+        test {
+            sql """ALTER TABLE ${tableName} ADD CONSTRAINT ${constraintName} UNIQUE (id)"""
+            exception "denied"
+        }
+    }
+
+    sql """use ${dbName}"""
+    sql """ALTER TABLE ${tableName} ADD CONSTRAINT ${constraintName} UNIQUE (id)"""
+
+    // dropping a constraint is refused as well, including when the user can not even resolve
+    // the table (the name-based fallback path must not skip the check)
+    connect(user, "${pwd}", context.config.jdbcUrl) {
+        sql """use ${dbName}"""
+        test {
+            sql """ALTER TABLE ${tableName} DROP CONSTRAINT ${constraintName}"""
+            exception "denied"
+        }
+    }
+    def constraints = sql """SHOW CONSTRAINTS FROM ${dbName}.${tableName}"""
+    assertTrue(constraints.size() == 1)
+
+    sql """grant ALTER_PRIV on ${dbName}.${tableName} to ${user}"""
+    connect(user, "${pwd}", context.config.jdbcUrl) {
+        sql """use ${dbName}"""
+        sql """ALTER TABLE ${tableName} DROP CONSTRAINT ${constraintName}"""
+        sql """ALTER TABLE ${tableName} ADD CONSTRAINT ${constraintName} UNIQUE (id)"""
+    }
+    constraints = sql """SHOW CONSTRAINTS FROM ${dbName}.${tableName}"""
+    assertTrue(constraints.size() == 1)
+
+    sql """drop database if exists ${dbName}"""
+    try_sql("DROP USER ${user}")
+}
diff --git a/regression-test/suites/auth_call/test_ddl_dictionary_auth.groovy b/regression-test/suites/auth_call/test_ddl_dictionary_auth.groovy
new file mode 100644
index 00000000000000..394969acf73b3f
--- /dev/null
+++ b/regression-test/suites/auth_call/test_ddl_dictionary_auth.groovy
@@ -0,0 +1,106 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_ddl_dictionary_auth", "p0,auth_call") {
+    String user = 'test_ddl_dictionary_auth_user'
+    String pwd = 'C123_567p'
+    String dbName = 'test_ddl_dictionary_auth_db'
+    String tableName = 'test_ddl_dictionary_auth_tb'
+    String dictName = 'test_ddl_dictionary_auth_dict'
+
+    try_sql("DROP USER ${user}")
+    try_sql """drop database if exists ${dbName}"""
+    sql """CREATE USER '${user}' IDENTIFIED BY '${pwd}'"""
+    sql """grant select_priv on regression_test to ${user}"""
+    //cloud-mode
+    if (isCloudMode()) {
+        def clusters = sql " SHOW CLUSTERS; "
+        assertTrue(!clusters.isEmpty())
+        def validCluster = clusters[0][0]
+        sql """GRANT USAGE_PRIV ON CLUSTER `${validCluster}` TO ${user}""";
+    }
+
+    sql """create database ${dbName}"""
+    sql """
+        CREATE TABLE IF NOT EXISTS ${dbName}.${tableName} (
+            id BIGINT,
+            username VARCHAR(30)
+        )
+        DISTRIBUTED BY HASH(id) BUCKETS 2
+        PROPERTIES ("replication_num" = "1");
+        """
+    sql """insert into ${dbName}.${tableName} values(1, 'doris')"""
+
+    def createDictSql = """
+        CREATE DICTIONARY ${dbName}.${dictName} USING ${dbName}.${tableName}
+        (
+            id KEY,
+            username VALUE
+        )
+        LAYOUT(HASH_MAP)
+        PROPERTIES('data_lifetime'='600');
+        """
+
+    // no privilege at all on the database
+    connect(user, "${pwd}", context.config.jdbcUrl) {
+        test {
+            sql createDictSql
+            exception "denied"
+        }
+        test {
+            sql """DROP DICTIONARY ${dbName}.${dictName};"""
+            exception "denied"
+        }
+    }
+
+    // CREATE on the database is not enough: the dictionary load task runs as ADMIN, so the
+    // creator must also be able to read the source table itself.
+    sql """grant CREATE_PRIV on ${dbName}.* to ${user}"""
+    connect(user, "${pwd}", context.config.jdbcUrl) {
+        test {
+            sql createDictSql
+            exception "denied"
+        }
+    }
+
+    sql """grant SELECT_PRIV on ${dbName}.${tableName} to ${user}"""
+    connect(user, "${pwd}", context.config.jdbcUrl) {
+        sql createDictSql
+    }
+    sql """use ${dbName}"""
+    def dictRes = sql """SHOW DICTIONARIES"""
+    assertTrue(dictRes.size() == 1)
+
+    // dropping needs DROP on the database
+    connect(user, "${pwd}", context.config.jdbcUrl) {
+        test {
+            sql """DROP DICTIONARY ${dbName}.${dictName};"""
+            exception "denied"
+        }
+    }
+
+    sql """grant DROP_PRIV on ${dbName}.* to ${user}"""
+    connect(user, "${pwd}", context.config.jdbcUrl) {
+        sql """DROP DICTIONARY ${dbName}.${dictName};"""
+    }
+    sql """use ${dbName}"""
+    dictRes = sql """SHOW DICTIONARIES"""
+    assertTrue(dictRes.size() == 0)
+
+    sql """drop database if exists ${dbName}"""
+    try_sql("DROP USER ${user}")
+}
diff --git a/regression-test/suites/auth_call/test_recycle_bin_auth.groovy b/regression-test/suites/auth_call/test_recycle_bin_auth.groovy
new file mode 100644
index 00000000000000..ea99772dabf02e
--- /dev/null
+++ b/regression-test/suites/auth_call/test_recycle_bin_auth.groovy
@@ -0,0 +1,82 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_recycle_bin_auth", "p0,auth_call") {
+    String user = 'test_recycle_bin_auth_user'
+    String pwd = 'C123_567p'
+    String dbName = 'test_recycle_bin_auth_db'
+    String tableName = 'test_recycle_bin_auth_tb'
+
+    try_sql("DROP USER ${user}")
+    try_sql """drop database if exists ${dbName}"""
+    sql """CREATE USER '${user}' IDENTIFIED BY '${pwd}'"""
+    sql """grant select_priv on regression_test to ${user}"""
+    //cloud-mode
+    if (isCloudMode()) {
+        def clusters = sql " SHOW CLUSTERS; "
+        assertTrue(!clusters.isEmpty())
+        def validCluster = clusters[0][0]
+        sql """GRANT USAGE_PRIV ON CLUSTER `${validCluster}` TO ${user}""";
+    }
+
+    sql """create database ${dbName}"""
+    sql """
+        CREATE TABLE IF NOT EXISTS ${dbName}.${tableName} (
+            id BIGINT,
+            username VARCHAR(30)
+        )
+        DISTRIBUTED BY HASH(id) BUCKETS 2
+        PROPERTIES ("replication_num" = "1");
+        """
+    // give the user full privileges on the database. erasing the recycle bin must still be
+    // refused: it takes a raw object id and is irreversible, so it is ADMIN only.
+    sql """grant ALL on ${dbName}.* to ${user}"""
+
+    sql """drop table ${dbName}.${tableName}"""
+    def binRes = sql """SHOW CATALOG RECYCLE BIN WHERE NAME = "${tableName}" """
+    assertTrue(binRes.size() > 0)
+    def tableId = binRes[0][3]
+    logger.info("recycled table id: " + tableId)
+
+    connect(user, "${pwd}", context.config.jdbcUrl) {
+        test {
+            sql """DROP CATALOG RECYCLE BIN WHERE 'TableId' = ${tableId};"""
+            exception "denied"
+        }
+        test {
+            sql """DROP CATALOG RECYCLE BIN WHERE 'DbId' = ${tableId};"""
+            exception "denied"
+        }
+        test {
+            sql """DROP CATALOG RECYCLE BIN WHERE 'PartitionId' = ${tableId};"""
+            exception "denied"
+        }
+    }
+    // still there after the denied attempts
+    def afterRes = sql """SHOW CATALOG RECYCLE BIN WHERE NAME = "${tableName}" """
+    assertTrue(afterRes.size() == binRes.size())
+
+    sql """grant admin_priv on *.*.* to ${user}"""
+    connect(user, "${pwd}", context.config.jdbcUrl) {
+        sql """DROP CATALOG RECYCLE BIN WHERE 'TableId' = ${tableId};"""
+    }
+    def erasedRes = sql """SHOW CATALOG RECYCLE BIN WHERE NAME = "${tableName}" """
+    assertTrue(erasedRes.size() < binRes.size())
+
+    sql """drop database if exists ${dbName}"""
+    try_sql("DROP USER ${user}")
+}