release memory allocated in agg function in vec stream load (#10739)

release memory allocated in agg function in vec stream load

When a load is cancelled, memory allocated by agg functions should
be freeed.

be/src/olap/memtable.cpp

@@ -118,6 +118,19 @@ void MemTable::_init_agg_functions(const vectorized::Block* block) {
 }
 
 MemTable::~MemTable() {
+    if (_vec_skip_list != nullptr && _keys_type != KeysType::DUP_KEYS) {
+        VecTable::Iterator it(_vec_skip_list.get());
+        for (it.SeekToFirst(); it.Valid(); it.Next()) {
+            // We should release agg_places here, because they are not relesed when a
+            // load is canceled.
+            for (size_t i = _schema->num_key_columns(); i < _schema->num_columns(); ++i) {
+                auto function = _agg_functions[i];
+                DCHECK(function != nullptr);
+                DCHECK(it.key()->agg_places(i) != nullptr);
+                function->destroy(it.key()->agg_places(i));
+            }
+        }
+    }
     std::for_each(_row_in_blocks.begin(), _row_in_blocks.end(), std::default_delete<RowInBlock>());
     _mem_tracker->release(_mem_usage);
     _buffer_mem_pool->free_all();
@@ -322,6 +335,10 @@ void MemTable::_collect_vskiplist_results() {
             _output_mutable_block.clear_column_data();
         }
     }
+
+    if (is_final) {
+        _vec_skip_list.reset();
+    }
 }
 
 void MemTable::shrink_memtable_by_agg() {

bin/stop_be.sh

@@ -23,7 +23,7 @@ export DORIS_HOME=`cd "$curdir/.."; pwd`
 export PID_DIR=`cd "$curdir"; pwd`
 
 signum=9
-if [[ $1 = "--grace" ]]; then
+if [ "x"$1 = "x--grace" ]; then
     signum=15
 fi