github-actions[bot] commented on code in PR #66802:
URL: https://github.com/apache/doris/pull/66802#discussion_r3791849952
##########
be/src/format_v2/parquet/parquet_reader.cpp:
##########
@@ -551,6 +574,83 @@ Status ParquetReader::init(RuntimeState* state) {
return Status::OK();
}
+Status ParquetReader::build_physical_splits(const FileScanSplit& source_split,
+ std::vector<FileScanSplit>* splits,
+ bool* was_split) const {
+ DORIS_CHECK(splits != nullptr);
+ DORIS_CHECK(was_split != nullptr);
+ splits->clear();
+ *was_split = false;
+ if (_state == nullptr || _state->file_context.native_metadata == nullptr ||
+ _state->file_context.shared_file_context == nullptr) {
+ return Status::Uninitialized("ParquetReader is not open");
+ }
+ if (!_state->file_context.shared_file_context->has_stable_identity) {
+ // A path and size do not identify a mutable remote object. Keep the
initialized parent
+ // reader instead of publishing children whose shared footer could
become stale.
+ return Status::OK();
+ }
+ if (!_state->file_context.can_refine_physical_splits()) {
+ return Status::OK();
+ }
+
+ ParquetScanRange scan_range {
+ .start_offset =
+ source_split.range.__isset.start_offset ?
source_split.range.start_offset : 0,
+ .size = source_split.range.__isset.size ? source_split.range.size
: -1,
+ .file_size = source_split.range.__isset.file_size ?
source_split.range.file_size
+ :
_file_description->file_size,
+ };
+ std::vector<int> selected_row_groups;
+ RETURN_IF_ERROR(detail::select_native_row_groups_by_scan_range(
+ _state->file_context.native_metadata->to_thrift(), scan_range,
+ _state->file_context.native_metadata->row_group_first_rows(),
&selected_row_groups));
+ const auto& metadata = _state->file_context.native_metadata->to_thrift();
+ const auto compat = native::parquet_reader_compat(
+ metadata.__isset.created_by ? metadata.created_by : std::string
{});
+ const size_t file_size = _state->file_context.native_file->size();
+ auto shared_source_range =
std::make_shared<TFileRangeDesc>(source_split.range);
+ splits->reserve(selected_row_groups.size());
+ for (const int row_group_id : selected_row_groups) {
+ const auto& row_group = metadata.row_groups[row_group_id];
+ if (row_group.num_rows == 0) {
+ // Empty row groups are valid and the ordinary scan planner
ignores them. Refinement
+ // must preserve that behavior before inspecting their potentially
empty chunks.
+ continue;
+ }
+ size_t group_start = std::numeric_limits<size_t>::max();
+ size_t group_end = 0;
+ for (size_t column_id = 0; column_id < row_group.columns.size();
++column_id) {
Review Comment:
[P1] Preserve positive-row root-only Parquet files when refining splits. The
native schema contract explicitly accepts a required root with zero physical
fields for metadata-only `COUNT(*)`, and a row group with `num_rows > 0` and no
chunks passes metadata validation. The ordinary full-file planner selects that
group and synthesizes its rows without column readers, but this loop leaves the
chunk-derived range empty and turns the same file into Corruption during split
planning. If a selected nonempty group has no chunks, clear any tentative
children, decline refinement, and keep the initialized source reader (or
support an exact child without a physical byte envelope); add a positive-row
root-only COUNT regression. This is distinct from the existing zero-row-group
thread: those groups can be skipped, while this one contains real rows.
##########
be/src/exec/scan/file_scanner_v2.cpp:
##########
@@ -585,15 +643,52 @@ Status FileScannerV2::_prepare_next_split(bool* eos) {
// advance exactly one scan range and preserve later files in the
same scan.
RETURN_IF_ERROR(_table_reader->abort_split());
COUNTER_UPDATE(_empty_file_counter, 1);
- _state->update_num_finished_scan_range(1);
+ RETURN_IF_ERROR(_complete_current_split());
continue;
}
RETURN_IF_ERROR(status);
if (_table_reader->current_split_pruned()) {
- _state->update_num_finished_scan_range(1);
+ RETURN_IF_ERROR(_complete_current_split());
continue;
}
- COUNTER_UPDATE(_file_counter, 1);
+ _update_file_counter(_file_counter, _current_split);
+ if (_current_split.is_source_split &&
can_refine_source_split(_current_range)) {
+ std::vector<FileScanSplit> generated_splits;
+ bool was_split = false;
+ const auto split_status = _table_reader->build_physical_splits(
Review Comment:
[P2] Skip row-group refinement when only one scanner can consume the
children. In serial mode, or when `max_file_scanners_concurrency=1` yields one
constructed scanner, this call initializes and closes a planning reader, then
makes that same scanner open the generated children one by one. That turns one
reader lifecycle into one planning lifecycle plus R child lifecycles without
any parallelism; the shared footer avoids reparsing metadata, but each child
still repeats physical-delegate creation, schema/mapping/filter/request setup,
and row-group planning. Gate refinement on the actual constructed scanner count
and keep the ordinary unsplit path when it is below two; cover a
multi-row-group file with one scanner.
--
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]
