[improvement](statistics)Support drop cached stats.

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

@@ -3280,6 +3280,10 @@ drop_stmt ::=
     {:
         RESULT = new DropStatsStmt(tbl, cols, partitionNames);
     :}
+    | KW_DROP KW_CACHED KW_STATS table_name:tbl
+    {:
+        RESULT = new DropCachedStatsStmt(tbl);
+    :}
     | KW_DROP KW_EXPIRED KW_STATS
     {:
         RESULT = new DropStatsStmt(true);

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

@@ -0,0 +1,121 @@
+// 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.DatabaseIf;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.qe.ConnectContext;
+
+/**
+ * Manually drop cached statistics for table and its mv.
+ * <p>
+ * syntax:
+ * DROP CACHED STATS TableName;
+ */
+public class DropCachedStatsStmt extends DdlStmt {
+
+    private final TableName tableName;
+
+    private long catalogId;
+    private long dbId;
+    private long tblId;
+
+    public DropCachedStatsStmt(TableName tableName) {
+        this.tableName = tableName;
+    }
+
+    @Override
+    public void analyze(Analyzer analyzer) throws UserException {
+        super.analyze(analyzer);
+        if (tableName == null) {
+            throw new UserException("Should specify a valid table name.");
+        }
+        tableName.analyze(analyzer);
+        String catalogName = tableName.getCtl();
+        String dbName = tableName.getDb();
+        String tblName = tableName.getTbl();
+        CatalogIf catalog = analyzer.getEnv().getCatalogMgr()
+                .getCatalogOrAnalysisException(catalogName);
+        DatabaseIf db = catalog.getDbOrAnalysisException(dbName);
+        TableIf table = db.getTableOrAnalysisException(tblName);
+        tblId = table.getId();
+        dbId = db.getId();
+        catalogId = catalog.getId();
+        // check permission
+        checkAnalyzePriv(catalogName, db.getFullName(), table.getName());
+    }
+
+    public long getTblId() {
+        return tblId;
+    }
+
+    public long getDbId() {
+        return dbId;
+    }
+
+    public long getCatalogIdId() {
+        return catalogId;
+    }
+
+    @Override
+    public String toSql() {
+        StringBuilder sb = new StringBuilder();
+        sb.append("DROP CACHED STATS ");
+
+        if (tableName != null) {
+            sb.append(tableName.toSql());
+        }
+
+        return sb.toString();
+    }
+
+    @Override
+    public String toString() {
+        return toSql();
+    }
+
+    private void checkAnalyzePriv(String catalogName, String dbName, String tblName) throws AnalysisException {
+        if (!Env.getCurrentEnv().getAccessManager()
+                .checkTblPriv(ConnectContext.get(), catalogName, dbName, tblName,
+                PrivPredicate.DROP)) {
+            ErrorReport.reportAnalysisException(
+                    ErrorCode.ERR_TABLEACCESS_DENIED_ERROR,
+                    "DROP",
+                    ConnectContext.get().getQualifiedUser(),
+                    ConnectContext.get().getRemoteIP(),
+                    dbName + "." + tblName);
+        }
+    }
+
+    @Override
+    public StmtType stmtType() {
+        return StmtType.DROP;
+    }
+
+    @Override
+    public RedirectStatus getRedirectStatus() {
+        return RedirectStatus.NO_FORWARD;
+    }
+}

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

@@ -87,6 +87,7 @@
 import org.apache.doris.analysis.CreateWorkloadSchedPolicyStmt;
 import org.apache.doris.analysis.DdlStmt;
 import org.apache.doris.analysis.DropAnalyzeJobStmt;
+import org.apache.doris.analysis.DropCachedStatsStmt;
 import org.apache.doris.analysis.DropCatalogStmt;
 import org.apache.doris.analysis.DropDbStmt;
 import org.apache.doris.analysis.DropEncryptKeyStmt;
@@ -406,6 +407,8 @@ public static void execute(Env env, DdlStmt ddlStmt) throws Exception {
             ProfileManager.getInstance().cleanProfile();
         } else if (ddlStmt instanceof DropStatsStmt) {
             env.getAnalysisManager().dropStats((DropStatsStmt) ddlStmt);
+        } else if (ddlStmt instanceof DropCachedStatsStmt) {
+            env.getAnalysisManager().dropCachedStats((DropCachedStatsStmt) ddlStmt);
         } else if (ddlStmt instanceof KillAnalysisJobStmt) {
             env.getAnalysisManager().handleKillAnalyzeStmt((KillAnalysisJobStmt) ddlStmt);
         } else if (ddlStmt instanceof CleanQueryStatsStmt) {

fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java

@@ -22,6 +22,7 @@
 import org.apache.doris.analysis.AnalyzeStmt;
 import org.apache.doris.analysis.AnalyzeTblStmt;
 import org.apache.doris.analysis.DropAnalyzeJobStmt;
+import org.apache.doris.analysis.DropCachedStatsStmt;
 import org.apache.doris.analysis.DropStatsStmt;
 import org.apache.doris.analysis.KillAnalysisJobStmt;
 import org.apache.doris.analysis.PartitionNames;
@@ -641,6 +642,13 @@ private ThreadPoolExecutor createThreadPoolForSyncAnalyze() {
                 StatisticsUtil.getAnalyzeTimeout()));
     }
 
+    public void dropCachedStats(DropCachedStatsStmt stmt) {
+        long catalogId = stmt.getCatalogIdId();
+        long dbId = stmt.getDbId();
+        long tblId = stmt.getTblId();
+        dropCachedStats(catalogId, dbId, tblId);
+    }
+
     public void dropStats(DropStatsStmt dropStatsStmt) throws DdlException {
         if (dropStatsStmt.dropExpired) {
             Env.getCurrentEnv().getStatisticsCleaner().clear();
@@ -719,6 +727,27 @@ public void submitAsyncDropStatsTask(long catalogId, long dbId, long tableId,
         }
     }
 
+    public void dropCachedStats(long catalogId, long dbId, long tableId) {
+        TableIf table = StatisticsUtil.findTable(catalogId, dbId, tableId);
+        StatisticsCache statsCache = Env.getCurrentEnv().getStatisticsCache();
+        Set<String> columns = table.getSchemaAllIndexes(false)
+                .stream().map(Column::getName).collect(Collectors.toSet());
+        for (String column : columns) {
+            List<Long> indexIds = Lists.newArrayList();
+            if (table instanceof OlapTable) {
+                indexIds = ((OlapTable) table).getMvColumnIndexIds(column);
+            } else {
+                indexIds.add(-1L);
+            }
+            for (long indexId : indexIds) {
+                statsCache.invalidateColumnStatsCache(catalogId, dbId, tableId, indexId, column);
+                for (String part : table.getPartitionNames()) {
+                    statsCache.invalidatePartitionColumnStatsCache(catalogId, dbId, tableId, indexId, part, column);
+                }
+            }
+        }
+    }
+
     public void invalidateLocalStats(long catalogId, long dbId, long tableId, Set<String> columns,
                                      TableStatsMeta tableStats, PartitionNames partitionNames) {
         if (tableStats == null) {

regression-test/suites/statistics/test_drop_cached_stats.groovy

@@ -0,0 +1,58 @@
+// 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_drop_cached_stats") {
+
+    sql """drop database if exists test_drop_cached_stats"""
+    sql """create database test_drop_cached_stats"""
+    sql """use test_drop_cached_stats"""
+    sql """set global force_sample_analyze=false"""
+    sql """set global enable_auto_analyze=false"""
+
+    sql """CREATE TABLE drop_cache_test (
+            key1 int NOT NULL,
+            value1 varchar(25) NOT NULL,
+            value2 varchar(125) NOT NULL
+        )ENGINE=OLAP
+        DUPLICATE KEY(`key1`)
+        COMMENT "OLAP"
+        DISTRIBUTED BY HASH(`key1`) BUCKETS 2
+        PROPERTIES (
+            "replication_num" = "1"
+        )
+    """
+    createMV("create materialized view mv1 as select key1 from drop_cache_test;")
+
+    sql """insert into drop_cache_test values (1, "1", "1")"""
+    sql """analyze table drop_cache_test with sync"""
+
+    def result = sql """show column stats drop_cache_test"""
+    assertEquals(4, result.size())
+    result = sql """show column cached stats drop_cache_test"""
+    assertEquals(4, result.size())
+
+    sql """drop cached stats drop_cache_test"""
+    result = sql """show column cached stats drop_cache_test"""
+    assertEquals(0, result.size())
+
+    result = sql """show column stats drop_cache_test"""
+    assertEquals(4, result.size())
+
+
+    sql """drop database if exists test_drop_cached_stats"""
+}
+