Antalya 25.6.5: Forward port of #795 - Parquet Metadata Caching

programs/server/Server.cpp

@@ -156,6 +156,10 @@
 #   include <azure/core/diagnostics/logger.hpp>
 #endif
 
+#if USE_PARQUET
+#   include <Processors/Formats/Impl/ParquetFileMetaDataCache.h>
+#endif
+
 
 #include <incbin.h>
 /// A minimal file used when the server is run without installation
@@ -326,6 +330,7 @@ namespace ServerSetting
     extern const ServerSettingsUInt64 os_cpu_busy_time_threshold;
     extern const ServerSettingsFloat min_os_cpu_wait_time_ratio_to_drop_connection;
     extern const ServerSettingsFloat max_os_cpu_wait_time_ratio_to_drop_connection;
+    extern const ServerSettingsUInt64 input_format_parquet_metadata_cache_max_size;
 }
 
 namespace ErrorCodes
@@ -2423,6 +2428,10 @@ try
 
     auto replicas_reconnector = ReplicasReconnector::init(global_context);
 
+#if USE_PARQUET
+    ParquetFileMetaDataCache::instance()->setMaxSizeInBytes(server_settings[ServerSetting::input_format_parquet_metadata_cache_max_size]);
+#endif
+
     /// Set current database name before loading tables and databases because
     /// system logs may copy global context.
     std::string default_database = server_settings[ServerSetting::default_database];

src/Access/Common/AccessType.h

@@ -184,6 +184,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_PARQUET_METADATA_CACHE, "SYSTEM DROP PARQUET METADATA 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) \

src/Common/ProfileEvents.cpp

@@ -1041,7 +1041,8 @@ The server successfully detected this situation and will download merged part fr
     M(IndexGenericExclusionSearchAlgorithm, "Number of times the generic exclusion search algorithm is used over the index marks", ValueType::Number) \
     M(ParallelReplicasQueryCount, "Number of (sub)queries executed using parallel replicas during a query execution", ValueType::Number) \
     M(DistributedConnectionReconnectCount, "Number of reconnects to other servers done during distributed query execution. It can happen when a stale connection has been acquired from connection pool", ValueType::Number) \
-
+    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) \
 
 #ifdef APPLY_FOR_EXTERNAL_EVENTS
     #define APPLY_FOR_EVENTS(M) APPLY_FOR_BUILTIN_EVENTS(M) APPLY_FOR_EXTERNAL_EVENTS(M)

src/Core/FormatFactorySettings.h

@@ -1348,8 +1348,7 @@ Limits the size of the blocks formed during data parsing in input formats in byt
     DECLARE(Bool, input_format_parquet_allow_geoparquet_parser, true, R"(
 Use geo column parser to convert Array(UInt8) into Point/Linestring/Polygon/MultiLineString/MultiPolygon types
 )", 0) \
-
-
+    DECLARE(Bool, input_format_parquet_use_metadata_cache, true, R"(Enable parquet file metadata caching)", 0) \
 // End of FORMAT_FACTORY_SETTINGS
 
 #define OBSOLETE_FORMAT_SETTINGS(M, ALIAS) \

src/Core/ServerSettings.cpp

@@ -1064,8 +1064,7 @@ The policy on how to perform a scheduling of CPU slots specified by `concurrent_
     See [Controlling behavior on server CPU overload](/operations/settings/server-overload) for more details.
     )", 0) \
     DECLARE(Float, distributed_cache_keep_up_free_connections_ratio, 0.1f, "Soft limit for number of active connection distributed cache will try to keep free. After the number of free connections goes below distributed_cache_keep_up_free_connections_ratio * max_connections, connections with oldest activity will be closed until the number goes above the limit.", 0) \
-
-
+    DECLARE(UInt64, input_format_parquet_metadata_cache_max_size, 500000000, "Maximum size of parquet file metadata cache", 0) \
 // clang-format on
 
 /// If you add a setting which can be updated at runtime, please update 'changeable_settings' map in dumpToSystemServerSettingsColumns below

src/Core/SettingsChangesHistory.cpp

@@ -182,6 +182,11 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"parallel_hash_join_threshold", 0, 0, "New setting"},
             /// Release closed. Please use 25.4
         });
+        addSettingsChanges(settings_changes_history, "24.12.2.20000",
+        {
+            // Altinity Antalya modifications atop of 24.12
+            {"input_format_parquet_use_metadata_cache", true, true, "New setting, turned ON by default"}, // https://github.com/Altinity/ClickHouse/pull/586
+        });
         addSettingsChanges(settings_changes_history, "25.2",
         {
             /// Release closed. Please use 25.3

src/Interpreters/InterpreterSystemQuery.cpp

@@ -79,6 +79,10 @@
 #include <Formats/ProtobufSchemas.h>
 #endif
 
+#if USE_PARQUET
+#include <Processors/Formats/Impl/ParquetFileMetaDataCache.h>
+#endif
+
 #if USE_AWS_S3
 #include <IO/S3/Client.h>
 #endif
@@ -433,6 +437,16 @@ BlockIO InterpreterSystemQuery::execute()
             getContext()->clearQueryResultCache(query.query_result_cache_tag);
             break;
         }
+        case Type::DROP_PARQUET_METADATA_CACHE:
+        {
+#if USE_PARQUET
+            getContext()->checkAccess(AccessType::SYSTEM_DROP_PARQUET_METADATA_CACHE);
+            ParquetFileMetaDataCache::instance()->clear();
+            break;
+#else
+            throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "The server was compiled without the support for Parquet");
+#endif
+        }
         case Type::DROP_COMPILED_EXPRESSION_CACHE:
 #if USE_EMBEDDED_COMPILER
             getContext()->checkAccess(AccessType::SYSTEM_DROP_COMPILED_EXPRESSION_CACHE);
@@ -1533,6 +1547,7 @@ AccessRightsElements InterpreterSystemQuery::getRequiredAccessForDDLOnCluster()
         case Type::DROP_PAGE_CACHE:
         case Type::DROP_SCHEMA_CACHE:
         case Type::DROP_FORMAT_SCHEMA_CACHE:
+        case Type::DROP_PARQUET_METADATA_CACHE:
         case Type::DROP_S3_CLIENT_CACHE:
         {
             required_access.emplace_back(AccessType::SYSTEM_DROP_CACHE);

src/Parsers/ASTSystemQuery.cpp

@@ -467,6 +467,7 @@ void ASTSystemQuery::formatImpl(WriteBuffer & ostr, const FormatSettings & setti
         case Type::DROP_COMPILED_EXPRESSION_CACHE:
         case Type::DROP_S3_CLIENT_CACHE:
         case Type::DROP_ICEBERG_METADATA_CACHE:
+        case Type::DROP_PARQUET_METADATA_CACHE:
         case Type::RESET_COVERAGE:
         case Type::RESTART_REPLICAS:
         case Type::JEMALLOC_PURGE:

src/Parsers/ASTSystemQuery.h

@@ -42,6 +42,7 @@ class ASTSystemQuery : public IAST, public ASTQueryWithOnCluster
         DROP_SCHEMA_CACHE,
         DROP_FORMAT_SCHEMA_CACHE,
         DROP_S3_CLIENT_CACHE,
+        DROP_PARQUET_METADATA_CACHE,
         STOP_LISTEN,
         START_LISTEN,
         RESTART_REPLICAS,

src/Processors/Formats/IInputFormat.h

@@ -70,6 +70,9 @@ class IInputFormat : public SourceWithKeyCondition
 
     void needOnlyCount() { need_only_count = true; }
 
+    /// Set additional info/key/id related to underlying storage of the ReadBuffer
+    virtual void setStorageRelatedUniqueKey(const Settings & /*settings*/, const String & /*key*/) {}
+
 protected:
     ReadBuffer & getReadBuffer() const { chassert(in); return *in; }
 

src/Processors/Formats/Impl/ParquetBlockInputFormat.cpp

@@ -3,6 +3,9 @@
 #if USE_PARQUET
 
 #include <Columns/ColumnNullable.h>
+#include <Core/Settings.h>
+#include <Core/ServerSettings.h>
+#include <Common/ProfileEvents.h>
 #include <Common/logger_useful.h>
 #include <Common/ThreadPool.h>
 #include <Formats/FormatFactory.h>
@@ -29,6 +32,7 @@
 #include <Processors/Formats/Impl/Parquet/ParquetRecordReader.h>
 #include <Processors/Formats/Impl/Parquet/parquetBloomFilterHash.h>
 #include <Interpreters/Context.h>
+#include <Processors/Formats/Impl/ParquetFileMetaDataCache.h>
 #include <Interpreters/convertFieldToType.h>
 
 #include <boost/algorithm/string/case_conv.hpp>
@@ -38,6 +42,8 @@ namespace ProfileEvents
     extern const Event ParquetFetchWaitTimeMicroseconds;
     extern const Event ParquetReadRowGroups;
     extern const Event ParquetPrunedRowGroups;
+    extern const Event ParquetMetaDataCacheHits;
+    extern const Event ParquetMetaDataCacheMisses;
 }
 
 namespace CurrentMetrics
@@ -54,6 +60,16 @@ namespace CurrentMetrics
 namespace DB
 {
 
+namespace Setting
+{
+    extern const SettingsBool input_format_parquet_use_metadata_cache;
+}
+
+namespace ServerSetting
+{
+    extern const ServerSettingsUInt64 input_format_parquet_metadata_cache_max_size;
+}
+
 namespace ErrorCodes
 {
     extern const int BAD_ARGUMENTS;
@@ -513,6 +529,49 @@ static std::vector<Range> getHyperrectangleForRowGroup(const parquet::FileMetaDa
     return hyperrectangle;
 }
 
+std::shared_ptr<parquet::FileMetaData> ParquetBlockInputFormat::readMetadataFromFile()
+{
+    createArrowFileIfNotCreated();
+    return parquet::ReadMetaData(arrow_file);
+}
+
+std::shared_ptr<parquet::FileMetaData> ParquetBlockInputFormat::getFileMetaData()
+{
+    // in-memory cache is not implemented for local file operations, only for remote files
+    // there is a chance the user sets `input_format_parquet_use_metadata_cache=1` for a local file operation
+    // and the cache_key won't be set. Therefore, we also need to check for metadata_cache.key
+    if (!metadata_cache.use_cache || metadata_cache.key.empty())
+    {
+        return readMetadataFromFile();
+    }
+
+    auto [parquet_file_metadata, loaded] = ParquetFileMetaDataCache::instance()->getOrSet(
+        metadata_cache.key,
+        [&]()
+        {
+            return readMetadataFromFile();
+        }
+    );
+    if (loaded)
+        ProfileEvents::increment(ProfileEvents::ParquetMetaDataCacheMisses);
+    else
+        ProfileEvents::increment(ProfileEvents::ParquetMetaDataCacheHits);
+    return parquet_file_metadata;
+}
+
+void ParquetBlockInputFormat::createArrowFileIfNotCreated()
+{
+    if (arrow_file)
+    {
+        return;
+    }
+
+    // Create arrow file adapter.
+    // TODO: Make the adapter do prefetching on IO threads, based on the full set of ranges that
+    //       we'll need to read (which we know in advance). Use max_download_threads for that.
+    arrow_file = asArrowFile(*in, format_settings, is_stopped, "Parquet", PARQUET_MAGIC_BYTES, /* avoid_buffering */ true);
+}
+
 std::unordered_set<std::size_t> getBloomFilterFilteringColumnKeys(const KeyCondition::RPN & rpn)
 {
     std::unordered_set<std::size_t> column_keys;
@@ -612,7 +671,7 @@ void ParquetBlockInputFormat::initializeIfNeeded()
     if (is_stopped)
         return;
 
-    metadata = parquet::ReadMetaData(arrow_file);
+    metadata = getFileMetaData();
     const bool prefetch_group = supportPrefetch();
 
     std::shared_ptr<arrow::Schema> schema;
@@ -712,6 +771,8 @@ void ParquetBlockInputFormat::initializeIfNeeded()
         }
     }
 
+    bool has_row_groups_to_read = false;
+
     auto skip_row_group_based_on_filters = [&](int row_group)
     {
         if (!format_settings.parquet.filter_push_down && !format_settings.parquet.bloom_filter_push_down)
@@ -763,7 +824,20 @@ void ParquetBlockInputFormat::initializeIfNeeded()
         row_group_batches.back().total_bytes_compressed += row_group_size;
         auto rows = adaptive_chunk_size(row_group);
         row_group_batches.back().adaptive_chunk_size = rows ? rows : format_settings.parquet.max_block_size;
+
+        has_row_groups_to_read = true;
     }
+
+    if (has_row_groups_to_read)
+    {
+        createArrowFileIfNotCreated();
+    }
+}
+
+void ParquetBlockInputFormat::setStorageRelatedUniqueKey(const Settings & settings, const String & key_)
+{
+    metadata_cache.key = key_;
+    metadata_cache.use_cache = settings[Setting::input_format_parquet_use_metadata_cache];
 }
 
 void ParquetBlockInputFormat::initializeRowGroupBatchReader(size_t row_group_batch_idx)

src/Processors/Formats/Impl/ParquetBlockInputFormat.h

@@ -72,6 +72,8 @@ class ParquetBlockInputFormat : public IInputFormat
 
     size_t getApproxBytesReadForChunk() const override { return previous_approx_bytes_read_for_chunk; }
 
+    void setStorageRelatedUniqueKey(const Settings & settings, const String & key_) override;
+
 private:
     Chunk read() override;
 
@@ -90,6 +92,11 @@ class ParquetBlockInputFormat : public IInputFormat
 
     void threadFunction(size_t row_group_batch_idx);
 
+    void createArrowFileIfNotCreated();
+    std::shared_ptr<parquet::FileMetaData> readMetadataFromFile();
+
+    std::shared_ptr<parquet::FileMetaData> getFileMetaData();
+
     inline bool supportPrefetch() const;
 
     // Data layout in the file:
@@ -338,6 +345,13 @@ class ParquetBlockInputFormat : public IInputFormat
     std::exception_ptr background_exception = nullptr;
     std::atomic<int> is_stopped{0};
     bool is_initialized = false;
+    struct Cache
+    {
+        String key;
+        bool use_cache = false;
+    };
+
+    Cache metadata_cache;
 };
 
 class ParquetSchemaReader : public ISchemaReader

src/Processors/Formats/Impl/ParquetFileMetaDataCache.cpp

@@ -0,0 +1,20 @@
+#include <Processors/Formats/Impl/ParquetFileMetaDataCache.h>
+
+#if USE_PARQUET
+
+namespace DB
+{
+
+ParquetFileMetaDataCache::ParquetFileMetaDataCache()
+    : CacheBase<String, parquet::FileMetaData>(CurrentMetrics::end(), CurrentMetrics::end(), 0)
+{}
+
+ParquetFileMetaDataCache * ParquetFileMetaDataCache::instance()
+{
+    static ParquetFileMetaDataCache instance;
+    return &instance;
+}
+
+}
+
+#endif

src/Processors/Formats/Impl/ParquetFileMetaDataCache.h

@@ -0,0 +1,30 @@
+#pragma once
+
+#include "config.h"
+
+#if USE_PARQUET
+
+namespace parquet
+{
+
+class FileMetaData;
+
+}
+
+#include <Common/CacheBase.h>
+
+namespace DB
+{
+
+class ParquetFileMetaDataCache : public CacheBase<String, parquet::FileMetaData>
+{
+public:
+    static ParquetFileMetaDataCache * instance();
+
+private:
+    ParquetFileMetaDataCache();
+};
+
+}
+
+#endif