Gabriel39 commented on code in PR #66802:
URL: https://github.com/apache/doris/pull/66802#discussion_r3795513982
##########
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:
Fixed in eef1413e7e7. TableReader now asks the opened planning reader for a
metadata-only aggregate result before publishing children and caches a
successful source-level result on that reader. Parquet accepts COUNT(*),
required primitive COUNT(col), and safe MIN/MAX; nullable/nested COUNT and
missing or unsafe MIN/MAX statistics return NotSupported and continue with
row-group refinement.
##########
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:
Fixed in eef1413e7e7. Parquet row-group readers share a synchronized
adaptive snapshot through the scan-local file context. Predicate statistics,
survival ratio, batch sequence, and empty-probe width are restored by an
independent semantic predicate digest; late-RF snapshots remain isolated, and
stale concurrent publications cannot regress warm state.
##########
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:
Fixed in eef1413e7e7. Generated siblings now publish a synchronized
source-level bytes-per-row hint, seed later scanner predictors from it, and
count the first source sample only once under concurrent publication. Parquet
also carries the widened empty-predicate probe across children so filtered
prefixes do not restart at 32 rows.
--
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]
