mrhhsg commented on code in PR #68032:
URL: https://github.com/apache/doris/pull/68032#discussion_r4296005934
##########
be/src/io/fs/s3_file_writer.h:
##########
@@ -21,6 +21,7 @@
#include <chrono>
#include <cstddef>
+#include <functional>
Review Comment:
Follow-up in e42e09d2298: `S3FileWriter` gets one addition, an opt-in
`abort()`. A failed or discarded spill part leaves the parts of its multipart
upload behind; deleting the spill prefix does not remove them (they are not
listed as objects), so they stay billed until the bucket lifecycle rule aborts
the upload. Before, the spill writer aborted it itself, rebuilding the bucket,
key and upload id from the writer's getters and calling the object storage
client directly. `abort()` keeps that inside `S3FileWriter`: it skips the
buffers not uploaded yet, waits for the running uploads and aborts the
multipart upload if one was started. It is only allowed on an open writer or
one whose `close()` failed. Destroying a writer still never aborts and no
existing caller changes, since an upload may be completed elsewhere (the S3
committer completes them on FE); only the spill writer calls it.
##########
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:
Correction to the reply above: since e42e09d2298, `s3_file_writer.{h,cpp}`
differ from master by one addition, an opt-in `abort()` used by spill to give
up a failed or discarded part (see the reply on the "no need to modify s3 file
writer" thread). It does not change this thread's conclusion:
`check_after_upload()` is called only from `_complete()` and `_put_object()`
inside `_close_impl()`, and `abort()` calls neither (it skips the queued
buffers, waits for the running uploads and sends AbortMultipartUpload;
`close()` is refused afterwards), so spill reaches `check_after_upload()`
exactly as every other S3 upload on master does.
--
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]
