Merge branch 'apache:master' into s7

Author: Tech-Circle-48

be/src/http/action/file_cache_action.cpp

@@ -58,6 +58,7 @@ constexpr static std::string_view BASE_PATH = "base_path";
 constexpr static std::string_view RELEASED_ELEMENTS = "released_elements";
 constexpr static std::string_view DUMP = "dump";
 constexpr static std::string_view VALUE = "value";
+constexpr static std::string_view RELOAD = "reload";
 
 Status FileCacheAction::_handle_header(HttpRequest* req, std::string* json_metrics) {
     req->add_output_header(HttpHeaders::CONTENT_TYPE, HEADER_JSON.data());
@@ -161,6 +162,41 @@ Status FileCacheAction::_handle_header(HttpRequest* req, std::string* json_metri
                 *json_metrics = json.ToString();
             }
         }
+    } else if (operation == RELOAD) {
+#ifdef BE_TEST
+        std::string doris_home = getenv("DORIS_HOME");
+        std::string conffile = std::string(doris_home) + "/conf/be.conf";
+        if (!doris::config::init(conffile.c_str(), true, true, true)) {
+            return Status::InternalError("Error reading config file");
+        }
+
+        std::string custom_conffile = doris::config::custom_config_dir + "/be_custom.conf";
+        if (!doris::config::init(custom_conffile.c_str(), true, false, false)) {
+            return Status::InternalError("Error reading custom config file");
+        }
+
+        if (!doris::config::enable_file_cache) {
+            return Status::InternalError("config::enbale_file_cache should be true!");
+        }
+
+        std::unordered_set<std::string> cache_path_set;
+        std::vector<doris::CachePath> cache_paths;
+        RETURN_IF_ERROR(doris::parse_conf_cache_paths(doris::config::file_cache_path, cache_paths));
+
+        std::vector<CachePath> cache_paths_no_dup;
+        cache_paths_no_dup.reserve(cache_paths.size());
+        for (const auto& cache_path : cache_paths) {
+            if (cache_path_set.contains(cache_path.path)) {
+                LOG(WARNING) << fmt::format("cache path {} is duplicate", cache_path.path);
+                continue;
+            }
+            cache_path_set.emplace(cache_path.path);
+            cache_paths_no_dup.emplace_back(cache_path);
+        }
+        RETURN_IF_ERROR(doris::io::FileCacheFactory::instance()->reload_file_cache(cache_paths));
+#else
+        return Status::InternalError("Do not use reload in production environment!!!!");
+#endif
     } else {
         st = Status::InternalError("invalid operation: {}", operation);
     }

be/src/io/cache/block_file_cache_factory.cpp

@@ -120,6 +120,42 @@ Status FileCacheFactory::create_file_cache(const std::string& cache_base_path,
     return Status::OK();
 }
 
+Status FileCacheFactory::reload_file_cache(const std::vector<CachePath>& cache_base_paths) {
+    {
+        std::unique_lock lock(_mtx);
+        for (const auto& cache_path : cache_base_paths) {
+            if (_path_to_cache.find(cache_path.path) == _path_to_cache.end()) {
+                return Status::InternalError(
+                        "Current file cache not support file cache num changes");
+            }
+        }
+
+        for (const auto& cache_path : cache_base_paths) {
+            auto cache_map_iter = _path_to_cache.find(cache_path.path);
+            auto cache_iter = std::find_if(_caches.begin(), _caches.end(),
+                                           [cache_map_iter](const auto& cache_uptr) {
+                                               return cache_uptr.get() == cache_map_iter->second;
+                                           });
+
+            if (cache_iter == _caches.end()) {
+                return Status::InternalError("Target relaod cache in path {} may has been released",
+                                             cache_path.path);
+            }
+
+            // deconstruct target reload first
+            *cache_iter = std::unique_ptr<BlockFileCache>();
+            // after deconstruct the BlockFileCache, construct the BlockFileCache again
+            *cache_iter =
+                    std::make_unique<BlockFileCache>(cache_path.path, cache_path.init_settings());
+            cache_map_iter->second = cache_iter->get();
+
+            RETURN_IF_ERROR(cache_iter->get()->initialize());
+        }
+    }
+
+    return Status::OK();
+}
+
 std::vector<doris::CacheBlockPB> FileCacheFactory::get_cache_data_by_path(const std::string& path) {
     auto cache_hash = BlockFileCache::hash(path);
     return get_cache_data_by_path(cache_hash);

be/src/io/cache/block_file_cache_factory.h

@@ -30,6 +30,7 @@
 #include "gen_cpp/internal_service.pb.h"
 #include "io/cache/block_file_cache.h"
 #include "io/cache/file_cache_common.h"
+#include "olap/options.h"
 namespace doris {
 class TUniqueId;
 
@@ -49,6 +50,8 @@ class FileCacheFactory {
     Status create_file_cache(const std::string& cache_base_path,
                              FileCacheSettings file_cache_settings);
 
+    Status reload_file_cache(const std::vector<CachePath>& cache_base_paths);
+
     size_t try_release();
 
     size_t try_release(const std::string& base_path);

be/src/io/cache/fs_file_cache_storage.cpp

@@ -26,6 +26,7 @@
 #include <filesystem>
 #include <mutex>
 #include <system_error>
+#include <vector>
 
 #include "common/logging.h"
 #include "common/status.h"
@@ -877,8 +878,8 @@ void FSFileCacheStorage::load_cache_info_into_memory(BlockFileCache* _mgr) const
     // If the difference is more than threshold, load from filesystem as well
     if (estimated_file_count > 100) {
         double difference_ratio =
-                static_cast<double>(estimated_file_count) -
-                static_cast<double>(db_block_count) / static_cast<double>(estimated_file_count);
+                (static_cast<double>(estimated_file_count) - static_cast<double>(db_block_count)) /
+                static_cast<double>(estimated_file_count);
 
         if (difference_ratio > config::file_cache_meta_store_vs_file_system_diff_num_threshold) {
             LOG(WARNING) << "Significant difference between DB blocks (" << db_block_count
@@ -983,13 +984,53 @@ size_t FSFileCacheStorage::estimate_file_count_from_statfs() const {
     // Get total size of cache directory to estimate file count
     std::error_code ec;
     uintmax_t total_size = 0;
-    for (const auto& entry : std::filesystem::recursive_directory_iterator(_cache_base_path, ec)) {
+    std::vector<std::filesystem::path> pending_dirs {std::filesystem::path(_cache_base_path)};
+    while (!pending_dirs.empty()) {
+        auto current_dir = pending_dirs.back();
+        pending_dirs.pop_back();
+
+        std::filesystem::directory_iterator it(current_dir, ec);
         if (ec) {
-            LOG(WARNING) << "Error accessing directory entry: " << ec.message();
+            LOG(WARNING) << "Failed to list directory while estimating file count, dir="
+                         << current_dir << ", err=" << ec.message();
+            ec.clear();
             continue;
         }
-        if (entry.is_regular_file()) {
-            total_size += entry.file_size();
+
+        for (; it != std::filesystem::directory_iterator(); ++it) {
+            std::error_code status_ec;
+            auto entry_status = it->symlink_status(status_ec);
+            TEST_SYNC_POINT_CALLBACK(
+                    "FSFileCacheStorage::estimate_file_count_from_statfs::AfterEntryStatus",
+                    &status_ec);
+            if (status_ec) {
+                LOG(WARNING) << "Failed to stat entry while estimating file count, path="
+                             << it->path() << ", err=" << status_ec.message();
+                continue;
+            }
+
+            if (std::filesystem::is_directory(entry_status)) {
+                auto next_dir = it->path();
+                TEST_SYNC_POINT_CALLBACK(
+                        "FSFileCacheStorage::estimate_file_count_from_statfs::OnDirectory",
+                        &next_dir);
+                pending_dirs.emplace_back(next_dir);
+                continue;
+            }
+
+            if (std::filesystem::is_regular_file(entry_status)) {
+                std::error_code size_ec;
+                auto file_size = it->file_size(size_ec);
+                TEST_SYNC_POINT_CALLBACK(
+                        "FSFileCacheStorage::estimate_file_count_from_statfs::AfterFileSize",
+                        &size_ec);
+                if (size_ec) {
+                    LOG(WARNING) << "Failed to get file size while estimating file count, path="
+                                 << it->path() << ", err=" << size_ec.message();
+                    continue;
+                }
+                total_size += file_size;
+            }
         }
     }