[feature](be) Extend range read-ahead to point queries

### What problem does this PR solve?

Issue Number: None

Related PR: #67579, #67292

Problem Summary: Short-circuit point queries read row-store data or missing column-store values through exact row IDs. Create one request-scoped range-read context, initialize the scalar row-store or missing top-level scalar iterators against the shared read-ahead reader, and submit the required pages before decoding. Complex columns and extracted variant paths retain Segment's original iterator/reader resolution. Preserve row-cache behavior, result padding, request-scoped constants and synchronous fallback. Append read-ahead Profile statistics to point-query diagnostics; BE cumulative counters retain unsampled totals. Include exact-row helpers and tests in this component. TopN second-stage integration remains identical to the master base and is outside this PR series.

Handle a dynamic switch closing after a point request creates its context: if create_for_query returns no read-ahead object, read row-store data synchronously through the already loaded segment. The deterministic row-store test closes either enable switch after segment resolution, verifies that no segment-loader call is made, and compares the result with the original reader.

Rebase compatibility: reuse the segment returned by primary-key lookup in both the read-ahead and synchronous row-store paths, preserving master's segment and acquired-rowset lifetime.

### Release note

Use query range read-ahead for scalar point-query data pages when enable_query_read_ahead and enable_async_file_cache_write are enabled.

### Check List (For Author)

- Test: Unit Test — pre-rebase integrated ASAN/PCH validation covered 225 tests from 10 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 — adds scalar point-query prefetch without changing complex-column decoding, scheduling limits or query results.
- Does this need documentation: Yes — scope and synchronization are recorded in the related active PR descriptions.

be/src/service/point_query_executor.cpp

@@ -27,6 +27,7 @@
 #include <stdlib.h>
 
 #include <memory>
+#include <sstream>
 #include <unordered_map>
 #include <vector>
 
@@ -43,6 +44,7 @@
 #include "exprs/vexpr_fwd.h"
 #include "exprs/vslot_ref.h"
 #include "io/cache/remote_scan_cache_write_limiter.h"
+#include "io/fs/read_ahead_metrics.h"
 #include "io/io_common.h"
 #include "runtime/descriptors.h"
 #include "runtime/exec_env.h"
@@ -55,6 +57,8 @@
 #include "storage/rowset/rowset_fwd.h"
 #include "storage/schema.h"
 #include "storage/segment/column_reader.h"
+#include "storage/segment/rowid_read_ahead.h"
+#include "storage/segment/segment_read_ahead.h"
 #include "storage/tablet/tablet_schema.h"
 #include "storage/utils.h"
 #include "util/defer_op.h"
@@ -417,13 +421,29 @@ void PointQueryExecutor::print_profile() {
             _profile_metrics.read_stats.file_cache_stats.remote_io_timer,
             _profile_metrics.read_stats.file_cache_stats.write_cache_io_timer);
 
+    const auto read_ahead_profile = [&]() {
+        if (_profile_metrics.read_stats.read_ahead_stats == nullptr &&
+            _read_stats.read_ahead_stats == nullptr) {
+            return std::string {};
+        }
+        RuntimeProfile profile("PageReadAhead");
+        for (auto* stats : {&_profile_metrics.read_stats, &_read_stats}) {
+            if (stats->read_ahead_stats != nullptr) {
+                stats->read_ahead_stats->update_profile(&profile);
+            }
+        }
+        std::ostringstream output;
+        profile.pretty_print(&output);
+        return "\n" + output.str();
+    };
+
     constexpr static int kSlowThreholdUs = 50 * 1000; // 50ms
     if (total_us > kSlowThreholdUs) {
-        LOG(WARNING) << "slow query, " << stats_str;
+        LOG(WARNING) << "slow query, " << stats_str << read_ahead_profile();
     } else if (VLOG_DEBUG_IS_ON) {
-        VLOG_DEBUG << stats_str;
+        VLOG_DEBUG << stats_str << read_ahead_profile();
     } else {
-        LOG_EVERY_N(INFO, 1000) << stats_str;
+        LOG_EVERY_N(INFO, 1000) << stats_str << read_ahead_profile();
     }
 }
 
@@ -528,6 +548,12 @@ Status PointQueryExecutor::_lookup_row_key() {
 Status PointQueryExecutor::_lookup_row_data() {
     // 3. get values
     SCOPED_TIMER(&_profile_metrics.lookup_data_ns);
+    std::shared_ptr<io::FileRangeReadContext> read_ahead_context;
+    if (segment_v2::SegmentReadAhead::enabled()) {
+        auto* scheduler = ExecEnv::GetInstance()->file_range_read_scheduler();
+        DORIS_CHECK(scheduler != nullptr);
+        read_ahead_context = scheduler->create_context();
+    }
     {
         auto result_columns_guard = _result_block->mutate_columns_scoped();
         MutableColumns& result_columns = result_columns_guard.mutable_columns();
@@ -552,10 +578,18 @@ Status PointQueryExecutor::_lookup_row_data() {
                 io_ctx.reader_type = ReaderType::READER_QUERY;
                 io_ctx.file_cache_stats = &_profile_metrics.read_stats.file_cache_stats;
                 io_ctx.remote_scan_cache_write_limiter = _remote_scan_cache_write_limiter.get();
-                RETURN_IF_ERROR(_tablet->lookup_row_data(
-                        _row_read_ctxs[i]._primary_key, _row_read_ctxs[i]._row_location.value(),
-                        _row_read_ctxs[i]._segment, *(_row_read_ctxs[i]._rowset_ptr),
-                        _profile_metrics.read_stats, value, use_row_cache, &io_ctx));
+                if (read_ahead_context != nullptr) {
+                    RETURN_IF_ERROR(_tablet->lookup_row_data_with_read_ahead(
+                            _row_read_ctxs[i]._primary_key, _row_read_ctxs[i]._row_location.value(),
+                            _row_read_ctxs[i]._segment, *(_row_read_ctxs[i]._rowset_ptr),
+                            _profile_metrics.read_stats, value, use_row_cache, &io_ctx,
+                            read_ahead_context));
+                } else {
+                    RETURN_IF_ERROR(_tablet->lookup_row_data(
+                            _row_read_ctxs[i]._primary_key, _row_read_ctxs[i]._row_location.value(),
+                            _row_read_ctxs[i]._segment, *(_row_read_ctxs[i]._rowset_ptr),
+                            _profile_metrics.read_stats, value, use_row_cache, &io_ctx));
+                }
                 // serialize value to block, currently only jsonb row format
                 RETURN_IF_ERROR(JsonbSerializeUtil::jsonb_to_columns(
                         _reusable->get_data_type_serdes(), value.data(), value.size(),
@@ -598,37 +632,80 @@ Status PointQueryExecutor::_lookup_row_data() {
                                            return seg->id() == row_loc.segment_id;
                                        });
                 const auto& segment = *it;
-                const auto tablet_schema = _tablet->tablet_schema();
-                for (int cid : _reusable->missing_col_uids()) {
-                    int pos = _reusable->get_col_uid_to_idx().at(cid);
+                StorageReadOptions read_ahead_options(_read_stats);
+                read_ahead_options.use_page_cache = !config::disable_storage_page_cache;
+                read_ahead_options.io_ctx = io_ctx;
+                // Pre-read iterators need the same rowset constants as the synchronous path:
+                // a version column in rowset [7-7] must produce 7, not its stored placeholder 0.
+                read_ahead_options.version = rowset->version();
+                read_ahead_options.commit_tso = rowset->commit_tso();
+                std::unique_ptr<segment_v2::SegmentReadAhead> read_ahead;
+                if (read_ahead_context != nullptr) {
+                    const auto status = segment_v2::SegmentReadAhead::create_for_query(
+                            segment->file_reader(), ExecEnv::GetInstance(), read_ahead_context,
+                            read_ahead_options, &read_ahead);
+                    if (!status.ok()) {
+                        LOG_EVERY_N(WARNING, 100)
+                                << "failed to initialize point-read column read-ahead; use the "
+                                   "original read path: "
+                                << status;
+                    }
+                }
+                if (read_ahead != nullptr) {
+                    const auto missing_column_count = _reusable->missing_col_uids().size();
+                    std::vector<StorageReadOptions> read_options(missing_column_count,
+                                                                 read_ahead_options);
+                    std::vector<std::unique_ptr<ColumnIterator>> iterators(missing_column_count);
+                    std::vector<segment_v2::RowIdColumnRead> column_reads;
+                    column_reads.reserve(missing_column_count);
+                    size_t read_index = 0;
+                    for (int cid : _reusable->missing_col_uids()) {
+                        int pos = _reusable->get_col_uid_to_idx().at(cid);
+                        SlotDescriptor* slot = _reusable->tuple_desc()->slots()[pos];
+                        column_reads.push_back({.slot = slot,
+                                                .result = &result_columns[pos],
+                                                .read_options = &read_options[read_index],
+                                                .iterator = &iterators[read_index]});
+                        ++read_index;
+                    }
                     std::vector<segment_v2::rowid_t> row_ids {
                             static_cast<segment_v2::rowid_t>(row_loc.row_id)};
-                    auto& column = result_columns[pos];
-                    std::unique_ptr<ColumnIterator> iter;
-                    SlotDescriptor* slot = _reusable->tuple_desc()->slots()[pos];
-                    int32_t index = slot->col_unique_id() >= 0
-                                            ? tablet_schema->field_index(slot->col_unique_id())
-                                            : tablet_schema->field_index(slot->col_name());
-                    // A light schema change adds a column without writing anything, and
-                    // BaseTablet::_max_version_schema only moves forward on write, so until the
-                    // next load the tablet schema here does not have it. No segment holds it
-                    // either, so describe it from the slot and let the segment answer with its
-                    // default -- the same answer it gives once a load does bring the column in.
-                    TabletColumn column_awaiting_a_load;
-                    if (index < 0) {
-                        column_awaiting_a_load = ReadSchema::tablet_column_from_slot(*slot);
+                    RETURN_IF_ERROR(segment_v2::read_columns_by_rowids_with_read_ahead(
+                            *segment, *_tablet->tablet_schema(), row_ids, column_reads,
+                            *read_ahead));
+                } else {
+                    const auto tablet_schema = _tablet->tablet_schema();
+                    for (int cid : _reusable->missing_col_uids()) {
+                        int pos = _reusable->get_col_uid_to_idx().at(cid);
+                        std::vector<segment_v2::rowid_t> row_ids {
+                                static_cast<segment_v2::rowid_t>(row_loc.row_id)};
+                        auto& column = result_columns[pos];
+                        std::unique_ptr<ColumnIterator> iter;
+                        SlotDescriptor* slot = _reusable->tuple_desc()->slots()[pos];
+                        int32_t index = slot->col_unique_id() >= 0
+                                                ? tablet_schema->field_index(slot->col_unique_id())
+                                                : tablet_schema->field_index(slot->col_name());
+                        // A light schema change adds a column without writing anything, and
+                        // BaseTablet::_max_version_schema only moves forward on write, so until the
+                        // next load the tablet schema here does not have it. No segment holds it
+                        // either, so describe it from the slot and let the segment answer with its
+                        // default -- the same answer it gives once a load does bring the column in.
+                        TabletColumn column_awaiting_a_load;
+                        if (index < 0) {
+                            column_awaiting_a_load = ReadSchema::tablet_column_from_slot(*slot);
+                        }
+                        const TabletColumn& read_column =
+                                index >= 0 ? tablet_schema->column(index) : column_awaiting_a_load;
+                        StorageReadOptions storage_read_options(_read_stats);
+                        storage_read_options.io_ctx = io_ctx;
+                        // A point lookup bypasses TabletReader, so supply the rowset context here. For
+                        // example, reading __DORIS_VERSION_COL__ from rowset [7-7] must return 7
+                        // instead of its on-disk placeholder 0.
+                        storage_read_options.version = rowset->version();
+                        storage_read_options.commit_tso = rowset->commit_tso();
+                        RETURN_IF_ERROR(segment->seek_and_read_by_rowid(
+                                read_column, slot, row_ids, column, storage_read_options, iter));
                     }
-                    const TabletColumn& read_column =
-                            index >= 0 ? tablet_schema->column(index) : column_awaiting_a_load;
-                    StorageReadOptions storage_read_options(_read_stats);
-                    storage_read_options.io_ctx = io_ctx;
-                    // A point lookup bypasses TabletReader, so supply the rowset context here. For
-                    // example, reading __DORIS_VERSION_COL__ from rowset [7-7] must return 7
-                    // instead of its on-disk placeholder 0.
-                    storage_read_options.version = rowset->version();
-                    storage_read_options.commit_tso = rowset->commit_tso();
-                    RETURN_IF_ERROR(segment->seek_and_read_by_rowid(
-                            read_column, slot, row_ids, column, storage_read_options, iter));
                 }
             }
         }

be/src/storage/segment/rowid_read_ahead.cpp

@@ -0,0 +1,158 @@
+// 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/rowid_read_ahead.h"
+
+#include <algorithm>
+#include <array>
+
+#include "common/config.h"
+#include "runtime/descriptors.h"
+#include "storage/schema.h"
+#include "storage/segment/column_reader.h"
+#include "storage/segment/segment.h"
+#include "storage/segment/segment_read_ahead.h"
+#include "storage/tablet/tablet_schema.h"
+#include "storage/types.h"
+
+namespace doris::segment_v2 {
+namespace {
+
+Status prepare_column_iterator(Segment& segment, const TabletColumn& column,
+                               StorageReadOptions& read_options,
+                               std::unique_ptr<ColumnIterator>& iterator,
+                               SegmentReadAhead& read_ahead) {
+    DORIS_CHECK(read_options.stats != nullptr);
+    if (iterator == nullptr) {
+        RETURN_IF_ERROR(segment.new_column_iterator(column, &iterator, &read_options));
+        auto io_ctx = read_options.io_ctx;
+        io_ctx.reader_type = ReaderType::READER_QUERY;
+        io_ctx.file_cache_stats = &read_options.stats->file_cache_stats;
+        ColumnIteratorOptions options {
+                .use_page_cache = !config::disable_storage_page_cache,
+                .file_reader = read_ahead.file_reader().get(),
+                .stats = read_options.stats,
+                .io_ctx = io_ctx,
+        };
+        RETURN_IF_ERROR(iterator->init(options));
+    }
+    return Status::OK();
+}
+
+Status prepare_slot_iterator(Segment& segment, const TabletSchema& schema, RowIdColumnRead& column,
+                             SegmentReadAhead& read_ahead, ColumnIterator** prepared) {
+    DORIS_CHECK(column.slot != nullptr);
+    DORIS_CHECK(column.read_options != nullptr);
+    DORIS_CHECK(column.iterator != nullptr);
+    DORIS_CHECK(prepared != nullptr);
+    *prepared = nullptr;
+
+    // Extracted variant paths use Segment's existing path resolution and casting logic.
+    if (!column.slot->column_paths().empty()) {
+        return Status::OK();
+    }
+    const int index = column.slot->col_unique_id() >= 0
+                              ? schema.field_index(column.slot->col_unique_id())
+                              : schema.field_index(column.slot->col_name());
+    if (index < 0) {
+        return Status::OK();
+    }
+    const auto& tablet_column = schema.column(index);
+    // Nested columns retain the original iterator and reader; their children can be read in
+    // different predicate/lazy phases, with independent positions.
+    if (!is_scalar_type(tablet_column.type())) {
+        return Status::OK();
+    }
+
+    RETURN_IF_ERROR(prepare_column_iterator(segment, tablet_column, *column.read_options,
+                                            *column.iterator, read_ahead));
+    *prepared = column.iterator->get();
+    return Status::OK();
+}
+
+} // namespace
+
+Status prepare_columns_by_rowids_with_read_ahead(Segment& segment, const TabletSchema& schema,
+                                                 const std::vector<uint32_t>& row_ids,
+                                                 std::span<RowIdColumnRead> columns,
+                                                 SegmentReadAhead& read_ahead) {
+    if (row_ids.empty()) {
+        return Status::OK();
+    }
+
+    std::vector<ColumnIterator*> prepared;
+    prepared.reserve(columns.size());
+    for (auto& column : columns) {
+        ColumnIterator* iterator = nullptr;
+        RETURN_IF_ERROR(prepare_slot_iterator(segment, schema, column, read_ahead, &iterator));
+        if (iterator != nullptr) {
+            prepared.push_back(iterator);
+        }
+    }
+    if (!prepared.empty()) {
+        static_cast<void>(read_ahead.prefetch_by_rowids(row_ids.data(), row_ids.size(), prepared));
+    }
+
+    return Status::OK();
+}
+
+Status read_columns_by_rowids_with_read_ahead(Segment& segment, const TabletSchema& schema,
+                                              const std::vector<uint32_t>& row_ids,
+                                              std::span<RowIdColumnRead> columns,
+                                              SegmentReadAhead& read_ahead) {
+    RETURN_IF_ERROR(prepare_columns_by_rowids_with_read_ahead(segment, schema, row_ids, columns,
+                                                              read_ahead));
+
+    for (auto& column : columns) {
+        DORIS_CHECK(column.result != nullptr);
+        const int index = column.slot->col_unique_id() >= 0
+                                  ? schema.field_index(column.slot->col_unique_id())
+                                  : schema.field_index(column.slot->col_name());
+        // A newly added column can be absent from the tablet schema until the next load.
+        // Preserve the point-query path's slot-derived default in that interval.
+        TabletColumn column_awaiting_a_load;
+        if (index < 0) {
+            column_awaiting_a_load = ReadSchema::tablet_column_from_slot(*column.slot);
+        }
+        const TabletColumn& read_column =
+                index >= 0 ? schema.column(index) : column_awaiting_a_load;
+        RETURN_IF_ERROR(segment.seek_and_read_by_rowid(read_column, column.slot, row_ids,
+                                                       *column.result, *column.read_options,
+                                                       *column.iterator));
+    }
+    return Status::OK();
+}
+
+Status read_column_by_rowids_with_read_ahead(Segment& segment, const TabletColumn& column,
+                                             const std::vector<uint32_t>& row_ids,
+                                             MutableColumnPtr& result,
+                                             StorageReadOptions& read_options,
+                                             std::unique_ptr<ColumnIterator>& iterator,
+                                             SegmentReadAhead& read_ahead) {
+    if (row_ids.empty()) {
+        return Status::OK();
+    }
+    DORIS_CHECK(std::is_sorted(row_ids.begin(), row_ids.end()));
+    DORIS_CHECK(std::adjacent_find(row_ids.begin(), row_ids.end()) == row_ids.end());
+    DORIS_CHECK(is_scalar_type(column.type()));
+    RETURN_IF_ERROR(prepare_column_iterator(segment, column, read_options, iterator, read_ahead));
+    std::array<ColumnIterator*, 1> columns {iterator.get()};
+    static_cast<void>(read_ahead.prefetch_by_rowids(row_ids.data(), row_ids.size(), columns));
+    return iterator->read_by_rowids(row_ids.data(), row_ids.size(), result);
+}
+
+} // namespace doris::segment_v2

be/src/storage/segment/rowid_read_ahead.h

@@ -0,0 +1,74 @@
+// 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 <cstdint>
+#include <memory>
+#include <span>
+#include <vector>
+
+#include "common/status.h"
+#include "core/column/column.h"
+
+namespace doris {
+
+class SlotDescriptor;
+class StorageReadOptions;
+class TabletColumn;
+class TabletSchema;
+
+namespace segment_v2 {
+
+class ColumnIterator;
+class Segment;
+class SegmentReadAhead;
+
+struct RowIdColumnRead {
+    /// Logical slot used to resolve the physical TabletColumn.
+    SlotDescriptor* slot {nullptr};
+    /// Destination column populated by the exact row-id read.
+    MutableColumnPtr* result {nullptr};
+    /// Per-column read options updated to use the read-ahead reader.
+    StorageReadOptions* read_options {nullptr};
+    /// Iterator initialized on demand and reused for decoding.
+    std::unique_ptr<ColumnIterator>* iterator {nullptr};
+};
+
+/// Initializes top-level scalar iterators on the read-ahead reader and submits their data pages
+/// together. Complex columns and extracted variant paths keep the original Segment path.
+Status prepare_columns_by_rowids_with_read_ahead(Segment& segment, const TabletSchema& schema,
+                                                 const std::vector<uint32_t>& row_ids,
+                                                 std::span<RowIdColumnRead> columns,
+                                                 SegmentReadAhead& read_ahead);
+
+/// Prepares all columns, then delegates decoding to Segment::seek_and_read_by_rowid().
+Status read_columns_by_rowids_with_read_ahead(Segment& segment, const TabletSchema& schema,
+                                              const std::vector<uint32_t>& row_ids,
+                                              std::span<RowIdColumnRead> columns,
+                                              SegmentReadAhead& read_ahead);
+
+/// Reads a physical scalar column whose TabletColumn is known, such as the row-store column.
+Status read_column_by_rowids_with_read_ahead(Segment& segment, const TabletColumn& column,
+                                             const std::vector<uint32_t>& row_ids,
+                                             MutableColumnPtr& result,
+                                             StorageReadOptions& read_options,
+                                             std::unique_ptr<ColumnIterator>& iterator,
+                                             SegmentReadAhead& read_ahead);
+
+} // namespace segment_v2
+} // namespace doris

be/src/storage/segment/segment_read_ahead.cpp

@@ -310,6 +310,36 @@ Status SegmentReadAhead::create_for_query(io::FileReaderSPtr source_reader, Exec
     return Status::OK();
 }
 
+SegmentReadAheadResult SegmentReadAhead::prefetch_by_rowids(
+        const rowid_t* rowids, size_t count, std::span<ColumnIterator* const> columns) {
+    DORIS_CHECK(rowids != nullptr);
+    DORIS_CHECK(count > 0);
+    DORIS_CHECK(std::is_sorted(rowids, rowids + count));
+    DORIS_CHECK(std::adjacent_find(rowids, rowids + count) == rowids + count);
+
+    roaring::Roaring scan_rowids;
+    scan_rowids.addMany(count, rowids);
+    std::vector<ColumnReadAheadPlan> plans;
+    const ColumnReadAheadRequest request {
+            .current_rowids = rowids,
+            .current_rowid_count = count,
+            .scan_rowids = &scan_rowids,
+            .context = &_column_context,
+            .role = ColumnReadAheadRole::EAGER,
+            .reverse = false,
+    };
+    for (auto* column : columns) {
+        DORIS_CHECK(column != nullptr);
+        const auto status = column->prepare_read_ahead(request, &plans);
+        if (!status.ok()) {
+            LOG_EVERY_N(WARNING, 100)
+                    << "failed to prepare exact row-id read-ahead; use the original read path: "
+                    << status;
+        }
+    }
+    return apply_plans(std::move(plans));
+}
+
 SegmentReadAheadResult SegmentReadAhead::apply_plans(std::vector<ColumnReadAheadPlan> plans) {
     COUNTER_UPDATE(&_statistics->plan_calls, 1);
     SCOPED_TIMER(&_statistics->plan_time);

be/src/storage/segment/segment_read_ahead.h

@@ -22,6 +22,7 @@
 #include <map>
 #include <memory>
 #include <mutex>
+#include <span>
 #include <vector>
 
 #include "io/cache/file_range_planner.h"
@@ -178,6 +179,10 @@ public:
     /// the original reader.
     SegmentReadAheadResult apply_plans(std::vector<ColumnReadAheadPlan> plans);
 
+    /// Plans all data pages touched by one exact row-id batch before any column starts decoding.
+    SegmentReadAheadResult prefetch_by_rowids(const rowid_t* rowids, size_t count,
+                                              std::span<ColumnIterator* const> columns);
+
     const std::shared_ptr<SegmentReadAheadFileReader>& file_reader() const { return _reader; }
 
     const ColumnReadAheadContext& column_context() const { return _column_context; }

be/src/storage/tablet/base_tablet.cpp

@@ -61,6 +61,8 @@
 #include "storage/rowset/rowset_reader.h"
 #include "storage/rowset/rowset_writer_context.h"
 #include "storage/segment/column_reader.h"
+#include "storage/segment/rowid_read_ahead.h"
+#include "storage/segment/segment_read_ahead.h"
 #include "storage/tablet/tablet_fwd.h"
 #include "storage/txn/txn_manager.h"
 #include "util/bvar_helper.h"
@@ -146,6 +148,41 @@ Status _get_segment_column_iterator(const BetaRowsetSharedPtr& rowset, uint32_t
                                          input_io_ctx);
 }
 
+Status _read_row_store_with_read_ahead(
+        const BetaRowsetSharedPtr& rowset, const RowLocation& row_location,
+        const segment_v2::SegmentSharedPtr& segment, const TabletColumn& column,
+        OlapReaderStatistics& stats, const io::IOContext* io_ctx,
+        const std::shared_ptr<io::FileRangeReadContext>& read_ahead_context,
+        MutableColumnPtr& result) {
+    DORIS_CHECK(read_ahead_context != nullptr);
+
+    StorageReadOptions read_options(stats);
+    read_options.version = rowset->version();
+    read_options.commit_tso = rowset->commit_tso();
+    read_options.use_page_cache = !config::disable_storage_page_cache;
+    if (io_ctx != nullptr) {
+        read_options.io_ctx = *io_ctx;
+    }
+    read_options.io_ctx.reader_type = ReaderType::READER_QUERY;
+    read_options.io_ctx.file_cache_stats = &stats.file_cache_stats;
+    std::unique_ptr<segment_v2::SegmentReadAhead> read_ahead;
+    RETURN_IF_ERROR(segment_v2::SegmentReadAhead::create_for_query(
+            segment->file_reader(), ExecEnv::GetInstance(), read_ahead_context, read_options,
+            &read_ahead));
+
+    std::vector<segment_v2::rowid_t> rowids {static_cast<segment_v2::rowid_t>(row_location.row_id)};
+    std::unique_ptr<segment_v2::ColumnIterator> column_iterator;
+    // A switch can close after the point request creates its context. The factory then returns
+    // no read-ahead object; reuse the loaded segment with the original synchronous reader.
+    if (read_ahead == nullptr) {
+        RETURN_IF_ERROR(_init_segment_column_iterator(rowset, segment, column, &column_iterator,
+                                                      stats, io_ctx));
+        return column_iterator->read_by_rowids(rowids.data(), rowids.size(), result);
+    }
+    return segment_v2::read_column_by_rowids_with_read_ahead(
+            *segment, column, rowids, result, read_options, column_iterator, *read_ahead);
+}
+
 } // namespace
 
 extern MetricPrototype METRIC_query_scan_bytes;
@@ -540,6 +577,39 @@ Status BaseTablet::_lookup_row_data(const Slice& encoded_key, const RowLocation&
     return Status::OK();
 }
 
+Status BaseTablet::lookup_row_data_with_read_ahead(
+        const Slice& encoded_key, const RowLocation& row_location,
+        const segment_v2::SegmentSharedPtr& segment, RowsetSharedPtr input_rowset,
+        OlapReaderStatistics& stats, std::string& values, bool write_to_cache,
+        const io::IOContext* io_ctx,
+        const std::shared_ptr<io::FileRangeReadContext>& read_ahead_context) {
+    DCHECK(segment != nullptr);
+    DCHECK_EQ(segment->id(), row_location.segment_id);
+    MonotonicStopWatch watch;
+    size_t row_size = 1;
+    watch.start();
+    Defer _defer([&]() {
+        LOG_EVERY_N(INFO, 500) << "get a single_row, cost(us):" << watch.elapsed_time() / 1000
+                               << ", row_size:" << row_size;
+    });
+
+    BetaRowsetSharedPtr rowset = std::static_pointer_cast<BetaRowset>(input_rowset);
+    CHECK(rowset);
+    const TabletSchemaSPtr tablet_schema = rowset->tablet_schema();
+    const auto& column = *DORIS_TRY(tablet_schema->column(BeConsts::ROW_STORE_COL));
+    MutableColumnPtr column_ptr = ColumnString::create();
+    RETURN_IF_ERROR(_read_row_store_with_read_ahead(rowset, row_location, segment, column, stats,
+                                                    io_ctx, read_ahead_context, column_ptr));
+    assert(column_ptr->size() == 1);
+    auto* string_column = static_cast<ColumnString*>(column_ptr.get());
+    StringRef value = string_column->get_data_at(0);
+    values = value.to_string();
+    if (write_to_cache) {
+        RowCache::instance()->insert({tablet_id(), encoded_key}, Slice {value.data, value.size});
+    }
+    return Status::OK();
+}
+
 // NOLINTNEXTLINE(readability-function-size): Keep the existing rowset lookup flow together.
 Status BaseTablet::lookup_row_key(const Slice& encoded_key, TabletSchema* latest_schema,
                                   bool with_seq_col,

be/src/storage/tablet/base_tablet.h

@@ -38,6 +38,10 @@
 #include "util/bthread_shared_mutex.h"
 
 namespace doris {
+namespace io {
+class FileRangeReadContext;
+}
+
 struct RowSetSplits;
 struct RowsetWriterContext;
 class RowsetWriter;
@@ -189,6 +193,15 @@ public:
                            const segment_v2::SegmentSharedPtr& segment, RowsetSharedPtr rowset,
                            OlapReaderStatistics& stats, std::string& values,
                            bool write_to_cache = false, const io::IOContext* io_ctx = nullptr);
+
+    // Read from the resolved segment with query-scoped range read-ahead admission. The caller
+    // keeps the segment and its acquired rowset alive as with lookup_row_data above.
+    Status lookup_row_data_with_read_ahead(
+            const Slice& encoded_key, const RowLocation& row_location,
+            const segment_v2::SegmentSharedPtr& segment, RowsetSharedPtr rowset,
+            OlapReaderStatistics& stats, std::string& values, bool write_to_cache,
+            const io::IOContext* io_ctx,
+            const std::shared_ptr<io::FileRangeReadContext>& read_ahead_context);
     // Lookup the row location of `encoded_key`, the function sets `row_location` on success.
     // NOTE: the method only works in unique key model with primary key index, you will got a
     //       not supported error in other data model.

be/test/service/point_query_exector_test.cpp

@@ -25,6 +25,7 @@
 
 #include "cloud/cloud_storage_engine.h"
 #include "cloud/cloud_tablet.h"
+#include "cloud/config.h"
 #include "common/config.h"
 #include "common/object_pool.h"
 #include "core/block/block.h"
@@ -39,16 +40,83 @@
 #include "runtime/memory/cache_manager.h"
 #include "runtime/runtime_state.h"
 #include "service/point_query_executor.h"
+#include "storage/mow/mow_transform_test_base.h"
 #include "storage/options.h"
 #include "storage/rowset/beta_rowset.h"
 #include "storage/rowset/beta_rowset_writer.h"
 #include "storage/segment/segment_loader.h"
+#include "storage/segment/segment_read_ahead.h"
 #include "storage/storage_engine.h"
 #include "storage/tablet/tablet_schema.h"
+#include "util/defer_op.h"
 #include "util/thrift_util.h"
 
 namespace doris {
 
+class PointQueryReadAheadTest : public MowTransformTestBase {};
+
+TEST_F(PointQueryReadAheadTest, RowStoreFallsBackWhenReadAheadSwitchChanges) {
+    auto schema = create_row_store_schema();
+    TabletSharedPtr tablet;
+    auto rowset = write_rowset(schema, 5101, 2, {{1, 11}, {2, 22}}, &tablet);
+    ASSERT_NE(rowset, nullptr);
+    const RowLocation location {rowset->rowset_id(), 0, 1};
+    const std::string key = "key";
+    OlapReaderStatistics expected_stats;
+    std::string expected;
+    ASSERT_TRUE(
+            tablet->lookup_row_data(Slice(key), location, rowset, expected_stats, expected).ok());
+    SegmentCacheHandle segment_cache;
+    ASSERT_TRUE(
+            SegmentLoader::instance()
+                    ->load_segments(std::static_pointer_cast<BetaRowset>(rowset), &segment_cache)
+                    .ok());
+    ASSERT_EQ(segment_cache.get_segments().size(), 1);
+    const auto& segment = segment_cache.get_segments().front();
+
+    const auto old_cloud_unique_id = config::cloud_unique_id;
+    const bool old_read_ahead = config::enable_query_read_ahead;
+    const bool old_async_write = config::enable_async_file_cache_write;
+    auto* sync_point = SyncPoint::get_instance();
+    Defer restore {[&] {
+        sync_point->disable_processing();
+        config::cloud_unique_id = old_cloud_unique_id;
+        config::enable_query_read_ahead = old_read_ahead;
+        config::enable_async_file_cache_write = old_async_write;
+    }};
+    config::cloud_unique_id = "PointQueryReadAheadTest";
+
+    for (int switch_index = 0; switch_index < 2; ++switch_index) {
+        SCOPED_TRACE(switch_index);
+        config::enable_query_read_ahead = true;
+        config::enable_async_file_cache_write = true;
+        ASSERT_TRUE(segment_v2::SegmentReadAhead::enabled());
+        auto context = std::make_shared<io::FileRangeReadContext>(1024 * 1024);
+
+        // Model a switch closing after the request created its context and resolved the segment.
+        // Both the read-ahead and fallback paths must reuse that segment without loading it again.
+        if (switch_index == 0) {
+            config::enable_query_read_ahead = false;
+        } else {
+            config::enable_async_file_cache_write = false;
+        }
+        size_t segment_loads = 0;
+        SyncPoint::CallbackGuard guard;
+        sync_point->set_call_back(
+                "SegmentLoader::load_segments", [&](auto&&) { ++segment_loads; }, &guard);
+        sync_point->enable_processing();
+        OlapReaderStatistics stats;
+        std::string actual;
+        const auto status = tablet->lookup_row_data_with_read_ahead(
+                Slice(key), location, segment, rowset, stats, actual, false, nullptr, context);
+        sync_point->disable_processing();
+        ASSERT_TRUE(status.ok()) << status;
+        EXPECT_EQ(segment_loads, 0);
+        EXPECT_EQ(actual, expected);
+        EXPECT_EQ(context->resident_bytes(), 0);
+    }
+}
+
 // Helper class for setting up Reusable objects to test LookupConnectionCache
 class ReusableTestHelper {
 public:

be/test/storage/segment/segment_read_ahead_test.cpp

@@ -20,6 +20,7 @@
 #include <butil/iobuf.h>
 #include <gtest/gtest.h>
 
+#include <array>
 #include <atomic>
 #include <cstring>
 #include <memory>
@@ -29,11 +30,22 @@
 #include <vector>
 
 #include "common/cast_set.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_struct.h"
 #include "io/fs/read_ahead_metrics.h"
+#include "runtime/descriptor_helper.h"
+#include "runtime/descriptors.h"
+#include "storage/iterators.h"
 #include "storage/olap_common.h"
+#include "storage/schema.h"
 #include "storage/segment/column_reader.h"
+#include "storage/segment/mock/mock_segment.h"
 #include "storage/segment/page_handle.h"
 #include "storage/segment/page_io.h"
+#include "storage/segment/rowid_read_ahead.h"
 #include "util/coding.h"
 #include "util/threadpool.h"
 
@@ -92,6 +104,44 @@ private:
     bool _closed {false};
 };
 
+class ExactRowIdColumn final : public ColumnIterator {
+public:
+    explicit ExactRowIdColumn(ColumnReadAhead* window) : _window(window) {
+        DORIS_CHECK(_window != nullptr);
+    }
+
+    Status seek_to_ordinal(ordinal_t) override { return Status::OK(); }
+
+    Status next_batch(size_t*, MutableColumnPtr&, bool*) override { return Status::OK(); }
+
+    Status read_by_rowids(const rowid_t*, size_t, MutableColumnPtr&) override {
+        return Status::OK();
+    }
+
+    ordinal_t get_current_ordinal() const override { return 0; }
+
+    Status prepare_read_ahead(const ColumnReadAheadRequest& request,
+                              std::vector<ColumnReadAheadPlan>* plans) override {
+        ++prepare_calls;
+        last_role = request.role;
+        scan_rowids = *request.scan_rowids;
+        ColumnReadAheadPlan plan;
+        _window->plan(request.current_rowids, request.current_rowid_count, *request.scan_rowids,
+                      &plan);
+        if (!plan.empty()) {
+            plans->push_back(std::move(plan));
+        }
+        return Status::OK();
+    }
+
+    size_t prepare_calls {0};
+    ColumnReadAheadRole last_role {ColumnReadAheadRole::LAZY};
+    roaring::Roaring scan_rowids;
+
+private:
+    ColumnReadAhead* _window;
+};
+
 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;
@@ -167,6 +217,89 @@ roaring::Roaring rows(size_t count) {
     return result;
 }
 
+TEST(SegmentReadAheadTest, ReadsSlotDefaultBeforeNewColumnReachesTabletSchema) {
+    auto source = std::make_shared<TestFileReader>(std::string(256, '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 schema = std::make_shared<TabletSchema>();
+    testing::NiceMock<MockSegment> segment(schema);
+    segment.set_file_reader_for_test(source);
+    ON_CALL(segment, file_reader()).WillByDefault(testing::Return(source));
+    auto footer = std::make_shared<SegmentFooterPB>();
+    footer->set_num_rows(3);
+    EXPECT_CALL(segment, _get_segment_footer(testing::_, testing::_, testing::_))
+            .WillOnce([&](std::shared_ptr<SegmentFooterPB>& result, OlapReaderStatistics*,
+                          const io::IOContext*) {
+                result = footer;
+                return Status::OK();
+            });
+
+    auto slot_data = TSlotDescriptorBuilder()
+                             .type(TYPE_INT)
+                             .column_name("new_column")
+                             .set_nullIndicatorBit(-1)
+                             .build();
+    slot_data.__set_col_unique_id(123);
+    slot_data.__set_col_default_value("42");
+    SlotDescriptor slot(slot_data);
+    MutableColumnPtr result = ColumnInt32::create();
+    OlapReaderStatistics stats;
+    StorageReadOptions options(stats);
+    std::unique_ptr<ColumnIterator> iterator;
+    RowIdColumnRead column {
+            .slot = &slot, .result = &result, .read_options = &options, .iterator = &iterator};
+    const std::vector<uint32_t> row_ids {0, 2};
+    ASSERT_TRUE(read_columns_by_rowids_with_read_ahead(segment, *schema, row_ids,
+                                                       std::span<RowIdColumnRead>(&column, 1),
+                                                       *read_ahead)
+                        .ok());
+    const auto& values = assert_cast<const ColumnInt32&>(*result).get_data();
+    ASSERT_EQ(values.size(), 2);
+    EXPECT_EQ(values[0], 42);
+    EXPECT_EQ(values[1], 42);
+    EXPECT_EQ(source->read_calls(), 0);
+}
+
+TEST(SegmentReadAheadTest, PointReadPreparationLeavesComplexIteratorsOnOriginalPath) {
+    auto source = std::make_shared<TestFileReader>(std::string(256, '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 = {}});
+    const auto integer = std::make_shared<DataTypeInt32>();
+    const DataTypes types {
+            std::make_shared<DataTypeArray>(integer),
+            std::make_shared<DataTypeMap>(integer, integer),
+            std::make_shared<DataTypeStruct>(DataTypes {integer}, Strings {"child"})};
+    for (const auto& type : types) {
+        SCOPED_TRACE(type->get_name());
+        auto slot_type = type->to_thrift();
+        auto slot_data = TSlotDescriptorBuilder()
+                                 .set_slotType(slot_type)
+                                 .column_name("complex_column")
+                                 .set_nullIndicatorBit(-1)
+                                 .build();
+        slot_data.__set_col_unique_id(123);
+        SlotDescriptor slot(slot_data);
+        auto schema = std::make_shared<TabletSchema>();
+        schema->append_column(ReadSchema::tablet_column_from_slot(slot));
+        testing::StrictMock<MockSegment> segment(schema);
+        OlapReaderStatistics stats;
+        StorageReadOptions options(stats);
+        std::unique_ptr<ColumnIterator> iterator;
+        RowIdColumnRead column {.slot = &slot, .read_options = &options, .iterator = &iterator};
+        const std::vector<uint32_t> row_ids {0, 2};
+        ASSERT_TRUE(prepare_columns_by_rowids_with_read_ahead(
+                            segment, *schema, row_ids, std::span<RowIdColumnRead>(&column, 1),
+                            *read_ahead)
+                            .ok());
+        EXPECT_EQ(iterator, nullptr);
+    }
+    EXPECT_EQ(source->read_calls(), 0);
+}
+
 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'));
@@ -233,6 +366,47 @@ TEST(SegmentReadAheadTest, CoalescesColumnsAndServesExactPageSlices) {
     EXPECT_GT(statistics.read_time.value(), 0);
 }
 
+TEST(SegmentReadAheadTest, PrefetchesExactRowIdsAcrossColumnsInOneSubmission) {
+    auto source = std::make_shared<TestFileReader>(std::string(128, '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 = {}},
+            {.window_bytes = 64}, {.window_bytes = 32});
+    auto first_window = make_window(1, 16, 0);
+    auto second_window = make_window(1, 16, 24);
+    ExactRowIdColumn first(first_window.get());
+    ExactRowIdColumn second(second_window.get());
+    std::array<ColumnIterator*, 2> columns {&first, &second};
+    const std::array<rowid_t, 1> rowids {0};
+
+    const auto result = read_ahead->prefetch_by_rowids(rowids.data(), rowids.size(), columns);
+
+    ASSERT_TRUE(result.accepted()) << result.status;
+    EXPECT_EQ(result.new_pages, 2);
+    EXPECT_EQ(result.submitted_ranges, 1);
+    EXPECT_EQ(result.submitted_bytes, 40);
+    EXPECT_EQ(first.prepare_calls, 1);
+    EXPECT_EQ(second.prepare_calls, 1);
+    EXPECT_EQ(first.last_role, ColumnReadAheadRole::EAGER);
+    EXPECT_EQ(second.last_role, ColumnReadAheadRole::EAGER);
+    EXPECT_TRUE(first.scan_rowids.contains(0));
+    EXPECT_TRUE(second.scan_rowids.contains(0));
+
+    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_EQ(bytes_read, sizeof(first_data));
+    ASSERT_TRUE(read_ahead->file_reader()
+                        ->read_at(24, Slice(second_data, sizeof(second_data)), &bytes_read)
+                        .ok());
+    ASSERT_EQ(bytes_read, sizeof(second_data));
+    EXPECT_EQ(source->read_calls(), 1);
+}
+
 TEST(SegmentReadAheadTest, SharesOnePhysicalPageAcrossColumnOwners) {
     auto source = std::make_shared<TestFileReader>(std::string(64, 'x'));
     auto scheduler = make_scheduler();