[fix](filecache) Support memory file cache resource management

be/src/io/cache/block_file_cache.h

@@ -467,6 +467,9 @@ class BlockFileCache {
     size_t get_used_cache_size_unlocked(FileCacheType type,
                                         std::lock_guard<std::mutex>& cache_lock) const;
 
+    bool is_memory_storage() const;
+    size_t calc_size_percentage_unlocked(std::lock_guard<std::mutex>&) const;
+
     void check_disk_resource_limit();
     void check_need_evict_cache_in_advance();
 
@@ -539,8 +542,8 @@ class BlockFileCache {
     std::thread _cache_background_block_lru_update_thread;
     std::atomic_bool _async_open_done {false};
     // disk space or inode is less than the specified value
-    bool _disk_resource_limit_mode {false};
-    bool _need_evict_cache_in_advance {false};
+    std::atomic_bool _disk_resource_limit_mode {false};
+    std::atomic_bool _need_evict_cache_in_advance {false};
     bool _is_initialized {false};
 
     // strategy

be/src/io/cache/block_file_cache_factory.cpp

@@ -36,6 +36,7 @@
 #include <ostream>
 #include <utility>
 
+#include "common/cast_set.h"
 #include "common/config.h"
 #include "core/block/block.h"
 #include "information_schema/schema_scanner_helper.h"
@@ -312,8 +313,8 @@ std::vector<std::string> FileCacheFactory::get_base_paths() {
     return paths;
 }
 
-std::string validate_capacity(const std::string& path, int64_t new_capacity,
-                              int64_t& valid_capacity) {
+std::string validate_disk_capacity(const std::string& path, int64_t new_capacity,
+                                   int64_t& valid_capacity) {
     struct statfs stat;
     if (statfs(path.c_str(), &stat) < 0) {
         auto ret = fmt::format("reset capacity {} statfs error {}. ", path, strerror(errno));
@@ -340,16 +341,35 @@ std::string validate_capacity(const std::string& path, int64_t new_capacity,
     return "";
 }
 
+std::string validate_capacity(BlockFileCache* cache, int64_t new_capacity,
+                              int64_t& valid_capacity) {
+    if (cache->get_storage()->get_type() == FileCacheStorageType::MEMORY) {
+        valid_capacity = new_capacity;
+        if (valid_capacity <= 0) {
+            return fmt::format("The memory cache {} capacity must be greater than zero. ",
+                               cache->get_base_path());
+        }
+        return "";
+    }
+    return validate_disk_capacity(cache->get_base_path(), new_capacity, valid_capacity);
+}
+
 std::string FileCacheFactory::reset_capacity(const std::string& path, int64_t new_capacity) {
     std::stringstream ss;
     size_t total_capacity = 0;
     if (path.empty()) {
-        for (auto& [p, cache] : _path_to_cache) {
+        std::vector<std::pair<BlockFileCache*, size_t>> reset_targets;
+        reset_targets.reserve(_path_to_cache.size());
+        for (auto& entry : _path_to_cache) {
+            auto* cache = entry.second;
             int64_t valid_capacity = 0;
-            ss << validate_capacity(p, new_capacity, valid_capacity);
+            ss << validate_capacity(cache, new_capacity, valid_capacity);
             if (valid_capacity <= 0) {
                 return ss.str();
             }
+            reset_targets.emplace_back(cache, cast_set<size_t>(valid_capacity));
+        }
+        for (auto& [cache, valid_capacity] : reset_targets) {
             ss << cache->reset_capacity(valid_capacity);
             total_capacity += cache->capacity();
         }
@@ -358,11 +378,11 @@ std::string FileCacheFactory::reset_capacity(const std::string& path, int64_t ne
     } else {
         if (auto iter = _path_to_cache.find(path); iter != _path_to_cache.end()) {
             int64_t valid_capacity = 0;
-            ss << validate_capacity(path, new_capacity, valid_capacity);
+            ss << validate_capacity(iter->second, new_capacity, valid_capacity);
             if (valid_capacity <= 0) {
                 return ss.str();
             }
-            ss << iter->second->reset_capacity(valid_capacity);
+            ss << iter->second->reset_capacity(cast_set<size_t>(valid_capacity));
 
             for (auto& [p, cache] : _path_to_cache) {
                 total_capacity += cache->capacity();

be/src/io/cache/lru_queue_recorder.cpp

@@ -35,6 +35,9 @@ size_t file_cache_type_index(FileCacheType type) {
 void LRUQueueRecorder::record_queue_event(FileCacheType type, CacheLRULogType log_type,
                                           const UInt128Wrapper hash, const size_t offset,
                                           const size_t size) {
+    if (_mgr->is_memory_storage()) {
+        return;
+    }
     if (config::file_cache_background_lru_dump_tail_record_num <= 0) {
         return;
     }