[fix](Parquet) add a memory tracker to parquet meta

be/src/runtime/exec_env.h

@@ -217,6 +217,7 @@ class ExecEnv {
         return _subcolumns_tree_tracker;
     }
     std::shared_ptr<MemTrackerLimiter> s3_file_buffer_tracker() { return _s3_file_buffer_tracker; }
+    std::shared_ptr<MemTrackerLimiter> parquet_meta_tracker() { return _parquet_meta_tracker; }
 
     ThreadPool* send_batch_thread_pool() { return _send_batch_thread_pool.get(); }
     ThreadPool* buffered_reader_prefetch_thread_pool() {
@@ -413,6 +414,9 @@ class ExecEnv {
     std::shared_ptr<MemTrackerLimiter> _subcolumns_tree_tracker;
     std::shared_ptr<MemTrackerLimiter> _s3_file_buffer_tracker;
 
+    // Tracking memory consumption of parquet meta
+    std::shared_ptr<MemTrackerLimiter> _parquet_meta_tracker;
+
     std::unique_ptr<ThreadPool> _send_batch_thread_pool;
     // Threadpool used to prefetch remote file for buffered reader
     std::unique_ptr<ThreadPool> _buffered_reader_prefetch_thread_pool;

be/src/runtime/exec_env_init.cpp

@@ -614,6 +614,8 @@ void ExecEnv::init_mem_tracker() {
             MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::GLOBAL, "S3FileBuffer");
     _stream_load_pipe_tracker =
             MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::LOAD, "StreamLoadPipe");
+    _parquet_meta_tracker =
+            MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::METADATA, "ParquetMeta");
 }
 
 Status ExecEnv::_check_deploy_mode() {

be/src/vec/exec/format/parquet/parquet_thrift_util.h

@@ -75,7 +75,7 @@ static Status parse_thrift_footer(io::FileReaderSPtr file, FileMetaData** file_m
     tparquet::FileMetaData t_metadata;
     // deserialize footer
     RETURN_IF_ERROR(deserialize_thrift_msg(meta_ptr, &metadata_size, true, &t_metadata));
-    *file_metadata = new FileMetaData(t_metadata);
+    *file_metadata = new FileMetaData(t_metadata, metadata_size);
     RETURN_IF_ERROR((*file_metadata)->init_schema());
     *meta_size = PARQUET_FOOTER_SIZE + metadata_size;
     return Status::OK();

be/src/vec/exec/format/parquet/vparquet_file_metadata.cpp

@@ -22,11 +22,21 @@
 #include <sstream>
 #include <vector>
 
+#include "runtime/exec_env.h"
+#include "runtime/memory/mem_tracker_limiter.h"
 #include "schema_desc.h"
 
 namespace doris::vectorized {
 #include "common/compile_check_begin.h"
-FileMetaData::FileMetaData(tparquet::FileMetaData& metadata) : _metadata(metadata) {}
+
+FileMetaData::FileMetaData(tparquet::FileMetaData& metadata, size_t mem_size)
+        : _metadata(metadata), _mem_size(mem_size) {
+    ExecEnv::GetInstance()->parquet_meta_tracker()->consume(mem_size);
+}
+
+FileMetaData::~FileMetaData() {
+    ExecEnv::GetInstance()->parquet_meta_tracker()->release(_mem_size);
+}
 
 Status FileMetaData::init_schema() {
     if (_metadata.schema[0].num_children <= 0) {

be/src/vec/exec/format/parquet/vparquet_file_metadata.h

@@ -27,19 +27,21 @@ namespace doris::vectorized {
 #include "common/compile_check_begin.h"
 class FileMetaData {
 public:
-    FileMetaData(tparquet::FileMetaData& metadata);
-    ~FileMetaData() = default;
+    FileMetaData(tparquet::FileMetaData& metadata, size_t mem_size);
+    ~FileMetaData();
     Status init_schema();
     const FieldDescriptor& schema() const { return _schema; }
     const tparquet::FileMetaData& to_thrift();
     void iceberg_sanitize(const std::vector<std::string>& read_columns) {
         _schema.iceberg_sanitize(read_columns);
     }
     std::string debug_string() const;
+    size_t get_mem_size() const { return _mem_size; }
 
 private:
     tparquet::FileMetaData _metadata;
     FieldDescriptor _schema;
+    size_t _mem_size;
 };
 #include "common/compile_check_end.h"