[feature](vparquet-reader) Implements parquet file page cache. (#59307)

### What problem does this PR solve?

Problem Summary:

### Release note

[Feature] Implementation of Parquet File Page Cache and Integration with
Unified Page Cache Framework

#### Solution Overview

This PR implements a page-level caching mechanism for Parquet files and
integrates it with Apache Doris's existing unified page cache framework,
significantly improving query performance by caching decompressed (or
compressed) data pages in memory.

Key Features

1. Unified Page Cache Integration

• Leverages Existing Framework: Directly integrates with Doris's
StoragePageCache infrastructure used for internal tables

• Shared Resource Management: Parquet cache shares memory pool and
eviction policies with internal table caches

• Consistent Monitoring: Reuses existing cache statistics and
RuntimeProfile for unified performance monitoring

• Cache Type Identification: Uses segment_v2::DATA_PAGE as cache page
type, consistent with internal table data page caching

2. Smart Caching Strategy

• Compression Ratio Awareness: Automatically chooses between caching
compressed or decompressed data based on
parquet_page_cache_decompress_threshold (default: 1.5)

• Flexible Storage: Caches decompressed data when
uncompressed_size/compressed_size ≤ threshold, otherwise caches
compressed data if enable_parquet_cache_compressed_pages=true

• Cache Key Design: Uses file_path::mtime::offset as key to ensure cache
consistency across file modifications

Author: kaka11chen

be/src/common/config.cpp

@@ -404,6 +404,12 @@ DEFINE_Int32(index_page_cache_percentage, "10");
 DEFINE_mBool(disable_storage_page_cache, "false");
 // whether to disable row cache feature in storage
 DEFINE_mBool(disable_storage_row_cache, "true");
+// Parquet page cache: threshold ratio for caching decompressed vs compressed pages
+// If uncompressed_size / compressed_size <= threshold, cache decompressed;
+// otherwise cache compressed if enable_parquet_cache_compressed_pages = true
+DEFINE_Double(parquet_page_cache_decompress_threshold, "1.5");
+// Parquet page cache: whether to enable caching compressed pages (when ratio exceeds threshold)
+DEFINE_Bool(enable_parquet_cache_compressed_pages, "false");
 // whether to disable pk page cache feature in storage
 DEFINE_Bool(disable_pk_storage_page_cache, "false");
 

be/src/common/config.h

@@ -453,6 +453,12 @@ DECLARE_Int32(index_page_cache_percentage);
 DECLARE_Bool(disable_storage_page_cache);
 // whether to disable row cache feature in storage
 DECLARE_mBool(disable_storage_row_cache);
+// Parquet page cache: threshold ratio for caching decompressed vs compressed pages
+// If uncompressed_size / compressed_size <= threshold, cache decompressed;
+// otherwise cache compressed if enable_parquet_cache_compressed_pages = true
+DECLARE_Double(parquet_page_cache_decompress_threshold);
+// Parquet page cache: whether to enable caching compressed pages (when ratio exceeds threshold)
+DECLARE_Bool(enable_parquet_cache_compressed_pages);
 // whether to disable pk page cache feature in storage
 DECLARE_Bool(disable_pk_storage_page_cache);
 

be/src/io/cache/cached_remote_file_reader.h

@@ -55,6 +55,8 @@ class CachedRemoteFileReader final : public FileReader {
 
     static std::pair<size_t, size_t> s_align_size(size_t offset, size_t size, size_t length);
 
+    int64_t mtime() const override { return _remote_file_reader->mtime(); }
+
 protected:
     Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
                         const IOContext* io_ctx) override;

be/src/io/file_factory.cpp

@@ -203,6 +203,18 @@ Result<io::FileReaderSPtr> FileFactory::create_file_reader(
         const io::FileSystemProperties& system_properties,
         const io::FileDescription& file_description, const io::FileReaderOptions& reader_options,
         RuntimeProfile* profile) {
+    auto reader_res = _create_file_reader_internal(system_properties, file_description,
+                                                   reader_options, profile);
+    if (!reader_res.has_value()) {
+        return unexpected(std::move(reader_res).error());
+    }
+    return std::move(reader_res).value();
+}
+
+Result<io::FileReaderSPtr> FileFactory::_create_file_reader_internal(
+        const io::FileSystemProperties& system_properties,
+        const io::FileDescription& file_description, const io::FileReaderOptions& reader_options,
+        RuntimeProfile* profile) {
     TFileType::type type = system_properties.system_type;
     switch (type) {
     case TFileType::FILE_LOCAL: {

be/src/io/file_factory.h

@@ -126,6 +126,12 @@ class FileFactory {
 
 private:
     static std::string _get_fs_name(const io::FileDescription& file_description);
+
+    /// Create FileReader without FS
+    static Result<io::FileReaderSPtr> _create_file_reader_internal(
+            const io::FileSystemProperties& system_properties,
+            const io::FileDescription& file_description,
+            const io::FileReaderOptions& reader_options, RuntimeProfile* profile = nullptr);
 };
 
 } // namespace doris

be/src/io/fs/broker_file_reader.cpp

@@ -39,12 +39,14 @@ struct IOContext;
 
 BrokerFileReader::BrokerFileReader(const TNetworkAddress& broker_addr, Path path, size_t file_size,
                                    TBrokerFD fd,
-                                   std::shared_ptr<BrokerServiceConnection> connection)
+                                   std::shared_ptr<BrokerServiceConnection> connection,
+                                   int64_t mtime)
         : _path(std::move(path)),
           _file_size(file_size),
           _broker_addr(broker_addr),
           _fd(fd),
-          _connection(std::move(connection)) {
+          _connection(std::move(connection)),
+          _mtime(mtime) {
     DorisMetrics::instance()->broker_file_open_reading->increment(1);
     DorisMetrics::instance()->broker_file_reader_total->increment(1);
 }

be/src/io/fs/broker_file_reader.h

@@ -38,7 +38,7 @@ struct IOContext;
 class BrokerFileReader final : public FileReader {
 public:
     BrokerFileReader(const TNetworkAddress& broker_addr, Path path, size_t file_size, TBrokerFD fd,
-                     std::shared_ptr<BrokerServiceConnection> connection);
+                     std::shared_ptr<BrokerServiceConnection> connection, int64_t mtime = 0);
 
     ~BrokerFileReader() override;
 
@@ -50,6 +50,8 @@ class BrokerFileReader final : public FileReader {
 
     bool closed() const override { return _closed.load(std::memory_order_acquire); }
 
+    int64_t mtime() const override { return _mtime; }
+
 protected:
     Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
                         const IOContext* io_ctx) override;
@@ -62,6 +64,7 @@ class BrokerFileReader final : public FileReader {
     TBrokerFD _fd;
 
     std::shared_ptr<BrokerServiceConnection> _connection;
+    int64_t _mtime;
     std::atomic<bool> _closed = false;
 };
 } // namespace doris::io

be/src/io/fs/broker_file_system.cpp

@@ -139,7 +139,7 @@ Status BrokerFileSystem::open_file_internal(const Path& file, FileReaderSPtr* re
                                error_msg(response->opStatus.message));
     }
     *reader = std::make_shared<BrokerFileReader>(_broker_addr, file, fsize, response->fd,
-                                                 _connection);
+                                                 _connection, opts.mtime);
     return Status::OK();
 }
 

be/src/io/fs/buffered_reader.h

@@ -160,6 +160,8 @@ class RangeCacheFileReader : public io::FileReader {
 
     bool closed() const override { return _closed; }
 
+    int64_t mtime() const override { return _inner_reader->mtime(); }
+
 protected:
     Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
                         const IOContext* io_ctx) override;
@@ -329,6 +331,8 @@ class MergeRangeFileReader : public io::FileReader {
 
     bool closed() const override { return _closed; }
 
+    int64_t mtime() const override { return _reader->mtime(); }
+
     // for test only
     size_t buffer_remaining() const { return _remaining; }
 
@@ -532,6 +536,8 @@ class PrefetchBufferedReader final : public io::FileReader {
 
     bool closed() const override { return _closed; }
 
+    int64_t mtime() const override { return _reader->mtime(); }
+
     void set_random_access_ranges(const std::vector<PrefetchRange>* random_access_ranges) {
         _random_access_ranges = random_access_ranges;
         for (auto& _pre_buffer : _pre_buffers) {
@@ -592,6 +598,8 @@ class InMemoryFileReader final : public io::FileReader {
 
     bool closed() const override { return _closed; }
 
+    int64_t mtime() const override { return _reader->mtime(); }
+
 protected:
     Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
                         const IOContext* io_ctx) override;
@@ -626,6 +634,8 @@ class BufferedStreamReader {
     virtual ~BufferedStreamReader() = default;
     // return the file path
     virtual std::string path() = 0;
+
+    virtual int64_t mtime() const = 0;
 };
 
 class BufferedFileStreamReader : public BufferedStreamReader, public ProfileCollector {
@@ -639,6 +649,8 @@ class BufferedFileStreamReader : public BufferedStreamReader, public ProfileColl
     Status read_bytes(Slice& slice, uint64_t offset, const IOContext* io_ctx) override;
     std::string path() override { return _file->path(); }
 
+    int64_t mtime() const override { return _file->mtime(); }
+
 protected:
     void _collect_profile_before_close() override {
         if (_file != nullptr) {

be/src/io/fs/file_reader.h

@@ -90,6 +90,9 @@ class FileReader : public doris::ProfileCollector {
 
     virtual const std::string& get_data_dir_path() { return VIRTUAL_REMOTE_DATA_DIR; }
 
+    // File modification time (seconds since epoch). Default to 0 meaning unknown.
+    virtual int64_t mtime() const = 0;
+
 protected:
     virtual Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
                                 const IOContext* io_ctx) = 0;