feat: support prefetch for orc (#77)

Author: lucasfang

cmake_modules/orc.diff

@@ -30,6 +30,20 @@ index a914e5f26..efe1d4933 100644
    };
 
    // Specializations for char
+diff --git a/c++/include/orc/Reader.hh b/c++/include/orc/Reader.hh
+index b015b6491..585e50ec5 100644
+--- a/c++/include/orc/Reader.hh
++++ b/c++/include/orc/Reader.hh
+@@ -659,6 +659,9 @@ namespace orc {
+     virtual void preBuffer(const std::vector<uint32_t>& stripes,
+                            const std::list<uint64_t>& includeTypes) = 0;
+
++    virtual std::vector<std::pair<uint64_t, uint64_t>> preBufferRange(
++        const std::vector<uint32_t>& stripes, const std::list<uint64_t>& includeTypes) = 0;
++
+     /**
+      * Release cached entries whose right boundary is less than or equal to the given boundary.
+      * @param boundary the boundary value to release cache entries
 diff --git a/c++/src/ColumnReader.cc b/c++/src/ColumnReader.cc
 index af434c37c..08393259c 100644
 --- a/c++/src/ColumnReader.cc
@@ -328,6 +342,86 @@ index ed7fee737..a8ee8a67c 100644
      reserve(newSize);
      if (newSize > currentSize_) {
        memset(buf_ + currentSize_, 0, newSize - currentSize_);
+diff --git a/c++/src/Reader.cc b/c++/src/Reader.cc
+index c93c62f6c..2a821b622 100644
+--- a/c++/src/Reader.cc
++++ b/c++/src/Reader.cc
+@@ -1531,8 +1531,8 @@ namespace orc {
+     }
+   }
+
+-  void ReaderImpl::preBuffer(const std::vector<uint32_t>& stripes,
+-                             const std::list<uint64_t>& includeTypes) {
++  std::vector<std::pair<uint64_t, uint64_t>> ReaderImpl::preBufferRange(
++      const std::vector<uint32_t>& stripes, const std::list<uint64_t>& includeTypes) {
+     std::vector<uint32_t> newStripes;
+     for (auto stripe : stripes) {
+       if (stripe < static_cast<uint32_t>(footer_->stripes_size())) newStripes.push_back(stripe);
+@@ -1544,7 +1544,7 @@ namespace orc {
+     }
+
+     if (newStripes.empty() || newIncludeTypes.empty()) {
+-      return;
++      return {};
+     }
+
+     orc::RowReaderOptions rowReaderOptions;
+@@ -1553,7 +1553,7 @@ namespace orc {
+     std::vector<bool> selectedColumns;
+     columnSelector.updateSelected(selectedColumns, rowReaderOptions);
+
+-    std::vector<ReadRange> ranges;
++    std::vector<std::pair<uint64_t, uint64_t>> ranges;
+     ranges.reserve(newIncludeTypes.size());
+     for (auto stripe : newStripes) {
+       // get stripe information
+@@ -1598,17 +1598,23 @@ namespace orc {
+
+         offset += stream.length();
+       }
++    }
++    return ranges;
++  }
+
+-      {
+-        std::lock_guard<std::mutex> lock(contents_->readCacheMutex);
+-
+-        if (!contents_->readCache) {
+-          contents_->readCache = std::make_shared<ReadRangeCache>(
+-              getStream(), options_.getCacheOptions(), contents_->pool, contents_->readerMetrics);
+-        }
+-        contents_->readCache->cache(std::move(ranges));
+-      }
++  void ReaderImpl::preBuffer(const std::vector<uint32_t>& stripes,
++                             const std::list<uint64_t>& includeTypes) {
++    auto ranges = preBufferRange(stripes, includeTypes);
++    std::vector<ReadRange> read_ranges;
++    for (const auto& range : ranges) {
++      read_ranges.emplace_back(range.first, range.second);
++    }
++    std::lock_guard<std::mutex> lock(contents_->readCacheMutex);
++    if (!contents_->readCache) {
++      contents_->readCache = std::make_shared<ReadRangeCache>(
++          getStream(), options_.getCacheOptions(), contents_->pool, contents_->readerMetrics);
+     }
++    contents_->readCache->cache(std::move(read_ranges));
+   }
+
+   RowReader::~RowReader() {
+diff --git a/c++/src/Reader.hh b/c++/src/Reader.hh
+index 39ca73967..13da45a49 100644
+--- a/c++/src/Reader.hh
++++ b/c++/src/Reader.hh
+@@ -387,6 +387,9 @@ namespace orc {
+     std::map<uint32_t, BloomFilterIndex> getBloomFilters(
+         uint32_t stripeIndex, const std::set<uint32_t>& included) const override;
+
++    std::vector<std::pair<uint64_t, uint64_t>> preBufferRange(
++        const std::vector<uint32_t>& stripes, const std::list<uint64_t>& includeTypes) override;
++
+     void preBuffer(const std::vector<uint32_t>& stripes,
+                    const std::list<uint64_t>& includeTypes) override;
+     void releaseBuffer(uint64_t boundary) override;
 diff --git a/cmake_modules/ThirdpartyToolchain.cmake b/cmake_modules/ThirdpartyToolchain.cmake
 index 9b2c829c7..434841224 100644
 --- a/cmake_modules/ThirdpartyToolchain.cmake

include/paimon/read_context.h

@@ -26,6 +26,7 @@
 #include "paimon/predicate/predicate.h"
 #include "paimon/result.h"
 #include "paimon/type_fwd.h"
+#include "paimon/utils/read_ahead_cache.h"
 #include "paimon/utils/special_field_ids.h"
 #include "paimon/visibility.h"
 
@@ -53,7 +54,8 @@ class PAIMON_EXPORT ReadContext {
                 const std::shared_ptr<Executor>& executor,
                 const std::shared_ptr<FileSystem>& specific_file_system,
                 const std::map<std::string, std::string>& fs_scheme_to_identifier_map,
-                const std::map<std::string, std::string>& options);
+                const std::map<std::string, std::string>& options, bool enable_prefetch_cache,
+                const CacheConfig& cache_config);
     ~ReadContext();
 
     const std::string& GetPath() const {
@@ -115,6 +117,14 @@ class PAIMON_EXPORT ReadContext {
         return specific_file_system_;
     }
 
+    bool EnablePrefetchCache() const {
+        return enable_prefetch_cache_;
+    }
+
+    const CacheConfig& GetCacheConfig() const {
+        return cache_config_;
+    }
+
  private:
     std::string path_;
     std::string branch_;
@@ -133,6 +143,8 @@ class PAIMON_EXPORT ReadContext {
     std::shared_ptr<FileSystem> specific_file_system_;
     std::map<std::string, std::string> fs_scheme_to_identifier_map_;
     std::map<std::string, std::string> options_;
+    bool enable_prefetch_cache_;
+    CacheConfig cache_config_;
 };
 
 /// `ReadContextBuilder` used to build a `ReadContext`, has input validation.
@@ -215,6 +227,20 @@ class PAIMON_EXPORT ReadContextBuilder {
     /// @return Reference to this builder for method chaining.
     ReadContextBuilder& EnablePrefetch(bool enabled);
 
+    /// Enable or disable prefetch cache for read operations.
+    ///
+    /// When enabled, a prefetch cache is used to prebuffer data ranges before they are needed,
+    /// which can improve read performance by reducing redundant I/O operations.
+    /// @param enabled Whether to enable prefetch cache (default: true)
+    /// @return Reference to this builder for method chaining.
+    ReadContextBuilder& EnablePrefetchCache(bool enabled);
+
+    /// Set the cache configuration for prefetch read operations.
+    ///
+    /// @param config The cache configuration to use.
+    /// @return Reference to this builder for method chaining.
+    ReadContextBuilder& WithCacheConfig(const CacheConfig& config);
+
     /// Set the total number of batches to prefetch across all files.
     ///
     /// This controls the memory usage and parallelism of the prefetching mechanism.

include/paimon/reader/prefetch_file_batch_reader.h

@@ -71,6 +71,26 @@ class PAIMON_EXPORT PrefetchFileBatchReader : public FileBatchReader {
     /// @param read_ranges A vector of pairs, where each pair defines a half-open interval
     /// `[start_row, end_row)`. The `start_row` is inclusive, and the `end_row` is exclusive.
     virtual Status SetReadRanges(const std::vector<std::pair<uint64_t, uint64_t>>& read_ranges) = 0;
+
+    /// Returns a list of file offset/length ranges that should be prefetched for the current read
+    /// scenario.
+    ///
+    /// This method should analyze the columns selected by the user and return the minimal set of
+    /// physical file ranges (offset, length) that need to be read, avoiding unnecessary IO
+    /// amplification. For example, if only a subset of columns is requested, the implementation
+    /// should only return the byte ranges corresponding to those columns, rather than the entire
+    /// row group or block.
+    ///
+    /// This enables the cache to prefetch only the required data, reducing disk and network load
+    /// and improving performance for columnar formats and selective queries.
+    ///
+    /// By default, returns an empty list (no prefetching). Format-specific implementations should
+    /// override this method to provide accurate offset/length hints for efficient IO.
+    /// @return A vector of pairs, where each pair contains the file offset and length to be
+    /// prefetched.
+    virtual Result<std::vector<std::pair<uint64_t, uint64_t>>> PreBufferRange() {
+        return std::vector<std::pair<uint64_t, uint64_t>>{};
+    }
 };
 
 }  // namespace paimon

src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp

@@ -35,6 +35,7 @@
 #include "paimon/testing/mock/mock_format_reader_builder.h"
 #include "paimon/testing/utils/read_result_collector.h"
 #include "paimon/testing/utils/testharness.h"
+#include "paimon/utils/read_ahead_cache.h"
 
 namespace arrow {
 class Array;
@@ -92,7 +93,8 @@ class ApplyBitmapIndexBatchReaderTest : public ::testing::Test,
                                          /*data_file_path=*/"DUMMY", &reader_builder, fs_,
                                          prefetch_batch_count, batch_size, prefetch_batch_count * 2,
                                          /*enable_adaptive_prefetch_strategy=*/false, executor_,
-                                         /*initialize_read_ranges=*/true));
+                                         /*initialize_read_ranges=*/true,
+                                         /*enable_prefetch_cache=*/true, CacheConfig(), pool_));
             } else {
                 file_batch_reader =
                     std::make_unique<MockFileBatchReader>(data, target_type_, batch_size);

src/paimon/common/io/cache_input_stream.h

@@ -0,0 +1,90 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstring>
+#include <memory>
+#include <string>
+
+#include "paimon/fs/file_system.h"
+#include "paimon/utils/read_ahead_cache.h"
+
+namespace paimon {
+
+class CacheInputStream : public InputStream {
+ public:
+    CacheInputStream(std::unique_ptr<InputStream> input_stream,
+                     const std::shared_ptr<ReadAheadCache>& cache)
+        : cache_(cache), input_stream_(std::move(input_stream)) {}
+
+    Status Seek(int64_t offset, SeekOrigin origin) override {
+        return input_stream_->Seek(offset, origin);
+    }
+    Result<int64_t> GetPos() const override {
+        return input_stream_->GetPos();
+    }
+    Result<int32_t> Read(char* buffer, uint32_t size) override {
+        return input_stream_->Read(buffer, size);
+    }
+    Result<int32_t> Read(char* buffer, uint32_t size, uint64_t offset) override {
+        if (cache_) {
+            ByteRange range{offset, static_cast<uint64_t>(size)};
+            PAIMON_ASSIGN_OR_RAISE(ByteSlice slice, cache_->Read(range));
+            if (slice.buffer) {
+                std::memcpy(buffer, slice.buffer->data() + slice.offset, slice.length);
+                return slice.length;
+            }
+        }
+        return input_stream_->Read(buffer, size, offset);
+    }
+    void ReadAsync(char* buffer, uint32_t size, uint64_t offset,
+                   std::function<void(Status)>&& callback) override {
+        if (cache_) {
+            ByteRange range{offset, static_cast<uint64_t>(size)};
+            Result<ByteSlice> slice = cache_->Read(range);
+            if (!slice.ok()) {
+                callback(slice.status());
+                return;
+            }
+            if (slice.value().buffer) {
+                std::memcpy(buffer, slice.value().buffer->data() + slice.value().offset,
+                            slice.value().length);
+                callback(Status::OK());
+                return;
+            }
+        }
+        return input_stream_->ReadAsync(buffer, size, offset, std::move(callback));
+    }
+
+    Status Close() override {
+        return input_stream_->Close();
+    }
+
+    Result<std::string> GetUri() const override {
+        return input_stream_->GetUri();
+    }
+
+    Result<uint64_t> Length() const override {
+        return input_stream_->Length();
+    }
+
+ private:
+    std::shared_ptr<ReadAheadCache> cache_;
+    std::unique_ptr<InputStream> input_stream_;
+};
+
+}  // namespace paimon