Allow getting cache configuration from named collection

src/Disks/ObjectStorages/Cached/registerDiskCache.cpp

@@ -4,6 +4,7 @@
 #include <Common/logger_useful.h>
 #include <Common/assert_cast.h>
 #include <Common/filesystemHelpers.h>
+#include <Common/NamedCollections/NamedCollections.h>
 #include <Disks/DiskFactory.h>
 #include <Disks/ObjectStorages/Cached/CachedObjectStorage.h>
 #include <Disks/ObjectStorages/DiskObjectStorage.h>
@@ -39,7 +40,11 @@ void registerDiskCache(DiskFactory & factory, bool /* global_skip_access_check *
         }
 
         FileCacheSettings file_cache_settings;
-        file_cache_settings.loadFromConfig(config, config_prefix);
+        auto predefined_configuration = config.has("cache_name") ? NamedCollectionFactory::instance().tryGet(config.getString("cache_name")) : nullptr;
+        if (predefined_configuration)
+            file_cache_settings.loadFromCollection(*predefined_configuration);
+        else
+            file_cache_settings.loadFromConfig(config, config_prefix);
 
         auto config_fs_caches_dir = context->getFilesystemCachesPath();
         if (config_fs_caches_dir.empty())

src/Interpreters/Cache/FileCacheSettings.cpp

@@ -2,6 +2,7 @@
 
 #include <Poco/Util/AbstractConfiguration.h>
 #include <Common/Exception.h>
+#include <Common/NamedCollections/NamedCollections.h>
 #include <IO/ReadHelpers.h>
 
 namespace DB
@@ -26,10 +27,6 @@ void FileCacheSettings::loadFromConfig(const Poco::Util::AbstractConfiguration &
     if (max_size == 0)
         throw Exception(ErrorCodes::BAD_ARGUMENTS, "Expected non-zero size for cache configuration");
 
-    auto path = config.getString(config_prefix + ".path", "");
-    if (path.empty())
-        throw Exception(ErrorCodes::BAD_ARGUMENTS, "Disk Cache requires non-empty `path` field (cache base path) in config");
-
     max_elements = config.getUInt64(config_prefix + ".max_elements", FILECACHE_DEFAULT_MAX_ELEMENTS);
 
     if (config.has(config_prefix + ".max_file_segment_size"))
@@ -38,7 +35,6 @@ void FileCacheSettings::loadFromConfig(const Poco::Util::AbstractConfiguration &
     cache_on_write_operations = config.getUInt64(config_prefix + ".cache_on_write_operations", false);
     enable_filesystem_query_cache_limit = config.getUInt64(config_prefix + ".enable_filesystem_query_cache_limit", false);
     cache_hits_threshold = config.getUInt64(config_prefix + ".cache_hits_threshold", FILECACHE_DEFAULT_HITS_THRESHOLD);
-
     enable_bypass_cache_with_threshold = config.getUInt64(config_prefix + ".enable_bypass_cache_with_threshold", false);
 
     if (config.has(config_prefix + ".bypass_cache_threshold"))
@@ -54,4 +50,41 @@ void FileCacheSettings::loadFromConfig(const Poco::Util::AbstractConfiguration &
         load_metadata_threads = config.getUInt(config_prefix + ".load_metadata_threads");
 }
 
+void FileCacheSettings::loadFromCollection(const NamedCollection & collection)
+{
+    if (!collection.has("path"))
+        throw Exception(ErrorCodes::BAD_ARGUMENTS, "Expected cache path (`path`) in configuration");
+
+    base_path = collection.get<String>("path");
+
+    if (!collection.has("max_size"))
+        throw Exception(ErrorCodes::BAD_ARGUMENTS, "Expected cache size (`max_size`) in configuration");
+
+    max_size = parseWithSizeSuffix<uint64_t>(collection.get<String>("max_size"));
+    if (max_size == 0)
+        throw Exception(ErrorCodes::BAD_ARGUMENTS, "Expected non-zero size for cache configuration");
+
+    max_elements = collection.getOrDefault<UInt64>("max_elements", FILECACHE_DEFAULT_MAX_ELEMENTS);
+
+    if (collection.has("max_file_segment_size"))
+        max_file_segment_size = parseWithSizeSuffix<uint64_t>(collection.get<String>("max_file_segment_size"));
+
+    cache_on_write_operations = collection.getOrDefault<UInt64>("cache_on_write_operations", false);
+    enable_filesystem_query_cache_limit = collection.getOrDefault<UInt64>("enable_filesystem_query_cache_limit", false);
+    cache_hits_threshold = collection.getOrDefault<UInt64>("cache_hits_threshold", FILECACHE_DEFAULT_HITS_THRESHOLD);
+    enable_bypass_cache_with_threshold = collection.getOrDefault<UInt64>("enable_bypass_cache_with_threshold", false);
+
+    if (collection.has("bypass_cache_threshold"))
+        bypass_cache_threshold = parseWithSizeSuffix<uint64_t>(collection.get<String>("bypass_cache_threshold"));
+
+    if (collection.has("boundary_alignment"))
+        boundary_alignment = parseWithSizeSuffix<uint64_t>(collection.get<String>("boundary_alignment"));
+
+    if (collection.has("background_download_threads"))
+        background_download_threads = collection.get<UInt64>("background_download_threads");
+
+    if (collection.has("load_metadata_threads"))
+        load_metadata_threads = collection.get<UInt64>("load_metadata_threads");
+}
+
 }

src/Interpreters/Cache/FileCacheSettings.h

@@ -8,6 +8,7 @@ namespace Poco { namespace Util { class AbstractConfiguration; } } // NOLINT(cpp
 
 namespace DB
 {
+class NamedCollection;
 
 struct FileCacheSettings
 {
@@ -31,6 +32,7 @@ struct FileCacheSettings
     size_t load_metadata_threads = FILECACHE_DEFAULT_LOAD_METADATA_THREADS;
 
     void loadFromConfig(const Poco::Util::AbstractConfiguration & config, const std::string & config_prefix);
+    void loadFromCollection(const NamedCollection & collection);
 };
 
 }

tests/config/config.d/named_collection.xml

@@ -37,5 +37,9 @@
             <access_key_id>test</access_key_id>
             <secret_access_key>testtest</secret_access_key>
         </s3_conn_db>
+        <cache_collection>
+            <max_size>1Mi</max_size>
+            <path>collection</path>
+        </cache_collection>
     </named_collections>
 </clickhouse>

tests/queries/0_stateless/02908_filesystem_cache_as_collection.reference

@@ -0,0 +1,2 @@
+1048576	10000000	33554432	4194304	0	0	0	0	/var/lib/clickhouse/filesystem_caches/collection_sql	2	0	1
+1048576	10000000	33554432	4194304	0	0	0	0	/var/lib/clickhouse/filesystem_caches/collection	2	0	1

tests/queries/0_stateless/02908_filesystem_cache_as_collection.sql

@@ -0,0 +1,8 @@
+CREATE NAMED COLLECTION IF NOT EXISTS cache_collection_sql AS path = 'collection_sql', max_size = '1Mi';
+DROP TABLE IF EXISTS test;
+CREATE TABLE test (a Int32, b String)
+ENGINE = MergeTree() ORDER BY a SETTINGS disk = disk(type = cache, disk = 's3_disk', name = '$CLICHOUSE_TEST_UNIQUE_NAME', cache_name='cache_collection_sql');
+DESCRIBE FILESYSTEM CACHE '$CLICHOUSE_TEST_UNIQUE_NAME';
+CREATE TABLE test2 (a Int32, b String)
+ENGINE = MergeTree() ORDER BY a SETTINGS disk = disk(type = cache, disk = 's3_disk', name = '$CLICHOUSE_TEST_UNIQUE_NAME_2', cache_name='cache_collection');
+DESCRIBE FILESYSTEM CACHE '$CLICHOUSE_TEST_UNIQUE_NAME_2';