[fix](outfile) Make distributed OUTFILE cleanup atomic - #67328
Conversation
Track every file created by an OUTFILE writer and defer exact-path cleanup until the coordinator confirms global success. Abort active multipart uploads and roll back all receivers when execution, finalization, or success marker creation fails. Add focused ASAN unit coverage for local files, ownership isolation, distributed result-buffer lifecycle, and multipart aborts. Issue Number: None
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
|
run buildall |
|
/review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Review status
REQUEST_CHANGES — capped/incomplete after Round 3.
I found 15 substantiated issues (13 P1, 2 P2), all attached inline. Two new valuable findings appeared in the third and final permitted round (lost S3 multipart-abort ownership and cross-thread memory-tracker mismatch), so the review contract forbids a fourth discovery round and I am not claiming convergence. Every candidate currently in the ledger has nevertheless been independently accepted, merged, or dismissed with evidence.
Required checkpoint conclusions
- Goal and proof: The goal is to remove OUTFILE artifacts for failed/cancelled queries without deleting successful output. The current code does not accomplish that across Nereids, Arrow Flight, mixed-version operation, distributed timeout/marker failures, delete/abort errors, or process lifecycle boundaries. The added BE unit tests prove only direct sequential success cases.
- Scope and clarity: The code is locally focused, but the behavior is a distributed commit/compensation protocol spanning writer publication, result buffers, FE participant discovery, marker creation, and client delivery. The abstraction is incomplete at those boundaries.
- Concurrency: The buffer mutex prevents duplicate callback execution, and remote I/O is outside the manager/buffer locks, so I found no lock-order deadlock. However, map removal versus the void state transition can acknowledge success after abort or discard provisional rollback state, and synchronous deletion can block the sole expiry worker.
- Lifecycle: Callback/path ownership is in-memory only. Timeout, graceful shutdown, S3 abort failure, and cross-thread callback destruction can lose cleanup state or corrupt memory accounting. No new static-initialization dependency was introduced.
- Configuration: No configuration item was added.
- Compatibility: The new unversioned FE/BE RPC handshake has no capability gate. Doris's supported BE-first rolling-upgrade order leaves old-FE/new-BE queries pending and later deletes acknowledged output; the inverse mix cannot roll old BEs back.
- Parallel paths and conditions: Legacy MySQL parallel/non-parallel buffer IDs are internally consistent. Nereids uses a different receiver owner, Arrow Flight uses endpoints and returns early, and both miss finalization. Point-query short circuit is not a reachable OUTFILE sibling. The new success/abort conditions do not establish a durable global decision.
- Tests and expected results: The new local-file, S3-abort, and result-buffer tests are deterministic and their asserted sequential results are correct, but there are no FE/end-to-end, Nereids, Arrow, mixed-version, multi-BE partial-failure, response-loss, deletion-failure, blocking-expiry, shutdown, or tracker-balance tests.
- Observability: Warnings exist, but cleanup Status is discarded and exact ownership is then destroyed, so logs cannot repair or retry the failure. Query/path identifiers, retry ownership, and metrics are insufficient for this protocol.
- Persistence and data writes: No EditLog path is involved, but the external-file ownership manifest and global decision are not durable. Data files and the success marker are not atomically committed/rolled back, and FE/BE lifecycle transitions can leak files or delete successful output.
- FE/BE variables and protocol: Proto, client, proxy, and BE service definitions match for the new RPC, but capability negotiation and participant transport are missing from Nereids/Arrow paths.
- Performance and resource use: Per-file Broker distribution is removed, FE finalization waits are unbounded and sequential, the sole expiry worker performs blocking remote cleanup, and deferred path vectors can be charged/freed under different trackers.
- Other correctness: OUTFILE fields and the summary row are sent before the fallible global decision, allowing some drivers to observe success for files that the catch path subsequently deletes.
User focus: No additional user-provided focus was supplied; the full PR was reviewed.
Validation: Static review of the authoritative bundle at head 2233b4e56dc2f32e5f10513b3179a39ab9f66aa1. Builds and tests were not run because the review prompt explicitly prohibited them.
| @Override | ||
| public void finishOutfile(boolean success) throws Exception { | ||
| Map<TNetworkAddress, InternalService.POutfileWriteFinishedRequest.Builder> requests = new HashMap<>(); | ||
| for (ResultReceiver receiver : receivers) { |
There was a problem hiding this 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.
| } | ||
| 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)] { |
There was a problem hiding this comment.
[P1] Gate this cleanup protocol on FE capability. Doris rolling-upgrade guidance 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.
| break; | ||
| } | ||
| } | ||
| if (isOutfileQuery) { |
There was a problem hiding this 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.
| std::vector<std::function<void()>> cleanups; | ||
| { | ||
| std::lock_guard<std::mutex> l(_lock); | ||
| if (success) { |
There was a problem hiding this 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.
| if (isOutfileQuery) { | ||
| coordBase.finishOutfile(true); | ||
| if (!Strings.isNullOrEmpty(outFileClause.getSuccessFileName())) { | ||
| outfileWriteSuccess(outFileClause); |
There was a problem hiding this 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.
| } | ||
| return exist; | ||
| // Outfile rollback can perform remote I/O, so it must not hold the manager-wide map lock. | ||
| buffer->cancel(reason); |
There was a problem hiding this 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.
| 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), |
There was a problem hiding this 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.
| } | ||
| } | ||
| if (isOutfileQuery) { | ||
| coordBase.finishOutfile(true); |
There was a problem hiding this 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.
| 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"); |
There was a problem hiding this 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.
|
|
||
| template <typename ResultCtxType> | ||
| void ResultBlockBuffer<ResultCtxType>::cancel(const Status& reason) { | ||
| release_outfile_cleanup(); |
There was a problem hiding this 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 UT Coverage ReportIncrement line coverage Increment coverage report
|
TPC-H: Total hot run time: 17055 ms |
TPC-DS: Total hot run time: 83612 ms |
ClickBench: Total hot run time: 14.76 s |
|
run buildall |
|
/review |
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
Gabriel39
changed the title
FE Regression Coverage ReportIncrement line coverage |
There was a problem hiding this comment.
Review status
REQUEST_CHANGES — current-head follow-up review completed after two convergence rounds. Eight distinct issues below remain in addition to the existing inline threads; the PR is not ready to merge.
Required checkpoint conclusions
- Goal/data correctness: the atomic OUTFILE protocol still exposes a success marker before the distributed commit decision, and legacy Config<14 execution calls an RPC unavailable on pre-14 BEs.
- Concurrency/lifecycle: result-buffer shutdown can miss senders created after the stop snapshot, and late ABORTED cleanup registration can strand failed deletion ownership.
- Remote storage/error handling: S3 completion verification can leave a completed multipart upload in an un-abortable retry state; marker tombstones can expire after failed deletion and permit resurrection.
- FE protocol/deadlines: Nereids finalization uses the zero-valued legacy deadline and therefore a 1 ms timeout.
- Compatibility: FEP-002 below is the actionable FE-configured-old-version path; the mixed participant variant was dismissed because pre-14 BEs reject execution version 14 before finalization.
- Tests: the changed files add BE unit coverage but no FE/end-to-end tests for Nereids deadlines, packet ordering, rollback, marker visibility, or old-version RPC compatibility. This is recorded as a coverage gap, not a separate inline blocker.
- Performance/durability: local OUTFILE now passes sync_file_data=false where the previous local factory defaulted to synchronous close; see BW-002 below.
- User focus: no additional user-provided focus was supplied; the full authoritative bundle was reviewed.
Builds and tests were not run because the review prompt prohibits them. Existing review comments are hard duplicate fences and are not repeated here.
| long timeoutMs = operation == InternalService.POutfileWriteOperation.OUTFILE_ABORT | ||
| ? Math.max(1, Math.min(Config.remote_fragment_exec_timeout_ms, OUTFILE_CLEANUP_TIMEOUT_MS)) | ||
| : Math.max(1, Math.min(Config.remote_fragment_exec_timeout_ms, | ||
| timeoutDeadline - System.currentTimeMillis())); |
There was a problem hiding this comment.
[P1] finishOutfile() derives the PREPARE/COMMIT timeout from Coordinator.timeoutDeadline, but that field is initialized only inside Coordinator.execInternal(). NereidsCoordinator.exec() overrides exec() and uses coordinatorContext.timeoutDeadline instead, so the inherited field stays at its default 0. Nereids finalization therefore reaches Math.max(1, timeoutDeadline - now) as 1 ms and ordinary BE RPC latency causes spurious failure. Use the coordinator-context deadline (or initialize the base field) for Nereids and add a normal-latency finalization test.
| } | ||
| // This also carries the legacy success acknowledgement needed by old BEs during | ||
| // rolling upgrades; atomic-capable BEs interpret it as the global COMMIT phase. | ||
| coordBase.finishOutfile(InternalService.POutfileWriteOperation.OUTFILE_COMMIT); |
There was a problem hiding this comment.
[P1] The call to finishOutfile(OUTFILE_COMMIT) is unconditional even when atomicOutfile is false (configured execution version below 14). In that compatibility mode a pre-14 BE has no outfile_write_finished RPC, so this fails after the legacy writer/marker path has run; abortOutfile() then returns immediately because atomic mode is disabled, potentially leaving the success marker behind and turning old-version OUTFILE queries into errors. Keep PREPARE/COMMIT/marker orchestration behind the atomic capability and preserve the pre-change path for old BEs, with a version-13 compatibility test.
| // The create response may be lost after the marker is durable, so rollback must | ||
| // conservatively delete it whenever this call does not complete successfully. | ||
| outfileMarkerMayExist = true; | ||
| outfileWriteSuccess(outFileClause, outfileMarkerBackend, |
There was a problem hiding this comment.
[P1] The atomic sequence creates and closes the success marker immediately after PREPARE, before finishOutfile(OUTFILE_COMMIT) acknowledges every receiver. The marker is thus externally visible while data is only prepared; if one COMMIT RPC then times out, rollback deletes files after a reader may already have treated the marker as durable completion. Publish the marker only after distributed commit succeeds (or keep it hidden until that decision) and test a partial-commit failure with a concurrent marker reader.
| if (_clean_thread) { | ||
| _clean_thread->join(); | ||
| } | ||
| std::vector<TUniqueId> remaining_ids; |
There was a problem hiding this comment.
[P1] stop() copies _buffer_map IDs under the lock, releases it, and only then cancels them. create_sender() has no stopping-state check, and ExecEnv::destroy() stops this manager before FragmentMgr/workload execution, so an in-flight fragment can insert a new buffer after the snapshot. That buffer is never canceled and its OUTFILE callbacks are lost when the manager is destroyed. Quiesce/reject new senders before the snapshot and drain until no buffers remain (or retain a shutdown cleanup owner).
| return Status::OK(); | ||
| } | ||
| if (run_cleanup) { | ||
| Status status = cleanup(); |
There was a problem hiding this comment.
[P1] When cancellation wins the race, ResultBufferMgr::cancel() erases the buffer and runs its only release_outfile_cleanup() pass before the asynchronous writer necessarily calls VFileResultWriter::close(). A later add_outfile_cleanup() executes inline in the ABORTED branch; if remote delete/abort transiently fails, the callback is merely reinserted into a buffer with no retry owner and is lost on destruction. Hand late registrations to a durable/bounded retry owner or keep cancellation draining callbacks until writers finish.
| } | ||
| } | ||
| for (auto it = outfile_marker_states.begin(); it != outfile_marker_states.end();) { | ||
| if (now - it->second.updated_at >= OUTFILE_MARKER_TOMBSTONE_TTL) { |
There was a problem hiding this comment.
[P1] cleanup_expired_outfile_marker_states() erases every marker state after one hour, including tombstones whose delete_file() failed and still retain owned_path. Once that entry is evicted, the rollback fence and ownership record are gone: a delayed CREATE can be accepted if the marker is absent, while a marker left behind by the failed delete has no state for a later DELETE to recover and is silently leaked. Do not expire tombstones until deletion is durably confirmed (and define recovery across BE restart), or late requests can resurrect or orphan rolled-back markers.
| RETURN_IF_ERROR(check_after_upload(client.get(), resp, _obj_storage_path_opts, _bytes_appended, | ||
| "complete_multipart")); | ||
|
|
||
| _multipart_upload_completed = true; |
There was a problem hiding this comment.
[P1] _multipart_upload_completed is set only after check_after_upload() returns. If CompleteMultipartUpload succeeds but the subsequent HEAD/size check transiently fails, this returns an error with the flag still false even though the object is already published. Cleanup then calls AbortMultipartUpload on a completed upload (typically NoSuchUpload) and retains the writer/ID forever. Mark completion immediately after the CompleteMultipartUpload response and treat an already-completed/NoSuchUpload abort as converged; add a post-complete HEAD failure test.
| // 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); | ||
| const io::FileWriterOptions options {.write_file_cache = false, .sync_file_data = false}; |
There was a problem hiding this comment.
[P2] The old FileFactory::create_file_writer(FILE_LOCAL, ..., options) ignored options, so local writers defaulted to sync_data=true and honored sync_file_on_close with fdatasync/directory sync. This path now passes .sync_file_data = false to LocalFileSystem, which propagates it to LocalFileWriter; successful local OUTFILE closes no longer get the prior crash-durability sync. Preserve the old local setting while retaining remote no-sync behavior, and add a local durability regression test.
FE UT Coverage ReportIncrement line coverage |
BE UT Coverage ReportIncrement line coverage Increment coverage report
|
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
FE Regression Coverage ReportIncrement line coverage |
### What problem does this PR solve? Issue Number: None Related PR: apache#67328 Problem Summary: Follow-up review found that atomic OUTFILE finalization could use the wrong Nereids deadline, contact old BEs, publish the marker before all receiver commits, miss late cleanup and shutdown races, and regress local or multipart durability. Keep the protocol version-gated, order the global decision before marker and client publication, and preserve bounded rollback ownership during the live query lifecycle. ### Release note None ### Check List (For Author) - Test: Unit Test - BE ASAN unit tests: 36/36 passed. - FE unit tests: 4/4 passed with Checkstyle and source/test compilation. - Build hygiene and clang-format 16 checks passed. - Clang-tidy was attempted but could not analyze the existing tree because of unrelated toolchain and header diagnostics. - Behavior changed: Yes. Supported execution versions publish OUTFILE success only after every receiver commits; older versions retain the legacy path. - Does this need documentation: No
|
run buildall |
|
/review |
There was a problem hiding this comment.
Requesting changes: six distinct current-head issues remain, with details inline.
Checkpoint conclusions:
- Goal and proof: the stable version-14 happy path orders writer ownership, EOS, PREPARE, COMMIT, marker publication, and client delivery coherently, but the six failure paths below prevent the atomic-cleanup contract from being established.
- Concurrency and lifecycle: registration-before-EOS and per-buffer locking were verified. Cross-token marker ownership, completed marker-state reclamation, and rejected Flight coordinator cleanup remain unsafe.
- Configuration and compatibility: fixed-version v13/v14 and optional wire fields are coherent, but the mutable execution-version capability is resampled across planning and finalization.
- Parallel and conditional paths: legacy/Nereids and parallel/non-parallel receiver IDs were traced; OUTFILE bypasses cache/retry. Arrow Flight rejection currently leaks its registered coordinator.
- Error handling, persistence, and performance: RPC statuses are generally checked and fan-out is bounded, but marker durability/deadline behavior and the unbounded marker-state scan remain correctness risks.
- Tests and validation: the added unit tests cover selected writer, S3, state-machine, ordering, and deadline mechanics, but not these six issues, live version transitions, true multi-BE fan-out, marker concurrency/durability, or Flight rejection cleanup. The pure-text header/build hygiene check passed; no local build or test was run under the review-only task. Live Clang Formatter and CheckStyle checks pass; broader CI remains pending.
Convergence: three complete review rounds were performed. Both final-round normal reviews and the independent risk pass returned NO_NEW_VALUABLE_FINDINGS after deduplication against all 23 existing inline threads. Exact head reviewed: aa474c1.
| { | ||
| std::lock_guard marker_guard(outfile_marker_lock); | ||
| cleanup_expired_outfile_marker_states(now); | ||
| operation_lock = outfile_marker_operation_locks[marker_token].lock(); |
There was a problem hiding this 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.
There was a problem hiding this 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.
| } | ||
|
|
||
| io::FileWriterPtr file_writer; | ||
| const io::FileWriterOptions options {.write_file_cache = false, .sync_file_data = false}; |
There was a problem hiding this 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.
There was a problem hiding this 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.
| 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() && |
There was a problem hiding this 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.
There was a problem hiding this 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.
| .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); |
There was a problem hiding this 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.
There was a problem hiding this 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.
| 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"); |
There was a problem hiding this 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.
There was a problem hiding this 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.
| sinkOptions.setDeleteExistingFiles(deleteExistingFiles); | ||
| sinkOptions.setFileSuffix(fileSuffix); | ||
| sinkOptions.setWithBom(withBom); | ||
| sinkOptions.setEnableAtomicOutfile(Config.be_exec_version >= SUPPORT_ATOMIC_OUTFILE_VERSION); |
There was a problem hiding this 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.
There was a problem hiding this 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.
Cloud UT Coverage ReportIncrement line coverage Increment coverage report
|
TPC-H: Total hot run time: 16774 ms |
TPC-DS: Total hot run time: 92347 ms |
ClickBench: Total hot run time: 16.39 s |
FE UT Coverage ReportIncrement line coverage |
|
run buildall |
|
/review |
There was a problem hiding this comment.
Review status: complete after two convergence rounds on head 290924d7c004ff5c18fffd2fc70eb22f9fe25c38. I found one distinct P1 correctness issue and am requesting changes.
Critical checkpoint conclusions:
- Goal and behavior: the nominal atomic OUTFILE sequence now prepares all participants, commits them, publishes the marker, and only then exposes the deferred client result. However, the generic inner retry can start a non-isolated attempt before the failed attempt is quiescent; the inline comment is blocking.
- Concurrency and lifecycle: I traced writer close and late cleanup registration, PENDING/PREPARED/COMMITTED/ABORTED transitions, expiry, lazy release, shutdown, marker-token locking, FE cancellation, and retry regeneration. Remaining buffer/marker lifecycle concerns are already covered by existing review threads; no distinct additional issue remained.
- Configuration, compatibility, and parallel paths: execution version 14, optional Thrift/protobuf fields, old-request fallback, legacy versus Nereids coordinators, and parallel query-ID versus non-parallel instance-ID buffer routing were checked. The still-separate Nereids capability snapshot is already covered by an existing thread, so I did not duplicate it. No persisted metadata or storage-format compatibility change is introduced.
- Data writes and rollback: ownership-before-create, file rotation, LOCAL/Broker/HDFS/S3 cleanup, Parquet/ORC abort order, multipart completion/abort, all-participant fan-out, and marker create/delete were reviewed. The accepted issue breaks attempt-level atomicity even though the individual writer and buffer mechanisms have cleanup paths.
- Conditions, performance, and observability: error/timeout branches, no-marker OUTFILE, Arrow Flight rejection, client-send ordering, and retry exception types were checked. RPC fan-out uses bounded waits and cleanup is moved off the expiry thread; I found no new performance or logging blocker.
- Tests: the added BE/FE tests cover the principal state transitions, writer cleanup, S3 behavior, marker state, commit ordering, capability stability, Flight rejection, and deadline arithmetic, but not the accepted retry/quiescence ordering. Per the review prompt I did not run builds or tests. Live CI currently shows BE UT failed, FE UT and compile pending, and formatter/CheckStyle passed; the TeamCity failure log requires authentication.
- User focus: no additional review focus was provided.
All Round 2 reviewers returned NO_NEW_VALUABLE_FINDINGS; every other candidate was either dismissed with code evidence or fenced by an existing inline thread.
| } | ||
| coordBase.cancel(internalErrorSt); | ||
| // set to null so that the retry logic will generate a new coordinator | ||
| this.coord = null; |
There was a problem hiding this comment.
[P1] Do not let atomic OUTFILE enter the generic inner retry here. Atomic mode deliberately leaves the MySQL channel unsent, and clearing this.coord removes the cancelled/timed-out guard, so an original fetch or finalization RpcException starts a new attempt with regenerated query/fragment IDs. The ABORT above is not a quiescence barrier: failures are suppressed, and even an OK response drains only callbacks already registered while fragment cancellation is asynchronous. The retry can therefore publish the common success marker in the same destination while old-attempt files or late cleanup are still active. Disable automatic retry for atomic OUTFILE unless attempts are namespace-isolated or compensation proves quiescence, and add a fault-injection test for this ordering.
There was a problem hiding this comment.
Fixed. Atomic OUTFILE now snapshots the query capability before entering the inner retry loop and immediately propagates RpcException/UserException instead of starting a new attempt. Ordinary queries and legacy OUTFILE retain the existing retry path. I added a fault-injection test that enables a retry, throws RpcException from the actual loop, and verifies that the attempt count remains one.
TPC-H: Total hot run time: 16974 ms |
TPC-DS: Total hot run time: 83277 ms |
ClickBench: Total hot run time: 14.73 s |
|
run buildall |
|
/review |
There was a problem hiding this comment.
Review status: complete and converged on head f627bc401812beb7a2d51a8f99a862c176846559 after all three allowed rounds. The final BE, FE/protocol, and risk-focused passes each returned NO_NEW_VALUABLE_FINDINGS. I am requesting changes for three P1 issues and one P2 issue, detailed inline.
Critical checkpoint conclusions:
- Goal and data correctness: the PR aims to make distributed OUTFILE cleanup atomic for live processes. The normal PREPARE/COMMIT/ABORT and marker ordering is substantially covered, but the goal is not yet met: concurrent drains can lose cleanup ownership (MF-1), and deferred Broker rollback can dereference destroyed configuration (MF-4).
- Scope and user focus: I reviewed all 38 changed files plus the relevant writer, filesystem, result-buffer, pipeline teardown, legacy/Nereids coordinator, retry, marker, and mixed-version paths. No additional user focus was supplied.
- Concurrency and error handling: heavy storage work is generally kept outside locks and failures are retained for retry, but FE ABORT can overlap timeout/shutdown cancellation; vector emptiness is not a valid quiescence signal while another drain owns the callbacks. This is the new MF-1 race.
- Lifecycle and memory safety: result buffers intentionally outlive pipeline operators. The retained Broker filesystem borrows address/properties from
ResultFileSinkOperatorX::_file_opts, creating the MF-4 use-after-free path. Cleanup also retains one filesystem wrapper per rotated part (MF-3). No circular ownership issue was otherwise found. - Configuration and compatibility: this adds protocol fields and moves the negotiated BE execution default to v14 rather than adding a new standalone setting. Legacy/new FE-BE routing was traced in both directions. The unconditional remote marker
exists()check changes field-absent/version-13 behavior (MF-2). The remaining Nereids query-wide snapshot mismatch is real but is already covered by existing review threads, so it was not duplicated inline. - Parallel paths: legacy and Nereids coordinators, parallel and non-parallel buffer IDs, Cloud inheritance, Arrow Flight rejection, point-query reachability, partial RPC submission, and client-result deferral were checked. No distinct additional issue remained after duplicate fencing.
- Tests: the added BE/FE tests cover several serial transitions and compatibility branches, but they do not cover the four accepted failures. Needed coverage is a latch-based ABORT/cancel overlap, field-absent/version-13 remote marker compatibility, many rotations with bounded retained filesystem identities, and operator-options destruction before Broker ABORT. This was a review-only run; no builds or tests were executed, and no generated result files were changed.
- Observability: status propagation, deadlines, warnings, and cleanup logging were reviewed; no distinct observability blocker remained. The accepted paths fail correctness before extra metrics would make them safe.
- Persistence, transactions, and writes: no EditLog/master-failover metadata path is changed. This is a live-process compensation protocol; hard crash/restart and cross-query same-destination atomicity are outside the stated scope. Within the stated scope, MF-1 and MF-4 still break rollback guarantees, and MF-2 breaks rolling compatibility.
- Performance: MF-3 is a concrete O(parts) retention regression through the default 300-second result-buffer window; the underlying S3/HDFS clients may be shared, but the per-part filesystem/configuration wrappers are not.
Existing inline discussions were treated as hard duplicate fences throughout; every ledger candidate is accepted, dismissed with evidence, or mapped to an existing thread.
| return Status::OK(); | ||
| } | ||
| if (operation == OutfileOperation::ABORT) { | ||
| if (_outfile_state == OutfileState::ABORTED && _outfile_cleanups.empty()) { |
There was a problem hiding this 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.
|
|
||
| // Never claim an existing marker path; rollback is restricted to token-owned paths below. | ||
| bool exists = true; | ||
| st = file_system->exists(file_name, &exists); |
There was a problem hiding this 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.
| _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); |
There was a problem hiding this 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.
| .sync_file_data = false, | ||
| })); | ||
| io::FSPropertiesRef properties(file_type); | ||
| properties.broker_addresses = &_file_opts->broker_addresses; |
There was a problem hiding this 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.
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
FE Regression Coverage ReportIncrement line coverage |
2 participants
What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
A distributed OUTFILE could leave files from successful receivers behind when another BE or receiver failed. Cleanup ownership could also be lost across partial finalization, late writer close, graceful result-buffer shutdown, multipart abort failure, or a lost success-marker response.
This change adds a version-gated distributed finalization and compensation protocol:
Atomicity scope
The atomicity guarantee applies when the negotiated execution version supports this protocol, the FE/BE processes remain alive during the query lifecycle, and cleanup storage operations succeed within the bounded retries: if any BE/receiver, finalization RPC, or success-marker publication fails, the task fails and all query-owned OUTFILE files are rolled back. The success marker and client result are not published before every receiver acknowledges COMMIT.
The following are intentionally outside this PR:
Release note
None
Check List (For Author)
Test
Behavior changed:
Does this need documentation?
Check List (For Reviewer who merge this PR)