[Improvement](coldhot) add statement to show objects which use storage policy

fe/fe-core/src/main/cup/sql_parser.cup

@@ -3939,6 +3939,14 @@ show_stmt ::=
     {:
         RESULT = new ShowPolicyStmt(PolicyTypeEnum.STORAGE, null, null);
     :}
+    | KW_SHOW KW_STORAGE KW_POLICY KW_USING
+    {:
+        RESULT = new ShowStoragePolicyUsingStmt(null);
+    :}
+    | KW_SHOW KW_STORAGE KW_POLICY KW_USING KW_FOR ident:policy
+    {:
+        RESULT = new ShowStoragePolicyUsingStmt(policy);
+    :}
     ;
 
 show_param ::=

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

@@ -0,0 +1,78 @@
+// 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.
+
+package org.apache.doris.analysis;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.ScalarType;
+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.qe.ConnectContext;
+import org.apache.doris.qe.ShowResultSetMetaData;
+
+import lombok.Getter;
+
+/**
+ * Show objects where storage policy is used
+ * syntax:
+ * SHOW STORAGE POLICY USING [for policy_name]
+ **/
+public class ShowStoragePolicyUsingStmt extends ShowStmt {
+
+    public static final ShowResultSetMetaData RESULT_META_DATA =
+            ShowResultSetMetaData.builder()
+                .addColumn(new Column("PolicyName", ScalarType.createVarchar(100)))
+                .addColumn(new Column("Database", ScalarType.createVarchar(20)))
+                .addColumn(new Column("Table", ScalarType.createVarchar(20)))
+                .addColumn(new Column("Partitions", ScalarType.createVarchar(60)))
+                .build();
+    @Getter
+    private final String policyName;
+
+    public ShowStoragePolicyUsingStmt(String policyName) {
+        this.policyName = policyName;
+    }
+
+    @Override
+    public void analyze(Analyzer analyzer) throws UserException {
+        super.analyze(analyzer);
+
+        // check auth
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
+        }
+    }
+
+    @Override
+    public String toSql() {
+        StringBuilder sb = new StringBuilder();
+        sb.append("SHOW STORAGE POLICY USING");
+        if (policyName != null) {
+            sb.append(" FOR ").append(policyName);
+        }
+
+        return sb.toString();
+    }
+
+    @Override
+    public ShowResultSetMetaData getMetaData() {
+        return RESULT_META_DATA;
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/policy/PolicyMgr.java

@@ -22,6 +22,7 @@
 import org.apache.doris.analysis.CreatePolicyStmt;
 import org.apache.doris.analysis.DropPolicyStmt;
 import org.apache.doris.analysis.ShowPolicyStmt;
+import org.apache.doris.analysis.ShowStoragePolicyUsingStmt;
 import org.apache.doris.analysis.UserIdentity;
 import org.apache.doris.catalog.Database;
 import org.apache.doris.catalog.Env;
@@ -55,7 +56,9 @@
 import java.io.DataOutput;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -485,6 +488,113 @@ public ShowResultSet showPolicy(ShowPolicyStmt showStmt) throws AnalysisExceptio
         }
     }
 
+    /**
+     * Show objects which is using the storage policy
+     **/
+    public ShowResultSet showStoragePolicyUsing(ShowStoragePolicyUsingStmt showStmt) throws AnalysisException {
+        List<List<String>> rows = Lists.newArrayList();
+        String targetPolicyName = showStmt.getPolicyName();
+
+        readLock();
+        try {
+            List<Database> databases = Env.getCurrentEnv().getInternalCatalog().getDbs();
+            // show for all storage policies
+            if (Strings.isNullOrEmpty(targetPolicyName)) {
+                for (Database db : databases) {
+                    List<Table> tables = db.getTables();
+                    for (Table table : tables) {
+                        if (!(table instanceof OlapTable)) {
+                            continue;
+                        }
+
+                        Map<String, List<String>> policyToPartitionsMap = new HashMap<>();
+                        OlapTable olapTable = (OlapTable) table;
+                        PartitionInfo partitionInfo = olapTable.getPartitionInfo();
+                        // classify a table's all partitions by storage policy
+                        for (Long partitionId : olapTable.getPartitionIds()) {
+                            String policyName = partitionInfo.getDataProperty(partitionId).getStoragePolicy();
+                            if (StringUtils.isEmpty(policyName)) {
+                                continue;
+                            }
+                            if (policyToPartitionsMap.containsKey(policyName)) {
+                                policyToPartitionsMap.get(policyName)
+                                        .add(olapTable.getPartition(partitionId).getName());
+                            } else {
+                                List<String> partitionList = new ArrayList<>();
+                                partitionList.add(olapTable.getPartition(partitionId).getName());
+                                policyToPartitionsMap.put(policyName, partitionList);
+                            }
+                        }
+
+                        //output, all partitions with same storage policy in a table will be shown in one line
+                        if (policyToPartitionsMap.size() == 1) {
+                            String[] policyArray = policyToPartitionsMap.keySet().toArray(new String[0]);
+                            List<String> partitionsList = new ArrayList<>(policyToPartitionsMap.values()).get(0);
+                            if (partitionsList.size() == olapTable.getPartitionNum()) {
+                                List<String> row = Arrays.asList(policyArray[0],
+                                        ClusterNamespace.getNameFromFullName(db.getFullName()), olapTable.getName(),
+                                        "ALL");
+                                rows.add(row);
+                            } else {
+                                List<String> row = Arrays.asList(policyArray[0],
+                                        ClusterNamespace.getNameFromFullName(db.getFullName()), olapTable.getName(),
+                                        String.join(",", partitionsList));
+                                rows.add(row);
+                            }
+                        } else {
+                            for (Map.Entry<String, List<String>> entry : policyToPartitionsMap.entrySet()) {
+                                List<String> row = Arrays.asList(entry.getKey(),
+                                        ClusterNamespace.getNameFromFullName(db.getFullName()), olapTable.getName(),
+                                        String.join(",", entry.getValue()));
+                                rows.add(row);
+                            }
+                        }
+                    }
+                }
+            } else {
+                // show for specific storage policy
+                for (Database db : databases) {
+                    List<Table> tables = db.getTables();
+                    for (Table table : tables) {
+                        if (!(table instanceof OlapTable)) {
+                            continue;
+                        }
+
+                        OlapTable olapTable = (OlapTable) table;
+                        int partitionMatchNum = 0;
+                        StringBuilder matchPartitionsSB = new StringBuilder();
+                        PartitionInfo partitionInfo = olapTable.getPartitionInfo();
+                        for (Long partitionId : olapTable.getPartitionIds()) {
+                            String policyName = partitionInfo.getDataProperty(partitionId).getStoragePolicy();
+                            if (policyName.equals(targetPolicyName)) {
+                                partitionMatchNum++;
+                                matchPartitionsSB.append(olapTable.getPartition(partitionId).getName()).append(",");
+                            }
+                        }
+
+                        if (partitionMatchNum == 0) {
+                            continue;
+                        }
+
+                        String matchPartitionsStr = "ALL";
+                        if (partitionMatchNum < olapTable.getPartitionNum()) {
+                            matchPartitionsStr = matchPartitionsSB.toString();
+                            matchPartitionsStr = matchPartitionsStr.substring(0, matchPartitionsStr.length() - 1);
+                        }
+
+                        List<String> row = Arrays.asList(targetPolicyName,
+                                ClusterNamespace.getNameFromFullName(db.getFullName()), olapTable.getName(),
+                                matchPartitionsStr);
+                        rows.add(row);
+                    }
+                }
+            }
+            return new ShowResultSet(showStmt.getMetaData(), rows);
+        } finally {
+            readUnlock();
+        }
+    }
+
     private void addTablePolicies(RowPolicy policy) {
         if (policy.getUser() != null) {
             policy.getUser().setIsAnalyzed();

fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java

@@ -89,6 +89,7 @@
 import org.apache.doris.analysis.ShowSnapshotStmt;
 import org.apache.doris.analysis.ShowSqlBlockRuleStmt;
 import org.apache.doris.analysis.ShowStmt;
+import org.apache.doris.analysis.ShowStoragePolicyUsingStmt;
 import org.apache.doris.analysis.ShowStreamLoadStmt;
 import org.apache.doris.analysis.ShowSyncJobStmt;
 import org.apache.doris.analysis.ShowTableCreationStmt;
@@ -422,6 +423,8 @@ public ShowResultSet execute() throws AnalysisException {
             handleShowCreateMaterializedView();
         } else if (stmt instanceof ShowPolicyStmt) {
             handleShowPolicy();
+        } else if (stmt instanceof ShowStoragePolicyUsingStmt) {
+            handleShowStoragePolicyUsing();
         } else if (stmt instanceof ShowCatalogStmt) {
             handleShowCatalogs();
         } else if (stmt instanceof ShowCreateCatalogStmt) {
@@ -2696,6 +2699,11 @@ public void handleShowPolicy() throws AnalysisException {
         resultSet = Env.getCurrentEnv().getPolicyMgr().showPolicy(showStmt);
     }
 
+    public void handleShowStoragePolicyUsing() throws AnalysisException {
+        ShowStoragePolicyUsingStmt showStmt = (ShowStoragePolicyUsingStmt) stmt;
+        resultSet = Env.getCurrentEnv().getPolicyMgr().showStoragePolicyUsing(showStmt);
+    }
+
     public void handleShowCatalogs() throws AnalysisException {
         ShowCatalogStmt showStmt = (ShowCatalogStmt) stmt;
         resultSet = Env.getCurrentEnv().getCatalogMgr().showCatalogs(showStmt, ctx.getCurrentCatalog() != null