github-actions[bot] commented on code in PR #66802:
URL: https://github.com/apache/doris/pull/66802#discussion_r3791121669
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -800,6 +1237,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(
Review Comment:
[P2] Avoid promoting a full 256 KiB region on every first local-cache hit by
default. This call receives the aligned `buffer_size`, so even the added
8-byte-read test performs a 256 KiB FileCache disk read and allocation. Sparse
footer/index/page probes can touch distinct regions, and generated children
have separate per-reader maps, so neighboring bytes need never be reused while
the default-on cache churns its 64 MiB per-scanner budget. This is distinct
from the admission-rejection counter thread: the fill succeeds but performs
excessive I/O. Admit after observed reuse, fill the requested/planned span, or
keep this disabled until representative warm sparse-cache benchmarks show the
default is safe.
##########
be/src/format_v2/parquet/parquet_reader.cpp:
##########
@@ -506,6 +515,76 @@ 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<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));
Review Comment:
[P2] Reuse the row-prefix calculation across generated children. This parent
call computes `row_group_first_rows` for the whole footer, but each child
receives only `format_split_id`; when that child opens,
`select_native_row_groups_by_scan_range` allocates the same R-entry vector and
walks all R row groups again even for an exact id. Refining a file into R
children therefore adds O(R^2) footer work and O(C*R) concurrent scratch for
files with many row groups. This is distinct from the Iceberg delete-state
thread because it applies to every refined Parquet file. Retain the immutable
prefix in the shared context or carry each child's `first_file_row` so exact-id
planning is O(1).
##########
be/src/io/fs/buffered_reader.cpp:
##########
@@ -397,6 +494,39 @@ Status MergeRangeFileReader::_fill_box(int range_index,
size_t start_offset, siz
return Status::OK();
}
+void MergeRangeFileReader::_record_merged_read(int range_index, size_t
start_offset,
+ size_t bytes_read) {
+ if (bytes_read == 0) {
+ return;
+ }
+ if (range_index < 0) {
+ _statistics.merged_useful_bytes += bytes_read;
+ return;
+ }
+ const size_t read_end = start_offset + bytes_read;
+ size_t useful_bytes = 0;
+ size_t future_predicate_bytes = 0;
+ for (size_t index = static_cast<size_t>(range_index);
+ index < _random_access_ranges.size() &&
+ _random_access_ranges[index].start_offset < read_end;
+ ++index) {
+ const auto& range = _random_access_ranges[index];
+ const size_t overlap_start = std::max(start_offset,
range.start_offset);
+ const size_t overlap_end = std::min(read_end, range.end_offset);
+ if (overlap_start >= overlap_end) {
+ continue;
+ }
+ const size_t overlap = overlap_end - overlap_start;
+ useful_bytes += overlap;
+ if (_range_stages[index] > _range_stages[range_index]) {
+ future_predicate_bytes += overlap;
+ }
+ }
+ _statistics.merged_useful_bytes += useful_bytes;
+ _statistics.merged_gap_bytes += bytes_read - useful_bytes;
Review Comment:
[P2] Compute useful bytes over the union of eager ranges. PARQUET-816
padding can make adjacent column-chunk ranges overlap, while the eager path
only sorts those ranges before the constructor; only staged additions are
coalesced. This loop then counts the overlap once per range, so an 8-byte read
in two padded ranges records 16 useful bytes and the unsigned `bytes_read -
useful_bytes` expression wraps before it is stored, publishing an invalid gap
counter. Coalesce constructor inputs with the staged path or measure union
coverage, and cover eager overlapping ranges in the counter test.
--
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]
