refactor(read)!: remove decoded warmup mode

BREAKING CHANGE: WarmupLevel::DECODED is removed. Use RAW for byte warmup or NONE to disable it.

Author: lucasfang

docs/source/user_guide/prefetch.rst

@@ -83,37 +83,34 @@ 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 and the read-ahead cache work independently of batch prefetch for
-Parquet and ORC. With ``EnablePrefetch(false)``, ``RAW`` still warms the cache,
-and ``DECODED`` prepares only the first batch on a background thread; later
-batches are read synchronously. The same behavior applies when the format or
-the adaptive strategy decides against prefetching. With batch prefetch active,
-``DECODED`` starts the normal parallel decode loop instead.
+Parquet and ORC. 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 but still permits
-``DECODED``. When the shared cache is active, Parquet disables its own
-pre-buffering to avoid buffering the same byte ranges twice.
+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

include/paimon/read_context.h

@@ -425,18 +425,15 @@ class PAIMON_EXPORT ReadContextBuilder {
     /// @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.
-    /// Both levels work with EnablePrefetch(false). Without prefetch, DECODED prepares only the
-    /// first batch; subsequent batches are read synchronously.
+    /// 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/utils/prefetch_cache_config.h

@@ -124,27 +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.
-    /// Without batch prefetch, prepares only the first batch and then continues synchronously.
-    DECODED,
 };
 
 }  // namespace paimon

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/reader/data_file_reader_factory.cpp

@@ -48,7 +48,7 @@ bool FormatSupportsPrefetch(const std::string& format_identifier) {
            format_identifier != "vortex";
 }
 
-bool FormatSupportsSynchronousWarmup(const std::string& format_identifier) {
+bool FormatSupportsReadAhead(const std::string& format_identifier) {
     // Other format plugins may expose only FileBatchReader. Keep their plain read path intact.
     return format_identifier == "parquet" || format_identifier == "orc";
 }
@@ -79,7 +79,7 @@ Result<std::unique_ptr<ReaderBuilder>> DataFileReaderFactory::CreateReaderBuilde
     read_hints.read_ahead_cache_enabled =
         read_options.read_ahead_cache_enabled &&
         ((read_options.prefetch_enabled && FormatSupportsPrefetch(format_identifier)) ||
-         FormatSupportsSynchronousWarmup(format_identifier));
+         FormatSupportsReadAhead(format_identifier));
     reader_builder->WithReadHints(read_hints);
     return reader_builder;
 }
@@ -102,7 +102,7 @@ Result<std::unique_ptr<FileBatchReader>> DataFileReaderFactory::Open(
                 read_options.warmup_level, pool, arrow_pool));
         return std::make_unique<DelegatingPrefetchReader>(std::move(prefetch_reader));
     }
-    if (FormatSupportsSynchronousWarmup(format_identifier)) {
+    if (FormatSupportsReadAhead(format_identifier)) {
         PAIMON_ASSIGN_OR_RAISE(
             FileReadAhead::OpenedReaders opened,
             FileReadAhead::Open(file_path, file_size, reader_builder, file_system,

src/paimon/common/reader/data_file_reader_factory_test.cpp

@@ -68,7 +68,7 @@ TEST(DataFileReaderFactoryTest, OpenForwardsWarmupLevelToPrefetchReader) {
             data_type->fields())
             .ValueOrDie();
 
-    for (WarmupLevel level : {WarmupLevel::NONE, WarmupLevel::RAW, WarmupLevel::DECODED}) {
+    for (WarmupLevel level : {WarmupLevel::NONE, WarmupLevel::RAW}) {
         MockFormatReaderBuilder reader_builder(data, data_type, /*read_batch_size=*/10);
         DataFileReadOptions read_options;
         read_options.read_batch_size = 10;

src/paimon/common/reader/late_materializing_file_batch_reader_test.cpp

@@ -706,7 +706,7 @@ TEST_F(LateMaterializingFileBatchReaderTest, WorksAsInnerOfPrefetchReader) {
             /*prefetch_max_parallel_num=*/1, /*batch_size=*/3, /*prefetch_batch_count=*/2,
             /*enable_adaptive_prefetch_strategy=*/false, executor,
             /*initialize_read_ranges=*/false, /*read_ahead_cache_enabled=*/false, CacheConfig(),
-            /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_, GetArrowPool(pool_)));
+            /*enable_io_metrics=*/false, WarmupLevel::RAW, pool_, GetArrowPool(pool_)));
     auto predicate =
         PredicateBuilder::GreaterOrEqual(/*field_index=*/0, "k", FieldType::BIGINT, Literal(4l));
     ::ArrowSchema c_schema;
@@ -742,7 +742,7 @@ TEST_F(LateMaterializingFileBatchReaderTest, PrefetchInnerReentrantSetReadSchema
             /*prefetch_max_parallel_num=*/1, /*batch_size=*/3, /*prefetch_batch_count=*/2,
             /*enable_adaptive_prefetch_strategy=*/false, executor,
             /*initialize_read_ranges=*/false, /*read_ahead_cache_enabled=*/false, CacheConfig(),
-            /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_, GetArrowPool(pool_)));
+            /*enable_io_metrics=*/false, WarmupLevel::RAW, pool_, GetArrowPool(pool_)));
 
     auto full_schema = arrow::schema(full_fields_);
     auto predicate1 =
@@ -805,7 +805,7 @@ TEST_F(LateMaterializingFileBatchReaderTest, PrefetchInnerParallelReadersWithSee
             /*prefetch_max_parallel_num=*/3, /*batch_size=*/3, /*prefetch_batch_count=*/6,
             /*enable_adaptive_prefetch_strategy=*/false, executor,
             /*initialize_read_ranges=*/false, /*read_ahead_cache_enabled=*/false, CacheConfig(),
-            /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_, GetArrowPool(pool_)));
+            /*enable_io_metrics=*/false, WarmupLevel::RAW, pool_, GetArrowPool(pool_)));
     auto predicate =
         PredicateBuilder::GreaterOrEqual(/*field_index=*/0, "k", FieldType::BIGINT, Literal(5l));
     ::ArrowSchema c_schema;
@@ -960,7 +960,7 @@ TEST_F(LateMaterializingFileBatchReaderTest, PrefetchInnerRegistersPayloadPreBuf
             /*prefetch_max_parallel_num=*/1, /*batch_size=*/3, /*prefetch_batch_count=*/2,
             /*enable_adaptive_prefetch_strategy=*/false, executor,
             /*initialize_read_ranges=*/false, /*read_ahead_cache_enabled=*/true, CacheConfig(),
-            /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_, GetArrowPool(pool_)));
+            /*enable_io_metrics=*/false, WarmupLevel::RAW, pool_, GetArrowPool(pool_)));
     auto predicate =
         PredicateBuilder::GreaterOrEqual(/*field_index=*/0, "k", FieldType::BIGINT, Literal(4l));
     ::ArrowSchema c_schema;

src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp

@@ -284,7 +284,7 @@ PrefetchFileBatchReaderImpl::DispatchReadRanges(
 }
 
 Status PrefetchFileBatchReaderImpl::CleanUp() {
-    synchronous_reader_->ResetWarmup();
+    synchronous_reader_->ResetReadState();
     auto clean_prefetch_queue = [this]() {
         for (auto& prefetch_queue : prefetch_queues_) {
             while (true) {
@@ -336,9 +336,8 @@ Status PrefetchFileBatchReaderImpl::CleanUp() {
 void PrefetchFileBatchReaderImpl::Workloop() {
     std::vector<std::future<void>> futures;
     futures.resize(readers_.size());
-    // Warm the read-ahead cache before decoding. At DECODED this is the first warmup; at RAW
-    // Warmup() already ran it on the caller's thread and the guard makes this a no-op, so the
-    // cache is never Init'd twice.
+    // Register the file's byte ranges before decoding. An earlier RAW Warmup() may have already
+    // fetched them, in which case the shared state makes this a no-op.
     WarmCacheOnce();
 
     while (true) {
@@ -587,7 +586,7 @@ void PrefetchFileBatchReaderImpl::ClearPreBufferRangeCallbacks() {
 }
 
 void PrefetchFileBatchReaderImpl::Warmup() {
-    if (warmup_level_ == WarmupLevel::NONE) {
+    if (warmup_level_ != WarmupLevel::RAW) {
         return;
     }
     // Ranges that are not set mean the reader is either not configured yet or already cleaned up,
@@ -597,18 +596,7 @@ void PrefetchFileBatchReaderImpl::Warmup() {
     if (!read_ranges_freshed_) {
         return;
     }
-    if (!need_prefetch_) {
-        synchronous_reader_->Warmup();
-        return;
-    }
-    if (warmup_level_ == WarmupLevel::RAW) {
-        // Prefetch the raw bytes only, without starting the decoder. The first real read starts the
-        // background thread, whose Workloop() finds the cache already warmed and skips re-Init.
-        WarmCacheOnce();
-        return;
-    }
-    // DECODED: start the background decode loop now so decoded batches are ready before the read.
-    EnsureBackgroundThread();
+    WarmCacheOnce();
 }
 
 Result<BatchReader::ReadBatchWithBitmap> PrefetchFileBatchReaderImpl::NextBatchWithBitmap() {

src/paimon/common/reader/prefetch_file_batch_reader_impl.h

@@ -132,8 +132,7 @@ class PrefetchFileBatchReaderImpl : public PrefetchFileBatchReader {
 
     Status CleanUp();
     void Workloop();
-    /// Starts the background prefetch thread if it is not running yet. The first read does this
-    /// itself; Warmup() is the same call made earlier, so the two must not diverge.
+    /// Starts the background batch prefetch thread lazily on the first read.
     void EnsureBackgroundThread();
     /// Shares byte-range registration with the synchronous warmup path.
     void WarmCacheOnce();