diff --git a/be/src/exec/operator/file_scan_operator.cpp b/be/src/exec/operator/file_scan_operator.cpp
index 5290e3078fb171..6e2fb81602e4cd 100644
--- a/be/src/exec/operator/file_scan_operator.cpp
+++ b/be/src/exec/operator/file_scan_operator.cpp
@@ -135,13 +135,13 @@ ScannerScheduler* FileScanLocalState::scan_scheduler(RuntimeState* state) const
 bool FileScanLocalState::TEST_should_use_file_scanner_v2(const TQueryOptions& query_options,
                                                          bool is_load,
                                                          const TFileScanRangeParams& scan_params) {
-    return _should_use_file_scanner_v2(query_options, is_load, scan_params);
+    return should_use_file_scanner_v2(query_options, is_load, scan_params);
 }
 #endif
 
-bool FileScanLocalState::_should_use_file_scanner_v2(const TQueryOptions& query_options,
-                                                     bool is_load,
-                                                     const TFileScanRangeParams& scan_params) {
+bool FileScanLocalState::should_use_file_scanner_v2(const TQueryOptions& query_options,
+                                                    bool is_load,
+                                                    const TFileScanRangeParams& scan_params) {
     // ADBC only has a FileScannerV2 reader, and enable_file_scanner_v2 is a session variable marked
     // fuzzy=true, so the regression harness flips it to false at random. Without letting adbc
     // through unconditionally, those queries land in v1, which has no "adbc" branch, and come back
@@ -190,7 +190,7 @@ Status FileScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
             state()->desc_tbl().get_tuple_descriptor(scan_params->src_tuple_id) != nullptr;
     // TODO: Use scanner v2 for all queries.
     const bool use_file_scanner_v2 =
-            _should_use_file_scanner_v2(state()->query_options(), is_load, *scan_params);
+            should_use_file_scanner_v2(state()->query_options(), is_load, *scan_params);
     _operator_profile->add_info_string("UseScannerV2", use_file_scanner_v2 ? "true" : "false");
     const auto* output_tuple_desc = state()->desc_tbl().get_tuple_descriptor(_output_tuple_id);
     DORIS_CHECK(output_tuple_desc != nullptr);
diff --git a/be/src/exec/operator/file_scan_operator.h b/be/src/exec/operator/file_scan_operator.h
index 6128715ddb7242..45901d828582ad 100644
--- a/be/src/exec/operator/file_scan_operator.h
+++ b/be/src/exec/operator/file_scan_operator.h
@@ -54,6 +54,8 @@ class FileScanLocalState final : public ScanLocalState<FileScanLocalState> {
     int max_scanners_concurrency(RuntimeState* state) const override;
     int min_scanners_concurrency(RuntimeState* state) const override;
     ScannerScheduler* scan_scheduler(RuntimeState* state) const override;
+    static bool should_use_file_scanner_v2(const TQueryOptions& query_options, bool is_load,
+                                           const TFileScanRangeParams& scan_params);
 #ifdef BE_TEST
     static bool TEST_should_use_file_scanner_v2(const TQueryOptions& query_options, bool is_load,
                                                 const TFileScanRangeParams& scan_params);
@@ -69,8 +71,6 @@ class FileScanLocalState final : public ScanLocalState<FileScanLocalState> {
         return PushDownType::PARTIAL_ACCEPTABLE;
     }
     bool _push_down_topn(const RuntimePredicate& predicate) override;
-    static bool _should_use_file_scanner_v2(const TQueryOptions& query_options, bool is_load,
-                                            const TFileScanRangeParams& scan_params);
 
     PushDownType _should_push_down_is_null_predicate(VectorizedFnCall* fn_call) const override {
         return fn_call->fn().name.function_name == "is_null_pred" ||
diff --git a/be/src/exec/rowid_fetcher.cpp b/be/src/exec/rowid_fetcher.cpp
index 05a62e70e1399c..fca26050618af2 100644
--- a/be/src/exec/rowid_fetcher.cpp
+++ b/be/src/exec/rowid_fetcher.cpp
@@ -42,7 +42,9 @@
 #include "core/column/column.h"
 #include "core/data_type/data_type_struct.h"
 #include "core/data_type_serde/data_type_serde.h"
+#include "exec/operator/file_scan_operator.h"
 #include "exec/scan/file_scanner.h"
+#include "exec/scan/file_scanner_v2.h"
 #include "format/orc/vorc_reader.h"
 #include "format/parquet/vparquet_reader.h"
 #include "io/io_common.h"
@@ -437,6 +439,28 @@ const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseRowsRead
 const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseSegmentsRead =
         "TopNLazyMaterializationSecondPhaseSegmentsRead";
 
+bool RowIdStorageReader::should_use_file_scanner_v2(const TQueryOptions& query_options,
+                                                    const TFileScanRangeParams& scan_params,
+                                                    const TFileRangeDesc& range) {
+    const auto format_type =
+            range.__isset.format_type ? range.format_type : scan_params.format_type;
+    // Phase two inherits the query options, including the Thrift presence bit. Reuse phase one's
+    // policy so disabling V2 (or an older payload omitting the option) also keeps row fetches on V1.
+    return FileScanLocalState::should_use_file_scanner_v2(query_options, false, scan_params) &&
+           (format_type == TFileFormatType::FORMAT_PARQUET ||
+            format_type == TFileFormatType::FORMAT_ORC) &&
+           FileScannerV2::is_supported(scan_params, range);
+}
+
+TFileRangeDesc RowIdStorageReader::build_external_fetch_range(const TFileRangeDesc& source_range) {
+    // Rows were selected after delete filtering. Preserve the original path and row lineage
+    // needed by virtual columns, and do not mutate the FileMapping shared by other fetches.
+    auto range = source_range;
+    range.table_format_params.iceberg_params.__set_delete_files({});
+    range.table_format_params.transactional_hive_params = TTransactionalHiveDesc {};
+    return range;
+}
+
 Status RowIdStorageReader::read_external_row_from_file_mapping(
         size_t idx, const std::multimap<segment_v2::rowid_t, size_t>& row_ids,
         const std::shared_ptr<FileMapping>& file_mapping,
@@ -467,25 +491,30 @@ Status RowIdStorageReader::read_external_row_from_file_mapping(
     scan_blocks[idx] = Block(scan_slots, read_ids.size());
 
     auto& external_info = file_mapping->get_external_file_info();
-    auto& scan_range_desc = external_info.scan_range_desc;
-
-    // Clear to avoid reading iceberg position delete file...
-    scan_range_desc.table_format_params.iceberg_params = TIcebergFileDesc {};
-
-    // Clear to avoid reading hive transactional delete delta file...
-    scan_range_desc.table_format_params.transactional_hive_params = TTransactionalHiveDesc {};
+    auto scan_range_desc = build_external_fetch_range(external_info.scan_range_desc);
 
     std::unique_ptr<RuntimeProfile> sub_runtime_profile =
             std::make_unique<RuntimeProfile>("ExternalRowIDFetcher");
     {
-        std::unique_ptr<FileScanner> vfile_scanner_ptr =
-                FileScanner::create_unique(runtime_state.get(), sub_runtime_profile.get(),
-                                           &rpc_scan_params, &colname_to_slot_id, &tuple_desc);
-
-        RETURN_IF_ERROR(vfile_scanner_ptr->prepare_for_read_lines(scan_range_desc));
-        RETURN_IF_ERROR(vfile_scanner_ptr->read_lines_from_range(
-                scan_range_desc, read_ids, &scan_blocks[idx], external_info,
-                &fetch_statistics[idx].init_reader_ms, &fetch_statistics[idx].get_block_ms));
+        if (should_use_file_scanner_v2(runtime_state->query_options(), rpc_scan_params,
+                                       scan_range_desc)) {
+            auto file_scanner = FileScannerV2::create_unique(
+                    runtime_state.get(), sub_runtime_profile.get(), &rpc_scan_params,
+                    &colname_to_slot_id, &tuple_desc);
+            RETURN_IF_ERROR(file_scanner->read_by_rows(scan_range_desc, read_ids, &scan_blocks[idx],
+                                                       &fetch_statistics[idx].init_reader_ms,
+                                                       &fetch_statistics[idx].get_block_ms));
+        } else {
+            // Phase one can still use V1 for table-format variants unsupported by V2, so keep the
+            // matching phase-two path instead of turning a previously valid query into an error.
+            auto file_scanner =
+                    FileScanner::create_unique(runtime_state.get(), sub_runtime_profile.get(),
+                                               &rpc_scan_params, &colname_to_slot_id, &tuple_desc);
+            RETURN_IF_ERROR(file_scanner->prepare_for_read_lines(scan_range_desc));
+            RETURN_IF_ERROR(file_scanner->read_lines_from_range(
+                    scan_range_desc, read_ids, &scan_blocks[idx], external_info,
+                    &fetch_statistics[idx].init_reader_ms, &fetch_statistics[idx].get_block_ms));
+        }
     }
 
     if (scan_blocks[idx].rows() != read_ids.size()) {
@@ -506,7 +535,7 @@ Status RowIdStorageReader::read_external_row_from_file_mapping(
     }
 
     auto file_read_bytes_counter =
-            sub_runtime_profile->get_counter(FileScanner::FileReadBytesProfile);
+            sub_runtime_profile->get_counter(FileScannerV2::FileReadBytesProfile);
 
     if (file_read_bytes_counter != nullptr) {
         fetch_statistics[idx].file_read_bytes = PrettyPrinter::print(
@@ -514,7 +543,7 @@ Status RowIdStorageReader::read_external_row_from_file_mapping(
     }
 
     auto file_read_times_counter =
-            sub_runtime_profile->get_counter(FileScanner::FileReadTimeProfile);
+            sub_runtime_profile->get_counter(FileScannerV2::FileReadTimeProfile);
     if (file_read_times_counter != nullptr) {
         fetch_statistics[idx].file_read_times = PrettyPrinter::print(
                 file_read_times_counter->value(), file_read_times_counter->type());
@@ -609,6 +638,47 @@ Status RowIdStorageReader::submit_external_scan_tasks(
     return scan_status.ok() ? Status::OK() : scan_status.status();
 }
 
+TFileScanRangeParams RowIdStorageReader::build_external_scan_params(
+        const TFileScanRangeParams& source_params, const TFileRangeDesc& range,
+        const std::vector<SlotDescriptor>& scan_slots,
+        const std::vector<uint32_t>& scan_column_idxs) {
+    DORIS_CHECK(scan_slots.size() == scan_column_idxs.size());
+    auto params = source_params;
+    params.required_slots.clear();
+    params.column_idxs.clear();
+    params.slot_name_to_schema_pos.clear();
+    const std::set partition_names(range.columns_from_path_keys.begin(),
+                                   range.columns_from_path_keys.end());
+    for (size_t slot_idx = 0; slot_idx < scan_slots.size(); ++slot_idx) {
+        const auto& slot = scan_slots[slot_idx];
+        const auto column_idx = scan_column_idxs[slot_idx];
+        TFileScanSlotInfo slot_info;
+        slot_info.__set_slot_id(slot.id());
+        // Hive V2 checks the Thrift presence bit before trusting is_file_slot. Without it,
+        // partition columns consume physical file indexes and invalidate the rebuilt projection.
+        bool is_file_slot = !partition_names.contains(slot.col_name());
+        if (source_params.__isset.column_name_to_category) {
+            // Lazy metadata slots may be absent from phase one's required_slots and have new
+            // slot IDs here. The pinned schema's name map preserves their original categories.
+            const auto it = source_params.column_name_to_category.find(slot.col_name());
+            const auto category = it != source_params.column_name_to_category.end()
+                                          ? it->second
+                                          : TColumnCategory::REGULAR;
+            slot_info.__set_category(category);
+            is_file_slot =
+                    category == TColumnCategory::REGULAR || category == TColumnCategory::GENERATED;
+        }
+        slot_info.__set_is_file_slot(is_file_slot);
+        if (is_file_slot) {
+            params.column_idxs.emplace_back(column_idx);
+        }
+        params.default_value_of_src_slot.emplace(slot.id(), TExpr {});
+        params.required_slots.emplace_back(slot_info);
+        params.slot_name_to_schema_pos.emplace(slot.col_name(), column_idx);
+    }
+    return params;
+}
+
 Status RowIdStorageReader::read_batch_external_row(
         const uint64_t workload_group_id, const PRequestBlockDesc& request_block_desc,
         std::shared_ptr<IdFileMap> id_file_map, std::vector<SlotDescriptor>& slots,
@@ -641,15 +711,6 @@ Status RowIdStorageReader::read_batch_external_row(
 
         DCHECK(id_file_map->get_external_scan_params().contains(plan_node_id));
         const auto* old_scan_params = &(id_file_map->get_external_scan_params().at(plan_node_id));
-        rpc_scan_params = *old_scan_params;
-
-        rpc_scan_params.required_slots.clear();
-        rpc_scan_params.column_idxs.clear();
-        rpc_scan_params.slot_name_to_schema_pos.clear();
-
-        std::set partition_name_set(first_scan_range_desc.columns_from_path_keys.begin(),
-                                    first_scan_range_desc.columns_from_path_keys.end());
-
         std::unordered_map<std::string, size_t> source_column_to_scan_idx;
 
         result_column_to_scan_column.reserve(slots.size());
@@ -668,24 +729,12 @@ Status RowIdStorageReader::read_batch_external_row(
             }
         }
 
-        for (auto slot_idx = 0; slot_idx < scan_slots.size(); ++slot_idx) {
-            auto& slot = scan_slots[slot_idx];
+        for (auto& slot : scan_slots) {
             tuple_desc.add_slot(&slot);
             colname_to_slot_id[slot.col_name()] = slot.id();
-            TFileScanSlotInfo slot_info;
-            slot_info.slot_id = slot.id();
-            auto column_idx = scan_column_idxs[slot_idx];
-            if (partition_name_set.contains(slot.col_name())) {
-                //This is partition column.
-                slot_info.is_file_slot = false;
-            } else {
-                rpc_scan_params.column_idxs.emplace_back(column_idx);
-                slot_info.is_file_slot = true;
-            }
-            rpc_scan_params.default_value_of_src_slot.emplace(slot.id(), TExpr {});
-            rpc_scan_params.required_slots.emplace_back(slot_info);
-            rpc_scan_params.slot_name_to_schema_pos.emplace(slot.col_name(), column_idx);
         }
+        rpc_scan_params = build_external_scan_params(*old_scan_params, first_scan_range_desc,
+                                                     scan_slots, scan_column_idxs);
 
         const auto& query_options = id_file_map->get_query_options();
         const auto& query_globals = id_file_map->get_query_globals();
@@ -866,9 +915,9 @@ Status RowIdStorageReader::read_batch_external_row(
                                          std::to_string(*init_reader_avg_ms) + "ms");
         runtime_profile->add_info_string(FileReadLinesProfile,
                                          fmt::to_string(file_read_lines_buffer));
-        runtime_profile->add_info_string(FileScanner::FileReadBytesProfile,
+        runtime_profile->add_info_string(FileScannerV2::FileReadBytesProfile,
                                          fmt::to_string(file_read_bytes_buffer));
-        runtime_profile->add_info_string(FileScanner::FileReadTimeProfile,
+        runtime_profile->add_info_string(FileScannerV2::FileReadTimeProfile,
                                          fmt::to_string(file_read_times_buffer));
     }
 
diff --git a/be/src/exec/rowid_fetcher.h b/be/src/exec/rowid_fetcher.h
index 7c2dd3d137827b..9b98d7b9ab141f 100644
--- a/be/src/exec/rowid_fetcher.h
+++ b/be/src/exec/rowid_fetcher.h
@@ -35,6 +35,7 @@
 namespace doris {
 
 class RuntimeState;
+class TQueryOptions;
 class TupleDescriptor;
 class ScannerScheduler;
 namespace io {
@@ -81,6 +82,9 @@ class RowIdStorageReader {
     static Status read_by_rowids(const PMultiGetRequestV2& request, PMultiGetResponseV2* response);
 
 private:
+    static bool should_use_file_scanner_v2(const TQueryOptions& query_options,
+                                           const TFileScanRangeParams& scan_params,
+                                           const TFileRangeDesc& range);
     struct ExternalFetchStatistics;
 
     static Status read_doris_format_row(
@@ -118,6 +122,13 @@ class RowIdStorageReader {
             const std::unordered_map<std::string, int>& colname_to_slot_id,
             std::counting_semaphore<>& semaphore, TupleDescriptor& tuple_desc);
 
+    static TFileRangeDesc build_external_fetch_range(const TFileRangeDesc& source_range);
+
+    static TFileScanRangeParams build_external_scan_params(
+            const TFileScanRangeParams& source_params, const TFileRangeDesc& range,
+            const std::vector<SlotDescriptor>& scan_slots,
+            const std::vector<uint32_t>& scan_column_idxs);
+
     static std::string source_column_key(const SlotDescriptor& slot, uint32_t column_idx);
 
     friend class RowIdStorageReaderTest;
diff --git a/be/src/exec/scan/file_scanner_v2.cpp b/be/src/exec/scan/file_scanner_v2.cpp
index 96387b92f6603c..c64d64fbd1fc6b 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -75,9 +75,13 @@
 #include "runtime/runtime_state.h"
 #include "service/backend_options.h"
 #include "storage/id_manager.h"
+#include "util/stopwatch.hpp"
 #include "util/string_util.h"
 
 namespace doris {
+const std::string FileScannerV2::FileReadBytesProfile = "FileReadBytes";
+const std::string FileScannerV2::FileReadTimeProfile = "FileReadTime";
+
 namespace {
 
 constexpr int kIcebergPositionDeleteContent = 1;
@@ -397,12 +401,12 @@ Status FileScannerV2::init(RuntimeState* state, const VExprContextSPtrs& conjunc
                                                            file_scan_profile::SCANNER, 1);
     _file_counter = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "FileNumber", TUnit::UNIT,
                                                  file_scan_profile::SCANNER, 1);
-    _file_read_bytes_counter = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "FileReadBytes", TUnit::BYTES,
-                                                            file_scan_profile::IO, 1);
+    _file_read_bytes_counter = ADD_CHILD_COUNTER_WITH_LEVEL(profile, FileReadBytesProfile,
+                                                            TUnit::BYTES, file_scan_profile::IO, 1);
     _file_read_calls_counter = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "FileReadCalls", TUnit::UNIT,
                                                             file_scan_profile::IO, 1);
     _file_read_time_counter =
-            ADD_CHILD_TIMER_WITH_LEVEL(profile, "FileReadTime", file_scan_profile::IO, 1);
+            ADD_CHILD_TIMER_WITH_LEVEL(profile, FileReadTimeProfile, file_scan_profile::IO, 1);
     _adaptive_batch_predicted_rows_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
             profile, "AdaptiveBatchPredictedRows", TUnit::UNIT, file_scan_profile::SCANNER, 1);
     _adaptive_batch_actual_bytes_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
@@ -593,18 +597,21 @@ Status FileScannerV2::_init_table_reader(const TFileRangeDesc& range) {
     VExprContextSPtrs table_conjuncts;
     RETURN_IF_ERROR(_build_table_conjuncts(&table_conjuncts));
     std::optional<std::vector<format::GlobalIndex>> push_down_count_columns;
-    const auto& push_down_count_slot_ids = _local_state->get_push_down_count_slot_ids();
-    if (push_down_count_slot_ids.has_value()) {
-        push_down_count_columns.emplace();
-        push_down_count_columns->reserve(push_down_count_slot_ids->size());
-        for (const auto slot_id : *push_down_count_slot_ids) {
-            const auto global_index_it = _slot_id_to_global_index.find(slot_id);
-            if (global_index_it == _slot_id_to_global_index.end()) {
-                return Status::InternalError(
-                        "Pushed-down COUNT argument is not a projected file scan slot, slot_id={}",
-                        slot_id);
+    if (_local_state != nullptr) {
+        const auto& push_down_count_slot_ids = _local_state->get_push_down_count_slot_ids();
+        if (push_down_count_slot_ids.has_value()) {
+            push_down_count_columns.emplace();
+            push_down_count_columns->reserve(push_down_count_slot_ids->size());
+            for (const auto slot_id : *push_down_count_slot_ids) {
+                const auto global_index_it = _slot_id_to_global_index.find(slot_id);
+                if (global_index_it == _slot_id_to_global_index.end()) {
+                    return Status::InternalError(
+                            "Pushed-down COUNT argument is not a projected file scan slot, "
+                            "slot_id={}",
+                            slot_id);
+                }
+                push_down_count_columns->push_back(global_index_it->second);
             }
-            push_down_count_columns->push_back(global_index_it->second);
         }
     }
     RETURN_IF_ERROR(_table_reader->init({
@@ -614,15 +621,97 @@ Status FileScannerV2::_init_table_reader(const TFileRangeDesc& range) {
             .scan_params = const_cast<TFileScanRangeParams*>(_params),
             .io_ctx = _io_ctx,
             .runtime_state = _state,
-            .scanner_profile = _local_state->scanner_profile(),
+            .scanner_profile = _local_state != nullptr ? _local_state->scanner_profile() : _profile,
             .file_slot_descs = &_file_slot_descs,
-            .push_down_agg_type = _local_state->get_push_down_agg_type(),
+            .push_down_agg_type = _local_state != nullptr ? _local_state->get_push_down_agg_type()
+                                                          : TPushAggOp::type::NONE,
             .push_down_count_columns = std::move(push_down_count_columns),
-            .condition_cache_digest = _local_state->get_condition_cache_digest(),
+            .condition_cache_digest =
+                    _local_state != nullptr ? _local_state->get_condition_cache_digest() : 0,
     }));
     return Status::OK();
 }
 
+Status FileScannerV2::read_by_rows(const TFileRangeDesc& range, const std::list<int64_t>& row_ids,
+                                   Block* result_block, int64_t* init_reader_ms,
+                                   int64_t* get_block_ms) {
+    DORIS_CHECK(result_block != nullptr);
+    DORIS_CHECK(init_reader_ms != nullptr);
+    DORIS_CHECK(get_block_ms != nullptr);
+    _current_range = range;
+    RETURN_IF_ERROR(_validate_scan_range(*_params, range));
+    const auto format_type = get_range_format_type(*_params, range);
+    if (format_type != TFileFormatType::FORMAT_PARQUET &&
+        format_type != TFileFormatType::FORMAT_ORC) {
+        return Status::NotSupported(
+                "FileScannerV2 row-id fetch supports only Parquet and ORC, file format={}",
+                to_string(format_type));
+    }
+
+    _file_cache_statistics = std::make_unique<io::FileCacheStatistics>();
+    _file_reader_stats = std::make_unique<io::FileReaderStats>();
+    _file_read_bytes_counter =
+            ADD_COUNTER_WITH_LEVEL(_profile, FileReadBytesProfile, TUnit::BYTES, 1);
+    _file_read_time_counter = ADD_TIMER_WITH_LEVEL(_profile, FileReadTimeProfile, 1);
+    RETURN_IF_ERROR(_init_io_ctx());
+    _io_ctx->file_cache_stats = _file_cache_statistics.get();
+    _io_ctx->file_reader_stats = _file_reader_stats.get();
+    _io_ctx->is_disposable = _state->query_options().disable_file_cache;
+
+    MonotonicStopWatch init_watch;
+    init_watch.start();
+    auto init_status = [&]() -> Status {
+        RETURN_IF_ERROR(_create_table_reader_for_format(range, &_table_reader));
+        DORIS_CHECK(_table_reader != nullptr);
+        RETURN_IF_ERROR(_init_expr_ctxes());
+        RETURN_IF_ERROR(_init_table_reader(range));
+        std::map<std::string, Field> partition_values;
+        RETURN_IF_ERROR(_generate_partition_values(range, &partition_values));
+        format::FileFormat current_split_format;
+        RETURN_IF_ERROR(_to_file_format(format_type, &current_split_format));
+        std::vector<int64_t> requested_rows(row_ids.begin(), row_ids.end());
+        _table_reader->set_batch_size(std::max<size_t>(requested_rows.size(), 1));
+        RETURN_IF_ERROR(_table_reader->prepare_split({
+                .partition_values = std::move(partition_values),
+                .conjuncts = std::nullopt,
+                .partition_prune_conjuncts = {},
+                .all_runtime_filters_applied = true,
+                .condition_cache_digest = 0,
+                .cache = nullptr,
+                .current_range = range,
+                .current_split_format = current_split_format,
+                .global_rowid_context = std::nullopt,
+                .row_ids = std::move(requested_rows),
+        }));
+        return Status::OK();
+    }();
+    *init_reader_ms += init_watch.elapsed_time() / 1000 / 1000;
+    RETURN_IF_ERROR(init_status);
+
+    MonotonicStopWatch read_watch;
+    read_watch.start();
+    auto read_status = [&]() -> Status {
+        Block read_block = result_block->clone_empty();
+        ScopedMutableBlock mutable_result(result_block);
+        bool eof = false;
+        while (!eof) {
+            RETURN_IF_ERROR(_table_reader->get_block(&read_block, &eof));
+            if (read_block.rows() > 0) {
+                RETURN_IF_ERROR(mutable_result.mutable_block().merge(read_block));
+            }
+        }
+        return Status::OK();
+    }();
+    *get_block_ms += read_watch.elapsed_time() / 1000 / 1000;
+    RETURN_IF_ERROR(read_status);
+
+    RETURN_IF_ERROR(_table_reader->close());
+    _table_reader.reset();
+    COUNTER_UPDATE(_file_read_bytes_counter, _file_reader_stats->read_bytes);
+    COUNTER_UPDATE(_file_read_time_counter, _file_reader_stats->read_time_ns);
+    return Status::OK();
+}
+
 Status FileScannerV2::_create_table_reader_for_format(
         const TFileRangeDesc& range, std::unique_ptr<format::TableReader>* reader) const {
     DORIS_CHECK(reader != nullptr);
@@ -693,6 +782,7 @@ Status FileScannerV2::_prepare_table_reader_split(const TFileRangeDesc& range,
             .current_range = range,
             .current_split_format = current_split_format,
             .global_rowid_context = _create_global_rowid_context(range),
+            .row_ids = std::nullopt,
     }));
     return Status::OK();
 }
diff --git a/be/src/exec/scan/file_scanner_v2.h b/be/src/exec/scan/file_scanner_v2.h
index ed1dce988219ad..13a6d593a3ca84 100644
--- a/be/src/exec/scan/file_scanner_v2.h
+++ b/be/src/exec/scan/file_scanner_v2.h
@@ -17,6 +17,7 @@
 
 #pragma once
 
+#include <list>
 #include <map>
 #include <memory>
 #include <optional>
@@ -53,6 +54,8 @@ class FileScannerV2 final : public Scanner {
 public:
     static constexpr const char* NAME = "FileScannerV2";
     static constexpr size_t ADAPTIVE_BATCH_INITIAL_PROBE_ROWS = 32;
+    static const std::string FileReadBytesProfile;
+    static const std::string FileReadTimeProfile;
 
     struct RealtimeCounterDeltas {
         int64_t scan_rows = 0;
@@ -108,6 +111,18 @@ class FileScannerV2 final : public Scanner {
                   ShardedKVCache* kv_cache,
                   const std::unordered_map<std::string, int>* colname_to_slot_id);
 
+    // Standalone scanner used by TopN two-phase materialization.
+    FileScannerV2(RuntimeState* state, RuntimeProfile* profile, const TFileScanRangeParams* params,
+                  const std::unordered_map<std::string, int>* colname_to_slot_id,
+                  TupleDescriptor* tuple_desc)
+            : Scanner(state, profile), _params(params) {
+        (void)colname_to_slot_id;
+        _output_tuple_desc = tuple_desc;
+    }
+
+    Status read_by_rows(const TFileRangeDesc& range, const std::list<int64_t>& row_ids,
+                        Block* result_block, int64_t* init_reader_ms, int64_t* get_block_ms);
+
     Status init(RuntimeState* state, const VExprContextSPtrs& conjuncts) override;
     Status _open_impl(RuntimeState* state) override;
     Status close(RuntimeState* state) override;
diff --git a/be/src/format_v2/file_reader.cpp b/be/src/format_v2/file_reader.cpp
index 8a9df5d7953454..2d94a23d673170 100644
--- a/be/src/format_v2/file_reader.cpp
+++ b/be/src/format_v2/file_reader.cpp
@@ -73,7 +73,13 @@ std::string FileScanRequest::debug_string() const {
         }
         out << column_id << ":" << block_position;
     }
-    out << "}, conjunct_count=" << conjuncts.size()
+    out << "}, row_ids=";
+    if (row_ids.has_value()) {
+        out << join_debug_strings(*row_ids, [](int64_t row_id) { return std::to_string(row_id); });
+    } else {
+        out << "nullopt";
+    }
+    out << ", conjunct_count=" << conjuncts.size()
         << ", metadata_pruning_safe_conjunct_count=" << metadata_pruning_safe_conjunct_count
         << ", constant_pruning_safe_table_filter_count=" << constant_pruning_safe_table_filter_count
         << ", delete_conjunct_count=" << delete_conjuncts.size() << ", variant_schema_overrides="
diff --git a/be/src/format_v2/file_reader.h b/be/src/format_v2/file_reader.h
index 91a7873d84df42..700ccf9439d546 100644
--- a/be/src/format_v2/file_reader.h
+++ b/be/src/format_v2/file_reader.h
@@ -18,6 +18,7 @@
 #include <algorithm>
 #include <cstddef>
 #include <cstdint>
+#include <functional>
 #include <limits>
 #include <map>
 #include <memory>
@@ -102,6 +103,12 @@ struct FileScanRequest {
     // predicate_columns, the value is semantically required and must still be validated and read.
     std::vector<LocalColumnId> count_star_placeholder_columns;
 
+    // Absolute zero-based file row positions selected by a row-id fetch. A present but empty
+    // vector means that no rows should be read; nullopt keeps the normal sequential scan path.
+    // Readers require strictly increasing positions so they can seek forward without duplicating
+    // output rows.
+    std::optional<std::vector<int64_t>> row_ids = std::nullopt;
+
     // Table formats may assign semantics that legacy physical files do not encode. Each path here
     // identifies an unannotated Parquet group that the physical reader must validate and decode as
     // Variant. Keeping this explicit prevents generic Parquet scans from guessing based on names.
@@ -374,8 +381,21 @@ class FileReader {
     virtual std::unique_ptr<TableColumnMapper> create_column_mapper(
             TableColumnMapperOptions options) const;
 
+    virtual bool supports_rowid_fetch() const { return false; }
+
     // Open the file reader with file-local scan request. The file reader should initialize its internal state according to the request, but does not need to interpret table/global schema semantics. For example, all schema change, filter localization, default/generated/partition columns should be handled in table reader layer. This method can only be called after init() successfully.
     virtual Status open(std::shared_ptr<FileScanRequest> request) {
+        if (request->row_ids.has_value()) {
+            if (!supports_rowid_fetch()) {
+                return Status::NotSupported("File reader does not support row-id fetch");
+            }
+            const auto& row_ids = *request->row_ids;
+            if (std::ranges::any_of(row_ids, [](int64_t row_id) { return row_id < 0; }) ||
+                std::ranges::adjacent_find(row_ids, std::greater_equal<>()) != row_ids.end()) {
+                return Status::InvalidArgument(
+                        "Row-id fetch requires non-negative, strictly increasing file row ids");
+            }
+        }
         _request = std::move(request);
         return Status::OK();
     }
diff --git a/be/src/format_v2/orc/orc_reader.cpp b/be/src/format_v2/orc/orc_reader.cpp
index 83cb929e18bccc..77654578c91d2e 100644
--- a/be/src/format_v2/orc/orc_reader.cpp
+++ b/be/src/format_v2/orc/orc_reader.cpp
@@ -778,6 +778,7 @@ struct OrcReaderScanState {
     std::vector<StripeRange> selected_stripe_ranges;
     size_t current_stripe_range = 0;
     bool stripe_pruning_applied = false;
+    size_t next_row_id = 0;
 
     bool row_reader_created = false;
 };
@@ -1232,6 +1233,7 @@ Status OrcReader::open(std::shared_ptr<format::FileScanRequest> request) {
         return Status::Uninitialized("OrcReader is not open");
     }
     RETURN_IF_ERROR(format::FileReader::open(std::move(request)));
+    _state->next_row_id = 0;
 
     if (_request->local_positions.empty()) {
         size_t next_position = 0;
@@ -1289,6 +1291,16 @@ Status OrcReader::open(std::shared_ptr<format::FileScanRequest> request) {
     _apply_current_stripe_range();
 
     RETURN_IF_ERROR(_create_row_reader());
+    if (_request->row_ids.has_value()) {
+        for (const int64_t row_id : *_request->row_ids) {
+            if (static_cast<uint64_t>(row_id) < _state->row_reader_range_first_row ||
+                static_cast<uint64_t>(row_id) >= _state->row_reader_range_end_row) {
+                return Status::InvalidArgument(
+                        "ORC row id {} is outside the current split row range [{}, {})", row_id,
+                        _state->row_reader_range_first_row, _state->row_reader_range_end_row);
+            }
+        }
+    }
     _eof = get_total_rows() == 0;
     return Status::OK();
 }
@@ -1643,7 +1655,12 @@ Status OrcReader::_create_row_reader() {
                 _state->orc_lazy_read_enabled ? _orc_filter.get() : nullptr);
         _state->selected_type = &_state->row_reader->getSelectedType();
         DORIS_CHECK(_state->selected_type->getKind() == ::orc::TypeKind::STRUCT);
-        _state->batch = _state->row_reader->createRowBatch(DEFAULT_ORC_READ_BATCH_SIZE);
+        // Row-id fetch seeks before every read; a one-row batch preserves exact selection instead
+        // of also returning the sequential rows that follow the requested position.
+        const uint64_t batch_size = _request != nullptr && _request->row_ids.has_value()
+                                            ? 1
+                                            : DEFAULT_ORC_READ_BATCH_SIZE;
+        _state->batch = _state->row_reader->createRowBatch(batch_size);
         _state->orc_lazy_selection_valid = false;
         _state->orc_lazy_selected_rows.clear();
         _state->orc_lazy_input_rows = 0;
@@ -1940,11 +1957,23 @@ Status OrcReader::get_block(Block* file_block, size_t* rows, bool* eof) {
     }
 
     bool has_next = false;
+    std::optional<uint64_t> fetched_row_id;
     while (true) {
         try {
+            if (_request->row_ids.has_value()) {
+                if (_state->next_row_id >= _request->row_ids->size()) {
+                    _eof = true;
+                    *eof = true;
+                    return Status::OK();
+                }
+                fetched_row_id = static_cast<uint64_t>((*_request->row_ids)[_state->next_row_id]);
+                _state->row_reader->seekToRow(*fetched_row_id);
+            }
             // Condition-cache seeks can perform I/O, so keep them in the same cancellation
             // boundary as next().
-            _skip_condition_cache_false_granules(rows, eof);
+            if (!_request->row_ids.has_value()) {
+                _skip_condition_cache_false_granules(rows, eof);
+            }
             if (*eof) {
                 return Status::OK();
             }
@@ -1952,6 +1981,9 @@ Status OrcReader::get_block(Block* file_block, size_t* rows, bool* eof) {
             _state->orc_lazy_selected_rows.clear();
             _state->orc_lazy_input_rows = 0;
             has_next = _state->row_reader->next(*_state->batch);
+            if (_request->row_ids.has_value() && has_next) {
+                ++_state->next_row_id;
+            }
         } catch (const std::exception& e) {
             if (is_orc_stop(_io_ctx.get(), e)) {
                 file_block->clear_column_data(file_block->columns());
@@ -1972,6 +2004,10 @@ Status OrcReader::get_block(Block* file_block, size_t* rows, bool* eof) {
             }
             break;
         }
+        if (_request->row_ids.has_value()) {
+            return Status::InternalError("ORC row id {} could not be read from the current split",
+                                         *fetched_row_id);
+        }
         bool advanced = false;
         RETURN_IF_ERROR(_advance_to_next_stripe_range(&advanced));
         if (!advanced) {
@@ -1982,7 +2018,7 @@ Status OrcReader::get_block(Block* file_block, size_t* rows, bool* eof) {
     }
 
     const auto batch_rows = static_cast<size_t>(_state->batch->numElements);
-    const auto batch_first_row = _state->row_reader->getRowNumber();
+    const auto batch_first_row = fetched_row_id.value_or(_state->row_reader->getRowNumber());
     _state->current_batch_first_row = batch_first_row;
     _state->condition_cache_next_row = _state->current_batch_first_row + batch_rows;
     auto* struct_batch = dynamic_cast<::orc::StructVectorBatch*>(_state->batch.get());
diff --git a/be/src/format_v2/orc/orc_reader.h b/be/src/format_v2/orc/orc_reader.h
index a67dc5ad7f5e3e..a58af5451a73e8 100644
--- a/be/src/format_v2/orc/orc_reader.h
+++ b/be/src/format_v2/orc/orc_reader.h
@@ -58,6 +58,7 @@ class OrcReader final : public format::FileReader {
     Status get_schema(std::vector<format::ColumnDefinition>* const file_schema) const override;
     std::unique_ptr<format::TableColumnMapper> create_column_mapper(
             format::TableColumnMapperOptions options) const override;
+    bool supports_rowid_fetch() const override { return true; }
     Status open(std::shared_ptr<format::FileScanRequest> request) override;
     Status get_block(Block* file_block, size_t* rows, bool* eof) override;
     Status get_aggregate_result(const format::FileAggregateRequest& request,
diff --git a/be/src/format_v2/parquet/parquet_reader.h b/be/src/format_v2/parquet/parquet_reader.h
index 5a1ebd64614b01..61aad3362bb4b2 100644
--- a/be/src/format_v2/parquet/parquet_reader.h
+++ b/be/src/format_v2/parquet/parquet_reader.h
@@ -68,6 +68,7 @@ class ParquetReader : public format::FileReader {
 
     std::unique_ptr<format::TableColumnMapper> create_column_mapper(
             format::TableColumnMapperOptions options) const override;
+    bool supports_rowid_fetch() const override { return true; }
 
     Status open(std::shared_ptr<format::FileScanRequest> request) override;
 
diff --git a/be/src/format_v2/parquet/parquet_scan.cpp b/be/src/format_v2/parquet/parquet_scan.cpp
index b4cd731f12bb4d..dc2741f7d3e324 100644
--- a/be/src/format_v2/parquet/parquet_scan.cpp
+++ b/be/src/format_v2/parquet/parquet_scan.cpp
@@ -556,7 +556,28 @@ Status build_native_row_group_read_plans(
         row_group_plan.row_group_id = row_group_idx;
         row_group_plan.first_file_row = row_group_first_rows[row_group_idx];
         row_group_plan.row_group_rows = row_group.num_rows;
-        row_group_plan.selected_ranges = {{.start = 0, .length = row_group.num_rows}};
+        if (request.row_ids.has_value()) {
+            const auto& row_ids = *request.row_ids;
+            const int64_t row_group_end = row_group_plan.first_file_row + row_group.num_rows;
+            auto row_id = std::ranges::lower_bound(row_ids, row_group_plan.first_file_row);
+            const auto row_id_end = std::ranges::lower_bound(row_id, row_ids.end(), row_group_end);
+            for (; row_id != row_id_end; ++row_id) {
+                const int64_t local_row = *row_id - row_group_plan.first_file_row;
+                if (!row_group_plan.selected_ranges.empty() &&
+                    row_group_plan.selected_ranges.back().start +
+                                    row_group_plan.selected_ranges.back().length ==
+                            local_row) {
+                    ++row_group_plan.selected_ranges.back().length;
+                } else {
+                    row_group_plan.selected_ranges.push_back({.start = local_row, .length = 1});
+                }
+            }
+            if (row_group_plan.selected_ranges.empty()) {
+                continue;
+            }
+        } else {
+            row_group_plan.selected_ranges = {{.start = 0, .length = row_group.num_rows}};
+        }
         row_group_plan.expensive_pruning_pending = true;
         plan->row_groups.push_back(std::move(row_group_plan));
     }
@@ -1214,14 +1235,14 @@ Status ParquetScanScheduler::open_next_row_group(
     RETURN_IF_ERROR(detail::build_native_prefetch_ranges(
             thrift_metadata, file_schema, request_scan_columns(request), row_group_idx,
             file_context.native_file->size(), compat.parquet_816_padding, &native_ranges));
-    if (request.non_predicate_positions.empty()) {
+    if (!request.row_ids.has_value() && request.non_predicate_positions.empty()) {
         _current_merge_range_active = file_context.set_native_random_access_ranges(
                 native_ranges, detail::average_prefetch_range_size(native_ranges), _profile,
                 _merge_read_slice_size);
     } else {
-        // Independent predicate/output readers may revisit the same physical leaf at different
-        // cursors. MergeRangeFileReader has one consumptive cache per range, so use the random
-        // access reader for this layout instead of sharing one sequential range cache.
+        // Row-ID reads must not merge whole chunks containing unselected rows. Independent
+        // predicate/output readers also need random access: they can revisit one physical leaf
+        // at different cursors, while MergeRangeFileReader has one consumptive cache per range.
         _current_merge_range_active = file_context.set_native_random_access_ranges(
                 {}, 0, _profile, _merge_read_slice_size);
     }
@@ -1263,8 +1284,9 @@ Status ParquetScanScheduler::open_next_row_group(
     // changes row/column materialization order.
     if (!_current_merge_range_active) {
         const auto prefetch_columns = adaptive_predicate_prefetch_columns(request);
-        RETURN_IF_ERROR(prefetch_current_row_group_columns(
-                file_context, file_schema, prefetch_columns, &_current_predicate_prefetched));
+        RETURN_IF_ERROR(prefetch_current_row_group_columns(file_context, file_schema, request,
+                                                           prefetch_columns,
+                                                           &_current_predicate_prefetched));
     }
     for (const auto& col : request.non_predicate_columns) {
         const auto local_id = col.column_id();
@@ -1302,7 +1324,7 @@ Status ParquetScanScheduler::open_next_row_group(
         // With no row-level filters there is no lazy-read decision to wait for, so start warming
         // output chunks immediately after their readers are created. Filtered scans still defer
         // this until at least one row survives the predicate phase.
-        RETURN_IF_ERROR(prefetch_current_row_group_columns(file_context, file_schema,
+        RETURN_IF_ERROR(prefetch_current_row_group_columns(file_context, file_schema, request,
                                                            physical_non_predicate_columns(request),
                                                            &_current_non_predicate_prefetched));
     }
@@ -2519,10 +2541,14 @@ Status ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
 Status ParquetScanScheduler::prefetch_current_row_group_columns(
         ParquetFileContext& file_context,
         const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+        const format::FileScanRequest& request,
         const std::vector<format::LocalColumnIndex>& scan_columns, bool* prefetched) {
     DORIS_CHECK(prefetched != nullptr);
-    if (_current_merge_range_active || *prefetched || scan_columns.empty() ||
-        _current_row_group_id < 0 || file_context.native_metadata == nullptr) {
+    // Row-ID requests remain selective even without conjuncts. Whole-chunk dry-run prefetch
+    // would download unselected bytes without query accounting; demand reads retain IOContext stats.
+    if (request.row_ids.has_value() || _current_merge_range_active || *prefetched ||
+        scan_columns.empty() || _current_row_group_id < 0 ||
+        file_context.native_metadata == nullptr) {
         return Status::OK();
     }
     *prefetched = true;
@@ -2640,7 +2666,7 @@ Status ParquetScanScheduler::read_current_row_group_batch(
         // Do not prefetch lazy output columns until at least one row survives filtering. This is
         // the same decision point where the v2 reader switches from predicate-only reads to
         // materializing non-predicate columns, so fully filtered batches avoid unnecessary IO.
-        RETURN_IF_ERROR(prefetch_current_row_group_columns(file_context, file_schema,
+        RETURN_IF_ERROR(prefetch_current_row_group_columns(file_context, file_schema, request,
                                                            physical_non_predicate_columns(request),
                                                            &_current_non_predicate_prefetched));
     }
diff --git a/be/src/format_v2/parquet/parquet_scan.h b/be/src/format_v2/parquet/parquet_scan.h
index 0cccbc31cb2398..25367c84595d4e 100644
--- a/be/src/format_v2/parquet/parquet_scan.h
+++ b/be/src/format_v2/parquet/parquet_scan.h
@@ -242,6 +242,7 @@ class ParquetScanScheduler {
     Status prefetch_current_row_group_columns(
             ParquetFileContext& file_context,
             const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+            const format::FileScanRequest& request,
             const std::vector<format::LocalColumnIndex>& scan_columns, bool* prefetched);
 
     Status read_current_row_group_batch(
diff --git a/be/src/format_v2/table_reader.cpp b/be/src/format_v2/table_reader.cpp
index 76aec6768934c1..3e654ab594a10f 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -1161,6 +1161,7 @@ Status TableReader::refresh_conjuncts(VExprContextSPtrs conjuncts) {
             _file_scan_request == nullptr ? nullptr : &_file_scan_request->local_positions,
             _file_scan_request == nullptr ? nullptr
                                           : &_file_scan_request->non_predicate_positions));
+    refreshed_request->row_ids = _row_ids;
     // A refresh does not prove that every future runtime filter has arrived. Keep carrier values
     // available whenever the split started with pending filters.
     if (_push_down_agg_type == TPushAggOp::type::COUNT && _push_down_count_columns.has_value() &&
@@ -1470,6 +1471,7 @@ Status TableReader::prepare_split(const SplitReadOptions& options) {
                                      ? std::make_optional(options.current_range.load_id)
                                      : std::nullopt;
     _global_rowid_context = options.global_rowid_context;
+    _row_ids = options.row_ids;
     _delete_rows = nullptr;
     _deletion_vector = nullptr;
     _aggregate_pushdown_tried = false;
@@ -1494,9 +1496,10 @@ Status TableReader::prepare_split(const SplitReadOptions& options) {
     // the NULL state of a COUNT argument. Require the new FE's explicit empty argument list, which
     // means COUNT(*)/COUNT(1). A non-empty list means COUNT(col), while nullopt comes from an old FE
     // whose COUNT semantics are unknown during a BE-first rolling upgrade.
-    if (_push_down_agg_type == TPushAggOp::type::COUNT && _push_down_count_columns.has_value() &&
-        _push_down_count_columns->empty() && options.all_runtime_filters_applied &&
-        _conjuncts.empty() && options.current_range.__isset.table_format_params &&
+    if (!_row_ids.has_value() && _push_down_agg_type == TPushAggOp::type::COUNT &&
+        _push_down_count_columns.has_value() && _push_down_count_columns->empty() &&
+        options.all_runtime_filters_applied && _conjuncts.empty() &&
+        options.current_range.__isset.table_format_params &&
         options.current_range.table_format_params.__isset.table_level_row_count) {
         DORIS_CHECK(options.current_range.table_format_params.table_level_row_count >= -1);
         _remaining_table_level_count =
diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h
index 097cee93734b2e..1f890a514fee98 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -186,6 +186,9 @@ struct SplitReadOptions {
     TFileRangeDesc current_range;
     FileFormat current_split_format = FileFormat::PARQUET;
     std::optional<GlobalRowIdContext> global_rowid_context;
+    // Optional absolute file-row selection used by TopN two-phase materialization. TableReader
+    // carries it unchanged into the format-neutral FileScanRequest.
+    std::optional<std::vector<int64_t>> row_ids = std::nullopt;
 };
 
 // Base class for table-level readers.
@@ -456,6 +459,7 @@ class TableReader {
         auto file_request = std::make_shared<FileScanRequest>();
         RETURN_IF_ERROR(_data_reader.column_mapper->create_scan_request(
                 _table_filters, _projected_columns, file_request.get(), _runtime_state));
+        file_request->row_ids = _row_ids;
         _constant_pruning_safe_filter_count =
                 std::min(_constant_pruning_safe_filter_count,
                          file_request->constant_pruning_safe_table_filter_count);
@@ -1048,7 +1052,7 @@ class TableReader {
         *pushed_down = false;
         block->clear_column_data(_projected_columns.size());
         _aggregate_pushdown_tried = true;
-        if (!_supports_aggregate_pushdown(_push_down_agg_type)) {
+        if (_row_ids.has_value() || !_supports_aggregate_pushdown(_push_down_agg_type)) {
             return Status::OK();
         }
 
@@ -2251,6 +2255,7 @@ class TableReader {
     // irreversible aggregate rows, not only the table-level row-count shortcut in prepare_split().
     bool _all_runtime_filters_applied_for_split = true;
     std::optional<GlobalRowIdContext> _global_rowid_context;
+    std::optional<std::vector<int64_t>> _row_ids;
     bool _aggregate_pushdown_tried = false;
     bool _current_split_pruned = false;
     TableColumnMapperOptions _mapper_options;
diff --git a/be/test/exec/rowid_fetcher_test.cpp b/be/test/exec/rowid_fetcher_test.cpp
index b2fcbbac23667b..8f14aa6654b136 100644
--- a/be/test/exec/rowid_fetcher_test.cpp
+++ b/be/test/exec/rowid_fetcher_test.cpp
@@ -22,8 +22,13 @@
 #include <string>
 #include <vector>
 
+#include "exec/operator/file_scan_operator.h"
+#include "exec/scan/file_scanner_v2.h"
+#include "format_v2/column_mapper.h"
+#include "format_v2/table/hive_reader.h"
 #include "runtime/descriptor_helper.h"
 #include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
 
 namespace doris {
 
@@ -40,6 +45,8 @@ class RowIdStorageReaderTest : public testing::Test {
 protected:
     struct SlotSpec {
         std::string col_name = "c";
+        int32_t slot_id = 0;
+        PrimitiveType type = TYPE_INT;
         int32_t col_unique_id = 1;
         std::vector<std::string> column_paths = {};
         TColumnAccessPaths access_paths = {};
@@ -47,11 +54,12 @@ class RowIdStorageReaderTest : public testing::Test {
 
     static SlotDescriptor make_slot(const SlotSpec& spec) {
         TSlotDescriptor tdesc = TSlotDescriptorBuilder()
-                                        .type(TYPE_INT)
+                                        .type(spec.type)
                                         .nullable(true)
                                         .column_name(spec.col_name)
                                         .column_pos(0)
                                         .build();
+        tdesc.__set_id(spec.slot_id);
         tdesc.__set_col_unique_id(spec.col_unique_id);
         tdesc.__set_column_paths(spec.column_paths);
         if (!spec.access_paths.empty()) {
@@ -76,6 +84,198 @@ class RowIdStorageReaderTest : public testing::Test {
     }
 };
 
+TEST_F(RowIdStorageReaderTest, ExternalScannerSelectionRespectsRolloutOption) {
+    for (auto format : {TFileFormatType::FORMAT_PARQUET, TFileFormatType::FORMAT_ORC}) {
+        TFileScanRangeParams params;
+        params.__set_format_type(format);
+        TFileRangeDesc range;
+        for (const auto& table_format : {"hive", "iceberg", "tvf"}) {
+            TTableFormatFileDesc table;
+            table.__set_table_format_type(table_format);
+            params.__set_table_format_params(table);
+            range.__set_table_format_params(table);
+            for (int option = 0; option < 3; ++option) {
+                TQueryOptions options;
+                if (option != 0) {
+                    options.__set_enable_file_scanner_v2(option == 2);
+                } else {
+                    // An absent Thrift field must not enable V2 even if its value defaults to true.
+                    options.enable_file_scanner_v2 = true;
+                    options.__isset.enable_file_scanner_v2 = false;
+                }
+                EXPECT_EQ(RowIdStorageReader::should_use_file_scanner_v2(options, params, range),
+                          FileScanLocalState::TEST_should_use_file_scanner_v2(options, false,
+                                                                              params));
+                EXPECT_EQ(RowIdStorageReader::should_use_file_scanner_v2(options, params, range),
+                          option == 2);
+            }
+        }
+    }
+}
+
+TEST_F(RowIdStorageReaderTest, ExternalScannerSelectionKeepsUnsupportedFormatsOnV1) {
+    TQueryOptions options;
+    options.__set_enable_file_scanner_v2(true);
+    TFileScanRangeParams params;
+    params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+    TFileRangeDesc range;
+    range.__set_format_type(TFileFormatType::FORMAT_JNI);
+    EXPECT_FALSE(RowIdStorageReader::should_use_file_scanner_v2(options, params, range));
+    range.__set_format_type(TFileFormatType::FORMAT_ORC);
+    TTableFormatFileDesc table;
+    table.__set_table_format_type("transactional_hive");
+    params.__set_table_format_params(table);
+    range.__set_table_format_params(table);
+    EXPECT_FALSE(RowIdStorageReader::should_use_file_scanner_v2(options, params, range));
+}
+
+TEST_F(RowIdStorageReaderTest, ExternalFetchPreservesIcebergFileMetadata) {
+    TFileRangeDesc range;
+    range.__set_path("normalized/data.parquet");
+    TIcebergFileDesc iceberg;
+    iceberg.__set_original_file_path("s3://bucket/data.parquet");
+    iceberg.__set_format_version(3);
+    iceberg.__set_first_row_id(128);
+    iceberg.__set_last_updated_sequence_number(7);
+    TIcebergDeleteFileDesc deletes;
+    deletes.__set_path("s3://bucket/deletes.parquet");
+    iceberg.__set_delete_files({deletes});
+    range.table_format_params.__set_iceberg_params(iceberg);
+
+    const auto fetch_range = RowIdStorageReader::build_external_fetch_range(range);
+    const auto& fetch_iceberg = fetch_range.table_format_params.iceberg_params;
+    EXPECT_TRUE(fetch_iceberg.__isset.original_file_path);
+    EXPECT_EQ(fetch_iceberg.original_file_path, iceberg.original_file_path);
+    EXPECT_EQ(fetch_iceberg.format_version, 3);
+    EXPECT_EQ(fetch_iceberg.first_row_id, 128);
+    EXPECT_EQ(fetch_iceberg.last_updated_sequence_number, 7);
+    EXPECT_TRUE(fetch_iceberg.delete_files.empty());
+    EXPECT_EQ(range.table_format_params.iceberg_params, iceberg);
+}
+
+TEST_F(RowIdStorageReaderTest, ExternalFetchPreservesPrunedMetadataCategories) {
+    TFileScanRangeParams source_params;
+    source_params.__set_column_name_to_category(
+            {{"_file", TColumnCategory::SYNTHESIZED},
+             {"_pos", TColumnCategory::SYNTHESIZED},
+             {"generated_col", TColumnCategory::GENERATED},
+             {"partition_col", TColumnCategory::PARTITION_KEY}});
+    // Phase one projects only the sort key; none of these fetch slots survives in required_slots.
+    TFileScanSlotInfo sort_slot;
+    sort_slot.__set_slot_id(99);
+    sort_slot.__set_category(TColumnCategory::REGULAR);
+    source_params.__set_required_slots({sort_slot});
+    source_params.__set_column_idxs({0});
+    std::vector<SlotDescriptor> slots;
+    for (const auto* name : {"_file", "_pos", "generated_col", "partition_col", "value"}) {
+        slots.emplace_back(
+                make_slot({.col_name = name,
+                           .slot_id = static_cast<int32_t>(slots.size()),
+                           .type = std::string_view(name) == "_file" ? TYPE_STRING : TYPE_BIGINT}));
+    }
+    const auto params = RowIdStorageReader::build_external_scan_params(
+            source_params, TFileRangeDesc {}, slots, {3, 4, 1, 2, 0});
+    EXPECT_EQ(params.column_idxs, (std::vector<int32_t> {1, 0}));
+    const std::vector<TColumnCategory::type> categories {
+            TColumnCategory::SYNTHESIZED, TColumnCategory::SYNTHESIZED, TColumnCategory::GENERATED,
+            TColumnCategory::PARTITION_KEY, TColumnCategory::REGULAR};
+    std::vector<format::ColumnDefinition> columns;
+    for (size_t i = 0; i < slots.size(); ++i) {
+        const auto& info = params.required_slots[i];
+        EXPECT_TRUE(info.__isset.category);
+        EXPECT_EQ(info.category, categories[i]);
+        EXPECT_EQ(info.is_file_slot, i == 2 || i == 4);
+        EXPECT_EQ(FileScannerV2::TEST_is_partition_slot(info, slots[i].col_name()), i == 3);
+        auto column = FileScannerV2::_build_table_column(&slots[i]);
+        column.is_synthesized =
+                info.__isset.category && info.category == TColumnCategory::SYNTHESIZED;
+        columns.emplace_back(std::move(column));
+    }
+    format::TableColumnMapper mapper({.mode = format::TableColumnMappingMode::BY_NAME,
+                                      .enable_iceberg_metadata_virtual_columns = true});
+    ASSERT_TRUE(mapper.create_mapping({columns[0], columns[1]}, {}, {}).ok());
+    EXPECT_EQ(mapper.mappings()[0].virtual_column_type,
+              format::TableVirtualColumnType::ICEBERG_FILE_PATH);
+    EXPECT_EQ(mapper.mappings()[1].virtual_column_type,
+              format::TableVirtualColumnType::ICEBERG_ROW_POSITION);
+
+    // An authoritative empty map means ordinary physical columns, even for metadata spellings.
+    source_params.__set_column_name_to_category({});
+    const auto physical_params = RowIdStorageReader::build_external_scan_params(
+            source_params, TFileRangeDesc {}, slots, {3, 4, 1, 2, 0});
+    for (const auto& info : physical_params.required_slots) {
+        EXPECT_EQ(info.category, TColumnCategory::REGULAR);
+        EXPECT_TRUE(info.is_file_slot);
+    }
+}
+
+// Row-id fetch rebuilds the projection after TopN. Hive's positional mapper must consume
+// indexes only for physical columns, including when partition columns precede file columns.
+TEST_F(RowIdStorageReaderTest, ExternalFetchPartitionSlotsPreserveHivePositionMapping) {
+    for (const auto format : {TFileFormatType::FORMAT_ORC, TFileFormatType::FORMAT_PARQUET}) {
+        TQueryOptions options;
+        options.__set_hive_orc_use_column_names(false);
+        options.__set_hive_parquet_use_column_names(false);
+        RuntimeState state(options, TQueryGlobals {});
+        TFileScanRangeParams source_params;
+        source_params.__set_format_type(format);
+        source_params.__set_column_idxs({0, 1, 2});
+        TFileScanSlotInfo old_slot;
+        old_slot.__set_slot_id(99);
+        source_params.__set_required_slots({old_slot});
+        source_params.__set_slot_name_to_schema_pos({{"old_column", 0}});
+        TFileRangeDesc range;
+        range.__set_columns_from_path_keys({"partition_col"});
+
+        for (const auto& names : {std::vector<std::string> {"value", "partition_col"},
+                                  std::vector<std::string> {"partition_col", "value", "id"},
+                                  std::vector<std::string> {"partition_col"},
+                                  std::vector<std::string> {"value", "id"}}) {
+            SCOPED_TRACE(fmt::format("format={}, columns={}", static_cast<int>(format),
+                                     fmt::join(names, ",")));
+            std::vector<SlotDescriptor> slots;
+            std::vector<uint32_t> indices;
+            std::vector<int32_t> file_indices;
+            for (const auto& name : names) {
+                slots.emplace_back(make_slot(
+                        {.col_name = name, .slot_id = static_cast<int32_t>(slots.size())}));
+                const uint32_t index = name == "partition_col" ? 3 : name == "value" ? 2 : 0;
+                indices.emplace_back(index);
+                if (name != "partition_col") {
+                    file_indices.emplace_back(index);
+                }
+            }
+            const auto params = RowIdStorageReader::build_external_scan_params(source_params, range,
+                                                                               slots, indices);
+            ASSERT_EQ(params.required_slots.size(), slots.size());
+            EXPECT_EQ(params.column_idxs, file_indices);
+            EXPECT_FALSE(params.slot_name_to_schema_pos.contains("old_column"));
+            format::ProjectedColumnBuildContext context {
+                    .scan_params = &params, .range = &range, .runtime_state = &state};
+            format::hive::HiveReader reader;
+            for (size_t i = 0; i < slots.size(); ++i) {
+                const auto& slot_info = params.required_slots[i];
+                const auto& name = names[i];
+                const bool is_partition = name == "partition_col";
+                EXPECT_TRUE(slot_info.__isset.slot_id);
+                EXPECT_EQ(slot_info.slot_id, slots[i].id());
+                EXPECT_TRUE(slot_info.__isset.is_file_slot);
+                EXPECT_EQ(FileScannerV2::TEST_is_partition_slot(slot_info, name), is_partition);
+                format::ColumnDefinition column;
+                column.name = name;
+                column.type = slots[i].get_data_type_ptr();
+                const auto status = reader.annotate_projected_column(slot_info, &context, &column);
+                ASSERT_TRUE(status.ok()) << status;
+                if (!is_partition) {
+                    EXPECT_EQ(column.get_identifier_position(), indices[i]);
+                }
+            }
+            EXPECT_EQ(context.next_file_column_idx, file_indices.size());
+            EXPECT_TRUE(reader.validate_projected_columns(context).ok());
+        }
+    }
+}
+
 TEST_F(RowIdStorageReaderTest, SameSourceColumnSharesKey) {
     // The bug case: one physical column projected twice must dedup onto one scan column.
     const SlotDescriptor first = make_slot({});
diff --git a/be/test/format_v2/orc/orc_reader_test.cpp b/be/test/format_v2/orc/orc_reader_test.cpp
index 1058d774d44e60..5de8a81204c481 100644
--- a/be/test/format_v2/orc/orc_reader_test.cpp
+++ b/be/test/format_v2/orc/orc_reader_test.cpp
@@ -10178,6 +10178,38 @@ TEST_F(NewOrcReaderTest, CloseClearsFileLocalState) {
     EXPECT_FALSE(reader->open(request).ok());
 }
 
+TEST_F(NewOrcReaderTest, ReadsOnlyRequestedAbsoluteFileRows) {
+    auto reader = create_reader();
+    RuntimeState state {TQueryOptions(), TQueryGlobals()};
+    ASSERT_TRUE(reader->init(&state).ok());
+
+    std::vector<format::ColumnDefinition> schema;
+    ASSERT_TRUE(reader->get_schema(&schema).ok());
+    auto request = std::make_shared<format::FileScanRequest>();
+    request->non_predicate_columns = {field_projection(0)};
+    request->row_ids = {0, 2, 4};
+    ASSERT_TRUE(reader->open(request).ok());
+
+    std::vector<int32_t> ids;
+    bool eof = false;
+    while (!eof) {
+        Block block = build_file_block({schema[0]});
+        size_t rows = 0;
+        ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+        if (rows == 0) {
+            continue;
+        }
+        const auto& id_column = assert_cast<const ColumnInt32&>(
+                assert_cast<const ColumnNullable&>(*block.get_by_position(0).column)
+                        .get_nested_column());
+        for (size_t row = 0; row < rows; ++row) {
+            ids.push_back(id_column.get_element(row));
+        }
+    }
+
+    EXPECT_EQ(ids, std::vector<int32_t>({1, 3, 5}));
+}
+
 TEST_F(NewOrcReaderTest, ReadPrimitiveTypesWithNulls) {
     const auto primitive_file_path = (_test_dir / "primitive.orc").string();
     write_primitive_orc_file(primitive_file_path);
diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp b/be/test/format_v2/parquet/parquet_scan_test.cpp
index 17d80897b7aa22..6d10bc2d427fa5 100644
--- a/be/test/format_v2/parquet/parquet_scan_test.cpp
+++ b/be/test/format_v2/parquet/parquet_scan_test.cpp
@@ -25,7 +25,9 @@
 #include <parquet/arrow/writer.h>
 #include <parquet/encoding.h>
 
+#include <atomic>
 #include <bit>
+#include <chrono>
 #include <cstdlib>
 #include <cstring>
 #include <filesystem>
@@ -36,6 +38,7 @@
 #include <optional>
 #include <set>
 #include <string>
+#include <thread>
 #include <utility>
 #include <vector>
 
@@ -72,6 +75,11 @@
 #include "format_v2/parquet/reader/native_column_reader.h"
 #include "gen_cpp/PlanNodes_types.h"
 #include "gen_cpp/Types_types.h"
+#include "io/cache/block_file_cache.h"
+#include "io/cache/block_file_cache_factory.h"
+#include "io/cache/cached_remote_file_reader.h"
+#include "io/cache/fs_file_cache_storage.h"
+#include "io/fs/local_file_system.h"
 #include "io/io_common.h"
 #include "runtime/runtime_state.h"
 #include "storage/index/zone_map/zonemap_eval_context.h"
@@ -79,6 +87,8 @@
 #include "storage/utils.h"
 #include "testutil/mock/mock_query_context.h"
 #include "util/coding.h"
+#include "util/defer_op.h"
+#include "util/threadpool.h"
 #include "util/thrift_util.h"
 
 namespace doris {
@@ -2513,6 +2523,167 @@ TEST_F(ParquetScanTest, GlobalRowIdUsesFileLocalPositionForScanRange) {
     EXPECT_EQ(row_ids, std::vector<uint32_t>({2, 3}));
 }
 
+TEST_F(ParquetScanTest, ReadsOnlyRequestedAbsoluteFileRowsAcrossRowGroups) {
+    write_int_pair_parquet_file(_file_path, 2);
+    auto reader = create_reader();
+    RuntimeState state {TQueryOptions(), TQueryGlobals()};
+    ASSERT_TRUE(reader->init(&state).ok());
+
+    std::vector<format::ColumnDefinition> schema;
+    ASSERT_TRUE(reader->get_schema(&schema).ok());
+    auto request = std::make_shared<format::FileScanRequest>();
+    request->non_predicate_columns = {field_projection(0)};
+    request->row_ids = {0, 3, 5};
+    ASSERT_TRUE(reader->open(request).ok());
+
+    std::vector<int32_t> ids;
+    bool eof = false;
+    while (!eof) {
+        Block block = build_file_block({schema[0]});
+        size_t rows = 0;
+        ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+        if (rows == 0) {
+            continue;
+        }
+        const auto& id_column = int32_data_column(*block.get_by_position(0).column);
+        for (size_t row = 0; row < rows; ++row) {
+            ids.push_back(id_column.get_element(row));
+        }
+    }
+
+    EXPECT_EQ(ids, std::vector<int32_t>({1, 4, 6}));
+}
+
+class CountingParquetRemoteReader final : public io::FileReader {
+public:
+    explicit CountingParquetRemoteReader(io::FileReaderSPtr reader) : _reader(std::move(reader)) {}
+    Status close() override { return _reader->close(); }
+    const io::Path& path() const override { return _reader->path(); }
+    size_t size() const override { return _reader->size(); }
+    bool closed() const override { return _reader->closed(); }
+    int64_t mtime() const override { return _reader->mtime(); }
+    std::atomic<size_t> remote_bytes {0};
+    std::atomic<size_t> dryrun_bytes {0};
+
+protected:
+    Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
+                        const io::IOContext* io_ctx) override {
+        RETURN_IF_ERROR(_reader->read_at(offset, result, bytes_read, io_ctx));
+        remote_bytes += *bytes_read;
+        if (io_ctx != nullptr && io_ctx->is_dryrun) {
+            dryrun_bytes += *bytes_read;
+        }
+        return Status::OK();
+    }
+
+private:
+    io::FileReaderSPtr _reader;
+};
+
+TEST_F(ParquetScanTest, SparseRowIdsAvoidCachedRemoteChunkPrefetch) {
+    using namespace format::parquet;
+    // Use multiple cache blocks so an eager chunk read cannot hide inside one demand read.
+    std::vector<int32_t> ids(1024 * 1024);
+    std::iota(ids.begin(), ids.end(), 0);
+    auto table = arrow::Table::Make(arrow::schema({arrow::field("id", arrow::int32(), false)}),
+                                    {build_int32_array(ids)});
+    write_table(_file_path, table, ids.size());
+
+    auto* env = ExecEnv::GetInstance();
+    auto* old_factory = env->file_cache_factory();
+    auto factory = std::make_unique<io::FileCacheFactory>();
+    const auto cache_path = (_test_dir / "cache").string();
+    io::FileCacheSettings settings;
+    settings.storage = "disk";
+    settings.capacity = 16 * 1024 * 1024;
+    settings.query_queue_size = settings.capacity;
+    settings.query_queue_elements = 1024;
+    settings.max_file_block_size = 64 * 1024;
+    const auto old_ttl_gc_interval = config::file_cache_background_ttl_gc_interval_ms;
+    const auto old_ttl_info_interval = config::file_cache_background_ttl_info_update_interval_ms;
+    // The TTL workers sleep between iterations, so bound the fixture's shutdown latency.
+    config::file_cache_background_ttl_gc_interval_ms = 100;
+    config::file_cache_background_ttl_info_update_interval_ms = 100;
+    const auto old_column_buffer = config::parquet_column_max_buffer_mb;
+    config::parquet_column_max_buffer_mb = 1;
+    const auto old_block_size = config::file_cache_each_block_size;
+    config::file_cache_each_block_size = settings.max_file_block_size;
+    auto old_fd_cache = std::move(env->_file_cache_open_fd_cache);
+    auto old_pool = std::move(env->_segment_prefetch_thread_pool);
+    Defer restore([&] {
+        env->_segment_prefetch_thread_pool.reset();
+        factory.reset();
+        env->set_file_cache_factory(old_factory);
+        env->_file_cache_open_fd_cache = std::move(old_fd_cache);
+        env->_segment_prefetch_thread_pool = std::move(old_pool);
+        config::file_cache_each_block_size = old_block_size;
+        config::parquet_column_max_buffer_mb = old_column_buffer;
+        config::file_cache_background_ttl_gc_interval_ms = old_ttl_gc_interval;
+        config::file_cache_background_ttl_info_update_interval_ms = old_ttl_info_interval;
+    });
+    env->set_file_cache_factory(factory.get());
+    env->_file_cache_open_fd_cache = std::make_unique<io::FDCache>();
+    ASSERT_TRUE(factory->create_file_cache(cache_path, settings).ok());
+    auto* cache = factory->get_by_path(cache_path);
+    ASSERT_NE(cache, nullptr);
+    for (int attempt = 0; attempt < 200 && !cache->get_async_open_success(); ++attempt) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(5));
+    }
+    ASSERT_TRUE(cache->get_async_open_success());
+    ASSERT_TRUE(ThreadPoolBuilder("parquet_test_prefetch")
+                        .set_min_threads(1)
+                        .set_max_threads(1)
+                        .build(&ExecEnv::GetInstance()->_segment_prefetch_thread_pool)
+                        .ok());
+    io::FileReaderSPtr local;
+    ASSERT_TRUE(io::global_local_filesystem()->open_file(_file_path, &local).ok());
+    auto remote = std::make_shared<CountingParquetRemoteReader>(local);
+    io::FileReaderOptions options;
+    options.cache_type = io::FileCachePolicy::FILE_BLOCK_CACHE;
+    options.cache_base_path = cache_path;
+    options.cache_write_mode = io::CacheWriteMode::SYNC_WRITE;
+    auto cached = std::make_shared<io::CachedRemoteFileReader>(remote, options);
+    io::FileCacheStatistics cache_stats;
+    io::IOContext io_ctx;
+    io_ctx.file_cache_stats = &cache_stats;
+    io::FileDescription description;
+    description.path = _file_path;
+    description.file_size = local->size();
+    ParquetFileContext context;
+    ASSERT_TRUE(context.open(cached, &io_ctx, false, description).ok());
+    std::vector<std::unique_ptr<ParquetColumnSchema>> schema;
+    ASSERT_TRUE(build_parquet_column_schema(context.native_metadata->schema(), &schema).ok());
+    auto request = std::make_shared<format::FileScanRequest>();
+    request->non_predicate_columns = {field_projection(0)};
+    request->local_positions.emplace(format::LocalColumnId(0), format::LocalIndex(0));
+    request->row_ids = {0};
+    RuntimeState state {TQueryOptions(), TQueryGlobals()};
+    RowGroupScanPlan plan;
+    ASSERT_TRUE(plan_parquet_row_groups(*context.native_metadata, schema, *request, {}, false,
+                                        &plan, &state.timezone_obj(), &state, &context)
+                        .ok());
+    ParquetScanScheduler scheduler;
+    scheduler.set_plan(std::move(plan));
+    scheduler.set_scan_request(request);
+    scheduler.set_runtime_state(&state);
+    scheduler.set_timezone(&state.timezone_obj());
+    Block block;
+    block.insert({schema[0]->type->create_column(), schema[0]->type, "id"});
+    size_t rows = 0;
+    bool eof = false;
+    ASSERT_TRUE(scheduler.read_next_batch(context, schema, &block, &rows, &eof).ok());
+    ExecEnv::GetInstance()->segment_prefetch_thread_pool()->wait();
+    ASSERT_EQ(rows, 1);
+    EXPECT_EQ(int32_data_column(*block.get_by_position(0).column).get_element(0), 0);
+    EXPECT_FALSE(scheduler._current_merge_range_active);
+    EXPECT_FALSE(scheduler._current_non_predicate_prefetched);
+    EXPECT_LT(remote->remote_bytes.load(), local->size() / 2);
+    EXPECT_EQ(remote->dryrun_bytes.load(), 0);
+    // Source counters measure copied bytes; downloads can include cache-block alignment padding.
+    EXPECT_GT(cache_stats.bytes_read_from_remote, 0);
+    EXPECT_LE(cache_stats.bytes_read_from_remote, remote->remote_bytes.load());
+}
+
 TEST_F(ParquetScanTest, PredicateOnlyGlobalRowIdKeepsSignedFileLocalId) {
     write_int_pair_parquet_file(_file_path, 6, false);
     format::GlobalRowIdContext context {.version = 7, .backend_id = 123456789, .file_id = 42};
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileQueryScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileQueryScanNode.java
index 775b811e5ee147..63d45e4868880b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileQueryScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileQueryScanNode.java
@@ -241,12 +241,16 @@ private void updateRequiredSlots() throws UserException {
         setColumnPositionMapping();
     }
 
+    protected TColumnCategory classifyColumn(SlotDescriptor slot, List<String> partitionKeys) {
+        return classifyColumn(slot.getColumn().getName(), partitionKeys);
+    }
+
     /**
-     * Classify a column's category for the BE reader.
+     * Classify projected and lazy columns with the same connector-specific rules.
      * Subclasses override this for format-specific classification.
      */
-    protected TColumnCategory classifyColumn(SlotDescriptor slot, List<String> partitionKeys) {
-        if (partitionKeys.contains(slot.getColumn().getName())) {
+    protected TColumnCategory classifyColumn(String columnName, List<String> partitionKeys) {
+        if (partitionKeys.contains(columnName)) {
             return TColumnCategory.PARTITION_KEY;
         }
         return TColumnCategory.REGULAR;
@@ -300,8 +304,24 @@ private void setColumnPositionMapping()
         // Pre-index columns into a Map for O(1) lookup
         List<Column> columns = getPinnedFullSchema();
         Map<String, Integer> columnNameMap = new HashMap<>(columns.size());
+        boolean needsRowIdFetch = desc.getSlots().stream()
+                .anyMatch(slot -> slot.getColumn().getName().startsWith(Column.GLOBAL_ROWID_COL));
+        List<String> partitionKeys = needsRowIdFetch ? getPathPartitionKeys() : Collections.emptyList();
+        Map<String, TColumnCategory> columnCategories = new HashMap<>();
         for (int i = 0; i < columns.size(); i++) {
-            columnNameMap.putIfAbsent(columns.get(i).getName(), i);
+            String columnName = columns.get(i).getName();
+            columnNameMap.putIfAbsent(columnName, i);
+            if (needsRowIdFetch) {
+                TColumnCategory category = classifyColumn(columnName, partitionKeys);
+                if (category != TColumnCategory.REGULAR) {
+                    columnCategories.put(columnName, category);
+                }
+            }
+        }
+        if (needsRowIdFetch) {
+            // Lazy slots are already absent from the scan tuple. Preserve their categories from
+            // the pinned full schema so phase two cannot mistake metadata for physical columns.
+            params.setColumnNameToCategory(columnCategories);
         }
 
         for (TFileScanSlotInfo slot : params.getRequiredSlots()) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
index 3229689f0175d4..9af04f0f83373b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
@@ -782,8 +782,7 @@ protected List<String> getPathPartitionKeys() throws UserException {
      * Everything else falls through to {@code super} (partition key / regular).
      */
     @Override
-    protected TColumnCategory classifyColumn(SlotDescriptor slot, List<String> partitionKeys) {
-        String name = slot.getColumn().getName();
+    protected TColumnCategory classifyColumn(String name, List<String> partitionKeys) {
         if (name.startsWith(Column.GLOBAL_ROWID_COL)) {
             return TColumnCategory.SYNTHESIZED;
         }
@@ -794,7 +793,7 @@ protected TColumnCategory classifyColumn(SlotDescriptor slot, List<String> parti
         if (category == ConnectorColumnCategory.GENERATED) {
             return TColumnCategory.GENERATED;
         }
-        return super.classifyColumn(slot, partitionKeys);
+        return super.classifyColumn(name, partitionKeys);
     }
 
     /**
diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FileQueryScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FileQueryScanNodeTest.java
index 5f608d7349785f..1afdc77defebea 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FileQueryScanNodeTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FileQueryScanNodeTest.java
@@ -264,6 +264,46 @@ public void testHiveParquetTimezoneIgnoresPluginDrivenQueryTableValuedFunction()
         Assertions.assertFalse(node.getFileScanRangeParams().isSetHiveParquetTimeZone());
     }
 
+    @Test
+    public void testRowIdFetchRetainsCategoriesOfPrunedColumns() throws Exception {
+        TestFileQueryScanNode node = new TestFileQueryScanNode(new SessionVariable()) {
+            @Override
+            protected TColumnCategory classifyColumn(String name, List<String> partitionKeys) {
+                if (name.equals("_file") || name.equals("_pos")) {
+                    return TColumnCategory.SYNTHESIZED;
+                }
+                if (name.equals("generated_col")) {
+                    return TColumnCategory.GENERATED;
+                }
+                return super.classifyColumn(name, partitionKeys);
+            }
+        };
+        node.setTargetTable(table);
+        TupleDescriptor desc = node.getTupleDescriptor();
+        desc.setTable(table);
+        SlotDescriptor sortSlot = new SlotDescriptor(new SlotId(1), desc.getId());
+        sortSlot.setColumn(new Column("id", Type.INT));
+        desc.addSlot(sortSlot);
+        SlotDescriptor rowIdSlot = new SlotDescriptor(new SlotId(2), desc.getId());
+        rowIdSlot.setColumn(new Column(Column.GLOBAL_ROWID_COL, Type.STRING));
+        desc.addSlot(rowIdSlot);
+        List<Column> fullSchema = Arrays.asList(sortSlot.getColumn(), new Column("_file", Type.STRING),
+                new Column("_pos", Type.BIGINT), new Column("generated_col", Type.BIGINT));
+        Mockito.when(table.getBaseSchema(false)).thenReturn(fullSchema);
+        Mockito.when(table.getFullSchema()).thenReturn(fullSchema);
+
+        node.initSchemaParamsForTest();
+        UPDATE_REQUIRED_SLOTS_METHOD.invoke(node);
+
+        TFileScanRangeParams params = node.getFileScanRangeParams();
+        Assertions.assertEquals(2, params.getRequiredSlotsSize());
+        Assertions.assertEquals(Arrays.asList(0), params.getColumnIdxs());
+        Assertions.assertEquals(TColumnCategory.SYNTHESIZED, params.getColumnNameToCategory().get("_file"));
+        Assertions.assertEquals(TColumnCategory.SYNTHESIZED, params.getColumnNameToCategory().get("_pos"));
+        Assertions.assertEquals(TColumnCategory.GENERATED, params.getColumnNameToCategory().get("generated_col"));
+        Assertions.assertFalse(params.getColumnNameToCategory().containsKey("id"));
+    }
+
     @Test
     public void testUpdateRequiredSlotsPreservesInlineDefaultValueExpr() throws Exception {
         SessionVariable sv = new SessionVariable();
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index 9867ab16052eff..f18de426d4aad9 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -596,6 +596,9 @@ struct TFileScanRangeParams {
     // values unchanged. When present, only INT96 TIMESTAMP values are converted with this zone.
     36: optional string hive_parquet_time_zone
     37: optional TLanceScanParams lance_scan_params
+    // Non-regular columns in the pinned full schema, including columns pruned from phase one.
+    // When present, omitted names are REGULAR. Used to rebuild row-id fetch projections.
+    38: optional map<string, TColumnCategory> column_name_to_category
 }
 
 struct TFileRangeDesc {
diff --git a/regression-test/suites/external_table_p0/iceberg/test_iceberg_file_metadata_columns.groovy b/regression-test/suites/external_table_p0/iceberg/test_iceberg_file_metadata_columns.groovy
index d4dde31313ed14..6a19f385668f02 100644
--- a/regression-test/suites/external_table_p0/iceberg/test_iceberg_file_metadata_columns.groovy
+++ b/regression-test/suites/external_table_p0/iceberg/test_iceberg_file_metadata_columns.groovy
@@ -163,5 +163,21 @@ suite("test_iceberg_file_metadata_columns", "p0,external,iceberg,external_docker
         """
 
         verifyFileMetadata(tableName, format)
+
+        // Compare against an eager scan: file paths depend on the writer, and metadata slots
+        // pruned from phase one must still be synthesized during the row-id fetch.
+        for (String projection : ["`_file`", "`_pos`", "`_file`, `_pos`",
+                                  "id, payload, `_file`, `_pos`"]) {
+            String query = "select ${projection} from ${tableName} order by id limit 3"
+            sql "set topn_lazy_materialization_threshold = -1"
+            def eagerRows = sql query
+            sql "set topn_lazy_materialization_threshold = 10"
+            explain {
+                sql query
+                contains "VMaterializeNode"
+            }
+            assertEquals(eagerRows, sql(query))
+        }
+
     }
 }