Gabriel39 commented on code in PR #67328:
URL: https://github.com/apache/doris/pull/67328#discussion_r3912518249
##########
be/src/runtime/result_buffer_mgr.cpp:
##########
@@ -58,6 +61,21 @@ void ResultBufferMgr::stop() {
if (_clean_thread) {
_clean_thread->join();
}
+ std::vector<TUniqueId> remaining_ids;
+ {
+ std::unique_lock<std::shared_mutex> wlock(_buffer_map_lock);
+ // Closing the registration gate under the map lock keeps the shutdown
snapshot complete.
+ _stopping = true;
+ remaining_ids.reserve(_buffer_map.size());
+ for (const auto& item : _buffer_map) {
+ remaining_ids.emplace_back(item.first);
+ }
+ }
Review Comment:
Fixed in f2362cfbc9. Pending asynchronous cleanup ownership is now
registered before the result buffer is removed from the manager.
ResultBufferMgr::stop() snapshots and synchronously drains those owners before
the lazy-release pool is shut down, while the pending-cleanup guard makes
queued, shutdown, and submit-fallback execution single-shot. Added
StopDrainsCleanupQueuedBehindBlockedWorker to cover a cancellation queued
behind a blocked worker.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -180,6 +181,7 @@
// first: Parse receive byte array to statement struct.
Review Comment:
Fixed in f2362cfbc9 and refined in df62412ef0. OutFileClause now captures
one execution-version snapshot, LogicalFileSink preserves it through rewrites,
the implementation rule transfers it to PhysicalFileSink, and both
LogicalPlanAdapter and PhysicalPlanTranslator construct their clauses from that
same value. The regression test completes physical planning, changes
Config.be_exec_version, and then verifies that both translation and the adapter
still use the planned atomic capability.
##########
be/src/service/internal_service.cpp:
##########
@@ -737,25 +773,42 @@ void
PInternalService::outfile_write_success(google::protobuf::RpcController* co
std::stringstream ss;
ss << file_options.file_path << file_options.success_file_name;
std::string file_name = ss.str();
- if (result_file_sink.storage_backend_type ==
TStorageBackendType::LOCAL) {
- // For local file writer, the file_path is a local dir.
- // Here we do a simple security verification by checking whether
the file exists.
- // Because the file path is currently arbitrarily specified by the
user,
- // Doris is not responsible for ensuring the correctness of the
path.
- // This is just to prevent overwriting the existing file.
- bool exists = true;
- st = io::global_local_filesystem()->exists(file_name, &exists);
- if (!st.ok()) {
- LOG(WARNING) << "outfile write success filefailed, errmsg = "
<< st;
- st.to_protobuf(result->mutable_status());
- return;
- }
- if (exists) {
- st = Status::InternalError("File already exists: {}",
file_name);
+ const bool atomic_outfile_marker = file_options.enable_atomic_outfile;
+ const std::string marker_token =
+ request->marker_token().empty() ? file_name :
request->marker_token();
+ const auto now = std::chrono::steady_clock::now();
+ std::shared_ptr<std::mutex> operation_lock;
+ if (atomic_outfile_marker) {
+ std::lock_guard marker_guard(outfile_marker_lock);
+ cleanup_expired_outfile_marker_states(now);
+ operation_lock =
outfile_marker_operation_locks[marker_token].lock();
+ if (operation_lock == nullptr) {
+ operation_lock = std::make_shared<std::mutex>();
+ outfile_marker_operation_locks[marker_token] = operation_lock;
}
- if (!st.ok()) {
- LOG(WARNING) << "outfile write success file failed, errmsg = "
<< st;
- st.to_protobuf(result->mutable_status());
+ }
+
+ // Serialize only requests for the same query marker. Remote
file-system calls can be slow,
+ // so unrelated OUTFILE queries must not block behind them.
+ std::unique_lock<std::mutex> operation_guard;
+ if (atomic_outfile_marker) {
+ operation_guard = std::unique_lock(*operation_lock);
+ }
+ std::string owned_marker_path;
Review Comment:
Fixed in f2362cfbc9. Atomic marker ownership is recorded immediately after
create_file() succeeds, before append or close can partially publish data.
Immediate and compensating deletes clear ownership only after success or
NOT_FOUND; a failed delete tombstones and retains the owner for a later retry.
Added state-machine coverage for an append failure, failed first delete, and
successful compensating retry.
--
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]
