[enhancement](Nereids) Enable parse sql from sql cache

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -17,8 +17,6 @@
 
 package org.apache.doris.common;
 
-import org.apache.doris.common.ConfigBase.ConfField;
-
 public class Config extends ConfigBase {
 
     @ConfField(description = {"用户自定义配置文件的路径，用于存放 fe_custom.conf。该文件中的配置会覆盖 fe.conf 中的配置",
@@ -2004,6 +2002,14 @@ public class Config extends ConfigBase {
                     + "the old load statement will be degraded."})
     public static boolean enable_nereids_load = false;
 
+    /**
+     * the plan cache num which can be reused for the next query
+     */
+    @ConfField(mutable = false, varType = VariableAnnotation.EXPERIMENTAL, description = {
+            "当前默认设置为 100，用来控制控制NereidsSqlCacheManager管理的sql cache数量。",
+            "Now default set to 100, this config is used to control the number of "
+                    + "sql cache managed by NereidsSqlCacheManager"})
+    public static int sql_cache_manage_num = 100;
 
     /**
      * Maximum number of events to poll in each RPC.

fe/fe-common/src/main/java/org/apache/doris/common/Pair.java

@@ -43,6 +43,10 @@ private Pair(F first, S second) {
         this.second = second;
     }
 
+    public static <P, K extends P> Pair<K, K> ofSame(K same) {
+        return new Pair<>(same, same);
+    }
+
     public static <F, S> Pair<F, S> of(F first, S second) {
         return new Pair<>(first, second);
     }

fe/fe-core/src/main/java/org/apache/doris/blockrule/SqlBlockRuleMgr.java

@@ -269,10 +269,10 @@ public void checkLimitations(Long partitionNum, Long tabletNum, Long cardinality
             return;
         }
         // match global rule
-        List<SqlBlockRule> globalRules =
-                nameToSqlBlockRuleMap.values().stream().filter(SqlBlockRule::getGlobal).collect(Collectors.toList());
-        for (SqlBlockRule rule : globalRules) {
-            checkLimitations(rule, partitionNum, tabletNum, cardinality);
+        for (SqlBlockRule rule : nameToSqlBlockRuleMap.values()) {
+            if (rule.getGlobal()) {
+                checkLimitations(rule, partitionNum, tabletNum, cardinality);
+            }
         }
         // match user rule
         String[] bindSqlBlockRules = Env.getCurrentEnv().getAuth().getSqlBlockRules(user);

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

@@ -105,6 +105,7 @@
 import org.apache.doris.common.FeConstants;
 import org.apache.doris.common.FeMetaVersion;
 import org.apache.doris.common.MetaNotFoundException;
+import org.apache.doris.common.NereidsSqlCacheManager;
 import org.apache.doris.common.Pair;
 import org.apache.doris.common.ThreadPoolManager;
 import org.apache.doris.common.UserException;
@@ -530,6 +531,8 @@ public class Env {
 
     private DNSCache dnsCache;
 
+    private final NereidsSqlCacheManager sqlCacheManager;
+
     public List<TFrontendInfo> getFrontendInfos() {
         List<TFrontendInfo> res = new ArrayList<>();
 
@@ -766,6 +769,9 @@ public Env(boolean isCheckpointCatalog) {
         this.mtmvService = new MTMVService();
         this.insertOverwriteManager = new InsertOverwriteManager();
         this.dnsCache = new DNSCache();
+        this.sqlCacheManager = new NereidsSqlCacheManager(
+                Config.sql_cache_manage_num, Config.cache_last_version_interval_second
+        );
     }
 
     public static void destroyCheckpoint() {
@@ -6092,6 +6098,10 @@ public MasterDaemon getTabletStatMgr() {
         return tabletStatMgr;
     }
 
+    public NereidsSqlCacheManager getSqlCacheManager() {
+        return sqlCacheManager;
+    }
+
     public void alterMTMVRefreshInfo(AlterMTMVRefreshInfo info) {
         AlterMTMV alter = new AlterMTMV(info.getMvName(), info.getRefreshInfo(), MTMVAlterOpType.ALTER_REFRESH_INFO);
         this.alter.processAlterMTMV(alter, false);