github-actions[bot] commented on code in PR #67328:
URL: https://github.com/apache/doris/pull/67328#discussion_r3912082611
##########
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:
**[P1] Retain marker ownership before fallible append/close.** After
`create_file()` succeeds, `owned_path` is not recorded until `close()` returns
OK. If a remote append/close partially publishes the marker and the immediate
abort/delete also fails, FE's compensating DELETE tombstones the token but sees
an empty `owned_path` and returns success, leaving the marker with no retry
owner. Record the created path before append/close and retain it until DELETE
succeeds or returns NOT_FOUND; add append/close failure-injection coverage with
a failed first delete and a later retry.
##########
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:
**[P1] Retain queued cleanup owners across BE shutdown.** `cancel(...,
cleanup_async=true)` erases the buffer from `_buffer_map` before enqueueing the
only `release_outfile_cleanup()` owner. The expiry thread uses this path, while
`ExecEnv::destroy()` drains the manager before shutting down the single-thread
lazy-release pool; any task queued behind a slow cleanup is then discarded by
`ThreadPool::shutdown()`, and its captured buffer destructor does not run the
cleanup callback. Track pending async owners or synchronously drain/transfer
them before pool shutdown, and add a blocked-worker/queued-cancellation
shutdown test.
##########
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:
**[P1] Reuse the planner's OUTFILE capability snapshot.**
`LogicalPlanAdapter.getOutFileClause()` creates and analyzes a fresh clause on
every call, so the physical translator can serialize
`enable_atomic_outfile=true` while this newly computed `atomicOutfile` is false
(or vice versa) if mutable `Config.be_exec_version` changes between planning
and execution. The first case leaves BE callbacks pending while FE follows the
legacy branch; the second asks a legacy sink to finalize atomically. Cache one
analyzed clause or propagate the planner's negotiated capability through the
sink/query options, and add a Nereids test that flips the config between
planning and execution.
--
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]
