github-actions[bot] commented on code in PR #66802:
URL: https://github.com/apache/doris/pull/66802#discussion_r3791511284
##########
be/src/format_v2/parquet/parquet_reader.cpp:
##########
@@ -551,6 +560,75 @@ 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();
+ }
+
+ 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];
+ 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) {
+ return Status::Corruption("Parquet row group {} column {} has
no metadata",
+ row_group_id, column_id);
+ }
+ native::ColumnChunkRange chunk_range;
+ RETURN_IF_ERROR(native::compute_column_chunk_range(
+ chunk.meta_data, file_size, compat.parquet_816_padding,
&chunk_range));
+ 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) {
Review Comment:
[P1] Skip zero-row groups before deriving a child range. Valid Parquet files
can contain an empty row group with zero-length column chunks; the ordinary
planner skips it, but refinement reaches this new Corruption branch first and
fails the whole source. Skip row_group.num_rows == 0 here and cover empty
groups before/between populated groups plus an all-empty file.
##########
be/src/format_v2/parquet/parquet_reader.cpp:
##########
@@ -551,6 +560,75 @@ 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) {
Review Comment:
[P2] Keep staged small HTTP files on the initialized reader. Iceberg marks
snapshot files immutable, so an HTTP(S) Parquet object at or below
in_memory_file_size passes this refinement gate; every generated child then
creates a private InMemoryFileReader and reloads/copies the entire object on
first access. Share the staged bytes across children or decline refinement for
this reader type, and cover a multi-row-group HTTP file.
##########
be/src/format_v2/parquet/parquet_file_context.cpp:
##########
@@ -299,24 +311,66 @@ Status ParquetFileContext::open(io::FileReaderSPtr
input_file_reader, io::IOCont
meta_cache_key.push_back(static_cast<char>(enable_mapping_varbinary));
meta_cache_key.push_back(static_cast<char>(enable_mapping_timestamp_tz));
}
- size_t native_footer_size = 0;
- if (has_stable_meta_cache_identity && meta_cache != nullptr &&
meta_cache->enabled() &&
- meta_cache->lookup(meta_cache_key, &native_meta_cache_handle)) {
- native_metadata =
native_meta_cache_handle.data<NativeParquetMetadata>();
- ++native_footer_cache_hits;
- } else {
- RETURN_IF_ERROR(parse_native_parquet_footer(
- native_file, &native_metadata_owner, &native_footer_size,
io_ctx,
- enable_mapping_varbinary, enable_mapping_timestamp_tz));
- ++native_footer_read_calls;
- if (has_stable_meta_cache_identity && meta_cache != nullptr &&
meta_cache->enabled()) {
- meta_cache->insert(meta_cache_key, native_metadata_owner.release(),
- &native_meta_cache_handle);
- native_metadata =
native_meta_cache_handle.data<NativeParquetMetadata>();
+ // The registry is scoped to one scan instance, whose splits describe one
planned snapshot.
+ // Normalize optional FE identity fields with reader values so equivalent
splits cannot miss
+ // single-flight merely because only one of them carried file size or
mtime.
+ const int64_t registry_mtime =
+ file_description.mtime != 0 ? file_description.mtime :
native_file->mtime();
+ const int64_t registry_file_size = file_description.file_size >= 0
+ ? file_description.file_size
+ :
cast_set<int64_t>(native_file->size());
+ const auto registry_path = native_file->path().native();
+ const std::string registry_key = fmt::format(
+
"fs[{}]={}::path[{}]={}::mtime={}::size={}::immutable={}::varbinary={}::timestamp_tz={"
+ "}",
+ file_description.fs_name.size(), file_description.fs_name,
registry_path.size(),
+ registry_path, registry_mtime, registry_file_size,
file_description.is_immutable,
+ enable_mapping_varbinary, enable_mapping_timestamp_tz);
+ auto load_context = [&](std::shared_ptr<const FileContext>* result) ->
Status {
+ auto loaded = std::make_shared<ParquetSharedFileContext>();
+ loaded->registry_key = registry_key;
+ loaded->has_stable_identity = has_stable_meta_cache_identity;
+ if (has_stable_meta_cache_identity && meta_cache != nullptr &&
meta_cache->enabled() &&
+ meta_cache->lookup(meta_cache_key,
&loaded->metadata_cache_handle)) {
+ loaded->metadata =
loaded->metadata_cache_handle.data<NativeParquetMetadata>();
+ ++native_footer_cache_hits;
+ } else {
+ size_t native_footer_size = 0;
+ RETURN_IF_ERROR(parse_native_parquet_footer(
+ native_file, &loaded->metadata_owner, &native_footer_size,
io_ctx,
+ enable_mapping_varbinary, enable_mapping_timestamp_tz));
+ ++native_footer_read_calls;
+ if (has_stable_meta_cache_identity && meta_cache != nullptr &&
meta_cache->enabled()) {
+ meta_cache->insert(meta_cache_key,
loaded->metadata_owner.release(),
+ &loaded->metadata_cache_handle);
+ loaded->metadata =
loaded->metadata_cache_handle.data<NativeParquetMetadata>();
+ } else {
+ loaded->metadata = loaded->metadata_owner.get();
+ }
+ }
+ DORIS_CHECK(loaded->metadata != nullptr);
+ *result = std::move(loaded);
+ return Status::OK();
+ };
+
+ std::shared_ptr<const FileContext> resolved_context =
std::move(file_context);
Review Comment:
[P2] Account for scan-local footer-context reuse. Supplied child contexts
and successful registry reuse bypass load_context(), so they increment neither
FileFooterReadCalls nor FileFooterHitCache; the new sibling/child test locks in
that invisible work. Add registry request/hit/wait/miss-or-load/bypass counters
while keeping physical reads and process-cache hits distinguishable.
##########
be/src/format_v2/parquet/parquet_reader.cpp:
##########
@@ -551,6 +560,75 @@ 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();
+ }
+
+ 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];
+ 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) {
+ return Status::Corruption("Parquet row group {} column {} has
no metadata",
+ row_group_id, column_id);
+ }
+ native::ColumnChunkRange chunk_range;
+ RETURN_IF_ERROR(native::compute_column_chunk_range(
+ chunk.meta_data, file_size, compat.parquet_816_padding,
&chunk_range));
+ 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) {
+ return Status::Corruption("Parquet row group {} has an empty
physical byte range",
+ row_group_id);
+ }
+ FileScanSplit child;
+ child.source_range = shared_source_range;
+ child.start_offset = cast_set<int64_t>(group_start);
Review Comment:
[P2] Separate GLOBAL_ROWID's file mapping from the physical child range.
This child start replaces the source start before
_create_global_rowid_context(), and IdFileMap plus RowIdStorageReader group by
path/start, so K row-group children create K second-phase scanner/file-reader
setups (and K footer reads when metadata cache is disabled) for one FE source.
Carry the source identity for GLOBAL_ROWID while retaining this child range for
first-phase reads, with an explicit batching policy and a multi-child TopN test.
##########
be/src/exec/operator/file_scan_operator.cpp:
##########
@@ -258,9 +258,54 @@ void FileScanLocalState::set_scan_ranges(RuntimeState*
state,
if (_split_source == nullptr) {
_split_source =
std::make_shared<LocalSplitSourceConnector>(scan_ranges, _max_scanners);
}
- // currently the total number of splits in the bach split mode cannot
be accurately obtained,
- // so we don't do it in the batch split mode.
- _max_scanners = std::min(_max_scanners,
_split_source->num_scan_ranges());
+ // A single FE Parquet split can publish many row-group children after
its footer is read.
+ // Keep the requested scanner concurrency in that case; capping it to
the initial range
+ // count would leave the generated children serial even though they
share one footer.
+ bool can_generate_parquet_splits = false;
+ const TFileScanRangeParams* common_params = nullptr;
+ if (state->get_query_ctx() != nullptr &&
+
state->get_query_ctx()->file_scan_range_params_map.contains(parent_id())) {
+ common_params =
&state->get_query_ctx()->file_scan_range_params_map[parent_id()];
+ }
+ for (const auto& scan_range_params : scan_ranges) {
+ const auto& file_scan_range =
+
scan_range_params.scan_range.ext_scan_range.file_scan_range;
+ const auto* params =
+ file_scan_range.__isset.params ? &file_scan_range.params :
common_params;
+ if (params == nullptr) {
+ continue;
+ }
+ const bool is_load =
+
state->desc_tbl().get_tuple_descriptor(params->src_tuple_id) != nullptr;
+ if (!_should_use_file_scanner_v2(state->query_options(), is_load,
*params)) {
+ continue;
+ }
+ can_generate_parquet_splits =
+ std::ranges::any_of(file_scan_range.ranges, [&](const
auto& range) {
+ const auto format =
+ range.__isset.format_type ? range.format_type
: params->format_type;
+ if (format == TFileFormatType::FORMAT_PARQUET) {
Review Comment:
[P2] Keep the range-count cap when downstream disables refinement. This
returns true for every Parquet range, but _can_refine_source_split() rejects an
Iceberg range with nonempty delete_files. A one-range scan can therefore
construct up to the configured scanner maximum; all but the source owner block
in get_next_split() while it initializes delete state, although no child can
ever be published. Mirror the delete guard here and test the resulting scanner
count.
--
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]
