github-actions[bot] commented on code in PR #67328:
URL: https://github.com/apache/doris/pull/67328#discussion_r3892106386
##########
fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java:
##########
@@ -1459,6 +1459,38 @@ protected void cancelInternal(Status cancelReason) {
cancelLatch();
}
+ @Override
+ public void finishOutfile(boolean success) throws Exception {
+ Map<TNetworkAddress,
InternalService.POutfileWriteFinishedRequest.Builder> requests = new
HashMap<>();
+ for (ResultReceiver receiver : receivers) {
Review Comment:
**[P1] Finalize the receiver set that Nereids actually uses.**
`NereidsCoordinator` never populates this legacy `receivers` list;
`QueryProcessor.build()` owns a separate list and inherited `finishOutfile()`
therefore sends zero RPCs. An ordinary Nereids OUTFILE can return success (and
write the marker) while every BE buffer stays `PENDING`, so scheduled
result-buffer cleanup later deletes the data files. Source participants from a
coordinator-wide abstraction or override this path, and reject an empty
participant set for OUTFILE.
##########
be/src/exec/sink/writer/vfile_result_writer.cpp:
##########
@@ -487,7 +492,43 @@ Status VFileResultWriter::close(Status exec_status) {
}
st = _close_file_writer(true);
}
+ if (!st.ok()) {
+ _cleanup_created_files();
+ } else {
+ _register_created_files_cleanup();
+ }
return st;
}
+void VFileResultWriter::_cleanup_created_files() {
+ _vfile_writer.reset();
+ if (_file_writer_impl != nullptr) {
+ if (_created_file_paths.empty()) {
+ _created_file_paths.emplace_back(_file_writer_impl->path());
+ }
+ WARN_IF_ERROR(_file_writer_impl->abort(), "failed to abort outfile
writer");
+ _file_writer_impl.reset();
+ }
+ if (_file_system == nullptr && _storage_type ==
TStorageBackendType::LOCAL) {
+ _file_system = io::global_local_filesystem();
+ }
+ if (_file_system != nullptr && !_created_file_paths.empty()) {
+ WARN_IF_ERROR(_file_system->batch_delete(_created_file_paths),
+ "failed to remove files from aborted outfile");
+ }
+ _created_file_paths.clear();
+}
+
+void VFileResultWriter::_register_created_files_cleanup() {
+ if (_sinker == nullptr || _file_system == nullptr ||
_created_file_paths.empty()) {
+ return;
+ }
+ auto file_system = _file_system;
+ auto paths = std::move(_created_file_paths);
+ _sinker->add_outfile_cleanup([file_system = std::move(file_system), paths
= std::move(paths)] {
Review Comment:
**[P1] Gate this cleanup protocol on FE capability.** [Doris rolling-upgrade
guidance](https://doris.apache.org/docs/4.x/admin-manual/cluster-management/upgrade/)
replaces BEs before the FE, so an older FE can successfully run OUTFILE on
this new BE but can never call the newly added `outfile_write_finished` RPC.
This callback remains `PENDING`, and scheduled buffer cancellation deletes
already-acknowledged output, often leaving the old FE's success marker.
Negotiate support and retain legacy successful behavior until the FE opts into
this handshake.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -1592,6 +1590,13 @@ public void executeAndSendResult(boolean isOutfileQuery,
boolean isSendFields,
break;
}
}
+ if (isOutfileQuery) {
Review Comment:
**[P1] Add the Arrow Flight finalization path before arming rollback.**
`ARROW_FLIGHT_SQL` returns earlier in this method, and
`finalizeArrowFlightQuery()` only closes the coordinator; it never reaches this
block. Both coordinator implementations also represent Flight OUTFILE with
endpoints rather than the `receivers` consumed by `finishOutfile()`. Successful
Flight output therefore stays `PENDING` and is deleted by BE timeout after
DoGet. Finalize from the DoGet/deferred lifecycle using endpoint buffer IDs, or
reject Flight OUTFILE until it can participate.
##########
be/src/runtime/result_block_buffer.cpp:
##########
@@ -118,6 +119,64 @@ void ResultBlockBuffer<ResultCtxType>::cancel(const
Status& reason) {
_result_batch_queue.clear();
}
+template <typename ResultCtxType>
+void
ResultBlockBuffer<ResultCtxType>::add_outfile_cleanup(std::function<void()>
cleanup) {
+ bool run_cleanup = false;
+ {
+ std::lock_guard<std::mutex> l(_lock);
+ if (_outfile_state == OutfileState::ABORTED) {
+ run_cleanup = true;
+ } else {
+ _outfile_cleanups.emplace_back(std::move(cleanup));
+ }
+ }
+ if (run_cleanup) {
+ cleanup();
+ }
+}
+
+template <typename ResultCtxType>
+void ResultBlockBuffer<ResultCtxType>::finish_outfile(bool success) {
+ std::vector<std::function<void()>> cleanups;
+ {
+ std::lock_guard<std::mutex> l(_lock);
+ if (success) {
Review Comment:
**[P1] Do not acknowledge success from an already-aborted buffer.**
`ResultBufferMgr::finish_outfile()` drops the map lock after finding a shared
pointer and later returns `true` regardless of this void transition. The
timeout thread can erase/abort the buffer first; this branch then sees
`ABORTED`, returns silently, and FE publishes success after files were deleted.
Conversely, `COMMITTED` expiry clears the only callbacks before another BE or
the marker fails, so compensating abort leaks that participant's files.
Propagate the actual transition result and keep provisional rollback state
until the global decision is final.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -1592,6 +1590,13 @@ public void executeAndSendResult(boolean isOutfileQuery,
boolean isSendFields,
break;
}
}
+ if (isOutfileQuery) {
+ coordBase.finishOutfile(true);
+ if
(!Strings.isNullOrEmpty(outFileClause.getSuccessFileName())) {
+ outfileWriteSuccess(outFileClause);
Review Comment:
**[P1] Include the success marker in the same rollback ownership.** The BE
can append and close the marker successfully and then lose the RPC response.
This call throws while `outfileCommitted` is still false, so the catch
broadcasts data-file rollback, but the already-closed marker has no cleanup
owner. The final directory then advertises success with no data. Make marker
publication idempotent/transactional with the data callbacks, and test response
loss after marker close.
##########
be/src/exec/sink/writer/vfile_result_writer.cpp:
##########
@@ -487,7 +492,43 @@ Status VFileResultWriter::close(Status exec_status) {
}
st = _close_file_writer(true);
}
+ if (!st.ok()) {
+ _cleanup_created_files();
+ } else {
+ _register_created_files_cleanup();
+ }
return st;
}
+void VFileResultWriter::_cleanup_created_files() {
+ _vfile_writer.reset();
Review Comment:
**[P1] Abort before destroying Parquet/ORC transformers.** Their
output-stream destructors call `FileWriter::close()`, so this reset can
flush/finalize a cancelled S3 file and complete its multipart upload before
explicit `abort()`. `_multipart_upload_completed` then makes abort a no-op,
exposing partial output until delete and leaving it published if delete fails.
Add a non-finalizing detach/cancel path, or abort safely while the raw writer
is alive, and test that cancelled Parquet/ORC never call
PutObject/CompleteMultipartUpload.
##########
be/src/runtime/result_buffer_mgr.cpp:
##########
@@ -130,15 +130,28 @@ Status ResultBufferMgr::find_buffer(const TUniqueId&
finst_id,
}
bool ResultBufferMgr::cancel(const TUniqueId& unique_id, const Status& reason)
{
- std::unique_lock<std::shared_mutex> wlock(_buffer_map_lock);
- auto iter = _buffer_map.find(unique_id);
-
- auto exist = _buffer_map.end() != iter;
- if (exist) {
- iter->second->cancel(reason);
+ std::shared_ptr<ResultBlockBufferBase> buffer;
+ {
+ std::unique_lock<std::shared_mutex> wlock(_buffer_map_lock);
+ auto iter = _buffer_map.find(unique_id);
+ if (iter == _buffer_map.end()) {
+ return false;
+ }
+ buffer = std::move(iter->second);
_buffer_map.erase(iter);
}
- return exist;
+ // Outfile rollback can perform remote I/O, so it must not hold the
manager-wide map lock.
+ buffer->cancel(reason);
Review Comment:
**[P1] Keep remote deletion off the sole expiry thread.** `cancel_thread()`
processes every due result buffer sequentially through this call, and
`ResultBlockBuffer::cancel()` now performs synchronous HDFS/Broker/S3
`batch_delete` before clearing the buffer or notifying waiters. One slow or
hung storage request can stop expiry for unrelated queries, retain their result
queues indefinitely, and hang `ResultBufferMgr::stop()` while it joins this
worker. Release local resources first and run bounded/retryable cleanup on an
I/O executor; test that a blocked first delete does not prevent a second buffer
from expiring.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -1592,6 +1590,13 @@ public void executeAndSendResult(boolean isOutfileQuery,
boolean isSendFields,
break;
}
}
+ if (isOutfileQuery) {
+ coordBase.finishOutfile(true);
Review Comment:
**[P1] Finalize before exposing the OUTFILE result set.** The loop above has
already sent the OUTFILE fields and summary row when this fallible distributed
commit runs. If a buffer expired or an RPC/mixed-version participant fails, the
catch path deletes the files and sends an error only after result packets; the
adjacent compatibility comment notes that some drivers treat an error after
fields as success. Buffer the summary until participant commit and marker
publication succeed, and test that no result-set packet precedes a finalization
failure.
##########
be/src/runtime/result_block_buffer.cpp:
##########
@@ -104,6 +104,7 @@ Status ResultBlockBuffer<ResultCtxType>::close(const
TUniqueId& id, Status exec_
template <typename ResultCtxType>
void ResultBlockBuffer<ResultCtxType>::cancel(const Status& reason) {
+ release_outfile_cleanup();
Review Comment:
**[P1] Run deferred cleanup under its owning memory tracker.** The path
vector/callback is allocated by `AsyncResultWriter` under the query's
`SCOPED_ATTACH_TASK`, but this call executes and destroys it before the
existing `_mem_tracker` switch below; explicit RPC rollback has no
buffer-tracker switch either. Doris charges alloc/free to the current thread
tracker, so expiry or rollback can leave the query tracker inflated, credit
Orphan/service tracking on free, and allocate `batch_delete`'s proportional
vector outside the query limit. Move the tracker scope above cleanup or
capture/transfer ownership, and test large-manifest rollback and
committed-expiry tracker balances.
##########
be/src/exec/sink/writer/vfile_result_writer.cpp:
##########
@@ -487,7 +492,43 @@ Status VFileResultWriter::close(Status exec_status) {
}
st = _close_file_writer(true);
}
+ if (!st.ok()) {
+ _cleanup_created_files();
+ } else {
+ _register_created_files_cleanup();
+ }
return st;
}
+void VFileResultWriter::_cleanup_created_files() {
+ _vfile_writer.reset();
+ if (_file_writer_impl != nullptr) {
+ if (_created_file_paths.empty()) {
+ _created_file_paths.emplace_back(_file_writer_impl->path());
+ }
+ WARN_IF_ERROR(_file_writer_impl->abort(), "failed to abort outfile
writer");
+ _file_writer_impl.reset();
+ }
+ if (_file_system == nullptr && _storage_type ==
TStorageBackendType::LOCAL) {
+ _file_system = io::global_local_filesystem();
+ }
+ if (_file_system != nullptr && !_created_file_paths.empty()) {
+ WARN_IF_ERROR(_file_system->batch_delete(_created_file_paths),
+ "failed to remove files from aborted outfile");
+ }
+ _created_file_paths.clear();
+}
+
+void VFileResultWriter::_register_created_files_cleanup() {
+ if (_sinker == nullptr || _file_system == nullptr ||
_created_file_paths.empty()) {
+ return;
+ }
+ auto file_system = _file_system;
+ auto paths = std::move(_created_file_paths);
+ _sinker->add_outfile_cleanup([file_system = std::move(file_system), paths
= std::move(paths)] {
+ WARN_IF_ERROR(file_system->batch_delete(paths),
Review Comment:
**[P1] Preserve and report failed cleanup ownership.** This one-shot `void`
callback only warns on `batch_delete` failure after `finish_outfile(false)` has
removed the callback/path manifest; the RPC reports OK and no retry owner
remains. Local/Broker/HDFS also stop at the first bad path, and S3 at the first
failed chunk, so later independently deletable outputs are never attempted.
Attempt every owned path, aggregate/propagate errors, and retain only failures
for bounded retry or orphan cleanup; cover total and first-of-many deletion
failures.
##########
be/src/exec/sink/writer/vfile_result_writer.cpp:
##########
@@ -123,13 +123,18 @@ 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,
- }));
+ if (_file_system == nullptr) {
Review Comment:
**[P2] Preserve per-file Broker selection.** Previously every rotated
filename passed through `get_broker_index(..., path)`, so suffixes could
distribute work across configured brokers. Retaining the filesystem created
from the first path pins all later creates, writes, and cleanup to that broker,
removing the existing distribution and making later rotations fail when it is
unavailable even if peers are healthy. Associate each owned path with its
per-path-selected Broker filesystem and add a multi-broker rotation test.
##########
be/src/exec/sink/writer/vfile_result_writer.cpp:
##########
@@ -123,13 +123,18 @@ 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,
- }));
+ if (_file_system == nullptr) {
+ 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));
+ }
+ const io::FileWriterOptions options {.write_file_cache = false,
.sync_file_data = false};
+ RETURN_IF_ERROR(_file_system->create_file(file_name, &_file_writer_impl,
&options));
Review Comment:
**[P1] Claim this query-owned path before the side-effecting create.**
Remote create/open is not failure-atomic: Broker retries `openWriter` after a
transport exception, and HDFS open can create/truncate the path before the
caller observes an error. Because the vector is still empty when this call
fails, `close(error)` has nothing to batch-delete and the first partial path
can survive. Record the deterministic part path before `create_file` (or always
delete it on failure) and cover an acknowledgement-lost create; success-marker
creation has the same ordering.
##########
fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java:
##########
@@ -113,6 +113,11 @@ public Future<InternalService.POutfileWriteSuccessResult>
outfileWriteSuccessAsy
return stub.outfileWriteSuccess(request);
}
+ public Future<InternalService.POutfileWriteFinishedResult>
outfileWriteFinishedAsync(
+ InternalService.POutfileWriteFinishedRequest request) {
+ return stub.outfileWriteFinished(request);
Review Comment:
**[P1] Bound the new finalization RPC.** This bare stub is consumed by
sequential `Future.get()` calls with no deadline. A live-but-stalled BE,
saturated heavy-work pool, or remote `batch_delete` can therefore hang beyond
the query deadline, prevent later participants from receiving commit/abort, and
delay `coordBase.cancel()` on the original error path. Apply a deadline derived
from remaining query/cleanup time and continue best-effort rollback for every
participant after one timeout.
##########
be/src/runtime/result_block_buffer.h:
##########
@@ -134,6 +141,10 @@ class ResultBlockBuffer : public ResultBlockBufferBase {
const int _be_exec_version;
const segment_v2::CompressionTypePB
_fragment_transmission_compression_type;
const int _buffer_limit;
+
+ enum class OutfileState : uint8_t { PENDING, COMMITTED, ABORTED };
+ OutfileState _outfile_state = OutfileState::PENDING;
+ std::vector<std::function<void()>> _outfile_cleanups;
Review Comment:
**[P2] Do not discard pending ownership on BE shutdown.** These callbacks
are the only manifest of published part paths, but `ResultBufferMgr::stop()`
merely stops/joins the expiry thread and the default buffer/vector destructors
never invoke them. A graceful BE shutdown after writer close but before FE's
global decision therefore makes the query fail while permanently leaking its
Local/Broker/HDFS/S3 files; restart cannot recover the IDs. Drain pending
buffers during graceful stop and define a durable/recoverable hard-restart
policy.
##########
be/src/exec/sink/writer/vfile_result_writer.cpp:
##########
@@ -487,7 +492,43 @@ Status VFileResultWriter::close(Status exec_status) {
}
st = _close_file_writer(true);
}
+ if (!st.ok()) {
+ _cleanup_created_files();
+ } else {
+ _register_created_files_cleanup();
+ }
return st;
}
+void VFileResultWriter::_cleanup_created_files() {
+ _vfile_writer.reset();
+ if (_file_writer_impl != nullptr) {
+ if (_created_file_paths.empty()) {
+ _created_file_paths.emplace_back(_file_writer_impl->path());
+ }
+ WARN_IF_ERROR(_file_writer_impl->abort(), "failed to abort outfile
writer");
Review Comment:
**[P1] Retain the multipart handle when abort fails.**
`S3FileWriter::abort()` can return a network/provider error after parts and an
`upload_id` exist, but this warning is followed by destroying the writer and
its only upload ID. The subsequent `DeleteObject` targets only the final key;
it can return OK while hidden multipart parts remain, and this immediate
failure path registers no callback for retry. Hand `{bucket, key, upload_id}`
to a bounded retry/orphan owner and clear it only after abort succeeds;
fault-test a failed first abort followed by retry with the same ID.
--
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]
