[fix](insert) fix memory leak for insert transaction

be/src/exec/plain_binary_line_reader.cpp

@@ -31,13 +31,14 @@ PlainBinaryLineReader::~PlainBinaryLineReader() {
 void PlainBinaryLineReader::close() {}
 
 Status PlainBinaryLineReader::read_line(const uint8_t** ptr, size_t* size, bool* eof) {
-    std::unique_ptr<uint8_t[]> file_buf;
     int64_t read_size = 0;
-    RETURN_IF_ERROR(_file_reader->read_one_message(&file_buf, &read_size));
-    *ptr = file_buf.release();
+    RETURN_IF_ERROR(_file_reader->read_one_message(&_file_buf, &read_size));
+    *ptr = _file_buf.get();
     *size = read_size;
     if (read_size == 0) {
         *eof = true;
+    } else {
+        _cur_row.reset(*reinterpret_cast<PDataRow**>(_file_buf.get()));
     }
     return Status::OK();
 }

be/src/exec/plain_binary_line_reader.h

@@ -17,6 +17,8 @@
 
 #pragma once
 
+#include <gen_cpp/internal_service.pb.h>
+
 #include "exec/line_reader.h"
 
 namespace doris {
@@ -35,6 +37,8 @@ class PlainBinaryLineReader : public LineReader {
 
 private:
     FileReader* _file_reader;
+    std::unique_ptr<uint8_t[]> _file_buf;
+    std::unique_ptr<PDataRow> _cur_row;
 };
 
 } // namespace doris

be/src/runtime/stream_load/stream_load_pipe.h

@@ -60,6 +60,16 @@ class StreamLoadPipe : public MessageBodySink, public FileReader {
         return _append(buf, proto_byte_size);
     }
 
+    Status append(std::unique_ptr<PDataRow>&& row) {
+        PDataRow* row_ptr = row.get();
+        {
+            std::unique_lock<std::mutex> l(_lock);
+            _data_row_ptrs.emplace_back(std::move(row));
+        }
+        return append_and_flush(reinterpret_cast<char*>(&row_ptr), sizeof(row_ptr),
+                                sizeof(PDataRow*) + row_ptr->ByteSizeLong());
+    }
+
     Status append(const char* data, size_t size) override {
         size_t pos = 0;
         if (_write_buf != nullptr) {
@@ -219,8 +229,11 @@ class StreamLoadPipe : public MessageBodySink, public FileReader {
         _buf_queue.pop_front();
         _buffered_bytes -= buf->limit;
         if (_use_proto) {
-            PDataRow** ptr = reinterpret_cast<PDataRow**>(data->get());
-            _proto_buffered_bytes -= (sizeof(PDataRow*) + (*ptr)->GetCachedSize());
+            auto row_ptr = std::move(_data_row_ptrs.front());
+            _proto_buffered_bytes -= (sizeof(PDataRow*) + row_ptr->GetCachedSize());
+            _data_row_ptrs.pop_front();
+            // PlainBinaryLineReader will hold the PDataRow
+            row_ptr.release();
         }
         _put_cond.notify_one();
         return Status::OK();
@@ -271,6 +284,7 @@ class StreamLoadPipe : public MessageBodySink, public FileReader {
     int64_t _total_length = -1;
     bool _use_proto = false;
     std::deque<ByteBufferPtr> _buf_queue;
+    std::deque<std::unique_ptr<PDataRow>> _data_row_ptrs;
     std::condition_variable _put_cond;
     std::condition_variable _get_cond;
 

be/src/service/internal_service.cpp

@@ -619,10 +619,14 @@ void PInternalServiceImpl::send_data(google::protobuf::RpcController* controller
         response->mutable_status()->add_error_msgs("pipe is null");
     } else {
         for (int i = 0; i < request->data_size(); ++i) {
-            PDataRow* row = new PDataRow();
+            std::unique_ptr<PDataRow> row(new PDataRow());
             row->CopyFrom(request->data(i));
-            pipe->append_and_flush(reinterpret_cast<char*>(&row), sizeof(row),
-                                   sizeof(row) + row->ByteSizeLong());
+            Status s = pipe->append(std::move(row));
+            if (!s.ok()) {
+                response->mutable_status()->set_status_code(1);
+                response->mutable_status()->add_error_msgs(s.to_string());
+                return;
+            }
         }
         response->mutable_status()->set_status_code(0);
     }

be/src/vec/exec/format/csv/csv_reader.cpp

@@ -402,16 +402,11 @@ Status CsvReader::_line_split_to_values(const Slice& line, bool* success) {
 void CsvReader::_split_line(const Slice& line) {
     _split_values.clear();
     if (_file_format_type == TFileFormatType::FORMAT_PROTO) {
-        PDataRow** ptr = reinterpret_cast<PDataRow**>(line.data);
-        PDataRow* row = *ptr;
-        for (const PDataColumn& col : (row)->col()) {
-            int len = col.value().size();
-            uint8_t* buf = new uint8_t[len];
-            memcpy(buf, col.value().c_str(), len);
-            _split_values.emplace_back(buf, len);
+        PDataRow** row_ptr = reinterpret_cast<PDataRow**>(line.data);
+        PDataRow* row = *row_ptr;
+        for (const PDataColumn& col : row->col()) {
+            _split_values.emplace_back(col.value());
         }
-        delete row;
-        delete[] ptr;
     } else {
         const char* value = line.data;
         size_t start = 0;     // point to the start pos of next col value.