Gabriel39 commented on code in PR #67328:
URL: https://github.com/apache/doris/pull/67328#discussion_r3900455915
##########
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:
Agreed that this is a real risk, but it is outside this PR's fix boundary.
Query-token-local ownership cannot make two independent queries sharing one
destination object atomic; that requires storage-native conditional/generation
semantics or a durable cross-query namespace transaction. I updated the PR's
Atomicity scope to explicitly exclude concurrent OUTFILE queries targeting the
same destination directory or success-marker path.
##########
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:
Fixed. Atomic Arrow Flight OUTFILE is now rejected before coordinator
construction and registration. The new test invokes executeAndSendResult and
verifies that no coordinator is registered in QeProcessorImpl for the rejected
query ID.
##########
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:
Fixed. OutFileClause now snapshots the negotiated BE execution version on
first planning use, copies that snapshot during clause cloning, and reuses it
for FE finalization. The coordinator query option is aligned with the same
snapshot before execution, and finalization no longer rereads the mutable
global.
##########
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:
Fixed. Marker creation now enables synchronous file data only for LOCAL
storage, preserving the historical local durability behavior without changing
remote object-store writes. Added focused coverage in
OutfileMarkerStateTest.SyncsOnlyLocalSuccessMarker.
##########
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:
Fixed. Ordinary successful marker ownership is now reclaimed after the
existing one-hour protection window. A tombstone with an owned path is retained
because it represents a failed compensating delete; once that path is cleared,
it becomes reclaimable. Added focused tests for both cases.
##########
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:
Fixed. Atomic marker CREATE now receives the coordinator's absolute OUTFILE
deadline and uses min(remaining query time, remote RPC timeout). An exhausted
deadline fails before starting the RPC. Cleanup DELETE remains separately
bounded so compensation can still run after the query deadline.
--
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]
