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

Author: kaka11chen

be/src/common/config.cpp

@@ -1000,6 +1000,11 @@ DEFINE_mInt32(parquet_header_max_size_mb, "1");
 DEFINE_mInt32(parquet_rowgroup_max_buffer_mb, "128");
 // Max buffer size for parquet chunk column
 DEFINE_mInt32(parquet_column_max_buffer_mb, "8");
+// Enable Parquet file page cache for Parquet page reader
+DEFINE_Bool(enable_parquet_file_page_cache, "true");
+// Threshold ratio to decide caching decompressed parquet page. If uncompressed_size <=
+// parquet_page_cache_decompress_threshold * compressed_size, we cache decompressed page.
+DEFINE_mDouble(parquet_page_cache_decompress_threshold, "1.5");
 DEFINE_mDouble(max_amplified_read_ratio, "0.8");
 DEFINE_mInt32(merged_oss_min_io_size, "1048576");
 DEFINE_mInt32(merged_hdfs_min_io_size, "8192");

be/src/common/config.h

@@ -144,6 +144,13 @@ DECLARE_String(memory_mode);
 // if true, process memory limit and memory usage based on cgroup memory info.
 DECLARE_mBool(enable_use_cgroup_memory_info);
 
+// Enable Parquet file page cache for Parquet page reader
+DECLARE_Bool(enable_parquet_file_page_cache);
+
+// Threshold ratio to decide caching decompressed parquet page. If uncompressed_size <=
+// parquet_page_cache_decompress_threshold * compressed_size, we cache decompressed page.
+DECLARE_mDouble(parquet_page_cache_decompress_threshold);
+
 // process memory limit specified as number of bytes
 // ('<int>[bB]?'), megabytes ('<float>[mM]'), gigabytes ('<float>[gG]'),
 // or percentage of the physical memory ('<int>%').

be/src/vec/exec/format/parquet/vparquet_column_chunk_reader.cpp

@@ -25,6 +25,8 @@
 #include <utility>
 
 #include "common/compiler_util.h" // IWYU pragma: keep
+#include "io/fs/buffered_reader.h"
+#include "olap/page_cache.h"
 #include "util/bit_util.h"
 #include "util/block_compression.h"
 #include "util/runtime_profile.h"
@@ -103,7 +105,7 @@ Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::_parse_first_page_header(
     RETURN_IF_ERROR(parse_page_header());
 
     const tparquet::PageHeader* header = nullptr;
-    RETURN_IF_ERROR(_page_reader->get_page_header(header));
+    RETURN_IF_ERROR(_page_reader->get_page_header(&header));
     if (header->type == tparquet::PageType::DICTIONARY_PAGE) {
         // the first page maybe directory page even if _metadata.__isset.dictionary_page_offset == false,
         // so we should parse the directory page in next_page()
@@ -124,8 +126,7 @@ Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::parse_page_header() {
     RETURN_IF_ERROR(_page_reader->parse_page_header());
 
     const tparquet::PageHeader* header = nullptr;
-    ;
-    RETURN_IF_ERROR(_page_reader->get_page_header(header));
+    RETURN_IF_ERROR(_page_reader->get_page_header(&header));
     int32_t page_num_values = _page_reader->is_header_v2() ? header->data_page_header_v2.num_values
                                                            : header->data_page_header.num_values;
     _remaining_rep_nums = page_num_values;
@@ -168,40 +169,161 @@ Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::load_page_data() {
     }
 
     const tparquet::PageHeader* header = nullptr;
-    RETURN_IF_ERROR(_page_reader->get_page_header(header));
+    RETURN_IF_ERROR(_page_reader->get_page_header(&header));
     int32_t uncompressed_size = header->uncompressed_page_size;
 
+    // First, try to reuse a cache handle previously discovered by PageReader
+    // (header-only lookup) to avoid a second lookup here. If no handle is
+    // attached, fall back to a StoragePageCache lookup for a decompressed page.
+    if (config::enable_parquet_file_page_cache && !config::disable_storage_page_cache &&
+        StoragePageCache::instance() != nullptr) {
+        if (_page_reader->has_page_cache_handle()) {
+            const PageCacheHandle& handle = _page_reader->page_cache_handle();
+            Slice cached = handle.data();
+            size_t header_size = _page_reader->header_bytes().size();
+            //size_t levels_size = 0;
+            size_t levels_size = 0;
+            if (header->__isset.data_page_header_v2) {
+                const tparquet::DataPageHeaderV2& header_v2 = header->data_page_header_v2;
+                size_t rl = header_v2.repetition_levels_byte_length;
+                size_t dl = header_v2.definition_levels_byte_length;
+                levels_size = rl + dl;
+                _v2_rep_levels =
+                        Slice(reinterpret_cast<const uint8_t*>(cached.data) + header_size, rl);
+                _v2_def_levels =
+                        Slice(reinterpret_cast<const uint8_t*>(cached.data) + header_size + rl, dl);
+            }
+            // payload_slice points to the bytes after header and levels
+            Slice payload_slice(cached.data + header_size + levels_size,
+                                cached.size - header_size - levels_size);
+            // Determine whether cached payload is compressed by comparing sizes
+            bool payload_is_compressed = false;
+            // if (header->__isset.data_page_header_v2) {
+            //     const tparquet::DataPageHeaderV2& header_v2 = header->data_page_header_v2;
+            //     levels_size = header_v2.repetition_levels_byte_length + header_v2.definition_levels_byte_length;
+            //     // compressed_page_size is stored on the top-level PageHeader
+            //     payload_is_compressed = (payload_slice.size == static_cast<size_t>(header->compressed_page_size) - levels_size);
+            // } else if (header->__isset.data_page_header) {
+            //     payload_is_compressed = (payload_slice.size == static_cast<size_t>(header->compressed_page_size));
+            // }
+
+            if (header->__isset.data_page_header_v2) {
+                const tparquet::DataPageHeaderV2& header_v2 = header->data_page_header_v2;
+                levels_size = header_v2.repetition_levels_byte_length +
+                              header_v2.definition_levels_byte_length;
+                // compressed_page_size is stored on the top-level PageHeader
+                payload_is_compressed =
+                        (!(payload_slice.size ==
+                           static_cast<size_t>(header->uncompressed_page_size) - levels_size)) &&
+                        (payload_slice.size ==
+                         static_cast<size_t>(header->compressed_page_size) - levels_size);
+            } else if (header->__isset.data_page_header) {
+                payload_is_compressed =
+                        (!(payload_slice.size ==
+                           static_cast<size_t>(header->uncompressed_page_size))) &&
+                        (payload_slice.size == static_cast<size_t>(header->compressed_page_size));
+            }
+
+            if (payload_is_compressed && _block_compress_codec != nullptr) {
+                // Decompress cached payload into _decompress_buf for decoding
+                size_t uncompressed_payload_size =
+                        header->__isset.data_page_header_v2
+                                ? static_cast<size_t>(header->uncompressed_page_size) - levels_size
+                                : static_cast<size_t>(header->uncompressed_page_size);
+                _reserve_decompress_buf(uncompressed_payload_size);
+                _page_data = Slice(_decompress_buf.get(), uncompressed_payload_size);
+                RETURN_IF_ERROR(_block_compress_codec->decompress(payload_slice, &_page_data));
+            } else {
+                // Cached payload is already uncompressed
+                _page_data = payload_slice;
+            }
+            // page cache counters were incremented when PageReader did the header-only
+            // cache lookup. Do not increment again to avoid double-counting.
+            goto data_loaded;
+        } else {
+        }
+    }
+
     if (_block_compress_codec != nullptr) {
         Slice compressed_data;
         RETURN_IF_ERROR(_page_reader->get_page_data(compressed_data));
+        std::vector<uint8_t> level_bytes;
         if (header->__isset.data_page_header_v2) {
             const tparquet::DataPageHeaderV2& header_v2 = header->data_page_header_v2;
             // uncompressed_size = rl + dl + uncompressed_data_size
             // compressed_size = rl + dl + compressed_data_size
             uncompressed_size -= header_v2.repetition_levels_byte_length +
                                  header_v2.definition_levels_byte_length;
+            // copy level bytes (rl + dl) so that we can cache header + levels + uncompressed payload
+            size_t rl = header_v2.repetition_levels_byte_length;
+            size_t dl = header_v2.definition_levels_byte_length;
+            level_bytes.resize(rl + dl);
+            memcpy(level_bytes.data(), compressed_data.data, rl + dl);
+            // now remove levels from compressed_data for decompression
             _get_uncompressed_levels(header_v2, compressed_data);
         }
         bool is_v2_compressed =
                 header->__isset.data_page_header_v2 && header->data_page_header_v2.is_compressed;
-        if (header->__isset.data_page_header || is_v2_compressed) {
-            // check decompressed buffer size
+        bool page_has_compression = header->__isset.data_page_header || is_v2_compressed;
+
+        if (page_has_compression) {
+            // Decompress payload for immediate decoding
             _reserve_decompress_buf(uncompressed_size);
             _page_data = Slice(_decompress_buf.get(), uncompressed_size);
             SCOPED_RAW_TIMER(&_chunk_statistics.decompress_time);
             _chunk_statistics.decompress_cnt++;
             RETURN_IF_ERROR(_block_compress_codec->decompress(compressed_data, &_page_data));
+
+            // Decide whether to cache decompressed payload or compressed payload based on threshold
+            bool should_cache_decompressed = false;
+            if (header->compressed_page_size > 0) {
+                should_cache_decompressed =
+                        (static_cast<double>(header->uncompressed_page_size) <=
+                         static_cast<double>(config::parquet_page_cache_decompress_threshold) *
+                                 static_cast<double>(header->compressed_page_size));
+            }
+
+            if (config::enable_parquet_file_page_cache && !config::disable_storage_page_cache &&
+                StoragePageCache::instance() != nullptr && !_page_reader->header_bytes().empty()) {
+                if (should_cache_decompressed) {
+                    _insert_page_into_cache(level_bytes, _page_data);
+                } else {
+                    // cache the compressed payload as-is (header | levels | compressed_payload)
+                    _insert_page_into_cache(level_bytes,
+                                            Slice(compressed_data.data, compressed_data.size));
+                }
+            }
         } else {
-            // Don't need decompress
+            // no compression on this page, use the data directly
             _page_data = Slice(compressed_data.data, compressed_data.size);
+            if (config::enable_parquet_file_page_cache && !config::disable_storage_page_cache &&
+                StoragePageCache::instance() != nullptr) {
+                _insert_page_into_cache(level_bytes, _page_data);
+            }
         }
     } else {
-        RETURN_IF_ERROR(_page_reader->get_page_data(_page_data));
+        // For uncompressed page, we may still need to extract v2 levels
+        std::vector<uint8_t> level_bytes;
+        Slice uncompressed_data;
+        RETURN_IF_ERROR(_page_reader->get_page_data(uncompressed_data));
         if (header->__isset.data_page_header_v2) {
-            _get_uncompressed_levels(header->data_page_header_v2, _page_data);
+            const tparquet::DataPageHeaderV2& header_v2 = header->data_page_header_v2;
+            size_t rl = header_v2.repetition_levels_byte_length;
+            size_t dl = header_v2.definition_levels_byte_length;
+            level_bytes.resize(rl + dl);
+            memcpy(level_bytes.data(), uncompressed_data.data, rl + dl);
+            _get_uncompressed_levels(header_v2, uncompressed_data);
+        }
+        // copy page data out
+        _page_data = Slice(uncompressed_data.data, uncompressed_data.size);
+        // Optionally cache uncompressed data for uncompressed pages
+        if (config::enable_parquet_file_page_cache && !config::disable_storage_page_cache &&
+            StoragePageCache::instance() != nullptr) {
+            _insert_page_into_cache(level_bytes, _page_data);
         }
     }
 
+data_loaded:
     // Initialize repetition level and definition level. Skip when level = 0, which means required field.
     if (_max_rep_level > 0) {
         SCOPED_RAW_TIMER(&_chunk_statistics.decode_level_time);
@@ -243,7 +365,15 @@ Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::load_page_data() {
         _decoders[static_cast<int>(encoding)] = std::move(page_decoder);
         _page_decoder = _decoders[static_cast<int>(encoding)].get();
     }
-    // Reset page data for each page
+    // Reset page data for each page.
+    // If this is a v2 data page, _page_data currently contains rl+dl followed by payload.
+    // The decoder expects payload-only, so strip the level bytes into a temporary Slice
+    // that points into the same cached memory (so ownership remains with the cache handle).
+    // Slice payload_slice = _page_data;
+    // if (header->__isset.data_page_header_v2) {
+    //     const tparquet::DataPageHeaderV2& header_v2 = header->data_page_header_v2;
+    //     _get_uncompressed_levels(header_v2, payload_slice);
+    // }
     RETURN_IF_ERROR(_page_decoder->set_data(&_page_data));
 
     _state = DATA_LOADED;
@@ -253,7 +383,7 @@ Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::load_page_data() {
 template <bool IN_COLLECTION, bool OFFSET_INDEX>
 Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::_decode_dict_page() {
     const tparquet::PageHeader* header = nullptr;
-    RETURN_IF_ERROR(_page_reader->get_page_header(header));
+    RETURN_IF_ERROR(_page_reader->get_page_header(&header));
     DCHECK_EQ(tparquet::PageType::DICTIONARY_PAGE, header->type);
     SCOPED_RAW_TIMER(&_chunk_statistics.decode_dict_time);
 
@@ -305,6 +435,33 @@ void ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::_reserve_decompress_buf(siz
     }
 }
 
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+void ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::_insert_page_into_cache(
+        const std::vector<uint8_t>& level_bytes, const Slice& payload) {
+    int64_t header_start = _page_reader->header_start_offset();
+    StoragePageCache::CacheKey key(_stream_reader->path(), _page_reader->file_end_offset(),
+                                   header_start);
+    const std::vector<uint8_t>& header_bytes = _page_reader->header_bytes();
+    size_t total = header_bytes.size() + level_bytes.size() + payload.size;
+    auto* page = new DataPage(total, true, segment_v2::DATA_PAGE);
+    size_t pos = 0;
+    memcpy(page->data() + pos, header_bytes.data(), header_bytes.size());
+    pos += header_bytes.size();
+    if (!level_bytes.empty()) {
+        memcpy(page->data() + pos, level_bytes.data(), level_bytes.size());
+        pos += level_bytes.size();
+    }
+    if (payload.size > 0) {
+        memcpy(page->data() + pos, payload.data, payload.size);
+        pos += payload.size;
+    }
+    page->reset_size(total);
+    PageCacheHandle handle;
+    StoragePageCache::instance()->insert(key, page, &handle, segment_v2::DATA_PAGE);
+    _page_reader->attach_page_cache_handle(std::move(handle));
+    _chunk_statistics.page_cache_write_counter += 1;
+}
+
 template <bool IN_COLLECTION, bool OFFSET_INDEX>
 Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::skip_values(size_t num_values,
                                                                    bool skip_data) {

be/src/vec/exec/format/parquet/vparquet_column_chunk_reader.h

@@ -61,6 +61,15 @@ struct ColumnChunkReaderStatistics {
     int64_t skip_page_header_num = 0;
     int64_t parse_page_header_num = 0;
     int64_t read_page_header_time = 0;
+    // page cache metrics
+    // total pages read (from cache or file)
+    int64_t page_read_counter = 0;
+    int64_t page_cache_write_counter = 0;
+    // number of cache hits (either compressed or decompressed)
+    int64_t page_cache_hit_counter = 0;
+    // per-hit breakdown
+    int64_t page_cache_compressed_hit_counter = 0;
+    int64_t page_cache_decompressed_hit_counter = 0;
 };
 
 /**
@@ -74,6 +83,15 @@ struct ColumnChunkReaderStatistics {
         int64_t decode_level_time = 0;
         int64_t skip_page_header_num = 0;
         int64_t parse_page_header_num = 0;
+        // page cache metrics
+        // total pages read (from cache or file)
+        int64_t page_read_counter = 0;
+        int64_t page_cache_write_counter = 0;
+        // number of cache hits (either compressed or decompressed)
+        int64_t page_cache_hit_counter = 0;
+        // per-hit breakdown
+        int64_t page_cache_compressed_hit_counter = 0;
+        int64_t page_cache_decompressed_hit_counter = 0;
     };
  * // Create chunk reader
  * ColumnChunkReader chunk_reader(BufferedStreamReader* reader,
@@ -155,6 +173,15 @@ class ColumnChunkReader {
                 _page_reader->page_statistics().parse_page_header_num;
         _chunk_statistics.read_page_header_time =
                 _page_reader->page_statistics().read_page_header_time;
+        _chunk_statistics.page_read_counter += _page_reader->page_statistics().page_read_counter;
+        _chunk_statistics.page_cache_write_counter +=
+                _page_reader->page_statistics().page_cache_write_counter;
+        _chunk_statistics.page_cache_hit_counter +=
+                _page_reader->page_statistics().page_cache_hit_counter;
+        _chunk_statistics.page_cache_compressed_hit_counter +=
+                _page_reader->page_statistics().page_cache_compressed_hit_counter;
+        _chunk_statistics.page_cache_decompressed_hit_counter +=
+                _page_reader->page_statistics().page_cache_decompressed_hit_counter;
         return _chunk_statistics;
     }
 
@@ -193,6 +220,12 @@ class ColumnChunkReader {
                                  size_t* result_rows, bool* cross_page);
     Status load_cross_page_nested_row(std::vector<level_t>& rep_levels, bool* cross_page);
 
+    // Test helpers / accessors
+    Slice get_page_data() const { return _page_data; }
+    const Slice& v2_rep_levels() const { return _v2_rep_levels; }
+    const Slice& v2_def_levels() const { return _v2_def_levels; }
+    ColumnChunkReaderStatistics& statistics() { return chunk_statistics(); }
+
 private:
     enum ColumnChunkReaderState { NOT_INIT, INITIALIZED, HEADER_PARSED, DATA_LOADED, PAGE_SKIPPED };
 
@@ -202,6 +235,7 @@ class ColumnChunkReader {
 
     void _reserve_decompress_buf(size_t size);
     int32_t _get_type_length();
+    void _insert_page_into_cache(const std::vector<uint8_t>& level_bytes, const Slice& payload);
 
     void _get_uncompressed_levels(const tparquet::DataPageHeaderV2& page_v2, Slice& page_data);
     Status _skip_nested_rows_in_page(size_t num_rows);