mrhhsg commented on code in PR #68032:
URL: https://github.com/apache/doris/pull/68032#discussion_r4279089206
##########
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:
Resolving: the fix from 559acdf1aa0 is in the current head. `_read_footer`
bounds the probe by `std::max(_coalesce_bytes, kFooterTailBytes)` (and by the
part size), and fetches the rest of a larger offset array with one more read.
##########
be/src/exec/spill/spill_file_manager.cpp:
##########
@@ -153,14 +199,16 @@ void SpillFileManager::stop() {
// The GC thread may observe the stop latch before processing a recently
queued failed deletion.
// Retry the pending directories after the thread exits; later failures
get one final retry in
// the destructor.
- _retry_pending_query_spill_directories();
+ _retry_pending_spill_directories();
}
Status SpillFileManager::init() {
LOG(INFO) << "init spill stream manager";
RETURN_IF_ERROR(_init_spill_store_map());
+ _remote_upload_budget =
+
std::make_shared<SpillRemoteUploadBudget>(config::spill_s3_max_inflight_upload_bytes);
- for (const auto& [path, store] : _spill_store_map) {
+ for (auto* store : _local_stores) {
Review Comment:
Resolving as a documented limitation (see "Known limitations" in the PR
description, which links this thread). Rediscovering residual objects would
need a listing of this BE's prefix at startup; the startup sweep and the
recycler sweep were removed on reviewer request, and the lifetime of objects
left by an abrupt exit is the bucket lifecycle rule's. The reported value is
documented as best-effort and not a billing source.
##########
be/src/exec/spill/spill_file_writer.cpp:
##########
@@ -90,42 +146,208 @@ Status SpillFileWriter::_close_current_part(const
std::shared_ptr<SpillFile>& sp
_part_meta.append((const char*)&_part_max_sub_block_size,
sizeof(_part_max_sub_block_size));
_part_meta.append((const char*)&_part_written_blocks,
sizeof(_part_written_blocks));
- {
+ int64_t meta_size = _part_meta.size();
+ // The footer must always be written so that the part can be closed;
account it
+ // without checking the capacity limit.
+ Status status = _data_dir->try_reserve(meta_size, /*force=*/true);
+ if (status.ok()) {
SCOPED_TIMER(_write_file_timer);
- RETURN_IF_ERROR(_file_writer->append(_part_meta));
+ status = _append_to_part(Slice(_part_meta));
+ if (!status.ok()) {
+ _data_dir->release(meta_size);
+ }
}
- int64_t meta_size = _part_meta.size();
- _part_written_bytes += meta_size;
- COUNTER_UPDATE(_write_file_total_size, meta_size);
+ if (status.ok()) {
+ _part_written_bytes += meta_size;
+ COUNTER_UPDATE(_write_file_total_size, meta_size);
+ if (_resource_ctx) {
+ if (_data_dir->is_remote()) {
+
_resource_ctx->io_context()->update_spill_write_bytes_to_remote_storage(meta_size);
+ } else {
+
_resource_ctx->io_context()->update_spill_write_bytes_to_local_storage(meta_size);
+ }
+ }
+ if (_write_file_current_size) {
+ COUNTER_UPDATE(_write_file_current_size, meta_size);
+ }
+
ExecEnv::GetInstance()->spill_file_mgr()->update_spill_write_bytes(meta_size);
+ // Incrementally update SpillFile's accounting so gc() can always
+ // decrement the correct amount, even if close() is never called.
+ if (spill_file) {
+ spill_file->update_written_bytes(meta_size);
+ }
+ }
+
+ // Close synchronously. Spilling already runs on a thread that may block
on every append
+ // (the pipeline execution thread running the revoke task, or the async
result-writer
+ // thread for Iceberg), and with the default part size the wait at a part
boundary (the
+ // tail of the uploads plus one CompleteMultipartUpload) is negligible
against the part
+ // itself, so overlapping it with the next part is not worth an async
close pipeline.
+ std::unique_ptr<io::FileWriter> writer = std::move(_file_writer);
+ MultipartUploadId upload = _multipart_upload_id(writer.get());
+ if (status.ok()) {
+ MonotonicStopWatch close_watch;
+ close_watch.start();
+ status = writer->close();
Review Comment:
Resolving, unchanged. `be/src/io/fs/s3_file_writer.{h,cpp}` are identical to
master in this PR (`git diff origin/master -- be/src/io/fs/s3_file_writer.cpp`
is empty), so `check_after_upload()` and its `DCHECK`s run for every S3 upload
on master already (loads, compaction, segment writes); spill does not add a new
assertion. In release builds the `IOError` is returned: the part close fails,
the multipart upload is aborted and the query fails with that error. Changing
the assertion policy of `S3FileWriter` for all its callers is a separate change.
##########
fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java:
##########
@@ -185,6 +185,10 @@ public TMasterResult handleReport(TReportRequest request)
throws TException {
reportType = ReportType.DISK;
}
+ if (request.isSetRemoteSpillBytes()) {
+ backend.setRemoteSpillBytes(request.getRemoteSpillBytes());
Review Comment:
Resolving, unchanged, as explained above: the stale value lasts at most one
disk report interval (`report_disk_state_interval_seconds`, 30 s by default)
until the first report of the new process overwrites it, the disk capacity in
the same request has the same ordering, and the value is documented as
best-effort.
--
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]
