github-actions[bot] commented on code in PR #66802:
URL: https://github.com/apache/doris/pull/66802#discussion_r3793892938
##########
be/src/format_v2/table_reader.cpp:
##########
@@ -1249,6 +1368,102 @@ Status TableReader::prepare_split(const
SplitReadOptions& options) {
return _parse_delete_predicates(options);
}
+Status TableReader::build_physical_splits(const FileScanSplit& source_split,
+ std::vector<FileScanSplit>* splits,
bool* was_split) {
+ SCOPED_TIMER(_profile.total_timer);
+ DORIS_CHECK(splits != nullptr);
+ DORIS_CHECK(was_split != nullptr);
+ splits->clear();
+ *was_split = false;
+ if (_format != FileFormat::PARQUET || _current_split_pruned ||
+ _current_split_uses_metadata_count || _current_task == nullptr) {
Review Comment:
[P2] Keep footer-only aggregates on the planning reader. This guard only
recognizes the FE table-level count, but after `open_reader()` the ready
Parquet reader can also answer `COUNT(*)`, required-primitive `COUNT(col)`, and
successful `MIN/MAX` from the complete source footer plan. Refinement closes it
and performs R child create/map/plan/aggregate/close lifecycles; `MIN/MAX`
additionally sends 2R synthetic rows to the upper aggregate instead of two.
Retain the planning reader when the source-level metadata aggregate succeeds
(falling back to refinement for level-reading COUNT or unsafe/missing MIN/MAX
stats), and cover these multi-row-group cases.
##########
be/src/exec/scan/file_scanner_v2.cpp:
##########
@@ -585,15 +656,53 @@ 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 &&
+ _should_refine_source_split(_current_range,
_constructed_scanners)) {
+ std::vector<FileScanSplit> generated_splits;
+ bool was_split = false;
+ const auto split_status = _table_reader->build_physical_splits(
+ _current_split, &generated_splits, &was_split);
+ const auto ignored_split_status = _classify_ignored_split_status(
+ split_status,
config::ignore_not_found_file_in_external_table,
+ _should_stop || _io_ctx->should_stop);
+ if (ignored_split_status == IgnoredSplitStatus::NOT_FOUND) {
+ RETURN_IF_ERROR(_table_reader->abort_split());
+ COUNTER_UPDATE(_not_found_file_counter, 1);
+ RETURN_IF_ERROR(_complete_current_split());
+ continue;
+ }
+ if (ignored_split_status == IgnoredSplitStatus::EMPTY) {
+ RETURN_IF_ERROR(_table_reader->abort_split());
+ COUNTER_UPDATE(_empty_file_counter, 1);
+ RETURN_IF_ERROR(_complete_current_split());
+ continue;
+ }
+ RETURN_IF_ERROR(split_status);
+ if (was_split) {
+ RETURN_IF_ERROR(_table_reader->abort_split());
+ const bool has_children = !generated_splits.empty();
+
RETURN_IF_ERROR(_retire_current_source_split(std::move(generated_splits)));
Review Comment:
[P2] Preserve adaptive batch history across row-group children. Every
generated child re-enters `_prepare_next_split()`, which resets the predictor
and starts a fresh 32-row probe; the unsplit reader learned table-block bytes
per row once and reused that cap across later row groups. A source with R
surviving groups can now pay R scanner learning probes, and a filtered prefix
makes each child repeat the cold 32/256/1024/... predicate ramp before its
first width sample. Retain or seed the scanner-owned predictor for siblings
with the same source identity (with a safe synchronized hint for other
scanners), and cover a multi-row-group source by bounding
`AdaptiveBatchProbeCount` per FE range.
##########
be/src/format_v2/parquet/parquet_reader.cpp:
##########
@@ -551,6 +574,97 @@ Status ParquetReader::init(RuntimeState* state) {
return Status::OK();
}
+Status ParquetReader::build_physical_splits(std::vector<PhysicalFileSplit>*
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 = _file_description->range_start_offset,
+ .size = _file_description->range_size,
+ .file_size = _file_description->file_size,
+ };
+ std::vector<int> selected_row_groups;
+ if (_state->scan_plan != nullptr) {
+ selected_row_groups.reserve(_state->scan_plan->row_groups.size());
+ for (const auto& row_group_plan : _state->scan_plan->row_groups) {
+ selected_row_groups.push_back(row_group_plan.row_group_id);
+ }
+ } else {
+ 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();
+ 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;
+ }
+ if (row_group.columns.empty()) {
+ // A root-only schema can still carry rows for metadata COUNT(*),
but it has no byte
+ // envelope that can identify a child. Keep the initialized source
reader so those
+ // rows are not turned into a corruption error or discarded after
tentative children.
+ splits->clear();
+ return Status::OK();
+ }
+ 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) {
+ const auto& chunk = row_group.columns[column_id];
+ if (!chunk.__isset.meta_data) {
+ // Scheduling envelopes span every chunk, including columns
the current request
+ // does not read. If an unused chunk cannot prove a safe
envelope, retain the
+ // initialized source reader so projected-column validation
remains authoritative.
+ splits->clear();
+ return Status::OK();
+ }
+ native::ColumnChunkRange chunk_range;
+ const auto range_status = native::compute_column_chunk_range(
+ chunk.meta_data, file_size, compat.parquet_816_padding,
&chunk_range);
+ if (!range_status.ok()) {
+ splits->clear();
+ return Status::OK();
+ }
+ group_start = std::min(group_start, chunk_range.offset);
+ group_end = std::max(group_end, chunk_range.offset +
chunk_range.length);
+ }
+ if (group_end <= group_start) {
+ splits->clear();
+ return Status::OK();
+ }
+ PhysicalFileSplit child;
+ child.start_offset = cast_set<int64_t>(group_start);
+ child.size = cast_set<int64_t>(group_end - group_start);
+ child.file_context = _state->file_context.shared_file_context;
+ child.format_split_id = row_group_id;
Review Comment:
[P2] Preserve Parquet adaptive state across row-group children. An unsplit
scheduler keeps `_predicate_runtime_stats` and `_predicate_survival_ratio`
across `reset_current_row_group()`, but each generated child constructs a fresh
scheduler. Each child with eligible staged predicates therefore restarts in
request order and repeats warmup sampling; on non-MergeRange prefetch paths it
also warms the full cold candidate set and cannot use learned high survival to
prefetch output chunks before its first surviving predicate batch. The source
previously reused those decisions on later groups. Carry a synchronized
source-level stats snapshot keyed to the predicate snapshot (or seed later
siblings), and test both ordering and output-prefetch reuse across multiple
children.
--
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]
