github-actions[bot] commented on code in PR #67328:
URL: https://github.com/apache/doris/pull/67328#discussion_r3899764069
##########
be/src/service/internal_service.cpp:
##########
@@ -737,25 +781,38 @@ 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 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;
+ {
+ std::lock_guard marker_guard(outfile_marker_lock);
+ cleanup_expired_outfile_marker_states(now);
+ operation_lock =
outfile_marker_operation_locks[marker_token].lock();
Review Comment:
[P1] Serialize marker ownership by the destination object, not only by the
query token. Two queries targeting the same success path take different
mutexes, can both pass the non-atomic exists/create sequence, and both record
the same owned path. If either CREATE response is lost, that token's
compensating DELETE can then remove the marker after the other query committed.
Use conditional, generation-aware path ownership across BEs and make DELETE
verify that exact generation; add a deterministic two-token same-path rollback
test.
##########
be/src/service/internal_service.cpp:
##########
@@ -135,6 +138,47 @@ 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();) {
+ // A failed marker delete retains the only in-process rollback fence
and ownership record.
+ // Expire state only after the owned path has been deleted
successfully.
+ if (it->second.owned_path.empty() &&
Review Comment:
[P2] Reclaim ordinary successful marker states. Every successful CREATE
stores a nonempty owned_path, the success path never sends DELETE, and this
predicate only expires entries after that path has been cleared, so one
query-ID-keyed map node and path string remain for the BE lifetime. The minute
cleanup also scans the ever-growing map under the global lock. Retain
unresolved tombstones as required, but give non-tombstoned committed state a
bounded terminal lifecycle and test both cases.
##########
be/src/service/internal_service.cpp:
##########
@@ -770,41 +827,154 @@ 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);
+ if (st.ok() && exists) {
+ st = Status::InternalError("File already exists: {}", file_name);
+ }
+ if (!st.ok()) {
+ st.to_protobuf(result->mutable_status());
+ return;
+ }
+
+ io::FileWriterPtr file_writer;
+ const io::FileWriterOptions options {.write_file_cache = false,
.sync_file_data = false};
Review Comment:
[P2] Preserve the previous local sync behavior for the success marker too.
The old FileFactory local branch ignored this false option and created a writer
with sync enabled, while the new direct filesystem call honors false, so
LocalFileWriter skips both fdatasync and the parent-directory sync. The
data-writer fix does not cover this marker site; a crash after this RPC returns
OK can leave durable data without its completion marker. Enable sync for LOCAL
here and add a marker-specific durability test.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -1686,28 +1800,17 @@ private void outfileWriteSuccess(OutFileClause
outFileClause) throws Exception {
sink.setStorageBackendType(storageType.toThrift());
// 4. get BE
- TNetworkAddress address = null;
- for (Backend be :
Env.getCurrentSystemInfo().getBackendsByCurrentCluster().values()) {
- if (be.isAlive()) {
- address = new TNetworkAddress(be.getHost(), be.getBrpcPort());
- break;
- }
- }
- if (address == null) {
- String computeGroupHints = "";
- if (Config.isCloudMode()) {
- // null: computeGroupNotFoundPromptMsg select cluster for hint
msg
- computeGroupHints =
ComputeGroupMgr.computeGroupNotFoundPromptMsg(null);
- }
- throw new AnalysisException("No Alive backends" +
computeGroupHints);
- }
-
// 5. send rpc to BE
POutfileWriteSuccessRequest request =
POutfileWriteSuccessRequest.newBuilder()
- .setResultFileSink(ByteString.copyFrom(new
TSerializer().serialize(sink))).build();
+ .setResultFileSink(ByteString.copyFrom(new
TSerializer().serialize(sink)))
+ .setOperation(operation)
+ .setMarkerToken(DebugUtil.printId(context.queryId())).build();
+ long timeoutMs = operation ==
InternalService.POutfileSuccessOperation.OUTFILE_MARKER_DELETE
+ ? Math.max(1, Math.min(Config.remote_fragment_exec_timeout_ms,
OUTFILE_CLEANUP_TIMEOUT_MS))
+ : Math.max(1, Config.remote_fragment_exec_timeout_ms);
Review Comment:
[P1] Keep marker CREATE inside the same absolute query deadline as PREPARE
and COMMIT. This branch starts a fresh remote_fragment_exec_timeout_ms wait
after those phases have consumed the query budget; the timeout checker only
cancels coord and neither cancels this future nor prevents the waiting thread
from publishing the marker, sending the deferred row, and setting EOF
afterward. Pass the remaining deadline into marker publication, fail before
CREATE when it is exhausted, and test a short query whose COMMIT consumes
nearly all of its budget.
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java:
##########
@@ -756,6 +757,7 @@ public TResultFileSinkOptions toSinkOptions() {
sinkOptions.setDeleteExistingFiles(deleteExistingFiles);
sinkOptions.setFileSuffix(fileSuffix);
sinkOptions.setWithBom(withBom);
+ sinkOptions.setEnableAtomicOutfile(Config.be_exec_version >=
SUPPORT_ATOMIC_OUTFILE_VERSION);
Review Comment:
[P1] Snapshot the atomic-OUTFILE capability once per query instead of
rereading this mutable global across phases. This planning-time value controls
whether BE writers retain cleanup ownership, but StmtExecutor recomputes the FE
transaction branch later and Coordinator checks the live value again for every
operation. A 14-to-13 change can publish legacy success while PENDING callbacks
later delete the acknowledged files; 13-to-14 can promise rollback after
writers already discarded their manifests. Carry one negotiated version through
sink options, query options, FE branching, and RPC encoding, and test both live
transitions.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -1519,8 +1521,18 @@ public void executeAndSendResult(boolean isOutfileQuery,
boolean isSendFields,
if (isOutfileQuery) {
outFileClause = queryStmt.getOutFileClause();
Preconditions.checkState(outFileClause != null, "OUTFILE query
must have OutFileClause");
+ if (Config.be_exec_version >=
OutFileClause.SUPPORT_ATOMIC_OUTFILE_VERSION
+ &&
context.getConnectType().equals(ConnectType.ARROW_FLIGHT_SQL)) {
+ throw new UserException("Atomic OUTFILE is not supported over
Arrow Flight SQL");
Review Comment:
[P1] Reject atomic Arrow Flight OUTFILE before constructing/registering the
coordinator. Both distributed branches have already called registerQuery here,
but this throw is before the method's cleanup try/finally and before
addFlightSqlDeferredExecutor. Because Flight already set
returnResultFromLocal=false, handleQueryWithRetry also skips finalizeQuery, so
each rejected request permanently retains the coordinator and QueryInfo in
QeProcessor. Move the capability check ahead of registration or put it under
the normal cleanup scope, and assert rejection leaves no registered query.
--
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]
