github-actions[bot] commented on code in PR #67328:
URL: https://github.com/apache/doris/pull/67328#discussion_r3916225877
##########
be/src/service/internal_service.cpp:
##########
@@ -770,41 +823,181 @@ 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) {
+ if (!atomic_outfile_marker) {
+ Status::InvalidArgument("legacy OUTFILE has no marker rollback
operation")
+ .to_protobuf(result->mutable_status());
+ return;
+ }
+ // 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);
+ const bool delete_succeeded = 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()) {
+ complete_outfile_marker_delete(&state_it->second,
owned_marker_path,
+ delete_succeeded,
+
std::chrono::steady_clock::now());
+ }
+ }
+ if (delete_succeeded) {
+ 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;
+ }
+
+ if (should_check_outfile_marker_existence(file_options,
+
result_file_sink.storage_backend_type)) {
+ // Never claim an existing marker path; rollback is restricted to
token-owned paths.
+ 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 =
should_sync_outfile_marker(
+
result_file_sink.storage_backend_type)};
+ // The create acknowledgement can be lost after publishing the marker,
so ownership is
+ // retained by marker_token for a compensating DELETE.
+ st = file_system->create_file(file_name, &file_writer, &options);
+ if (!st.ok()) {
+ // A failed create does not prove ownership; deleting here could
remove a file that won
+ // a concurrent exclusive create or existed before this query.
st.to_protobuf(result->mutable_status());
return;
}
- std::unique_ptr<doris::io::FileWriter> _file_writer_impl =
std::forward<T>(res).value();
+ if (atomic_outfile_marker) {
+ // Claim the path before append/close because either call can
publish data before
+ // reporting failure, and a failed immediate delete must remain
retryable.
+ std::lock_guard marker_guard(outfile_marker_lock);
+
record_outfile_marker_ownership(&outfile_marker_states[marker_token], file_name,
+ std::chrono::steady_clock::now());
+ }
+
+ auto remove_incomplete_marker = [&]() {
+ Status delete_status = file_system->delete_file(file_name);
Review Comment:
**[P1] Do not delete an unowned legacy marker after a failed write.** For
`enable_atomic_outfile=false`, remote requests intentionally bypass the
existence/ownership policy above. On S3, `create_file()` only constructs the
writer, so if this marker key already exists and the replacement `PutObject`
fails before publishing anything, the old marker is still intact here; this
unconditional `delete_file()` then removes that pre-existing object even though
this request never owned it. The earlier legacy thread covered the pre-write
existence rejection, not this failure cleanup. Restrict removal to atomic
token-owned markers (or otherwise prove this request published the path), and
cover an existing S3 marker plus a failed replacement write.
##########
be/src/exec/sink/writer/vfile_result_writer.cpp:
##########
@@ -67,6 +67,60 @@ namespace doris {
static double nons_to_second = 1000000000.00;
+namespace {
+
+using OutfileFileSystemId = int32_t;
+using OwnedOutfile = std::pair<OutfileFileSystemId, io::Path>;
+
+struct OutfileCleanupState {
+ std::shared_ptr<io::FileWriter> writer;
+ std::unordered_map<OutfileFileSystemId, std::shared_ptr<io::FileSystem>>
file_systems;
+ std::vector<OwnedOutfile> files;
+
+ Status cleanup() {
+ Status first_failure = Status::OK();
+ if (writer != nullptr) {
+ Status status = writer->abort();
+ if (status.ok()) {
+ writer.reset();
+ } else {
+ first_failure = status;
+ }
+ }
+
+ std::vector<OwnedOutfile> failed_files;
+ for (auto& [file_system_id, path] : files) {
+ auto file_system = file_systems.find(file_system_id);
+ Status status = file_system == file_systems.end()
+ ? Status::InternalError(
+ "missing filesystem for OUTFILE
cleanup path {}",
+ path.string())
+ : file_system->second->delete_file(path);
Review Comment:
**[P1] Batch S3 cleanup before the five-second ABORT deadline.** This loop
turns every retained part into a separate synchronous `DeleteObject` RPC. The
new rotation test explicitly treats 1,000 paths as supported, but even 10 ms
per successful request makes rollback take about 10 seconds before retries,
beyond FE's fixed five-second ABORT budget and monopolizing the heavy-work or
lazy-release worker. `S3FileSystem::batch_delete()` already sends up to 1,000
keys per request. Group paths by retained filesystem and use the batch API for
the success path, falling back to per-key diagnosis/retention only when a batch
fails; add a many-part test that bounds request count under injected latency.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileSink.java:
##########
@@ -45,32 +46,45 @@ public class LogicalFileSink<CHILD_TYPE extends Plan>
extends LogicalSink<CHILD_
private final String filePath;
private final String format;
private final Map<String, String> properties;
+ private final int beExecVersion;
public LogicalFileSink(String filePath, String format,
Map<String, String> properties, List<NamedExpression> outputExprs,
CHILD_TYPE child) {
- this(filePath, format, properties, outputExprs, Optional.empty(),
Optional.empty(), child);
+ this(filePath, format, properties, Config.be_exec_version, outputExprs,
Review Comment:
**[P1] Snapshot this capability at execution time, not retained-plan
construction.** `PrepareCommand` keeps this `LogicalFileSink` and
`ExecuteCommand` reuses the same logical plan for later EXECUTEs; queued EXPORT
jobs likewise build the sink before their task runs. Every rewrite preserves
this final value, so a cluster transition after PREPARE/job creation leaves a
new execution serializing the stale `enable_atomic_outfile` value, forcing the
stale query option, and selecting the corresponding finalization RPC path. A
14-to-13 transition can therefore select protocol behavior the current
participants do not support, while 13-to-14 silently remains legacy. Make an
execution-local plan copy that reads the version once immediately before
planning, then preserve that value only for that execution; cover prepared
EXECUTE and queued EXPORT transitions in both directions.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -1686,28 +1840,23 @@ 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;
+ if (operation ==
InternalService.POutfileSuccessOperation.OUTFILE_MARKER_DELETE) {
+ timeoutMs = Math.max(1,
Math.min(Config.remote_fragment_exec_timeout_ms, OUTFILE_CLEANUP_TIMEOUT_MS));
+ } else if (deadlineMs == Long.MAX_VALUE) {
Review Comment:
**[P1] Preserve the uncompensated legacy marker wait.** The `Long.MAX_VALUE`
overload is the version-13 path, but this helper now gives it a fresh
`remote_fragment_exec_timeout_ms` deadline and bounded `Future.get()`. If
remote CREATE runs longer, FE returns an error while the BE's already-queued
lambda ignores controller cancellation and can later publish the marker.
Because `atomicOutfile` is false, ABORT is a no-op and no marker ownership is
recorded for DELETE; the pre-PR code instead kept waiting for this legacy
operation. Apply the bounded/absolute deadline only to atomic operations (or
add a legacy-compatible publication fence), and test a slow version-13 remote
marker cannot appear after FE has failed the 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]
