branch-3.1: [fix](memleak) fix memory leak for tabletschema and result cache and fix memleak in arrow input stream

be/src/olap/tablet_schema.cpp

@@ -871,9 +871,7 @@ void TabletIndex::to_schema_pb(TabletIndexPB* index) const {
 
 TabletSchema::TabletSchema() = default;
 
-TabletSchema::~TabletSchema() {
-    clear_column_cache_handlers();
-}
+TabletSchema::~TabletSchema() = default;
 
 int64_t TabletSchema::get_metadata_size() const {
     return sizeof(TabletSchema) + _vl_field_mem_size;
@@ -968,14 +966,6 @@ void TabletSchema::clear_columns() {
     _num_null_columns = 0;
     _num_key_columns = 0;
     _cols.clear();
-    clear_column_cache_handlers();
-}
-
-void TabletSchema::clear_column_cache_handlers() {
-    for (auto* cache_handle : _column_cache_handlers) {
-        TabletColumnObjectPool::instance()->release(cache_handle);
-    }
-    _column_cache_handlers.clear();
 }
 
 void TabletSchema::init_from_pb(const TabletSchemaPB& schema, bool ignore_extracted_columns,
@@ -990,7 +980,6 @@ void TabletSchema::init_from_pb(const TabletSchemaPB& schema, bool ignore_extrac
     _field_name_to_index.clear();
     _field_id_to_index.clear();
     _cluster_key_idxes.clear();
-    clear_column_cache_handlers();
     for (const auto& i : schema.cluster_key_idxes()) {
         _cluster_key_idxes.push_back(i);
     }
@@ -1000,7 +989,10 @@ void TabletSchema::init_from_pb(const TabletSchemaPB& schema, bool ignore_extrac
             auto pair = TabletColumnObjectPool::instance()->insert(
                     deterministic_string_serialize(column_pb));
             column = pair.second;
-            _column_cache_handlers.push_back(pair.first);
+            // Release the handle quickly, because we use shared ptr to manage column.
+            // It often core during tablet schema copy to another schema because handle's
+            // reference count should be managed mannually.
+            TabletColumnObjectPool::instance()->release(pair.first);
         } else {
             column = std::make_shared<TabletColumn>();
             column->init_from_pb(column_pb);
@@ -1089,8 +1081,6 @@ void TabletSchema::shawdow_copy_without_columns(const TabletSchema& tablet_schem
     _num_null_columns = 0;
     _num_key_columns = 0;
     _cols.clear();
-    // notice : do not ref columns
-    _column_cache_handlers.clear();
 }
 
 void TabletSchema::update_index_info_from(const TabletSchema& tablet_schema) {
@@ -1153,7 +1143,6 @@ void TabletSchema::build_current_tablet_schema(int64_t index_id, int32_t version
     _sequence_col_idx = -1;
     _version_col_idx = -1;
     _cluster_key_idxes.clear();
-    clear_column_cache_handlers();
     for (const auto& i : ori_tablet_schema._cluster_key_idxes) {
         _cluster_key_idxes.push_back(i);
     }

be/src/olap/tablet_schema.h

@@ -544,13 +544,10 @@ class TabletSchema : public MetadataAdder<TabletSchema> {
     friend bool operator!=(const TabletSchema& a, const TabletSchema& b);
     TabletSchema(const TabletSchema&) = default;
 
-    void clear_column_cache_handlers();
-
     KeysType _keys_type = DUP_KEYS;
     SortType _sort_type = SortType::LEXICAL;
     size_t _sort_col_num = 0;
     std::vector<TabletColumnPtr> _cols;
-    std::vector<Cache::Handle*> _column_cache_handlers;
 
     std::vector<TabletIndex> _indexes;
     std::unordered_map<StringRef, int32_t, StringRefHash> _field_name_to_index;

be/src/runtime/cache/result_cache.h

@@ -79,7 +79,11 @@ class ResultCache {
         _partition_count = 0;
     }
 
-    virtual ~ResultCache() {}
+    virtual ~ResultCache() {
+        _node_list.clear();
+        _node_map.clear();
+    }
+
     void update(const PUpdateCacheRequest* request, PCacheResponse* response);
     void fetch(const PFetchCacheRequest* request, PFetchCacheResult* result);
     bool contains(const UniqueId& sql_key);

be/src/runtime/exec_env_init.cpp

@@ -722,6 +722,7 @@ void ExecEnv::destroy() {
     // Free resource after threads are stopped.
     // Some threads are still running, like threads created by _new_load_stream_mgr ...
     SAFE_DELETE(_tablet_schema_cache);
+    SAFE_DELETE(_tablet_column_object_pool);
 
     // _scanner_scheduler must be desotried before _storage_page_cache
     SAFE_DELETE(_scanner_scheduler);

be/src/vec/exec/format/arrow/arrow_pip_input_stream.cpp

@@ -32,7 +32,7 @@
 namespace doris::vectorized {
 
 ArrowPipInputStream::ArrowPipInputStream(io::FileReaderSPtr file_reader)
-        : _file_reader(file_reader), _pos(0), _begin(true), _read_buf(new uint8_t[4]) {
+        : _file_reader(file_reader), _pos(0), _begin(true) {
     set_mode(arrow::io::FileMode::READ);
 }
 

be/src/vec/exec/format/arrow/arrow_pip_input_stream.h

@@ -59,7 +59,8 @@ class ArrowPipInputStream : public arrow::io::InputStream {
     io::FileReaderSPtr _file_reader;
     int64_t _pos;
     bool _begin;
-    uint8_t* _read_buf;
+    // The read buf is very small, so use stack memory directly.
+    uint8_t _read_buf[4];
 };
 
 } // namespace vectorized