github-actions[bot] commented on code in PR #66602:
URL: https://github.com/apache/doris/pull/66602#discussion_r3801073085
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -1004,8 +1636,10 @@ Status
CachedRemoteFileReader::_read_remaining_blocks_from_cache(
SCOPED_RAW_TIMER(&stats.local_read_timer);
SCOPED_CONCURRENCY_COUNT(
ConcurrencyStatsManager::instance().cached_remote_reader_local_read);
- st = block->read(Slice(result.data + (current_offset -
offset), read_size),
- file_offset);
+ st = _read_local_block(block, file_offset, current_offset,
+ Slice(result.data + (current_offset -
offset), read_size),
+ stats,
+ io_ctx != nullptr &&
io_ctx->bypass_reader_local_cache);
Review Comment:
[P2] Split reader-local hits out of local-storage bytes — The all-or-nothing
probe can miss for a request spanning one resident reader-local cell and one
cold cell. This helper then serves the hot part from RAM and fills only the
cold part from FileCache, but the caller still adds the full `read_size` to
`source_read_breakdown.local_bytes`. The hot prefix is therefore counted both
in `bytes_read_from_reader_local_cache` and as physical local-storage I/O;
`FileScannerV2::_collect_realtime_counter_deltas()` propagates that overcount
into local-storage scan bytes. Please have `_read_local_block()` return the
actual disk-served byte split (and cover the existing cross-block partial-hit
test), rather than charging every successful byte to disk.
##########
be/src/exec/scan/split_source_connector.cpp:
##########
@@ -85,4 +168,65 @@ Status RemoteSplitSourceConnector::get_next(bool* has_next,
TFileRangeDesc* rang
return Status::OK();
}
+Status RemoteSplitSourceConnector::get_next_split(bool* has_next,
FileScanSplitTask* task) {
+ DORIS_CHECK(has_next != nullptr && task != nullptr);
+ std::unique_lock lock(_scan_range_lock);
+ while (true) {
+ if (_take_generated_split(task)) {
+ *has_next = true;
+ return Status::OK();
+ }
+ if (_scan_index == _scan_ranges.size() && !_last_batch) {
+ SCOPED_TIMER(_get_split_timer);
+ Status coord_status;
+ FrontendServiceConnection
coord(_state->exec_env()->frontend_client_cache(),
+
_state->get_query_ctx()->coord_addr, &coord_status);
+ RETURN_IF_ERROR(coord_status);
+ TFetchSplitBatchRequest request;
+ request.__set_split_source_id(_split_source_id);
+
request.__set_max_num_splits(config::remote_split_source_batch_size);
+ TFetchSplitBatchResult result;
+ try {
+ coord->fetchSplitBatch(result, request);
Review Comment:
[P2] Release the split lock while fetching the next remote batch — This
synchronous RPC runs while `_scan_range_lock` is held. Once the current remote
batch has handed out its parent tasks, the next scanner can enter here while
sibling scanners are still expanding those parents; their
`finish_file_parent()` calls then block on the same lock, so ready Row Group
children cannot be enqueued and the parent count/waiter state cannot advance
until FE returns. A slow coordinator fetch therefore stalls otherwise-ready
local work. Please serialize fetchers separately and drop this
state/publication lock around the RPC, then merge the returned batch under the
lock.
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -1151,8 +1910,11 @@ Status
CachedRemoteFileReader::_read_remote_only_on_cache_miss(
g_skip_local_cache_io_sum_bytes << read_size;
} else {
SCOPED_RAW_TIMER(&stats.local_read_timer);
- Status st = block->read(Slice(result.data + (read_left - offset),
read_size),
- read_left - block_range.left);
+ // Remote-only governs misses, not downloaded hits. Promote the
full cache cell so
+ // repeated Parquet page/header reads can reuse memory without
another disk read.
+ Status st =
+ _read_local_block(block, read_left - block_range.left,
read_left,
Review Comment:
[P2] Honor the MergeRange promotion bypass on remote-only hits —
`MergeRangeFileReader::_fill_box()` sets `bypass_reader_local_cache` because it
already retains the merged slice, but this downloaded-hit branch calls
`_read_local_block()` with the default `false`. When the exact block-aligned
probe misses while the unaligned merged extent is fully cached, a remote-only
scan reaches this branch and promotes reader-local cells in addition to the
MergeRange boxes, defeating the explicit single-owner memory bound. Please pass
`io_ctx->bypass_reader_local_cache` here and add a CachedRemote + MergeRange
remote-only regression that keeps reader-local fill bytes at zero.
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -587,6 +596,83 @@ Status build_native_prefetch_ranges(
namespace detail {
+Status build_native_row_group_split_ranges(const tparquet::FileMetaData&
metadata, size_t file_size,
+ int64_t target_split_size,
+ std::vector<ParquetScanRange>*
ranges) {
+ DORIS_CHECK(ranges != nullptr);
+ ranges->clear();
+ ranges->reserve(metadata.row_groups.size());
+ for (size_t row_group_idx = 0; row_group_idx < metadata.row_groups.size();
++row_group_idx) {
+ const auto& row_group = metadata.row_groups[row_group_idx];
+ if (row_group.columns.empty()) {
+ return Status::Corruption("Parquet row group {} has no column
chunks", row_group_idx);
+ }
+ size_t group_start = std::numeric_limits<size_t>::max();
+ size_t group_end = 0;
+ for (size_t column_idx = 0; column_idx < row_group.columns.size();
++column_idx) {
+ const auto& chunk = row_group.columns[column_idx];
+ if (!chunk.__isset.meta_data) {
+ return Status::Corruption("Parquet row group {} column {} has
no metadata",
+ row_group_idx, column_idx);
+ }
+ native::ColumnChunkRange chunk_range;
+
RETURN_IF_ERROR(native::compute_column_chunk_range(chunk.meta_data, file_size,
false,
+ &chunk_range));
+ group_start = std::min(group_start, chunk_range.offset);
+ group_end = std::max(group_end, chunk_range.offset +
chunk_range.length);
+ }
+ // Compatibility padding belongs to physical reads, not scheduling
ownership; keeping the
+ // split boundary raw preserves the non-overlapping scan-range
ownership invariant.
+ ranges->push_back({.start_offset = cast_set<int64_t>(group_start),
Review Comment:
[P2] Reject zero-length children for nonempty Row Groups —
`compute_column_chunk_range()` accepts `total_compressed_size == 0`, so a
malformed one-column Row Group with `num_rows > 0` reaches this line with
`group_start == group_end` and publishes a zero-length child. Child selection
later requires `group_start < range_end`; for this range that is false, so the
Row Group disappears and the scan returns successful EOF instead of rejecting
the malformed payload. Please reject a zero raw extent for a nonempty Row Group
before publishing tasks, and add a flat one-column regression with matching
positive `num_rows`/`num_values` and zero compressed size.
--
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]
