github-actions[bot] commented on code in PR #67328:
URL: https://github.com/apache/doris/pull/67328#discussion_r3893938504
##########
be/src/runtime/result_buffer_mgr.cpp:
##########
@@ -58,6 +61,19 @@ void ResultBufferMgr::stop() {
if (_clean_thread) {
_clean_thread->join();
}
+ std::vector<TUniqueId> remaining_ids;
Review Comment:
[P1] `stop()` copies `_buffer_map` IDs under the lock, releases it, and only
then cancels them. `create_sender()` has no stopping-state check, and
`ExecEnv::destroy()` stops this manager before FragmentMgr/workload execution,
so an in-flight fragment can insert a new buffer after the snapshot. That
buffer is never canceled and its OUTFILE callbacks are lost when the manager is
destroyed. Quiesce/reject new senders before the snapshot and drain until no
buffers remain (or retain a shutdown cleanup owner).
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -1592,6 +1609,33 @@ public void executeAndSendResult(boolean isOutfileQuery,
boolean isSendFields,
break;
}
}
+ if (isOutfileQuery) {
+ if (atomicOutfile) {
+
coordBase.finishOutfile(InternalService.POutfileWriteOperation.OUTFILE_PREPARE);
+ }
+ if (atomicOutfile &&
!Strings.isNullOrEmpty(outFileClause.getSuccessFileName())) {
+ outfileMarkerBackend = selectOutfileSuccessBackend();
+ // The create response may be lost after the marker is
durable, so rollback must
+ // conservatively delete it whenever this call does not
complete successfully.
+ outfileMarkerMayExist = true;
+ outfileWriteSuccess(outFileClause, outfileMarkerBackend,
Review Comment:
[P1] The atomic sequence creates and closes the success marker immediately
after PREPARE, before `finishOutfile(OUTFILE_COMMIT)` acknowledges every
receiver. The marker is thus externally visible while data is only prepared; if
one COMMIT RPC then times out, rollback deletes files after a reader may
already have treated the marker as durable completion. Publish the marker only
after distributed commit succeeds (or keep it hidden until that decision) and
test a partial-commit failure with a concurrent marker reader.
##########
be/src/runtime/result_block_buffer.cpp:
##########
@@ -103,19 +104,136 @@ Status ResultBlockBuffer<ResultCtxType>::close(const
TUniqueId& id, Status exec_
}
template <typename ResultCtxType>
-void ResultBlockBuffer<ResultCtxType>::cancel(const Status& reason) {
- std::unique_lock<std::mutex> l(_lock);
+void ResultBlockBuffer<ResultCtxType>::cancel(const Status& reason, bool
release_outfile) {
SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);
- if (_status.ok()) {
- _status = reason;
+ {
+ std::unique_lock<std::mutex> l(_lock);
+ if (_status.ok()) {
+ _status = reason;
+ }
+ _arrow_data_arrival.notify_all();
+ for (auto& ctx : _waiting_rpc) {
+ ctx->on_failure(reason);
+ }
+ _waiting_rpc.clear();
+ _update_dependency();
+ _result_batch_queue.clear();
}
- _arrow_data_arrival.notify_all();
- for (auto& ctx : _waiting_rpc) {
- ctx->on_failure(reason);
+ if (release_outfile) {
+ // Release query result memory before rollback performs potentially
slow remote I/O.
+ release_outfile_cleanup();
}
- _waiting_rpc.clear();
- _update_dependency();
- _result_batch_queue.clear();
+}
+
+template <typename ResultCtxType>
+Status ResultBlockBuffer<ResultCtxType>::add_outfile_cleanup(OutfileCleanup
cleanup) {
+ SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);
+ bool run_cleanup = false;
+ bool discard_cleanup = false;
+ {
+ std::lock_guard<std::mutex> l(_lock);
+ if (_outfile_state == OutfileState::ABORTED) {
+ run_cleanup = true;
+ } else if (_outfile_state == OutfileState::COMMITTED) {
+ discard_cleanup = true;
+ } else {
+ _outfile_cleanups.emplace_back(std::move(cleanup));
+ }
+ }
+ if (discard_cleanup) {
+ cleanup = nullptr;
+ return Status::OK();
+ }
+ if (run_cleanup) {
+ Status status = cleanup();
Review Comment:
[P1] When cancellation wins the race, `ResultBufferMgr::cancel()` erases the
buffer and runs its only `release_outfile_cleanup()` pass before the
asynchronous writer necessarily calls `VFileResultWriter::close()`. A later
`add_outfile_cleanup()` executes inline in the `ABORTED` branch; if remote
delete/abort transiently fails, the callback is merely reinserted into a buffer
with no retry owner and is lost on destruction. Hand late registrations to a
durable/bounded retry owner or keep cancellation draining callbacks until
writers finish.
##########
be/src/io/fs/s3_file_writer.cpp:
##########
@@ -514,6 +541,7 @@ Status S3FileWriter::_complete() {
RETURN_IF_ERROR(check_after_upload(client.get(), resp,
_obj_storage_path_opts, _bytes_appended,
"complete_multipart"));
+ _multipart_upload_completed = true;
Review Comment:
[P1] `_multipart_upload_completed` is set only after `check_after_upload()`
returns. If `CompleteMultipartUpload` succeeds but the subsequent HEAD/size
check transiently fails, this returns an error with the flag still false even
though the object is already published. Cleanup then calls
`AbortMultipartUpload` on a completed upload (typically `NoSuchUpload`) and
retains the writer/ID forever. Mark completion immediately after the
CompleteMultipartUpload response and treat an already-completed/NoSuchUpload
abort as converged; add a post-complete HEAD failure test.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java:
##########
@@ -1459,6 +1461,65 @@ protected void cancelInternal(Status cancelReason) {
cancelLatch();
}
+ protected List<ResultReceiver> getOutfileResultReceivers() {
+ return receivers;
+ }
+
+ @Override
+ public void finishOutfile(InternalService.POutfileWriteOperation
operation) throws Exception {
+ Map<TNetworkAddress,
InternalService.POutfileWriteFinishedRequest.Builder> requests = new
HashMap<>();
+ for (ResultReceiver receiver : getOutfileResultReceivers()) {
+ InternalService.POutfileWriteFinishedRequest.Builder request =
requests.computeIfAbsent(
+ receiver.getAddress(), address ->
InternalService.POutfileWriteFinishedRequest.newBuilder()
+ // Old BEs ignore operation, so preserve their
success field during upgrades.
+ .setSuccess(operation ==
InternalService.POutfileWriteOperation.OUTFILE_COMMIT));
+ if (Config.be_exec_version >=
OutFileClause.SUPPORT_ATOMIC_OUTFILE_VERSION) {
+ request.setOperation(operation);
+ }
+ request.addBufferIds(receiver.getRealFinstId());
+ }
+ if (requests.isEmpty()) {
+ throw new UserException("No OUTFILE result receiver participates
in finalization");
+ }
+ long timeoutMs = operation ==
InternalService.POutfileWriteOperation.OUTFILE_ABORT
+ ? Math.max(1, Math.min(Config.remote_fragment_exec_timeout_ms,
OUTFILE_CLEANUP_TIMEOUT_MS))
+ : Math.max(1, Math.min(Config.remote_fragment_exec_timeout_ms,
+ timeoutDeadline - System.currentTimeMillis()));
Review Comment:
[P1] `finishOutfile()` derives the PREPARE/COMMIT timeout from
`Coordinator.timeoutDeadline`, but that field is initialized only inside
`Coordinator.execInternal()`. `NereidsCoordinator.exec()` overrides `exec()`
and uses `coordinatorContext.timeoutDeadline` instead, so the inherited field
stays at its default 0. Nereids finalization therefore reaches `Math.max(1,
timeoutDeadline - now)` as 1 ms and ordinary BE RPC latency causes spurious
failure. Use the coordinator-context deadline (or initialize the base field)
for Nereids and add a normal-latency finalization test.
##########
be/src/service/internal_service.cpp:
##########
@@ -135,6 +138,44 @@ class RpcController;
} // namespace google
namespace doris {
+
+namespace {
+
+std::mutex outfile_marker_lock;
+struct OutfileMarkerState {
+ std::chrono::steady_clock::time_point updated_at;
+ std::string owned_path;
+ bool tombstoned = false;
+};
+std::unordered_map<std::string, std::weak_ptr<std::mutex>>
outfile_marker_operation_locks;
+std::unordered_map<std::string, OutfileMarkerState> outfile_marker_states;
+constexpr auto OUTFILE_MARKER_TOMBSTONE_TTL = std::chrono::hours(1);
+constexpr auto OUTFILE_MARKER_CLEANUP_INTERVAL = std::chrono::minutes(1);
+auto outfile_marker_last_cleanup = std::chrono::steady_clock::now();
+
+void
cleanup_expired_outfile_marker_states(std::chrono::steady_clock::time_point
now) {
+ if (now - outfile_marker_last_cleanup < OUTFILE_MARKER_CLEANUP_INTERVAL) {
+ return;
+ }
+ outfile_marker_last_cleanup = now;
+ for (auto it = outfile_marker_operation_locks.begin();
+ it != outfile_marker_operation_locks.end();) {
+ if (it->second.expired()) {
+ it = outfile_marker_operation_locks.erase(it);
+ } else {
+ ++it;
+ }
+ }
+ for (auto it = outfile_marker_states.begin(); it !=
outfile_marker_states.end();) {
+ if (now - it->second.updated_at >= OUTFILE_MARKER_TOMBSTONE_TTL) {
Review Comment:
[P1] `cleanup_expired_outfile_marker_states()` erases every marker state
after one hour, including tombstones whose `delete_file()` failed and still
retain `owned_path`. Once that entry is evicted, the rollback fence and
ownership record are gone: a delayed CREATE can be accepted if the marker is
absent, while a marker left behind by the failed delete has no state for a
later DELETE to recover and is silently leaked. Do not expire tombstones until
deletion is durably confirmed (and define recovery across BE restart), or late
requests can resurrect or orphan rolled-back markers.
##########
be/src/exec/sink/writer/vfile_result_writer.cpp:
##########
@@ -123,13 +170,17 @@ Status VFileResultWriter::_create_next_file_writer() {
Status VFileResultWriter::_create_file_writer(const std::string& file_name) {
auto file_type =
DORIS_TRY(FileFactory::convert_storage_type(_storage_type));
- _file_writer_impl = DORIS_TRY(FileFactory::create_file_writer(
- file_type, _state->exec_env(), _file_opts->broker_addresses,
- _file_opts->broker_properties, file_name,
- {
- .write_file_cache = false,
- .sync_file_data = false,
- }));
+ io::FSPropertiesRef properties(file_type);
+ properties.broker_addresses = &_file_opts->broker_addresses;
+ properties.properties = &_file_opts->broker_properties;
+ io::FileDescription file_description;
+ file_description.path = file_name;
+ _file_system = DORIS_TRY(FileFactory::create_fs(properties,
file_description));
+ // Create/open can publish a path before returning an error, so claim
deterministic ownership
+ // first. A separate filesystem preserves Broker's existing per-path
endpoint selection.
+ _created_files.emplace_back(_file_system, file_name);
+ const io::FileWriterOptions options {.write_file_cache = false,
.sync_file_data = false};
Review Comment:
[P2] The old `FileFactory::create_file_writer(FILE_LOCAL, ..., options)`
ignored `options`, so local writers defaulted to `sync_data=true` and honored
`sync_file_on_close` with `fdatasync`/directory sync. This path now passes
`.sync_file_data = false` to `LocalFileSystem`, which propagates it to
`LocalFileWriter`; successful local OUTFILE closes no longer get the prior
crash-durability sync. Preserve the old local setting while retaining remote
no-sync behavior, and add a local durability regression test.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -1592,6 +1609,33 @@ public void executeAndSendResult(boolean isOutfileQuery,
boolean isSendFields,
break;
}
}
+ if (isOutfileQuery) {
+ if (atomicOutfile) {
+
coordBase.finishOutfile(InternalService.POutfileWriteOperation.OUTFILE_PREPARE);
+ }
+ if (atomicOutfile &&
!Strings.isNullOrEmpty(outFileClause.getSuccessFileName())) {
+ outfileMarkerBackend = selectOutfileSuccessBackend();
+ // The create response may be lost after the marker is
durable, so rollback must
+ // conservatively delete it whenever this call does not
complete successfully.
+ outfileMarkerMayExist = true;
+ outfileWriteSuccess(outFileClause, outfileMarkerBackend,
+
InternalService.POutfileSuccessOperation.OUTFILE_MARKER_CREATE);
+ }
+ // This also carries the legacy success acknowledgement needed
by old BEs during
+ // rolling upgrades; atomic-capable BEs interpret it as the
global COMMIT phase.
+
coordBase.finishOutfile(InternalService.POutfileWriteOperation.OUTFILE_COMMIT);
Review Comment:
[P1] The call to `finishOutfile(OUTFILE_COMMIT)` is unconditional even when
`atomicOutfile` is false (configured execution version below 14). In that
compatibility mode a pre-14 BE has no `outfile_write_finished` RPC, so this
fails after the legacy writer/marker path has run; `abortOutfile()` then
returns immediately because atomic mode is disabled, potentially leaving the
success marker behind and turning old-version OUTFILE queries into errors. Keep
PREPARE/COMMIT/marker orchestration behind the atomic capability and preserve
the pre-change path for old BEs, with a version-13 compatibility test.
--
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]
