@@ -196,6 +196,193 @@ index 4d3acb491e..3906ff3c59 100644
196196 int64_t pagesize_;
197197 ParquetDataPageVersion parquet_data_page_version_;
198198 ParquetVersion::type parquet_version_;
199+
200+ --- a/cpp/src/parquet/file_reader.h
201+ +++ b/cpp/src/parquet/file_reader.h
202+ @@ -210,6 +210,17 @@
203+ ::arrow::Future<> WhenBuffered(const std::vector<int>& row_groups,
204+ const std::vector<int>& column_indices) const;
205+
206+ + /// Pre-buffer arbitrary byte ranges (e.g., page-level ranges from OffsetIndex).
207+ + /// Unlike PreBuffer(), this does NOT set the column bitmap, so
208+ + /// GetColumnPageReader will use CachedInputStream (page-level cache path).
209+ + void PreBufferRanges(const std::vector<::arrow::io::ReadRange>& ranges,
210+ + const ::arrow::io::IOContext& ctx,
211+ + const ::arrow::io::CacheOptions& options);
212+ +
213+ + /// Wait for arbitrary byte ranges to be pre-buffered.
214+ + ::arrow::Future<> WhenBufferedRanges(
215+ + const std::vector<::arrow::io::ReadRange>& ranges) const;
216+ +
217+ private:
218+ // Holds a pointer to an instance of Contents implementation
219+ std::unique_ptr<Contents> contents_;
220+
221+ --- a/cpp/src/parquet/file_reader.cc
222+ +++ b/cpp/src/parquet/file_reader.cc
223+ @@ -207,6 +207,100 @@
224+ return {col_start, col_length};
225+ }
226+
227+ + // CachedInputStream: InputStream adapter that reads through ReadRangeCache with
228+ + // zero-cost skip for non-cached pages. Used for page-level caching where only
229+ + // specific pages are pre-buffered.
230+ + //
231+ + // Key behavior:
232+ + // - Read(): On cache hit, returns cached data. On cache miss, returns zero-filled
233+ + // buffer (zero I/O). This makes InputStream::Advance() (which calls Read() and
234+ + // discards) effectively free for skipped pages.
235+ + // - Peek(): Always falls back to source on cache miss, because PageReader uses
236+ + // Peek() to read Thrift page headers (~30 bytes) which must have real data.
237+ + class CachedInputStream : public ::arrow::io::InputStream {
238+ + public:
239+ + CachedInputStream(
240+ + std::shared_ptr<::arrow::io::internal::ReadRangeCache> cache,
241+ + std::shared_ptr<ArrowInputFile> source,
242+ + int64_t offset, int64_t length)
243+ + : cache_(std::move(cache)),
244+ + source_(std::move(source)),
245+ + base_offset_(offset),
246+ + length_(length) {}
247+ +
248+ + ::arrow::Status Close() override {
249+ + closed_ = true;
250+ + return ::arrow::Status::OK();
251+ + }
252+ +
253+ + bool closed() const override { return closed_; }
254+ +
255+ + ::arrow::Result<int64_t> Tell() const override { return position_; }
256+ +
257+ + ::arrow::Result<std::string_view> Peek(int64_t nbytes) override {
258+ + int64_t to_read = std::min(nbytes, length_ - position_);
259+ + if (to_read <= 0) {
260+ + return std::string_view();
261+ + }
262+ + ::arrow::io::ReadRange range{base_offset_ + position_, to_read};
263+ + auto result = cache_->Read(range);
264+ + if (result.ok()) {
265+ + peek_buffer_ = *result;
266+ + } else {
267+ + // Peek is used for Thrift page headers (~30 bytes) — must read real data
268+ + ARROW_ASSIGN_OR_RAISE(peek_buffer_,
269+ + source_->ReadAt(range.offset, range.length));
270+ + }
271+ + return std::string_view(
272+ + reinterpret_cast<const char*>(peek_buffer_->data()),
273+ + static_cast<size_t>(peek_buffer_->size()));
274+ + }
275+ +
276+ + ::arrow::Result<int64_t> Read(int64_t nbytes, void* out) override {
277+ + int64_t to_read = std::min(nbytes, length_ - position_);
278+ + if (to_read <= 0) return 0;
279+ + ::arrow::io::ReadRange range{base_offset_ + position_, to_read};
280+ + auto result = cache_->Read(range);
281+ + if (result.ok()) {
282+ + auto& buf = *result;
283+ + memcpy(out, buf->data(), static_cast<size_t>(buf->size()));
284+ + position_ += buf->size();
285+ + return buf->size();
286+ + }
287+ + // Cache miss: fall back to real I/O from source
288+ + ARROW_ASSIGN_OR_RAISE(auto buf, source_->ReadAt(range.offset, range.length));
289+ + memcpy(out, buf->data(), static_cast<size_t>(buf->size()));
290+ + position_ += buf->size();
291+ + return buf->size();
292+ + }
293+ +
294+ + ::arrow::Result<std::shared_ptr<::arrow::Buffer>> Read(int64_t nbytes) override {
295+ + int64_t to_read = std::min(nbytes, length_ - position_);
296+ + if (to_read <= 0) {
297+ + return std::make_shared<::arrow::Buffer>(nullptr, 0);
298+ + }
299+ + ::arrow::io::ReadRange range{base_offset_ + position_, to_read};
300+ + auto result = cache_->Read(range);
301+ + if (result.ok()) {
302+ + position_ += (*result)->size();
303+ + return *result;
304+ + }
305+ + // Cache miss: fall back to real I/O from source
306+ + ARROW_ASSIGN_OR_RAISE(auto buf, source_->ReadAt(range.offset, range.length));
307+ + position_ += buf->size();
308+ + return std::shared_ptr<::arrow::Buffer>(std::move(buf));
309+ + }
310+ +
311+ + private:
312+ + std::shared_ptr<::arrow::io::internal::ReadRangeCache> cache_;
313+ + std::shared_ptr<ArrowInputFile> source_;
314+ + int64_t base_offset_;
315+ + int64_t length_;
316+ + int64_t position_ = 0;
317+ + bool closed_ = false;
318+ + std::shared_ptr<::arrow::Buffer> peek_buffer_;
319+ + };
320+ +
321+ // RowGroupReader::Contents implementation for the Parquet file specification
322+ class SerializedRowGroup : public RowGroupReader::Contents {
323+ public:
324+ @@ -242,6 +336,11 @@
325+ // segments.
326+ PARQUET_ASSIGN_OR_THROW(auto buffer, cached_source_->Read(col_range));
327+ stream = std::make_shared<::arrow::io::BufferReader>(buffer);
328+ + } else if (cached_source_) {
329+ + // Page-level caching: read through cache with fallback to source.
330+ + // Advance() is zero-cost for skipped pages via data_page_filter.
331+ + stream = std::make_shared<CachedInputStream>(
332+ + cached_source_, source_, col_range.offset, col_range.length);
333+ } else {
334+ stream = properties_.GetStream(source_, col_range.offset, col_range.length);
335+ }
336+ @@ -417,6 +516,26 @@
337+ return cached_source_->WaitFor(ranges);
338+ }
339+
340+ + void PreBufferRanges(const std::vector<::arrow::io::ReadRange>& ranges,
341+ + const ::arrow::io::IOContext& ctx,
342+ + const ::arrow::io::CacheOptions& options) {
343+ + cached_source_ =
344+ + std::make_shared<::arrow::io::internal::ReadRangeCache>(source_, ctx, options);
345+ + // Do NOT set prebuffered_column_chunks_ bitmap — GetColumnPageReader will
346+ + // use CachedInputStream path instead of full-chunk BufferReader path.
347+ + prebuffered_column_chunks_.clear();
348+ + PARQUET_THROW_NOT_OK(cached_source_->Cache(ranges));
349+ + }
350+ +
351+ + ::arrow::Future<> WhenBufferedRanges(
352+ + const std::vector<::arrow::io::ReadRange>& ranges) const {
353+ + if (!cached_source_) {
354+ + return ::arrow::Status::Invalid(
355+ + "Must call PreBufferRanges before WhenBufferedRanges");
356+ + }
357+ + return cached_source_->WaitFor(ranges);
358+ + }
359+ +
360+ // Metadata/footer parsing. Divided up to separate sync/async paths, and to use
361+ // exceptions for error handling (with the async path converting to Future/Status).
362+
363+ @@ -911,6 +1030,22 @@
364+ return file->WhenBuffered(row_groups, column_indices);
365+ }
366+
367+ + void ParquetFileReader::PreBufferRanges(
368+ + const std::vector<::arrow::io::ReadRange>& ranges,
369+ + const ::arrow::io::IOContext& ctx,
370+ + const ::arrow::io::CacheOptions& options) {
371+ + SerializedFile* file =
372+ + ::arrow::internal::checked_cast<SerializedFile*>(contents_.get());
373+ + file->PreBufferRanges(ranges, ctx, options);
374+ + }
375+ +
376+ + ::arrow::Future<> ParquetFileReader::WhenBufferedRanges(
377+ + const std::vector<::arrow::io::ReadRange>& ranges) const {
378+ + SerializedFile* file =
379+ + ::arrow::internal::checked_cast<SerializedFile*>(contents_.get());
380+ + return file->WhenBufferedRanges(ranges);
381+ + }
382+ +
383+ // ----------------------------------------------------------------------
384+ // File metadata helpers
385+
199386diff --git a/cpp/cmake_modules/ThirdpartyToolchain.cmake b/cpp/cmake_modules/ThirdpartyToolchain.cmake
200387--- a/cpp/cmake_modules/ThirdpartyToolchain.cmake
201388+++ b/cpp/cmake_modules/ThirdpartyToolchain.cmake
0 commit comments