[bugfix](memtable) arena is freed early and will cause use after free (apache#46997)

### What problem does this PR solve?

Related PR: apache#40912

Problem Summary:

Do not reset _arena in MemTable::to_block(), because it is still used in
~MemTable() when releasing agg places

Fix the following use-after-free

Use:

==3628099==ERROR: AddressSanitizer: heap-use-after-free on address
0x52100381be60 at pc 0x5648f30893f8 bp 0x7f8842433310 sp 0x7f8842433308
READ of size 8 at 0x52100381be60 thread T4767 (wg_flush_broker)
#0 0x5648f30893f7 in
phmap::priv::raw_hash_set<phmap::priv::FlatHashSetPolicy<unsigned long>,
phmap::Hash<unsigned long>, phmap::EqualTo<unsigned long>,
std::allocator<unsigned long>>::destroy_slots()
doris/thirdparty/installed/include/parallel_hashmap/phmap.h:1992:14
#1 0x5648f30936f6 in
phmap::priv::raw_hash_set<phmap::priv::FlatHashSetPolicy<unsigned long>,
phmap::Hash<unsigned long>, phmap::EqualTo<unsigned long>,
std::allocator<unsigned long>>::~raw_hash_set()
doris/thirdparty/installed/include/parallel_hashmap/phmap.h:1236:23
#2 0x5648f3089276 in phmap::flat_hash_set<unsigned long,
phmap::Hash<unsigned long>, phmap::EqualTo<unsigned long>,
std::allocator<unsigned long>>::~flat_hash_set()
doris/thirdparty/installed/include/parallel_hashmap/phmap.h:4577:7
#3 0x5648f308922a in doris::BitmapValue::~BitmapValue()
doris/be/src/util/bitmap_value.h:824:7
#4 0x56490d319fa6 in
doris::vectorized::AggregateFunctionBitmapData<doris::vectorized::AggregateFunctionBitmapUnionOp>::~AggregateFunctionBitmapData()
doris/be/src/vec/aggregate_functions/aggregate_function_bitmap.h:127:8
#5 0x56490d49636a in
doris::vectorized::IAggregateFunctionDataHelper<doris::vectorized::AggregateFunctionBitmapData<doris::vectorized::AggregateFunctionBitmapUnionOp>,
doris::vectorized::AggregateFunctionBitmapOp<doris::vectorized::AggregateFunctionBitmapUnionOp>>::destroy(char*)
const doris/be/src/vec/aggregate_functions/aggregate_function.h:563:92
#6 0x5648f68376e9 in doris::MemTable::~MemTable()
doris/be/src/olap/memtable.cpp:159:27
Free:

0x52100381be60 is located 352 bytes inside of 4096-byte region
[0x52100381bd00,0x52100381cd00)
freed by thread T4767 (wg_flush_broker) here:
#0 0x5648f2f3ee46 in free (doris/output/be/lib/doris_be+0x57418e46)
(BuildId: 298b9c91a1ec8fe0)
#1 0x5648f3080dfc in DefaultMemoryAllocator::free(void*)
doris/be/src/vec/common/allocator.h:108:41
#2 0x5648f3080b3f in Allocator<false, false, false,
DefaultMemoryAllocator>::free(void*, unsigned long)
doris/be/src/vec/common/allocator.h:323:13
#3 0x5648f30b6dee in doris::vectorized::Arena::Chunk::~Chunk()
doris/be/src/vec/common/arena.h:77:31
#4 0x5648f30b6d1f in doris::vectorized::Arena::~Arena()
doris/be/src/vec/common/arena.h:151:16
#5 0x5648f30b695a in
std::default_delete<doris::vectorized::Arena>::operator()(doris::vectorized::Arena*)
const
env/ldb_toolchain/bin/../lib/gcc/x86_64-linux-gnu/13/../../../../include/c++/13/bits/unique_ptr.h:99:2
#6 0x5648f30b67c8 in std::__uniq_ptr_impl<doris::vectorized::Arena,
std::default_delete<doris::vectorized::Arena>>::reset(doris::vectorized::Arena*)
env/ldb_toolchain/bin/../lib/gcc/x86_64-linux-gnu/13/../../../../include/c++/13/bits/unique_ptr.h:211:4
#7 0x5648f30b5d8c in std::unique_ptr<doris::vectorized::Arena,
std::default_delete<doris::vectorized::Arena>>::reset(doris::vectorized::Arena*)
env/ldb_toolchain/bin/../lib/gcc/x86_64-linux-gnu/13/../../../../include/c++/13/bits/unique_ptr.h:509:7
apache#8 0x5648f684253b in
doris::MemTable::_to_block(std::unique_ptr<doris::vectorized::Block,
std::default_delete<doris::vectorized::Block>>*)
doris/be/src/olap/memtable.cpp:522:12
apache#9 0x5648f6842ac5 in
doris::MemTable::to_block(std::unique_ptr<doris::vectorized::Block,
std::default_delete<doris::vectorized::Block>>*)
doris/be/src/olap/memtable.cpp:528:5
apache#10 0x5648f6907a72 in
doris::FlushToken::_do_flush_memtable(doris::MemTable*, int, long*)
doris/be/src/olap/memtable_flush_executor.cpp:144:9
apache#11 0x5648f690932c in
doris::FlushToken::_flush_memtable(std::shared_ptr<doris::MemTable>,
int, long) doris/be/src/olap/memtable_flush_executor.cpp:183:16
apache#12 0x5648f6915d18 in doris::MemtableFlushTask::run()
doris/be/src/olap/memtable_flush_executor.cpp:60:20

Author: yiguolei

be/src/olap/memtable.cpp

@@ -144,35 +144,41 @@ void MemTable::_init_agg_functions(const vectorized::Block* block) {
 
 MemTable::~MemTable() {
     SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_query_thread_context.query_mem_tracker);
+    {
+        SCOPED_CONSUME_MEM_TRACKER(_mem_tracker);
+        g_memtable_cnt << -1;
+        if (_keys_type != KeysType::DUP_KEYS) {
+            for (auto it = _row_in_blocks.begin(); it != _row_in_blocks.end(); it++) {
+                if (!(*it)->has_init_agg()) {
+                    continue;
+                }
+                // We should release agg_places here, because they are not released when a
+                // load is canceled.
+                for (size_t i = _tablet_schema->num_key_columns(); i < _num_columns; ++i) {
+                    auto function = _agg_functions[i];
+                    DCHECK(function != nullptr);
+                    function->destroy((*it)->agg_places(i));
+                }
+            }
+        }
+        std::for_each(_row_in_blocks.begin(), _row_in_blocks.end(),
+                      std::default_delete<RowInBlock>());
+        // Arena has to be destroyed after agg state, because some agg state's memory may be
+        // allocated in arena.
+        _arena.reset();
+        _vec_row_comparator.reset();
+        _row_in_blocks.clear();
+        _agg_functions.clear();
+        _input_mutable_block.clear();
+        _output_mutable_block.clear();
+    }
     if (_is_flush_success) {
         // If the memtable is flush success, then its memtracker's consumption should be 0
         if (_mem_tracker->consumption() != 0 && config::crash_in_memory_tracker_inaccurate) {
             LOG(FATAL) << "memtable flush success but cosumption is not 0, it is "
                        << _mem_tracker->consumption();
         }
     }
-    g_memtable_cnt << -1;
-    if (_keys_type != KeysType::DUP_KEYS) {
-        for (auto it = _row_in_blocks.begin(); it != _row_in_blocks.end(); it++) {
-            if (!(*it)->has_init_agg()) {
-                continue;
-            }
-            // We should release agg_places here, because they are not released when a
-            // load is canceled.
-            for (size_t i = _tablet_schema->num_key_columns(); i < _num_columns; ++i) {
-                auto function = _agg_functions[i];
-                DCHECK(function != nullptr);
-                function->destroy((*it)->agg_places(i));
-            }
-        }
-    }
-    std::for_each(_row_in_blocks.begin(), _row_in_blocks.end(), std::default_delete<RowInBlock>());
-    _arena.reset();
-    _vec_row_comparator.reset();
-    _row_in_blocks.clear();
-    _agg_functions.clear();
-    _input_mutable_block.clear();
-    _output_mutable_block.clear();
 }
 
 int RowInBlockComparator::operator()(const RowInBlock* left, const RowInBlock* right) const {
@@ -629,8 +635,6 @@ Status MemTable::_to_block(std::unique_ptr<vectorized::Block>* res) {
         RETURN_IF_ERROR(_sort_by_cluster_keys());
     }
     _input_mutable_block.clear();
-    // After to block, all data in arena is saved in the block
-    _arena.reset();
     *res = vectorized::Block::create_unique(_output_mutable_block.to_block());
     return Status::OK();
 }