[fix](be) Align CSV V2 enclosed field parsing

be/src/format/file_reader/new_plain_text_line_reader.cpp

@@ -46,11 +46,31 @@
 
 namespace doris {
 const uint8_t* EncloseCsvLineReaderCtx::read_line_impl(const uint8_t* start, const size_t length) {
+    if (_skip_utf8_bom && !_first_record_prefix_checked && _idx == 0) {
+        constexpr uint8_t UTF8_BOM[] = {0xEF, 0xBB, 0xBF};
+        constexpr size_t UTF8_BOM_SIZE = sizeof(UTF8_BOM);
+        const size_t prefix_size = std::min(length, UTF8_BOM_SIZE);
+        if (std::memcmp(start, UTF8_BOM, prefix_size) != 0) {
+            _first_record_prefix_checked = true;
+        } else if (length < UTF8_BOM_SIZE) {
+            // The input buffer can end inside the BOM. Wait for the remaining prefix bytes instead
+            // of feeding a partial BOM into START, which would permanently select NORMAL state.
+            return nullptr;
+        } else {
+            // Keep offsets relative to the original buffer. CsvReader removes these three bytes
+            // from the returned line and shifts separator positions by the same amount.
+            _idx = UTF8_BOM_SIZE;
+            _first_record_prefix_checked = true;
+        }
+    }
     // Avoid part bytes of the multi-char column separator have already been parsed,
     // causing parse column separator error.
     if (_state.curr_state == ReaderState::NORMAL ||
         _state.curr_state == ReaderState::MATCH_ENCLOSE) {
-        _idx -= std::min(_column_sep_len - 1, _idx);
+        const size_t last_column_sep_end =
+                _column_sep_positions.empty() ? 0 : _column_sep_positions.back() + _column_sep_len;
+        DORIS_CHECK_LE(last_column_sep_end, _idx);
+        _idx -= std::min(_column_sep_len - 1, _idx - last_column_sep_end);
     }
     _total_len = length;
     size_t bound = update_reading_bound(start);
@@ -134,7 +154,8 @@ void EncloseCsvLineReaderCtx::_on_start(const uint8_t* start, size_t& len) {
 
 void EncloseCsvLineReaderCtx::_on_normal(const uint8_t* start, size_t& len) {
     const uint8_t* curr_start = start + _idx;
-    size_t curr_len = len - _idx;
+    const size_t field_search_bound = _result == nullptr ? len : _result - start;
+    const size_t curr_len = field_search_bound - _idx;
     const uint8_t* col_sep_pos =
             find_col_sep_func(curr_start, curr_len, _column_sep.c_str(), _column_sep_len);
 
@@ -190,7 +211,8 @@ void EncloseCsvLineReaderCtx::_on_pre_match_enclose(const uint8_t* start, size_t
 
 void EncloseCsvLineReaderCtx::_on_match_enclose(const uint8_t* start, size_t& len) {
     const uint8_t* curr_start = start + _idx;
-    size_t curr_len = len - _idx;
+    const size_t field_search_bound = _result == nullptr ? len : _result - start;
+    const size_t curr_len = field_search_bound - _idx;
     const uint8_t* delim_pos =
             find_col_sep_func(curr_start, curr_len, _column_sep.c_str(), _column_sep_len);
 

be/src/format/file_reader/new_plain_text_line_reader.h

@@ -156,10 +156,12 @@ class EncloseCsvLineReaderCtx final : public BaseTextLineReaderContext<EncloseCs
     explicit EncloseCsvLineReaderCtx(const std::string& line_delimiter_,
                                      const size_t line_delimiter_len_, std::string column_sep_,
                                      const size_t column_sep_len_, size_t col_sep_num,
-                                     const char enclose, const char escape, const bool keep_cr_)
+                                     const char enclose, const char escape, const bool keep_cr_,
+                                     const bool skip_utf8_bom_ = false)
             : BaseTextLineReaderContext(line_delimiter_, line_delimiter_len_, keep_cr_),
               _enclose(enclose),
               _escape(escape),
+              _skip_utf8_bom(skip_utf8_bom_),
               _column_sep_len(column_sep_len_),
               _column_sep(std::move(column_sep_)) {
         if (column_sep_len_ == 1) {
@@ -210,6 +212,11 @@ class EncloseCsvLineReaderCtx final : public BaseTextLineReaderContext<EncloseCs
     ReaderStateWrapper _state;
     const char _enclose;
     const char _escape;
+    // Only V2 readers starting at file offset zero enable this. The line reader must recognize the
+    // BOM before deciding whether the first field is enclosed; the format reader later removes the
+    // same bytes from the returned Slice and adjusts the recorded separator positions.
+    const bool _skip_utf8_bom;
+    bool _first_record_prefix_checked = false;
     const uint8_t* _result = nullptr;
 
     size_t _total_len;

be/src/format/hive_text_util.h

@@ -0,0 +1,33 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <cstddef>
+
+namespace doris {
+
+inline bool is_hive_text_separator_escaped(const char* data, size_t separator_pos,
+                                           char escape_char) {
+    size_t escape_count = 0;
+    while (separator_pos > escape_count && data[separator_pos - escape_count - 1] == escape_char) {
+        ++escape_count;
+    }
+    return escape_count % 2 == 1;
+}
+
+} // namespace doris

be/src/format/text/text_reader.cpp

@@ -34,6 +34,7 @@
 #include "exec/scan/scanner.h"
 #include "format/csv/csv_reader.h"
 #include "format/file_reader/new_plain_text_line_reader.h"
+#include "format/hive_text_util.h"
 #include "format/line_reader.h"
 #include "io/file_factory.h"
 #include "io/fs/buffered_reader.h"
@@ -60,7 +61,7 @@ void HiveTextFieldSplitter::_split_field_single_char(const Slice& line,
     for (size_t i = 0; i < size; ++i) {
         if (data[i] == _value_sep[0]) {
             // hive will escape the field separator in string
-            if (_escape_char != 0 && i > 0 && data[i - 1] == _escape_char) {
+            if (_escape_char != 0 && is_hive_text_separator_escaped(data, i, _escape_char)) {
                 continue;
             }
             process_value_func(data, value_start, i - value_start, _trimming_char, splitted_values);
@@ -98,7 +99,7 @@ void HiveTextFieldSplitter::_split_field_multi_char(const Slice& line,
         }
         if (j == (int)_value_sep_len - 1) {
             size_t curpos = i - _value_sep_len + 1;
-            if (_escape_char != 0 && curpos > 0 && data[curpos - 1] == _escape_char) {
+            if (_escape_char != 0 && is_hive_text_separator_escaped(data, curpos, _escape_char)) {
                 j = next[j];
                 continue;
             }

be/src/format_v2/delimited_text/csv_reader.cpp

@@ -134,14 +134,13 @@ Status CsvReader::_create_line_reader() {
             text_line_reader_ctx = std::make_shared<PlainTextLineReaderCtx>(
                     _line_delimiter, _line_delimiter.size(), _keep_cr);
         } else {
-            // The enclosed-line context finds logical records that may span physical newlines.
-            // Field slicing still happens in `_split_line()` because the v2 scan request may ask
-            // for CSV ordinals in a different order from the physical file.
             const size_t col_sep_num =
                     _source_file_slot_descs.size() > 1 ? _source_file_slot_descs.size() - 1 : 0;
-            text_line_reader_ctx = std::make_shared<EncloseCsvLineReaderCtx>(
+            _enclose_reader_ctx = std::make_shared<EncloseCsvLineReaderCtx>(
                     _line_delimiter, _line_delimiter.size(), _value_separator,
-                    _value_separator.size(), col_sep_num, _enclose, _escape, _keep_cr);
+                    _value_separator.size(), col_sep_num, _enclose, _escape, _keep_cr,
+                    _start_offset == 0);
+            text_line_reader_ctx = _enclose_reader_ctx;
         }
         _line_reader = NewPlainTextLineReader::create_unique(
                 _profile, _file_reader, _decompressor.get(), std::move(text_line_reader_ctx), _size,
@@ -174,77 +173,40 @@ void CsvReader::_split_line(const Slice& line) {
         return;
     }
 
-    // The text line reader is responsible for split boundaries and multi-line quoted fields.
-    // Field slicing still happens here because FileScannerV2 asks columns by file-local id, so we
-    // must be able to materialize only the requested CSV ordinals without building a row object.
-    // Example: for `1,"a,b",10` and column separator `,`, this loop returns three slices:
-    // `1`, `a,b`, and `10`; the comma inside quotes does not create an extra field.
-    bool in_quote = false;
-    bool escaped = false;
-    size_t start = 0;
-    size_t i = 0;
-    while (i < line.size) {
-        const char ch = line.data[i];
-        if (_enclose != 0) {
-            if (escaped) {
-                escaped = false;
-                ++i;
-                continue;
-            }
-            if (_escape != 0 && ch == _escape) {
-                escaped = true;
-                ++i;
-                continue;
-            }
-            if (ch == _enclose) {
-                if (in_quote && i + 1 < line.size && line.data[i + 1] == _enclose) {
-                    i += 2;
-                    continue;
-                }
-                in_quote = !in_quote;
-                ++i;
-                continue;
-            }
+    const auto append_value = [&](size_t value_start, size_t value_len) {
+        while (_trim_tailing_spaces && value_len > 0 &&
+               line.data[value_start + value_len - 1] == ' ') {
+            --value_len;
         }
-        if (!in_quote && starts_with_at(line, i, _value_separator)) {
-            size_t value_start = start;
-            size_t value_len = i - start;
-            while (_trim_tailing_spaces && value_len > 0 &&
-                   line.data[value_start + value_len - 1] == ' ') {
-                --value_len;
-            }
-            if (_trim_double_quotes && value_len > 1 && line.data[value_start] == '"' &&
-                line.data[value_start + value_len - 1] == '"') {
-                ++value_start;
-                value_len -= 2;
-            } else if (_enclose != 0 && value_len > 1 && line.data[value_start] == _enclose &&
-                       line.data[value_start + value_len - 1] == _enclose) {
-                ++value_start;
-                value_len -= 2;
-            }
-            _split_values.emplace_back(line.data + value_start, value_len);
-            i += _value_separator.size();
-            start = i;
-            continue;
+        if (_enclose != 0 && value_len > 1 && line.data[value_start] == _enclose &&
+            line.data[value_start + value_len - 1] == _enclose) {
+            ++value_start;
+            value_len -= 2;
         }
-        ++i;
-    }
+        _split_values.emplace_back(line.data + value_start, value_len);
+    };
 
-    size_t value_start = start;
-    size_t value_len = line.size - start;
-    while (_trim_tailing_spaces && value_len > 0 && line.data[value_start + value_len - 1] == ' ') {
-        --value_len;
-    }
-    if (_trim_double_quotes && value_len > 1 && line.data[value_start] == '"' &&
-        line.data[value_start + value_len - 1] == '"') {
-        ++value_start;
-        value_len -= 2;
-    } else if (_enclose != 0 && value_len > 1 && line.data[value_start] == _enclose &&
-               line.data[value_start + value_len - 1] == _enclose) {
-        ++value_start;
-        value_len -= 2;
+    size_t value_start = 0;
+    if (_enclose_reader_ctx != nullptr) {
+        for (const size_t separator_position : _enclose_reader_ctx->column_sep_positions()) {
+            DORIS_CHECK_LE(value_start, separator_position);
+            DORIS_CHECK_LE(separator_position, line.size);
+            append_value(value_start, separator_position - value_start);
+            value_start = separator_position + _value_separator.size();
+        }
+    } else {
+        for (size_t i = 0; i < line.size;) {
+            if (starts_with_at(line, i, _value_separator)) {
+                append_value(value_start, i - value_start);
+                i += _value_separator.size();
+                value_start = i;
+            } else {
+                ++i;
+            }
+        }
     }
-    _split_values.emplace_back(line.data + value_start, value_len);
+    DORIS_CHECK_LE(value_start, line.size);
+    append_value(value_start, line.size - value_start);
 }
 
 Status CsvReader::_deserialize_one_cell(const RequestedColumn& column, IColumn* output,
@@ -261,7 +223,8 @@ Status CsvReader::_deserialize_one_cell(const RequestedColumn& column, IColumn*
         // CSV keeps empty-field handling separate from null_format matching. An empty
         // null_format must not turn every empty CSV field into NULL unless FE explicitly sets
         // empty_field_as_null; OpenCSV-compatible tables expect empty fields to stay empty strings.
-        if (_options.null_len > 0 && value.size == _options.null_len &&
+        const bool quoted = _options.converted_from_string && value.trim_double_quotes();
+        if (!quoted && _options.null_len > 0 && value.size == _options.null_len &&
             std::memcmp(value.data, _options.null_format, value.size) == 0) {
             null_column.insert_data(nullptr, 0);
             return Status::OK();
@@ -292,4 +255,10 @@ bool CsvReader::_can_split() const {
             _file_format_type == TFileFormatType::FORMAT_CSV_PLAIN);
 }
 
+void CsvReader::_on_bom_removed(size_t bom_size) {
+    if (_enclose_reader_ctx != nullptr) {
+        _enclose_reader_ctx->adjust_column_sep_positions(bom_size);
+    }
+}
+
 } // namespace doris::format::csv

be/src/format_v2/delimited_text/csv_reader.h

@@ -25,6 +25,7 @@
 #include "util/slice.h"
 
 namespace doris {
+class EncloseCsvLineReaderCtx;
 class SlotDescriptor;
 } // namespace doris
 
@@ -61,13 +62,15 @@ class CsvReader final : public ::doris::format::DelimitedTextReader {
                                  Slice value) override;
     Slice _normalize_value(Slice value) const override;
     bool _can_split() const override;
+    void _on_bom_removed(size_t bom_size) override;
 
     TFileFormatType::type _file_format_type = TFileFormatType::FORMAT_CSV_PLAIN;
     char _enclose = 0;
     bool _trim_double_quotes = false;
     bool _trim_tailing_spaces = false;
     bool _empty_field_as_null = false;
     bool _keep_cr = false;
+    std::shared_ptr<EncloseCsvLineReaderCtx> _enclose_reader_ctx;
 };
 
 } // namespace doris::format::csv

be/src/format_v2/delimited_text/delimited_text_reader.cpp

@@ -624,6 +624,10 @@ bool DelimitedTextReader::_can_split() const {
     return _file_compress_type == TFileCompressType::PLAIN;
 }
 
+void DelimitedTextReader::_on_bom_removed(size_t bom_size) {
+    (void)bom_size;
+}
+
 Status DelimitedTextReader::_append_null(IColumn* output) {
     DORIS_CHECK(output != nullptr);
     auto* nullable = assert_cast<ColumnNullable*>(output);
@@ -635,8 +639,10 @@ const uint8_t* DelimitedTextReader::_remove_bom(const uint8_t* ptr, size_t* size
     DORIS_CHECK(size != nullptr);
     if (ptr != nullptr && *size >= 3 && static_cast<uint8_t>(ptr[0]) == 0xEF &&
         static_cast<uint8_t>(ptr[1]) == 0xBB && static_cast<uint8_t>(ptr[2]) == 0xBF) {
-        *size -= 3;
-        return ptr + 3;
+        constexpr size_t BOM_SIZE = 3;
+        *size -= BOM_SIZE;
+        _on_bom_removed(BOM_SIZE);
+        return ptr + BOM_SIZE;
     }
     return ptr;
 }

be/src/format_v2/delimited_text/delimited_text_reader.h

@@ -123,6 +123,8 @@ class DelimitedTextReader : public FileReader {
     // Whether this file can start at a non-zero split offset. Compressed delimited files cannot be
     // split because the decompressor needs the stream from the beginning.
     virtual bool _can_split() const;
+    // Let formats adjust parser metadata that was computed before the common reader removed a BOM.
+    virtual void _on_bom_removed(size_t bom_size);
 
     Status _append_null(IColumn* output);
     // Match the generic nullable serde semantics exactly: a field is NULL when its raw slice is

be/src/format_v2/delimited_text/text_reader.cpp

@@ -25,6 +25,7 @@
 #include "core/data_type/data_type_string.h"
 #include "core/data_type_serde/data_type_string_serde.h"
 #include "format/file_reader/new_plain_text_line_reader.h"
+#include "format/hive_text_util.h"
 #include "runtime/descriptors.h"
 #include "util/decompressor.h"
 
@@ -104,7 +105,7 @@ void TextReader::_split_line_single_char(const Slice& line) {
         if (line.data[i] == _value_separator[0]) {
             // Hive text lets a string escape the field separator. The backslash remains in the
             // field slice so deserialize_one_cell_from_hive_text() can unescape the final value.
-            if (_escape != 0 && i > 0 && line.data[i - 1] == _escape) {
+            if (_escape != 0 && is_hive_text_separator_escaped(line.data, i, _escape)) {
                 continue;
             }
             _split_values.emplace_back(line.data + value_start, i - value_start);
@@ -119,7 +120,7 @@ void TextReader::_split_line_multi_char(const Slice& line) {
     size_t i = 0;
     while (i < line.size) {
         if (starts_with_at(line, i, _value_separator)) {
-            if (_escape != 0 && i > 0 && line.data[i - 1] == _escape) {
+            if (_escape != 0 && is_hive_text_separator_escaped(line.data, i, _escape)) {
                 ++i;
                 continue;
             }

be/test/format/text/hive_text_field_splitter_test.cpp

@@ -81,7 +81,9 @@ TEST_F(HiveTextFieldSplitterTest, OverlappingPatterns) {
 // Test escape character functionality
 TEST_F(HiveTextFieldSplitterTest, EscapeCharacter) {
     verify_field_split("a\\,b,c", ",", {"a\\,b", "c"}, '\\');
+    verify_field_split(R"(a\\,b)", ",", {R"(a\\)", "b"}, '\\');
     verify_field_split("a\\||b||c", "||", {"a\\||b", "c"}, '\\');
+    verify_field_split(R"(a\\||b)", "||", {R"(a\\)", "b"}, '\\');
     verify_field_split("field1\\|+|field2|+|field3", "|+|", {"field1\\|+|field2", "field3"}, '\\');
 }
 
@@ -94,4 +96,4 @@ TEST_F(HiveTextFieldSplitterTest, RealWorldScenarios) {
     verify_field_split("a|+||+|c", "|+|", {"a", "", "c"});
 }
 
-} // namespace doris
+} // namespace doris