github-actions[bot] commented on code in PR #66802: URL: https://github.com/apache/doris/pull/66802#discussion_r3790993143
########## be/src/format_v2/file_scan_context.h: ########## @@ -0,0 +1,87 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include <condition_variable> +#include <cstdint> +#include <functional> +#include <memory> +#include <mutex> +#include <string> +#include <unordered_map> + +#include "common/status.h" +#include "gen_cpp/PlanNodes_types.h" + +namespace doris { + +// Opaque immutable metadata shared by all physical splits of one file. Concrete file formats own +// the derived type so the scanner and split-source layers do not depend on format internals. +class FileContext { +public: + virtual ~FileContext() = default; +}; + +class FileContextRegistry { +public: + using Loader = std::function<Status(std::shared_ptr<const FileContext>*)>; + + Status get_or_create(const std::string& key, const Loader& loader, + std::shared_ptr<const FileContext>* context); + +private: + struct Entry { + std::mutex lock; + std::condition_variable ready; + bool loading = true; + Status status; + std::shared_ptr<const FileContext> context; + }; + + std::mutex _lock; + std::unordered_map<std::string, std::shared_ptr<Entry>> _entries; Review Comment: [P1] Bound the scan-local footer registry. This map and each Entry retain strong ownership until FileScanLocalState teardown, while ParquetSharedFileContext owns either the parsed footer or an ObjLRUCache CacheHandle. A remote split source can therefore stream many unique files and make metadata memory/cache pins grow with the total file count, even after all children for earlier files finish. Please expire weak entries or explicitly retire/bound them once their children are no longer queued or active. ########## be/src/format_v2/parquet/parquet_file_context.cpp: ########## @@ -299,24 +301,65 @@ 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; + 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); + if (resolved_context == nullptr) { + if (file_context_registry != nullptr) { Review Comment: [P1] Do not reuse this registry entry for a mutable file with no stable version identity. `build_native_file_cache_key()` intentionally returns an empty key when mtime is zero and `!is_immutable`, because an overwrite can preserve path and size while changing both footer semantics and page bytes; this unconditional scan-local lookup nevertheless keys that unstable tuple. Ordinary Hive/TVF files can take this path, and generated children reopen the path while carrying the planner's old footer, so pruning or decoding can use old metadata against new bytes. Gate registry reuse on `has_stable_meta_cache_identity`; for refined children, pin/revalidate the opened version or decline splitting when no stable identity exists. ########## be/src/io/cache/cached_remote_file_reader.cpp: ########## @@ -800,6 +1236,145 @@ bool CachedRemoteFileReader::_try_read_from_cached_files_directly( return false; } +bool CachedRemoteFileReader::_read_from_memory_block_cache(size_t offset, Slice result, + ReadStatistics* stats) { + if (!_enable_reader_local_cache || _reader_local_file_cache == nullptr || _cache == nullptr) { + return false; + } + size_t current_offset = offset; + const size_t request_end = offset + result.size; + struct PinnedRead { + size_t block_offset; + size_t read_offset; + size_t result_offset; + size_t read_size; + FileScannerV2ReaderLocalFileCache::LookupResult lookup; + }; + // Parquet metadata and page reads normally span very few cache blocks. Keep their pins on the + // stack so the direct-memory hot path does not replace FileCache locking with heap allocation. + constexpr size_t INLINE_PINNED_READS = 4; + std::array<PinnedRead, INLINE_PINNED_READS> inline_pinned_reads {}; + std::vector<PinnedRead> overflow_pinned_reads; + size_t pinned_read_count = 0; + while (current_offset < request_end) { + const size_t block_offset = + current_offset / READER_LOCAL_CACHE_BLOCK_BYTES * READER_LOCAL_CACHE_BLOCK_BYTES; + const size_t read_end = + std::min(request_end, block_offset + READER_LOCAL_CACHE_BLOCK_BYTES); + const size_t read_size = read_end - current_offset; + PinnedRead pinned {.block_offset = block_offset, + .read_offset = current_offset, + .result_offset = current_offset - offset, + .read_size = read_size, + .lookup = {}}; + if (!_reader_local_file_cache->pin_if_present(block_offset, current_offset, read_size, + &pinned.lookup)) { + if (stats != nullptr && pinned_read_count != 0) { + stats->num_reader_local_cache_partial_miss++; + } + return false; + } + try { + if (pinned_read_count < INLINE_PINNED_READS) { + inline_pinned_reads[pinned_read_count] = std::move(pinned); + } else { + overflow_pinned_reads.push_back(std::move(pinned)); + } + } catch (...) { + // Optional hot-cache bookkeeping must never fail the scan under memory pressure. + return false; + } + ++pinned_read_count; + current_offset = read_end; + } + // Pin the complete request before copying. A partial probe must leave the caller's buffer + // untouched because the FileCache fallback will restart the request from its original offset. + auto copy_pinned_read = [&](const PinnedRead& pinned) { + memcpy(result.data + pinned.result_offset, + pinned.lookup.data->data() + pinned.read_offset - pinned.block_offset, + pinned.read_size); + if (pinned.lookup.file_block_to_touch != nullptr) { + _cache->add_need_update_lru_block(pinned.lookup.file_block_to_touch); + if (stats != nullptr) { + stats->num_reader_local_cache_disk_lru_touch++; + } + } + }; + for (size_t i = 0; i < std::min(pinned_read_count, INLINE_PINNED_READS); ++i) { + copy_pinned_read(inline_pinned_reads[i]); + } + for (const auto& pinned : overflow_pinned_reads) { + copy_pinned_read(pinned); + } + if (stats != nullptr) { + stats->num_reader_local_cache_total += cast_set<int64_t>(pinned_read_count); + stats->num_reader_local_cache_hit += cast_set<int64_t>(pinned_read_count); + stats->bytes_reader_local_cache_request += cast_set<int64_t>(result.size); + stats->bytes_read_from_reader_local_cache += cast_set<int64_t>(result.size); + } + return true; +} + +Status CachedRemoteFileReader::_read_local_block(const FileBlockSPtr& block, size_t file_offset, + size_t absolute_offset, Slice result, + ReadStatistics& stats, + bool bypass_reader_local_cache) { + if (!_enable_reader_local_cache || _reader_local_file_cache == nullptr || + bypass_reader_local_cache) { + return block->read(result, file_offset); + } + + size_t current_offset = absolute_offset; + const size_t request_end = absolute_offset + result.size; + while (current_offset < request_end) { + const size_t aligned_offset = + current_offset / READER_LOCAL_CACHE_BLOCK_BYTES * READER_LOCAL_CACHE_BLOCK_BYTES; + const size_t buffer_offset = std::max(aligned_offset, block->range().left); + const size_t buffer_end = std::min({aligned_offset + READER_LOCAL_CACHE_BLOCK_BYTES, + block->range().right + 1, size()}); + const size_t copy_end = std::min(request_end, buffer_end); + const size_t copy_size = copy_end - current_offset; + const size_t result_offset = current_offset - absolute_offset; + + stats.num_reader_local_cache_total++; + stats.bytes_reader_local_cache_request += cast_set<int64_t>(copy_size); + + const size_t buffer_size = buffer_end - buffer_offset; + FileScannerV2ReaderLocalFileCache::LookupResult lookup; + RETURN_IF_ERROR(_reader_local_file_cache->get_or_load( + buffer_offset, buffer_size, block, buffer_offset - block->range().left, &lookup)); + stats.num_reader_local_cache_evict += cast_set<int64_t>(lookup.evicted); + stats.num_reader_local_cache_admission_reject += lookup.admission_rejected ? 1 : 0; Review Comment: [P2] Record reader-local admission rejection before consuming this flag. LookupResult::admission_rejected defaults to false and is never set anywhere; in particular, the `_reserve()` failure path aborts and returns without changing it. Consequently the newly exposed ReaderLocalCacheAdmissionRejects counter is always zero under capacity/query/process memory pressure. Set the flag on policy/reservation denial and cover it in the existing rejection tests. ########## be/src/format_v2/parquet/parquet_reader.cpp: ########## @@ -506,6 +515,71 @@ 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"); + } + + 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<int64_t> row_group_first_rows; + 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, &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; Review Comment: [P1] Do not rebuild Iceberg delete indexes for every row-group child. Each child reuses the full source table-format descriptor, so its independent `prepare_split()` recopies/sorts the data file's complete position-delete vector and constructs a fresh equality predicate that rehashes every delete row; the cache shares only the raw mapping/Block. A file with R row groups and D deletes now pays O(R*D) setup and can retain one full delete vector/hash map per concurrent child, erasing the split benefit or exhausting memory. Share immutable prepared delete state across the children, or decline refinement for Iceberg splits with delete files until it can be shared. ########## be/src/format_v2/parquet/parquet_scan.cpp: ########## @@ -2310,6 +2376,9 @@ Status ParquetScanScheduler::read_filter_columns(int64_t batch_rows, const size_t idx = _predicate_indices_by_position_scratch.at(position); const auto& col = request.predicate_columns[idx]; const auto fid = col.column_id(); + if (_current_merge_range_reader != nullptr) { + RETURN_IF_ERROR(activate_merge_ranges_for_columns({fid})); Review Comment: [P2] Activate the columns that must be skipped after early rejection. This stages only the current predicate. If it rejects the whole batch, the code later calls `NativeColumnReader::skip()` for every unmaterialized predicate without activating their ranges; inside a selected span that skip parses headers and often loads page data. With the new empty initial MergeRange list, those reads delegate directly to remote storage and bypass both coalescing and exact-cache probing. Activate the skipped columns as one stage, or keep their cursor lag logical until they become reachable. ########## be/src/exec/scan/file_scanner_v2.cpp: ########## @@ -585,14 +616,29 @@ 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; } + if (_current_split.is_source_split) { + std::vector<FileScanSplit> generated_splits; + bool was_split = false; + RETURN_IF_ERROR(_table_reader->build_physical_splits(_current_split, &generated_splits, Review Comment: [P1] Preserve missing-file handling around split planning. `prepare_split()` has not opened the Parquet data file yet; `build_physical_splits()` now creates and initializes the temporary reader, so a missing object or footer EOF can first surface here. This direct RETURN_IF_ERROR bypasses the `ignore_not_found_file_in_external_table` and empty-file branches just above (and the equivalent get_block branches that handled this lazy init before the PR), causing configured ignorable files to fail scanner open. Classify this status through the same skip/abort/retire path before returning other errors. ########## be/src/format_v2/parquet/parquet_reader.cpp: ########## @@ -506,6 +515,71 @@ 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"); + } + + 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<int64_t> row_group_first_rows; + 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, &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); + child.size = cast_set<int64_t>(group_end - group_start); + // A source-level count is not valid for one generated row group. Child readers can still + // derive an exact count from the shared footer when aggregate pushdown is eligible. + child.clear_table_level_row_count = true; + child.file_context = _state->file_context.shared_file_context; + child.format_split_id = row_group_id; + splits->push_back(std::move(child)); + } + *was_split = true; Review Comment: [P2] Reuse the planning reader for a single selected row group. This method has already initialized the temporary Parquet reader, but returning a one-element child list closes it, aborts the source state, and repeats physical-reader, schema, and table-format/delete setup for the child with no parallelism to gain. This is common for one-row-group files. Retain or transfer that initialized reader into the prepared TableReader when there is exactly one selection; merely returning `was_split=false` without retaining it would still reinitialize the file later. ########## be/src/exec/scan/file_scanner_v2.cpp: ########## @@ -1095,7 +1158,10 @@ void FileScannerV2::update_realtime_counters() { _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_bytes_from_remote_storage( deltas.scan_bytes_from_remote_storage); - COUNTER_SET(_file_read_bytes_counter, bytes_read); + // Scanner instances share the profile counter, so publishing an absolute value would erase + // bytes already reported by sibling scanners. + COUNTER_UPDATE(_file_read_bytes_counter, Review Comment: [P2] Aggregate calls and time with deltas too. These counters are shared by sibling scanners for the same reason FileReadBytes is, and this PR makes one FE source produce row-group work that those scanners can consume, but the following COUNTER_SET calls publish each scanner's private cumulative values. With scanners at 10 and 20 calls, the final profile can show either 10 or 20 instead of 30; FileReadTime has the same loss, and the close path repeats it. Give both metrics per-scanner reported watermarks and COUNTER_UPDATE their deltas alongside bytes. ########## be/src/exec/scan/file_scanner_v2.cpp: ########## @@ -448,17 +458,37 @@ Status FileScannerV2::_open_impl(RuntimeState* state) { DORIS_CHECK(_table_reader != nullptr); RETURN_IF_ERROR(_init_expr_ctxes()); RETURN_IF_ERROR(_init_table_reader(_current_range)); + // Refine the first source split before yielding the scanner worker. Other scanners may be + // waiting for its row-group children, so deferring publication until a later get_block() + // turn could let those waiters occupy the scan thread pool ahead of the producer. + bool eos = false; + RETURN_IF_ERROR(_prepare_next_split(&eos)); } return Status::OK(); } Status FileScannerV2::_get_next_scan_range(bool* has_next) { SCOPED_TIMER(_get_next_range_timer); DORIS_CHECK(has_next != nullptr); - RETURN_IF_ERROR(_split_source->get_next(has_next, &_current_range)); + RETURN_IF_ERROR(_split_source->get_next_split(has_next, &_current_split)); if (*has_next) { + _current_range = _current_split.materialize_range(); RETURN_IF_ERROR(_validate_scan_range(*_params, _current_range)); + } else { + _current_split = {}; + _current_range = {}; + } + return Status::OK(); +} + +Status FileScannerV2::_complete_current_split(std::vector<FileScanSplit> generated_splits) { + if (!_current_split.is_source_split) { + return Status::OK(); } + RETURN_IF_ERROR( + _split_source->finish_source_split(_current_split, std::move(generated_splits))); + _current_split.is_source_split = false; + _state->update_num_finished_scan_range(1); Review Comment: [P2] Separate reservation retirement from finished-range progress. This increment now runs when an ordinary source is merely prepared, and for a refined parent before any queued row-group child is read; generated children never increment later. FE load progress periodically publishes this value, so it can report every range finished/100% while the actual data is still scanning. Retire the connector reservation here, but count the FE range only at unsplit EOF or after the final child completes. -- 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]
