mrhhsg commented on code in PR #68032:
URL: https://github.com/apache/doris/pull/68032#discussion_r4256992219
##########
be/src/exec/spill/spill_file_manager.cpp:
##########
@@ -431,10 +622,10 @@ void SpillFileManager::gc(int32_t max_work_time_ms) {
LOG(INFO) << msg;
}
}};
- _retry_pending_query_spill_directories();
- for (const auto& [path, store_dir] : _spill_store_map) {
+ _retry_pending_spill_directories();
Review Comment:
Fixed in b4be610c4f7: retrying queued failed deletions stops once the GC
time slice (`spill_gc_work_time_ms`) is used up, after at least one directory;
the directories not tried are queued again ahead of the ones that just failed,
so a slow failing prefix cannot starve the others. `stop()` and the destructor
use the same time slice. Covered by
`PendingRemoteDeletesAreBoundedByGcWorkTime` with a slow mock object store. The
known-limitation entry is removed from the PR description.
##########
be/src/io/fs/s3_file_writer.h:
##########
@@ -21,6 +21,7 @@
#include <chrono>
#include <cstddef>
+#include <functional>
Review Comment:
Update: since d5afdb0c50b the spill writer no longer slices appends itself.
`S3FileWriter` already splits appended data into upload buffers; the spill
writer checks the upload queue once per serialized block and hands the whole
block over, so later appends cannot run ahead of object storage.
##########
be/src/cloud/cloud_meta_mgr.cpp:
##########
@@ -1908,6 +1909,9 @@ Status
CloudMetaMgr::get_storage_vault_info(StorageVaultInfos* vault_infos, bool
}
*is_vault_mode = resp.enable_storage_vault();
+ if (default_vault_id != nullptr) {
+ *default_vault_id = resp.default_storage_vault_id();
Review Comment:
Update: changed in 7a97727319c to the S3 vault with the largest numeric ID,
as @gavinchou suggested. Spill follows the newest S3 vault; changing the
default vault still does not move it, and files already spilled stay on the
vault they were written to. When a fixed destination is wanted,
`spill_s3_storage_vault_id` pins the vault explicitly.
##########
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:
Not changed in this PR. `check_after_upload()` and its `DCHECK`s are master
code in `S3FileWriter`, which this PR keeps identical to master as requested in
review; every S3 upload (loads, compaction, segment writes) goes through the
same check. In release builds the DCHECKs are compiled out and the `IOError` is
returned: the spill part close fails, the multipart upload is aborted and the
query fails with that error. Changing the assertion policy of `S3FileWriter`
belongs to 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:
Acknowledged, not changed. The window is one disk-report interval
(`report_disk_state_interval_seconds`, 30 s by default): the first report of
the new process overwrites the value. Disk capacity reported in the same
request has the same ordering. The value is documented as best-effort, so
adding a start epoch to `TReportRequest` is not worth it here.
--
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]
