[fix](outfile) Address atomic cleanup review findings

### 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

Author: Gabriel39

be/src/exec/sink/writer/vfile_result_writer.cpp

@@ -179,7 +179,11 @@ Status VFileResultWriter::_create_file_writer(const std::string& file_name) {
     // 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};
+    // Local OUTFILE historically synced successful closes; preserve that durability while remote
+    // writers keep the existing no-sync option to avoid redundant flushes.
+    const io::FileWriterOptions options {
+            .write_file_cache = false,
+            .sync_file_data = _storage_type == TStorageBackendType::LOCAL};
     RETURN_IF_ERROR(_file_system->create_file(file_name, &_file_writer_impl, &options));
     switch (_file_opts->file_format) {
     case TFileFormatType::FORMAT_CSV_PLAIN:

be/src/exec/sink/writer/vfile_result_writer.h

@@ -80,6 +80,7 @@ class VFileResultWriter final : public AsyncResultWriter {
 private:
     FRIEND_TEST(VFileResultWriterTest, FailedCloseRemovesClosedOutputFile);
     FRIEND_TEST(VFileResultWriterTest, FailedCloseRemovesOnlyOwnedOutputFiles);
+    FRIEND_TEST(VFileResultWriterTest, LocalOutfilePreservesSynchronousClose);
 
     Status _write_file(const Block& block);
 

be/src/io/fs/s3_file_writer.cpp

@@ -538,10 +538,12 @@ Status S3FileWriter::_complete() {
         return {resp.status.code, std::move(resp.status.msg)};
     }
 
+    // CompleteMultipartUpload publishes the object and consumes the upload ID. A later HEAD
+    // failure must be cleaned up as an object, not retried as an already-finished multipart upload.
+    _multipart_upload_completed = true;
     RETURN_IF_ERROR(check_after_upload(client.get(), resp, _obj_storage_path_opts, _bytes_appended,
                                        "complete_multipart"));
 
-    _multipart_upload_completed = true;
     s3_file_created_total << 1;
     return Status::OK();
 }

be/src/runtime/result_block_buffer.cpp

@@ -145,11 +145,15 @@ Status ResultBlockBuffer<ResultCtxType>::add_outfile_cleanup(OutfileCleanup clea
         return Status::OK();
     }
     if (run_cleanup) {
-        Status status = cleanup();
-        if (!status.ok()) {
-            std::lock_guard<std::mutex> l(_lock);
-            _outfile_cleanups.emplace_back(std::move(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();

be/src/runtime/result_buffer_mgr.cpp

@@ -63,7 +63,9 @@ void ResultBufferMgr::stop() {
     }
     std::vector<TUniqueId> remaining_ids;
     {
-        std::shared_lock<std::shared_mutex> rlock(_buffer_map_lock);
+        std::unique_lock<std::shared_mutex> wlock(_buffer_map_lock);
+        // Closing the registration gate under the map lock keeps the shutdown snapshot complete.
+        _stopping = true;
         remaining_ids.reserve(_buffer_map.size());
         for (const auto& item : _buffer_map) {
             remaining_ids.emplace_back(item.first);
@@ -89,6 +91,9 @@ Status ResultBufferMgr::create_sender(const TUniqueId& unique_id, int buffer_siz
                                       std::shared_ptr<arrow::Schema> schema) {
     {
         std::shared_lock<std::shared_mutex> rlock(_buffer_map_lock);
+        if (_stopping) {
+            return Status::Cancelled("ResultBufferMgr is stopping");
+        }
         auto iter = _buffer_map.find(unique_id);
 
         if (_buffer_map.end() != iter) {
@@ -108,6 +113,9 @@ Status ResultBufferMgr::create_sender(const TUniqueId& unique_id, int buffer_siz
 
     {
         std::unique_lock<std::shared_mutex> wlock(_buffer_map_lock);
+        if (_stopping) {
+            return Status::Cancelled("ResultBufferMgr is stopping");
+        }
         _buffer_map.insert(std::make_pair(unique_id, control_block));
         // ResultBlockBufferBase should destroy after max_timeout
         // for exceed max_timeout FE will return timeout to client

be/src/runtime/result_buffer_mgr.h

@@ -90,6 +90,7 @@ class ResultBufferMgr {
     std::shared_mutex _buffer_map_lock;
     // buffer block map
     BufferMap _buffer_map;
+    bool _stopping = false;
 
     // lock for timeout map
     std::mutex _timeout_lock;

be/src/service/internal_service.cpp

@@ -167,7 +167,10 @@ void cleanup_expired_outfile_marker_states(std::chrono::steady_clock::time_point
         }
     }
     for (auto it = outfile_marker_states.begin(); it != outfile_marker_states.end();) {
-        if (now - it->second.updated_at >= OUTFILE_MARKER_TOMBSTONE_TTL) {
+        // 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() &&
+            now - it->second.updated_at >= OUTFILE_MARKER_TOMBSTONE_TTL) {
             it = outfile_marker_states.erase(it);
         } else {
             ++it;

be/test/exec/sink/writer/vfile_result_writer_test.cpp

@@ -21,10 +21,13 @@
 
 #include <filesystem>
 
+#include "exec/operator/result_sink_operator.h"
 #include "format/transformer/vorc_transformer.h"
 #include "format/transformer/vparquet_transformer.h"
 #include "io/fs/file_writer.h"
 #include "io/fs/local_file_system.h"
+#include "io/fs/local_file_writer.h"
+#include "runtime/runtime_state.h"
 #include "util/slice.h"
 
 namespace doris {
@@ -131,4 +134,34 @@ TEST(VFileResultWriterTest, AbortedFormatStreamsDoNotCloseRawWriter) {
     EXPECT_EQ(orc_writer.close_count, 0);
 }
 
+TEST(VFileResultWriterTest, LocalOutfilePreservesSynchronousClose) {
+    const auto directory =
+            std::filesystem::temp_directory_path() / "doris_vfile_result_writer_sync_close";
+    const auto path = directory / "part.csv";
+    std::filesystem::remove_all(directory);
+    std::filesystem::create_directories(directory);
+
+    TResultFileSinkOptions thrift_options;
+    thrift_options.file_path = directory.string() + "/";
+    thrift_options.file_format = TFileFormatType::FORMAT_CSV_PLAIN;
+    thrift_options.file_suffix = "csv";
+    thrift_options.with_bom = false;
+    ResultFileOptions file_options(thrift_options);
+    RuntimeState state;
+    VExprContextSPtrs output_exprs;
+    VFileResultWriter writer(TDataSink {}, output_exprs, nullptr, nullptr);
+    writer._state = &state;
+    writer._file_opts = &file_options;
+    writer._storage_type = TStorageBackendType::LOCAL;
+
+    ASSERT_TRUE(writer._create_file_writer(path.string()).ok());
+    const auto* local_writer =
+            dynamic_cast<const io::LocalFileWriter*>(writer._file_writer_impl.get());
+    ASSERT_NE(local_writer, nullptr);
+    EXPECT_TRUE(local_writer->_sync_data);
+
+    ASSERT_FALSE(writer.close(Status::Cancelled("test cleanup")).ok());
+    std::filesystem::remove_all(directory);
+}
+
 } // namespace doris

be/test/io/fs/s3_file_writer_test.cpp

@@ -1208,6 +1208,9 @@ class SimpleMockObjStorageClient : public io::ObjStorageClientTestStub {
 
     ObjStorageHeadResult head_object(const ObjStoragePath& opts) override {
         std::lock_guard lock(_mutex);
+        if (on_head) {
+            on_head();
+        }
         return {.resp = ObjStorageResponse::OK(),
                 .file_size = static_cast<int64_t>(objects[opts.path.native()].size())};
     }
@@ -1272,6 +1275,7 @@ class SimpleMockObjStorageClient : public io::ObjStorageClientTestStub {
     int complete_multipart_count = 0;
     int abort_multipart_count = 0;
     int abort_failures_remaining = 0;
+    std::function<void()> on_head;
     std::vector<std::string> abort_upload_ids;
 
     // Structures to store input parameters for each call
@@ -1314,6 +1318,7 @@ class SimpleMockObjStorageClient : public io::ObjStorageClientTestStub {
         complete_multipart_count = 0;
         abort_multipart_count = 0;
         abort_failures_remaining = 0;
+        on_head = nullptr;
         abort_upload_ids.clear();
 
         create_multipart_params.clear();
@@ -1382,6 +1387,25 @@ TEST_F(S3FileWriterTest, AbortRetryReusesMultipartUploadId) {
     EXPECT_EQ(mock_client->complete_multipart_count, 0);
 }
 
+TEST_F(S3FileWriterTest, MarksMultipartCompletedBeforePostUploadCheck) {
+    auto [mock_client, writer] = create_s3_client("complete-before-post-upload-check");
+    std::string content(config::s3_write_buffer_size, 'a');
+    const bool original_check_after_upload = config::enable_s3_object_check_after_upload;
+    Defer restore_config {
+            [&] { config::enable_s3_object_check_after_upload = original_check_after_upload; }};
+    config::enable_s3_object_check_after_upload = true;
+    bool completion_flag_during_head = false;
+    mock_client->on_head = [&] {
+        completion_flag_during_head = writer->_multipart_upload_completed;
+    };
+
+    ASSERT_TRUE(writer->append(content).ok());
+    ASSERT_TRUE(writer->close().ok());
+
+    EXPECT_EQ(mock_client->complete_multipart_count, 1);
+    EXPECT_TRUE(completion_flag_during_head);
+}
+
 /**
  * Generate test data for S3FileWriter boundary tests.
  * Returns a vector of sizes that we'll use to generate data on demand.

be/test/runtime/result_buffer_mgr_test.cpp

@@ -115,6 +115,41 @@ TEST_F(ResultBufferMgrTest, cancel_no_block) {
     EXPECT_FALSE(buffer_mgr.cancel(query_id, Status::InternalError("")));
 }
 
+TEST_F(ResultBufferMgrTest, RejectsNewSenderAfterStop) {
+    ResultBufferMgr buffer_mgr;
+    buffer_mgr.stop();
+
+    TUniqueId query_id;
+    query_id.lo = 10;
+    query_id.hi = 100;
+    std::shared_ptr<ResultBlockBufferBase> control_block;
+
+    EXPECT_FALSE(buffer_mgr.create_sender(query_id, 1024, &control_block, &_state, false).ok());
+    EXPECT_EQ(control_block, nullptr);
+}
+
+TEST_F(ResultBufferMgrTest, LateOutfileCleanupRetriesAfterCancellation) {
+    ResultBufferMgr buffer_mgr;
+    TUniqueId query_id;
+    query_id.lo = 11;
+    query_id.hi = 101;
+
+    std::shared_ptr<ResultBlockBufferBase> control_block;
+    ASSERT_TRUE(buffer_mgr.create_sender(query_id, 1024, &control_block, &_state, false).ok());
+    ASSERT_TRUE(buffer_mgr.cancel(query_id, Status::Cancelled("injected cancellation")));
+
+    int cleanup_attempts = 0;
+    EXPECT_TRUE(control_block
+                        ->add_outfile_cleanup([&] {
+                            ++cleanup_attempts;
+                            return cleanup_attempts == 1
+                                           ? Status::IOError("injected transient cleanup failure")
+                                           : Status::OK();
+                        })
+                        .ok());
+    EXPECT_EQ(cleanup_attempts, 2);
+}
+
 TEST_F(ResultBufferMgrTest, OutfileAbortCleansRegisteredAndLateFiles) {
     ResultBufferMgr buffer_mgr;
     TUniqueId query_id;