[feature](load) refactor CSV reading process during scanning, and support enclose and escape for stream load

be/src/exec/line_reader.h

@@ -24,7 +24,7 @@ namespace doris {
 namespace io {
 class IOContext;
 }
-// This class is used for CSV scanner, to read content line by line
+// This class is used to read content line by line
 class LineReader {
 public:
     virtual ~LineReader() = default;

be/src/exec/text_converter.cpp

@@ -52,9 +52,12 @@ TextConverter::TextConverter(char escape_char, char collection_delimiter, char m
 
 void TextConverter::write_string_column(const SlotDescriptor* slot_desc,
                                         vectorized::MutableColumnPtr* column_ptr, const char* data,
-                                        size_t len) {
+                                        size_t len, bool need_escape) {
     DCHECK(column_ptr->get()->is_nullable());
     auto* nullable_column = reinterpret_cast<vectorized::ColumnNullable*>(column_ptr->get());
+    if (need_escape) {
+        unescape_string_on_spot(data, &len);
+    }
     if ((len == 2 && data[0] == '\\' && data[1] == 'N') || len == SQL_NULL_DATA) {
         nullable_column->get_null_map_data().push_back(1);
         reinterpret_cast<vectorized::ColumnString&>(nullable_column->get_nested_column())

be/src/exec/text_converter.h

@@ -17,7 +17,7 @@
 
 #pragma once
 
-#include <stddef.h>
+#include <cstddef>
 
 #include "vec/columns/column.h"
 
@@ -33,9 +33,15 @@ class TextConverter {
 
     TextConverter(char escape_char, char collection_delimiter = '\2', char map_kv_delimiter = '\3');
 
+    inline void write_string_column(const SlotDescriptor* slot_desc,
+                                    vectorized::MutableColumnPtr* column_ptr, const char* data,
+                                    size_t len) {
+        return write_string_column(slot_desc, column_ptr, data, len, false);
+    }
+
     void write_string_column(const SlotDescriptor* slot_desc,
-                             vectorized::MutableColumnPtr* column_ptr, const char* data,
-                             size_t len);
+                             vectorized::MutableColumnPtr* column_ptr, const char* data, size_t len,
+                             bool need_escape);
 
     inline bool write_column(const SlotDescriptor* slot_desc,
                              vectorized::MutableColumnPtr* column_ptr, const char* data, size_t len,
@@ -62,6 +68,8 @@ class TextConverter {
     }
     void set_map_kv_delimiter(char mapkv_delimiter) { _map_kv_delimiter = mapkv_delimiter; }
 
+    inline void set_escape_char(const char escape) { this->_escape_char = escape; }
+
 private:
     bool _write_data(const TypeDescriptor& type_desc, vectorized::IColumn* nullable_col_ptr,
                      const char* data, size_t len, bool copy_string, bool need_escape, size_t rows,

be/src/http/action/stream_load.cpp

@@ -375,6 +375,12 @@ Status StreamLoadAction::_process_put(HttpRequest* http_req,
     if (!http_req->header(HTTP_LINE_DELIMITER).empty()) {
         request.__set_line_delimiter(http_req->header(HTTP_LINE_DELIMITER));
     }
+    if (!http_req->header(HTTP_ENCLOSE).empty() && http_req->header(HTTP_ENCLOSE).size() > 0) {
+        request.__set_enclose(http_req->header(HTTP_ENCLOSE)[0]);
+    }
+    if (!http_req->header(HTTP_ESCAPE).empty() && http_req->header(HTTP_ESCAPE).size() > 0) {
+        request.__set_escape(http_req->header(HTTP_ESCAPE)[0]);
+    }
     if (!http_req->header(HTTP_PARTITIONS).empty()) {
         request.__set_partitions(http_req->header(HTTP_PARTITIONS));
         request.__set_isTempPartition(false);

be/src/http/http_common.h

@@ -29,6 +29,8 @@ static const std::string HTTP_COLUMNS = "columns";
 static const std::string HTTP_WHERE = "where";
 static const std::string HTTP_COLUMN_SEPARATOR = "column_separator";
 static const std::string HTTP_LINE_DELIMITER = "line_delimiter";
+static const std::string HTTP_ENCLOSE = "enclose";
+static const std::string HTTP_ESCAPE = "escape";
 static const std::string HTTP_MAX_FILTER_RATIO = "max_filter_ratio";
 static const std::string HTTP_TIMEOUT = "timeout";
 static const std::string HTTP_PARTITIONS = "partitions";