[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,18 @@ 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文件目录",
+        "Directory path for storing admission rules JSON files"
+    })
+    public static String file_cache_admission_control_json_dir = "";
+
     @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;
@@ -221,6 +222,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

fe/fe-core/src/main/java/org/apache/doris/common/ConfigWatcher.java

@@ -32,6 +32,7 @@
 import java.nio.file.WatchEvent.Kind;
 import java.nio.file.WatchKey;
 import java.nio.file.WatchService;
+import java.util.function.Consumer;
 
 /*
  * used for watch config changed
@@ -41,6 +42,10 @@ public class ConfigWatcher extends Daemon {
 
     public final Path configPath;
 
+    private Consumer<Path> onCreateConsumer = null;
+    private Consumer<Path> onModifyConsumer = null;
+    private Consumer<Path> onDeleteConsumer = null;
+
     public ConfigWatcher(String configPathStr) {
         super("config watcher");
         Preconditions.checkState(!Strings.isNullOrEmpty(configPathStr));
@@ -95,16 +100,58 @@ protected void runOneCycle() {
         }
     }
 
+    public void setOnCreateConsumer(Consumer<Path> consumer) {
+        this.onCreateConsumer = consumer;
+    }
+
+    public void setOnModifyConsumer(Consumer<Path> consumer) {
+        this.onModifyConsumer = consumer;
+    }
+
+    public void setOnDeleteConsumer(Consumer<Path> consumer) {
+        this.onDeleteConsumer = consumer;
+    }
+
     private void handleCreate(Path filePath) {
-        // TODO(cmy): implement if needed
+        if (onCreateConsumer != null) {
+            try {
+                onCreateConsumer.accept(filePath);
+            } catch (Exception e) {
+                LOG.error("Error in onCreateConsumer for file created in directory: " + filePath, e);
+            }
+        } else {
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("File created in directory but no onCreateConsumer set: " + filePath);
+            }
+        }
     }
 
     private void handleDelete(Path filePath) {
-        // TODO(cmy): implement if needed
+        if (onDeleteConsumer != null) {
+            try {
+                onDeleteConsumer.accept(filePath);
+            } catch (Exception e) {
+                LOG.error("Error in onDeleteConsumer for file deleted from directory: " + filePath, e);
+            }
+        } else {
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("File deleted from directory but no onDeleteConsumer set: " + filePath);
+            }
+        }
     }
 
     private void handleModify(Path filePath) {
-        // TODO(cmy): implement if needed
+        if (onModifyConsumer != null) {
+            try {
+                onModifyConsumer.accept(filePath);
+            } catch (Exception e) {
+                LOG.error("Error in onModifyConsumer for file modified in directory: " + filePath, e);
+            }
+        } else {
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("File modified in directory but no onModifyConsumer set: " + filePath);
+            }
+        }
     }
 
     // for test