@@ -22,6 +22,7 @@
 #include "exec/operator/iceberg_table_sink_operator.h"
 #include "exec/sink/writer/iceberg/viceberg_sort_writer.h"
 #include "exec/sink/writer/iceberg/viceberg_table_writer.h"
+#include "runtime/runtime_profile_counter_names.h"
 
 namespace doris {
 
@@ -196,6 +197,11 @@ void SpillIcebergTableSinkLocalState::_init_spill_counters() {
     ADD_COUNTER_WITH_LEVEL(profile, "SpillWriteFileBytes", TUnit::BYTES, 1);
     ADD_COUNTER_WITH_LEVEL(profile, "SpillWriteRows", TUnit::UNIT, 1);
     ADD_COUNTER_WITH_LEVEL(profile, "SpillWriteFileTotalCount", TUnit::UNIT, 1);
+    ADD_COUNTER_WITH_LEVEL(profile, profile::SPILL_REMOTE_WRITE_REQUESTS, TUnit::UNIT, 1);
+    ADD_COUNTER_WITH_LEVEL(profile, profile::SPILL_REMOTE_UPLOAD_PART_REQUESTS, TUnit::UNIT, 1);
+    ADD_COUNTER_WITH_LEVEL(profile, profile::SPILL_REMOTE_UPLOAD_BYTES, TUnit::BYTES, 1);
+    ADD_TIMER_WITH_LEVEL(profile, profile::SPILL_REMOTE_UPLOAD_TIME, 1);
+    ADD_TIMER_WITH_LEVEL(profile, profile::SPILL_REMOTE_UPLOAD_WAIT_TIME, 1);
 
     //seems init_spill_read_counters()
     ADD_TIMER_WITH_LEVEL(profile, "SpillTotalTime", 1);
@@ -210,6 +216,7 @@ void SpillIcebergTableSinkLocalState::_init_spill_counters() {
     ADD_COUNTER_WITH_LEVEL(profile, "SpillReadFileBytes", TUnit::BYTES, 1);
     ADD_COUNTER_WITH_LEVEL(profile, "SpillReadRows", TUnit::UNIT, 1);
     ADD_COUNTER_WITH_LEVEL(profile, "SpillReadFileCount", TUnit::UNIT, 1);
+    ADD_COUNTER_WITH_LEVEL(profile, profile::SPILL_REMOTE_READ_REQUESTS, TUnit::UNIT, 1);
     ADD_COUNTER_WITH_LEVEL(profile, "SpillWriteFileCurrentBytes", TUnit::BYTES, 1);
 }
 
 
@@ -274,6 +274,9 @@ Status VIcebergSortWriter::_do_spill() {
         RETURN_IF_ERROR(writer->write_block(_runtime_state, block));
         block.clear_column_data();
     }
+    // On object storage close() uploads the tail of the file; a failure must fail the spill
+    // instead of being logged by the destructor.
+    RETURN_IF_ERROR(writer->close());
     // Reset the sorter to free memory and accept new data
     _sorter->reset();
     return Status::OK();
@@ -337,6 +340,8 @@ Status VIcebergSortWriter::_do_intermediate_merge() {
         RETURN_IF_ERROR(_merger->get_next(&merge_sorted_block, &eos));
         RETURN_IF_ERROR(tmp_spill_writer->write_block(_runtime_state, merge_sorted_block));
     }
+    // The merged output must be complete before its inputs are deleted.
+    RETURN_IF_ERROR(tmp_spill_writer->close());
 
     // Clean up the files that were consumed during this intermediate merge
     for (auto& file : _current_merging_spill_files) {
 
@@ -43,6 +43,12 @@ SpillFile::~SpillFile() {
 }
 
 void SpillFile::gc() {
+    // The writer may outlive this file (e.g. when an error unwinds the file's owner first).
+    // Drop its unfinished part now: closing it later would charge a footer to no owner and
+    // could publish an object after the prefix below was deleted.
+    if (_active_writer != nullptr) {
+        _active_writer->_discard(this);
+    }
     const int64_t written_bytes = std::exchange(_total_written_bytes, 0);
     if (!_dir_created) {
         _data_dir->release(written_bytes);
 
@@ -114,7 +114,8 @@ class SpillFile : public std::enable_shared_from_this<SpillFile> {
     bool _dir_created = false;
     bool _ready_for_reading = false;
     // Pointer to the currently-active writer. Mutable to allow checks from const
-    // methods like create_reader(). Only one writer may be active at a time.
+    // methods like create_reader(). Only one writer may be active at a time. Cleared when
+    // the writer finished, was discarded by gc(), or was destroyed.
     mutable SpillFileWriter* _active_writer = nullptr;
 };
 using SpillFileSPtr = std::shared_ptr<SpillFile>;
 
@@ -81,7 +81,8 @@ void SpillFileReader::_record_read(size_t bytes_read) {
     COUNTER_UPDATE(_read_file_size, bytes_read);
     ExecEnv::GetInstance()->spill_file_mgr()->update_spill_read_bytes(bytes_read);
     if (_is_remote) {
-        // One read_at() is exactly one GET request on object storage.
+        // One successful read_at() is one logical GET. Retries inside the object storage client
+        // and reads that finally failed are not counted (see RemoteWriteStats).
         if (_remote_read_requests != nullptr) {
             COUNTER_UPDATE(_remote_read_requests, 1);
         }
 
@@ -75,14 +75,36 @@ SpillFileWriter::SpillFileWriter(const std::shared_ptr<SpillFile>& spill_file, R
 }
 
 SpillFileWriter::~SpillFileWriter() {
-    if (_closed) {
-        return;
+    if (!_closed) {
+        Status st = close();
+        if (!st.ok()) {
+            LOG(WARNING) << "SpillFileWriter::~SpillFileWriter() failed: " << st.to_string()
+                         << ", spill_dir=" << _spill_dir;
+        }
+    }
+    // A failed close() leaves this writer registered; the SpillFile must not reach it anymore.
+    if (auto spill_file = _spill_file_wptr.lock();
+        spill_file && spill_file->_active_writer == this) {
+        spill_file->_active_writer = nullptr;
     }
-    Status st = close();
-    if (!st.ok()) {
-        LOG(WARNING) << "SpillFileWriter::~SpillFileWriter() failed: " << st.to_string()
-                     << ", spill_dir=" << _spill_dir;
+}
+
+void SpillFileWriter::_discard(SpillFile* spill_file) {
+    DCHECK_EQ(spill_file->_active_writer, this);
+    spill_file->_active_writer = nullptr;
+    _closed = true;
+    if (!_file_writer) {
+        return;
     }
+    // The part is dropped without a footer and without close(): destroying an unclosed
+    // writer waits for its in-flight uploads and publishes nothing, so the ledger below is
+    // complete afterwards and no object can appear after the SpillFile deleted its prefix.
+    std::unique_ptr<io::FileWriter> writer = std::move(_file_writer);
+    MultipartUploadId upload = _multipart_upload_id(writer.get());
+    writer.reset();
+    _reconcile_part();
+    _abort_multipart_upload(upload);
+    _advance_part();
 }
 
 Status SpillFileWriter::_open_next_part(const std::shared_ptr<SpillFile>& spill_file) {
@@ -192,6 +214,20 @@ Status SpillFileWriter::_close_current_part(const std::shared_ptr<SpillFile>& sp
         writer.reset();
     }
 
+    _reconcile_part();
+
+    if (!status.ok()) {
+        LOG(WARNING) << "failed to close spill part " << _current_part_path << ": " << status;
+        _abort_multipart_upload(upload);
+    } else if (spill_file) {
+        spill_file->add_part(_part_written_bytes);
+    }
+
+    _advance_part();
+    return status;
+}
+
+void SpillFileWriter::_reconcile_part() {
     // Budget: everything the gate took for this part but the upload callback never gave
     // back (buffers that failed before their upload started) is released here. The writer is
     // in its final state at this point, so every callback that will ever fire has fired.
@@ -225,24 +261,16 @@ Status SpillFileWriter::_close_current_part(const std::shared_ptr<SpillFile>& sp
         ExecEnv::GetInstance()->spill_file_mgr()->update_spill_remote_write(uploaded_bytes,
                                                                             data_requests);
     }
+}
 
-    if (!status.ok()) {
-        LOG(WARNING) << "failed to close spill part " << _current_part_path << ": " << status;
-        _abort_multipart_upload(upload);
-    } else if (spill_file) {
-        spill_file->add_part(_part_written_bytes);
-    }
-
-    // Advance to next part
+void SpillFileWriter::_advance_part() {
     ++_current_part_index;
     _part_written_blocks = 0;
     _part_written_bytes = 0;
     _part_max_sub_block_size = 0;
     _part_meta.clear();
     _part_ledger.reset();
     _part_stats.reset();
-
-    return status;
 }
 
 SpillFileWriter::MultipartUploadId SpillFileWriter::_multipart_upload_id(io::FileWriter* writer) {
 
@@ -44,9 +44,8 @@ class SpillRemoteUploadBudget;
 ///
 /// Part rotation is fully internal. Each part file has its own footer with
 /// block offset metadata. Parts are named 0, 1, 2, ... within the SpillFile's
-/// directory. Parts are closed non-blocking (FileWriter::close(true)) so that the
-/// upload of a finished part overlaps with writing the next one; close() waits for
-/// all of them.
+/// directory. A part is closed synchronously when it is rotated out or when close() is called.
+/// If the SpillFile is destroyed before the writer is closed, the unfinished part is discarded.
 ///
 /// Files are created on the SpillDataDir's file system, which is either a local disk or
 /// the object storage of a cloud storage vault. For object storage the writer additionally
@@ -69,6 +68,8 @@ class SpillFileWriter {
     Status close();
 
 private:
+    friend class SpillFile;
+
     /// Remote only: budget bytes taken and given back for one part. Updated from the
     /// appending thread (gate) and the upload threads (done callback); shared by value with
     /// the FileWriterOptions lambdas so that it outlives the writer.
@@ -88,6 +89,18 @@ class SpillFileWriter {
     /// or not.
     Status _close_current_part(const std::shared_ptr<SpillFile>& spill_file);
 
+    /// Budget and statistics of the current part, once its writer reached a final state.
+    void _reconcile_part();
+
+    /// Reset the per-part state and move on to the next part index.
+    void _advance_part();
+
+    /// Called by SpillFile::gc() when the SpillFile goes away before this writer was closed
+    /// (e.g. an error unwound the owner of the file first). Drops the current part without
+    /// a footer and without publishing it, and unregisters from the SpillFile, so that nothing
+    /// is charged to or written under a SpillFile that no longer exists.
+    void _discard(SpillFile* spill_file);
+
     /// If current part size >= _max_part_size, close it.
     Status _rotate_if_needed(const std::shared_ptr<SpillFile>& spill_file);
 
 
@@ -36,7 +36,10 @@ struct FileCacheAllocatorBuilder;
 struct EncryptionInfo;
 
 // Request statistics reported by remote file writers when the caller passes an instance
-// through FileWriterOptions::remote_write_stats. All fields are cumulative.
+// through FileWriterOptions::remote_write_stats. All fields are cumulative. They count logical
+// requests, one per call into the object storage client: attempts retried inside the client
+// (throttling, transient errors) are not seen by the writer and not counted, so these are
+// observability numbers, not the provider's billed request count.
 struct RemoteWriteStats {
     std::atomic<int64_t> put_object_requests {0};
     std::atomic<int64_t> create_multipart_requests {0};
 
@@ -100,6 +100,8 @@ inline constexpr char SPILL_WRITE_BLOCK_COUNT[] = "SpillWriteBlockCount";
 inline constexpr char SPILL_WRITE_BLOCK_BYTES[] = "SpillWriteBlockBytes";
 inline constexpr char SPILL_WRITE_ROWS[] = "SpillWriteRows";
 // Remote (object storage) spill write counters. Zero when spill is written to local disks.
+// Request counters count logical requests; retries inside the object storage client are not
+// included.
 inline constexpr char SPILL_REMOTE_WRITE_REQUESTS[] = "SpillRemoteWriteRequests";
 inline constexpr char SPILL_REMOTE_UPLOAD_PART_REQUESTS[] = "SpillRemoteUploadPartRequests";
 inline constexpr char SPILL_REMOTE_UPLOAD_BYTES[] = "SpillRemoteUploadBytes";
 
@@ -1479,6 +1479,71 @@ TEST_F(SpillFileS3Test, CancelledQueryCloseAbortsOpenMultipart) {
     ASSERT_EQ(_data_dir->get_spill_data_bytes(), 0);
 }
 
+// A writer can outlive its SpillFile when an error unwinds the owner of the file first (the
+// repartitioner keeps its writers while the output files live in the caller). Destroying the
+// file discards the unfinished part: the multipart upload is aborted, no footer is charged,
+// and destroying the writer afterwards publishes nothing under the deleted prefix.
+TEST_F(SpillFileS3Test, SpillFileDestroyedBeforeWriterDiscardsPart) {
+    config::spill_file_part_size_bytes = 1024 * 1024;
+    _create_manager();
+    auto* budget = _manager->remote_upload_budget();
+    std::mt19937 rng(47);
+    SpillFileSPtr spill_file;
+    ASSERT_TRUE(_manager->create_spill_file("query_16/join-1-0-1", spill_file).ok());
+    SpillFileWriterSPtr writer;
+    ASSERT_TRUE(spill_file->create_writer(_runtime_state.get(), _profile.get(), writer).ok());
+    for (int i = 0; i < 4; ++i) {
+        ASSERT_TRUE(writer->write_block(_runtime_state.get(), _random_string_block(rng, 100, 100))
+                            .ok());
+    }
+    ASSERT_EQ(mock_store().create_multipart_requests, 1);
+    ASSERT_GT(_data_dir->get_spill_data_bytes(), 0);
+
+    spill_file.reset();
+    ASSERT_EQ(mock_store().abort_multipart_requests, 1);
+    ASSERT_EQ(_data_dir->get_spill_data_bytes(), 0);
+    ASSERT_EQ(budget->inflight_bytes(), 0);
+    ASSERT_EQ(budget->total_acquired_bytes(), budget->total_released_bytes());
+
+    const int64_t puts_before = mock_store().put_requests;
+    ASSERT_TRUE(writer->close().ok());
+    writer.reset();
+    ASSERT_EQ(mock_store().put_requests, puts_before);
+    ASSERT_TRUE(mock_store().keys_with_prefix(kBucket, spill_root() + "/query_16/").empty());
+    ASSERT_EQ(_data_dir->get_spill_data_bytes(), 0);
+}
+
+// A writer whose close() failed stays registered with its SpillFile; destroying the file
+// afterwards must neither touch the writer's (already reconciled) state nor leak capacity, and
+// the writer destroyed first must not leave a dangling registration behind.
+TEST_F(SpillFileS3Test, FailedCloseThenSpillFileDestroyed) {
+    config::spill_file_part_size_bytes = 1024 * 1024;
+    _create_manager();
+    std::mt19937 rng(53);
+    for (bool writer_first : {false, true}) {
+        SpillFileSPtr spill_file;
+        ASSERT_TRUE(_manager->create_spill_file(fmt::format("query_17/sort-1-0-{}", writer_first),
+                                                spill_file)
+                            .ok());
+        SpillFileWriterSPtr writer;
+        ASSERT_TRUE(spill_file->create_writer(_runtime_state.get(), _profile.get(), writer).ok());
+        ASSERT_TRUE(writer->write_block(_runtime_state.get(), _random_string_block(rng, 100, 100))
+                            .ok());
+        mock_store().fail_uploads = true;
+        ASSERT_FALSE(writer->close().ok());
+        mock_store().fail_uploads = false;
+        ASSERT_FALSE(spill_file->ready_for_reading());
+        if (writer_first) {
+            writer.reset();
+            spill_file.reset();
+        } else {
+            spill_file.reset();
+            writer.reset();
+        }
+        ASSERT_EQ(_data_dir->get_spill_data_bytes(), 0);
+    }
+}
+
 // The scenario behind the fast-path cancellation check: a query cancelled before close() must
 // not start a new upload for its last buffer; the part is refused, nothing is put, the budget
 // stays balanced.