github-actions[bot] commented on code in PR #66602:
URL: https://github.com/apache/doris/pull/66602#discussion_r3747038696
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -889,9 +903,27 @@ private CloseableIterable<FileScanTask>
splitFiles(TableScan scan) {
}
targetSplitSize = determineTargetFileSplitSize(fileScanTaskList);
+ if (sessionVariable.enableFileScannerV2) {
Review Comment:
In the ordinary non-batch path, preserving this Parquet task leaves one FE
range for a backend. `FileScanLocalState::set_scan_ranges()` then clamps
`_max_scanners` to `LocalSplitSourceConnector::num_scan_ranges()`, so a
one-file scan creates exactly one `FileScannerV2`; its generated Row Group
children are all consumed serially and the former multi-split scan can collapse
to one core. Batch mode avoids this clamp, but local mode does not. Please make
the scanner-count calculation expansion-aware (and cover one local parent
through production scanner construction); the native Paimon parent path needs
the same treatment.
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -118,15 +638,25 @@ static bool use_remote_only_on_cache_miss(const
IOContext* io_ctx) {
CachedRemoteFileReader::CachedRemoteFileReader(FileReaderSPtr
remote_file_reader,
const FileReaderOptions& opts)
: _is_doris_table(opts.is_doris_table),
+ _enable_reader_local_cache(opts.enable_reader_local_cache &&
+ opts.reader_local_cache != nullptr &&
!opts.is_doris_table),
_tablet_id(opts.tablet_id),
_storage_resource_id(opts.storage_resource_id),
- _remote_file_reader(std::move(remote_file_reader)) {
+ _remote_file_reader(std::move(remote_file_reader)),
+ _reader_local_cache(opts.reader_local_cache) {
DCHECK(!_is_doris_table || _tablet_id > 0);
if (_is_doris_table) {
_init_doris_table_cache();
} else {
_init_external_table_cache(opts);
}
+ if (_enable_reader_local_cache) {
+ // Path, version, and size form a stable physical-file identity across
the row-group
+ // readers created from one parent task without extending the
aggregate reader options.
+ _reader_local_file_cache =
_reader_local_cache->get_or_create_file_cache(
+ fmt::format("{}:{}:{}", path().native(), opts.mtime,
opts.file_size));
Review Comment:
This shared map needs a stable physical-file identity, but the key drops
`fs_name`/storage identity and accepts the unknown sentinels `mtime == 0` and
`file_size == -1`. In particular, `HdfsFileReader` normalizes away the
nameservice, while `TFileRangeDesc` permits different `fs_name` values in one
scan, so equal paths/mtime/size on two filesystems alias here; the second
reader can return the first file's promoted bytes before touching its own
storage. Please use the canonical filesystem/resource plus path, reliable
version, and actual-size fallback, and disable sharing for a mutable file whose
version is unknown (as `build_native_file_cache_key()` already does).
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -1177,17 +1211,38 @@ Status ParquetScanScheduler::open_next_row_group(
RETURN_IF_ERROR(detail::build_native_prefetch_ranges(
thrift_metadata, file_schema, request_scan_columns(request),
row_group_idx,
file_context.native_file->size(), compat.parquet_816_padding,
&native_ranges));
+ // Local readers benefit from one eager coalescing plan; splitting their
ranges by predicate
+ // stage only adds small reads. Remote readers can avoid future-stage IO,
and exact cache hits
+ // can bypass their merge path altogether.
+ const bool defer_merge_ranges =
file_context.native_file_should_defer_merge_ranges();
if (request.non_predicate_positions.empty()) {
_current_merge_range_active =
file_context.set_native_random_access_ranges(
native_ranges,
detail::average_prefetch_range_size(native_ranges), _profile,
- _merge_read_slice_size);
+ _merge_read_slice_size, !defer_merge_ranges);
} else {
// Independent predicate/output readers may revisit the same physical
leaf at different
// cursors. MergeRangeFileReader has one consumptive cache per range,
so use the random
// access reader for this layout instead of sharing one sequential
range cache.
_current_merge_range_active =
file_context.set_native_random_access_ranges(
Review Comment:
This is the predicate-plus-lazy-output layout that needs staged activation,
but passing `{}` makes `should_use_merge_range_reader()` return false.
Consequently `_current_merge_range_reader` remains null, the per-column ranges
below are never recorded, and every predicate/lazy activation hook later in the
scheduler is a no-op; remote scans with projected payload still fall back to
small direct reads. Please make the staged path reachable with a design that
preserves the independent predicate/output cursor contract (the current single
consumptive cache cannot simply be enabled for revisited leaves), and add an
integration test proving predicate activation plus survivor-only lazy
activation.
##########
be/src/format_v2/parquet/parquet_reader.cpp:
##########
@@ -506,6 +522,50 @@ Status ParquetReader::init(RuntimeState* state) {
return Status::OK();
}
+Status ParquetReader::build_split_tasks(const TFileRangeDesc& parent,
+ std::vector<FileScanSplitTask>*
children) {
+ DORIS_CHECK(children != nullptr);
+ children->clear();
+ if (_state == nullptr || _state->file_context.native_metadata == nullptr ||
+ _state->file_context.shared_metadata == nullptr) {
+ return Status::Uninitialized("ParquetReader is not open");
+ }
+ 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 = _file_description->file_size < 0
+ ? _state->file_context.native_file->size()
+ :
cast_set<size_t>(_file_description->file_size);
+ auto context =
std::make_shared<ParquetFileSplitContext>(_state->file_context.shared_metadata);
+ children->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(
Review Comment:
The compatibility padding is a decoder read extent, not a disjoint ownership
boundary. For an affected file with adjacent one-column chunks
`(offset=100,size=20)` and `(150,20)`, this call pads them to child ranges
`[100,220)` and `[150,270)`. Selection later recomputes padded Row Group
midpoints 160 and 210, both of which lie in both children, so each child reads
both Row Groups and duplicates the rows. Please keep padding out of task
ownership (for example, carry the exact Row Group ordinal in the child context)
and add an old parquet-mr fixture that asserts one distinct group per child.
##########
be/src/format_v2/parquet/parquet_reader.cpp:
##########
@@ -506,6 +522,50 @@ Status ParquetReader::init(RuntimeState* state) {
return Status::OK();
}
+Status ParquetReader::build_split_tasks(const TFileRangeDesc& parent,
+ std::vector<FileScanSplitTask>*
children) {
+ DORIS_CHECK(children != nullptr);
+ children->clear();
+ if (_state == nullptr || _state->file_context.native_metadata == nullptr ||
+ _state->file_context.shared_metadata == nullptr) {
+ return Status::Uninitialized("ParquetReader is not open");
+ }
+ 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 = _file_description->file_size < 0
+ ? _state->file_context.native_file->size()
+ :
cast_set<size_t>(_file_description->file_size);
+ auto context =
std::make_shared<ParquetFileSplitContext>(_state->file_context.shared_metadata);
+ children->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, 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);
+ }
+ TFileRangeDesc child = parent;
Review Comment:
Each generated child opens a fresh `ParquetFileContext`, but small HTTP(S)
files are wrapped in a fresh `InMemoryFileReader` before the shared-footer
check. The parent therefore stages the complete object to parse its footer, and
each of N Row Group children stages the complete object again on first data
access—N+1 full downloads without FileCache, or N+1 full cache reads and copies
with it. Please share the staged immutable buffer across children or skip Row
Group expansion for this whole-object HTTP path, and add a multi-Row-Group HTTP
test that asserts one full-object load.
##########
be/src/exec/scan/file_scanner_v2.cpp:
##########
@@ -584,12 +647,34 @@ Status FileScannerV2::_prepare_next_split(bool* eos) {
// get_block() follows the symmetric branch in _get_block_impl().
Both paths must
// advance exactly one scan range and preserve later files in the
same scan.
RETURN_IF_ERROR(_table_reader->abort_split());
+ if (is_file_parent) {
+ RETURN_IF_ERROR(finish_file_parent({}));
+ }
COUNTER_UPDATE(_empty_file_counter, 1);
_state->update_num_finished_scan_range(1);
continue;
}
- RETURN_IF_ERROR(status);
+ if (!status.ok()) {
+ if (is_file_parent) {
+ RETURN_IF_ERROR(finish_file_parent({}));
+ }
+ return status;
+ }
if (_table_reader->current_split_pruned()) {
+ if (is_file_parent) {
+ RETURN_IF_ERROR(finish_file_parent({}));
+ }
+ _state->update_num_finished_scan_range(1);
+ continue;
+ }
+ if (is_file_parent) {
+ std::vector<FileScanSplitTask> children;
+ const auto build_status =
_table_reader->build_file_split_tasks(&children);
Review Comment:
A parent is deliberately not opened by `prepare_split()`, so a missing file
is first observed here. This branch releases the parent and returns
`build_status` directly, bypassing the `_should_skip_not_found()` policy above;
the same stale listing is skipped as an ordinary split when
`ignore_not_found_file_in_external_table` is enabled but now fails the query
when it is marked as a parent. Please route parent-build failures through the
same NOT_FOUND handling, then continue to the next split after finishing the
parent.
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -107,6 +113,520 @@ bvar::Adder<uint64_t>
g_peer_cross_compute_group_read("peer_cross_compute_group_
bvar::Adder<uint64_t>
g_peer_same_compute_group_read("peer_same_compute_group_read");
bvar::Adder<uint64_t> g_peer_lazy_fetch_triggered("peer_lazy_fetch_triggered");
+FileScannerV2ReaderLocalCache::FileScannerV2ReaderLocalCache(
+ size_t capacity, std::shared_ptr<doris::MemTrackerLimiter>
query_mem_tracker)
+ : _capacity(capacity),
+ _query_mem_tracker(std::move(query_mem_tracker)),
+
_memory_tracker(std::make_shared<doris::MemTracker>("FileScannerV2ReaderLocalCache"))
{}
+
+FileScannerV2ReaderLocalCache::~FileScannerV2ReaderLocalCache() {
+ // Destructors are noexcept; drain the registry in place so teardown
cannot allocate a snapshot
+ // vector and terminate the query process under memory pressure.
+ try {
+ std::lock_guard registry_lock(_registry_mutex);
+ for (const auto& file : _files) {
+ if (auto live_file = file.lock(); live_file != nullptr) {
+ live_file->_drain(this);
+ }
+ }
+ for (const auto& [_, file] : _file_cache_by_key) {
+ file->_drain(this);
+ }
+ } catch (...) {
+ return;
+ }
+ std::lock_guard lock(_budget_mutex);
+ DORIS_CHECK(_memory_bytes == 0);
+ DORIS_CHECK(_reserved_bytes == 0);
+}
+
+std::shared_ptr<FileScannerV2ReaderLocalFileCache>
+FileScannerV2ReaderLocalCache::get_or_create_file_cache(const std::string&
file_key) {
+ if (_capacity == 0 || file_key.empty()) {
+ return nullptr;
+ }
+ std::lock_guard lock(_registry_mutex);
+ if (auto it = _file_cache_by_key.find(file_key); it !=
_file_cache_by_key.end()) {
+ return it->second;
+ }
+ try {
+ auto file_cache = std::shared_ptr<FileScannerV2ReaderLocalFileCache>(
+ new FileScannerV2ReaderLocalFileCache(shared_from_this()));
+ _file_cache_by_key.emplace(file_key, file_cache);
Review Comment:
The configured capacity only accounts promoted vector bytes, while this
strong registry entry, its key, mutex/map object, and later entry metadata are
retained outside `_memory_bytes` and the cache tracker. `_file_cache_by_key` is
pruned only from payload-reservation pressure, so a many-file scan whose files
stay cold or whose MergeRange reads bypass reader-local promotion can grow one
empty retained object per file without entering that path. Please bound and
track the registry itself (or use an eviction/weak-retention scheme independent
of payload pressure) and add a no-promotions many-file test.
##########
be/benchmark/parquet/AGENTS.md:
##########
@@ -162,7 +162,7 @@ rows, at least 10% NULLs, and materially fragmented
definition-level runs.
`ParquetSelection` contains 25 cases that isolate the selection-vector work
used by Parquet
predicate evaluation. It measures identity initialization, one raw-row filter,
and two successive
-filters. The filter matrix covers 0%, 1%, 10%, 50%, 90%, and 100% selectivity
with clustered and
+filters. The filter matrix covers 0%, 1%, 5%, 10%, 50%, 90%, and 100%
selectivity with clustered and
Review Comment:
The guide now names 5% Selection and Reader scenarios, but their generators
still use `{0,1,10,50,90,100}`; only the Decoder matrix gained 5%.
`ParquetSelection` still has 25 cases, which is also incompatible with the
seven selectivities listed here. Please either keep these two descriptions
aligned with the registered matrices or add the missing scenarios and update
their counts/invariants.
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -1318,25 +2122,40 @@ void CachedRemoteFileReader::_update_stats(const
ReadStatistics& read_stats,
const bool has_source_bytes = source_read_breakdown.local_bytes != 0 ||
source_read_breakdown.remote_bytes != 0 ||
source_read_breakdown.peer_bytes != 0;
- if (has_source_bytes) {
+ const bool has_source_activity = has_source_bytes ||
+ source_read_breakdown.remote_requests !=
0 ||
+ source_read_breakdown.peer_requests != 0;
+ const bool exact_probe_miss_without_io =
+ read_stats.num_exact_cache_probe_miss != 0 && !has_source_bytes;
+ const bool reader_local_only =
+ read_stats.bytes_read != 0 &&
+ read_stats.bytes_read_from_reader_local_cache ==
read_stats.bytes_read;
+ if (has_source_activity) {
if (source_read_breakdown.local_bytes != 0) {
statis->num_local_io_total++;
statis->bytes_read_from_local += source_read_breakdown.local_bytes;
}
- if (source_read_breakdown.peer_bytes != 0 ||
read_stats.from_peer_cache) {
+ if (source_read_breakdown.peer_bytes != 0 ||
source_read_breakdown.peer_requests != 0 ||
+ read_stats.from_peer_cache) {
// Count peer IO whenever peer was used, even if its fetched
blocks were entirely
// outside the copy range (e.g., backward-aligned prefetch block
before
// offset+already_read). In that case peer_bytes==0 but the peer
RPC did happen
// and wrote data into the local file cache.
- statis->num_peer_io_total++;
+ statis->num_peer_io_total +=
std::max<int64_t>(source_read_breakdown.peer_requests, 1);
statis->bytes_read_from_peer += source_read_breakdown.peer_bytes;
statis->peer_io_timer += read_stats.peer_read_timer;
}
- if (source_read_breakdown.remote_bytes != 0) {
- statis->num_remote_io_total++;
+ if (source_read_breakdown.remote_bytes != 0 ||
source_read_breakdown.remote_requests != 0) {
+ statis->num_remote_io_total +=
+ std::max<int64_t>(source_read_breakdown.remote_requests,
1);
Review Comment:
`_read_from_indirect_cache()` can now issue one remote/peer request per
disjoint miss run, and the general profile correctly adds
`source_read_breakdown.*_requests`. This specialized aggregation still
increments only once when any source bytes exist, so a `[miss][hit][miss]`
index/footer read reports two physical RPCs generally but one in its
index/footer subprofile. Please use the request counts here as well and extend
the split-miss test to cover both specialized profiles.
--
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]
