[fix](auth) Add privilege checks to SHOW/EXPLAIN/REFRESH DICTIONARY

### What problem does this PR solve?

Issue Number: None

Related PR: #66218

Problem Summary:

`SHOW DICTIONARIES` and `EXPLAIN DICTIONARY` did not check any privilege. Any
user who can `USE` a database (which only needs a privilege on some table of
that database) could list every dictionary of the database together with its
source table name, status and BE data distribution, and describe its columns.
`REFRESH DICTIONARY` only failed inside the internal `INSERT INTO`, after the
dictionary had been looked up and switched to `LOADING`. Reading values through
`dict_get()` / `dict_get_many()` never checked any privilege at all.

This is inconsistent with `SHOW TABLES`, which hides tables the user cannot
show, and with `CREATE/DROP DICTIONARY`, which already require privileges on the
dictionary name (#66218).

Dictionaries are authorized like tables of the internal catalog, so:

- `SHOW DICTIONARIES` now skips dictionaries the user has no `SHOW` privilege
  on, the same way `SHOW TABLES` filters tables.
- `EXPLAIN DICTIONARY` requires `SHOW` on the dictionary, like `DESCRIBE` on a
  table.
- `REFRESH DICTIONARY` checks `LOAD` on the dictionary and column-aware
  `SELECT` on the dictionary's source columns up front (`checkColumnsPriv` on
  the source columns' own names, the same contract the generated `INSERT`
  enforces through `CheckPrivileges`, so column-level grants keep working even
  when the dictionary definition spells the columns differently). These are the
  privileges the reload already required; the checks now happen before the
  dictionary is resolved and before its status is flipped to `LOADING`, so an
  unauthorized request can no longer block concurrent refreshes while its
  INSERT is being planned.
- `dict_get()` / `dict_get_many()` require `SELECT` on the dictionary before
  resolving it, so a user who cannot see a dictionary cannot read its values or
  probe whether it exists.
- Statements that read a dictionary are kept out of the sql cache and are not
  short-circuited as reusable point-query plans of a server prepared statement:
  neither mechanism records the dictionary's privilege or version, so a cached
  result / reused plan could survive a revoke or a refresh. Dictionaries are a
  cache already, so the lost sql-cache hit is not a real cost.
- `SHOW DICTIONARIES` returns early when the visible set is empty: an empty id
  list means "all dictionaries" to the BE status RPC (`get_dictionary_status`),
  so the previous code fanned status RPCs to every alive BE (also reachable
  before this change via an empty database or a non-matching `LIKE`).

The checks run before the dictionary is looked up, so a denied user cannot
probe whether a dictionary exists either.

### Release note

None

### Check List (For Author)

- Test
    - [x] Regression test: `auth_call/test_ddl_dictionary_auth` now covers a
      user with a privilege on another table of the database (must not see,
      describe or refresh the dictionary), `SHOW_VIEW` on the database (sees
      the dictionary and its source table, may describe it, still cannot
      refresh), `LOAD` on the database without `SELECT` on the source table
      (rejected before the dictionary starts loading), column-level `SELECT`
      on the source columns (may refresh), and `dict_get()` denied until
      `SELECT` on the dictionary is granted. New
      `auth_call/test_dictionary_read_auth_cache` (nonConcurrent) covers a
      dictionary spelling its columns differently from the source table, a
      primed sql-cache statement that is denied right after the revoke, and a
      server prepared point query whose second EXECUTE is denied after the
      revoke.
    - [ ] Unit Test
    - [ ] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:

- Behavior changed:
    - [ ] No.
    - [x] Yes. Users without `SHOW` on a dictionary no longer see it in
      `SHOW DICTIONARIES` and cannot `EXPLAIN DICTIONARY` it. `REFRESH
      DICTIONARY` still needs `LOAD` on the dictionary and (column-aware)
      `SELECT` on the source columns, but is now rejected before the dictionary
      is touched. `dict_get()` / `dict_get_many()` now require `SELECT` on the
      dictionary; internal paths without a user context are unaffected.
      Statements reading a dictionary no longer use the sql cache or a reused
      short-circuit point-query plan.

- Does this need documentation?
    - [ ] No.
    - [x] Yes. The privilege requirements of the three statements should be
      documented.

Claude-Session: https://claude.ai/code/session_01X9KukfTLYxHmP6iYyEnQtW

Author: mrhhsg

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/ExpressionAnalyzer.java

@@ -78,6 +78,8 @@
 import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
 import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
 import org.apache.doris.nereids.trees.expressions.functions.agg.SupportMultiDistinct;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.DictGet;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.DictGetMany;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.ElementAt;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.Lambda;
 import org.apache.doris.nereids.trees.expressions.functions.udf.AliasUdfBuilder;
@@ -588,8 +590,12 @@ public Expression visitUnboundFunction(UnboundFunction unboundFunction, Expressi
         }
         if (wantToParseSqlFromSqlCache) {
             sqlCacheContext = context.cascadesContext.getStatementContext().getSqlCacheContext();
+            // Dictionary reads depend on the dictionary's privilege and version, which the sql cache
+            // neither records nor revalidates on a hit, so keep such statements out of it.
             if (builder instanceof AliasUdfBuilder
-                    || buildResult.second instanceof Udf) {
+                    || buildResult.second instanceof Udf
+                    || buildResult.second instanceof DictGet
+                    || buildResult.second instanceof DictGetMany) {
                 if (sqlCacheContext.isPresent()) {
                     sqlCacheContext.get().setCannotProcessExpression(true);
                 }

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/LogicalResultSinkToShortCircuitPointQuery.java

@@ -26,9 +26,12 @@
 import org.apache.doris.nereids.trees.expressions.EqualTo;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.DictGet;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.DictGetMany;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
 import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
 import org.apache.doris.qe.ConnectContext;
 
 import com.google.common.collect.ImmutableList;
@@ -72,6 +75,14 @@ private boolean scanMatchShortCircuitCondition(LogicalOlapScan olapScan) {
                         && olapTable.storeRowColumn();
     }
 
+    // A short-circuit plan of a server prepared statement is reused by later EXECUTEs without
+    // re-analysis. Dictionary reads depend on the dictionary's privilege and version, neither of
+    // which the reuse check tracks, so such point queries take the normal path every time.
+    private boolean projectReadsDictionary(LogicalProject<?> project) {
+        return project.getProjects().stream().anyMatch(
+                expr -> expr.anyMatch(e -> e instanceof DictGet || e instanceof DictGetMany));
+    }
+
     // set short circuit flag and return the original plan
     private Plan shortCircuit(Plan root, OlapTable olapTable,
                 Set<Expression> conjuncts, StatementContext statementContext) {
@@ -98,8 +109,10 @@ public List<Rule> buildRules() {
                             .when(this::scanMatchShortCircuitCondition)
                     ).when(this::filterMatchShortCircuitCondition)))
                         .thenApply(ctx -> {
+                            if (projectReadsDictionary(ctx.root.child())) {
+                                return ctx.root;
+                            }
                             return shortCircuit(ctx.root, ctx.root.child().child().child().getTable(),
-
                                         ctx.root.child().child().getConjuncts(), ctx.statementContext);
                         })),
                 RuleType.SHOR_CIRCUIT_POINT_QUERY.build(

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/DictGet.java

@@ -21,9 +21,11 @@
 import org.apache.doris.catalog.FunctionSignature;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.Pair;
+import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.dictionary.Dictionary;
 import org.apache.doris.dictionary.DictionaryManager;
 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.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.functions.AlwaysNullable;
@@ -32,6 +34,7 @@
 import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
 import org.apache.doris.nereids.types.DataType;
 import org.apache.doris.nereids.util.TypeCoercionUtils;
+import org.apache.doris.qe.ConnectContext;
 
 import com.google.common.base.Preconditions;
 
@@ -93,6 +96,16 @@ public Pair<FunctionSignature, Dictionary> customSignatureDict() {
         String dictName = firstNames[1];
         String colName = ((Literal) getArgument(1)).getStringValue();
 
+        // Reading dictionary values must be authorized like reading its source data. Check before
+        // the lookup so a caller without the privilege cannot even probe whether the dictionary
+        // exists. ConnectContext may be absent on internal paths, which carry no user to check.
+        ConnectContext connectContext = ConnectContext.get();
+        if (connectContext != null && !Env.getCurrentEnv().getAccessManager().checkTblPriv(connectContext,
+                InternalCatalog.INTERNAL_CATALOG_NAME, dbName, dictName, PrivPredicate.SELECT)) {
+            throw new AnalysisException("SELECT command denied to user " + connectContext.getQualifiedUser()
+                    + " for dictionary '" + dbName + "." + dictName + "'");
+        }
+
         Dictionary dictionary;
         try {
             dictionary = dicMgr.getDictionary(dbName, dictName);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/DictGetMany.java

@@ -22,9 +22,11 @@
 import org.apache.doris.catalog.FunctionSignature;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.Pair;
+import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.dictionary.Dictionary;
 import org.apache.doris.dictionary.DictionaryManager;
 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.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.functions.AlwaysNotNullable;
@@ -36,6 +38,7 @@
 import org.apache.doris.nereids.types.StructField;
 import org.apache.doris.nereids.types.StructType;
 import org.apache.doris.nereids.util.TypeCoercionUtils;
+import org.apache.doris.qe.ConnectContext;
 
 import com.google.common.base.Preconditions;
 
@@ -99,6 +102,16 @@ public Pair<FunctionSignature, Dictionary> customSignatureDict() {
         String dictName = firstNames[1];
         List<Literal> colNames = ((ArrayLiteral) getArgument(1)).getValue();
 
+        // Reading dictionary values must be authorized like reading its source data. Check before
+        // the lookup so a caller without the privilege cannot even probe whether the dictionary
+        // exists. ConnectContext may be absent on internal paths, which carry no user to check.
+        ConnectContext connectContext = ConnectContext.get();
+        if (connectContext != null && !Env.getCurrentEnv().getAccessManager().checkTblPriv(connectContext,
+                InternalCatalog.INTERNAL_CATALOG_NAME, dbName, dictName, PrivPredicate.SELECT)) {
+            throw new AnalysisException("SELECT command denied to user " + connectContext.getQualifiedUser()
+                    + " for dictionary '" + dbName + "." + dictName + "'");
+        }
+
         Dictionary dictionary;
         try {
             dictionary = dicMgr.getDictionary(dbName, dictName);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExplainDictionaryCommand.java

@@ -18,10 +18,16 @@
 package org.apache.doris.nereids.trees.plans.commands;
 
 import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.DdlException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.dictionary.Dictionary;
 import org.apache.doris.dictionary.DictionaryManager;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.commands.info.DictionaryColumnDefinition;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
@@ -62,11 +68,17 @@ public ShowResultSetMetaData getMetaData() {
     }
 
     @Override
-    public ShowResultSet doRun(ConnectContext ctx, StmtExecutor executor) throws DdlException {
+    public ShowResultSet doRun(ConnectContext ctx, StmtExecutor executor) throws DdlException, AnalysisException {
         List<List<String>> rows = Lists.newArrayList();
 
         DictionaryManager dictionaryManager = ctx.getEnv().getDictionaryManager();
         String db = dbName == null ? ctx.getDatabase() : dbName;
+        // Describing a dictionary exposes its schema, so require SHOW on it like DESCRIBE on a table.
+        if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, InternalCatalog.INTERNAL_CATALOG_NAME,
+                db, dictionaryName, PrivPredicate.SHOW)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR, "SHOW",
+                    ctx.getQualifiedUser(), ctx.getRemoteIP(), db + ": " + dictionaryName);
+        }
         Dictionary dictionary = dictionaryManager.getDictionary(db, dictionaryName);
 
         for (DictionaryColumnDefinition column : dictionary.getDicColumns()) {

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowDictionariesCommand.java

@@ -19,12 +19,15 @@
 
 import org.apache.doris.analysis.RedirectStatus;
 import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.ScalarType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.PatternMatcher;
 import org.apache.doris.common.PatternMatcherWrapper;
+import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.dictionary.Dictionary;
 import org.apache.doris.dictionary.DictionaryManager;
+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;
@@ -73,14 +76,31 @@ public ShowResultSet doRun(ConnectContext ctx, StmtExecutor executor) throws Ana
 
         DictionaryManager dictionaryManager = ctx.getEnv().getDictionaryManager();
         List<Dictionary> queryDicts = Lists.newArrayList();
+        String dbName = ctx.getDatabase();
         // getDictionaries() already have read lock
-        Map<String, Dictionary> dbDictionaries = dictionaryManager.getDictionaries(ctx.getDatabase());
+        Map<String, Dictionary> dbDictionaries = dictionaryManager.getDictionaries(dbName);
         for (Map.Entry<String, Dictionary> entry : dbDictionaries.entrySet()) {
             String dictionaryName = entry.getKey();
             // Apply wild condition filtering if wild pattern is provided
-            if (wild == null || matcher.match(dictionaryName)) {
-                queryDicts.add(entry.getValue());
+            if (wild != null && !matcher.match(dictionaryName)) {
+                continue;
             }
+            // Dictionaries are authorized like tables of the internal catalog. Hide the ones the user
+            // may not show, the same way SHOW TABLES hides tables, so the source table name, status
+            // and data distribution are not exposed to users without privileges on the dictionary.
+            if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, InternalCatalog.INTERNAL_CATALOG_NAME,
+                    dbName, dictionaryName, PrivPredicate.SHOW)) {
+                continue;
+            }
+            queryDicts.add(entry.getValue());
+        }
+
+        // An empty id list means "all dictionaries" to the BE status RPC (get_dictionary_status in
+        // BackendService.thrift), so an empty visible set must return before status collection:
+        // otherwise it fans RPCs to every alive BE for dictionaries this user may not see, logs
+        // them as missing, and a bad response from any BE would fail the whole command.
+        if (queryDicts.isEmpty()) {
+            return new ShowResultSet(getMetaData(), rows);
         }
 
         // ignore its return value because we dont update it, just show.

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/refresh/RefreshDictionaryCommand.java

@@ -18,15 +18,24 @@
 package org.apache.doris.nereids.trees.plans.commands.refresh;
 
 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.Dictionary;
 import org.apache.doris.dictionary.DictionaryManager;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.commands.Command;
 import org.apache.doris.nereids.trees.plans.commands.ForwardWithSync;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.StmtExecutor;
 
+import java.util.LinkedHashSet;
+import java.util.Set;
+import java.util.stream.Collectors;
+
 /**
  * Refresh table command.
  */
@@ -45,7 +54,28 @@ public RefreshDictionaryCommand(String dbName, String dicName) {
     public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         DictionaryManager dictionaryManager = ctx.getEnv().getDictionaryManager();
         String db = dbName == null ? ctx.getDatabase() : dbName;
+        // The reload is an INSERT INTO the dictionary executed as the current user, which requires
+        // LOAD on the dictionary (and SELECT on the source table). Check LOAD up front so a user
+        // without it can neither probe whether the dictionary exists nor flip it to LOADING.
+        if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, InternalCatalog.INTERNAL_CATALOG_NAME,
+                db, dictionaryName, PrivPredicate.LOAD)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR, "LOAD",
+                    ctx.getQualifiedUser(), ctx.getRemoteIP(), db + ": " + dictionaryName);
+        }
         Dictionary dictionary = dictionaryManager.getDictionary(db, dictionaryName);
+        // The reload also reads the source table as the current user, and its INSERT authorizes
+        // exactly the dictionary's source columns through the column-aware contract (BindSink
+        // projects the source to the dictionary schema before CheckPrivileges runs). Use the same
+        // contract and the same column spelling here: the dictionary definition may spell a column
+        // differently from the source table, but column grants are compared against the source
+        // column's own name. Reject a missing privilege before dataLoad() publishes the LOADING
+        // status and blocks concurrent refreshes.
+        Set<String> sourceColumns = dictionary.getDicColumns().stream()
+                .map(definition -> definition.getOriginColumn().getName())
+                .collect(Collectors.toCollection(LinkedHashSet::new));
+        Env.getCurrentEnv().getAccessManager().checkColumnsPriv(ctx, dictionary.getSourceCtlName(),
+                dictionary.getSourceDbName(), dictionary.getSourceTableName(), sourceColumns,
+                PrivPredicate.SELECT);
         dictionaryManager.dataLoad(ctx, dictionary, false);
     }