diff --git a/be/src/exec/operator/file_scan_operator.cpp b/be/src/exec/operator/file_scan_operator.cpp index c33fe2f8bd881c..f8a72456e8494a 100644 --- a/be/src/exec/operator/file_scan_operator.cpp +++ b/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* 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 862ff4d438d7b8..80c16e1177ddb2 100644 --- a/be/src/exec/operator/file_scan_operator.h +++ b/be/src/exec/operator/file_scan_operator.h @@ -56,6 +56,8 @@ class FileScanLocalState final : public ScanLocalState { 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 { 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); diff --git a/be/src/exec/rowid_fetcher.cpp b/be/src/exec/rowid_fetcher.cpp index 61c5a529d73785..8c3b0023831899 100644 --- a/be/src/exec/rowid_fetcher.cpp +++ b/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& scan_slots, + const std::vector& 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& row_ids, const std::vector& 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 sub_runtime_profile = std::make_unique("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 legacy_read_ids; for (const auto row_id : read_ids) { if (row_id > static_cast(std::numeric_limits::max())) { @@ -879,18 +936,29 @@ Status RowIdStorageReader::read_external_row_from_file_mapping( } legacy_read_ids.emplace_back(static_cast(row_id)); } - std::unique_ptr 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 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( @@ -898,7 +966,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()); @@ -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 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, diff --git a/be/src/exec/rowid_fetcher.h b/be/src/exec/rowid_fetcher.h index ceb7a51c1a0d38..ab8019e8f32641 100644 --- a/be/src/exec/rowid_fetcher.h +++ b/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& scan_slots, + const std::vector& scan_column_idxs); struct ExternalFetchStatistics; static Status read_doris_format_row( diff --git a/be/src/exec/scan/file_scanner_v2.cpp b/be/src/exec/scan/file_scanner_v2.cpp index ac775fd94bff76..c0cdfaf7b78d4c 100644 --- a/be/src/exec/scan/file_scanner_v2.cpp +++ b/be/src/exec/scan/file_scanner_v2.cpp @@ -75,8 +75,12 @@ #include "runtime/runtime_state.h" #include "service/backend_options.h" #include "storage/id_manager.h" +#include "util/stopwatch.hpp" namespace doris { +const std::string FileScannerV2::FileReadBytesProfile = "FileReadBytes"; +const std::string FileScannerV2::FileReadTimeProfile = "FileReadTime"; + namespace { constexpr int kIcebergPositionDeleteContent = 1; @@ -432,12 +436,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( @@ -715,18 +719,21 @@ Status FileScannerV2::_init_table_reader(const TFileRangeDesc& range) { VExprContextSPtrs table_conjuncts; RETURN_IF_ERROR(_build_table_conjuncts(&table_conjuncts)); std::optional> 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({ @@ -736,15 +743,106 @@ Status FileScannerV2::_init_table_reader(const TFileRangeDesc& range) { .scan_params = const_cast(_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& 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(); + _file_reader_stats = std::make_unique(); + _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 partition_values; + RETURN_IF_ERROR(_generate_partition_values(range, &partition_values)); + format::FileFormat current_split_format; + RETURN_IF_ERROR(_to_file_format(format_type, ¤t_split_format)); + std::vector requested_rows(row_ids.begin(), row_ids.end()); + _table_reader->set_batch_size(std::max(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, + .file_context = nullptr, + .condition_cache_source_range = std::nullopt, + .condition_cache_split_context = nullptr, + .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); + // Reordering uses a dense position for every requested ID. A replaced or truncated file + // must fail here instead of exposing a short source column to unchecked indexed inserts. + if (result_block->rows() != row_ids.size()) { + return Status::Corruption("FileScannerV2 row-ID fetch returned {} rows, expected {}", + result_block->rows(), row_ids.size()); + } + return Status::OK(); +} + Status FileScannerV2::_create_table_reader_for_format( const TFileRangeDesc& range, std::unique_ptr* reader) const { DORIS_CHECK(reader != nullptr); @@ -825,6 +923,7 @@ Status FileScannerV2::_prepare_table_reader_split(const TFileRangeDesc& range, .format_split_id_end = _current_split.format_split_id_end, .global_rowid_context = _create_global_rowid_context(_current_split.source_identity_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 5f6869a8312d96..dd10d2952b1912 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 #include #include #include @@ -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; @@ -139,6 +142,18 @@ class FileScannerV2 final : public Scanner { ShardedKVCache* kv_cache, const std::unordered_map* colname_to_slot_id); + // Standalone scanner used by TopN two-phase materialization. + FileScannerV2(RuntimeState* state, RuntimeProfile* profile, const TFileScanRangeParams* params, + const std::unordered_map* 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& 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 34fd4e5b273368..894cd49fa6b043 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() << ", residual_predicate_columns=" + 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() << ", residual_predicate_columns=" << join_debug_strings( residual_predicate_columns, [](LocalColumnId column_id) { return std::to_string(column_id.value()); }) diff --git a/be/src/format_v2/file_reader.h b/be/src/format_v2/file_reader.h index 586f9dd4ff5e36..8d0cc487676a33 100644 --- a/be/src/format_v2/file_reader.h +++ b/be/src/format_v2/file_reader.h @@ -18,6 +18,7 @@ #include #include #include +#include #include #include #include @@ -117,6 +118,12 @@ struct FileScanRequest { // predicate_columns, the value is semantically required and must still be validated and read. std::vector 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> 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. @@ -409,8 +416,21 @@ class FileReader { virtual std::unique_ptr 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 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 868f451963fc4f..85754b9997a8bf 100644 --- a/be/src/format_v2/orc/orc_reader.cpp +++ b/be/src/format_v2/orc/orc_reader.cpp @@ -755,6 +755,7 @@ struct OrcReaderScanState { std::vector 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; }; @@ -1285,6 +1286,7 @@ Status OrcReader::open(std::shared_ptr 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; @@ -1342,6 +1344,16 @@ Status OrcReader::open(std::shared_ptr 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(row_id) < _state->row_reader_range_first_row || + static_cast(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(); } @@ -1757,7 +1769,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; @@ -2054,11 +2071,23 @@ Status OrcReader::get_block(Block* file_block, size_t* rows, bool* eof) { } bool has_next = false; + std::optional 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((*_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(); } @@ -2066,6 +2095,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()); @@ -2086,6 +2118,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) { @@ -2096,7 +2132,7 @@ Status OrcReader::get_block(Block* file_block, size_t* rows, bool* eof) { } const auto batch_rows = static_cast(_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 0c42b32f617b46..5994ad6993c439 100644 --- a/be/src/format_v2/orc/orc_reader.h +++ b/be/src/format_v2/orc/orc_reader.h @@ -77,6 +77,7 @@ class OrcReader final : public format::FileReader { Status get_schema(std::vector* const file_schema) const override; std::unique_ptr create_column_mapper( format::TableColumnMapperOptions options) const override; + bool supports_rowid_fetch() const override { return true; } Status open(std::shared_ptr 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 b912af0a1654c7..1b162b48bd6d61 100644 --- a/be/src/format_v2/parquet/parquet_reader.h +++ b/be/src/format_v2/parquet/parquet_reader.h @@ -62,6 +62,7 @@ class ParquetReader : public format::FileReader { std::unique_ptr create_column_mapper( format::TableColumnMapperOptions options) const override; + bool supports_rowid_fetch() const override { return true; } Status open(std::shared_ptr request) override; diff --git a/be/src/format_v2/parquet/parquet_scan.cpp b/be/src/format_v2/parquet/parquet_scan.cpp index 2652e3e5f75044..254e43bbc9b20f 100644 --- a/be/src/format_v2/parquet/parquet_scan.cpp +++ b/be/src/format_v2/parquet/parquet_scan.cpp @@ -842,7 +842,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; prepare_row_group_physical_projection(row_group, file_schema, request, &row_group_plan); plan->row_groups.push_back(std::move(row_group_plan)); @@ -1580,14 +1601,14 @@ Status ParquetScanScheduler::open_next_row_group( RETURN_IF_ERROR(detail::build_native_prefetch_ranges( thrift_metadata, file_schema, request_scan_columns(row_group_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); } @@ -1621,7 +1642,7 @@ Status ParquetScanScheduler::open_next_row_group( file_context.native_io_ctx, _runtime_state, file_context.native_page_cache_enabled, file_context.native_page_cache_file_key, _current_dictionary_filters.contains(local_id), _scan_profile.column_reader_profile, - &column_reader)); + &column_reader, !request.row_ids.has_value())); _current_predicate_columns[local_id] = std::move(column_reader); } // Start warming filter-column chunks as soon as their row group is selected. The native @@ -1630,8 +1651,9 @@ Status ParquetScanScheduler::open_next_row_group( if (!_current_merge_range_active) { const auto prefetch_columns = adaptive_predicate_prefetch_columns(request, row_group_request.predicate_columns); - 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 : row_group_request.non_predicate_columns) { const auto local_id = col.column_id(); @@ -1660,7 +1682,7 @@ Status ParquetScanScheduler::open_next_row_group( row_group_idx, _current_selected_ranges, _current_offset_indexes, _timezone, file_context.native_io_ctx, _runtime_state, file_context.native_page_cache_enabled, file_context.native_page_cache_file_key, false, _scan_profile.column_reader_profile, - &column_reader)); + &column_reader, !request.row_ids.has_value())); _current_non_predicate_columns[local_id] = std::move(column_reader); } if (!_current_merge_range_active && @@ -1670,7 +1692,8 @@ Status ParquetScanScheduler::open_next_row_group( // 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, physical_non_predicate_columns(row_group_request), + file_context, file_schema, request, + physical_non_predicate_columns(row_group_request), &_current_non_predicate_prefetched)); } if (_parquet_profile != nullptr) { @@ -2206,7 +2229,7 @@ Status ParquetScanScheduler::prepare_current_dictionary_filters( row_group_idx, _current_selected_ranges, _current_offset_indexes, _timezone, file_context.native_io_ctx, _runtime_state, file_context.native_page_cache_enabled, file_context.native_page_cache_file_key, true, _scan_profile.column_reader_profile, - &column_reader)); + &column_reader, !request.row_ids.has_value())); MutableColumnPtr dictionary_values; { SCOPED_TIMER(_scan_profile.dict_filter_read_dict_time); @@ -2898,10 +2921,14 @@ Status ParquetScanScheduler::read_filter_columns(int64_t batch_rows, Status ParquetScanScheduler::prefetch_current_row_group_columns( ParquetFileContext& file_context, const std::vector>& file_schema, + const format::FileScanRequest& request, const std::vector& 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; @@ -3022,9 +3049,10 @@ Status ParquetScanScheduler::read_current_row_group_batch( // materializing non-predicate columns, so fully filtered batches avoid unnecessary IO. const auto& physical_request = _current_row_group_request != nullptr ? *_current_row_group_request : request; - RETURN_IF_ERROR(prefetch_current_row_group_columns( - file_context, file_schema, physical_non_predicate_columns(physical_request), - &_current_non_predicate_prefetched)); + RETURN_IF_ERROR( + prefetch_current_row_group_columns(file_context, file_schema, request, + physical_non_predicate_columns(physical_request), + &_current_non_predicate_prefetched)); } if (selected_rows > _batch_size) { @@ -3185,6 +3213,13 @@ Status ParquetScanScheduler::read_next_batch( *eof = false; return Status::OK(); } + // Phase-two IDs have already survived filtering. Append sparse ranges directly to one + // output block so TableReader finalization and result merging happen once per caller batch. + // Filtered requests and pending projection changes retain their single-batch coordinates. + const bool append_row_id_ranges = + _active_request->row_ids.has_value() && _active_request->predicate_columns.empty() && + _active_request->conjuncts.empty() && _active_request->delete_conjuncts.empty() && + _active_request->count_star_placeholder_columns.empty() && _pending_request == nullptr; int64_t predicate_batch_rows = std::max(_batch_size, _empty_predicate_batch_rows); const int64_t max_predicate_batch_rows = std::min( std::numeric_limits::max(), @@ -3237,13 +3272,17 @@ Status ParquetScanScheduler::read_next_batch( continue; } - const int64_t batch_rows = std::min(predicate_batch_rows, remaining_rows); + const int64_t row_cap = append_row_id_ranges ? _batch_size - static_cast(*rows) + : predicate_batch_rows; + const int64_t batch_rows = std::min(row_cap, remaining_rows); const int64_t physical_rows_read = batch_rows; const int64_t batch_first_file_row = _current_row_group_first_row + _current_row_group_rows_read; + size_t batch_output_rows = 0; RETURN_IF_ERROR(read_current_row_group_batch(file_context, file_schema, batch_rows, *_active_request, batch_first_file_row, - file_block, rows)); + file_block, &batch_output_rows)); + *rows += batch_output_rows; _current_row_group_rows_read += physical_rows_read; _current_range_rows_read += physical_rows_read; if (_current_range_rows_read >= current_range.length) { @@ -3259,6 +3298,9 @@ Status ParquetScanScheduler::read_next_batch( _publish_adaptive_state(*_active_request); continue; } + if (append_row_id_ranges && *rows < static_cast(_batch_size)) { + continue; + } *eof = false; return Status::OK(); } diff --git a/be/src/format_v2/parquet/parquet_scan.h b/be/src/format_v2/parquet/parquet_scan.h index dff822fe21c14f..7046d342304162 100644 --- a/be/src/format_v2/parquet/parquet_scan.h +++ b/be/src/format_v2/parquet/parquet_scan.h @@ -313,6 +313,7 @@ class ParquetScanScheduler { Status prefetch_current_row_group_columns( ParquetFileContext& file_context, const std::vector>& file_schema, + const format::FileScanRequest& request, const std::vector& scan_columns, bool* prefetched); Status read_current_row_group_batch( diff --git a/be/src/format_v2/parquet/reader/native_column_reader.cpp b/be/src/format_v2/parquet/reader/native_column_reader.cpp index 273964e5d1f451..e0fab04951be42 100644 --- a/be/src/format_v2/parquet/reader/native_column_reader.cpp +++ b/be/src/format_v2/parquet/reader/native_column_reader.cpp @@ -266,7 +266,7 @@ Status NativeColumnReader::create( const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState* runtime_state, bool enable_page_cache, const std::string& page_cache_file_key, bool enable_dictionary_filter, ParquetColumnReaderProfile profile, - std::unique_ptr* reader) { + std::unique_ptr* reader, bool enable_read_ahead) { if (reader == nullptr) { return Status::InvalidArgument("Native parquet reader result is null"); } @@ -307,10 +307,11 @@ Status NativeColumnReader::create( auto native_reader = std::unique_ptr( new NativeColumnReader(column_schema, std::move(logical_type), std::move(native_type), std::move(variant_plan), profile)); - RETURN_IF_ERROR(native_reader->init( - std::move(file), metadata, row_group_id, field, std::move(schema_node), - std::move(projected_ids), selected_ranges, offset_indexes, timezone, io_ctx, - runtime_state, enable_page_cache, page_cache_file_key, enable_dictionary_filter)); + RETURN_IF_ERROR(native_reader->init(std::move(file), metadata, row_group_id, field, + std::move(schema_node), std::move(projected_ids), + selected_ranges, offset_indexes, timezone, io_ctx, + runtime_state, enable_page_cache, page_cache_file_key, + enable_dictionary_filter, enable_read_ahead)); *reader = std::move(native_reader); return Status::OK(); } @@ -322,7 +323,7 @@ Status NativeColumnReader::init( const std::unordered_map& offset_indexes, const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState* runtime_state, bool enable_page_cache, const std::string& page_cache_file_key, - bool enable_dictionary_filter) { + bool enable_dictionary_filter, bool enable_read_ahead) { DORIS_CHECK(file != nullptr); DORIS_CHECK(metadata != nullptr); DORIS_CHECK(field != nullptr); @@ -347,7 +348,10 @@ Status NativeColumnReader::init( const size_t max_group_buffer = config::parquet_rowgroup_max_buffer_mb << 20; const size_t max_column_buffer = config::parquet_column_max_buffer_mb << 20; - const size_t max_buffer_size = std::min(max_group_buffer, max_column_buffer); + // Sparse exact-row fetches need demand pages, not one read-ahead buffer per physical leaf. + // Passing zero through the native tree also prevents wide nested projections multiplying it. + const size_t max_buffer_size = + enable_read_ahead ? std::min(max_group_buffer, max_column_buffer) : 0; RuntimeState* native_runtime_state = runtime_state; const bool runtime_page_cache_enabled = runtime_state == nullptr || diff --git a/be/src/format_v2/parquet/reader/native_column_reader.h b/be/src/format_v2/parquet/reader/native_column_reader.h index 150ab4352d00c6..69588484696b56 100644 --- a/be/src/format_v2/parquet/reader/native_column_reader.h +++ b/be/src/format_v2/parquet/reader/native_column_reader.h @@ -76,7 +76,8 @@ class NativeColumnReader final : public ParquetColumnReader { RuntimeState* runtime_state, bool enable_page_cache, const std::string& page_cache_file_key, bool enable_dictionary_filter, ParquetColumnReaderProfile profile, - std::unique_ptr* reader); + std::unique_ptr* reader, + bool enable_read_ahead = true); ~NativeColumnReader() override; @@ -115,7 +116,7 @@ class NativeColumnReader final : public ParquetColumnReader { const std::unordered_map& offset_indexes, const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState* runtime_state, bool enable_page_cache, const std::string& page_cache_file_key, - bool enable_dictionary_filter); + bool enable_dictionary_filter, bool enable_read_ahead); Status read_with_filter(int64_t rows, const uint8_t* filter_data, bool filter_all, MutableColumnPtr& column, const DataTypePtr& output_type, diff --git a/be/src/format_v2/table_reader.cpp b/be/src/format_v2/table_reader.cpp index 8e03d3d2282634..2c7ffe933c70bf 100644 --- a/be/src/format_v2/table_reader.cpp +++ b/be/src/format_v2/table_reader.cpp @@ -1343,6 +1343,7 @@ Status TableReader::refresh_conjuncts(VExprContextSPtrs conjuncts, _table_filters, _projected_columns, refreshed_request.get(), _runtime_state, _file_scan_request == nullptr ? nullptr : &_file_scan_request->local_positions)); refreshed_request->predicate_snapshot_digest = _predicate_snapshot_digest; + 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() && @@ -1751,6 +1752,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; @@ -1779,9 +1781,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 797bba2a0a1999..273ef9820219e6 100644 --- a/be/src/format_v2/table_reader.h +++ b/be/src/format_v2/table_reader.h @@ -194,6 +194,9 @@ struct SplitReadOptions { int64_t format_split_id = -1; int64_t format_split_id_end = -1; std::optional 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> row_ids = std::nullopt; }; // Base class for table-level readers. @@ -478,6 +481,7 @@ class TableReader { RETURN_IF_ERROR(_data_reader.column_mapper->create_scan_request( _table_filters, _projected_columns, file_request.get(), _runtime_state)); file_request->predicate_snapshot_digest = _predicate_snapshot_digest; + 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); @@ -1072,7 +1076,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(); } @@ -2060,6 +2064,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 _global_rowid_context; + std::optional> _row_ids; bool _aggregate_pushdown_tried = false; std::optional _metadata_aggregate_result; bool _current_split_pruned = false; diff --git a/be/test/exec/rowid_fetcher_test.cpp b/be/test/exec/rowid_fetcher_test.cpp new file mode 100644 index 00000000000000..a1ba94cf7020e6 --- /dev/null +++ b/be/test/exec/rowid_fetcher_test.cpp @@ -0,0 +1,243 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "exec/rowid_fetcher.h" + +#include + +#include +#include + +#include "common/consts.h" +#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 { + +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; + }; + + static SlotDescriptor make_slot(const SlotSpec& spec) { + TSlotDescriptor tdesc = TSlotDescriptorBuilder() + .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); + return SlotDescriptor(tdesc); + } +}; + +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_LANCE); + 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; + // Branch-4.1 recognizes these built-in virtual names; arbitrary synthesized names are + // not virtual columns and fall through to the legacy partition-slot classification. + source_params.__set_column_name_to_category( + {{BeConsts::GLOBAL_ROWID_COL, TColumnCategory::SYNTHESIZED}, + {BeConsts::ICEBERG_ROWID_COL, 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 slots; + for (const auto& name : + {BeConsts::GLOBAL_ROWID_COL, BeConsts::ICEBERG_ROWID_COL, std::string("generated_col"), + std::string("partition_col"), std::string("value")}) { + slots.emplace_back(make_slot( + {.col_name = name, + .slot_id = static_cast(slots.size()), + .type = name == BeConsts::GLOBAL_ROWID_COL ? 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 {1, 0})); + const std::vector categories { + TColumnCategory::SYNTHESIZED, TColumnCategory::SYNTHESIZED, TColumnCategory::GENERATED, + TColumnCategory::PARTITION_KEY, TColumnCategory::REGULAR}; + 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); + } + // 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 {"value", "partition_col"}, + std::vector {"partition_col", "value", "id"}, + std::vector {"partition_col"}, + std::vector {"value", "id"}}) { + SCOPED_TRACE(fmt::format("format={}, columns={}", static_cast(format), + fmt::join(names, ","))); + std::vector slots; + std::vector indices; + std::vector file_indices; + for (const auto& name : names) { + slots.emplace_back(make_slot( + {.col_name = name, .slot_id = static_cast(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 = ¶ms, .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()); + } + } +} + +} // namespace doris diff --git a/be/test/format_v2/orc/orc_reader_test.cpp b/be/test/format_v2/orc/orc_reader_test.cpp index 1884afaf73cdfd..10f3a524aebab7 100644 --- a/be/test/format_v2/orc/orc_reader_test.cpp +++ b/be/test/format_v2/orc/orc_reader_test.cpp @@ -10079,6 +10079,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 schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + auto request = std::make_shared(); + request->non_predicate_columns = {field_projection(0)}; + request->row_ids = {0, 2, 4}; + ASSERT_TRUE(reader->open(request).ok()); + + std::vector 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( + assert_cast(*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({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 1292b90cb497a0..c045683e424d27 100644 --- a/be/test/format_v2/parquet/parquet_scan_test.cpp +++ b/be/test/format_v2/parquet/parquet_scan_test.cpp @@ -25,7 +25,10 @@ #include #include +#include #include +#include +#include #include #include #include @@ -35,6 +38,7 @@ #include #include #include +#include #include #include @@ -50,6 +54,7 @@ #include "core/data_type/data_type_string.h" #include "core/data_type/data_type_struct.h" #include "core/field.h" +#include "exec/scan/file_scanner_v2.h" #include "exprs/bloom_filter_func.h" #include "exprs/create_predicate_function.h" #include "exprs/runtime_filter_expr.h" @@ -69,16 +74,25 @@ #include "format_v2/parquet/parquet_column_schema.h" #include "format_v2/parquet/parquet_reader.h" #include "format_v2/parquet/reader/native/block_split_bloom_filter.h" +#include "format_v2/parquet/reader/native/column_reader.h" #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/descriptor_helper.h" #include "runtime/runtime_state.h" #include "storage/index/zone_map/zonemap_eval_context.h" #include "storage/index/zone_map/zonemap_filter_result.h" #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 { @@ -2596,6 +2610,381 @@ TEST_F(ParquetScanTest, GlobalRowIdUsesFileLocalPositionForScanRange) { EXPECT_EQ(row_ids, std::vector({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 schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + auto request = std::make_shared(); + request->non_predicate_columns = {field_projection(0)}; + request->row_ids = {0, 3, 5}; + ASSERT_TRUE(reader->open(request).ok()); + + std::vector 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({1, 4, 6})); +} + +TEST_F(ParquetScanTest, StandaloneRowIdFetchRejectsMissingRows) { + write_int_pair_parquet_file(_file_path, 2); + TDescriptorTableBuilder descriptors_builder; + TTupleDescriptorBuilder tuple_builder; + tuple_builder.add_slot(TSlotDescriptorBuilder() + .type(TYPE_INT) + .nullable(false) + .column_name("id") + .column_pos(0) + .build()); + tuple_builder.build(&descriptors_builder); + ObjectPool pool; + DescriptorTbl* descriptors = nullptr; + ASSERT_TRUE(DescriptorTbl::create(&pool, descriptors_builder.desc_tbl(), &descriptors).ok()); + auto* tuple = descriptors->get_tuple_descriptor(0); + TFileScanRangeParams params; + params.__set_format_type(TFileFormatType::FORMAT_PARQUET); + params.__set_file_type(TFileType::FILE_LOCAL); + params.__set_src_tuple_id(0); + params.__set_dest_tuple_id(0); + TFileScanSlotInfo slot; + slot.__set_slot_id(0); + slot.__set_is_file_slot(true); + slot.__set_category(TColumnCategory::REGULAR); + params.__set_required_slots({slot}); + params.__set_column_idxs({0}); + params.__set_slot_name_to_schema_pos({{"id", 0}}); + TFileRangeDesc range; + range.__set_path(_file_path); + range.__set_start_offset(0); + range.__set_size(std::filesystem::file_size(_file_path)); + range.__set_file_size(range.size); + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + state.set_desc_tbl(descriptors); + // EOF is valid for a scan, but exact fetches must not expose missing rows to reordering. + for (const auto& requested : {std::list {}, {0, 3, 5}, {0, 3, 6}, {6}}) { + RuntimeProfile profile("row_id_fetch"); + FileScannerV2 scanner(&state, &profile, ¶ms, nullptr, tuple); + Block block; + block.insert({ColumnInt32::create(), std::make_shared(), "id"}); + int64_t init_ms = 0; + int64_t read_ms = 0; + auto status = scanner.read_by_rows(range, requested, &block, &init_ms, &read_ms); + if (requested.empty() || requested.back() < 6) { + ASSERT_TRUE(status.ok()) << status.to_string(); + ASSERT_EQ(block.rows(), requested.size()); + const auto& values = int32_data_column(*block.get_by_position(0).column); + size_t row = 0; + for (auto id : requested) { + EXPECT_EQ(values.get_element(row++), id + 1); + } + } else { + EXPECT_FALSE(status.ok()); + EXPECT_NE(status.to_string().find("row-ID fetch"), std::string::npos); + } + } +} + +TEST_F(ParquetScanTest, SparseRowIdsFillBatchesAcrossRangesAndRowGroups) { + std::vector ids(4096); + 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, 500); + auto reader = create_reader(); + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + ASSERT_TRUE(reader->init(&state).ok()); + reader->set_batch_size(128); + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + auto request = std::make_shared(); + request->non_predicate_columns = {field_projection(0), + field_projection(format::ROW_POSITION_COLUMN_ID)}; + request->local_positions.emplace(format::LocalColumnId(0), format::LocalIndex(0)); + request->local_positions.emplace(format::LocalColumnId(format::ROW_POSITION_COLUMN_ID), + format::LocalIndex(1)); + request->row_ids.emplace(); + for (int64_t id = 0; id < 4096; id += 4) { + request->row_ids->push_back(id); + } + ASSERT_TRUE(reader->open(request).ok()); + size_t fetched = 0; + size_t batches = 0; + bool eof = false; + while (!eof) { + Block block = build_file_block({schema[0], format::row_position_column_definition()}); + size_t rows = 0; + ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok()); + if (rows == 0) { + continue; + } + EXPECT_LE(rows, 128); + const auto& column = int32_data_column(*block.get_by_position(0).column); + const auto& positions = int64_data_column(*block.get_by_position(1).column); + for (size_t row = 0; row < rows; ++row) { + ASSERT_LT(fetched, request->row_ids->size()); + EXPECT_EQ(positions.get_element(row), (*request->row_ids)[fetched]); + EXPECT_EQ(column.get_element(row), (*request->row_ids)[fetched++]); + } + ++batches; + } + RecordProperty("nonempty_batches", std::to_string(batches)); + EXPECT_EQ(fetched, 1024); + EXPECT_EQ(batches, 8); +} + +TEST_F(ParquetScanTest, SparseRowIdsAppendNestedColumnsAcrossRanges) { + write_int_list_parquet_file(_file_path); + auto reader = create_reader(); + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + ASSERT_TRUE(reader->init(&state).ok()); + reader->set_batch_size(3); + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + auto request = std::make_shared(); + request->non_predicate_columns = {field_projection(0), field_projection(1)}; + request->row_ids = {0, 2}; + ASSERT_TRUE(reader->open(request).ok()); + Block block = build_file_block(schema); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok()); + ASSERT_EQ(rows, 2); + EXPECT_EQ(int32_data_column(*block.get_by_position(0).column).get_element(0), 1); + EXPECT_EQ(int32_data_column(*block.get_by_position(0).column).get_element(1), 3); + const IColumn* list_column = block.get_by_position(1).column.get(); + if (const auto* nullable = check_and_get_column(*list_column)) { + list_column = &nullable->get_nested_column(); + } + const auto& lists = assert_cast(*list_column); + ASSERT_EQ(lists.get_offsets().size(), 2); + EXPECT_EQ(lists.get_offsets()[0], 2); + EXPECT_EQ(lists.get_offsets()[1], 2); + const auto& elements = int32_data_column(lists.get_data()); + ASSERT_EQ(elements.size(), 2); + EXPECT_EQ(elements.get_element(0), 1); + EXPECT_EQ(elements.get_element(1), 2); +} + +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 remote_bytes {0}; + std::atomic 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, SparseRowIdsBoundWideProjectionReadAhead) { + using namespace format::parquet; + // Twenty leaves exceed the default aggregate budget if each retains 8 MiB of read-ahead. + constexpr int leaf_count = 20; + std::vector ids(3 * 1024 * 1024); + std::iota(ids.begin(), ids.end(), 0); + auto values = build_int32_array(ids); + std::vector> fields; + std::vector> arrays; + for (int leaf = 0; leaf < leaf_count; ++leaf) { + fields.push_back(arrow::field("c" + std::to_string(leaf), arrow::int32(), false)); + arrays.push_back(values); + } + auto table = arrow::Table::Make(arrow::schema(fields), arrays); + auto output = arrow::io::FileOutputStream::Open(_file_path).ValueOrDie(); + ::parquet::WriterProperties::Builder writer_properties; + writer_properties.disable_dictionary(); + writer_properties.compression(::parquet::Compression::UNCOMPRESSED); + writer_properties.data_page_version(::parquet::ParquetDataPageVersion::V2); + // Arrow otherwise caps each row group at 1M rows, making every chunk smaller than 8 MiB. + writer_properties.max_row_group_length(ids.size()); + PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table, arrow::default_memory_pool(), output, + ids.size(), writer_properties.build())); + ASSERT_TRUE(output->Close().ok()); + io::FileReaderSPtr local; + ASSERT_TRUE(io::global_local_filesystem()->open_file(_file_path, &local).ok()); + auto remote = std::make_shared(local); + io::IOContext io_ctx; + io::FileDescription description; + description.path = _file_path; + description.file_size = local->size(); + ParquetFileContext context; + ASSERT_TRUE(context.open(remote, &io_ctx, false, description).ok()); + std::vector> schema; + ASSERT_TRUE(build_parquet_column_schema(context.native_metadata->schema(), &schema).ok()); + auto request = std::make_shared(); + Block block; + for (int leaf = 0; leaf < leaf_count; ++leaf) { + request->non_predicate_columns.push_back(field_projection(leaf)); + request->local_positions.emplace(format::LocalColumnId(leaf), format::LocalIndex(leaf)); + block.insert({schema[leaf]->type->create_column(), schema[leaf]->type, schema[leaf]->name}); + } + request->row_ids = {0}; + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + auto plan = std::make_shared(); + ASSERT_TRUE(plan_parquet_row_groups(*context.native_metadata, schema, *request, {}, false, + plan.get(), &state.timezone_obj(), &state, &context) + .ok()); + ParquetScanScheduler scheduler; + scheduler.set_plan(plan); + scheduler.set_batch_size(1); + scheduler.set_scan_request(request); + scheduler.set_runtime_state(&state); + scheduler.set_timezone(&state.timezone_obj()); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(scheduler.read_next_batch(context, schema, &block, &rows, &eof).ok()); + ASSERT_EQ(rows, 1); + for (int leaf = 0; leaf < leaf_count; ++leaf) { + EXPECT_EQ(int32_data_column(*block.get_by_position(leaf).column).get_element(0), 0); + } + size_t retained_buffer_bytes = 0; + ASSERT_EQ(scheduler._current_non_predicate_columns.size(), leaf_count); + for (const auto& [id, column_reader] : scheduler._current_non_predicate_columns) { + const auto* adapter = dynamic_cast(column_reader.get()); + ASSERT_NE(adapter, nullptr); + const auto* scalar = dynamic_cast*>( + adapter->_native_reader.get()); + ASSERT_NE(scalar, nullptr); + retained_buffer_bytes += scalar->_stream_reader->_buf_size; + } + RecordProperty("retained_buffer_bytes", std::to_string(retained_buffer_bytes)); + RecordProperty("remote_bytes", std::to_string(remote->remote_bytes.load())); + EXPECT_LT(retained_buffer_bytes, size_t(config::parquet_rowgroup_max_buffer_mb) << 20); + EXPECT_LT(remote->remote_bytes.load(), size_t(config::parquet_rowgroup_max_buffer_mb) << 20); + EXPECT_EQ(remote->dryrun_bytes.load(), 0); +} + +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 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(); + 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_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::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(); + 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(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(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> schema; + ASSERT_TRUE(build_parquet_column_schema(context.native_metadata->schema(), &schema).ok()); + auto request = std::make_shared(); + 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::make_shared(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/FileQueryScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java index f57d84ca273caf..e95a938848321c 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java @@ -276,12 +276,16 @@ private void updateRequiredSlots() throws UserException { setColumnPositionMapping(); } + protected TColumnCategory classifyColumn(SlotDescriptor slot, List 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 partitionKeys) { - if (partitionKeys.contains(slot.getColumn().getName())) { + protected TColumnCategory classifyColumn(String columnName, List partitionKeys) { + if (partitionKeys.contains(columnName)) { return TColumnCategory.PARTITION_KEY; } return TColumnCategory.REGULAR; @@ -339,6 +343,25 @@ private void setColumnPositionMapping() columnNameMap.putIfAbsent(columnNames.get(i), i); } + boolean needsRowIdFetch = desc.getSlots().stream() + .anyMatch(slot -> slot.getColumn().getName().startsWith(Column.GLOBAL_ROWID_COL)); + if (needsRowIdFetch) { + // Lazy slots are absent from the scan tuple. Use the relation's full schema so + // metadata categories survive pruning without changing physical file positions. + List columns = desc.getTable() instanceof ExternalTable + ? ((ExternalTable) desc.getTable()).getFullSchema(getRelationSnapshot()) + : desc.getTable().getFullSchema(); + List partitionKeys = getPathPartitionKeys(); + Map columnCategories = new HashMap<>(); + for (Column column : columns) { + TColumnCategory category = classifyColumn(column.getName(), partitionKeys); + if (category != TColumnCategory.REGULAR) { + columnCategories.put(column.getName(), category); + } + } + params.setColumnNameToCategory(columnCategories); + } + for (TFileScanSlotInfo slot : params.getRequiredSlots()) { if (!slot.isIsFileSlot()) { continue; diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java index 2a45a1b749f794..ebae13b7d4c030 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java @@ -2570,17 +2570,17 @@ private static Object getPartitionJsonValue(Type type, String partitionValue) { } @Override - protected TColumnCategory classifyColumn(SlotDescriptor slot, List partitionKeys) { - if (Column.ICEBERG_ROWID_COL.equalsIgnoreCase(slot.getColumn().getName())) { + protected TColumnCategory classifyColumn(String columnName, List partitionKeys) { + if (Column.ICEBERG_ROWID_COL.equalsIgnoreCase(columnName)) { return TColumnCategory.SYNTHESIZED; } - if (slot.getColumn().getName().startsWith(Column.GLOBAL_ROWID_COL)) { + if (columnName.startsWith(Column.GLOBAL_ROWID_COL)) { return TColumnCategory.SYNTHESIZED; } - if (IcebergUtils.isIcebergRowLineageColumn(slot.getColumn())) { + if (IcebergUtils.isIcebergRowLineageColumn(columnName)) { return TColumnCategory.GENERATED; } - return super.classifyColumn(slot, partitionKeys); + return super.classifyColumn(columnName, partitionKeys); } private List doGetSplits(int numBackends) throws UserException { diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/FileQueryScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/FileQueryScanNodeTest.java index a29471aff20233..bb3a25c5c975bc 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/FileQueryScanNodeTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/FileQueryScanNodeTest.java @@ -253,6 +253,46 @@ public void testDedicatedSplitSizesHaveCoarseAndFineDefaults() { Assert.assertEquals(64 * MB, sv.getFileSplitSizeOnBe()); } + @Test + public void testRowIdFetchRetainsCategoriesOfPrunedColumns() throws Exception { + TestFileQueryScanNode node = new TestFileQueryScanNode(new SessionVariable()) { + @Override + protected TColumnCategory classifyColumn(String name, List partitionKeys) { + if (name.equals("metadata_path") || name.equals("metadata_position")) { + 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); + sortSlot.setColumn(new Column("id", Type.INT)); + desc.addSlot(sortSlot); + SlotDescriptor rowIdSlot = new SlotDescriptor(new SlotId(2), desc); + rowIdSlot.setColumn(new Column(Column.GLOBAL_ROWID_COL, Type.STRING)); + desc.addSlot(rowIdSlot); + List fullSchema = Arrays.asList(sortSlot.getColumn(), new Column("metadata_path", Type.STRING), + new Column("metadata_position", Type.BIGINT), new Column("generated_col", Type.BIGINT)); + Mockito.when(table.getBaseSchema(false)).thenReturn(fullSchema); + Mockito.when(table.getFullSchema()).thenReturn(fullSchema); + + node.params = new TFileScanRangeParams(); + UPDATE_REQUIRED_SLOTS_METHOD.invoke(node); + + TFileScanRangeParams params = node.getFileScanRangeParams(); + Assert.assertEquals(2, params.getRequiredSlotsSize()); + Assert.assertEquals(Arrays.asList(0), params.getColumnIdxs()); + Assert.assertEquals(TColumnCategory.SYNTHESIZED, params.getColumnNameToCategory().get("metadata_path")); + Assert.assertEquals(TColumnCategory.SYNTHESIZED, params.getColumnNameToCategory().get("metadata_position")); + Assert.assertEquals(TColumnCategory.GENERATED, params.getColumnNameToCategory().get("generated_col")); + Assert.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 35a04f3f99fba8..45b65a0d63d361 100644 --- a/gensrc/thrift/PlanNodes.thrift +++ b/gensrc/thrift/PlanNodes.thrift @@ -693,6 +693,9 @@ struct TFileScanRangeParams { 35: optional string serialized_table_cache_key // 31-33 and 36 are used in master; do not allocate them in branch-4.1. 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 column_name_to_category } struct TFileRangeDesc { diff --git a/regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_query_insert.groovy b/regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_query_insert.groovy index 28e28bdd8870b8..a4579535a9eacf 100644 --- a/regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_query_insert.groovy +++ b/regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_query_insert.groovy @@ -118,6 +118,29 @@ suite("test_iceberg_v3_row_lineage_query_insert", "p0,external,iceberg,external_ "_last_updated_sequence_number should be non-null for ${tableName}, row=${rowLineageRows[i]}") } + def originalThreshold = sql("select @@topn_lazy_materialization_threshold")[0][0] + def originalScannerV2 = sql("select @@enable_file_scanner_v2")[0][0] + try { + sql "set enable_file_scanner_v2 = true" + // Phase two must retain generated-column categories and file-level lineage metadata. + for (String projection : ["_row_id", "_last_updated_sequence_number", + "id, _row_id, _last_updated_sequence_number"]) { + String query = "select ${projection} from ${tableName} order by id limit 2" + 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)) + } + } finally { + // UNSET requires the experimental prefix in 4.1; preserve the caller's values even on failure. + sql "set topn_lazy_materialization_threshold = ${originalThreshold}" + sql "set enable_file_scanner_v2 = ${originalScannerV2}" + } + long firstRowId = rowLineageRows[0][1].toString().toLong() long secondRowId = rowLineageRows[1][1].toString().toLong() assertTrue(firstRowId < secondRowId,