Gabriel39 commented on code in PR #67922:
URL: https://github.com/apache/doris/pull/67922#discussion_r4002258097
##########
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:
Fixed in a4abc561e1. The standalone V2 fetch now rejects a result whose row
count differs from the requested ID count, after closing the reader and
publishing its counters. The real Parquet scanner test covers empty/valid
requests and both partially and entirely missing IDs; the missing-ID cases
failed before the guard and now pass.
##########
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:
Fixed in a4abc561e1. Predicate-free row-ID requests append sparse ranges
directly into the same output block across row-group boundaries, capped by the
remaining caller capacity. Filtered requests, COUNT placeholders, and pending
projection changes retain the existing flow. The 1,024-disjoint-ID regression
now returns 8 nonempty batches at a 128-row cap instead of 1,024; it also
verifies original row positions. A separate nested-array case verifies appended
values and offsets.
##########
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:
Fixed in a4abc561e1. Row-ID requests disable extra buffered read-ahead in
native predicate, output, and dictionary-probe readers; the policy propagates
to physical leaves while preserving demand-page reads. With default settings,
the 20-column one-row regression reduced retained stream buffers from
167,772,160 to 1,600,000 bytes and counted I/O from 167,821,312 to 1,649,752
bytes. The cached-remote test also passes without the 1 MiB override. All 274
related ASAN tests pass.
--
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]
