fix: reset merge_function_wrapper in SortMergeReader::Close to prevent use-after-free

Move the merge_function_wrapper_->Reset() into SortMergeReader::Close()
(both LoserTree and MinHeap variants) so that cached KeyValue data is
released before underlying readers are destroyed. This prevents Arrow
Buffer objects from calling Free() on a dangling pool pointer when
SpillReader is later destructed.

Also reorder the cleanup in MergeTreeCompactRewriter::RewriteCompaction
to call reader->Close() before merge_file_split_read_.reset(), ensuring
the Reset() triggered by Close() runs while all resource providers are
still alive.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: zjw1111

src/paimon/core/mergetree/compact/merge_tree_compact_rewriter.cpp

@@ -263,10 +263,10 @@ Result<CompactResult> MergeTreeCompactRewriter::RewriteCompaction(
 
     ScopeGuard write_guard([&]() -> void {
         rolling_writer->Abort();
-        merge_file_split_read_.reset();
         for (const auto& reader : reader_holders) {
             reader->Close();
         }
+        merge_file_split_read_.reset();
     });
 
     for (const auto& section : sections) {

src/paimon/core/mergetree/compact/sort_merge_reader_with_loser_tree.h

@@ -51,6 +51,9 @@ class SortMergeReaderWithLoserTree : public SortMergeReader {
 
     void Close() override {
         loser_tree_->Close();
+        if (merge_function_wrapper_) {
+            merge_function_wrapper_->Reset();
+        }
     }
 
     class Iterator : public SortMergeReader::Iterator {

src/paimon/core/mergetree/compact/sort_merge_reader_with_min_heap.h

@@ -70,6 +70,9 @@ class SortMergeReaderWithMinHeap : public SortMergeReader {
         for (const auto& reader : readers_holder_) {
             reader->Close();
         }
+        if (merge_function_wrapper_) {
+            merge_function_wrapper_->Reset();
+        }
     }
 
     std::shared_ptr<Metrics> GetReaderMetrics() const override {

src/paimon/core/mergetree/merge_tree_writer.cpp

@@ -273,7 +273,6 @@ Status MergeTreeWriter::FlushWriteBuffer(bool wait_for_latest_compaction,
         ScopeGuard write_guard([&]() -> void {
             rolling_writer->Abort();
             async_key_value_producer_consumer->Close();
-            merge_function_wrapper_->Reset();
         });
         while (true) {
             PAIMON_ASSIGN_OR_RAISE(KeyValueBatch key_value_batch,