[feature](cache) Add FileCache for RemoteFile

be/src/common/config.h

@@ -781,7 +781,16 @@ CONF_mBool(enable_function_pushdown, "false");
 CONF_Int32(cooldown_thread_num, "5");
 CONF_mInt64(generate_cooldown_task_interval_sec, "20");
 CONF_Int32(concurrency_per_dir, "2");
-CONF_mInt64(cooldown_lag_time_sec, "10800"); // 3h
+CONF_mInt64(cooldown_lag_time_sec, "10800");        // 3h
+CONF_mInt64(max_sub_cache_file_size, "1073741824"); // 1GB
+CONF_mInt64(file_cache_alive_time_sec, "604800");   // 1 week
+// file_cache_type is used to set the type of file cache for remote files.
+// "": no cache, "sub_file_cache": split sub files from remote file.
+// "whole_file_cache": the whole file.
+CONF_mString(file_cache_type, "");
+CONF_Validator(file_cache_type, [](const std::string config) -> bool {
+    return config == "sub_file_cache" || config == "whole_file_cache" || config == "";
+});
 
 CONF_Int32(s3_transfer_executor_pool_size, "2");
 

be/src/io/CMakeLists.txt

@@ -40,6 +40,9 @@ set(IO_FILES
     fs/local_file_writer.cpp
     fs/s3_file_reader.cpp
     fs/s3_file_system.cpp
+    cache/file_cache_manager.cpp
+    cache/sub_file_cache.cpp
+    cache/whole_file_cache.cpp
 )
 
 add_library(IO STATIC

be/src/io/cache/file_cache.h

@@ -0,0 +1,51 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <memory>
+#include <shared_mutex>
+
+#include "common/status.h"
+#include "io/fs/file_reader.h"
+#include "io/fs/path.h"
+
+namespace doris {
+namespace io {
+
+class FileCache : public FileReader {
+public:
+    FileCache() = default;
+    virtual ~FileCache() = default;
+
+    DISALLOW_COPY_AND_ASSIGN(FileCache);
+
+    virtual const Path& cache_dir() const = 0;
+
+    virtual size_t cache_file_size() const = 0;
+
+    virtual io::FileReaderSPtr remote_file_reader() const = 0;
+
+    virtual Status clean_timeout_cache() = 0;
+
+    virtual Status clean_all_cache() = 0;
+};
+
+using FileCachePtr = std::shared_ptr<FileCache>;
+
+} // namespace io
+} // namespace doris

be/src/io/cache/file_cache_manager.cpp

@@ -0,0 +1,62 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "io/cache/file_cache_manager.h"
+
+#include "io/cache/sub_file_cache.h"
+#include "io/cache/whole_file_cache.h"
+
+namespace doris {
+namespace io {
+
+void FileCacheManager::add_file_cache(const Path& cache_path, FileCachePtr file_cache) {
+    std::lock_guard<std::shared_mutex> wrlock(_cache_map_lock);
+    _file_cache_map.emplace(cache_path.native(), file_cache);
+}
+
+void FileCacheManager::remove_file_cache(const Path& cache_path) {
+    std::lock_guard<std::shared_mutex> wrlock(_cache_map_lock);
+    _file_cache_map.erase(cache_path.native());
+}
+
+void FileCacheManager::clean_timeout_caches() {
+    std::shared_lock<std::shared_mutex> rdlock(_cache_map_lock);
+    for (std::map<std::string, FileCachePtr>::const_iterator iter = _file_cache_map.cbegin();
+         iter != _file_cache_map.cend(); ++iter) {
+        iter->second->clean_timeout_cache();
+    }
+}
+
+FileCachePtr FileCacheManager::new_file_cache(const Path& cache_dir, int64_t alive_time_sec,
+                                              io::FileReaderSPtr remote_file_reader,
+                                              const std::string& file_cache_type) {
+    if (file_cache_type == "whole_file_cache") {
+        return std::make_unique<WholeFileCache>(cache_dir, alive_time_sec, remote_file_reader);
+    } else if (file_cache_type == "sub_file_cache") {
+        return std::make_unique<SubFileCache>(cache_dir, alive_time_sec, remote_file_reader);
+    } else {
+        return nullptr;
+    }
+}
+
+FileCacheManager* FileCacheManager::instance() {
+    static FileCacheManager cache_manager;
+    return &cache_manager;
+}
+
+} // namespace io
+} // namespace doris

be/src/io/cache/file_cache_manager.h

@@ -0,0 +1,53 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <memory>
+
+#include "common/config.h"
+#include "common/status.h"
+#include "io/cache/file_cache.h"
+
+namespace doris {
+namespace io {
+
+class FileCacheManager {
+public:
+    FileCacheManager() = default;
+    ~FileCacheManager() = default;
+
+    static FileCacheManager* instance();
+
+    void add_file_cache(const Path& cache_path, FileCachePtr file_cache);
+
+    void remove_file_cache(const Path& cache_path);
+
+    void clean_timeout_caches();
+
+    FileCachePtr new_file_cache(const Path& cache_dir, int64_t alive_time_sec,
+                                io::FileReaderSPtr remote_file_reader,
+                                const std::string& file_cache_type);
+
+private:
+    std::shared_mutex _cache_map_lock;
+    // cache_path -> FileCache
+    std::map<std::string, FileCachePtr> _file_cache_map;
+};
+
+} // namespace io
+} // namespace doris