feat(parquet): support parquet metadata cache

docs/source/user_guide.rst

@@ -25,6 +25,7 @@ User Guide
    user_guide/snapshot
    user_guide/manifest
    user_guide/manifest_cache
+   user_guide/parquet_metadata_cache
    user_guide/data_types
    user_guide/primary_key_table
    user_guide/append_only_table

docs/source/user_guide/parquet_metadata_cache.rst

@@ -0,0 +1,107 @@
+.. Copyright 2026-present Alibaba Inc.
+
+.. Licensed 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.
+
+Parquet Metadata Cache
+======================
+
+Overview
+--------
+
+paimon-cpp can cache serialized Parquet metadata footer bytes for Parquet data files.
+The cache is used by ``ParquetReaderBuilder`` before opening the Arrow Parquet
+reader. On a cache miss, paimon-cpp loads the Parquet file metadata, serializes
+it as a complete metadata footer, and stores those bytes in the public
+``Cache`` abstraction. On a cache hit, paimon-cpp parses the cached footer bytes into
+``parquet::FileMetaData`` and passes the metadata to the Parquet reader.
+
+The cache stores serialized metadata footer bytes instead of caching a
+``parquet::FileMetaData`` instance. This keeps the cache value compact and
+similar to manifest cache values: the cache weight follows the actual cached
+bytes, while the Parquet library still owns metadata parsing and validation.
+
+This optimization is useful when the same Parquet files are opened repeatedly
+in the same process, for example repeated ``get`` or ``scan`` requests over the
+same snapshot. On a cache hit, the read path avoids reading the Parquet footer
+bytes from the filesystem again. paimon-cpp still parses the cached footer bytes
+into ``parquet::FileMetaData`` for each reader open. Data pages, page indexes,
+and column chunks are still read from the file as usual.
+
+Configuration
+-------------
+
+Parquet metadata caching is disabled by default. Embedding applications that
+need it can provide a custom ``Cache`` implementation and inject it through
+``ScanContextBuilder`` or ``ReadContextBuilder``. Parquet reader builders
+receive the cache from the read context and create cache keys with
+``CacheKind::PARQUET_METADATA`` internally.
+
+The cache key represents the file footer and is created from the file URI with
+position ``-1`` and length ``-1``. Callers do not need to construct this key
+directly; they only need to route ``CacheKind::PARQUET_METADATA`` entries to an
+appropriate cache backend.
+
+Example:
+
+.. code-block:: cpp
+
+   class RoutingCache : public paimon::Cache {
+    public:
+     RoutingCache(std::shared_ptr<paimon::Cache> default_cache,
+                  std::shared_ptr<paimon::Cache> parquet_metadata_cache)
+         : default_cache_(std::move(default_cache)),
+           parquet_metadata_cache_(std::move(parquet_metadata_cache)) {}
+
+     paimon::Result<std::shared_ptr<paimon::CacheValue>> Get(
+         const std::shared_ptr<paimon::CacheKey>& key,
+         std::function<paimon::Result<std::shared_ptr<paimon::CacheValue>>(
+             const std::shared_ptr<paimon::CacheKey>&)> supplier) override {
+       return Select(key)->Get(key, std::move(supplier));
+     }
+
+     // Put(), Invalidate(), InvalidateAll(), and Size() route in the same way.
+
+    private:
+     std::shared_ptr<paimon::Cache> Select(
+         const std::shared_ptr<paimon::CacheKey>& key) const {
+       return key && key->GetKind() == paimon::CacheKind::PARQUET_METADATA
+                  ? parquet_metadata_cache_
+                  : default_cache_;
+     }
+
+     std::shared_ptr<paimon::Cache> default_cache_;
+     std::shared_ptr<paimon::Cache> parquet_metadata_cache_;
+   };
+
+   auto cache = std::make_shared<RoutingCache>(
+       std::make_shared<MyDefaultCache>(),
+       std::make_shared<MyParquetMetadataCache>());
+
+   paimon::ScanContextBuilder scan_builder(table_path);
+   scan_builder.WithCache(cache);
+
+   paimon::ReadContextBuilder read_builder(table_path);
+   read_builder.WithCache(cache);
+
+Passing ``nullptr`` or omitting ``WithCache()`` leaves Parquet metadata caching
+disabled. If a file URI cannot be obtained, paimon-cpp also bypasses the cache
+and opens the Parquet file normally.
+
+Future Optimizations
+--------------------
+
+- Add hit, miss, bypass, and eviction metrics for Parquet metadata cache.
+- Add single-flight loading for high-concurrency misses on the same Parquet
+  file.
+- Evaluate sharing cached metadata footer bytes with page-index prefetch logic when
+  those read paths can use the same cache abstraction.

include/paimon/cache/cache.h

@@ -32,6 +32,7 @@ class CacheValue;
 enum class CacheKind {
     DEFAULT,
     MANIFEST,
+    PARQUET_METADATA,
 };
 
 class PAIMON_EXPORT CacheKey {
@@ -40,6 +41,7 @@ class PAIMON_EXPORT CacheKey {
                                                  int32_t length, bool is_index);
     static std::shared_ptr<CacheKey> ForKind(const std::string& file_path, int64_t position,
                                              int32_t length, CacheKind kind);
+    static std::shared_ptr<CacheKey> ForParquetMeta(const std::string& file_uri);
 
  public:
     virtual ~CacheKey() = default;

include/paimon/format/reader_builder.h

@@ -24,6 +24,7 @@
 #include "paimon/type_fwd.h"
 
 namespace paimon {
+class Cache;
 
 /// Create a file batch reader based on the file path. Allows you to specify memory pool.
 class PAIMON_EXPORT ReaderBuilder {
@@ -33,6 +34,12 @@ class PAIMON_EXPORT ReaderBuilder {
     /// Set memory pool to use.
     virtual ReaderBuilder* WithMemoryPool(const std::shared_ptr<MemoryPool>& pool) = 0;
 
+    /// Inject a cache for reader-specific immutable metadata.
+    virtual ReaderBuilder* WithCache(const std::shared_ptr<Cache>& cache) {
+        (void)cache;
+        return this;
+    }
+
     /// Build a file batch reader based on the created `InputStream`.
     virtual Result<std::unique_ptr<FileBatchReader>> Build(
         const std::shared_ptr<InputStream>& path) const = 0;

src/paimon/common/io/cache/cache_key.cpp

@@ -31,6 +31,10 @@ std::shared_ptr<CacheKey> CacheKey::ForKind(const std::string& file_path, int64_
     return key;
 }
 
+std::shared_ptr<CacheKey> CacheKey::ForParquetMeta(const std::string& file_uri) {
+    return ForKind(file_uri, /*position=*/-1, /*length=*/-1, CacheKind::PARQUET_METADATA);
+}
+
 bool PositionCacheKey::IsIndex() const {
     return is_index_;
 }

src/paimon/core/operation/abstract_split_read.cpp

@@ -117,6 +117,7 @@ Result<std::unique_ptr<ReaderBuilder>> AbstractSplitRead::PrepareReaderBuilder(
     PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReaderBuilder> reader_builder,
                            file_format->CreateReaderBuilder(options_.GetReadBatchSize()));
     reader_builder->WithMemoryPool(pool_);
+    reader_builder->WithCache(options_.GetCache());
     return reader_builder;
 }
 

src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp

@@ -120,7 +120,8 @@ class PageFilteredRowGroupReaderTest : public ::testing::Test {
         options[PARQUET_READ_ENABLE_PAGE_INDEX_FILTER] = "true";
         ASSERT_OK_AND_ASSIGN(
             auto batch_reader,
-            ParquetFileBatchReader::Create(std::move(in_stream), arrow_pool_, options, batch_size));
+            ParquetFileBatchReader::Create(std::move(in_stream), arrow_pool_, options, batch_size,
+                                           /*file_metadata=*/nullptr));
         auto c_schema = std::make_unique<ArrowSchema>();
         ASSERT_TRUE(arrow::ExportSchema(*read_schema, c_schema.get()).ok());
         ASSERT_OK(batch_reader->SetReadSchema(c_schema.get(), predicate,

src/paimon/format/parquet/parquet_file_batch_reader.cpp

@@ -71,7 +71,8 @@ ParquetFileBatchReader::ParquetFileBatchReader(
 Result<std::unique_ptr<ParquetFileBatchReader>> ParquetFileBatchReader::Create(
     std::shared_ptr<arrow::io::RandomAccessFile>&& input_stream,
     const std::shared_ptr<arrow::MemoryPool>& pool,
-    const std::map<std::string, std::string>& options, int32_t batch_size) {
+    const std::map<std::string, std::string>& options, int32_t batch_size,
+    std::shared_ptr<::parquet::FileMetaData> file_metadata) {
     try {
         assert(input_stream);
         PAIMON_ASSIGN_OR_RAISE(::parquet::ReaderProperties reader_properties,
@@ -81,7 +82,8 @@ Result<std::unique_ptr<ParquetFileBatchReader>> ParquetFileBatchReader::Create(
                                CreateArrowReaderProperties(pool, options, batch_size));
 
         ::parquet::arrow::FileReaderBuilder file_reader_builder;
-        PAIMON_RETURN_NOT_OK_FROM_ARROW(file_reader_builder.Open(input_stream, reader_properties));
+        PAIMON_RETURN_NOT_OK_FROM_ARROW(
+            file_reader_builder.Open(input_stream, reader_properties, std::move(file_metadata)));
 
         std::unique_ptr<::parquet::arrow::FileReader> file_reader;
         PAIMON_RETURN_NOT_OK_FROM_ARROW(file_reader_builder.memory_pool(pool.get())

src/paimon/format/parquet/parquet_file_batch_reader.h

@@ -51,6 +51,9 @@ namespace io {
 class RandomAccessFile;
 }  // namespace io
 }  // namespace arrow
+namespace parquet {
+class FileMetaData;
+}  // namespace parquet
 namespace paimon {
 class Metrics;
 class Predicate;
@@ -64,7 +67,12 @@ class ParquetFileBatchReader : public PrefetchFileBatchReader {
     static Result<std::unique_ptr<ParquetFileBatchReader>> Create(
         std::shared_ptr<arrow::io::RandomAccessFile>&& input_stream,
         const std::shared_ptr<arrow::MemoryPool>& pool,
-        const std::map<std::string, std::string>& options, int32_t batch_size);
+        const std::map<std::string, std::string>& options, int32_t batch_size,
+        std::shared_ptr<::parquet::FileMetaData> file_metadata);
+
+    static Result<::parquet::ReaderProperties> CreateReaderProperties(
+        const std::shared_ptr<arrow::MemoryPool>& pool,
+        const std::map<std::string, std::string>& options);
 
     // For timestamp type, we return the schema stored in file, e.g., second in parquet file will
     // store as milli.
@@ -128,10 +136,6 @@ class ParquetFileBatchReader : public PrefetchFileBatchReader {
                            const std::map<std::string, std::string>& options,
                            const std::shared_ptr<arrow::MemoryPool>& arrow_pool);
 
-    static Result<::parquet::ReaderProperties> CreateReaderProperties(
-        const std::shared_ptr<arrow::MemoryPool>& pool,
-        const std::map<std::string, std::string>& options);
-
     static Result<::parquet::ArrowReaderProperties> CreateArrowReaderProperties(
         const std::shared_ptr<arrow::MemoryPool>& pool,
         const std::map<std::string, std::string>& options, int32_t batch_size);