Merge pull request #743 from Altinity/list_objects_cache

Enmk · arthurpassos · commit e92ba1984a44 · 2025-05-23T14:45:08.000-03:00
Antalya: Cache the list objects operation on object storage using a TTL + prefix matching cache implementation
diff --git a/programs/server/Server.cpp b/programs/server/Server.cpp
@@ -84,6 +84,7 @@
 #include <Storages/System/attachInformationSchemaTables.h>
 #include <Storages/Cache/ExternalDataSourceCache.h>
 #include <Storages/Cache/registerRemoteFileMetadatas.h>
+#include <Storages/Cache/ObjectStorageListObjectsCache.h>
 #include <AggregateFunctions/registerAggregateFunctions.h>
 #include <Functions/UserDefined/IUserDefinedSQLObjectsStorage.h>
 #include <Functions/registerFunctions.h>
@@ -316,6 +317,10 @@ namespace ServerSetting
     extern const ServerSettingsUInt64 page_cache_max_size;
     extern const ServerSettingsDouble page_cache_free_memory_ratio;
     extern const ServerSettingsUInt64 page_cache_lookahead_blocks;
+    extern const ServerSettingsUInt64 input_format_parquet_metadata_cache_max_size;
+    extern const ServerSettingsUInt64 object_storage_list_objects_cache_size;
+    extern const ServerSettingsUInt64 object_storage_list_objects_cache_max_entries;
+    extern const ServerSettingsUInt64 object_storage_list_objects_cache_ttl;
 }
 
 }
@@ -2375,6 +2380,10 @@ try
     if (dns_cache_updater)
         dns_cache_updater->start();
 
+    ObjectStorageListObjectsCache::instance().setMaxSizeInBytes(server_settings[ServerSetting::object_storage_list_objects_cache_size]);
+    ObjectStorageListObjectsCache::instance().setMaxCount(server_settings[ServerSetting::object_storage_list_objects_cache_max_entries]);
+    ObjectStorageListObjectsCache::instance().setTTL(server_settings[ServerSetting::object_storage_list_objects_cache_ttl]);
+
     auto replicas_reconnector = ReplicasReconnector::init(global_context);
 
     /// Set current database name before loading tables and databases because
diff --git a/src/Access/Common/AccessType.h b/src/Access/Common/AccessType.h
@@ -182,6 +182,7 @@ enum class AccessType : uint8_t
     M(SYSTEM_DROP_SCHEMA_CACHE, "SYSTEM DROP SCHEMA CACHE, DROP SCHEMA CACHE", GLOBAL, SYSTEM_DROP_CACHE) \
     M(SYSTEM_DROP_FORMAT_SCHEMA_CACHE, "SYSTEM DROP FORMAT SCHEMA CACHE, DROP FORMAT SCHEMA CACHE", GLOBAL, SYSTEM_DROP_CACHE) \
     M(SYSTEM_DROP_S3_CLIENT_CACHE, "SYSTEM DROP S3 CLIENT, DROP S3 CLIENT CACHE", GLOBAL, SYSTEM_DROP_CACHE) \
+    M(SYSTEM_DROP_OBJECT_STORAGE_LIST_OBJECTS_CACHE, "SYSTEM DROP OBJECT STORAGE LIST OBJECTS CACHE", GLOBAL, SYSTEM_DROP_CACHE) \
     M(SYSTEM_DROP_CACHE, "DROP CACHE", GROUP, SYSTEM) \
     M(SYSTEM_RELOAD_CONFIG, "RELOAD CONFIG", GLOBAL, SYSTEM_RELOAD) \
     M(SYSTEM_RELOAD_USERS, "RELOAD USERS", GLOBAL, SYSTEM_RELOAD) \
diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp
@@ -964,11 +964,22 @@ The server successfully detected this situation and will download merged part fr
     M(MemoryWorkerRun, "Number of runs done by MemoryWorker in background", ValueType::Number) \
     M(MemoryWorkerRunElapsedMicroseconds, "Total time spent by MemoryWorker for background work", ValueType::Microseconds) \
     \
+<<<<<<< HEAD
     M(ParquetFetchWaitTimeMicroseconds, "Time of waiting fetching parquet data", ValueType::Microseconds) \
     M(FilterTransformPassedRows, "Number of rows that passed the filter in the query", ValueType::Number) \
     M(FilterTransformPassedBytes, "Number of bytes that passed the filter in the query", ValueType::Bytes) \
     M(QueryPreempted, "How many times tasks are paused and waiting due to 'priority' setting", ValueType::Number) \
 
+=======
+    M(ParquetFetchWaitTimeMicroseconds, "Time of waiting fetching parquet data", ValueType::Microseconds)                                                                                                                                                                                                                                               \
+    \
+    M(ParquetMetaDataCacheHits, "Number of times the read from filesystem cache hit the cache.", ValueType::Number) \
+    M(ParquetMetaDataCacheMisses, "Number of times the read from filesystem cache miss the cache.", ValueType::Number)                                                                                                                                                                                                                                  \
+    M(ObjectStorageListObjectsCacheHits, "Number of times object storage list objects operation hit the cache.", ValueType::Number) \
+    M(ObjectStorageListObjectsCacheMisses, "Number of times object storage list objects operation miss the cache.", ValueType::Number)                                                                                                                                                                                                                                  \
+    M(ObjectStorageListObjectsCacheExactMatchHits, "Number of times object storage list objects operation hit the cache with an exact match.", ValueType::Number) \
+    M(ObjectStorageListObjectsCachePrefixMatchHits, "Number of times object storage list objects operation miss the cache using prefix matching.", ValueType::Number) \
+>>>>>>> 60282dacee4 (Merge pull request #743 from Altinity/list_objects_cache)
 
 #ifdef APPLY_FOR_EXTERNAL_EVENTS
     #define APPLY_FOR_EVENTS(M) APPLY_FOR_BUILTIN_EVENTS(M) APPLY_FOR_EXTERNAL_EVENTS(M)
diff --git a/src/Common/TTLCachePolicy.h b/src/Common/TTLCachePolicy.h
@@ -248,10 +248,10 @@ class TTLCachePolicy : public ICachePolicy<Key, Mapped, HashFunction, WeightFunc
         return res;
     }
 
-private:
+protected:
     using Cache = std::unordered_map<Key, MappedPtr, HashFunction>;
     Cache cache;
-
+private:
     /// TODO To speed up removal of stale entries, we could also add another container sorted on expiry times which maps keys to iterators
     /// into the cache. To insert an entry, add it to the cache + add the iterator to the sorted container. To remove stale entries, do a
     /// binary search on the sorted container and erase all left of the found key.
diff --git a/src/Core/ServerSettings.cpp b/src/Core/ServerSettings.cpp
@@ -1020,8 +1020,11 @@ namespace DB
     ```
     )", 0) \
     DECLARE(Bool, storage_shared_set_join_use_inner_uuid, false, "If enabled, an inner UUID is generated during the creation of SharedSet and SharedJoin. ClickHouse Cloud only", 0)
-
-
+    DECLARE(Bool, storage_shared_set_join_use_inner_uuid, false, "If enabled, an inner UUID is generated during the creation of SharedSet and SharedJoin. ClickHouse Cloud only", 0) \
+    DECLARE(UInt64, input_format_parquet_metadata_cache_max_size, 500000000, "Maximum size of parquet file metadata cache", 0)                                                                                \
+    DECLARE(UInt64, object_storage_list_objects_cache_size, 500000000, "Maximum size of ObjectStorage list objects cache in bytes. Zero means disabled.", 0)                            \
+    DECLARE(UInt64, object_storage_list_objects_cache_max_entries, 1000, "Maximum size of ObjectStorage list objects cache in entries. Zero means disabled.", 0)                      \
+    DECLARE(UInt64, object_storage_list_objects_cache_ttl, 3600, "Time to live of records in ObjectStorage list objects cache in seconds. Zero means unlimited", 0) \
 // clang-format on
 
 /// If you add a setting which can be updated at runtime, please update 'changeable_settings' map in dumpToSystemServerSettingsColumns below
diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp
@@ -6116,6 +6116,9 @@ Trigger processor to spill data into external storage adpatively. grace join is
     /** Experimental tsToGrid aggregate function. */ \
     DECLARE(Bool, allow_experimental_ts_to_grid_aggregate_function, false, R"(
 Experimental tsToGrid aggregate function for Prometheus-like timeseries resampling. Cloud only
+)", EXPERIMENTAL) \
+    DECLARE(Bool, use_object_storage_list_objects_cache, false, R"(
+Cache the list of objects returned by list objects calls in object storage
 )", EXPERIMENTAL) \
     \
     /* ####################################################### */ \
diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp
@@ -68,6 +68,13 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         /// Note: please check if the key already exists to prevent duplicate entries.
         addSettingsChanges(settings_changes_history, "25.4",
         {
+            // Altinity Antalya modifications atop of 25.2
+            {"object_storage_cluster", "", "", "New setting"},
+            {"object_storage_max_nodes", 0, 0, "New setting"},
+            {"use_iceberg_metadata_files_cache", true, true, "New setting"},
+            {"iceberg_timestamp_ms", 0, 0, "New setting."},
+            {"iceberg_snapshot_id", 0, 0, "New setting."},
+            {"use_object_storage_list_objects_cache", true, false, "New setting."},
         });
         addSettingsChanges(settings_changes_history, "25.3",
         {
diff --git a/src/Disks/ObjectStorages/AzureBlobStorage/AzureObjectStorage.h b/src/Disks/ObjectStorages/AzureBlobStorage/AzureObjectStorage.h
@@ -31,6 +31,8 @@ class AzureObjectStorage : public IObjectStorage
         const String & object_namespace_,
         const String & description_);
 
+    bool supportsListObjectsCache() override { return true; }
+
     void listObjects(const std::string & path, RelativePathsWithMetadata & children, size_t max_keys) const override;
 
     ObjectStorageIteratorPtr iterate(const std::string & path_prefix, size_t max_keys) const override;
diff --git a/src/Disks/ObjectStorages/IObjectStorage.h b/src/Disks/ObjectStorages/IObjectStorage.h
@@ -274,6 +274,8 @@ class IObjectStorage
 #endif
 
 
+    virtual bool supportsListObjectsCache() { return false; }
+
 private:
     mutable std::mutex throttlers_mutex;
     ThrottlerPtr remote_read_throttler;
diff --git a/src/Disks/ObjectStorages/S3/S3ObjectStorage.h b/src/Disks/ObjectStorages/S3/S3ObjectStorage.h
@@ -81,6 +81,8 @@ class S3ObjectStorage : public IObjectStorage
 
     ObjectStorageType getType() const override { return ObjectStorageType::S3; }
 
+    bool supportsListObjectsCache() override { return true; }
+
     bool exists(const StoredObject & object) const override;
 
     std::unique_ptr<ReadBufferFromFileBase> readObject( /// NOLINT
diff --git a/src/Interpreters/InterpreterSystemQuery.cpp b/src/Interpreters/InterpreterSystemQuery.cpp
@@ -47,6 +47,7 @@
 #include <Access/Common/AllowedClientHosts.h>
 #include <Databases/DatabaseReplicated.h>
 #include <Disks/ObjectStorages/IMetadataStorage.h>
+#include <Storages/Cache/ObjectStorageListObjectsCache.h>
 #include <Storages/StorageDistributed.h>
 #include <Storages/StorageReplicatedMergeTree.h>
 #include <Storages/Freeze.h>
@@ -419,6 +420,12 @@ BlockIO InterpreterSystemQuery::execute()
             getContext()->clearQueryResultCache(query.query_result_cache_tag);
             break;
         }
+        case Type::DROP_OBJECT_STORAGE_LIST_OBJECTS_CACHE:
+        {
+            getContext()->checkAccess(AccessType::SYSTEM_DROP_OBJECT_STORAGE_LIST_OBJECTS_CACHE);
+            ObjectStorageListObjectsCache::instance().clear();
+            break;
+        }
         case Type::DROP_COMPILED_EXPRESSION_CACHE:
 #if USE_EMBEDDED_COMPILER
             getContext()->checkAccess(AccessType::SYSTEM_DROP_COMPILED_EXPRESSION_CACHE);
@@ -1464,6 +1471,11 @@ AccessRightsElements InterpreterSystemQuery::getRequiredAccessForDDLOnCluster()
         case Type::DROP_PAGE_CACHE:
         case Type::DROP_SCHEMA_CACHE:
         case Type::DROP_FORMAT_SCHEMA_CACHE:
+<<<<<<< HEAD
+=======
+        case Type::DROP_PARQUET_METADATA_CACHE:
+        case Type::DROP_OBJECT_STORAGE_LIST_OBJECTS_CACHE:
+>>>>>>> 60282dacee4 (Merge pull request #743 from Altinity/list_objects_cache)
         case Type::DROP_S3_CLIENT_CACHE:
         {
             required_access.emplace_back(AccessType::SYSTEM_DROP_CACHE);
diff --git a/src/Parsers/ASTSystemQuery.cpp b/src/Parsers/ASTSystemQuery.cpp
@@ -456,6 +456,7 @@ void ASTSystemQuery::formatImpl(WriteBuffer & ostr, const FormatSettings & setti
         case Type::DROP_SKIPPING_INDEX_CACHE:
         case Type::DROP_COMPILED_EXPRESSION_CACHE:
         case Type::DROP_S3_CLIENT_CACHE:
+        case Type::DROP_OBJECT_STORAGE_LIST_OBJECTS_CACHE:
         case Type::RESET_COVERAGE:
         case Type::RESTART_REPLICAS:
         case Type::JEMALLOC_PURGE:
diff --git a/src/Parsers/ASTSystemQuery.h b/src/Parsers/ASTSystemQuery.h
@@ -41,6 +41,7 @@ class ASTSystemQuery : public IAST, public ASTQueryWithOnCluster
         DROP_SCHEMA_CACHE,
         DROP_FORMAT_SCHEMA_CACHE,
         DROP_S3_CLIENT_CACHE,
+        DROP_OBJECT_STORAGE_LIST_OBJECTS_CACHE,
         STOP_LISTEN,
         START_LISTEN,
         RESTART_REPLICAS,
diff --git a/src/Storages/Cache/ObjectStorageListObjectsCache.cpp b/src/Storages/Cache/ObjectStorageListObjectsCache.cpp
@@ -0,0 +1,210 @@
+#include <Storages/Cache/ObjectStorageListObjectsCache.h>
+#include <Common/TTLCachePolicy.h>
+#include <Common/ProfileEvents.h>
+#include <boost/functional/hash.hpp>
+
+namespace ProfileEvents
+{
+extern const Event ObjectStorageListObjectsCacheHits;
+extern const Event ObjectStorageListObjectsCacheMisses;
+extern const Event ObjectStorageListObjectsCacheExactMatchHits;
+extern const Event ObjectStorageListObjectsCachePrefixMatchHits;
+}
+
+namespace DB
+{
+
+template <typename Key, typename Mapped, typename HashFunction, typename WeightFunction, typename IsStaleFunction>
+class ObjectStorageListObjectsCachePolicy : public TTLCachePolicy<Key, Mapped, HashFunction, WeightFunction, IsStaleFunction>
+{
+public:
+    using BasePolicy = TTLCachePolicy<Key, Mapped, HashFunction, WeightFunction, IsStaleFunction>;
+    using typename BasePolicy::MappedPtr;
+    using typename BasePolicy::KeyMapped;
+    using BasePolicy::cache;
+
+    ObjectStorageListObjectsCachePolicy()
+        : BasePolicy(std::make_unique<NoCachePolicyUserQuota>())
+    {
+    }
+
+    std::optional<KeyMapped> getWithKey(const Key & key) override
+    {
+        if (const auto it = cache.find(key); it != cache.end())
+        {
+            if (!IsStaleFunction()(it->first))
+            {
+                return std::make_optional<KeyMapped>({it->first, it->second});
+            }
+            // found a stale entry, remove it but don't return. We still want to perform the prefix matching search
+            BasePolicy::remove(it->first);
+        }
+
+        if (const auto it = findBestMatchingPrefixAndRemoveExpiredEntries(key); it != cache.end())
+        {
+            return std::make_optional<KeyMapped>({it->first, it->second});
+        }
+
+        return std::nullopt;
+    }
+
+private:
+    auto findBestMatchingPrefixAndRemoveExpiredEntries(Key key)
+    {
+        while (!key.prefix.empty())
+        {
+            if (const auto it = cache.find(key); it != cache.end())
+            {
+                if (IsStaleFunction()(it->first))
+                {
+                    BasePolicy::remove(it->first);
+                }
+                else
+                {
+                    return it;
+                }
+            }
+
+            key.prefix.pop_back();
+        }
+
+        return cache.end();
+    }
+};
+
+ObjectStorageListObjectsCache::Key::Key(
+    const String & storage_description_,
+    const String & bucket_,
+    const String & prefix_,
+    const std::chrono::steady_clock::time_point & expires_at_,
+    std::optional<UUID> user_id_)
+    : storage_description(storage_description_), bucket(bucket_), prefix(prefix_), expires_at(expires_at_), user_id(user_id_) {}
+
+bool ObjectStorageListObjectsCache::Key::operator==(const Key & other) const
+{
+    return storage_description == other.storage_description && bucket == other.bucket && prefix == other.prefix;
+}
+
+size_t ObjectStorageListObjectsCache::KeyHasher::operator()(const Key & key) const
+{
+    std::size_t seed = 0;
+
+    boost::hash_combine(seed, key.storage_description);
+    boost::hash_combine(seed, key.bucket);
+    boost::hash_combine(seed, key.prefix);
+
+    return seed;
+}
+
+bool ObjectStorageListObjectsCache::IsStale::operator()(const Key & key) const
+{
+    return key.expires_at < std::chrono::steady_clock::now();
+}
+
+size_t ObjectStorageListObjectsCache::WeightFunction::operator()(const Value & value) const
+{
+    std::size_t weight = 0;
+
+    for (const auto & object : value)
+    {
+        const auto object_metadata = object->metadata;
+        weight += object->relative_path.capacity() + sizeof(object_metadata);
+
+        // variable size
+        if (object_metadata)
+        {
+            weight += object_metadata->etag.capacity();
+            weight += object_metadata->attributes.size() * (sizeof(std::string) * 2);
+
+            for (const auto & [k, v] : object_metadata->attributes)
+            {
+                weight += k.capacity() + v.capacity();
+            }
+        }
+    }
+
+    return weight;
+}
+
+ObjectStorageListObjectsCache::ObjectStorageListObjectsCache()
+    : cache(std::make_unique<ObjectStorageListObjectsCachePolicy<Key, Value, KeyHasher, WeightFunction, IsStale>>())
+{
+}
+
+void ObjectStorageListObjectsCache::set(
+    const Key & key,
+    const std::shared_ptr<Value> & value)
+{
+    auto key_with_ttl = key;
+    key_with_ttl.expires_at = std::chrono::steady_clock::now() + std::chrono::seconds(ttl_in_seconds);
+
+    cache.set(key_with_ttl, value);
+}
+
+void ObjectStorageListObjectsCache::clear()
+{
+    cache.clear();
+}
+
+std::optional<ObjectStorageListObjectsCache::Value> ObjectStorageListObjectsCache::get(const Key & key, bool filter_by_prefix)
+{
+    const auto pair = cache.getWithKey(key);
+
+    if (!pair)
+    {
+        ProfileEvents::increment(ProfileEvents::ObjectStorageListObjectsCacheMisses);
+        return {};
+    }
+
+    ProfileEvents::increment(ProfileEvents::ObjectStorageListObjectsCacheHits);
+
+    if (pair->key == key)
+    {
+        ProfileEvents::increment(ProfileEvents::ObjectStorageListObjectsCacheExactMatchHits);
+        return *pair->mapped;
+    }
+
+    ProfileEvents::increment(ProfileEvents::ObjectStorageListObjectsCachePrefixMatchHits);
+
+    if (!filter_by_prefix)
+    {
+        return *pair->mapped;
+    }
+
+    Value filtered_objects;
+
+    filtered_objects.reserve(pair->mapped->size());
+
+    for (const auto & object : *pair->mapped)
+    {
+        if (object->relative_path.starts_with(key.prefix))
+        {
+            filtered_objects.push_back(object);
+        }
+    }
+
+    return filtered_objects;
+}
+
+void ObjectStorageListObjectsCache::setMaxSizeInBytes(std::size_t size_in_bytes_)
+{
+    cache.setMaxSizeInBytes(size_in_bytes_);
+}
+
+void ObjectStorageListObjectsCache::setMaxCount(std::size_t count)
+{
+    cache.setMaxCount(count);
+}
+
+void ObjectStorageListObjectsCache::setTTL(std::size_t ttl_in_seconds_)
+{
+    ttl_in_seconds = ttl_in_seconds_;
+}
+
+ObjectStorageListObjectsCache & ObjectStorageListObjectsCache::instance()
+{
+    static ObjectStorageListObjectsCache instance;
+    return instance;
+}
+
+}
diff --git a/src/Storages/Cache/ObjectStorageListObjectsCache.h b/src/Storages/Cache/ObjectStorageListObjectsCache.h
diff --git a/src/Storages/Cache/tests/gtest_object_storage_list_objects_cache.cpp b/src/Storages/Cache/tests/gtest_object_storage_list_objects_cache.cpp
diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp
diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.h b/src/Storages/ObjectStorage/StorageObjectStorageSource.h
diff --git a/tests/queries/0_stateless/01271_show_privileges.reference b/tests/queries/0_stateless/01271_show_privileges.reference
diff --git a/tests/queries/0_stateless/03377_object_storage_list_objects_cache.reference b/tests/queries/0_stateless/03377_object_storage_list_objects_cache.reference
diff --git a/tests/queries/0_stateless/03377_object_storage_list_objects_cache.sql b/tests/queries/0_stateless/03377_object_storage_list_objects_cache.sql

Original file line number	Diff line number	Diff line change
`@@ -68,6 +68,13 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()`
`68`	`68`	`/// Note: please check if the key already exists to prevent duplicate entries.`
`69`	`69`	`addSettingsChanges(settings_changes_history, "25.4",`
`70`	`70`	`{`
	`71`	`+ // Altinity Antalya modifications atop of 25.2`
	`72`	`+ {"object_storage_cluster", "", "", "New setting"},`
	`73`	`+ {"object_storage_max_nodes", 0, 0, "New setting"},`
	`74`	`+ {"use_iceberg_metadata_files_cache", true, true, "New setting"},`
	`75`	`+ {"iceberg_timestamp_ms", 0, 0, "New setting."},`
	`76`	`+ {"iceberg_snapshot_id", 0, 0, "New setting."},`
	`77`	`+ {"use_object_storage_list_objects_cache", true, false, "New setting."},`
`71`	`78`	`});`
`72`	`79`	`addSettingsChanges(settings_changes_history, "25.3",`
`73`	`80`	`{`