[feature](be) Add segment range read-ahead coordination

### What problem does this PR solve?

Issue Number: #67611 (tracking)

Related PR: #67292, #67579, #67612

Problem Summary: Coordinate all resolvable column plans, probe Page Cache, plan and submit ranges, and register exact page slices. Reuse a physical page's existing buffer when independent column windows overlap it. Keep synchronous fallback and per-page ownership accounting; expose planner and consumption statistics.

Register only ranges admitted by the scheduler, preserving the planner's original range indices and buffer offsets. Complete unadmitted pages immediately so PageIO uses the original reader, and publish consumed-range callbacks only for admitted buffers. The window, coalescing, block-fill and hole-fill policies are unchanged.

The full read-ahead feature adds six translation-unit users of the RLE header: its segment/row-id helpers, query/segment tests, and cached-reader coverage. Preserve master 8d88f1cc6f7's updated baseline and add only those six users, from 828 to 834 (limit 918), while keeping the dependency rules and 10% slack unchanged. The complete rebased feature measures 916 users. This avoids mixing a global expression/reader header refactor into the feature.

### Release note

Internal segment-level coordination for concurrent data-page reads.

### Check List (For Author)

- Test: Unit Test — integrated ASAN/PCH validation covers 141 tests from 8 suites through ./run-be-ut.sh --run -j100. Includes partial budget admission, original page offsets, cancellation, allocation rollback, executor rejection, consumed-range callbacks and row-store switch fallback. Individual intermediate commits are not built separately; target-branch validation is recorded on the PR. Clang-format 16, header hygiene and diff checks pass. No cluster or clang-tidy run.
- Behavior changed: Yes — introduces the opt-in component; this synchronization preserves existing defaults and IO/writeback semantics.
- Does this need documentation: Yes — component scope and current behavior are tracked in #67611 and the related PR descriptions.

be/src/storage/olap_common.h

@@ -51,6 +51,10 @@
 #include "util/uid_util.h"
 
 namespace doris {
+namespace io {
+struct ReadAheadStatistics;
+}
+
 using SchemaHash = int32_t;
 
 using TabletUid = UniqueId;
@@ -398,6 +402,8 @@ struct OlapReaderStatistics {
     int64_t total_segment_number = 0;
 
     io::FileCacheStatistics file_cache_stats;
+    // Shared with range tasks so asynchronous completion never writes into this reader's lifetime.
+    std::shared_ptr<io::ReadAheadStatistics> read_ahead_stats;
     int64_t load_segments_timer = 0;
 
     int64_t collect_iterator_merge_next_timer = 0;

be/src/storage/segment/page_io.cpp

@@ -36,11 +36,13 @@
 #include "io/cache/cached_remote_file_reader.h"
 #include "io/fs/file_reader.h"
 #include "io/fs/file_writer.h"
+#include "io/fs/read_ahead_metrics.h"
 #include "runtime/runtime_profile.h"
 #include "storage/cache/page_cache.h"
 #include "storage/olap_common.h"
 #include "storage/segment/encoding_info.h"
 #include "storage/segment/page_handle.h"
+#include "storage/segment/segment_read_ahead.h"
 #include "util/block_compression.h"
 #include "util/coding.h"
 #include "util/concurrency_stats.h"
@@ -170,8 +172,31 @@ Status PageIO::read_and_decompress_page_(const PageReadOptions& opts, PageHandle
     {
         SCOPED_RAW_TIMER(&opts.stats->io_ns);
         size_t bytes_read = 0;
-        RETURN_IF_ERROR(opts.file_reader->read_at(opts.page_pointer.offset, page_slice, &bytes_read,
-                                                  &opts.io_ctx));
+        auto read_page = [&]() -> Status {
+            auto* reader = opts.file_reader;
+            if (opts.type == DATA_PAGE && opts.stats->read_ahead_stats != nullptr) {
+                if (auto* read_ahead = dynamic_cast<SegmentReadAheadFileReader*>(reader)) {
+                    Status status;
+                    if (read_ahead->try_read_page(opts.page_pointer.offset, page_slice, &status) &&
+                        status.ok()) {
+                        bytes_read = page_size;
+                        return Status::OK();
+                    }
+                    reader = read_ahead->inner_reader().get();
+                }
+                // Count only data-page source reads in a read-ahead-enabled reader. This also
+                // covers admission rejection and retries using the original reader after decode
+                // failure; index/dictionary reads and Page Cache hits take their usual paths.
+                auto* statistics = opts.stats->read_ahead_stats.get();
+                COUNTER_UPDATE(&statistics->fallback_pages, 1);
+                COUNTER_UPDATE(&statistics->fallback_bytes, page_size);
+                SCOPED_TIMER(&statistics->fallback_time);
+                return reader->read_at(opts.page_pointer.offset, page_slice, &bytes_read,
+                                       &opts.io_ctx);
+            }
+            return reader->read_at(opts.page_pointer.offset, page_slice, &bytes_read, &opts.io_ctx);
+        };
+        RETURN_IF_ERROR(read_page());
         DCHECK_EQ(bytes_read, page_size);
         opts.stats->compressed_bytes_read += page_size;
     }

be/src/storage/segment/segment_read_ahead.cpp

@@ -0,0 +1,322 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#include "storage/segment/segment_read_ahead.h"
+
+#include <butil/iobuf.h>
+
+#include <algorithm>
+#include <cstring>
+#include <utility>
+
+#include "common/logging.h"
+#include "io/fs/read_ahead_metrics.h"
+#include "storage/cache/page_cache.h"
+
+namespace doris::segment_v2 {
+namespace {
+
+struct PendingPage {
+    ColumnReadAhead* column {nullptr};
+    ColumnReadAheadPage page;
+};
+
+} // namespace
+
+SegmentReadAheadFileReader::SegmentReadAheadFileReader(
+        io::FileReaderSPtr inner, std::shared_ptr<io::ReadAheadStatistics> statistics)
+        : _inner(std::move(inner)), _statistics(std::move(statistics)) {
+    DORIS_CHECK(_inner != nullptr);
+}
+
+void SegmentReadAheadFileReader::_register_page(ColumnReadAhead* column,
+                                                const ColumnReadAheadPage& page,
+                                                std::shared_ptr<BufferedRange> range,
+                                                size_t buffer_offset) {
+    DORIS_CHECK(column != nullptr);
+    DORIS_CHECK(range != nullptr);
+    std::lock_guard lock(_mutex);
+    const PageKey key {.offset = page.range.offset, .size = page.range.size};
+    auto [slot, inserted] = _pages.try_emplace(key);
+    if (inserted) {
+        slot->second.range = std::move(range);
+        slot->second.buffer_offset = buffer_offset;
+    } else {
+        DORIS_CHECK(slot->second.range == range);
+        DORIS_CHECK(slot->second.buffer_offset == buffer_offset);
+    }
+    const PageOwner owner {.column = column, .page_index = page.page_index};
+    DORIS_CHECK(std::ranges::find(slot->second.owners, owner) == slot->second.owners.end());
+    slot->second.owners.push_back(owner);
+}
+
+bool SegmentReadAheadFileReader::_reuse_page(ColumnReadAhead* column,
+                                             const ColumnReadAheadPage& page) {
+    std::lock_guard lock(_mutex);
+    auto slot = _pages.find({.offset = page.range.offset, .size = page.range.size});
+    if (slot == _pages.end()) {
+        return false;
+    }
+    const PageOwner owner {.column = column, .page_index = page.page_index};
+    DCHECK(std::ranges::find(slot->second.owners, owner) == slot->second.owners.end());
+    slot->second.owners.push_back(owner);
+    return true;
+}
+
+void SegmentReadAheadFileReader::_release_page(ColumnReadAhead* column,
+                                               const ColumnReadAheadPage& page) {
+    DORIS_CHECK(column != nullptr);
+    std::lock_guard lock(_mutex);
+    const PageKey key {.offset = page.range.offset, .size = page.range.size};
+    auto slot = _pages.find(key);
+    if (slot == _pages.end()) {
+        return;
+    }
+    const PageOwner owner {.column = column, .page_index = page.page_index};
+    auto entry = std::ranges::find(slot->second.owners, owner);
+    if (entry == slot->second.owners.end()) {
+        return;
+    }
+    slot->second.owners.erase(entry);
+    if (slot->second.owners.empty()) {
+        _pages.erase(slot);
+    }
+}
+
+bool SegmentReadAheadFileReader::try_read_page(size_t offset, Slice output, Status* status) {
+    const PageKey key {.offset = offset, .size = output.size};
+    std::shared_ptr<BufferedRange> range;
+    size_t buffer_offset = 0;
+    {
+        std::lock_guard lock(_mutex);
+        const auto slot = _pages.find(key);
+        if (slot == _pages.end()) {
+            return false;
+        }
+        range = slot->second.range;
+        buffer_offset = slot->second.buffer_offset;
+    }
+
+    bool ready = false;
+    {
+        SCOPED_TIMER(&_statistics->wait_time);
+        *status = range->read->wait(&ready);
+    }
+    if (!status->ok()) {
+        _finish_page(key, false);
+        return true;
+    }
+    DORIS_CHECK(output.size == key.size);
+    const Slice source = range->read->slice(buffer_offset, key.size);
+    std::memcpy(output.data, source.data, source.size);
+    auto& hits = ready ? _statistics->ready_page_hits : _statistics->wait_page_hits;
+    COUNTER_UPDATE(&hits, 1);
+    COUNTER_UPDATE(&_statistics->consumed_page_bytes, static_cast<int64_t>(source.size));
+    io::read_ahead_bvars().consumed_page_bytes << static_cast<int64_t>(source.size);
+    _finish_page(key, true);
+    return true;
+}
+
+void SegmentReadAheadFileReader::_finish_page(const PageKey& key, bool consumed) {
+    std::vector<PageOwner> owners;
+    std::shared_ptr<BufferedRange> consumed_range;
+    {
+        std::lock_guard lock(_mutex);
+        auto slot = _pages.find(key);
+        if (slot == _pages.end()) {
+            return;
+        }
+        owners = std::move(slot->second.owners);
+        auto range = std::move(slot->second.range);
+        _pages.erase(slot);
+        if (consumed && !range->consumed) {
+            range->consumed = true;
+            COUNTER_UPDATE(&_statistics->consumed_range_bytes,
+                           static_cast<int64_t>(range->read->range().size));
+            consumed_range = std::move(range);
+        }
+    }
+
+    for (const auto& owner : owners) {
+        owner.column->complete(owner.page_index);
+    }
+    if (consumed_range != nullptr && consumed_range->consumer) {
+        auto consumer = std::move(consumed_range->consumer);
+        consumer(consumed_range->read->range(), consumed_range->read->data());
+    }
+}
+
+Status SegmentReadAheadFileReader::read_at_impl(size_t offset, Slice result, size_t* bytes_read,
+                                                const io::IOContext* io_ctx) {
+    Status read_ahead_status;
+    if (try_read_page(offset, result, &read_ahead_status)) {
+        if (read_ahead_status.ok()) {
+            *bytes_read = result.size;
+            return Status::OK();
+        }
+    }
+    return _inner->read_at(offset, result, bytes_read, io_ctx);
+}
+
+Status SegmentReadAheadFileReader::read_at_iobuf_impl(size_t offset, size_t bytes_req,
+                                                      butil::IOBuf* out, size_t* bytes_read,
+                                                      const io::IOContext* io_ctx) {
+    return _inner->read_at_iobuf(offset, bytes_req, out, bytes_read, io_ctx);
+}
+
+SegmentReadAhead::SegmentReadAhead(io::FileReaderSPtr source_reader,
+                                   io::FileRangeReadScheduler* scheduler,
+                                   std::shared_ptr<io::FileRangeReadContext> context,
+                                   io::FileRangeReadIOContext io_context,
+                                   SegmentReadAheadOptions options)
+        : _source_reader(std::move(source_reader)),
+          _scheduler(scheduler),
+          _context(std::move(context)),
+          _io_context(std::move(io_context)),
+          _options(std::move(options)),
+          _statistics(_io_context.statistics != nullptr
+                              ? _io_context.statistics
+                              : std::make_shared<io::ReadAheadStatistics>()),
+          _reader(std::shared_ptr<SegmentReadAheadFileReader>(
+                  new SegmentReadAheadFileReader(_source_reader, _statistics))) {
+    DORIS_CHECK(_source_reader != nullptr);
+    DORIS_CHECK(_scheduler != nullptr);
+    DORIS_CHECK(_context != nullptr);
+    _io_context.statistics = _statistics;
+}
+
+SegmentReadAheadResult SegmentReadAhead::apply_plans(std::vector<ColumnReadAheadPlan> plans) {
+    COUNTER_UPDATE(&_statistics->plan_calls, 1);
+    SCOPED_TIMER(&_statistics->plan_time);
+    SegmentReadAheadResult result;
+    std::vector<PendingPage> misses;
+
+    for (auto& plan : plans) {
+        DORIS_CHECK(plan.column != nullptr);
+        for (const auto& page : plan.released_pages) {
+            _reader->_release_page(plan.column, page);
+        }
+        COUNTER_UPDATE(&_statistics->released_pages,
+                       static_cast<int64_t>(plan.released_pages.size()));
+        COUNTER_UPDATE(&_statistics->candidate_pages, static_cast<int64_t>(plan.new_pages.size()));
+        for (const auto& page : plan.new_pages) {
+            ++result.new_pages;
+            COUNTER_UPDATE(&_statistics->candidate_bytes, static_cast<int64_t>(page.range.size));
+            if (_options.page_cache_probe && _options.page_cache_probe(page.range)) {
+                plan.column->complete(page.page_index);
+                ++result.page_cache_hits;
+                COUNTER_UPDATE(&_statistics->page_cache_hits, 1);
+                COUNTER_UPDATE(&_statistics->page_cache_hit_bytes,
+                               static_cast<int64_t>(page.range.size));
+            } else if (!_reader->_reuse_page(plan.column, page)) {
+                misses.push_back({.column = plan.column, .page = page});
+            }
+        }
+    }
+
+    if (misses.empty()) {
+        return result;
+    }
+
+    std::ranges::sort(misses, [](const PendingPage& left, const PendingPage& right) {
+        return left.page.range.offset < right.page.range.offset ||
+               (left.page.range.offset == right.page.range.offset &&
+                left.page.range.size < right.page.range.size);
+    });
+    std::vector<io::FileRange> input_ranges;
+    input_ranges.reserve(misses.size());
+    for (const auto& miss : misses) {
+        if (input_ranges.empty() || input_ranges.back() != miss.page.range) {
+            input_ranges.push_back(miss.page.range);
+        }
+    }
+
+    io::FileRangePlan range_plan;
+    result.status = io::FileRangePlanner::plan(input_ranges, _source_reader->size(),
+                                               _options.range_plan, &range_plan, _statistics.get());
+    if (!result.status.ok()) {
+        for (const auto& miss : misses) {
+            miss.column->complete(miss.page.page_index);
+        }
+        return result;
+    }
+
+    ReadAheadRangeConsumer consumer;
+    if (_options.range_consumer_factory) {
+        consumer = _options.range_consumer_factory();
+    }
+    auto submit_result =
+            _scheduler->try_submit(range_plan.ranges, _source_reader, _io_context, _context);
+    result.reject_reason = submit_result.reject_reason;
+    result.status = submit_result.status;
+    if (!submit_result.accepted()) {
+        for (const auto& miss : misses) {
+            miss.column->complete(miss.page.page_index);
+        }
+        return result;
+    }
+
+    DORIS_CHECK(submit_result.reads.size() == range_plan.ranges.size());
+    std::vector<std::shared_ptr<SegmentReadAheadFileReader::BufferedRange>> ranges(
+            range_plan.ranges.size());
+    for (size_t index = 0; index < range_plan.ranges.size(); ++index) {
+        // Byte-budget admission may skip individual ranges while preserving their input slots.
+        if (submit_result.reads[index] == nullptr) {
+            continue;
+        }
+        ++result.submitted_ranges;
+        result.submitted_bytes += range_plan.ranges[index].size;
+        ranges[index] = std::make_shared<SegmentReadAheadFileReader::BufferedRange>(
+                SegmentReadAheadFileReader::BufferedRange {
+                        .read = std::move(submit_result.reads[index]), .consumer = consumer});
+    }
+
+    DORIS_CHECK(range_plan.input_locations.size() == input_ranges.size());
+    size_t input_index = 0;
+    for (size_t index = 0; index < misses.size(); ++index) {
+        if (index > 0 && misses[index - 1].page.range != misses[index].page.range) {
+            ++input_index;
+        }
+        DORIS_CHECK(input_index < range_plan.input_locations.size());
+        const auto& location = range_plan.input_locations[input_index];
+        DORIS_CHECK(location.range_index < ranges.size());
+        // No buffer was admitted for this page. Clear pending now; PageIO uses the original reader.
+        if (ranges[location.range_index] == nullptr) {
+            misses[index].column->complete(misses[index].page.page_index);
+            continue;
+        }
+        _reader->_register_page(misses[index].column, misses[index].page,
+                                ranges[location.range_index], location.buffer_offset);
+    }
+    return result;
+}
+
+ReadAheadPageCacheProbe make_storage_page_cache_probe(const io::FileReaderSPtr& file_reader) {
+    DORIS_CHECK(file_reader != nullptr);
+    return [file_reader](const io::FileRange& page_range) {
+        auto* cache = StoragePageCache::instance();
+        if (cache == nullptr) {
+            return false;
+        }
+        PageCacheHandle handle;
+        StoragePageCache::CacheKey key(file_reader->path().native(), file_reader->size(),
+                                       cast_set<int64_t>(page_range.offset));
+        return cache->lookup(key, &handle, DATA_PAGE);
+    };
+}
+
+} // namespace doris::segment_v2

be/src/storage/segment/segment_read_ahead.h

@@ -0,0 +1,176 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 <cstddef>
+#include <functional>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <vector>
+
+#include "io/cache/file_range_planner.h"
+#include "io/fs/file_range_read_scheduler.h"
+#include "io/fs/file_reader.h"
+#include "storage/segment/column_read_ahead.h"
+
+namespace doris::segment_v2 {
+
+/// Invoked with the whole physical range at most once, when any registered page successfully
+/// consumes bytes from it.
+using ReadAheadRangeConsumer = std::function<void(const io::FileRange&, Slice)>;
+/// Creates submission-scoped state immediately before asynchronous range tasks are admitted.
+using ReadAheadRangeConsumerFactory = std::function<ReadAheadRangeConsumer()>;
+/// Returns true when the exact compressed data-page range is already in StoragePageCache.
+using ReadAheadPageCacheProbe = std::function<bool(const io::FileRange& page_range)>;
+
+struct SegmentReadAheadOptions {
+    io::FileRangePlanOptions range_plan;
+    ReadAheadPageCacheProbe page_cache_probe;
+    /// Capture per-submission writeback state immediately before asynchronous IO starts. The
+    /// returned consumer is retained only by ranges accepted in that submission.
+    ReadAheadRangeConsumerFactory range_consumer_factory;
+};
+
+struct SegmentReadAheadResult {
+    /// Data pages newly requested by all column windows before the page-cache probe.
+    size_t new_pages {0};
+    /// Newly requested pages already available from StoragePageCache.
+    size_t page_cache_hits {0};
+    /// Physical ranges admitted to buffering; byte-budget skips are excluded.
+    size_t submitted_ranges {0};
+    /// Sum of admitted range sizes, including coalescing and block-fill bytes.
+    size_t submitted_bytes {0};
+    /// Batch-level scheduler rejection; NONE also covers plans that required no source reads.
+    io::FileRangeReadRejectReason reject_reason {io::FileRangeReadRejectReason::NONE};
+    /// Planning or batch-submission status. Per-range execution failures surface on page reads.
+    Status status;
+
+    bool accepted() const {
+        return reject_reason == io::FileRangeReadRejectReason::NONE && status.ok();
+    }
+};
+
+/// FileReader view used by PageIO. Exact data-page reads reuse an asynchronous range buffer;
+/// unrelated reads and failed prefetches transparently use the original reader.
+class SegmentReadAheadFileReader final : public io::FileReader {
+public:
+    Status close() override { return _inner->close(); }
+    const io::Path& path() const override { return _inner->path(); }
+    size_t size() const override { return _inner->size(); }
+    bool closed() const override { return _inner->closed(); }
+    int64_t mtime() const override { return _inner->mtime(); }
+    const std::string& get_data_dir_path() override { return _inner->get_data_dir_path(); }
+
+    const io::FileReaderSPtr& inner_reader() const { return _inner; }
+
+    /// Try an exact data page, waiting for its range when necessary. A true return means the page
+    /// was registered; a failed status leaves the caller to use the original reader. PageIO uses
+    /// this entry to distinguish buffer reuse from synchronous data-page fallback.
+    bool try_read_page(size_t offset, Slice output, Status* status);
+
+protected:
+    Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
+                        const io::IOContext* io_ctx) override;
+    Status read_at_iobuf_impl(size_t offset, size_t bytes_req, butil::IOBuf* out,
+                              size_t* bytes_read, const io::IOContext* io_ctx) override;
+    void _collect_profile_at_runtime() override { _inner->collect_profile_at_runtime(); }
+    void _collect_profile_before_close() override { _inner->collect_profile_before_close(); }
+
+private:
+    friend class SegmentReadAhead;
+
+    struct PageKey {
+        size_t offset {0};
+        size_t size {0};
+
+        bool operator<(const PageKey& other) const {
+            return offset < other.offset || (offset == other.offset && size < other.size);
+        }
+    };
+
+    struct PageOwner {
+        ColumnReadAhead* column {nullptr};
+        int32_t page_index {0};
+
+        bool operator==(const PageOwner&) const = default;
+    };
+
+    struct BufferedRange {
+        std::shared_ptr<io::FileRangeRead> read;
+        ReadAheadRangeConsumer consumer;
+        /// Guards the range-level consumer when several registered pages share this buffer.
+        bool consumed {false};
+    };
+
+    struct PageSlot {
+        std::vector<PageOwner> owners;
+        std::shared_ptr<BufferedRange> range;
+        size_t buffer_offset {0};
+    };
+
+    SegmentReadAheadFileReader(io::FileReaderSPtr inner,
+                               std::shared_ptr<io::ReadAheadStatistics> statistics);
+    /// Associate one exact page read with its location in a scheduled physical range.
+    void _register_page(ColumnReadAhead* column, const ColumnReadAheadPage& page,
+                        std::shared_ptr<BufferedRange> range, size_t buffer_offset);
+    /// Attach a column's independently advanced window to an already buffered physical page.
+    bool _reuse_page(ColumnReadAhead* column, const ColumnReadAheadPage& page);
+    /// Drop one column's prediction after its scan has passed the page.
+    void _release_page(ColumnReadAhead* column, const ColumnReadAheadPage& page);
+    /// Remove a page from every owning column and invoke the range consumer once after a successful
+    /// read. Callbacks run without `_mutex`.
+    void _finish_page(const PageKey& key, bool consumed);
+
+    const io::FileReaderSPtr _inner;
+    const std::shared_ptr<io::ReadAheadStatistics> _statistics;
+    std::mutex _mutex;
+    std::map<PageKey, PageSlot> _pages;
+};
+
+/// Coordinates the newly added pages of all physical columns in one segment submission. It does
+/// not decide column roles or window sizes and does not own the BE-level asynchronous reader.
+class SegmentReadAhead final {
+public:
+    SegmentReadAhead(io::FileReaderSPtr source_reader, io::FileRangeReadScheduler* scheduler,
+                     std::shared_ptr<io::FileRangeReadContext> context,
+                     io::FileRangeReadIOContext io_context, SegmentReadAheadOptions options);
+
+    SegmentReadAhead(const SegmentReadAhead&) = delete;
+    SegmentReadAhead& operator=(const SegmentReadAhead&) = delete;
+
+    /// Apply released pages first, probe new pages, then coalesce and submit all misses together.
+    /// Planning or admission failure completes those pages and leaves PageIO to read them through
+    /// the original reader.
+    SegmentReadAheadResult apply_plans(std::vector<ColumnReadAheadPlan> plans);
+
+    const std::shared_ptr<SegmentReadAheadFileReader>& file_reader() const { return _reader; }
+
+private:
+    const io::FileReaderSPtr _source_reader;
+    io::FileRangeReadScheduler* const _scheduler;
+    const std::shared_ptr<io::FileRangeReadContext> _context;
+    io::FileRangeReadIOContext _io_context;
+    const SegmentReadAheadOptions _options;
+    const std::shared_ptr<io::ReadAheadStatistics> _statistics;
+    const std::shared_ptr<SegmentReadAheadFileReader> _reader;
+};
+
+ReadAheadPageCacheProbe make_storage_page_cache_probe(const io::FileReaderSPtr& file_reader);
+
+} // namespace doris::segment_v2

be/test/storage/segment/segment_read_ahead_test.cpp

@@ -0,0 +1,568 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#include "storage/segment/segment_read_ahead.h"
+
+#include <butil/iobuf.h>
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <cstring>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "common/cast_set.h"
+#include "io/fs/read_ahead_metrics.h"
+#include "storage/olap_common.h"
+#include "storage/segment/page_handle.h"
+#include "storage/segment/page_io.h"
+#include "util/coding.h"
+#include "util/threadpool.h"
+
+namespace doris::segment_v2 {
+namespace {
+
+class TestFileReader final : public io::FileReader {
+public:
+    explicit TestFileReader(std::string data) : _data(std::move(data)) {}
+
+    Status close() override {
+        _closed = true;
+        return Status::OK();
+    }
+    const io::Path& path() const override { return _path; }
+    size_t size() const override { return _data.size(); }
+    bool closed() const override { return _closed; }
+    int64_t mtime() const override { return 0; }
+
+    void fail_next_read() { _fail_next_read = true; }
+    size_t read_calls() const { return _read_calls; }
+    const std::vector<io::FileRange>& reads() const { return _reads; }
+
+protected:
+    Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
+                        const io::IOContext* io_context) override {
+        std::lock_guard lock(_mutex);
+        ++_read_calls;
+        _reads.push_back({.offset = offset, .size = result.size});
+        if (_fail_next_read.exchange(false)) {
+            *bytes_read = 0;
+            return Status::IOError("injected read failure");
+        }
+        DORIS_CHECK(offset <= _data.size());
+        const size_t length = std::min(result.size, _data.size() - offset);
+        std::memcpy(result.data, _data.data() + offset, length);
+        *bytes_read = length;
+        if (io_context != nullptr && io_context->file_cache_stats != nullptr) {
+            ++io_context->file_cache_stats->num_remote_io_total;
+            io_context->file_cache_stats->bytes_read_from_remote += length;
+        }
+        if (io_context != nullptr && io_context->file_reader_stats != nullptr) {
+            ++io_context->file_reader_stats->read_calls;
+            io_context->file_reader_stats->read_bytes += length;
+        }
+        return Status::OK();
+    }
+
+private:
+    const io::Path _path {"test-segment.dat"};
+    const std::string _data;
+    mutable std::mutex _mutex;
+    std::vector<io::FileRange> _reads;
+    std::atomic<bool> _fail_next_read {false};
+    size_t _read_calls {0};
+    bool _closed {false};
+};
+
+std::vector<ColumnReadAheadPage> make_pages(size_t page_count, size_t page_size,
+                                            size_t first_offset = 0, size_t rows_per_page = 100) {
+    std::vector<ColumnReadAheadPage> pages;
+    for (size_t index = 0; index < page_count; ++index) {
+        pages.push_back({.page_index = cast_set<int32_t>(index),
+                         .first_ordinal = cast_set<ordinal_t>(index * rows_per_page),
+                         .last_ordinal = cast_set<ordinal_t>((index + 1) * rows_per_page - 1),
+                         .range = {.offset = first_offset + index * page_size, .size = page_size}});
+    }
+    return pages;
+}
+
+std::unique_ptr<ColumnReadAhead> make_window(size_t page_count, size_t page_size,
+                                             size_t first_offset = 0, size_t rows_per_page = 100) {
+    std::unique_ptr<ColumnReadAhead> window;
+    EXPECT_TRUE(
+            ColumnReadAhead::create(make_pages(page_count, page_size, first_offset, rows_per_page),
+                                    {.window_bytes = page_size * page_count}, false, &window)
+                    .ok());
+    return window;
+}
+
+ColumnReadAheadPlan plan_window(ColumnReadAhead* window, const roaring::Roaring& rows,
+                                rowid_t current) {
+    ColumnReadAheadPlan plan;
+    window->plan(&current, 1, rows, &plan);
+    return plan;
+}
+
+io::FileRangePlanOptions plan_options(size_t max_gap = 16) {
+    return {.coalesce_options = {.max_gap_bytes = max_gap,
+                                 .max_range_bytes = 1024,
+                                 .max_read_amplification_ratio = 2.0},
+            .cache_block_size = 1024,
+            .block_fill_min_coverage = 1.0};
+}
+
+ThreadPool* read_ahead_test_executor() {
+    static std::unique_ptr<ThreadPool> executor = []() {
+        std::unique_ptr<ThreadPool> result;
+        Status status = ThreadPoolBuilder("SegmentReadAheadTest")
+                                .set_min_threads(4)
+                                .set_max_threads(16)
+                                .build(&result);
+        DORIS_CHECK(status.ok());
+        return result;
+    }();
+    return executor.get();
+}
+
+std::unique_ptr<io::FileRangeReadScheduler> make_scheduler(size_t max_bytes_per_query = 4096) {
+    std::unique_ptr<io::FileRangeReadScheduler> scheduler;
+    EXPECT_TRUE(io::FileRangeReadScheduler::create(
+                        {.max_bytes_per_query = max_bytes_per_query, .max_bytes_per_be = 64 * 1024},
+                        read_ahead_test_executor(), &scheduler)
+                        .ok());
+    return scheduler;
+}
+
+std::unique_ptr<SegmentReadAhead> make_segment_read_ahead(const io::FileReaderSPtr& source,
+                                                          io::FileRangeReadScheduler* scheduler,
+                                                          SegmentReadAheadOptions options) {
+    auto context = scheduler->create_context();
+    return std::make_unique<SegmentReadAhead>(source, scheduler, std::move(context),
+                                              io::FileRangeReadIOContext {}, std::move(options));
+}
+
+roaring::Roaring rows(size_t count) {
+    roaring::Roaring result;
+    result.addRange(0, count);
+    return result;
+}
+
+TEST(SegmentReadAheadTest, CoalescesColumnsAndServesExactPageSlices) {
+    const auto consumed_before = io::read_ahead_bvars().consumed_page_bytes.get_value();
+    auto source = std::make_shared<TestFileReader>(std::string(256, 'x'));
+    auto scheduler = make_scheduler();
+    size_t consumer_factory_calls = 0;
+    std::vector<io::FileRange> consumed_ranges;
+    auto read_ahead = make_segment_read_ahead(
+            source, scheduler.get(),
+            {.range_plan = plan_options(), .page_cache_probe = {}, .range_consumer_factory = [&]() {
+                 ++consumer_factory_calls;
+                 return [&](const io::FileRange& range, Slice) {
+                     consumed_ranges.push_back(range);
+                 };
+             }});
+    auto first = make_window(1, 16, 0);
+    auto second = make_window(1, 16, 24);
+    const auto scan_rows = rows(100);
+    auto first_plan = plan_window(first.get(), scan_rows, 0);
+    auto second_plan = plan_window(second.get(), scan_rows, 0);
+
+    const auto result = read_ahead->apply_plans({std::move(second_plan), std::move(first_plan)});
+
+    ASSERT_TRUE(result.accepted()) << result.status;
+    EXPECT_EQ(consumer_factory_calls, 1);
+    EXPECT_EQ(result.submitted_ranges, 1);
+    EXPECT_EQ(result.submitted_bytes, 40);
+    char first_data[16] {};
+    size_t bytes_read = 0;
+    ASSERT_TRUE(read_ahead->file_reader()
+                        ->read_at(0, Slice(first_data, sizeof(first_data)), &bytes_read)
+                        .ok());
+    EXPECT_EQ(bytes_read, sizeof(first_data));
+    ASSERT_EQ(consumed_ranges.size(), 1);
+    EXPECT_EQ(consumed_ranges[0], (io::FileRange {.offset = 0, .size = 40}));
+
+    char second_data[16] {};
+    ASSERT_TRUE(read_ahead->file_reader()
+                        ->read_at(24, Slice(second_data, sizeof(second_data)), &bytes_read)
+                        .ok());
+    ASSERT_EQ(consumed_ranges.size(), 1);
+    EXPECT_EQ(source->read_calls(), 1);
+    EXPECT_EQ(source->reads(), (std::vector<io::FileRange> {{.offset = 0, .size = 40}}));
+
+    const auto& statistics = *read_ahead->_statistics;
+    EXPECT_EQ(statistics.plan_calls.value(), 1);
+    EXPECT_EQ(statistics.candidate_pages.value(), 2);
+    EXPECT_EQ(statistics.candidate_bytes.value(), 32);
+    EXPECT_EQ(statistics.input_bytes.value(), 32);
+    EXPECT_EQ(statistics.coalesced_bytes.value(), 40);
+    EXPECT_EQ(statistics.submitted_ranges.value(), 1);
+    EXPECT_EQ(statistics.remote_requests.value(), 1);
+    EXPECT_EQ(statistics.remote_bytes.value(), 40);
+    EXPECT_EQ(statistics.ready_page_hits.value() + statistics.wait_page_hits.value(), 2);
+    EXPECT_EQ(statistics.consumed_page_bytes.value(), 32);
+    EXPECT_EQ(statistics.consumed_range_bytes.value(), 40);
+    EXPECT_EQ(io::read_ahead_bvars().consumed_page_bytes.get_value() - consumed_before, 32);
+    EXPECT_GT(statistics.plan_time.value(), 0);
+    EXPECT_GT(statistics.read_time.value(), 0);
+}
+
+TEST(SegmentReadAheadTest, SharesOnePhysicalPageAcrossColumnOwners) {
+    auto source = std::make_shared<TestFileReader>(std::string(64, 'x'));
+    auto scheduler = make_scheduler();
+    auto read_ahead = make_segment_read_ahead(
+            source, scheduler.get(),
+            {.range_plan = plan_options(), .page_cache_probe = {}, .range_consumer_factory = {}});
+    auto first = make_window(1, 16, 8);
+    auto second = make_window(1, 16, 8);
+    const auto scan_rows = rows(100);
+    auto first_plan = plan_window(first.get(), scan_rows, 0);
+    auto second_plan = plan_window(second.get(), scan_rows, 0);
+
+    const auto result = read_ahead->apply_plans({std::move(first_plan), std::move(second_plan)});
+
+    ASSERT_TRUE(result.accepted()) << result.status;
+    EXPECT_EQ(result.submitted_ranges, 1);
+    EXPECT_EQ(result.submitted_bytes, 16);
+    char data[16] {};
+    size_t bytes_read = 0;
+    ASSERT_TRUE(read_ahead->file_reader()->read_at(8, Slice(data, sizeof(data)), &bytes_read).ok());
+    EXPECT_FALSE(first->pending(0));
+    EXPECT_FALSE(second->pending(0));
+    EXPECT_EQ(source->reads(), (std::vector<io::FileRange> {{.offset = 8, .size = 16}}));
+}
+
+TEST(SegmentReadAheadTest, UpdatesColumnsInOneCoalescedSubmission) {
+    auto source = std::make_shared<TestFileReader>(std::string(256, 's'));
+    auto scheduler = make_scheduler();
+    auto read_ahead = make_segment_read_ahead(
+            source, scheduler.get(),
+            {.range_plan = plan_options(), .page_cache_probe = {}, .range_consumer_factory = {}});
+    auto first = make_window(1, 16, 0);
+    auto second = make_window(1, 16, 24);
+    const auto scan_rows = rows(100);
+    std::vector<ColumnReadAheadPlan> plans;
+    plans.push_back(plan_window(first.get(), scan_rows, 0));
+    plans.push_back(plan_window(second.get(), scan_rows, 0));
+
+    static_cast<void>(read_ahead->apply_plans(std::move(plans)));
+
+    char first_data[16] {};
+    char second_data[16] {};
+    size_t bytes_read = 0;
+    ASSERT_TRUE(read_ahead->file_reader()
+                        ->read_at(0, Slice(first_data, sizeof(first_data)), &bytes_read)
+                        .ok());
+    ASSERT_TRUE(read_ahead->file_reader()
+                        ->read_at(24, Slice(second_data, sizeof(second_data)), &bytes_read)
+                        .ok());
+    EXPECT_EQ(source->reads(), (std::vector<io::FileRange> {{.offset = 0, .size = 40}}));
+}
+
+TEST(SegmentReadAheadTest, ReusesBufferedPageAcrossIndependentWindowSubmissions) {
+    auto source = std::make_shared<TestFileReader>(std::string(64, 'x'));
+    auto scheduler = make_scheduler();
+    auto read_ahead = make_segment_read_ahead(
+            source, scheduler.get(),
+            {.range_plan = plan_options(), .page_cache_probe = {}, .range_consumer_factory = {}});
+    auto first = make_window(1, 16, 8);
+    auto second = make_window(1, 16, 8);
+    ColumnReadAheadPlan first_plan;
+    ColumnReadAheadPlan second_plan;
+    first->start(rows(100), &first_plan);
+    second->start(rows(100), &second_plan);
+    auto result = read_ahead->apply_plans({std::move(first_plan)});
+    ASSERT_TRUE(result.accepted()) << result.status;
+    ASSERT_EQ(result.submitted_ranges, 1);
+
+    result = read_ahead->apply_plans({std::move(second_plan)});
+    ASSERT_TRUE(result.accepted()) << result.status;
+    EXPECT_EQ(result.submitted_ranges, 0);
+    char data[16] {};
+    size_t bytes_read = 0;
+    ASSERT_TRUE(read_ahead->file_reader()->read_at(8, Slice(data, sizeof(data)), &bytes_read).ok());
+    EXPECT_FALSE(first->pending(0));
+    EXPECT_FALSE(second->pending(0));
+    EXPECT_EQ(source->reads(), (std::vector<io::FileRange> {{.offset = 8, .size = 16}}));
+}
+
+TEST(SegmentReadAheadTest, PageCacheHitCompletesWindowWithoutSubmission) {
+    auto source = std::make_shared<TestFileReader>(std::string(64, 'p'));
+    auto scheduler = make_scheduler();
+    auto read_ahead =
+            make_segment_read_ahead(source, scheduler.get(),
+                                    {.range_plan = plan_options(),
+                                     .page_cache_probe = [](const io::FileRange&) { return true; },
+                                     .range_consumer_factory = {}});
+    auto window = make_window(1, 16);
+    const auto scan_rows = rows(100);
+    auto plan = plan_window(window.get(), scan_rows, 0);
+
+    const auto result = read_ahead->apply_plans({std::move(plan)});
+
+    EXPECT_TRUE(result.accepted());
+    EXPECT_EQ(result.page_cache_hits, 1);
+    EXPECT_EQ(result.submitted_ranges, 0);
+    EXPECT_EQ(window->pending_bytes(), 0);
+    EXPECT_FALSE(window->pending(0));
+    EXPECT_EQ(read_ahead->_statistics->candidate_bytes.value(), 16);
+    EXPECT_EQ(read_ahead->_statistics->page_cache_hit_bytes.value(), 16);
+}
+
+TEST(SegmentReadAheadTest, ReportsCandidateBytesBeforeCacheProbeAndDeduplication) {
+    auto source = std::make_shared<TestFileReader>(std::string(64, 'p'));
+    auto scheduler = make_scheduler();
+    auto read_ahead = make_segment_read_ahead(
+            source, scheduler.get(),
+            {.range_plan = plan_options(),
+             .page_cache_probe = [](const io::FileRange& page) { return page.offset == 0; },
+             .range_consumer_factory = {}});
+    auto cached = make_window(1, 16);
+    auto first = make_window(1, 8, 24);
+    auto second = make_window(1, 8, 24);
+    const auto scan_rows = rows(100);
+    ASSERT_TRUE(read_ahead
+                        ->apply_plans({plan_window(cached.get(), scan_rows, 0),
+                                       plan_window(first.get(), scan_rows, 0),
+                                       plan_window(second.get(), scan_rows, 0)})
+                        .accepted());
+    EXPECT_EQ(read_ahead->_statistics->candidate_bytes.value(), 32);
+    EXPECT_EQ(read_ahead->_statistics->page_cache_hit_bytes.value(), 16);
+    EXPECT_EQ(read_ahead->_statistics->input_bytes.value(), 8);
+    // Keep the column owners alive until their buffered page mappings are released.
+    read_ahead.reset();
+}
+
+TEST(SegmentReadAheadTest, ConsumedPageReportsTheWholeRangeBeforeClose) {
+    auto source = std::make_shared<TestFileReader>(std::string(64, 'c'));
+    auto scheduler = make_scheduler();
+    auto options = plan_options();
+    options.cache_block_size = 64;
+    options.block_fill_min_coverage = 0.25;
+    auto read_ahead = make_segment_read_ahead(
+            source, scheduler.get(),
+            {.range_plan = options, .page_cache_probe = {}, .range_consumer_factory = {}});
+    auto window = make_window(1, 16, 8);
+    ASSERT_TRUE(read_ahead->apply_plans({plan_window(window.get(), rows(100), 0)}).accepted());
+    char data[16];
+    size_t bytes_read = 0;
+    ASSERT_TRUE(read_ahead->file_reader()->read_at(8, Slice(data, sizeof(data)), &bytes_read).ok());
+    RuntimeProfile profile("ClosingScanner");
+    read_ahead->_statistics->update_profile(&profile);
+    EXPECT_EQ(profile.get_counter("ReadAheadSourceBytes")->value(), 64);
+    EXPECT_EQ(profile.get_counter("ReadAheadSuccessfulSourceBytes")->value(), 64);
+    EXPECT_EQ(profile.get_counter("ReadAheadConsumedPageBytes")->value(), 16);
+    EXPECT_EQ(profile.get_counter("ReadAheadConsumedRangeBytes")->value(), 64);
+    EXPECT_EQ(profile.get_counter("ReadAheadBlockFillBytes")->value(), 48);
+}
+
+TEST(SegmentReadAheadTest, PageIOCountsDataFallbackAfterRejectionOrReadFailure) {
+    PageFooterPB written_footer;
+    written_footer.set_type(DATA_PAGE);
+    written_footer.set_uncompressed_size(4);
+    written_footer.mutable_data_page_footer()->set_first_ordinal(0);
+    written_footer.mutable_data_page_footer()->set_num_values(1);
+    written_footer.mutable_data_page_footer()->set_nullmap_size(0);
+    const std::string serialized_footer = written_footer.SerializeAsString();
+    char trailer[8] {};
+    encode_fixed32_le(reinterpret_cast<uint8_t*>(trailer),
+                      cast_set<uint32_t>(serialized_footer.size()));
+    const std::string content = "data" + serialized_footer + std::string(trailer, sizeof(trailer));
+
+    for (bool reject : {false, true}) {
+        auto source = std::make_shared<TestFileReader>(content);
+        auto scheduler = make_scheduler(reject ? 1 : 4096);
+        auto read_ahead = make_segment_read_ahead(source, scheduler.get(),
+                                                  {.range_plan = plan_options(),
+                                                   .page_cache_probe = {},
+                                                   .range_consumer_factory = {}});
+        auto window = make_window(1, content.size());
+        if (!reject) {
+            source->fail_next_read();
+        }
+        const auto submitted = read_ahead->apply_plans({plan_window(window.get(), rows(100), 0)});
+        ASSERT_EQ(submitted.accepted(), !reject);
+
+        OlapReaderStatistics stats;
+        stats.read_ahead_stats = read_ahead->_statistics;
+        PageReadOptions options(io::IOContext {});
+        options.file_reader = read_ahead->file_reader().get();
+        options.stats = &stats;
+        options.type = DATA_PAGE;
+        options.page_pointer = PagePointer(0, cast_set<uint32_t>(content.size()));
+        options.verify_checksum = false;
+        options.pre_decode = false;
+        PageHandle handle;
+        Slice body;
+        PageFooterPB footer;
+        ASSERT_TRUE(PageIO::read_and_decompress_page(options, &handle, &body, &footer).ok());
+        EXPECT_EQ(body.to_string(), "data");
+        EXPECT_EQ(stats.read_ahead_stats->fallback_pages.value(), 1);
+        EXPECT_EQ(stats.read_ahead_stats->fallback_bytes.value(), content.size());
+        EXPECT_GT(stats.read_ahead_stats->fallback_time.value(), 0);
+        EXPECT_EQ(stats.read_ahead_stats->consumed_page_bytes.value(), 0);
+        EXPECT_EQ(stats.read_ahead_stats->ready_page_hits.value(), 0);
+        EXPECT_EQ(stats.read_ahead_stats->wait_page_hits.value(), 0);
+        EXPECT_EQ(source->read_calls(), reject ? 1 : 2);
+
+        // Index and dictionary reads use the INDEX_PAGE cache category and are not fallbacks.
+        options.type = INDEX_PAGE;
+        ASSERT_TRUE(PageIO::read_and_decompress_page(options, &handle, &body, &footer).ok());
+        EXPECT_EQ(stats.read_ahead_stats->fallback_pages.value(), 1);
+
+        options.type = DATA_PAGE;
+        source->fail_next_read();
+        EXPECT_FALSE(PageIO::read_and_decompress_page(options, &handle, &body, &footer).ok());
+        EXPECT_EQ(stats.read_ahead_stats->fallback_pages.value(), 2);
+        EXPECT_EQ(stats.read_ahead_stats->fallback_bytes.value(), 2 * content.size());
+    }
+}
+
+TEST(SegmentReadAheadTest, AdmissionRejectionFallsBackWithoutWaiting) {
+    auto source = std::make_shared<TestFileReader>(std::string(64, 'r'));
+    auto scheduler = make_scheduler(8);
+    auto read_ahead = make_segment_read_ahead(
+            source, scheduler.get(),
+            {.range_plan = plan_options(), .page_cache_probe = {}, .range_consumer_factory = {}});
+    auto window = make_window(1, 16);
+    const auto scan_rows = rows(100);
+    auto plan = plan_window(window.get(), scan_rows, 0);
+
+    const auto result = read_ahead->apply_plans({std::move(plan)});
+
+    EXPECT_FALSE(result.accepted());
+    EXPECT_EQ(result.reject_reason, io::FileRangeReadRejectReason::QUERY_BYTE_LIMIT);
+    EXPECT_FALSE(window->pending(0));
+    EXPECT_EQ(window->pending_bytes(), 0);
+    char data[16] {};
+    size_t bytes_read = 0;
+    EXPECT_TRUE(read_ahead->file_reader()->read_at(0, Slice(data, sizeof(data)), &bytes_read).ok());
+    EXPECT_EQ(source->read_calls(), 1);
+}
+
+TEST(SegmentReadAheadTest, PartialAdmissionPreservesPageMappingAndWriteback) {
+    const std::string content = std::string(8, 'a') + std::string(8, 'b') + std::string(64, 'c') +
+                                std::string(8, 'd') + std::string(40, 'e');
+    auto source = std::make_shared<TestFileReader>(content);
+    auto scheduler = make_scheduler(24);
+    std::vector<io::FileRange> consumed;
+    auto read_ahead = make_segment_read_ahead(
+            source, scheduler.get(),
+            {.range_plan = plan_options(0),
+             .page_cache_probe = {},
+             .range_consumer_factory = [&]() -> ReadAheadRangeConsumer {
+                 return [&](const io::FileRangeRead& read) { consumed.push_back(read.range()); };
+             }});
+    auto first = make_window(2, 8, 0);
+    auto second = make_window(1, 16, 48, 200);
+    auto third = make_window(1, 8, 80, 200);
+    const auto scan_rows = rows(200);
+
+    // Coalescing yields [0,16), [48,64), [80,88). The 24-byte budget admits the first and
+    // third ranges. Both pages in the first range must keep their original buffer offsets.
+    const auto result = read_ahead->apply_plans({plan_window(first.get(), scan_rows, 0),
+                                                 plan_window(second.get(), scan_rows, 0),
+                                                 plan_window(third.get(), scan_rows, 0)});
+    ASSERT_TRUE(result.accepted()) << result.status;
+    EXPECT_EQ(result.new_pages, 4);
+    EXPECT_EQ(result.submitted_ranges, 2);
+    EXPECT_EQ(result.submitted_bytes, 24);
+    EXPECT_TRUE(first->pending(0));
+    EXPECT_TRUE(first->pending(1));
+    EXPECT_FALSE(second->pending(0));
+    EXPECT_TRUE(third->pending(0));
+
+    char data[16] {};
+    Status status;
+    EXPECT_FALSE(read_ahead->file_reader()->try_read_page(48, Slice(data, 16), &status));
+    size_t bytes_read = 0;
+    ASSERT_TRUE(read_ahead->file_reader()->read_at(48, Slice(data, 16), &bytes_read).ok());
+    EXPECT_EQ(bytes_read, 16);
+    EXPECT_EQ(Slice(data, bytes_read).to_string(), content.substr(48, 16));
+    EXPECT_TRUE(consumed.empty());
+
+    ASSERT_TRUE(read_ahead->file_reader()->read_at(0, Slice(data, 8), &bytes_read).ok());
+    EXPECT_EQ(Slice(data, bytes_read).to_string(), content.substr(0, 8));
+    EXPECT_FALSE(first->pending(0));
+    EXPECT_TRUE(first->pending(1));
+    ASSERT_TRUE(read_ahead->file_reader()->read_at(8, Slice(data, 8), &bytes_read).ok());
+    EXPECT_EQ(Slice(data, bytes_read).to_string(), content.substr(8, 8));
+    ASSERT_TRUE(read_ahead->file_reader()->read_at(80, Slice(data, 8), &bytes_read).ok());
+    EXPECT_EQ(Slice(data, bytes_read).to_string(), content.substr(80, 8));
+    EXPECT_FALSE(first->pending(1));
+    EXPECT_FALSE(third->pending(0));
+    EXPECT_EQ(consumed,
+              (std::vector<io::FileRange> {{.offset = 0, .size = 16}, {.offset = 80, .size = 8}}));
+    scheduler->shutdown();
+    // The thread-pool runnable releases its final buffer reference after scheduler completion.
+    read_ahead_test_executor()->wait();
+    EXPECT_EQ(source->read_calls(), 3);
+    EXPECT_EQ(read_ahead->_statistics->submitted_ranges.value(), 2);
+    EXPECT_EQ(read_ahead->_statistics->submitted_bytes.value(), 24);
+    EXPECT_EQ(read_ahead->_context->resident_bytes(), 0);
+}
+
+TEST(SegmentReadAheadTest, AsyncReadFailureUsesOriginalReader) {
+    auto source = std::make_shared<TestFileReader>(std::string(64, 'f'));
+    source->fail_next_read();
+    auto scheduler = make_scheduler();
+    auto read_ahead = make_segment_read_ahead(
+            source, scheduler.get(),
+            {.range_plan = plan_options(), .page_cache_probe = {}, .range_consumer_factory = {}});
+    auto window = make_window(1, 16);
+    const auto scan_rows = rows(100);
+    auto plan = plan_window(window.get(), scan_rows, 0);
+    ASSERT_TRUE(read_ahead->apply_plans({std::move(plan)}).accepted());
+
+    char data[16] {};
+    size_t bytes_read = 0;
+    const auto status =
+            read_ahead->file_reader()->read_at(0, Slice(data, sizeof(data)), &bytes_read);
+
+    EXPECT_TRUE(status.ok()) << status;
+    EXPECT_EQ(bytes_read, sizeof(data));
+    EXPECT_EQ(source->read_calls(), 2);
+    EXPECT_FALSE(window->pending(0));
+}
+
+TEST(SegmentReadAheadTest, DiscardedUnusedRangeIsNotPublished) {
+    auto source = std::make_shared<TestFileReader>(std::string(64, 'd'));
+    auto scheduler = make_scheduler();
+    size_t published = 0;
+    auto read_ahead = make_segment_read_ahead(
+            source, scheduler.get(),
+            {.range_plan = plan_options(), .page_cache_probe = {}, .range_consumer_factory = [&]() {
+                 return [&](const io::FileRange&, Slice) { ++published; };
+             }});
+    auto window = make_window(2, 16);
+    const auto scan_rows = rows(200);
+    auto first_plan = plan_window(window.get(), scan_rows, 0);
+    ASSERT_TRUE(read_ahead->apply_plans({std::move(first_plan)}).accepted());
+
+    const rowid_t later = 199;
+    ColumnReadAheadPlan discard_plan;
+    window->plan(&later, 1, scan_rows, &discard_plan);
+    static_cast<void>(read_ahead->apply_plans({std::move(discard_plan)}));
+
+    EXPECT_EQ(published, 0);
+}
+
+} // namespace
+} // namespace doris::segment_v2

build-support/check-header-deps.py

@@ -591,7 +591,7 @@ REVERSE_REACH_BASELINES = {
     # parsed_page.h holds RleDecoder<bool> by value and constructs it in the
     # header, so the whole storage read stack sees the RLE machinery; a live
     # dependency, budgeted as-is (trimming it is a refactor, not a gate).
-    "util/rle_encoding.h": 828,
+    "util/rle_encoding.h": 834,
     "gen_cpp/internal_service.pb.h": 741,
     "gen_cpp/FrontendService_types.h": 576,
     "storage/options.h": 464,