[feature](cache) support file cache admission control

be/src/io/file_factory.cpp

@@ -67,7 +67,7 @@ io::FileReaderOptions FileFactory::get_reader_options(RuntimeState* state,
     };
     if (config::enable_file_cache && state != nullptr &&
         state->query_options().__isset.enable_file_cache &&
-        state->query_options().enable_file_cache) {
+        state->query_options().enable_file_cache && fd.file_cache_admission) {
         opts.cache_type = io::FileCachePolicy::FILE_BLOCK_CACHE;
     }
     if (state != nullptr && state->query_options().__isset.file_cache_base_path &&

be/src/io/file_factory.h

@@ -71,6 +71,7 @@ struct FileDescription {
     // because for a hive table, differenet partitions may have different
     // locations(or fs), so different files may have different fs.
     std::string fs_name;
+    bool file_cache_admission = true;
 };
 
 } // namespace io

be/src/vec/exec/format/csv/csv_reader.cpp

@@ -223,6 +223,7 @@ void CsvReader::_init_file_description() {
     if (_range.__isset.fs_name) {
         _file_description.fs_name = _range.fs_name;
     }
+    _file_description.file_cache_admission = _range.file_cache_admission;
 }
 
 Status CsvReader::init_reader(bool is_load) {

be/src/vec/exec/format/json/new_json_reader.cpp

@@ -164,6 +164,7 @@ void NewJsonReader::_init_file_description() {
     if (_range.__isset.fs_name) {
         _file_description.fs_name = _range.fs_name;
     }
+    _file_description.file_cache_admission = _range.file_cache_admission;
 }
 
 Status NewJsonReader::init_reader(

be/src/vec/exec/format/orc/vorc_reader.cpp

@@ -1437,6 +1437,7 @@ void OrcReader::_init_file_description() {
     if (_scan_range.__isset.fs_name) {
         _file_description.fs_name = _scan_range.fs_name;
     }
+    _file_description.file_cache_admission = _scan_range.file_cache_admission;
 }
 
 DataTypePtr OrcReader::convert_to_doris_type(const orc::Type* orc_type) {

be/src/vec/exec/format/parquet/vparquet_reader.cpp

@@ -321,6 +321,7 @@ void ParquetReader::_init_file_description() {
     if (_scan_range.__isset.fs_name) {
         _file_description.fs_name = _scan_range.fs_name;
     }
+    _file_description.file_cache_admission = _scan_range.file_cache_admission;
 }
 
 Status ParquetReader::init_reader(

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

@@ -3750,6 +3750,30 @@ public static int metaServiceRpcRetryTimes() {
     })
     public static int first_error_msg_max_length = 256;
 
+    @ConfField(mutable = false, description = {
+        "是否启用文件缓存准入控制",
+        "Whether to enable file cache admission control"
+    })
+    public static boolean enable_file_cache_admission_control = false;
+
+    @ConfField(mutable = true, description = {
+        "存储准入规则的JSON文件路径",
+        "JSON file path for storing admission rules"
+    })
+    public static String file_cache_admission_control_json_file_path = "";
+
+    @ConfField(mutable = true, description = {
+        "准入规则自动刷新间隔（秒），设置为0关闭自动刷新",
+        "Auto-refresh interval for admission policies (seconds). Set to 0 to disable auto refresh"
+    })
+    public static int file_cache_admission_control_fresh_interval_s = 300;
+
+    @ConfField(mutable = true, description = {
+        "当没有匹配的准入规则时的默认行为：true=允许，false=拒绝",
+        "Default behavior when no matching admission rule is found: true=allow, false=deny"
+    })
+    public static boolean file_cache_admission_control_default_allow = false;
+
     @ConfField
     public static String cloud_snapshot_handler_class = "org.apache.doris.cloud.snapshot.CloudSnapshotHandler";
     @ConfField

fe/fe-core/src/main/java/org/apache/doris/DorisFE.java

@@ -31,6 +31,7 @@
 import org.apache.doris.common.util.JdkUtils;
 import org.apache.doris.common.util.NetUtils;
 import org.apache.doris.common.util.Util;
+import org.apache.doris.datasource.FileCacheAdmissionManager;
 import org.apache.doris.httpv2.HttpServer;
 import org.apache.doris.journal.bdbje.BDBDebugger;
 import org.apache.doris.journal.bdbje.BDBTool;
@@ -160,6 +161,10 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star
                 serverReady.set(false);
                 gracefulShutdown();
 
+                if (Config.enable_file_cache_admission_control) {
+                    FileCacheAdmissionManager.getInstance().shutdown();
+                }
+
                 // Shutdown HTTP server after main process graceful shutdown is complete
                 if (httpServer != null) {
                     httpServer.shutdown();
@@ -221,6 +226,10 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star
             Env.getCurrentEnv().initialize(args);
             Env.getCurrentEnv().waitForReady();
 
+            if (Config.enable_file_cache_admission_control) {
+                FileCacheAdmissionManager.getInstance().loadOnStartup();
+            }
+
             // init and start:
             // 1. HttpServer for HTTP Server
             // 2. FeServer for Thrift Server