github-actions[bot] commented on code in PR #67922:
URL: https://github.com/apache/doris/pull/67922#discussion_r4002100082
##########
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});
Review Comment:
[P2] Batch disjoint row IDs across range boundaries
For a normal TopN whose order is unrelated to file position, this creates a
length-one selected range for almost every requested ID. `read_next_batch()`
limits `batch_rows` to the current range and returns after the first nonempty
row, so a default 1,024-row lazy fetch can require roughly 1,024
TableReader/finalization/mutable-block-merge cycles even though the requested
batch capacity is 1,024. V1 fills a batch while walking its complete RowRanges
set. Please keep the exact sparse ranges but let one scheduler call accumulate
across range boundaries up to the caller's row cap.
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -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()) {
Review Comment:
[P2] Bound row-ID read-ahead across projected leaves
This row-ID branch avoids whole-chunk MergeRange/prefetch, but the base
native reader still gives every physical leaf its own
`BufferedFileStreamReader` with up to the default 8 MiB read-ahead. The first
page request fills that buffer, so fetching even one row from a wide lazy
projection can synchronously download and retain `8 MiB * leaf_count`,
exceeding the nominal 128 MiB row-group budget. V1 divides that budget across
projected columns. The new sparse-I/O test masks this by setting
`parquet_column_max_buffer_mb` to 1 and reading one column. Please propagate
row-ID selectivity here and enforce an aggregate read-ahead cap (or disable it)
across the projected leaves.
##########
be/src/exec/scan/file_scanner_v2.cpp:
##########
@@ -736,15 +743,100 @@ Status FileScannerV2::_init_table_reader(const
TFileRangeDesc& range) {
.scan_params = const_cast<TFileScanRangeParams*>(_params),
.io_ctx = _io_ctx,
.runtime_state = _state,
- .scanner_profile = _local_state->scanner_profile(),
+ .scanner_profile = _local_state != nullptr ?
_local_state->scanner_profile() : _profile,
.file_slot_descs = &_file_slot_descs,
- .push_down_agg_type = _local_state->get_push_down_agg_type(),
+ .push_down_agg_type = _local_state != nullptr ?
_local_state->get_push_down_agg_type()
+ :
TPushAggOp::type::NONE,
.push_down_count_columns = std::move(push_down_count_columns),
- .condition_cache_digest =
_local_state->get_condition_cache_digest(),
+ .condition_cache_digest =
+ _local_state != nullptr ?
_local_state->get_condition_cache_digest() : 0,
}));
return Status::OK();
}
+Status FileScannerV2::read_by_rows(const TFileRangeDesc& range, const
std::list<int64_t>& row_ids,
+ Block* result_block, int64_t*
init_reader_ms,
+ int64_t* get_block_ms) {
+ DORIS_CHECK(result_block != nullptr);
+ DORIS_CHECK(init_reader_ms != nullptr);
+ DORIS_CHECK(get_block_ms != nullptr);
+ _current_range = range;
+ RETURN_IF_ERROR(_validate_scan_range(*_params, range));
+ const auto format_type = get_range_format_type(*_params, range);
+ if (format_type != TFileFormatType::FORMAT_PARQUET &&
+ format_type != TFileFormatType::FORMAT_ORC) {
+ return Status::NotSupported(
+ "FileScannerV2 row-id fetch supports only Parquet and ORC,
file format={}",
+ to_string(format_type));
+ }
+
+ _file_cache_statistics = std::make_unique<io::FileCacheStatistics>();
+ _file_reader_stats = std::make_unique<io::FileReaderStats>();
+ _file_read_bytes_counter =
+ ADD_COUNTER_WITH_LEVEL(_profile, FileReadBytesProfile,
TUnit::BYTES, 1);
+ _file_read_time_counter = ADD_TIMER_WITH_LEVEL(_profile,
FileReadTimeProfile, 1);
+ RETURN_IF_ERROR(_init_io_ctx());
+ _io_ctx->file_cache_stats = _file_cache_statistics.get();
+ _io_ctx->file_reader_stats = _file_reader_stats.get();
+ _io_ctx->is_disposable = _state->query_options().disable_file_cache;
+
+ MonotonicStopWatch init_watch;
+ init_watch.start();
+ auto init_status = [&]() -> Status {
+ RETURN_IF_ERROR(_create_table_reader_for_format(range,
&_table_reader));
+ DORIS_CHECK(_table_reader != nullptr);
+ RETURN_IF_ERROR(_init_expr_ctxes());
+ RETURN_IF_ERROR(_init_table_reader(range));
+ std::map<std::string, Field> partition_values;
+ RETURN_IF_ERROR(_generate_partition_values(range, &partition_values));
+ format::FileFormat current_split_format;
+ RETURN_IF_ERROR(_to_file_format(format_type, ¤t_split_format));
+ std::vector<int64_t> requested_rows(row_ids.begin(), row_ids.end());
+ _table_reader->set_batch_size(std::max<size_t>(requested_rows.size(),
1));
+ RETURN_IF_ERROR(_table_reader->prepare_split({
+ .partition_values = std::move(partition_values),
+ .conjuncts = std::nullopt,
+ .partition_prune_conjuncts = {},
+ .all_runtime_filters_applied = true,
+ .condition_cache_digest = 0,
+ .cache = nullptr,
+ .current_range = range,
+ .current_split_format = current_split_format,
+ .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);
Review Comment:
[P1] Reject short row-ID reads before reordering
This loop treats EOF as success without proving that it materialized one row
per requested ID. Hive/TVF inputs are explicitly allowed to be overwritten in
place, and Parquet silently ignores IDs outside its selected Row Groups, so a
file replacement or truncation between TopN's two phases can leave
`result_block` short. The caller has already assigned a dense position for
every ID and later passes all of them to unchecked
`insert_from_multi_column()`, which then indexes past the shorter source
column. Please require `result_block->rows() == row_ids.size()` (or otherwise
account for every requested ID) before returning success.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
