branch-4.1: [Enhancement](file scanner) Support row-id fetch in FileScannerV2 #67906

be/src/exec/operator/file_scan_operator.cpp

@@ -137,13 +137,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) {
     const bool is_transactional_hive =
             scan_params.__isset.table_format_params &&
             scan_params.table_format_params.table_format_type == "transactional_hive";
@@ -156,7 +156,7 @@ bool FileScanLocalState::_can_generate_physical_splits(const TQueryOptions& quer
                                                        bool is_load,
                                                        const TFileScanRangeParams& scan_params,
                                                        const TFileRangeDesc& range) {
-    if (!_should_use_file_scanner_v2(query_options, is_load, scan_params)) {
+    if (!should_use_file_scanner_v2(query_options, is_load, scan_params)) {
         return false;
     }
     const auto format = range.__isset.format_type ? range.format_type : scan_params.format_type;
@@ -210,7 +210,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);

be/src/exec/operator/file_scan_operator.h

@@ -56,6 +56,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);
@@ -80,8 +82,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);
     static bool _can_generate_physical_splits(const TQueryOptions& query_options, bool is_load,
                                               const TFileScanRangeParams& scan_params,
                                               const TFileRangeDesc& range);

be/src/exec/rowid_fetcher.cpp

@@ -56,7 +56,9 @@
 #include "core/data_type_serde/data_type_serde.h"
 #include "core/string_ref.h"
 #include "core/types.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 "format_v2/table/lance_reader.h"
@@ -776,6 +778,69 @@ 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;
+}
+
+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_lance_rows_by_row_ids(
         const TFileRangeDesc& scan_range_desc, const std::vector<uint64_t>& row_ids,
         const std::vector<SlotDescriptor>& slots, RuntimeState* runtime_state,
@@ -851,13 +916,7 @@ Status RowIdStorageReader::read_external_row_from_file_mapping(
     scan_blocks[idx] = Block(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");
@@ -868,9 +927,7 @@ Status RowIdStorageReader::read_external_row_from_file_mapping(
                 scan_range_desc, read_ids, slots, runtime_state.get(), sub_runtime_profile.get(),
                 rpc_scan_params, &scan_blocks[idx], &fetch_statistics[idx]));
     } else {
-        // Parquet/ORC row IDs are consumed as row ordinals within the exact physical file range
-        // recorded by phase one. Keep using FileScanner so the format reader can resolve those
-        // ordinals against that range; unlike Lance, ranges cannot be merged at dataset level.
+        // Parquet/ORC IDs remain signed physical file positions; Lance keeps native uint64 IDs.
         std::list<int64_t> legacy_read_ids;
         for (const auto row_id : read_ids) {
             if (row_id > static_cast<uint64_t>(std::numeric_limits<int64_t>::max())) {
@@ -879,26 +936,37 @@ Status RowIdStorageReader::read_external_row_from_file_mapping(
             }
             legacy_read_ids.emplace_back(static_cast<int64_t>(row_id));
         }
-        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, legacy_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, legacy_read_ids, &scan_blocks[idx],
+                    &fetch_statistics[idx].init_reader_ms, &fetch_statistics[idx].get_block_ms));
+        } else {
+            // Keep phase two on V1 whenever the rollout policy or table format requires it.
+            std::unique_ptr<FileScanner> 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, legacy_read_ids, &scan_blocks[idx], external_info,
+                    &fetch_statistics[idx].init_reader_ms, &fetch_statistics[idx].get_block_ms));
+        }
     }
 
     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(
                 file_read_bytes_counter->value(), file_read_bytes_counter->type());
     }
 
     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());
@@ -947,30 +1015,16 @@ Status RowIdStorageReader::read_batch_external_row(
             }
         }
 
-        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());
-        for (auto slot_idx = 0; slot_idx < slots.size(); ++slot_idx) {
+        std::vector<uint32_t> scan_column_idxs;
+        scan_column_idxs.reserve(slots.size());
+        for (int slot_idx = 0; slot_idx < slots.size(); ++slot_idx) {
             auto& slot = slots[slot_idx];
             tuple_desc.add_slot(&slot);
             colname_to_slot_id.emplace(slot.col_name(), slot.id());
-            TFileScanSlotInfo slot_info;
-            slot_info.slot_id = slot.id();
-            auto column_idx = request_block_desc.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);
+            scan_column_idxs.emplace_back(request_block_desc.column_idxs(slot_idx));
         }
+        rpc_scan_params = build_external_scan_params(rpc_scan_params, first_scan_range_desc, slots,
+                                                     scan_column_idxs);
 
         const auto& query_options = id_file_map->get_query_options();
         const auto& query_globals = id_file_map->get_query_globals();
@@ -1174,9 +1228,9 @@ Status RowIdStorageReader::read_batch_external_row(
                                          std::to_string(*get_block_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));
         for (const auto& [time_name, time_value] : lance_fetch_times_ns) {
             runtime_profile->add_info_string(time_name,

be/src/exec/rowid_fetcher.h

@@ -40,6 +40,7 @@ namespace doris {
 class DorisNodesInfo;
 class RuntimeProfile;
 class RuntimeState;
+class TQueryOptions;
 class TupleDescriptor;
 namespace io {
 enum class FileCacheMissPolicy : uint8_t;
@@ -122,7 +123,17 @@ class RowIdStorageReader {
     static Status read_by_rowids(const PMultiGetRequest& request, PMultiGetResponse* response);
     static Status read_by_rowids(const PMultiGetRequestV2& request, PMultiGetResponseV2* response);
 
+    static bool should_use_file_scanner_v2(const TQueryOptions& query_options,
+                                           const TFileScanRangeParams& scan_params,
+                                           const TFileRangeDesc& range);
+
 private:
+    friend class RowIdStorageReaderTest;
+    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);
     struct ExternalFetchStatistics;
 
     static Status read_doris_format_row(