feat(read)!: decouple raw warmup and read-ahead caching from prefetch

docs/source/user_guide/metrics.rst

@@ -86,8 +86,9 @@ These metrics are C++-only and have no counterparts in Java Paimon.
 Prefetch I/O
 ------------
 
-``PrefetchIoMetrics`` describes only I/O that passes through the prefetch reader's instrumented
-input streams. It is not a whole-query or whole-table I/O total. All counters accumulate for the
+``PrefetchIoMetrics`` describes only I/O that passes through the instrumented
+file input streams, including synchronous reads with batch prefetch disabled.
+It is not a whole-query or whole-table I/O total. All counters accumulate for the
 reader lifetime and are retained across ``SetReadSchema()`` and cache reset. Latency uses relaxed
 atomic count and sum counters instead of per-I/O histograms to reduce hot-path cost. Collection is
 disabled by default; set ``prefetch.io-metrics.enabled`` to ``true`` in the read options to enable

docs/source/user_guide/prefetch.rst

@@ -83,33 +83,37 @@ the default ``loser-tree`` sort engine additionally warms the first file of
 every sorted run before merging. It applies to append-table reads and append
 compaction as well, but does not cross split boundaries.
 
-How far the next file is prepared is set with
+Whether the next file's raw bytes are fetched early is set with
 ``ReadContextBuilder::SetWarmupLevel()``. Append compaction builds its own read
 context and always uses the default level.
 
 - ``WarmupLevel::NONE``: do not warm up.
 - ``WarmupLevel::RAW`` (default): fetch the next file's raw bytes through the
   read-ahead cache without decoding them. It has no effect when the read-ahead
   cache is disabled with ``ReadContextBuilder::SetReadAheadCacheEnabled()``.
-- ``WarmupLevel::DECODED``: also start decoding the next file, which hides the
-  most latency but uses the most memory.
-
-Warmup goes through the prefetch reader, so it only applies to Parquet and ORC
-files that are actually read with prefetching, and enabling prefetch alone does
-not guarantee that. With the adaptive strategy, a file whose first read range
-holds more batches than one prefetch queue can buffer is read without
-prefetching, and so without warmup; the ``prefetch.adaptive-disabled-count``
-metric counts how often this happens. Append compaction uses a prefetch queue
-of three batches, so it only warms files whose first read range spans no more
-than three full batches, such as files with small row groups.
+
+Warmup and the read-ahead cache work independently of batch prefetch for readers
+implementing the ``PrefetchFileBatchReader`` byte-range interface, including
+Parquet and ORC. Support is determined by the reader's capabilities rather than
+the format name; other readers keep their plain read path. With
+``EnablePrefetch(false)``, ``RAW`` still warms the cache
+and batches are read synchronously. The same behavior applies when the format
+or the adaptive strategy decides against prefetching. Warmup only fetches raw
+bytes; it never decodes batches. When batch prefetch is active, the first
+actual read starts the parallel decode loop.
+
+The defaults are batch prefetch off, read-ahead cache on, and ``RAW`` warmup.
+``WarmupLevel::NONE`` disables early warmup while retaining byte caching during
+normal reads. Disabling the cache makes ``RAW`` a no-op. When the shared cache
+is active, Parquet disables its own pre-buffering to avoid buffering the same
+byte ranges twice.
 
 ``WarmupLevel::RAW`` issues its fetches from the reading thread through the
 file system's asynchronous read, so it only overlaps I/O when that read is
 really asynchronous. The local file system serves it synchronously, so there
 ``RAW`` warmup reads the next file on the reading thread before the current
 batch is returned, adding latency instead of hiding it. On such file systems,
-prefer ``WarmupLevel::DECODED``, which fetches on the background decode thread,
-or ``WarmupLevel::NONE``.
+use ``WarmupLevel::NONE`` to disable early fetching.
 
 Warmup preserves the returned rows and their order, but it costs memory: under
 ``WarmupLevel::RAW`` the next file may hold up to the read-ahead cache's
@@ -118,3 +122,20 @@ on top of the file being read. It may also fetch data from a file that the
 reader never consumes: a read that stops early, for example because of a LIMIT,
 may already have warmed the next file, and closing it waits for that file's
 in-flight fetches.
+
+For synchronous reads, including adaptive fallback, the pre-buffer limit also
+bounds retained range buffers. Completed buffers outside the current window can
+be evicted; later reads can fetch them again. In-flight I/O and buffers retained
+by callbacks count against that budget, so a full budget can temporarily cause a
+read to bypass range caching. The block cache has its separate limit. Parallel
+prefetch retains its existing range-buffer retention policy.
+
+ORC discovers byte ranges one stripe at a time and reports later stripes as
+reading advances. Preparing the first batch does not require warmup to parse
+every stripe footer.
+
+Late-materialization payloads and later ORC stripes prepare their ranges through
+the same internal cache context. Format readers expose their initial ranges
+through ``PreBufferRange()``; the public prefetch reader has no range callback
+setter. Plugins overriding the former callback setter must remove that override
+and rebuild against the updated interface.

include/paimon/defs.h

@@ -236,8 +236,8 @@ struct PAIMON_EXPORT Options {
     /// Applies to supported manifest read paths with or without caching. Default value is true.
     static const char SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED[];
 
-    /// "prefetch.io-metrics.enabled" - Whether to collect per-I/O metrics for prefetch reads.
-    /// Default value is false.
+    /// "prefetch.io-metrics.enabled" - Whether to collect per-I/O metrics for Parquet and ORC
+    /// reads, with or without batch prefetch. Default value is false.
     static const char PREFETCH_IO_METRICS_ENABLED[];
 
     /// "read.batch-size" - Read batch size for any file format if it supports.

include/paimon/format/read_hints.h

@@ -28,7 +28,7 @@ namespace paimon {
 struct PAIMON_EXPORT ReadHints {
     /// Whether framework-level prefetch is enabled for this read.
     bool prefetch_enabled = false;
-    /// Whether the shared read-ahead cache is enabled for this read.
+    /// Whether the shared read-ahead cache is active for this read, independently of prefetch.
     bool read_ahead_cache_enabled = false;
 };
 

include/paimon/read_context.h

@@ -414,26 +414,27 @@ class PAIMON_EXPORT ReadContextBuilder {
     ///
     /// A read-ahead cache is used to prebuffer data ranges before they are needed,
     /// which can improve read performance by reducing redundant I/O operations.
+    /// It also works without batch prefetch for readers exposing byte ranges through
+    /// PrefetchFileBatchReader, including Parquet and ORC.
     /// @param enabled Whether to enable the read-ahead cache (default: true)
     /// @return Reference to this builder for method chaining.
     ReadContextBuilder& SetReadAheadCacheEnabled(bool enabled);
 
-    /// Set the cache configuration for prefetch read operations.
+    /// Set the read-ahead cache configuration for synchronous and prefetch reads.
     ///
     /// @param config The cache configuration to use.
     /// @return Reference to this builder for method chaining.
     ReadContextBuilder& WithCacheConfig(const CacheConfig& config);
 
-    /// Set how far the reader prepares the next file before it is read.
+    /// Set whether the reader fetches the next file's raw bytes before it is read.
     ///
-    /// Warmup overlaps remote-storage latency with the read of the current file. Higher levels hide
-    /// more latency but use more memory, and may warm files that a query never reads (for example
-    /// when a LIMIT stops the scan early).
+    /// Warmup can overlap remote-storage latency with reading the current file. It may fetch bytes
+    /// for files that a query never reads, for example when a LIMIT stops the scan early.
     /// @param level The warmup level to use (default: WarmupLevel::RAW).
     /// @return Reference to this builder for method chaining.
     /// @note WarmupLevel::RAW warms the read-ahead cache, so it has no effect and behaves like
     /// WarmupLevel::NONE when the cache is off (see SetReadAheadCacheEnabled()).
-    /// WarmupLevel::DECODED does not depend on the cache.
+    /// RAW also works with EnablePrefetch(false) and never starts batch decoding.
     /// @see WarmupLevel
     ReadContextBuilder& SetWarmupLevel(WarmupLevel level);
 

include/paimon/reader/file_batch_reader.h

@@ -64,9 +64,8 @@ class PAIMON_EXPORT FileBatchReader : public BatchReader {
     /// Get whether or not support read precisely while bitmap pushed down.
     virtual bool SupportPreciseBitmapSelection() const = 0;
 
-    /// Starts whatever background work this reader would otherwise start on its first read, so a
-    /// caller that knows this reader is next can pay that startup while still consuming the
-    /// previous one.
+    /// Fetches this file's raw bytes ahead of its first read, so a caller that knows this reader
+    /// is next can overlap the fetch with consuming the previous file.
     ///
     /// This is an optional hint and never changes what the reader returns: ordering, filtering and
     /// metrics are the same with or without it. A reader with nothing to start, or one that is not
@@ -78,8 +77,8 @@ class PAIMON_EXPORT FileBatchReader : public BatchReader {
     /// that cannot start its work leaves it to be started by the first read, which reports the
     /// failure itself.
     ///
-    /// What an implementation starts is up to the implementation - a reader that prefetches on a
-    /// background thread starts that thread, one that reads synchronously may do nothing at all.
+    /// A reader may fetch through its read-ahead cache using the filesystem's asynchronous I/O.
+    /// Warmup does not start batch decoding. A reader without byte-warmup support does nothing.
     /// @warning Whatever it starts, the call itself is not thread-safe: make it from the same
     /// thread that reads this reader, and not concurrently with any other call on it.
     virtual void Warmup() {}

include/paimon/reader/prefetch_file_batch_reader.h

@@ -19,7 +19,6 @@
 
 #pragma once
 
-#include <functional>
 #include <memory>
 #include <utility>
 #include <vector>
@@ -48,7 +47,8 @@ class PAIMON_EXPORT PrefetchMetrics {
     static constexpr char CONSUMER_WAIT_LATENCY_US[] = "prefetch.consumer-wait-latency-us";
 };
 
-/// C++-only metric names for I/O observed by the prefetch reader's instrumented input streams.
+/// C++-only metric names for I/O observed by instrumented file input streams, with or without
+/// batch prefetch.
 /// Java Paimon has no corresponding metrics.
 /// These metrics do not represent whole-query or whole-table I/O.
 class PAIMON_EXPORT PrefetchIoMetrics {
@@ -136,20 +136,6 @@ class PAIMON_EXPORT PrefetchFileBatchReader : public FileBatchReader {
     virtual Result<std::vector<std::pair<uint64_t, uint64_t>>> PreBufferRange() {
         return std::vector<std::pair<uint64_t, uint64_t>>{};
     }
-
-    /// Callback a reader reports byte ranges through when they only become known after reading has
-    /// started, so the prefetch layer can register them with its shared read-ahead cache.
-    ///
-    /// PreBufferRange() covers what is known up front; a reader whose ranges depend on data it
-    /// has already read - the late-materialization payload pass only knows which pages hold the
-    /// matched rows once the probe pass has run - reports them through this callback instead.
-    using PreBufferRangeCallback =
-        std::function<void(std::vector<std::pair<uint64_t, uint64_t>>&&)>;
-
-    /// Installs the callback above, or clears it when `callback` is empty. By default a reader has
-    /// no late byte ranges to report and ignores the callback.
-    /// @param callback The callback to report late byte ranges through.
-    virtual void SetPreBufferRangeCallback(PreBufferRangeCallback callback) {}
 };
 
 }  // namespace paimon

include/paimon/utils/prefetch_cache_config.h

@@ -124,26 +124,18 @@ class PAIMON_EXPORT CacheConfig {
     uint64_t block_cache_limit_ = 1024 * 1024;
 };
 
-/// Controls how far a reader prepares the next file before that file is actually read.
+/// Controls whether a reader fetches the next file's raw bytes before that file is read.
 ///
-/// Warmup overlaps remote-storage latency with the read of the current file. Each level takes the
-/// next file one step further along the read pipeline: `RAW` covers the remote fetch, `DECODED`
-/// adds the decode on top of it. Higher levels hide more latency, but commit more memory and
-/// background I/O to files that a query may end up never reading (for example when a LIMIT stops
-/// the scan early). Callers can trade latency against memory by picking a level.
+/// Warmup can overlap remote-storage latency with reading the current file. It may fetch bytes
+/// for a file a query never consumes, for example when a LIMIT stops the scan early.
 enum class PAIMON_EXPORT WarmupLevel {
     /// Do not warm up. The next file's I/O starts only when it is actually read. This is the
     /// behavior from before warmup existed and uses no extra memory or background threads.
     NONE,
     /// Fetch only the next file's raw, still-compressed bytes into memory, and leave the decoder
-    /// alone. Overlaps the remote fetch while keeping memory lower than `DECODED`, because no
-    /// decoded batches are materialized ahead of the read. It fetches through the read-ahead
-    /// cache, so it falls back to `NONE` when that cache is disabled. This is the default, as it
-    /// hides the remote fetch without committing memory to decoded batches.
+    /// alone. No decoded batches are materialized ahead of the read. It uses the read-ahead cache,
+    /// so it falls back to `NONE` when that cache is disabled. This is the default level.
     RAW,
-    /// Fetch the raw bytes and start the background decode loop as well, so decoded batches are
-    /// ready before the file is read. Hides the most latency but uses the most memory.
-    DECODED,
 };
 
 }  // namespace paimon

src/paimon/CMakeLists.txt

@@ -154,6 +154,8 @@ set(PAIMON_COMMON_SRCS
     common/reader/data_file_reader_factory.cpp
     common/reader/predicate_batch_reader.cpp
     common/reader/prefetch_file_batch_reader_impl.cpp
+    common/reader/file_read_ahead.cpp
+    common/reader/warmup_file_batch_reader.cpp
     common/reader/reader_utils.cpp
     common/reader/blob_fallback_batch_reader.cpp
     common/reader/blob_view_resolving_batch_reader.cpp
@@ -711,6 +713,7 @@ if(PAIMON_BUILD_TESTS)
                     common/reader/late_materializing_file_batch_reader_test.cpp
                     common/reader/predicate_batch_reader_test.cpp
                     common/reader/prefetch_file_batch_reader_impl_test.cpp
+                    common/reader/warmup_file_batch_reader_test.cpp
                     common/reader/reader_utils_test.cpp
                     common/reader/complete_row_kind_batch_reader_test.cpp
                     common/reader/blob_fallback_batch_reader_test.cpp

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

@@ -128,8 +128,7 @@ class ApplyBitmapIndexBatchReaderTest : public ::testing::Test,
                         /*enable_adaptive_prefetch_strategy=*/false, executor_,
                         /*initialize_read_ranges=*/true,
                         /*read_ahead_cache_enabled=*/true, CacheConfig(),
-                        /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
-                        GetArrowPool(pool_)));
+                        /*enable_io_metrics=*/false, WarmupLevel::RAW, pool_, GetArrowPool(pool_)));
             } else {
                 file_batch_reader =
                     std::make_unique<MockFileBatchReader>(data, target_type_, batch_size);

src/paimon/common/io/cache_input_stream.h

@@ -18,6 +18,7 @@
 
 #pragma once
 
+#include <functional>
 #include <memory>
 #include <string>
 
@@ -27,11 +28,49 @@
 
 namespace paimon {
 
+class FileReadAhead;
+
 class CacheInputStream : public InputStream {
  public:
+    using CacheFactory = std::function<Result<std::shared_ptr<ReadAheadCache>>()>;
+
     CacheInputStream(std::unique_ptr<InputStream> input_stream,
                      const std::shared_ptr<ReadAheadCache>& cache)
-        : cache_(cache), input_stream_(std::move(input_stream)) {}
+        : CacheInputStream(std::move(input_stream), cache, CacheFactory(), nullptr) {}
+
+    CacheInputStream(std::unique_ptr<InputStream> input_stream,
+                     const std::shared_ptr<ReadAheadCache>& cache, CacheFactory cache_factory)
+        : CacheInputStream(std::move(input_stream), cache, std::move(cache_factory), nullptr) {}
+
+    CacheInputStream(std::unique_ptr<InputStream> input_stream,
+                     const std::shared_ptr<ReadAheadCache>& cache, CacheFactory cache_factory,
+                     const std::shared_ptr<FileReadAhead>& read_ahead)
+        : cache_(cache),
+          input_stream_(std::move(input_stream)),
+          cache_factory_(std::move(cache_factory)),
+          read_ahead_(read_ahead) {}
+
+    const std::shared_ptr<FileReadAhead>& GetReadAhead() const {
+        return read_ahead_;
+    }
+
+    // Configure before the reader starts consuming data. Capable builders may enable the cache
+    // early to coordinate their own buffering; otherwise the factory enables it after Build().
+    Result<bool> EnableCache() {
+        if (cache_) {
+            return true;
+        }
+        if (!cache_factory_) {
+            return false;
+        }
+        PAIMON_ASSIGN_OR_RAISE(cache_, cache_factory_());
+        cache_factory_ = nullptr;
+        return cache_ != nullptr;
+    }
+
+    void ClearCacheFactory() {
+        cache_factory_ = nullptr;
+    }
 
     Status Seek(int64_t offset, SeekOrigin origin) override {
         return input_stream_->Seek(offset, origin);
@@ -96,6 +135,8 @@ class CacheInputStream : public InputStream {
  private:
     std::shared_ptr<ReadAheadCache> cache_;
     std::unique_ptr<InputStream> input_stream_;
+    CacheFactory cache_factory_;
+    std::shared_ptr<FileReadAhead> read_ahead_;
 };
 
 }  // namespace paimon

src/paimon/common/io/cache_input_stream_test.cpp

@@ -26,6 +26,7 @@
 
 #include "gtest/gtest.h"
 #include "paimon/common/factories/io_hook.h"
+#include "paimon/common/metrics/metrics_impl.h"
 #include "paimon/common/utils/read_ahead_cache.h"
 #include "paimon/common/utils/scope_guard.h"
 #include "paimon/fs/file_system.h"
@@ -82,6 +83,40 @@ class CacheInputStreamTest : public ::testing::Test {
     std::string file_path_;
 };
 
+TEST_F(CacheInputStreamTest, EnableCacheLazilyOnce) {
+    uint32_t factory_calls = 0;
+    auto cache = CreateCache({{2, 5}});
+    CacheInputStream stream(OpenFile(), nullptr, [&]() -> Result<std::shared_ptr<ReadAheadCache>> {
+        ++factory_calls;
+        return cache;
+    });
+    std::string buffer(5, '\0');
+    ASSERT_OK(stream.Read(buffer.data(), 5, 2));
+    ASSERT_EQ(factory_calls, 0);
+    ASSERT_EQ(buffer, "cdefg");
+    ASSERT_OK_AND_ASSIGN(bool enabled, stream.EnableCache());
+    ASSERT_TRUE(enabled);
+    ASSERT_OK_AND_ASSIGN(bool enabled_again, stream.EnableCache());
+    ASSERT_TRUE(enabled_again);
+    ASSERT_EQ(factory_calls, 1);
+    ASSERT_OK(stream.Read(buffer.data(), 5, 2));
+    ASSERT_EQ(buffer, "cdefg");
+    std::shared_ptr<Metrics> metrics = std::make_shared<MetricsImpl>();
+    cache->CollectMetrics(&metrics);
+    ASSERT_OK_AND_ASSIGN(uint64_t io_count, metrics->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
+    ASSERT_EQ(io_count, 1);
+}
+
+TEST_F(CacheInputStreamTest, DisableCacheFactoryAndPropagateFailure) {
+    CacheInputStream stream(OpenFile(), nullptr, []() -> Result<std::shared_ptr<ReadAheadCache>> {
+        return Status::IOError("cache initialization failed");
+    });
+    ASSERT_NOK_WITH_MSG(stream.EnableCache(), "cache initialization failed");
+    stream.ClearCacheFactory();
+    ASSERT_OK_AND_ASSIGN(bool enabled, stream.EnableCache());
+    ASSERT_FALSE(enabled);
+}
+
 // Test proxy methods: Seek, GetPos, Read (sequential), Close, GetUri, Length
 TEST_F(CacheInputStreamTest, TestProxyMethods) {
     auto underlying = OpenFile();