github-actions[bot] commented on code in PR #67328:
URL: https://github.com/apache/doris/pull/67328#discussion_r3901872558
##########
be/src/runtime/result_block_buffer.cpp:
##########
@@ -103,19 +104,140 @@ 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;
+ for (int attempt = 0; attempt < 3; ++attempt) {
+ status = cleanup();
+ if (status.ok()) {
+ return status;
+ }
+ }
+ // Cancellation removed this buffer from the manager, so bounded
inline retries are the
+ // last live-query owner for a cleanup registered after that point.
+ return status;
+ }
+ return Status::OK();
+}
+
+template <typename ResultCtxType>
+Status ResultBlockBuffer<ResultCtxType>::finish_outfile(OutfileOperation
operation) {
+ SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);
+ std::vector<OutfileCleanup> cleanups;
+ {
+ std::lock_guard<std::mutex> l(_lock);
+ if (operation == OutfileOperation::PREPARE) {
+ if (_outfile_state == OutfileState::ABORTED) {
+ return Status::Cancelled("outfile result buffer was already
aborted");
+ }
+ if (_outfile_state == OutfileState::PENDING) {
+ _outfile_state = OutfileState::PREPARED;
+ }
+ return Status::OK();
+ }
+ if (operation == OutfileOperation::COMMIT) {
+ if (_outfile_state == OutfileState::COMMITTED) {
+ return Status::OK();
+ }
+ if (_outfile_state != OutfileState::PREPARED) {
+ return Status::InternalError("outfile result buffer is not
prepared");
+ }
+ // Keep rollback ownership after this provisional acknowledgement
so an explicit ABORT
+ // can compensate a COMMIT that reached only a subset of receivers.
+ _outfile_state = OutfileState::COMMITTED;
+ return Status::OK();
+ }
+ if (operation == OutfileOperation::ABORT) {
+ if (_outfile_state == OutfileState::ABORTED &&
_outfile_cleanups.empty()) {
Review Comment:
[P1] Do not treat an in-flight ABORT drain as empty/finished. FE
`OUTFILE_ABORT` can set `ABORTED`, swap the registered callbacks out, and block
in cleanup while timeout/shutdown erases the manager entry and
`release_outfile_cleanup()` re-enters here. This branch then returns OK because
the vector is only temporarily empty; if the first cleanup fails, it reinserts
after the manager owner is gone and no retry can find the buffer. This differs
from the prior late-registration thread: this callback was already registered
and is temporarily owned by the first ABORT stack. Track/join an active drain
(or transfer it to durable retry ownership), and add a latch test overlapping
ABORT with cancellation and a cleanup that fails before succeeding.
##########
be/src/service/internal_service.cpp:
##########
@@ -770,41 +819,156 @@ void
PInternalService::outfile_write_success(google::protobuf::RpcController* co
return;
}
- auto&& res = FileFactory::create_file_writer(file_type_res.value(),
ExecEnv::GetInstance(),
-
file_options.broker_addresses,
-
file_options.broker_properties, file_name,
- {
- .write_file_cache
= false,
- .sync_file_data =
false,
- });
- using T = std::decay_t<decltype(res)>;
- if (!res.has_value()) [[unlikely]] {
- st = std::forward<T>(res).error();
+ io::FSPropertiesRef properties(file_type_res.value());
+ properties.broker_addresses = &file_options.broker_addresses;
+ properties.properties = &file_options.broker_properties;
+ io::FileDescription file_description;
+ file_description.path = file_name;
+ auto fs_res = FileFactory::create_fs(properties, file_description);
+ if (!fs_res.has_value()) [[unlikely]] {
+ st = std::move(fs_res).error();
+ st.to_protobuf(result->mutable_status());
+ return;
+ }
+ auto file_system = std::move(fs_res).value();
+
+ if (request->operation() == OUTFILE_MARKER_DELETE) {
+ // Delete only a path created by this token. A blind rollback
could otherwise remove a
+ // pre-existing user file when CREATE failed before acquiring
ownership.
+ if (owned_marker_path.empty()) {
+ Status::OK().to_protobuf(result->mutable_status());
+ return;
+ }
+ st = file_system->delete_file(owned_marker_path);
+ if (st.ok() || st.is<ErrorCode::NOT_FOUND>()) {
+ std::lock_guard marker_guard(outfile_marker_lock);
+ auto state_it = outfile_marker_states.find(marker_token);
+ if (state_it != outfile_marker_states.end() &&
+ state_it->second.owned_path == owned_marker_path) {
+ state_it->second.owned_path.clear();
+ state_it->second.updated_at =
std::chrono::steady_clock::now();
+ }
+ st = Status::OK();
+ }
+ st.to_protobuf(result->mutable_status());
+ return;
+ }
+ if (request->operation() != OUTFILE_MARKER_CREATE) {
+ Status::InvalidArgument("unknown OUTFILE success marker operation")
+ .to_protobuf(result->mutable_status());
+ return;
+ }
+
+ // Never claim an existing marker path; rollback is restricted to
token-owned paths below.
+ bool exists = true;
+ st = file_system->exists(file_name, &exists);
Review Comment:
[P1] Preserve legacy remote success-marker behavior when atomic OUTFILE is
disabled. This unconditional existence check also runs for an old FE (the new
fields are absent and `marker_token` falls back to the path) and for a new FE
below v14. Previously only LOCAL rejected an existing marker; S3/HDFS/Broker
went directly through their existing writer semantics, so a BE-first rolling
upgrade can make an otherwise legacy query fail here after its data files were
written. Gate the new nonexistence/ownership policy on
`file_options.enable_atomic_outfile` (or an explicit protocol bit), and add
field-absent/version-13 remote-marker compatibility coverage.
##########
be/src/exec/sink/writer/vfile_result_writer.cpp:
##########
@@ -123,13 +170,21 @@ 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);
Review Comment:
[P2] Avoid retaining one initialized filesystem wrapper per rotated file.
This call runs for every part and `_created_files` keeps each shared pointer;
atomic success then moves the whole vector into a cleanup callback retained
until the result buffer expires (300 seconds by default). With the supported 5
MB split size, a large export can therefore pin O(parts)
filesystem/configuration/client-holder objects even though S3/HDFS clients are
cache-shared and Broker cleanup needs at most one filesystem per selected
endpoint. Keep the O(parts) path manifest, but deduplicate cleanup filesystems
by storage identity/endpoint, and add a many-rotation test that verifies
retained filesystem instances stay bounded.
##########
be/src/exec/sink/writer/vfile_result_writer.cpp:
##########
@@ -123,13 +170,21 @@ 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;
Review Comment:
[P1] Make deferred Broker cleanup own this configuration. These pointers
refer into `ResultFileSinkOperatorX::_file_opts`, and `BrokerFileSystem` stores
the selected `TNetworkAddress` and property map as `const&`. Atomic close moves
that filesystem into a result-buffer cleanup callback, but pipeline teardown
destroys `_file_opts`; a later timeout, partial-COMMIT failure, or marker
failure then calls `delete_file()` through dangling references, risking a BE
crash and failed rollback. Copy the selected address/properties into the
retained filesystem or cleanup manifest, and add a lifetime test that destroys
the operator-owned options before invoking ABORT.
--
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]
