github-actions[bot] commented on code in PR #69716:
URL: https://github.com/apache/doris/pull/69716#discussion_r4396835123
##########
be/src/format_v2/delimited_text/delimited_text_reader.cpp:
##########
@@ -561,8 +562,14 @@ Status DelimitedTextReader::_open_file() {
} else {
auto reader_options =
FileFactory::get_reader_options(_runtime_state->query_options(),
*_file_description);
- auto file_reader = DORIS_TRY(FileFactory::create_file_reader(
- *_system_properties, *_file_description, reader_options,
_profile));
+ // Include the prefix used to align non-first splits. Unknown-size
first splits read to
+ // EOF; PrefetchBufferedReader bounds this range by the opened file's
physical size.
+ const auto prefetch_end = _size < 0 ?
std::numeric_limits<size_t>::max()
+ : cast_set<size_t>(_start_offset +
_size);
+ auto file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
+ _profile, *_system_properties, *_file_description,
reader_options,
Review Comment:
[P2] Publish the prefetch counters before dropping this reader
This call now registers `PrefetchBufferedReader` profile counters, but
`DelimitedTextReader::close()` only closes/resets the line reader and drops the
file reader. Neither those close/destructor paths nor `ProfileCollector`
invokes `collect_profile_before_close()`. The callback that publishes
`ReadTime`, `PreRequestIO` and `PreRequestBytes` is therefore never called, so
completed S3 CSV/Text scans display zero prefetch I/O. Drain the background
reads and collect the reader profile before releasing it, and assert these
counters in the new S3 test; its current assertions check only the outer
tracing statistics.
##########
be/src/format_v2/delimited_text/delimited_text_reader.cpp:
##########
@@ -561,8 +562,14 @@ Status DelimitedTextReader::_open_file() {
} else {
auto reader_options =
FileFactory::get_reader_options(_runtime_state->query_options(),
*_file_description);
- auto file_reader = DORIS_TRY(FileFactory::create_file_reader(
- *_system_properties, *_file_description, reader_options,
_profile));
+ // Include the prefix used to align non-first splits. Unknown-size
first splits read to
+ // EOF; PrefetchBufferedReader bounds this range by the opened file's
physical size.
+ const auto prefetch_end = _size < 0 ?
std::numeric_limits<size_t>::max()
+ : cast_set<size_t>(_start_offset +
_size);
+ auto file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
+ _profile, *_system_properties, *_file_description,
reader_options,
+ io::DelegateReader::AccessMode::SEQUENTIAL, _io_ctx,
+ io::PrefetchRange(_start_offset, prefetch_end)));
Review Comment:
[P1] Keep scanner state alive until timed-out reads finish
Sharing the `IOContext` does not own its raw `file_cache_stats` pointer.
`PrefetchBuffer::close()` returns without joining when its wait times out, but
`PrefetchBufferedReader::_close_internal()` still returns the underlying
reader's successful close; `DelimitedTextReader::close()` can then discard the
wrappers and the scanner can destroy its statistics. A pending
`CachedRemoteFileReader` read subsequently runs its deferred `_update_stats()`
through that freed pointer. For example, a cached 2 MiB object with
`buffered_reader_read_timeout_ms=10` and a GET delayed 100 ms can outlive both
the foreground-read timeout and the close timeout; only one buffer is in range.
Ensure every pending task is drained before releasing scanner-owned pointees,
or retain ownership and propagate a retryable close failure. This is separate
from concurrent counter updates and occurs with one pending task.
##########
be/src/format_v2/delimited_text/delimited_text_reader.cpp:
##########
@@ -561,8 +562,14 @@ Status DelimitedTextReader::_open_file() {
} else {
auto reader_options =
FileFactory::get_reader_options(_runtime_state->query_options(),
*_file_description);
- auto file_reader = DORIS_TRY(FileFactory::create_file_reader(
- *_system_properties, *_file_description, reader_options,
_profile));
+ // Include the prefix used to align non-first splits. Unknown-size
first splits read to
+ // EOF; PrefetchBufferedReader bounds this range by the opened file's
physical size.
+ const auto prefetch_end = _size < 0 ?
std::numeric_limits<size_t>::max()
+ : cast_set<size_t>(_start_offset +
_size);
+ auto file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
+ _profile, *_system_properties, *_file_description,
reader_options,
+ io::DelegateReader::AccessMode::SEQUENTIAL, _io_ctx,
Review Comment:
[P1] Isolate the cache statistics used by prefetch workers
For cached S3 CSV/Text, this call now queues four 4 MiB reads by default,
all sharing `_io_ctx->file_cache_stats`.
`CachedRemoteFileReader::_update_stats()` performs unsynchronized plain `+=`
writes to `bytes_read_from_remote/local`, while
`FileScannerV2::_collect_realtime_counter_deltas()` reads those fields and
`DORIS_CHECK`s that they never decrease. A worker can load an old total,
another worker can publish a larger total that the scanner samples, and the
first worker can then store its stale smaller sum; the next scanner sample
aborts BE. Previously this v2 path read synchronously. Use worker-local
statistics merged on the scanner thread, or synchronize both updates and
snapshots. The added test disables file cache and uses one worker, so it misses
this path.
##########
be/src/format_v2/delimited_text/delimited_text_reader.cpp:
##########
@@ -561,8 +562,14 @@ Status DelimitedTextReader::_open_file() {
} else {
auto reader_options =
FileFactory::get_reader_options(_runtime_state->query_options(),
*_file_description);
- auto file_reader = DORIS_TRY(FileFactory::create_file_reader(
- *_system_properties, *_file_description, reader_options,
_profile));
+ // Include the prefix used to align non-first splits. Unknown-size
first splits read to
+ // EOF; PrefetchBufferedReader bounds this range by the opened file's
physical size.
+ const auto prefetch_end = _size < 0 ?
std::numeric_limits<size_t>::max()
+ : cast_set<size_t>(_start_offset +
_size);
+ auto file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
Review Comment:
[P1] Handle accepted buffer sizes before selecting prefetch
This also moves v2 CSV/Text onto a buffer-ring assumption that the previous
raw-reader path did not require. `remote_storage_read_buffer_mb` accepts an
integer such as `6` without validation. `PrefetchBufferedReader` then creates
`floor(6/4) = 1` buffers but uses 6 MiB as its ring modulus. On an 8 MiB S3
CSV/Text scan, offset 4 MiB computes `get_buffer_pos() = (4 % 6) / 4 = 1` and
indexes the nonexistent `_pre_buffers[1]`; a non-first split can hit it during
initialization. Normalize the ring geometry or validate the setting
consistently before enabling this path, and cover a non-multiple-of-four
setting with a scan crossing 4 MiB.
--
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]
