mrhhsg commented on code in PR #68032:
URL: https://github.com/apache/doris/pull/68032#discussion_r4240593992
##########
be/src/exec/spill/spill_file_reader.cpp:
##########
@@ -62,77 +68,177 @@ SpillFileReader::SpillFileReader(RuntimeState* state,
RuntimeProfile* profile,
_read_file_size = get_counter(profile::SPILL_READ_FILE_BYTES);
_read_rows_count = get_counter(profile::SPILL_READ_ROWS);
_read_file_count = get_counter(profile::SPILL_READ_FILE_COUNT);
+ // Optional: older profiles may not register it.
+ _remote_read_requests =
custom_profile->get_counter(profile::SPILL_REMOTE_READ_REQUESTS);
+ if (_is_remote) {
+ _coalesce_bytes = static_cast<size_t>(std::max<int64_t>(
+ 0,
+ std::min(config::spill_s3_read_coalesce_bytes,
state->spill_buffer_size_bytes())));
+ }
+}
+
+void SpillFileReader::_record_read(size_t bytes_read) {
+ COUNTER_UPDATE(_read_file_size, bytes_read);
+
ExecEnv::GetInstance()->spill_file_mgr()->update_spill_read_bytes(bytes_read);
+ if (_is_remote) {
+ // One successful read_at() is one logical GET. Retries inside the
object storage client
+ // and reads that finally failed are not counted (see
RemoteWriteStats).
+ if (_remote_read_requests != nullptr) {
+ COUNTER_UPDATE(_remote_read_requests, 1);
+ }
+ if (_resource_ctx) {
+
_resource_ctx->io_context()->update_spill_read_bytes_from_remote_storage(bytes_read);
+ _resource_ctx->io_context()->update_spill_remote_read_requests(1);
+ }
+
ExecEnv::GetInstance()->spill_file_mgr()->update_spill_remote_read(bytes_read,
1);
+ } else if (_resource_ctx) {
+
_resource_ctx->io_context()->update_spill_read_bytes_from_local_storage(bytes_read);
+ }
}
Status SpillFileReader::open() {
if (_is_open || _part_count == 0) {
return Status::OK();
}
- RETURN_IF_ERROR(_open_part(0));
+ RETURN_IF_ERROR(_open_part(0, true));
_is_open = true;
return Status::OK();
}
-Status SpillFileReader::_open_part(size_t part_index) {
+Status SpillFileReader::_open_part(size_t part_index, bool fetch_small_part) {
_close_current_part();
_current_part_index = part_index;
_part_opened = true;
std::string part_path = _spill_dir + "/" + std::to_string(part_index);
- SCOPED_TIMER(_read_file_timer);
COUNTER_UPDATE(_read_file_count, 1);
- RETURN_IF_ERROR(io::global_local_filesystem()->open_file(part_path,
&_file_reader));
-
- size_t file_size = _file_reader->size();
- DCHECK(file_size >= 16); // max_sub_block_size + block count
-
- Slice result((char*)&_part_block_count, sizeof(size_t));
+ auto fs = _data_dir != nullptr ? _data_dir->fs() :
io::global_local_filesystem();
+ if (fs == nullptr) {
+ return Status::InternalError("spill store {} is not ready",
_data_dir->path());
+ }
+ io::FileReaderOptions opts;
+ opts.cache_type = io::FileCachePolicy::NO_CACHE;
+ // The writer recorded the part size; on object storage this saves a HEAD
request.
+ opts.file_size = _part_sizes[part_index];
+ {
+ SCOPED_TIMER(_read_file_timer);
+ RETURN_IF_ERROR(fs->open_file(part_path, &_file_reader, &opts));
+ }
+ RETURN_IF_ERROR(_read_footer(_file_reader->size(), fetch_small_part));
+ _part_read_block_index = 0;
+ return Status::OK();
+}
- // read block count
- size_t bytes_read = 0;
- RETURN_IF_ERROR(_file_reader->read_at(file_size - sizeof(size_t), result,
&bytes_read));
- DCHECK(bytes_read == 8);
-
- // read max sub block size
- bytes_read = 0;
- result.data = (char*)&_part_max_sub_block_size;
- RETURN_IF_ERROR(_file_reader->read_at(file_size - sizeof(size_t) * 2,
result, &bytes_read));
- DCHECK(bytes_read == 8);
-
- // The buffer is used for two purposes:
- // 1. Reading the block start offsets array (needs _part_block_count *
sizeof(size_t) bytes)
- // 2. Reading a single block's serialized data (needs up to
_part_max_sub_block_size bytes)
- // We must ensure the buffer is large enough for either case, so take the
maximum.
- size_t buff_size = std::max(_part_block_count * sizeof(size_t),
_part_max_sub_block_size);
- if (buff_size > _read_buff.size()) {
- _read_buff.reserve(buff_size);
- }
-
- // Read the block start offsets array from the end of the file.
- // The file layout (from end backwards) is:
+Status SpillFileReader::_read_footer(size_t file_size, bool fetch_small_part) {
+ // The part layout (from the end backwards) is:
// [block count (size_t)]
// [max sub block size (size_t)]
// [block start offsets array (_part_block_count * size_t)]
- // So the offsets array starts at:
- // file_size - (_part_block_count + 2) * sizeof(size_t)
- size_t read_offset = file_size - (_part_block_count + 2) * sizeof(size_t);
- result.data = _read_buff.data();
- result.size = _part_block_count * sizeof(size_t);
+ // [serialized blocks]
+ constexpr size_t kFooterTailBytes = 2 * sizeof(size_t);
+ // Enough for the offsets of a few thousand blocks, so the footer usually
takes one GET.
+ constexpr size_t kRemoteFooterProbeBytes = 64 * 1024;
+ if (file_size < kFooterTailBytes) {
+ return Status::InternalError("spill part {} is too small: {} bytes",
+ _file_reader->path().native(), file_size);
+ }
- RETURN_IF_ERROR(_file_reader->read_at(read_offset, result, &bytes_read));
- DCHECK(bytes_read == _part_block_count * sizeof(size_t));
+ // Without coalescing (local disk) the footer is read exactly. Otherwise a
larger tail is
+ // read in one request, or the whole part when it fits in one coalesced
read.
+ size_t probe_size = kFooterTailBytes;
+ if (_coalesce_bytes > 0) {
+ probe_size = fetch_small_part && file_size <= _coalesce_bytes
Review Comment:
Fixed in 559acdf1aa0: the footer probe is bounded by the effective coalesced
read size (at least the 16-byte tail), and the rest of a larger block offset
array is fetched with one more read.
--
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]
